Browse Source

使用宏来开关是否支持多线程多调度器

tags/v2.9.7
tearshark 6 years ago
parent
commit
9212bfcdf7

+ 4
- 4
librf/src/channel.h View File

template<class _Ty> template<class _Ty>
struct channel_impl : public std::enable_shared_from_this<channel_impl<_Ty>> struct channel_impl : public std::enable_shared_from_this<channel_impl<_Ty>>
{ {
typedef _awaker<channel_impl<_Ty>, _Ty *, future_error> channel_read_awaker;
typedef _awaker<channel_impl<_Ty>, _Ty *, error_code> channel_read_awaker;
typedef std::shared_ptr<channel_read_awaker> channel_read_awaker_ptr; typedef std::shared_ptr<channel_read_awaker> channel_read_awaker_ptr;
typedef _awaker<channel_impl<_Ty>> channel_write_awaker; typedef _awaker<channel_impl<_Ty>> channel_write_awaker;
auto val = std::move(_values.front()); auto val = std::move(_values.front());
_values.pop_front(); _values.pop_front();
r_awaker->awake(this, 1, &val, future_error::none);
r_awaker->awake(this, 1, &val, error_code::none);
ret_value = true; ret_value = true;
} }
else else
auto r_awaker = *iter; auto r_awaker = *iter;
iter = _read_awakes.erase(iter); iter = _read_awakes.erase(iter);
if (r_awaker->awake(this, 1, _values.size() ? &_values.front() : nullptr, future_error::read_before_write))
if (r_awaker->awake(this, 1, _values.size() ? &_values.front() : nullptr, error_code::read_before_write))
{ {
if(_values.size()) _values.pop_front(); if(_values.size()) _values.pop_front();
awaitable_t<_Ty> awaitable; awaitable_t<_Ty> awaitable;
auto awaker = std::make_shared<channel_read_awaker>( auto awaker = std::make_shared<channel_read_awaker>(
[st = awaitable._state](channel_impl_type *, _Ty * val, future_error fe) -> bool
[st = awaitable._state](channel_impl_type *, _Ty * val, error_code fe) -> bool
{ {
if(val) if(val)
st->set_value(std::move(*val)); st->set_value(std::move(*val));

+ 9
- 16
librf/src/def.h View File

#define _coro_promise_ptr(T) _coro_promise_ptr__<resumef::promise_t<T> >(_coro_frame_ptr()) #define _coro_promise_ptr(T) _coro_promise_ptr__<resumef::promise_t<T> >(_coro_frame_ptr())
enum struct future_error
enum struct error_code
{ {
none, none,
not_ready, // get_value called when value not available not_ready, // get_value called when value not available
max__ max__
}; };
const char * get_error_string(future_error fe, const char * classname);
const char * get_error_string(error_code fe, const char * classname);
//const char * future_error_string[size_t(future_error::max__)]; //const char * future_error_string[size_t(future_error::max__)];
struct future_exception : std::exception struct future_exception : std::exception
{ {
future_error _error;
future_exception(future_error fe)
error_code _error;
future_exception(error_code fe)
: exception(get_error_string(fe, "future_exception")) : exception(get_error_string(fe, "future_exception"))
, _error(fe) , _error(fe)
{ {
struct lock_exception : std::exception struct lock_exception : std::exception
{ {
future_error _error;
lock_exception(future_error fe)
error_code _error;
lock_exception(error_code fe)
: exception(get_error_string(fe, "lock_exception")) : exception(get_error_string(fe, "lock_exception"))
, _error(fe) , _error(fe)
{ {
struct channel_exception : std::exception struct channel_exception : std::exception
{ {
future_error _error;
channel_exception(future_error fe)
error_code _error;
channel_exception(error_code fe)
: exception(get_error_string(fe, "channel_exception")) : exception(get_error_string(fe, "channel_exception"))
, _error(fe) , _error(fe)
{ {
struct state_base; struct state_base;
//获得当前线程下的调度器 //获得当前线程下的调度器
extern scheduler * this_scheduler();
//获得当前线程下,正在由调度器调度的协程
//extern state_base * this_coroutine();
//namespace detail
//{
// extern state_base * current_coroutine();
//}
scheduler * this_scheduler();
} }
#define co_yield_void co_yield nullptr #define co_yield_void co_yield nullptr

+ 8
- 2
librf/src/future.h View File

promise_impl_t() promise_impl_t()
: _state(make_counted<state_type>()) : _state(make_counted<state_type>())
{ {
#if RESUMEF_ENABLE_MULT_SCHEDULER
_state->this_promise(this); _state->this_promise(this);
#endif
} }
promise_impl_t(promise_impl_t&& _Right) promise_impl_t(promise_impl_t&& _Right)
: _state(std::move(_Right._state)) : _state(std::move(_Right._state))
{ {
#if RESUMEF_ENABLE_MULT_SCHEDULER
_state->this_promise(this); _state->this_promise(this);
#endif
} }
promise_impl_t & operator = (promise_impl_t&& _Right) promise_impl_t & operator = (promise_impl_t&& _Right)
{ {
if (this != _Right) if (this != _Right)
{ {
_state = std::move(_Right._state); _state = std::move(_Right._state);
#if RESUMEF_ENABLE_MULT_SCHEDULER
_state->this_promise(this); _state->this_promise(this);
#endif
} }
return *this; return *this;
} }
using awaitable_vt = awaitable_t<void>; using awaitable_vt = awaitable_t<void>;
#if RESUMEF_ENABLE_MULT_SCHEDULER
inline promise_t<void> * state_base::parent_promise() const inline promise_t<void> * state_base::parent_promise() const
{ {
if (_coro) return _coro_promise_ptr__<promise_t<void>>(_coro.address()); if (_coro) return _coro_promise_ptr__<promise_t<void>>(_coro.address());
return nullptr; return nullptr;
} }
/*
inline scheduler * state_base::parent_scheduler() const inline scheduler * state_base::parent_scheduler() const
{ {
auto promise_ = parent_promise(); auto promise_ = parent_promise();
return promise_->_state->current_scheduler(); return promise_->_state->current_scheduler();
return nullptr; return nullptr;
} }
*/
#endif
} }

+ 6
- 0
librf/src/rf_task.h View File

virtual bool go_next(scheduler *) = 0; virtual bool go_next(scheduler *) = 0;
virtual void cancel() = 0; virtual void cancel() = 0;
virtual void * get_id() = 0; virtual void * get_id() = 0;
#if RESUMEF_ENABLE_MULT_SCHEDULER
virtual void bind(scheduler *) = 0; virtual void bind(scheduler *) = 0;
#endif
}; };
//---------------------------------------------------------------------------------------------- //----------------------------------------------------------------------------------------------
{ {
return nullptr; return nullptr;
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
virtual void bind(scheduler *) override virtual void bind(scheduler *) override
{ {
} }
#endif
}; };
template<class _Ty> template<class _Ty>
{ {
return _future._state.get(); return _future._state.get();
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
virtual void bind(scheduler * schdler) override virtual void bind(scheduler * schdler) override
{ {
auto * _state = _future._state.get(); auto * _state = _future._state.get();
_state->current_scheduler(schdler); _state->current_scheduler(schdler);
} }
#endif
}; };
//---------------------------------------------------------------------------------------------- //----------------------------------------------------------------------------------------------

+ 14
- 17
librf/src/scheduler.cpp View File

namespace resumef namespace resumef
{ {
static const char * future_error_string[(size_t)future_error::max__]
static const char * future_error_string[(size_t)error_code::max__]
{ {
"none", "none",
"not_ready", "not_ready",
static char sz_future_error_buffer[256]; static char sz_future_error_buffer[256];
const char * get_error_string(future_error fe, const char * classname)
const char * get_error_string(error_code fe, const char * classname)
{ {
if (classname) if (classname)
{ {
return future_error_string[(size_t)(fe)]; return future_error_string[(size_t)(fe)];
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
thread_local scheduler * th_scheduler_ptr = nullptr; thread_local scheduler * th_scheduler_ptr = nullptr;
//获得当前线程下的调度器 //获得当前线程下的调度器
{ {
return th_scheduler_ptr ? th_scheduler_ptr : &scheduler::g_scheduler; return th_scheduler_ptr ? th_scheduler_ptr : &scheduler::g_scheduler;
} }
#endif
//获得当前线程下,正在由调度器调度的协程
/*
namespace detail
{
state_base * current_coroutine()
{
scheduler * schdler = this_scheduler();
if (schdler->current_state)
return schdler->current_state;
return schdler->top_state();
}
}
*/
local_scheduler::local_scheduler() local_scheduler::local_scheduler()
{ {
#if RESUMEF_ENABLE_MULT_SCHEDULER
if (th_scheduler_ptr == nullptr) if (th_scheduler_ptr == nullptr)
{ {
_scheduler_ptr = new scheduler; _scheduler_ptr = new scheduler;
th_scheduler_ptr = _scheduler_ptr; th_scheduler_ptr = _scheduler_ptr;
} }
#endif
} }
local_scheduler::~local_scheduler() local_scheduler::~local_scheduler()
{ {
#if RESUMEF_ENABLE_MULT_SCHEDULER
if (th_scheduler_ptr == _scheduler_ptr) if (th_scheduler_ptr == _scheduler_ptr)
th_scheduler_ptr = nullptr; th_scheduler_ptr = nullptr;
delete _scheduler_ptr; delete _scheduler_ptr;
#endif
} }
scheduler::scheduler() scheduler::scheduler()
scheduler::~scheduler() scheduler::~scheduler()
{ {
cancel_all_task_(); cancel_all_task_();
#if RESUMEF_ENABLE_MULT_SCHEDULER
if (th_scheduler_ptr == this) if (th_scheduler_ptr == this)
th_scheduler_ptr = nullptr; th_scheduler_ptr = nullptr;
#endif
} }
void scheduler::new_task(task_base * task) void scheduler::new_task(task_base * task)
if (task) if (task)
{ {
scoped_lock<std::recursive_mutex> __guard(_mtx_ready); scoped_lock<std::recursive_mutex> __guard(_mtx_ready);
#if RESUMEF_ENABLE_MULT_SCHEDULER
task->bind(this); task->bind(this);
#endif
this->_ready_task.push_back(task); this->_ready_task.push_back(task);
} }
} }
void scheduler::run_one_batch() void scheduler::run_one_batch()
{ {
#if RESUMEF_ENABLE_MULT_SCHEDULER
if (th_scheduler_ptr == nullptr) if (th_scheduler_ptr == nullptr)
th_scheduler_ptr = this; th_scheduler_ptr = this;
#endif
{ {
scoped_lock<std::recursive_mutex> __guard(_mtx_task); scoped_lock<std::recursive_mutex> __guard(_mtx_task);

+ 9
- 0
librf/src/scheduler.h View File

local_scheduler & operator = (local_scheduler && right_) = delete; local_scheduler & operator = (local_scheduler && right_) = delete;
local_scheduler(const local_scheduler &) = delete; local_scheduler(const local_scheduler &) = delete;
local_scheduler & operator = (const local_scheduler &) = delete; local_scheduler & operator = (const local_scheduler &) = delete;
#if RESUMEF_ENABLE_MULT_SCHEDULER
private: private:
scheduler * _scheduler_ptr; scheduler * _scheduler_ptr;
#endif
}; };
//-------------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------------
#if !RESUMEF_ENABLE_MULT_SCHEDULER
//获得当前线程下的调度器
inline scheduler * this_scheduler()
{
return &scheduler::g_scheduler;
}
#endif
#if !defined(_DISABLE_RESUMEF_GO_MACRO) #if !defined(_DISABLE_RESUMEF_GO_MACRO)
#define go (*::resumef::this_scheduler()) + #define go (*::resumef::this_scheduler()) +

+ 16
- 12
librf/src/state.cpp View File

{ {
return _state.get(); return _state.get();
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
virtual void bind(scheduler * ) override virtual void bind(scheduler * ) override
{ {
} }
#endif
}; };
state_base::~state_base() state_base::~state_base()


if (_coro) if (_coro)
{ {
// auto sch_ = this->current_scheduler();
auto sch_ = this_scheduler();
/*
#if RESUMEF_ENABLE_MULT_SCHEDULER
auto sch_ = this->current_scheduler();
if (sch_ == nullptr) if (sch_ == nullptr)
sch_ = this_scheduler(); sch_ = this_scheduler();
*/
#else
auto sch_ = this_scheduler();
#endif
sch_->push_task_internal(new awaitable_task_t<state_base>(this)); sch_->push_task_internal(new awaitable_task_t<state_base>(this));
} }
} }


if (_coro) if (_coro)
{ {
// auto sch_ = this->current_scheduler();
auto sch_ = this_scheduler();
/*
#if RESUMEF_ENABLE_MULT_SCHEDULER
auto sch_ = this->current_scheduler();
if (sch_ == nullptr) if (sch_ == nullptr)
sch_ = this_scheduler(); sch_ = this_scheduler();
*/
#else
auto sch_ = this_scheduler();
#endif
sch_->push_task_internal(new awaitable_task_t<state_base>(this)); sch_->push_task_internal(new awaitable_task_t<state_base>(this));
} }
} }
_coro = resume_cb; _coro = resume_cb;
/*
#if RESUMEF_ENABLE_MULT_SCHEDULER
if (_current_scheduler == nullptr) if (_current_scheduler == nullptr)
{ {
auto * promise_ = this->parent_promise(); auto * promise_ = this->parent_promise();
stptr->current_scheduler(_current_scheduler); stptr->current_scheduler(_current_scheduler);
_depend_states.clear(); _depend_states.clear();
} }
*/
#endif
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
void state_base::current_scheduler(scheduler * sch_) void state_base::current_scheduler(scheduler * sch_)
{ {
/*
scoped_lock<lock_type> __guard(_mtx); scoped_lock<lock_type> __guard(_mtx);
_current_scheduler = sch_; _current_scheduler = sch_;
for (auto & stptr : _depend_states) for (auto & stptr : _depend_states)
stptr->current_scheduler(sch_); stptr->current_scheduler(sch_);
_depend_states.clear(); _depend_states.clear();
*/
} }
#endif
} }

+ 16
- 6
librf/src/state.h View File

typedef std::recursive_mutex lock_type; typedef std::recursive_mutex lock_type;
lock_type _mtx; //for value, _exception lock_type _mtx; //for value, _exception
RF_API void set_value_none_lock(); RF_API void set_value_none_lock();
#if RESUMEF_ENABLE_MULT_SCHEDULER
private: private:
void * _this_promise = nullptr; void * _this_promise = nullptr;
scheduler * _current_scheduler = nullptr; scheduler * _current_scheduler = nullptr;
std::vector<counted_ptr<state_base>> _depend_states; std::vector<counted_ptr<state_base>> _depend_states;
#endif
public: public:
coroutine_handle<> _coro; coroutine_handle<> _coro;
std::atomic<intptr_t> _count = 0; // tracks reference count of state object std::atomic<intptr_t> _count = 0; // tracks reference count of state object
{ {
#if RESUMEF_DEBUG_COUNTER #if RESUMEF_DEBUG_COUNTER
{ {
scoped_lock<lock_type> __lock(g_resumef_cout_mutex);
scoped_lock<std::mutex> __lock(g_resumef_cout_mutex);
std::cout
<< "coro=" << _coro.address()
<< ",thread=" << std::this_thread::get_id()
#if RESUMEF_ENABLE_MULT_SCHEDULER
std::cout << "scheduler=" << current_scheduler()
<< ",coro=" << _coro.address()
<< ",scheduler=" << current_scheduler()
<< ",this_promise=" << this_promise() << ",this_promise=" << this_promise()
<< ",parent_promise=" << parent_promise() << ",parent_promise=" << parent_promise()
<< ",thread=" << std::this_thread::get_id()
#endif
<< std::endl; << std::endl;
} }
#endif #endif
delete this; delete this;
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
promise_t<void> * parent_promise() const; promise_t<void> * parent_promise() const;
//scheduler * parent_scheduler() const; //scheduler * parent_scheduler() const;
return _current_scheduler; return _current_scheduler;
} }
void current_scheduler(scheduler * sch_); void current_scheduler(scheduler * sch_);
#endif
//------------------------------------------------------------------------------------------ //------------------------------------------------------------------------------------------
//以下是通过future_t/promise_t, 与编译器生成的resumable function交互的接口 //以下是通过future_t/promise_t, 与编译器生成的resumable function交互的接口
scoped_lock<lock_type> __guard(_mtx); scoped_lock<lock_type> __guard(_mtx);
if (!_ready) if (!_ready)
throw future_exception{ future_error::not_ready };
throw future_exception{ error_code::not_ready };
return _value; return _value;
} }
void reset() void reset()
scoped_lock<lock_type> __guard(_mtx); scoped_lock<lock_type> __guard(_mtx);
if (!_ready) if (!_ready)
throw future_exception{ future_error::not_ready };
throw future_exception{ error_code::not_ready };
} }
void reset() void reset()
{ {
reset_none_lock(); reset_none_lock();
} }
#if RESUMEF_ENABLE_MULT_SCHEDULER
promise_t<void> * parent_promise() const promise_t<void> * parent_promise() const
{ {
return reinterpret_cast<promise_t<void> *>(state_base::parent_promise()); return reinterpret_cast<promise_t<void> *>(state_base::parent_promise());
} }
#endif
}; };
} }

+ 7
- 7
tutorial/test_async_channel_mult_thread.cpp View File

{ {
channel_t<std::string> c(MAX_CHANNEL_QUEUE); channel_t<std::string> c(MAX_CHANNEL_QUEUE);
std::thread wth([&]
std::thread write_th([&]
{ {
//local_scheduler my_scheduler; //2017/11/27日,仍然存在BUG。真多线程下调度,存在有协程无法被调度完成的BUG //local_scheduler my_scheduler; //2017/11/27日,仍然存在BUG。真多线程下调度,存在有协程无法被调度完成的BUG
go test_channel_producer(c, BATCH * N); go test_channel_producer(c, BATCH * N);
this_scheduler()->run_until_notask(); this_scheduler()->run_until_notask();
std::cout << "Write OK" << std::endl;
std::cout << "Write OK\r\n";
}); });
//std::this_thread::sleep_for(100ms); //std::this_thread::sleep_for(100ms);
std::thread rth[N];
std::thread read_th[N];
for (size_t i = 0; i < N; ++i) for (size_t i = 0; i < N; ++i)
{ {
rth[i] = std::thread([&]
read_th[i] = std::thread([&]
{ {
//local_scheduler my_scheduler; //2017/11/27日,仍然存在BUG。真多线程下调度,存在有协程无法被调度完成的BUG //local_scheduler my_scheduler; //2017/11/27日,仍然存在BUG。真多线程下调度,存在有协程无法被调度完成的BUG
go test_channel_consumer(c, BATCH); go test_channel_consumer(c, BATCH);
this_scheduler()->run_until_notask(); this_scheduler()->run_until_notask();
std::cout << "Read OK" << std::endl;
std::cout << "Read OK\r\n";
}); });
} }
for(auto & th : rth)
for(auto & th : read_th)
th.join(); th.join();
wth.join();
write_th.join();
std::cout << "OK" << std::endl; std::cout << "OK" << std::endl;
_getch(); _getch();

+ 1
- 1
vs_proj/librf.vcxproj View File

<ClCompile> <ClCompile>
<WarningLevel>Level3</WarningLevel> <WarningLevel>Level3</WarningLevel>
<Optimization>Disabled</Optimization> <Optimization>Disabled</Optimization>
<PreprocessorDefinitions>_DEBUG;_CONSOLE;RESUMEF_DEBUG_COUNTER=0;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<PreprocessorDefinitions>_DEBUG;_CONSOLE;RESUMEF_DEBUG_COUNTER=1;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<SDLCheck>true</SDLCheck> <SDLCheck>true</SDLCheck>
<AdditionalIncludeDirectories>..\librf;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories> <AdditionalIncludeDirectories>..\librf;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories>
<AdditionalOptions>/await /std:c++latest </AdditionalOptions> <AdditionalOptions>/await /std:c++latest </AdditionalOptions>

Loading…
Cancel
Save