diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/Async.cc | 171 | ||||
| -rw-r--r-- | src/TaskMan.cc | 30 |
2 files changed, 201 insertions, 0 deletions
diff --git a/src/Async.cc b/src/Async.cc new file mode 100644 index 0000000..829c677 --- /dev/null +++ b/src/Async.cc @@ -0,0 +1,171 @@ +#include "Async.h" + +namespace { + +class AsyncStopper : public Balau::AsyncOperation { + public: + virtual bool needsSynchronousCallback() { return false; } + virtual void done() { delete this; } +}; + +}; + +void Balau::AsyncManager::setIdleReadyCallback(void (*callback)(void *), void * param) { + while (!m_ready); + TLS * tls = getTLS(); + tls->idleReadyCallback = callback; + tls->idleReadyParam = param; +} + +void Balau::AsyncManager::queueOp(AsyncOperation * op) { + if (m_stopperPushed) { + Printer::elog(E_ASYNC, "AsyncManager's queue has been stopped; running operation %p on this thread instead.", op); + op->run(); + op->finalize(); + return; + } + while (!m_ready); + TLS * tls = getTLS(); + Printer::elog(E_ASYNC, "Queuing operation at %p", op); + if (op->needsSynchronousCallback()) { + Printer::elog(E_ASYNC, "Operation at %p needs synchronous callback, copying values; idleQueue = %p; idleReadyCallback = %p; idleReadyParam = %p", &tls->idleQueue, tls->idleReadyCallback, tls->idleReadyParam); + op->m_idleQueue = &tls->idleQueue; + op->m_idleReadyCallback = tls->idleReadyCallback; + op->m_idleReadyParam = tls->idleReadyParam; + } + if (op->needsMainQueue()) + m_queue.push(op); + else + m_finished.push(op); +} + +void Balau::AsyncManager::checkIdle() { + if (m_numFinishersIdle > m_maxIdle) + killOneFinisher(); + if (m_numFinishersIdle < m_minIdle) + startOneFinisher(); + joinStoppedFinishers(); +} + +void Balau::AsyncManager::killOneFinisher() { + Printer::elog(E_ASYNC, "Too many workers idle (%i / %i), killing one.", m_numFinishersIdle, m_maxIdle); + m_finished.push(new AsyncStopper()); +} + +void Balau::AsyncManager::startOneFinisher() { + AsyncFinishWorker * worker = new AsyncFinishWorker(this, &m_finished); + Printer::elog(E_ASYNC, "Not enough workers idle (%i / %i), starting one at %p.", m_numFinishersIdle, m_minIdle, worker); + m_workers.push_back(worker); + m_numFinishers++; + worker->threadStart(); +} + +void Balau::AsyncManager::joinStoppedFinishers() { + for (auto i = m_workers.begin(); i != m_workers.end(); i++) { + AsyncFinishWorker * worker = *i; + if (!worker->m_stopped) + continue; + Printer::elog(E_ASYNC, "Joining stopped worker at %p", worker); + m_numFinishers--; + m_workers.erase(i); + worker->join(); + delete worker; + break; + } +} + +void * Balau::AsyncManager::proc() { + Printer::elog(E_ASYNC, "AsyncManager thread starting up"); + m_tlsManager.init(); + m_ready = true; + while (!m_stopping) { + checkIdle(); + AsyncOperation * op = m_queue.pop(); + Printer::elog(E_ASYNC, "AsyncManager got an operation at %p", op); + if (dynamic_cast<AsyncStopper *>(op)) { + Printer::elog(E_ASYNC, "AsyncManager got a stopper operation"); + m_stopping = true; + } + Printer::elog(E_ASYNC, "AsyncManager running operation at %p", op); + op->run(); + if (op->needsFinishWorker()) { + Printer::elog(E_ASYNC, "AsyncManager pushing operation at %p in the finisher's queue", op); + m_finished.push(op); + } else { + Printer::elog(E_ASYNC, "AsyncManager finalizing operation at %p", op); + op->finalize(); + } + } + stopAllWorkers(); + + while (Atomic::Prefetch::Decrement(&m_numTLSes)) { + TLS * tls = m_TLSes.pop(); + while (!tls->idleQueue.isEmpty()); + } + + return NULL; +} + +void * Balau::AsyncFinishWorker::proc() { + Printer::elog(E_ASYNC, "AsyncFinishWorker thread starting up"); + AsyncOperation * op; + while (!m_stopping) { + m_async->incIdle(); + op = m_queue->pop(); + m_async->decIdle(); + Printer::elog(E_ASYNC, "AsyncFinishWorker got operation at %p", op); + if (dynamic_cast<AsyncStopper *>(op)) { + Printer::elog(E_ASYNC, "AsyncFinishWorker got a stopper operation"); + m_stopping = true; + } + op->finalize(); + } + + m_stopped = true; + Printer::elog(E_ASYNC, "AsyncFinishWorker thread stopping"); + + return NULL; +} + +void Balau::AsyncOperation::finalize() { + Printer::elog(E_ASYNC, "AsyncOperation::finalize() is finishing operation %p", this); + finish(); + if (needsSynchronousCallback()) { + Printer::elog(E_ASYNC, "AsyncOperation::finalize() is pushing operation %p to its idle queue", this); + bool wasEmpty = m_idleQueue->isEmpty(); + m_idleQueue->push(this); + Printer::elog(E_ASYNC, "AsyncOperation::finalize() has pushed operation %p to its idle queue; wasEmpty = %s; callback = %p", this, wasEmpty ? "true" : "false", m_idleReadyCallback); + if (wasEmpty && m_idleReadyCallback) { + Printer::elog(E_ASYNC, "AsyncOperation::finalize() is calling ready callback to wake up main loop"); + m_idleReadyCallback(m_idleReadyParam); + } + } else { + Printer::elog(E_ASYNC, "AsyncOperation::finalize() is wrapping up operation %p", this); + done(); + } +} + +void Balau::AsyncManager::idle() { + Printer::elog(E_ASYNC, "AsyncManager::idle() is running"); + while (!m_ready); + AsyncOperation * op; + TLS * tls = getTLS(); + while ((op = tls->idleQueue.pop(false))) { + Printer::elog(E_ASYNC, "AsyncManager::idle() is wrapping up operation %p", op); + op->done(); + } +} + +void Balau::AsyncManager::threadExit() { + Printer::elog(E_ASYNC, "AsyncManager thread is being asked to stop; creating stopper"); + if (Atomic::CmpXChgBool(&m_stopperPushed, true, false)) + m_queue.push(new AsyncStopper()); +} + +void Balau::AsyncManager::stopAllWorkers() { + Printer::elog(E_ASYNC, "AsyncManager thread is being stopping and joining %i workers", m_numFinishers); + for (int i = 0; i < m_numFinishers; i++) + m_finished.push(new AsyncStopper()); + for (auto worker : m_workers) + worker->join(); +} diff --git a/src/TaskMan.cc b/src/TaskMan.cc index 17f0a40..49c326a 100644 --- a/src/TaskMan.cc +++ b/src/TaskMan.cc @@ -1,8 +1,24 @@ +#include "Async.h" #include "TaskMan.h" #include "Task.h" #include "Main.h" #include "Local.h" +static Balau::AsyncManager s_async; + +namespace { + +class AsyncStarter : public Balau::AtStart, Balau::AtExit { + public: + AsyncStarter() : AtStart(1000), AtExit(0) { } + void doStart() { + s_async.threadStart(); + } + void doExit() { + s_async.join(); + } +}; + class Stopper : public Balau::Task { public: Stopper(int code) : m_code(code) { } @@ -12,6 +28,10 @@ class Stopper : public Balau::Task { int m_code; }; +}; + +static AsyncStarter s_asyncStarter; + void Stopper::Do() { getTaskMan()->stopMe(m_code); } @@ -207,6 +227,8 @@ int Balau::TaskMan::mainLoop() { if (t->getStatus() == Task::STARTING) starting.insert(t); + s_async.setIdleReadyCallback(asyncIdleReady, this); + do { bool noWait = false; @@ -239,6 +261,9 @@ int Balau::TaskMan::mainLoop() { ev_run(m_loop, noWait || m_stopped ? EVRUN_NOWAIT : EVRUN_ONCE); Printer::elog(E_TASK, "TaskMan at %p Getting out of libev main loop", this); + // calling async's idle loop here + s_async.idle(); + // let's check what task got stopped, and signal them for (Task * t : stopped) { IAssert((t->getStatus() == Task::STOPPED) || (t->getStatus() == Task::FAULTED), "Task %p in stopped list but isn't stopped.", t); @@ -316,6 +341,7 @@ int Balau::TaskMan::mainLoop() { } while (!m_stopped); Printer::elog(E_TASK, "TaskManager at %p stopping.", this); + s_async.setIdleReadyCallback(NULL, NULL); return m_stopCode; } @@ -329,6 +355,10 @@ void Balau::TaskMan::registerTask(Balau::Task * t, Balau::Task * stick) { } } +void Balau::TaskMan::registerAsyncOp(Balau::AsyncOperation * op) { + s_async.queueOp(op); +} + void Balau::TaskMan::addToPending(Balau::Task * t) { m_pendingAdd.push(t); } |