From cc78db0520ff384892cbee69557a38c3a1b04613 Mon Sep 17 00:00:00 2001 From: Pixel Date: Sun, 8 Apr 2012 22:03:50 -0700 Subject: Redifining yielding for operations a bit, making further refinements. --- src/Handle.cc | 6 +++--- src/HttpServer.cc | 2 +- src/Input.cc | 8 ++++---- src/LuaTask.cc | 2 +- src/Output.cc | 8 ++++---- src/Socket.cc | 10 +++++----- src/Task.cc | 24 +++++++++++++++++------- src/ZHandle.cc | 4 ++-- 8 files changed, 37 insertions(+), 27 deletions(-) (limited to 'src') diff --git a/src/Handle.cc b/src/Handle.cc index dacd30d..9bcb419 100644 --- a/src/Handle.cc +++ b/src/Handle.cc @@ -97,7 +97,7 @@ ssize_t Balau::Handle::forceRead(void * _buf, size_t count, Events::BaseEvent * catch (EAgain e) { if (evt && evt->gotSignal()) return total; - Task::yield(e.getEvent()); + Task::operationYield(e.getEvent()); continue; } if (r < 0) @@ -124,7 +124,7 @@ ssize_t Balau::Handle::forceWrite(const void * _buf, size_t count, Events::BaseE catch (EAgain e) { if (evt && evt->gotSignal()) return total; - Task::yield(e.getEvent()); + Task::operationYield(e.getEvent()); continue; } if (r < 0) @@ -242,7 +242,7 @@ int Balau::FileSystem::mkdir(const char * path) throw (GeneralException) { cbResults_t cbResults; eio_req * r = eio_mkdir(path, 0755, 0, eioDone, &cbResults); EAssert(r != NULL, "eio_mkdir returned a NULL eio_req"); - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); char str[4096]; if (cbResults.result < 0) diff --git a/src/HttpServer.cc b/src/HttpServer.cc index 69d58cd..751b01a 100644 --- a/src/HttpServer.cc +++ b/src/HttpServer.cc @@ -607,7 +607,7 @@ void Balau::HttpServer::stop() { Task::prepare(&event); listener->stop(); m_started = false; - Task::yield(&event); + Task::operationYield(&event); IAssert(event.gotSignal(), "HttpServer::stop didn't actually get the listener to stop"); event.ack(); } diff --git a/src/Input.cc b/src/Input.cc index 14aed91..46b9b94 100644 --- a/src/Input.cc +++ b/src/Input.cc @@ -56,7 +56,7 @@ Balau::Input::Input(const char * fname) throw (GeneralException) : m_fd(-1), m_s cbResults_t cbResults; eio_req * r = eio_open(fname, O_RDONLY, 0, 0, eioDone, &cbResults); EAssert(r != NULL, "eio_open returned a NULL eio_req"); - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); if (cbResults.result < 0) { char str[4096]; if (cbResults.errorno == ENOENT) { @@ -71,7 +71,7 @@ Balau::Input::Input(const char * fname) throw (GeneralException) : m_fd(-1), m_s cbStatsResults_t cbStatsResults; r = eio_fstat(m_fd, 0, eioStatsDone, &cbStatsResults); EAssert(r != NULL, "eio_fstat returned a NULL eio_req"); - Task::yield(&cbStatsResults.evt); + Task::operationYield(&cbStatsResults.evt); if (cbStatsResults.result == 0) { m_size = cbStatsResults.statdata.st_size; m_mtime = cbStatsResults.statdata.st_mtime; @@ -85,7 +85,7 @@ void Balau::Input::close() throw (GeneralException) { eio_req * r = eio_close(m_fd, 0, eioDone, &cbResults); EAssert(r != NULL, "eio_close returned a NULL eio_req"); m_fd = -1; - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); if (cbResults.result < 0) { char str[4096]; strerror_r(cbResults.errorno, str, sizeof(str)); @@ -99,7 +99,7 @@ ssize_t Balau::Input::read(void * buf, size_t count) throw (GeneralException) { cbResults_t cbResults; eio_req * r = eio_read(m_fd, buf, count, getROffset(), 0, eioDone, &cbResults); EAssert(r != NULL, "eio_read returned a NULL eio_req"); - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); if (cbResults.result > 0) { rseek(cbResults.result, SEEK_CUR); } else { diff --git a/src/LuaTask.cc b/src/LuaTask.cc index 687c81a..b023f45 100644 --- a/src/LuaTask.cc +++ b/src/LuaTask.cc @@ -69,7 +69,7 @@ void Balau::LuaExecCell::exec(LuaMainTask * mainTask) { Task::prepare(&m_event); mainTask->exec(this); if (!m_detached) - Task::yield(&m_event); + Task::operationYield(&m_event); } void Balau::LuaExecString::run(Lua & L) { diff --git a/src/Output.cc b/src/Output.cc index 6de404e..c585ce5 100644 --- a/src/Output.cc +++ b/src/Output.cc @@ -56,7 +56,7 @@ Balau::Output::Output(const char * fname, bool truncate) throw (GeneralException cbResults_t cbResults; eio_req * r = eio_open(fname, O_WRONLY | O_CREAT | (truncate ? O_TRUNC : 0), 0755, 0, eioDone, &cbResults); EAssert(r != NULL, "eio_open returned a NULL eio_req"); - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); if (cbResults.result < 0) { char str[4096]; if (cbResults.errorno == ENOENT) { @@ -71,7 +71,7 @@ Balau::Output::Output(const char * fname, bool truncate) throw (GeneralException cbStatsResults_t cbStatsResults; r = eio_fstat(m_fd, 0, eioStatsDone, &cbStatsResults); EAssert(r != NULL, "eio_fstat returned a NULL eio_req"); - Task::yield(&cbStatsResults.evt); + Task::operationYield(&cbStatsResults.evt); if (cbStatsResults.result == 0) { m_size = cbStatsResults.statdata.st_size; m_mtime = cbStatsResults.statdata.st_mtime; @@ -85,7 +85,7 @@ void Balau::Output::close() throw (GeneralException) { eio_req * r = eio_close(m_fd, 0, eioDone, &cbResults); EAssert(r != NULL, "eio_close returned a NULL eio_req"); m_fd = -1; - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); if (cbResults.result < 0) { char str[4096]; strerror_r(cbResults.errorno, str, sizeof(str)); @@ -99,7 +99,7 @@ ssize_t Balau::Output::write(const void * buf, size_t count) throw (GeneralExcep cbResults_t cbResults; eio_req * r = eio_write(m_fd, const_cast(buf), count, getWOffset(), 0, eioDone, &cbResults); EAssert(r != NULL, "eio_write returned a NULL eio_req"); - Task::yield(&cbResults.evt); + Task::operationYield(&cbResults.evt); if (cbResults.result > 0) { wseek(cbResults.result, SEEK_CUR); } else { diff --git a/src/Socket.cc b/src/Socket.cc index d78c6e5..100b94e 100644 --- a/src/Socket.cc +++ b/src/Socket.cc @@ -243,7 +243,7 @@ static DNSRequest resolveName(const char * name, const char * service = NULL, st Balau::Printer::elog(Balau::E_SOCKET, "Sending a request to the resolver thread"); Balau::Task::prepare(&evt); resolverThread.pushRequest(&req); - Balau::Task::yield(&evt); + Balau::Task::operationYield(&evt); Balau::Atomic::MemoryFence(); @@ -476,7 +476,7 @@ bool Balau::Socket::connect(const char * hostname, int port) { IAssert(spins == 0, "We shouldn't have spinned..."); } - Task::yield(m_evtW, true); + Task::operationYield(m_evtW, Task::INTERRUPTIBLE); // if we're still here, it means the parent task doesn't want to be thrown an exception IAssert(m_evtW->gotSignal(), "We shouldn't have been awoken without getting our event signalled"); @@ -535,7 +535,7 @@ Balau::IO Balau::Socket::accept() throw (GeneralException) { if (s < 0) { if ((errno == EAGAIN) || (errno == EINTR) || (errno == EWOULDBLOCK)) { - Task::yield(m_evtR, true); + Task::operationYield(m_evtR, Task::INTERRUPTIBLE); } else { String msg = getErrorMessage(); throw GeneralException(String("Unexpected error accepting a connection: #") + errno + "(" + msg + ")"); @@ -567,7 +567,7 @@ ssize_t Balau::Socket::read(void * buf, size_t count) throw (GeneralException) { } if ((errno == EAGAIN) || (errno == EINTR) || (errno == EWOULDBLOCK)) { - Task::yield(m_evtR, true); + Task::operationYield(m_evtR, Task::INTERRUPTIBLE); } else { m_evtR->stop(); return r; @@ -602,7 +602,7 @@ ssize_t Balau::Socket::write(const void * buf, size_t count) throw (GeneralExcep #endif if ((errno == EAGAIN) || (errno == EINTR) || (errno == EWOULDBLOCK)) { - Task::yield(m_evtW, true); + Task::operationYield(m_evtW, Task::INTERRUPTIBLE); } else { m_evtW->stop(); return r; diff --git a/src/Task.cc b/src/Task.cc index f01bb18..df8eee2 100644 --- a/src/Task.cc +++ b/src/Task.cc @@ -260,27 +260,36 @@ void Balau::Events::Custom::gotOwner(Task * task) { m_loop = task->getLoop(); } -void Balau::Task::yield(Events::BaseEvent * evt, bool interruptible) throw (GeneralException) { +void Balau::Task::operationYield(Events::BaseEvent * evt, enum OperationYieldType yieldType) throw (GeneralException) { Task * t = getCurrentTask(); if (evt) t->waitFor(evt); if (t->m_stackless) { - AAssert(interruptible, "You can't run non-interruptible operations from a stackless task."); + AAssert(yieldType != SIMPLE, "You can't run simple operations from a stackless task."); + } + + if (yieldType == STACKLESS) { + AAssert(t->m_okayToEAgain, "You can't run a stackless operation from a non-okay-to-eagain task."); } bool gotSignal; do { t->yield(evt == NULL); - Printer::elog(E_TASK, "operation back from yielding; interruptible = %s; okayToEAgain = %s", interruptible ? "true" : "false", t->m_okayToEAgain ? "true" : "false"); + static const char * YieldTypeToString[] = { + "SIMPLE", + "INTERRUPTIBLE", + "STACKLESS", + }; + Printer::elog(E_TASK, "operation back from yielding; yieldType = %s; okayToEAgain = %s", YieldTypeToString[yieldType], t->m_okayToEAgain ? "true" : "false"); gotSignal = evt ? evt->gotSignal() : true; - } while ((!interruptible || !t->m_okayToEAgain) && !gotSignal); + } while (((yieldType == SIMPLE) || !t->m_okayToEAgain) && !gotSignal); if (!evt) return; - if (interruptible && t->m_okayToEAgain && !evt->gotSignal()) { + if ((yieldType != SIMPLE) && t->m_okayToEAgain && !evt->gotSignal()) { Printer::elog(E_TASK, "operation is throwing an exception."); throw EAgain(evt); } @@ -302,12 +311,12 @@ void Balau::QueueBase::iPush(void * t, Events::Async * event) { } void * Balau::QueueBase::iPop(Events::Async * event) { - ScopeLock sl(m_lock); + m_lock.enter(); while (!m_front) { if (event) { Task::prepare(event); m_lock.leave(); - Task::yield(event); + Task::operationYield(event, Task::INTERRUPTIBLE); m_lock.enter(); } else { pthread_cond_wait(&m_cond, &m_lock.m_lock); @@ -323,5 +332,6 @@ void * Balau::QueueBase::iPop(Events::Async * event) { m_back = NULL; void * t = c->m_elem; delete c; + m_lock.leave(); return t; } diff --git a/src/ZHandle.cc b/src/ZHandle.cc index 7ab0133..3a14d7d 100644 --- a/src/ZHandle.cc +++ b/src/ZHandle.cc @@ -79,7 +79,7 @@ ssize_t Balau::ZStream::read(void * buf, size_t count) throw (GeneralException) if (r <= 0) return readTotal; } - Task::yield(nullptr); + Task::operationYield(); int r = inflate(&m_zin, Z_SYNC_FLUSH); size_t didRead = count - m_zin.avail_out; readTotal += didRead; @@ -108,7 +108,7 @@ ssize_t Balau::ZStream::write(const void * buf, size_t count) throw (GeneralExce m_zout.avail_out = BLOCK_SIZE; int r = deflate(&m_zout, Z_NO_FLUSH); EAssert(r == Z_OK, "deflate() didn't return Z_OK but %i", r); - Task::yield(nullptr); + Task::operationYield(); size_t compressed = BLOCK_SIZE - m_zout.avail_out; if (compressed) { size_t w = m_h->forceWrite(obuf, compressed); -- cgit v1.2.3