@@ -18,7 +18,8 @@ import (
1818 "time"
1919)
2020
21- func benchmarkRoutine (radixClient Client , ruedisClient rueidis.Client , useRuedis , useCSC , enableMultiExec bool , datapointsChan chan datapoint , continueOnError bool , cmdS [][]string , commandsCDF []float32 , keyspacelen , datasize , number_samples uint64 , loop bool , debug_level int , wg * sync.WaitGroup , keyplace , dataplace []int , readOnly []bool , useLimiter bool , rateLimiter * rate.Limiter , waitReplicas , waitReplicasMs int , cscDuration time.Duration ) {
21+ func benchmarkRoutine (radixClient Client , ruedisClient rueidis.Client , useRuedis , useCSC , enableMultiExec bool , datapointsChan chan datapoint , continueOnError bool , cmdS [][]string , commandsCDF []float32 , keyspacelen , datasize , number_samples uint64 , loop bool , debug_level int , wg * sync.WaitGroup , keyplace , dataplace []int , readOnly []bool , useLimiter bool , rateLimiter * rate.Limiter , waitReplicas , waitReplicasMs int , cacheOptions * rueidis.CacheOptions ) {
22+
2223 defer wg .Done ()
2324 for i := 0 ; uint64 (i ) < number_samples || loop ; i ++ {
2425 cmdPos := sample (commandsCDF )
@@ -32,15 +33,15 @@ func benchmarkRoutine(radixClient Client, ruedisClient rueidis.Client, useRuedis
3233 time .Sleep (r .Delay ())
3334 }
3435 if useRuedis {
35- sendCmdLogicRuedis (ruedisClient , newCmdS , enableMultiExec , datapointsChan , continueOnError , debug_level , useCSC , isReadOnly , cscDuration , waitReplicas , waitReplicasMs )
36+ sendCmdLogicRuedis (ruedisClient , newCmdS , enableMultiExec , datapointsChan , continueOnError , debug_level , useCSC , isReadOnly , cacheOptions , waitReplicas , waitReplicasMs )
3637 } else {
3738 sendCmdLogicRadix (radixClient , newCmdS , enableMultiExec , key , datapointsChan , continueOnError , debug_level , waitReplicas , waitReplicasMs )
3839
3940 }
4041 }
4142}
4243
43- func sendCmdLogicRuedis (ruedisClient rueidis.Client , newCmdS []string , enableMultiExec bool , datapointsChan chan datapoint , continueOnError bool , debug_level int , useCSC , isReadOnly bool , cscDuration time. Duration , waitReplicas , waitReplicasMs int ) {
44+ func sendCmdLogicRuedis (ruedisClient rueidis.Client , newCmdS []string , enableMultiExec bool , datapointsChan chan datapoint , continueOnError bool , debug_level int , useCSC , isReadOnly bool , cacheOptions * rueidis. CacheOptions , waitReplicas , waitReplicasMs int ) {
4445 ctx := context .Background ()
4546 var startT time.Time
4647 var endT time.Time
@@ -55,9 +56,15 @@ func sendCmdLogicRuedis(ruedisClient rueidis.Client, newCmdS []string, enableMul
5556 }
5657 }
5758 if useCSC && isReadOnly {
58- startT = time .Now ()
59- redisResult = ruedisClient .DoCache (ctx , arbitrary .Cache (), cscDuration )
60- endT = time .Now ()
59+ if cacheOptions .UseMultiExec && cacheOptions .UseServerPTTL {
60+ startT = time .Now ()
61+ redisResult = ruedisClient .DoCache (ctx , arbitrary .Cache (), cacheOptions .ClientTTL )
62+ endT = time .Now ()
63+ } else {
64+ startT = time .Now ()
65+ redisResult = ruedisClient .DoCacheWithOptions (ctx , arbitrary .Cache (), * cacheOptions )
66+ endT = time .Now ()
67+ }
6168 } else if enableMultiExec {
6269 cmds := make (rueidis.Commands , 0 , 3 )
6370 cmds = append (cmds , ruedisClient .B ().Multi ().Build ())
@@ -190,8 +197,13 @@ func main() {
190197 betweenClientsDelay := flag .Duration ("between-clients-duration" , time .Millisecond * 0 , "Between each client creation, wait this time." )
191198 version := flag .Bool ("v" , false , "Output version and exit" )
192199 verbose := flag .Bool ("verbose" , false , "Output verbose info" )
193- cscEnabled := flag .Bool ("csc" , false , "Enable client side caching" )
194200 useRuedis := flag .Bool ("rueidis" , false , "Use rueidis as the vanilla underlying client." )
201+ cscEnabled := flag .Bool ("csc" , false , "Enable client side caching" )
202+ // TODO: add this feature to check locking overhead
203+ // cscDisableTrackInvalidations := flag.Bool("csc-disable-track-invalidations", false, "Disable CSC tracking")
204+ cscUseMultiExec := flag .Bool ("csc-use-multi-exec" , false , "Use CSC wrapped in MULTI/EXEC" )
205+ // TODO: add this feature
206+ // cscUseServerPTTL := flag.Bool("csc-use-server-pttl", false, "Use CSC wrapped in with PTTL expiration info")
195207 cscDuration := flag .Duration ("csc-ttl" , time .Minute , "Client side cache ttl for cached entries" )
196208 clientKeepAlive := flag .Duration ("client-keepalive" , time .Minute , "Client keepalive" )
197209 cscSizeBytes := flag .Int ("csc-per-client-bytes" , rueidis .DefaultCacheBytes , "client side cache size that bind to each TCP connection to a single redis instance" )
@@ -335,27 +347,30 @@ func main() {
335347 AlwaysRESP2 : alwaysRESP2 ,
336348 DisableCache : ! * cscEnabled ,
337349 BlockingPoolSize : 0 ,
338- PipelineMultiplex : 0 ,
350+ PipelineMultiplex : - 1 ,
339351 RingScaleEachConn : 1 ,
340352 ReadBufferEachConn : 1024 ,
341353 WriteBufferEachConn : 1024 ,
342354 CacheSizeEachConn : * cscSizeBytes ,
343355 OnInvalidations : invalidationFunction ,
356+ ForceSingleClient : ! * clusterMode ,
344357 }
345358 clientOptions .Dialer .KeepAlive = * clientKeepAlive
346359 ruedisClient , err = rueidis .NewClient (clientOptions )
360+ cacheOptions := rueidis.CacheOptions {UseMultiExec : * cscUseMultiExec , UseServerPTTL : * cscUseMultiExec , ClientTTL : * cscDuration }
361+
347362 if err != nil {
348363 panic (err )
349364 }
350- go benchmarkRoutine (radixStandalone , ruedisClient , * useRuedis , * cscEnabled , * multi , datapointsChan , * continueonerror , cmds , cdf , * keyspacelen , * datasize , samplesPerClient , * loop , int (* debug ), & wg , cmdKeyplaceHolderPos , cmdDataplaceHolderPos , cmdReadOnly , useRateLimiter , rateLimiter , * waitReplicas , * waitReplicasMs , * cscDuration )
365+ go benchmarkRoutine (radixStandalone , ruedisClient , * useRuedis , * cscEnabled , * multi , datapointsChan , * continueonerror , cmds , cdf , * keyspacelen , * datasize , samplesPerClient , * loop , int (* debug ), & wg , cmdKeyplaceHolderPos , cmdDataplaceHolderPos , cmdReadOnly , useRateLimiter , rateLimiter , * waitReplicas , * waitReplicasMs , & cacheOptions )
351366 } else {
352367 // legacy radix code
353368 if * clusterMode {
354369 cluster = getOSSClusterConn (connectionStr , opts , 1 )
355- go benchmarkRoutine (cluster , ruedisClient , * useRuedis , * cscEnabled , * multi , datapointsChan , * continueonerror , cmds , cdf , * keyspacelen , * datasize , samplesPerClient , * loop , int (* debug ), & wg , cmdKeyplaceHolderPos , cmdDataplaceHolderPos , cmdReadOnly , useRateLimiter , rateLimiter , * waitReplicas , * waitReplicasMs , * cscDuration )
370+ go benchmarkRoutine (cluster , ruedisClient , * useRuedis , * cscEnabled , * multi , datapointsChan , * continueonerror , cmds , cdf , * keyspacelen , * datasize , samplesPerClient , * loop , int (* debug ), & wg , cmdKeyplaceHolderPos , cmdDataplaceHolderPos , cmdReadOnly , useRateLimiter , rateLimiter , * waitReplicas , * waitReplicasMs , nil )
356371 } else {
357372 radixStandalone = getStandaloneConn (connectionStr , opts , 1 )
358- go benchmarkRoutine (radixStandalone , ruedisClient , * useRuedis , * cscEnabled , * multi , datapointsChan , * continueonerror , cmds , cdf , * keyspacelen , * datasize , samplesPerClient , * loop , int (* debug ), & wg , cmdKeyplaceHolderPos , cmdDataplaceHolderPos , cmdReadOnly , useRateLimiter , rateLimiter , * waitReplicas , * waitReplicasMs , * cscDuration )
373+ go benchmarkRoutine (radixStandalone , ruedisClient , * useRuedis , * cscEnabled , * multi , datapointsChan , * continueonerror , cmds , cdf , * keyspacelen , * datasize , samplesPerClient , * loop , int (* debug ), & wg , cmdKeyplaceHolderPos , cmdDataplaceHolderPos , cmdReadOnly , useRateLimiter , rateLimiter , * waitReplicas , * waitReplicasMs , nil )
359374 }
360375 }
361376
0 commit comments