lock_client.go 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210
  1. package cluster
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
  6. "github.com/seaweedfs/seaweedfs/weed/glog"
  7. "github.com/seaweedfs/seaweedfs/weed/pb"
  8. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  9. "github.com/seaweedfs/seaweedfs/weed/util"
  10. "google.golang.org/grpc"
  11. "time"
  12. )
  13. type LockClient struct {
  14. grpcDialOption grpc.DialOption
  15. maxLockDuration time.Duration
  16. sleepDuration time.Duration
  17. seedFiler pb.ServerAddress
  18. }
  19. func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *LockClient {
  20. return &LockClient{
  21. grpcDialOption: grpcDialOption,
  22. maxLockDuration: 5 * time.Second,
  23. sleepDuration: 2473 * time.Millisecond,
  24. seedFiler: seedFiler,
  25. }
  26. }
  27. type LiveLock struct {
  28. key string
  29. renewToken string
  30. expireAtNs int64
  31. filer pb.ServerAddress
  32. cancelCh chan struct{}
  33. grpcDialOption grpc.DialOption
  34. isLocked bool
  35. owner string
  36. lc *LockClient
  37. }
  38. // NewLock creates a lock with a very long duration
  39. func (lc *LockClient) NewLock(key string, owner string) (lock *LiveLock) {
  40. return lc.doNewLock(key, lock_manager.MaxDuration, owner)
  41. }
  42. // StartLock starts a goroutine to lock the key and returns immediately.
  43. func (lc *LockClient) StartLock(key string, owner string) (lock *LiveLock) {
  44. lock = &LiveLock{
  45. key: key,
  46. filer: lc.seedFiler,
  47. cancelCh: make(chan struct{}),
  48. expireAtNs: time.Now().Add(lock_manager.MaxDuration).UnixNano(),
  49. grpcDialOption: lc.grpcDialOption,
  50. owner: owner,
  51. lc: lc,
  52. }
  53. go func() {
  54. util.RetryForever("create lock:"+key, func() error {
  55. errorMessage, err := lock.doLock(lock_manager.MaxDuration)
  56. if err != nil {
  57. glog.Infof("create lock %s: %s", key, err)
  58. time.Sleep(time.Second)
  59. return err
  60. }
  61. if errorMessage != "" {
  62. glog.Infof("create lock %s: %s", key, errorMessage)
  63. time.Sleep(time.Second)
  64. return fmt.Errorf("%v", errorMessage)
  65. }
  66. lock.isLocked = true
  67. return nil
  68. }, func(err error) (shouldContinue bool) {
  69. if err != nil {
  70. glog.Warningf("create lock %s: %s", key, err)
  71. time.Sleep(time.Second)
  72. }
  73. return lock.renewToken == ""
  74. })
  75. lc.keepLock(lock)
  76. }()
  77. return
  78. }
  79. func (lc *LockClient) doNewLock(key string, lockDuration time.Duration, owner string) (lock *LiveLock) {
  80. lock = &LiveLock{
  81. key: key,
  82. filer: lc.seedFiler,
  83. cancelCh: make(chan struct{}),
  84. expireAtNs: time.Now().Add(lockDuration).UnixNano(),
  85. grpcDialOption: lc.grpcDialOption,
  86. owner: owner,
  87. lc: lc,
  88. }
  89. var needRenewal bool
  90. if lockDuration > lc.maxLockDuration {
  91. lockDuration = lc.maxLockDuration
  92. needRenewal = true
  93. }
  94. util.RetryForever("create lock:"+key, func() error {
  95. errorMessage, err := lock.doLock(lockDuration)
  96. if err != nil {
  97. time.Sleep(time.Second)
  98. return err
  99. }
  100. if errorMessage != "" {
  101. time.Sleep(time.Second)
  102. return fmt.Errorf("%v", errorMessage)
  103. }
  104. lock.isLocked = true
  105. return nil
  106. }, func(err error) (shouldContinue bool) {
  107. if err != nil {
  108. glog.Warningf("create lock %s: %s", key, err)
  109. }
  110. return lock.renewToken == ""
  111. })
  112. if needRenewal {
  113. go lc.keepLock(lock)
  114. }
  115. return
  116. }
  117. func (lock *LiveLock) IsLocked() bool {
  118. return lock.isLocked
  119. }
  120. func (lock *LiveLock) StopLock() error {
  121. close(lock.cancelCh)
  122. if !lock.isLocked {
  123. return nil
  124. }
  125. return pb.WithFilerClient(false, 0, lock.filer, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  126. _, err := client.DistributedUnlock(context.Background(), &filer_pb.UnlockRequest{
  127. Name: lock.key,
  128. RenewToken: lock.renewToken,
  129. })
  130. return err
  131. })
  132. }
  133. func (lc *LockClient) keepLock(lock *LiveLock) {
  134. ticker := time.Tick(lc.sleepDuration)
  135. for {
  136. select {
  137. case <-ticker:
  138. // renew the lock if lock.expireAtNs is still greater than now
  139. util.RetryForever("keep lock:"+lock.key, func() error {
  140. lockDuration := time.Duration(lock.expireAtNs-time.Now().UnixNano()) * time.Nanosecond
  141. if lockDuration > lc.maxLockDuration {
  142. lockDuration = lc.maxLockDuration
  143. }
  144. if lockDuration <= 0 {
  145. return nil
  146. }
  147. errorMessage, err := lock.doLock(lockDuration)
  148. if err != nil {
  149. lock.isLocked = false
  150. time.Sleep(time.Second)
  151. return err
  152. }
  153. if errorMessage != "" {
  154. lock.isLocked = false
  155. time.Sleep(time.Second)
  156. return fmt.Errorf("%v", errorMessage)
  157. }
  158. return nil
  159. }, func(err error) (shouldContinue bool) {
  160. if err == nil {
  161. return false
  162. }
  163. glog.Warningf("keep lock %s: %v", lock.key, err)
  164. return true
  165. })
  166. if !lock.isLocked {
  167. return
  168. }
  169. case <-lock.cancelCh:
  170. return
  171. }
  172. }
  173. }
  174. func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, err error) {
  175. err = pb.WithFilerClient(false, 0, lock.filer, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  176. resp, err := client.DistributedLock(context.Background(), &filer_pb.LockRequest{
  177. Name: lock.key,
  178. SecondsToLock: int64(lockDuration.Seconds()),
  179. RenewToken: lock.renewToken,
  180. IsMoved: false,
  181. Owner: lock.owner,
  182. })
  183. if err == nil {
  184. lock.renewToken = resp.RenewToken
  185. }
  186. if resp != nil {
  187. errorMessage = resp.Error
  188. if resp.MovedTo != "" {
  189. lock.filer = pb.ServerAddress(resp.MovedTo)
  190. lock.lc.seedFiler = lock.filer
  191. }
  192. }
  193. return err
  194. })
  195. return
  196. }