X-Git-Url: http://g0dil.de/git?a=blobdiff_plain;f=Scheduler%2FScheduler.cc;h=3d61946a8fdae0f6c6b68bbbb895e71d29a96b16;hb=1004d55779889e573ab51cf007ea5e25434165a7;hp=9f222bfd789304d1f8347afb8c41e3fd62e10f1a;hpb=17e24d84603667395e9ffa786a9cdbb722bf9c1f;p=senf.git diff --git a/Scheduler/Scheduler.cc b/Scheduler/Scheduler.cc index 9f222bf..3d61946 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,11 +39,13 @@ //#include "Scheduler.ih" // Custom includes +#include "../Utils/senfassert.hh" #include #include #include #include #include "../Utils/Exception.hh" +#include "../Utils/Backtrace.hh" static const int EPollInitialSize = 16; @@ -54,28 +53,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("::epoll_create()"); if (::pipe(sigpipe_) < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION("::pipe()"); int flags (::fcntl(sigpipe_[1],F_GETFL)); if (flags < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION("::fcntl(F_GETFL)"); flags |= O_NONBLOCK; if (::fcntl(sigpipe_[1], F_SETFL, flags) < 0) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION("::fcntl(F_SETFL)"); ::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("::epoll_ctl(EPOLL_CTL_ADD)"); } prefix_ void senf::Scheduler::registerSignal(unsigned signal, SimpleCallback const & cb) @@ -104,12 +103,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; } + else 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 +127,31 @@ 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); + for (;;) { + if ( (!i->second.file) && (epoll_ctl(epollFd_, action, fd, &ev) < 0) ) { + switch (errno) { + case EPERM : + // Argh ... epoll does not support ordinary files :-( :-( + i->second.file = true; + ++ files_; + return; + case ENOENT : + action = EPOLL_CTL_ADD; + break; + default: + SENF_THROW_SYSTEM_EXCEPTION("::epoll_ctl()"); + } + } + else + return; + } } 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 +166,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 && errno != ENOENT) + 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 +189,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("::sigaction()"); } + } } prefix_ void senf::Scheduler::sigHandler(int signal, ::siginfo_t * siginfo, void *) @@ -204,27 +234,37 @@ 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 struct epoll_event ev; - ::sigprocmask(SIG_UNBLOCK, &sigset_, 0); - int events (epoll_wait(epollFd_, &ev, 1, timeout)); - ::sigprocmask(SIG_BLOCK, &sigset_, 0); + if (::sigprocmask(SIG_UNBLOCK, &sigset_, 0) < 0) + SENF_THROW_SYSTEM_EXCEPTION("::sigprocmask(SIG_UNBLOCK)"); + int events (::epoll_wait(epollFd_, &ev, 1, timeout)); + if (::sigprocmask(SIG_BLOCK, &sigset_, 0) < 0) + SENF_THROW_SYSTEM_EXCEPTION("::sigprocmask(SIG_BLOCK)"); if (events<0) if (errno != EINTR) - throw SystemException(errno); + SENF_THROW_SYSTEM_EXCEPTION("::epoll_wait()"); eventTime_ = ClockService::now(); @@ -244,11 +284,8 @@ 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))) { // We ignore truncated records which may only occur if the signal @@ -261,37 +298,42 @@ prefix_ void senf::Scheduler::process() continue; } - FdTable::iterator i = fdTable_.find(ev.data.fd); - BOOST_ASSERT (i != fdTable_.end() ); - EventSpec spec (i->second); + for (FdTable::iterator i = fdTable_.begin(); i != fdTable_.end(); ++i) { + EventSpec & spec (i->second); - unsigned extraFlags (0); - if (ev.events & EPOLLHUP) extraFlags |= EV_HUP; - if (ev.events & EPOLLERR) extraFlags |= EV_ERR; + if (! (spec.file || (events > 0 && i->first == ev.data.fd))) + continue; + + unsigned extraFlags (0); + unsigned mask (spec.file ? spec.epollMask() : ev.events); - 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 (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)); + } } } } @@ -299,10 +341,10 @@ prefix_ void senf::Scheduler::process() /////////////////////////////////////////////////////////////////////////// // senf::SchedulerLogTimeSource -prefix_ boost::posix_time::ptime senf::SchedulerLogTimeSource::operator()() +prefix_ senf::log::time_type senf::SchedulerLogTimeSource::operator()() const { - return ClockService::abstime(Scheduler::instance().eventTime()); + return Scheduler::instance().eventTime(); } ///////////////////////////////cc.e////////////////////////////////////////