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