Utils/Daemon/Daemon.cc

   1 // $Id$
   2 //
   3 // Copyright (C) 2007
   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 Daemon non-inline non-template implementation */
  25
  26 #include "Daemon.hh"
  27 #include "Daemon.ih"
  28
  29 // Custom includes
  30 #include <sys/types.h>
  31 #include <sys/stat.h>
  32 #include <sys/wait.h>
  33 #include <unistd.h>
  34 #include <fcntl.h>
  35 #include <errno.h>
  36 #include <signal.h>
  37 #include <execinfo.h>
  38 #include <sstream>
  39 #include <algorithm>
  40 #include <boost/algorithm/string/predicate.hpp>
  41 #include <boost/algorithm/string/trim.hpp>
  42 #include <boost/format.hpp>
  43 #include "../Exception.hh"
  44 #include "../membind.hh"
  45 #include "../Backtrace.hh"
  46
  47 // #define __USE_GNU
  48 #include <ucontext.h>
  49
  50 //#include "Daemon.mpp"
  51 #define prefix_
  52 ///////////////////////////////cc.p////////////////////////////////////////
  53
  54 #define LIBC_CALL(fn, args) if (fn args < 0) \
  55     SENF_THROW_SYSTEM_EXCEPTION(#fn "()")
  56
  57 #define LIBC_CALL_RV(var, fn, args) \
  58     int var (fn args); if (var < 0) SENF_THROW_SYSTEM_EXCEPTION(#fn "()")
  59
  60 ///////////////////////////////////////////////////////////////////////////
  61 // senf::Daemon
  62
  63 prefix_ senf::Daemon::~Daemon()
  64 {
  65     if (pidfileCreated_) {
  66         try {
  67             LIBC_CALL( ::unlink, (pidfile_.c_str()) );
  68         } catch (Exception e) {
  69             // e << "; could not unlink " << pidfile_.c_str();
  70             // throw;
  71         }
  72     }
  73 }
  74
  75 prefix_ void senf::Daemon::daemonize(bool v)
  76 {
  77     daemonize_ = v;
  78 }
  79
  80 prefix_ bool senf::Daemon::daemon()
  81 {
  82     return daemonize_;
  83 }
  84
  85 prefix_ int senf::Daemon::argc()
  86 {
  87     return argc_;
  88 }
  89
  90 prefix_ char ** senf::Daemon::argv()
  91 {
  92     return argv_;
  93 }
  94
  95 prefix_ void senf::Daemon::consoleLog(std::string const & path, StdStream which)
  96 {
  97     switch (which) {
  98     case StdOut : stdoutLog_ = path; break;
  99     case StdErr : stderrLog_ = path; break;
 100     case Both : stdoutLog_ = path; stderrLog_ = path; break;
 101     }
 102 }
 103
 104
 105 prefix_ void senf::Daemon::openLog()
 106 {
 107     int fd (-1);
 108     if (! stdoutLog_.empty()) {
 109         fd = ::open(stdoutLog_.c_str(), O_WRONLY | O_APPEND | O_CREAT, 0666);
 110         if (fd < 0)
 111             SENF_THROW_SYSTEM_EXCEPTION("")
 112                   << " Could not open \"" << stdoutLog_ << "\" for redirecting stdout.";
 113         stdout_ = fd;
 114     }
 115     if (stderrLog_ == stdoutLog_)
 116         stderr_ = fd;
 117     else if (! stderrLog_.empty()) {
 118         fd = ::open(stdoutLog_.c_str(), O_WRONLY | O_APPEND | O_CREAT, 0666);
 119         if (fd < 0)
 120             SENF_THROW_SYSTEM_EXCEPTION("")
 121                   << " Could not open \"" << stderrLog_ << "\" for redirecting stderr.";
 122         stderr_ = fd;
 123     }
 124 }
 125
 126 prefix_ void senf::Daemon::logReopen()
 127 {
 128     if (! stdoutLog_.empty()) {
 129         int fd (::open(stdoutLog_.c_str(), O_WRONLY | O_APPEND | O_CREAT, 0666));
 130         if (fd < 0)
 131             goto error;
 132         if (::dup2(fd, 1) < 0)
 133             goto error;
 134         if (stderrLog_ == stdoutLog_) {
 135             if (::dup2(fd, 2) < 0)
 136                 goto error;
 137             return;
 138         }
 139     }
 140     if (! stderrLog_.empty()) {
 141         int fd (::open(stderrLog_.c_str(), O_WRONLY | O_APPEND | O_CREAT, 0666));
 142         if (fd < 0)
 143             goto error;
 144         if (::dup2(fd, 2) < 0)
 145             goto error;
 146     }
 147     return;
 148
 149  error:
 150     SENF_LOG(
 151         (senf::log::CRITICAL)
 152         ("log-file reopen failed: (" << errno << ") " << ::strerror(errno)) );
 153 }
 154
 155 prefix_ void senf::Daemon::pidFile(std::string const & f)
 156 {
 157     pidfile_ = f;
 158 }
 159
 160 namespace {
 161     bool signaled (false);
 162     void waitusr(int) {
 163         signaled = true;
 164     }
 165 }
 166
 167 prefix_ void senf::Daemon::detach()
 168 {
 169     if (daemonize_ && ! detached_) {
 170         // Wow .. ouch ..
 171         // To ensure all data is written to the console log file in the correct order, we suspend
 172         // execution here until the parent process tells us to continue via SIGUSR1: We block
 173         // SIGUSR1 and install our own signal handler saving the old handler and signal mask. Then
 174         // we close stdin/stderr which will send a HUP condition to the parent process. We wait for
 175         // SIGUSR1 and reinstall the old signal mask and action.
 176         ::sigset_t oldsig;
 177         ::sigset_t usrsig;
 178         ::sigemptyset(&usrsig);
 179         LIBC_CALL( ::sigaddset, (&usrsig, SIGUSR1) );
 180         LIBC_CALL( ::sigprocmask, (SIG_BLOCK, &usrsig, &oldsig) );
 181         struct ::sigaction oldact;
 182         struct ::sigaction usract;
 183         ::memset(&usract, 0, sizeof(usract));
 184         usract.sa_handler = &waitusr;
 185         LIBC_CALL( ::sigaction, (SIGUSR1, &usract, &oldact) );
 186         ::sigset_t waitsig (oldsig);
 187         LIBC_CALL( ::sigdelset, (&waitsig, SIGUSR1) );
 188
 189         LIBC_CALL_RV( nul, ::open, ("/dev/null", O_WRONLY) );
 190         LIBC_CALL( ::dup2, (stdout_ == -1 ? nul : stdout_, 1) );
 191         LIBC_CALL( ::dup2, (stderr_ == -1 ? nul : stderr_, 2) );
 192         LIBC_CALL( ::close, (nul) );
 193
 194         signaled = false;
 195         while (! signaled) {
 196             ::sigsuspend(&waitsig);
 197             if (errno != EINTR)
 198                 SENF_THROW_SYSTEM_EXCEPTION("::sigsuspend()");
 199         }
 200
 201         LIBC_CALL( ::sigaction, (SIGUSR1, &oldact, 0) );
 202         LIBC_CALL( ::sigprocmask, (SIG_SETMASK, &oldsig, 0) );
 203
 204         detached_ = true;
 205     }
 206 }
 207
 208 namespace {
 209     /* Purposely *not* derived from std::exception */
 210     struct DaemonExitException {
 211         DaemonExitException(unsigned c) : code(c) {}
 212         unsigned code;
 213     };
 214 }
 215
 216 prefix_ void senf::Daemon::exit(unsigned code)
 217 {
 218     throw DaemonExitException(code);
 219 }
 220
 221 prefix_ int senf::Daemon::start(int argc, char ** argv)
 222 {
 223     argc_ = argc;
 224     argv_ = argv;
 225
 226     try {
 227         configure();
 228
 229         if (daemonize_) {
 230             openLog();
 231             fork();
 232         }
 233         installSighandlers();
 234         if (! pidfile_.empty()) {
 235             if (pidfileCreate())
 236                 pidfileCreated_ = true;
 237             else {
 238                 std::cerr << "PID file '" << pidfile_
 239                           << "' creation failed. Daemon running ?" << std::endl;
 240                 return 1;
 241             }
 242         }
 243
 244         main();
 245     }
 246     catch (DaemonExitException & e) {
 247         return e.code;
 248     }
 249
 250 #ifndef SENF_DEBUG
 251
 252     catch (std::exception & e) {
 253         std::cerr << "\n*** Fatal exception: " << e.what() << "\n" << std::endl;
 254         return 1;
 255     }
 256     catch (...) {
 257         std::cerr << "\n*** Fatal exception: (unknown)" << "\n" << std::endl;
 258         return 1;
 259     }
 260
 261 #   endif
 262
 263     return 0;
 264 }
 265
 266 prefix_ senf::Daemon & senf::Daemon::instance()
 267 {
 268     BOOST_ASSERT( instance_ );
 269     return *instance_;
 270 }
 271
 272 ////////////////////////////////////////
 273 // protected members
 274
 275 prefix_ senf::Daemon::Daemon()
 276     : argc_(0), argv_(0), daemonize_(true), stdout_(-1), stderr_(-1), pidfile_(""),
 277       pidfileCreated_(false), detached_(false)
 278 {
 279     BOOST_ASSERT( ! instance_ );
 280     instance_ = this;
 281 }
 282
 283 senf::Daemon * senf::Daemon::instance_ (0);
 284
 285 ////////////////////////////////////////
 286 // private members
 287
 288 prefix_ void senf::Daemon::configure()
 289 {
 290     for (int i (1); i<argc_; ++i) {
 291         if (argv_[i] == std::string("--no-daemon"))
 292             daemonize(false);
 293         else if (boost::starts_with(argv_[i], std::string("--console-log="))) {
 294             std::string arg (std::string(argv_[i]), 14u);
 295             std::string::size_type komma (arg.find(','));
 296             if (komma == std::string::npos) {
 297                 boost::trim(arg);
 298                 if (arg == std::string("none")) consoleLog("");
 299                 else if (!arg.empty())          consoleLog(arg);
 300             } else {
 301                 std::string arg1 (arg,0,komma);
 302                 std::string arg2 (arg,komma+1);
 303                 boost::trim(arg1);
 304                 boost::trim(arg2);
 305                 if (arg1 == std::string("none")) consoleLog("",StdOut);
 306                 else if (! arg1.empty() )        consoleLog(arg1, StdOut);
 307                 if (arg2 == std::string("none")) consoleLog("",StdErr);
 308                 else if (! arg2.empty() )        consoleLog(arg2, StdErr);
 309             }
 310         }
 311         else if (boost::starts_with(argv_[i], std::string("--pid-file=")))
 312             pidFile(std::string(std::string(argv_[i]), 11u));
 313     }
 314 }
 315
 316 prefix_ void senf::Daemon::main()
 317 {
 318     init();
 319     detach();
 320     run();
 321 }
 322
 323 prefix_ void senf::Daemon::init()
 324 {}
 325
 326 prefix_ void senf::Daemon::run()
 327 {}
 328
 329 prefix_ void senf::Daemon::fork()
 330 {
 331     int coutpipe[2];
 332     int cerrpipe[2];
 333
 334     LIBC_CALL_RV( nul, ::open, ("/dev/null", O_RDONLY) );
 335     LIBC_CALL( ::dup2, (nul, 0) );
 336     LIBC_CALL( ::close, (nul) );
 337     LIBC_CALL( ::pipe, (coutpipe) );
 338     LIBC_CALL( ::pipe, (cerrpipe) );
 339
 340     // We need to block the SIGCHLD signal here so we don't miss it, if the child
 341     // dies immediately
 342     ::sigset_t oldsig;
 343     ::sigset_t cldsig;
 344     ::sigemptyset(&cldsig);
 345     LIBC_CALL( ::sigaddset, (&cldsig, SIGCHLD) );
 346     LIBC_CALL( ::sigprocmask, (SIG_BLOCK, &cldsig, &oldsig) );
 347
 348     LIBC_CALL_RV( pid, ::fork, () );
 349
 350     if (pid == 0) {
 351         // Daemon process
 352
 353         LIBC_CALL( ::dup2, (coutpipe[1],1) );
 354         LIBC_CALL( ::dup2, (cerrpipe[1],2) );
 355         LIBC_CALL( ::close, (coutpipe[0]) );
 356         LIBC_CALL( ::close, (coutpipe[1]) );
 357         LIBC_CALL( ::close, (cerrpipe[0]) );
 358         LIBC_CALL( ::close, (cerrpipe[1]) );
 359         LIBC_CALL( ::setsid, () );
 360         LIBC_CALL( ::sigprocmask, (SIG_SETMASK, &oldsig, 0) );
 361         return;
 362     }
 363
 364     // Ouch ... ensure, the daemon watcher does not remove the pidfile ...
 365     pidfile_ = "";
 366
 367     LIBC_CALL( ::close, (coutpipe[1]) );
 368     LIBC_CALL( ::close, (cerrpipe[1]) );
 369
 370     detail::DaemonWatcher watcher (pid, coutpipe[0], cerrpipe[0], stdout_, stderr_);
 371     watcher.run();
 372
 373     ::_exit(0);
 374 }
 375
 376 prefix_ bool senf::Daemon::pidfileCreate()
 377 {
 378     // Create temporary file pidfile_.hostname.pid and hard-link it to pidfile_ If the hardlink
 379     // fails, the pidfile exists. If the link count of the temporary file is not 2 after this, there
 380     // was some race condition, probably over NFS.
 381
 382     std::string tempname;
 383     boost::format linkErrorFormat(" Could not link \"%1%\" to \"%2%\".");
 384
 385     {
 386         char hostname[HOST_NAME_MAX+1];
 387         LIBC_CALL( ::gethostname, (hostname, HOST_NAME_MAX+1) );
 388         hostname[HOST_NAME_MAX] = 0;
 389         std::stringstream tempname_s;
 390         tempname_s << pidfile_ << "." << hostname << "." << ::getpid();
 391         tempname = tempname_s.str();
 392     }
 393
 394     while (1) {
 395         {
 396             std::ofstream pidf (tempname.c_str());
 397             if (! pidf)
 398                 SENF_THROW_SYSTEM_EXCEPTION("")
 399                       << " Could not open pidfile \"" << tempname << "\" for output.";
 400             pidf << ::getpid() << std::endl;
 401             if (! pidf)
 402                 SENF_THROW_SYSTEM_EXCEPTION("")
 403                       << " Could not write to pidfile \"" << tempname << "\".";
 404         }
 405
 406         if (::link(tempname.c_str(), pidfile_.c_str()) < 0) {
 407             if (errno != EEXIST)
 408                 SENF_THROW_SYSTEM_EXCEPTION("") << linkErrorFormat % pidfile_ % tempname;
 409         }
 410         else {
 411             struct ::stat s;
 412             LIBC_CALL( ::stat, (tempname.c_str(), &s) );
 413             LIBC_CALL( ::unlink, (tempname.c_str()) );
 414             return s.st_nlink == 2;
 415         }
 416
 417         // pidfile exists. Check, whether the pid in the pidfile still exists.
 418         {
 419             int old_pid (-1);
 420             std::ifstream pidf (pidfile_.c_str());
 421             if ( ! (pidf >> old_pid)
 422                  || old_pid < 0
 423                  || ::kill(old_pid, 0) >= 0
 424                  || errno == EPERM ) {
 425                 LIBC_CALL( ::unlink, (tempname.c_str()) );
 426                 return false;
 427             }
 428         }
 429
 430         // If we reach this point, the pid file exists but the process mentioned within the
 431         // pid file does *not* exists. We assume, the pid file to be stale.
 432
 433         // I hope, the following procedure is without race condition: We remove our generated
 434         // temporary pid file and recreate it as hard-link to the old pid file. Now we check, that
 435         // the hard-link count of this file is 2. If it is not, we terminate, since someone else
 436         // must have already created his hardlink. We then truncate the file and write our pid.
 437
 438         LIBC_CALL( ::unlink, (tempname.c_str() ));
 439         if (::link(pidfile_.c_str(), tempname.c_str()) < 0) {
 440             if (errno != ENOENT)
 441                 SENF_THROW_SYSTEM_EXCEPTION("") << linkErrorFormat % tempname % pidfile_;
 442             // Hmm ... the pidfile mysteriously disappeared ... try again.
 443             continue;
 444         }
 445
 446         {
 447             struct ::stat s;
 448             LIBC_CALL( ::stat, (tempname.c_str(), &s) );
 449             if (s.st_nlink != 2) {
 450                 LIBC_CALL( ::unlink, (tempname.c_str()) );
 451                 return false;
 452             }
 453         }
 454
 455         {
 456             std::ofstream pidf (tempname.c_str());
 457             pidf << ::getpid() << std::endl;
 458         }
 459
 460         LIBC_CALL( ::unlink, (tempname.c_str()) );
 461         break;
 462     }
 463     return true;
 464 }
 465
 466
 467 #ifdef SENF_DEBUG
 468
 469 namespace {
 470     void fatalSignalsHandler(int sig, ::siginfo_t * info, void * arg)
 471     {
 472         static char const * const signames[] = {
 473             "",
 474             "SIGHUP", "SIGINT", "SIGQUIT", "SIGILL", "SIGTRAP", "SIGABRT", "SIGBUS", "SIGFPE",
 475             "SIGKILL", "SIGUSR1", "SIGSEGV", "SIGUSR2", "SIGPIPE", "SIGALRM", "SIGTERM",
 476             "SIGSTKFLT", "SIGCHLD", "SIGCONT", "SIGSTOP", "SIGTSTP", "SIGTTIN", "SIGTTOU",
 477             "SIGURG", "SIGXCPU", "SIGXFSZ", "SIGVTALRM", "SIGPROF", "SIGWINCH", "SIGIO",
 478             "SIGPWR", "SIGSYS" };
 479
 480         // ::ucontext_t * ucontext = static_cast<ucontext_t*>(arg);
 481         std::cerr << "\n" << "Signal " << sig;
 482         if (unsigned(sig) < sizeof(signames) / sizeof(signames[0]))
 483             std::cerr << " (" << signames[unsigned(sig)] << ")";
 484         std::cerr << " received\n";
 485
 486         if (sig == SIGSEGV)
 487             std::cerr << "Invalid memory access at " << info->si_addr << "\n";
 488
 489         static void * entries[SENF_DEBUG_BACKTRACE_NUMCALLERS];
 490         unsigned nEntries( ::backtrace(entries, SENF_DEBUG_BACKTRACE_NUMCALLERS) );
 491
 492         // Hack the callers address into the backtrace
 493         // entries[1] = reinterpret_cast<void *>(ucontext->uc_mcontext.gregs[REG_EIP]);
 494
 495         std::cerr << "Backtrace:\n";
 496         senf::formatBacktrace(std::cerr, entries, nEntries);
 497         std::cerr << "-- \n";
 498
 499         if (sig != SIGUSR2) {
 500             ::signal(sig, SIG_DFL);
 501             ::kill(::getpid(), sig);
 502         }
 503     }
 504
 505 }
 506
 507 #endif
 508
 509 namespace {
 510     void sighupHandler(int sig)
 511     {
 512         senf::Daemon::instance().logReopen();
 513     }
 514 }
 515
 516 prefix_ void senf::Daemon::installSighandlers()
 517 {
 518     struct ::sigaction sa;
 519
 520     ::sigemptyset(&sa.sa_mask);
 521     sa.sa_handler = &sighupHandler;
 522     sa.sa_flags = SA_RESTART;
 523
 524     ::sigaction(SIGHUP,   &sa, NULL);
 525
 526 #ifdef SENF_DEBUG
 527     sa.sa_sigaction = &fatalSignalsHandler;
 528     sa.sa_flags = SA_RESTART | SA_SIGINFO;
 529
 530     ::sigaction(SIGILL,    &sa, NULL);
 531     ::sigaction(SIGTRAP,   &sa, NULL);
 532     ::sigaction(SIGABRT,   &sa, NULL);
 533     ::sigaction(SIGFPE,    &sa, NULL);
 534     ::sigaction(SIGBUS,    &sa, NULL);
 535     ::sigaction(SIGSEGV,   &sa, NULL);
 536     ::sigaction(SIGSTKFLT, &sa, NULL);
 537     ::sigaction(SIGSYS,    &sa, NULL);
 538     ::sigaction(SIGUSR2,   &sa, NULL);
 539 #endif
 540 }
 541
 542 ///////////////////////////////////////////////////////////////////////////
 543 // senf::detail::DaemonWatcher
 544
 545 prefix_ senf::detail::DaemonWatcher::DaemonWatcher(int pid, int coutpipe, int cerrpipe,
 546                                                    int stdout, int stderr)
 547     : childPid_(pid), coutpipe_(coutpipe), cerrpipe_(cerrpipe), stdout_(stdout),
 548       stderr_(stderr), sigChld_(false),
 549       coutForwarder_(coutpipe_, boost::bind(&DaemonWatcher::pipeClosed, this, 1)),
 550       cerrForwarder_(cerrpipe_, boost::bind(&DaemonWatcher::pipeClosed, this, 2))
 551 {
 552     coutForwarder_.addTarget(1);
 553     if (stdout_ >= 0)
 554         coutForwarder_.addTarget(stdout_);
 555     cerrForwarder_.addTarget(2);
 556     if (stderr_ >= 0)
 557         cerrForwarder_.addTarget(stderr_);
 558 }
 559
 560 prefix_ void senf::detail::DaemonWatcher::run()
 561 {
 562     Scheduler::instance().registerSignal(SIGCHLD, senf::membind(&DaemonWatcher::sigChld, this));
 563     Scheduler::instance().process();
 564 }
 565
 566 ////////////////////////////////////////
 567 // private members
 568
 569 prefix_ void senf::detail::DaemonWatcher::pipeClosed(int id)
 570 {
 571     switch (id) {
 572     case 1 : coutpipe_ = -1; break;
 573     case 2 : cerrpipe_ = -1; break;
 574     }
 575
 576     if (coutpipe_ == -1 && cerrpipe_ == -1) {
 577         if (sigChld_)
 578             childDied(); // does not return
 579         if (::kill(childPid_, SIGUSR1) < 0)
 580             if (errno != ESRCH) SENF_THROW_SYSTEM_EXCEPTION("::kill()");
 581         Scheduler::instance().timeout(
 582             Scheduler::instance().eventTime() + ClockService::seconds(1),
 583             senf::membind(&DaemonWatcher::childOk, this));
 584     }
 585 }
 586
 587 prefix_ void senf::detail::DaemonWatcher::sigChld()
 588 {
 589     sigChld_ = true;
 590     if (coutpipe_ == -1 && cerrpipe_ == -1)
 591         childDied(); // does not return
 592 }
 593
 594 prefix_ void senf::detail::DaemonWatcher::childDied()
 595 {
 596     int status (0);
 597     if (::waitpid(childPid_,&status,0) < 0) SENF_THROW_SYSTEM_EXCEPTION("::waitpid()");
 598     if (WIFSIGNALED(status)) {
 599         ::signal(WTERMSIG(status),SIG_DFL);
 600         ::kill(::getpid(), WTERMSIG(status));
 601         // should not be reached
 602         ::_exit(1);
 603     }
 604     if (WEXITSTATUS(status) == 0)
 605         ::_exit(1);
 606     ::_exit(WEXITSTATUS(status));
 607 }
 608
 609 prefix_ void senf::detail::DaemonWatcher::childOk()
 610 {
 611     Scheduler::instance().terminate();
 612 }
 613
 614 ///////////////////////////////////////////////////////////////////////////
 615 // senf::detail::DaemonWatcher::Forwarder
 616
 617 prefix_ senf::detail::DaemonWatcher::Forwarder::Forwarder(int src, Callback cb)
 618     : src_(src), cb_(cb)
 619 {
 620     Scheduler::instance().add(src_, senf::membind(&Forwarder::readData, this),
 621                               Scheduler::EV_READ);
 622 }
 623
 624 prefix_ senf::detail::DaemonWatcher::Forwarder::~Forwarder()
 625 {
 626     if (src_ != -1)
 627         Scheduler::instance().remove(src_);
 628
 629     for (Targets::iterator i (targets_.begin()); i != targets_.end(); ++i)
 630         if (i->offset >= buffer_.size())
 631             Scheduler::instance().remove(i->fd);
 632 }
 633
 634 prefix_ void senf::detail::DaemonWatcher::Forwarder::addTarget(int fd)
 635 {
 636     Target target = { fd, 0 };
 637     targets_.push_back(target);
 638 }
 639
 640 prefix_ void senf::detail::DaemonWatcher::Forwarder::readData(Scheduler::EventId event)
 641 {
 642     char buf[1024];
 643     int n (0);
 644
 645     while (1) {
 646         n = ::read(src_,buf,1024);
 647         if (n<0) {
 648             if (errno != EINTR) SENF_THROW_SYSTEM_EXCEPTION("::read()");
 649         } else
 650             break;
 651     }
 652
 653     if (n == 0) {
 654         // Hangup
 655         Scheduler::instance().remove(src_);
 656         if (buffer_.empty())
 657             cb_();
 658         src_ = -1;
 659         return;
 660     }
 661
 662     if (targets_.empty())
 663         return;
 664
 665     for (Targets::iterator i (targets_.begin()); i != targets_.end(); ++i)
 666         if (i->offset >= buffer_.size())
 667             Scheduler::instance().add( i->fd,
 668                                        boost::bind(&Forwarder::writeData, this, _1, i),
 669                                        Scheduler::EV_WRITE );
 670
 671     buffer_.insert(buffer_.end(), buf, buf+n);
 672 }
 673
 674 prefix_ void senf::detail::DaemonWatcher::Forwarder::writeData(Scheduler::EventId event,
 675                                                                Targets::iterator target)
 676 {
 677     if (event != Scheduler::EV_WRITE) {
 678         // Broken pipe while writing data ? Not much, we can do here, we just drop the data
 679         Scheduler::instance().remove(target->fd);
 680         targets_.erase(target);
 681         if (targets_.empty() && src_ == -1)
 682             cb_();
 683         return;
 684     }
 685
 686     char buf[1024];
 687     int n (buffer_.size() - target->offset > 1024 ? 1024 : buffer_.size() - target->offset);
 688     std::copy(buffer_.begin() + target->offset, buffer_.begin() + target->offset + n, buf);
 689
 690     int w (::write(target->fd, buf, n));
 691     if (w < 0) {
 692         if (errno != EINTR) SENF_THROW_SYSTEM_EXCEPTION("::write()");
 693         return;
 694     }
 695     target->offset += w;
 696
 697     n = std::min_element(
 698         targets_.begin(), targets_.end(),
 699         boost::bind(&Target::offset, _1) < boost::bind(&Target::offset, _2))->offset;
 700
 701     buffer_.erase(buffer_.begin(), buffer_.begin()+n);
 702
 703     for (Targets::iterator i (targets_.begin()); i != targets_.end(); ++i)
 704         i->offset -= n;
 705
 706     if (target->offset >= buffer_.size())
 707         Scheduler::instance().remove(target->fd);
 708     if (src_ == -1 && (buffer_.empty() || targets_.empty()))
 709         cb_();
 710 }
 711
 712 #undef LIBC_CALL
 713 #undef LIBC_CALL_RV
 714
 715 ///////////////////////////////cc.e////////////////////////////////////////
 716 #undef prefix_
 717 //#include "Daemon.mpp"
 718
 719 \f
 720 // Local Variables:
 721 // mode: c++
 722 // fill-column: 100
 723 // comment-column: 40
 724 // c-file-style: "senf"
 725 // indent-tabs-mode: nil
 726 // ispell-local-dictionary: "american"
 727 // compile-command: "scons -u test"
 728 // End: