vendor: bump yunion.io/x/pkg

go.mod

@@ -144,7 +144,7 @@ require (
 	yunion.io/x/jsonutils v0.0.0-20201110084044-3e4e1cb49769
 	yunion.io/x/log v0.0.0-20200313080802-57a4ce5966b3
 	yunion.io/x/ovsdb v0.0.0-20200526071744-27bf0940cbc7
-	yunion.io/x/pkg v0.0.0-20201028134817-3ed15ee169bc
+	yunion.io/x/pkg v0.0.0-20201123083159-ca3aea986ff2
 	yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e
 	yunion.io/x/sqlchemy v0.0.0-20201116041103-013e56cab959
 	yunion.io/x/structarg v0.0.0-20200720093445-9f850fa222ce

go.sum

@@ -1138,6 +1138,8 @@ yunion.io/x/pkg v0.0.0-20200814072949-4f1b541857d6 h1:UarEDTBGkgcgc+nc+PZ75uo9M9
 yunion.io/x/pkg v0.0.0-20200814072949-4f1b541857d6/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
 yunion.io/x/pkg v0.0.0-20201028134817-3ed15ee169bc h1:cgwWI6k9LsXpq6yiOuEIprsXmPyPGUa60sOZu57e8Jo=
 yunion.io/x/pkg v0.0.0-20201028134817-3ed15ee169bc/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
+yunion.io/x/pkg v0.0.0-20201123083159-ca3aea986ff2 h1:NeCr2J8HjcIuJvEhP0rwWA1UKP8ReOv6HVf5k9YPtyA=
+yunion.io/x/pkg v0.0.0-20201123083159-ca3aea986ff2/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
 yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e h1:v+EzIadodSwkdZ/7bremd7J8J50Cise/HCylsOJngmo=
 yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e/go.mod h1:0iFKpOs1y4lbCxeOmq3Xx/0AcQoewVPwj62eRluioEo=
 yunion.io/x/sqlchemy v0.0.0-20201029091740-cd5e77b56d4b h1:xBAzDPvvmsOBjxAj0PLHMNLVdPNOlYFHAm7T3IZ2ehs=

vendor/modules.txt

@@ -1077,7 +1077,7 @@ yunion.io/x/log/hooks
 yunion.io/x/ovsdb/cli_util
 yunion.io/x/ovsdb/schema/ovn_nb
 yunion.io/x/ovsdb/types
-# yunion.io/x/pkg v0.0.0-20201028134817-3ed15ee169bc
+# yunion.io/x/pkg v0.0.0-20201123083159-ca3aea986ff2
 yunion.io/x/pkg/errors
 yunion.io/x/pkg/gotypes
 yunion.io/x/pkg/prettytable
@@ -1087,6 +1087,7 @@ yunion.io/x/pkg/tristate
 yunion.io/x/pkg/util/cache
 yunion.io/x/pkg/util/clock
 yunion.io/x/pkg/util/compare
+yunion.io/x/pkg/util/delayedwork
 yunion.io/x/pkg/util/errors
 yunion.io/x/pkg/util/fileutils
 yunion.io/x/pkg/util/filterclause

vendor/yunion.io/x/pkg/util/delayedwork/delayedwork.go

@@ -0,0 +1,139 @@
+package delayedwork
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+var (
+	maxDuration = time.Duration(290 * 365 * 24 * time.Hour)
+	maxTime     = time.Now().Add(maxDuration)
+)
+
+type DelayedWorkFunc func(context.Context)
+type delayedWork struct {
+	id       string
+	created  time.Time
+	interval time.Duration
+	deadline time.Time
+	f        DelayedWorkFunc
+
+	last    time.Time
+	dueTime time.Time
+}
+
+func (dw *delayedWork) initDueTime() {
+	t0 := dw.last.Add(dw.interval)
+	if t0.Before(dw.deadline) {
+		dw.dueTime = t0
+		return
+	}
+	dw.dueTime = dw.deadline
+}
+
+type DelayedWorkManager struct {
+	works   map[string]*delayedWork
+	worksMu *sync.Mutex
+	sigch   chan struct{}
+}
+
+func NewDelayedWorkManager() *DelayedWorkManager {
+	dwm := &DelayedWorkManager{
+		works:   map[string]*delayedWork{},
+		worksMu: &sync.Mutex{},
+		sigch:   make(chan struct{}),
+	}
+	return dwm
+}
+
+func (dwm *DelayedWorkManager) pendingCount() int {
+	dwm.worksMu.Lock()
+	defer dwm.worksMu.Unlock()
+	return len(dwm.works)
+}
+
+func (dwm *DelayedWorkManager) Start(ctx context.Context) {
+	var (
+		tmr *time.Timer
+		dw  *delayedWork
+	)
+	for {
+		tmr, dw = dwm.calRecentWork(ctx)
+		select {
+		case <-tmr.C:
+			if dw != nil {
+				dwm.worksMu.Lock()
+				delete(dwm.works, dw.id)
+				dwm.worksMu.Unlock()
+
+				go dw.f(ctx)
+			}
+		case <-dwm.sigch:
+			if !tmr.Stop() {
+				<-tmr.C
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (dwm *DelayedWorkManager) calRecentWork(ctx context.Context) (*time.Timer, *delayedWork) {
+	dwm.worksMu.Lock()
+	defer dwm.worksMu.Unlock()
+
+	var (
+		rt  = maxTime
+		rdw *delayedWork
+	)
+	// use container/heap should it be huge
+	for _, dw := range dwm.works {
+		if rt.After(dw.dueTime) {
+			rt = dw.dueTime
+			rdw = dw
+		}
+	}
+	return time.NewTimer(rt.Sub(time.Now())), rdw
+}
+
+// DelayedWorkRequest serves as an argument to DelayedWorkManager.Submit
+//
+// ID should be unique among works managed here
+//
+// Every submit of the same work as identified by ID will be delayed for at
+// most SoftDelay time
+//
+// Func will be called after either SoftDelay since each submit, or HardDelay
+// since first submit, whichever comes first
+type DelayedWorkRequest struct {
+	ID        string
+	SoftDelay time.Duration
+	HardDelay time.Duration
+	Func      DelayedWorkFunc
+}
+
+func (dwm *DelayedWorkManager) Submit(ctx context.Context, req DelayedWorkRequest) {
+	dwm.worksMu.Lock()
+	dw, ok := dwm.works[req.ID]
+	if !ok {
+		now := time.Now()
+		dw = &delayedWork{
+			id:       req.ID,
+			created:  now,
+			interval: req.SoftDelay,
+			deadline: now.Add(req.HardDelay),
+			f:        req.Func,
+
+			last: now,
+		}
+		dwm.works[req.ID] = dw
+		dw.initDueTime()
+	} else {
+		dw.last = time.Now()
+		dw.initDueTime()
+	}
+	dwm.worksMu.Unlock()
+
+	dwm.sigch <- struct{}{}
+}