ring_buffer: add mutex to prevent data race

[nokute78]

This patch is to add pthread_mutex_t 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
    ==================

    ==================
    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
    ==================

---

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

• Example configuration file for the change
• Debug log output from testing the change

• Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

• [N/A] Run local packaging test showing all targets (including any new ones) build.
• [N/A] Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

• [N/A] Documentation required for this feature

Backporting

• [] Backport to latest stable release.

Configuration

1. Run some container
2. Build fluent-bit with -DSANITIZE_THREAD=on
3. Run fluent-bit with following configuration

[SERVICE]
     Storage.path hoge

[INPUT]
     Name tail
     Alias container-logs
#     Path /var/log/syslog
     Path /var/lib/docker/containers/63e1107469fd01febb7e667a3cae782e87a54d46592f683ca39c91417b75216a/63e1107469fd01febb7e667a3cae782e87a54d46592f683ca39c91417b75216a-json.log
     multiline.parser docker, cri
     Tag kube.*
     threaded on
     Buffer_Chunk_Size 64KB
     Buffer_Max_Size 128KB
     Mem_Buf_Limit 10MB
     storage.type filesystem
     storage.pause_on_chunks_overlimit On
     Skip_Long_Lines On
     DB a.db
     DB.locking true
     Read_From_Head on

[OUTPUT]
     Name stdout
     Match *

Debug/Valgrind output

# valgrind --leak-check=full ../../bin/fluent-bit -c b.conf 
==56790== Memcheck, a memory error detector
==56790== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==56790== Using Valgrind-3.18.1 and LibVEX; rerun with -h for copyright info
==56790== Command: ../../bin/fluent-bit -c b.conf
==56790== 
Fluent Bit v2.0.9
* Copyright (C) 2015-2022 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2023/02/02 09:57:36] [ info] [fluent bit] version=2.0.9, commit=d55530fa80, pid=56790
[2023/02/02 09:57:36] [ info] [storage] created root path hoge
[2023/02/02 09:57:36] [ info] [storage] ver=1.2.0, type=memory+filesystem, sync=normal, checksum=off, max_chunks_up=128
[2023/02/02 09:57:36] [ info] [storage] backlog input plugin: storage_backlog.1
[2023/02/02 09:57:36] [ info] [cmetrics] version=0.5.8
[2023/02/02 09:57:36] [ info] [ctraces ] version=0.2.7
[2023/02/02 09:57:36] [ info] [input:tail:container-logs] initializing
[2023/02/02 09:57:36] [ info] [input:tail:container-logs] storage_strategy='filesystem' (memory + filesystem)
[2023/02/02 09:57:37] [ info] [input:tail:container-logs] multiline core started
[2023/02/02 09:57:37] [ info] [input:tail:container-logs] thread instance initialized
(snip)
[2023/02/02 09:57:40] [ info] [input:tail:container-logs] inotify_fs_remove(): inode=5133186 watch_fd=1
[2023/02/02 09:57:40] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/02/02 09:57:40] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==56790== 
==56790== HEAP SUMMARY:
==56790==     in use at exit: 0 bytes in 0 blocks
==56790==   total heap usage: 36,434 allocs, 36,434 frees, 146,724,432 bytes allocated
==56790== 
==56790== All heap blocks were freed -- no leaks are possible
==56790== 
==56790== For lists of detected and suppressed errors, rerun with: -s
==56790== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

---

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.

[pwhelan]

Adding a mutex to a ring buffer kind of defeats the purpose of a lock-free ring buffer.

[edsiper]

is this a real dead-lock or something that valgrind detects as a potential deadlock ? this ring-buffer is lock-free by design

[nokute78]

This is to fix following data race issue not for dead lock.
It was detected by thread sanitizer.

How is it implemented lock free?

    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)

[leonardo-albertovich]

The idea behind that mechanism not being "race safe" was that in the worse case scenario the consumer code would try to flush twice with the second flush returning immediately which was not deemed a problem and particularly not something that would warrant the overhead of having a real lock.

[edsiper]

@nokute78 can you take a look pls at the latest changes in the library, not sure if there something already enhanced/fixed in there (if there is an issue):

https://github.com/MaJerle/lwrb/

[nokute78]

Thank you for comment.

@edsiper I'll check it.
So, should lwrb be implemented as lock free ?

[nokute78]

https://github.com/MaJerle/lwrb/blob/fc8044d99fcd3950c18172fec0edc850b8dfe565/docs/examples_src/example_thread_safety.c
The example of lwrb example_thread_safety.c uses mutex before/after lwrb_write and lwrb_read .

[nokute78]

Too many races are detected while testing #7061.

We can test using following configuration file.

cd build
cmake .. -DSANITIZE_THREAD=on && make
bin/fluent-bit -c a.conf

[SERVICE]
    Flush        0.5
    Daemon       Off
    Log_Level    error

[INPUT]
    name  event_type
    type  logs
    interval_sec 0
    interval_nsec 500

[OUTPUT]
    Name  null
    Match *

Error is

==================
WARNING: ThreadSanitizer: data race (pid=181318)
  Read of size 8 at 0x7b0c00004858 by thread T1:
    #0 flb_input_chunk_ring_buffer_collector /home/taka/git/fluent-bit/src/flb_input_chunk.c:1941 (fluent-bit+0x20a493)
    #1 flb_engine_start /home/taka/git/fluent-bit/src/flb_engine.c:1000 (fluent-bit+0x1276e2)
    #2 flb_lib_worker /home/taka/git/fluent-bit/src/flb_lib.c:629 (fluent-bit+0xa1ced)

  Previous write of size 8 at 0x7b0c00004858 by thread T3:
    #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:95 (fluent-bit+0x1ffd82)
    #2 append_to_ring_buffer /home/taka/git/fluent-bit/src/flb_input_chunk.c:1853 (fluent-bit+0x209f84)
    #3 flb_input_chunk_append_raw /home/taka/git/fluent-bit/src/flb_input_chunk.c:1966 (fluent-bit+0x20a6dd)
    #4 input_log_append /home/taka/git/fluent-bit/src/flb_input_log.c:33 (fluent-bit+0x20fa63)
    #5 flb_input_log_append /home/taka/git/fluent-bit/src/flb_input_log.c:47 (fluent-bit+0x20fae6)
    #6 send_logs /home/taka/git/fluent-bit/plugins/in_event_type/event_type.c:95 (fluent-bit+0x3f78bc)
    #7 cb_collector_time /home/taka/git/fluent-bit/plugins/in_event_type/event_type.c:349 (fluent-bit+0x3f89ec)
    #8 input_pre_cb_collect /home/taka/git/fluent-bit/include/fluent-bit/flb_input.h:486 (fluent-bit+0xdfc58)
    #9 co_init /home/taka/git/fluent-bit/lib/monkey/deps/flb_libco/amd64.c:117 (fluent-bit+0xc45eca)
    #10 engine_handle_event /home/taka/git/fluent-bit/src/flb_input_thread.c:175 (fluent-bit+0xe1f0c)

[edsiper]

https://github.com/MaJerle/lwrb/blob/fc8044d99fcd3950c18172fec0edc850b8dfe565/docs/examples_src/example_thread_safety.c The example of lwrb example_thread_safety.c uses mutex before/after lwrb_write and lwrb_read .

In the example listed, there are 2 threads that writes and 2 threads that reads, that's why a mutex is required. In our case we always have 1 writer and 1 reader.

[github-actions]

This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days.

[github-actions]

This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days.