srcs/toolbox/rb.c

/* 
 * Copyright (c) 2005-2010 by KoanLogic s.r.l.
 */

#include <sys/types.h>
#include <sys/mman.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <u/toolbox/rb.h>
#include <u/toolbox/misc.h>
#include <u/toolbox/carpal.h>

#if defined(HAVE_SYSCONF) && defined(_SC_PAGE_SIZE)
  #define u_vm_page_sz  sysconf(_SC_PAGE_SIZE)
#elif defined (HAVE_SYSCONF) && defined(_SC_PAGESIZE)
  #define u_vm_page_sz  sysconf(_SC_PAGESIZE)
#elif defined(HAVE_GETPAGESIZE)
  #define u_vm_page_sz  getpagesize()
#else
  #error "don't know how to get page size.  reconfigure with --no_ringbuffer."
#endif

struct u_rb_s
{
    char *base;     /* base address of the mmap'd region */
    size_t sz;      /* ring buffer size */
    size_t wr_off;  /* write offset */
    size_t rd_off;  /* read offset */
    int opts;       /* options */
};

static char *write_addr (u_rb_t *rb);
static void write_incr (u_rb_t *rb, size_t cnt);
static char *read_addr (u_rb_t *rb);
static void read_incr (u_rb_t *rb, size_t cnt);
static size_t round_sz (size_t sz);

int u_rb_create (size_t hint_sz, int opts, u_rb_t **prb)
{
    int fd = -1;
    u_rb_t *rb = NULL;
    char path[] = "/tmp/rb-XXXXXX";

    dbg_err_sif ((rb = u_zalloc(sizeof(u_rb_t))) == NULL);
    dbg_err_sif ((fd = mkstemp(path)) == -1);
    dbg_err_sif (u_remove(path));
 
    /* round the supplied size to a page multiple (mmap is quite picky
     * about page boundary alignement) */
    rb->sz = round_sz(hint_sz);
    rb->wr_off = 0;
    rb->rd_off = 0;
    rb->opts = opts;
 
    dbg_err_sif (ftruncate(fd, rb->sz) == -1);

    /* mmap 2*rb->sz bytes. this is just a commodity map that will be 
     * discarded by the two following "half" maps.  we use it just to let the 
     * system choose a suitable base address (one that we are sure it's 
     * a multiple of the page size) that we can safely reuse later on when
     * pretending to MAP_FIXED */
    rb->base = mmap(NULL, rb->sz << 1, PROT_NONE, 
            MAP_ANON | MAP_PRIVATE, -1, 0);
    dbg_err_sif (rb->base == MAP_FAILED);

    /* POSIX: "The mapping established by mmap() shall replace any previous 
     * mappings for those whole pages containing any part of the address space 
     * of the process starting at pa and continuing for len bytes."
     * So, next two mappings replace in-toto the first 'rb->base' map which
     * does not need to be explicitly munmap'd */
 
    /* first half of the mmap'd region.  use MAP_SHARED to "twin" the two
     * mmap'd regions: each byte stored at a given offset in the first half
     * will show up at the same offset in the second half and viceversa */
    dbg_err_sif (mmap(rb->base, rb->sz, PROT_READ | PROT_WRITE, 
            MAP_FIXED | MAP_SHARED, fd, 0) != rb->base);
  
    /* second half is first's twin: they are attached to the same file 
     * descriptor 'fd', hence their pairing is handled at the OS level */
    dbg_err_sif (mmap(rb->base + rb->sz, rb->sz, PROT_READ | PROT_WRITE,
            MAP_FIXED | MAP_SHARED, fd, 0) != rb->base + rb->sz);
 
    /* dispose the file descriptor */
    dbg_err_sif (close(fd) == -1);

    *prb = rb;

    return 0;
err:
    u_rb_free(rb);
    U_CLOSE(fd);
    return -1;
}
 
void u_rb_free (u_rb_t *rb)
{
    nop_return_if (rb == NULL, );

    /* "All pages containing a part of the indicated range are unmapped",
     * hence the following single munmap with a double length should be ok for 
     * both previous (contiguous) mmap's */
    dbg_return_sif (rb->base && (munmap(rb->base, rb->sz << 1) == -1), );
    u_free(rb);

    return;
}

size_t u_rb_size (u_rb_t *rb)
{
    return rb->sz;
}

ssize_t u_rb_write (u_rb_t *rb, const void *b, size_t b_sz)
{
    size_t to_be_written;

    dbg_return_if (rb == NULL, -1);
    dbg_return_if (b == NULL, -1);
    dbg_return_if (b_sz > u_rb_size(rb), -1);

    nop_goto_if (!(to_be_written = U_MIN(u_rb_avail(rb), b_sz)), end);

    memcpy(write_addr(rb), b, to_be_written);
    write_incr(rb, to_be_written);

    /* fall through */
end:
    return to_be_written;
}

ssize_t u_rb_read (u_rb_t *rb, void *b, size_t b_sz)
{
    size_t to_be_read;

    dbg_return_if (b == NULL, -1);
    dbg_return_if (rb == NULL, -1);
    dbg_return_if (b_sz > u_rb_size(rb), -1);

    /* if there is nothing ready to be read go out immediately */
    nop_goto_if (!(to_be_read = U_MIN(u_rb_ready(rb), b_sz)), end);

    memcpy(b, read_addr(rb), to_be_read);
    read_incr(rb, to_be_read);

    /* fall through */
end:
    return to_be_read;
}

void *u_rb_fast_read (u_rb_t *rb, size_t *pb_sz)
{
    void *data = NULL;

    dbg_return_if (rb == NULL, NULL);
    dbg_return_if (!(rb->opts & U_RB_OPT_USE_CONTIGUOUS_MEM), NULL);
    dbg_return_if (pb_sz == NULL, NULL);
    dbg_return_if (*pb_sz > u_rb_size(rb), NULL);

    /* if there is nothing ready to be read go out immediately */
    nop_goto_if (!(*pb_sz = U_MIN(u_rb_ready(rb), *pb_sz)), end);

    data = read_addr(rb);
    read_incr(rb, *pb_sz);

    /* fall through */
end:
    return data;
}

int u_rb_clear (u_rb_t *rb)
{
    dbg_return_if (rb == NULL, -1);
    rb->wr_off = rb->rd_off = 0;
    return 0;
}
 
size_t u_rb_ready (u_rb_t *rb)
{
    return rb->wr_off - rb->rd_off;
}
 
size_t u_rb_avail (u_rb_t *rb)
{
    return rb->sz - u_rb_ready(rb);
}

/* round requested size to a multiple of PAGESIZE */
static size_t round_sz (size_t sz)
{
    size_t pg_sz = (size_t) u_vm_page_sz;

    return !sz ? pg_sz : (((sz - 1) / pg_sz) + 1) * pg_sz;
}

/* address for write op */
static char *write_addr (u_rb_t *rb)
{
    return rb->base + rb->wr_off;
}
 
/* shift the write pointer */
static void write_incr (u_rb_t *rb, size_t cnt)
{
    rb->wr_off += cnt;
}
 
/* address for next read op */
static char *read_addr (u_rb_t *rb)
{
    return rb->base + rb->rd_off;
}
 
/* shift the read pointer of 'cnt' positions */
static void read_incr (u_rb_t *rb, size_t cnt)
{
    /* 
     * assume 'rb' and 'cnt' sanitized 
     */
    rb->rd_off += cnt;

    /* When the read offset is advanced into the second virtual-memory region, 
     * both offsets - read and write - are decremented by the length of the 
     * underlying buffer */
    if (rb->rd_off >= rb->sz)
    {
        rb->rd_off -= rb->sz;
        rb->wr_off -= rb->sz;
    }

    return;
}