ring_buffer: add mutex to prevent race

nokute78 · nokute78 · commit f0762385528e · 2023-02-02T07:20:02.000+09:00
To prevent following data race. ================== WARNING: ThreadSanitizer: data race (pid=4506) Read of size 8 at 0x7b9400000010 by thread T1: #0 memcpy ../../../../src/libsanitizer/sanitizer_common/sanitizer_common_interceptors.inc:827 (libtsan.so.0+0x6243e) #1 memcpy ../../../../src/libsanitizer/sanitizer_common/sanitizer_common_interceptors.inc:819 (libtsan.so.0+0x6243e) #2 lwrb_read /home/taka/git/fluent-bit/lib/lwrb/lwrb/src/lwrb/lwrb.c:199 (fluent-bit+0x8d3d1) #3 flb_ring_buffer_read /home/taka/git/fluent-bit/src/flb_ring_buffer.c:197 (fluent-bit+0x176143) #4 flb_input_chunk_ring_buffer_collector /home/taka/git/fluent-bit/src/flb_input_chunk.c:1930 (fluent-bit+0x209432) #5 flb_sched_event_handler /home/taka/git/fluent-bit/src/flb_scheduler.c:428 (fluent-bit+0x139c76) #6 flb_engine_start /home/taka/git/fluent-bit/src/flb_engine.c:937 (fluent-bit+0x12615a) #7 flb_lib_worker /home/taka/git/fluent-bit/src/flb_lib.c:629 (fluent-bit+0xa0ced) Previous write of size 8 at 0x7b9400000010 by thread T3: #0 memcpy ../../../../src/libsanitizer/sanitizer_common/sanitizer_common_interceptors.inc:827 (libtsan.so.0+0x6243e) #1 memcpy ../../../../src/libsanitizer/sanitizer_common/sanitizer_common_interceptors.inc:819 (libtsan.so.0+0x6243e) #2 lwrb_write /home/taka/git/fluent-bit/lib/lwrb/lwrb/src/lwrb/lwrb.c:145 (fluent-bit+0x8d13d) #3 flb_ring_buffer_write /home/taka/git/fluent-bit/src/flb_ring_buffer.c:175 (fluent-bit+0x176025) #4 append_to_ring_buffer /home/taka/git/fluent-bit/src/flb_input_chunk.c:1904 (fluent-bit+0x20914f) #5 flb_input_chunk_append_raw /home/taka/git/fluent-bit/src/flb_input_chunk.c:1966 (fluent-bit+0x20954c) #6 input_log_append /home/taka/git/fluent-bit/src/flb_input_log.c:33 (fluent-bit+0x20e8d2) #7 flb_input_log_append /home/taka/git/fluent-bit/src/flb_input_log.c:47 (fluent-bit+0x20e955) #8 ml_stream_buffer_flush /home/taka/git/fluent-bit/plugins/in_tail/tail_file.c:356 (fluent-bit+0x348975) #9 process_content /home/taka/git/fluent-bit/plugins/in_tail/tail_file.c:575 (fluent-bit+0x3494cf) #10 flb_tail_file_chunk /home/taka/git/fluent-bit/plugins/in_tail/tail_file.c:1341 (fluent-bit+0x34d428) #11 in_tail_collect_event /home/taka/git/fluent-bit/plugins/in_tail/tail.c:328 (fluent-bit+0x32bec0) #12 tail_fs_event /home/taka/git/fluent-bit/plugins/in_tail/tail_fs_inotify.c:267 (fluent-bit+0x3329f6) #13 input_collector_fd /home/taka/git/fluent-bit/src/flb_input_thread.c:168 (fluent-bit+0xe0395) #14 engine_handle_event /home/taka/git/fluent-bit/src/flb_input_thread.c:183 (fluent-bit+0xe1111) #15 input_thread /home/taka/git/fluent-bit/src/flb_input_thread.c:384 (fluent-bit+0xe1111) #16 step_callback /home/taka/git/fluent-bit/src/flb_worker.c:43 (fluent-bit+0x157a2e) Location is heap block of size 8193 at 0x7b9400000000 allocated by main thread: #0 calloc ../../../../src/libsanitizer/tsan/tsan_interceptors_posix.cpp:672 (libtsan.so.0+0x31edc) #1 flb_calloc /home/taka/git/fluent-bit/include/fluent-bit/flb_mem.h:89 (fluent-bit+0x175969) #2 flb_ring_buffer_create /home/taka/git/fluent-bit/src/flb_ring_buffer.c:67 (fluent-bit+0x175ab2) #3 flb_input_new /home/taka/git/fluent-bit/src/flb_input.c:326 (fluent-bit+0xd3c48) #4 service_configure_plugin /home/taka/git/fluent-bit/src/fluent-bit.c:710 (fluent-bit+0x8540d) #5 service_configure /home/taka/git/fluent-bit/src/fluent-bit.c:878 (fluent-bit+0x8625f) #6 flb_main /home/taka/git/fluent-bit/src/fluent-bit.c:1182 (fluent-bit+0x87038) #7 main /home/taka/git/fluent-bit/src/fluent-bit.c:1257 (fluent-bit+0x8734e) Thread T1 'flb-pipeline' (tid=4508, running) created by main thread at: #0 pthread_create ../../../../src/libsanitizer/tsan/tsan_interceptors_posix.cpp:969 (libtsan.so.0+0x605b8) #1 mk_utils_worker_spawn /home/taka/git/fluent-bit/lib/monkey/mk_core/mk_utils.c:284 (fluent-bit+0xc06aa0) #2 flb_main /home/taka/git/fluent-bit/src/fluent-bit.c:1231 (fluent-bit+0x871f1) #3 main /home/taka/git/fluent-bit/src/fluent-bit.c:1257 (fluent-bit+0x8734e) Thread T3 'flb-in-tail.0-w' (tid=4510, running) created by thread T1 at: #0 pthread_create ../../../../src/libsanitizer/tsan/tsan_interceptors_posix.cpp:969 (libtsan.so.0+0x605b8) #1 mk_utils_worker_spawn /home/taka/git/fluent-bit/lib/monkey/mk_core/mk_utils.c:284 (fluent-bit+0xc06aa0) #2 flb_tp_thread_start /home/taka/git/fluent-bit/src/flb_thread_pool.c:123 (fluent-bit+0x170263) #3 flb_input_thread_instance_init /home/taka/git/fluent-bit/src/flb_input_thread.c:543 (fluent-bit+0xe1cf6) #4 flb_input_instance_init /home/taka/git/fluent-bit/src/flb_input.c:1130 (fluent-bit+0xd73d0) #5 flb_input_init_all /home/taka/git/fluent-bit/src/flb_input.c:1217 (fluent-bit+0xd78a3) #6 flb_engine_start /home/taka/git/fluent-bit/src/flb_engine.c:717 (fluent-bit+0x12532a) #7 flb_lib_worker /home/taka/git/fluent-bit/src/flb_lib.c:629 (fluent-bit+0xa0ced) SUMMARY: ThreadSanitizer: data race /home/taka/git/fluent-bit/lib/lwrb/lwrb/src/lwrb/lwrb.c:199 in lwrb_read ================== Signed-off-by: Takahiro Yamashita <nokute78@gmail.com>
diff --git a/include/fluent-bit/flb_ring_buffer.h b/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 */
 };
 
diff --git a/src/flb_ring_buffer.c b/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,15 +192,16 @@ 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;
     }
