-
-
Notifications
You must be signed in to change notification settings - Fork 1.4k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[FIXED] Possible delays in delivering messages #895
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -202,6 +202,7 @@ type outbound struct { | |
pb int64 // Total pending/queued bytes. | ||
pm int64 // Total pending/queued messages. | ||
sg *sync.Cond // Flusher conditional for signaling. | ||
sgw bool // Indicate flusher is waiting on condition wait. | ||
wdl time.Duration // Snapshot fo write deadline. | ||
mp int64 // snapshot of max pending. | ||
fsp int // Flush signals that are pending from readLoop's pcd. | ||
|
@@ -631,7 +632,9 @@ func (c *client) writeLoop() { | |
c.mu.Lock() | ||
if waitOk && (c.out.pb == 0 || c.out.fsp > 0) && len(c.out.nb) == 0 && !c.flags.isSet(clearConnection) { | ||
// Wait on pending data. | ||
c.out.sgw = true | ||
c.out.sg.Wait() | ||
c.out.sgw = false | ||
} | ||
// Flush data | ||
waitOk = c.flushOutbound() | ||
|
@@ -917,13 +920,21 @@ func (c *client) flushOutbound() bool { | |
} | ||
} | ||
|
||
// Check that if there is still data to send and writeLoop is in wait, | ||
// we need to signal | ||
if c.out.pb > 0 && c.out.sgw { | ||
c.out.sg.Signal() | ||
} | ||
|
||
return true | ||
} | ||
|
||
// flushSignal will use server to queue the flush IO operation to a pool of flushers. | ||
// Lock must be held. | ||
func (c *client) flushSignal() { | ||
c.out.sg.Signal() | ||
if c.out.sgw { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we think this is saving us much? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Did a quick and dirty benchmark of simulating signaling a routine going into a wait, with and without a check with boolean:
But more importantly, even if we don't gain much, why not checking? Signaling while not in wait is just wasted because a signal while not in wait is lost anyway. Since we use mutex to protect wait()/signal(), it makes sense to me to know if we are in wait or not. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok that is convincing! |
||
c.out.sg.Signal() | ||
} | ||
} | ||
|
||
func (c *client) traceMsg(msg []byte) { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why not just signal, why do we need the bool? I think we may just want to signal on partial?