Scheduler/Scheduler.cc

   1 // $Id$
   2 //
   3 // Copyright (C) 2006
   4 // Fraunhofer Institute for Open Communication Systems (FOKUS)
   5 // Competence Center NETwork research (NET), St. Augustin, GERMANY
   6 //     Stefan Bund <g0dil@berlios.de>
   7 //
   8 // This program is free software; you can redistribute it and/or modify
   9 // it under the terms of the GNU General Public License as published by
  10 // the Free Software Foundation; either version 2 of the License, or
  11 // (at your option) any later version.
  12 //
  13 // This program is distributed in the hope that it will be useful,
  14 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 // GNU General Public License for more details.
  17 //
  18 // You should have received a copy of the GNU General Public License
  19 // along with this program; if not, write to the
  20 // Free Software Foundation, Inc.,
  21 // 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  22
  23 /** \file
  24     \brief Scheduler non-inline non-template implementation
  25
  26     \idea Multithreading support: To support multithreading, the
  27     static member Scheduler::instance() must return a thread-local
  28     value (that is Scheduler::instance() must allocate one Scheduler
  29     instance per thread). Another possibility would be to distribute
  30     the async load unto several threads (one scheduler for multiple
  31     threads)
  32  */
  33
  34 // Here a basic concept of how to add signal support to the scheduler:
  35 //
  36 // ... no, I had overlooked one race condition. So back to the signal-pipe approach ...
  37
  38 #include "Scheduler.hh"
  39 //#include "Scheduler.ih"
  40
  41 // Custom includes
  42 #include "../Utils/senfassert.hh"
  43 #include <errno.h>
  44 #include <sys/epoll.h>
  45 #include <unistd.h>
  46 #include <fcntl.h>
  47 #include "../Utils/Exception.hh"
  48
  49 static const int EPollInitialSize = 16;
  50
  51 #define prefix_
  52 ///////////////////////////////cc.p////////////////////////////////////////
  53
  54 prefix_ senf::Scheduler::Scheduler()
  55     : files_(0), timerIdCounter_(0), epollFd_ (epoll_create(EPollInitialSize)), terminate_(false),
  56       eventTime_(0), eventEarly_(ClockService::milliseconds(11)), eventAdjust_(0)
  57 {
  58     if (epollFd_<0)
  59         throwErrno();
  60
  61     if (::pipe(sigpipe_) < 0)
  62         throwErrno();
  63
  64     int flags (::fcntl(sigpipe_[1],F_GETFL));
  65     if (flags < 0)
  66         throwErrno();
  67     flags |= O_NONBLOCK;
  68     if (::fcntl(sigpipe_[1], F_SETFL, flags) < 0)
  69         throwErrno();
  70
  71     ::epoll_event ev;
  72     ::memset(&ev, 0, sizeof(ev));
  73     ev.events = EV_READ;
  74     ev.data.fd = sigpipe_[0];
  75     if (::epoll_ctl(epollFd_, EPOLL_CTL_ADD, sigpipe_[0], &ev) < 0)
  76         throwErrno();
  77 }
  78
  79 prefix_ void senf::Scheduler::registerSignal(unsigned signal, SimpleCallback const & cb)
  80 {
  81     ::sigset_t sig;
  82     ::sigemptyset(&sig);
  83     if (::sigaddset(&sig, signal) < 0)
  84         throw InvalidSignalNumberException();
  85     ::sigprocmask(SIG_BLOCK, &sig, 0);
  86     ::sigaddset(&sigset_, signal);
  87     if (sigHandlers_.size() <= signal)
  88         sigHandlers_.resize(signal+1);
  89     sigHandlers_[signal] = cb;
  90
  91     registerSigHandlers();
  92 }
  93
  94 prefix_ void senf::Scheduler::unregisterSignal(unsigned signal)
  95 {
  96     if (::sigdelset(&sigset_, signal) < 0)
  97         throw InvalidSignalNumberException();
  98     sigHandlers_[signal] = 0;
  99     ::signal(signal, SIG_DFL);
 100     registerSigHandlers();
 101 }
 102
 103 prefix_ void senf::Scheduler::do_add(int fd, FdCallback const & cb, int eventMask)
 104 {
 105     if (eventMask == 0)
 106         return;
 107
 108     FdTable::iterator i (fdTable_.find(fd));
 109     int action (EPOLL_CTL_MOD);
 110     if (i == fdTable_.end()) {
 111         action = EPOLL_CTL_ADD;
 112         i = fdTable_.insert(std::make_pair(fd, EventSpec())).first;
 113     }
 114     if (i->second.epollMask() == 0) {
 115         action = EPOLL_CTL_ADD;
 116         fdErase_.erase( std::remove(fdErase_.begin(), fdErase_.end(), unsigned(fd)),
 117                         fdErase_.end() );
 118     }
 119
 120     if (eventMask & EV_READ)  i->second.cb_read = cb;
 121     if (eventMask & EV_PRIO)  i->second.cb_prio = cb;
 122     if (eventMask & EV_WRITE) i->second.cb_write = cb;
 123
 124     epoll_event ev;
 125     memset(&ev,0,sizeof(ev));
 126     ev.events = i->second.epollMask();
 127     ev.data.fd = fd;
 128
 129     if (! i->second.file && epoll_ctl(epollFd_, action, fd, &ev) < 0) {
 130         if (errno == EPERM) {
 131             // Argh ... epoll does not support ordinary files :-( :-(
 132             i->second.file = true;
 133             ++ files_;
 134         }
 135         else
 136             throwErrno("::epoll_ctl()");
 137     }
 138 }
 139
 140 prefix_ void senf::Scheduler::do_remove(int fd, int eventMask)
 141 {
 142     if (eventMask == 0)
 143         return;
 144
 145     FdTable::iterator i (fdTable_.find(fd));
 146     if (i == fdTable_.end())
 147         return;
 148
 149     if (eventMask & EV_READ)  i->second.cb_read = 0;
 150     if (eventMask & EV_PRIO)  i->second.cb_prio = 0;
 151     if (eventMask & EV_WRITE) i->second.cb_write = 0;
 152
 153     epoll_event ev;
 154     memset(&ev,0,sizeof(ev));
 155     ev.events = i->second.epollMask();
 156     ev.data.fd = fd;
 157
 158     int action (EPOLL_CTL_MOD);
 159     bool file (i->second.file);
 160     if (ev.events==0) {
 161         action = EPOLL_CTL_DEL;
 162         fdErase_.push_back(fd);
 163     }
 164
 165     if (! file && epoll_ctl(epollFd_, action, fd, &ev) < 0)
 166         throwErrno("::epoll_ctl()");
 167     if (file)
 168         -- files_;
 169 }
 170
 171 prefix_ void senf::Scheduler::registerSigHandlers()
 172 {
 173     for (unsigned signal (1); signal < sigHandlers_.size(); ++signal) {
 174         if (sigHandlers_[signal]) {
 175             struct ::sigaction sa;
 176             sa.sa_sigaction = & Scheduler::sigHandler;
 177             sa.sa_mask = sigset_;
 178             sa.sa_flags = SA_SIGINFO;
 179             if (signal == SIGCHLD)
 180                 sa.sa_flags |= SA_NOCLDSTOP;
 181             if (::sigaction(signal, &sa, 0) < 0)
 182                 throwErrno();
 183         }
 184     }
 185 }
 186
 187 prefix_ void senf::Scheduler::sigHandler(int signal, ::siginfo_t * siginfo, void *)
 188 {
 189     // This is a bit unsafe. Better write single bytes and place the siginfo into an explicit
 190     // queue. Since signals are only unblocked during epoll_wait, we even wouldn't need to
 191     // synchronize access to that queue any further.
 192
 193     ::write(instance().sigpipe_[1], siginfo, sizeof(*siginfo));
 194
 195     // We ignore errors. The file handle is set to non-blocking IO. If any failure occurs (pipe
 196     // full), the signal will be dropped. That's like kernel signal handling which may also drop
 197     // signals.
 198 }
 199
 200 prefix_ int senf::Scheduler::EventSpec::epollMask()
 201     const
 202 {
 203     int mask (0);
 204     if (cb_read)  mask |= EPOLLIN;
 205     if (cb_prio)  mask |= EPOLLPRI;
 206     if (cb_write) mask |= EPOLLOUT;
 207     return mask;
 208 }
 209
 210 prefix_ void senf::Scheduler::process()
 211 {
 212     terminate_ = false;
 213     eventTime_ = ClockService::now();
 214     while (! terminate_) {
 215
 216         // Since a callback may have disabled further timers, we need to check for canceled timeouts
 217         // again.
 218
 219         while (! timerQueue_.empty()) {
 220             TimerMap::iterator i (timerQueue_.top());
 221             if (! i->second.canceled)
 222                 break;
 223             timerMap_.erase(i);
 224             timerQueue_.pop();
 225         }
 226
 227         for (FdEraseList::iterator i (fdErase_.begin()); i != fdErase_.end(); ++i)
 228             fdTable_.erase(*i);
 229         fdErase_.clear();
 230
 231         int timeout (-1);
 232         if (files_ > 0)
 233             timeout = 0;
 234         else {
 235             if (timerQueue_.empty()) {
 236                 if (fdTable_.empty())
 237                     break;
 238             }
 239             else {
 240                 ClockService::clock_type delta (
 241                     (timerQueue_.top()->second.timeout - eventTime_ + eventAdjust_)/1000000UL);
 242                 timeout = delta < 0 ? 0 : delta;
 243             }
 244         }
 245
 246         ///\todo Handle more than one epoll_event per call
 247         struct epoll_event ev;
 248
 249         ::sigprocmask(SIG_UNBLOCK, &sigset_, 0);
 250         int events (epoll_wait(epollFd_, &ev, 1, timeout));
 251         ::sigprocmask(SIG_BLOCK, &sigset_, 0);
 252
 253         if (events<0)
 254             if (errno != EINTR)
 255                 throwErrno();
 256
 257         eventTime_ = ClockService::now();
 258
 259         // We always run timeout handlers. This is important, even if a file-descriptor is signaled
 260         // since some descriptors (e.g. real files) will *always* be ready and we still may want to
 261         // handle timers.  Time handlers are run before file events to not delay them unnecessarily.
 262
 263         while (! timerQueue_.empty()) {
 264             TimerMap::iterator i (timerQueue_.top());
 265             if (i->second.canceled)
 266                 ;
 267             else if (i->second.timeout <= eventTime_ + eventEarly_)
 268                 i->second.cb();
 269             else
 270                 break;
 271             timerQueue_.pop();
 272             timerMap_.erase(i);
 273         }
 274
 275         // Check the signal queue
 276         if (events > 0 && ev.data.fd == sigpipe_[0]) {
 277             ::siginfo_t siginfo;
 278             if (::read(sigpipe_[0], &siginfo, sizeof(siginfo)) < int(sizeof(siginfo))) {
 279                 // We ignore truncated records which may only occur if the signal
 280                 // queue became filled up
 281                 SENF_LOG((senf::log::IMPORTANT)("Truncated signal record!"));
 282                 continue;
 283             }
 284             if (siginfo.si_signo < int(sigHandlers_.size()) && sigHandlers_[siginfo.si_signo])
 285                 sigHandlers_[siginfo.si_signo]();
 286             continue;
 287         }
 288
 289         for (FdTable::iterator i = fdTable_.begin(); i != fdTable_.end(); ++i) {
 290             EventSpec & spec (i->second);
 291
 292             if (! (spec.file || (events > 0 && i->first == ev.data.fd)))
 293                 continue;
 294
 295             unsigned extraFlags (0);
 296             unsigned mask (spec.file ? spec.epollMask() : ev.events);
 297
 298             if (mask & EPOLLHUP) extraFlags |= EV_HUP;
 299             if (mask & EPOLLERR) extraFlags |= EV_ERR;
 300
 301             if (mask & EPOLLIN) {
 302                 SENF_ASSERT(spec.cb_read);
 303                 spec.cb_read(EventId(EV_READ | extraFlags));
 304             }
 305             else if (mask & EPOLLPRI) {
 306                 SENF_ASSERT(spec.cb_prio);
 307                 spec.cb_prio(EventId(EV_PRIO | extraFlags));
 308             }
 309             else if (mask & EPOLLOUT) {
 310                 SENF_ASSERT(spec.cb_write);
 311                 spec.cb_write(EventId(EV_WRITE | extraFlags));
 312             }
 313             else {
 314                 // This branch is only taken, if HUP or ERR is signaled but none of IN/OUT/PRI.
 315                 // In this case we will signal all registered callbacks. The callbacks must be
 316                 // prepared to be called multiple times if they are registered to more than
 317                 // one event.
 318                 if (spec.cb_write)
 319                     spec.cb_write(EventId(extraFlags));
 320                 if (spec.cb_prio)
 321                     spec.cb_prio(EventId(extraFlags));
 322                 if (spec.cb_read)
 323                     spec.cb_read(EventId(extraFlags));
 324             }
 325         }
 326     }
 327 }
 328
 329 ///////////////////////////////////////////////////////////////////////////
 330 // senf::SchedulerLogTimeSource
 331
 332 prefix_ boost::posix_time::ptime senf::SchedulerLogTimeSource::operator()()
 333     const
 334 {
 335     return ClockService::abstime(Scheduler::instance().eventTime());
 336 }
 337
 338 ///////////////////////////////cc.e////////////////////////////////////////
 339 #undef prefix_
 340
 341 \f
 342 // Local Variables:
 343 // mode: c++
 344 // fill-column: 100
 345 // c-file-style: "senf"
 346 // indent-tabs-mode: nil
 347 // ispell-local-dictionary: "american"
 348 // compile-command: "scons -u test"
 349 // comment-column: 40
 350 // End: