Skip to content
This repository has been archived by the owner on Oct 17, 2023. It is now read-only.

Commit

Permalink
fix growing allocation of pendingOffsets when no offsetManager is in …
Browse files Browse the repository at this point in the history
…use (#436)

* fix growing allocation of pendingOffsets when no offsetManager is in use

* dont close writeResult channel, avoids potential panic if used as a lib
  • Loading branch information
jipperinbham authored Nov 13, 2017
1 parent 612eda1 commit 61a4566
Show file tree
Hide file tree
Showing 2 changed files with 30 additions and 3 deletions.
8 changes: 5 additions & 3 deletions pipeline/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -544,9 +544,11 @@ func (n *Node) write(msg message.Msg, off offset.Offset) (message.Msg, error) {
return nil, err
}

n.offsetLock.Lock()
n.pendingOffsets = append(n.pendingOffsets, off)
n.offsetLock.Unlock()
if n.om != nil {
n.offsetLock.Lock()
n.pendingOffsets = append(n.pendingOffsets, off)
n.offsetLock.Unlock()
}
ctx, cancel := context.WithTimeout(context.Background(), n.writeTimeout)
defer cancel()
c := make(chan writeResult)
Expand Down
25 changes: 25 additions & 0 deletions pipeline/node_bench_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
package pipeline

import (
"testing"

"github.com/compose/transporter/offset"

"github.com/compose/transporter/message"
)

func BenchmarkNodeWrite(b *testing.B) {
node, err := NewNodeWithOptions("benchwriter", "test", ".*",
WithWriteTimeout("500ms"),
)
if err != nil {
b.Error(err)
}

msg := &message.Base{}
off := offset.Offset{}
b.ReportAllocs()
for i := 0; i < b.N; i++ {
node.write(msg, off)
}
}

0 comments on commit 61a4566

Please sign in to comment.