add locks functions

pkg/incrservice/store_sql_test.go

@@ -138,6 +138,11 @@ func (tls *testLockService) GetWaitingList(ctx context.Context, txnID []byte) (b
 	panic("implement me")
 }
 
+func (tls *testLockService) GetLockHolder(ctx context.Context, tableID uint64, row []byte, options lock.LockOptions) (lock.WaitTxn, bool, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (tls *testLockService) ForceRefreshLockTableBinds(targets []uint64, matcher func(bind lock.LockTable) bool) {
 	//TODO implement me
 	panic("implement me")

pkg/lockservice/lock_table_local.go

@@ -354,6 +354,26 @@ func (l *localLockTable) getLock(
 	}
 }
 
+func (l *localLockTable) getLockHolder(ctx context.Context, key []byte) (pb.WaitTxn, bool, error) {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+	if l.mu.closed {
+		return pb.WaitTxn{}, false, nil
+	}
+	lock, ok := l.mu.store.Get(key)
+	if !ok {
+		return pb.WaitTxn{}, false, nil
+	}
+	var holder pb.WaitTxn
+	var found bool
+	lock.IterHolders(func(v pb.WaitTxn) bool {
+		holder = v
+		found = true
+		return false
+	})
+	return holder, found, nil
+}
+
 func (l *localLockTable) getBind() pb.LockTable {
 	return l.bind
 }

pkg/lockservice/lock_table_proxy.go

@@ -202,6 +202,10 @@ func (lp *localLockTableProxy) getLock(
 	lp.remote.getLock(key, txn, fn)
 }
 
+func (lp *localLockTableProxy) getLockHolder(ctx context.Context, key []byte) (pb.WaitTxn, bool, error) {
+	return lp.remote.getLockHolder(ctx, key)
+}
+
 func (lp *localLockTableProxy) getBind() pb.LockTable {
 	return lp.remote.getBind()
 }

pkg/lockservice/lock_table_remote.go

@@ -242,12 +242,52 @@ func (l *remoteLockTable) getLock(
 	}
 }
 
+func (l *remoteLockTable) getLockHolder(ctx context.Context, key []byte) (pb.WaitTxn, bool, error) {
+	backoff := remoteRetryInitialBackoff
+	for {
+		if err := ctx.Err(); err != nil {
+			return pb.WaitTxn{}, false, err
+		}
+		holder, ok, err := l.doGetLockHolder(ctx, key)
+		if err == nil {
+			return holder, ok, nil
+		}
+		if err := ctx.Err(); err != nil {
+			return pb.WaitTxn{}, false, err
+		}
+		if err = l.handleError(err, false); err == nil {
+			// The bind-change handler replaces the lock-table object in service.tableGroups.
+			// This in-flight remote table still carries the stale bind, so let the service
+			// reacquire the current table before retrying the holder lookup.
+			return pb.WaitTxn{}, false, ErrLockTableBindChanged
+		}
+		if err := waitRemoteRetryBackoffWithContext(ctx, backoff); err != nil {
+			return pb.WaitTxn{}, false, err
+		}
+		backoff = nextRemoteRetryBackoff(backoff)
+	}
+}
+
 func waitRemoteRetryBackoff(backoff time.Duration) {
 	if backoff > 0 {
 		time.Sleep(backoff)
 	}
 }
 
+func waitRemoteRetryBackoffWithContext(ctx context.Context, backoff time.Duration) error {
+	if backoff <= 0 {
+		return ctx.Err()
+	}
+	timer := time.NewTimer(backoff)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
 func nextRemoteRetryBackoff(backoff time.Duration) time.Duration {
 	if backoff <= 0 {
 		return remoteRetryInitialBackoff
@@ -320,6 +360,32 @@ func (l *remoteLockTable) doGetLock(key []byte, txn pb.WaitTxn) (Lock, bool, err
 	return Lock{}, false, moerr.AttachCause(ctx, err)
 }
 
+func (l *remoteLockTable) doGetLockHolder(ctx context.Context, key []byte) (pb.WaitTxn, bool, error) {
+	ctx, cancel := context.WithTimeoutCause(ctx, defaultRPCTimeout, moerr.CauseDoGetLock)
+	defer cancel()
+
+	req := acquireRequest()
+	defer releaseRequest(req)
+
+	req.Method = pb.Method_GetLockHolder
+	req.LockTable = l.bind
+	req.GetLockHolder.Row = key
+	req.GetLockHolder.Sharding = l.bind.Sharding
+
+	resp, err := l.client.Send(ctx, req)
+	if err == nil {
+		defer releaseResponse(resp)
+		if err := l.maybeHandleBindChanged(resp); err != nil {
+			return pb.WaitTxn{}, false, err
+		}
+		if len(resp.GetLockHolder.Holder.TxnID) == 0 {
+			return pb.WaitTxn{}, false, nil
+		}
+		return resp.GetLockHolder.Holder, true, nil
+	}
+	return pb.WaitTxn{}, false, moerr.AttachCause(ctx, err)
+}
+
 func (l *remoteLockTable) getBind() pb.LockTable {
 	return l.bind
 }

pkg/lockservice/lock_table_remote_test.go

@@ -384,6 +384,133 @@ func TestGetLockRemoteWithRetry(t *testing.T) {
 	)
 }
 
+func TestGetLockHolderRemoteReturnsBindChangedAfterBindRefresh(t *testing.T) {
+	oldInitial := remoteRetryInitialBackoff
+	oldMaxBackoff := remoteRetryMaxBackoff
+	remoteRetryInitialBackoff = time.Millisecond
+	remoteRetryMaxBackoff = time.Millisecond
+	defer func() {
+		remoteRetryInitialBackoff = oldInitial
+		remoteRetryMaxBackoff = oldMaxBackoff
+	}()
+
+	n := 0
+	refreshedBind := pb.LockTable{ServiceID: "s1", Table: 1, Version: 2}
+	runRemoteLockTableTests(
+		t,
+		pb.LockTable{ServiceID: "s1", Table: 1, Version: 1},
+		func(s Server) {
+			s.RegisterMethodHandler(
+				pb.Method_GetLockHolder,
+				func(
+					ctx context.Context,
+					cancel context.CancelFunc,
+					req *pb.Request,
+					resp *pb.Response,
+					cs morpc.ClientSession) {
+					n++
+					writeResponse(getLogger(""), cancel, resp, moerr.NewRPCTimeout(ctx), cs)
+				},
+			)
+			s.RegisterMethodHandler(
+				pb.Method_GetBind,
+				func(
+					ctx context.Context,
+					cancel context.CancelFunc,
+					req *pb.Request,
+					resp *pb.Response,
+					cs morpc.ClientSession) {
+					resp.GetBind.LockTable = refreshedBind
+					writeResponse(getLogger(""), cancel, resp, nil, cs)
+				},
+			)
+		},
+		func(l *remoteLockTable, s Server) {
+			_, ok, err := l.getLockHolder(context.Background(), []byte("row1"))
+			require.True(t, moerr.IsMoErrCode(err, moerr.ErrLockTableBindChanged))
+			require.False(t, ok)
+			require.Equal(t, 1, n)
+		},
+		func(lt pb.LockTable) {},
+	)
+}
+
+func TestGetLockHolderRemoteCarriesSharding(t *testing.T) {
+	runRemoteLockTableTests(
+		t,
+		pb.LockTable{ServiceID: "s1", Table: 1, Sharding: pb.Sharding_ByRow},
+		func(s Server) {
+			s.RegisterMethodHandler(
+				pb.Method_GetLockHolder,
+				func(
+					ctx context.Context,
+					cancel context.CancelFunc,
+					req *pb.Request,
+					resp *pb.Response,
+					cs morpc.ClientSession) {
+					require.Equal(t, pb.Sharding_ByRow, req.GetLockHolder.Sharding)
+					writeResponse(getLogger(""), cancel, resp, nil, cs)
+				},
+			)
+		},
+		func(l *remoteLockTable, s Server) {
+			_, ok, err := l.getLockHolder(context.Background(), []byte("row1"))
+			require.NoError(t, err)
+			require.False(t, ok)
+		},
+		func(lt pb.LockTable) {},
+	)
+}
+
+func TestGetLockHolderRemoteStopsRetryWhenContextCanceled(t *testing.T) {
+	oldInitial := remoteRetryInitialBackoff
+	oldMaxBackoff := remoteRetryMaxBackoff
+	remoteRetryInitialBackoff = time.Millisecond
+	remoteRetryMaxBackoff = time.Millisecond
+	defer func() {
+		remoteRetryInitialBackoff = oldInitial
+		remoteRetryMaxBackoff = oldMaxBackoff
+	}()
+
+	runRemoteLockTableTests(
+		t,
+		pb.LockTable{ServiceID: "s1", Table: 1, Version: 1},
+		func(s Server) {
+			s.RegisterMethodHandler(
+				pb.Method_GetLockHolder,
+				func(
+					ctx context.Context,
+					cancel context.CancelFunc,
+					req *pb.Request,
+					resp *pb.Response,
+					cs morpc.ClientSession) {
+					writeResponse(getLogger(""), cancel, resp, moerr.NewRPCTimeout(ctx), cs)
+				},
+			)
+			s.RegisterMethodHandler(
+				pb.Method_GetBind,
+				func(
+					ctx context.Context,
+					cancel context.CancelFunc,
+					req *pb.Request,
+					resp *pb.Response,
+					cs morpc.ClientSession) {
+					resp.GetBind.LockTable = req.LockTable
+					writeResponse(getLogger(""), cancel, resp, nil, cs)
+				},
+			)
+		},
+		func(l *remoteLockTable, s Server) {
+			ctx, cancel := context.WithCancel(context.Background())
+			cancel()
+			_, ok, err := l.getLockHolder(ctx, []byte("row1"))
+			require.ErrorIs(t, err, context.Canceled)
+			require.False(t, ok)
+		},
+		func(lt pb.LockTable) {},
+	)
+}
+
 func TestRemoteWithBindChanged(t *testing.T) {
 	newBind := pb.LockTable{
 		ServiceID: "s2",

pkg/lockservice/rpc.go

@@ -170,6 +170,7 @@ func (c *client) AsyncSend(ctx context.Context, request *pb.Request) (*morpc.Fut
 		case pb.Method_Lock,
 			pb.Method_Unlock,
 			pb.Method_GetTxnLock,
+			pb.Method_GetLockHolder,
 			pb.Method_KeepRemoteLock:
 			sid = getUUIDFromServiceIdentifier(request.LockTable.ServiceID)
 			c.cluster.GetCNServiceWithoutWorkingState(

pkg/lockservice/service_observability.go

@@ -17,6 +17,7 @@ package lockservice
 import (
 	"context"
 
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/lock"
 )
 
@@ -49,6 +50,30 @@ func (s *service) GetWaitingList(
 	return true, waitingList, nil
 }
 
+func (s *service) GetLockHolder(
+	ctx context.Context,
+	tableID uint64,
+	row []byte,
+	options pb.LockOptions) (pb.WaitTxn, bool, error) {
+	s.wait()
+	for {
+		if err := ctx.Err(); err != nil {
+			return pb.WaitTxn{}, false, err
+		}
+		s.bindChangeMu.RLock()
+		l, err := s.getLockTableWithCreate(options.Group, tableID, [][]byte{row}, options.Sharding)
+		if err != nil {
+			s.bindChangeMu.RUnlock()
+			return pb.WaitTxn{}, false, err
+		}
+		holder, ok, err := l.getLockHolder(ctx, row)
+		s.bindChangeMu.RUnlock()
+		if !moerr.IsMoErrCode(err, moerr.ErrLockTableBindChanged) {
+			return holder, ok, err
+		}
+	}
+}
+
 func (s *service) ForceRefreshLockTableBinds(
 	targets []uint64,
 	matcher func(bind pb.LockTable) bool) {