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