root/opal/class/opal_ring_buffer.c

DEFINITIONS

1. opal_ring_buffer_construct
2. opal_ring_buffer_destruct
3. opal_ring_buffer_init
4. opal_ring_buffer_push
5. opal_ring_buffer_pop
6. opal_ring_buffer_poke

   1 /* -*- Mode: C; c-basic-offset:4 ; -*- */
   2 /*
   3  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
   4  *                         University Research and Technology
   5  *                         Corporation.  All rights reserved.
   6  * Copyright (c) 2004-2007 The University of Tennessee and The University
   7  *                         of Tennessee Research Foundation.  All rights
   8  *                         reserved.
   9  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
  10  *                         University of Stuttgart.  All rights reserved.
  11  * Copyright (c) 2004-2005 The Regents of the University of California.
  12  *                         All rights reserved.
  13  * Copyright (c) 2010      Cisco Systems, Inc. All rights reserved.
  14  * $COPYRIGHT$
  15  *
  16  * Additional copyrights may follow
  17  *
  18  * $HEADER$
  19  */
  20 
  21 #include "opal_config.h"
  22 
  23 #include <stdlib.h>
  24 #include <stdio.h>
  25 #include <assert.h>
  26 
  27 #include "opal/constants.h"
  28 #include "opal/class/opal_ring_buffer.h"
  29 #include "opal/util/output.h"
  30 
  31 static void opal_ring_buffer_construct(opal_ring_buffer_t *);
  32 static void opal_ring_buffer_destruct(opal_ring_buffer_t *);
  33 
  34 OBJ_CLASS_INSTANCE(opal_ring_buffer_t, opal_object_t,
  35                    opal_ring_buffer_construct,
  36                    opal_ring_buffer_destruct);
  37 
  38 /*
  39  * opal_ring_buffer constructor
  40  */
  41 static void opal_ring_buffer_construct(opal_ring_buffer_t *ring)
  42 {
  43     OBJ_CONSTRUCT(&ring->lock, opal_mutex_t);
  44     OBJ_CONSTRUCT(&ring->cond, opal_condition_t);
  45     ring->in_use = false;
  46     ring->head = 0;
  47     ring->tail = -1;
  48     ring->size = 0;
  49     ring->addr = NULL;
  50 }
  51 
  52 /*
  53  * opal_ring_buffer destructor
  54  */
  55 static void opal_ring_buffer_destruct(opal_ring_buffer_t *ring)
  56 {
  57     if( NULL != ring->addr) {
  58         free(ring->addr);
  59         ring->addr = NULL;
  60     }
  61 
  62     ring->size = 0;
  63 
  64     OBJ_DESTRUCT(&ring->lock);
  65     OBJ_DESTRUCT(&ring->cond);
  66 }
  67 
  68 /**
  69  * initialize a ring object
  70  */
  71 int opal_ring_buffer_init(opal_ring_buffer_t* ring, int size)
  72 {
  73     /* check for errors */
  74     if (NULL == ring) {
  75         return OPAL_ERR_BAD_PARAM;
  76     }
  77 
  78     /* Allocate and set the ring to NULL */
  79     ring->addr = (char **)calloc(size * sizeof(char*), 1);
  80     if (NULL == ring->addr) { /* out of memory */
  81         return OPAL_ERR_OUT_OF_RESOURCE;
  82     }
  83     ring->size = size;
  84 
  85     return OPAL_SUCCESS;
  86 }
  87 
  88 void* opal_ring_buffer_push(opal_ring_buffer_t *ring, void *ptr)
  89 {
  90     char *p=NULL;
  91 
  92     OPAL_ACQUIRE_THREAD(&(ring->lock), &(ring->cond), &(ring->in_use));
  93     if (NULL != ring->addr[ring->head]) {
  94         p = (char*)ring->addr[ring->head];
  95         if (ring->tail == ring->size - 1) {
  96             ring->tail = 0;
  97         } else {
  98             ring->tail = ring->head + 1;
  99         }
 100     }
 101     ring->addr[ring->head] = (char*)ptr;
 102     if (ring->tail < 0) {
 103         ring->tail = ring->head;
 104     }
 105     if (ring->head == ring->size - 1) {
 106         ring->head = 0;
 107     } else {
 108         ring->head++;
 109     }
 110     OPAL_RELEASE_THREAD(&(ring->lock), &(ring->cond), &(ring->in_use));
 111     return (void*)p;
 112 }
 113 
 114 void* opal_ring_buffer_pop(opal_ring_buffer_t *ring)
 115 {
 116     char *p=NULL;
 117 
 118     OPAL_ACQUIRE_THREAD(&(ring->lock), &(ring->cond), &(ring->in_use));
 119     if (-1 == ring->tail) {
 120         /* nothing has been put on the ring yet */
 121         p = NULL;
 122     } else {
 123         p = (char*)ring->addr[ring->tail];
 124         ring->addr[ring->tail] = NULL;
 125         if (ring->tail == ring->size-1) {
 126             ring->tail = 0;
 127         } else {
 128             ring->tail++;
 129         }
 130         /* see if the ring is empty */
 131         if (ring->tail == ring->head) {
 132             ring->tail = -1;
 133         }
 134     }
 135     OPAL_RELEASE_THREAD(&(ring->lock), &(ring->cond), &(ring->in_use));
 136     return (void*)p;
 137 }
 138 
 139  void* opal_ring_buffer_poke(opal_ring_buffer_t *ring, int i)
 140  {
 141     char *p=NULL;
 142     int offset;
 143 
 144     OPAL_ACQUIRE_THREAD(&(ring->lock), &(ring->cond), &(ring->in_use));
 145     if (ring->size <= i || -1 == ring->tail) {
 146         p = NULL;
 147     } else if (i < 0) {
 148         /* return the value at the head of the ring */
 149         if (ring->head == 0) {
 150             p = ring->addr[ring->size - 1];
 151         } else {
 152             p = ring->addr[ring->head - 1];
 153         }
 154     } else {
 155         /* calculate the offset of the tail in the ring */
 156         offset = ring->tail + i;
 157         /* correct for wrap-around */
 158         if (ring->size <= offset) {
 159             offset -= ring->size;
 160         }
 161         p = ring->addr[offset];
 162     }
 163     OPAL_RELEASE_THREAD(&(ring->lock), &(ring->cond), &(ring->in_use));
 164     return (void*)p;
 165 }