Add std::function overload to RoutingWorker::broadcast

This makes it easier to collect distributed results from multiple workers.
This commit is contained in:
Markus Mäkelä
2018-08-31 10:04:16 +03:00
parent ce7a59fb56
commit 19b059d303
2 changed files with 36 additions and 0 deletions

View File

@ -214,6 +214,21 @@ public:
*/ */
static size_t broadcast(std::unique_ptr<DisposableTask> sTask); static size_t broadcast(std::unique_ptr<DisposableTask> sTask);
/**
* Posts a function to all workers for execution.
*
* @param pSem If non-NULL, will be posted once the task's `execute` return.
* @param mode Execution mode
*
* @return How many workers the task was posted to.
*/
static size_t broadcast(std::function<void ()> func, mxb::Semaphore* pSem, execute_mode_t mode);
static size_t broadcast(std::function<void ()> func, enum execute_mode_t mode)
{
return broadcast(func, NULL, mode);
}
/** /**
* Executes a task on all workers in serial mode (the task is executed * Executes a task on all workers in serial mode (the task is executed
* on at most one worker thread at a time). When the function returns * on at most one worker thread at a time). When the function returns

View File

@ -634,6 +634,27 @@ size_t RoutingWorker::broadcast(std::unique_ptr<DisposableTask> sTask)
return n; return n;
} }
//static
size_t RoutingWorker::broadcast(std::function<void ()> func, mxb::Semaphore* pSem,
mxb::Worker::execute_mode_t mode)
{
size_t n = 0;
int nWorkers = this_unit.next_worker_id;
for (int i = 0; i < nWorkers; ++i)
{
RoutingWorker* pWorker = this_unit.ppWorkers[i];
mxb_assert(pWorker);
if (pWorker->execute(func, pSem, mode))
{
++n;
}
}
return n;
}
//static //static
size_t RoutingWorker::execute_serially(Task& task) size_t RoutingWorker::execute_serially(Task& task)
{ {