Skip to content

Change from radix/v4 to go-redis due to high lock overhead on EncodeDecode. Current best performance vs all previous versions #14

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

Merged
merged 3 commits into from
May 24, 2023
Merged

Change from radix/v4 to go-redis due to high lock overhead on EncodeDecode. Current best performance vs all previous versions #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4d68790
Enabled pprof cpuprofile collection
filipecosta90 May 24, 2023
b36213c
Avoid context locking on pubsub due to PingInterval (disabled it)
filipecosta90 May 24, 2023
26a6e82
Using go-redis for SUBSCRIBE flow
filipecosta90 May 24, 2023
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
6 changes: 5 additions & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,4 +90,8 @@ Thumbs.db

# Coverage Results #
####################
coverage.txt
coverage.txt

# Profiler Results #
####################
*.pprof
64 changes: 24 additions & 40 deletions subscriber.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package main
import (
"context"
"encoding/json"
"errors"
"flag"
"fmt"
radix "github.com/mediocregopher/radix/v4"
Expand All @@ -12,6 +11,7 @@ import (
"log"
"os"
"os/signal"
"runtime/pprof"
"strings"
"sync"
"sync/atomic"
Expand Down Expand Up @@ -40,14 +40,14 @@ type testResult struct {
func subscriberRoutine(addr string, mode, subscriberName string, channel string, printMessages bool, ctx context.Context, wg *sync.WaitGroup, opts radix.Dialer, protocolVersion int) {
// tell the caller we've stopped
defer wg.Done()
client := redis.NewClient(&redis.Options{
Addr: addr,
Password: opts.AuthPass,
ClientName: subscriberName,
ProtocolVersion: protocolVersion,
})
switch mode {
case "ssubscribe":
client := redis.NewClient(&redis.Options{
Addr: addr,
Password: opts.AuthPass,
ClientName: subscriberName,
ProtocolVersion: protocolVersion,
})
spubsub := client.SSubscribe(ctx, channel)
defer spubsub.Close()
for {
Expand All @@ -64,53 +64,26 @@ func subscriberRoutine(addr string, mode, subscriberName string, channel string,
case "subscribe":
fallthrough
default:
_, _, ps, _ := bootstrapPubSub(addr, subscriberName, channel, opts)
defer ps.Close()
pubsub := client.Subscribe(ctx, channel)
defer pubsub.Close()
for {
msg, err := ps.Next(ctx)
if errors.Is(err, context.Canceled) {
break
} else if err != nil {
msg, err := pubsub.ReceiveMessage(ctx)
if err != nil {
panic(err)
}
if printMessages {
fmt.Println(fmt.Sprintf("received message in channel %s. Message: %s", msg.Channel, msg.Message))
fmt.Println(fmt.Sprintf("received message in channel %s. Message: %s", msg.Channel, msg.Payload))
}
atomic.AddUint64(&totalMessages, 1)
}

}

}

func bootstrapPubSub(addr string, subscriberName string, channel string, opts radix.Dialer) (radix.Conn, error, radix.PubSubConn, *time.Ticker) {
// Create a normal redis connection
ctx := context.Background()
conn, err := opts.Dial(ctx, "tcp", addr)

if err != nil {
log.Fatal(err)
}

err = conn.Do(ctx, radix.FlatCmd(nil, "CLIENT", "SETNAME", subscriberName))
if err != nil {
log.Fatal(err)
}

// Pass that connection into PubSub, conn should never get used after this
ps := radix.PubSubConfig{}.New(conn)

err = ps.Subscribe(ctx, channel)
if err != nil {
log.Fatal(err)
}

return conn, err, ps, nil
}

func main() {
host := flag.String("host", "127.0.0.1", "redis host.")
port := flag.String("port", "6379", "redis port.")
cpuprofile := flag.String("cpuprofile", "", "write cpu profile to file")
password := flag.String("a", "", "Password for Redis Auth.")
mode := flag.String("mode", "subscribe", "Subscribe mode. Either 'subscribe' or 'ssubscribe'.")
username := flag.String("user", "", "Used to send ACL style 'AUTH username pass'. Needs -a.")
Expand Down Expand Up @@ -186,6 +159,13 @@ func main() {
total_subscriptions := total_channels * *subscribers_per_channel
total_messages := int64(total_subscriptions) * *messages_per_channel_subscriber
fmt.Println(fmt.Sprintf("Total subcriptions: %d. Subscriptions per node %d. Total messages: %d", total_subscriptions, subscriptions_per_node, total_messages))
if *cpuprofile != "" {
f, err := os.Create(*cpuprofile)
if err != nil {
log.Fatal(err)
}
pprof.StartCPUProfile(f)
}

if strings.Compare(*subscribers_placement, "dense") == 0 {
for channel_id := *channel_minimum; channel_id <= *channel_maximum; channel_id++ {
Expand All @@ -210,6 +190,10 @@ func main() {
closed, start_time, duration, totalMessages, messageRateTs := updateCLI(tick, c, total_messages, w, *test_time)
messageRate := float64(totalMessages) / float64(duration.Seconds())

if *cpuprofile != "" {
pprof.StopCPUProfile()
}

fmt.Fprint(w, fmt.Sprintf("#################################################\nTotal Duration %f Seconds\nMessage Rate %f\n#################################################\n", duration.Seconds(), messageRate))
fmt.Fprint(w, "\r\n")
w.Flush()
Expand Down