X-Git-Url: http://g0dil.de/git?a=blobdiff_plain;f=Scheduler%2FScheduler.cc;h=a79307ac5ca989e8ccc909552429e078672dcb0b;hb=a1fdb7bb122f0b05be809a922d4b7ef5e125fa67;hp=e6e8b855d3c093da078b83bf1b9256cedee8240f;hpb=8d2d26f114d3df0a60c5c516fcf40671b1e55558;p=senf.git diff --git a/Scheduler/Scheduler.cc b/Scheduler/Scheduler.cc index e6e8b85..a79307a 100644 --- a/Scheduler/Scheduler.cc +++ b/Scheduler/Scheduler.cc @@ -1,9 +1,9 @@ // $Id$ // // Copyright (C) 2006 -// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS) -// Kompetenzzentrum fuer Satelitenkommunikation (SatCom) -// Stefan Bund +// Fraunhofer Institute for Open Communication Systems (FOKUS) +// Competence Center NETwork research (NET), St. Augustin, GERMANY +// Stefan Bund // // This program is free software; you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by @@ -23,9 +23,6 @@ /** \file \brief Scheduler non-inline non-template implementation - \idea Implement signal handling (See source for more discussion - about this) - \idea Multithreading support: To support multithreading, the static member Scheduler::instance() must return a thread-local value (that is Scheduler::instance() must allocate one Scheduler @@ -34,213 +31,163 @@ threads) */ -// Here a basic concept of how to add signal support to the scheduler: -// -// Every signal to be reported by the scheduler will be asigned a -// generic signal handler by the scheduler. This signal handler will -// use longjmp (juck) to report this signal back to the scheduler -// main-loop. -// -// To make this safe, the main-loop will look something like: -// -// int signal = setjmp(jmpBuffer_); -// if (signal == 0) { -// // unblock all signals which are registered with the -// // scheduler -// // call epoll -// // block all relevant signals again -// } -// -// // now handle the event -// -// The signal handler is then simply defined as -// -// static void Scheduler::sigHandler(int signal) -// { -// // make sure to restore the signal handler here if -// // necessary -// longjmp(Scheduler::instance().jmpBuffer_,signal); -// } -// -// You should use sigaction to register the signal handlers and define -// a sa_mask so all Scheduler-registered signals are automatically -// *blocked* whenever one of the signals is called (including the -// called signal!) (This also means, we will have to re-register all -// signals if we change the registration of some signal since the -// sa_mask changes). This ensures, that no two signals can be -// delivered on top of each other. And of course any signal registered -// with the scheduler must be blocked as soon as it is registered with -// the scheduler. - #include "Scheduler.hh" //#include "Scheduler.ih" // Custom includes -#include -#include -#include "../Utils/Exception.hh" - -static const int EPollInitialSize = 16; #define prefix_ ///////////////////////////////cc.p//////////////////////////////////////// -prefix_ senf::Scheduler::Scheduler() - : timerIdCounter_(0), epollFd_ (epoll_create(EPollInitialSize)), terminate_(false), - eventTime_(0) -{ - if (epollFd_<0) - throw SystemException(errno); +namespace { + bool terminate_ (false); + bool running_ (false); } -prefix_ void senf::Scheduler::do_add(int fd, SimpleCallback const & cb, int eventMask) +prefix_ void senf::scheduler::terminate() { - FdTable::iterator i (fdTable_.find(fd)); - int action (EPOLL_CTL_MOD); - if (i == fdTable_.end()) { - action = EPOLL_CTL_ADD; - i = fdTable_.insert(std::make_pair(fd, EventSpec())).first; - } - - if (eventMask & EV_READ) i->second.cb_read = cb; - if (eventMask & EV_PRIO) i->second.cb_prio = cb; - if (eventMask & EV_WRITE) i->second.cb_write = cb; - - epoll_event ev; - memset(&ev,0,sizeof(ev)); - ev.events = i->second.epollMask(); - ev.data.fd = fd; + terminate_ = true; +} - if (epoll_ctl(epollFd_, action, fd, &ev)<0) - throw SystemException(errno); +prefix_ void senf::scheduler::yield() +{ + senf::scheduler::detail::FIFORunner::instance().yield(); } -prefix_ void senf::Scheduler::do_remove(int fd, int eventMask) +prefix_ bool senf::scheduler::running() { - FdTable::iterator i (fdTable_.find(fd)); - if (i == fdTable_.end()) - return; + return running_; +} - if (eventMask & EV_READ) i->second.cb_read = 0; - if (eventMask & EV_PRIO) i->second.cb_prio = 0; - if (eventMask & EV_WRITE) i->second.cb_write = 0; +prefix_ senf::ClockService::clock_type senf::scheduler::now() +{ + return running() ? eventTime() : ClockService::now(); +} - epoll_event ev; - memset(&ev,0,sizeof(ev)); - ev.events = i->second.epollMask(); - ev.data.fd = fd; +namespace { + + // We don't want try { } catch(...) { ... throw; } since that will make debugging more + // difficult: the stack backtrace for an unexpected exception would always end here. + struct SchedulerScopedInit + { + SchedulerScopedInit() + { + senf::scheduler::detail::FIFORunner::instance().startWatchdog(); + senf::scheduler::detail::SignalDispatcher::instance().unblockSignals(); + senf::scheduler::detail::TimerDispatcher::instance().enable(); + running_ = true; + } + + ~SchedulerScopedInit() + { + senf::scheduler::detail::TimerDispatcher::instance().disable(); + senf::scheduler::detail::SignalDispatcher::instance().blockSignals(); + senf::scheduler::detail::FIFORunner::instance().stopWatchdog(); + running_ = false; + } + }; +} - int action (EPOLL_CTL_MOD); - if (ev.events==0) { - action = EPOLL_CTL_DEL; - fdTable_.erase(i); +prefix_ void senf::scheduler::process() +{ + SchedulerScopedInit initScheduler; + terminate_ = false; + running_ = true; + detail::TimerDispatcher::instance().reschedule(); + while(! terminate_ && ! (detail::FdDispatcher::instance().empty() && + detail::TimerDispatcher::instance().empty() && + detail::FileDispatcher::instance().empty())) { + detail::FdManager::instance().processOnce(); + detail::FileDispatcher::instance().prepareRun(); + detail::EventHookDispatcher::instance().prepareRun(); + detail::TimerDispatcher::instance().prepareRun(); + detail::FIFORunner::instance().run(); + detail::TimerDispatcher::instance().reschedule(); } - - if (epoll_ctl(epollFd_, action, fd, &ev)<0) - throw SystemException(errno); } - -prefix_ int senf::Scheduler::EventSpec::epollMask() - const +prefix_ void senf::scheduler::restart() { - int mask (0); - if (cb_read) mask |= EPOLLIN; - if (cb_prio) mask |= EPOLLPRI; - if (cb_write) mask |= EPOLLOUT; - return mask; + detail::FdManager* fdm (&detail::FdManager::instance()); + detail::FIFORunner* ffr (&detail::FIFORunner::instance()); + detail::FdDispatcher* fdd (&detail::FdDispatcher::instance()); + detail::TimerDispatcher* tdd (&detail::TimerDispatcher::instance()); + detail::SignalDispatcher* sdd (&detail::SignalDispatcher::instance()); + detail::FileDispatcher* fld (&detail::FileDispatcher::instance()); + detail::EventHookDispatcher* eed (&detail::EventHookDispatcher::instance()); + + eed->~EventHookDispatcher(); + fld->~FileDispatcher(); + sdd->~SignalDispatcher(); + tdd->~TimerDispatcher(); + fdd->~FdDispatcher(); + ffr->~FIFORunner(); + fdm->~FdManager(); + + new (fdm) detail::FdManager(); + new (ffr) detail::FIFORunner(); + new (fdd) detail::FdDispatcher(); + new (tdd) detail::TimerDispatcher(); + new (sdd) detail::SignalDispatcher(); + new (fld) detail::FileDispatcher(); + new (eed) detail::EventHookDispatcher(); } -prefix_ void senf::Scheduler::process() +prefix_ bool senf::scheduler::empty() { - terminate_ = false; - eventTime_ = ClockService::now(); - while (! terminate_) { - - // Since a callback may have disabled further timers, we need to check for canceled timeouts - // again. - - while (! timerQueue_.empty()) { - TimerMap::iterator i (timerQueue_.top()); - if (! i->second.canceled) - break; - timerMap_.erase(i); - timerQueue_.pop(); - } - - int timeout (-1); - if (timerQueue_.empty()) { - if (fdTable_.empty()) - break; - } - else { - ClockService::clock_type delta ( - (timerQueue_.top()->second.timeout - eventTime_)/1000000UL); - timeout = delta < 0 ? 0 : delta; - } - - ///\todo Handle more than one epoll_event per call - struct epoll_event ev; - int events = epoll_wait(epollFd_, &ev, 1, timeout); - if (events<0) - if (errno != EINTR) - throw SystemException(errno); + return detail::FdDispatcher::instance().empty() + && detail::TimerDispatcher::instance().empty() + && detail::FileDispatcher::instance().empty() + && detail::SignalDispatcher::instance().empty() + && detail::EventHookDispatcher::instance().empty(); +} - eventTime_ = ClockService::now(); +prefix_ void senf::scheduler::hiresTimers() +{ +#ifdef HAVE_TIMERFD + if (haveScalableHiresTimers()) + detail::TimerDispatcher::instance().timerSource( + std::auto_ptr(new detail::TimerFDTimerSource())); + else +#endif + detail::TimerDispatcher::instance().timerSource( + std::auto_ptr(new detail::POSIXTimerSource())); +} - // We always run event handlers. This is important, even if a file-descriptor is signaled - // since some descriptors (e.g. real files) will *always* be ready and we still may want to - // handle timers. - // Time handlers are run before file events to not delay them unnecessarily. +/////////////////////////////////////////////////////////////////////////// +// senf::schedulerLogTimeSource - while (! timerQueue_.empty()) { - TimerMap::iterator i (timerQueue_.top()); - if (i->second.canceled) - ; - else if (i->second.timeout <= eventTime_) - i->second.cb(); - else - break; - timerMap_.erase(i); - timerQueue_.pop(); - } +prefix_ senf::log::time_type senf::scheduler::LogTimeSource::operator()() + const +{ + return eventTime(); +} - if (events <= 0) - continue; +/////////////////////////////////////////////////////////////////////////// +// senf::scheduler::BlockSignals - FdTable::iterator i = fdTable_.find(ev.data.fd); - BOOST_ASSERT (i != fdTable_.end() ); - EventSpec spec (i->second); +prefix_ senf::scheduler::BlockSignals::BlockSignals(bool initiallyBlocked) + : blocked_ (false) +{ + ::sigfillset(&allSigs_); + if (initiallyBlocked) + block(); +} - unsigned extraFlags (0); - if (ev.events & EPOLLHUP) extraFlags |= EV_HUP; - if (ev.events & EPOLLERR) extraFlags |= EV_ERR; +prefix_ void senf::scheduler::BlockSignals::block() +{ + if (blocked_) + return; + ::sigprocmask(SIG_BLOCK, &allSigs_, &savedSigs_); + blocked_ = true; +} - if (ev.events & EPOLLIN) { - BOOST_ASSERT(spec.cb_read); - spec.cb_read(EventId(EV_READ | extraFlags)); - } - else if (ev.events & EPOLLPRI) { - BOOST_ASSERT(spec.cb_prio); - spec.cb_prio(EventId(EV_PRIO | extraFlags)); - } - else if (ev.events & EPOLLOUT) { - BOOST_ASSERT(spec.cb_write); - spec.cb_write(EventId(EV_WRITE | extraFlags)); - } - else { - // This branch is only taken, if HUP or ERR is signaled but none of IN/OUT/PRI. - // In this case we will signal all registered callbacks. The callbacks must be - // prepared to be called multiple times if they are registered to more than - // one event. - if (spec.cb_write) - spec.cb_write(EventId(extraFlags)); - if (spec.cb_prio) - spec.cb_prio(EventId(extraFlags)); - if (spec.cb_read) - spec.cb_read(EventId(extraFlags)); - } - } +prefix_ void senf::scheduler::BlockSignals::unblock() +{ + if (!blocked_) + return; + ::sigprocmask(SIG_SETMASK, &savedSigs_, 0); + blocked_ = false; } ///////////////////////////////cc.e////////////////////////////////////////