Merge branch 'main' into feature/reduce-lock

Author: bootjp

adapter/distribution_server.go

@@ -21,6 +21,7 @@ type DistributionServer struct {
 	engine      *distribution.Engine
 	catalog     *distribution.CatalogStore
 	coordinator kv.Coordinator
+	readTracker *kv.ActiveTimestampTracker
 	reloadRetry struct {
 		attempts int
 		interval time.Duration
@@ -39,6 +40,12 @@ func WithDistributionCoordinator(coordinator kv.Coordinator) DistributionServerO
 	}
 }
 
+func WithDistributionActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) DistributionServerOption {
+	return func(s *DistributionServer) {
+		s.readTracker = tracker
+	}
+}
+
 // WithCatalogReloadRetryPolicy configures the retry policy used after split
 // commit when waiting for the local catalog snapshot to become visible.
 func WithCatalogReloadRetryPolicy(attempts int, interval time.Duration) DistributionServerOption {
@@ -140,6 +147,8 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR
 	if err != nil {
 		return nil, err
 	}
+	readPin := s.pinReadTS(snapshot.ReadTS)
+	defer readPin.Release()
 	if err := validateExpectedCatalogVersion(snapshot.Version, req.GetExpectedCatalogVersion()); err != nil {
 		return nil, err
 	}
@@ -175,6 +184,13 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR
 	}, nil
 }
 
+func (s *DistributionServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
+	if s == nil || s.readTracker == nil {
+		return nil
+	}
+	return s.readTracker.Pin(ts)
+}
+
 func (s *DistributionServer) verifyCatalogLeader() error {
 	if s.coordinator == nil {
 		return grpcStatusError(codes.FailedPrecondition, errDistributionCoordinatorRequired.Error())

adapter/grpc.go

@@ -12,6 +12,8 @@ import (
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"github.com/spaolacci/murmur3"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 var _ pb.RawKVServer = (*GRPCServer)(nil)
@@ -89,6 +91,9 @@ func (r *GRPCServer) RawGet(ctx context.Context, req *pb.RawGetRequest) (*pb.Raw
 	if errors.Is(err, store.ErrKeyNotFound) {
 		return &pb.RawGetResponse{Value: nil, Exists: false}, nil
 	}
+	if errors.Is(err, store.ErrReadTSCompacted) {
+		return nil, errors.WithStack(status.Error(codes.FailedPrecondition, store.ErrReadTSCompacted.Error()))
+	}
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -135,6 +140,9 @@ func (r *GRPCServer) RawScanAt(ctx context.Context, req *pb.RawScanAtRequest) (*
 		res, err = r.store.ScanAt(ctx, req.StartKey, req.EndKey, limit, readTS)
 	}
 	if err != nil {
+		if errors.Is(err, store.ErrReadTSCompacted) {
+			return &pb.RawScanAtResponse{Kv: nil}, errors.WithStack(status.Error(codes.FailedPrecondition, store.ErrReadTSCompacted.Error()))
+		}
 		return &pb.RawScanAtResponse{Kv: nil}, errors.WithStack(err)
 	}
 

adapter/redis.go

@@ -243,6 +243,7 @@ type RedisServer struct {
 	listen          net.Listener
 	store           store.MVCCStore
 	coordinator     kv.Coordinator
+	readTracker     *kv.ActiveTimestampTracker
 	redisTranscoder *redisTranscoder
 	pubsub          *redisPubSub
 	scriptMu        sync.RWMutex
@@ -258,6 +259,14 @@ type RedisServer struct {
 	route map[string]func(conn redcon.Conn, cmd redcon.Command)
 }
 
+type RedisServerOption func(*RedisServer)
+
+func WithRedisActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) RedisServerOption {
+	return func(r *RedisServer) {
+		r.readTracker = tracker
+	}
+}
+
 type connState struct {
 	inTxn bool
 	queue []redcon.Command
@@ -283,7 +292,7 @@ type redisResult struct {
 	err     error
 }
 
-func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore, coordinate kv.Coordinator, leaderRedis map[raft.ServerAddress]string, relay *RedisPubSubRelay) *RedisServer {
+func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore, coordinate kv.Coordinator, leaderRedis map[raft.ServerAddress]string, relay *RedisPubSubRelay, opts ...RedisServerOption) *RedisServer {
 	if relay == nil {
 		relay = NewRedisPubSubRelay()
 	}
@@ -383,6 +392,11 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		cmdZRevRangeByScore: r.zrevrangebyscore,
 		cmdZScore:           r.zscore,
 	}
+	for _, opt := range opts {
+		if opt != nil {
+			opt(r)
+		}
+	}
 
 	return r
 }
@@ -402,6 +416,13 @@ func (r *RedisServer) readTS() uint64 {
 	return snapshotTS(r.coordinator.Clock(), r.store)
 }
 
+func (r *RedisServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
+	if r == nil || r.readTracker == nil {
+		return nil
+	}
+	return r.readTracker.Pin(ts)
+}
+
 func (r *RedisServer) Run() error {
 	err := redcon.Serve(r.listen,
 		func(conn redcon.Conn, cmd redcon.Command) {
@@ -1905,6 +1926,8 @@ func (r *RedisServer) runTransaction(queue []redcon.Command) ([]redisResult, err
 	if err != nil {
 		return nil, err
 	}
+	readPin := r.pinReadTS(startTS)
+	defer readPin.Release()
 
 	ctx := &txnContext{
 		server:     r,

adapter/redis_lua.go

@@ -118,6 +118,7 @@ func (r *RedisServer) runLuaScript(conn redcon.Conn, script string, evalArgs [][
 	var reply luaReply
 	err = r.retryRedisWrite(ctx, func() error {
 		scriptCtx := newLuaScriptContext(r)
+		defer scriptCtx.Close()
 		state := newRedisLuaState()
 		defer state.Close()
 		state.SetContext(ctx)
@@ -847,6 +848,7 @@ func (r *RedisServer) execLuaCompat(conn redcon.Conn, command string, args [][]b
 	var reply luaReply
 	err := r.retryRedisWrite(ctx, func() error {
 		scriptCtx := newLuaScriptContext(r)
+		defer scriptCtx.Close()
 		nextReply, err := scriptCtx.exec(command, stringArgs)
 		if err != nil {
 			return err

adapter/redis_lua_context.go

@@ -16,6 +16,7 @@ import (
 type luaScriptContext struct {
 	server  *RedisServer
 	startTS uint64
+	readPin *kv.ActiveTimestampToken
 
 	touched map[string]struct{}
 	deleted map[string]bool
@@ -161,9 +162,11 @@ var luaRenameHandlers = map[redisValueType]luaRenameHandler{
 }
 
 func newLuaScriptContext(server *RedisServer) *luaScriptContext {
+	startTS := server.readTS()
 	return &luaScriptContext{
 		server:  server,
-		startTS: server.readTS(),
+		startTS: startTS,
+		readPin: server.pinReadTS(startTS),
 		touched: map[string]struct{}{},
 		deleted: map[string]bool{},
 		strings: map[string]*luaStringState{},
@@ -176,6 +179,13 @@ func newLuaScriptContext(server *RedisServer) *luaScriptContext {
 	}
 }
 
+func (c *luaScriptContext) Close() {
+	if c == nil || c.readPin == nil {
+		return
+	}
+	c.readPin.Release()
+}
+
 func (c *luaScriptContext) exec(command string, args []string) (luaReply, error) {
 	if handler, ok := luaCommandHandlers[command]; ok {
 		return handler(c, args)

adapter/test_util.go

@@ -13,6 +13,7 @@ import (
 	"github.com/Jille/raft-grpc-leader-rpc/leaderhealth"
 	transport "github.com/Jille/raft-grpc-transport"
 	"github.com/Jille/raftadmin"
+	internalutil "github.com/bootjp/elastickv/internal"
 	"github.com/bootjp/elastickv/kv"
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
@@ -23,7 +24,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sys/unix"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 func shutdown(nodes []Node) {
@@ -350,7 +350,7 @@ func setupNodes(t *testing.T, ctx context.Context, n int, ports []portsAdress) (
 		r, tm, err := newRaft(strconv.Itoa(i), port.raftAddress, fsm, i == 0, cfg, electionTimeout)
 		assert.NoError(t, err)
 
-		s := grpc.NewServer()
+		s := grpc.NewServer(internalutil.GRPCServerOptions()...)
 		trx := kv.NewTransaction(r)
 		coordinator := kv.NewCoordinator(trx, r)
 		relay := NewRedisPubSubRelay()
@@ -416,9 +416,7 @@ func newRaft(myID string, myAddress string, fsm raft.FSM, bootstrap bool, cfg ra
 		Level: hclog.LevelFromString("WARN"),
 	})
 
-	tm := transport.New(raft.ServerAddress(myAddress), []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	})
+	tm := transport.New(raft.ServerAddress(myAddress), internalutil.GRPCDialOptions())
 
 	r, err := raft.NewRaft(c, fsm, ldb, sdb, fss, tm.Transport())
 	if err != nil {

cmd/server/demo.go

@@ -30,7 +30,6 @@ import (
 	"github.com/hashicorp/raft"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 var (
@@ -206,7 +205,7 @@ func joinCluster(ctx context.Context, nodes []config) error {
 	}
 
 	// Connect to leader
-	conn, err := grpc.NewClient(leader.address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.NewClient(leader.address, internalutil.GRPCDialOptions()...)
 	if err != nil {
 		return fmt.Errorf("failed to dial leader: %w", err)
 	}
@@ -322,7 +321,7 @@ func setupStorage(dir string) (raft.LogStore, raft.StableStore, raft.SnapshotSto
 }
 
 func setupGRPC(r *raft.Raft, st store.MVCCStore, tm *transport.Manager, coordinator *kv.Coordinate, distServer *adapter.DistributionServer, relay *adapter.RedisPubSubRelay) (*grpc.Server, *adapter.GRPCServer) {
-	s := grpc.NewServer()
+	s := grpc.NewServer(internalutil.GRPCServerOptions()...)
 	trx := kv.NewTransaction(r)
 	routedStore := kv.NewLeaderRoutedStore(st, coordinator)
 	gs := adapter.NewGRPCServer(routedStore, coordinator, adapter.WithCloseStore())
@@ -336,7 +335,7 @@ func setupGRPC(r *raft.Raft, st store.MVCCStore, tm *transport.Manager, coordina
 	return s, gs
 }
 
-func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, coordinator *kv.Coordinate, addr, redisAddr, raftRedisMapStr string, relay *adapter.RedisPubSubRelay) (*adapter.RedisServer, error) {
+func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, coordinator *kv.Coordinate, addr, redisAddr, raftRedisMapStr string, relay *adapter.RedisPubSubRelay, readTracker *kv.ActiveTimestampTracker) (*adapter.RedisServer, error) {
 	leaderRedis := make(map[raft.ServerAddress]string)
 	if raftRedisMapStr != "" {
 		parts := strings.SplitSeq(raftRedisMapStr, ",")
@@ -355,7 +354,7 @@ func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, co
 		return nil, errors.WithStack(err)
 	}
 	routedStore := kv.NewLeaderRoutedStore(st, coordinator)
-	return adapter.NewRedisServer(l, redisAddr, routedStore, coordinator, leaderRedis, relay), nil
+	return adapter.NewRedisServer(l, redisAddr, routedStore, coordinator, leaderRedis, relay, adapter.WithRedisActiveTimestampTracker(readTracker)), nil
 }
 
 func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
@@ -370,6 +369,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 
 	st := store.NewMVCCStore()
 	fsm := kv.NewKvFSM(st)
+	readTracker := kv.NewActiveTimestampTracker()
 
 	// Config
 	c := raft.DefaultConfig()
@@ -381,9 +381,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 	})
 
 	// Transport
-	tm := transport.New(raft.ServerAddress(cfg.address), []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	})
+	tm := transport.New(raft.ServerAddress(cfg.address), internalutil.GRPCDialOptions())
 
 	r, err := raft.NewRaft(c, fsm, ldb, sdb, fss, tm.Transport())
 	if err != nil {
@@ -405,9 +403,18 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 		distEngine,
 		distCatalog,
 		adapter.WithDistributionCoordinator(coordinator),
+		adapter.WithDistributionActiveTimestampTracker(readTracker),
 	)
 	metricsRegistry := monitoring.NewRegistry(cfg.raftID, cfg.address)
 	metricsRegistry.RaftObserver().Start(ctx, []monitoring.RaftRuntime{{GroupID: 1, Raft: r}}, raftObserveInterval)
+	compactor := kv.NewFSMCompactor(
+		[]kv.FSMCompactRuntime{{
+			GroupID: 1,
+			Raft:    r,
+			Store:   st,
+		}},
+		kv.WithFSMCompactorActiveTimestampTracker(readTracker),
+	)
 	relay := adapter.NewRedisPubSubRelay()
 
 	s, grpcSvc := setupGRPC(r, st, tm, coordinator, distServer, relay)
@@ -420,7 +427,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 		_ = grpcSock.Close()
 	})
 
-	rd, err := setupRedis(ctx, lc, st, coordinator, cfg.address, cfg.redisAddress, cfg.raftRedisMap, relay)
+	rd, err := setupRedis(ctx, lc, st, coordinator, cfg.address, cfg.redisAddress, cfg.raftRedisMap, relay, readTracker)
 	if err != nil {
 		return err
 	}
@@ -447,6 +454,7 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 	}
 
 	eg.Go(catalogWatcherTask(ctx, distCatalog, distEngine))
+	eg.Go(func() error { return compactor.Run(ctx) })
 	eg.Go(grpcShutdownTask(ctx, s, grpcSock, cfg.address, grpcSvc))
 	eg.Go(grpcServeTask(s, grpcSock, cfg.address))
 	eg.Go(redisShutdownTask(ctx, rd, cfg.redisAddress))