clientv3/concurrency: ctx-ize Leader(), Resign(), and Unlock()
parent
2e93c65c96
commit
393725fe5f
|
@ -89,19 +89,19 @@ func (e *Election) Proclaim(ctx context.Context, val string) error {
|
|||
}
|
||||
|
||||
// Resign lets a leader start a new election.
|
||||
func (e *Election) Resign() (err error) {
|
||||
func (e *Election) Resign(ctx context.Context) (err error) {
|
||||
if e.leaderSession == nil {
|
||||
return nil
|
||||
}
|
||||
_, err = e.client.Delete(e.client.Ctx(), e.leaderKey)
|
||||
_, err = e.client.Delete(ctx, e.leaderKey)
|
||||
e.leaderKey = ""
|
||||
e.leaderSession = nil
|
||||
return err
|
||||
}
|
||||
|
||||
// Leader returns the leader value for the current election.
|
||||
func (e *Election) Leader() (string, error) {
|
||||
resp, err := e.client.Get(e.client.Ctx(), e.keyPrefix, v3.WithFirstCreate()...)
|
||||
func (e *Election) Leader(ctx context.Context) (string, error) {
|
||||
resp, err := e.client.Get(ctx, e.keyPrefix, v3.WithFirstCreate()...)
|
||||
if err != nil {
|
||||
return "", err
|
||||
} else if len(resp.Kvs) == 0 {
|
||||
|
|
|
@ -63,14 +63,14 @@ func (m *Mutex) Lock(ctx context.Context) error {
|
|||
// release lock key if cancelled
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
m.Unlock()
|
||||
m.Unlock(m.client.Ctx())
|
||||
default:
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *Mutex) Unlock() error {
|
||||
if _, err := m.client.Delete(m.client.Ctx(), m.myKey); err != nil {
|
||||
func (m *Mutex) Unlock(ctx context.Context) error {
|
||||
if _, err := m.client.Delete(ctx, m.myKey); err != nil {
|
||||
return err
|
||||
}
|
||||
m.myKey = "\x00"
|
||||
|
@ -92,7 +92,7 @@ func (lm *lockerMutex) Lock() {
|
|||
}
|
||||
}
|
||||
func (lm *lockerMutex) Unlock() {
|
||||
if err := lm.Mutex.Unlock(); err != nil {
|
||||
if err := lm.Mutex.Unlock(lm.client.Ctx()); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -128,5 +128,5 @@ func campaign(c *clientv3.Client, election string, prop string) error {
|
|||
return errors.New("elect: session expired")
|
||||
}
|
||||
|
||||
return e.Resign()
|
||||
return e.Resign(context.TODO())
|
||||
}
|
||||
|
|
|
@ -80,7 +80,7 @@ func lockUntilSignal(c *clientv3.Client, lockname string) error {
|
|||
|
||||
select {
|
||||
case <-donec:
|
||||
return m.Unlock()
|
||||
return m.Unlock(context.TODO())
|
||||
case <-s.Done():
|
||||
}
|
||||
|
||||
|
|
|
@ -72,7 +72,7 @@ func TestElectionWait(t *testing.T) {
|
|||
}
|
||||
}
|
||||
// let next leader take over
|
||||
if err := e.Resign(); err != nil {
|
||||
if err := e.Resign(context.TODO()); err != nil {
|
||||
t.Fatalf("failed resign (%v)", err)
|
||||
}
|
||||
// tell followers to start listening for next leader
|
||||
|
|
|
@ -69,7 +69,7 @@ func testMutex(t *testing.T, waiters int, chooseClient func() *clientv3.Client)
|
|||
t.Fatalf("lock %d followers did not wait", i)
|
||||
default:
|
||||
}
|
||||
if err := m.Unlock(); err != nil {
|
||||
if err := m.Unlock(context.TODO()); err != nil {
|
||||
t.Fatalf("could not release lock (%v)", err)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -95,7 +95,7 @@ func runElection(eps []string, rounds int) {
|
|||
}
|
||||
}
|
||||
rcs[i].validate = func() error {
|
||||
if l, err := e.Leader(); err == nil && l != observedLeader {
|
||||
if l, err := e.Leader(context.TODO()); err == nil && l != observedLeader {
|
||||
return fmt.Errorf("expected leader %q, got %q", observedLeader, l)
|
||||
}
|
||||
validatec <- struct{}{}
|
||||
|
@ -110,7 +110,7 @@ func runElection(eps []string, rounds int) {
|
|||
return fmt.Errorf("waiting on followers")
|
||||
}
|
||||
}
|
||||
if err := e.Resign(); err != nil {
|
||||
if err := e.Resign(context.TODO()); err != nil {
|
||||
return err
|
||||
}
|
||||
if observedLeader == v {
|
||||
|
@ -182,7 +182,7 @@ func runRacer(eps []string, round int) {
|
|||
return nil
|
||||
}
|
||||
rcs[i].release = func() error {
|
||||
if err := m.Unlock(); err != nil {
|
||||
if err := m.Unlock(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
cnt = 0
|
||||
|
|
Loading…
Reference in New Issue