second ring buffer implementation

5 files changed, 432 insertions, 1 deletions

diff --git a/.gitignore b/.gitignore
index 68f8a12..2bdb0e4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,4 +4,5 @@ cas
 container_of
 lf_fifo_test
 lf_ring_buffer_test
+lf_ring_buffer_test2
 lock_free_queue_test
diff --git a/Makefile b/Makefile
index ba02f07..79b79ab 100644
--- a/Makefile
+++ b/Makefile
@@ -3,7 +3,7 @@
 CC     = gcc
 STD    = _GNU_SOURCE
 CFLAGS = -DDEBUG
-BIN    = cas container_of lock_free_queue_test lf_fifo_test lf_ring_buffer_test
+BIN    = cas container_of lock_free_queue_test lf_fifo_test lf_ring_buffer_test lf_ring_buffer_test2
 
 .c.o:
 	$(CC) -march=i686 -O2 -c -Wall -I. $(CFLAGS) -D$(STD) $<
@@ -35,5 +35,10 @@ lf_ring_buffer_test.o: lf_ring_buffer_test.c
 lf_ring_buffer_test: lf_ring_buffer.o lf_ring_buffer_test.o
 	$(CC) -lrt lf_ring_buffer.o lf_ring_buffer_test.o -o lf_ring_buffer_test
 
+lf_ring_buffer2.o: lf_ring_buffer2.h lf_ring_buffer2.c lf_ring_buffer_data.h lf_portable_cas.h
+lf_ring_buffer_test2.o: lf_ring_buffer_test2.c
+lf_ring_buffer_test2: lf_ring_buffer2.o lf_ring_buffer_test2.o
+	$(CC) -lrt lf_ring_buffer2.o lf_ring_buffer_test2.o -o lf_ring_buffer_test2
+
 clean:
 	rm -f *~ *.o *.s core $(BIN)
diff --git a/lf_ring_buffer2.c b/lf_ring_buffer2.c
new file mode 100644
index 0000000..d4ff4d8
--- /dev/null
+++ b/lf_ring_buffer2.c
@@ -0,0 +1,168 @@
+/*
+ * File     : lf_ring_buffer.c
+ * Author   : Jérémy Zurcher  <jeremy@asynk.ch>
+ * Date     : 05/01/010
+ * License  :
+ *
+ *  Permission is hereby granted, free of charge, to any person obtaining
+ *  a copy of this software and associated documentation files (the
+ *  "Software"), to deal in the Software without restriction, including
+ *  without limitation the rights to use, copy, modify, merge, publish,
+ *  distribute, sublicense, and/or sell copies of the Software, and to
+ *  permit persons to whom the Software is furnished to do so, subject to
+ *  the following conditions:
+ *  
+ *  The above copyright notice and this permission notice shall be
+ *  included in all copies or substantial portions of the Software.
+ *  
+ *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ *  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ *  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ *  NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ *  LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+ *  OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ *  WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ *
+ */
+
+#include "lf_ring_buffer2.h"
+#include "lf_portable_cas.h"
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+
+//#define DEBUG_LFRB_KO 1
+//#define DEBUG_LFRB_CAS 1
+//#define DEBUG_LFRB 1
+
+#ifdef DEBUG_LFRB
+    #include <stdio.h>
+    #define _LOG_KO_( ... ) 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__)
+#elif defined DEBUG_LFRB_KO
+    #include <stdio.h>
+    #define _LOG_KO_( ... ) fprintf(stdout,__VA_ARGS__)
+    #define _LOG_CAS_( ... )
+#else
+    #define _LOG_KO_( ... )
+    #define _LOG_CAS_( ... )
+#endif
+
+#define BACKOFF_NANO_SLEEP  100000
+
+#define USHORTMAX 0xffff
+
+/* initialize an empty lf_ring_buffer struct */
+lf_ring_buffer_t* lf_ring_buffer_create( size_t n_buf ) {
+    if(n_buf>=USHORTMAX) {
+        return NULL;
+    }
+    /* alloc ring_buffer struct */
+    lf_ring_buffer_t *r = malloc(sizeof(lf_ring_buffer_t));
+    if(r==NULL) return NULL;
+    /* */
+    r->buffer = malloc(LFRB_BUFFER_SIZE*n_buf);
+    if(r->buffer==NULL) {
+        free(r);
+        return NULL;
+    }
+    memset(r->buffer,0,LFRB_BUFFER_SIZE*n_buf);
+    r->n_buf = n_buf;
+    r->indexes = (unsigned int)(USHORTMAX<<16) | (unsigned int)0;
+    return r;
+}
+
+/* destroy an lf_ring_buffer strcture */
+void lf_ring_buffer_destroy( lf_ring_buffer_t *r ) {
+    free(r->buffer);
+    free(r);
+}
+
+/* return 1 if is empty */
+int lf_ring_buffer_empty( lf_ring_buffer_t *r ) { return (r->indexes>>16)==USHORTMAX; }
+
+/* write data into the ring buffer */
+int lf_ring_buffer_write( lf_ring_buffer_t *r, void *data, int flags ) {
+    unsigned int current, next;
+    unsigned int write_to, read_from;
+    struct timespec backoff;
+    int backoff_time = BACKOFF_NANO_SLEEP;
+    /* reserve a buffer */
+    for(;;){
+        /* copy indexes and split it */
+        current = r->indexes;
+        write_to = current&0xffff;
+        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( LFRB_IS_AVAILABLE( r->buffer[write_to] ) && read_from!=write_to ) {
+            next = write_to+1;
+            if (next==r->n_buf) next=0;
+            /* set read_from to write_to if needed */
+            if (read_from==USHORTMAX) {
+                next |= write_to<<16;
+            } else {
+                next |= read_from<<16;
+            }
+            /* try to update indexes */
+            _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);
+            if(IS_NOT_BLOCKING(flags)) return -1;
+        }
+        backoff.tv_sec = 0;
+        backoff.tv_nsec = backoff_time;
+        nanosleep(&backoff,NULL);
+        backoff_time += BACKOFF_NANO_SLEEP;
+    }
+    /* fill this buffer and mark it as filled */
+    memcpy( LFRB_DATA_PTR(r->buffer[write_to]), data, LFRB_DATA_SIZE );
+    LFRB_MARK_AS_FILLED( r->buffer[write_to] );
+    return 0;
+}
+
+/* read data from the ring buffer */
+int lf_ring_buffer_read( lf_ring_buffer_t *r, void *data, int flags ) {
+    unsigned int current, next;
+    unsigned int write_to, read_from, tmp;
+    struct timespec backoff;
+    int backoff_time = BACKOFF_NANO_SLEEP;
+    for(;;) {
+        current = r->indexes;
+        write_to = current&0xffff;
+        read_from = current>>16;
+        if( !(LFRB_IS_AVAILABLE( r->buffer[read_from] )) && read_from!=USHORTMAX ) {
+            tmp = read_from +1;
+            if (tmp==r->n_buf) tmp=0;
+            /* is the buffer empty */
+            if ( tmp==write_to) {
+                tmp = USHORTMAX;
+            }
+            next = tmp<<16 | write_to;
+            _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;
+        }
+        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 !! */
+    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
new file mode 100644
index 0000000..6f0cee1
--- /dev/null
+++ b/lf_ring_buffer2.h
@@ -0,0 +1,72 @@
+/*
+ * File     : lf_ring_buffer.h
+ * Author   : Jérémy Zurcher  <jeremy@asynk.ch>
+ * Date     : 05/01/010
+ * License  :
+ *
+ *  Permission is hereby granted, free of charge, to any person obtaining
+ *  a copy of this software and associated documentation files (the
+ *  "Software"), to deal in the Software without restriction, including
+ *  without limitation the rights to use, copy, modify, merge, publish,
+ *  distribute, sublicense, and/or sell copies of the Software, and to
+ *  permit persons to whom the Software is furnished to do so, subject to
+ *  the following conditions:
+ *  
+ *  The above copyright notice and this permission notice shall be
+ *  included in all copies or substantial portions of the Software.
+ *  
+ *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ *  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ *  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ *  NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ *  LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+ *  OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ *  WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ *
+ */
+
+#ifndef _LF_RING_BUFFER_H_
+#define _LF_RING_BUFFER_H_
+
+# ifdef __cplusplus
+extern "C" {
+# endif /* __cplusplus */
+
+#include <sys/types.h>
+#include "lf_ring_buffer_data.h"
+
+#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 )
+
+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 */
+} lf_ring_buffer_t;
+
+/* return an initialized lf_ring_buffer_t struct */
+lf_ring_buffer_t* lf_ring_buffer_create( size_t n_buf );
+
+/* destroy an lf_ring_buffer_t struct */
+void lf_ring_buffer_destroy( lf_ring_buffer_t *r );
+
+/* return 1 if is empty */
+int lf_ring_buffer_empty( lf_ring_buffer_t *r );
+
+/* write data into the ring buffer
+ * return 0 on success
+ * return -1 if IS_NOT_BLOCKING and buffer is full
+ */
+int lf_ring_buffer_write( lf_ring_buffer_t *r, void *data, int flags );
+
+/* read data from the ring buffer
+ * return 0 on success
+ * return -1 if IS_NOT_BLOCKING and buffer is empty
+ */
+int lf_ring_buffer_read( lf_ring_buffer_t *r, void *data, int flags );
+
+# ifdef __cplusplus
+}
+# endif /* __cplusplus */
+
+# endif /* _LF_RING_BUFFER_H_ */
diff --git a/lf_ring_buffer_test2.c b/lf_ring_buffer_test2.c
new file mode 100644
index 0000000..ecceede
--- /dev/null
+++ b/lf_ring_buffer_test2.c
@@ -0,0 +1,185 @@
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdint.h>
+
+#include "lf_ring_buffer.h"
+
+#define BUFFER_LEN 65000
+
+//static rb_data_t data[BUFFER_LEN][RB_DATA_LEN];
+
+static int64_t time_diff(struct timespec *t0, struct timespec *t1)
+{
+    return ((t1->tv_sec * 1000000000) + t1->tv_nsec) - ((t0->tv_sec * 1000000000) + t0->tv_nsec);
+}
+
+/*
+static void print_now(char* s) {
+    fprintf(stdout,s);
+    fflush(stdout);
+}
+*/
+
+static void report( char* op, int n, uint64_t dt, int redo ) {
+    fprintf(stdout,"\t%9d %s operations in %4d [ms] => %7d [us] => %10d [ns]\t >>> %6d [ns/op]\t%4d redone operations\n",
+            n, op, (int)(dt/1000000), (int)(dt/1000), (int)dt, (int)(dt/n), redo );
+}
+
+struct thread_params {
+    lf_ring_buffer_t *ring;
+    int n;
+    int flags;
+};
+
+/*
+static void feed_data( int n){
+    int i;
+    for(i=0; i<n; i++){
+        sprintf(data[i],"hello world %04d\n",i);
+    }
+}
+*/
+
+static uint64_t sequential_writes( lf_ring_buffer_t *ring, int n, int flags ) {
+    int i, redo=0;
+    rb_data_t data[RB_DATA_LEN];
+    struct timespec start, end;
+    clock_gettime(CLOCK_MONOTONIC, &start);
+    if(flags==0) {
+        for(i=0; i<n; i++) lf_ring_buffer_write( ring, data, flags );
+    } else {
+        for(i=0; i<n;) {
+            if(lf_ring_buffer_write( ring, data, flags )==0) { i++; } else { redo+=1; }
+        }
+    }
+    clock_gettime(CLOCK_MONOTONIC, &end);
+    report( "write", n, time_diff( &start, &end ), redo );
+    return time_diff( &start, &end );
+}
+
+static uint64_t sequential_reads( lf_ring_buffer_t *ring, int n, int flags ) {
+    int i,redo=0;
+    rb_data_t data[RB_DATA_LEN];
+    struct timespec start, end;
+    clock_gettime(CLOCK_MONOTONIC, &start);
+    if(flags==0) {
+        for(i=0; i<n; i++) lf_ring_buffer_read( ring, data, flags );
+    } else {
+        for(i=0; i<n;) {
+            if(lf_ring_buffer_read( ring, data, flags )==0) { i++; } else { redo+=1; }
+        }
+    }
+    clock_gettime(CLOCK_MONOTONIC, &end);
+    report( "read ", n, time_diff( &start, &end ), redo );
+    return time_diff( &start, &end );
+}
+
+void* writer_thread( void* param ) {
+    struct thread_params *params = (struct thread_params*)param;
+    //report( "write", params->n, sequential_writes( params->ring, params->n, params->flags ) );
+    sequential_writes( params->ring, params->n, params->flags );
+    return NULL;
+}
+
+void* reader_thread( void* param ) {
+    struct thread_params *params = (struct thread_params*)param;
+    //report( "read ", params->n, sequential_reads( params->ring, params->n, params->flags ) );
+    sequential_reads( params->ring, params->n, params->flags );
+    return NULL;
+}
+
+static void parallel_op( int op, int nt, lf_ring_buffer_t *ring, int n, int flags ) {
+    int i;
+    
+    pthread_t *threads = malloc( sizeof(pthread_t)*nt);
+    struct thread_params *params = malloc( sizeof(struct thread_params)*nt);
+
+    for(i=0; i<nt; i++) {
+        params[i].ring = ring;
+        params[i].n = n/nt;
+        params[i].flags = flags;
+        if(op==0) {
+            if (pthread_create( &threads[i], NULL, writer_thread, &params[i])) {
+                fprintf(stderr,"Failed to create writer thread[%d]\n",i);
+                exit(1);
+            }
+        } else if(op==1) {
+            if (pthread_create( &threads[i], NULL, reader_thread, &params[i])) {
+                fprintf(stderr,"Failed to create reader thread[%d]\n",i);
+                exit(1);
+            }
+        } else {
+            params[i].n /=2;
+            if (pthread_create( &threads[i], NULL, writer_thread, &params[i])) {
+                fprintf(stderr,"Failed to create writer thread[%d]\n",i);
+                exit(1);
+            }
+            if (pthread_create( &threads[i], NULL, reader_thread, &params[i])) {
+                fprintf(stderr,"Failed to create reader thread[%d]\n",i);
+                exit(1);
+            }
+        }
+    }
+    for(i=0; i<nt; i++) {
+        pthread_join( threads[i], NULL );
+    }
+    /* empty the ring */
+    free(threads);
+    free(params);
+}
+
+int main( int argc, char** argv ) {
+    
+    int i;
+    int b_len = BUFFER_LEN;
+    lf_ring_buffer_t *ring;
+
+    ring = lf_ring_buffer_create( b_len );
+    if(ring==NULL){
+        fprintf(stderr,"ERROR : lf_ring_buffer_create( %d );\n",b_len);
+        exit( EXIT_FAILURE );
+    }
+
+    /*
+    print_now("feed the data ... ");
+    feed_data(b_len);
+    printf("done.\n");
+    */
+
+    printf("sequential non blocking write operations ...\n");
+    sequential_writes( ring, b_len, 0 );
+    printf("sequential non blocking read operations ...\n");
+    sequential_reads( ring, b_len, 0 );
+    if(!lf_ring_buffer_empty(ring)) { fprintf(stderr,"ring should be empty but is not\n"); exit( EXIT_FAILURE ); }
+    printf("sequential blocking write operations ...\n");
+    sequential_writes( ring, b_len, LFRB_NO_BLOCK );
+    printf("sequential blocking read operations ...\n");
+    sequential_reads( ring, b_len, LFRB_NO_BLOCK );
+    if(!lf_ring_buffer_empty(ring)) { fprintf(stderr,"ring should be empty but is not\n"); exit( EXIT_FAILURE ); }
+
+    for(i=5; i<=50;i*=2) {
+        printf("%d parallel blocking with backoff inc write operations .... \n",i);
+        parallel_op( 0, i, ring, b_len, 0 );
+        printf("parallel blocking read operations ...\n");
+        sequential_reads( ring, b_len, 0 );
+        if(!lf_ring_buffer_empty(ring)) { fprintf(stderr,"ring should be empty but is not\n"); exit( EXIT_FAILURE ); }
+    }
+    for(i=5; i<=50;i*=2) {
+        printf("%d parallel non blocking write operations .... \n",i);
+        parallel_op( 0, i, ring, b_len, LFRB_NO_BLOCK );
+        printf("non blocking read operations ...\n");
+        sequential_reads( ring, b_len, LFRB_NO_BLOCK );
+        if(!lf_ring_buffer_empty(ring)) { fprintf(stderr,"ring should be empty but is not\n"); exit( EXIT_FAILURE ); }
+    }
+    for(i=10; i<=50;i*=2) {
+        printf("%d parallel blocking write and read operations .... \n",i*2);
+        parallel_op( 3, i, ring, b_len, 0 );
+        if(!lf_ring_buffer_empty(ring)) { fprintf(stderr,"ring should be empty but is not\n"); exit( EXIT_FAILURE ); }
+    }
+
+    lf_ring_buffer_destroy( ring );
+
+    return EXIT_SUCCESS;
+}
+