diff options
| -rw-r--r-- | lf_ring_buffer2.c | 45 | ||||
| -rw-r--r-- | lf_ring_buffer2.h | 7 |
2 files changed, 33 insertions, 19 deletions
diff --git a/lf_ring_buffer2.c b/lf_ring_buffer2.c index d4ff4d8..4a1a3cd 100644 --- a/lf_ring_buffer2.c +++ b/lf_ring_buffer2.c @@ -38,11 +38,11 @@ #ifdef DEBUG_LFRB #include <stdio.h> #define _LOG_KO_( ... ) fprintf(stdout,__VA_ARGS__) - #define _LOG_CAS_( ... ) fprintf(stdout,__VA_ARGS__) + #define _LOG_CAS_( ... )fprintf(stdout,__VA_ARGS__) #elif defined DEBUG_LFRB_CAS #include <stdio.h> #define _LOG_KO_( ... ) - #define _LOG_CAS_( ... ) fprintf(stdout,__VA_ARGS__) + #define _LOG_CAS_( ... )fprintf(stdout,__VA_ARGS__) #elif defined DEBUG_LFRB_KO #include <stdio.h> #define _LOG_KO_( ... ) fprintf(stdout,__VA_ARGS__) @@ -52,10 +52,17 @@ #define _LOG_CAS_( ... ) #endif -#define BACKOFF_NANO_SLEEP 100000 +#define BACKOFF_NANO_SLEEP 10 #define USHORTMAX 0xffff +/* An unsigned int (indexes) is used to store both read_from and write_to indexes + * so that 32 bits compare and swap may be used to update both in one atomic instruction call. + * To be able to differenciate an empty buffer from a full buffer, + * read_from part of indexes is set to 0xffff (USHORTMAX) when the buffer is empty. + * Thus the maximum size of the buffer is 0xffff-1 = 65534 elements + */ + /* initialize an empty lf_ring_buffer struct */ lf_ring_buffer_t* lf_ring_buffer_create( size_t n_buf ) { if(n_buf>=USHORTMAX) { @@ -64,7 +71,7 @@ lf_ring_buffer_t* lf_ring_buffer_create( size_t n_buf ) { /* alloc ring_buffer struct */ lf_ring_buffer_t *r = malloc(sizeof(lf_ring_buffer_t)); if(r==NULL) return NULL; - /* */ + /* alloc buffer */ r->buffer = malloc(LFRB_BUFFER_SIZE*n_buf); if(r->buffer==NULL) { free(r); @@ -99,9 +106,9 @@ int lf_ring_buffer_write( lf_ring_buffer_t *r, void *data, int flags ) { read_from = current>>16; /* * check if the buffer is available, - * if it is but read_from==write, - * it means that the buffer is full and that a writer thread which at first reserved this buffer - * hasn't had enough CPU cycles to call MARK_AS_FILLED + * if read_from==write_to the buffer is full + * if it is available and full, it means that a writer thread which reserved this buffer + * hadn't had enough CPU cycles to call MARK_AS_FILLED */ if( LFRB_IS_AVAILABLE( r->buffer[write_to] ) && read_from!=write_to ) { next = write_to+1; @@ -116,9 +123,10 @@ int lf_ring_buffer_write( lf_ring_buffer_t *r, void *data, int flags ) { _LOG_CAS_( "write: CAS %u %u %u\n", r->indexes, current, next ); if( CompareAndSwapInt( &r->indexes, current, next ) ) break; } else { - _LOG_KO_("write: fail : %d %d %d\n",LFRB_IS_AVAILABLE(r->buffer[write_to]),write_to,read_from); + _LOG_KO_("write: impossible : wt:%d rf:%d wa:%d\n",write_to,read_from,LFRB_IS_AVAILABLE(r->buffer[write_to])); if(IS_NOT_BLOCKING(flags)) return -1; } + /* sleep */ backoff.tv_sec = 0; backoff.tv_nsec = backoff_time; nanosleep(&backoff,NULL); @@ -137,31 +145,36 @@ int lf_ring_buffer_read( lf_ring_buffer_t *r, void *data, int flags ) { struct timespec backoff; int backoff_time = BACKOFF_NANO_SLEEP; for(;;) { + /* copy indexes and split it */ current = r->indexes; write_to = current&0xffff; read_from = current>>16; - if( !(LFRB_IS_AVAILABLE( r->buffer[read_from] )) && read_from!=USHORTMAX ) { + /* + * the buffer may be non empty but available if a writer thread hadn't had enough CPU cycles to mark the buffer as filled + */ + if( read_from==USHORTMAX || LFRB_IS_AVAILABLE( r->buffer[read_from] ) ) { + _LOG_KO_("read : impossible : wt:%d rf:%d ra:%d\n",write_to,read_from,LFRB_IS_AVAILABLE(r->buffer[read_from])); + if(IS_NOT_BLOCKING(flags)) return -1; + } else { tmp = read_from +1; if (tmp==r->n_buf) tmp=0; - /* is the buffer empty */ - if ( tmp==write_to) { + /* set the buffer empty if needed */ + if (tmp==write_to) { tmp = USHORTMAX; } next = tmp<<16 | write_to; + /* try to update indexes */ _LOG_CAS_( "read: CAS %u %u %u\n", r->indexes, current, next ); if( CompareAndSwapInt( &r->indexes, current , next ) ) break; - } else { - _LOG_KO_("read: ring empty\n"); - if(IS_NOT_BLOCKING(flags)) return -1; } + /* sleep */ backoff.tv_sec = 0; backoff.tv_nsec = backoff_time; nanosleep(&backoff,NULL); backoff_time += BACKOFF_NANO_SLEEP; } - /* will do bad things if data dst buffer is too small !! */ + /* copy this buffer and mark it as read */ memcpy( data, LFRB_DATA_PTR(r->buffer[read_from]), LFRB_DATA_SIZE ); - /* finish the read process */ LFRB_MARK_AS_READ( r->buffer[read_from] ); return 0; } diff --git a/lf_ring_buffer2.h b/lf_ring_buffer2.h index 6f0cee1..acf9974 100644 --- a/lf_ring_buffer2.h +++ b/lf_ring_buffer2.h @@ -38,13 +38,14 @@ extern "C" { #define LFRB_NO_BLOCK 1 /* if buffer is full, leave instead of try again and again */ #define IS_NOT_BLOCKING( flags ) ( (flags)&LFRB_NO_BLOCK ) +/* the ring buffer structure */ typedef struct ring_buffer { LFRB_BUFFER_TYPE *buffer; /* buffer data */ - size_t n_buf; /* number of buffers */ - unsigned int indexes; /* indexes where to read from and write data to */ + size_t n_buf; /* number of buffers, max 65534, see implementation for details */ + unsigned int indexes; /* indexes where to read_from and write_to */ } lf_ring_buffer_t; -/* return an initialized lf_ring_buffer_t struct */ +/* return an initialized lf_ring_buffer_t struct, size is limited to 65534, see implementation for details */ lf_ring_buffer_t* lf_ring_buffer_create( size_t n_buf ); /* destroy an lf_ring_buffer_t struct */ |