Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

simplify lock

Browse Source
This commit is contained in:
Xiang Li 2013-09-30 23:18:52 -07:00
parent d2407dff9f
commit 558d30f33f
3 changed files with 14 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
store/node.go
View File

@ -3,7 +3,6 @@ package store
import ( import (
"path" "path"
"sort" "sort"
"sync"
"time" "time"
etcdErr "github.com/coreos/etcd/error" etcdErr "github.com/coreos/etcd/error"
@ -30,7 +29,6 @@ type Node struct {
Value string // for key-value pair Value string // for key-value pair
Children map[string]*Node // for directory Children map[string]*Node // for directory
status int status int
mu sync.Mutex
stopExpire chan bool // stop expire routine channel stopExpire chan bool // stop expire routine channel
} }
@ -66,10 +64,7 @@ func newDir(nodePath string, createIndex uint64, createTerm uint64, parent *Node
// If the node is a directory and recursive is true, the function will recursively remove // If the node is a directory and recursive is true, the function will recursively remove
// add nodes under the receiver node. // add nodes under the receiver node.
func (n *Node) Remove(recursive bool, callback func(path string)) error { func (n *Node) Remove(recursive bool, callback func(path string)) error {
n.mu.Lock() if n.status == removed { // check race between remove and expire
defer n.mu.Unlock()
if n.status == removed {
return nil return nil
} }
@ -144,8 +139,6 @@ func (n *Node) Write(value string, index uint64, term uint64) error {
// List function return a slice of nodes under the receiver node. // List function return a slice of nodes under the receiver node.
// If the receiver node is not a directory, a "Not A Directory" error will be returned. // If the receiver node is not a directory, a "Not A Directory" error will be returned.
func (n *Node) List() ([]*Node, error) { func (n *Node) List() ([]*Node, error) {
n.mu.Lock()
defer n.mu.Unlock()
if !n.IsDir() { if !n.IsDir() {
return nil, etcdErr.NewError(etcdErr.EcodeNotDir, "") return nil, etcdErr.NewError(etcdErr.EcodeNotDir, "")
} }
@ -168,9 +161,6 @@ func (n *Node) List() ([]*Node, error) {
// If the node corresponding to the name string is not file, it returns // If the node corresponding to the name string is not file, it returns
// Not File Error // Not File Error
func (n *Node) GetFile(name string) (*Node, error) { func (n *Node) GetFile(name string) (*Node, error) {
n.mu.Lock()
defer n.mu.Unlock()
if !n.IsDir() { if !n.IsDir() {
return nil, etcdErr.NewError(etcdErr.EcodeNotDir, n.Path) return nil, etcdErr.NewError(etcdErr.EcodeNotDir, n.Path)
} }
@ -193,12 +183,6 @@ func (n *Node) GetFile(name string) (*Node, error) {
// If there is a existing node with the same name under the directory, a "Already Exist" // If there is a existing node with the same name under the directory, a "Already Exist"
// error will be returned // error will be returned
func (n *Node) Add(child *Node) error { func (n *Node) Add(child *Node) error {
n.mu.Lock()
defer n.mu.Unlock()
if n.status == removed {
return etcdErr.NewError(etcdErr.EcodeKeyNotFound, "")
}
if !n.IsDir() { if !n.IsDir() {
return etcdErr.NewError(etcdErr.EcodeNotDir, "") return etcdErr.NewError(etcdErr.EcodeNotDir, "")
} }
@ -220,8 +204,6 @@ func (n *Node) Add(child *Node) error {
// If the node is a directory, it will clone all the content under this directory. // If the node is a directory, it will clone all the content under this directory.
// If the node is a key-value pair, it will clone the pair. // If the node is a key-value pair, it will clone the pair.
func (n *Node) Clone() *Node { func (n *Node) Clone() *Node {
n.mu.Lock()
defer n.mu.Unlock()
if !n.IsDir() { if !n.IsDir() {
return newFile(n.Path, n.Value, n.CreateIndex, n.CreateTerm, n.Parent, n.ACL, n.ExpireTime) return newFile(n.Path, n.Value, n.CreateIndex, n.CreateTerm, n.Parent, n.ACL, n.ExpireTime)
} }
@ -256,7 +238,6 @@ func (n *Node) Expire(s *Store) {
expired, duration := n.IsExpired() expired, duration := n.IsExpired()
if expired { // has been expired if expired { // has been expired
// since the parent function of Expire() runs serially, // since the parent function of Expire() runs serially,
// there is no need for lock here // there is no need for lock here
e := newEvent(Expire, n.Path, UndefIndex, UndefTerm) e := newEvent(Expire, n.Path, UndefIndex, UndefTerm)
@ -277,9 +258,8 @@ func (n *Node) Expire(s *Store) {
// if timeout, delete the node // if timeout, delete the node
case <-time.After(duration): case <-time.After(duration):
// Lock the worldLock to avoid race on s.WatchHub,
// and the race with other slibling nodes on their common parent.
s.worldLock.Lock() s.worldLock.Lock()
defer s.worldLock.Unlock()
e := newEvent(Expire, n.Path, UndefIndex, UndefTerm) e := newEvent(Expire, n.Path, UndefIndex, UndefTerm)
s.WatcherHub.notify(e) s.WatcherHub.notify(e)
@ -287,8 +267,6 @@ func (n *Node) Expire(s *Store) {
n.Remove(true, nil) n.Remove(true, nil)
s.Stats.Inc(ExpireCount) s.Stats.Inc(ExpireCount)
s.worldLock.Unlock()
return return
// if stopped, return // if stopped, return

18
store/store.go
View File

@ -18,7 +18,7 @@ type Store struct {
Index uint64 Index uint64
Term uint64 Term uint64
Stats *Stats Stats *Stats
worldLock sync.RWMutex // stop the world lock. Used to do snapshot worldLock sync.RWMutex // stop the world lock
} }
func New() *Store { func New() *Store {
@ -95,8 +95,8 @@ func (s *Store) Get(nodePath string, recursive, sorted bool, index uint64, term
// If the node has already existed, create will fail. // If the node has already existed, create will fail.
// If any node on the path is a file, create will fail. // If any node on the path is a file, create will fail.
func (s *Store) Create(nodePath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) { func (s *Store) Create(nodePath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
s.worldLock.RLock() s.worldLock.Lock()
defer s.worldLock.RUnlock() defer s.worldLock.Unlock()
nodePath = path.Clean(path.Join("/", nodePath)) nodePath = path.Clean(path.Join("/", nodePath))
@ -164,8 +164,8 @@ func (s *Store) Create(nodePath string, value string, expireTime time.Time, inde
// If the node is a file, the value and the ttl can be updated. // If the node is a file, the value and the ttl can be updated.
// If the node is a directory, only the ttl can be updated. // If the node is a directory, only the ttl can be updated.
func (s *Store) Update(nodePath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) { func (s *Store) Update(nodePath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
s.worldLock.RLock() s.worldLock.Lock()
defer s.worldLock.RUnlock() defer s.worldLock.Unlock()
n, err := s.internalGet(nodePath, index, term) n, err := s.internalGet(nodePath, index, term)
@ -209,8 +209,8 @@ func (s *Store) Update(nodePath string, value string, expireTime time.Time, inde
func (s *Store) TestAndSet(nodePath string, prevValue string, prevIndex uint64, func (s *Store) TestAndSet(nodePath string, prevValue string, prevIndex uint64,
value string, expireTime time.Time, index uint64, term uint64) (*Event, error) { value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
s.worldLock.RLock() s.worldLock.Lock()
defer s.worldLock.RUnlock() defer s.worldLock.Unlock()
n, err := s.internalGet(nodePath, index, term) n, err := s.internalGet(nodePath, index, term)
@ -246,8 +246,8 @@ func (s *Store) TestAndSet(nodePath string, prevValue string, prevIndex uint64,
// Delete function deletes the node at the given path. // Delete function deletes the node at the given path.
// If the node is a directory, recursive must be true to delete it. // If the node is a directory, recursive must be true to delete it.
func (s *Store) Delete(nodePath string, recursive bool, index uint64, term uint64) (*Event, error) { func (s *Store) Delete(nodePath string, recursive bool, index uint64, term uint64) (*Event, error) {
s.worldLock.RLock() s.worldLock.Lock()
defer s.worldLock.RUnlock() defer s.worldLock.Unlock()
n, err := s.internalGet(nodePath, index, term) n, err := s.internalGet(nodePath, index, term)

6
store/store_test.go
View File

@ -244,7 +244,7 @@ func TestExpire(t *testing.T) {
s.Create("/foo", "bar", expire, 1, 1) s.Create("/foo", "bar", expire, 1, 1)
_, err := s.internalGet("/foo", 1, 1) _, err := s.Get("/foo", false, false, 1, 1)
if err != nil { if err != nil {
t.Fatalf("can not get the node") t.Fatalf("can not get the node")
@ -252,7 +252,7 @@ func TestExpire(t *testing.T) {
time.Sleep(time.Second * 2) time.Sleep(time.Second * 2)
_, err = s.internalGet("/foo", 1, 1) _, err = s.Get("/foo", false, false, 1, 1)
if err == nil { if err == nil {
t.Fatalf("can get the node after expiration time") t.Fatalf("can get the node after expiration time")
@ -263,7 +263,7 @@ func TestExpire(t *testing.T) {
s.Create("/foo", "bar", expire, 1, 1) s.Create("/foo", "bar", expire, 1, 1)
time.Sleep(time.Millisecond * 50) time.Sleep(time.Millisecond * 50)
_, err = s.internalGet("/foo", 1, 1) _, err = s.Get("/foo", false, false, 1, 1)
if err != nil { if err != nil {
t.Fatalf("cannot get the node before expiration", err.Error()) t.Fatalf("cannot get the node before expiration", err.Error())