boost::asio和活动对象
boost::asio and Active Object
我实现了一些基于模块的活动对象设计模式。这是一个非常简单的实现。我有调度程序,激活列表,请求和未来得到响应。我的要求是这样的:
- 对活动对象的访问应该通过执行其方法来序列化在它自己的线程内(主请求和活动对象的假设)设计模式)
- 调用者应该能够指定请求执行的优先级。这意味着如果等待执行的请求多于零,则应按分配给每个请求的优先级对它们进行排序。具有较高优先级的请求应首先执行,因此,如果ActivationList上总是有一些待处理的请求,并且它们的优先级将高于给定的请求,则该请求将永远不会被执行-这对我来说是OK的
- 应该可以指定列表中待处理请求的最大数量(限制内存使用)
- 应该有可能使所有未处理的请求无效
- 请求应该能够返回值(阻塞调用者),或者只是在没有返回值的情况下执行,但是调用者应该被阻塞直到请求被处理,或者调用者不应该被阻塞,对于它来说,给定的请求是否被处理并不重要li>在请求执行之前,应该执行一些保护方法来检查给定的请求是否应该执行。如果没有-它将返回一些未定义的值调用者(在我目前的实现是boost::none,因为每个请求返回类型是boost::可选的)
现在的问题是:是否有可能使用boost::asio并满足我的所有需求?我的实现正在工作,但我想使用的东西可能是实现比我做得更好的方式。我也想知道它的未来,不要再"重新发明轮子"。
Asio可以用来包含活动对象的意图:将方法执行与方法调用解耦。额外的需求需要在更高的级别上处理,但在使用Boost时并不会过于复杂。Asio与其他Boost库一起使用
Scheduler
可以使用:
-
boost::thread
用于线程抽象。 -
boost::thread_group
管理线程的生存期。 -
boost::asio::io_service
提供线程池。当没有工作挂起时,可能想要使用boost::asio::io_service::work
来保持线程存活。
ActivationList
可以实现为:
- 。用于获取最高优先级方法请求的MultiIndex。使用提示位置
insert()
,将保留具有相同优先级的请求的插入顺序。 -
std::multiset
或std::multimap
可选。然而,在c++ 03中,对于具有相同键(优先级)的请求的顺序没有指定。 - 如果
Request
不需要保护方法,则可以使用std::priority_queue
。
Request
可以是未指定的类型:
-
boost::function
和boost::bind
可以用来提供类型擦除,同时绑定到可调用类型而不引入Request
层次结构。
Futures
可以使用Boost。线程的期货支持。
-
future.valid()
将返回true,如果Request
已经添加到ActivationList
。 -
future.wait()
将阻塞等待结果。 -
future.get()
将阻塞等待结果。 - 如果呼叫者没有使用
future
,则呼叫者不会被阻塞。 使用Boost的另一个好处。线程的未来是,从
Request
内产生的异常将传递给Future
。下面是一个利用各种Boost库的完整示例,应该满足要求:
// Standard includes
#include <algorithm> // std::find_if
#include <iostream>
#include <string>
// 3rd party includes
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/make_shared.hpp>
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/utility/result_of.hpp>
/// @brief scheduler that provides limits with prioritized jobs.
template <typename Priority,
typename Compare = std::less<Priority> >
class scheduler
{
public:
typedef Priority priority_type;
private:
/// @brief method_request is used to couple the guard and call
/// functions for a given method.
struct method_request
{
typedef boost::function<bool()> ready_func_type;
typedef boost::function<void()> run_func_type;
template <typename ReadyFunctor,
typename RunFunctor>
method_request(ReadyFunctor ready,
RunFunctor run)
: ready(ready),
run(run)
{}
ready_func_type ready;
run_func_type run;
};
/// @brief Pair type used to associate a request with its priority.
typedef std::pair<priority_type,
boost::shared_ptr<method_request> > pair_type;
static bool is_method_ready(const pair_type& pair)
{
return pair.second->ready();
}
public:
/// @brief Construct scheduler.
///
/// @param max_threads Maximum amount of concurrent task.
/// @param max_request Maximum amount of request.
scheduler(std::size_t max_threads,
std::size_t max_request)
: work_(io_service_),
max_request_(max_request),
request_count_(0)
{
// Spawn threads, dedicating them to the io_service.
for (std::size_t i = 0; i < max_threads; ++i)
threads_.create_thread(
boost::bind(&boost::asio::io_service::run, &io_service_));
}
/// @brief Destructor.
~scheduler()
{
// Release threads from the io_service.
io_service_.stop();
// Cleanup.
threads_.join_all();
}
/// @brief Insert a method request into the scheduler.
///
/// @param priority Priority of job.
/// @param ready_func Invoked to check if method is ready to run.
/// @param run_func Invoked when ready to run.
///
/// @return future associated with the method.
template <typename ReadyFunctor,
typename RunFunctor>
boost::unique_future<typename boost::result_of<RunFunctor()>::type>
insert(priority_type priority,
const ReadyFunctor& ready_func,
const RunFunctor& run_func)
{
typedef typename boost::result_of<RunFunctor()>::type result_type;
typedef boost::unique_future<result_type> future_type;
boost::unique_lock<mutex_type> lock(mutex_);
// If max request has been reached, then return an invalid future.
if (max_request_ &&
(request_count_ == max_request_))
return future_type();
++request_count_;
// Use a packaged task to handle populating promise and future.
typedef boost::packaged_task<result_type> task_type;
// Bind does not work with rvalue, and packaged_task is only moveable,
// so allocate a shared pointer.
boost::shared_ptr<task_type> task =
boost::make_shared<task_type>(run_func);
// Create method request.
boost::shared_ptr<method_request> request =
boost::make_shared<method_request>(
ready_func,
boost::bind(&task_type::operator(), task));
// Insert into priority. Hint to inserting as close to the end as
// possible to preserve insertion order for request with same priority.
activation_list_.insert(activation_list_.end(),
pair_type(priority, request));
// There is now an outstanding request, so post to dispatch.
io_service_.post(boost::bind(&scheduler::dispatch, this));
return task->get_future();
}
/// @brief Insert a method request into the scheduler.
///
/// @param ready_func Invoked to check if method is ready to run.
/// @param run_func Invoked when ready to run.
///
/// @return future associated with the method.
template <typename ReadyFunctor,
typename RunFunctor>
boost::unique_future<typename boost::result_of<RunFunctor()>::type>
insert(const ReadyFunctor& ready_func,
const RunFunctor& run_func)
{
return insert(priority_type(), ready_func, run_func);
}
/// @brief Insert a method request into the scheduler.
///
/// @param priority Priority of job.
/// @param run_func Invoked when ready to run.
///
/// @return future associated with the method.
template <typename RunFunctor>
boost::unique_future<typename boost::result_of<RunFunctor()>::type>
insert(priority_type priority,
const RunFunctor& run_func)
{
return insert(priority, &always_ready, run_func);
}
/// @brief Insert a method request with default priority into the
/// scheduler.
///
/// @param run_func Invoked when ready to run.
///
/// @param functor Job to run.
///
/// @return future associated with the job.
template <typename RunFunc>
boost::unique_future<typename boost::result_of<RunFunc()>::type>
insert(const RunFunc& run_func)
{
return insert(&always_ready, run_func);
}
/// @brief Cancel all outstanding request.
void cancel()
{
boost::unique_lock<mutex_type> lock(mutex_);
activation_list_.clear();
request_count_ = 0;
}
private:
/// @brief Dispatch a request.
void dispatch()
{
// Get the current highest priority request ready to run from the queue.
boost::unique_lock<mutex_type> lock(mutex_);
if (activation_list_.empty()) return;
// Find the highest priority method ready to run.
typedef typename activation_list_type::iterator iterator;
iterator end = activation_list_.end();
iterator result = std::find_if(
activation_list_.begin(), end, &is_method_ready);
// If no methods are ready, then post into dispatch, as the
// method may have become ready.
if (end == result)
{
io_service_.post(boost::bind(&scheduler::dispatch, this));
return;
}
// Take ownership of request.
boost::shared_ptr<method_request> method = result->second;
activation_list_.erase(result);
// Run method without mutex.
lock.unlock();
method->run();
lock.lock();
// Perform bookkeeping.
--request_count_;
}
static bool always_ready() { return true; }
private:
/// @brief List of outstanding request.
typedef boost::multi_index_container<
pair_type,
boost::multi_index::indexed_by<
boost::multi_index::ordered_non_unique<
boost::multi_index::member<pair_type,
typename pair_type::first_type,
&pair_type::first>,
Compare
>
>
> activation_list_type;
activation_list_type activation_list_;
/// @brief Thread group managing threads servicing pool.
boost::thread_group threads_;
/// @brief io_service used to function as a thread pool.
boost::asio::io_service io_service_;
/// @brief Work is used to keep threads servicing io_service.
boost::asio::io_service::work work_;
/// @brief Maximum amount of request.
const std::size_t max_request_;
/// @brief Count of outstanding request.
std::size_t request_count_;
/// @brief Synchronize access to the activation list.
typedef boost::mutex mutex_type;
mutex_type mutex_;
};
typedef scheduler<unsigned int,
std::greater<unsigned int> > high_priority_scheduler;
/// @brief adder is a simple proxy that will delegate work to
/// the scheduler.
class adder
{
public:
adder(high_priority_scheduler& scheduler)
: scheduler_(scheduler)
{}
/// @brief Add a and b with a priority.
///
/// @return Return future result.
template <typename T>
boost::unique_future<T> add(
high_priority_scheduler::priority_type priority,
const T& a, const T& b)
{
// Insert method request
return scheduler_.insert(
priority,
boost::bind(&adder::do_add<T>, a, b));
}
/// @brief Add a and b.
///
/// @return Return future result.
template <typename T>
boost::unique_future<T> add(const T& a, const T& b)
{
return add(high_priority_scheduler::priority_type(), a, b);
}
private:
/// @brief Actual add a and b.
template <typename T>
static T do_add(const T& a, const T& b)
{
std::cout << "Starting addition of '" << a
<< "' and '" << b << "'" << std::endl;
// Mimic busy work.
boost::this_thread::sleep_for(boost::chrono::seconds(2));
std::cout << "Finished addition" << std::endl;
return a + b;
}
private:
high_priority_scheduler& scheduler_;
};
bool get(bool& value) { return value; }
void guarded_call()
{
std::cout << "guarded_call" << std::endl;
}
int main()
{
const unsigned int max_threads = 1;
const unsigned int max_request = 4;
// Sscheduler
high_priority_scheduler scheduler(max_threads, max_request);
// Proxy
adder adder(scheduler);
// Client
// Add guarded method to scheduler.
bool ready = false;
std::cout << "Add guarded method." << std::endl;
boost::unique_future<void> future1 = scheduler.insert(
boost::bind(&get, boost::ref(ready)),
&guarded_call);
// Add 1 + 100 with default priority.
boost::unique_future<int> future2 = adder.add(1, 100);
// Force sleep to try to get scheduler to run request 2 first.
boost::this_thread::sleep_for(boost::chrono::seconds(1));
// Add:
// 2 + 200 with low priority (5)
// "test" + "this" with high priority (99)
boost::unique_future<int> future3 = adder.add(5, 2, 200);
boost::unique_future<std::string> future4 = adder.add(99,
std::string("test"), std::string("this"));
// Max request should have been reached, so add another.
boost::unique_future<int> future5 = adder.add(3, 300);
// Check if request was added.
std::cout << "future1 is valid: " << future1.valid()
<< "nfuture2 is valid: " << future2.valid()
<< "nfuture3 is valid: " << future3.valid()
<< "nfuture4 is valid: " << future4.valid()
<< "nfuture5 is valid: " << future5.valid()
<< std::endl;
// Get results for future2 and future3. Do nothing with future4's results.
std::cout << "future2 result: " << future2.get()
<< "nfuture3 result: " << future3.get()
<< std::endl;
std::cout << "Unguarding method." << std::endl;
ready = true;
future1.wait();
}
执行使用1个线程池,最多4个请求。
- request1被保护到程序结束,并且应该是最后一个运行。
- request2(1 + 100)以默认优先级插入,应该首先运行。
- request3(2 + 200)插入低优先级,应该在request4之后运行。
- request4 ('test' + 'this')以高优先级插入,应该在request3之前运行。
- request5应该由于最大请求而无法插入,并且应该无效。
输出如下:
添加保护方法。开始添加'1'和'100'Future1有效:1Future2有效:1Future3有效:1Future4有效:1Future5有效:0完成添加开始添加test和this完成添加开始添加'2'和'200'完成添加Future2结果:101Future3 result: 202使无防备的方法。guarded_call
相关文章:
- 什么时候调用组成单元对象的析构函数
- 对RValue对象调用的LValue ref限定成员函数
- CMake-按正确顺序将项目与C运行时对象文件链接
- 如何保持异步函数中使用成员的shared_ptr对象的活动状态?
- 使用 std::launder 从指向非活动对象的指针获取指向活动对象成员的指针?
- 类的堆分配对象是否在其作用域之后但在 C++ 中调用其析构函数之前处于活动状态
- 过滤器抱怨对象保持活动状态
- 在活动对象的实现中支持unique_ptr
- 我可以在 QP(量子平台)活动对象中等待 Windows 事件(WaitForMultipleObjects)吗?
- 对对象的数据成员的shared_ptr是否使数据成员在对象的生存期之后保持活动状态
- 活动对象 - 从函数对象类创建线程
- JNI保留对对象的全局引用,并使用其他JNI方法访问它.在多个JNI调用中保持C++对象的活动状态
- 通过多个jni调用保持某种c++对象的活动性
- 如何保持作为 SWIG shared_ptr传递给非托管代码的托管代码对象的活动状态?
- boost::asio和活动对象
- 当使用std::shared_ptr的函数返回堆分配的对象时,是否保证该对象是活动的
- 使对象保持活动状态时出现问题
- 检测对象是否仍处于活动状态或已被销毁
- 返回活动对象的值
- 如何使对象保持活动状态以供以后在另一个函数中使用