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. The default is to NOT create a lock; if the user
  66         wants read locking then uta_config_ring() can be used to setup the
  67         mutex. (We use several rings internally and the assumption is that
  68         there is no locking for these.)
  69 */
  70 static void* uta_mk_ring( int size ) {
  71         ring_t* r;
  72         uint16_t max;
  73
  74         if( size <= 0 ) {
  75                 return NULL;
  76         }
  77         if( (r = (ring_t *) malloc( sizeof( *r ) )) == NULL ) {
  78                 return NULL;
  79         }
  80
  81         r->flags = 0;
  82         r->rgate = NULL;
  83         r->wgate = NULL;
  84         r->head = r->tail = 0;
  85
  86         max = (r->head - 1);
  87         if( size >= max ) {
  88                 size--;
  89         }
  90
  91         r->nelements = size;            // because we always have an empty element when full
  92         if( (r->data = (void **) malloc( sizeof( void* ) * (r->nelements + 1) )) == NULL ) {
  93                 free( r );
  94                 return NULL;
  95         }
  96
  97         memset( r->data, 0, sizeof( void* ) * r->nelements );
  98         r->pfd = eventfd( 0, EFD_SEMAPHORE | EFD_NONBLOCK );            // in semaphore mode counter is maintained with each insert/extract
  99         return (void *) r;
 100 }
 101
 102 /*
 103         Allows for configuration of a ring after it has been allocated.
 104         Options are RING_* options that allow for things like setting/clearing
 105         read locking. Returns 0 for failure 1 on success.
 106
 107         Options can be ORd together and all made effective at the same time, but
 108         it will be impossible to determine a specific failure if invoked this
 109         way.  Control is returned on the first error, and no provision is made
 110         to "undo" previously set options if an error occurs.
 111 */
 112 static int uta_ring_config( void* vr, int options ) {
 113         ring_t* r;
 114
 115         if( (r = (ring_t*) vr) == NULL ) {
 116                 errno = EINVAL;
 117                 return 0;
 118         }
 119
 120         if( options & RING_WLOCK  &&  r->wgate == NULL ) {              // don't realloc if we have one
 121                 r->wgate = (pthread_mutex_t *) malloc( sizeof( *r->wgate ) );
 122                 if( r->wgate == NULL ) {
 123                         return 0;
 124                 }
 125
 126                 pthread_mutex_init( r->wgate, NULL );
 127         }
 128
 129         if( options & (RING_RLOCK | RING_FRLOCK) ) {                            // read locking
 130                 if( r->rgate == NULL ) {                // don't realloc
 131                         r->rgate = (pthread_mutex_t *) malloc( sizeof( *r->rgate ) );
 132                         if( r->rgate == NULL ) {
 133                                 return 0;
 134                         }
 135
 136                         pthread_mutex_init( r->rgate, NULL );
 137                 }
 138                 if( options & RING_FRLOCK ) {
 139                         r->flags |= RING_FL_FLOCK;
 140                 }
 141         }
 142
 143         return 1;
 144 }
 145
 146 /*
 147         Ditch the ring. The caller is responsible for extracting any remaining
 148         pointers and freeing them as needed.
 149 */
 150 static void uta_ring_free( void* vr ) {
 151         ring_t* r;
 152
 153         if( (r = (ring_t*) vr) == NULL ) {
 154                 return;
 155         }
 156         if( r->data ){
 157                 free( r->data );
 158         }
 159         if( r->rgate ){
 160                 free( r->rgate );
 161         }
 162         if( r->wgate ){
 163                 free( r->wgate );
 164         }
 165         free( r );
 166 }
 167
 168
 169 /*
 170         Pull the next data pointer from the ring; null if there isn't
 171         anything to be pulled.
 172
 173         If the read lock exists for the ring, then this will BLOCK until
 174         it gets the lock.  There is always a chance that once the lock
 175         is obtained that the ring is empty, so the caller MUST handle
 176         a nil pointer as the return.
 177 */
 178 static inline void* uta_ring_extract( void* vr ) {
 179         ring_t*         r;
 180         uint16_t        ti;             // real index in data
 181         int64_t ctr;            // pfd counter
 182         void*           data;
 183
 184         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 185                 if( (r = (ring_t*) vr) == NULL ) {
 186                         return 0;
 187                 }
 188         } else {
 189                 r = (ring_t*) vr;
 190         }
 191
 192         if( r->tail == r->head ) {                                              // empty ring we can bail out quickly
 193                 return NULL;
 194         }
 195
 196         if( r->rgate != NULL ) {                                                        // if lock exists we must honour it
 197                 if( r->flags & RING_FL_FLOCK ) {                                // fast read locking try once and return nil if we cant lock
 198                         if( pthread_mutex_trylock( r->rgate ) != 0 ) {  // quick fail if not able to get a lock
 199                                 return NULL;
 200                         }
 201                 } else {
 202                         if( pthread_mutex_lock( r->rgate ) != 0 ) {
 203                                 return NULL;
 204                         }
 205                 }
 206                 if( r->tail == r->head ) {                                      // ensure ring didn't go empty while waiting
 207                         pthread_mutex_unlock( r->rgate );
 208                         return NULL;
 209                 }
 210         }
 211
 212         ti = r->tail;
 213         r->tail++;
 214         if( r->tail >= r->nelements ) {
 215                 r->tail = 0;
 216         }
 217
 218         read( r->pfd, &ctr, sizeof( ctr )  );                           // when not in semaphore, this zeros the counter and value is meaningless
 219 /*
 220 future -- investigate if it's possible only to set/clear when empty or going to empty
 221         if( r->tail == r->head ) {                                                              // if this emptied the ring, turn off ready
 222         }
 223 */
 224
 225         data = r->data[ti];                                             // secure data and clear before letting go of the lock
 226         r->data[ti] = NULL;
 227
 228         if( r->rgate != NULL ) {                                                        // if locked above...
 229                 pthread_mutex_unlock( r->rgate );
 230         }
 231
 232         return data;
 233 }
 234
 235
 236 /*
 237         Insert the pointer at the next open space in the ring.
 238         Returns 1 if the inert was ok, and 0 if there is an error;
 239         errno will be set to EXFULL if  the ring is full, if the attempt
 240         fails with anyt other error that indicates the inability to obtain
 241         a lock on the ring.
 242 */
 243 static inline int uta_ring_insert( void* vr, void* new_data ) {
 244         ring_t*         r;
 245         int64_t inc = 1;                                // used to set the counter in the pfd
 246
 247         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 248                 if( (r = (ring_t*) vr) == NULL ) {
 249                         return 0;
 250                 }
 251         } else {
 252                 r = (ring_t*) vr;
 253         }
 254
 255         if( r->wgate != NULL ) {                                                // if lock exists we must honour it
 256                 if( pthread_mutex_lock( r->wgate ) != 0 ) {
 257                         return 0;                                                               // leave mutex reason in place
 258                 }
 259         }
 260
 261         if( r->head+1 == r->tail || (r->head+1 >= r->nelements && !r->tail) ) {         // ring is full
 262                 if( r->wgate != NULL ) {                                                                                                // ensure released if needed
 263                         pthread_mutex_unlock( r->wgate );
 264                 }
 265                 errno = EXFULL;
 266                 return 0;
 267         }
 268
 269         r->data[r->head] = new_data;
 270         r->head++;
 271         if( r->head >= r->nelements ) {
 272                 r->head = 0;
 273         }
 274
 275         write( r->pfd, &inc, sizeof( inc ) );
 276 /*
 277 future -- investigate if it's possible only to set/clear when empty or going to empty
 278         if( r->tail == r->head ) {                                                              // turn on ready if ring was empty
 279         }
 280 */
 281
 282         if( r->wgate != NULL ) {                                                // if lock exists we must unlock before going
 283                 pthread_mutex_unlock( r->wgate );
 284         }
 285         return 1;
 286 }
 287
 288
 289
 290 #endif