diff options
Diffstat (limited to 'weed/server')
| -rw-r--r-- | weed/server/filer_grpc_server_dlm.go | 46 | ||||
| -rw-r--r-- | weed/server/filer_server.go | 4 |
2 files changed, 15 insertions, 35 deletions
diff --git a/weed/server/filer_grpc_server_dlm.go b/weed/server/filer_grpc_server_dlm.go index fdd299947..6618285e0 100644 --- a/weed/server/filer_grpc_server_dlm.go +++ b/weed/server/filer_grpc_server_dlm.go @@ -3,7 +3,6 @@ package weed_server import ( "context" "fmt" - "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -18,21 +17,17 @@ func (fs *FilerServer) Lock(ctx context.Context, req *filer_pb.LockRequest) (res if snapshot == nil { resp.Error = "no lock server found" return - } else { - server := lock_manager.HashKeyToServer(req.Name, snapshot) - if server != fs.option.Host { - resp.Error = fmt.Sprintf("not the lock server for %s", req.Name) - resp.MovedTo = string(server) - return - } } - renewToken, err := fs.dlm.Lock(req.Name, time.Duration(req.SecondsToLock)*time.Second, req.PreviousLockToken) + var movedTo pb.ServerAddress + expiredAtNs := time.Now().Add(time.Duration(req.SecondsToLock) * time.Second).UnixNano() + resp.RenewToken, movedTo, err = fs.dlm.Lock(fs.option.Host, req.Name, expiredAtNs, req.PreviousLockToken, snapshot) if err != nil { resp.Error = fmt.Sprintf("%v", err) - } else { - resp.RenewToken = renewToken + } + if movedTo != "" { + resp.MovedTo = string(movedTo) } return resp, nil @@ -46,16 +41,9 @@ func (fs *FilerServer) Unlock(ctx context.Context, req *filer_pb.UnlockRequest) if snapshot == nil { resp.Error = "no lock server found" return - } else { - server := lock_manager.HashKeyToServer(req.Name, snapshot) - if server != fs.option.Host { - resp.Error = fmt.Sprintf("not the lock server for %s", req.Name) - resp.MovedTo = string(server) - return - } } - _, err = fs.dlm.Unlock(req.Name, req.LockToken) + _, err = fs.dlm.Unlock(fs.option.Host, req.Name, req.LockToken, snapshot) if err != nil { resp.Error = fmt.Sprintf("%v", err) } @@ -67,13 +55,8 @@ func (fs *FilerServer) Unlock(ctx context.Context, req *filer_pb.UnlockRequest) // TransferLocks is a grpc handler to handle FilerServer's TransferLocksRequest func (fs *FilerServer) TransferLocks(ctx context.Context, req *filer_pb.TransferLocksRequest) (*filer_pb.TransferLocksResponse, error) { - now := time.Now() for _, lock := range req.Locks { - duration := time.Duration(lock.ExpirationNs - now.UnixNano()) - if _, err := fs.dlm.Lock(lock.Name, duration, lock.RenewToken); err != nil { - glog.Errorf("receive transferred lock %v to %v: %v", lock.Name, fs.option.Host, err) - return nil, err - } + fs.dlm.InsertLock(lock.Name, lock.ExpiredAtNs, lock.RenewToken) } return &filer_pb.TransferLocksResponse{}, nil @@ -81,23 +64,20 @@ func (fs *FilerServer) TransferLocks(ctx context.Context, req *filer_pb.Transfer } func (fs *FilerServer) OnDlmChangeSnapshot(snapshot []pb.ServerAddress) { - locks := fs.dlm.TakeOutLocksByKey(func(key string) bool { - server := lock_manager.HashKeyToServer(key, snapshot) - return server != fs.option.Host - }) + locks := fs.dlm.SelectNotOwnedLocks(fs.option.Host, snapshot) if len(locks) == 0 { return } for _, lock := range locks { - server := lock_manager.HashKeyToServer(lock.Key, snapshot) + server := fs.dlm.CalculateTargetServer(lock.Key, snapshot) if err := pb.WithFilerClient(false, 0, server, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { _, err := client.TransferLocks(context.Background(), &filer_pb.TransferLocksRequest{ Locks: []*filer_pb.Lock{ { - Name: lock.Key, - RenewToken: lock.Token, - ExpirationNs: lock.ExpirationNs, + Name: lock.Key, + RenewToken: lock.Token, + ExpiredAtNs: lock.ExpiredAtNs, }, }, }) diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index e22dfd737..adb1dd85b 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -97,7 +97,7 @@ type FilerServer struct { knownListeners map[int32]int32 // distributed lock manager - dlm *lock_manager.LockManager + dlm *lock_manager.DistributedLockManager } func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) { @@ -185,7 +185,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) fs.filer.Shutdown() }) - fs.dlm = lock_manager.NewLockManager() + fs.dlm = lock_manager.NewDistributedLockManager() fs.filer.LockRing.SetTakeSnapshotCallback(fs.OnDlmChangeSnapshot) return fs, nil |