use a dedicated per-stream writer goroutine, send messages to it over a 1-buffered channel, this eliminates most of the false positive blocking that causes drops

This commit is contained in:
Arceliar 2020-05-17 08:22:02 -05:00
parent 527d443916
commit 15ac2595aa

View File

@ -256,6 +256,11 @@ func (intf *linkInterface) handler() error {
intf.link.core.log.Infof("Disconnected %s: %s, source %s", intf.link.core.log.Infof("Disconnected %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local) strings.ToUpper(intf.info.linkType), themString, intf.info.local)
} }
intf.writer.Act(nil, func() {
if intf.writer.worker != nil {
close(intf.writer.worker)
}
})
return err return err
} }
@ -429,6 +434,7 @@ func (intf *linkInterface) notifyDoKeepAlive() {
type linkWriter struct { type linkWriter struct {
phony.Inbox phony.Inbox
intf *linkInterface intf *linkInterface
worker chan [][]byte
} }
func (w *linkWriter) sendFrom(from phony.Actor, bss [][]byte, isLinkTraffic bool) { func (w *linkWriter) sendFrom(from phony.Actor, bss [][]byte, isLinkTraffic bool) {
@ -437,8 +443,19 @@ func (w *linkWriter) sendFrom(from phony.Actor, bss [][]byte, isLinkTraffic bool
for _, bs := range bss { for _, bs := range bss {
size += len(bs) size += len(bs)
} }
w.intf.notifySending(size, isLinkTraffic) if w.worker == nil {
w.worker = make(chan [][]byte, 1)
go func() {
for bss := range w.worker {
w.intf.msgIO.writeMsgs(bss) w.intf.msgIO.writeMsgs(bss)
}
}()
}
w.intf.notifySending(size, isLinkTraffic)
func() {
defer func() { recover() }()
w.worker <- bss
}()
w.intf.notifySent(size, isLinkTraffic) w.intf.notifySent(size, isLinkTraffic)
}) })
} }