feat: reuse etcd-mutex instance

go.mod

@@ -66,6 +66,7 @@ require (
 	github.com/ClickHouse/ch-go v0.61.3 // indirect
 	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/bluele/gcache v0.0.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/coreos/go-semver v0.3.1 // indirect

go.sum

@@ -33,6 +33,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 h1:mXoPYz/Ul5HYEDvkta6I8/rnYM5gSdSV2tJ6XbZuEtY=
 github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932/go.mod h1:NOuUCSz6Q9T7+igc/hlvDOUdtWKryOrtFyIVABv/p7k=
+github.com/bluele/gcache v0.0.2 h1:WcbfdXICg7G/DGBh1PFfcirkWOQV+v077yF1pSy3DGw=
+github.com/bluele/gcache v0.0.2/go.mod h1:m15KV+ECjptwSPxKhOhQoAFQVtUFjTVkc3H8o0t/fp0=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
 github.com/brahma-adshonor/gohook v1.1.9 h1:YuQFj8rhAj1kvtGHUc5BlLvAELw98M/ydpBz/MvBXNU=

providers/etcd-mutex/examples/examples.yaml

@@ -1,5 +1,5 @@
 etcd:
-    endpoints: "https://127.0.0.1:2379"
+    endpoints: "http://127.0.0.1:2379"
     # tls:
     #     cert_file: "etcd-client.pem"
     #     cert_key_file: "etcd-client-key.pem"

providers/etcd-mutex/examples/main.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"sync"
 	"time"
 
@@ -31,8 +32,9 @@ type provider struct {
 }
 
 func (p *provider) Run(ctx context.Context) error {
+	fmt.Println("running ...")
 	go func() {
-		time.Sleep(10 * time.Second)
+		time.Sleep(10000 * time.Second)
 		err := p.Lock.Close()
 		if err != nil {
 			fmt.Println("Close err: ", err)
@@ -44,47 +46,46 @@ func (p *provider) Run(ctx context.Context) error {
 			return
 		}
 	}()
-	lock := p.Lock
-	sleep := func(d time.Duration) bool {
-		select {
-		case <-time.After(d):
-		case <-ctx.Done():
-			return false
-		}
-		return true
+	//ctx, cancelFunc := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	mu, err := p.Mutex.New(ctx, "keyAAAA")
+	if err != nil {
+		return err
 	}
-	doTaskInLock := func(prefix string) bool {
-		err := lock.Lock(ctx)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := mu.Lock(ctx)
 		if err != nil {
-			fmt.Println("Lock err: ", err)
-			return false
-		}
-		defer func() {
-			err := lock.Unlock(context.TODO())
-			if err != nil {
-				fmt.Println("Unlock err: ", err)
-			}
-		}()
-		fmt.Println(prefix+" {", time.Now())
-		if !sleep(1 * time.Second) {
-			return false
+			return
 		}
-		fmt.Println(prefix+"    ", time.Now())
-		if !sleep(1 * time.Second) {
-			return false
+		defer mu.Unlock(ctx)
+		for i := 0; i < 10; i++ {
+			fmt.Printf("A wait %vs\n", i+1)
+			time.Sleep(1 * time.Second)
 		}
-		fmt.Println(prefix+" }", time.Now())
-		return true
-	}
-	var wg sync.WaitGroup
-	for _, elem := range []string{"A", "B", "C"} {
-		wg.Add(1)
-		go func(prefix string) {
-			defer wg.Done()
-			for doTaskInLock(prefix) {
-			}
-		}(elem)
+		fmt.Println("AAA===>")
+	}()
+	time.Sleep(1 * time.Second)
+	ctx, cancelFunc = context.WithCancel(context.Background())
+	defer cancelFunc()
+	mu, err = p.Mutex.New(ctx, "keyAAAA")
+	if err != nil {
+		return err
 	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := mu.Lock(ctx)
+		if err != nil {
+			return
+		}
+		defer mu.Unlock(ctx)
+		time.Sleep(5 * time.Second)
+		fmt.Println("BBB===>")
+	}()
 	wg.Wait()
 	return nil
 }
@@ -101,8 +102,9 @@ func init() {
 }
 
 func main() {
+	dir, _ := os.Getwd()
 	hub := servicehub.New()
-	hub.Run("examples", "", os.Args...)
+	hub.Run("examples", filepath.Join(dir, "providers", "etcd-mutex", "examples", "examples.yaml"), os.Args...)
 }
 
 // OUTPUT:

providers/etcd-mutex/mutex.go

@@ -17,6 +17,7 @@
 import (
 	"context"
 	"errors"
+	"github.com/bluele/gcache"
 	"path/filepath"
 	"reflect"
 	"sync"
@@ -57,14 +58,16 @@
 	Cfg         *config
 	Log         logs.Logger
 	etcd        etcd.Interface
-	instances   map[string]Mutex
+	instances   gcache.Cache
 	inProcMutex *inProcMutex
+	lock        *sync.Mutex
 }
 
 // Init .
 func (p *provider) Init(ctx servicehub.Context) error {
 	p.etcd = ctx.Service("etcd").(etcd.Interface)
 	p.Cfg.RootPath = filepath.Clean("/" + p.Cfg.RootPath)
+	p.instances = gcache.New(30).LRU().Build()
 	return nil
 }
 
@@ -76,20 +79,32 @@
 	if seconds > 0 {
 		opts = append(opts, concurrency.WithTTL(seconds))
 	}
-	return &etcdMutex{
+	mutex := &etcdMutex{
 		log:        p.Log,
 		key:        filepath.Clean(filepath.Join(p.Cfg.RootPath, key)),
 		client:     p.etcd.Client(),
 		opts:       opts,
 		inProcLock: make(chan struct{}, 1),
 		ctx:        ctx,
 		cancel:     cancel,
-	}, nil
+	}
+	return mutex, p.instances.SetWithExpire(key, mutex, time.Second*3)
 }
 
 // New .
 func (p *provider) New(ctx context.Context, key string) (Mutex, error) {
-	return p.NewWithTTL(ctx, key, time.Duration(0))
+	ins, err := p.instances.Get(key)
+	if err != nil {
+		if errors.Is(err, gcache.KeyNotFoundError) {
+			return p.NewWithTTL(ctx, key, time.Duration(0))
+		}
+		return nil, err
+	}
+	if mutex, ok := ins.(Mutex); ok {
+		return mutex, nil
+	} else {
+		return nil, errors.New("can't convert to mutex")
+	}
 }
 
 // Provide .
@@ -291,8 +306,8 @@
 		ConfigFunc:   func() interface{} { return &config{} },
 		Creator: func() servicehub.Provider {
 			return &provider{
-				instances:   make(map[string]Mutex),
 				inProcMutex: &inProcMutex{lock: make(chan struct{}, 1)},
+				lock:        &sync.Mutex{},
 			}
 		},
 	})