Race condition / data race

/* PoC demonstrating a data race between a signal handler and the main thread. Compile with ThreadSanitizer to observe the race. */ #include <stdio.h> #include <stdlib.h> #include <signal.h> #include <unistd.h> #include <pthread.h> /* Shared state without atomic protection (demonstrates data race) */ volatile int g_shutdown_asap = 0; volatile int g_last_sig = 0; void sigint_handler(int sig) { /* Deliberate data race: non-atomic writes from a signal handler */ g_shutdown_asap = 1; g_last_sig = sig; } void* raiser(void* arg) { sleep(2); raise(SIGINT); return NULL; } int main() { struct sigaction sa; sa.sa_handler = sigint_handler; sigemptyset(&sa.sa_mask); sa.sa_flags = 0; if (sigaction(SIGINT, &sa, NULL) != 0) { perror("sigaction"); return 1; } pthread_t t; if (pthread_create(&t, NULL, raiser, NULL) != 0) { perror("pthread_create"); return 1; } while (!g_shutdown_asap) { usleep(1000); } printf("Shutdown ASAP requested by signal: last_sig=%d\n", g_last_sig); pthread_join(t, NULL); return 0; }

diff --git a/.github/workflows/daily.yml b/.github/workflows/daily.yml index 7c41dc8d9d5..89f6e61e53e 100644 --- a/.github/workflows/daily.yml +++ b/.github/workflows/daily.yml @@ -697,6 +697,52 @@ jobs: if: true && !contains(github.event.inputs.skiptests, 'unittest') run: ./src/redis-server test all --accurate + test-sanitizer-thread: + runs-on: ubuntu-latest + if: | + (github.event_name == 'workflow_dispatch' || (github.event_name != 'workflow_dispatch' && github.repository == 'redis/redis')) && + !contains(github.event.inputs.skipjobs, 'sanitizer') + timeout-minutes: 14400 + strategy: + matrix: + compiler: [ gcc, clang ] + env: + CC: ${{ matrix.compiler }} + steps: + - name: prep + if: github.event_name == 'workflow_dispatch' + run: | + echo "GITHUB_REPOSITORY=${{github.event.inputs.use_repo}}" >> $GITHUB_ENV + echo "GITHUB_HEAD_REF=${{github.event.inputs.use_git_ref}}" >> $GITHUB_ENV + echo "skipjobs: ${{github.event.inputs.skipjobs}}" + echo "skiptests: ${{github.event.inputs.skiptests}}" + echo "test_args: ${{github.event.inputs.test_args}}" + echo "cluster_test_args: ${{github.event.inputs.cluster_test_args}}" + - uses: actions/checkout@v4 + with: + repository: ${{ env.GITHUB_REPOSITORY }} + ref: ${{ env.GITHUB_HEAD_REF }} + - name: make + # TODO Investigate why jemalloc with clang TSan crash on start; + # with gcc TSan, jemalloc works modulo sentinel tests hanging. + run: make SANITIZER=thread USE_JEMALLOC=no REDIS_CFLAGS='-DREDIS_TEST -Werror -DDEBUG_ASSERTIONS' + - name: testprep + run: | + sudo apt-get update + sudo apt-get install tcl8.6 tclx -y + - name: test + if: true && !contains(github.event.inputs.skiptests, 'redis') + run: ./runtest --tsan --clients 1 --config io-threads 4 --accurate --verbose --dump-logs ${{github.event.inputs.test_args}} + - name: module api test + if: true && !contains(github.event.inputs.skiptests, 'modules') + run: CFLAGS='-Werror' ./runtest-moduleapi --tsan --clients 1 --config io-threads 4 --accurate --verbose --dump-logs ${{github.event.inputs.test_args}} + - name: sentinel tests + if: true && !contains(github.event.inputs.skiptests, 'sentinel') + run: ./runtest-sentinel --config io-threads 2 ${{github.event.inputs.cluster_test_args}} + - name: cluster tests + if: true && !contains(github.event.inputs.skiptests, 'cluster') + run: ./runtest-cluster --config io-threads 2 ${{github.event.inputs.cluster_test_args}} + test-centos-jemalloc: runs-on: ubuntu-latest if: | diff --git a/src/debug.c b/src/debug.c index 43b6cf13af6..5be265e99b5 100644 --- a/src/debug.c +++ b/src/debug.c @@ -2501,7 +2501,7 @@ void removeSigSegvHandlers(void) { } void printCrashReport(void) { - server.crashing = 1; + atomicSet(server.crashing, 1); /* Log INFO and CLIENT LIST */ logServerInfo(); @@ -2583,6 +2583,9 @@ void serverLogHexDump(int level, char *descr, void *value, size_t len) { #include <sys/time.h> void sigalrmSignalHandler(int sig, siginfo_t *info, void *secret) { + /* Save and restore errno to avoid spoiling it's value as caught by + * WARNING: ThreadSanitizer: signal handler spoils errno */ + int save_errno = errno; #ifdef HAVE_BACKTRACE ucontext_t *uc = (ucontext_t*) secret; #else @@ -2603,6 +2606,7 @@ void sigalrmSignalHandler(int sig, siginfo_t *info, void *secret) { serverLogRawFromHandler(LL_WARNING,"Sorry: no support for backtrace()."); #endif serverLogRawFromHandler(LL_WARNING,"--------\n"); + errno = save_errno; } /* Schedule a SIGALRM delivery after the specified period in milliseconds. diff --git a/src/networking.c b/src/networking.c index 0af0a1cef9d..028eae68b03 100644 --- a/src/networking.c +++ b/src/networking.c @@ -3111,6 +3111,12 @@ char *getClientSockname(client *c) { return c->sockname; } +static inline int isCrashing(void) { + int crashing; + atomicGet(server.crashing, crashing); + return crashing; +} + /* Concatenate a string representing the state of a client in a human * readable format, into the sds string 's'. */ sds catClientInfoString(sds s, client *client) { @@ -3120,7 +3126,7 @@ sds catClientInfoString(sds s, client *client) { int paused = 0; if (client->running_tid != IOTHREAD_MAIN_THREAD_ID && pthread_equal(server.main_thread_id, pthread_self()) && - !server.crashing) + !isCrashing()) { paused = 1; pauseIOThread(client->running_tid); @@ -3221,7 +3227,7 @@ sds getAllClientsInfoString(int type) { /* Pause all IO threads to access data of clients safely, and pausing the * specific IO thread will not repeatedly execute in catClientInfoString. */ int allpaused = 0; - if (server.io_threads_num > 1 && !server.crashing && + if (server.io_threads_num > 1 && !isCrashing() && pthread_equal(server.main_thread_id, pthread_self())) { allpaused = 1; diff --git a/src/server.c b/src/server.c index 3e6b076ed55..cbb5220238a 100644 --- a/src/server.c +++ b/src/server.c @@ -108,6 +108,12 @@ static inline int isCommandReusable(struct redisCommand *cmd, robj *commandArg) * function of Redis may be called from other threads. */ void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst); +static inline int shouldShutdownAsap(void) { + int shutdown_asap; + atomicGet(server.shutdown_asap, shutdown_asap); + return shutdown_asap; +} + /* Low level logging. To use only for very big messages, otherwise * serverLog() is to prefer. */ void serverLogRaw(int level, const char *msg) { @@ -1484,11 +1490,13 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { /* We received a SIGTERM or SIGINT, shutting down here in a safe way, as it is * not ok doing so inside the signal handler. */ - if (server.shutdown_asap && !isShutdownInitiated()) { + if (shouldShutdownAsap() && !isShutdownInitiated()) { int shutdownFlags = SHUTDOWN_NOFLAGS; - if (server.last_sig_received == SIGINT && server.shutdown_on_sigint) + int last_sig_received; + atomicGet(server.last_sig_received, last_sig_received); + if (last_sig_received == SIGINT && server.shutdown_on_sigint) shutdownFlags = server.shutdown_on_sigint; - else if (server.last_sig_received == SIGTERM && server.shutdown_on_sigterm) + else if (last_sig_received == SIGTERM && server.shutdown_on_sigterm) shutdownFlags = server.shutdown_on_sigterm; if (prepareForShutdown(shutdownFlags) == C_OK) exit(0); @@ -1730,11 +1738,11 @@ void whileBlockedCron(void) { /* We received a SIGTERM during loading, shutting down here in a safe way, * as it isn't ok doing so inside the signal handler. */ - if (server.shutdown_asap && server.loading) { + if (shouldShutdownAsap() && server.loading) { if (prepareForShutdown(SHUTDOWN_NOSAVE) == C_OK) exit(0); serverLog(LL_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information"); - server.shutdown_asap = 0; - server.last_sig_received = 0; + atomicSet(server.shutdown_asap, 0); + atomicSet(server.last_sig_received, 0); } } @@ -4616,10 +4624,10 @@ int isReadyToShutdown(void) { } static void cancelShutdown(void) { - server.shutdown_asap = 0; + atomicSet(server.shutdown_asap, 0); server.shutdown_flags = 0; server.shutdown_mstime = 0; - server.last_sig_received = 0; + atomicSet(server.last_sig_received, 0); replyToClientsBlockedOnShutdown(); unpauseActions(PAUSE_DURING_SHUTDOWN); } @@ -4628,10 +4636,10 @@ static void cancelShutdown(void) { int abortShutdown(void) { if (isShutdownInitiated()) { cancelShutdown(); - } else if (server.shutdown_asap) { + } else if (shouldShutdownAsap()) { /* Signal handler has requested shutdown, but it hasn't been initiated * yet. Just clear the flag. */ - server.shutdown_asap = 0; + atomicSet(server.shutdown_asap, 0); } else { /* Shutdown neither initiated nor requested. */ return C_ERR; @@ -6772,7 +6780,7 @@ static void sigShutdownHandler(int sig) { * If we receive the signal the second time, we interpret this as * the user really wanting to quit ASAP without waiting to persist * on disk and without waiting for lagging replicas. */ - if (server.shutdown_asap && sig == SIGINT) { + if (shouldShutdownAsap() && sig == SIGINT) { serverLogRawFromHandler(LL_WARNING, "You insist... exiting now."); rdbRemoveTempFile(getpid(), 1); exit(1); /* Exit with an error since this was not a clean shutdown. */ @@ -6781,8 +6789,8 @@ static void sigShutdownHandler(int sig) { } serverLogRawFromHandler(LL_WARNING, msg); - server.shutdown_asap = 1; - server.last_sig_received = sig; + atomicSet(server.shutdown_asap, 1); + atomicSet(server.last_sig_received, sig); } void setupSignalHandlers(void) { diff --git a/src/server.h b/src/server.h index a03c59169d1..d631cc19c95 100644 --- a/src/server.h +++ b/src/server.h @@ -1762,10 +1762,10 @@ struct redisServer { rax *errors; /* Errors table */ int errors_enabled; /* If true, errorstats is enabled, and we will add new errors. */ unsigned int lruclock; /* Clock for LRU eviction */ - volatile sig_atomic_t shutdown_asap; /* Shutdown ordered by signal handler. */ - volatile sig_atomic_t crashing; /* Server is crashing report. */ + redisAtomic int shutdown_asap; /* Shutdown ordered by signal handler. */ + redisAtomic int crashing; /* Server is crashing report. */ mstime_t shutdown_mstime; /* Timestamp to limit graceful shutdown. */ - int last_sig_received; /* Indicates the last SIGNAL received, if any (e.g., SIGINT or SIGTERM). */ + redisAtomic int last_sig_received; /* Indicates the last SIGNAL received, if any (e.g., SIGINT or SIGTERM). */ int shutdown_flags; /* Flags passed to prepareForShutdown(). */ int activerehashing; /* Incremental rehash in serverCron() */ int active_defrag_running; /* Active defragmentation running (holds current scan aggressiveness) */ diff --git a/src/threads_mngr.c b/src/threads_mngr.c index 606f39c2fdf..dec83e381a1 100644 --- a/src/threads_mngr.c +++ b/src/threads_mngr.c @@ -24,8 +24,8 @@ static const clock_t RUN_ON_THREADS_TIMEOUT = 2; /*================================= Globals ================================= */ -static run_on_thread_cb g_callback = NULL; -static volatile size_t g_tids_len = 0; +static redisAtomic run_on_thread_cb g_callback = NULL; +static redisAtomic size_t g_tids_len = 0; static redisAtomic size_t g_num_threads_done = 0; /* This flag is set while ThreadsManager_runOnThreads is running */ @@ -63,15 +63,15 @@ int ThreadsManager_runOnThreads(pid_t *tids, size_t tids_len, run_on_thread_cb c } /* Update g_callback */ - g_callback = callback; + atomicSet(g_callback, callback); /* Set g_tids_len */ - g_tids_len = tids_len; + atomicSet(g_tids_len, tids_len); /* set g_num_threads_done to 0 To handler the case where in the previous run we reached the timeout and called ThreadsManager_cleanups before one or more threads were done and increased (the already set to 0) g_num_threads_done */ - g_num_threads_done = 0; + atomicSet(g_num_threads_done, 0); /* Send signal to all the threads in tids */ pid_t pid = getpid(); @@ -103,7 +103,9 @@ static int test_and_start(void) { __attribute__ ((noinline)) static void invoke_callback(int sig) { UNUSED(sig); - run_on_thread_cb callback = g_callback; + + run_on_thread_cb callback; + atomicGet(g_callback, callback); if (callback) { callback(); atomicIncr(g_num_threads_done, 1); @@ -122,6 +124,7 @@ static void wait_threads(void) { /* Wait until all threads are done to invoke the callback or until we reached the timeout */ size_t curr_done_count; struct timespec curr_time; + size_t tids_len; do { struct timeval tv = { @@ -132,7 +135,8 @@ static void wait_threads(void) { select(0, NULL, NULL, NULL, &tv); atomicGet(g_num_threads_done, curr_done_count); clock_gettime(CLOCK_REALTIME, &curr_time); - } while (curr_done_count < g_tids_len && + atomicGet(g_tids_len, tids_len); + } while (curr_done_count < tids_len && curr_time.tv_sec <= timeout_time.tv_sec); if (curr_time.tv_sec > timeout_time.tv_sec) { @@ -142,9 +146,9 @@ static void wait_threads(void) { } static void ThreadsManager_cleanups(void) { - g_callback = NULL; - g_tids_len = 0; - g_num_threads_done = 0; + atomicSet(g_callback, NULL); + atomicSet(g_tids_len, 0); + atomicSet(g_num_threads_done, 0); /* Lastly, turn off g_in_progress */ atomicSet(g_in_progress, 0); diff --git a/src/tsan.sup b/src/tsan.sup new file mode 100644 index 00000000000..786d1d3df87 --- /dev/null +++ b/src/tsan.sup @@ -0,0 +1,8 @@ +# collect_stacktrace_data() calls backtrace() from a signal handler but +# backtrace() is signal-unsafe since it might allocate memory, at least on +# glibc 2.39 it does through a call to _dl_map_object_deps(). +signal:collect_stacktrace_data +signal:printCrashReport +# TODO Investigate this race in jemalloc probably related to +# https://github.com/jemalloc/jemalloc/issues/2621 +race:malloc_mutex_trylock_final diff --git a/tests/integration/aof.tcl b/tests/integration/aof.tcl index 7fbf520da86..d4a556e3bc5 100644 --- a/tests/integration/aof.tcl +++ b/tests/integration/aof.tcl @@ -208,7 +208,7 @@ tags {"aof external:skip"} { } } - start_server {overrides {appendonly {yes} appendfsync always}} { + start_server {tags {"tsan:skip"} overrides {appendonly {yes} appendfsync always}} { test {AOF fsync always barrier issue} { set rd [redis_deferring_client] # Set a sleep when aof is flushed, so that we have a chance to look diff --git a/tests/integration/corrupt-dump.tcl b/tests/integration/corrupt-dump.tcl index ee044c71a79..8d25cdd85ba 100644 --- a/tests/integration/corrupt-dump.tcl +++ b/tests/integration/corrupt-dump.tcl @@ -14,7 +14,7 @@ tags {"dump" "corruption" "external:skip"} { # iterating as many as 2^61 hash table slots. set arch_name [exec uname -m] -set run_oom_tests [expr {$arch_name == "x86_64" || $arch_name == "aarch64"}] +set run_oom_tests [expr {($arch_name == "x86_64" || $arch_name == "aarch64") && !$::tsan}] set corrupt_payload_7445 "\x0E\x01\x1D\x1D\x00\x00\x00\x16\x00\x00\x00\x03\x00 ... [truncated]