Subversion Repositories f9daq

/********************************************************************\

  Name:         rb.c

  Created by:   Stefan Ritt

  $Id: rb.cpp 21437 2014-07-30 14:13:29Z ritt $

\********************************************************************/

#include <stdio.h>

#ifdef OS_DARWIN

#include <sys/malloc.h>

#else

#include <malloc.h>

#endif

#include <string.h>

#include <assert.h>

#include "rb.h"

/********************************************************************\

*                                                                    *

*                 Ring buffer functions                              *

*                                                                    *

* Provide an inter-thread buffer scheme for handling front-end       *

* events. This code allows concurrent data acquisition, calibration  *

* and network transfer on a multi-CPU machine. One thread reads      *

* out the data, passes it vis the ring buffer functions              *

* to another thread running on the other CPU, which can then         *

* calibrate and/or send the data over the network.                   *

*                                                                    *

\********************************************************************/

typedef struct {

   unsigned char *buffer;

   unsigned int size;

   unsigned int max_event_size;

   unsigned char *rp;

   unsigned char *wp;

   unsigned char *ep;

} RING_BUFFER;

#define MAX_RING_BUFFER 100

RING_BUFFER rb[MAX_RING_BUFFER];

volatile int _rb_nonblocking = 0;

extern void ss_sleep(int ms);

int rb_set_nonblocking()

/********************************************************************\

  Routine: rb_set_nonblocking

  Purpose: Set all rb_get_xx to nonblocking. Needed in multi-thread

           environments for stopping all theads without deadlock

  Input:

    NONE

  Output:

    NONE

  Function value:

    RB_SUCCESS       Successful completion

\********************************************************************/

{

   _rb_nonblocking = 1;

   return RB_SUCCESS;

}

int rb_create(int size, int max_event_size, int *handle)

/********************************************************************\

  Routine: rb_create

  Purpose: Create a ring buffer with a given size

  Input:

    int size             Size of ring buffer, must be larger than

                         2*max_event_size

    int max_event_size   Maximum event size to be placed into

                         ring buffer

  Output:

    int *handle          Handle to ring buffer

  Function value:

    DB_SUCCESS           Successful completion

    DB_NO_MEMORY         Maximum number of ring buffers exceeded

    DB_INVALID_PARAM     Invalid event size specified

\********************************************************************/

{

   int i;

   for (i = 0; i < MAX_RING_BUFFER; i++)

      if (rb[i].buffer == NULL)

         break;

   if (i == MAX_RING_BUFFER)

      return RB_NO_MEMORY;

   if (size < max_event_size * 2)

      return RB_INVALID_PARAM;

   memset(&rb[i], 0, sizeof(RING_BUFFER));

   rb[i].buffer = (unsigned char *) malloc(size);

   assert(rb[i].buffer);

   rb[i].size = size;

   rb[i].max_event_size = max_event_size;

   rb[i].rp = rb[i].buffer;

   rb[i].wp = rb[i].buffer;

   rb[i].ep = rb[i].buffer;

   *handle = i + 1;

   return RB_SUCCESS;

}

int rb_delete(int handle)

/********************************************************************\

  Routine: rb_delete

  Purpose: Delete a ring buffer

  Input:

    none

  Output:

    int handle       Handle to ring buffer

  Function value:

    DB_SUCCESS       Successful completion

\********************************************************************/

{

   if (handle < 0 || handle >= MAX_RING_BUFFER || rb[handle - 1].buffer == NULL)

      return RB_INVALID_HANDLE;

   free(rb[handle - 1].buffer);

   memset(&rb[handle - 1], 0, sizeof(RING_BUFFER));

   return RB_SUCCESS;

}

int rb_get_wp(int handle, void **p, int millisec)

/********************************************************************\

Routine: rb_get_wp

  Purpose: Retrieve write pointer where new data can be written

  Input:

     int handle               Ring buffer handle

     int millisec             Optional timeout in milliseconds if

                              buffer is full. Zero to not wait at

                              all (non-blocking)

  Output:

    char **p                  Write pointer

  Function value:

    DB_SUCCESS       Successful completion

\********************************************************************/

{

   int h, i;

   unsigned char *rp;

   if (handle < 1 || handle > MAX_RING_BUFFER || rb[handle - 1].buffer == NULL)

      return RB_INVALID_HANDLE;

   h = handle - 1;

   for (i = 0; i <= millisec / 10; i++) {

      rp = rb[h].rp;            // keep local copy, rb[h].rp might be changed by other thread

      /* check if enough size for wp >= rp without wrap-around */

      if (rb[h].wp >= rp

          && rb[h].wp + rb[h].max_event_size <= rb[h].buffer + rb[h].size - rb[h].max_event_size) {

         *p = rb[h].wp;

         return RB_SUCCESS;

      }

      /* check if enough size for wp >= rp with wrap-around */

      if (rb[h].wp >= rp && rb[h].wp + rb[h].max_event_size > rb[h].buffer + rb[h].size - rb[h].max_event_size && rb[h].rp > rb[h].buffer) {    // next increment of wp wraps around, so need space at beginning

         *p = rb[h].wp;

         return RB_SUCCESS;

      }

      /* check if enough size for wp < rp */

      if (rb[h].wp < rp && rb[h].wp + rb[h].max_event_size < rp) {

         *p = rb[h].wp;

         return RB_SUCCESS;

      }

      if (millisec == 0)

         return RB_TIMEOUT;

      if (_rb_nonblocking)

         return RB_TIMEOUT;

      /* wait one time slice */

      ss_sleep(10);

   }

   return RB_TIMEOUT;

}

int rb_increment_wp(int handle, int size)

/********************************************************************\

  Routine: rb_increment_wp

  Purpose: Increment current write pointer, making the data at

           the write pointer available to the receiving thread

  Input:

     int handle               Ring buffer handle

     int size                 Number of bytes placed at the WP

  Output:

    NONE

  Function value:

    RB_SUCCESS                Successful completion

    RB_INVALID_PARAM          Event size too large or invalid handle

\********************************************************************/

{

   int h;

   unsigned char *new_wp;

   if (handle < 1 || handle > MAX_RING_BUFFER || rb[handle - 1].buffer == NULL)

      return RB_INVALID_HANDLE;

   h = handle - 1;

   if ((unsigned int) size > rb[h].max_event_size)

      return RB_INVALID_PARAM;

   new_wp = rb[h].wp + size;

   /* wrap around wp if not enough space */

   if (new_wp > rb[h].buffer + rb[h].size - rb[h].max_event_size) {

      rb[h].ep = new_wp;

      new_wp = rb[h].buffer;

      assert(rb[h].rp != rb[h].buffer);

   }

   rb[h].wp = new_wp;

   return RB_SUCCESS;

}

int rb_get_rp(int handle, void **p, int millisec)

/********************************************************************\

  Routine: rb_get_rp

  Purpose: Obtain the current read pointer at which new data is

           available with optional timeout

  Input:

    int handle               Ring buffer handle

    int millisec             Optional timeout in milliseconds if

                             buffer is full. Zero to not wait at

                             all (non-blocking)

  Output:

    char **p                 Address of pointer pointing to newly

                             available data. If p == NULL, only

                             return status.

  Function value:

    RB_SUCCESS       Successful completion

\********************************************************************/

{

   int i, h;

   if (handle < 1 || handle > MAX_RING_BUFFER || rb[handle - 1].buffer == NULL)

      return RB_INVALID_HANDLE;

   h = handle - 1;

   for (i = 0; i <= millisec / 10; i++) {

      if (rb[h].wp != rb[h].rp) {

         if (p != NULL)

            *p = rb[handle - 1].rp;

         return RB_SUCCESS;

      }

      if (millisec == 0)

         return RB_TIMEOUT;

      if (_rb_nonblocking)

         return RB_TIMEOUT;

      /* wait one time slice */

      ss_sleep(10);

   }

   return RB_TIMEOUT;

}

int rb_increment_rp(int handle, int size)

/********************************************************************\

  Routine: rb_increment_rp

  Purpose: Increment current read pointer, freeing up space for

           the writing thread.

  Input:

     int handle               Ring buffer handle

     int size                 Number of bytes to free up at current

                              read pointer

  Output:

    NONE

  Function value:

    RB_SUCCESS                Successful completion

    RB_INVALID_PARAM          Event size too large or invalid handle

\********************************************************************/

{

   int h;

   unsigned char *new_rp;

   if (handle < 1 || handle > MAX_RING_BUFFER || rb[handle - 1].buffer == NULL)

      return RB_INVALID_HANDLE;

   h = handle - 1;

   if ((unsigned int) size > rb[h].max_event_size)

      return RB_INVALID_PARAM;

   new_rp = rb[h].rp + size;

   /* wrap around if not enough space left */

   if (new_rp + rb[h].max_event_size > rb[h].buffer + rb[h].size)

      new_rp = rb[h].buffer;

   rb[handle - 1].rp = new_rp;

   return RB_SUCCESS;

}

int rb_get_buffer_level(int handle, int *n_bytes)

/********************************************************************\

  Routine: rb_get_buffer_level

  Purpose: Return number of bytes in a ring buffer

  Input:

    int handle              Handle of the buffer to get the info

  Output:

    int *n_bytes            Number of bytes in buffer

  Function value:

    RB_SUCCESS              Successful completion

    RB_INVALID_HANDLE       Buffer handle is invalid

\********************************************************************/

{

   int h;

   if (handle < 1 || handle > MAX_RING_BUFFER || rb[handle - 1].buffer == NULL)

      return RB_INVALID_HANDLE;

   h = handle - 1;

   if (rb[h].wp >= rb[h].rp)

      *n_bytes = rb[h].wp - rb[h].rp;

   else

      *n_bytes = rb[h].ep - rb[h].rp + rb[h].wp - rb[h].buffer;

   return RB_SUCCESS;

}