boost::asio を使用して制限されたスレッド プール クラスを作成しようとしています。しかし、ある時点で行き詰まってしまいました。誰か助けてくれませんか。
唯一の問題は、カウンターをどこで減らすべきかということです。
コードが期待どおりに動作しません。
問題は、スレッドの実行がいつ終了するか、またスレッドがプールに戻ったことをどうやって知るかが分からないことです。
#include <boost/asio.hpp>
#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/bind.hpp>
#include <boost/thread/mutex.hpp>
#include <stack>
using namespace std;
using namespace boost;
class ThreadPool
{
static int count;
int NoOfThread;
thread_group grp;
mutex mutex_;
asio::io_service io_service;
int counter;
stack<thread*> thStk ;
public:
ThreadPool(int num)
{
NoOfThread = num;
counter = 0;
mutex::scoped_lock lock(mutex_);
if(count == 0)
count++;
else
return;
for(int i=0 ; i<num ; ++i)
{
thStk.push(grp.create_thread(boost::bind(&asio::io_service::run, &io_service)));
}
}
~ThreadPool()
{
io_service.stop();
grp.join_all();
}
thread* getThread()
{
if(counter > NoOfThread)
{
cout<<"run out of threads \n";
return NULL;
}
counter++;
thread* ptr = thStk.top();
thStk.pop();
return ptr;
}
};
int ThreadPool::count = 0;
struct callable
{
void operator()()
{
cout<<"some task for thread \n";
}
};
int main( int argc, char * argv[] )
{
callable x;
ThreadPool pool(10);
thread* p = pool.getThread();
cout<<p->get_id();
//how i can assign some function to thread pointer ?
//how i can return thread pointer after work done so i can add
//it back to stack?
return 0;
}
ベストアンサー1
つまり、ユーザーが指定したタスクを、次の機能を実行する別の関数でラップする必要があります。
- ユーザー関数または呼び出し可能なオブジェクトを呼び出します。
- ミューテックスをロックし、カウンターを減らします。
このスレッド プールのすべての要件を理解しているわけではないかもしれません。したがって、明確にするために、私が要件だと考えているものを明示的にリストします。
- プールはスレッドの有効期間を管理します。ユーザーはプール内に存在するスレッドを削除できません。
- ユーザーは、非侵入的な方法でプールにタスクを割り当てることができます。
- タスクが割り当てられているときに、プール内のすべてのスレッドが現在他のタスクを実行している場合、そのタスクは破棄されます。
実装を説明する前に、強調したい重要なポイントがいくつかあります。
- スレッドは、一度起動されると、完了、キャンセル、または終了するまで実行されます。スレッドが実行している関数は、再割り当てできません。1 つのスレッドがその存続期間中に複数の関数を実行できるようにするには、スレッドを などのキューから読み取る関数で起動し、
io_service::run()などの呼び出し可能型を からイベント キューに投稿する必要がありますio_service::post()。 io_service::run()に保留中の作業がない場合io_service、 がio_service停止した場合、またはスレッドが実行していたハンドラから例外がスローされた場合は、 が戻ります。io_serivce::run()未完了の作業がない場合に が戻らないようにするには、io_service::workクラスを使用できます。object()型を要求する (つまり、タスクは を継承する必要がある) のではなく、タスクの型要件を定義する (つまり、タスクの型は構文によって呼び出し可能である必要があるprocess) と、ユーザーにさらなる柔軟性が提供されます。これにより、ユーザーはタスクを関数ポインターまたは nullary を提供する型として提供できますoperator()。
実装には以下を使用しますboost::asio:
#include <boost/asio.hpp>
#include <boost/thread.hpp>
class thread_pool
{
private:
boost::asio::io_service io_service_;
boost::asio::io_service::work work_;
boost::thread_group threads_;
std::size_t available_;
boost::mutex mutex_;
public:
/// @brief Constructor.
thread_pool( std::size_t pool_size )
: work_( io_service_ ),
available_( pool_size )
{
for ( std::size_t i = 0; i < pool_size; ++i )
{
threads_.create_thread( boost::bind( &boost::asio::io_service::run,
&io_service_ ) );
}
}
/// @brief Destructor.
~thread_pool()
{
// Force all threads to return from io_service::run().
io_service_.stop();
// Suppress all exceptions.
try
{
threads_.join_all();
}
catch ( const std::exception& ) {}
}
/// @brief Adds a task to the thread pool if a thread is currently available.
template < typename Task >
void run_task( Task task )
{
boost::unique_lock< boost::mutex > lock( mutex_ );
// If no threads are available, then return.
if ( 0 == available_ ) return;
// Decrement count, indicating thread is no longer available.
--available_;
// Post a wrapped task into the queue.
io_service_.post( boost::bind( &thread_pool::wrap_task, this,
boost::function< void() >( task ) ) );
}
private:
/// @brief Wrap a task so that the available count can be increased once
/// the user provided task has completed.
void wrap_task( boost::function< void() > task )
{
// Run the user supplied task.
try
{
task();
}
// Suppress all exceptions.
catch ( const std::exception& ) {}
// Task has finished, so increment count of available threads.
boost::unique_lock< boost::mutex > lock( mutex_ );
++available_;
}
};
実装に関するコメントをいくつか紹介します。
- 例外処理はユーザーのタスクの周囲で行われる必要があります。ユーザーの関数または呼び出し可能オブジェクトが 型ではない例外をスローした場合
boost::thread_interrupted、std::terminate()が呼び出されます。これは Boost.Thread の結果です。スレッド関数の例外動作。Boost.Asioのハンドラからスローされた例外の影響。 - ユーザーが
taskviaを指定した場合boost::bind、ネストされた部分はboost::bindコンパイルに失敗します。次のオプションのいずれかが必要です。taskによって作成されたものはサポートされていませんboost::bind。- メタプログラミングは、ユーザーの型に基づいてコンパイル時の分岐を実行し、
boost::bindその結果boost::protectを使用できるようにします。これにより、boost::protect特定の関数オブジェクトでのみ適切に機能します。 - 別の型を使用して
taskオブジェクトを間接的に渡します。私は、boost::function正確な型を失うことを犠牲にして、可読性のために を使用することを選択しました。boost::tupleは、少し読みにくくなりますが、Boost.Asio のシリアル化例。
アプリケーション コードでは、このthread_pool型を非侵入的に使用できるようになりました。
void work() {};
struct worker
{
void operator()() {};
};
void more_work( int ) {};
int main()
{
thread_pool pool( 2 );
pool.run_task( work ); // Function pointer.
pool.run_task( worker() ); // Callable object.
pool.run_task( boost::bind( more_work, 5 ) ); // Callable object.
}
はBoost.Asio なしで作成でき、いつ が返されるか、オブジェクトが何であるかなどの動作thread_poolについて知る必要がなくなるため、メンテナーにとっては少し簡単になるかもしれません。Boost.Asioio_service::run()io_service::work
#include <queue>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
class thread_pool
{
private:
std::queue< boost::function< void() > > tasks_;
boost::thread_group threads_;
std::size_t available_;
boost::mutex mutex_;
boost::condition_variable condition_;
bool running_;
public:
/// @brief Constructor.
thread_pool( std::size_t pool_size )
: available_( pool_size ),
running_( true )
{
for ( std::size_t i = 0; i < pool_size; ++i )
{
threads_.create_thread( boost::bind( &thread_pool::pool_main, this ) ) ;
}
}
/// @brief Destructor.
~thread_pool()
{
// Set running flag to false then notify all threads.
{
boost::unique_lock< boost::mutex > lock( mutex_ );
running_ = false;
condition_.notify_all();
}
try
{
threads_.join_all();
}
// Suppress all exceptions.
catch ( const std::exception& ) {}
}
/// @brief Add task to the thread pool if a thread is currently available.
template < typename Task >
void run_task( Task task )
{
boost::unique_lock< boost::mutex > lock( mutex_ );
// If no threads are available, then return.
if ( 0 == available_ ) return;
// Decrement count, indicating thread is no longer available.
--available_;
// Set task and signal condition variable so that a worker thread will
// wake up andl use the task.
tasks_.push( boost::function< void() >( task ) );
condition_.notify_one();
}
private:
/// @brief Entry point for pool threads.
void pool_main()
{
while( running_ )
{
// Wait on condition variable while the task is empty and the pool is
// still running.
boost::unique_lock< boost::mutex > lock( mutex_ );
while ( tasks_.empty() && running_ )
{
condition_.wait( lock );
}
// If pool is no longer running, break out.
if ( !running_ ) break;
// Copy task locally and remove from the queue. This is done within
// its own scope so that the task object is destructed immediately
// after running the task. This is useful in the event that the
// function contains shared_ptr arguments bound via bind.
{
boost::function< void() > task = tasks_.front();
tasks_.pop();
lock.unlock();
// Run the task.
try
{
task();
}
// Suppress all exceptions.
catch ( const std::exception& ) {}
}
// Task has finished, so increment count of available threads.
lock.lock();
++available_;
} // while running_
}
};