From 1dbf2ccaeb56c3fd17e2a7bf724d232926cd5ba3 Mon Sep 17 00:00:00 2001 From: xyctruth <398041993@qq.com> Date: Thu, 24 Feb 2022 15:42:50 +0800 Subject: [PATCH] resetCronTime add has_remaining succeed_count --- dtmsvr/api_http.go | 7 ++++++- dtmsvr/storage/boltdb/boltdb.go | 18 +++++++++++------- dtmsvr/storage/redis/redis.go | 32 +++++++++++++++++++------------- dtmsvr/storage/sql/sql.go | 13 +++++++++++-- dtmsvr/storage/store.go | 2 +- test/api_test.go | 13 ++++++++++--- test/store_test.go | 25 ++++++++++++++++++------- 7 files changed, 76 insertions(+), 34 deletions(-) diff --git a/dtmsvr/api_http.go b/dtmsvr/api_http.go index e3507b8..2d8dea0 100644 --- a/dtmsvr/api_http.go +++ b/dtmsvr/api_http.go @@ -89,5 +89,10 @@ func resetCronTime(c *gin.Context) interface{} { sTimeoutSecond := dtmimp.OrString(c.Query("timeout"), strconv.FormatInt(3*conf.TimeoutToFail, 10)) sLimit := dtmimp.OrString(c.Query("limit"), "100") timeout := time.Duration(dtmimp.MustAtoi(sTimeoutSecond)) * time.Second - return GetStore().ResetCronTime(timeout, int64(dtmimp.MustAtoi(sLimit))) + + succeedCount, hasRemaining, err := GetStore().ResetCronTime(timeout, int64(dtmimp.MustAtoi(sLimit))) + if err != nil { + return err + } + return map[string]interface{}{"has_remaining": hasRemaining, "succeed_count": succeedCount} } diff --git a/dtmsvr/storage/boltdb/boltdb.go b/dtmsvr/storage/boltdb/boltdb.go index a09eb12..21abfb1 100644 --- a/dtmsvr/storage/boltdb/boltdb.go +++ b/dtmsvr/storage/boltdb/boltdb.go @@ -411,15 +411,19 @@ func (s *Store) LockOneGlobalTrans(expireIn time.Duration) *storage.TransGlobalS // ResetCronTime rest nextCronTime // Prevent multiple backoff from causing NextCronTime to be too long -func (s *Store) ResetCronTime(timeout time.Duration, limit int64) error { +func (s *Store) ResetCronTime(timeout time.Duration, limit int64) (succeedCount int64, hasRemaining bool, err error) { next := time.Now() var trans *storage.TransGlobalStore min := fmt.Sprintf("%d", time.Now().Add(timeout).Unix()) - err := s.boltDb.Update(func(t *bolt.Tx) error { - + err = s.boltDb.Update(func(t *bolt.Tx) error { cursor := t.Bucket(bucketIndex).Cursor() - i := 0 - for k, v := cursor.Seek([]byte(min)); k != nil && i < int(limit); k, v = cursor.Next() { + succeedCount = 0 + for k, v := cursor.Seek([]byte(min)); k != nil && succeedCount <= limit; k, v = cursor.Next() { + if succeedCount == limit { + hasRemaining = true + break + } + trans = tGetGlobal(t, string(v)) err := t.Bucket(bucketIndex).Delete(k) dtmimp.E2P(err) @@ -431,9 +435,9 @@ func (s *Store) ResetCronTime(timeout time.Duration, limit int64) error { trans.NextCronTime = &next tPutGlobal(t, trans) tPutIndex(t, next.Unix(), trans.Gid) - i++ + succeedCount++ } return nil }) - return err + return } diff --git a/dtmsvr/storage/redis/redis.go b/dtmsvr/storage/redis/redis.go index d4200c7..0dcf992 100644 --- a/dtmsvr/storage/redis/redis.go +++ b/dtmsvr/storage/redis/redis.go @@ -262,28 +262,34 @@ return gid // ResetCronTime rest nextCronTime // Prevent multiple backoff from causing NextCronTime to be too long -func (s *Store) ResetCronTime(timeout time.Duration, limit int64) error { +func (s *Store) ResetCronTime(timeout time.Duration, limit int64) (succeedCount int64, hasRemaining bool, err error) { next := time.Now().Unix() timeoutTimestamp := time.Now().Add(timeout).Unix() args := newArgList().AppendGid("").AppendRaw(timeoutTimestamp).AppendRaw(next).AppendRaw(limit) lua := `-- ResetCronTime -local r = redis.call('ZRANGEBYSCORE', KEYS[3], ARGV[3], '+inf', 'WITHSCORES', 'LIMIT', 0, ARGV[5]) -local index = 1 -while(true) -do - local gid = r[index] - if gid == nil then - break +local r = redis.call('ZRANGEBYSCORE', KEYS[3], ARGV[3], '+inf', 'LIMIT', 0, ARGV[5]+1) +local i = 0 +for score,gid in pairs(r) do + if i == tonumber(ARGV[5]) then + i = i + 1 + break end redis.call('ZADD', KEYS[3], ARGV[4], gid) - index = index + 2 + i = i + 1 end +return tostring(i) ` - _, err := callLua(args, lua) - if errors.Is(err, storage.ErrNotFound) { - return nil + r := "" + r, err = callLua(args, lua) + if err != nil { + return } - return err + succeedCount = int64(dtmimp.MustAtoi(r)) + if succeedCount > limit { + hasRemaining = true + succeedCount = limit + } + return } // TouchCronTime updates cronTime diff --git a/dtmsvr/storage/sql/sql.go b/dtmsvr/storage/sql/sql.go index d263955..566d2f8 100644 --- a/dtmsvr/storage/sql/sql.go +++ b/dtmsvr/storage/sql/sql.go @@ -158,7 +158,7 @@ func (s *Store) LockOneGlobalTrans(expireIn time.Duration) *storage.TransGlobalS // ResetCronTime rest nextCronTime // Prevent multiple backoff from causing NextCronTime to be too long -func (s *Store) ResetCronTime(timeout time.Duration, limit int64) error { +func (s *Store) ResetCronTime(timeout time.Duration, limit int64) (succeedCount int64, hasRemaining bool, err error) { db := dbGet() getTime := func(second int) string { return map[string]string{ @@ -176,7 +176,16 @@ func (s *Store) ResetCronTime(timeout time.Duration, limit int64) error { Updates(&storage.TransGlobalStore{ NextCronTime: dtmutil.GetNextTime(0), }) - return dbr.Error + succeedCount = dbr.RowsAffected + if succeedCount == limit { + var count int64 + db.Must().Model(global).Where(whereTime + "and status in ('prepared', 'aborting', 'submitted')").Limit(1).Count(&count) + if count > 0 { + hasRemaining = true + } + } + + return succeedCount, hasRemaining, dbr.Error } // SetDBConn sets db conn pool diff --git a/dtmsvr/storage/store.go b/dtmsvr/storage/store.go index 55e822e..391c4dd 100644 --- a/dtmsvr/storage/store.go +++ b/dtmsvr/storage/store.go @@ -30,5 +30,5 @@ type Store interface { ChangeGlobalStatus(global *TransGlobalStore, newStatus string, updates []string, finished bool) TouchCronTime(global *TransGlobalStore, nextCronInterval int64, nextCronTime *time.Time) LockOneGlobalTrans(expireIn time.Duration) *TransGlobalStore - ResetCronTime(timeout time.Duration, limit int64) error + ResetCronTime(timeout time.Duration, limit int64) (succeedCount int64, hasRemaining bool, err error) } diff --git a/test/api_test.go b/test/api_test.go index 5db7fc3..54e869d 100644 --- a/test/api_test.go +++ b/test/api_test.go @@ -82,14 +82,21 @@ func TestDtmMetrics(t *testing.T) { } func TestAPIResetCronTime(t *testing.T) { - testStoreResetCronTime(t, dtmimp.GetFuncName(), func(timeout int64, limit int64) error { + testStoreResetCronTime(t, dtmimp.GetFuncName(), func(timeout int64, limit int64) (int64, bool, error) { sTimeout := strconv.FormatInt(timeout, 10) sLimit := strconv.FormatInt(limit, 10) - _, err := dtmimp.RestyClient.R().SetQueryParams(map[string]string{ + resp, err := dtmimp.RestyClient.R().SetQueryParams(map[string]string{ "timeout": sTimeout, "limit": sLimit, }).Get(dtmutil.DefaultHTTPServer + "/resetCronTime") - return err + + m := map[string]interface{}{} + dtmimp.MustUnmarshalString(resp.String(), &m) + hasRemaining, ok := m["has_remaining"].(bool) + assert.Equal(t, ok, true) + succeedCount, ok := m["succeed_count"].(float64) + assert.Equal(t, ok, true) + return int64(succeedCount), hasRemaining, err }) } diff --git a/test/store_test.go b/test/store_test.go index 75503f1..b9bb31e 100644 --- a/test/store_test.go +++ b/test/store_test.go @@ -95,12 +95,12 @@ func TestStoreLockTrans(t *testing.T) { func TestStoreResetCronTime(t *testing.T) { s := registry.GetStore() - testStoreResetCronTime(t, dtmimp.GetFuncName(), func(timeout int64, limit int64) error { + testStoreResetCronTime(t, dtmimp.GetFuncName(), func(timeout int64, limit int64) (int64, bool, error) { return s.ResetCronTime(time.Duration(timeout)*time.Second, limit) }) } -func testStoreResetCronTime(t *testing.T, funcName string, restCronHandler func(expire int64, limit int64) error) { +func testStoreResetCronTime(t *testing.T, funcName string, restCronHandler func(expire int64, limit int64) (int64, bool, error)) { s := registry.GetStore() var restTimeTimeout, lockExpireIn, limit, i int64 restTimeTimeout = 100 //The time that will be ResetCronTime @@ -122,7 +122,9 @@ func testStoreResetCronTime(t *testing.T, funcName string, restCronHandler func( assert.Nil(t, g) // Rest limit-1 count - err := restCronHandler(restTimeTimeout, limit-1) + succeedCount, hasRemaining, err := restCronHandler(restTimeTimeout, limit-1) + assert.Equal(t, hasRemaining, true) + assert.Equal(t, succeedCount, limit-1) assert.Nil(t, err) // Fount limit-1 count for i = 0; i < limit-1; i++ { @@ -136,8 +138,10 @@ func testStoreResetCronTime(t *testing.T, funcName string, restCronHandler func( assert.Nil(t, g) // Rest 1 count - err1 := restCronHandler(restTimeTimeout, limit) - assert.Nil(t, err1) + succeedCount, hasRemaining, err = restCronHandler(restTimeTimeout, limit) + assert.Equal(t, hasRemaining, false) + assert.Equal(t, succeedCount, int64(1)) + assert.Nil(t, err) // Fount 1 count g = s.LockOneGlobalTrans(time.Duration(lockExpireIn) * time.Second) assert.NotNil(t, g) @@ -148,8 +152,10 @@ func testStoreResetCronTime(t *testing.T, funcName string, restCronHandler func( assert.Nil(t, g) // reduce the restTimeTimeout, Rest 1 count - err2 := restCronHandler(restTimeTimeout-12, limit) - assert.Nil(t, err2) + succeedCount, hasRemaining, err = restCronHandler(restTimeTimeout-12, limit) + assert.Equal(t, hasRemaining, false) + assert.Equal(t, succeedCount, int64(1)) + assert.Nil(t, err) // Fount 1 count g = s.LockOneGlobalTrans(time.Duration(lockExpireIn) * time.Second) assert.NotNil(t, g) @@ -159,6 +165,11 @@ func testStoreResetCronTime(t *testing.T, funcName string, restCronHandler func( g = s.LockOneGlobalTrans(time.Duration(lockExpireIn) * time.Second) assert.Nil(t, g) + // Not Fount + succeedCount, hasRemaining, err = restCronHandler(restTimeTimeout-12, limit) + assert.Equal(t, hasRemaining, false) + assert.Equal(t, succeedCount, int64(0)) + assert.Nil(t, err) } func TestUpdateBranches(t *testing.T) {