Allow coordinator to serve multiple clients

coordinator/app/app.go

@@ -14,15 +14,23 @@ import (
 	"github.com/pg-sharding/spqr/coordinator/provider"
 	"github.com/pg-sharding/spqr/pkg/config"
 	protos "github.com/pg-sharding/spqr/pkg/protos"
+
+	"golang.org/x/sync/semaphore"
 )
 
 type App struct {
 	coordinator coordinator.Coordinator
+	sem         *semaphore.Weighted
 }
 
+const (
+	maxWorkers = 50
+)
+
 func NewApp(c coordinator.Coordinator) *App {
 	return &App{
 		coordinator: c,
+		sem:         semaphore.NewWeighted(int64(maxWorkers)),
 	}
 }
 
@@ -83,8 +91,24 @@ func (app *App) ServeCoordinator(wg *sync.WaitGroup) error {
 
 			for {
 				conn, err := listener.Accept()
-				spqrlog.Zero.Error().Err(err).Msg("")
-				_ = app.coordinator.ProcClient(context.TODO(), conn)
+				if err != nil {
+					spqrlog.Zero.Error().Err(err).Msg("")
+					continue
+				}
+
+				if err := app.sem.Acquire(context.Background(), 1); err != nil {
+					spqrlog.Zero.Error().Err(err).Msg("")
+					continue
+				}
+
+				go func() {
+					defer app.sem.Release(1)
+
+					err := app.coordinator.ProcClient(context.TODO(), conn)
+					if err != nil {
+						spqrlog.Zero.Error().Err(err).Msg("failed to serve client")
+					}
+				}()
 			}
 		}(l)
 	}

qdb/etcdqdb.go

@@ -63,7 +63,7 @@ const (
 )
 
 func keyLockPath(key string) string {
-	return path.Join("lock", key)
+	return path.Join("/lock", key)
 }
 
 func keyRangeNodePath(key string) string {

qdb/memqdb.go

@@ -459,6 +459,7 @@ func (q *MemQDB) UpdateCoordinator(ctx context.Context, address string) error {
 }
 
 func (q *MemQDB) GetCoordinator(ctx context.Context) (string, error) {
+	spqrlog.Zero.Debug().Str("address", q.Coordinator).Msg("memqdb: get coordinator address")
 	return q.Coordinator, nil
 }