Merge pull request #133 from rfyiamcool/feat/with_fail_fast_mode

feat: add with fail fast mode

mutex.go

@@ -30,6 +30,7 @@ type Mutex struct {
 	value        string
 	until        time.Time
 	shuffle      bool
+	failFast     bool
 
 	pools []redis.Pool
 }
@@ -263,17 +264,21 @@ func (m *Mutex) actOnPoolsAsync(actFn func(redis.Pool) (bool, error)) (int, erro
 		err      error
 	}
 
-	ch := make(chan result)
+	ch := make(chan result, len(m.pools))
 	for node, pool := range m.pools {
 		go func(node int, pool redis.Pool) {
 			r := result{node: node}
 			r.statusOK, r.err = actFn(pool)
 			ch <- r
 		}(node, pool)
 	}
-	n := 0
-	var taken []int
-	var err error
+
+	var (
+		n     = 0
+		taken []int
+		err   error
+	)
+
 	for range m.pools {
 		r := <-ch
 		if r.statusOK {
@@ -284,6 +289,18 @@ func (m *Mutex) actOnPoolsAsync(actFn func(redis.Pool) (bool, error)) (int, erro
 			taken = append(taken, r.node)
 			err = multierror.Append(err, &ErrNodeTaken{Node: r.node})
 		}
+
+		if m.failFast {
+			// fast retrun
+			if n >= m.quorum {
+				return n, err
+			}
+
+			// fail fast
+			if len(taken) >= m.quorum {
+				return n, &ErrTaken{Nodes: taken}
+			}
+		}
 	}
 
 	if len(taken) >= m.quorum {

redsync.go

@@ -125,6 +125,15 @@ func WithValue(v string) Option {
 	})
 }
 
+// WithFailFast can be used to quickly acquire and release the locker.
+// When some redis servers are blocking, we do not need to wait for all redis servers response. as long as the quorum is met, it will be returned immediately.
+// The effect of this parameter is to achieve low latency, avoid redis blocking causing lock/unlock to not return for a long time.
+func WithFailFast(b bool) Option {
+	return OptionFunc(func(m *Mutex) {
+		m.failFast = b
+	})
+}
+
 // WithShufflePools can be used to shuffle Redis pools to reduce centralized access in concurrent scenarios.
 func WithShufflePools(b bool) Option {
 	return OptionFunc(func(m *Mutex) {