From 7c2ebcad1399922a10bf80fb8ef59565b13b0710 Mon Sep 17 00:00:00 2001 From: Sydonian <794346190@qq.com> Date: Tue, 24 Oct 2023 15:29:07 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E4=BC=98=E5=8C=96imfs=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sdks/imfs/models.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdks/imfs/models.go b/sdks/imfs/models.go index ac53719..8565719 100644 --- a/sdks/imfs/models.go +++ b/sdks/imfs/models.go @@ -1,5 +1,6 @@ package imsdk const ( - EnvPackageList = "IMFS_PACKAGE_LIST" + EnvPackageList = "IMFS_PACKAGE_LIST" + EnvServiceAddress = "IMFS_SERVICE_ADDRESS" ) From 0eb03a6f2a430e37a5f737744de7e745fad25c33 Mon Sep 17 00:00:00 2001 From: Sydonian <794346190@qq.com> Date: Thu, 26 Oct 2023 15:53:53 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E6=8B=86=E5=88=86MainActor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkgs/distlock/config.go | 7 +- .../service/internal/acquire_actor.go | 233 +++++++++++++ pkgs/distlock/service/internal/lease_actor.go | 16 +- pkgs/distlock/service/internal/main_actor.go | 313 ------------------ .../service/internal/providers_actor.go | 26 +- .../service/internal/release_actor.go | 144 ++++++++ pkgs/distlock/service/internal/retry_actor.go | 128 ------- .../service/internal/watch_etcd_actor.go | 4 +- pkgs/distlock/service/mutex.go | 4 +- pkgs/distlock/service/service.go | 115 ++++--- pkgs/future/set_value_future.go | 2 + 11 files changed, 472 insertions(+), 520 deletions(-) create mode 100644 pkgs/distlock/service/internal/acquire_actor.go delete mode 100644 pkgs/distlock/service/internal/main_actor.go create mode 100644 pkgs/distlock/service/internal/release_actor.go delete mode 100644 pkgs/distlock/service/internal/retry_actor.go diff --git a/pkgs/distlock/config.go b/pkgs/distlock/config.go index 65ffe67..ef2acb1 100644 --- a/pkgs/distlock/config.go +++ b/pkgs/distlock/config.go @@ -5,9 +5,6 @@ type Config struct { EtcdUsername string `json:"etcdUsername"` EtcdPassword string `json:"etcdPassword"` - EtcdLockLeaseTimeSec int64 `json:"etcdLockLeaseTimeSec"` // 全局锁的租约时间。锁服务会在这个时间内自动续约锁，但如果服务崩溃，则其他服务在租约到期后能重新获得锁。 - - // 写入锁请求数据到的ETCD的时候，不设置租约。开启此选项之后，请求锁的服务崩溃， - // 锁请求数据会依然留在ETCD中。仅供调试使用。 - SubmitLockRequestWithoutLease bool `json:"submitLockRequestWithoutLease"` + EtcdLockLeaseTimeSec int64 `json:"etcdLockLeaseTimeSec"` // 全局锁的租约时间。锁服务会在这个时间内自动续约锁，但如果服务崩溃，则其他服务在租约到期后能重新获得锁。 + RandomReleasingDelayMs int64 `json:"randomReleasingDelayMs"` // 释放锁失败，随机延迟之后再次尝试。延迟时间=random(0, RandomReleasingDelayMs) + 最少延迟时间(1000ms) } diff --git a/pkgs/distlock/service/internal/acquire_actor.go b/pkgs/distlock/service/internal/acquire_actor.go new file mode 100644 index 0000000..e7a2a39 --- /dev/null +++ b/pkgs/distlock/service/internal/acquire_actor.go @@ -0,0 +1,233 @@ +package internal + +import ( + "context" + "errors" + "fmt" + "strconv" + "sync" + + "gitlink.org.cn/cloudream/common/pkgs/distlock" + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/pkgs/logger" + mylo "gitlink.org.cn/cloudream/common/utils/lo" + "gitlink.org.cn/cloudream/common/utils/serder" + clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/client/v3/concurrency" +) + +var ErrAcquiringTimeout = errors.New("acquiring timeout") + +const ( + EtcdLockRequestData = "/distlock/lockRequest/data" + EtcdLockRequestIndex = "/distlock/lockRequest/index" + EtcdLockRequestLock = "/distlock/lockRequest/lock" +) + +type lockData struct { + Path []string `json:"path"` + Name string `json:"name"` + Target string `json:"target"` +} +type LockRequestData struct { + ID string `json:"id"` + Locks []lockData `json:"locks"` +} + +type acquireInfo struct { + Request distlock.LockRequest + Callback *future.SetValueFuture[string] + LastErr error +} + +type AcquireActor struct { + cfg *distlock.Config + etcdCli *clientv3.Client + providersActor *ProvidersActor + + acquirings []*acquireInfo + lock sync.Mutex +} + +func NewAcquireActor(cfg *distlock.Config, etcdCli *clientv3.Client) *AcquireActor { + return &AcquireActor{ + cfg: cfg, + etcdCli: etcdCli, + } +} + +func (a *AcquireActor) Init(providersActor *ProvidersActor) { + a.providersActor = providersActor +} + +// Acquire 请求一批锁。成功后返回锁请求ID +func (a *AcquireActor) Acquire(ctx context.Context, req distlock.LockRequest) (string, error) { + info := &acquireInfo{ + Request: req, + Callback: future.NewSetValue[string](), + } + + func() { + a.lock.Lock() + defer a.lock.Unlock() + + a.acquirings = append(a.acquirings, info) + // TODO 处理错误 + err := a.doAcquiring() + if err != nil { + logger.Std.Debugf("doing acquiring: %s", err.Error()) + } + }() + + go func() { + info.Callback.Wait(ctx) + + a.lock.Lock() + defer a.lock.Unlock() + + // 调用Callback时都加了锁，所以此处的IsComplete判断可以作为后续操作的依据 + if info.Callback.IsComplete() { + return + } + + a.acquirings = mylo.Remove(a.acquirings, info) + if info.LastErr != nil { + info.Callback.SetError(info.LastErr) + } else { + info.Callback.SetError(ErrAcquiringTimeout) + } + }() + + // 此处不能直接用ctx去等Callback，原因是Wait超时不代表锁没有获取到，这会导致锁泄露。 + return info.Callback.WaitValue(context.Background()) +} + +// TryAcquireNow 重试一下内部还没有成功的锁请求。不会阻塞调用者 +func (a *AcquireActor) TryAcquireNow() { + go func() { + a.lock.Lock() + defer a.lock.Unlock() + + err := a.doAcquiring() + if err != nil { + logger.Std.Debugf("doing acquiring: %s", err.Error()) + } + }() +} + +func (a *AcquireActor) doAcquiring() error { + ctx := context.Background() + + if len(a.acquirings) == 0 { + return nil + } + + // TODO 根据不同的错误设置不同的错误类型，方便上层进行后续处理 + unlock, err := acquireEtcdRequestDataLock(ctx, a.etcdCli, a.cfg.EtcdLockLeaseTimeSec) + if err != nil { + return fmt.Errorf("acquire etcd request data lock failed, err: %w", err) + } + defer unlock() + + index, err := getEtcdLockRequestIndex(ctx, a.etcdCli) + if err != nil { + return err + } + + // 等待本地状态同步到最新 + // TODO 配置等待时间 + err = a.providersActor.WaitIndexUpdated(ctx, index) + if err != nil { + return err + } + + // TODO 可以考虑一次性获得多个锁 + for i := 0; i < len(a.acquirings); i++ { + // 测试锁，并获得锁数据 + reqData, err := a.providersActor.TestLockRequestAndMakeData(a.acquirings[i].Request) + if err != nil { + a.acquirings[i].LastErr = err + continue + } + + nextIndexStr := strconv.FormatInt(index+1, 10) + reqData.ID = nextIndexStr + + // 锁成功，提交锁数据 + err = a.submitLockRequest(ctx, nextIndexStr, reqData) + if err != nil { + a.acquirings[i].LastErr = err + continue + } + + a.acquirings[i].Callback.SetValue(reqData.ID) + a.acquirings = mylo.RemoveAt(a.acquirings, i) + break + } + + return nil +} + +func (a *AcquireActor) submitLockRequest(ctx context.Context, index string, reqData LockRequestData) error { + reqBytes, err := serder.ObjectToJSON(reqData) + if err != nil { + return fmt.Errorf("serialize lock request data failed, err: %w", err) + } + + etcdOps := []clientv3.Op{ + clientv3.OpPut(EtcdLockRequestIndex, index), + clientv3.OpPut(makeEtcdLockRequestKey(reqData.ID), string(reqBytes)), + } + txResp, err := a.etcdCli.Txn(ctx).Then(etcdOps...).Commit() + if err != nil { + return fmt.Errorf("submit lock request data failed, err: %w", err) + } + if !txResp.Succeeded { + return fmt.Errorf("submit lock request data failed for lock request data index changed") + } + + return nil +} + +func acquireEtcdRequestDataLock(ctx context.Context, etcdCli *clientv3.Client, etcdLockLeaseTimeSec int64) (unlock func(), err error) { + lease, err := etcdCli.Grant(context.Background(), etcdLockLeaseTimeSec) + if err != nil { + return nil, fmt.Errorf("grant lease failed, err: %w", err) + } + + session, err := concurrency.NewSession(etcdCli, concurrency.WithLease(lease.ID)) + if err != nil { + return nil, fmt.Errorf("new session failed, err: %w", err) + } + + mutex := concurrency.NewMutex(session, EtcdLockRequestLock) + + err = mutex.Lock(ctx) + if err != nil { + session.Close() + return nil, fmt.Errorf("acquire lock failed, err: %w", err) + } + + return func() { + mutex.Unlock(context.Background()) + session.Close() + }, nil +} + +func getEtcdLockRequestIndex(ctx context.Context, etcdCli *clientv3.Client) (int64, error) { + indexKv, err := etcdCli.Get(ctx, EtcdLockRequestIndex) + if err != nil { + return 0, fmt.Errorf("get lock request index failed, err: %w", err) + } + + if len(indexKv.Kvs) == 0 { + return 0, nil + } + + index, err := strconv.ParseInt(string(indexKv.Kvs[0].Value), 0, 64) + if err != nil { + return 0, fmt.Errorf("parse lock request index failed, err: %w", err) + } + + return index, nil +} diff --git a/pkgs/distlock/service/internal/lease_actor.go b/pkgs/distlock/service/internal/lease_actor.go index 98eecf0..132430a 100644 --- a/pkgs/distlock/service/internal/lease_actor.go +++ b/pkgs/distlock/service/internal/lease_actor.go @@ -21,7 +21,7 @@ type LeaseActor struct { commandChan *actor.CommandChannel - mainActor *MainActor + releaseActor *ReleaseActor } func NewLeaseActor() *LeaseActor { @@ -31,8 +31,8 @@ func NewLeaseActor() *LeaseActor { } } -func (a *LeaseActor) Init(mainActor *MainActor) { - a.mainActor = mainActor +func (a *LeaseActor) Init(releaseActor *ReleaseActor) { + a.releaseActor = releaseActor } func (a *LeaseActor) StartChecking() error { @@ -113,15 +113,7 @@ func (a *LeaseActor) Serve() error { // TODO 可以考虑打个日志 logger.Std.Infof("lock request %s is timeout, will release it", reqID) - // TODO 可以考虑让超时时间可配置 - ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) - err := a.mainActor.Release(ctx, reqID) - cancel() - if err == nil { - delete(a.leases, reqID) - } else { - logger.Std.Warnf("releasing lock request: %s", err.Error()) - } + a.releaseActor.DelayRelease([]string{reqID}) } } diff --git a/pkgs/distlock/service/internal/main_actor.go b/pkgs/distlock/service/internal/main_actor.go deleted file mode 100644 index c8fbff7..0000000 --- a/pkgs/distlock/service/internal/main_actor.go +++ /dev/null @@ -1,313 +0,0 @@ -package internal - -import ( - "context" - "fmt" - "strconv" - - "gitlink.org.cn/cloudream/common/pkgs/actor" - "gitlink.org.cn/cloudream/common/pkgs/distlock" - "gitlink.org.cn/cloudream/common/utils/serder" - clientv3 "go.etcd.io/etcd/client/v3" - "go.etcd.io/etcd/client/v3/clientv3util" - "go.etcd.io/etcd/client/v3/concurrency" -) - -const ( - EtcdLockRequestData = "/distlock/lockRequest/data" - EtcdLockRequestIndex = "/distlock/lockRequest/index" - EtcdLockRequestLock = "/distlock/lockRequest/lock" -) - -type lockData struct { - Path []string `json:"path"` - Name string `json:"name"` - Target string `json:"target"` -} - -type acquireManyResult struct { - IsTried bool - RequestID string - Err error -} - -type LockRequestData struct { - ID string `json:"id"` - Locks []lockData `json:"locks"` -} - -type MainActor struct { - cfg *distlock.Config - etcdCli *clientv3.Client - - commandChan *actor.CommandChannel - - providersActor *ProvidersActor - - lockRequestLeaseID clientv3.LeaseID -} - -func NewMainActor(cfg *distlock.Config, etcdCli *clientv3.Client) *MainActor { - return &MainActor{ - cfg: cfg, - etcdCli: etcdCli, - commandChan: actor.NewCommandChannel(), - } -} - -func (a *MainActor) Init(providersActor *ProvidersActor) { - a.providersActor = providersActor -} - -// Acquire 请求一批锁。成功后返回锁请求ID -func (a *MainActor) Acquire(ctx context.Context, req distlock.LockRequest) (reqID string, err error) { - rets, err := a.AcquireMany(ctx, []distlock.LockRequest{req}) - if err != nil { - return "", err - } - - if rets[0].Err != nil { - return "", rets[0].Err - } - - return rets[0].RequestID, nil -} - -// AcquireAny 尝试多个锁请求。目前的实现会在第一个获取成功后就直接返回 -func (a *MainActor) AcquireMany(ctx context.Context, reqs []distlock.LockRequest) (rets []acquireManyResult, err error) { - return actor.WaitValue(context.TODO(), a.commandChan, func() ([]acquireManyResult, error) { - // TODO 根据不同的错误设置不同的错误类型，方便上层进行后续处理 - unlock, err := a.acquireEtcdRequestDataLock(ctx) - if err != nil { - return nil, fmt.Errorf("acquire etcd request data lock failed, err: %w", err) - } - defer unlock() - - index, err := a.getEtcdLockRequestIndex(ctx) - if err != nil { - return nil, err - } - - // 等待本地状态同步到最新 - // TODO 配置等待时间 - err = a.providersActor.WaitIndexUpdated(ctx, index) - if err != nil { - return nil, err - } - - rets := make([]acquireManyResult, len(reqs)) - for i := 0; i < len(reqs); i++ { - // 测试锁，并获得锁数据 - reqData, err := a.providersActor.TestLockRequestAndMakeData(reqs[i]) - if err == nil { - nextIndexStr := strconv.FormatInt(index+1, 10) - reqData.ID = nextIndexStr - - // 锁成功，提交锁数据 - err := a.submitLockRequest(ctx, reqData) - - rets[i] = acquireManyResult{ - IsTried: true, - RequestID: nextIndexStr, - Err: err, - } - - break - - } else { - rets[i] = acquireManyResult{ - IsTried: true, - Err: err, - } - } - } - - return rets, nil - }) -} - -func (a *MainActor) submitLockRequest(ctx context.Context, reqData LockRequestData) error { - reqBytes, err := serder.ObjectToJSON(reqData) - if err != nil { - return fmt.Errorf("serialize lock request data failed, err: %w", err) - } - - var etcdOps []clientv3.Op - if a.cfg.SubmitLockRequestWithoutLease { - etcdOps = []clientv3.Op{ - clientv3.OpPut(EtcdLockRequestIndex, reqData.ID), - clientv3.OpPut(makeEtcdLockRequestKey(reqData.ID), string(reqBytes)), - } - - } else { - etcdOps = []clientv3.Op{ - clientv3.OpPut(EtcdLockRequestIndex, reqData.ID), - // 归属到当前连接的租约，在当前连接断开后，能自动解锁 - // TODO 不能直接给RequestData上租约，因为如果在别的服务已经获取到锁的情况下， - // 如果当前服务崩溃，删除消息会立刻发送出去，这就破坏了锁的约定（在锁定期间其他服务不能修改数据） - clientv3.OpPut(makeEtcdLockRequestKey(reqData.ID), string(reqBytes)), //, clientv3.WithLease(a.lockRequestLeaseID)), - } - } - txResp, err := a.etcdCli.Txn(ctx).Then(etcdOps...).Commit() - if err != nil { - return fmt.Errorf("submit lock request data failed, err: %w", err) - } - if !txResp.Succeeded { - return fmt.Errorf("submit lock request data failed for lock request data index changed") - } - - return nil -} - -// Release 释放锁 -func (a *MainActor) Release(ctx context.Context, reqID string) error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - // TODO 根据不同的错误设置不同的错误类型，方便上层进行后续处理 - unlock, err := a.acquireEtcdRequestDataLock(ctx) - if err != nil { - return fmt.Errorf("acquire etcd request data lock failed, err: %w", err) - } - defer unlock() - - index, err := a.getEtcdLockRequestIndex(ctx) - if err != nil { - return err - } - - lockReqKey := makeEtcdLockRequestKey(reqID) - - txResp, err := a.etcdCli.Txn(ctx). - If(clientv3util.KeyExists(lockReqKey)). - Then(clientv3.OpDelete(lockReqKey), clientv3.OpPut(EtcdLockRequestIndex, strconv.FormatInt(index+1, 10))).Commit() - if err != nil { - return fmt.Errorf("updating lock request data index: %w", err) - } - if !txResp.Succeeded { - return fmt.Errorf("updating lock request data failed") - } - - return nil - }) -} - -func (a *MainActor) acquireEtcdRequestDataLock(ctx context.Context) (unlock func(), err error) { - lease, err := a.etcdCli.Grant(context.Background(), a.cfg.EtcdLockLeaseTimeSec) - if err != nil { - return nil, fmt.Errorf("grant lease failed, err: %w", err) - } - - session, err := concurrency.NewSession(a.etcdCli, concurrency.WithLease(lease.ID)) - if err != nil { - return nil, fmt.Errorf("new session failed, err: %w", err) - } - - mutex := concurrency.NewMutex(session, EtcdLockRequestLock) - - err = mutex.Lock(ctx) - if err != nil { - session.Close() - return nil, fmt.Errorf("acquire lock failed, err: %w", err) - } - - return func() { - mutex.Unlock(context.Background()) - session.Close() - }, nil -} - -func (a *MainActor) getEtcdLockRequestIndex(ctx context.Context) (int64, error) { - indexKv, err := a.etcdCli.Get(ctx, EtcdLockRequestIndex) - if err != nil { - return 0, fmt.Errorf("get lock request index failed, err: %w", err) - } - - if len(indexKv.Kvs) == 0 { - return 0, nil - } - - index, err := strconv.ParseInt(string(indexKv.Kvs[0].Value), 0, 64) - if err != nil { - return 0, fmt.Errorf("parse lock request index failed, err: %w", err) - } - - return index, nil -} - -func (a *MainActor) ReloadEtcdData() error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - // 使用事务一次性获取index和锁数据，就不需要加全局锁了 - txResp, err := a.etcdCli.Txn(context.Background()). - Then( - clientv3.OpGet(EtcdLockRequestIndex), - clientv3.OpGet(EtcdLockRequestData, clientv3.WithPrefix()), - ). - Commit() - if err != nil { - return fmt.Errorf("get etcd data failed, err: %w", err) - } - if !txResp.Succeeded { - return fmt.Errorf("get etcd data failed") - } - - indexKvs := txResp.Responses[0].GetResponseRange().Kvs - lockKvs := txResp.Responses[1].GetResponseRange().Kvs - - var index int64 - var reqData []LockRequestData - - // 解析Index - if len(indexKvs) > 0 { - val, err := strconv.ParseInt(string(indexKvs[0].Value), 0, 64) - if err != nil { - return fmt.Errorf("parse lock request index failed, err: %w", err) - } - index = val - - } else { - index = 0 - } - - // 解析锁请求数据 - for _, kv := range lockKvs { - var req LockRequestData - err := serder.JSONToObject(kv.Value, &req) - if err != nil { - return fmt.Errorf("parse lock request data failed, err: %w", err) - } - - reqData = append(reqData, req) - } - - err = a.providersActor.ResetState(index, reqData) - if err != nil { - return fmt.Errorf("reset lock providers state failed, err: %w", err) - } - - return nil - }) -} - -func (a *MainActor) Serve() error { - lease, err := a.etcdCli.Grant(context.Background(), a.cfg.EtcdLockLeaseTimeSec) - if err != nil { - return fmt.Errorf("grant lease failed, err: %w", err) - } - a.lockRequestLeaseID = lease.ID - - cmdChan := a.commandChan.BeginChanReceive() - defer a.commandChan.CloseChanReceive() - - for { - select { - case cmd, ok := <-cmdChan: - if !ok { - return fmt.Errorf("command channel closed") - } - - // TODO Actor启动时，如果第一个调用的是Acquire，那么就会在Acquire中等待本地锁数据同步到最新。 - // 此时命令的执行也会被阻塞，导致ReloadEtcdData命令无法执行，因此产生死锁，最后Acquire超时失败。 - // 此处暂时使用单独的goroutine的来执行命令，避免阻塞。 - go cmd() - } - } -} diff --git a/pkgs/distlock/service/internal/providers_actor.go b/pkgs/distlock/service/internal/providers_actor.go index fa56a5b..c81d75c 100644 --- a/pkgs/distlock/service/internal/providers_actor.go +++ b/pkgs/distlock/service/internal/providers_actor.go @@ -7,6 +7,7 @@ import ( "gitlink.org.cn/cloudream/common/pkgs/actor" "gitlink.org.cn/cloudream/common/pkgs/distlock" "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/pkgs/trie" ) @@ -56,30 +57,31 @@ func (a *ProvidersActor) WaitIndexUpdated(ctx context.Context, index int64) erro return fut.Wait(ctx) } -func (a *ProvidersActor) ApplyLockRequestEvents(events []LockRequestEvent) error { - return actor.Wait(context.TODO(), a.commandChan, func() error { +func (a *ProvidersActor) ApplyLockRequestEvents(events []LockRequestEvent) { + a.commandChan.Send(func() { for _, op := range events { if op.IsLocking { err := a.lockLockRequest(op.Data) if err != nil { - return fmt.Errorf("lock by lock request data failed, err: %w", err) + // TODO 发生这种错误需要重新加载全量状态，下同 + logger.Std.Warnf("applying locking event: %s", err.Error()) + return } } else { err := a.unlockLockRequest(op.Data) if err != nil { - return fmt.Errorf("unlock by lock request data failed, err: %w", err) + logger.Std.Warnf("applying unlocking event: %s", err.Error()) + return } } - } - // 处理了多少事件，Index就往后移动多少个 - a.localLockReqIndex += int64(len(events)) + // 处理了多少事件，Index就往后移动多少个 + a.localLockReqIndex++ + } // 检查是否有等待同步进度的需求 - a.checkIndexWaiter() - - return nil + a.wakeUpIndexWaiter() }) } @@ -181,13 +183,13 @@ func (a *ProvidersActor) ResetState(index int64, lockRequestData []LockRequestDa a.localLockReqIndex = index // 检查是否有等待同步进度的需求 - a.checkIndexWaiter() + a.wakeUpIndexWaiter() return nil }) } -func (a *ProvidersActor) checkIndexWaiter() { +func (a *ProvidersActor) wakeUpIndexWaiter() { var resetWaiters []indexWaiter for _, waiter := range a.indexWaiters { if waiter.Index <= a.localLockReqIndex { diff --git a/pkgs/distlock/service/internal/release_actor.go b/pkgs/distlock/service/internal/release_actor.go new file mode 100644 index 0000000..c633d53 --- /dev/null +++ b/pkgs/distlock/service/internal/release_actor.go @@ -0,0 +1,144 @@ +package internal + +import ( + "context" + "fmt" + "math/rand" + "strconv" + "sync" + "time" + + "gitlink.org.cn/cloudream/common/pkgs/distlock" + "gitlink.org.cn/cloudream/common/pkgs/logger" + clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/client/v3/clientv3util" +) + +const ( + DefaultMaxReleaseingDelayMs = 4000 + BaseReleaseingDelayMs = 1000 +) + +type ReleaseActor struct { + cfg *distlock.Config + etcdCli *clientv3.Client + + releasingLockRequestIDs map[string]bool + timer *time.Timer + timerSetuped bool + lock sync.Mutex +} + +func NewReleaseActor(cfg *distlock.Config, etcdCli *clientv3.Client) *ReleaseActor { + return &ReleaseActor{ + cfg: cfg, + etcdCli: etcdCli, + releasingLockRequestIDs: make(map[string]bool), + } +} + +// 立刻尝试释放这些锁。一般用于在用户主动释放了一个锁之后 +func (a *ReleaseActor) Release(reqIDs []string) { + a.lock.Lock() + defer a.lock.Unlock() + + for _, id := range reqIDs { + a.releasingLockRequestIDs[id] = true + } + + // TODO 处理错误 + err := a.doReleasing() + if err != nil { + logger.Std.Debugf("doing releasing: %s", err.Error()) + } + + a.setupTimer() +} + +// 延迟释放锁。一般用于清理崩溃的锁服务遗留下来的锁 +func (a *ReleaseActor) DelayRelease(reqIDs []string) { + a.lock.Lock() + defer a.lock.Unlock() + + for _, id := range reqIDs { + a.releasingLockRequestIDs[id] = true + } + + a.setupTimer() +} + +func (a *ReleaseActor) doReleasing() error { + ctx := context.TODO() + + if len(a.releasingLockRequestIDs) == 0 { + return nil + } + + // TODO 根据不同的错误设置不同的错误类型，方便上层进行后续处理 + unlock, err := acquireEtcdRequestDataLock(ctx, a.etcdCli, a.cfg.EtcdLockLeaseTimeSec) + if err != nil { + return fmt.Errorf("acquire etcd request data lock failed, err: %w", err) + } + defer unlock() + + index, err := getEtcdLockRequestIndex(ctx, a.etcdCli) + if err != nil { + return err + } + + // TODO 可以考虑优化成一次性删除多个锁 + for id := range a.releasingLockRequestIDs { + lockReqKey := makeEtcdLockRequestKey(id) + + txResp, err := a.etcdCli.Txn(ctx). + If(clientv3util.KeyExists(lockReqKey)). + Then(clientv3.OpDelete(lockReqKey), clientv3.OpPut(EtcdLockRequestIndex, strconv.FormatInt(index+1, 10))).Commit() + if err != nil { + return fmt.Errorf("updating lock request data: %w", err) + } + // 只有确实删除了锁数据，才更新index + if txResp.Succeeded { + index++ + } + delete(a.releasingLockRequestIDs, id) + } + + return nil +} + +func (a *ReleaseActor) setupTimer() { + if len(a.releasingLockRequestIDs) == 0 { + return + } + + if a.timerSetuped { + return + } + a.timerSetuped = true + + delay := int64(0) + if a.cfg.RandomReleasingDelayMs == 0 { + delay = rand.Int63n(DefaultMaxReleaseingDelayMs) + } else { + delay = rand.Int63n(a.cfg.RandomReleasingDelayMs) + } + + if a.timer == nil { + a.timer = time.NewTimer(time.Duration(delay+BaseReleaseingDelayMs) * time.Millisecond) + } else { + a.timer.Reset(time.Duration(delay+BaseReleaseingDelayMs) * time.Millisecond) + } + + go func() { + <-a.timer.C + a.timerSetuped = false + + // TODO 处理错误 + err := a.doReleasing() + if err != nil { + logger.Std.Debugf("doing releasing: %s", err.Error()) + } + + a.setupTimer() + }() +} diff --git a/pkgs/distlock/service/internal/retry_actor.go b/pkgs/distlock/service/internal/retry_actor.go deleted file mode 100644 index 6f6e304..0000000 --- a/pkgs/distlock/service/internal/retry_actor.go +++ /dev/null @@ -1,128 +0,0 @@ -package internal - -import ( - "context" - "fmt" - - "github.com/samber/lo" - "gitlink.org.cn/cloudream/common/pkgs/actor" - "gitlink.org.cn/cloudream/common/pkgs/distlock" - "gitlink.org.cn/cloudream/common/pkgs/future" - "gitlink.org.cn/cloudream/common/pkgs/logger" - mylo "gitlink.org.cn/cloudream/common/utils/lo" -) - -type retryInfo struct { - Callback *future.SetValueFuture[string] - LastErr error -} - -type RetryActor struct { - retrys []distlock.LockRequest - retryInfos []*retryInfo - - commandChan *actor.CommandChannel - - mainActor *MainActor -} - -func NewRetryActor() *RetryActor { - return &RetryActor{ - commandChan: actor.NewCommandChannel(), - } -} - -func (a *RetryActor) Init(mainActor *MainActor) { - a.mainActor = mainActor -} - -func (a *RetryActor) Retry(ctx context.Context, req distlock.LockRequest, lastErr error) (future.ValueFuture[string], error) { - fut := future.NewSetValue[string]() - - var info *retryInfo - err := actor.Wait(ctx, a.commandChan, func() error { - a.retrys = append(a.retrys, req) - info = &retryInfo{ - Callback: fut, - LastErr: lastErr, - } - a.retryInfos = append(a.retryInfos, info) - return nil - }) - if err != nil { - return nil, err - } - - go func() { - <-ctx.Done() - a.commandChan.Send(func() { - // 由于只可能在cmd中修改future状态，所以此处的IsComplete判断可以作为后续操作的依据 - if fut.IsComplete() { - return - } - - index := lo.IndexOf(a.retryInfos, info) - if index == -1 { - return - } - - a.retryInfos[index].Callback.SetError(a.retryInfos[index].LastErr) - - a.retrys = mylo.RemoveAt(a.retrys, index) - a.retryInfos = mylo.RemoveAt(a.retryInfos, index) - }) - }() - - return fut, nil -} - -func (a *RetryActor) OnLocalStateUpdated() { - a.commandChan.Send(func() { - if len(a.retrys) == 0 { - return - } - - rets, err := a.mainActor.AcquireMany(context.Background(), a.retrys) - if err != nil { - // TODO 处理错误 - logger.Std.Warnf("acquire many lock requests failed, err: %s", err.Error()) - return - } - - // 根据尝试的结果更新状态 - delCnt := 0 - for i, ret := range rets { - a.retrys[i-delCnt] = a.retrys[i] - a.retryInfos[i-delCnt] = a.retryInfos[i] - - if !ret.IsTried { - continue - } - - if ret.Err != nil { - a.retryInfos[i].LastErr = ret.Err - } else { - a.retryInfos[i].Callback.SetValue(ret.RequestID) - delCnt++ - } - } - a.retrys = a.retrys[:len(a.retrys)-delCnt] - a.retryInfos = a.retryInfos[:len(a.retryInfos)-delCnt] - }) -} - -func (a *RetryActor) Serve() error { - cmdChan := a.commandChan.BeginChanReceive() - defer a.commandChan.CloseChanReceive() - - for { - select { - case cmd, ok := <-cmdChan: - if !ok { - return fmt.Errorf("command channel closed") - } - - cmd() - } - } -} diff --git a/pkgs/distlock/service/internal/watch_etcd_actor.go b/pkgs/distlock/service/internal/watch_etcd_actor.go index 5d0530c..2d0b791 100644 --- a/pkgs/distlock/service/internal/watch_etcd_actor.go +++ b/pkgs/distlock/service/internal/watch_etcd_actor.go @@ -37,9 +37,9 @@ func NewWatchEtcdActor(etcdCli *clientv3.Client) *WatchEtcdActor { func (a *WatchEtcdActor) Init() { } -func (a *WatchEtcdActor) StartWatching() error { +func (a *WatchEtcdActor) StartWatching(revision int64) error { return actor.Wait(context.TODO(), a.commandChan, func() error { - a.watchChan = a.etcdCli.Watch(context.Background(), EtcdLockRequestData, clientv3.WithPrefix(), clientv3.WithPrevKV()) + a.watchChan = a.etcdCli.Watch(context.Background(), EtcdLockRequestData, clientv3.WithPrefix(), clientv3.WithPrevKV(), clientv3.WithRev(revision)) return nil }) } diff --git a/pkgs/distlock/service/mutex.go b/pkgs/distlock/service/mutex.go index b386ff1..abba728 100644 --- a/pkgs/distlock/service/mutex.go +++ b/pkgs/distlock/service/mutex.go @@ -25,6 +25,6 @@ func (m *Mutex) Lock() error { return nil } -func (m *Mutex) Unlock() error { - return m.svc.Release(m.lockReqID) +func (m *Mutex) Unlock() { + m.svc.Release(m.lockReqID) } diff --git a/pkgs/distlock/service/service.go b/pkgs/distlock/service/service.go index a655a7d..8cb8e34 100644 --- a/pkgs/distlock/service/service.go +++ b/pkgs/distlock/service/service.go @@ -3,11 +3,13 @@ package service import ( "context" "fmt" + "strconv" "time" "gitlink.org.cn/cloudream/common/pkgs/distlock" "gitlink.org.cn/cloudream/common/pkgs/distlock/service/internal" "gitlink.org.cn/cloudream/common/pkgs/logger" + "gitlink.org.cn/cloudream/common/utils/serder" clientv3 "go.etcd.io/etcd/client/v3" ) @@ -46,11 +48,11 @@ type Service struct { cfg *distlock.Config etcdCli *clientv3.Client - mainActor *internal.MainActor + acquireActor *internal.AcquireActor + releaseActor *internal.ReleaseActor providersActor *internal.ProvidersActor watchEtcdActor *internal.WatchEtcdActor leaseActor *internal.LeaseActor - retryActor *internal.RetryActor lockReqEventWatcher internal.LockRequestEventWatcher } @@ -66,17 +68,16 @@ func NewService(cfg *distlock.Config, initProvs []PathProvider) (*Service, error return nil, fmt.Errorf("new etcd client failed, err: %w", err) } - mainActor := internal.NewMainActor(cfg, etcdCli) + acquireActor := internal.NewAcquireActor(cfg, etcdCli) + releaseActor := internal.NewReleaseActor(cfg, etcdCli) providersActor := internal.NewProvidersActor() watchEtcdActor := internal.NewWatchEtcdActor(etcdCli) leaseActor := internal.NewLeaseActor() - retryActor := internal.NewRetryActor() - mainActor.Init(providersActor) + acquireActor.Init(providersActor) providersActor.Init() watchEtcdActor.Init() - leaseActor.Init(mainActor) - retryActor.Init(mainActor) + leaseActor.Init(releaseActor) for _, prov := range initProvs { providersActor.AddProvider(prov.Provider, prov.Path...) @@ -85,11 +86,11 @@ func NewService(cfg *distlock.Config, initProvs []PathProvider) (*Service, error return &Service{ cfg: cfg, etcdCli: etcdCli, - mainActor: mainActor, + acquireActor: acquireActor, + releaseActor: releaseActor, providersActor: providersActor, watchEtcdActor: watchEtcdActor, leaseActor: leaseActor, - retryActor: retryActor, }, nil } @@ -109,18 +110,9 @@ func (svc *Service) Acquire(req distlock.LockRequest, opts ...AcquireOptionFn) ( defer cancel() } - reqID, err := svc.mainActor.Acquire(ctx, req) + reqID, err := svc.acquireActor.Acquire(ctx, req) if err != nil { - fut, err := svc.retryActor.Retry(ctx, req, err) - if err != nil { - return "", fmt.Errorf("retrying failed, err: %w", err) - } - - // Retry 如果超时，Retry内部会设置fut为Failed，所以这里可以用Background无限等待 - reqID, err = fut.WaitValue(context.Background()) - if err != nil { - return "", err - } + return "", err } if opt.Lease > 0 { @@ -140,16 +132,14 @@ func (svc *Service) Renew(reqID string) error { } // Release 释放锁 -func (svc *Service) Release(reqID string) error { - err := svc.mainActor.Release(context.TODO(), reqID) +func (svc *Service) Release(reqID string) { + svc.releaseActor.Release([]string{reqID}) // TODO 不影响结果，但考虑打日志 - err2 := svc.leaseActor.Remove(reqID) - if err2 != nil { - logger.Std.Warnf("removing lease: %s", err2.Error()) + err := svc.leaseActor.Remove(reqID) + if err != nil { + logger.Std.Warnf("removing lease: %s", err.Error()) } - - return err } func (svc *Service) Serve() error { @@ -174,14 +164,6 @@ func (svc *Service) Serve() error { } }() - go func() { - // TODO 处理错误 - err := svc.mainActor.Serve() - if err != nil { - logger.Std.Warnf("serving main actor failed, err: %s", err.Error()) - } - }() - go func() { // TODO 处理错误 err := svc.leaseActor.Serve() @@ -190,23 +172,15 @@ func (svc *Service) Serve() error { } }() - go func() { - // TODO 处理错误 - err := svc.retryActor.Serve() - if err != nil { - logger.Std.Warnf("serving retry actor failed, err: %s", err.Error()) - } - }() - - err := svc.mainActor.ReloadEtcdData() + revision, err := svc.loadState() if err != nil { // TODO 关闭其他的Actor，或者更好的错误处理方式 return fmt.Errorf("init data failed, err: %w", err) } svc.lockReqEventWatcher.OnEvent = func(events []internal.LockRequestEvent) { + svc.acquireActor.TryAcquireNow() svc.providersActor.ApplyLockRequestEvents(events) - svc.retryActor.OnLocalStateUpdated() } err = svc.watchEtcdActor.AddEventWatcher(&svc.lockReqEventWatcher) if err != nil { @@ -214,7 +188,7 @@ func (svc *Service) Serve() error { return fmt.Errorf("add lock request event watcher failed, err: %w", err) } - err = svc.watchEtcdActor.StartWatching() + err = svc.watchEtcdActor.StartWatching(revision) if err != nil { // TODO 关闭其他的Actor，或者更好的错误处理方式 return fmt.Errorf("start watching etcd failed, err: %w", err) @@ -232,3 +206,52 @@ func (svc *Service) Serve() error { return nil } + +func (svc *Service) loadState() (int64, error) { + // 使用事务一次性获取index和锁数据，就不需要加全局锁了 + txResp, err := svc.etcdCli.Txn(context.Background()). + Then( + clientv3.OpGet(internal.EtcdLockRequestIndex), + clientv3.OpGet(internal.EtcdLockRequestData, clientv3.WithPrefix()), + ). + Commit() + if err != nil { + return 0, fmt.Errorf("get etcd data failed, err: %w", err) + } + + indexKvs := txResp.Responses[0].GetResponseRange().Kvs + lockKvs := txResp.Responses[1].GetResponseRange().Kvs + + var index int64 + var reqData []internal.LockRequestData + + // 解析Index + if len(indexKvs) > 0 { + val, err := strconv.ParseInt(string(indexKvs[0].Value), 0, 64) + if err != nil { + return 0, fmt.Errorf("parse lock request index failed, err: %w", err) + } + index = val + + } else { + index = 0 + } + + // 解析锁请求数据 + for _, kv := range lockKvs { + var req internal.LockRequestData + err := serder.JSONToObject(kv.Value, &req) + if err != nil { + return 0, fmt.Errorf("parse lock request data failed, err: %w", err) + } + + reqData = append(reqData, req) + } + + err = svc.providersActor.ResetState(index, reqData) + if err != nil { + return 0, fmt.Errorf("reset lock providers state failed, err: %w", err) + } + + return txResp.Header.Revision, nil +} diff --git a/pkgs/future/set_value_future.go b/pkgs/future/set_value_future.go index fad7d36..4aea9e8 100644 --- a/pkgs/future/set_value_future.go +++ b/pkgs/future/set_value_future.go @@ -48,6 +48,8 @@ func (f *SetValueFuture[T]) IsComplete() bool { return f.isCompleted } +// 等待直到Complete或者ctx被取消。 +// 注：返回ErrContextCancelled不代表产生结果的过程没有执行过，甚至不代表Future没有Complete func (f *SetValueFuture[T]) Wait(ctx context.Context) error { select { case <-f.completeChan: From ede0aa5f14cb67ca4bcd1886c9c2a280208f91fa Mon Sep 17 00:00:00 2001 From: Sydonian <794346190@qq.com> Date: Fri, 27 Oct 2023 09:19:17 +0800 Subject: [PATCH 3/6] =?UTF-8?q?=E8=B0=83=E6=95=B4=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkgs/distlock/distlock.go | 38 +++----------- .../{service => }/internal/acquire_actor.go | 25 ++------- pkgs/distlock/{ => internal}/config.go | 2 +- .../{service => }/internal/lease_actor.go | 0 pkgs/distlock/internal/models.go | 52 +++++++++++++++++++ .../{service => }/internal/providers_actor.go | 13 +++-- .../{service => }/internal/release_actor.go | 5 +- pkgs/distlock/{service => }/internal/utils.go | 0 .../{service => }/internal/utils_test.go | 0 .../internal/watch_etcd_actor.go | 0 pkgs/distlock/{service => }/mutex.go | 8 +-- pkgs/distlock/{service => }/service.go | 17 +++--- 12 files changed, 85 insertions(+), 75 deletions(-) rename pkgs/distlock/{service => }/internal/acquire_actor.go (88%) rename pkgs/distlock/{ => internal}/config.go (97%) rename pkgs/distlock/{service => }/internal/lease_actor.go (100%) create mode 100644 pkgs/distlock/internal/models.go rename pkgs/distlock/{service => }/internal/providers_actor.go (92%) rename pkgs/distlock/{service => }/internal/release_actor.go (95%) rename pkgs/distlock/{service => }/internal/utils.go (100%) rename pkgs/distlock/{service => }/internal/utils_test.go (100%) rename pkgs/distlock/{service => }/internal/watch_etcd_actor.go (100%) rename pkgs/distlock/{service => }/mutex.go (64%) rename pkgs/distlock/{service => }/service.go (92%) diff --git a/pkgs/distlock/distlock.go b/pkgs/distlock/distlock.go index 93091bb..2972c74 100644 --- a/pkgs/distlock/distlock.go +++ b/pkgs/distlock/distlock.go @@ -1,40 +1,18 @@ package distlock -import "fmt" +import ( + "fmt" -type Lock struct { - Path []string // 锁路径，存储的是路径的每一部分 - Name string // 锁名 - Target any // 锁对象，由具体的Provider去解析 -} - -type LockRequest struct { - Locks []Lock -} - -func (b *LockRequest) Add(lock Lock) { - b.Locks = append(b.Locks, lock) -} - -type LockProvider interface { - // CanLock 判断这个锁能否锁定成功 - CanLock(lock Lock) error + "gitlink.org.cn/cloudream/common/pkgs/distlock/internal" +) - // Lock 锁定。由于同一个锁请求内的锁不检查冲突，因此这个函数必须支持有冲突的锁进行锁定。 - Lock(reqID string, lock Lock) error +type Lock = internal.Lock - // 解锁 - Unlock(reqID string, lock Lock) error +type LockRequest = internal.LockRequest - // GetTargetString 将锁对象序列化为字符串，方便存储到ETCD - GetTargetString(target any) (string, error) +type LockProvider = internal.LockProvider - // ParseTargetString 解析字符串格式的锁对象数据 - ParseTargetString(targetStr string) (any, error) - - // Clear 清除内部所有状态 - Clear() -} +type Config = internal.Config type LockTargetBusyError struct { lockName string diff --git a/pkgs/distlock/service/internal/acquire_actor.go b/pkgs/distlock/internal/acquire_actor.go similarity index 88% rename from pkgs/distlock/service/internal/acquire_actor.go rename to pkgs/distlock/internal/acquire_actor.go index e7a2a39..770ad9c 100644 --- a/pkgs/distlock/service/internal/acquire_actor.go +++ b/pkgs/distlock/internal/acquire_actor.go @@ -7,7 +7,6 @@ import ( "strconv" "sync" - "gitlink.org.cn/cloudream/common/pkgs/distlock" "gitlink.org.cn/cloudream/common/pkgs/future" "gitlink.org.cn/cloudream/common/pkgs/logger" mylo "gitlink.org.cn/cloudream/common/utils/lo" @@ -18,30 +17,14 @@ import ( var ErrAcquiringTimeout = errors.New("acquiring timeout") -const ( - EtcdLockRequestData = "/distlock/lockRequest/data" - EtcdLockRequestIndex = "/distlock/lockRequest/index" - EtcdLockRequestLock = "/distlock/lockRequest/lock" -) - -type lockData struct { - Path []string `json:"path"` - Name string `json:"name"` - Target string `json:"target"` -} -type LockRequestData struct { - ID string `json:"id"` - Locks []lockData `json:"locks"` -} - type acquireInfo struct { - Request distlock.LockRequest + Request LockRequest Callback *future.SetValueFuture[string] LastErr error } type AcquireActor struct { - cfg *distlock.Config + cfg *Config etcdCli *clientv3.Client providersActor *ProvidersActor @@ -49,7 +32,7 @@ type AcquireActor struct { lock sync.Mutex } -func NewAcquireActor(cfg *distlock.Config, etcdCli *clientv3.Client) *AcquireActor { +func NewAcquireActor(cfg *Config, etcdCli *clientv3.Client) *AcquireActor { return &AcquireActor{ cfg: cfg, etcdCli: etcdCli, @@ -61,7 +44,7 @@ func (a *AcquireActor) Init(providersActor *ProvidersActor) { } // Acquire 请求一批锁。成功后返回锁请求ID -func (a *AcquireActor) Acquire(ctx context.Context, req distlock.LockRequest) (string, error) { +func (a *AcquireActor) Acquire(ctx context.Context, req LockRequest) (string, error) { info := &acquireInfo{ Request: req, Callback: future.NewSetValue[string](), diff --git a/pkgs/distlock/config.go b/pkgs/distlock/internal/config.go similarity index 97% rename from pkgs/distlock/config.go rename to pkgs/distlock/internal/config.go index ef2acb1..b6219f3 100644 --- a/pkgs/distlock/config.go +++ b/pkgs/distlock/internal/config.go @@ -1,4 +1,4 @@ -package distlock +package internal type Config struct { EtcdAddress string `json:"etcdAddress"` diff --git a/pkgs/distlock/service/internal/lease_actor.go b/pkgs/distlock/internal/lease_actor.go similarity index 100% rename from pkgs/distlock/service/internal/lease_actor.go rename to pkgs/distlock/internal/lease_actor.go diff --git a/pkgs/distlock/internal/models.go b/pkgs/distlock/internal/models.go new file mode 100644 index 0000000..b0c7eeb --- /dev/null +++ b/pkgs/distlock/internal/models.go @@ -0,0 +1,52 @@ +package internal + +const ( + EtcdLockRequestData = "/distlock/lockRequest/data" + EtcdLockRequestIndex = "/distlock/lockRequest/index" + EtcdLockRequestLock = "/distlock/lockRequest/lock" +) + +type Lock struct { + Path []string // 锁路径，存储的是路径的每一部分 + Name string // 锁名 + Target any // 锁对象，由具体的Provider去解析 +} + +type LockRequest struct { + Locks []Lock +} + +func (b *LockRequest) Add(lock Lock) { + b.Locks = append(b.Locks, lock) +} + +type LockProvider interface { + // CanLock 判断这个锁能否锁定成功 + CanLock(lock Lock) error + + // Lock 锁定。由于同一个锁请求内的锁不检查冲突，因此这个函数必须支持有冲突的锁进行锁定。 + Lock(reqID string, lock Lock) error + + // 解锁 + Unlock(reqID string, lock Lock) error + + // GetTargetString 将锁对象序列化为字符串，方便存储到ETCD + GetTargetString(target any) (string, error) + + // ParseTargetString 解析字符串格式的锁对象数据 + ParseTargetString(targetStr string) (any, error) + + // Clear 清除内部所有状态 + Clear() +} + +type lockData struct { + Path []string `json:"path"` + Name string `json:"name"` + Target string `json:"target"` +} + +type LockRequestData struct { + ID string `json:"id"` + Locks []lockData `json:"locks"` +} diff --git a/pkgs/distlock/service/internal/providers_actor.go b/pkgs/distlock/internal/providers_actor.go similarity index 92% rename from pkgs/distlock/service/internal/providers_actor.go rename to pkgs/distlock/internal/providers_actor.go index c81d75c..23be57e 100644 --- a/pkgs/distlock/service/internal/providers_actor.go +++ b/pkgs/distlock/internal/providers_actor.go @@ -5,7 +5,6 @@ import ( "fmt" "gitlink.org.cn/cloudream/common/pkgs/actor" - "gitlink.org.cn/cloudream/common/pkgs/distlock" "gitlink.org.cn/cloudream/common/pkgs/future" "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/pkgs/trie" @@ -18,8 +17,8 @@ type indexWaiter struct { type ProvidersActor struct { localLockReqIndex int64 - provdersTrie trie.Trie[distlock.LockProvider] - allProviders []distlock.LockProvider + provdersTrie trie.Trie[LockProvider] + allProviders []LockProvider indexWaiters []indexWaiter @@ -32,7 +31,7 @@ func NewProvidersActor() *ProvidersActor { } } -func (a *ProvidersActor) AddProvider(prov distlock.LockProvider, path ...any) { +func (a *ProvidersActor) AddProvider(prov LockProvider, path ...any) { a.provdersTrie.Create(path).Value = prov a.allProviders = append(a.allProviders, prov) } @@ -97,7 +96,7 @@ func (svc *ProvidersActor) lockLockRequest(reqData LockRequestData) error { return fmt.Errorf("parse target data failed, err: %w", err) } - err = node.Value.Lock(reqData.ID, distlock.Lock{ + err = node.Value.Lock(reqData.ID, Lock{ Path: lockData.Path, Name: lockData.Name, Target: target, @@ -121,7 +120,7 @@ func (svc *ProvidersActor) unlockLockRequest(reqData LockRequestData) error { return fmt.Errorf("parse target data failed, err: %w", err) } - err = node.Value.Unlock(reqData.ID, distlock.Lock{ + err = node.Value.Unlock(reqData.ID, Lock{ Path: lockData.Path, Name: lockData.Name, Target: target, @@ -135,7 +134,7 @@ func (svc *ProvidersActor) unlockLockRequest(reqData LockRequestData) error { // TestLockRequestAndMakeData 判断锁能否锁成功，并生成锁数据的字符串表示。注：不会生成请求ID。 // 在检查单个锁是否能上锁时，不会考虑同一个锁请求中的其他的锁影响。简单来说，就是同一个请求中的锁可以互相冲突。 -func (a *ProvidersActor) TestLockRequestAndMakeData(req distlock.LockRequest) (LockRequestData, error) { +func (a *ProvidersActor) TestLockRequestAndMakeData(req LockRequest) (LockRequestData, error) { return actor.WaitValue(context.TODO(), a.commandChan, func() (LockRequestData, error) { reqData := LockRequestData{} diff --git a/pkgs/distlock/service/internal/release_actor.go b/pkgs/distlock/internal/release_actor.go similarity index 95% rename from pkgs/distlock/service/internal/release_actor.go rename to pkgs/distlock/internal/release_actor.go index c633d53..4105350 100644 --- a/pkgs/distlock/service/internal/release_actor.go +++ b/pkgs/distlock/internal/release_actor.go @@ -8,7 +8,6 @@ import ( "sync" "time" - "gitlink.org.cn/cloudream/common/pkgs/distlock" "gitlink.org.cn/cloudream/common/pkgs/logger" clientv3 "go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/client/v3/clientv3util" @@ -20,7 +19,7 @@ const ( ) type ReleaseActor struct { - cfg *distlock.Config + cfg *Config etcdCli *clientv3.Client releasingLockRequestIDs map[string]bool @@ -29,7 +28,7 @@ type ReleaseActor struct { lock sync.Mutex } -func NewReleaseActor(cfg *distlock.Config, etcdCli *clientv3.Client) *ReleaseActor { +func NewReleaseActor(cfg *Config, etcdCli *clientv3.Client) *ReleaseActor { return &ReleaseActor{ cfg: cfg, etcdCli: etcdCli, diff --git a/pkgs/distlock/service/internal/utils.go b/pkgs/distlock/internal/utils.go similarity index 100% rename from pkgs/distlock/service/internal/utils.go rename to pkgs/distlock/internal/utils.go diff --git a/pkgs/distlock/service/internal/utils_test.go b/pkgs/distlock/internal/utils_test.go similarity index 100% rename from pkgs/distlock/service/internal/utils_test.go rename to pkgs/distlock/internal/utils_test.go diff --git a/pkgs/distlock/service/internal/watch_etcd_actor.go b/pkgs/distlock/internal/watch_etcd_actor.go similarity index 100% rename from pkgs/distlock/service/internal/watch_etcd_actor.go rename to pkgs/distlock/internal/watch_etcd_actor.go diff --git a/pkgs/distlock/service/mutex.go b/pkgs/distlock/mutex.go similarity index 64% rename from pkgs/distlock/service/mutex.go rename to pkgs/distlock/mutex.go index abba728..a7c4d23 100644 --- a/pkgs/distlock/service/mutex.go +++ b/pkgs/distlock/mutex.go @@ -1,14 +1,14 @@ -package service +package distlock -import "gitlink.org.cn/cloudream/common/pkgs/distlock" +import "gitlink.org.cn/cloudream/common/pkgs/distlock/internal" type Mutex struct { svc *Service - lockReq distlock.LockRequest + lockReq internal.LockRequest lockReqID string } -func NewMutex(svc *Service, lockReq distlock.LockRequest) *Mutex { +func NewMutex(svc *Service, lockReq internal.LockRequest) *Mutex { return &Mutex{ svc: svc, lockReq: lockReq, diff --git a/pkgs/distlock/service/service.go b/pkgs/distlock/service.go similarity index 92% rename from pkgs/distlock/service/service.go rename to pkgs/distlock/service.go index 8cb8e34..6fb6430 100644 --- a/pkgs/distlock/service/service.go +++ b/pkgs/distlock/service.go @@ -1,4 +1,4 @@ -package service +package distlock import ( "context" @@ -6,8 +6,7 @@ import ( "strconv" "time" - "gitlink.org.cn/cloudream/common/pkgs/distlock" - "gitlink.org.cn/cloudream/common/pkgs/distlock/service/internal" + "gitlink.org.cn/cloudream/common/pkgs/distlock/internal" "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/utils/serder" clientv3 "go.etcd.io/etcd/client/v3" @@ -34,10 +33,10 @@ func WithLease(time time.Duration) AcquireOptionFn { type PathProvider struct { Path []any - Provider distlock.LockProvider + Provider internal.LockProvider } -func NewPathProvider(prov distlock.LockProvider, path ...any) PathProvider { +func NewPathProvider(prov internal.LockProvider, path ...any) PathProvider { return PathProvider{ Path: path, Provider: prov, @@ -45,7 +44,7 @@ func NewPathProvider(prov distlock.LockProvider, path ...any) PathProvider { } type Service struct { - cfg *distlock.Config + cfg *internal.Config etcdCli *clientv3.Client acquireActor *internal.AcquireActor @@ -57,7 +56,7 @@ type Service struct { lockReqEventWatcher internal.LockRequestEventWatcher } -func NewService(cfg *distlock.Config, initProvs []PathProvider) (*Service, error) { +func NewService(cfg *internal.Config, initProvs []PathProvider) (*Service, error) { etcdCli, err := clientv3.New(clientv3.Config{ Endpoints: []string{cfg.EtcdAddress}, Username: cfg.EtcdUsername, @@ -95,7 +94,7 @@ func NewService(cfg *distlock.Config, initProvs []PathProvider) (*Service, error } // Acquire 请求一批锁。成功后返回锁请求ID -func (svc *Service) Acquire(req distlock.LockRequest, opts ...AcquireOptionFn) (string, error) { +func (svc *Service) Acquire(req internal.LockRequest, opts ...AcquireOptionFn) (string, error) { var opt = AcquireOption{ Timeout: time.Second * 10, } @@ -145,7 +144,7 @@ func (svc *Service) Release(reqID string) { func (svc *Service) Serve() error { // TODO 需要停止service的方法 // 目前已知问题： - // 1. client退出时直接中断进程，此时RetryActor可能正在进行Retry，于是导致Etcd锁没有解除就退出了进程。 + // 1. client退出时直接中断进程，此时AcquireActor可能正在进行重试，于是导致Etcd锁没有解除就退出了进程。 // 虽然由于租约的存在不会导致系统长期卡死，但会影响client的使用 go func() { From 763bf968ff36054ba426db1f2d362cdac574f140 Mon Sep 17 00:00:00 2001 From: Sydonian <794346190@qq.com> Date: Mon, 30 Oct 2023 10:07:47 +0800 Subject: [PATCH 4/6] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E9=94=81=E6=B3=A8?= =?UTF-8?q?=E5=86=8C=E6=9C=8D=E5=8A=A1=E4=BF=A1=E6=81=AF=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkgs/distlock/internal/acquire_actor.go | 32 +++- pkgs/distlock/internal/lease_actor.go | 11 +- pkgs/distlock/internal/models.go | 36 +++- pkgs/distlock/internal/providers_actor.go | 179 +++++++++---------- pkgs/distlock/internal/release_actor.go | 35 +++- pkgs/distlock/internal/service_info_actor.go | 136 ++++++++++++++ pkgs/distlock/internal/utils.go | 116 ------------ pkgs/distlock/internal/utils_test.go | 61 ------- pkgs/distlock/internal/watch_etcd_actor.go | 164 ++++++++++------- pkgs/distlock/service.go | 142 ++++++++------- 10 files changed, 496 insertions(+), 416 deletions(-) create mode 100644 pkgs/distlock/internal/service_info_actor.go delete mode 100644 pkgs/distlock/internal/utils.go delete mode 100644 pkgs/distlock/internal/utils_test.go diff --git a/pkgs/distlock/internal/acquire_actor.go b/pkgs/distlock/internal/acquire_actor.go index 770ad9c..28e2bb6 100644 --- a/pkgs/distlock/internal/acquire_actor.go +++ b/pkgs/distlock/internal/acquire_actor.go @@ -6,6 +6,7 @@ import ( "fmt" "strconv" "sync" + "time" "gitlink.org.cn/cloudream/common/pkgs/future" "gitlink.org.cn/cloudream/common/pkgs/logger" @@ -28,6 +29,7 @@ type AcquireActor struct { etcdCli *clientv3.Client providersActor *ProvidersActor + serviceID string acquirings []*acquireInfo lock sync.Mutex } @@ -98,6 +100,21 @@ func (a *AcquireActor) TryAcquireNow() { }() } +func (a *AcquireActor) ResetState(serviceID string) { + a.lock.Lock() + defer a.lock.Unlock() + + a.serviceID = serviceID + for _, info := range a.acquirings { + if info.LastErr != nil { + info.Callback.SetError(info.LastErr) + } else { + info.Callback.SetError(ErrAcquiringTimeout) + } + } + a.acquirings = nil +} + func (a *AcquireActor) doAcquiring() error { ctx := context.Background() @@ -126,24 +143,29 @@ func (a *AcquireActor) doAcquiring() error { // TODO 可以考虑一次性获得多个锁 for i := 0; i < len(a.acquirings); i++ { + req := a.acquirings[i] + // 测试锁，并获得锁数据 - reqData, err := a.providersActor.TestLockRequestAndMakeData(a.acquirings[i].Request) + reqData, err := a.providersActor.TestLockRequestAndMakeData(req.Request) if err != nil { - a.acquirings[i].LastErr = err + req.LastErr = err continue } nextIndexStr := strconv.FormatInt(index+1, 10) reqData.ID = nextIndexStr + reqData.SerivceID = a.serviceID + reqData.Reason = req.Request.Reason + reqData.Timestamp = time.Now().Unix() // 锁成功，提交锁数据 err = a.submitLockRequest(ctx, nextIndexStr, reqData) if err != nil { - a.acquirings[i].LastErr = err + req.LastErr = err continue } - a.acquirings[i].Callback.SetValue(reqData.ID) + req.Callback.SetValue(reqData.ID) a.acquirings = mylo.RemoveAt(a.acquirings, i) break } @@ -159,7 +181,7 @@ func (a *AcquireActor) submitLockRequest(ctx context.Context, index string, reqD etcdOps := []clientv3.Op{ clientv3.OpPut(EtcdLockRequestIndex, index), - clientv3.OpPut(makeEtcdLockRequestKey(reqData.ID), string(reqBytes)), + clientv3.OpPut(MakeEtcdLockRequestKey(reqData.ID), string(reqBytes)), } txResp, err := a.etcdCli.Txn(ctx).Then(etcdOps...).Commit() if err != nil { diff --git a/pkgs/distlock/internal/lease_actor.go b/pkgs/distlock/internal/lease_actor.go index 132430a..5b6045f 100644 --- a/pkgs/distlock/internal/lease_actor.go +++ b/pkgs/distlock/internal/lease_actor.go @@ -35,14 +35,14 @@ func (a *LeaseActor) Init(releaseActor *ReleaseActor) { a.releaseActor = releaseActor } -func (a *LeaseActor) StartChecking() error { +func (a *LeaseActor) Start() error { return actor.Wait(context.TODO(), a.commandChan, func() error { a.ticker = time.NewTicker(time.Second) return nil }) } -func (a *LeaseActor) StopChecking() error { +func (a *LeaseActor) Stop() error { return actor.Wait(context.TODO(), a.commandChan, func() error { if a.ticker != nil { a.ticker.Stop() @@ -91,6 +91,13 @@ func (a *LeaseActor) Remove(reqID string) error { }) } +func (a *LeaseActor) ResetState() { + actor.Wait(context.Background(), a.commandChan, func() error { + a.leases = make(map[string]*lockRequestLease) + return nil + }) +} + func (a *LeaseActor) Serve() error { cmdChan := a.commandChan.BeginChanReceive() defer a.commandChan.CloseChanReceive() diff --git a/pkgs/distlock/internal/models.go b/pkgs/distlock/internal/models.go index b0c7eeb..08eb593 100644 --- a/pkgs/distlock/internal/models.go +++ b/pkgs/distlock/internal/models.go @@ -1,9 +1,13 @@ package internal +import "strings" + const ( - EtcdLockRequestData = "/distlock/lockRequest/data" - EtcdLockRequestIndex = "/distlock/lockRequest/index" - EtcdLockRequestLock = "/distlock/lockRequest/lock" + EtcdLockRequestDataPrefix = "/distlock/lockRequest/data" + EtcdLockRequestIndex = "/distlock/lockRequest/index" + EtcdLockRequestLock = "/distlock/lockRequest/lock" + EtcdServiceInfoPrefix = "/distlock/services" + EtcdWatchPrefix = "/distlock" ) type Lock struct { @@ -13,7 +17,8 @@ type Lock struct { } type LockRequest struct { - Locks []Lock + Reason string + Locks []Lock } func (b *LockRequest) Add(lock Lock) { @@ -47,6 +52,25 @@ type lockData struct { } type LockRequestData struct { - ID string `json:"id"` - Locks []lockData `json:"locks"` + ID string `json:"id"` + SerivceID string `json:"serviceID"` + Reason string `json:"reason"` + Timestamp int64 `json:"timestamp"` + Locks []lockData `json:"locks"` +} + +func MakeEtcdLockRequestKey(reqID string) string { + return EtcdLockRequestDataPrefix + "/" + reqID +} + +func GetLockRequestID(key string) string { + return strings.TrimPrefix(key, EtcdLockRequestDataPrefix+"/") +} + +func MakeServiceInfoKey(svcID string) string { + return EtcdServiceInfoPrefix + "/" + svcID +} + +type ServiceInfo struct { + ID string `json:"id"` } diff --git a/pkgs/distlock/internal/providers_actor.go b/pkgs/distlock/internal/providers_actor.go index 23be57e..c9f426c 100644 --- a/pkgs/distlock/internal/providers_actor.go +++ b/pkgs/distlock/internal/providers_actor.go @@ -2,17 +2,20 @@ package internal import ( "context" + "errors" "fmt" + "sync" - "gitlink.org.cn/cloudream/common/pkgs/actor" "gitlink.org.cn/cloudream/common/pkgs/future" "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/pkgs/trie" ) +var ErrWaitIndexUpdateTimeout = errors.New("waitting local index updating timeout") + type indexWaiter struct { - Index int64 - Future *future.SetVoidFuture + Index int64 + Callback *future.SetVoidFuture } type ProvidersActor struct { @@ -21,14 +24,11 @@ type ProvidersActor struct { allProviders []LockProvider indexWaiters []indexWaiter - - commandChan *actor.CommandChannel + lock sync.Mutex } func NewProvidersActor() *ProvidersActor { - return &ProvidersActor{ - commandChan: actor.NewCommandChannel(), - } + return &ProvidersActor{} } func (a *ProvidersActor) AddProvider(prov LockProvider, path ...any) { @@ -42,46 +42,46 @@ func (a *ProvidersActor) Init() { func (a *ProvidersActor) WaitIndexUpdated(ctx context.Context, index int64) error { fut := future.NewSetVoid() - a.commandChan.Send(func() { - if index <= a.localLockReqIndex { - fut.SetVoid() - } else { - a.indexWaiters = append(a.indexWaiters, indexWaiter{ - Index: index, - Future: fut, - }) - } - }) + a.lock.Lock() + if index <= a.localLockReqIndex { + fut.SetVoid() + } else { + a.indexWaiters = append(a.indexWaiters, indexWaiter{ + Index: index, + Callback: fut, + }) + } + a.lock.Unlock() return fut.Wait(ctx) } -func (a *ProvidersActor) ApplyLockRequestEvents(events []LockRequestEvent) { - a.commandChan.Send(func() { - for _, op := range events { - if op.IsLocking { - err := a.lockLockRequest(op.Data) - if err != nil { - // TODO 发生这种错误需要重新加载全量状态，下同 - logger.Std.Warnf("applying locking event: %s", err.Error()) - return - } - - } else { - err := a.unlockLockRequest(op.Data) - if err != nil { - logger.Std.Warnf("applying unlocking event: %s", err.Error()) - return - } +func (a *ProvidersActor) OnLockRequestEvent(evt LockRequestEvent) { + func() { + a.lock.Lock() + defer a.lock.Unlock() + + if evt.IsLocking { + err := a.lockLockRequest(evt.Data) + if err != nil { + // TODO 发生这种错误需要重新加载全量状态，下同 + logger.Std.Warnf("applying locking event: %s", err.Error()) + return } - // 处理了多少事件，Index就往后移动多少个 - a.localLockReqIndex++ + } else { + err := a.unlockLockRequest(evt.Data) + if err != nil { + logger.Std.Warnf("applying unlocking event: %s", err.Error()) + return + } } - // 检查是否有等待同步进度的需求 - a.wakeUpIndexWaiter() - }) + a.localLockReqIndex++ + }() + + // 检查是否有等待同步进度的需求 + a.wakeUpIndexWaiter() } func (svc *ProvidersActor) lockLockRequest(reqData LockRequestData) error { @@ -135,83 +135,74 @@ func (svc *ProvidersActor) unlockLockRequest(reqData LockRequestData) error { // TestLockRequestAndMakeData 判断锁能否锁成功，并生成锁数据的字符串表示。注：不会生成请求ID。 // 在检查单个锁是否能上锁时，不会考虑同一个锁请求中的其他的锁影响。简单来说，就是同一个请求中的锁可以互相冲突。 func (a *ProvidersActor) TestLockRequestAndMakeData(req LockRequest) (LockRequestData, error) { - return actor.WaitValue(context.TODO(), a.commandChan, func() (LockRequestData, error) { - reqData := LockRequestData{} + a.lock.Lock() + defer a.lock.Unlock() - for _, lock := range req.Locks { - n, ok := a.provdersTrie.WalkEnd(lock.Path) - if !ok || n.Value == nil { - return LockRequestData{}, fmt.Errorf("lock provider not found for path %v", lock.Path) - } + reqData := LockRequestData{} - err := n.Value.CanLock(lock) - if err != nil { - return LockRequestData{}, err - } + for _, lock := range req.Locks { + n, ok := a.provdersTrie.WalkEnd(lock.Path) + if !ok || n.Value == nil { + return LockRequestData{}, fmt.Errorf("lock provider not found for path %v", lock.Path) + } - targetStr, err := n.Value.GetTargetString(lock.Target) - if err != nil { - return LockRequestData{}, fmt.Errorf("get lock target string failed, err: %w", err) - } + err := n.Value.CanLock(lock) + if err != nil { + return LockRequestData{}, err + } - reqData.Locks = append(reqData.Locks, lockData{ - Path: lock.Path, - Name: lock.Name, - Target: targetStr, - }) + targetStr, err := n.Value.GetTargetString(lock.Target) + if err != nil { + return LockRequestData{}, fmt.Errorf("get lock target string failed, err: %w", err) } - return reqData, nil - }) + reqData.Locks = append(reqData.Locks, lockData{ + Path: lock.Path, + Name: lock.Name, + Target: targetStr, + }) + } + + return reqData, nil } -// ResetState 重置内部状态 func (a *ProvidersActor) ResetState(index int64, lockRequestData []LockRequestData) error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - for _, p := range a.allProviders { - p.Clear() - } + a.lock.Lock() + defer a.lock.Unlock() - for _, reqData := range lockRequestData { - err := a.lockLockRequest(reqData) - if err != nil { - return fmt.Errorf("lock by lock request data failed, err: %w", err) - } + var err error + + for _, p := range a.allProviders { + p.Clear() + } + + for _, reqData := range lockRequestData { + err = a.lockLockRequest(reqData) + if err != nil { + err = fmt.Errorf("applying lock request data: %w", err) + break } + } - a.localLockReqIndex = index + a.localLockReqIndex = index - // 检查是否有等待同步进度的需求 - a.wakeUpIndexWaiter() + // 内部状态已被破坏，停止所有监听器 + for _, w := range a.indexWaiters { + w.Callback.SetError(ErrWaitIndexUpdateTimeout) + } + a.indexWaiters = nil - return nil - }) + return err } func (a *ProvidersActor) wakeUpIndexWaiter() { var resetWaiters []indexWaiter for _, waiter := range a.indexWaiters { if waiter.Index <= a.localLockReqIndex { - waiter.Future.SetVoid() + waiter.Callback.SetVoid() } else { resetWaiters = append(resetWaiters, waiter) } } a.indexWaiters = resetWaiters } - -func (a *ProvidersActor) Serve() error { - cmdChan := a.commandChan.BeginChanReceive() - defer a.commandChan.CloseChanReceive() - - for { - select { - case cmd, ok := <-cmdChan: - if !ok { - return fmt.Errorf("command channel closed") - } - - cmd() - } - } -} diff --git a/pkgs/distlock/internal/release_actor.go b/pkgs/distlock/internal/release_actor.go index 4105350..fea93a9 100644 --- a/pkgs/distlock/internal/release_actor.go +++ b/pkgs/distlock/internal/release_actor.go @@ -24,7 +24,7 @@ type ReleaseActor struct { releasingLockRequestIDs map[string]bool timer *time.Timer - timerSetuped bool + timerSetup bool lock sync.Mutex } @@ -66,6 +66,27 @@ func (a *ReleaseActor) DelayRelease(reqIDs []string) { a.setupTimer() } +func (a *ReleaseActor) ResetState(reqIDs []string) { + a.lock.Lock() + defer a.lock.Unlock() + + a.releasingLockRequestIDs = make(map[string]bool) + for _, id := range reqIDs { + a.releasingLockRequestIDs[id] = true + } + + a.setupTimer() +} + +func (a *ReleaseActor) OnLockRequestEvent(event LockRequestEvent) { + a.lock.Lock() + defer a.lock.Unlock() + + if !event.IsLocking { + delete(a.releasingLockRequestIDs, event.Data.ID) + } +} + func (a *ReleaseActor) doReleasing() error { ctx := context.TODO() @@ -87,7 +108,7 @@ func (a *ReleaseActor) doReleasing() error { // TODO 可以考虑优化成一次性删除多个锁 for id := range a.releasingLockRequestIDs { - lockReqKey := makeEtcdLockRequestKey(id) + lockReqKey := MakeEtcdLockRequestKey(id) txResp, err := a.etcdCli.Txn(ctx). If(clientv3util.KeyExists(lockReqKey)). @@ -110,10 +131,10 @@ func (a *ReleaseActor) setupTimer() { return } - if a.timerSetuped { + if a.timerSetup { return } - a.timerSetuped = true + a.timerSetup = true delay := int64(0) if a.cfg.RandomReleasingDelayMs == 0 { @@ -130,7 +151,11 @@ func (a *ReleaseActor) setupTimer() { go func() { <-a.timer.C - a.timerSetuped = false + + a.lock.Lock() + defer a.lock.Unlock() + + a.timerSetup = false // TODO 处理错误 err := a.doReleasing() diff --git a/pkgs/distlock/internal/service_info_actor.go b/pkgs/distlock/internal/service_info_actor.go new file mode 100644 index 0000000..baad890 --- /dev/null +++ b/pkgs/distlock/internal/service_info_actor.go @@ -0,0 +1,136 @@ +package internal + +import ( + "context" + "fmt" + "sync" + + "github.com/google/uuid" + mylo "gitlink.org.cn/cloudream/common/utils/lo" + "gitlink.org.cn/cloudream/common/utils/serder" + clientv3 "go.etcd.io/etcd/client/v3" +) + +type serviceStatus struct { + Info ServiceInfo + LockRequestIDs []string +} + +type ServiceInfoActor struct { + cfg *Config + etcdCli *clientv3.Client + releaseActor *ReleaseActor + + lock sync.Mutex + selfInfo ServiceInfo + leaseID *clientv3.LeaseID + services map[string]*serviceStatus +} + +func NewServiceInfoActor(cfg *Config, etcdCli *clientv3.Client) *ServiceInfoActor { + return &ServiceInfoActor{ + cfg: cfg, + etcdCli: etcdCli, + } +} + +func (a *ServiceInfoActor) GetSelfInfo() *ServiceInfo { + return &a.selfInfo +} + +func (a *ServiceInfoActor) ResetState(ctx context.Context, currentServices []ServiceInfo, currentLocks []LockRequestData) ([]string, error) { + a.lock.Lock() + defer a.lock.Unlock() + + if a.leaseID != nil { + a.etcdCli.Revoke(ctx, *a.leaseID) + a.leaseID = nil + } + + // 生成并注册服务信息 + a.selfInfo.ID = uuid.NewString() + + infoData, err := serder.ObjectToJSON(a.selfInfo) + if err != nil { + return nil, fmt.Errorf("service info to json: %w", err) + } + + lease, err := a.etcdCli.Grant(ctx, a.cfg.EtcdLockLeaseTimeSec) + if err != nil { + return nil, fmt.Errorf("granting lease: %w", err) + } + + a.leaseID = &lease.ID + + _, err = a.etcdCli.Put(ctx, MakeServiceInfoKey(a.selfInfo.ID), string(infoData), clientv3.WithLease(lease.ID)) + if err != nil { + a.etcdCli.Revoke(ctx, lease.ID) + return nil, fmt.Errorf("putting service info to etcd: %w", err) + } + + // 导入当前已有的服务信息和锁信息 + a.services = make(map[string]*serviceStatus) + for _, svc := range currentServices { + a.services[svc.ID] = &serviceStatus{ + Info: svc, + } + } + + // 导入锁信息的过程中可能会发现未注册信息的锁服务的锁，把他们挑出来释放掉 + var willReleaseIDs []string + for _, lock := range currentLocks { + svc, ok := a.services[lock.SerivceID] + if !ok { + willReleaseIDs = append(willReleaseIDs, lock.ID) + continue + } + + svc.LockRequestIDs = append(svc.LockRequestIDs, lock.ID) + } + + return willReleaseIDs, nil +} + +func (a *ServiceInfoActor) OnServiceEvent(evt ServiceEvent) { + a.lock.Lock() + defer a.lock.Unlock() + + // TODO 可以考虑打印一点日志 + + if evt.IsNew { + a.services[evt.Info.ID] = &serviceStatus{ + Info: evt.Info, + } + } else { + status, ok := a.services[evt.Info.ID] + if !ok { + return + } + + a.releaseActor.DelayRelease(status.LockRequestIDs) + + delete(a.services, evt.Info.ID) + + // TODO 处理收到自己崩溃的消息 + } +} + +func (a *ServiceInfoActor) OnLockRequestEvent(evt LockRequestEvent) { + a.lock.Lock() + defer a.lock.Unlock() + + status, ok := a.services[evt.Data.SerivceID] + if !ok { + // 加锁的是一个没有注册过的锁服务，大概率是因为这个锁服务之前网络发生了波动， + // 在波动期间它注册的信息过期，于是被当前的服务删除了。 + // 为了防止它加了这个锁之后又崩溃，导致的无限锁定，它加的锁我们都直接释放。 + a.releaseActor.Release([]string{evt.Data.ID}) + return + } + + if evt.IsLocking { + status.LockRequestIDs = append(status.LockRequestIDs, evt.Data.ID) + } else { + status.LockRequestIDs = mylo.Remove(status.LockRequestIDs, evt.Data.ID) + } +} diff --git a/pkgs/distlock/internal/utils.go b/pkgs/distlock/internal/utils.go deleted file mode 100644 index d77e5e3..0000000 --- a/pkgs/distlock/internal/utils.go +++ /dev/null @@ -1,116 +0,0 @@ -package internal - -import ( - "strings" -) - -func makeEtcdLockRequestKey(reqID string) string { - return EtcdLockRequestData + "/" + reqID -} - -func getLockRequestID(key string) string { - return strings.TrimPrefix(key, EtcdLockRequestData+"/") -} - -/* -func parseLockData(str string) (lock lockData, err error) { - sb := strings.Builder{} - var comps []string - - escaping := false - for _, ch := range strings.TrimSpace(str) { - if escaping { - if ch == 'n' { - sb.WriteRune('\n') - } else { - sb.WriteRune(ch) - } - - escaping = false - continue - } - - if ch == '/' { - comps = append(comps, sb.String()) - sb.Reset() - } else if ch == '\\' { - escaping = true - } else { - sb.WriteRune(ch) - } - } - - comps = append(comps, sb.String()) - - if len(comps) < 3 { - return lockData{}, fmt.Errorf("string must includes 3 components devided by /") - } - - return lockData{ - Path: comps[0 : len(comps)-2], - Name: comps[len(comps)-2], - Target: comps[len(comps)-1], - }, nil -} - -func lockDataToString(lock lockData) string { - sb := strings.Builder{} - - for _, s := range lock.Path { - sb.WriteString(lockDataEncoding(s)) - sb.WriteRune('/') - } - - sb.WriteString(lockDataEncoding(lock.Name)) - sb.WriteRune('/') - - sb.WriteString(lockDataEncoding(lock.Target)) - - return sb.String() -} - -func lockDataEncoding(str string) string { - sb := strings.Builder{} - - for _, ch := range str { - if ch == '\\' { - sb.WriteString("\\\\") - } else if ch == '/' { - sb.WriteString("\\/") - } else if ch == '\n' { - sb.WriteString("\\n") - } else { - sb.WriteRune(ch) - } - } - - return sb.String() -} - -func lockDataDecoding(str string) string { - sb := strings.Builder{} - - escaping := false - for _, ch := range str { - if escaping { - if ch == 'n' { - sb.WriteRune('\n') - } else { - sb.WriteRune(ch) - } - - escaping = false - continue - } - - if ch == '\\' { - escaping = true - - } else { - sb.WriteRune(ch) - } - } - - return sb.String() -} -*/ diff --git a/pkgs/distlock/internal/utils_test.go b/pkgs/distlock/internal/utils_test.go deleted file mode 100644 index 6c9f95c..0000000 --- a/pkgs/distlock/internal/utils_test.go +++ /dev/null @@ -1,61 +0,0 @@ -package internal - -/* -import ( - . "github.com/smartystreets/goconvey/convey" -) - -func Test_parseLockData_lockDataToString(t *testing.T) { - cases := []struct { - title string - data lockData - }{ - { - title: "多段路径", - data: lockData{ - Path: []string{"a", "b", "c"}, - Name: "d", - Target: "e", - }, - }, - - { - title: "包含分隔符", - data: lockData{ - Path: []string{"a/", "b", "c/c"}, - Name: "/d", - Target: "///e//d/", - }, - }, - - { - title: "包含转义符", - data: lockData{ - Path: []string{"a\\/", "b", "\\c/c"}, - Name: "/d", - Target: "///e\\//d/\\", - }, - }, - - { - title: "包含换行符", - data: lockData{ - Path: []string{"a\n", "\nb", "c\nc"}, - Name: "/d", - Target: "e\nd\n", - }, - }, - } - - for _, ca := range cases { - Convey(ca.title, t, func() { - str := lockDataToString(ca.data) - - data, err := parseLockData(str) - - So(err, ShouldBeNil) - So(data, ShouldResemble, ca.data) - }) - } -} -*/ diff --git a/pkgs/distlock/internal/watch_etcd_actor.go b/pkgs/distlock/internal/watch_etcd_actor.go index 2d0b791..5928da6 100644 --- a/pkgs/distlock/internal/watch_etcd_actor.go +++ b/pkgs/distlock/internal/watch_etcd_actor.go @@ -3,9 +3,9 @@ package internal import ( "context" "fmt" + "strings" "gitlink.org.cn/cloudream/common/pkgs/actor" - mylo "gitlink.org.cn/cloudream/common/utils/lo" "gitlink.org.cn/cloudream/common/utils/serder" clientv3 "go.etcd.io/etcd/client/v3" ) @@ -15,16 +15,23 @@ type LockRequestEvent struct { Data LockRequestData } -type LockRequestEventWatcher struct { - OnEvent func(events []LockRequestEvent) +type ServiceEvent struct { + IsNew bool + Info ServiceInfo } +type OnLockRequestEventFn func(event LockRequestEvent) + +type OnServiceEventFn func(event ServiceEvent) + type WatchEtcdActor struct { - etcdCli *clientv3.Client - watchChan clientv3.WatchChan - lockReqWatchers []*LockRequestEventWatcher + etcdCli *clientv3.Client - commandChan *actor.CommandChannel + watchChan clientv3.WatchChan + watchChanCancel func() + onLockRequestEventFn OnLockRequestEventFn + onServiceEventFn OnServiceEventFn + commandChan *actor.CommandChannel } func NewWatchEtcdActor(etcdCli *clientv3.Client) *WatchEtcdActor { @@ -34,33 +41,32 @@ func NewWatchEtcdActor(etcdCli *clientv3.Client) *WatchEtcdActor { } } -func (a *WatchEtcdActor) Init() { +func (a *WatchEtcdActor) Init(onLockRequestEvent OnLockRequestEventFn, onServiceDown OnServiceEventFn) { + a.onLockRequestEventFn = onLockRequestEvent + a.onServiceEventFn = onServiceDown } -func (a *WatchEtcdActor) StartWatching(revision int64) error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - a.watchChan = a.etcdCli.Watch(context.Background(), EtcdLockRequestData, clientv3.WithPrefix(), clientv3.WithPrevKV(), clientv3.WithRev(revision)) - return nil - }) -} - -func (a *WatchEtcdActor) StopWatching() error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - a.watchChan = nil - return nil - }) -} +func (a *WatchEtcdActor) Start(revision int64) { + actor.Wait(context.Background(), a.commandChan, func() error { + if a.watchChanCancel != nil { + a.watchChanCancel() + a.watchChanCancel = nil + } -func (a *WatchEtcdActor) AddEventWatcher(watcher *LockRequestEventWatcher) error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - a.lockReqWatchers = append(a.lockReqWatchers, watcher) + ctx, cancel := context.WithCancel(context.Background()) + a.watchChan = a.etcdCli.Watch(ctx, EtcdWatchPrefix, clientv3.WithPrefix(), clientv3.WithPrevKV(), clientv3.WithRev(revision)) + a.watchChanCancel = cancel return nil }) } -func (a *WatchEtcdActor) RemoveEventWatcher(watcher *LockRequestEventWatcher) error { - return actor.Wait(context.TODO(), a.commandChan, func() error { - a.lockReqWatchers = mylo.Remove(a.lockReqWatchers, watcher) +func (a *WatchEtcdActor) Stop() { + actor.Wait(context.Background(), a.commandChan, func() error { + if a.watchChanCancel != nil { + a.watchChanCancel() + a.watchChanCancel = nil + } + a.watchChan = nil return nil }) } @@ -85,14 +91,10 @@ func (a *WatchEtcdActor) Serve() error { return fmt.Errorf("watch etcd channel closed") } - events, err := a.parseEvents(msg) + err := a.dispatchEtcdEvent(msg) if err != nil { // TODO 更好的错误处理 - return fmt.Errorf("parse etcd lock request data failed, err: %w", err) - } - - for _, w := range a.lockReqWatchers { - w.OnEvent(events) + return err } } @@ -109,41 +111,79 @@ func (a *WatchEtcdActor) Serve() error { } } -func (a *WatchEtcdActor) parseEvents(watchResp clientv3.WatchResponse) ([]LockRequestEvent, error) { - var events []LockRequestEvent - +func (a *WatchEtcdActor) dispatchEtcdEvent(watchResp clientv3.WatchResponse) error { for _, e := range watchResp.Events { + key := string(e.Kv.Key) - shouldParseData := false - isLocking := true - var valueData []byte - - // 只监听新建和删除的事件，因为在设计上约定只有这两种事件才会影响Index - if e.Type == clientv3.EventTypeDelete { - shouldParseData = true - isLocking = false - valueData = e.PrevKv.Value - } else if e.IsCreate() { - shouldParseData = true - isLocking = true - valueData = e.Kv.Value - } + if strings.HasPrefix(key, EtcdLockRequestDataPrefix) { + if err := a.applyLockRequestEvent(e); err != nil { + return fmt.Errorf("parsing lock request event: %w", err) + } - if !shouldParseData { - continue + } else if strings.HasPrefix(key, EtcdServiceInfoPrefix) { + if err := a.applyServiceEvent(e); err != nil { + return fmt.Errorf("parsing service event: %w", err) + } } + } - var reqData LockRequestData - err := serder.JSONToObject(valueData, &reqData) - if err != nil { - return nil, fmt.Errorf("parse lock request data failed, err: %w", err) - } + return nil +} + +func (a *WatchEtcdActor) applyLockRequestEvent(evt *clientv3.Event) error { + isLocking := true + var valueData []byte + + // 只监听新建和删除的事件，因为在设计上约定只有这两种事件才会影响Index + if evt.Type == clientv3.EventTypeDelete { + isLocking = false + valueData = evt.PrevKv.Value + } else if evt.IsCreate() { + isLocking = true + valueData = evt.Kv.Value + } else { + return nil + } + + var reqData LockRequestData + err := serder.JSONToObject(valueData, &reqData) + if err != nil { + return fmt.Errorf("parse lock request data failed, err: %w", err) + } - events = append(events, LockRequestEvent{ - IsLocking: isLocking, - Data: reqData, - }) + a.onLockRequestEventFn(LockRequestEvent{ + IsLocking: isLocking, + Data: reqData, + }) + + return nil +} + +func (a *WatchEtcdActor) applyServiceEvent(evt *clientv3.Event) error { + isNew := true + var valueData []byte + + // 只监听新建和删除的事件，因为在设计上约定只有这两种事件才会影响Index + if evt.Type == clientv3.EventTypeDelete { + isNew = false + valueData = evt.PrevKv.Value + } else if evt.IsCreate() { + isNew = true + valueData = evt.Kv.Value + } else { + return nil } - return events, nil + var svcInfo ServiceInfo + err := serder.JSONToObject(valueData, &svcInfo) + if err != nil { + return fmt.Errorf("parsing service info: %w", err) + } + + a.onServiceEventFn(ServiceEvent{ + IsNew: isNew, + Info: svcInfo, + }) + + return nil } diff --git a/pkgs/distlock/service.go b/pkgs/distlock/service.go index 6fb6430..ec08875 100644 --- a/pkgs/distlock/service.go +++ b/pkgs/distlock/service.go @@ -47,13 +47,12 @@ type Service struct { cfg *internal.Config etcdCli *clientv3.Client - acquireActor *internal.AcquireActor - releaseActor *internal.ReleaseActor - providersActor *internal.ProvidersActor - watchEtcdActor *internal.WatchEtcdActor - leaseActor *internal.LeaseActor - - lockReqEventWatcher internal.LockRequestEventWatcher + acquireActor *internal.AcquireActor + releaseActor *internal.ReleaseActor + providersActor *internal.ProvidersActor + watchEtcdActor *internal.WatchEtcdActor + leaseActor *internal.LeaseActor + serviceInfoActor *internal.ServiceInfoActor } func NewService(cfg *internal.Config, initProvs []PathProvider) (*Service, error) { @@ -72,24 +71,36 @@ func NewService(cfg *internal.Config, initProvs []PathProvider) (*Service, error providersActor := internal.NewProvidersActor() watchEtcdActor := internal.NewWatchEtcdActor(etcdCli) leaseActor := internal.NewLeaseActor() + serviceInfoActor := internal.NewServiceInfoActor(cfg, etcdCli) acquireActor.Init(providersActor) - providersActor.Init() - watchEtcdActor.Init() leaseActor.Init(releaseActor) + providersActor.Init() + watchEtcdActor.Init( + func(event internal.LockRequestEvent) { + providersActor.OnLockRequestEvent(event) + acquireActor.TryAcquireNow() + releaseActor.OnLockRequestEvent(event) + serviceInfoActor.OnLockRequestEvent(event) + }, + func(event internal.ServiceEvent) { + serviceInfoActor.OnServiceEvent(event) + }, + ) for _, prov := range initProvs { providersActor.AddProvider(prov.Provider, prov.Path...) } return &Service{ - cfg: cfg, - etcdCli: etcdCli, - acquireActor: acquireActor, - releaseActor: releaseActor, - providersActor: providersActor, - watchEtcdActor: watchEtcdActor, - leaseActor: leaseActor, + cfg: cfg, + etcdCli: etcdCli, + acquireActor: acquireActor, + releaseActor: releaseActor, + providersActor: providersActor, + watchEtcdActor: watchEtcdActor, + leaseActor: leaseActor, + serviceInfoActor: serviceInfoActor, }, nil } @@ -147,14 +158,6 @@ func (svc *Service) Serve() error { // 1. client退出时直接中断进程，此时AcquireActor可能正在进行重试，于是导致Etcd锁没有解除就退出了进程。 // 虽然由于租约的存在不会导致系统长期卡死，但会影响client的使用 - go func() { - // TODO 处理错误 - err := svc.providersActor.Serve() - if err != nil { - logger.Std.Warnf("serving providers actor failed, err: %s", err.Error()) - } - }() - go func() { // TODO 处理错误 err := svc.watchEtcdActor.Serve() @@ -171,34 +174,13 @@ func (svc *Service) Serve() error { } }() - revision, err := svc.loadState() + // TODO context + err := svc.resetState(context.Background()) if err != nil { // TODO 关闭其他的Actor，或者更好的错误处理方式 return fmt.Errorf("init data failed, err: %w", err) } - svc.lockReqEventWatcher.OnEvent = func(events []internal.LockRequestEvent) { - svc.acquireActor.TryAcquireNow() - svc.providersActor.ApplyLockRequestEvents(events) - } - err = svc.watchEtcdActor.AddEventWatcher(&svc.lockReqEventWatcher) - if err != nil { - // TODO 关闭其他的Actor，或者更好的错误处理方式 - return fmt.Errorf("add lock request event watcher failed, err: %w", err) - } - - err = svc.watchEtcdActor.StartWatching(revision) - if err != nil { - // TODO 关闭其他的Actor，或者更好的错误处理方式 - return fmt.Errorf("start watching etcd failed, err: %w", err) - } - - err = svc.leaseActor.StartChecking() - if err != nil { - // TODO 关闭其他的Actor，或者更好的错误处理方式 - return fmt.Errorf("start checking lease failed, err: %w", err) - } - // TODO 防止退出的临时解决办法 ch := make(chan any) <-ch @@ -206,51 +188,81 @@ func (svc *Service) Serve() error { return nil } -func (svc *Service) loadState() (int64, error) { - // 使用事务一次性获取index和锁数据，就不需要加全局锁了 - txResp, err := svc.etcdCli.Txn(context.Background()). +// ResetState 重置内部状态。注：只要调用到了此函数，无论在哪一步出的错， +// 都要将内部状态视为已被破坏，直到成功调用了此函数才能继续后面的步骤 +func (svc *Service) resetState(ctx context.Context) error { + // 必须使用事务一次性获取所有数据 + txResp, err := svc.etcdCli.Txn(ctx). Then( clientv3.OpGet(internal.EtcdLockRequestIndex), - clientv3.OpGet(internal.EtcdLockRequestData, clientv3.WithPrefix()), + clientv3.OpGet(internal.EtcdLockRequestDataPrefix, clientv3.WithPrefix()), + clientv3.OpGet(internal.EtcdServiceInfoPrefix, clientv3.WithPrefix()), ). Commit() if err != nil { - return 0, fmt.Errorf("get etcd data failed, err: %w", err) + return fmt.Errorf("getting etcd data: %w", err) } - indexKvs := txResp.Responses[0].GetResponseRange().Kvs - lockKvs := txResp.Responses[1].GetResponseRange().Kvs - - var index int64 - var reqData []internal.LockRequestData - // 解析Index + var index int64 = 0 + indexKvs := txResp.Responses[0].GetResponseRange().Kvs if len(indexKvs) > 0 { val, err := strconv.ParseInt(string(indexKvs[0].Value), 0, 64) if err != nil { - return 0, fmt.Errorf("parse lock request index failed, err: %w", err) + return fmt.Errorf("parsing lock request index: %w", err) } index = val - - } else { - index = 0 } // 解析锁请求数据 + var reqData []internal.LockRequestData + lockKvs := txResp.Responses[1].GetResponseRange().Kvs for _, kv := range lockKvs { var req internal.LockRequestData err := serder.JSONToObject(kv.Value, &req) if err != nil { - return 0, fmt.Errorf("parse lock request data failed, err: %w", err) + return fmt.Errorf("parsing lock request data: %w", err) } reqData = append(reqData, req) } + // 解析服务信息数据 + var svcInfo []internal.ServiceInfo + svcInfoKvs := txResp.Responses[2].GetResponseRange().Kvs + for _, kv := range svcInfoKvs { + var info internal.ServiceInfo + err := serder.JSONToObject(kv.Value, &info) + if err != nil { + return fmt.Errorf("parsing service info data: %w", err) + } + + svcInfo = append(svcInfo, info) + } + + // 先停止监听等定时事件 + svc.watchEtcdActor.Stop() + svc.leaseActor.Stop() + + // 然后将新获取到的状态装填到Actor中。注：执行顺序需要考虑Actor会被谁调用，不会被调用的优先Reset。 + releasingIDs, err := svc.serviceInfoActor.ResetState(ctx, svcInfo, reqData) + if err != nil { + return fmt.Errorf("reseting service info actor: %w", err) + } + + svc.acquireActor.ResetState(svc.serviceInfoActor.GetSelfInfo().ID) + + svc.leaseActor.ResetState() + err = svc.providersActor.ResetState(index, reqData) if err != nil { - return 0, fmt.Errorf("reset lock providers state failed, err: %w", err) + return fmt.Errorf("reseting providers actor: %w", err) } - return txResp.Header.Revision, nil + svc.releaseActor.ResetState(releasingIDs) + + // 重置完了之后再启动监听 + svc.watchEtcdActor.Start(txResp.Header.Revision) + svc.leaseActor.Start() + return nil } From b41724c9a685a5be79df3d61bbf1ba59a7ad48aa Mon Sep 17 00:00:00 2001 From: Sydonian <794346190@qq.com> Date: Mon, 30 Oct 2023 15:09:37 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E5=AE=8C=E5=96=84=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkgs/distlock/internal/acquire_actor.go | 49 +++++-- pkgs/distlock/internal/config.go | 5 +- pkgs/distlock/internal/lease_actor.go | 22 +-- pkgs/distlock/internal/models.go | 3 +- pkgs/distlock/internal/providers_actor.go | 19 +-- pkgs/distlock/internal/release_actor.go | 48 +++++- pkgs/distlock/internal/service_info_actor.go | 26 +++- pkgs/distlock/internal/watch_etcd_actor.go | 33 +++-- pkgs/distlock/service.go | 146 +++++++++++-------- 9 files changed, 217 insertions(+), 134 deletions(-) diff --git a/pkgs/distlock/internal/acquire_actor.go b/pkgs/distlock/internal/acquire_actor.go index 28e2bb6..ce161b1 100644 --- a/pkgs/distlock/internal/acquire_actor.go +++ b/pkgs/distlock/internal/acquire_actor.go @@ -29,15 +29,17 @@ type AcquireActor struct { etcdCli *clientv3.Client providersActor *ProvidersActor - serviceID string - acquirings []*acquireInfo - lock sync.Mutex + isMaintenance bool + serviceID string + acquirings []*acquireInfo + lock sync.Mutex } func NewAcquireActor(cfg *Config, etcdCli *clientv3.Client) *AcquireActor { return &AcquireActor{ - cfg: cfg, - etcdCli: etcdCli, + cfg: cfg, + etcdCli: etcdCli, + isMaintenance: true, } } @@ -57,6 +59,12 @@ func (a *AcquireActor) Acquire(ctx context.Context, req LockRequest) (string, er defer a.lock.Unlock() a.acquirings = append(a.acquirings, info) + + // 如果处于维护模式，那么只接受请求，不实际去处理 + if a.isMaintenance { + return + } + // TODO 处理错误 err := a.doAcquiring() if err != nil { @@ -93,6 +101,11 @@ func (a *AcquireActor) TryAcquireNow() { a.lock.Lock() defer a.lock.Unlock() + // 处于维护模式中时，即使是主动触发Acqurire也不予理会 + if a.isMaintenance { + return + } + err := a.doAcquiring() if err != nil { logger.Std.Debugf("doing acquiring: %s", err.Error()) @@ -100,19 +113,27 @@ func (a *AcquireActor) TryAcquireNow() { }() } +// 进入维护模式。维护模式期间只接受请求，不处理请求。 +func (a *AcquireActor) EnterMaintenance() { + a.lock.Lock() + defer a.lock.Unlock() + + a.isMaintenance = true +} + +// 退出维护模式。退出之后建议调用一下TryAcquireNow。 +func (a *AcquireActor) LeaveMaintenance() { + a.lock.Lock() + defer a.lock.Unlock() + + a.isMaintenance = false +} + func (a *AcquireActor) ResetState(serviceID string) { a.lock.Lock() defer a.lock.Unlock() a.serviceID = serviceID - for _, info := range a.acquirings { - if info.LastErr != nil { - info.Callback.SetError(info.LastErr) - } else { - info.Callback.SetError(ErrAcquiringTimeout) - } - } - a.acquirings = nil } func (a *AcquireActor) doAcquiring() error { @@ -136,7 +157,7 @@ func (a *AcquireActor) doAcquiring() error { // 等待本地状态同步到最新 // TODO 配置等待时间 - err = a.providersActor.WaitIndexUpdated(ctx, index) + err = a.providersActor.WaitLocalIndexTo(ctx, index) if err != nil { return err } diff --git a/pkgs/distlock/internal/config.go b/pkgs/distlock/internal/config.go index b6219f3..cf97cad 100644 --- a/pkgs/distlock/internal/config.go +++ b/pkgs/distlock/internal/config.go @@ -5,6 +5,7 @@ type Config struct { EtcdUsername string `json:"etcdUsername"` EtcdPassword string `json:"etcdPassword"` - EtcdLockLeaseTimeSec int64 `json:"etcdLockLeaseTimeSec"` // 全局锁的租约时间。锁服务会在这个时间内自动续约锁，但如果服务崩溃，则其他服务在租约到期后能重新获得锁。 - RandomReleasingDelayMs int64 `json:"randomReleasingDelayMs"` // 释放锁失败，随机延迟之后再次尝试。延迟时间=random(0, RandomReleasingDelayMs) + 最少延迟时间(1000ms) + EtcdLockLeaseTimeSec int64 `json:"etcdLockLeaseTimeSec"` // 全局锁的租约时间。锁服务会在这个时间内自动续约锁，但如果服务崩溃，则其他服务在租约到期后能重新获得锁。 + RandomReleasingDelayMs int64 `json:"randomReleasingDelayMs"` // 释放锁失败，随机延迟之后再次尝试。延迟时间=random(0, RandomReleasingDelayMs) + 最少延迟时间(1000ms) + ServiceDescription string `json:"serviceDescription"` // 锁服务描述信息，锁服务启动后会注册到Etcd中 } diff --git a/pkgs/distlock/internal/lease_actor.go b/pkgs/distlock/internal/lease_actor.go index 5b6045f..41c6800 100644 --- a/pkgs/distlock/internal/lease_actor.go +++ b/pkgs/distlock/internal/lease_actor.go @@ -91,26 +91,14 @@ func (a *LeaseActor) Remove(reqID string) error { }) } -func (a *LeaseActor) ResetState() { - actor.Wait(context.Background(), a.commandChan, func() error { - a.leases = make(map[string]*lockRequestLease) - return nil - }) -} - -func (a *LeaseActor) Serve() error { +func (a *LeaseActor) Serve() { cmdChan := a.commandChan.BeginChanReceive() defer a.commandChan.CloseChanReceive() for { if a.ticker != nil { select { - case cmd, ok := <-cmdChan: - if !ok { - a.ticker.Stop() - return fmt.Errorf("command chan closed") - } - + case cmd := <-cmdChan: cmd() case now := <-a.ticker.C: @@ -127,11 +115,7 @@ func (a *LeaseActor) Serve() error { } } else { select { - case cmd, ok := <-cmdChan: - if !ok { - return fmt.Errorf("command chan closed") - } - + case cmd := <-cmdChan: cmd() } } diff --git a/pkgs/distlock/internal/models.go b/pkgs/distlock/internal/models.go index 08eb593..808cbad 100644 --- a/pkgs/distlock/internal/models.go +++ b/pkgs/distlock/internal/models.go @@ -72,5 +72,6 @@ func MakeServiceInfoKey(svcID string) string { } type ServiceInfo struct { - ID string `json:"id"` + ID string `json:"id"` + Description string `json:"description"` } diff --git a/pkgs/distlock/internal/providers_actor.go b/pkgs/distlock/internal/providers_actor.go index c9f426c..149a842 100644 --- a/pkgs/distlock/internal/providers_actor.go +++ b/pkgs/distlock/internal/providers_actor.go @@ -7,7 +7,6 @@ import ( "sync" "gitlink.org.cn/cloudream/common/pkgs/future" - "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/pkgs/trie" ) @@ -39,7 +38,7 @@ func (a *ProvidersActor) AddProvider(prov LockProvider, path ...any) { func (a *ProvidersActor) Init() { } -func (a *ProvidersActor) WaitIndexUpdated(ctx context.Context, index int64) error { +func (a *ProvidersActor) WaitLocalIndexTo(ctx context.Context, index int64) error { fut := future.NewSetVoid() a.lock.Lock() @@ -56,32 +55,34 @@ func (a *ProvidersActor) WaitIndexUpdated(ctx context.Context, index int64) erro return fut.Wait(ctx) } -func (a *ProvidersActor) OnLockRequestEvent(evt LockRequestEvent) { - func() { +func (a *ProvidersActor) OnLockRequestEvent(evt LockRequestEvent) error { + err := func() error { a.lock.Lock() defer a.lock.Unlock() if evt.IsLocking { err := a.lockLockRequest(evt.Data) if err != nil { - // TODO 发生这种错误需要重新加载全量状态，下同 - logger.Std.Warnf("applying locking event: %s", err.Error()) - return + return fmt.Errorf("applying locking event: %w", err) } } else { err := a.unlockLockRequest(evt.Data) if err != nil { - logger.Std.Warnf("applying unlocking event: %s", err.Error()) - return + return fmt.Errorf("applying unlocking event: %w", err) } } a.localLockReqIndex++ + return nil }() + if err != nil { + return err + } // 检查是否有等待同步进度的需求 a.wakeUpIndexWaiter() + return nil } func (svc *ProvidersActor) lockLockRequest(reqData LockRequestData) error { diff --git a/pkgs/distlock/internal/release_actor.go b/pkgs/distlock/internal/release_actor.go index fea93a9..7cac546 100644 --- a/pkgs/distlock/internal/release_actor.go +++ b/pkgs/distlock/internal/release_actor.go @@ -22,16 +22,18 @@ type ReleaseActor struct { cfg *Config etcdCli *clientv3.Client + lock sync.Mutex + isMaintenance bool releasingLockRequestIDs map[string]bool timer *time.Timer timerSetup bool - lock sync.Mutex } func NewReleaseActor(cfg *Config, etcdCli *clientv3.Client) *ReleaseActor { return &ReleaseActor{ cfg: cfg, etcdCli: etcdCli, + isMaintenance: true, releasingLockRequestIDs: make(map[string]bool), } } @@ -45,6 +47,10 @@ func (a *ReleaseActor) Release(reqIDs []string) { a.releasingLockRequestIDs[id] = true } + if a.isMaintenance { + return + } + // TODO 处理错误 err := a.doReleasing() if err != nil { @@ -63,21 +69,48 @@ func (a *ReleaseActor) DelayRelease(reqIDs []string) { a.releasingLockRequestIDs[id] = true } + if a.isMaintenance { + return + } + a.setupTimer() } -func (a *ReleaseActor) ResetState(reqIDs []string) { +// 重试一下内部的解锁请求。不会阻塞调用者 +func (a *ReleaseActor) TryReleaseNow() { a.lock.Lock() defer a.lock.Unlock() - a.releasingLockRequestIDs = make(map[string]bool) - for _, id := range reqIDs { - a.releasingLockRequestIDs[id] = true + // 如果处于维护模式，那么即使主动进行释放操作，也不予理会 + if a.isMaintenance { + return + } + + // TODO 处理错误 + err := a.doReleasing() + if err != nil { + logger.Std.Debugf("doing releasing: %s", err.Error()) } a.setupTimer() } +// 进入维护模式。在维护模式期间只接受请求，不处理请求，包括延迟释放请求。 +func (a *ReleaseActor) EnterMaintenance() { + a.lock.Lock() + defer a.lock.Unlock() + + a.isMaintenance = true +} + +// 退出维护模式。退出之后建议调用一下TryReleaseNow。 +func (a *ReleaseActor) LeaveMaintenance() { + a.lock.Lock() + defer a.lock.Unlock() + + a.isMaintenance = false +} + func (a *ReleaseActor) OnLockRequestEvent(event LockRequestEvent) { a.lock.Lock() defer a.lock.Unlock() @@ -157,6 +190,11 @@ func (a *ReleaseActor) setupTimer() { a.timerSetup = false + // 如果处于维护模式，那么即使是定时器要求的释放操作，也不予理会 + if a.isMaintenance { + return + } + // TODO 处理错误 err := a.doReleasing() if err != nil { diff --git a/pkgs/distlock/internal/service_info_actor.go b/pkgs/distlock/internal/service_info_actor.go index baad890..c00354d 100644 --- a/pkgs/distlock/internal/service_info_actor.go +++ b/pkgs/distlock/internal/service_info_actor.go @@ -2,6 +2,7 @@ package internal import ( "context" + "errors" "fmt" "sync" @@ -11,6 +12,8 @@ import ( clientv3 "go.etcd.io/etcd/client/v3" ) +var ErrSelfServiceDown = errors.New("self service is down, need to restart") + type serviceStatus struct { Info ServiceInfo LockRequestIDs []string @@ -27,10 +30,11 @@ type ServiceInfoActor struct { services map[string]*serviceStatus } -func NewServiceInfoActor(cfg *Config, etcdCli *clientv3.Client) *ServiceInfoActor { +func NewServiceInfoActor(cfg *Config, etcdCli *clientv3.Client, baseSelfInfo ServiceInfo) *ServiceInfoActor { return &ServiceInfoActor{ - cfg: cfg, - etcdCli: etcdCli, + cfg: cfg, + etcdCli: etcdCli, + selfInfo: baseSelfInfo, } } @@ -91,7 +95,7 @@ func (a *ServiceInfoActor) ResetState(ctx context.Context, currentServices []Ser return willReleaseIDs, nil } -func (a *ServiceInfoActor) OnServiceEvent(evt ServiceEvent) { +func (a *ServiceInfoActor) OnServiceEvent(evt ServiceEvent) error { a.lock.Lock() defer a.lock.Unlock() @@ -104,15 +108,20 @@ func (a *ServiceInfoActor) OnServiceEvent(evt ServiceEvent) { } else { status, ok := a.services[evt.Info.ID] if !ok { - return + return nil } a.releaseActor.DelayRelease(status.LockRequestIDs) delete(a.services, evt.Info.ID) - // TODO 处理收到自己崩溃的消息 + // 如果收到的被删除服务信息是自己的，那么自己要重启，重新获取全量数据 + if evt.Info.ID == a.selfInfo.ID { + return ErrSelfServiceDown + } } + + return nil } func (a *ServiceInfoActor) OnLockRequestEvent(evt LockRequestEvent) { @@ -121,8 +130,9 @@ func (a *ServiceInfoActor) OnLockRequestEvent(evt LockRequestEvent) { status, ok := a.services[evt.Data.SerivceID] if !ok { - // 加锁的是一个没有注册过的锁服务，大概率是因为这个锁服务之前网络发生了波动， - // 在波动期间它注册的信息过期，于是被当前的服务删除了。 + // 加锁的是一个没有注册过的锁服务，可能是因为这个锁服务之前网络发生了波动， + // 在波动期间它注册的信息过期，于是被大家认为服务下线，清理掉了它管理的锁， + // 而在网络恢复回来之后，它还没有意识到自己被认为下线了，于是还在提交锁请求。 // 为了防止它加了这个锁之后又崩溃，导致的无限锁定，它加的锁我们都直接释放。 a.releaseActor.Release([]string{evt.Data.ID}) return diff --git a/pkgs/distlock/internal/watch_etcd_actor.go b/pkgs/distlock/internal/watch_etcd_actor.go index 5928da6..254ff17 100644 --- a/pkgs/distlock/internal/watch_etcd_actor.go +++ b/pkgs/distlock/internal/watch_etcd_actor.go @@ -24,6 +24,8 @@ type OnLockRequestEventFn func(event LockRequestEvent) type OnServiceEventFn func(event ServiceEvent) +type OnWatchFailedFn func(err error) + type WatchEtcdActor struct { etcdCli *clientv3.Client @@ -31,6 +33,7 @@ type WatchEtcdActor struct { watchChanCancel func() onLockRequestEventFn OnLockRequestEventFn onServiceEventFn OnServiceEventFn + onWatchFailedFn OnWatchFailedFn commandChan *actor.CommandChannel } @@ -41,9 +44,10 @@ func NewWatchEtcdActor(etcdCli *clientv3.Client) *WatchEtcdActor { } } -func (a *WatchEtcdActor) Init(onLockRequestEvent OnLockRequestEventFn, onServiceDown OnServiceEventFn) { +func (a *WatchEtcdActor) Init(onLockRequestEvent OnLockRequestEventFn, onServiceDown OnServiceEventFn, onWatchFailed OnWatchFailedFn) { a.onLockRequestEventFn = onLockRequestEvent a.onServiceEventFn = onServiceDown + a.onWatchFailedFn = onWatchFailed } func (a *WatchEtcdActor) Start(revision int64) { @@ -71,40 +75,37 @@ func (a *WatchEtcdActor) Stop() { }) } -func (a *WatchEtcdActor) Serve() error { +func (a *WatchEtcdActor) Serve() { cmdChan := a.commandChan.BeginChanReceive() defer a.commandChan.CloseChanReceive() for { if a.watchChan != nil { select { - case cmd, ok := <-cmdChan: - if !ok { - return fmt.Errorf("command channel closed") - } - + case cmd := <-cmdChan: cmd() case msg := <-a.watchChan: + // 只要发生错误，就停止监听，通知外部处理 if msg.Canceled { - // TODO 更好的错误处理 - return fmt.Errorf("watch etcd channel closed") + a.onWatchFailedFn(fmt.Errorf("watch etcd channel closed")) + a.watchChanCancel() + a.watchChan = nil + continue } err := a.dispatchEtcdEvent(msg) if err != nil { - // TODO 更好的错误处理 - return err + a.onWatchFailedFn(err) + a.watchChanCancel() + a.watchChan = nil + continue } } } else { select { - case cmd, ok := <-cmdChan: - if !ok { - return fmt.Errorf("command channel closed") - } - + case cmd := <-cmdChan: cmd() } } diff --git a/pkgs/distlock/service.go b/pkgs/distlock/service.go index ec08875..4c36bac 100644 --- a/pkgs/distlock/service.go +++ b/pkgs/distlock/service.go @@ -6,6 +6,7 @@ import ( "strconv" "time" + "gitlink.org.cn/cloudream/common/pkgs/actor" "gitlink.org.cn/cloudream/common/pkgs/distlock/internal" "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/utils/serder" @@ -47,6 +48,7 @@ type Service struct { cfg *internal.Config etcdCli *clientv3.Client + cmdChan *actor.CommandChannel acquireActor *internal.AcquireActor releaseActor *internal.ReleaseActor providersActor *internal.ProvidersActor @@ -66,42 +68,55 @@ func NewService(cfg *internal.Config, initProvs []PathProvider) (*Service, error return nil, fmt.Errorf("new etcd client failed, err: %w", err) } - acquireActor := internal.NewAcquireActor(cfg, etcdCli) - releaseActor := internal.NewReleaseActor(cfg, etcdCli) - providersActor := internal.NewProvidersActor() - watchEtcdActor := internal.NewWatchEtcdActor(etcdCli) - leaseActor := internal.NewLeaseActor() - serviceInfoActor := internal.NewServiceInfoActor(cfg, etcdCli) - - acquireActor.Init(providersActor) - leaseActor.Init(releaseActor) - providersActor.Init() - watchEtcdActor.Init( + svc := &Service{ + cfg: cfg, + etcdCli: etcdCli, + cmdChan: actor.NewCommandChannel(), + } + + svc.acquireActor = internal.NewAcquireActor(cfg, etcdCli) + svc.releaseActor = internal.NewReleaseActor(cfg, etcdCli) + svc.providersActor = internal.NewProvidersActor() + svc.watchEtcdActor = internal.NewWatchEtcdActor(etcdCli) + svc.leaseActor = internal.NewLeaseActor() + svc.serviceInfoActor = internal.NewServiceInfoActor(cfg, etcdCli, internal.ServiceInfo{ + Description: cfg.ServiceDescription, + }) + + svc.acquireActor.Init(svc.providersActor) + svc.leaseActor.Init(svc.releaseActor) + svc.providersActor.Init() + svc.watchEtcdActor.Init( func(event internal.LockRequestEvent) { - providersActor.OnLockRequestEvent(event) - acquireActor.TryAcquireNow() - releaseActor.OnLockRequestEvent(event) - serviceInfoActor.OnLockRequestEvent(event) + err := svc.providersActor.OnLockRequestEvent(event) + if err != nil { + logger.Std.Warnf("%s, will reset service state", err.Error()) + svc.cmdChan.Send(func() { svc.doResetState() }) + return + } + + svc.acquireActor.TryAcquireNow() + svc.releaseActor.OnLockRequestEvent(event) + svc.serviceInfoActor.OnLockRequestEvent(event) }, func(event internal.ServiceEvent) { - serviceInfoActor.OnServiceEvent(event) + err := svc.serviceInfoActor.OnServiceEvent(event) + if err != nil { + logger.Std.Warnf("%s, will reset service state", err.Error()) + svc.cmdChan.Send(func() { svc.doResetState() }) + } + }, + func(err error) { + logger.Std.Warnf("%s, will reset service state", err.Error()) + svc.cmdChan.Send(func() { svc.doResetState() }) }, ) for _, prov := range initProvs { - providersActor.AddProvider(prov.Provider, prov.Path...) + svc.providersActor.AddProvider(prov.Provider, prov.Path...) } - return &Service{ - cfg: cfg, - etcdCli: etcdCli, - acquireActor: acquireActor, - releaseActor: releaseActor, - providersActor: providersActor, - watchEtcdActor: watchEtcdActor, - leaseActor: leaseActor, - serviceInfoActor: serviceInfoActor, - }, nil + return svc, nil } // Acquire 请求一批锁。成功后返回锁请求ID @@ -158,39 +173,48 @@ func (svc *Service) Serve() error { // 1. client退出时直接中断进程，此时AcquireActor可能正在进行重试，于是导致Etcd锁没有解除就退出了进程。 // 虽然由于租约的存在不会导致系统长期卡死，但会影响client的使用 - go func() { - // TODO 处理错误 - err := svc.watchEtcdActor.Serve() - if err != nil { - logger.Std.Warnf("serving watch etcd actor actor failed, err: %s", err.Error()) - } - }() + go svc.watchEtcdActor.Serve() - go func() { - // TODO 处理错误 - err := svc.leaseActor.Serve() - if err != nil { - logger.Std.Warnf("serving lease actor failed, err: %s", err.Error()) + go svc.leaseActor.Serve() + + svc.cmdChan.Send(func() { svc.doResetState() }) + + cmdChan := svc.cmdChan.BeginChanReceive() + defer svc.cmdChan.CloseChanReceive() + + for { + select { + case cmd := <-cmdChan: + cmd() } - }() + } + return nil +} + +func (svc *Service) doResetState() { + logger.Std.Infof("start reset state") // TODO context err := svc.resetState(context.Background()) if err != nil { - // TODO 关闭其他的Actor，或者更好的错误处理方式 - return fmt.Errorf("init data failed, err: %w", err) + logger.Std.Warnf("reseting state: %s, will try again after 3 seconds", err.Error()) + <-time.After(time.Second * 3) + svc.cmdChan.Send(func() { svc.doResetState() }) + return } - - // TODO 防止退出的临时解决办法 - ch := make(chan any) - <-ch - - return nil + logger.Std.Infof("reset state success") } // ResetState 重置内部状态。注：只要调用到了此函数，无论在哪一步出的错， -// 都要将内部状态视为已被破坏，直到成功调用了此函数才能继续后面的步骤 +// 都要将内部状态视为已被破坏，直到成功调用了此函数才能继续后面的步骤。 +// 如果调用失败，服务将进入维护模式，届时可以接受请求，但不会处理请求，直到调用成功为止。 func (svc *Service) resetState(ctx context.Context) error { + // 让服务都进入维护模式 + svc.watchEtcdActor.Stop() + svc.leaseActor.Stop() + svc.acquireActor.EnterMaintenance() + svc.releaseActor.EnterMaintenance() + // 必须使用事务一次性获取所有数据 txResp, err := svc.etcdCli.Txn(ctx). Then( @@ -240,29 +264,31 @@ func (svc *Service) resetState(ctx context.Context) error { svcInfo = append(svcInfo, info) } - // 先停止监听等定时事件 - svc.watchEtcdActor.Stop() - svc.leaseActor.Stop() - - // 然后将新获取到的状态装填到Actor中。注：执行顺序需要考虑Actor会被谁调用，不会被调用的优先Reset。 + // 然后将新获取到的状态装填到Actor中 releasingIDs, err := svc.serviceInfoActor.ResetState(ctx, svcInfo, reqData) if err != nil { return fmt.Errorf("reseting service info actor: %w", err) } - svc.acquireActor.ResetState(svc.serviceInfoActor.GetSelfInfo().ID) - - svc.leaseActor.ResetState() - + // 要在acquireActor之前，因为acquireActor会调用它的WaitLocalIndexTo err = svc.providersActor.ResetState(index, reqData) if err != nil { return fmt.Errorf("reseting providers actor: %w", err) } - svc.releaseActor.ResetState(releasingIDs) + svc.acquireActor.ResetState(svc.serviceInfoActor.GetSelfInfo().ID) - // 重置完了之后再启动监听 + // ReleaseActor没有什么需要Reset的状态 + svc.releaseActor.Release(releasingIDs) + + // 重置完了之后再退出维护模式 svc.watchEtcdActor.Start(txResp.Header.Revision) svc.leaseActor.Start() + svc.acquireActor.LeaveMaintenance() + svc.releaseActor.LeaveMaintenance() + + svc.acquireActor.TryAcquireNow() + svc.releaseActor.TryReleaseNow() + return nil } From d775fd31137e9fb0241bffa7eda0073535b5e579 Mon Sep 17 00:00:00 2001 From: Sydonian <794346190@qq.com> Date: Tue, 31 Oct 2023 14:57:52 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=B0=83=E8=AF=95?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkgs/distlock/internal/acquire_actor.go | 51 ++++++++---- pkgs/distlock/internal/providers_actor.go | 31 +++----- pkgs/distlock/internal/release_actor.go | 59 +++++++++----- pkgs/distlock/internal/service_info_actor.go | 82 ++++++++++++++++---- pkgs/distlock/service.go | 10 ++- 5 files changed, 161 insertions(+), 72 deletions(-) diff --git a/pkgs/distlock/internal/acquire_actor.go b/pkgs/distlock/internal/acquire_actor.go index ce161b1..35fd1a0 100644 --- a/pkgs/distlock/internal/acquire_actor.go +++ b/pkgs/distlock/internal/acquire_actor.go @@ -29,17 +29,19 @@ type AcquireActor struct { etcdCli *clientv3.Client providersActor *ProvidersActor - isMaintenance bool - serviceID string - acquirings []*acquireInfo - lock sync.Mutex + isMaintenance bool + serviceID string + acquirings []*acquireInfo + lock sync.Mutex + doAcquiringChan chan any } func NewAcquireActor(cfg *Config, etcdCli *clientv3.Client) *AcquireActor { return &AcquireActor{ - cfg: cfg, - etcdCli: etcdCli, - isMaintenance: true, + cfg: cfg, + etcdCli: etcdCli, + isMaintenance: true, + doAcquiringChan: make(chan any), } } @@ -65,10 +67,9 @@ func (a *AcquireActor) Acquire(ctx context.Context, req LockRequest) (string, er return } - // TODO 处理错误 - err := a.doAcquiring() - if err != nil { - logger.Std.Debugf("doing acquiring: %s", err.Error()) + select { + case a.doAcquiringChan <- nil: + default: } }() @@ -106,9 +107,9 @@ func (a *AcquireActor) TryAcquireNow() { return } - err := a.doAcquiring() - if err != nil { - logger.Std.Debugf("doing acquiring: %s", err.Error()) + select { + case a.doAcquiringChan <- nil: + default: } }() } @@ -136,13 +137,30 @@ func (a *AcquireActor) ResetState(serviceID string) { a.serviceID = serviceID } +func (a *AcquireActor) Serve() { + for { + select { + case <-a.doAcquiringChan: + err := a.doAcquiring() + if err != nil { + logger.Std.Debugf("doing acquiring: %s", err.Error()) + } + } + } +} + func (a *AcquireActor) doAcquiring() error { ctx := context.Background() + // 先看一眼，如果没有需要请求的锁，就不用走后面的流程了 + a.lock.Lock() if len(a.acquirings) == 0 { + a.lock.Unlock() return nil } + a.lock.Unlock() + // 在获取全局锁的时候不用锁Actor，只有获取成功了，才加锁 // TODO 根据不同的错误设置不同的错误类型，方便上层进行后续处理 unlock, err := acquireEtcdRequestDataLock(ctx, a.etcdCli, a.cfg.EtcdLockLeaseTimeSec) if err != nil { @@ -155,6 +173,8 @@ func (a *AcquireActor) doAcquiring() error { return err } + logger.Std.Infof("wait to: %d", index) + // 等待本地状态同步到最新 // TODO 配置等待时间 err = a.providersActor.WaitLocalIndexTo(ctx, index) @@ -162,12 +182,15 @@ func (a *AcquireActor) doAcquiring() error { return err } + a.lock.Lock() + defer a.lock.Unlock() // TODO 可以考虑一次性获得多个锁 for i := 0; i < len(a.acquirings); i++ { req := a.acquirings[i] // 测试锁，并获得锁数据 reqData, err := a.providersActor.TestLockRequestAndMakeData(req.Request) + logger.Std.Infof("6") if err != nil { req.LastErr = err continue diff --git a/pkgs/distlock/internal/providers_actor.go b/pkgs/distlock/internal/providers_actor.go index 149a842..c2cd6d6 100644 --- a/pkgs/distlock/internal/providers_actor.go +++ b/pkgs/distlock/internal/providers_actor.go @@ -56,30 +56,23 @@ func (a *ProvidersActor) WaitLocalIndexTo(ctx context.Context, index int64) erro } func (a *ProvidersActor) OnLockRequestEvent(evt LockRequestEvent) error { - err := func() error { - a.lock.Lock() - defer a.lock.Unlock() - - if evt.IsLocking { - err := a.lockLockRequest(evt.Data) - if err != nil { - return fmt.Errorf("applying locking event: %w", err) - } + a.lock.Lock() + defer a.lock.Unlock() - } else { - err := a.unlockLockRequest(evt.Data) - if err != nil { - return fmt.Errorf("applying unlocking event: %w", err) - } + if evt.IsLocking { + err := a.lockLockRequest(evt.Data) + if err != nil { + return fmt.Errorf("applying locking event: %w", err) } - a.localLockReqIndex++ - return nil - }() - if err != nil { - return err + } else { + err := a.unlockLockRequest(evt.Data) + if err != nil { + return fmt.Errorf("applying unlocking event: %w", err) + } } + a.localLockReqIndex++ // 检查是否有等待同步进度的需求 a.wakeUpIndexWaiter() return nil diff --git a/pkgs/distlock/internal/release_actor.go b/pkgs/distlock/internal/release_actor.go index 7cac546..72e9145 100644 --- a/pkgs/distlock/internal/release_actor.go +++ b/pkgs/distlock/internal/release_actor.go @@ -27,6 +27,7 @@ type ReleaseActor struct { releasingLockRequestIDs map[string]bool timer *time.Timer timerSetup bool + doReleasingChan chan any } func NewReleaseActor(cfg *Config, etcdCli *clientv3.Client) *ReleaseActor { @@ -35,6 +36,7 @@ func NewReleaseActor(cfg *Config, etcdCli *clientv3.Client) *ReleaseActor { etcdCli: etcdCli, isMaintenance: true, releasingLockRequestIDs: make(map[string]bool), + doReleasingChan: make(chan any), } } @@ -51,13 +53,10 @@ func (a *ReleaseActor) Release(reqIDs []string) { return } - // TODO 处理错误 - err := a.doReleasing() - if err != nil { - logger.Std.Debugf("doing releasing: %s", err.Error()) + select { + case a.doReleasingChan <- nil: + default: } - - a.setupTimer() } // 延迟释放锁。一般用于清理崩溃的锁服务遗留下来的锁 @@ -86,13 +85,10 @@ func (a *ReleaseActor) TryReleaseNow() { return } - // TODO 处理错误 - err := a.doReleasing() - if err != nil { - logger.Std.Debugf("doing releasing: %s", err.Error()) + select { + case a.doReleasingChan <- nil: + default: } - - a.setupTimer() } // 进入维护模式。在维护模式期间只接受请求，不处理请求，包括延迟释放请求。 @@ -112,21 +108,41 @@ func (a *ReleaseActor) LeaveMaintenance() { } func (a *ReleaseActor) OnLockRequestEvent(event LockRequestEvent) { + if event.IsLocking { + return + } + a.lock.Lock() defer a.lock.Unlock() - if !event.IsLocking { - delete(a.releasingLockRequestIDs, event.Data.ID) + delete(a.releasingLockRequestIDs, event.Data.ID) +} + +func (a *ReleaseActor) Serve() { + for { + select { + case <-a.doReleasingChan: + err := a.doReleasing() + if err != nil { + logger.Std.Debugf("doing releasing: %s", err.Error()) + } + } } } func (a *ReleaseActor) doReleasing() error { - ctx := context.TODO() + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + // 先看一眼，如果没有需要释放的锁，就不用走后面的流程了 + a.lock.Lock() if len(a.releasingLockRequestIDs) == 0 { + a.lock.Unlock() return nil } + a.lock.Unlock() + // 在获取全局锁的时候不用锁Actor，只有获取成功了，才加锁 // TODO 根据不同的错误设置不同的错误类型，方便上层进行后续处理 unlock, err := acquireEtcdRequestDataLock(ctx, a.etcdCli, a.cfg.EtcdLockLeaseTimeSec) if err != nil { @@ -139,6 +155,10 @@ func (a *ReleaseActor) doReleasing() error { return err } + a.lock.Lock() + defer a.lock.Unlock() + defer a.setupTimer() + // TODO 可以考虑优化成一次性删除多个锁 for id := range a.releasingLockRequestIDs { lockReqKey := MakeEtcdLockRequestKey(id) @@ -195,12 +215,9 @@ func (a *ReleaseActor) setupTimer() { return } - // TODO 处理错误 - err := a.doReleasing() - if err != nil { - logger.Std.Debugf("doing releasing: %s", err.Error()) + select { + case a.doReleasingChan <- nil: + default: } - - a.setupTimer() }() } diff --git a/pkgs/distlock/internal/service_info_actor.go b/pkgs/distlock/internal/service_info_actor.go index c00354d..a940574 100644 --- a/pkgs/distlock/internal/service_info_actor.go +++ b/pkgs/distlock/internal/service_info_actor.go @@ -7,6 +7,7 @@ import ( "sync" "github.com/google/uuid" + "gitlink.org.cn/cloudream/common/pkgs/logger" mylo "gitlink.org.cn/cloudream/common/utils/lo" "gitlink.org.cn/cloudream/common/utils/serder" clientv3 "go.etcd.io/etcd/client/v3" @@ -20,14 +21,15 @@ type serviceStatus struct { } type ServiceInfoActor struct { - cfg *Config - etcdCli *clientv3.Client - releaseActor *ReleaseActor - - lock sync.Mutex - selfInfo ServiceInfo - leaseID *clientv3.LeaseID - services map[string]*serviceStatus + cfg *Config + etcdCli *clientv3.Client + + lock sync.Mutex + selfInfo ServiceInfo + leaseID *clientv3.LeaseID + leaseKeepAlive chan any + services map[string]*serviceStatus + releaseActor *ReleaseActor } func NewServiceInfoActor(cfg *Config, etcdCli *clientv3.Client, baseSelfInfo ServiceInfo) *ServiceInfoActor { @@ -38,6 +40,10 @@ func NewServiceInfoActor(cfg *Config, etcdCli *clientv3.Client, baseSelfInfo Ser } } +func (a *ServiceInfoActor) Init(releaseActor *ReleaseActor) { + a.releaseActor = releaseActor +} + func (a *ServiceInfoActor) GetSelfInfo() *ServiceInfo { return &a.selfInfo } @@ -48,6 +54,7 @@ func (a *ServiceInfoActor) ResetState(ctx context.Context, currentServices []Ser if a.leaseID != nil { a.etcdCli.Revoke(ctx, *a.leaseID) + close(a.leaseKeepAlive) a.leaseID = nil } @@ -63,9 +70,36 @@ func (a *ServiceInfoActor) ResetState(ctx context.Context, currentServices []Ser if err != nil { return nil, fmt.Errorf("granting lease: %w", err) } - a.leaseID = &lease.ID + keepAliveChan, err := a.etcdCli.Lease.KeepAlive(context.Background(), lease.ID) + if err != nil { + a.etcdCli.Revoke(ctx, lease.ID) + return nil, fmt.Errorf("starting keep lease alive: %w", err) + } + a.leaseKeepAlive = make(chan any) + + go func() { + for { + select { + case _, ok := <-keepAliveChan: + if !ok { + logger.Std.Warnf("lease keep alive channel closed, will try to open again") + + var err error + keepAliveChan, err = a.etcdCli.Lease.KeepAlive(context.Background(), lease.ID) + if err != nil { + logger.Std.Warnf("starting keep lease alive: %s", err.Error()) + return + } + } + + case <-a.leaseKeepAlive: + return + } + } + }() + _, err = a.etcdCli.Put(ctx, MakeServiceInfoKey(a.selfInfo.ID), string(infoData), clientv3.WithLease(lease.ID)) if err != nil { a.etcdCli.Revoke(ctx, lease.ID) @@ -79,6 +113,10 @@ func (a *ServiceInfoActor) ResetState(ctx context.Context, currentServices []Ser Info: svc, } } + // 直接添加自己的信息 + a.services[a.selfInfo.ID] = &serviceStatus{ + Info: a.selfInfo, + } // 导入锁信息的过程中可能会发现未注册信息的锁服务的锁，把他们挑出来释放掉 var willReleaseIDs []string @@ -102,10 +140,16 @@ func (a *ServiceInfoActor) OnServiceEvent(evt ServiceEvent) error { // TODO 可以考虑打印一点日志 if evt.IsNew { - a.services[evt.Info.ID] = &serviceStatus{ - Info: evt.Info, + if evt.Info.ID != a.selfInfo.ID { + logger.Std.WithField("ID", evt.Info.ID).Infof("new service up") + a.services[evt.Info.ID] = &serviceStatus{ + Info: evt.Info, + } } + } else { + logger.Std.WithField("ID", evt.Info.ID).Infof("service down, will release all its locks") + status, ok := a.services[evt.Info.ID] if !ok { return nil @@ -130,11 +174,17 @@ func (a *ServiceInfoActor) OnLockRequestEvent(evt LockRequestEvent) { status, ok := a.services[evt.Data.SerivceID] if !ok { - // 加锁的是一个没有注册过的锁服务，可能是因为这个锁服务之前网络发生了波动， - // 在波动期间它注册的信息过期，于是被大家认为服务下线，清理掉了它管理的锁， - // 而在网络恢复回来之后，它还没有意识到自己被认为下线了，于是还在提交锁请求。 - // 为了防止它加了这个锁之后又崩溃，导致的无限锁定，它加的锁我们都直接释放。 - a.releaseActor.Release([]string{evt.Data.ID}) + if evt.IsLocking { + // 加锁的是一个没有注册过的锁服务，可能是因为这个锁服务之前网络发生了波动， + // 在波动期间它注册的信息过期，于是被大家认为服务下线，清理掉了它管理的锁， + // 而在网络恢复回来之后，它还没有意识到自己被认为下线了，于是还在提交锁请求。 + // 为了防止它加了这个锁之后又崩溃，导致的无限锁定，它加的锁我们都直接释放。 + logger.Std.WithField("RequestID", evt.Data.ID). + WithField("ServiceID", evt.Data.SerivceID). + Warnf("the lock request is from an unknow service, will release it") + + a.releaseActor.Release([]string{evt.Data.ID}) + } return } diff --git a/pkgs/distlock/service.go b/pkgs/distlock/service.go index 4c36bac..354373b 100644 --- a/pkgs/distlock/service.go +++ b/pkgs/distlock/service.go @@ -111,6 +111,7 @@ func NewService(cfg *internal.Config, initProvs []PathProvider) (*Service, error svc.cmdChan.Send(func() { svc.doResetState() }) }, ) + svc.serviceInfoActor.Init(svc.releaseActor) for _, prov := range initProvs { svc.providersActor.AddProvider(prov.Provider, prov.Path...) @@ -177,6 +178,10 @@ func (svc *Service) Serve() error { go svc.leaseActor.Serve() + go svc.acquireActor.Serve() + + go svc.releaseActor.Serve() + svc.cmdChan.Send(func() { svc.doResetState() }) cmdChan := svc.cmdChan.BeginChanReceive() @@ -202,7 +207,8 @@ func (svc *Service) doResetState() { svc.cmdChan.Send(func() { svc.doResetState() }) return } - logger.Std.Infof("reset state success") + logger.Std.WithField("ID", svc.serviceInfoActor.GetSelfInfo().ID). + Infof("reset state success") } // ResetState 重置内部状态。注：只要调用到了此函数，无论在哪一步出的错， @@ -279,7 +285,7 @@ func (svc *Service) resetState(ctx context.Context) error { svc.acquireActor.ResetState(svc.serviceInfoActor.GetSelfInfo().ID) // ReleaseActor没有什么需要Reset的状态 - svc.releaseActor.Release(releasingIDs) + svc.releaseActor.DelayRelease(releasingIDs) // 重置完了之后再退出维护模式 svc.watchEtcdActor.Start(txResp.Header.Revision)