libplumbing/reactor.cc

Go to the documentation of this file.

//
// plumbing - C++ wrapper facades for Berkeley sockets
// Copyright (C) 2009 Peter Miller
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU General Public License, version 3, as published
// by the Free Software Foundation.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along with
// this program. If not, see <http://www.gnu.org/licenses/>.
//

#include <libplumbing/config.h>
#include <cerrno>
#include <signal.h>
#include <libexplain/select.h>

#include <libplumbing/reactor.h>
#include <libplumbing/endpoint.h>
#include <libplumbing/endpoint/functor.h>
#include <libplumbing/logging.h>
#include <libplumbing/time_value.h>


plumbing::reactor::~reactor()
{
}


void
plumbing::reactor::fatal_error(const char *fmt, ...)
{
    va_list ap;
    va_start(ap, fmt);
    fatal_error_v(fmt, ap);
    va_end(ap);
}


void
plumbing::reactor::fatal_error_v(const char *fmt, va_list ap)
{
    log().fatal_error_v(fmt, ap);
}


plumbing::reactor::reactor()
{
    //
    // Ignore pipe signals, so that we get EPIPE errors instead.
    //
    signal(SIGPIPE, SIG_IGN);

    //
    // Prevents zombies being created when children exit with no wait()
    //
    signal(SIGCHLD, SIG_IGN);
}


void
plumbing::reactor::kill_suicidal_endpoints()
{
    endpoints_t survivors;
    for
    (
        endpoints_t::iterator it = endpoints.begin();
        it != endpoints.end();
        ++it
    )
    {
        endpoint::pointer ep = *it;
        if (!ep->is_ready_to_die())
            survivors.push_back(ep);
    }
    endpoints.swap(survivors);
}


void
plumbing::reactor::process(bool block)
{
    fd_set readable;
    FD_ZERO(&readable);
    fd_set writable;
    FD_ZERO(&writable);

    //
    // Ask all of the endpoints to register their interest in their file
    // descriptors.
    //
    time_value maximum_sleep = at.get_maximum_sleep();
    int num_waiting = 0;
    for
    (
        endpoints_t::iterator it = endpoints.begin();
        it != endpoints.end();
        ++it
    )
    {
        endpoint::pointer ep = *it;
        assert(ep);
        int rfd = ep->get_read_file_descriptor();
        if (rfd >= 0)
        {
            FD_SET(rfd, &readable);
            ++num_waiting;
        }
        int wfd = ep->get_write_file_descriptor();
        if (wfd >= 0)
        {
            FD_SET(wfd, &writable);
            ++num_waiting;
        }

        time_value dt = ep->get_maximum_sleep();
        if (maximum_sleep > dt)
            maximum_sleep = dt;
    }

    //
    // See if any endpoints are ready to give up the ghost.
    // We test this here, after the file descriptos are calculated,
    // so that connect errors can call kill_me_now().
    //
    kill_suicidal_endpoints();

    //
    // If there are no end points and no at jobs, don't bother with
    // select, just bail.
    //
    if (endpoints.empty() && at.empty())
        return;

    //
    // The timeout is zero if we were told not to block, otherwise it is
    // the maximum_sleep calculated above.
    //
    time_value time_out(block ? maximum_sleep : time_value(0));

    //
    // Hello?  Anyone there?
    //
    int n = select(FD_SETSIZE, &readable, &writable, 0, &time_out);
    if (n < 0)
    {
        int err = errno;
        if (err != EINTR && err != EBADF)
        {
            //
            // EBADF can happen if an endpoint returns a file descriptor,
            // but gets removed by the kill_suicidal_endpoints method.
            //
            log().fatal_error
            (
                "%s",
                explain_errno_select
                (
                    err,
                    FD_SETSIZE,
                    &readable,
                    &writable,
                    0,
                    &time_out
                )
            );
        }
        return;
    }

    //
    // From the select(2) man page:
    // "On success, select() returns the number of descriptors contained
    // in the three returned descriptor sets (that is, the total number
    // of one bits in readfds, writefds, exceptfds) which may be zero if
    // the timeout expires before anything interesting happens."
    //
    if (n == 0)
    {
        for
        (
            endpoints_t::iterator it = endpoints.begin();
            it != endpoints.end();
            ++it
        )
        {
            endpoint::pointer ep = *it;
            ep->process_timeout();
        }
    }
    if (n > 0)
    {
        //
        // See if any of the endpoints need to do something.
        //
        for
        (
            endpoints_t::iterator it = endpoints.begin();
            it != endpoints.end();
            ++it
        )
        {
            endpoint::pointer ep = *it;

            //
            // NOTE: it is very important that you DO NOT change
            // the file descriptor to be -1 within an endpoint's
            // get_read_file_descriptor or get_write_file_descriptor
            // methods.  These methods MUST consistently return the
            // same thing before and after the select, but before
            // process_read and/or process_write are called.
            //
            int rfd = ep->get_read_file_descriptor();
            if (rfd >= 0 && FD_ISSET(rfd, &readable))
            {
                FD_CLR(rfd, &readable);
                ep->process_read();
                --n;
            }
            int wfd = ep->get_write_file_descriptor();
            if (wfd >= 0 && FD_ISSET(wfd, &writable))
            {
                FD_CLR(wfd, &writable);
                ep->process_write();
                --n;
            }
            if (n <= 0)
                break;
        }
    }

    //
    // See if there are any time-specific jobs to be run.
    //
    // We run them *after* the file descriptor processing, in case they
    // destroy at_job objects, giving the jobs a chance to be
    // removed from the queue elegantly.
    //
    at.process();
}


void
plumbing::reactor::add_endpoint(const endpoint::pointer &sp)
{
    endpoints.push_back(sp);
}


void
plumbing::reactor::for_each_endpoint(endpoint_functor &func)
{
    for
    (
        endpoints_t::iterator it = endpoints.begin();
        it != endpoints.end();
        ++it
    )
    {
        endpoint::pointer ep = *it;
        func(ep);
    }
}


// vim: set ts=8 sw=4 et

---