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
 170         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 171                 if( (r = (ring_t*) vr) == NULL ) {
 172                         return 0;
 173                 }
 174         } else {
 175                 r = (ring_t*) vr;
 176         }
 177
 178         if( r->tail == r->head ) {                                              // empty ring we can bail out quickly
 179                 return NULL;
 180         }
 181
 182         if( r->rgate != NULL ) {                                                // if lock exists we must honour it
 183                 pthread_mutex_lock( r->rgate );
 184                 if( r->tail == r->head ) {                                      // ensure ring didn't go empty while waiting
 185                         return NULL;
 186                 }
 187         }
 188
 189         ti = r->tail;
 190         r->tail++;
 191         if( r->tail >= r->nelements ) {
 192                 r->tail = 0;
 193         }
 194
 195         read( r->pfd, &ctr, sizeof( ctr )  );                           // when not in semaphore, this zeros the counter and value is meaningless
 196 /*
 197 future -- investigate if it's possible only to set/clear when empty or going to empty
 198         if( r->tail == r->head ) {                                                              // if this emptied the ring, turn off ready
 199         }
 200 */
 201
 202         if( r->rgate != NULL ) {                                                        // if locked above...
 203                 pthread_mutex_unlock( r->rgate );
 204         }
 205         return r->data[ti];
 206 }
 207
 208 /*
 209         Insert the pointer at the next open space in the ring.
 210         Returns 1 if the inert was ok, and 0 if the ring is full.
 211 */
 212 static inline int uta_ring_insert( void* vr, void* new_data ) {
 213         ring_t*         r;
 214         int64_t inc = 1;                                // used to set the counter in the pfd
 215
 216         if( !RING_FAST ) {                                                              // compiler should drop the conditional when always false
 217                 if( (r = (ring_t*) vr) == NULL ) {
 218                         return 0;
 219                 }
 220         } else {
 221                 r = (ring_t*) vr;
 222         }
 223
 224         if( r->wgate != NULL ) {                                                // if lock exists we must honour it
 225                 pthread_mutex_lock( r->wgate );
 226         }
 227
 228         if( r->head+1 == r->tail || (r->head+1 >= r->nelements && !r->tail) ) {         // ring is full
 229                 if( r->wgate != NULL ) {                                        // ensure released if needed
 230                         pthread_mutex_unlock( r->wgate );
 231                 }
 232                 return 0;
 233         }
 234
 235         write( r->pfd, &inc, sizeof( inc ) );
 236 /*
 237 future -- investigate if it's possible only to set/clear when empty or going to empty
 238         if( r->tail == r->head ) {                                                              // turn on ready if ring was empty
 239         }
 240 */
 241
 242         r->data[r->head] = new_data;
 243         r->head++;
 244         if( r->head >= r->nelements ) {
 245                 r->head = 0;
 246         }
 247
 248         if( r->wgate != NULL ) {                                                // if lock exists we must unlock before going
 249                 pthread_mutex_unlock( r->wgate );
 250         }
 251         return 1;
 252 }
 253
 254
 255
 256 #endif