<repo_name>cran/revoIPC<file_sep>/test/testrunner.R
library(RUnit)

res <- runTestFile('test.R')
printTextProtocol(res)

e <- getErrors(res)
if (e$nErr + e$nFail > 0) {
    quit('no', 1)
} else {
    quit('no', 0)
}
<file_sep>/test/test.R
library(RUnit)
library(revoIPC)
library(nws)

force.create.sem <- function(name, count) {
    sem <- tryCatch(ipcSemaphoreCreate(name, count), error=function(e) NULL)
    if (is.null(sem)) {
        sem <- ipcSemaphoreOpen(name)
        ipcSemaphoreDestroy(sem)
        sem <- ipcSemaphoreCreate(name, count)
    }
    sem
}

force.create.mutex <- function(name, rec) {
    mut <- tryCatch(ipcMutexCreate(name, rec), error=function(e) NULL)
    if (is.null(mut)) {
        mut <- ipcMutexOpen(name, rec)
        ipcMutexDestroy(mut)
        mut <- ipcMutexCreate(name, rec)
    }
    mut
}

force.create.mq <- function(name, maxmsg, maxsize) {
    mq <- tryCatch(ipcMsgQueueCreate(name, maxmsg, maxsize), error=function(e) NULL)
    if (is.null(mq)) {
        mq <- ipcMsgQueueOpen(name)
        ipcMsgQueueDestroy(mq)
        mq <- ipcMsgQueueCreate(name, maxmsg, maxsize)
    }
    mq
}

#####################################################
## Semaphores
#####################################################

test.sem.allnonblock <- function() {
    master <- force.create.sem('test_semaphore', 3)
    checkTrue(! is.null(master))
    worker <- ipcSemaphoreOpen('test_semaphore')
    checkTrue(ipcSemaphoreWait(worker))
    checkTrue(ipcSemaphoreWaitTry(worker))
    checkTrue(ipcSemaphoreWait(worker))
    checkTrue(! ipcSemaphoreWaitTry(worker))
    checkTrue(ipcSemaphorePost(master))
    checkTrue(ipcSemaphoreWaitTry(worker))
    checkTrue(! ipcSemaphoreWaitTry(worker))
    checkTrue(ipcSemaphorePost(master))
    checkTrue(ipcSemaphorePost(master))
    checkTrue(ipcSemaphoreWaitTry(worker))
    checkTrue(ipcSemaphoreWaitTry(worker))
    checkTrue(! ipcSemaphoreWaitTry(worker))
    checkTrue(! ipcSemaphoreWaitTry(master))
    checkTrue(ipcSemaphorePost(worker))
    checkTrue(ipcSemaphorePost(worker))
    checkTrue(ipcSemaphorePost(worker))
    checkTrue(ipcSemaphoreWaitTry(master))
    checkTrue(ipcSemaphoreWaitTry(master))
    checkTrue(ipcSemaphoreWaitTry(master))
    checkTrue(! ipcSemaphoreWaitTry(master))
    ipcSemaphoreRelease(worker)
    ipcSemaphoreDestroy(master)
}

test.sem.stale <- function() {
    sem <- force.create.sem('test_semaphore', 3)
    options(show.error.messages=FALSE)
    checkException(ipcSemaphoreCreate('test_semaphore', 3))
    checkException(ipcSemaphoreCreate('test_semaphore', 0))
    options(show.error.messages=TRUE)
    ipcSemaphoreDestroy(sem)
    options(show.error.messages=FALSE)
    ipcSemaphoreRelease(sem)           # Not currently an error, but shouldn't crash
    ipcSemaphoreDestroy(sem)           # Not currently an error, but shouldn't crash
    checkException(ipcSemaphorePost(sem))
    checkException(ipcSemaphoreWait(sem))
    checkException(ipcSemaphoreWaitTry(sem))
    checkException(ipcSemaphoreOpen('test_semaphore'))
    options(show.error.messages=TRUE)
}

helper.sem.allblock <- function() {
    library(revoIPC)
    # publishEvent <- function(ev) { nwsStore(SleighUserNws, 'events', ev)
    # publishEvent('begin')
    sem0 <- ipcSemaphoreOpen('test_semaphore0')
    sem1 <- ipcSemaphoreOpen('test_semaphore1')
    # publishEvent('semaphores')
    ipcSemaphoreWait(sem0)
    # publishEvent('gotsem0')
    ipcSemaphorePost(sem1)
    # publishEvent('postsem1')
    ipcSemaphoreRelease(sem0)
    ipcSemaphoreRelease(sem1)
    # publishEvent('end')
}

test.sem.allblock <- function() {
    sem0 <- force.create.sem('test_semaphore0', 0)
    sem1 <- force.create.sem('test_semaphore1', 0)

    # Use sleigh to start 3 workers running "helper"
    s <- sleigh()
    wsHandle <- userNws(s)
    sp <- eachWorker(s, helper.sem.allblock, eo=list(closure=FALSE, blocking=FALSE))
    # printEvents <- function() {
    #     while (TRUE) {
    #         val <- nwsFetchTry(wsHandle, 'events')
    #         if (is.null(val)) break
    #         else cat('Event: ', val, '\n')
    #     }
    # }

    # Initially, we cannot wait on sem1
    checkTrue(! ipcSemaphoreWaitTry(sem0))
    checkTrue(! ipcSemaphoreWaitTry(sem1))
    # printEvents()

    # Wake one worker and wait for response
    ipcSemaphorePost(sem0)
    ipcSemaphoreWait(sem1)
    checkTrue(! ipcSemaphoreWaitTry(sem0))
    checkTrue(! ipcSemaphoreWaitTry(sem1))
    # printEvents()

    # Wake second worker and wait for response
    ipcSemaphorePost(sem0)
    ipcSemaphoreWait(sem1)
    checkTrue(! ipcSemaphoreWaitTry(sem0))
    checkTrue(! ipcSemaphoreWaitTry(sem1))
    # printEvents()

    # Wake third worker and wait for response
    ipcSemaphorePost(sem0)
    ipcSemaphoreWait(sem1)
    checkTrue(! ipcSemaphoreWaitTry(sem0))
    checkTrue(! ipcSemaphoreWaitTry(sem1))
    # printEvents()

    # Sleigh workers should be done now.
    waitSleigh(sp)
    # printEvents()

    # Clean up
    close(s)
    ipcSemaphoreDestroy(sem0)
    ipcSemaphoreDestroy(sem1)
}

## It's not clear that this can be a useful test.  Posix semaphores can be
## "unlinked" while handles are still open, to be deleted when all references
## are closed.  Windows semaphores almost certainly cannot be deleted until all
## references are closed.
##
# test.sem.destroyed <- function() {
#     sem0 <- force.create.sem('test_semaphore', 3)
#     sem1 <- ipcSemaphoreOpen('test_semaphore')
#     ipcSemaphoreDestroy(sem0)
#     options(show.error.messages=FALSE)
#     ipcSemaphorePost(sem1)
#     ipcSemaphoreWait(sem1)
#     ipcSemaphoreWaitTry(sem1)
#     ipcSemaphoreDestroy(sem1)
#     options(show.error.messages=TRUE)
# }

#####################################################
## Mutexes
#####################################################

test.mutex.allnonblock <- function() {
    master <- force.create.mutex('test_mutex', FALSE)
    checkTrue(! is.null(master))
    worker <- ipcMutexOpen('test_mutex', FALSE)
    checkTrue(ipcMutexLock(worker))
    checkTrue(! ipcMutexLockTry(worker))
    checkTrue(! ipcMutexLockTry(master))
    checkTrue(ipcMutexUnlock(worker))
    checkTrue(ipcMutexLockTry(master))
    checkTrue(! ipcMutexLockTry(worker))
    checkTrue(! ipcMutexLockTry(master))
    checkTrue(ipcMutexUnlock(master))
    ipcMutexRelease(worker)
    ipcMutexDestroy(master)
}

test.rmutex.allnonblock <- function() {
    master <- force.create.mutex('test_rmutex', TRUE)
    checkTrue(! is.null(master))
    worker <- ipcMutexOpen('test_rmutex', TRUE)
    checkTrue(ipcMutexLock(worker))
    checkTrue(ipcMutexLockTry(worker))
    ## SIGH...  if pid(master) == pid(worker), 'master' and 'worker' are
    ## equivalent on Posix.
    # checkTrue(! ipcMutexLockTry(master))
    checkTrue(ipcMutexUnlock(worker))
    # checkTrue(! ipcMutexLockTry(master))
    checkTrue(ipcMutexUnlock(worker))
    checkTrue(ipcMutexLockTry(master))
    checkTrue(ipcMutexLock(master))
    checkTrue(ipcMutexLock(master))
    # checkTrue(! ipcMutexLockTry(worker))
    checkTrue(ipcMutexUnlock(master))
    # checkTrue(! ipcMutexLockTry(worker))
    checkTrue(ipcMutexUnlock(master))
    # checkTrue(! ipcMutexLockTry(worker))
    checkTrue(ipcMutexUnlock(master))
    checkTrue(ipcMutexLockTry(worker))
    checkTrue(ipcMutexUnlock(worker))
    ipcMutexRelease(worker)
    ipcMutexDestroy(master)
}

test.mutex.stale <- function() {
    master <- force.create.mutex('test_mutex', FALSE)
    checkTrue(! is.null(master))
    ipcMutexDestroy(master)
    options(show.error.messages=FALSE)
    checkException(ipcMutexLock(master))
    checkException(ipcMutexLockTry(master))
    checkException(ipcMutexUnlock(master))
    options(show.error.messages=TRUE)
    ipcMutexRelease(master)
    ipcMutexDestroy(master)
}

test.rmutex.stale <- function() {
    master <- force.create.mutex('test_rmutex', TRUE)
    checkTrue(! is.null(master))
    ipcMutexDestroy(master)
    options(show.error.messages=FALSE)
    checkException(ipcMutexLock(master))
    checkException(ipcMutexLockTry(master))
    checkException(ipcMutexUnlock(master))
    options(show.error.messages=TRUE)
    ipcMutexRelease(master)
    ipcMutexDestroy(master)
}

helper.mutex.allblock <- function() {
    library(revoIPC)
    # publishEvent <- function(ev) { nwsStore(SleighUserNws, 'events', ev)
    # publishEvent('begin')
    mtx <- ipcMutexOpen('test_mutex', FALSE)
    for (i in 1:250) {
        ipcMutexLock(mtx)
        value <- nwsFind(SleighUserNws, 'counter')
        nwsStore(SleighUserNws, 'counter', value + 1)
        ipcMutexUnlock(mtx)
    }
    ipcMutexRelease(mtx)
    # publishEvent('end')
}

test.mutex.allblock <- function() {
    mtx <- force.create.mutex('test_mutex', FALSE)

    # Use sleigh to start 3 workers running "helper"
    s <- sleigh()
    wsHandle <- userNws(s)
    nwsDeclare(wsHandle, 'counter', 'single')
    nwsStore(wsHandle, 'counter', 0)
    sp <- eachWorker(s, helper.mutex.allblock, eo=list(closure=FALSE, blocking=FALSE))
    # # printEvents <- function() {
    # #     while (TRUE) {
    # #         val <- nwsFetchTry(wsHandle, 'events')
    # #         if (is.null(val)) break
    # #         else cat('Event: ', val, '\n')
    # #     }
    # # }

    # Sleigh workers should be done now.
    waitSleigh(sp)
    checkEquals(nwsFetch(wsHandle, 'counter'), 750)
    # printEvents()

    # Clean up
    close(s)
    ipcMutexDestroy(mtx)
}

helper.rmutex.allblock <- function() {
    library(revoIPC)

    # publishEvent <- function(ev) { nwsStore(SleighUserNws, 'events', ev)
    # publishEvent('begin')
    mtx <- ipcMutexOpen('test_mutex', TRUE)
    for (i in 1:250) {
        ipcMutexLock(mtx)
        ipcMutexLock(mtx)   # The mutex so nice, they locked it twice!
        value <- nwsFind(SleighUserNws, 'counter')
        nwsStore(SleighUserNws, 'counter', value + 1)
        ipcMutexUnlock(mtx)
        ipcMutexUnlock(mtx)
    }
    ipcMutexRelease(mtx)
    # publishEvent('end')
}

test.rmutex.allblock <- function() {
    mtx <- force.create.mutex('test_mutex', TRUE)

    # Use sleigh to start 3 workers running "helper"
    s <- sleigh()
    wsHandle <- userNws(s)
    nwsDeclare(wsHandle, 'counter', 'single')
    nwsStore(wsHandle, 'counter', 0)
    sp <- eachWorker(s, helper.rmutex.allblock, eo=list(closure=FALSE, blocking=FALSE))
    # # printEvents <- function() {
    # #     while (TRUE) {
    # #         val <- nwsFetchTry(wsHandle, 'events')
    # #         if (is.null(val)) break
    # #         else cat('Event: ', val, '\n')
    # #     }
    # # }

    # Sleigh workers should be done now.
    waitSleigh(sp)
    value <- nwsFetch(wsHandle, 'counter')
    checkEquals(value, 750)
    # printEvents()

    # Clean up
    close(s)
    ipcMutexDestroy(mtx)
}

#####################################################
## Message Queues
#####################################################

test.mq.allnonblock <- function() {
    mq <- force.create.mq('test_mq', 16, 100)
    mqR <- ipcMsgQueueOpen('test_mq')
    checkEquals(16, ipcMsgQueueGetCapacity(mq))
    checkEquals(0, ipcMsgQueueGetCount(mq))
    checkTrue(ipcMsgQueueGetMaxSize(mq) >= 100)

    # Fill the queue
    for (i in 1:16) {
        checkTrue(ipcMsgSendTry(mq, i))
    }
    checkEquals(16, ipcMsgQueueGetCount(mq))

    # Next send should fail
    checkTrue(! ipcMsgSendTry(mq, 100))

    # First 16 receives should match first 16 sends
    for (i in 1:16) {
        checkEquals(i, ipcMsgReceiveTry(mqR))
    }
    checkEquals(0, ipcMsgQueueGetCount(mq))

    # Next receive should fail
    checkTrue(is.null(ipcMsgReceiveTry(mqR)))

    # Fill the queue again, blocking ops
    for (i in 17:32) {
        checkTrue(ipcMsgSend(mq, i))
    }
    checkEquals(16, ipcMsgQueueGetCount(mq))

    # Next send should fail
    checkTrue(! ipcMsgSendTry(mq, 100))

    # Next 16 receives should match second 16 sends
    for (i in 17:32) {
        checkEquals(i, ipcMsgReceive(mqR))
    }
    checkEquals(0, ipcMsgQueueGetCount(mq))

    # Next receive should fail
    checkTrue(is.null(ipcMsgReceiveTry(mqR)))

    ipcMsgQueueRelease(mqR)
    ipcMsgQueueDestroy(mq)
}

test.mq.stale <- function() {
    mq <- force.create.mq('test_mq', 16, 100)
    ipcMsgQueueDestroy(mq)
    options(show.error.messages=FALSE)
    checkException(ipcMsgSend(mq, 100))
    checkException(ipcMsgSendTry(mq, 100))
    checkException(ipcMsgReceive(mq, 100))
    checkException(ipcMsgReceiveTry(mq, 100))
    checkException(ipcMsgQueueGetCapacity(mq))
    checkException(ipcMsgQueueGetCount(mq))
    checkException(ipcMsgQueueGetMaxSize(mq))
    options(show.error.messages=TRUE)
}

helper.mq.allblock <- function() {
    library(revoIPC)
    # publishEvent <- function(ev) { nwsStore(SleighUserNws, 'events', ev) }
    # publishEvent('begin')
    mq1 <- ipcMsgQueueOpen('test_mq1')
    mq2 <- ipcMsgQueueOpen('test_mq2')
    while (TRUE) {
        # publishEvent('waiting')
        val <- ipcMsgReceive(mq1)
        # publishEvent(val)
        if (val == -1) break
        # publishEvent('replying')
        ipcMsgSend(mq2, val + 100)
    }
    ipcMsgQueueRelease(mq1)
    ipcMsgQueueRelease(mq2)
    # publishEvent('end')
}

test.mq.block <- function() {
    mq1 <- force.create.mq('test_mq1', 16, 100)
    mq2 <- force.create.mq('test_mq2', 16, 100)

    # Use sleigh to start 3 workers running "helper"
    s <- sleigh()
    wsHandle <- userNws(s)
    sp <- eachWorker(s, helper.mq.allblock, eo=list(closure=FALSE, blocking=FALSE))
    # printEvents <- function() {
    #     while (TRUE) {
    #         val <- nwsFetchTry(wsHandle, 'events')
    #         if (is.null(val)) { print('End of events.'); break }
    #         else cat('Event: ', val, '\n')
    #     }
    # }

    # print('Number 1')
    # printEvents()

    # Send the first batch
    for (i in 1:16) {
        ipcMsgSend(mq1, i)
    }

    # print('Number 2')
    # printEvents()

    # Drain them
    sum <- 0
    for (i in 1:16) {
        sum <- sum + ipcMsgReceive(mq2)
    }
    checkEquals(sum, 1736)

    # print('Number 3')
    # printEvents()

    # Kill a worker
    ipcMsgSend(mq1, -1)

    # print('Number 4')

    # Send the second batch
    for (i in 17:32) {
        ipcMsgSend(mq1, i)
    }

    # print('Number 5')

    # Drain them
    sum <- 0
    for (i in 1:16) {
        sum <- sum + ipcMsgReceive(mq2)
    }
    checkEquals(sum, 1992)

    # print('Number 6')

    # Kill a worker
    ipcMsgSend(mq1, -1)

    # print('Number 7')

    # Send the third batch
    for (i in 33:48) {
        ipcMsgSend(mq1, i)
    }

    # print('Number 8')

    # Drain them
    sum <- 0
    for (i in 1:16) {
        sum <- sum + ipcMsgReceive(mq2)
    }
    checkEquals(sum, 2248)

    # print('Number 9')

    # Kill the last worker
    ipcMsgSend(mq1, -1)

    # print('Number 10')

    # Sleigh workers should be done now.
    waitSleigh(sp)
    # printEvents()

    # print('Number 11')

    # Clean up
    close(s)
    ipcMsgQueueDestroy(mq1)
    ipcMsgQueueDestroy(mq2)
}
<file_sep>/src/queue.h
/*****************************************************************
 * Shared task queue structure.
 *
 * This queue structure makes a few assumptions.  First, there will be only a
 * single master (i.e. only a single agent which is enqueuing tasks and
 * processing returned results at any given time).  This minimizes the required
 * synchronization.  It isn't essential that the master always be the same
 * process, but enqueue and wait_results operations may not safely run
 * concurrently with other enqueue or wait_results operations.
 *
 * dequeue and return_results are concurrency-safe.
 *****************************************************************/

#ifndef INCLUDED_TASKQUEUE_QUEUE_H
#define INCLUDED_TASKQUEUE_QUEUE_H

#include "boost/interprocess/managed_mapped_file.hpp"
#include "boost/interprocess/managed_shared_memory.hpp"
#include "boost/interprocess/sync/named_semaphore.hpp"
#include "boost/interprocess/sync/named_mutex.hpp"
#include "boost/interprocess/allocators/allocator.hpp"
#include "boost/interprocess/containers/string.hpp"

#ifndef WIN32
#if ! defined(_POSIX_SEMAPHORES) || ((_XOPEN_VERSION >= 600) && (_POSIX_SEMAPHORES - 0 <= 0))
#define RIPC_USE_SYSV_SEMAPHORES
#endif
#endif

#ifdef RIPC_USE_SYSV_SEMAPHORES
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#endif

using boost::interprocess::read_write;
using boost::interprocess::mapped_region;
using boost::interprocess::file_mapping;
using boost::interprocess::managed_mapped_file;
using boost::interprocess::managed_shared_memory;
using boost::interprocess::offset_ptr;
using boost::interprocess::named_semaphore;
using boost::interprocess::named_mutex;

namespace queue {

    typedef boost::interprocess::allocator<char, managed_shared_memory::segment_manager> 
                      CharAllocator;
    typedef boost::interprocess::basic_string<char,
            std::char_traits<char>,
            CharAllocator> shared_string_t;


    /* Exception thrown to indicate a requested shutdown. */
    class ShutdownException : public std::exception
    {
    public:
        ShutdownException() : std::exception() { }
        ~ShutdownException() throw() { }

        char const *what() const throw() { return "shutdown requested"; }
    };

    class MappedFile
    {
    public:
        MappedFile(std::string const &fname)
            : _file(fname.c_str(), read_write),
              _region(_file, read_write)
        { }

        ~MappedFile() { }

        void *get_address() { return _region.get_address(); }
        std::size_t get_size() { return _region.get_size(); }

    private:
        file_mapping    _file;
        mapped_region   _region;
    };

    /* Buffer to hold a single task/result. */
    class Message
    {
    public:

        /* Construct a new task buffer. */
        Message()
            : _next(NULL),
              _data(NULL),
              _id(0),
              _length(0),
              _refs(0)
        {
        }

        /* Get the id of this task. */
        int get_id() const { return _id; }

        /* Set the id of this task. */
        void set_id(int id) { _id = id; }

        /* Get the "next" pointer for this message in the current queue. */
        Message *get_next() const { return _next.get(); }

        /* Set the "next" pointer for this message in the current queue. */
        void set_next(Message *m) { _next = m; }

        /* Set the data for this message.  This is a raw pointer to data
         * allocated in the shared segment, and a raw length, which should be
         * negated if the data refers to a filename rather than the direct
         * data.
         */
        void set_data(char *msg, int len)
        {
            _data = msg;
            _length = len;
        }

        /* Clear the data for this message.  Caller is responsible for freeing
         * the memory in the shared segment.
         */
        void clear_data()
        {
            _data = NULL;
            _length = 0;
        }

        /* Get the "raw" length of this message.  Positive values indicate that
         * the data is stored in shared memory, and give the actual data
         * length.  Negative values indicate that the shared memory contains a
         * filename, and the length is the negation of the filename length.
         */
        inline int get_length() const { return _length; }

        /* Set the raw length of this message.  Caveat callor: this doesn't
         * change the allocation, so...  be wise.
         */
        inline void set_length(int len) { _length = len; }

        /* Get the "raw" data of this message.  If the length is negative, the
         * raw data actually contains a filename containing the data.
         */
        inline char *get_data() const { return _data.get(); }

    private:

        /* Pointer to next task in the current queue. */
        offset_ptr<Message>         _next;

        /* Pointer to the raw data for this message. */
        offset_ptr<char>            _data;

        /* Id for this task. */
        int                         _id;

        /* Raw length of this message. */
        int                         _length;

        /* Number of completions (broadcast message) */
        int                         _refs;
    };

    class QueueHandle;

    /* Handle to a message.  Message will be stored in shared memory, but
     * MessageHandle will not. */
    class MessageHandle
    {
    public:
        /* Create a new message handle. */
        MessageHandle(Message *msg,
                      QueueHandle *queue,
                      MessageHandle *env)
            : _message(msg),
              _queue(queue),
              _file_temp(NULL),
              _environment(env)
        {
        }

        /* Destructor for message handles. */
        ~MessageHandle() { }

        /* Get the id of this message. */
        int get_id() const { return _message->get_id(); }

        /* Set the data for this message.
         */
        void set_data(char const *msg, int len);

        /* Clear the data for this message, freeing the underlying resources.
         */
        void clear_data();

        /* Get the length of this message. */
        int get_length();

        /* Get the data of this message. */
        char *get_data();

        /* Release the data of this message.  This MUST be called after calling
         * get_data. */
        void release_data();

        /* Get the underlying message object. */
        Message *get_message() const { return _message; }

        /* Get the prevailing environment, if any. */
        MessageHandle *get_environment() const { return _environment; }

    private:
        /* Pointer to the message data itself. */
        Message            *_message;

        /* Pointer to the queue handle for this message. */
        QueueHandle        *_queue;

        /* Temporary pointer to a managed mapped file. */
        MappedFile         *_file_temp;

        /* The new environment, if one is required. */
        MessageHandle      *_environment;
    };

    /* The task queue.  Consists of an "outbound" task queue and an "inbound"
     * result queue.  enqueue/dequeue may be used to access the task queue, and
     * return_result/wait_result may be used to access toe result queue. */
    class Queue
    {
    public:
        /* Create a task queue, cleaning up any leftover resources from an old queue. */
        static Queue *create(managed_shared_memory &segment,
                             std::string const &name,
                             std::string const &tmpdir,
                             long               mem_size,
                             int                max_tasks,
                             int                file_thresh);

        /* Connect to the task queue. */
        static Queue *open(managed_shared_memory &segment);

        /* Get the name of this queue. */
        std::string get_name() const
        {
            return std::string(_name.begin(), _name.end());
        }

        /* Get the location of the temp dir for this queue. */
        std::string get_temp_dir() const
        {
            return std::string(_tmpdir.begin(), _tmpdir.end());
        }

        /* Get the total space in bytes available in task queue. */
        long get_mem_size() const { return _mem_size; }

        /* Get the maximum number of tasks in queue. */
        int get_max_tasks() const { return _max_tasks; }

        /* Threshold for sending data via file. */
        int get_file_threshold() const { return _file_threshold; }

        /* Get the number of outstanding tasks. */
        int get_num_outstanding_tasks() const { return _num_outstanding; }

        /* Initiate a shutdown of the task queue, signalling shutdown to the
         * workers.  Note that it may take a little while for the workers to
         * shutdown, so it's a good idea to delay briefly before destroying the
         * task queue.
         */
        void shutdown() { _shutdown = true; }

        /* Check if a shutdown has been requested for this task queue.
         */
        bool shutdown_signalled() { return _shutdown; }

        /* Allocate a task.  Once a task is allocated, it MUST be returned to
         * the queue using either 'enqueue', 'return_result', or
         * 'discard_result', or it will be leaked.
         */
        Message *allocate_task();

        /* Enqueue a task, appending it to the queue of tasks.  Returns the
         * unique id for this particular task. */
        int enqueue(Message *m);

        /* Dequeue the first available task, blocking until one becomes
         * available. */
        Message *dequeue();

        /* Return a result to the task queue. */
        void return_result(Message *msg);

        /* Dequeue the first available result. */
        Message *get_result();

        /* Discard a result.  This must be called upon the value returned from
         * wait_result to clean up the results after they have been digested.
         */
        void discard_result(Message *msg);

        /* Refresh the "environment" task. */
        Message *new_environment();

        /* Get the "environment" task. */
        Message *get_environment() const { return _environment.get(); }

        /* Get the current environment generation. */
        int get_environment_generation() const { return _env_generation; }

        /* Construct a new queue. (DO NOT USE.) */
        Queue(managed_shared_memory &segment,
              std::string const &name,
              std::string const &tmpdir,
              long mem_size,
              int buf_size,
              int file_thresh);

        /* Destroy a queue. (DO NOT USE.) */
        ~Queue();

    private:

        /* Next task id to allocate */
        int                             _next_id;

        /* Number of tasks which have been submitted, but whose results have
         * not been collected. */
        int                             _num_outstanding;

        /* Number of tasks which have been submitted, but which have not been
         * claimed by workers. */
        int                             _num_tasks;

        /* Number of results which have been returned, but which have not been
         * retrieved. */
        int                             _num_results;

        /* Number of results which are in the "free" queue -- i.e. which may be
         * allocated to hold new tasks.  This is the maximum number of new
         * tasks which can be immediately enqueued at this moment. */
        int                             _num_free;

        /* Generation counter for environments. */
        int                             _env_generation;

        /* Pointer to array of all task structures. */
        offset_ptr<Message>             _buffer;

        /* Head and tail pointer for task queue. */
        offset_ptr<Message>             _task_head;
        offset_ptr<Message>             _task_tail;

        /* Head and tail pointer for result queue. */
        offset_ptr<Message>             _result_head;
        offset_ptr<Message>             _result_tail;

        /* Linked list of free task objects. */
        offset_ptr<Message>             _free;

        /* Current environment. */
        offset_ptr<Message>             _environment;

        /* Name of this queue. */
        shared_string_t                 _name;

        /* Location of temp dir for this queue. */
        shared_string_t                 _tmpdir;

        /* Total space available in task queue. */
        long                            _mem_size;

        /* Maximum number of tasks in queue. */
        int                             _max_tasks;

        /* Threshold for sending data via file. */
        int                             _file_threshold;

        /* Shutdown flag */
        bool                            _shutdown;
    };

#ifdef WIN32
    typedef HANDLE SemaphoreType;
    typedef HANDLE MutexType;
#elif defined(RIPC_USE_SYSV_SEMAPHORES)
    typedef struct {
        int sem_handle;
        int sem_idx;
    } SemaphoreType;
    typedef SemaphoreType MutexType;
#else
    typedef named_semaphore *SemaphoreType;
    typedef named_mutex *MutexType;
#endif

    /* Handle to a task queue. */
    class QueueHandle
    {
    public:
        QueueHandle(std::string const &name);

        QueueHandle(std::string const &name,
                    std::string const &tmpdir,
                    long mem_size,
                    int  max_tasks,
                    int  inline_thresh);

        ~QueueHandle()
        {
#ifdef WIN32
            if (_task_count)
                ::CloseHandle(_task_count);
            if (_result_count)
                ::CloseHandle(_result_count);
            if (_lock)
                ::CloseHandle(_lock);
#elif defined(RIPC_USE_SYSV_SEMAPHORES)
            /* No cleanup needed. */
#else
            delete _task_count;
            delete _result_count;
            delete _lock;
#endif
            delete _segment;
            delete _environment;
        }

        /* Destroy a task queue, cleaning up any resources. */
        void destroy();

        /* Get the name of this queue. */
        std::string get_name() const
        {
            check_destroyed();
            return _queue->get_name();
        }

        /* Get the location of the temp dir for this queue. */
        std::string get_temp_dir() const
        {
            check_destroyed();
            return _queue->get_temp_dir();
        }

        /* Get the total space in bytes available in task queue. */
        long get_mem_size() const
        {
            check_destroyed();
            return _queue->get_mem_size();
        }

        /* Get the maximum number of tasks in queue. */
        int get_max_tasks() const
        {
            check_destroyed();
            return _queue->get_max_tasks();
        }

        /* Threshold for sending data via file. */
        int get_file_threshold() const
        {
            check_destroyed();
            return _queue->get_file_threshold();
        }

        /* Get the number of outstanding tasks. */
        int get_num_outstanding_tasks() const
        {
            check_destroyed();
            return _queue->get_num_outstanding_tasks();
        }

        /* Get raw access to the shared memory segment.  This may be used to
         * pass more complex data structures as part of a task message. */
        managed_shared_memory &segment()
        {
            check_destroyed();
            return * _segment;
        }

        /* Initiate a shutdown of the task queue, signalling shutdown to the
         * workers.  Note that it may take a little while for the workers to
         * shutdown, so it's a good idea to delay briefly before destroying the
         * task queue.
         */
        void shutdown();

        /* Allocate a task.  Once a task is allocated, it MUST be returned to
         * the queue using either 'enqueue', 'return_result', or
         * 'discard_result', or it will be leaked.
         */
        MessageHandle *allocate_task()
        {
            check_destroyed();
            return new MessageHandle(_queue->allocate_task(),
                                     this,
                                     NULL);
        }

        /* Enqueue a task, appending it to the queue of tasks.  Returns the
         * unique id for this particular task. */
        int enqueue(MessageHandle *m);

        /* Enqueue a task, allocating it, and copying its data from 'msg',
         * returning the task's unique id. */
        int enqueue(char const *msg, int length);

        /* Dequeue the first available task, blocking until one becomes
         * available. */
        MessageHandle *dequeue();

        /* Return a result to the task queue. */
        void return_result(MessageHandle *msg);

        /* Dequeue the first available result, returning NULL immediately if
         * none is available. */
        MessageHandle *check_result();

        /* Dequeue the first available result, blocking until one becomes
         * available. */
        MessageHandle *wait_result();

        /* Discard a result.  This must be called upon the value returned from
         * wait_result to clean up the results after they have been digested.
         */
        void discard_result(MessageHandle *msg)
        {
            check_destroyed();
            msg->clear_data();
            _queue->discard_result(msg->get_message());
            delete msg;
        }

        /* Add a new "environment" task. */
        void set_environment(char const *data, int length);

        /* Get the environment. */
        MessageHandle *get_environment();

    private:
        void check_destroyed() const
        {
            if (! _queue)
                throw ShutdownException();
        }

    private:

        /* The queue. */
        Queue                          *_queue;

        /* Semaphore containing available task count. */
        SemaphoreType                   _task_count;

        /* Semaphore containing available result count. */
        SemaphoreType                   _result_count;

        /* Semaphore used as a lock for the queues. */
        MutexType                       _lock;

        /* Shared memory segment containing the task queue. */
        managed_shared_memory          *_segment;

        /* Current generation number for environment tasks. */
        int                             _env_generation;

        /* Message handle for environment. */
        MessageHandle                  *_environment;
    };
};

#endif
<file_sep>/src/queue.cc
#include "queue.h"
#include <fstream>
#include <exception>
#include <cassert>
#include <cstdlib>
#include <sys/types.h>
#include <sys/stat.h>
#ifdef WIN32
#include <winbase.h>
#include <io.h>
#include <fcntl.h>
#include <share.h>
#include <direct.h>
#else
#include <dirent.h>
#include <unistd.h>
#endif

using boost::interprocess::mapped_region;
using boost::interprocess::file_mapping;
using boost::interprocess::shared_memory_object;
using boost::interprocess::create_only;
using boost::interprocess::open_only;
using boost::interprocess::open_or_create;

#ifdef WIN32
#define RIPC_ACQUIRE_MUTEX(mtx)  ::WaitForSingleObject((mtx), INFINITE)
#define RIPC_RELEASE_MUTEX(mtx)  ::ReleaseMutex((mtx))

#define RIPC_WAIT_SEMAPHORE(sem) ::WaitForSingleObject((sem), INFINITE)
#define RIPC_POLL_SEMAPHORE(sem) (0 == ::WaitForSingleObject((sem), 0))
#define RIPC_POST_SEMAPHORE(sem) ::ReleaseSemaphore((sem), 1, NULL);
#elif defined(RIPC_USE_SYSV_SEMAPHORES)

#ifdef _SEM_SEMUN_UNDEFINED
union semun {
    int              val;    /* Value for SETVAL */
    struct semid_ds *buf;    /* Buffer for IPC_STAT, IPC_SET */
    unsigned short  *array;  /* Array for GETALL, SETALL */
};
#endif

static void sysv_sem_wait(queue::SemaphoreType *mtx)
{
    struct sembuf ctl_buf = { mtx->sem_idx, -1, SEM_UNDO };
    semop(mtx->sem_handle, & ctl_buf, 1);
}

static bool sysv_sem_poll(queue::SemaphoreType *mtx)
{
    struct sembuf ctl_buf = { mtx->sem_idx, -1, SEM_UNDO | IPC_NOWAIT };
    bool success = (semop(mtx->sem_handle, & ctl_buf, 1) != -1);
    return success;
}

static void sysv_sem_post(queue::SemaphoreType *mtx)
{
    struct sembuf ctl_buf = { mtx->sem_idx, 1, SEM_UNDO };
    semop(mtx->sem_handle, & ctl_buf, 1);
}

#define RIPC_ACQUIRE_MUTEX(mtx)  sysv_sem_wait(& (mtx))
#define RIPC_RELEASE_MUTEX(mtx)  sysv_sem_post(& (mtx))

#define RIPC_WAIT_SEMAPHORE(sem) sysv_sem_wait(& (sem))
#define RIPC_POLL_SEMAPHORE(sem) sysv_sem_poll(& (sem))
#define RIPC_POST_SEMAPHORE(sem) sysv_sem_post(& (sem))
#else
#define RIPC_ACQUIRE_MUTEX(mtx) (mtx)->lock()
#define RIPC_RELEASE_MUTEX(mtx) (mtx)->unlock()

#define RIPC_WAIT_SEMAPHORE(sem) (sem)->wait()
#define RIPC_POLL_SEMAPHORE(sem) (sem)->try_wait()
#define RIPC_POST_SEMAPHORE(sem) (sem)->post()
#endif

namespace {

    bool is_directory(std::string const &dir)
    {
#ifdef WIN32
        WIN32_FIND_DATA ffd;    // Directory traversal state
        HANDLE sh = FindFirstFile(dir.c_str(), &ffd);
        if (INVALID_HANDLE_VALUE == sh)
            return false;
        bool is_dir = (ffd.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) != 0;
        FindClose(sh);
        return is_dir;
#else
        struct stat sb;
        if (stat(dir.c_str(), &sb) == -1)
            return false;
        return S_ISDIR(sb.st_mode);
#endif
    }

    /* Destroy a directory and all of its subdirectories. */
    void nuke(std::string const &dir)
    {
#ifdef WIN32
        std::string searchPath(dir + "\\*");

        WIN32_FIND_DATA ffd; // file information struct
        HANDLE sh = FindFirstFile(searchPath.c_str(), &ffd);
        if (INVALID_HANDLE_VALUE == sh)
            return;

        do {
            if (strcmp(ffd.cFileName, ".") == 0  ||
                strcmp(ffd.cFileName, "..") == 0)
                continue;

            std::string fullPath(dir + '\\' + ffd.cFileName);
            if (ffd.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
                nuke(fullPath);
            else
                std::remove(fullPath.c_str());
        } while (FindNextFile(sh, &ffd));

        FindClose(sh);
#else
        DIR *d = opendir(dir.c_str());
        if (d == NULL)
            return;

        struct dirent *de = readdir(d);
        /* XXX: Use readdir_r if it is available. */
        do {
            std::string path(dir + '/' + de->d_name);
            if (is_directory(path))
            {
                if (de->d_name[0] == '.')
                {
                    if (de->d_name[1] == '\0'  ||
                        (de->d_name[1] == '.'  &&  de->d_name[2] == '\0'))
                        continue;
                }
                nuke(path);
            }
            else
                std::remove(path.c_str());
        }
        while ((de = readdir(d)) != NULL);
        closedir(d);
        std::remove(dir.c_str());
#endif
    }

    /* Need to add portable code for making a directory and any non-existent
     * parent directories. */
    void mkdirs(std::string const &dir)
    {
        if (is_directory(dir))
            return;

        /* XXX */
#ifdef WIN32
        _mkdir(dir.c_str());
#else
        mkdir(dir.c_str(), 0700);
#endif
    }

    /* Create a temporary file, optionally pre-populating it with specific
     * contents. */
    std::string make_temp_file(int id, std::string const &dir, char const *data, int len)
    {
        char buffer[100];
        sprintf(buffer, "%08d", id);
        std::string filepath(dir + '/' + buffer);

        std::filebuf fbuf;
        fbuf.open(filepath.c_str(),
                    std::ios_base::in    | std::ios_base::out 
                  | std::ios_base::trunc | std::ios_base::binary); 

        // Set the size
        fbuf.pubseekoff(len - 1, std::ios_base::beg);
        fbuf.sputc(0);
        fbuf.close();

        // If we have data, copy it in
        if (data != NULL)
        {
            file_mapping m_file(filepath.c_str(),  read_write);
            mapped_region region(m_file, read_write);
            memcpy(region.get_address(), data, len);
        }

        return filepath;
    }

    /* Get the length of a file in bytes. */
    int get_file_length(std::string const &filename)
    {
        FILE *f = fopen(filename.c_str(), "rb");
        if (f == NULL)
            return 0;

        fseek(f, 0, SEEK_END);
        int len = ftell(f);
        fclose(f);
        return len;
    }

    /* Segment details */
    char const SEGMENT_NAME[]               = "(Shm)";

    /* Semaphores */
    char const SEMAPHORE_NAME_NUMTASKS[]    = "(SemNumTasks)";
    char const SEMAPHORE_NAME_NUMREPLIES[]  = "(SemNumReplies)";
    char const MUTEX_NAME_LOCK[]            = "(MtxLock)";

    /* Buffer details */
    char const BUFFER_NAME[]                = "QueueBuf";
};

namespace queue {

    void MessageHandle::set_data(char const *msg, int len)
    {
        /* Free the old data.
           NOTE: This means we cannot copy the response data directly out of
                 the task data.  Insofar as this is used from R, this isn't a
                 problem, as we can't directly use the data in the task buffer
                 anyway, but must deserialize it.
        */
        clear_data();

        /* If the data is below the threshold, try to stash it in the shared
         * memory directly. */
        if (len <= _queue->get_file_threshold())
        {
            try {
                /* Allocate the new buffer and copy over the data. */
                char *new_d = static_cast<char *>(_queue->segment().allocate(len));
                memcpy(new_d, msg, len);

                /* Now put in the new data. */
                _message->set_data(new_d, len);
                return;
            }
            catch (std::bad_alloc b)
            {
                /* Recovery: we'll fail over to mmapped files. */
            }
        }

        /* Copy the data to a file and stash its filename in the shared memory.
         * XXX: This could be an epic fail if we don't even have enough room in
         *      the shared memory for the filename.  Hopefully it won't come to
         *      that... */
        std::string fname = make_temp_file(_message->get_id(),
                                           _queue->get_temp_dir(),
                                           msg, len);
        char *data = static_cast<char *>(_queue->segment().allocate(fname.size() + 1));
        memcpy(data, fname.c_str(), fname.size() + 1);
        _message->set_data(data, - (int) (1 + fname.size()));
    }

    void MessageHandle::clear_data(void)
    {
        if (_message->get_data())
        {
            /* _file_temp != NULL -> file stored in mmapped file. */
            if (_file_temp)
            {
                delete _file_temp;
                _file_temp = NULL;
                std::string path(_message->get_data());
                std::remove(path.c_str());
            }

            _queue->segment().deallocate(_message->get_data());
            _message->set_data(NULL, 0);
        }
    }

    int MessageHandle::get_length()
    {
        int const len = _message->get_length();

        /* Length >= 0 is the actual data length. */
        if (len >= 0)
            return len;

        /* Length < 0 means we get the length from a file. */
        return get_file_length(_message->get_data());
    }

    char *MessageHandle::get_data()
    {
        int const len = _message->get_length();

        /* Length >= 0 means the data is in shared memory. */
        if (len >= 0)
            return _message->get_data();

        /* Otherwise, let's map the file containing the data and return a
         * pointer to it. */
        if (_file_temp == NULL)
            _file_temp = new MappedFile(_message->get_data());
        return reinterpret_cast<char *>(_file_temp->get_address());
    }

    void MessageHandle::release_data()
    {
        if (_file_temp != NULL)
        {
            delete _file_temp;
            _file_temp = NULL;
        }
    }

    Queue *Queue::create(managed_shared_memory &segment,
                         std::string const &name,
                         std::string const &tmpdir,
                         long mem_size,
                         int max_tasks,
                         int file_thresh)
    {
        /* Construct the task queue in the shared memory segment. */
        return segment.construct<Queue>(BUFFER_NAME)(segment,
                                                     name,
                                                     tmpdir,
                                                     mem_size,
                                                     max_tasks,
                                                     file_thresh);
    }

    Queue *Queue::open(managed_shared_memory &segment)
    {
        /* Find the task queue in the shared memory segment. */
        std::pair<Queue *, std::size_t> res = segment.find<Queue>(BUFFER_NAME);
        Queue *q = res.first;
        return q;
    }

    Queue::Queue(managed_shared_memory &segment,
                 std::string const &name,
                 std::string const &tmpdir,
                 long mem_size,
                 int max_tasks,
                 int file_thresh)
        : _next_id(1000),
          _num_outstanding(0),
          _num_tasks(0),
          _num_results(0),
          _num_free(max_tasks),
          _env_generation(0),
          _buffer(),
          _task_head(),
          _task_tail(),
          _result_head(),
          _result_tail(),
          _free(),
          _environment(NULL),
          _name(name.c_str(), CharAllocator(segment.get_segment_manager())),
          _tmpdir(tmpdir.c_str(), CharAllocator(segment.get_segment_manager())),
          _mem_size(mem_size),
          _max_tasks(max_tasks),
          _file_threshold(file_thresh),
          _shutdown(false)
    {
        /* Allocate the task structures, and link them together into the free
         * queue. */
        _buffer = segment.construct<Message>(boost::interprocess::anonymous_instance)[max_tasks]();
        for (int i=0; i<max_tasks - 1; ++i)
            _buffer[i].set_next(&_buffer[i+1]);
        _free = &_buffer[0];
    }

    Queue::~Queue()
    {
    }

    Message *Queue::allocate_task(void)
    {
        /* If our free queue is empty, we cannot allocate another task. */
        if (! _free)
            return NULL;

        /* Remove empty task structure from the free list. */
        Message *msgbuf = _free.get();
        _free = msgbuf->get_next();
        msgbuf->set_next(NULL);
        -- _num_free;

        /* Allocate id */
        msgbuf->set_id(_next_id ++);
        return msgbuf;
    }

    int Queue::enqueue(Message *m)
    {
        int id = m->get_id();
        if (! _task_head)
            _task_head = m;
        else
            _task_tail->set_next(m);
        _task_tail = m;
        ++ _num_tasks;
        ++ _num_outstanding;
        return id;
    }

    Message *Queue::dequeue(void)
    {
        Message *msg = _task_head.get();
        -- _num_tasks;
        _task_head = msg->get_next();
        if (! _task_head)
            _task_tail = NULL;
        msg->set_next(NULL);
        return msg;
    }

    void Queue::return_result(Message *msg)
    {
        msg->set_next(NULL);
        if (! _result_head)
            _result_head = msg;
        else
            _result_tail->set_next(msg);
        _result_tail = msg;
        ++ _num_results;
    }

    Message *Queue::get_result()
    {
        Message *msg = _result_head.get();
        _result_head = msg->get_next();
        if (! _result_head)
            _result_tail = NULL;
        msg->set_next(NULL);
        -- _num_results;
        -- _num_outstanding;
        return msg;
    }

    void Queue::discard_result(Message *msg)
    {
        /* Free the contents of this task structure and add it to the free
         * list. */
        msg->clear_data();
        msg->set_next(_free.get());
        _free = msg;
        ++ _num_free;
    }

    Message *Queue::new_environment()
    {
        Message *m = _environment.get();
        if (m == NULL)
            _environment = m = allocate_task();
        ++ _env_generation;
        return m;
    }

    QueueHandle::QueueHandle(std::string const &name)
      : _queue(NULL),
#if defined(RIPC_USE_SYSV_SEMAPHORES)
        _task_count(),
        _result_count(),
        _lock(),
#else
        _task_count(NULL),
        _result_count(NULL),
        _lock(NULL),
#endif
        _segment(NULL),
        _env_generation(-1),
        _environment(NULL)
    {
        /* Find our shared memory. */
        _segment      = new managed_shared_memory(open_only, (name + SEGMENT_NAME).c_str());

        /* Find the task queue in the shared memory segment. */
        _queue        = Queue::open(*_segment);

        /* Find our semaphores/mutexes. */
#ifdef WIN32
        _task_count   = ::OpenSemaphoreA(SEMAPHORE_ALL_ACCESS, FALSE, (name + SEMAPHORE_NAME_NUMTASKS).c_str());
        _result_count = ::OpenSemaphoreA(SEMAPHORE_ALL_ACCESS, FALSE, (name + SEMAPHORE_NAME_NUMREPLIES).c_str());
        _lock         = ::OpenMutexA(SYNCHRONIZE, FALSE, (name + MUTEX_NAME_LOCK).c_str());
#elif defined(RIPC_USE_SYSV_SEMAPHORES)
        std::string sem_file(_queue->get_temp_dir() + "/" + name + "(semaphores)");
        key_t semkey = ftok(sem_file.c_str(), 'R');
        int handle = semget(semkey, 0, 0600);
        if (handle == -1)
        {
            /* XXX: Raise the alarm!  You fail at semaphores! */
        }
        _task_count.sem_handle   = handle;
        _task_count.sem_idx      = 0;
        _result_count.sem_handle = handle;
        _result_count.sem_idx    = 1;
        _lock.sem_handle         = handle;
        _lock.sem_idx            = 2;
#else
        _task_count   = new named_semaphore(open_only, (name + SEMAPHORE_NAME_NUMTASKS).c_str());
        _result_count = new named_semaphore(open_only, (name + SEMAPHORE_NAME_NUMREPLIES).c_str());
        _lock         = new named_mutex(open_only, (name + MUTEX_NAME_LOCK).c_str());
#endif
    }

    QueueHandle::QueueHandle(std::string const &name,
                             std::string const &tmpdir,
                             long mem_size,
                             int  max_tasks,
                             int  inline_thresh)
      : _queue(NULL),
#if defined(RIPC_USE_SYSV_SEMAPHORES)
        _task_count(),
        _result_count(),
        _lock(),
#else
        _task_count(NULL),
        _result_count(NULL),
        _lock(NULL),
#endif
        _segment(NULL),
        _env_generation(-1),
        _environment(NULL)
    {
        /* Create a directory for our mmapped files. */
        mkdirs(tmpdir.c_str());

        /* Create our shared memory and semaphore resources. */
#ifdef WIN32
        _task_count   = ::CreateSemaphoreA(NULL, 0, INT_MAX, (name + SEMAPHORE_NAME_NUMTASKS).c_str());
        _result_count = ::CreateSemaphoreA(NULL, 0, INT_MAX, (name + SEMAPHORE_NAME_NUMREPLIES).c_str());
        _lock         = ::CreateMutexA(NULL, FALSE, (name + MUTEX_NAME_LOCK).c_str());
#elif defined(RIPC_USE_SYSV_SEMAPHORES)
        std::string sem_file(tmpdir + "/" + name + "(semaphores)");
        ::close(::open(sem_file.c_str(), O_CREAT | O_TRUNC, 0600));
        key_t semkey = ftok(sem_file.c_str(), 'R');
        int handle = semget(semkey, 3, IPC_CREAT | IPC_EXCL | 0600);
        if (handle == -1)
        {
            /* XXX: Raise the alarm!  You fail at semaphores! */
        }
        static unsigned short values[] = { 0, 0, 1 };
        union semun arg;
        arg.array = values;
        semctl(handle, 0, SETALL, arg);
        _task_count.sem_handle   = handle;
        _task_count.sem_idx      = 0;
        _result_count.sem_handle = handle;
        _result_count.sem_idx    = 1;
        _lock.sem_handle         = handle;
        _lock.sem_idx            = 2;
#else
        _task_count   = new named_semaphore(create_only, (name + SEMAPHORE_NAME_NUMTASKS).c_str(),   0);
        _result_count = new named_semaphore(create_only, (name + SEMAPHORE_NAME_NUMREPLIES).c_str(), 0);
        _lock         = new named_mutex(create_only,     (name + MUTEX_NAME_LOCK).c_str());
#endif
        _segment      = new managed_shared_memory(create_only, (name + SEGMENT_NAME).c_str(), mem_size);

        /* Construct the task queue in the shared memory segment. */
        _queue        = Queue::create(*_segment,
                                      name,
                                      tmpdir,
                                      mem_size,
                                      max_tasks,
                                      inline_thresh);
    }

    void QueueHandle::destroy()
    {
        /* In case it hasn't been shutdown, do so now. */
        if (! _queue)
            return;
        shutdown();

        /* Release our shared memory. */
        std::string name(_queue->get_name());
        std::string tmpdir(_queue->get_temp_dir());
        delete _segment;
        _queue = NULL;
        _segment = NULL;
        _environment = NULL;

        /* Destroy the semaphores. */
#ifdef WIN32
        if (_task_count)
            ::CloseHandle(_task_count);
        if (_result_count)
            ::CloseHandle(_result_count);
        if (_lock)
            ::CloseHandle(_lock);
#elif defined(RIPC_USE_SYSV_SEMAPHORES)
        semctl(_task_count.sem_handle, 0, IPC_RMID, 0);
#else
        try { named_semaphore::remove((name + SEMAPHORE_NAME_NUMTASKS).c_str());   } catch (...) { }
        try { named_semaphore::remove((name + SEMAPHORE_NAME_NUMREPLIES).c_str()); } catch (...) { }
        try { named_mutex::remove((name + MUTEX_NAME_LOCK).c_str());               } catch (...) { }
#endif

        /* Destroy the shared memory segment. */
        try { shared_memory_object::remove((name + SEGMENT_NAME).c_str()); } catch (...) { }

        /* Destroy the mmapped files directory. */
        try { nuke(tmpdir); } catch (...) { }

        /* Be sanitary: NULL our pointers. */
#if defined(RIPC_USE_SYSV_SEMAPHORES)
        _task_count.sem_handle   = -1;
        _task_count.sem_idx      = -1;
        _result_count.sem_handle = -1;
        _result_count.sem_idx    = -1;
        _lock.sem_handle         = -1;
        _lock.sem_idx            = -1;
#else
        _task_count = _result_count = NULL;
        _lock = NULL;
#endif
    }

    void QueueHandle::shutdown()
    {
        check_destroyed();

        /* Acquire lock, set shutdown flag, release lock. */
        RIPC_ACQUIRE_MUTEX(_lock);
        _queue->shutdown();
        RIPC_RELEASE_MUTEX(_lock);

        /* Signal the task queue to wake any waiters. */
        RIPC_POST_SEMAPHORE(_task_count);
    }

    int QueueHandle::enqueue(MessageHandle *m)
    {
        check_destroyed();

        /* Lock, add to task queue, unlock. */
        RIPC_ACQUIRE_MUTEX(_lock);
        int id = _queue->enqueue(m->get_message());
        RIPC_RELEASE_MUTEX(_lock);

        /* Post task notification to the semaphore */
        RIPC_POST_SEMAPHORE(_task_count);
        return id;
    }

    int QueueHandle::enqueue(char const *msg, int length)
    {
        check_destroyed();

        /* Allocate an empty task. */
        MessageHandle *m = allocate_task();

        /* Populate task. */
        m->set_data(msg, length);

        /* Enqueue the task. */
        int id = enqueue(m);
        delete m;
        return id;
    }

    MessageHandle *QueueHandle::dequeue(void)
    {
        check_destroyed();

        /* Wait for a task to become available.  Claim it atomically.  Because
         * the semaphore has been signalled, and because we acquired the
         * semaphore successfully, we can be certain that either there will be
         * a task waiting for us immediately, or that a shutdown has been
         * signalled. */
        RIPC_WAIT_SEMAPHORE(_task_count);

        /* Lock, remove next task from queue, unlock. */
        RIPC_ACQUIRE_MUTEX(_lock);
        if (_queue->shutdown_signalled())
        {
            /* Post to the task count again to wake the next waiter, if there
             * are any. */
            RIPC_POST_SEMAPHORE(_task_count);
            RIPC_RELEASE_MUTEX(_lock);
            throw ShutdownException();
        }

        /* If there is a new prevailing environment, fetch it. */
        MessageHandle *env = NULL;
        int cur_generation = _queue->get_environment_generation();
        if (_env_generation != cur_generation)
        {
            env = get_environment();
            _env_generation = cur_generation;
        }

        /* Dequeue the message itself. */
        Message *msg = _queue->dequeue();
        RIPC_RELEASE_MUTEX(_lock);

        return new MessageHandle(msg, this, env);
    }

    void QueueHandle::return_result(MessageHandle *msg)
    {
        check_destroyed();

        /* Lock, append to results queue, unlock. */
        RIPC_ACQUIRE_MUTEX(_lock);
        if (_queue->shutdown_signalled())
        {
            RIPC_POST_SEMAPHORE(_task_count);
            RIPC_RELEASE_MUTEX(_lock);
            throw ShutdownException();
        }
        _queue->return_result(msg->get_message());
        RIPC_RELEASE_MUTEX(_lock);

        /* Post result notification to the semaphore. */
        RIPC_POST_SEMAPHORE(_result_count);
        delete msg;
    }

    MessageHandle *QueueHandle::check_result()
    {
        check_destroyed();

        /* If we have no outstanding tasks, there is no result to retrieve.
         * Fail fast.  No locking required because we assume that only the
         * master will update this counter. */
        if (_queue->get_num_outstanding_tasks() <= 0)
            return NULL;

        /* Block for result return. */
        if (! RIPC_POLL_SEMAPHORE(_result_count))
            return NULL;

        /* Lock, append to results queue, unlock. */
        RIPC_ACQUIRE_MUTEX(_lock);
        Message *msg = _queue->get_result();
        RIPC_RELEASE_MUTEX(_lock);

        return new MessageHandle(msg, this, NULL);
    }

    MessageHandle *QueueHandle::wait_result()
    {
        check_destroyed();

        /* If we have no outstanding tasks, there is no result to retrieve.
         * Fail fast.  No locking required because we assume that only the
         * master will update this counter. */
        if (_queue->get_num_outstanding_tasks() <= 0)
            return NULL;

        /* Block for result return. */
        RIPC_WAIT_SEMAPHORE(_result_count);

        /* Lock, append to results queue, unlock. */
        RIPC_ACQUIRE_MUTEX(_lock);
        Message *msg = _queue->get_result();
        RIPC_RELEASE_MUTEX(_lock);

        return new MessageHandle(msg, this, NULL);
    }

    void QueueHandle::set_environment(char const *data, int length)
    {
        check_destroyed();
        RIPC_ACQUIRE_MUTEX(_lock);
        Message *env = _queue->new_environment();
        if (_environment == NULL)
            _environment = new MessageHandle(env, this, NULL);
        _environment->set_data(data, length);
        RIPC_RELEASE_MUTEX(_lock);
    }

    MessageHandle *QueueHandle::get_environment()
    {
        if (_environment == NULL)
        {
            Message *env = _queue->get_environment();
            if (env != NULL)
                _environment = new MessageHandle(env, this, NULL);
        }
        return _environment;
    }
};
<file_sep>/test/runtests.sh
#!/bin/bash

if [ x"$REXECUTABLE" == x ]; then
    REXECUTABLE=R
fi

if [ x"$RARGUMENTS" == x ]; then
    RARGUMENTS="--vanilla --slave"
fi

if [ x"$RTESTLOG" == x ]; then
    RTESTLOG="ripctest.log"
fi

exec 1>"$RTESTLOG"

if "${REXECUTABLE}" "${RARGUMENTS}" < testrunner.R; then
    if [ -t 2 ]; then
        echo -e '\x1b[1;32mSuccess\x1b[0m' >&2
    else
        echo -e 'Success' >&2
    fi
    exit 0
else
    if [ -t 1 ]; then
        echo -e '\x1b[1;31mFailure\x1b[0m' >&2
    else
        echo -e 'Failure' >&2
    fi
    echo "See ${RTESTLOG} for details." >&2
    exit 1
fi
