Implement CLUSTER PUBSUB (#866)

* skip Interlocked.Exchange and FlushConfig when no update detected on merge

* special case of zero epoch necessitates tracking of explicit slot update

* move key spec extraction to clusterProvider

* implement CLUSTER PUBLISH api

* implement SSUBSCRIBE api

* implement SPUBLISH api

* implement separate code path for CLUSTER SPUBLISH

* expose max number of outstanding pubsub forwarding tasks parameter

* add API to forward published messages using GarnetClient

* add methods to extract nodeIds of all nodes in cluster and all nodes in a shard

* hook up published message forwarding

* PUBLISH and SPUBLISH benchmark addition

* fix check for getting most recent config instance

* optimization 1: calculate candidates for published message forwarding only when config gets updated

* optimization2: ignore response from cluster publish

* add publish BDN

* minimize task creation

* load no response task

* change to pinned array

* simplify config update for publish forward

* wip; wait for response vs fire and forget

* add protected connection initialization for gossip-publish

* remove BDN because not useful

* remove unused NoResponse flag

* remove option for tasks

* ensure noResponse for slot verification of cluster commands

* add SPUBLISH and PUBLISH docs

* correct typo

* addressing comments v1

* addressing comments v2

* rename new lock interface to indicate clear semantics

---------

Co-authored-by: Tal Zaccai <[email protected]>

benchmark/BDN.benchmark/Cluster/ClusterContext.cs

@@ -68,7 +68,7 @@ public void CreateGetSet(int keySize = 8, int valueSize = 32, int batchSize = 10
                 benchUtils.RandomBytes(ref pairs[i].Item2);
             }
 
-            var setByteCount = batchSize * ("*2\r\n$3\r\nSET\r\n"u8.Length + 1 + NumUtils.CountDigits(keySize) + 2 + keySize + 2 + 1 + NumUtils.CountDigits(valueSize) + 2 + valueSize + 2);
+            var setByteCount = batchSize * ("*3\r\n$3\r\nSET\r\n"u8.Length + 1 + NumUtils.CountDigits(keySize) + 2 + keySize + 2 + 1 + NumUtils.CountDigits(valueSize) + 2 + valueSize + 2);
             var setReq = new Request(setByteCount);
             var curr = setReq.ptr;
             var end = curr + setReq.buffer.Length;

benchmark/Resp.benchmark/OpType.cs

@@ -12,6 +12,7 @@ public enum OpType
         DBSIZE,
         READ_TXN, WRITE_TXN, READWRITETX, WATCH_TXN, SAMPLEUPDATETX, SAMPLEDELETETX,
         SCRIPTSET, SCRIPTGET, SCRIPTRETKEY,
+        PUBLISH, SPUBLISH,
         READONLY = 8888,
         AUTH = 9999,
     }

benchmark/Resp.benchmark/Program.cs

@@ -219,7 +219,7 @@ static void RunBasicCommandsBenchmark(Options opts)
             int keyLen = opts.KeyLength;
             int valueLen = opts.ValueLength;
 
-            if (opts.Op == OpType.ZADD || opts.Op == OpType.ZREM || opts.Op == OpType.ZADDREM || opts.Op == OpType.PING || opts.Op == OpType.GEOADD || opts.Op == OpType.GEOADDREM || opts.Op == OpType.SETEX || opts.Op == OpType.ZCARD || opts.Op == OpType.ZADDCARD)
+            if (opts.Op == OpType.PUBLISH || opts.Op == OpType.SPUBLISH || opts.Op == OpType.ZADD || opts.Op == OpType.ZREM || opts.Op == OpType.ZADDREM || opts.Op == OpType.PING || opts.Op == OpType.GEOADD || opts.Op == OpType.GEOADDREM || opts.Op == OpType.SETEX || opts.Op == OpType.ZCARD || opts.Op == OpType.ZADDCARD)
                 opts.SkipLoad = true;
 
             //if we have scripts ops we need to load them in memory

benchmark/Resp.benchmark/ReqGen.cs

@@ -204,6 +204,8 @@ public static (int, int) OnResponse(byte* buf, int bytesRead, int opType)
                 case OpType.DEL:
                 case OpType.INCR:
                 case OpType.DBSIZE:
+                case OpType.PUBLISH:
+                case OpType.SPUBLISH:
                     for (int i = 0; i < bytesRead; i++)
                         if (buf[i] == ':') count++;
                     break;

benchmark/Resp.benchmark/ReqGenLoadBuffers.cs

@@ -133,6 +133,8 @@ private bool GenerateBatch(int i, int start, int end, OpType opType)
                 OpType.SCRIPTSET => System.Text.Encoding.ASCII.GetBytes($"*5\r\n$7\r\nEVALSHA\r\n{BenchUtils.sha1SetScript}\r\n$1\r\n1\r\n"),
                 OpType.SCRIPTGET => System.Text.Encoding.ASCII.GetBytes($"*4\r\n$7\r\nEVALSHA\r\n{BenchUtils.sha1GetScript}\r\n$1\r\n1\r\n"),
                 OpType.SCRIPTRETKEY => System.Text.Encoding.ASCII.GetBytes($"*4\r\n$7\r\nEVALSHA\r\n{BenchUtils.sha1RetKeyScript}\r\n$1\r\n1\r\n"),
+                OpType.PUBLISH => System.Text.Encoding.ASCII.GetBytes($"*3\r\n$7\r\nPUBLISH\r\n"),
+                OpType.SPUBLISH => System.Text.Encoding.ASCII.GetBytes($"*3\r\n$8\r\nSPUBLISH\r\n"),
                 _ => null
             };
 
@@ -174,6 +176,8 @@ private bool GenerateBatch(int i, int start, int end, OpType opType)
                 case OpType.SCRIPTSET:
                 case OpType.SCRIPTGET:
                 case OpType.SCRIPTRETKEY:
+                case OpType.PUBLISH:
+                case OpType.SPUBLISH:
                     writeSuccess = GenerateSingleKeyValueOp(i, opHeader, start, end, opType);
                     return writeSuccess;
                 default:

benchmark/Resp.benchmark/ReqGenUtils.cs

@@ -94,6 +94,8 @@ private bool WriteOp(ref byte* curr, byte* vend, OpType opType)
                 case OpType.SCRIPTSET:
                 case OpType.SCRIPTGET:
                 case OpType.SCRIPTRETKEY:
+                case OpType.PUBLISH:
+                case OpType.SPUBLISH:
                     if (!WriteKey(ref curr, vend, out keyData))
                         return false;
                     break;
@@ -189,6 +191,8 @@ private bool WriteOp(ref byte* curr, byte* vend, OpType opType)
                 case OpType.MPFADD:
                 case OpType.SET:
                 case OpType.SCRIPTSET:
+                case OpType.PUBLISH:
+                case OpType.SPUBLISH:
                     RandomString();
                     if (!WriteStringBytes(ref curr, vend, valueBuffer))
                         return false;

benchmark/Resp.benchmark/RespOnlineBench.cs

@@ -595,7 +595,6 @@ public async void OpRunnerGarnetClientSession(int thread_id)
                     var op = SelectOpType(rand);
                     var startTimestamp = Stopwatch.GetTimestamp();
                     var c = opts.Pool ? await gcsPool.GetAsync() : client;
-
                     _ = op switch
                     {
                         OpType.PING => await c.ExecuteAsync(["PING"]),
@@ -605,6 +604,8 @@ public async void OpRunnerGarnetClientSession(int thread_id)
                         OpType.DEL => await c.ExecuteAsync(["DEL", req.GenerateKey()]),
                         OpType.SETBIT => await c.ExecuteAsync(["SETBIT", req.GenerateKey(), req.GenerateBitOffset()]),
                         OpType.GETBIT => await c.ExecuteAsync(["GETBIT", req.GenerateKey(), req.GenerateBitOffset()]),
+                        OpType.PUBLISH => await c.ExecuteAsync(["PUBLISH", req.GenerateKey(), req.GenerateValue()]),
+                        OpType.SPUBLISH => await c.ExecuteAsync(["SPUBLISH", req.GenerateKey(), req.GenerateValue()]),
                         OpType.ZADD => await ZADD(),
                         OpType.ZREM => await ZREM(),
                         OpType.ZCARD => await ZCARD(),
@@ -717,6 +718,12 @@ public async void OpRunnerGarnetClientSessionParallel(int thread_id, int paralle
                     case OpType.GETBIT:
                         c.ExecuteBatch(["GETBIT", req.GenerateKey(), req.GenerateBitOffset()]);
                         break;
+                    case OpType.PUBLISH:
+                        c.ExecuteBatch(["PUBLISH", req.GenerateKey(), req.GenerateValue()]);
+                        break;
+                    case OpType.SPUBLISH:
+                        c.ExecuteBatch(["SPUBLISH", req.GenerateKey(), req.GenerateValue()]);
+                        break;
 
                     default:
                         throw new Exception($"opType: {op} benchmark not supported with {opts.Client} ClientType!");
@@ -1046,6 +1053,12 @@ public async void OpRunnerSERedis(int thread_id)
                         case OpType.DEL:
                             await db.KeyDeleteAsync(req.GenerateKey());
                             break;
+                        case OpType.PUBLISH:
+                            await db.PublishAsync(RedisChannel.Literal(req.GenerateKey()), req.GenerateValue());
+                            break;
+                        case OpType.SPUBLISH:
+                            await db.ExecuteAsync("SPUBLISH", req.GenerateKey(), req.GenerateValue());
+                            break;
                         case OpType.ZADD:
                             {
                                 var key = req.GenerateKey();
@@ -1121,6 +1134,12 @@ public async void OpRunnerSERedisParallel(int thread_id, int parallel)
                         case OpType.SET:
                             tasks[offset++] = db.StringSetAsync(req.GenerateKey(), req.GenerateValue());
                             break;
+                        case OpType.PUBLISH:
+                            tasks[offset++] = db.PublishAsync(RedisChannel.Literal(req.GenerateKey()), req.GenerateValue());
+                            break;
+                        case OpType.SPUBLISH:
+                            tasks[offset++] = db.ExecuteAsync("SPUBLISH", req.GenerateKey(), req.GenerateValue());
+                            break;
                         case OpType.SETEX:
                             tasks[offset++] = db.StringSetAsync(req.GenerateKey(), req.GenerateValue(), TimeSpan.FromSeconds(opts.Ttl));
                             break;