summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/Handle.cc2
-rw-r--r--src/HttpServer.cc2
-rw-r--r--src/Task.cc45
-rw-r--r--src/TaskMan.cc124
4 files changed, 136 insertions, 37 deletions
diff --git a/src/Handle.cc b/src/Handle.cc
index 839c58a..b8dd45e 100644
--- a/src/Handle.cc
+++ b/src/Handle.cc
@@ -43,7 +43,7 @@ void eioInterface::readyCB(ev::async & w, int revents) {
void eioInterface::doStart() {
Balau::Printer::elog(Balau::E_HANDLE, "Starting the eio interface");
- Balau::TaskMan * taskMan = Balau::TaskMan::getTaskMan();
+ Balau::TaskMan * taskMan = Balau::TaskMan::getDefaultTaskMan();
Assert(taskMan);
struct ev_loop * loop = taskMan->getLoop();
diff --git a/src/HttpServer.cc b/src/HttpServer.cc
index 608b8a8..b027824 100644
--- a/src/HttpServer.cc
+++ b/src/HttpServer.cc
@@ -354,7 +354,7 @@ typedef Balau::Listener<Balau::HttpWorker> HttpListener;
void Balau::HttpServer::start() {
Assert(!m_started);
- m_listenerPtr = new HttpListener(m_port, m_local.to_charp());
+ m_listenerPtr = createTask(new HttpListener(m_port, m_local.to_charp()));
m_started = true;
}
diff --git a/src/Task.cc b/src/Task.cc
index 3772dd5..3a6f8ad 100644
--- a/src/Task.cc
+++ b/src/Task.cc
@@ -7,28 +7,29 @@
static Balau::LocalTmpl<Balau::Task> localTask;
Balau::Task::Task() {
+ m_status = STARTING;
+ m_okayToEAgain = false;
+
+ Printer::elog(E_TASK, "Created a Task at %p");
+}
+
+void Balau::Task::setup(TaskMan * taskMan) {
size_t size = stackSize();
#ifndef _WIN32
m_stack = malloc(size);
- coro_create(&m_ctx, coroutine, this, m_stack, size);
+ coro_create(&m_ctx, coroutineTrampoline, this, m_stack, size);
#else
m_stack = NULL;
- m_fiber = CreateFiber(size, coroutine, this);
+ m_fiber = CreateFiber(size, coroutineTrampoline, this);
#endif
- m_taskMan = TaskMan::getTaskMan();
- m_taskMan->registerTask(this);
+ m_taskMan = taskMan;
m_tls = g_tlsManager->createTLS();
void * oldTLS = g_tlsManager->getTLS();
g_tlsManager->setTLS(m_tls);
localTask.set(this);
g_tlsManager->setTLS(oldTLS);
-
- m_status = STARTING;
- m_okayToEAgain = false;
-
- Printer::elog(E_TASK, "Created a Task at %p");
}
Balau::Task::~Task() {
@@ -37,27 +38,31 @@ Balau::Task::~Task() {
free(m_tls);
}
-void Balau::Task::coroutine(void * arg) {
+void Balau::Task::coroutineTrampoline(void * arg) {
Task * task = reinterpret_cast<Task *>(arg);
Assert(task);
- Assert(task->m_status == STARTING);
+ task->coroutine();
+}
+
+void Balau::Task::coroutine() {
+ Assert(m_status == STARTING);
try {
- task->m_status = RUNNING;
- task->Do();
- task->m_status = STOPPED;
+ m_status = RUNNING;
+ Do();
+ m_status = STOPPED;
}
catch (GeneralException & e) {
- Printer::log(M_WARNING, "Task %s at %p caused an exception: `%s' - stopping.", task->getName(), task, e.getMsg());
- task->m_status = FAULTED;
+ Printer::log(M_WARNING, "Task %s at %p caused an exception: `%s' - stopping.", getName(), this, e.getMsg());
+ m_status = FAULTED;
}
catch (...) {
- Printer::log(M_WARNING, "Task %s at %p caused an unknown exception - stopping.", task->getName(), task);
- task->m_status = FAULTED;
+ Printer::log(M_WARNING, "Task %s at %p caused an unknown exception - stopping.", getName(), this);
+ m_status = FAULTED;
}
#ifndef _WIN32
- coro_transfer(&task->m_ctx, &task->m_taskMan->m_returnContext);
+ coro_transfer(&m_ctx, &m_taskMan->m_returnContext);
#else
- SwitchToFiber(task->m_taskMan->m_fiber);
+ SwitchToFiber(m_taskMan->m_fiber);
#endif
}
diff --git a/src/TaskMan.cc b/src/TaskMan.cc
index 0fc4668..8c99e52 100644
--- a/src/TaskMan.cc
+++ b/src/TaskMan.cc
@@ -3,9 +3,94 @@
#include "Main.h"
#include "Local.h"
+class Stopper : public Balau::Task {
+ virtual void Do();
+ virtual const char * getName();
+};
+
+void Stopper::Do() {
+ getMyTaskMan()->stopMe();
+}
+
+const char * Stopper::getName() {
+ return "Stopper";
+}
+
static Balau::DefaultTmpl<Balau::TaskMan> defaultTaskMan(50);
static Balau::LocalTmpl<Balau::TaskMan> localTaskMan;
+namespace Balau {
+
+class TaskScheduler : public Thread, public AtStart, public AtExit {
+ public:
+ TaskScheduler() : AtStart(100), m_stopping(false) { }
+ void registerTask(Task * t);
+ virtual void * proc();
+ virtual void doStart();
+ virtual void doExit();
+ void registerTaskMan(TaskMan * t);
+ void unregisterTaskMan(TaskMan * t);
+ void stopAll();
+ private:
+ Queue<Task *> m_queue;
+ volatile bool m_stopping;
+};
+
+};
+
+static Balau::TaskScheduler s_scheduler;
+
+void Balau::TaskScheduler::registerTask(Task * t) {
+ Printer::elog(E_TASK, "TaskScheduler::registerTask with t = %p", t);
+ m_queue.push(t);
+}
+
+void Balau::TaskScheduler::registerTaskMan(TaskMan * t) {
+ // meh. We need a round-robin queue system.
+}
+
+void Balau::TaskScheduler::unregisterTaskMan(TaskMan * t) {
+ // and here, we need to remove that taskman from the round robin queue.
+}
+
+void Balau::TaskScheduler::stopAll() {
+ m_stopping = true;
+ // and finally, we need to crawl the whole list and stop all of them.
+ TaskMan * tm = localTaskMan.getGlobal();
+ tm->addToPending(new Stopper());
+}
+
+void * Balau::TaskScheduler::proc() {
+ while (true) {
+ Printer::elog(E_TASK, "TaskScheduler waiting for a task to pop");
+ Task * t = m_queue.pop();
+ if (!t)
+ break;
+ if (dynamic_cast<Stopper *>(t) || m_stopping)
+ break;
+ // pick up a task manager here... for now let's take the global one.
+ // but we need some sort of round robin across all of the threads, as described above.
+ TaskMan * tm = localTaskMan.getGlobal();
+ Printer::elog(E_TASK, "TaskScheduler popped task %s at %p; adding to TaskMan %p", t->getName(), t, tm);
+ tm->addToPending(t);
+ tm->m_evt.send();
+ }
+ Printer::elog(E_TASK, "TaskScheduler stopping.");
+ return NULL;
+}
+
+void Balau::TaskScheduler::doStart() {
+ threadStart();
+}
+
+void Balau::TaskScheduler::doExit() {
+ Task * s = NULL;
+ m_queue.push(s);
+ join();
+}
+
+void asyncDummy(ev::async & w, int revents) { }
+
Balau::TaskMan::TaskMan() : m_stopped(false), m_allowedToSignal(false) {
#ifndef _WIN32
coro_create(&m_returnContext, 0, 0, 0, 0);
@@ -13,12 +98,17 @@ Balau::TaskMan::TaskMan() : m_stopped(false), m_allowedToSignal(false) {
m_fiber = ConvertThreadToFiber(NULL);
Assert(m_fiber);
#endif
- if (!localTaskMan.getGlobal()) {
+ TaskMan * global = localTaskMan.getGlobal();
+ if (!global) {
localTaskMan.setGlobal(this);
m_loop = ev_default_loop(EVFLAG_AUTO);
} else {
m_loop = ev_loop_new(EVFLAG_AUTO);
}
+ m_evt.set(m_loop);
+ m_evt.set<asyncDummy>();
+ m_evt.start();
+ s_scheduler.registerTaskMan(this);
}
#ifdef _WIN32
@@ -35,17 +125,18 @@ class WinSocketStartup : public Balau::AtStart {
static WinSocketStartup wsa;
#endif
-Balau::TaskMan * Balau::TaskMan::getTaskMan() { return localTaskMan.get(); }
+Balau::TaskMan * Balau::TaskMan::getDefaultTaskMan() { return localTaskMan.get(); }
Balau::TaskMan::~TaskMan() {
Assert(localTaskMan.getGlobal() != this);
+ s_scheduler.unregisterTaskMan(this);
+ // probably way more work to do here in order to clean up tasks from that thread
ev_loop_destroy(m_loop);
}
void Balau::TaskMan::mainLoop() {
// We need at least one round before bailing :)
do {
- taskList_t::iterator iL;
taskHash_t::iterator iH;
Task * t;
bool noWait = false;
@@ -65,15 +156,13 @@ void Balau::TaskMan::mainLoop() {
if (m_tasks.size() == 0)
noWait = true;
- m_pendingLock.enter();
if (m_pendingAdd.size() != 0)
noWait = true;
- m_pendingLock.leave();
// libev's event "loop". We always runs it once though.
m_allowedToSignal = true;
Printer::elog(E_TASK, "Going to libev main loop");
- ev_run(m_loop, noWait ? EVRUN_NOWAIT : EVRUN_ONCE);
+ ev_run(m_loop, noWait || m_stopped ? EVRUN_NOWAIT : EVRUN_ONCE);
Printer::elog(E_TASK, "Getting out of libev main loop");
// let's check what task got stopped, and signal them
@@ -99,15 +188,13 @@ void Balau::TaskMan::mainLoop() {
}
m_signaledTasks.clear();
- m_pendingLock.enter();
// Adding tasks that were added, maybe from other threads
- for (iL = m_pendingAdd.begin(); iL != m_pendingAdd.end(); iL++) {
- t = *iL;
+ while ((m_pendingAdd.size() != 0) || (m_tasks.size() == 0) && !m_stopped) {
+ t = m_pendingAdd.pop();
Assert(m_tasks.find(t) == m_tasks.end());
+ t->setup(this);
m_tasks.insert(t);
}
- m_pendingAdd.clear();
- m_pendingLock.leave();
// Finally, let's destroy tasks that no longer are necessary.
bool didDelete;
@@ -125,13 +212,16 @@ void Balau::TaskMan::mainLoop() {
}
} while (didDelete);
- } while (!m_stopped && m_tasks.size() != 0);
+ } while (!m_stopped);
+ Printer::elog(E_TASK, "TaskManager stopping.");
}
void Balau::TaskMan::registerTask(Balau::Task * t) {
- m_pendingLock.enter();
- m_pendingAdd.push_back(t);
- m_pendingLock.leave();
+ s_scheduler.registerTask(t);
+}
+
+void Balau::TaskMan::addToPending(Balau::Task * t) {
+ m_pendingAdd.push(t);
}
void Balau::TaskMan::signalTask(Task * t) {
@@ -139,3 +229,7 @@ void Balau::TaskMan::signalTask(Task * t) {
Assert(m_allowedToSignal);
m_signaledTasks.insert(t);
}
+
+void Balau::TaskMan::stop() {
+ s_scheduler.stopAll();
+}
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-14 12:13:28 +0000