BidirMMapPipe.h

Go to the documentation of this file.

/** @file BidirMMapPipe.h
 *
 * header file for BidirMMapPipe, a class which forks off a child process and
 * serves as communications channel between parent and child
 *
 * @author Manuel Schiller <manuel.schiller@nikhef.nl>
 * @date 2013-07-07
 */

#ifndef BIDIRMMAPPIPE_H
#define BIDIRMMAPPIPE_H

#include <list>
#include <vector>
#include <cassert>
#include <cstring>
#include <unistd.h>

#define BEGIN_NAMESPACE_ROOFIT namespace RooFit {
#define END_NAMESPACE_ROOFIT }

BEGIN_NAMESPACE_ROOFIT

/// namespace for implementation details of BidirMMapPipe
namespace BidirMMapPipe_impl {
    // forward declarations
    class BidirMMapPipeException;
    class Page;
    class PagePool;
    class Pages;

    /** @brief class representing a chunk of pages
     *
     * @author Manuel Schiller <manuel.schiller@nikhef.nl>
     * @date 2013-07-24
     *
     * allocating pages from the OS happens in chunks in order to not exhaust
     * the maximum allowed number of memory mappings per process; this class
     * takes care of such a chunk
     *
     * a page chunk allows callers to obtain or release pages in groups of
     * continuous pages of fixed size
     */
    class PageChunk {
   public:
       /// type of mmap support found
       typedef enum {
      Unknown, ///< don't know yet what'll work
      Copy,    ///< mmap doesn't work, have to copy back and forth
      FileBacked, ///< mmapping a temp file works
      DevZero, ///< mmapping /dev/zero works
      Anonymous   ///< anonymous mmap works
       } MMapVariety;

   private:
       static unsigned s_pagesize; ///< system page size (run-time determined)
       /// mmap variety that works on this system
       static MMapVariety s_mmapworks;
   
       /// convenience typedef
       typedef BidirMMapPipeException Exception;

       void* m_begin;   ///< pointer to start of mmapped area
       void* m_end;  ///< pointer one behind end of mmapped area
       // FIXME: cannot keep freelist inline - other end may need that
       //        data, and we'd end up overwriting the page header
       std::list<void*> m_freelist; ///< free pages list
       PagePool* m_parent; ///< parent page pool
       unsigned m_nPgPerGrp; ///< number of pages per group
       unsigned m_nUsedGrp; ///< number of used page groups

       /// determine page size at run time
       static unsigned getPageSize();

       /// mmap pages, len is length of mmapped area in bytes
       static void* dommap(unsigned len);
       /// munmap pages p, len is length of mmapped area in bytes
       static void domunmap(void* p, unsigned len);
       /// forbid copying
       PageChunk(const PageChunk&) {}
       /// forbid assignment
       PageChunk& operator=(const PageChunk&) { return *this; }
   public:
       /// return the page size of the system
       static unsigned pagesize() { return s_pagesize; }
       /// return mmap variety support found
       static MMapVariety mmapVariety() { return s_mmapworks; }

       /// constructor
       PageChunk(PagePool* parent, unsigned length, unsigned nPgPerGroup);

       /// destructor
       ~PageChunk();

       /// return if p is contained in this PageChunk
       bool contains(const Pages& p) const;

       /// pop a group of pages off the free list
       Pages pop();

       /// push a group of pages onto the free list
       void push(const Pages& p);

       /// return length of chunk
       unsigned len() const
       {
      return reinterpret_cast<unsigned char*>(m_end) -
          reinterpret_cast<unsigned char*>(m_begin);
       }
       /// return number of pages per page group
       unsigned nPagesPerGroup() const { return m_nPgPerGrp; }

       /// return true if no used page groups in this chunk
       bool empty() const { return !m_nUsedGrp; }

       /// return true if no free page groups in this chunk
       bool full() const { return m_freelist.empty(); }

       /// free all pages except for those pointed to by p
       void zap(Pages& p);
    };

    /** @brief handle class for a number of Pages
     *
     * @author Manuel Schiller <manuel.schiller@nikhef.nl>
     * @date 2013-07-24
     *
     * the associated pages are continuous in memory
     */
    class Pages {
   private:
       /// implementation
       typedef struct {
      PageChunk *m_parent; ///< pointer to parent pool
      Page* m_pages; ///< pointer to first page
      unsigned m_refcnt; ///< reference counter
      unsigned char m_npages; ///< length in pages
       } impl;
   public:
       /// default constructor
       Pages() : m_pimpl(0) { }

       /// destructor
       ~Pages();

       /** @brief copy constructor
        *
        * copy Pages handle to new object - old object loses ownership,
        * and becomes a dangling handle
        */
       Pages(const Pages& other);

       /** @brief assignment operator
        *
        * assign Pages handle to new object - old object loses ownership,
        * and becomes a dangling handle
        */
       Pages& operator=(const Pages& other);

       /// return page size
       static unsigned pagesize();

       /// return number of pages accessible
       unsigned npages() const { return m_pimpl->m_npages; }

       /// return page number pageno
       Page* page(unsigned pgno) const;

       /// return page number pageno
       Page* operator[](unsigned pgno) const { return page(pgno); }

       /// perform page to page number mapping
       unsigned pageno(Page* p) const;

       /// perform page to page number mapping
       unsigned operator[](Page* p) const { return pageno(p); }

       /// swap with other's contents
       void swap(Pages& other)
       {
      impl* tmp = other.m_pimpl;
      other.m_pimpl = m_pimpl;
      m_pimpl = tmp;
       }

   private:
       /// page pool is our friend - it's allowed to construct Pages
       friend class BidirMMapPipe_impl::PageChunk;

       /// pointer to implementation
       impl* m_pimpl;

       /// constructor
       Pages(PageChunk* parent, Page* pages, unsigned npg);
    };
}

/** @brief BidirMMapPipe creates a bidirectional channel between the current
 * process and a child it forks.
 *
 * @author Manuel Schiller <manuel.schiller@nikhef.nl>
 * @date 2013-07-07
 *
 * This class creates a bidirectional channel between this process and a child
 * it creates with fork().
 *
 * The channel is comrised of a small shared pool of buffer memory mmapped into
 * both process spaces, and two pipes to synchronise the exchange of data. The
 * idea behind using the pipes at all is to have some primitive which we can
 * block on without having to worry about atomic operations or polling, leaving
 * these tasks to the OS. In case the anonymous mmap cannot be performed on the
 * OS the code is running on (for whatever reason), the code falls back to
 * mmapping /dev/zero, mmapping a temporary file, or (if those all fail), a
 * dynamically allocated buffer which is then transmitted through the pipe(s),
 * a slightly slower alternative (because the data is copied more often).
 *
 * The channel supports five major operations: read(), write(), flush(),
 * purge() and close(). Reading and writing may block until the required buffer
 * space is available. Writes may queue up data to be sent to the other end
 * until either enough pages are full, or the user calls flush which forces
 * any unsent buffers to be sent to the other end. flush forces any data that
 * is to be sent to be sent. purge discards any buffered data waiting to be
 * read and/or sent. Closing the channel on the child returns zero, closing it
 * on the parent returns the child's exit status.
 *
 * The class also provides operator<< and operator>> for C++-style I/O for
 * basic data types (bool, char, short, int, long, long long, float, double
 * and their unsigned counterparts). Data is transmitted binary (i.e. no
 * formatting to strings like std::cout does). There are also overloads to
 * support C-style zero terminated strings and std::string. In terms of
 * performance, the former is to be preferred.
 *
 * If the caller needs to multiplex input and output to/from several pipes, the
 * class provides the poll() method which allows to block until an event occurs
 * on any of the polled pipes.
 *
 * After the BidirMMapPipe is closed, no further operations may be performed on
 * that object, save for the destructor which may still be called.
 *
 * If the BidirMMapPipe has not properly been closed, the destructor will call
 * close. However, the exit code of the child is lost in that case.
 *
 * Closing the object causes the mmapped memory to be unmapped and the two
 * pipes to be closed. We also install an atexit handler in the process of
 * creating BidirMMapPipes. This ensures that when the current process
 * terminates, a SIGTERM signal is sent to the child processes created for all
 * unclosed pipes to avoid leaving zombie processes in the OS's process table.
 *
 * BidirMMapPipe creation, closing and destruction are thread safe. If the
 * BidirMMapPipe is used in more than one thread, the other operations have to
 * be protected with a mutex (or something similar), though.
 *
 * End of file (other end closed its pipe, or died) is indicated with the eof()
 * method, serious I/O errors set a flags (bad(), fail()), and also throw
 * exceptions. For normal read/write operations, they can be suppressed (i.e.
 * error reporting only using flags) with a constructor argument.
 *
 * Technicalities:
 * - there is a pool of mmapped pages, half the pages are allocated to the
 *   parent process, half to the child 
 * - when one side has accumulated enough data (or a flush forces dirty pages
 *   out to the other end), it sends these pages to the other end by writing a
 *   byte containing the page number into the pipe
 * - the other end (which has the pages mmapped, too) reads the page number(s)
 *   and puts the corresponding pages on its busy list
 * - as the other ends reads, it frees busy pages, and eventually tries to put
 *   them on the its list; if a page belongs to the other end of the
 *   connection, it is sent back
 * - lists of pages are sent across the pipe, not individual pages, in order
 *   to minimise the number of read/write operations needed
 * - when mmap works properly, only one bytes containing the page number of
 *   the page list head is sent back and forth; the contents of that page
 *   allow to access the rest of the page list sent, and page headers on the
 *   list tell the receiving end if the page is free or has to be added to the
 *   busy list
 * - when mmap does not work, we transfer one byte to indicate the head of the
 *   page list sent, and for each page on the list of sent pages, the page
 *   header and the page payload is sent (if the page is free, we only
 *   transmit the page header, and we never transmit more payload than
 *   the page actually contains)
 * - in the child, all open BidirMMapPipes but the current one are closed. this
 *   is done for two reasons: first, to conserve file descriptors and address
 *   space. second, if more than one process is meant to use such a
 *   BidirMMapPipe, synchronisation issues arise which can lead to bugs that
 *   are hard to find and understand. it's much better to come up with a design
 *   which does not need pipes to be shared among more than two processes.
 *
 * Here is a trivial example of a parent and a child talking to each other over
 * a BidirMMapPipe:
 * @code
 * #include <string>
 * #include <iostream>
 * #include <cstdlib>
 *
 * #include "BidirMMapPipe.h"
 *
 * int simplechild(BidirMMapPipe& pipe)
 * {
 *     // child does an echo loop
 *     while (pipe.good() && !pipe.eof()) {
 *         // read a string
 *       std::string str;
 *       pipe >> str;
 *       if (!pipe) return -1;
 *       if (pipe.eof()) break;
 *       // check if parent wants us to shut down
 *       if (!str.empty()) {
 *           std::cout << "[CHILD] :  read: " << str << std::endl;
 *           str = "... early in the morning?";
 *       }
 *         pipe << str << BidirMMapPipe::flush;
 *         if (str.empty()) break;
 *       if (!pipe) return -1;
 *       std::cout << "[CHILD] : wrote: " << str << std::endl;
 *     }
 *     // send shutdown request acknowledged
 *     pipe << "" << BidirMMapPipe::flush;
 *     
 *     pipe.close();
 *     return 0;
 * }
 *  
 * BidirMMapPipe* spawnChild(int (*childexec)(BidirMMapPipe&))
 * {
 *     BidirMMapPipe *p = new BidirMMapPipe();
 *     if (p->isChild()) {
 *         int retVal = childexec(*p);
 *         delete p;
 *         std::exit(retVal);
 *     }
 *     return p;
 * }
 *  
 * int main()
 * {
 *     std::cout << "[PARENT]: simple challenge-response test, one child:" <<
 *             std::endl;
 *     BidirMMapPipe* pipe = spawnChild(simplechild);
 *     for (int i = 0; i < 5; ++i) {
 *         std::string str("What shall we do with a drunken sailor...");
 *         *pipe << str << BidirMMapPipe::flush;
 *         if (!*pipe) return -1;
 *         std::cout << "[PARENT]: wrote: " << str << std::endl;
 *         *pipe >> str;
 *         if (!*pipe) return -1;
 *         std::cout << "[PARENT]:  read: " << str << std::endl;
 *     }
 *     // ask child to shut down
 *     pipe << "" << BidirMMapPipe::flush;
 *     // wait for it to see the shutdown request
 *     std::string s;
 *     pipe >> s;
 *     std::cout << "[PARENT]: exit status of child: " << pipe->close() <<
 *             std::endl;
 *     delete pipe;
 *     return 0;
 * }
 * @endcode
 *
 * When designing your own protocols to use over the pipe, there are a few
 * things to bear in mind:
 * - Do as http does: When building a request, send all the options and
 *   properties of that request with the request itself in a single go (one
 *   flush). Then, the server has everything it needs, and hopefully, it'll
 *   shut up for a while and to let the client do something useful in the
 *   meantime... The same goes when the server replies to the request: include
 *   everything there is to know about the result of the request in the reply.
 * - The expensive operation should be the request that is made, all other
 *   operations should somehow be formulated as options or properties to that
 *   request.
 * - Include a shutdown handshake in whatever protocol you send over the
 *   pipe. That way, you can shut things down in a controlled way. Otherwise,
 *   and depending on your OS's scheduling quirks, you may catch a SIGPIPE if
 *   one end closes its pipe while the other is still trying to read.
 */
class BidirMMapPipe {
#ifndef _WIN32
    public:
   /// type used to represent sizes
   typedef std::size_t size_type;
   /// convenience typedef for BidirMMapPipeException
   typedef BidirMMapPipe_impl::BidirMMapPipeException Exception;
   /// flag bits for partial C++ iostream compatibility
   enum {
       eofbit = 1, ///< end of file reached
       failbit = 2, ///< logical failure (e.g. pipe closed)
       rderrbit = 4, ///< read error
       wrerrbit = 8, ///< write error
       badbit = rderrbit | wrerrbit, ///< general I/O error
       exceptionsbit = 16 ///< error reporting with exceptions
   };

   /** @brief constructor (forks!)
    *
    * Creates a bidirectional communications channel between this process
    * and a child the constructor forks. On return from the constructor,
    * isParent() and isChild() can be used to tell the parent end from the
    * child end of the pipe. In the child, all other open BidirMMapPipes
    * are closed.
    *
    * @param useExceptions read()/write() error reporting also done using
    *          exceptions
    * @param useSocketpair use a socketpair instead of a pair or pipes
    *
    * Normally, exceptions are thrown for all serious I/O errors (apart
    * from end of file). Setting useExceptions to false will force the
    * read() and write() methods to only report serious I/O errors using
    * flags.
    *
    * When useSocketpair is true, use a pair of Unix domain sockets
    * created using socketpair instead a pair of pipes. The advantage is
    * that only one pair of file descriptors is needed instead of two
    * pairs which are needed for the pipe pair. Performance should very
    * similar on most platforms, especially if mmap works, since only
    * very little data is sent through the pipe(s)/socketpair.
    */
   BidirMMapPipe(bool useExceptions = true, bool useSocketpair = false);

   /** @brief destructor
    *
    * closes this end of pipe
    */
   ~BidirMMapPipe();

   /** @brief read from pipe
    *
    * @param addr where to put read data
    * @param sz   size of data to read (in bytes)
    * @returns size of data read, or 0 in case of end-of-file
    *
    * read may block until data from other end is available. It will
    * return 0 if the other end closed the pipe.
    */
   size_type read(void* addr, size_type sz);

   /** @brief wirte to pipe
    *
    * @param addr where to get data to write from
    * @param sz   size of data to write (in bytes)
    * @returns size of data written, or 0 in case of end-of-file
    *
    * write may block until data can be written to other end (depends a
    * bit on available buffer space). It will return 0 if the other end
    * closed the pipe. The data is queued to be written on the next
    * convenient occasion, or it can be forced out with flush().
    */
   size_type write(const void* addr, size_type sz);

   /** @brief flush buffers with unwritten data
    *
    * This forces unwritten data to be written to the other end. The call
    * will block until this has been done (or the attempt failed with an
    * error).
    */
   void flush();

   /** @brief purge buffered data waiting to be read and/or written
    *
    * Discards all internal buffers.
    */
   void purge();

   /** @brief number of bytes that can be read without blocking
    *
    * @returns number of bytes that can be read without blocking
    */
   size_type bytesReadableNonBlocking();

   /** @brief number of bytes that can be written without blocking
    *
    * @returns number of bytes that can be written without blocking
    */
   size_type bytesWritableNonBlocking();

   /** @brief flush buffers, close pipe
    *
    * Flush buffers, discard unread data, closes the pipe. If the pipe is
    * in the parent process, it waits for the child.
    *
    * @returns exit code of child process in parent, zero in child
    */
   int close();

   /** @brief return PID of the process on the other end of the pipe
    *
    * @returns PID of the process running on the remote end
    */
   pid_t pidOtherEnd() const
   { return isChild() ? m_parentPid : m_childPid; }

   /// condition flags for poll
   enum PollFlags {
       None = 0,     ///< nothing special on this pipe
       Readable = 1, ///< pipe has data for reading
       Writable = 2, ///< pipe can be written to
       ReadError = 4,   ///< pipe error read end
       WriteError = 8,  ///< pipe error Write end
       Error = ReadError | WriteError, ///< pipe error
       ReadEndOfFile = 32, ///< read pipe in end-of-file state
       WriteEndOfFile = 64,///< write pipe in end-of-file state
       EndOfFile = ReadEndOfFile | WriteEndOfFile, ///< end of file
       ReadInvalid = 64,   ///< read end of pipe invalid
       WriteInvalid = 128, ///< write end of pipe invalid
       Invalid = ReadInvalid | WriteInvalid ///< invalid pipe
   };

   /// for poll() interface
   class PollEntry {
       public:
      BidirMMapPipe* pipe; ///< pipe of interest
      unsigned events;  ///< events of interest (or'ed bitmask)
      unsigned revents; ///< events that happened (or'ed bitmask)
      /// poll a pipe for all events
      PollEntry(BidirMMapPipe* _pipe) :
          pipe(_pipe), events(None), revents(None) { }
      /// poll a pipe for specified events
      PollEntry(BidirMMapPipe* _pipe, int _events) :
          pipe(_pipe), events(_events), revents(None) { }
   };
   /// convenience typedef for poll() interface
   typedef std::vector<PollEntry> PollVector;

   /** @brief poll a set of pipes for events (ready to read from, ready to
    * write to, error)
    *
    * @param pipes      set of pipes to check
    * @param timeout timeout in milliseconds
    * @returns    positive number: number of pipes which have
    *          status changes, 0: timeout, or no pipes with
    *          status changed, -1 on error
    *
    * Timeout can be zero (check for specified events, and return), finite
    * (wait at most timeout milliseconds before returning), or -1
    * (infinite). The poll method returns when the timeout has elapsed,
    * or if an event occurs on one of the pipes being polled, whichever
    * happens earlier.
    *
    * Pipes is a vector of one or more PollEntries, which each list a pipe
    * and events to poll for. If events is left empty (zero), all
    * conditions are polled for, otherwise only the indicated ones. On
    * return, the revents fields contain the events that occurred for each
    * pipe; error Error, EndOfFile or Invalid events are always set,
    * regardless of wether they were in the set of requested events.
    *
    * poll may block slightly longer than specified by timeout due to OS
    * timer granularity and OS scheduling. Due to its implementation, the
    * poll call can also return early if the remote end of the page sends
    * a free page while polling (which is put on that pipe's freelist),
    * while that pipe is polled for e.g Reading. The status of the pipe is
    * indicated correctly in revents, and the caller can simply poll
    * again. (The reason this is done this way is because it helps to
    * replenish the pool of free pages and queue busy pages without
    * blocking.)
    *
    * Here's a piece of example code waiting on two pipes; if they become
    * readable they are read:
    * @code
    * #include <unistd.h>
    * #include <cstdlib>
    * #include <string>
    * #include <sstream>
    * #include <iostream>
    *
    * #include "BidirMMapPipe.h"
    *
    * // what to execute in the child
         * int randomchild(BidirMMapPipe& pipe)
         * {
         *     ::srand48(::getpid());
         *     for (int i = 0; i < 5; ++i) {
         *         // sleep a random time between 0 and .9 seconds
         *        ::usleep(int(1e6 * ::drand48()));
         *        std::ostringstream buf;
         *        buf << "child pid " << ::getpid() << " sends message " << i;
         *        std::cout << "[CHILD] : " << buf.str() << std::endl;
         *        pipe << buf.str() << BidirMMapPipe::flush;
         *        if (!pipe) return -1;
         *        if (pipe.eof()) break;
         *     }
    *     // tell parent we're done
    *     pipe << "" << BidirMMapPipe::flush;
    *     // wait for parent to acknowledge
    *     std::string s;
    *     pipe >> s;
         *     pipe.close();
         *     return 0;
         * }
         * 
    * // function to spawn a child
         * BidirMMapPipe* spawnChild(int (*childexec)(BidirMMapPipe&))
         * {
         *     BidirMMapPipe *p = new BidirMMapPipe();
         *     if (p->isChild()) {
         *         int retVal = childexec(*p);
         *         delete p;
         *         std::exit(retVal);
         *     }
         *     return p;
         * }
         * 
         * int main()
         * {
         *     typedef BidirMMapPipe::PollEntry PollEntry;
         *     // poll data structure
         *     BidirMMapPipe::PollVector pipes;
    *     pipes.reserve(3);
         *     // spawn children
         *     for (int i = 0; i < 3; ++i) {
         *         pipes.push_back(PollEntry(spawnChild(randomchild),
         *           BidirMMapPipe::Readable));
         *     }
         *     // while at least some children alive
         *     while (!pipes.empty()) {
         *         // poll, wait until status change (infinite timeout)
         *         int npipes = BidirMMapPipe::poll(pipes, -1);
         *         // scan for pipes with changed status
         *         for (std::vector<PollEntry>::iterator it = pipes.begin();
         *                   npipes && pipes.end() != it; ) {
         *             if (!it->revents) {
         *                 // unchanged, next one
         *                 ++it;
         *                 continue;
         *             }
         *             --npipes; // maybe we can stop early...
         *             // read from pipes which are readable
         *             if (it->revents & BidirMMapPipe::Readable) {
         *                 std::string s;
         *                 *(it->pipe) >> s;
    *                 if (!s.empty()) {
         *                 std::cout << "[PARENT]: Read from pipe " <<
    *                     it->pipe << ": " << s << std::endl;
         *                 ++it;
         *                 continue;
    *                 } else {
         *                     // child is shutting down...
    *                     *(it->pipe) << "" << BidirMMapPipe::flush;
    *                     goto childcloses;
         *                 }
         *             }
         *             // retire pipes with error or end-of-file condition
         *             if (it->revents & (BidirMMapPipe::Error |
         *                     BidirMMapPipe::EndOfFile |
    *                       BidirMMapPipe::Invalid)) {
         *                 std::cout << "[PARENT]: Error on pipe " <<
         *                     it->pipe << " revents " << it->revents <<
    *                     std::endl;
    * childcloses:
         *                 std::cout << "[PARENT]:\tchild exit status: " <<
         *                     it->pipe->close() << std::endl;
    *                   if (retVal) return retVal;
         *                 delete it->pipe;
         *                 it = pipes.erase(it);
         *                 continue;
         *             }
         *         }
         *     }
         *     return 0;
         * }
    * @endcode
    */
   static int poll(PollVector& pipes, int timeout);

        /** @brief return if this end of the pipe is the parent end
    *
    * @returns true if parent end of pipe
    */
   bool isParent() const { return m_childPid; }

        /** @brief return if this end of the pipe is the child end
    *
    * @returns true if child end of pipe
    */
   bool isChild() const { return !m_childPid; }

   /** @brief if BidirMMapPipe uses a socketpair for communications
    *
    * @returns true if BidirMMapPipe uses a socketpair for communications
    */
   bool usesSocketpair() const { return m_inpipe == m_outpipe; }

   /** @brief if BidirMMapPipe uses a pipe pair for communications
    *
    * @returns true if BidirMMapPipe uses a pipe pair for communications
    */
   bool usesPipepair() const { return m_inpipe != m_outpipe; }

   /** @brief return flags (end of file, BidirMMapPipe closed, ...)
    *
    * @returns flags (end of file, BidirMMapPipe closed, ...)
    */
   int rdstate() const { return m_flags; }

   /** @brief true if end-of-file
    *
    * @returns true if end-of-file
    */
   bool eof() const { return m_flags & eofbit; }

   /** @brief logical failure (e.g. I/O on closed BidirMMapPipe)
    *
    * @returns true in case of grave logical error (I/O on closed pipe,...)
    */
   bool fail() const { return m_flags & failbit; }

   /** @brief true on I/O error
    *
    * @returns true on I/O error
    */
   bool bad() const { return m_flags & badbit; }

   /** @brief status of stream is good
    *
    * @returns true if pipe is good (no errors, eof, ...)
    */
   bool good() const { return !(m_flags & (eofbit | failbit | badbit)); }

   /** @brief true if closed
    *
    * @returns true if stream is closed
    */
   bool closed() const { return m_flags & failbit; }

   /** @brief return true if not serious error (fail/bad)
    *
    * @returns true if stream is does not have serious error (fail/bad)
    *
    * (if EOF, this is still true)
    */
   operator bool() const { return !fail() && !bad(); }

   /** @brief return true if serious error (fail/bad)
    *
    * @returns true if stream has a serious error (fail/bad)
    */
   bool operator!() const { return fail() || bad(); }

#ifdef STREAMOP
#undef STREAMOP
#endif
#define STREAMOP(TYPE) \
   BidirMMapPipe& operator<<(const TYPE& val) \
   { write(&val, sizeof(TYPE)); return *this; } \
   BidirMMapPipe& operator>>(TYPE& val) \
   { read(&val, sizeof(TYPE)); return *this; }
   STREAMOP(bool); ///< C++ style stream operators for bool
   STREAMOP(char); ///< C++ style stream operators for char
   STREAMOP(short); ///< C++ style stream operators for short
   STREAMOP(int); ///< C++ style stream operators for int
   STREAMOP(long); ///< C++ style stream operators for long
   STREAMOP(long long); ///< C++ style stream operators for long long 
   STREAMOP(unsigned char); ///< C++ style stream operators for unsigned char
   STREAMOP(unsigned short); ///< C++ style stream operators for unsigned short
   STREAMOP(unsigned int); ///< C++ style stream operators for unsigned int
   STREAMOP(unsigned long); ///< C++ style stream operators for unsigned long
   STREAMOP(unsigned long long); ///< C++ style stream operators for unsigned long long
   STREAMOP(float); ///< C++ style stream operators for float
   STREAMOP(double); ///< C++ style stream operators for double
#undef STREAMOP

   /** @brief write a C-style string
    *
    * @param str C-style string
    * @returns pipe written to
    */
   BidirMMapPipe& operator<<(const char* str);

   /** @brief read a C-style string
    *
    * @param str pointer to string (space allocated with malloc!)
    * @returns pipe read from
    *
    * since this is for C-style strings, we use malloc/realloc/free for
    * strings. passing in a NULL pointer is valid here, and the routine
    * will use realloc to allocate a chunk of memory of the right size.
    */
   BidirMMapPipe& operator>>(char* (&str));

   /** @brief write a std::string object
    *
    * @param str string to write
    * @returns pipe written to
    */
   BidirMMapPipe& operator<<(const std::string& str);

   /** @brief read a std::string object
    *
    * @param str string to be read
    * @returns pipe read from
    */
   BidirMMapPipe& operator>>(std::string& str);

   /** @brief write raw pointer to T to other side
    *
    * NOTE: This will not write the pointee! Only the value of the
    * pointer is transferred.
    *
    * @param tptr pointer to be written
    * @returns pipe written to
    */
   template<class T> BidirMMapPipe& operator<<(const T* tptr)
   { write(&tptr, sizeof(tptr)); return *this; }

   /** @brief read raw pointer to T from other side
    *
    * NOTE: This will not read the pointee! Only the value of the
    * pointer is transferred.
    *
    * @param tptr pointer to be read
    * @returns pipe read from
    */
   template<class T> BidirMMapPipe& operator>>(T* &tptr)
   { read(&tptr, sizeof(tptr)); return *this; }

   /** @brief I/O manipulator support
    *
    * @param f manipulator
    * @returns pipe with manipulator applied
    *
    * example:
    * @code
    * pipe << BidirMMapPipe::flush;
    * @endcode
    */
   BidirMMapPipe& operator<<(BidirMMapPipe& (*manip)(BidirMMapPipe&))
   { return manip(*this); }

   /** @brief I/O manipulator support
    *
    * @param f manipulator
    * @returns pipe with manipulator applied
    *
    * example:
    * @code
    * pipe >> BidirMMapPipe::purge;
    * @endcode
    */
   BidirMMapPipe& operator>>(BidirMMapPipe& (*manip)(BidirMMapPipe&))
   { return manip(*this); }

   /// for usage a la "pipe << flush;"
   static BidirMMapPipe& flush(BidirMMapPipe& pipe) { pipe.flush(); return pipe; }
   /// for usage a la "pipe << purge;"
   static BidirMMapPipe& purge(BidirMMapPipe& pipe) { pipe.purge(); return pipe; }

    private:
   /// copy-construction forbidden
   BidirMMapPipe(const BidirMMapPipe&);
   /// assignment forbidden
   BidirMMapPipe& operator=(const BidirMMapPipe&) { return *this; }

   /// page is our friend
   friend class BidirMMapPipe_impl::Page;
   /// convenience typedef for Page
   typedef BidirMMapPipe_impl::Page Page;
   
   /// tuning constants
   enum {
       // TotPages = 16 will give 32k buffers at 4k page size for both
       // parent and child; if your average message to send is larger
       // than this, consider raising the value (max 256)
       TotPages = 16, ///< pages shared (child + parent)

       PagesPerEnd = TotPages / 2, ///< pages per pipe end

       // if FlushThresh pages are filled, the code forces a flush; 3/4
       // of the pages available seems to work quite well
       FlushThresh = (3 * PagesPerEnd) / 4 ///< flush threshold
   };

   // per-class members
   static pthread_mutex_t s_openpipesmutex; ///< protects s_openpipes
   /// list of open BidirMMapPipes
   static std::list<BidirMMapPipe*> s_openpipes;
   /// pool of mmapped pages
   static BidirMMapPipe_impl::PagePool* s_pagepool;
   /// page pool reference counter
   static unsigned s_pagepoolrefcnt;

   /// return page pool
   static BidirMMapPipe_impl::PagePool& pagepool();

   // per-instance members
   BidirMMapPipe_impl::Pages m_pages; ///< mmapped pages
   Page* m_busylist; ///< linked list: busy pages (data to be read)
   Page* m_freelist; ///< linked list: free pages
   Page* m_dirtylist; ///< linked list: dirty pages (data to be sent)
   int m_inpipe; ///< pipe end from which data may be read
   int m_outpipe; ///< pipe end to which data may be written
   int m_flags; ///< flags (e.g. end of file)
   pid_t m_childPid; ///< pid of the child (zero if we're child)
   pid_t m_parentPid; ///< pid of the parent

   /// cleanup routine - at exit, we want our children to get a SIGTERM...
   static void teardownall(void);

   /// return length of a page list
   static unsigned lenPageList(const Page* list);

   /** "feed" the busy and free lists with a list of pages
    *
    * @param plist linked list of pages
    *
    * goes through plist, puts free pages from plist onto the freelist
    * (or sends them to the remote end if they belong there), and puts
    * non-empty pages on plist onto the busy list
    */
   void feedPageLists(Page* plist);

   /// put on dirty pages list
   void markPageDirty(Page* p);

   /// transfer bytes through the pipe (reading, writing, may block)
   static size_type xferraw(int fd, void* addr, size_type len,
      ssize_t (*xferfn)(int, void*, std::size_t));
   /// transfer bytes through the pipe (reading, writing, may block)
   static size_type xferraw(int fd, void* addr, const size_type len,
      ssize_t (*xferfn)(int, const void*, std::size_t))
   {
       return xferraw(fd, addr, len,
          reinterpret_cast<ssize_t (*)(
         int, void*, std::size_t)>(xferfn));
   }

   /** @brief send page(s) to the other end (may block)
    *
    * @param plist linked list of pages to send
    *
    * the implementation gathers the different write(s) whereever
    * possible; if mmap works, this results in a single write to transfer
    * the list of pages sent, if we need to copy things through the pipe,
    * we have one write to transfer which pages are sent, and then one
    * write per page.
    */
   void sendpages(Page* plist);

   /** @brief receive a pages from the other end (may block), queue them
    *
    * @returns number of pages received
    *
    * this is an application-level scatter read, which gets the list of
    * pages to read from the pipe. if mmap works, it needs only one read
    * call (to get the head of the list of pages transferred). if we need
    * to copy pages through the pipe, we need to add one read for each
    * empty page, and two reads for each non-empty page.
    */
   unsigned recvpages();

   /** @brief receive pages from other end (non-blocking)
    *
    * @returns number of pages received
    *
    * like recvpages(), but does not block if nothing is available for
    * reading
    */
   unsigned recvpages_nonblock();

   /// get a busy page to read data from (may block)
   Page* busypage();
   /// get a dirty page to write data to (may block)
   Page* dirtypage();

   /// close the pipe (no flush if forced)
   int doClose(bool force, bool holdlock = false);
   /// perform the flush
   void doFlush(bool forcePartialPages = true);
#endif //_WIN32
};

END_NAMESPACE_ROOFIT

#undef BEGIN_NAMESPACE_ROOFIT
#undef END_NAMESPACE_ROOFIT

#endif // BIDIRMMAPPIPE_H

// vim: ft=cpp:sw=4:tw=78