95 data mover keeps crashing because of something in redis adapter - #96
Merged
wsulli merged 3 commits intoJun 25, 2026
Merged
Conversation
There was a problem hiding this comment.
Pull request overview
This PR addresses production crashes in the redis adapter by hardening connection/reconnect behavior under concurrent access, and adds a regression test intended to reproduce the original failure mode.
Changes:
- Make
RedisConnection::connect()safe to call concurrently with other Redis operations by swappingshared_ptrclient instances under a mutex and snapshotting per-call. - Add additional synchronization in
RedisAdapteraround reader management and reconnect threading (tracked reconnect thread, shutdown flag, reader mutex, safer reader-start signaling). - Add a new concurrency stress test covering concurrent traffic + repeated
connect()calls.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
test.cpp |
Adds a RedisConnection concurrency stress test. |
RedisConnection.hpp |
Converts internal client storage to shared_ptr + snapshotting for thread-safe concurrent connect() and usage. |
RedisAdapter.hpp |
Introduces new synchronization primitives/state for reconnect and reader lifecycle management. |
RedisAdapter.cpp |
Implements shutdown-aware reconnect thread handling and adds locking around reader operations and reader start signaling. |
Comments suppressed due to low confidence (2)
RedisAdapter.cpp:174
- In addGenericReader(), subs/keyids are mutated before stopping an existing reader thread for the same token. The reader thread iterates over info.subs/info.keyids without _reader_mtx, so this ordering can cause a data race/UB if the reader is running. Stop the reader first (when token is valid), then mutate the per-token state, then restart.
uint32_t token = reader_token(key);
reader_info& info = _reader[token];
info.subs[key].push_back(make_reader_callback(func));
info.keyids[key] = "$";
RedisAdapter.cpp:292
- In add_reader_helper(), subs/keyids are updated before stopping an existing reader thread for the token. Since the reader thread reads info.subs/info.keyids without _reader_mtx, this can race if the reader is currently running. Stop the reader first (when token is valid), then update state, then restart.
uint32_t token = reader_token(key);
reader_info& info = _reader[token];
info.subs[key].push_back(func);
info.keyids[key] = "$";
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
412
to
+415
| // wait until notified that thread is running (or timeout) | ||
| bool nto = cv.wait_for(lk, THREAD_START_CONFIRM) == cv_status::no_timeout; | ||
| if ( ! nto) syslog(LOG_WARNING, "start_reader timeout waiting for thread start"); | ||
| return nto; | ||
| bool nto = info.start_cv.wait_for(lk, THREAD_START_CONFIRM) == cv_status::no_timeout; | ||
| if ( ! nto) syslog(LOG_WARNING, "start_reader timeout waiting for thread start"); | ||
| return nto; |
Contributor
Author
There was a problem hiding this comment.
I don't think we've actually experienced this
Comment on lines
496
to
500
| @@ -492,6 +498,12 @@ class RedisAdapter | |||
| std::unordered_map<std::string, std::string> keyids; | |||
| std::string stop; | |||
| bool run = false; | |||
Comment on lines
+357
to
360
| info.start_cv.notify_all(); // notify about to enter loop (NOT in loop) | ||
|
|
||
| for (Streams out; info.run; out.clear()) | ||
| { |
Contributor
Author
There was a problem hiding this comment.
I think this is addressed by making run atomic, though it's a bool, so it doesn't probably need to be but we'll do it anyway.
Comment on lines
+547
to
+562
| atomic<bool> stop{false}; | ||
| vector<thread> users; | ||
| for (int i = 0; i < 8; i++) | ||
| { | ||
| users.emplace_back([&]() | ||
| { | ||
| while ( ! stop) | ||
| { | ||
| conn.ping(); | ||
| conn.del("redis-connection-concurrent-connect-test-key"); | ||
| } | ||
| } | ||
| ); | ||
| } | ||
|
|
||
| for (int i = 0; i < 500; i++) { conn.connect(opts); } |
Contributor
Author
There was a problem hiding this comment.
I don't care if we fix this or not
Member
|
I'm with @wsulli : this PR is a bit beyond my paygrade. That being said two things stand out: (1) it's pretty limited and well-scoped. (2) it would slightly smaller and less complex if we dropped cluster support - should make that a priority moving forward. It's incremental improvement; LGTM once we assess AI comments for merit |
derekste
approved these changes
Jun 25, 2026
wsulli
deleted the
95-data-mover-keeps-crashing-because-of-something-in-redis-adapter
branch
June 25, 2026 17:11
This was referenced Jul 15, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
I don't understand redis adapter well enough to know if this is a good idea or not, but it does fix the crashes. I do want to see the benchmark results if those are still a thing.