X-Git-Url: http://g0dil.de/git?a=blobdiff_plain;f=Scheduler%2FScheduler.cc;h=758b980de29c7ac51eddd6d3d4450bbc6f74236a;hb=be33ff96c5b89738694da272d8610564cce48bfb;hp=09628f1af2f76bbd245407aa1889d1033f159117;hpb=e67faa588ac375105c0a6d12c79ef93d5499bcab;p=senf.git diff --git a/Scheduler/Scheduler.cc b/Scheduler/Scheduler.cc index 09628f1..758b980 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 @@ -42,6 +39,7 @@ //#include "Scheduler.ih" // Custom includes +#include "../Utils/senfassert.hh" #include #include #include @@ -54,28 +52,28 @@ static const int EPollInitialSize = 16; ///////////////////////////////cc.p//////////////////////////////////////// prefix_ senf::Scheduler::Scheduler() - : timerIdCounter_(0), epollFd_ (epoll_create(EPollInitialSize)), terminate_(false), + : files_(0), timerIdCounter_(0), epollFd_ (epoll_create(EPollInitialSize)), terminate_(false), eventTime_(0), eventEarly_(ClockService::milliseconds(11)), eventAdjust_(0) { if (epollFd_<0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); if (::pipe(sigpipe_) < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); int flags (::fcntl(sigpipe_[1],F_GETFL)); if (flags < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); flags |= O_NONBLOCK; if (::fcntl(sigpipe_[1], F_SETFL, flags) < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); ::epoll_event ev; ::memset(&ev, 0, sizeof(ev)); ev.events = EV_READ; ev.data.fd = sigpipe_[0]; if (::epoll_ctl(epollFd_, EPOLL_CTL_ADD, sigpipe_[0], &ev) < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); } prefix_ void senf::Scheduler::registerSignal(unsigned signal, SimpleCallback const & cb) @@ -104,12 +102,20 @@ prefix_ void senf::Scheduler::unregisterSignal(unsigned signal) prefix_ void senf::Scheduler::do_add(int fd, FdCallback const & cb, int eventMask) { + if (eventMask == 0) + return; + 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 (i->second.epollMask() == 0) { + action = EPOLL_CTL_ADD; + fdErase_.erase( std::remove(fdErase_.begin(), fdErase_.end(), unsigned(fd)), + fdErase_.end() ); + } if (eventMask & EV_READ) i->second.cb_read = cb; if (eventMask & EV_PRIO) i->second.cb_prio = cb; @@ -120,12 +126,22 @@ prefix_ void senf::Scheduler::do_add(int fd, FdCallback const & cb, int eventMas ev.events = i->second.epollMask(); ev.data.fd = fd; - if (epoll_ctl(epollFd_, action, fd, &ev)<0) - throw SystemException(errno); + if (! i->second.file && epoll_ctl(epollFd_, action, fd, &ev) < 0) { + if (errno == EPERM) { + // Argh ... epoll does not support ordinary files :-( :-( + i->second.file = true; + ++ files_; + } + else + SENF_THROW_SYSTEM_EXCEPTION("::epoll_ctl()"); + } } prefix_ void senf::Scheduler::do_remove(int fd, int eventMask) { + if (eventMask == 0) + return; + FdTable::iterator i (fdTable_.find(fd)); if (i == fdTable_.end()) return; @@ -140,18 +156,21 @@ prefix_ void senf::Scheduler::do_remove(int fd, int eventMask) ev.data.fd = fd; int action (EPOLL_CTL_MOD); + bool file (i->second.file); if (ev.events==0) { action = EPOLL_CTL_DEL; - fdTable_.erase(i); + fdErase_.push_back(fd); } - if (epoll_ctl(epollFd_, action, fd, &ev)<0) - throw SystemException(errno); + if (! file && epoll_ctl(epollFd_, action, fd, &ev) < 0) + SENF_THROW_SYSTEM_EXCEPTION("::epoll_ctl()"); + if (file) + -- files_; } prefix_ void senf::Scheduler::registerSigHandlers() { - for (unsigned signal; signal < sigHandlers_.size(); ++signal) + for (unsigned signal (1); signal < sigHandlers_.size(); ++signal) { if (sigHandlers_[signal]) { struct ::sigaction sa; sa.sa_sigaction = & Scheduler::sigHandler; @@ -160,8 +179,9 @@ prefix_ void senf::Scheduler::registerSigHandlers() if (signal == SIGCHLD) sa.sa_flags |= SA_NOCLDSTOP; if (::sigaction(signal, &sa, 0) < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); } + } } prefix_ void senf::Scheduler::sigHandler(int signal, ::siginfo_t * siginfo, void *) @@ -170,7 +190,7 @@ prefix_ void senf::Scheduler::sigHandler(int signal, ::siginfo_t * siginfo, void // queue. Since signals are only unblocked during epoll_wait, we even wouldn't need to // synchronize access to that queue any further. - ::write(instance().sigpipe_[1], siginfo, sizeof(siginfo)); + ::write(instance().sigpipe_[1], siginfo, sizeof(*siginfo)); // We ignore errors. The file handle is set to non-blocking IO. If any failure occurs (pipe // full), the signal will be dropped. That's like kernel signal handling which may also drop @@ -204,15 +224,23 @@ prefix_ void senf::Scheduler::process() timerQueue_.pop(); } + for (FdEraseList::iterator i (fdErase_.begin()); i != fdErase_.end(); ++i) + fdTable_.erase(*i); + fdErase_.clear(); + int timeout (-1); - if (timerQueue_.empty()) { - if (fdTable_.empty()) - break; - } + if (files_ > 0) + timeout = 0; else { - ClockService::clock_type delta ( - (timerQueue_.top()->second.timeout - eventTime_ + eventAdjust_)/1000000UL); - timeout = delta < 0 ? 0 : delta; + if (timerQueue_.empty()) { + if (fdTable_.empty()) + break; + } + else { + ClockService::clock_type delta ( + (timerQueue_.top()->second.timeout - eventTime_ + eventAdjust_)/1000000UL); + timeout = delta < 0 ? 0 : delta; + } } ///\todo Handle more than one epoll_event per call @@ -224,7 +252,7 @@ prefix_ void senf::Scheduler::process() if (events<0) if (errno != EINTR) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION(""); eventTime_ = ClockService::now(); @@ -244,52 +272,56 @@ prefix_ void senf::Scheduler::process() timerMap_.erase(i); } - if (events <= 0) - continue; - // Check the signal queue - if (ev.data.fd == sigpipe_[0]) { + if (events > 0 && ev.data.fd == sigpipe_[0]) { ::siginfo_t siginfo; - if (::read(sigpipe_[0], &siginfo, sizeof(siginfo)) < int(sizeof(siginfo))) + if (::read(sigpipe_[0], &siginfo, sizeof(siginfo)) < int(sizeof(siginfo))) { // We ignore truncated records which may only occur if the signal // queue became filled up + SENF_LOG((senf::log::IMPORTANT)("Truncated signal record!")); continue; + } if (siginfo.si_signo < int(sigHandlers_.size()) && sigHandlers_[siginfo.si_signo]) sigHandlers_[siginfo.si_signo](); continue; } - FdTable::iterator i = fdTable_.find(ev.data.fd); - BOOST_ASSERT (i != fdTable_.end() ); - EventSpec spec (i->second); - - unsigned extraFlags (0); - if (ev.events & EPOLLHUP) extraFlags |= EV_HUP; - if (ev.events & EPOLLERR) extraFlags |= EV_ERR; + for (FdTable::iterator i = fdTable_.begin(); i != fdTable_.end(); ++i) { + EventSpec & spec (i->second); - 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)); + if (! (spec.file || (events > 0 && i->first == ev.data.fd))) + continue; + + unsigned extraFlags (0); + unsigned mask (spec.file ? spec.epollMask() : ev.events); + + if (mask & EPOLLHUP) extraFlags |= EV_HUP; + if (mask & EPOLLERR) extraFlags |= EV_ERR; + + if (mask & EPOLLIN) { + SENF_ASSERT(spec.cb_read); + spec.cb_read(EventId(EV_READ | extraFlags)); + } + else if (mask & EPOLLPRI) { + SENF_ASSERT(spec.cb_prio); + spec.cb_prio(EventId(EV_PRIO | extraFlags)); + } + else if (mask & EPOLLOUT) { + SENF_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)); + } } } }