ring_buffer: add mutex to prevent data race

include/fluent-bit/flb_ring_buffer.h

@@ -21,6 +21,7 @@
 #define FLB_RING_BUFFER_H
 
 #include <fluent-bit/flb_pipe.h>
+#include <fluent-bit/flb_pthread.h>
 
 struct flb_ring_buffer {
     void *ctx;                        /* pointer to backend context */
@@ -30,6 +31,7 @@ struct flb_ring_buffer {
     flb_pipefd_t signal_channels[2];  /* flush request signaling channel */
     uint64_t data_window;             /* 0% - 100% occupancy window flush request */
     uint64_t data_size;               /* ring buffer size */
+    pthread_mutex_t pth_mutex;        /* mutex */
     void *data_buf;                   /* ring buffer */
 };
 
@@ -40,5 +42,6 @@ int flb_ring_buffer_add_event_loop(struct flb_ring_buffer *rb, void *evl, uint8_
 
 int flb_ring_buffer_write(struct flb_ring_buffer *rb, void *ptr, size_t size);
 int flb_ring_buffer_read(struct flb_ring_buffer *rb, void *ptr, size_t size);
+void flb_ring_buffer_set_flush_pending(struct flb_ring_buffer *rb, int val);
 
 #endif

src/flb_input_chunk.c

@@ -1945,8 +1945,7 @@ void flb_input_chunk_ring_buffer_collector(struct flb_config *ctx, void *data)
             }
             cr = NULL;
         }
-
-        ins->rb->flush_pending = FLB_FALSE;
+        flb_ring_buffer_set_flush_pending(ins->rb, FLB_FALSE);
     }
 }
 

src/flb_ring_buffer.c

@@ -27,6 +27,7 @@
 #include <fluent-bit/flb_mem.h>
 #include <fluent-bit/flb_log.h>
 #include <fluent-bit/flb_pipe.h>
+#include <fluent-bit/flb_pthread.h>
 #include <fluent-bit/flb_ring_buffer.h>
 #include <fluent-bit/flb_engine_macros.h>
 
@@ -52,6 +53,7 @@ struct flb_ring_buffer *flb_ring_buffer_create(uint64_t size)
         return NULL;
     }
     rb->data_size = size;
+    pthread_mutex_init(&rb->pth_mutex, NULL);
 
     /* lwrb context */
     lwrb = flb_malloc(sizeof(lwrb_t));
@@ -165,15 +167,19 @@ int flb_ring_buffer_write(struct flb_ring_buffer *rb, void *ptr, size_t size)
     size_t ret;
     size_t av;
 
+    pthread_mutex_lock(&rb->pth_mutex);
+
     /* make sure there is enough space available */
     av = lwrb_get_free(rb->ctx);
     if (av < size) {
+        pthread_mutex_unlock(&rb->pth_mutex);
         return -1;
     }
 
     /* write the content */
     ret = lwrb_write(rb->ctx, ptr, size);
     if (ret == 0) {
+        pthread_mutex_unlock(&rb->pth_mutex);
         return -1;
     }
 
@@ -186,20 +192,26 @@ int flb_ring_buffer_write(struct flb_ring_buffer *rb, void *ptr, size_t size)
             flb_pipe_write_all(rb->signal_channels[1], ".", 1);
         }
     }
-
+    pthread_mutex_unlock(&rb->pth_mutex);
     return 0;
 }
 
 int flb_ring_buffer_read(struct flb_ring_buffer *rb, void *ptr, size_t size)
 {
     size_t ret;
-
+    pthread_mutex_lock(&rb->pth_mutex);
     ret = lwrb_read(rb->ctx, ptr, size);
+    pthread_mutex_unlock(&rb->pth_mutex);
     if (ret == 0) {
         return -1;
     }
 
     return 0;
 }
 
-
+void flb_ring_buffer_set_flush_pending(struct flb_ring_buffer *rb, int bool_val)
+{
+    pthread_mutex_lock(&rb->pth_mutex);
+    rb->flush_pending = bool_val;
+    pthread_mutex_unlock(&rb->pth_mutex);
+}