From 6316fba335c00a01d5304c018480b4a653fbb3b4 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Fri, 18 Aug 2023 12:17:54 +0200 Subject: [PATCH] prevent a channel from blocking all others by using a ring buffer --- channel.go | 157 +++++++++++++++--------------- channel_accepter.go | 9 +- node.go | 83 ++++++++++------ node_streamrequest.go | 7 +- pkg/ringbuffer/event.go | 38 ++++++++ pkg/ringbuffer/ringbuffer.go | 77 +++++++++++++++ pkg/ringbuffer/ringbuffer_test.go | 147 ++++++++++++++++++++++++++++ 7 files changed, 398 insertions(+), 120 deletions(-) create mode 100644 pkg/ringbuffer/event.go create mode 100644 pkg/ringbuffer/ringbuffer.go create mode 100644 pkg/ringbuffer/ringbuffer_test.go diff --git a/channel.go b/channel.go index 9b430f173..cdb7e5622 100644 --- a/channel.go +++ b/channel.go @@ -6,6 +6,13 @@ import ( "github.com/bluenviron/gomavlib/v2/pkg/frame" "github.com/bluenviron/gomavlib/v2/pkg/message" + "github.com/bluenviron/gomavlib/v2/pkg/ringbuffer" +) + +const ( + // this is low in order to avoid accumulating messages + // when a channel is reconnecting + writeBufferSize = 8 ) func randomByte() (byte, error) { @@ -19,15 +26,15 @@ func randomByte() (byte, error) { // For instance, a TCP client endpoint creates a single channel, while a TCP // server endpoint creates a channel for each incoming connection. type Channel struct { - e Endpoint - label string - rwc io.ReadWriteCloser - n *Node - frw *frame.ReadWriter - running bool + e Endpoint + label string + rwc io.Closer + n *Node + frw *frame.ReadWriter + running bool + writeBuffer *ringbuffer.RingBuffer // in - write chan interface{} terminate chan struct{} } @@ -56,14 +63,19 @@ func newChannel(n *Node, e Endpoint, label string, rwc io.ReadWriteCloser) (*Cha return nil, err } + writeBuffer, err := ringbuffer.New(writeBufferSize) + if err != nil { + return nil, err + } + return &Channel{ - e: e, - label: label, - rwc: rwc, - n: n, - frw: frw, - write: make(chan interface{}), - terminate: make(chan struct{}), + e: e, + label: label, + rwc: rwc, + n: n, + frw: frw, + writeBuffer: writeBuffer, + terminate: make(chan struct{}), }, nil } @@ -85,88 +97,73 @@ func (ch *Channel) run() { defer ch.n.channelsWg.Done() readerDone := make(chan struct{}) - go func() { - defer close(readerDone) - - // wait client here, in order to allow the writer goroutine to start - // and allow clients to write messages before starting listening to events - select { - case ch.n.events <- &EventChannelOpen{ch}: - case <-ch.n.terminate: - } + go ch.runReader(readerDone) - for { - fr, err := ch.frw.Read() - if err != nil { - // ignore parse errors - if _, ok := err.(*frame.ReadError); ok { - select { - case ch.n.events <- &EventParseError{err, ch}: - case <-ch.n.terminate: - } - continue - } - return - } + writerDone := make(chan struct{}) + go ch.runWriter(writerDone) - evt := &EventFrame{fr, ch} + select { + case <-readerDone: + ch.rwc.Close() - if ch.n.nodeStreamRequest != nil { - ch.n.nodeStreamRequest.onEventFrame(evt) - } + ch.writeBuffer.Close() + <-writerDone - select { - case ch.n.events <- evt: - case <-ch.n.terminate: - } - } - }() + case <-ch.terminate: + ch.writeBuffer.Close() + <-writerDone - writerDone := make(chan struct{}) - go func() { - defer close(writerDone) + ch.rwc.Close() + <-readerDone + } + + ch.n.pushEvent(&EventChannelClose{ch}) + ch.n.closeChannel(ch) +} + +func (ch *Channel) runReader(readerDone chan struct{}) { + defer close(readerDone) - for what := range ch.write { - switch wh := what.(type) { - case message.Message: - ch.frw.WriteMessage(wh) //nolint:errcheck + // wait client here, in order to allow the writer goroutine to start + // and allow clients to write messages before starting listening to events + ch.n.pushEvent(&EventChannelOpen{ch}) - case frame.Frame: - ch.frw.WriteFrame(wh) //nolint:errcheck + for { + fr, err := ch.frw.Read() + if err != nil { + if _, ok := err.(*frame.ReadError); ok { + ch.n.pushEvent(&EventParseError{err, ch}) + continue } + return } - }() - select { - case <-readerDone: - select { - case ch.n.events <- &EventChannelClose{ch}: - case <-ch.n.terminate: - } + evt := &EventFrame{fr, ch} - select { - case ch.n.channelClose <- ch: - case <-ch.n.terminate: + if ch.n.nodeStreamRequest != nil { + ch.n.nodeStreamRequest.onEventFrame(evt) } - <-ch.terminate - - close(ch.write) - <-writerDone + ch.n.pushEvent(evt) + } +} - ch.rwc.Close() +func (ch *Channel) runWriter(writerDone chan struct{}) { + defer close(writerDone) - case <-ch.terminate: - select { - case ch.n.events <- &EventChannelClose{ch}: - case <-ch.n.terminate: + for { + what, ok := ch.writeBuffer.Pull() + if !ok { + return } - close(ch.write) - <-writerDone + switch wh := what.(type) { + case message.Message: + ch.frw.WriteMessage(wh) //nolint:errcheck - ch.rwc.Close() - <-readerDone + case frame.Frame: + ch.frw.WriteFrame(wh) //nolint:errcheck + } } } @@ -179,3 +176,7 @@ func (ch *Channel) String() string { func (ch *Channel) Endpoint() Endpoint { return ch.e } + +func (ch *Channel) write(what interface{}) { + ch.writeBuffer.Push(what) +} diff --git a/channel_accepter.go b/channel_accepter.go index eb5781d69..3132d35fe 100644 --- a/channel_accepter.go +++ b/channel_accepter.go @@ -22,10 +22,10 @@ func (ca *channelAccepter) close() { func (ca *channelAccepter) start() { ca.n.channelAcceptersWg.Add(1) - go ca.runSingle() + go ca.run() } -func (ca *channelAccepter) runSingle() { +func (ca *channelAccepter) run() { defer ca.n.channelAcceptersWg.Done() for { @@ -42,9 +42,6 @@ func (ca *channelAccepter) runSingle() { panic(fmt.Errorf("newChannel unexpected error: %s", err)) } - select { - case ca.n.channelNew <- ch: - case <-ca.n.terminate: - } + ca.n.newChannel(ch) } } diff --git a/node.go b/node.go index ae45bbf2a..e97420cca 100644 --- a/node.go +++ b/node.go @@ -100,16 +100,16 @@ type Node struct { nodeStreamRequest *nodeStreamRequest // in - channelNew chan *Channel - channelClose chan *Channel - writeTo chan writeToReq - writeAll chan interface{} - writeExcept chan writeExceptReq - terminate chan struct{} + chNewChannel chan *Channel + chCloseChannel chan *Channel + chWriteTo chan writeToReq + chWriteAll chan interface{} + chWriteExcept chan writeExceptReq + terminate chan struct{} // out - events chan Event - done chan struct{} + chEvent chan Event + done chan struct{} } // NewNode allocates a Node. See NodeConf for the options. @@ -169,13 +169,13 @@ func NewNode(conf NodeConf) (*Node, error) { dialectRW: dialectRW, channelAccepters: make(map[*channelAccepter]struct{}), channels: make(map[*Channel]struct{}), - channelNew: make(chan *Channel), - channelClose: make(chan *Channel), - writeTo: make(chan writeToReq), - writeAll: make(chan interface{}), - writeExcept: make(chan writeExceptReq), + chNewChannel: make(chan *Channel), + chCloseChannel: make(chan *Channel), + chWriteTo: make(chan writeToReq), + chWriteAll: make(chan interface{}), + chWriteExcept: make(chan writeExceptReq), terminate: make(chan struct{}), - events: make(chan Event), + chEvent: make(chan Event), done: make(chan struct{}), } @@ -256,15 +256,14 @@ func (n *Node) run() { outer: for { select { - case ch := <-n.channelNew: + case ch := <-n.chNewChannel: n.channels[ch] = struct{}{} ch.start() - case ch := <-n.channelClose: + case ch := <-n.chCloseChannel: delete(n.channels, ch) - ch.close() - case req := <-n.writeTo: + case req := <-n.chWriteTo: if _, ok := n.channels[req.ch]; !ok { continue } @@ -272,25 +271,25 @@ outer: var err error req.what, err = n.encodeMessage(req.what) if err == nil { - req.ch.write <- req.what + req.ch.write(req.what) } - case what := <-n.writeAll: + case what := <-n.chWriteAll: var err error what, err = n.encodeMessage(what) if err == nil { for ch := range n.channels { - ch.write <- what + ch.write(what) } } - case req := <-n.writeExcept: + case req := <-n.chWriteExcept: var err error req.what, err = n.encodeMessage(req.what) if err == nil { for ch := range n.channels { if ch != req.except { - ch.write <- req.what + ch.write(req.what) } } } @@ -318,7 +317,7 @@ outer: } n.channelsWg.Wait() - close(n.events) + close(n.chEvent) } // FixFrame recomputes the Frame checksum and signature. @@ -409,13 +408,13 @@ func (n *Node) encodeMessage(what interface{}) (interface{}, error) { // // See individual events for details. func (n *Node) Events() chan Event { - return n.events + return n.chEvent } // WriteMessageTo writes a message to given channel. func (n *Node) WriteMessageTo(channel *Channel, m message.Message) { select { - case n.writeTo <- writeToReq{channel, m}: + case n.chWriteTo <- writeToReq{channel, m}: case <-n.terminate: } } @@ -423,7 +422,7 @@ func (n *Node) WriteMessageTo(channel *Channel, m message.Message) { // WriteMessageAll writes a message to all channels. func (n *Node) WriteMessageAll(m message.Message) { select { - case n.writeAll <- m: + case n.chWriteAll <- m: case <-n.terminate: } } @@ -431,7 +430,7 @@ func (n *Node) WriteMessageAll(m message.Message) { // WriteMessageExcept writes a message to all channels except specified channel. func (n *Node) WriteMessageExcept(exceptChannel *Channel, m message.Message) { select { - case n.writeExcept <- writeExceptReq{exceptChannel, m}: + case n.chWriteExcept <- writeExceptReq{exceptChannel, m}: case <-n.terminate: } } @@ -441,7 +440,7 @@ func (n *Node) WriteMessageExcept(exceptChannel *Channel, m message.Message) { // since all frame fields must be filled manually. func (n *Node) WriteFrameTo(channel *Channel, fr frame.Frame) { select { - case n.writeTo <- writeToReq{channel, fr}: + case n.chWriteTo <- writeToReq{channel, fr}: case <-n.terminate: } } @@ -451,7 +450,7 @@ func (n *Node) WriteFrameTo(channel *Channel, fr frame.Frame) { // since all frame fields must be filled manually. func (n *Node) WriteFrameAll(fr frame.Frame) { select { - case n.writeAll <- fr: + case n.chWriteAll <- fr: case <-n.terminate: } } @@ -461,7 +460,29 @@ func (n *Node) WriteFrameAll(fr frame.Frame) { // since all frame fields must be filled manually. func (n *Node) WriteFrameExcept(exceptChannel *Channel, fr frame.Frame) { select { - case n.writeExcept <- writeExceptReq{exceptChannel, fr}: + case n.chWriteExcept <- writeExceptReq{exceptChannel, fr}: + case <-n.terminate: + } +} + +func (n *Node) pushEvent(evt Event) { + select { + case n.chEvent <- evt: + case <-n.terminate: + } +} + +func (n *Node) newChannel(ch *Channel) { + select { + case n.chNewChannel <- ch: + case <-n.terminate: + ch.close() + } +} + +func (n *Node) closeChannel(ch *Channel) { + select { + case n.chCloseChannel <- ch: case <-n.terminate: } } diff --git a/node_streamrequest.go b/node_streamrequest.go index 608397649..d5a49ea85 100644 --- a/node_streamrequest.go +++ b/node_streamrequest.go @@ -174,13 +174,10 @@ func (sr *nodeStreamRequest) onEventFrame(evt *EventFrame) { sr.n.WriteMessageTo(evt.Channel, m.Interface().(message.Message)) } - select { - case sr.n.events <- &EventStreamRequested{ + sr.n.pushEvent(&EventStreamRequested{ Channel: evt.Channel, SystemID: evt.SystemID(), ComponentID: evt.ComponentID(), - }: - case <-sr.n.terminate: - } + }) } } diff --git a/pkg/ringbuffer/event.go b/pkg/ringbuffer/event.go new file mode 100644 index 000000000..0b2bea5f1 --- /dev/null +++ b/pkg/ringbuffer/event.go @@ -0,0 +1,38 @@ +package ringbuffer + +import ( + "sync" +) + +type event struct { + mutex sync.Mutex + cond *sync.Cond + value bool +} + +func newEvent() *event { + cv := &event{} + cv.cond = sync.NewCond(&cv.mutex) + return cv +} + +func (cv *event) signal() { + func() { + cv.mutex.Lock() + defer cv.mutex.Unlock() + cv.value = true + }() + + cv.cond.Broadcast() +} + +func (cv *event) wait() { + cv.mutex.Lock() + defer cv.mutex.Unlock() + + if !cv.value { + cv.cond.Wait() + } + + cv.value = false +} diff --git a/pkg/ringbuffer/ringbuffer.go b/pkg/ringbuffer/ringbuffer.go new file mode 100644 index 000000000..8e9fb7ad7 --- /dev/null +++ b/pkg/ringbuffer/ringbuffer.go @@ -0,0 +1,77 @@ +// Package ringbuffer contains a ring buffer. +package ringbuffer + +import ( + "fmt" + "sync/atomic" + "unsafe" +) + +// RingBuffer is a ring buffer. +type RingBuffer struct { + size uint64 + readIndex uint64 + writeIndex uint64 + closed int64 + buffer []unsafe.Pointer + event *event +} + +// New allocates a RingBuffer. +func New(size uint64) (*RingBuffer, error) { + // when writeIndex overflows, if size is not a power of + // two, only a portion of the buffer is used. + if (size & (size - 1)) != 0 { + return nil, fmt.Errorf("size must be a power of two") + } + + return &RingBuffer{ + size: size, + readIndex: 1, + writeIndex: 0, + buffer: make([]unsafe.Pointer, size), + event: newEvent(), + }, nil +} + +// Close makes Pull() return false. +func (r *RingBuffer) Close() { + atomic.StoreInt64(&r.closed, 1) + r.event.signal() +} + +// Reset restores Pull() behavior after a Close(). +func (r *RingBuffer) Reset() { + for i := uint64(0); i < r.size; i++ { + atomic.SwapPointer(&r.buffer[i], nil) + } + atomic.SwapUint64(&r.writeIndex, 0) + r.readIndex = 1 + atomic.StoreInt64(&r.closed, 0) +} + +// Push pushes data at the end of the buffer. +func (r *RingBuffer) Push(data interface{}) { + writeIndex := atomic.AddUint64(&r.writeIndex, 1) + i := writeIndex % r.size + atomic.SwapPointer(&r.buffer[i], unsafe.Pointer(&data)) + r.event.signal() +} + +// Pull pulls data from the beginning of the buffer. +func (r *RingBuffer) Pull() (interface{}, bool) { + for { + i := r.readIndex % r.size + res := (*interface{})(atomic.SwapPointer(&r.buffer[i], nil)) + if res == nil { + if atomic.SwapInt64(&r.closed, 0) == 1 { + return nil, false + } + r.event.wait() + continue + } + + r.readIndex++ + return *res, true + } +} diff --git a/pkg/ringbuffer/ringbuffer_test.go b/pkg/ringbuffer/ringbuffer_test.go new file mode 100644 index 000000000..25a9635a9 --- /dev/null +++ b/pkg/ringbuffer/ringbuffer_test.go @@ -0,0 +1,147 @@ +package ringbuffer + +import ( + "bytes" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestCreateError(t *testing.T) { + _, err := New(1000) + require.EqualError(t, err, "size must be a power of two") +} + +func TestPushBeforePull(t *testing.T) { + r, err := New(1024) + require.NoError(t, err) + defer r.Close() + + data := bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 1024/4) + + r.Push(data) + ret, ok := r.Pull() + require.Equal(t, true, ok) + require.Equal(t, data, ret) +} + +func TestPullBeforePush(t *testing.T) { + r, err := New(1024) + require.NoError(t, err) + defer r.Close() + + data := bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 1024/4) + + done := make(chan struct{}) + go func() { + defer close(done) + ret, ok := r.Pull() + require.Equal(t, true, ok) + require.Equal(t, data, ret) + }() + + time.Sleep(100 * time.Millisecond) + + r.Push(data) + <-done +} + +func TestClose(t *testing.T) { + r, err := New(1024) + require.NoError(t, err) + + done := make(chan struct{}) + go func() { + defer close(done) + + _, ok := r.Pull() + require.Equal(t, true, ok) + + _, ok = r.Pull() + require.Equal(t, false, ok) + }() + + r.Push([]byte{0x01, 0x02, 0x03, 0x04}) + + r.Close() + <-done + + r.Reset() + + r.Push([]byte{0x05, 0x06, 0x07, 0x08}) + + _, ok := r.Pull() + require.Equal(t, true, ok) +} + +func BenchmarkPushPullContinuous(b *testing.B) { + r, _ := New(1024 * 8) + defer r.Close() + + data := make([]byte, 1024) + + for n := 0; n < b.N; n++ { + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 1024*8; i++ { + r.Push(data) + } + }() + + for i := 0; i < 1024*8; i++ { + r.Pull() + } + + <-done + } +} + +func BenchmarkPushPullPaused5(b *testing.B) { + r, _ := New(128) + defer r.Close() + + data := make([]byte, 1024) + + for n := 0; n < b.N; n++ { + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 128; i++ { + r.Push(data) + time.Sleep(5 * time.Millisecond) + } + }() + + for i := 0; i < 128; i++ { + r.Pull() + } + + <-done + } +} + +func BenchmarkPushPullPaused10(b *testing.B) { + r, _ := New(1024 * 8) + defer r.Close() + + data := make([]byte, 1024) + + for n := 0; n < b.N; n++ { + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 128; i++ { + r.Push(data) + time.Sleep(10 * time.Millisecond) + } + }() + + for i := 0; i < 128; i++ { + r.Pull() + } + + <-done + } +}