init: 1.0.0

pkg/syncx/atomicbool.go

@@ -0,0 +1,46 @@
+package syncx
+
+import "sync/atomic"
+
+// An AtomicBool is an atomic implementation for boolean values.
+type AtomicBool uint32
+
+// NewAtomicBool returns an AtomicBool.
+func NewAtomicBool() *AtomicBool {
+	return new(AtomicBool)
+}
+
+// ForAtomicBool returns an AtomicBool with given val.
+func ForAtomicBool(val bool) *AtomicBool {
+	b := NewAtomicBool()
+	b.Set(val)
+	return b
+}
+
+// CompareAndSwap compares current value with given old, if equals, set to given val.
+func (b *AtomicBool) CompareAndSwap(old, val bool) bool {
+	var ov, nv uint32
+
+	if old {
+		ov = 1
+	}
+	if val {
+		nv = 1
+	}
+
+	return atomic.CompareAndSwapUint32((*uint32)(b), ov, nv)
+}
+
+// Set sets the value to v.
+func (b *AtomicBool) Set(v bool) {
+	if v {
+		atomic.StoreUint32((*uint32)(b), 1)
+	} else {
+		atomic.StoreUint32((*uint32)(b), 0)
+	}
+}
+
+// True returns true if current value is true.
+func (b *AtomicBool) True() bool {
+	return atomic.LoadUint32((*uint32)(b)) == 1
+}

pkg/syncx/atomicbool_test.go

@@ -0,0 +1,27 @@
+package syncx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAtomicBool(t *testing.T) {
+	val := ForAtomicBool(true)
+	assert.True(t, val.True())
+	val.Set(false)
+	assert.False(t, val.True())
+	val.Set(true)
+	assert.True(t, val.True())
+	val.Set(false)
+	assert.False(t, val.True())
+	ok := val.CompareAndSwap(false, true)
+	assert.True(t, ok)
+	assert.True(t, val.True())
+	ok = val.CompareAndSwap(true, false)
+	assert.True(t, ok)
+	assert.False(t, val.True())
+	ok = val.CompareAndSwap(true, false)
+	assert.False(t, ok)
+	assert.False(t, val.True())
+}

pkg/syncx/atomicduration.go

@@ -0,0 +1,36 @@
+package syncx
+
+import (
+	"sync/atomic"
+	"time"
+)
+
+// An AtomicDuration is an implementation of atomic duration.
+type AtomicDuration int64
+
+// NewAtomicDuration returns an AtomicDuration.
+func NewAtomicDuration() *AtomicDuration {
+	return new(AtomicDuration)
+}
+
+// ForAtomicDuration returns an AtomicDuration with given value.
+func ForAtomicDuration(val time.Duration) *AtomicDuration {
+	d := NewAtomicDuration()
+	d.Set(val)
+	return d
+}
+
+// CompareAndSwap compares current value with old, if equals, set the value to val.
+func (d *AtomicDuration) CompareAndSwap(old, val time.Duration) bool {
+	return atomic.CompareAndSwapInt64((*int64)(d), int64(old), int64(val))
+}
+
+// Load loads the current duration.
+func (d *AtomicDuration) Load() time.Duration {
+	return time.Duration(atomic.LoadInt64((*int64)(d)))
+}
+
+// Set sets the value to val.
+func (d *AtomicDuration) Set(val time.Duration) {
+	atomic.StoreInt64((*int64)(d), int64(val))
+}

pkg/syncx/atomicduration_test.go

@@ -0,0 +1,19 @@
+package syncx
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAtomicDuration(t *testing.T) {
+	d := ForAtomicDuration(time.Duration(100))
+	assert.Equal(t, time.Duration(100), d.Load())
+	d.Set(time.Duration(200))
+	assert.Equal(t, time.Duration(200), d.Load())
+	assert.True(t, d.CompareAndSwap(time.Duration(200), time.Duration(300)))
+	assert.Equal(t, time.Duration(300), d.Load())
+	assert.False(t, d.CompareAndSwap(time.Duration(200), time.Duration(400)))
+	assert.Equal(t, time.Duration(300), d.Load())
+}

pkg/syncx/atomicfloat64.go

@@ -0,0 +1,47 @@
+package syncx
+
+import (
+	"math"
+	"sync/atomic"
+)
+
+// An AtomicFloat64 is an implementation of atomic float64.
+type AtomicFloat64 uint64
+
+// NewAtomicFloat64 returns an AtomicFloat64.
+func NewAtomicFloat64() *AtomicFloat64 {
+	return new(AtomicFloat64)
+}
+
+// ForAtomicFloat64 returns an AtomicFloat64 with given val.
+func ForAtomicFloat64(val float64) *AtomicFloat64 {
+	f := NewAtomicFloat64()
+	f.Set(val)
+	return f
+}
+
+// Add adds val to current value.
+func (f *AtomicFloat64) Add(val float64) float64 {
+	for {
+		old := f.Load()
+		nv := old + val
+		if f.CompareAndSwap(old, nv) {
+			return nv
+		}
+	}
+}
+
+// CompareAndSwap compares current value with old, if equals, set the value to val.
+func (f *AtomicFloat64) CompareAndSwap(old, val float64) bool {
+	return atomic.CompareAndSwapUint64((*uint64)(f), math.Float64bits(old), math.Float64bits(val))
+}
+
+// Load loads the current value.
+func (f *AtomicFloat64) Load() float64 {
+	return math.Float64frombits(atomic.LoadUint64((*uint64)(f)))
+}
+
+// Set sets the current value to val.
+func (f *AtomicFloat64) Set(val float64) {
+	atomic.StoreUint64((*uint64)(f), math.Float64bits(val))
+}

pkg/syncx/atomicfloat64_test.go

@@ -0,0 +1,24 @@
+package syncx
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAtomicFloat64(t *testing.T) {
+	f := ForAtomicFloat64(100)
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			for i := 0; i < 100; i++ {
+				f.Add(1)
+			}
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	assert.Equal(t, float64(600), f.Load())
+}

pkg/syncx/barrier.go

@@ -0,0 +1,20 @@
+package syncx
+
+import "sync"
+
+// A Barrier is used to facility the barrier on a resource.
+type Barrier struct {
+	lock sync.Mutex
+}
+
+// Guard guards the given fn on the resource.
+func (b *Barrier) Guard(fn func()) {
+	Guard(&b.lock, fn)
+}
+
+// Guard guards the given fn with lock.
+func Guard(lock sync.Locker, fn func()) {
+	lock.Lock()
+	defer lock.Unlock()
+	fn()
+}

pkg/syncx/barrier_test.go

@@ -0,0 +1,56 @@
+package syncx
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBarrier_Guard(t *testing.T) {
+	const total = 10000
+	var barrier Barrier
+	var count int
+	var wg sync.WaitGroup
+	wg.Add(total)
+	for i := 0; i < total; i++ {
+		go barrier.Guard(func() {
+			count++
+			wg.Done()
+		})
+	}
+	wg.Wait()
+	assert.Equal(t, total, count)
+}
+
+func TestBarrierPtr_Guard(t *testing.T) {
+	const total = 10000
+	barrier := new(Barrier)
+	var count int
+	wg := new(sync.WaitGroup)
+	wg.Add(total)
+	for i := 0; i < total; i++ {
+		go barrier.Guard(func() {
+			count++
+			wg.Done()
+		})
+	}
+	wg.Wait()
+	assert.Equal(t, total, count)
+}
+
+func TestGuard(t *testing.T) {
+	const total = 10000
+	var count int
+	var lock sync.Mutex
+	wg := new(sync.WaitGroup)
+	wg.Add(total)
+	for i := 0; i < total; i++ {
+		go Guard(&lock, func() {
+			count++
+			wg.Done()
+		})
+	}
+	wg.Wait()
+	assert.Equal(t, total, count)
+}

pkg/syncx/cond.go

@@ -0,0 +1,49 @@
+package syncx
+
+import (
+	"time"
+
+	"github.com/perfect-panel/ppanel-server/pkg/lang"
+	"github.com/perfect-panel/ppanel-server/pkg/timex"
+)
+
+// A Cond is used to wait for conditions.
+type Cond struct {
+	signal chan lang.PlaceholderType
+}
+
+// NewCond returns a Cond.
+func NewCond() *Cond {
+	return &Cond{
+		signal: make(chan lang.PlaceholderType),
+	}
+}
+
+// WaitWithTimeout wait for signal return remain wait time or timed out.
+func (cond *Cond) WaitWithTimeout(timeout time.Duration) (time.Duration, bool) {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	begin := timex.Now()
+	select {
+	case <-cond.signal:
+		elapsed := timex.Since(begin)
+		remainTimeout := timeout - elapsed
+		return remainTimeout, true
+	case <-timer.C:
+		return 0, false
+	}
+}
+
+// Wait waits for signals.
+func (cond *Cond) Wait() {
+	<-cond.signal
+}
+
+// Signal wakes one goroutine waiting on c, if there is any.
+func (cond *Cond) Signal() {
+	select {
+	case cond.signal <- lang.Placeholder:
+	default:
+	}
+}

pkg/syncx/cond_test.go

@@ -0,0 +1,69 @@
+package syncx
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTimeoutCondWait(t *testing.T) {
+	var wait sync.WaitGroup
+	cond := NewCond()
+	wait.Add(2)
+	go func() {
+		cond.Wait()
+		wait.Done()
+	}()
+	time.Sleep(time.Duration(50) * time.Millisecond)
+	go func() {
+		cond.Signal()
+		wait.Done()
+	}()
+	wait.Wait()
+}
+
+func TestTimeoutCondWaitTimeout(t *testing.T) {
+	var wait sync.WaitGroup
+	cond := NewCond()
+	wait.Add(1)
+	go func() {
+		cond.WaitWithTimeout(time.Duration(500) * time.Millisecond)
+		wait.Done()
+	}()
+	wait.Wait()
+}
+
+func TestTimeoutCondWaitTimeoutRemain(t *testing.T) {
+	var wait sync.WaitGroup
+	cond := NewCond()
+	wait.Add(2)
+	ch := make(chan time.Duration, 1)
+	defer close(ch)
+	timeout := time.Duration(2000) * time.Millisecond
+	go func() {
+		remainTimeout, _ := cond.WaitWithTimeout(timeout)
+		ch <- remainTimeout
+		wait.Done()
+	}()
+	sleep(200)
+	go func() {
+		cond.Signal()
+		wait.Done()
+	}()
+	wait.Wait()
+	remainTimeout := <-ch
+	assert.True(t, remainTimeout < timeout, "expect remainTimeout %v < %v", remainTimeout, timeout)
+	assert.True(t, remainTimeout >= time.Duration(200)*time.Millisecond,
+		"expect remainTimeout %v >= 200 millisecond", remainTimeout)
+}
+
+func TestSignalNoWait(t *testing.T) {
+	cond := NewCond()
+	cond.Signal()
+}
+
+func sleep(millisecond int) {
+	time.Sleep(time.Duration(millisecond) * time.Millisecond)
+}

pkg/syncx/donechan.go

@@ -0,0 +1,32 @@
+package syncx
+
+import (
+	"sync"
+
+	"github.com/perfect-panel/ppanel-server/pkg/lang"
+)
+
+// A DoneChan is used as a channel that can be closed multiple times and wait for done.
+type DoneChan struct {
+	done chan lang.PlaceholderType
+	once sync.Once
+}
+
+// NewDoneChan returns a DoneChan.
+func NewDoneChan() *DoneChan {
+	return &DoneChan{
+		done: make(chan lang.PlaceholderType),
+	}
+}
+
+// Close closes dc, it's safe to close more than once.
+func (dc *DoneChan) Close() {
+	dc.once.Do(func() {
+		close(dc.done)
+	})
+}
+
+// Done returns a channel that can be notified on dc closed.
+func (dc *DoneChan) Done() chan lang.PlaceholderType {
+	return dc.done
+}

pkg/syncx/donechan_test.go

@@ -0,0 +1,31 @@
+package syncx
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestDoneChanClose(t *testing.T) {
+	doneChan := NewDoneChan()
+
+	for i := 0; i < 5; i++ {
+		doneChan.Close()
+	}
+}
+
+func TestDoneChanDone(t *testing.T) {
+	var waitGroup sync.WaitGroup
+	doneChan := NewDoneChan()
+
+	waitGroup.Add(1)
+	go func() {
+		<-doneChan.Done()
+		waitGroup.Done()
+	}()
+
+	for i := 0; i < 5; i++ {
+		doneChan.Close()
+	}
+
+	waitGroup.Wait()
+}

pkg/syncx/immutableresource.go

@@ -0,0 +1,82 @@
+package syncx
+
+import (
+	"sync"
+	"time"
+
+	"github.com/perfect-panel/ppanel-server/pkg/timex"
+)
+
+const defaultRefreshInterval = time.Second
+
+type (
+	// ImmutableResourceOption defines the method to customize an ImmutableResource.
+	ImmutableResourceOption func(resource *ImmutableResource)
+
+	// An ImmutableResource is used to manage an immutable resource.
+	ImmutableResource struct {
+		fetch           func() (any, error)
+		resource        any
+		err             error
+		lock            sync.RWMutex
+		refreshInterval time.Duration
+		lastTime        *AtomicDuration
+	}
+)
+
+// NewImmutableResource returns an ImmutableResource.
+func NewImmutableResource(fn func() (any, error), opts ...ImmutableResourceOption) *ImmutableResource {
+	// cannot use executors.LessExecutor because of cycle imports
+	ir := ImmutableResource{
+		fetch:           fn,
+		refreshInterval: defaultRefreshInterval,
+		lastTime:        NewAtomicDuration(),
+	}
+	for _, opt := range opts {
+		opt(&ir)
+	}
+	return &ir
+}
+
+// Get gets the immutable resource, fetches automatically if not loaded.
+func (ir *ImmutableResource) Get() (any, error) {
+	ir.lock.RLock()
+	resource := ir.resource
+	ir.lock.RUnlock()
+	if resource != nil {
+		return resource, nil
+	}
+
+	ir.maybeRefresh(func() {
+		res, err := ir.fetch()
+		ir.lock.Lock()
+		if err != nil {
+			ir.err = err
+		} else {
+			ir.resource, ir.err = res, nil
+		}
+		ir.lock.Unlock()
+	})
+
+	ir.lock.RLock()
+	resource, err := ir.resource, ir.err
+	ir.lock.RUnlock()
+	return resource, err
+}
+
+func (ir *ImmutableResource) maybeRefresh(execute func()) {
+	now := timex.Now()
+	lastTime := ir.lastTime.Load()
+	if lastTime == 0 || lastTime+ir.refreshInterval < now {
+		ir.lastTime.Set(now)
+		execute()
+	}
+}
+
+// WithRefreshIntervalOnFailure sets refresh interval on failure.
+// Set interval to 0 to enforce refresh every time if not succeeded, default is time.Second.
+func WithRefreshIntervalOnFailure(interval time.Duration) ImmutableResourceOption {
+	return func(resource *ImmutableResource) {
+		resource.refreshInterval = interval
+	}
+}

pkg/syncx/immutableresource_test.go

@@ -0,0 +1,78 @@
+package syncx
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestImmutableResource(t *testing.T) {
+	var count int
+	r := NewImmutableResource(func() (any, error) {
+		count++
+		return "hello", nil
+	})
+
+	res, err := r.Get()
+	assert.Equal(t, "hello", res)
+	assert.Equal(t, 1, count)
+	assert.Nil(t, err)
+
+	// again
+	res, err = r.Get()
+	assert.Equal(t, "hello", res)
+	assert.Equal(t, 1, count)
+	assert.Nil(t, err)
+}
+
+func TestImmutableResourceError(t *testing.T) {
+	var count int
+	r := NewImmutableResource(func() (any, error) {
+		count++
+		return nil, errors.New("any")
+	})
+
+	res, err := r.Get()
+	assert.Nil(t, res)
+	assert.NotNil(t, err)
+	assert.Equal(t, "any", err.Error())
+	assert.Equal(t, 1, count)
+
+	// again
+	res, err = r.Get()
+	assert.Nil(t, res)
+	assert.NotNil(t, err)
+	assert.Equal(t, "any", err.Error())
+	assert.Equal(t, 1, count)
+
+	r.refreshInterval = 0
+	time.Sleep(time.Millisecond)
+	res, err = r.Get()
+	assert.Nil(t, res)
+	assert.NotNil(t, err)
+	assert.Equal(t, "any", err.Error())
+	assert.Equal(t, 2, count)
+}
+
+func TestImmutableResourceErrorRefreshAlways(t *testing.T) {
+	var count int
+	r := NewImmutableResource(func() (any, error) {
+		count++
+		return nil, errors.New("any")
+	}, WithRefreshIntervalOnFailure(0))
+
+	res, err := r.Get()
+	assert.Nil(t, res)
+	assert.NotNil(t, err)
+	assert.Equal(t, "any", err.Error())
+	assert.Equal(t, 1, count)
+
+	// again
+	res, err = r.Get()
+	assert.Nil(t, res)
+	assert.NotNil(t, err)
+	assert.Equal(t, "any", err.Error())
+	assert.Equal(t, 2, count)
+}

pkg/syncx/limit.go

@@ -0,0 +1,48 @@
+package syncx
+
+import (
+	"errors"
+
+	"github.com/perfect-panel/ppanel-server/pkg/lang"
+)
+
+// ErrLimitReturn indicates that the more than borrowed elements were returned.
+var ErrLimitReturn = errors.New("discarding limited token, resource pool is full, someone returned multiple times")
+
+// Limit controls the concurrent requests.
+type Limit struct {
+	pool chan lang.PlaceholderType
+}
+
+// NewLimit creates a Limit that can borrow n elements from it concurrently.
+func NewLimit(n int) Limit {
+	return Limit{
+		pool: make(chan lang.PlaceholderType, n),
+	}
+}
+
+// Borrow borrows an element from Limit in blocking mode.
+func (l Limit) Borrow() {
+	l.pool <- lang.Placeholder
+}
+
+// Return returns the borrowed resource, returns error only if returned more than borrowed.
+func (l Limit) Return() error {
+	select {
+	case <-l.pool:
+		return nil
+	default:
+		return ErrLimitReturn
+	}
+}
+
+// TryBorrow tries to borrow an element from Limit, in non-blocking mode.
+// If success, true returned, false for otherwise.
+func (l Limit) TryBorrow() bool {
+	select {
+	case l.pool <- lang.Placeholder:
+		return true
+	default:
+		return false
+	}
+}

pkg/syncx/limit_test.go

@@ -0,0 +1,17 @@
+package syncx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLimit(t *testing.T) {
+	limit := NewLimit(2)
+	limit.Borrow()
+	assert.True(t, limit.TryBorrow())
+	assert.False(t, limit.TryBorrow())
+	assert.Nil(t, limit.Return())
+	assert.Nil(t, limit.Return())
+	assert.Equal(t, ErrLimitReturn, limit.Return())
+}

pkg/syncx/lockedcalls.go

@@ -0,0 +1,57 @@
+package syncx
+
+import "sync"
+
+type (
+	// LockedCalls makes sure the calls with the same key to be called sequentially.
+	// For example, A called F, before it's done, B called F, then B's call would not blocked,
+	// after A's call finished, B's call got executed.
+	// The calls with the same key are independent, not sharing the returned values.
+	// A ------->calls F with key and executes<------->returns
+	// B ------------------>calls F with key<--------->executes<---->returns
+	LockedCalls interface {
+		Do(key string, fn func() (any, error)) (any, error)
+	}
+
+	lockedGroup struct {
+		mu sync.Mutex
+		m  map[string]*sync.WaitGroup
+	}
+)
+
+// NewLockedCalls returns a LockedCalls.
+func NewLockedCalls() LockedCalls {
+	return &lockedGroup{
+		m: make(map[string]*sync.WaitGroup),
+	}
+}
+
+func (lg *lockedGroup) Do(key string, fn func() (any, error)) (any, error) {
+begin:
+	lg.mu.Lock()
+	if wg, ok := lg.m[key]; ok {
+		lg.mu.Unlock()
+		wg.Wait()
+		goto begin
+	}
+
+	return lg.makeCall(key, fn)
+}
+
+func (lg *lockedGroup) makeCall(key string, fn func() (any, error)) (any, error) {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	lg.m[key] = &wg
+	lg.mu.Unlock()
+
+	defer func() {
+		// delete key first, done later. can't reverse the order, because if reverse,
+		// another Do call might wg.Wait() without get notified with wg.Done()
+		lg.mu.Lock()
+		delete(lg.m, key)
+		lg.mu.Unlock()
+		wg.Done()
+	}()
+
+	return fn()
+}

pkg/syncx/lockedcalls_test.go

@@ -0,0 +1,82 @@
+package syncx
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestLockedCallDo(t *testing.T) {
+	g := NewLockedCalls()
+	v, err := g.Do("key", func() (any, error) {
+		return "bar", nil
+	})
+	if got, want := fmt.Sprintf("%v (%T)", v, v), "bar (string)"; got != want {
+		t.Errorf("Do = %v; want %v", got, want)
+	}
+	if err != nil {
+		t.Errorf("Do error = %v", err)
+	}
+}
+
+func TestLockedCallDoErr(t *testing.T) {
+	g := NewLockedCalls()
+	someErr := errors.New("some error")
+	v, err := g.Do("key", func() (any, error) {
+		return nil, someErr
+	})
+	if !errors.Is(err, someErr) {
+		t.Errorf("Do error = %v; want someErr", err)
+	}
+	if v != nil {
+		t.Errorf("unexpected non-nil value %#v", v)
+	}
+}
+
+func TestLockedCallDoDupSuppress(t *testing.T) {
+	g := NewLockedCalls()
+	c := make(chan string)
+	var calls int
+	fn := func() (any, error) {
+		calls++
+		ret := calls
+		<-c
+		calls--
+		return ret, nil
+	}
+
+	const n = 10
+	var results []int
+	var lock sync.Mutex
+	var wg sync.WaitGroup
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go func() {
+			v, err := g.Do("key", fn)
+			if err != nil {
+				t.Errorf("Do error: %v", err)
+			}
+
+			lock.Lock()
+			results = append(results, v.(int))
+			lock.Unlock()
+			wg.Done()
+		}()
+	}
+	time.Sleep(100 * time.Millisecond) // let goroutines above block
+	for i := 0; i < n; i++ {
+		c <- "bar"
+	}
+	wg.Wait()
+
+	lock.Lock()
+	defer lock.Unlock()
+
+	for _, item := range results {
+		if item != 1 {
+			t.Errorf("number of calls = %d; want 1", item)
+		}
+	}
+}

pkg/syncx/managedresource.go

@@ -0,0 +1,48 @@
+package syncx
+
+import "sync"
+
+// A ManagedResource is used to manage a resource that might be broken and refetched, like a connection.
+type ManagedResource struct {
+	resource any
+	lock     sync.RWMutex
+	generate func() any
+	equals   func(a, b any) bool
+}
+
+// NewManagedResource returns a ManagedResource.
+func NewManagedResource(generate func() any, equals func(a, b any) bool) *ManagedResource {
+	return &ManagedResource{
+		generate: generate,
+		equals:   equals,
+	}
+}
+
+// MarkBroken marks the resource broken.
+func (mr *ManagedResource) MarkBroken(resource any) {
+	mr.lock.Lock()
+	defer mr.lock.Unlock()
+
+	if mr.equals(mr.resource, resource) {
+		mr.resource = nil
+	}
+}
+
+// Take takes the resource, if not loaded, generates it.
+func (mr *ManagedResource) Take() any {
+	mr.lock.RLock()
+	resource := mr.resource
+	mr.lock.RUnlock()
+
+	if resource != nil {
+		return resource
+	}
+
+	mr.lock.Lock()
+	defer mr.lock.Unlock()
+	// maybe another Take() call already generated the resource.
+	if mr.resource == nil {
+		mr.resource = mr.generate()
+	}
+	return mr.resource
+}

pkg/syncx/managedresource_test.go

@@ -0,0 +1,22 @@
+package syncx
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestManagedResource(t *testing.T) {
+	var count int32
+	resource := NewManagedResource(func() any {
+		return atomic.AddInt32(&count, 1)
+	}, func(a, b any) bool {
+		return a == b
+	})
+
+	assert.Equal(t, resource.Take(), resource.Take())
+	old := resource.Take()
+	resource.MarkBroken(old)
+	assert.NotEqual(t, old, resource.Take())
+}

pkg/syncx/once.go

@@ -0,0 +1,11 @@
+package syncx
+
+import "sync"
+
+// Once returns a func that guarantees fn can only called once.
+func Once(fn func()) func() {
+	once := new(sync.Once)
+	return func() {
+		once.Do(fn)
+	}
+}

pkg/syncx/once_test.go

@@ -0,0 +1,33 @@
+package syncx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOnce(t *testing.T) {
+	var v int
+	add := Once(func() {
+		v++
+	})
+
+	for i := 0; i < 5; i++ {
+		add()
+	}
+
+	assert.Equal(t, 1, v)
+}
+
+func BenchmarkOnce(b *testing.B) {
+	var v int
+	add := Once(func() {
+		v++
+	})
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		add()
+	}
+	assert.Equal(b, 1, v)
+}

pkg/syncx/onceguard.go

@@ -0,0 +1,18 @@
+package syncx
+
+import "sync/atomic"
+
+// An OnceGuard is used to make sure a resource can be taken once.
+type OnceGuard struct {
+	done uint32
+}
+
+// Taken checks if the resource is taken.
+func (og *OnceGuard) Taken() bool {
+	return atomic.LoadUint32(&og.done) == 1
+}
+
+// Take takes the resource, returns true on success, false for otherwise.
+func (og *OnceGuard) Take() bool {
+	return atomic.CompareAndSwapUint32(&og.done, 0, 1)
+}

pkg/syncx/onceguard_test.go

@@ -0,0 +1,17 @@
+package syncx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOnceGuard(t *testing.T) {
+	var guard OnceGuard
+
+	assert.False(t, guard.Taken())
+	assert.True(t, guard.Take())
+	assert.True(t, guard.Taken())
+	assert.False(t, guard.Take())
+	assert.True(t, guard.Taken())
+}

pkg/syncx/pool.go

@@ -0,0 +1,108 @@
+package syncx
+
+import (
+	"sync"
+	"time"
+
+	"github.com/perfect-panel/ppanel-server/pkg/timex"
+)
+
+type (
+	// PoolOption defines the method to customize a Pool.
+	PoolOption func(*Pool)
+
+	node struct {
+		item     any
+		next     *node
+		lastUsed time.Duration
+	}
+
+	// A Pool is used to pool resources.
+	// The difference between sync.Pool is that:
+	//  1. the limit of the resources
+	//  2. max age of the resources can be set
+	//  3. the method to destroy resources can be customized
+	Pool struct {
+		limit   int
+		created int
+		maxAge  time.Duration
+		lock    sync.Locker
+		cond    *sync.Cond
+		head    *node
+		create  func() any
+		destroy func(any)
+	}
+)
+
+// NewPool returns a Pool.
+func NewPool(n int, create func() any, destroy func(any), opts ...PoolOption) *Pool {
+	if n <= 0 {
+		panic("pool size can't be negative or zero")
+	}
+
+	lock := new(sync.Mutex)
+	pool := &Pool{
+		limit:   n,
+		lock:    lock,
+		cond:    sync.NewCond(lock),
+		create:  create,
+		destroy: destroy,
+	}
+
+	for _, opt := range opts {
+		opt(pool)
+	}
+
+	return pool
+}
+
+// Get gets a resource.
+func (p *Pool) Get() any {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	for {
+		if p.head != nil {
+			head := p.head
+			p.head = head.next
+			if p.maxAge > 0 && head.lastUsed+p.maxAge < timex.Now() {
+				p.created--
+				p.destroy(head.item)
+				continue
+			} else {
+				return head.item
+			}
+		}
+
+		if p.created < p.limit {
+			p.created++
+			return p.create()
+		}
+
+		p.cond.Wait()
+	}
+}
+
+// Put puts a resource back.
+func (p *Pool) Put(x any) {
+	if x == nil {
+		return
+	}
+
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	p.head = &node{
+		item:     x,
+		next:     p.head,
+		lastUsed: timex.Now(),
+	}
+	p.cond.Signal()
+}
+
+// WithMaxAge returns a function to customize a Pool with given max age.
+func WithMaxAge(duration time.Duration) PoolOption {
+	return func(pool *Pool) {
+		pool.maxAge = duration
+	}
+}

pkg/syncx/pool_test.go

@@ -0,0 +1,115 @@
+package syncx
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/perfect-panel/ppanel-server/pkg/lang"
+	"github.com/stretchr/testify/assert"
+)
+
+const limit = 10
+
+func TestPoolGet(t *testing.T) {
+	stack := NewPool(limit, create, destroy)
+	ch := make(chan lang.PlaceholderType)
+
+	for i := 0; i < limit; i++ {
+		var fail AtomicBool
+		go func() {
+			v := stack.Get()
+			if v.(int) != 1 {
+				fail.Set(true)
+			}
+			ch <- lang.Placeholder
+		}()
+
+		select {
+		case <-ch:
+		case <-time.After(time.Second):
+			t.Fail()
+		}
+
+		if fail.True() {
+			t.Fatal("unmatch value")
+		}
+	}
+}
+
+func TestPoolPopTooMany(t *testing.T) {
+	stack := NewPool(limit, create, destroy)
+	ch := make(chan lang.PlaceholderType, 1)
+
+	for i := 0; i < limit; i++ {
+		var wait sync.WaitGroup
+		wait.Add(1)
+		go func() {
+			stack.Get()
+			ch <- lang.Placeholder
+			wait.Done()
+		}()
+
+		wait.Wait()
+		select {
+		case <-ch:
+		default:
+			t.Fail()
+		}
+	}
+
+	var waitGroup, pushWait sync.WaitGroup
+	waitGroup.Add(1)
+	pushWait.Add(1)
+	go func() {
+		pushWait.Done()
+		stack.Get()
+		waitGroup.Done()
+	}()
+
+	pushWait.Wait()
+	stack.Put(1)
+	waitGroup.Wait()
+}
+
+func TestPoolPopFirst(t *testing.T) {
+	var value int32
+	stack := NewPool(limit, func() any {
+		return atomic.AddInt32(&value, 1)
+	}, destroy)
+
+	for i := 0; i < 100; i++ {
+		v := stack.Get().(int32)
+		assert.Equal(t, 1, int(v))
+		stack.Put(v)
+	}
+}
+
+func TestPoolWithMaxAge(t *testing.T) {
+	var value int32
+	stack := NewPool(limit, func() any {
+		return atomic.AddInt32(&value, 1)
+	}, destroy, WithMaxAge(time.Millisecond))
+
+	v1 := stack.Get().(int32)
+	// put nil should not matter
+	stack.Put(nil)
+	stack.Put(v1)
+	time.Sleep(time.Millisecond * 10)
+	v2 := stack.Get().(int32)
+	assert.NotEqual(t, v1, v2)
+}
+
+func TestNewPoolPanics(t *testing.T) {
+	assert.Panics(t, func() {
+		NewPool(0, create, destroy)
+	})
+}
+
+func create() any {
+	return 1
+}
+
+func destroy(_ any) {
+}

pkg/syncx/refresource.go

@@ -0,0 +1,53 @@
+package syncx
+
+import (
+	"errors"
+	"sync"
+)
+
+// ErrUseOfCleaned is an error that indicates using a cleaned resource.
+var ErrUseOfCleaned = errors.New("using a cleaned resource")
+
+// A RefResource is used to reference counting a resource.
+type RefResource struct {
+	lock    sync.Mutex
+	ref     int32
+	cleaned bool
+	clean   func()
+}
+
+// NewRefResource returns a RefResource.
+func NewRefResource(clean func()) *RefResource {
+	return &RefResource{
+		clean: clean,
+	}
+}
+
+// Use uses the resource with reference count incremented.
+func (r *RefResource) Use() error {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	if r.cleaned {
+		return ErrUseOfCleaned
+	}
+
+	r.ref++
+	return nil
+}
+
+// Clean cleans a resource with reference count decremented.
+func (r *RefResource) Clean() {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	if r.cleaned {
+		return
+	}
+
+	r.ref--
+	if r.ref == 0 {
+		r.cleaned = true
+		r.clean()
+	}
+}

pkg/syncx/refresource_test.go

@@ -0,0 +1,27 @@
+package syncx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRefCleaner(t *testing.T) {
+	var count int
+	clean := func() {
+		count += 1
+	}
+
+	cleaner := NewRefResource(clean)
+	err := cleaner.Use()
+	assert.Nil(t, err)
+	err = cleaner.Use()
+	assert.Nil(t, err)
+	cleaner.Clean()
+	cleaner.Clean()
+	assert.Equal(t, 1, count)
+	cleaner.Clean()
+	cleaner.Clean()
+	assert.Equal(t, 1, count)
+	assert.Equal(t, ErrUseOfCleaned, cleaner.Use())
+}

pkg/syncx/resourcemanager.go

@@ -0,0 +1,78 @@
+package syncx
+
+import (
+	"io"
+	"sync"
+
+	"github.com/perfect-panel/ppanel-server/pkg/errorx"
+)
+
+// A ResourceManager is a manager that used to manage resources.
+type ResourceManager struct {
+	resources    map[string]io.Closer
+	singleFlight SingleFlight
+	lock         sync.RWMutex
+}
+
+// NewResourceManager returns a ResourceManager.
+func NewResourceManager() *ResourceManager {
+	return &ResourceManager{
+		resources:    make(map[string]io.Closer),
+		singleFlight: NewSingleFlight(),
+	}
+}
+
+// Close closes the manager.
+// Don't use the ResourceManager after Close() called.
+func (manager *ResourceManager) Close() error {
+	manager.lock.Lock()
+	defer manager.lock.Unlock()
+
+	var be errorx.BatchError
+	for _, resource := range manager.resources {
+		if err := resource.Close(); err != nil {
+			be.Add(err)
+		}
+	}
+
+	// release resources to avoid using it later
+	manager.resources = nil
+
+	return be.Err()
+}
+
+// GetResource returns the resource associated with given key.
+func (manager *ResourceManager) GetResource(key string, create func() (io.Closer, error)) (
+	io.Closer, error) {
+	val, err := manager.singleFlight.Do(key, func() (any, error) {
+		manager.lock.RLock()
+		resource, ok := manager.resources[key]
+		manager.lock.RUnlock()
+		if ok {
+			return resource, nil
+		}
+
+		resource, err := create()
+		if err != nil {
+			return nil, err
+		}
+
+		manager.lock.Lock()
+		defer manager.lock.Unlock()
+		manager.resources[key] = resource
+
+		return resource, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return val.(io.Closer), nil
+}
+
+// Inject injects the resource associated with given key.
+func (manager *ResourceManager) Inject(key string, resource io.Closer) {
+	manager.lock.Lock()
+	manager.resources[key] = resource
+	manager.lock.Unlock()
+}

pkg/syncx/resourcemanager_test.go

@@ -0,0 +1,99 @@
+package syncx
+
+import (
+	"errors"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type dummyResource struct {
+	age int
+}
+
+func (dr *dummyResource) Close() error {
+	return errors.New("close")
+}
+
+func TestResourceManager_GetResource(t *testing.T) {
+	manager := NewResourceManager()
+	defer manager.Close()
+
+	var age int
+	for i := 0; i < 10; i++ {
+		val, err := manager.GetResource("key", func() (io.Closer, error) {
+			age++
+			return &dummyResource{
+				age: age,
+			}, nil
+		})
+		assert.Nil(t, err)
+		assert.Equal(t, 1, val.(*dummyResource).age)
+	}
+}
+
+func TestResourceManager_GetResourceError(t *testing.T) {
+	manager := NewResourceManager()
+	defer manager.Close()
+
+	for i := 0; i < 10; i++ {
+		_, err := manager.GetResource("key", func() (io.Closer, error) {
+			return nil, errors.New("fail")
+		})
+		assert.NotNil(t, err)
+	}
+}
+
+func TestResourceManager_Close(t *testing.T) {
+	manager := NewResourceManager()
+	defer manager.Close()
+
+	for i := 0; i < 10; i++ {
+		_, err := manager.GetResource("key", func() (io.Closer, error) {
+			return nil, errors.New("fail")
+		})
+		assert.NotNil(t, err)
+	}
+
+	if assert.NoError(t, manager.Close()) {
+		assert.Equal(t, 0, len(manager.resources))
+	}
+}
+
+func TestResourceManager_UseAfterClose(t *testing.T) {
+	manager := NewResourceManager()
+	defer manager.Close()
+
+	_, err := manager.GetResource("key", func() (io.Closer, error) {
+		return nil, errors.New("fail")
+	})
+	assert.NotNil(t, err)
+	if assert.NoError(t, manager.Close()) {
+		_, err = manager.GetResource("key", func() (io.Closer, error) {
+			return nil, errors.New("fail")
+		})
+		assert.NotNil(t, err)
+
+		assert.Panics(t, func() {
+			_, err = manager.GetResource("key", func() (io.Closer, error) {
+				return &dummyResource{age: 123}, nil
+			})
+		})
+	}
+}
+
+func TestResourceManager_Inject(t *testing.T) {
+	manager := NewResourceManager()
+	defer manager.Close()
+
+	manager.Inject("key", &dummyResource{
+		age: 10,
+	})
+
+	val, err := manager.GetResource("key", func() (io.Closer, error) {
+		return nil, nil
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, 10, val.(*dummyResource).age)
+}

pkg/syncx/singleflight.go

@@ -0,0 +1,81 @@
+package syncx
+
+import "sync"
+
+type (
+	// SingleFlight lets the concurrent calls with the same key to share the call result.
+	// For example, A called F, before it's done, B called F. Then B would not execute F,
+	// and shared the result returned by F which called by A.
+	// The calls with the same key are dependent, concurrent calls share the returned values.
+	// A ------->calls F with key<------------------->returns val
+	// B --------------------->calls F with key------>returns val
+	SingleFlight interface {
+		Do(key string, fn func() (any, error)) (any, error)
+		DoEx(key string, fn func() (any, error)) (any, bool, error)
+	}
+
+	call struct {
+		wg  sync.WaitGroup
+		val any
+		err error
+	}
+
+	flightGroup struct {
+		calls map[string]*call
+		lock  sync.Mutex
+	}
+)
+
+// NewSingleFlight returns a SingleFlight.
+func NewSingleFlight() SingleFlight {
+	return &flightGroup{
+		calls: make(map[string]*call),
+	}
+}
+
+func (g *flightGroup) Do(key string, fn func() (any, error)) (any, error) {
+	c, done := g.createCall(key)
+	if done {
+		return c.val, c.err
+	}
+
+	g.makeCall(c, key, fn)
+	return c.val, c.err
+}
+
+func (g *flightGroup) DoEx(key string, fn func() (any, error)) (val any, fresh bool, err error) {
+	c, done := g.createCall(key)
+	if done {
+		return c.val, false, c.err
+	}
+
+	g.makeCall(c, key, fn)
+	return c.val, true, c.err
+}
+
+func (g *flightGroup) createCall(key string) (c *call, done bool) {
+	g.lock.Lock()
+	if c, ok := g.calls[key]; ok {
+		g.lock.Unlock()
+		c.wg.Wait()
+		return c, true
+	}
+
+	c = new(call)
+	c.wg.Add(1)
+	g.calls[key] = c
+	g.lock.Unlock()
+
+	return c, false
+}
+
+func (g *flightGroup) makeCall(c *call, key string, fn func() (any, error)) {
+	defer func() {
+		g.lock.Lock()
+		delete(g.calls, key)
+		g.lock.Unlock()
+		c.wg.Done()
+	}()
+
+	c.val, c.err = fn()
+}

pkg/syncx/singleflight_test.go

@@ -0,0 +1,141 @@
+package syncx
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestExclusiveCallDo(t *testing.T) {
+	g := NewSingleFlight()
+	v, err := g.Do("key", func() (any, error) {
+		return "bar", nil
+	})
+	if got, want := fmt.Sprintf("%v (%T)", v, v), "bar (string)"; got != want {
+		t.Errorf("Do = %v; want %v", got, want)
+	}
+	if err != nil {
+		t.Errorf("Do error = %v", err)
+	}
+}
+
+func TestExclusiveCallDoErr(t *testing.T) {
+	g := NewSingleFlight()
+	someErr := errors.New("some error")
+	v, err := g.Do("key", func() (any, error) {
+		return nil, someErr
+	})
+	if !errors.Is(err, someErr) {
+		t.Errorf("Do error = %v; want someErr", err)
+	}
+	if v != nil {
+		t.Errorf("unexpected non-nil value %#v", v)
+	}
+}
+
+func TestExclusiveCallDoDupSuppress(t *testing.T) {
+	g := NewSingleFlight()
+	c := make(chan string)
+	var calls int32
+	fn := func() (any, error) {
+		atomic.AddInt32(&calls, 1)
+		return <-c, nil
+	}
+
+	const n = 10
+	var wg sync.WaitGroup
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go func() {
+			v, err := g.Do("key", fn)
+			if err != nil {
+				t.Errorf("Do error: %v", err)
+			}
+			if v.(string) != "bar" {
+				t.Errorf("got %q; want %q", v, "bar")
+			}
+			wg.Done()
+		}()
+	}
+	time.Sleep(100 * time.Millisecond) // let goroutines above block
+	c <- "bar"
+	wg.Wait()
+	if got := atomic.LoadInt32(&calls); got != 1 {
+		t.Errorf("number of calls = %d; want 1", got)
+	}
+}
+
+func TestExclusiveCallDoDiffDupSuppress(t *testing.T) {
+	g := NewSingleFlight()
+	broadcast := make(chan struct{})
+	var calls int32
+	tests := []string{"e", "a", "e", "a", "b", "c", "b", "a", "c", "d", "b", "c", "d"}
+
+	var wg sync.WaitGroup
+	for _, key := range tests {
+		wg.Add(1)
+		go func(k string) {
+			<-broadcast // get all goroutines ready
+			_, err := g.Do(k, func() (any, error) {
+				atomic.AddInt32(&calls, 1)
+				time.Sleep(10 * time.Millisecond)
+				return nil, nil
+			})
+			if err != nil {
+				t.Errorf("Do error: %v", err)
+			}
+			wg.Done()
+		}(key)
+	}
+
+	time.Sleep(100 * time.Millisecond) // let goroutines above block
+	close(broadcast)
+	wg.Wait()
+
+	if got := atomic.LoadInt32(&calls); got != 5 {
+		// five letters
+		t.Errorf("number of calls = %d; want 5", got)
+	}
+}
+
+func TestExclusiveCallDoExDupSuppress(t *testing.T) {
+	g := NewSingleFlight()
+	c := make(chan string)
+	var calls int32
+	fn := func() (any, error) {
+		atomic.AddInt32(&calls, 1)
+		return <-c, nil
+	}
+
+	const n = 10
+	var wg sync.WaitGroup
+	var freshes int32
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go func() {
+			v, fresh, err := g.DoEx("key", fn)
+			if err != nil {
+				t.Errorf("Do error: %v", err)
+			}
+			if fresh {
+				atomic.AddInt32(&freshes, 1)
+			}
+			if v.(string) != "bar" {
+				t.Errorf("got %q; want %q", v, "bar")
+			}
+			wg.Done()
+		}()
+	}
+	time.Sleep(100 * time.Millisecond) // let goroutines above block
+	c <- "bar"
+	wg.Wait()
+	if got := atomic.LoadInt32(&calls); got != 1 {
+		t.Errorf("number of calls = %d; want 1", got)
+	}
+	if got := atomic.LoadInt32(&freshes); got != 1 {
+		t.Errorf("freshes = %d; want 1", got)
+	}
+}

pkg/syncx/spinlock.go

@@ -0,0 +1,28 @@
+package syncx
+
+import (
+	"runtime"
+	"sync/atomic"
+)
+
+// A SpinLock is used as a lock a fast execution.
+type SpinLock struct {
+	lock uint32
+}
+
+// Lock locks the SpinLock.
+func (sl *SpinLock) Lock() {
+	for !sl.TryLock() {
+		runtime.Gosched()
+	}
+}
+
+// TryLock tries to lock the SpinLock.
+func (sl *SpinLock) TryLock() bool {
+	return atomic.CompareAndSwapUint32(&sl.lock, 0, 1)
+}
+
+// Unlock unlocks the SpinLock.
+func (sl *SpinLock) Unlock() {
+	atomic.StoreUint32(&sl.lock, 0)
+}

pkg/syncx/spinlock_test.go

@@ -0,0 +1,70 @@
+package syncx
+
+import (
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/perfect-panel/ppanel-server/pkg/lang"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTryLock(t *testing.T) {
+	var lock SpinLock
+	assert.True(t, lock.TryLock())
+	assert.False(t, lock.TryLock())
+	lock.Unlock()
+	assert.True(t, lock.TryLock())
+}
+
+func TestSpinLock(t *testing.T) {
+	var lock SpinLock
+	lock.Lock()
+	assert.False(t, lock.TryLock())
+	lock.Unlock()
+	assert.True(t, lock.TryLock())
+}
+
+func TestSpinLockRace(t *testing.T) {
+	var lock SpinLock
+	lock.Lock()
+	var wait sync.WaitGroup
+	wait.Add(1)
+	go func() {
+		wait.Done()
+	}()
+	time.Sleep(time.Millisecond * 100)
+	lock.Unlock()
+	wait.Wait()
+	assert.True(t, lock.TryLock())
+}
+
+func TestSpinLock_TryLock(t *testing.T) {
+	var lock SpinLock
+	var count int32
+	var wait sync.WaitGroup
+	wait.Add(2)
+	sig := make(chan lang.PlaceholderType)
+
+	go func() {
+		lock.TryLock()
+		sig <- lang.Placeholder
+		atomic.AddInt32(&count, 1)
+		runtime.Gosched()
+		lock.Unlock()
+		wait.Done()
+	}()
+
+	go func() {
+		<-sig
+		lock.Lock()
+		atomic.AddInt32(&count, 1)
+		lock.Unlock()
+		wait.Done()
+	}()
+
+	wait.Wait()
+	assert.Equal(t, int32(2), atomic.LoadInt32(&count))
+}

pkg/syncx/timeoutlimit.go

@@ -0,0 +1,57 @@
+package syncx
+
+import (
+	"errors"
+	"time"
+)
+
+// ErrTimeout is an error that indicates the borrow timeout.
+var ErrTimeout = errors.New("borrow timeout")
+
+// A TimeoutLimit is used to borrow with timeouts.
+type TimeoutLimit struct {
+	limit Limit
+	cond  *Cond
+}
+
+// NewTimeoutLimit returns a TimeoutLimit.
+func NewTimeoutLimit(n int) TimeoutLimit {
+	return TimeoutLimit{
+		limit: NewLimit(n),
+		cond:  NewCond(),
+	}
+}
+
+// Borrow borrows with given timeout.
+func (l TimeoutLimit) Borrow(timeout time.Duration) error {
+	if l.TryBorrow() {
+		return nil
+	}
+
+	var ok bool
+	for {
+		timeout, ok = l.cond.WaitWithTimeout(timeout)
+		if ok && l.TryBorrow() {
+			return nil
+		}
+
+		if timeout <= 0 {
+			return ErrTimeout
+		}
+	}
+}
+
+// Return returns a borrow.
+func (l TimeoutLimit) Return() error {
+	if err := l.limit.Return(); err != nil {
+		return err
+	}
+
+	l.cond.Signal()
+	return nil
+}
+
+// TryBorrow tries a borrow.
+func (l TimeoutLimit) TryBorrow() bool {
+	return l.limit.TryBorrow()
+}

pkg/syncx/timeoutlimit_test.go

@@ -0,0 +1,52 @@
+package syncx
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTimeoutLimit(t *testing.T) {
+	tests := []struct {
+		name     string
+		interval time.Duration
+	}{
+		{
+			name: "no wait",
+		},
+		{
+			name:     "wait",
+			interval: time.Millisecond * 100,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			limit := NewTimeoutLimit(2)
+			assert.Nil(t, limit.Borrow(time.Millisecond*200))
+			assert.Nil(t, limit.Borrow(time.Millisecond*200))
+			var wait1, wait2, wait3 sync.WaitGroup
+			wait1.Add(1)
+			wait2.Add(1)
+			wait3.Add(1)
+			go func() {
+				wait1.Wait()
+				wait2.Done()
+				time.Sleep(test.interval)
+				assert.Nil(t, limit.Return())
+				wait3.Done()
+			}()
+			wait1.Done()
+			wait2.Wait()
+			assert.Nil(t, limit.Borrow(time.Second))
+			wait3.Wait()
+			assert.Equal(t, ErrTimeout, limit.Borrow(time.Millisecond*100))
+			assert.Nil(t, limit.Return())
+			assert.Nil(t, limit.Return())
+			assert.Equal(t, ErrLimitReturn, limit.Return())
+		})
+	}
+}