Skip to content
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.

Sign up for GitHub

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

Jump to bottom

Use critical section for batch counters in producer default_router #684

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Use critical section for batch counters in producer default_router #684

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 22 additions & 19 deletions pulsar/default_router.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package pulsar
import (
"math"
"math/rand"
"sync"
"sync/atomic"
"time"
)
Expand All @@ -30,6 +31,7 @@ type defaultRouter struct {
lastChangeTimestamp int64
msgCounter uint32
cumulativeBatchSize uint32
sync.RWMutex
Copy link
Contributor

@dferstay dferstay Dec 26, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see that we only ever acquire write access; can this be a regular sync.Mutex?

Copy link
Contributor

@dferstay dferstay Dec 26, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, I'm concerned that the use of a mutex will incur overhead and affect performance, especially under concurrent publishing. I've added a bench test to get quantitative numbers in #693

Copy link
Contributor

@dferstay dferstay Dec 27, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@zzzming , I've attempted to address the races when updating the default router state in a lock-free manner here: Please take a look and see if you are satisfied: #694

Comparing the results from the parallel default router bench test we can see that the use of a mutex has an impact on performance (old==this PR, new==#694):

name                      old time/op  new time/op  delta
DefaultRouterParallel     27.4ns ± 3%  14.8ns ± 2%  -45.79%  (p=0.000 n=10+8)
DefaultRouterParallel-2   35.2ns ± 1%  41.9ns ± 0%  +18.96%  (p=0.000 n=9+7)
DefaultRouterParallel-4   49.4ns ± 6%  44.1ns ± 8%  -10.84%  (p=0.000 n=10+9)
DefaultRouterParallel-8   58.8ns ± 6%  53.2ns ± 3%   -9.53%  (p=0.000 n=10+8)
DefaultRouterParallel-16  69.7ns ± 3%  51.3ns ± 0%  -26.43%  (p=0.000 n=10+8)

EDIT: I can't explain why DefaultRouterParallel-2 is slower on the branch that doesn't use the mutex.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Update: the implementation in #694 was modified slightly and now DefaultRouterParallel-2 is no longer slower in the lock-free implementation. New bench comparison is below:

name                      old time/op  new time/op  delta
DefaultRouterParallel     27.4ns ± 3%  14.9ns ± 6%  -45.51%  (p=0.000 n=10+9)
DefaultRouterParallel-2   35.2ns ± 1%  33.7ns ± 8%     ~     (p=0.161 n=9+9)
DefaultRouterParallel-4   49.4ns ± 6%  28.8ns ± 4%  -41.66%  (p=0.000 n=10+9)
DefaultRouterParallel-8   58.8ns ± 6%  36.3ns ± 1%  -38.32%  (p=0.000 n=10+8)
DefaultRouterParallel-16  69.7ns ± 3%  39.5ns ±21%  -43.38%  (p=0.000 n=10+10)

}

// NewDefaultRouter set the message routing mode for the partitioned producer.
Expand Down Expand Up @@ -75,36 +77,37 @@ func NewDefaultRouter(
// If there's no key, we do round-robin across partition, sticking with a given
// partition for a certain amount of messages or volume buffered or the max delay to batch is reached so that
// we ensure having a decent amount of batching of the messages.
// Note that it is possible that we skip more than one partition if multiple goroutines increment
// currentPartitionCursor at the same time. If that happens it shouldn't be a problem because we only want to
// spread the data on different partitions but not necessarily in a specific sequence.
// Use critical section to protect a group of counters and increment the currentPartitionCursor
var now int64
batchReached := false
size := uint32(len(message.Payload))
previousMessageCount := atomic.LoadUint32(&state.msgCounter)
previousBatchingMaxSize := atomic.LoadUint32(&state.cumulativeBatchSize)
previousLastChange := atomic.LoadInt64(&state.lastChangeTimestamp)
state.Lock()
defer state.Unlock()

messageCountReached := previousMessageCount >= uint32(maxBatchingMessages-1)
sizeReached := (size >= uint32(maxBatchingSize)-previousBatchingMaxSize)
durationReached := false
if readClockAfterNumMessages == 0 || previousMessageCount%readClockAfterNumMessages == 0 {
if state.msgCounter >= uint32(maxBatchingMessages-1) {
batchReached = true
} else if size >= (uint32(maxBatchingSize) - state.cumulativeBatchSize) {
batchReached = true
} else if readClockAfterNumMessages == 0 || state.msgCounter%readClockAfterNumMessages == 0 {
now = time.Now().UnixNano()
durationReached = now-previousLastChange >= maxBatchingDelay.Nanoseconds()
batchReached = now-state.lastChangeTimestamp >= maxBatchingDelay.Nanoseconds()
}
if messageCountReached || sizeReached || durationReached {
atomic.AddUint32(&state.currentPartitionCursor, 1)
atomic.StoreUint32(&state.msgCounter, 0)
atomic.StoreUint32(&state.cumulativeBatchSize, 0)
if batchReached {
// only the current partition cursor when the current batch is ready to flush
// so that we move to another partition for the next batch
state.currentPartitionCursor++
state.msgCounter = 0
state.cumulativeBatchSize = 0
if now != 0 {
atomic.StoreInt64(&state.lastChangeTimestamp, now)
state.lastChangeTimestamp = now
}
return int(state.currentPartitionCursor % numPartitions)
}

atomic.AddUint32(&state.msgCounter, 1)
atomic.AddUint32(&state.cumulativeBatchSize, size)
state.msgCounter++
state.cumulativeBatchSize += size
if now != 0 {
atomic.StoreInt64(&state.lastChangeTimestamp, now)
state.lastChangeTimestamp = now
}
return int(state.currentPartitionCursor % numPartitions)
}
Expand Down