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         if( r->data ){
 152                 free( r->data );
 153         }
 154         if( r->rgate ){
 155                 free( r->rgate );
 156         }
 157         if( r->wgate ){
 158                 free( r->wgate );
 159         }
 160         free( r );
 161 }
 162
 163
 164 /*
 165         Pull the next data pointer from the ring; null if there isn't
 166         anything to be pulled.
 167
 168         If the read lock exists for the ring, then this will BLOCK until
 169         it gets the lock.  There is always a chance that once the lock
 170         is obtained that the ring is empty, so the caller MUST handle
 171         a nil pointer as the return.
 172 */
 173 static inline void* uta_ring_extract( void* vr ) {
 174         ring_t*         r;
 175         uint16_t        ti;             // real index in data
 176         int64_t ctr;            // pfd counter
 177         void*           data;
 178
 179         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 180                 if( (r = (ring_t*) vr) == NULL ) {
 181                         return 0;
 182                 }
 183         } else {
 184                 r = (ring_t*) vr;
 185         }
 186
 187         if( r->tail == r->head ) {                                              // empty ring we can bail out quickly
 188                 return NULL;
 189         }
 190
 191         if( r->rgate != NULL ) {                                                // if lock exists we must honour it
 192                 pthread_mutex_lock( r->rgate );
 193                 if( r->tail == r->head ) {                                      // ensure ring didn't go empty while waiting
 194                         pthread_mutex_unlock( r->rgate );
 195                         return NULL;
 196                 }
 197         }
 198
 199         ti = r->tail;
 200         r->tail++;
 201         if( r->tail >= r->nelements ) {
 202                 r->tail = 0;
 203         }
 204
 205         read( r->pfd, &ctr, sizeof( ctr )  );                           // when not in semaphore, this zeros the counter and value is meaningless
 206 /*
 207 future -- investigate if it's possible only to set/clear when empty or going to empty
 208         if( r->tail == r->head ) {                                                              // if this emptied the ring, turn off ready
 209         }
 210 */
 211
 212         data = r->data[ti];                                             // secure data and clear before letting go of the lock
 213         r->data[ti] = NULL;
 214
 215         if( r->rgate != NULL ) {                                                        // if locked above...
 216                 pthread_mutex_unlock( r->rgate );
 217         }
 218
 219         return data;
 220 }
 221
 222 /*
 223         Insert the pointer at the next open space in the ring.
 224         Returns 1 if the inert was ok, and 0 if the ring is full.
 225 */
 226 static inline int uta_ring_insert( void* vr, void* new_data ) {
 227         ring_t*         r;
 228         int64_t inc = 1;                                // used to set the counter in the pfd
 229
 230         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 231                 if( (r = (ring_t*) vr) == NULL ) {
 232                         return 0;
 233                 }
 234         } else {
 235                 r = (ring_t*) vr;
 236         }
 237
 238         if( r->wgate != NULL ) {                                                // if lock exists we must honour it
 239                 pthread_mutex_lock( r->wgate );
 240         }
 241
 242         if( r->head+1 == r->tail || (r->head+1 >= r->nelements && !r->tail) ) {         // ring is full
 243                 if( r->wgate != NULL ) {                                        // ensure released if needed
 244                         pthread_mutex_unlock( r->wgate );
 245                 }
 246                 return 0;
 247         }
 248
 249         r->data[r->head] = new_data;
 250         r->head++;
 251         if( r->head >= r->nelements ) {
 252                 r->head = 0;
 253         }
 254
 255         write( r->pfd, &inc, sizeof( inc ) );
 256 /*
 257 future -- investigate if it's possible only to set/clear when empty or going to empty
 258         if( r->tail == r->head ) {                                                              // turn on ready if ring was empty
 259         }
 260 */
 261
 262         if( r->wgate != NULL ) {                                                // if lock exists we must unlock before going
 263                 pthread_mutex_unlock( r->wgate );
 264         }
 265         return 1;
 266 }
 267
 268
 269
 270 #endif