hanzoai/psrpc
2
0
Fork 0
mirror of https://github.com/hanzoai/psrpc.git synced 2026-09-30 23:47:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use redis pipeline to deal with longer latency connections to Redis. (#111)

Browse Source 
* Use redis pipeline to deal with longer latency connections to Redis.

- Using hashed publish queue (probably not super helpful as RelaySignal
  probably overwhelms others)
- Use `Pipeline`. Can potentially use `TxPipeline`, but that block other
  pipeline clients while it is executing. In tests though (trying to
  start 200 rooms with 200ms latency to redis server), the performance was
  comparable, i. e. all the rooms were created, but participant join
  needed some retries in some runs. In some other runs, participant join
  also worked without retries for that setting.
- Execute pipeline immediately (on a publish) and let the RTT processing
  opportunistically build up commands for the next pipeline execute.
  Maybe, it is possible to run each execute in a separate goroutine to
  handle cases of channels like RelaySignal running multiple in parallel,
  but ordering could cause subtle issues.

* clean up

* tidy

* add slack notifier

* comment pipeline usage
This commit is contained in:
Raja Subramanian
2025-11-05 22:28:26 +05:30
committed by GitHub
parent de9e56f4c5
commit 0d6eb61a92
4 changed files with 130 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/slack-notifier.yaml
+25
View File
@@ -0,0 +1,25 @@
name: PR Slack Notifier
on:
pull_request:
types: [review_requested, reopened, closed]
pull_request_review:
types: [submitted]
permissions:
contents: read
pull-requests: write
issues: write
concurrency:
group: pr-slack-${{ github.event.pull_request.number }}-${{ github.workflow }}
cancel-in-progress: false
jobs:
notify-devs:
runs-on: ubuntu-latest
steps:
- uses: livekit/slack-notifier-action@main
with:
config_json: ${{ secrets.SLACK_NOTIFY_CONFIG_JSON }}
slack_token: ${{ secrets.SLACK_PR_NOTIFIER_TOKEN }}
go.mod
+2
View File
@@ -15,6 +15,7 @@ require (
github.com/redis/go-redis/v9 v9.3.0
github.com/stretchr/testify v1.10.0
github.com/twitchtv/twirp v8.1.3+incompatible
github.com/zeebo/xxh3 v1.0.2
go.opentelemetry.io/otel v1.37.0
go.opentelemetry.io/otel/trace v1.37.0
go.uber.org/atomic v1.11.0
@@ -44,6 +45,7 @@ require (
github.com/gogo/protobuf v1.3.2 // indirect
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
github.com/klauspost/compress v1.17.2 // indirect
github.com/klauspost/cpuid/v2 v2.0.9 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
github.com/moby/docker-image-spec v1.3.1 // indirect
github.com/moby/term v0.5.0 // indirect
go.sum
+6
View File
@@ -56,6 +56,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -110,6 +112,10 @@ github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17
github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
internal/bus/bus_redis.go
+97 -49
View File
@@ -24,6 +24,7 @@ import (
"github.com/gammazero/deque"
"github.com/redis/go-redis/v9"
"github.com/zeebo/xxh3"
"go.uber.org/multierr"
"golang.org/x/exp/maps"
"golang.org/x/exp/slices"
@@ -32,10 +33,13 @@ import (
"github.com/livekit/psrpc/internal/logger"
)
const lockExpiration = time.Second * 5
const reconcilerRetryInterval = time.Second
const minReadRetryInterval = time.Millisecond * 100
const maxReadRetryInterval = time.Second
const (
lockExpiration = time.Second * 5
reconcilerRetryInterval = time.Second
minReadRetryInterval = time.Millisecond * 100
maxReadRetryInterval = time.Second
publishBuckets = 17
)
type redisMessageBus struct {
rc redis.UniversalClient
@@ -48,9 +52,10 @@ type redisMessageBus struct {
wakeup chan struct{}
ops *redisWriteOpQueue
publishOps map[string]*redisWriteOpQueue
dirtyChannels map[string]struct{}
currentChannels map[string]struct{}
publishQueues [publishBuckets]*redisPublishQueue
}
func NewRedisMessageBus(rc redis.UniversalClient) MessageBus {
@@ -64,10 +69,12 @@ func NewRedisMessageBus(rc redis.UniversalClient) MessageBus {
wakeup: make(chan struct{}, 1),
ops: &redisWriteOpQueue{},
publishOps: map[string]*redisWriteOpQueue{},
dirtyChannels: map[string]struct{}{},
currentChannels: map[string]struct{}{},
}
for i := range len(r.publishQueues) {
r.publishQueues[i] = newRedisPublishQueue(r.ctx, r.rc)
}
go r.readWorker()
go r.writeWorker()
return r
@@ -79,18 +86,8 @@ func (r *redisMessageBus) Publish(_ context.Context, channel Channel, msg proto.
return err
}
r.mu.Lock()
ops, ok := r.publishOps[channel.Legacy]
if !ok {
ops = &redisWriteOpQueue{}
r.publishOps[channel.Legacy] = ops
}
ops.push(&redisPublishOp{r, channel.Legacy, b})
r.mu.Unlock()
if !ok {
r.enqueueWriteOp(&redisExecPublishOp{r, channel.Legacy, ops})
}
bucket := xxh3.HashString(channel.Legacy) % publishBuckets
r.publishQueues[bucket].Enqueue(channel.Legacy, b)
return nil
}
@@ -202,6 +199,8 @@ func (r *redisMessageBus) writeWorker() {
}
}
// ----------------------------------------------
type redisWriteOpQueue struct {
mu sync.Mutex
ops deque.Deque[redisWriteOp]
@@ -232,41 +231,13 @@ func (q *redisWriteOpQueue) drain() {
q.mu.Unlock()
}
//-----------------------------------------------------
type redisWriteOp interface {
run() error
}
type redisPublishOp struct {
*redisMessageBus
channel string
message []byte
}
func (r *redisPublishOp) run() error {
return r.rc.Publish(r.ctx, r.channel, r.message).Err()
}
type redisExecPublishOp struct {
*redisMessageBus
channel string
ops *redisWriteOpQueue
}
func (r *redisExecPublishOp) run() error {
go r.exec()
return nil
}
func (r *redisExecPublishOp) exec() {
r.mu.Lock()
for !r.ops.empty() {
r.mu.Unlock()
r.ops.drain()
r.mu.Lock()
}
delete(r.publishOps, r.channel)
r.mu.Unlock()
}
// ----------------------------------------------------
type redisReconcileSubscriptionsOp struct {
*redisMessageBus
@@ -320,6 +291,8 @@ func (r *redisReconcileSubscriptionsOp) run() error {
return nil
}
// ----------------------------------------------------
type redisSubList struct {
subs []*redisSubscription
next int
@@ -339,6 +312,8 @@ func (r *redisSubList) dispatch(msg *redis.Message) {
}
}
// ----------------------------------------------------
type redisSubscription struct {
bus *redisMessageBus
ctx context.Context
@@ -387,3 +362,76 @@ func (r *redisSubscription) Close() error {
close(r.msgChan)
return nil
}
// ----------------------------------------------------
type redisPublishMessage struct {
channel string
payload []byte
}
type redisPublishQueue struct {
ctx context.Context
rc redis.UniversalClient
lock sync.Mutex
messages []redisPublishMessage
wakeup chan struct{}
}
func newRedisPublishQueue(ctx context.Context, rc redis.UniversalClient) *redisPublishQueue {
r := &redisPublishQueue{
ctx: ctx,
rc: rc,
wakeup: make(chan struct{}, 1),
}
go r.worker()
return r
}
func (r *redisPublishQueue) Enqueue(channel string, payload []byte) {
r.lock.Lock()
defer r.lock.Unlock()
r.messages = append(r.messages, redisPublishMessage{channel, payload})
select {
case r.wakeup <- struct{}{}:
default:
}
}
func (r *redisPublishQueue) worker() {
for {
select {
case <-r.wakeup:
case <-r.ctx.Done():
return
}
r.lock.Lock()
messages := r.messages
r.messages = nil
r.lock.Unlock()
// using a pipeline to handle redis servers with a high RTT
// (https://redis.io/docs/latest/develop/using-commands/pipelining/).
//
// This is doing oppotunistic batching + pipelining.
// When a message is published, this worker is signalled immediately and
// the message will be sent immediately. While the pipeline is executing,
// messages will get queued as the pipeline execution takes one round trip
// exchange with the server. The next round will batch all those queued
// messages. For small RTTs, this will send messages without any delay.
pipeline := r.rc.Pipeline()
for _, msg := range messages {
pipeline.Publish(r.ctx, msg.channel, msg.payload)
}
cmds, err := pipeline.Exec(r.ctx)
if err != nil {
logger.Error(err, "pipeline execution failed", "numCommands", len(cmds))
}
}
}
// ----------------------------------------------------