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 #include "../Utils/Backtrace.hh"
  49
  50 static const int EPollInitialSize = 16;
  51
  52 #define prefix_
  53 ///////////////////////////////cc.p////////////////////////////////////////
  54
  55 prefix_ senf::Scheduler::Scheduler()
  56     : files_(0), timerIdCounter_(0), epollFd_ (epoll_create(EPollInitialSize)), terminate_(false),
  57       eventTime_(0), eventEarly_(ClockService::milliseconds(11)), eventAdjust_(0)
  58 {
  59     if (epollFd_<0)
  60         SENF_THROW_SYSTEM_EXCEPTION("::epoll_create()");
  61
  62     if (::pipe(sigpipe_) < 0)
  63         SENF_THROW_SYSTEM_EXCEPTION("::pipe()");
  64
  65     int flags (::fcntl(sigpipe_[1],F_GETFL));
  66     if (flags < 0)
  67         SENF_THROW_SYSTEM_EXCEPTION("::fcntl(F_GETFL)");
  68     flags |= O_NONBLOCK;
  69     if (::fcntl(sigpipe_[1], F_SETFL, flags) < 0)
  70         SENF_THROW_SYSTEM_EXCEPTION("::fcntl(F_SETFL)");
  71
  72     ::epoll_event ev;
  73     ::memset(&ev, 0, sizeof(ev));
  74     ev.events = EV_READ;
  75     ev.data.fd = sigpipe_[0];
  76     if (::epoll_ctl(epollFd_, EPOLL_CTL_ADD, sigpipe_[0], &ev) < 0)
  77         SENF_THROW_SYSTEM_EXCEPTION("::epoll_ctl(EPOLL_CTL_ADD)");
  78 }
  79
  80 prefix_ void senf::Scheduler::registerSignal(unsigned signal, SimpleCallback const & cb)
  81 {
  82     ::sigset_t sig;
  83     ::sigemptyset(&sig);
  84     if (::sigaddset(&sig, signal) < 0)
  85         throw InvalidSignalNumberException();
  86     ::sigprocmask(SIG_BLOCK, &sig, 0);
  87     ::sigaddset(&sigset_, signal);
  88     if (sigHandlers_.size() <= signal)
  89         sigHandlers_.resize(signal+1);
  90     sigHandlers_[signal] = cb;
  91
  92     registerSigHandlers();
  93 }
  94
  95 prefix_ void senf::Scheduler::unregisterSignal(unsigned signal)
  96 {
  97     if (::sigdelset(&sigset_, signal) < 0)
  98         throw InvalidSignalNumberException();
  99     sigHandlers_[signal] = 0;
 100     ::signal(signal, SIG_DFL);
 101     registerSigHandlers();
 102 }
 103
 104 prefix_ void senf::Scheduler::do_add(int fd, FdCallback const & cb, int eventMask)
 105 {
 106     if (eventMask == 0)
 107         return;
 108
 109     FdTable::iterator i (fdTable_.find(fd));
 110     int action (EPOLL_CTL_MOD);
 111     if (i == fdTable_.end()) {
 112         action = EPOLL_CTL_ADD;
 113         i = fdTable_.insert(std::make_pair(fd, EventSpec())).first;
 114     }
 115     else if (i->second.epollMask() == 0) {
 116         action = EPOLL_CTL_ADD;
 117         fdErase_.erase( std::remove(fdErase_.begin(), fdErase_.end(), unsigned(fd)),
 118                         fdErase_.end() );
 119     }
 120
 121     if (eventMask & EV_READ)  i->second.cb_read = cb;
 122     if (eventMask & EV_PRIO)  i->second.cb_prio = cb;
 123     if (eventMask & EV_WRITE) i->second.cb_write = cb;
 124
 125     epoll_event ev;
 126     memset(&ev,0,sizeof(ev));
 127     ev.events = i->second.epollMask();
 128     ev.data.fd = fd;
 129
 130     for (;;) {
 131         if ( (!i->second.file) && (epoll_ctl(epollFd_, action, fd, &ev) < 0) ) {
 132             switch (errno) {
 133             case EPERM :
 134                 // Argh ... epoll does not support ordinary files :-( :-(
 135                 i->second.file = true;
 136                 ++ files_;
 137                 return;
 138             case ENOENT :
 139                 action = EPOLL_CTL_ADD;
 140                 break;
 141             default:
 142                 SENF_THROW_SYSTEM_EXCEPTION("::epoll_ctl()");
 143             }
 144         }
 145         else
 146             return;
 147     }
 148 }
 149
 150 prefix_ void senf::Scheduler::do_remove(int fd, int eventMask)
 151 {
 152     if (eventMask == 0)
 153         return;
 154
 155     FdTable::iterator i (fdTable_.find(fd));
 156     if (i == fdTable_.end())
 157         return;
 158
 159     if (eventMask & EV_READ)  i->second.cb_read = 0;
 160     if (eventMask & EV_PRIO)  i->second.cb_prio = 0;
 161     if (eventMask & EV_WRITE) i->second.cb_write = 0;
 162
 163     epoll_event ev;
 164     memset(&ev,0,sizeof(ev));
 165     ev.events = i->second.epollMask();
 166     ev.data.fd = fd;
 167
 168     int action (EPOLL_CTL_MOD);
 169     bool file (i->second.file);
 170     if (ev.events==0) {
 171         action = EPOLL_CTL_DEL;
 172         fdErase_.push_back(fd);
 173     }
 174
 175     if (! file && epoll_ctl(epollFd_, action, fd, &ev) < 0 && errno != ENOENT)
 176         SENF_THROW_SYSTEM_EXCEPTION("::epoll_ctl()");
 177     if (file)
 178         -- files_;
 179 }
 180
 181 prefix_ void senf::Scheduler::registerSigHandlers()
 182 {
 183     for (unsigned signal (1); signal < sigHandlers_.size(); ++signal) {
 184         if (sigHandlers_[signal]) {
 185             struct ::sigaction sa;
 186             sa.sa_sigaction = & Scheduler::sigHandler;
 187             sa.sa_mask = sigset_;
 188             sa.sa_flags = SA_SIGINFO;
 189             if (signal == SIGCHLD)
 190                 sa.sa_flags |= SA_NOCLDSTOP;
 191             if (::sigaction(signal, &sa, 0) < 0)
 192                 SENF_THROW_SYSTEM_EXCEPTION("::sigaction()");
 193         }
 194     }
 195 }
 196
 197 prefix_ void senf::Scheduler::sigHandler(int signal, ::siginfo_t * siginfo, void *)
 198 {
 199     // This is a bit unsafe. Better write single bytes and place the siginfo into an explicit
 200     // queue. Since signals are only unblocked during epoll_wait, we even wouldn't need to
 201     // synchronize access to that queue any further.
 202
 203     ::write(instance().sigpipe_[1], siginfo, sizeof(*siginfo));
 204
 205     // We ignore errors. The file handle is set to non-blocking IO. If any failure occurs (pipe
 206     // full), the signal will be dropped. That's like kernel signal handling which may also drop
 207     // signals.
 208 }
 209
 210 prefix_ int senf::Scheduler::EventSpec::epollMask()
 211     const
 212 {
 213     int mask (0);
 214     if (cb_read)  mask |= EPOLLIN;
 215     if (cb_prio)  mask |= EPOLLPRI;
 216     if (cb_write) mask |= EPOLLOUT;
 217     return mask;
 218 }
 219
 220 prefix_ void senf::Scheduler::process()
 221 {
 222     terminate_ = false;
 223     eventTime_ = ClockService::now();
 224     while (! terminate_) {
 225
 226         // Since a callback may have disabled further timers, we need to check for canceled timeouts
 227         // again.
 228
 229         while (! timerQueue_.empty()) {
 230             TimerMap::iterator i (timerQueue_.top());
 231             if (! i->second.canceled)
 232                 break;
 233             timerMap_.erase(i);
 234             timerQueue_.pop();
 235         }
 236
 237         for (FdEraseList::iterator i (fdErase_.begin()); i != fdErase_.end(); ++i)
 238             fdTable_.erase(*i);
 239         fdErase_.clear();
 240
 241         int timeout (-1);
 242         if (files_ > 0)
 243             timeout = 0;
 244         else {
 245             if (timerQueue_.empty()) {
 246                 if (fdTable_.empty())
 247                     break;
 248             }
 249             else {
 250                 ClockService::clock_type delta (
 251                     (timerQueue_.top()->second.timeout - eventTime_ + eventAdjust_)/1000000UL);
 252                 timeout = delta < 0 ? 0 : delta;
 253             }
 254         }
 255
 256         ///\todo Handle more than one epoll_event per call
 257         struct epoll_event ev;
 258
 259         if (::sigprocmask(SIG_UNBLOCK, &sigset_, 0) < 0)
 260             SENF_THROW_SYSTEM_EXCEPTION("::sigprocmask(SIG_UNBLOCK)");
 261         int events (::epoll_wait(epollFd_, &ev, 1, timeout));
 262         if (::sigprocmask(SIG_BLOCK, &sigset_, 0) < 0)
 263             SENF_THROW_SYSTEM_EXCEPTION("::sigprocmask(SIG_BLOCK)");
 264
 265         if (events<0)
 266             if (errno != EINTR)
 267                 SENF_THROW_SYSTEM_EXCEPTION("::epoll_wait()");
 268
 269         eventTime_ = ClockService::now();
 270
 271         // We always run timeout handlers. This is important, even if a file-descriptor is signaled
 272         // since some descriptors (e.g. real files) will *always* be ready and we still may want to
 273         // handle timers.  Time handlers are run before file events to not delay them unnecessarily.
 274
 275         while (! timerQueue_.empty()) {
 276             TimerMap::iterator i (timerQueue_.top());
 277             if (i->second.canceled)
 278                 ;
 279             else if (i->second.timeout <= eventTime_ + eventEarly_)
 280                 i->second.cb();
 281             else
 282                 break;
 283             timerQueue_.pop();
 284             timerMap_.erase(i);
 285         }
 286
 287         // Check the signal queue
 288         if (events > 0 && ev.data.fd == sigpipe_[0]) {
 289             ::siginfo_t siginfo;
 290             if (::read(sigpipe_[0], &siginfo, sizeof(siginfo)) < int(sizeof(siginfo))) {
 291                 // We ignore truncated records which may only occur if the signal
 292                 // queue became filled up
 293                 SENF_LOG((senf::log::IMPORTANT)("Truncated signal record!"));
 294                 continue;
 295             }
 296             if (siginfo.si_signo < int(sigHandlers_.size()) && sigHandlers_[siginfo.si_signo])
 297                 sigHandlers_[siginfo.si_signo]();
 298             continue;
 299         }
 300
 301         for (FdTable::iterator i = fdTable_.begin(); i != fdTable_.end(); ++i) {
 302             EventSpec & spec (i->second);
 303
 304             if (! (spec.file || (events > 0 && i->first == ev.data.fd)))
 305                 continue;
 306
 307             unsigned extraFlags (0);
 308             unsigned mask (spec.file ? spec.epollMask() : ev.events);
 309
 310             if (mask & EPOLLHUP) extraFlags |= EV_HUP;
 311             if (mask & EPOLLERR) extraFlags |= EV_ERR;
 312
 313             if (mask & EPOLLIN) {
 314                 SENF_ASSERT(spec.cb_read);
 315                 spec.cb_read(EventId(EV_READ | extraFlags));
 316             }
 317             else if (mask & EPOLLPRI) {
 318                 SENF_ASSERT(spec.cb_prio);
 319                 spec.cb_prio(EventId(EV_PRIO | extraFlags));
 320             }
 321             else if (mask & EPOLLOUT) {
 322                 SENF_ASSERT(spec.cb_write);
 323                 spec.cb_write(EventId(EV_WRITE | extraFlags));
 324             }
 325             else {
 326                 // This branch is only taken, if HUP or ERR is signaled but none of IN/OUT/PRI.
 327                 // In this case we will signal all registered callbacks. The callbacks must be
 328                 // prepared to be called multiple times if they are registered to more than
 329                 // one event.
 330                 if (spec.cb_write)
 331                     spec.cb_write(EventId(extraFlags));
 332                 if (spec.cb_prio)
 333                     spec.cb_prio(EventId(extraFlags));
 334                 if (spec.cb_read)
 335                     spec.cb_read(EventId(extraFlags));
 336             }
 337         }
 338     }
 339 }
 340
 341 ///////////////////////////////////////////////////////////////////////////
 342 // senf::SchedulerLogTimeSource
 343
 344 prefix_ senf::log::time_type senf::SchedulerLogTimeSource::operator()()
 345     const
 346 {
 347     return Scheduler::instance().eventTime();
 348 }
 349
 350 ///////////////////////////////cc.e////////////////////////////////////////
 351 #undef prefix_
 352
 353 \f
 354 // Local Variables:
 355 // mode: c++
 356 // fill-column: 100
 357 // c-file-style: "senf"
 358 // indent-tabs-mode: nil
 359 // ispell-local-dictionary: "american"
 360 // compile-command: "scons -u test"
 361 // comment-column: 40
 362 // End: