parallel tests

1 files changed, 87 insertions, 25 deletions

diff --git a/lf_ring_buffer_test.c b/lf_ring_buffer_test.c
index e9e199a..b3ec016 100644
--- a/lf_ring_buffer_test.c
+++ b/lf_ring_buffer_test.c
@@ -1,4 +1,3 @@
-#include <unistd.h>
 #include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -8,22 +7,31 @@
 
 #define BUFFER_LEN 5000000
 
-static rb_data_t data[BUFFER_LEN][RB_DATA_LEN];
+//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( int n, uint64_t dt ) {
-    fprintf(stdout,"%d operations in %d [ms] => %d [us] => %d [ns]\t >>> %d [ns/op]\n", n, (int)(dt/1000000), (int)(dt/1000), (int)dt, (int)(dt/n) );
+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;
@@ -34,41 +42,68 @@ static void feed_data( int n){
 */
 
 static uint64_t sequential_writes( lf_ring_buffer_t *ring, int n, int flags ) {
-    int i;
+    int i, redo=0;
     rb_data_t data[RB_DATA_LEN];
     struct timespec start, end;
-    
-    print_now("sequential writes ... ");
-
     clock_gettime(CLOCK_MONOTONIC, &start);
-
-    for(i=0; i<n; i++) lf_ring_buffer_write( ring, data, flags );
-
+    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);
-    printf("done.\n");
-    
+    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;
+    int i,redo=0;
     rb_data_t data[RB_DATA_LEN];
     struct timespec start, end;
-    
-    print_now("sequential reads ... ");
-
     clock_gettime(CLOCK_MONOTONIC, &start);
+    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 );
+}
 
-    for(i=0; i<n; i++) lf_ring_buffer_read( ring, data, flags );
+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;
+}
 
-    clock_gettime(CLOCK_MONOTONIC, &end);
-    printf("done.\n");
+static void parallel_writes( int nt, lf_ring_buffer_t *ring, int n, int flags ) {
+    int i;
     
-    return time_diff( &start, &end );
+    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 (pthread_create( &threads[i], NULL, writer_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;
 
@@ -84,10 +119,37 @@ int main( int argc, char** argv ) {
     printf("done.\n");
     */
 
-    report( b_len, sequential_writes( ring, b_len, 0 ) );
-    report( b_len, sequential_reads( ring, b_len, 0 ) );
-    report( b_len, sequential_writes( ring, b_len, 0 ) );
-    report( b_len, sequential_reads( ring, b_len, 0 ) );
+    printf("sequential non blocking write operations ...\n");
+    //report( "write", b_len, sequential_writes( ring, b_len, 0 ) );
+    sequential_writes( ring, b_len, 0 );
+    printf("sequential non blocking read operations ...\n");
+    //report( "read ", b_len, sequential_reads( ring, b_len, 0 ) );
+    sequential_reads( ring, b_len, 0 );
+    printf("sequential blocking write operations ...\n");
+    //report( "write", b_len, sequential_writes( ring, b_len, 0 ) );
+    sequential_writes( ring, b_len, LFRB_NO_BLOCK );
+    printf("sequential blocking read operations ...\n");
+    //report( "read ", b_len, sequential_reads( ring, b_len, 0 ) );
+    sequential_reads( ring, b_len, LFRB_NO_BLOCK );
+
+    for(i=5; i<=100;i*=22) {
+        printf("%d parallel blocking with backoff inc write operations .... \n",i);
+        parallel_writes( i, ring, b_len, 0 );
+        printf("parallel blocking read operations ...\n");
+        sequential_reads( ring, b_len, 0 );
+    }
+    for(i=5; i<=100;i*=22) {
+        printf("%d parallel blocking with no backoff inc write operations .... \n",i);
+        parallel_writes( i, ring, b_len, LFRB_NO_BACKOFF_INC );
+        printf("parallel blocking read operations ...\n");
+        sequential_reads( ring, b_len, LFRB_NO_BACKOFF_INC );
+    }
+    for(i=5; i<=100;i*=22) {
+        printf("%d parallel non blocking write operations .... \n",i);
+        parallel_writes( i, ring, b_len, LFRB_NO_BLOCK );
+        printf("non blocking read operations ...\n");
+        sequential_reads( ring, b_len, LFRB_NO_BLOCK );
+    }
 
     lf_ring_buffer_destroy( ring );