11package main
22
33import (
4+ "context"
45 "flag"
56 "fmt"
67 hdrhistogram "github.com/HdrHistogram/hdrhistogram-go"
7- "github.com/mediocregopher/radix/v3 "
8+ "github.com/mediocregopher/radix/v4 "
89 "golang.org/x/time/rate"
910 "log"
1011 "math"
@@ -37,15 +38,15 @@ func stringWithCharset(length int, charset string) string {
3738 return string (b )
3839}
3940
40- func ingestionRoutine (conn radix. Client , enableMultiExec bool , datapointsChan chan datapoint , continueOnError bool , cmdS []string , keyspacelen , datasize , number_samples uint64 , loop bool , debug_level int , wg * sync.WaitGroup , keyplace , dataplace int , useLimiter bool , rateLimiter * rate.Limiter , waitReplicas , waitReplicasMs int ) {
41+ func ingestionRoutine (conn Client , enableMultiExec bool , datapointsChan chan datapoint , continueOnError bool , cmdS []string , keyspacelen , datasize , number_samples uint64 , loop bool , debug_level int , wg * sync.WaitGroup , keyplace , dataplace int , useLimiter bool , rateLimiter * rate.Limiter , waitReplicas , waitReplicasMs int ) {
4142 defer wg .Done ()
4243 for i := 0 ; uint64 (i ) < number_samples || loop ; i ++ {
4344 rawCurrentCmd , key , _ := keyBuildLogic (keyplace , dataplace , datasize , keyspacelen , cmdS )
4445 sendCmdLogic (conn , rawCurrentCmd , enableMultiExec , key , datapointsChan , continueOnError , debug_level , useLimiter , rateLimiter , waitReplicas , waitReplicasMs )
4546 }
4647}
4748
48- func keyBuildLogic (keyPos int , dataPos int , datasize , keyspacelen uint64 , cmdS []string ) (cmd radix.CmdAction , key string , keySlot uint16 ) {
49+ func keyBuildLogic (keyPos int , dataPos int , datasize , keyspacelen uint64 , cmdS []string ) (cmd radix.Action , key string , keySlot uint16 ) {
4950 newCmdS := cmdS
5051 if keyPos > - 1 {
5152 newCmdS [keyPos ] = fmt .Sprintf ("%d" , rand .Int63n (int64 (keyspacelen )))
@@ -54,22 +55,23 @@ func keyBuildLogic(keyPos int, dataPos int, datasize, keyspacelen uint64, cmdS [
5455 newCmdS [dataPos ] = stringWithCharset (int (datasize ), charset )
5556 }
5657 rawCmd := radix .Cmd (nil , newCmdS [0 ], newCmdS [1 :]... )
57-
5858 return rawCmd , key , radix .ClusterSlot ([]byte (newCmdS [1 ]))
5959}
6060
61- func sendCmdLogic (conn radix.Client , cmd radix.CmdAction , enableMultiExec bool , key string , datapointsChan chan datapoint , continueOnError bool , debug_level int , useRateLimiter bool , rateLimiter * rate.Limiter , waitReplicas , waitReplicasMs int ) {
61+ func sendCmdLogic (conn Client , cmd radix.Action , enableMultiExec bool , key string , datapointsChan chan datapoint , continueOnError bool , debug_level int , useRateLimiter bool , rateLimiter * rate.Limiter , waitReplicas , waitReplicasMs int ) {
62+ ctx := context .Background ()
63+
6264 if useRateLimiter {
6365 r := rateLimiter .ReserveN (time .Now (), int (1 ))
6466 time .Sleep (r .Delay ())
6567 }
6668 var err error
6769 startT := time .Now ()
6870 if enableMultiExec {
69- err = conn .Do (radix .WithConn (key , func (c radix.Conn ) error {
71+ err = conn .Do (ctx , radix .WithConn (key , func (ctx context. Context , c radix.Conn ) error {
7072
7173 // Begin the transaction with a MULTI command
72- if err := conn .Do (radix .Cmd (nil , "MULTI" )); err != nil {
74+ if err := conn .Do (ctx , radix .Cmd (nil , "MULTI" )); err != nil {
7375 log .Fatalf ("Received an error while preparing for MULTI: %v, error: %v" , cmd , err )
7476 }
7577
@@ -83,25 +85,28 @@ func sendCmdLogic(conn radix.Client, cmd radix.CmdAction, enableMultiExec bool,
8385 // The return from DISCARD doesn't matter. If it's an error then
8486 // it's a network error and the Conn will be closed by the
8587 // client.
86- conn .Do (radix .Cmd (nil , "DISCARD" ))
88+ conn .Do (ctx , radix .Cmd (nil , "DISCARD" ))
8789 log .Fatalf ("Received an error while in multi: %v, error: %v" , cmd , err )
8890 }
8991 }()
9092
9193 // queue up the transaction's commands
92- err = conn .Do (cmd )
94+ err = conn .Do (ctx , cmd )
9395
9496 // execute the transaction, capturing the result in a Tuple. We only
9597 // care about the first element (the result from GET), so we discard the
9698 // second by setting nil.
97- return conn .Do (radix .Cmd (nil , "EXEC" ))
99+ return conn .Do (ctx , radix .Cmd (nil , "EXEC" ))
98100 }))
99101 } else if waitReplicas > 0 {
100- // pipeline the command + wait
101- err = conn .Do (radix .Pipeline (cmd ,
102- radix .Cmd (nil , "WAIT" , fmt .Sprintf ("%d" , waitReplicas ), fmt .Sprintf ("%d" , waitReplicasMs ))))
102+ // Create a new pipeline for the WAIT command
103+ p := radix .NewPipeline ()
104+ // Pass both cmd and waitCmd to the original pipeline
105+ p .Append (cmd )
106+ p .Append (radix .Cmd (nil , "WAIT" , fmt .Sprintf ("%d" , waitReplicas ), fmt .Sprintf ("%d" , waitReplicasMs )))
107+ err = conn .Do (ctx , p )
103108 } else {
104- err = conn .Do (cmd )
109+ err = conn .Do (ctx , cmd )
105110 }
106111 endT := time .Now ()
107112 if err != nil {
@@ -134,6 +139,7 @@ func main() {
134139 clusterMode := flag .Bool ("oss-cluster" , false , "Enable OSS cluster mode." )
135140 loop := flag .Bool ("l" , false , "Loop. Run the tests forever." )
136141 version := flag .Bool ("v" , false , "Output version and exit" )
142+ resp := flag .Int ("resp" , 2 , "redis command response protocol (2 - RESP 2, 3 - RESP 3)" )
137143 flag .Parse ()
138144 git_sha := toolGitSHA1 ()
139145 git_dirty_str := ""
@@ -173,9 +179,14 @@ func main() {
173179 samplesPerClient := * numberRequests / * clients
174180 client_update_tick := 1
175181 latencies = hdrhistogram .New (1 , 90000000 , 3 )
176- opts := make ([] radix.DialOpt , 0 )
182+ opts := radix.Dialer {}
177183 if * password != "" {
178- opts = append (opts , radix .DialAuthPass (* password ))
184+ opts .AuthPass = * password
185+ }
186+ if * resp == 2 {
187+ opts .Protocol = "2"
188+ } else if * resp == 3 {
189+ opts .Protocol = "3"
179190 }
180191 ips , _ := net .LookupIP (* host )
181192 fmt .Printf ("IPs %v\n " , ips )
@@ -191,7 +202,6 @@ func main() {
191202 fmt .Printf ("Using random seed: %d\n " , * seed )
192203 rand .Seed (* seed )
193204 var cluster * radix.Cluster
194-
195205 datapointsChan := make (chan datapoint , * numberRequests )
196206 for channel_id := 1 ; uint64 (channel_id ) <= * clients ; channel_id ++ {
197207 wg .Add (1 )
0 commit comments