add several changes, see commit message.

.github/workflows/cmake-multi-platform.yml

@@ -25,8 +25,11 @@ jobs:
       # todo: add windows-latest
       matrix:
         os: [ubuntu-latest, ubuntu-22.04-arm]
-        build_type: [Release, Debug, Tsan]
+        build_type: [Release, Debug, Tsan, Asan]
         c_compiler: [gcc, clang]
+        exclude:
+            - c_compiler: gcc
+              build_type: Asan
 
     steps:
     - uses: actions/checkout@v4

CMakeLists.txt

@@ -16,6 +16,8 @@
 
 cmake_minimum_required(VERSION 3.12)
 
+set(CMAKE_EXPORT_COMPILE_COMMANDS ON) # for compile commands used by lsp
+
 message(STATUS "path : ${CMAKE_FIND_ROOT_PATH}")
 project(zlog)
 message(STATUS "path : ${CMAKE_FIND_ROOT_PATH}")
@@ -32,17 +34,29 @@ SET(ZLOG_SO_VERSION ${CPACK_PACKAGE_VERSION_MAJOR})
 
 message(STATUS "platform : ${CMAKE_SYSTEM}")
 
-add_definitions("-g -Wall -Wstrict-prototypes")
-set(CMAKE_C_FLAGS "-std=c11 -pedantic -D_DEFAULT_SOURCE")
-set(CMAKE_C_FLAGS_DEBUG "-ggdb3 -DDEBUG")
-set(CMAKE_C_FLAGS_RELEASE "-O2")
+set(CMAKE_C_STANDARD 11)
+set(CMAKE_C_STANDARD_REQUIRED ON)
+set(CMAKE_C_EXTENSIONS OFF) # Optional: set to ON for extensions
 
+set(CMAKE_C_FLAGS_RELEASE "-O2")
 if (CMAKE_VERSION VERSION_GREATER_EQUAL "3.24")
-set(CMAKE_COMPILE_WARNING_AS_ERROR ON)
+    set(CMAKE_COMPILE_WARNING_AS_ERROR ON)
 else()
-set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror")
+    add_compile_options(-Werror)
 endif()
 
+add_compile_options(-g -Wall -Wstrict-prototypes -pedantic -D_DEFAULT_SOURCE)
+add_compile_options(
+    "$<$<CONFIG:Release>:>"
+    "$<$<CONFIG:Debug>:-ggdb3;-DDEBUG>"
+    "$<$<CONFIG:Asan>:-fsanitize=address;-g;-fno-omit-frame-pointer>"
+    "$<$<CONFIG:Tsan>:-fsanitize=thread;-g;-O1;-fno-omit-frame-pointer>"
+)
+add_link_options(
+    "$<$<CONFIG:Asan>:-fsanitize=address>"
+    "$<$<CONFIG:Tsan>:-fsanitize=thread>"
+)
+
 if (WIN32)
     set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -DWINVER=0x0500 -D_WIN32_WINNT=0x0500 ")
 endif ()

scripts/test.sh

@@ -27,6 +27,11 @@ test_press_perf()
 consumer_static_file_single()
 {
     eval "$valgrind_cmd $asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 100 --threadN=10"
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
 }
 
 # varify_static_file_single - check if normal mode and consumer mode outputs identical
@@ -63,33 +68,80 @@ varify_static_file_single()
 test_multi_thread()
 {
     eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 10 -m 10 --threadN=10"
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
+}
+
+test_multi_thread_ftrue()
+{
+    eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/consumer_static_file_single_cmp.conf -n 1000 --threadN=50"
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
+
+    md5target="dc0877b6f5c669ed259fbc650be11ead"
+    for file in zlog.txt.*; do
+        md5output=$(md5sum $file | awk '{ print $1 }')
+        if [ "$md5output" != "$md5target" ]; then
+            echo "failed, $file $md5output != $md5target"
+            return 1
+        fi
+        echo "$file match, $md5output == $md5target"
+    done
+
+    return 0
 }
 
 test_multi_thread_record()
 {
-    eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 10 -m 10 --threadN=10 -r"
+    eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 10 -m 10 --threadN=10 -r > output"
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
 }
 
 test_multi_thread_reload()
 {
-    cmd="$valgrind_cmd $asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 1000 -m 10 --threadN=10 --reloadcnt=10 --reloadms=400 \
+    cmd="$valgrind_cmd $asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 500 -m 10 --threadN=10 --reloadcnt=8 --reloadms=400 \
         -l $conf_dir/test_consumer_static_file_single.conf \
         -l $conf_dir/test_consumer_static_file_single.conf \
         -l $conf_dir/test_static_file_single.conf \
         -l $conf_dir/test_dynamic_file.conf"
     echo "run cmd:"
     echo $cmd
     eval $cmd
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
 }
 
 test_multi_thread_recordms()
 {
-    eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 2 --threadN=10 -r --recordms=100"
+    eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 2 --threadN=10 -r --recordms=100 > output"
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
 }
 
 test_simple()
 {
     eval "$asan_pre $bin_dir/test_dzlog_conf -f $conf_dir/test_consumer_static_file_single.conf -n 10"
+    ret=$?
+    if [[ "$ret" -ne 0 ]]; then
+        echo "failed to test ${FUNCNAME[0]}"
+        return 1
+    fi
 }
 
 fifo()
@@ -103,12 +155,6 @@ fifo()
         echo "failed to test ${FUNCNAME[0]}"
         return 1
     fi
-    # rm -f $target ; touch $target
-    # i=0
-    # while [ "$i" -lt "$testcnt" ]; do
-        # echo "data $i" >> $target
-        # i=$(($i + 1))
-    # done
     md5output=$(md5sum $output | awk '{ print $1 }')
     if [ "$md5output" = "$target" ]; then
         echo "match, $md5output == $target"
@@ -118,6 +164,17 @@ fifo()
     return 1
 }
 
+dump_loop_targetfile()
+{
+    target=outputi
+    rm -f $target ; touch $target
+    i=999
+    while [ "$i" -ge "0" ]; do
+        echo "loglog $i" >> $target
+        i=$(($i - 1))
+    done
+}
+
 while getopts "t:a::v" opt; do
   case $opt in
     t)

src/CMakeLists.txt

@@ -26,13 +26,6 @@ if (CMAKE_BUILD_TYPE STREQUAL "Release")
     set_target_properties(zlog PROPERTIES C_VISIBILITY_PRESET hidden)
 endif()
 
-if (CMAKE_BUILD_TYPE STREQUAL "Asan")
-    if(CMAKE_C_COMPILER_ID MATCHES "GNU|Clang")
-        target_compile_options(zlog PRIVATE -fsanitize=address -g -fno-omit-frame-pointer)
-        target_link_options(zlog PRIVATE -fsanitize=address)
-    endif()
-endif()
-
 add_library(zlog_s
         STATIC
         ${SRCS}
@@ -66,12 +59,3 @@ install(FILES
         COMPONENT zlog
         DESTINATION include
 )
-
-if (CMAKE_BUILD_TYPE STREQUAL "Tsan")
-    if(CMAKE_C_COMPILER_ID MATCHES "GNU|Clang")
-        target_compile_options(zlog PRIVATE -fsanitize=thread -g -O1  -fno-omit-frame-pointer)
-        target_link_options(zlog PRIVATE -fsanitize=thread)
-        target_compile_options(zlog-chk-conf PRIVATE -fsanitize=thread -g -O1  -fno-omit-frame-pointer)
-        target_link_options(zlog-chk-conf PRIVATE -fsanitize=thread)
-    endif()
-endif()

src/consumer.c

@@ -70,8 +70,6 @@ static void handle_log(struct log_consumer *logc, struct msg_head *head, bool *e
                 logc->flush.done = true;
                 assert(!pthread_cond_signal(&logc->flush.cond));
                 assert(!pthread_mutex_unlock(&logc->flush.siglock));
-                zc_debug("consumer flush rec %d, send %d\n", logc->event.sig_recv,
-                         logc->event.sig_send);
             }
             offset += msg_cmd_size();
             break;
@@ -109,42 +107,39 @@ static void *logc_func(void *arg)
 {
     struct log_consumer *logc = arg;
     bool exit = false;
-    unsigned prev_sig_send = 0;
-    for (bool prev_sig_send_valid = false; !exit;) {
+
+    for (; !exit;) {
+        unsigned int sig_send_cache;
         pthread_mutex_lock(&logc->event.siglock);
-        /* impossible */
-        if (logc->event.sig_recv > logc->event.sig_send) {
-            assert(0);
-        }
         /* empty */
-        if (logc->event.sig_recv == logc->event.sig_send ||
-            (prev_sig_send_valid && prev_sig_send == logc->event.sig_send)) {
+        if (logc->event.sig_recv == logc->event.sig_send) {
             pthread_cond_wait(&logc->event.cond, &logc->event.siglock);
         }
-        /* todo: optimize prev_sig_send readability */
-        prev_sig_send = logc->event.sig_send;
+        sig_send_cache = logc->event.sig_send;
         pthread_mutex_unlock(&logc->event.siglock);
         /* has data */
 
-        struct msg_head *head = fifo_peek(logc->event.queue);
-        assert(head);
-        unsigned flag = atomic_load_explicit(&head->flags, memory_order_acquire);
-        if (flag == MSG_HEAD_FLAG_RESERVED) {
-            /* not commit yet, continue wait */
-            prev_sig_send_valid = true;
-            continue;
-        }
+        for (struct msg_head *head = fifo_peek(logc->event.queue); head;
+             head = fifo_peek(logc->event.queue)) {
+            logc->event.sig_recv++;
+
+            unsigned flag = atomic_load_explicit(&head->flags, memory_order_acquire);
+            if (flag == MSG_HEAD_FLAG_RESERVED) {
+                if (logc->event.sig_recv == sig_send_cache) {
+                    /* goto wait til fist commited */
+                    break;
+                }
+                continue;
+            }
 
-        logc->event.sig_recv++;
-        if (flag == MSG_HEAD_FLAG_COMMITED) {
-            handle_log(logc, head, &exit);
-        } else if (flag == MSG_HEAD_FLAG_DISCARDED) {
-        } else {
-            assert(1);
+            if (flag == MSG_HEAD_FLAG_COMMITED) {
+                handle_log(logc, head, &exit);
+            } else if (flag == MSG_HEAD_FLAG_DISCARDED) {
+            } else {
+                assert(1);
+            }
+            fifo_out(logc->event.queue, head);
         }
-        prev_sig_send_valid = false;
-
-        fifo_out(logc->event.queue, head);
     }
     return NULL;
 }

src/fifo.h

@@ -4,6 +4,19 @@
 #include <stdatomic.h>
 #include <stdio.h>
 
+enum msg_head_flag {
+    MSG_HEAD_FLAG_RESERVED = 1,
+    MSG_HEAD_FLAG_COMMITED,
+    MSG_HEAD_FLAG_DISCARDED,
+};
+
+struct msg_head {
+    unsigned total_size;
+    atomic_uint flags;
+
+    char data[];
+};
+
 /* todo: optimize page size macro */
 #define PAGE_SIZE 4096
 
@@ -62,4 +75,9 @@ static inline unsigned int fifo_unused(struct fifo *fifo)
     return fifo_size(fifo) - fifo_used(fifo);
 }
 
+static inline unsigned int msg_head_size(void)
+{
+    return sizeof(struct msg_head);
+}
+
 #endif

src/misc.h

@@ -3,7 +3,6 @@
 
 #include <pthread.h>
 #include <stdalign.h>
-#include <stdatomic.h>
 #include <stddef.h>
 
 #define container_of(ptr, type, member) ((type *)((char *)(ptr) - offsetof(type, member)))
@@ -26,19 +25,6 @@ enum _msg_cmd {
     MSG_CMD_FLUSH,
 };
 
-enum msg_head_flag {
-    MSG_HEAD_FLAG_RESERVED = 1,
-    MSG_HEAD_FLAG_COMMITED,
-    MSG_HEAD_FLAG_DISCARDED,
-};
-
-struct msg_head {
-    unsigned total_size;
-    atomic_uint flags;
-
-    char data[];
-};
-
 struct msg_type {
     unsigned long val;
 };
@@ -83,11 +69,6 @@ struct zlog_output_data {
     } time_str;
 };
 
-static inline unsigned int msg_head_size(void)
-{
-    return sizeof(struct msg_head);
-}
-
 static inline unsigned int msg_usr_str_size(void)
 {
     return sizeof(struct msg_usr_str);

src/rule.c

@@ -146,8 +146,8 @@ static int zlog_rule_output_static_file_single(zlog_rule_t * a_rule, zlog_thread
 		return -1;
 	}
 
-	/* not so thread safe here, as multiple thread may ++fsync_count at the same time */
-	if (a_rule->fsync_period && ++a_rule->fsync_count >= a_rule->fsync_period) {
+	/* achieve thread safe by define fsync_count as atomic and and only the == one will do sync */
+	if (a_rule->fsync_period && ++a_rule->fsync_count == a_rule->fsync_period) {
 		a_rule->fsync_count = 0;
 		if (fsync(a_rule->static_fd)) {
 			zc_error("fsync[%d] fail, errno[%d]", a_rule->static_fd, errno);
@@ -226,7 +226,7 @@ static int zlog_rule_output_static_file_rotate(zlog_rule_t * a_rule, zlog_thread
 		return -1;
 	}
 
-	if (a_rule->fsync_period && ++a_rule->fsync_count >= a_rule->fsync_period) {
+	if (a_rule->fsync_period && ++a_rule->fsync_count == a_rule->fsync_period) {
 		a_rule->fsync_count = 0;
 		if (fsync(fd)) zc_error("fsync[%d] fail, errno[%d]", fd, errno);
 	}
@@ -331,7 +331,7 @@ static int zlog_rule_output_dynamic_file_single(zlog_rule_t * a_rule, zlog_threa
 		return -1;
 	}
 
-	if (a_rule->fsync_period && ++a_rule->fsync_count >= a_rule->fsync_period) {
+	if (a_rule->fsync_period && ++a_rule->fsync_count == a_rule->fsync_period) {
 		a_rule->fsync_count = 0;
 		if (fsync(fd)) zc_error("fsync[%d] fail, errno[%d]", fd, errno);
 	}
@@ -381,7 +381,7 @@ static int zlog_rule_output_dynamic_file_rotate(zlog_rule_t * a_rule, zlog_threa
 		return -1;
 	}
 
-	if (a_rule->fsync_period && ++a_rule->fsync_count >= a_rule->fsync_period) {
+	if (a_rule->fsync_period && ++a_rule->fsync_count == a_rule->fsync_period) {
 		a_rule->fsync_count = 0;
 		if (fsync(fd)) zc_error("fsync[%d] fail, errno[%d]", fd, errno);
 	}