contrib/recipes/double_barrier.go
func (b *DoubleBarrier) Leave() error {
	client := b.s.Client()
	resp, err := client.Get(b.ctx, b.key+"/waiters", clientv3.WithPrefix())
	if err != nil {
		return err
	}
	if len(resp.Kvs) == 0 {
		return nil
	}

	lowest, highest := resp.Kvs[0], resp.Kvs[0]
	for _, k := range resp.Kvs {
		if k.ModRevision < lowest.ModRevision {
			lowest = k
		}
		if k.ModRevision > highest.ModRevision {
			highest = k
		}
	}
	isLowest := string(lowest.Key) == b.myKey.Key()

	if len(resp.Kvs) == 1 {
		// this is the only node in the barrier; finish up
		if _, err = client.Delete(b.ctx, b.key+"/ready"); err != nil {
			return err
		}
		return b.myKey.Delete()
	}

	// this ensures that if a process fails, the ephemeral lease will be
	// revoked, its barrier key is removed, and the barrier can resume

	// lowest process in node => wait on highest process
	if isLowest {
		_, err = WaitEvents(
			client,
			string(highest.Key),
			highest.ModRevision,
			[]mvccpb.Event_EventType{mvccpb.DELETE})
		if err != nil {
			return err
		}
		return b.Leave()
	}

	// delete self and wait on lowest process
	if err = b.myKey.Delete(); err != nil {
		return err
	}

	key := string(lowest.Key)
	_, err = WaitEvents(
		client,
		key,
		lowest.ModRevision,
		[]mvccpb.Event_EventType{mvccpb.DELETE})
	if err != nil {
		return err
	}
	return b.Leave()
}
func (b *DoubleBarrier) Leave() error {
	client := b.s.Client()
	resp, err := client.Get(b.ctx, b.key+"/waiters", clientv3.WithPrefix())
	if err != nil {
		return err
	}
	if len(resp.Kvs) == 0 {
		return nil
	}

	lowest, highest := resp.Kvs[0], resp.Kvs[0]
	for _, k := range resp.Kvs {
		if k.ModRevision < lowest.ModRevision {
			lowest = k
		}
		if k.ModRevision > highest.ModRevision {
			highest = k
		}
	}
	isLowest := string(lowest.Key) == b.myKey.Key()

	if len(resp.Kvs) == 1 {
		// this is the only node in the barrier; finish up
		if _, err = client.Delete(b.ctx, b.key+"/ready"); err != nil {
			return err
		}
		return b.myKey.Delete()
	}

	// this ensures that if a process fails, the ephemeral lease will be
	// revoked, its barrier key is removed, and the barrier can resume

	// lowest process in node => wait on highest process
	if isLowest {
		_, err = WaitEvents(
			client,
			string(highest.Key),
			highest.ModRevision,
			[]mvccpb.Event_EventType{mvccpb.DELETE})
		if err != nil {
			return err
		}
		return b.Leave()
	}

	// delete self and wait on lowest process
	if err = b.myKey.Delete(); err != nil {
		return err
	}

	key := string(lowest.Key)
	_, err = WaitEvents(
		client,
		key,
		lowest.ModRevision,
		[]mvccpb.Event_EventType{mvccpb.DELETE})
	if err != nil {
		return err
	}
	return b.Leave()
}
func ( b * DoubleBarrier ) Leave ( ) error { client := b . s . Client ( ) resp , err := client . Get ( b . ctx , b . key + " " , clientv3 . WithPrefix ( ) ) if err != nil { return err } if len ( resp . Kvs ) == 0 { return nil } lowest , highest := resp . Kvs [ 0 ] , resp . Kvs [ 0 ] for _ , k := range resp . Kvs { if k . ModRevision < lowest . ModRevision { lowest = k } if k . ModRevision > highest . ModRevision { highest = k } } isLowest := string ( lowest . Key ) == b . myKey . Key ( ) if len ( resp . Kvs ) == 1 { // this is the only node in the barrier; finish up if _ , err = client . Delete ( b . ctx , b . key + " " ) ; err != nil { return err } return b . myKey . Delete ( ) } // this ensures that if a process fails, the ephemeral lease will be // revoked, its barrier key is removed, and the barrier can resume // lowest process in node => wait on highest process if isLowest { _ , err = WaitEvents ( client , string ( highest . Key ) , highest . ModRevision , [ ] mvccpb . Event_EventType { mvccpb . DELETE } ) if err != nil { return err } return b . Leave ( ) } // delete self and wait on lowest process if err = b . myKey . Delete ( ) ; err != nil { return err } key := string ( lowest . Key ) _ , err = WaitEvents ( client , key , lowest . ModRevision , [ ] mvccpb . Event_EventType { mvccpb . DELETE } ) if err != nil { return err } return b . Leave ( ) }
// Leave waits for "count" processes to leave the barrier then returns
Leave waits for count processes to leave the barrier then returns