Changeset cf30639

Timestamp:

12/17/08 15:50:25 (12 years ago)

Author:

Perry Lorier <perry@…>

Branches:

4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, getfragoff, help, libtrace4, master, ndag_format, pfring, rc-4.0.1, rc-4.0.2, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance, ringtimestampfixes

Children:

417a136

Parents:

d4a1691

Message:

Threading cleanups (and some memory leaks addressed)

Location:

lib

Files:

• ior-stdio.c (modified) (1 diff)
• ior-thread.c (modified) (12 diffs)
• iow-thread.c (modified) (1 diff)
• wandio.c (modified) (2 diffs)

lib/ior-stdio.c

@@ -52 +52 @@
 {
         close(DATA(io)->fd);
+        free(io->data);
         free(io);
 }

lib/ior-thread.c

@@ -18 +18 @@
         int len;
         enum { EMPTY = 0, FULL = 1 } state;
-        pthread_cond_t dataready;
-        pthread_cond_t spaceavail;
-        pthread_mutex_t mutex;
 };
 
@@ -28 +25 @@
         int offset;
         pthread_t producer;
+        pthread_cond_t space_avail;
+        pthread_cond_t data_ready;
+        pthread_mutex_t mutex;
         bool closing;
+        bool closed;
         io_t *io;
 };
@@ -42 +43 @@
         bool running = true;
 
+        pthread_mutex_lock(&DATA(state)->mutex);
         do {
-                pthread_mutex_lock(&DATA(state)->buffer[buffer].mutex);
                 while (DATA(state)->buffer[buffer].state == FULL) {
                         if (DATA(state)->closing)
                                 break;
-                        pthread_cond_wait(&DATA(state)->buffer[buffer].spaceavail,
-                                        &DATA(state)->buffer[buffer].mutex);
+                        pthread_cond_wait(&DATA(state)->space_avail, &DATA(state)->mutex);
                 }
+
+                if (DATA(state)->closing) {
+                        break;
+                }
+                pthread_mutex_unlock(&DATA(state)->mutex);
+
                 /* Fill the buffer */
                 DATA(state)->buffer[buffer].len=wandio_read(
@@ -56 +62 @@
                                 sizeof(DATA(state)->buffer[buffer].buffer));
 
+                pthread_mutex_lock(&DATA(state)->mutex);
+
                 DATA(state)->buffer[buffer].state = FULL;
 
@@ -61 +69 @@
                 running = (DATA(state)->buffer[buffer].len > 0 );
 
-                pthread_cond_signal(&DATA(state)->buffer[buffer].dataready);
-
-                pthread_mutex_unlock(&DATA(state)->buffer[buffer].mutex);
+                pthread_cond_signal(&DATA(state)->data_ready);
 
                 /* Flip buffers */
@@ -70 +76 @@
         } while(running);
 
+
         wandio_destroy(DATA(state)->io);
+
+        DATA(state)->closed = true;
+        pthread_cond_signal(&DATA(state)->data_ready);
+        pthread_mutex_unlock(&DATA(state)->mutex);
 
         return NULL;
@@ -90 +101 @@
         DATA(state)->in_buffer = 0;
         DATA(state)->offset = 0;
-        pthread_mutex_init(&DATA(state)->buffer[0].mutex,NULL);
-        pthread_cond_init(&DATA(state)->buffer[0].dataready,NULL);
-        pthread_cond_init(&DATA(state)->buffer[0].spaceavail,NULL);
+        pthread_mutex_init(&DATA(state)->mutex,NULL);
+        pthread_cond_init(&DATA(state)->data_ready,NULL);
+        pthread_cond_init(&DATA(state)->space_avail,NULL);
 
         DATA(state)->io = parent;
         DATA(state)->closing = false;
+        DATA(state)->closed = false;
 
         pthread_create(&DATA(state)->producer,NULL,thread_producer,state);
@@ -109 +121 @@
 
         while(len>0) {
-                pthread_mutex_lock(&INBUFFER(state).mutex);
+                pthread_mutex_lock(&DATA(state)->mutex);
                 while (INBUFFER(state).state == EMPTY) {
-                        pthread_cond_wait(&INBUFFER(state).dataready,
-                                        &INBUFFER(state).mutex);
+                        pthread_cond_wait(&DATA(state)->data_ready, &DATA(state)->mutex);
 
                 }
@@ -121 +132 @@
                                 copied = INBUFFER(state).len;
 
-                        pthread_mutex_unlock(&INBUFFER(state).mutex);
+                        pthread_mutex_unlock(&DATA(state)->mutex);
                         return copied;
                 }
 
                 slice=min( INBUFFER(state).len-DATA(state)->offset,len);
+
+                pthread_mutex_unlock(&DATA(state)->mutex);
                                 
                 memcpy(
@@ -136 +149 @@
                 len-=slice;
                 copied+=slice;
+
+                pthread_mutex_lock(&DATA(state)->mutex);
                 DATA(state)->offset+=slice;
                 newbuffer = DATA(state)->in_buffer;
@@ -141 +156 @@
                 if (DATA(state)->offset >= INBUFFER(state).len) {
                         INBUFFER(state).state = EMPTY;
-                        pthread_cond_signal(&INBUFFER(state).spaceavail);
+                        pthread_cond_signal(&DATA(state)->space_avail);
                         newbuffer = (newbuffer+1) % BUFFERS;
                         DATA(state)->offset = 0;
                 }
 
-                pthread_mutex_unlock(&INBUFFER(state).mutex);
+                pthread_mutex_unlock(&DATA(state)->mutex);
 
                 DATA(state)->in_buffer = newbuffer;
@@ -155 +170 @@
 static void thread_close(io_t *io)
 {
-        pthread_mutex_lock(&INBUFFER(io).mutex);
+        pthread_mutex_lock(&DATA(io)->mutex);
         DATA(io)->closing = true;
-        pthread_cond_signal(&INBUFFER(io).spaceavail);
-        pthread_mutex_unlock(&INBUFFER(io).mutex);
+        pthread_cond_signal(&DATA(io)->space_avail);
+
+        /* Wait until the producer thread dies */
+        while (!DATA(io)->closed) {
+                pthread_cond_wait(&DATA(io)->data_ready, &DATA(io)->mutex);
+        }
+        pthread_mutex_unlock(&DATA(io)->mutex);
+        free(DATA(io));
         free(io);
 }

lib/iow-thread.c

@@ -157 +157 @@
         pthread_mutex_unlock(&OUTBUFFER(iow).mutex);
         pthread_join(DATA(iow)->consumer,NULL);
+        free(iow->data);
         free(iow);
 }

lib/wandio.c

@@ -18 +18 @@
         io_t *io = peek_open(stdio_open(filename));
         char buffer[1024];
+        int len;
         if (!io)
                 return NULL;
-        wandio_peek(io, buffer, sizeof(buffer));
+        len = wandio_peek(io, buffer, sizeof(buffer));
 #if HAVE_LIBZ
         /* auto detect gzip compressed data */
-        if (buffer[0] == '\037' && buffer[1] == '\213') { 
+        if (len>=2 && buffer[0] == '\037' && buffer[1] == '\213') { 
                 io = zlib_open(io);
         }
         /* auto detect compress(1) compressed data (gzip can read this) */
-        if (buffer[0] == '\037' && buffer[1] == '\235') {
+        if (len>=2 && buffer[0] == '\037' && buffer[1] == '\235') {
                 io = zlib_open(io);
         }
@@ -33 +34 @@
 #if HAVE_LIBBZ2
         /* auto detect bzip compressed data */
-        else if (buffer[0] == 'B' && buffer[1] == 'Z' && buffer[2] == 'h') { 
+        if (len>=3 && buffer[0] == 'B' && buffer[1] == 'Z' && buffer[2] == 'h') { 
                 io = bz_open(io);
         }