src/rmr/common/src/ring_static.c

   1 // :vi sw=4 ts=4 noet:
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2019 Nokia
   5         Copyright (c) 2018-2019 AT&T Intellectual Property.
   6
   7    Licensed under the Apache License, Version 2.0 (the "License");
   8    you may not use this file except in compliance with the License.
   9    You may obtain a copy of the License at
  10
  11            http://www.apache.org/licenses/LICENSE-2.0
  12
  13    Unless required by applicable law or agreed to in writing, software
  14    distributed under the License is distributed on an "AS IS" BASIS,
  15    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16    See the License for the specific language governing permissions and
  17    limitations under the License.
  18 ==================================================================================
  19 */
  20 /*
  21         Mnemonic:       ring_static.c
  22         Abstract:       Implements a ring of information (probably to act as a
  23                                 message queue).
  24         Author:         E. Scott Daniels
  25         Date:           31 August 2017
  26 */
  27
  28 #ifndef _ring_static_c
  29 #define _ring_static_c
  30
  31 #include <unistd.h>
  32 #include <stdio.h>
  33 #include <stdlib.h>
  34 #include <strings.h>
  35 #include <errno.h>
  36 #include <string.h>
  37 #include <stdint.h>
  38 #include <sys/eventfd.h>
  39
  40 #define RING_FAST 1                     // when set we skip nil pointer checks on the ring pointer
  41
  42 /*
  43         This returns the ring's pollable file descriptor. If one does not exist, then
  44         it is created.
  45 */
  46 static int uta_ring_getpfd( void* vr ) {
  47         ring_t*         r;
  48
  49         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
  50                 if( (r = (ring_t*) vr) == NULL ) {
  51                         return 0;
  52                 }
  53         } else {
  54                 r = (ring_t*) vr;
  55         }
  56
  57         if( r->pfd < 0 ) {
  58                 r->pfd = eventfd( 0, EFD_SEMAPHORE | EFD_NONBLOCK );
  59         }
  60
  61         return r->pfd;
  62 }
  63
  64 /*
  65         Make a new ring.
  66 */
  67 static void* uta_mk_ring( int size ) {
  68         ring_t* r;
  69         uint16_t max;
  70
  71         if( size <= 0 || (r = (ring_t *) malloc( sizeof( *r ) )) == NULL ) {
  72                 return NULL;
  73         }
  74
  75         r->head = r->tail = 0;
  76
  77         max = (r->head - 1);
  78         if( size >= max ) {
  79                 size--;
  80         }
  81
  82         r->nelements = size;            // because we always have an empty element when full
  83         if( (r->data = (void **) malloc( sizeof( void** ) * (r->nelements + 1) )) == NULL ) {
  84                 free( r );
  85                 return NULL;
  86         }
  87
  88         memset( r->data, 0, sizeof( void** ) * r->nelements );
  89         r->pfd = eventfd( 0, EFD_SEMAPHORE | EFD_NONBLOCK );            // in semaphore mode counter is maintained with each insert/extract
  90         return (void *) r;
  91 }
  92
  93 /*
  94         Ditch the ring. The caller is responsible for extracting any remaining
  95         pointers and freeing them as needed.
  96 */
  97 static void uta_ring_free( void* vr ) {
  98         ring_t* r;
  99
 100         if( (r = (ring_t*) vr) == NULL ) {
 101                 return;
 102         }
 103
 104         free( r );
 105 }
 106
 107
 108 /*
 109         Pull the next data pointer from the ring; null if there isn't
 110         anything to be pulled.
 111 */
 112 static inline void* uta_ring_extract( void* vr ) {
 113         ring_t*         r;
 114         uint16_t        ti;             // real index in data
 115         int64_t ctr;            // pfd counter
 116
 117         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 118                 if( (r = (ring_t*) vr) == NULL ) {
 119                         return 0;
 120                 }
 121         } else {
 122                 r = (ring_t*) vr;
 123         }
 124
 125         if( r->tail == r->head ) {                      // empty ring
 126                 return NULL;
 127         }
 128
 129         ti = r->tail;
 130         r->tail++;
 131         if( r->tail >= r->nelements ) {
 132                 r->tail = 0;
 133         }
 134
 135         read( r->pfd, &ctr, sizeof( ctr )  );                           // when not in semaphore, this zeros the counter and value is meaningless
 136 /*
 137 future -- investigate if it's possible only to set/clear when empty or going to empty
 138         if( r->tail == r->head ) {                                                              // if this emptied the ring, turn off ready
 139         }
 140 */
 141         return r->data[ti];
 142 }
 143
 144 /*
 145         Insert the pointer at the next open space in the ring.
 146         Returns 1 if the inert was ok, and 0 if the ring is full.
 147 */
 148 static inline int uta_ring_insert( void* vr, void* new_data ) {
 149         ring_t*         r;
 150         int64_t inc = 1;                                // used to set the counter in the pfd
 151
 152         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 153                 if( (r = (ring_t*) vr) == NULL ) {
 154                         return 0;
 155                 }
 156         } else {
 157                 r = (ring_t*) vr;
 158         }
 159
 160         if( r->head+1 == r->tail || (r->head+1 >= r->nelements && !r->tail) ) {         // ring is full
 161                 return 0;
 162         }
 163
 164         write( r->pfd, &inc, sizeof( inc ) );
 165 /*
 166 future -- investigate if it's possible only to set/clear when empty or going to empty
 167         if( r->tail == r->head ) {                                                              // turn on ready if ring was empty
 168         }
 169 */
 170
 171         r->data[r->head] = new_data;
 172         r->head++;
 173         if( r->head >= r->nelements ) {
 174                 r->head = 0;
 175         }
 176
 177         return 1;
 178 }
 179
 180
 181
 182 #endif