Skip to content

Fix client concurrency - #163

Open
mkuratczyk wants to merge 5 commits into
masterfrom
fix-client-concurrency
Open

mkuratczyk wants to merge 5 commits into
masterfrom
fix-client-concurrency

Conversation

@mkuratczyk

Copy link
Copy Markdown
Collaborator

A few concurrency fixes

Send, sendFrame, Subscribe and Disconnect hold closeMutex while
pushing to writeCh. If processLoop exited with writeCh full, the writer
stayed blocked (for msgSendTimeout, or forever when it is 0) while
processLoop's deferred MustDisconnect, or its ERROR handler, waited for
the same mutex.

Close a processLoopDone channel before processLoop takes closeMutex and
have writers give up with ErrClosedUnexpectedly once it is closed.
Disconnect keeps returning nil when the connection is already going
away.
Unsubscribe is commonly called from the goroutine that reads C. With more
messages in flight than C holds, readLoop blocked on the full C,
processLoop blocked handing it the next frame, and the UNSUBSCRIBE never
got written, so Unsubscribe waited forever (or until its receipt timeout,
after which it blocked sending to C itself).

Once Unsubscribe or abandon starts, readLoop still delivers to C while
there is room and drops what doesn't fit. This also makes abandon's
draining goroutine unnecessary.
Unsubscribe closed C itself when the connection was already closed, and
sent an error message to it on receipt timeout, while readLoop could
still be sending to C. After a Disconnect that timed out waiting for its
receipt, processLoop keeps delivering already buffered MESSAGEs, so a
following Unsubscribe crashed readLoop with "send on closed channel".

Unsubscribe now only reports the error; readLoop, the only sender, closes
C when the connection is torn down.
processLoop handed MESSAGEs to subscriptions with a blocking send, so a
subscriber that didn't keep up with C stopped everything else on the
connection: frames queued by Send were not written, heart-beats were not
sent, and the read timeout couldn't fire. If the connection failed in
the meantime, sendError blocked the same way and processLoop never
exited, so the loss went unnoticed.

Queue frames for subscriptions instead and stop reading while any are
pending, which keeps them in order and still pushes back on the server,
but keep serving writes and heart-beats. The read timer is paused while
processLoop isn't reading by choice. processLoop also exits once
Disconnect closes the connection.

On the way out, close the subscription channels and record the reason
in closeFrame instead of sending an ERROR frame to each subscription,
so a subscriber that stopped reading can't keep processLoop from
exiting. Reply-to UNSUBSCRIBE now ends the subscription with a made-up
RECEIPT, queued behind any pending deliveries, instead of closing the
channel directly, which no longer reports a spurious "channel read
failed" error on C.
readLoop's only way out was a read error. If processLoop exited (read
timeout, write error, ERROR frame) while readLoop was blocked handing it
a frame with readCh full, readLoop blocked forever, leaking the
goroutine and the frames it held.
@mkuratczyk
mkuratczyk force-pushed the fix-client-concurrency branch from ab3488b to 5a3e28f Compare October 16, 2026 10:33
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant