add: benchmark test for start method in worker.go

go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/cockroachdb/swiss v0.0.0-20240612210725-f4de07ae6964
 	github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da
-	github.com/dicedb/dicedb-go v0.0.0-20241015181607-d31c1df12107
+	github.com/dicedb/dicedb-go v0.0.0-20241026093718-570de4575be3
 	github.com/gobwas/glob v0.2.3
 	github.com/google/btree v1.1.3
 	github.com/google/go-cmp v0.6.0

go.sum

@@ -32,6 +32,8 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/r
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dicedb/dicedb-go v0.0.0-20241015181607-d31c1df12107 h1:sL5dXtCsogSMP/afS2K2vVMMYFqJy02EezeRXpZnGy0=
 github.com/dicedb/dicedb-go v0.0.0-20241015181607-d31c1df12107/go.mod h1:iaOsphlvjJ87VL/5d32ZgeQYxhYS51k7/bvFKro7lWk=
+github.com/dicedb/dicedb-go v0.0.0-20241026093718-570de4575be3 h1:JvnAibMNGA0vQH+T47Y/d5/POURIvfJl3fFk0GIEBkQ=
+github.com/dicedb/dicedb-go v0.0.0-20241026093718-570de4575be3/go.mod h1:p7x5/3S6wBEmiRMwxavj1I1P1xsSVQS6fcSbeai5ic4=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=

integration_tests/server/keep_alive_test.go

@@ -0,0 +1,117 @@
+package server
+
+import(
+	"testing"
+	"runtime"
+	"context"
+	"sync"
+	"log/slog"
+	"fmt"
+	"errors"
+	"net/http"
+	"time"
+
+
+	dice "github.com/dicedb/dicedb-go"
+	diceerrors "github.com/dicedb/dice/internal/errors"
+	"github.com/dicedb/dice/internal/shard"
+	"github.com/dicedb/dice/config"
+	dstore "github.com/dicedb/dice/internal/store"
+	"github.com/dicedb/dice/internal/server/resp"
+	"github.com/dicedb/dice/internal/worker"
+	"github.com/dicedb/dice/internal/server/abstractserver"
+
+)
+
+func BenchmarkStart(b *testing.B){
+	numShards := runtime.NumCPU()
+	runtime.GOMAXPROCS(numShards)
+
+	var queryWatchChan chan dstore.QueryWatchEvent
+	var cmdWatchChan   chan dstore.CmdWatchEvent
+	var serverErrCh = make(chan error, 2)
+
+	shardManager := shard.NewShardManager(uint8(numShards), queryWatchChan, cmdWatchChan, serverErrCh)
+
+	var ctx, cancel = context.WithCancel(context.Background())
+
+	var wg = sync.WaitGroup{}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		shardManager.Run(ctx)
+	}()
+
+	var serverWg sync.WaitGroup
+
+	workerManager := worker.NewWorkerManager(config.DiceConfig.Performance.MaxClients, shardManager)
+	respServer := resp.NewServer(shardManager, workerManager, cmdWatchChan, serverErrCh)
+
+	serverWg.Add(1)
+	go runServer(ctx, &serverWg, respServer, serverErrCh)
+
+	time.Sleep(10 * time.Second)
+
+	commands := []string{"PING", "SET foo bar", "GET foo", "INCR counter", "DEL foo"}
+	concurrencyLevels := []int{10, 50, 100, 200}
+
+	for _, concurrency := range concurrencyLevels {
+		b.Run("Concurrency="+fmt.Sprintf("%d",concurrency), func(b *testing.B) {
+			client := dice.NewClient(&dice.Options{
+				Addr: "localhost:7379",
+				Password: "",
+				DB: 0,
+			})
+			defer client.Close()
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				var cmdWg sync.WaitGroup
+				cmdWg.Add(concurrency)
+				for j := 0; j < concurrency; j++ {
+					go func() {
+						defer cmdWg.Done()
+						for _, cmd := range commands {
+							switch cmd {
+							case "PING":
+								client.Ping(ctx)
+							case "SET foo bar":
+								client.Set(ctx, "foo", "bar", 0)
+							case "GET foo":
+								client.Get(ctx, "foo")
+							case "INCR counter":
+								client.Incr(ctx, "counter")
+							case "DEL foo":
+								client.Del(ctx, "foo")
+							}
+						}
+					}()
+				}
+				cmdWg.Wait()
+			}
+		})
+	}
+	cancel()
+	wg.Wait()
+	serverWg.Wait()
+}
+
+func runServer(ctx context.Context, wg *sync.WaitGroup, srv abstractserver.AbstractServer, errCh chan<- error) {
+	defer wg.Done()
+	if err := srv.Run(ctx); err != nil {
+		switch {
+		case errors.Is(err, context.Canceled):
+			slog.Debug(fmt.Sprintf("%T was canceled", srv))
+		case errors.Is(err, diceerrors.ErrAborted):
+			slog.Debug(fmt.Sprintf("%T received abort command", srv))
+		case errors.Is(err, http.ErrServerClosed):
+			slog.Debug(fmt.Sprintf("%T received abort command", srv))
+		default:
+			slog.Error(fmt.Sprintf("%T error", srv), slog.Any("error", err))
+		}
+		errCh <- err
+	} else {
+		slog.Debug("bye.")
+	}
+}