Worker thread based async event processing (full duplex bidi) #2
Conversation
📝 WalkthroughWalkthroughEvent handling moved from synchronous callbacks to an asynchronous model: events are enqueued as EventItem objects and a dedicated worker thread dequeues and dispatches CONNECT_SUCCESS, CONNECTION_DROPPED, CONNECT_ERROR, and MESSAGE events to session handlers; shutdown is signaled and joined safely. Changes
Sequence Diagram(s)sequenceDiagram
participant Ext as External Event Source
participant CB as eventCallback
participant Q as Event Queue
participant WT as workerThread
participant Sess as Session Manager
participant Msg as processMessage
Ext->>CB: Trigger event (CONNECT_SUCCESS / MESSAGE / ...)
CB->>CB: if not m_shutdown
CB->>Q: Enqueue EventItem
CB->>Q: Notify m_queue_cv
loop worker loop
WT->>Q: Wait / Dequeue EventItem
WT->>Sess: Lookup session by id
alt Session Found
alt MESSAGE
WT->>Msg: processMessage(message)
Msg-->>WT: return
else CONNECT_SUCCESS / CONNECTION_DROPPED / CONNECT_ERROR
WT->>Sess: invoke handler (send_initial_metadata / m_notify / media_bug_close)
Sess-->>WT: return
end
else Session Not Found
WT-->>WT: log missing session
end
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. 📜 Recent review detailsConfiguration used: defaults Review profile: CHILL Plan: Pro 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
✏️ Tip: You can disable this entire section by setting Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
audio_streamer_glue.cpp (1)
449-453: Consider draining behavior during shutdown.The current logic drains all queued events before exiting on shutdown. If many events are queued or an event takes a long time to process (e.g., waiting on a lock in
processMessageat line 272-283), the destructor'sjoin()will block.This is likely intentional for graceful shutdown, but if fast shutdown is ever needed, you could add an early exit check within the inner loop:
while (!m_event_queue.empty()) { if (m_shutdown.load()) break; // Optional: immediate exit // ... rest of processing }
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
audio_streamer_glue.cpp(7 hunks)
🔇 Additional comments (4)
audio_streamer_glue.cpp (4)
180-196: LGTM - Thread-safe event enqueueing.The lock guard ensures thread safety, the shutdown check prevents queuing during destruction, and copying the message to a
std::stringavoids dangling pointer issues when the original buffer is freed.
375-388: LGTM - Proper worker thread shutdown sequence.The shutdown flag is set under the lock (consistent with the worker's wait predicate), the condition variable is notified, and the thread is joined before the destructor completes. Since
clientis a member, its destructor runs after this body, and any callbacks will be safely rejected by them_shutdowncheck ineventCallback.
465-497: Session lifecycle handling looks correct.The
switch_core_session_locate/switch_core_session_rwunlockpattern correctly handles the case where the session might be destroyed between event queuing and processing. Events for non-existent sessions are silently dropped, which is appropriate behavior.
720-729: Advisory: Detached thread infinish()interacts with the new async model.The existing
finish()function creates a detached thread that callsdisconnect(). With the new async model:
disconnect()may triggersetCloseCallbackwhich callseventCallback(CONNECTION_DROPPED, ...)- Since
m_shutdownis stillfalseat this point, the event will be queued- The worker thread will try to process it via
switch_core_session_locate()- If the session is already gone, the event is safely dropped
This appears safe due to the session lookup guard, but the AudioStreamer might outlive
stream_session_cleanupsince the detached thread holds the only reference. Consider whether this detached pattern is still desirable, or if the disconnect should be synchronous now that you have a worker thread that can handle it asynchronously.
kingster
changed the title
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
audio_streamer_glue.cpp (1)
412-437: Add mutex protection for WebSocket client operations.The
clientobject is accessed from multiple threads without synchronization:
- The write_frame_thread calls
writeBinary()at multiple points (lines 1020, 1056, 1071, 1117, 1134)- Main/session threads call
writeText()(lines 197, 877, 1176)- A detached thread calls
disconnect()(line 749)- WebSocket callbacks invoke
eventCallback()from the library's internal threadThe check-then-act patterns in
writeBinary()andwriteText()(checkingisConnected()then callingclient.sendBinary()/client.sendMessage()) are not atomic. Another thread could calldisconnect()between the check and the send. Add a dedicated mutex to protect all client operations:isConnected(),sendBinary(),sendMessage(), anddisconnect().
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
audio_streamer_glue.cpp(7 hunks)
🔇 Additional comments (7)
audio_streamer_glue.cpp (7)
11-15: LGTM!The threading-related includes are appropriate for the worker thread implementation.
136-162: Exception safety properly implemented!The try-catch block correctly addresses the past review comment. If
client.connect()throws, the worker thread is cleanly shut down before the exception propagates, preventingstd::terminate()from being called. The cleanup sequence (lock, signal, notify, join, rethrow) is correct.
202-218: LGTM!The event queueing logic is thread-safe and correct. The use of
std::moveat line 216 optimizes performance by avoiding unnecessary string copies, and the shutdown check prevents events from being queued during teardown.
397-410: LGTM!The destructor correctly implements graceful shutdown by signaling the worker thread and waiting for it to complete all queued work before terminating.
427-429: LGTM!The debug log addition is helpful for troubleshooting and respects the suppress_log setting.
452-457: LGTM!The
EventItemstruct is a clean, simple container for queueing events.
534-540: LGTM!The member variables are appropriately typed for the worker thread implementation. The use of
std::atomic<bool>form_shutdownis a safe choice, though technically unnecessary since it's always accessed underm_queue_mutex.
Introduces a worker thread and event queue to handle callback events asynchronously in AudioStreamer, preventing blocking of the event loop and enabling full-duplex communication. Thread safety is ensured with mutexes and condition variables, and proper shutdown is handled in the destructor.
Introduces a worker thread and event queue to handle callback events asynchronously in AudioStreamer, preventing blocking of the event loop and enabling full-duplex communication. Thread safety is ensured with mutexes and condition variables, and proper shutdown is handled in the destructor.
Solves voxcom-us#1
Summary by CodeRabbit
Refactor
Chores
✏️ Tip: You can customize this high-level summary in your review settings.