ring_buffer: add flb_ring_buffer_set_flush_pending

To prevent following data race.

==================
WARNING: ThreadSanitizer: data race (pid=31138)
  Read of size 4 at 0x7b1c000002b0 by thread T3 (mutexes: write M12):
    #0 flb_ring_buffer_write /home/taka/git/fluent-bit/src/flb_ring_buffer.c:186 (fluent-bit+0x17608f)
    #1 append_to_ring_buffer /home/taka/git/fluent-bit/src/flb_input_chunk.c:1904 (fluent-bit+0x2091c4)
    #2 flb_input_chunk_append_raw /home/taka/git/fluent-bit/src/flb_input_chunk.c:1966 (fluent-bit+0x2095c1)
    #3 input_log_append /home/taka/git/fluent-bit/src/flb_input_log.c:33 (fluent-bit+0x20e947)
    #4 flb_input_log_append /home/taka/git/fluent-bit/src/flb_input_log.c:47 (fluent-bit+0x20e9ca)
    #5 ml_stream_buffer_flush /home/taka/git/fluent-bit/plugins/in_tail/tail_file.c:356 (fluent-bit+0x3489ea)
    #6 process_content /home/taka/git/fluent-bit/plugins/in_tail/tail_file.c:575 (fluent-bit+0x349544)
    #7 flb_tail_file_chunk /home/taka/git/fluent-bit/plugins/in_tail/tail_file.c:1341 (fluent-bit+0x34d49d)
    #8 in_tail_collect_static /home/taka/git/fluent-bit/plugins/in_tail/tail.c:188 (fluent-bit+0x32b489)
    #9 input_collector_fd /home/taka/git/fluent-bit/src/flb_input_thread.c:168 (fluent-bit+0xe0395)
    #10 engine_handle_event /home/taka/git/fluent-bit/src/flb_input_thread.c:183 (fluent-bit+0xe1111)
    #11 input_thread /home/taka/git/fluent-bit/src/flb_input_thread.c:384 (fluent-bit+0xe1111)
    #12 step_callback /home/taka/git/fluent-bit/src/flb_worker.c:43 (fluent-bit+0x157a2e)

  Previous write of size 4 at 0x7b1c000002b0 by thread T1:
    #0 flb_input_chunk_ring_buffer_collector /home/taka/git/fluent-bit/src/flb_input_chunk.c:1949 (fluent-bit+0x2094dd)
    #1 flb_sched_event_handler /home/taka/git/fluent-bit/src/flb_scheduler.c:428 (fluent-bit+0x139c76)
    #2 flb_engine_start /home/taka/git/fluent-bit/src/flb_engine.c:937 (fluent-bit+0x12615a)
    #3 flb_lib_worker /home/taka/git/fluent-bit/src/flb_lib.c:629 (fluent-bit+0xa0ced)

  Location is heap block of size 104 at 0x7b1c000002a0 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:50 (fluent-bit+0x1759d2)
    #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)

  Mutex M12 (0x7b1c000002d8) created at:
    #0 pthread_mutex_init ../../../../src/libsanitizer/tsan/tsan_interceptors_posix.cpp:1227 (libtsan.so.0+0x4bee1)
    #1 flb_ring_buffer_create /home/taka/git/fluent-bit/src/flb_ring_buffer.c:56 (fluent-bit+0x175a40)
    #2 flb_input_new /home/taka/git/fluent-bit/src/flb_input.c:326 (fluent-bit+0xd3c48)
    #3 service_configure_plugin /home/taka/git/fluent-bit/src/fluent-bit.c:710 (fluent-bit+0x8540d)
    #4 service_configure /home/taka/git/fluent-bit/src/fluent-bit.c:878 (fluent-bit+0x8625f)
    #5 flb_main /home/taka/git/fluent-bit/src/fluent-bit.c:1182 (fluent-bit+0x87038)
    #6 main /home/taka/git/fluent-bit/src/fluent-bit.c:1257 (fluent-bit+0x8734e)

  Thread T3 'flb-in-tail.0-w' (tid=31142, 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+0xc06b10)
    #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)

  Thread T1 'flb-pipeline' (tid=31140, 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+0xc06b10)
    #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)

SUMMARY: ThreadSanitizer: data race /home/taka/git/fluent-bit/src/flb_ring_buffer.c:186 in flb_ring_buffer_write
==================

Signed-off-by: Takahiro Yamashita <[email protected]>

Author: nokute78

include/fluent-bit/flb_ring_buffer.h

@@ -42,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

@@ -209,4 +209,9 @@ int flb_ring_buffer_read(struct flb_ring_buffer *rb, void *ptr, size_t size)
     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);
+}