etcd-io__etcd/contrib/recipes/mutex.go

// Copyright 2016 CoreOS, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package recipe

import (
	"sync"

	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
	v3 "github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/storage/storagepb"
)

// Mutex implements the sync Locker interface with etcd
type Mutex struct {
	client *v3.Client
	kv     v3.KV
	ctx    context.Context

	key   string
	myKey *EphemeralKV
}

func NewMutex(client *v3.Client, key string) *Mutex {
	return &Mutex{client, v3.NewKV(client), context.TODO(), key, nil}
}

func (m *Mutex) Lock() (err error) {
	// put self in lock waiters via myKey; oldest waiter holds lock
	m.myKey, err = NewUniqueEphemeralKey(m.client, m.key)
	if err != nil {
		return err
	}
	// find oldest element in waiters via revision of insertion
	resp, err := m.kv.Get(m.ctx, m.key, withFirstRev()...)
	if err != nil {
		return err
	}
	// if myKey is oldest in waiters, then myKey holds the lock
	if m.myKey.Revision() == resp.Kvs[0].CreateRevision {
		return nil
	}
	// otherwise myKey isn't lowest, so there must be a key prior to myKey
	opts := append(withLastRev(), v3.WithRev(m.myKey.Revision()-1))
	lastKey, err := m.kv.Get(m.ctx, m.key, opts...)
	if err != nil {
		return err
	}
	// wait for release on prior key
	_, err = WaitEvents(
		m.client,
		string(lastKey.Kvs[0].Key),
		m.myKey.Revision()-1,
		[]storagepb.Event_EventType{storagepb.DELETE})
	// myKey now oldest
	return err
}

func (m *Mutex) Unlock() error {
	err := m.myKey.Delete()
	m.myKey = nil
	return err
}

type lockerMutex struct{ *Mutex }

func (lm *lockerMutex) Lock() {
	if err := lm.Mutex.Lock(); err != nil {
		panic(err)
	}
}
func (lm *lockerMutex) Unlock() {
	if err := lm.Mutex.Unlock(); err != nil {
		panic(err)
	}
}

func NewLocker(client *v3.Client, key string) sync.Locker {
	return &lockerMutex{NewMutex(client, key)}
}