Merge lp:~rogpeppe/gozk/safe-close into lp:gozk/zookeeper

[Roger Peppe]

zookeeper: make Conn.Close safe to call concurrently with other operations.

I am concerned at how complex the test for this is,
relative to the simplicity of the actual code change.
(and it's also potentially fragile on a heavily loaded machine).
That said, I can't think of another way to test the
operations directly, so I'm leaving it in.
Better suggestions welcome.

https://codereview.appspot.com/5699093/

[Roger Peppe (rogpeppe)]

Reviewers: mp+94812_code.launchpad.net,

Message:
Please take a look.

Description:
I am concerned at how complex the test for this is,
relative to the simplicity of the actual code change.
(and it's also potentially fragile on a heavily loaded machine).
That said, I can't think of another way to test the
operations directly, so I'm leaving it in.
Better suggestions welcome.

https://code.launchpad.net/~rogpeppe/gozk/safe-close/+merge/94812

(do not edit description out of merge proposal)

Please review this at https://codereview.appspot.com/5699093/

Affected files:
   A close_test.go
   M retry_test.go
   M suite_test.go
   M zk.go
   M zk_test.go

[Gustavo Niemeyer (niemeyer)]

https://codereview.appspot.com/5699093/diff/1001/close_test.go
File close_test.go (right):

https://codereview.appspot.com/5699093/diff/1001/close_test.go#newcode96
close_test.go:96: time.Sleep(0.05e9)
How can you tell this is really unblocking at the right time and
exercising what you intend it to at all?

https://codereview.appspot.com/5699093/

[Roger Peppe (rogpeppe)]

Please take a look.

https://codereview.appspot.com/5699093/

[lp:~rogpeppe/gozk/safe-close]

33. By Roger Peppe on 2012-02-28

improve close_test comment.
simplify io.Copy logic.

[William Reade (fwereade)]

This essentially LGTM; I don't think that it's a serious problem that
the tests can theoretically fail under "enough" load. So it comes down
to a question of practical reliability: for the sake of argument, what
if we shoot for an observed failure rate of <1% on your local machine
under "normal" load, and bump up the sleeps as needed if they turn out
not to be good enough for clean testing in practice (say, on ARM ;))?

https://codereview.appspot.com/5699093/

[Gustavo Niemeyer (niemeyer)]

LGTM, sorry for the delay.

https://codereview.appspot.com/5699093/

[Roger Peppe (rogpeppe)]

i forgot to publish this comment.

https://codereview.appspot.com/5699093/diff/1001/close_test.go
File close_test.go (right):

https://codereview.appspot.com/5699093/diff/1001/close_test.go#newcode96
close_test.go:96: time.Sleep(0.05e9)
On 2012/02/27 18:03:17, niemeyer wrote:
> How can you tell this is really unblocking at the right time and
exercising what
> you intend it to at all?

the idea is that any request that requests or changes a zookeeper node
must make at least one round trip to the server. so we interpose a proxy
between the client and the server which can stop all incoming traffic on
demand, thus hopefully blocking the request until we want it to unblock.

we assume that all requests take less than 0.1s to complete, thus when
we wait below, neither of the above goroutines should complete within
the allotted time (the request because it's waiting for a reply from the
server and the close because it's waiting for the request to complete).
if the locking doesn't work, the Close will return early. if the proxy
blocking doesn't work, the request will return early.

when we reenable incoming messages from the server, both goroutines
should complete (we can't tell which completes first - i don't think
that's observable) but i think that the fact that the close blocked
waiting for the request is sufficient.

i've changed the above comment to try to make this clearer.

as i said in the description of the CL, i think this a lot of mechanism
to test something that's quite simple to verify by eye, but i haven't
yet thought of a better way.
one way could be to test Close without testing all the individual
operations - simpler but less complete.

https://codereview.appspot.com/5699093/

[Roger Peppe (rogpeppe)]

*** Submitted:

zookeeper: make Conn.Close safe to call concurrently with other
operations.

I am concerned at how complex the test for this is,
relative to the simplicity of the actual code change.
(and it's also potentially fragile on a heavily loaded machine).
That said, I can't think of another way to test the
operations directly, so I'm leaving it in.
Better suggestions welcome.

R=niemeyer, fwereade
CC=
https://codereview.appspot.com/5699093

https://codereview.appspot.com/5699093/

=== added file 'close_test.go'
--- close_test.go	1970-01-01 00:00:00 +0000
+++ close_test.go	2012-02-28 09:18:17 +0000
@@ -0,0 +1,220 @@
+package zookeeper_test
+
+import (
+	"io"
+	. "launchpad.net/gocheck"
+	zk "launchpad.net/gozk/zookeeper"
+	"log"
+	"net"
+	"time"
+)
+
+// requestFuncs holds all the requests that take a read lock
+// on the zk connection except those that don't actually
+// make a round trip to the server.
+var requestFuncs = []func(conn *zk.Conn, path string) error{
+	func(conn *zk.Conn, path string) error {
+		_, err := conn.Create(path, "", 0, nil)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, err := conn.Exists(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, _, err := conn.ExistsW(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, _, err := conn.Get(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, _, _, err := conn.GetW(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, _, err := conn.Children(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, _, _, err := conn.ChildrenW(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, err := conn.Set(path, "", 0)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		_, _, err := conn.ACL(path)
+		return err
+	},
+	func(conn *zk.Conn, path string) error {
+		return conn.SetACL(path, []zk.ACL{{
+			Perms:  zk.PERM_ALL,
+			Scheme: "digest",
+			Id:     "foo",
+		}}, 0)
+	},
+	func(conn *zk.Conn, path string) error {
+		return conn.Delete(path, 0)
+	},
+}
+
+func (s *S) TestConcurrentClose(c *C) {
+	// make sure the server is ready to receive connections.
+	s.init(c)
+
+	// Close should wait until all outstanding requests have
+	// completed before returning.  The idea of this test is that
+	// any request that requests or changes a zookeeper node must
+	// make at least one round trip to the server, so we interpose a
+	// proxy between the client and the server which can stop all
+	// incoming traffic on demand, thus blocking the request until
+	// we want it to unblock.
+	//
+	// We assume that all requests take less than 0.1s to complete,
+	// thus when we wait below, neither of the above goroutines
+	// should complete within the allotted time (the request because
+	// it's waiting for a reply from the server and the close
+	// because it's waiting for the request to complete).  If the
+	// locking doesn't work, the Close will return early.  If the
+	// proxy blocking doesn't work, the request will return early.
+	//
+	// When we reenable incoming messages from the server, both
+	// goroutines should complete.  We can't tell which completes
+	// first, but the fact that the close blocked is sufficient to
+	// tell that the locking is working correctly.
+	for i, f := range requestFuncs {
+		c.Logf("iter %d", i)
+		p := newProxy(c, s.zkAddr)
+		conn, watch, err := zk.Dial(p.addr(), 5e9)
+		c.Assert(err, IsNil)
+		c.Assert((<-watch).Ok(), Equals, true)
+
+		// sanity check that the connection is actually
+		// up and running.
+		_, err = conn.Exists("/nothing")
+		c.Assert(err, IsNil)
+
+		p.stopIncoming()
+		reqDone := make(chan bool)
+		closeDone := make(chan bool)
+		go func() {
+			f(conn, "/closetest")
+			reqDone <- true
+		}()
+		go func() {
+			// sleep for long enough for the request to be initiated and the read lock taken.
+			time.Sleep(0.05e9)
+			conn.Close()
+			closeDone <- true
+		}()
+		select {
+		case <-reqDone:
+			c.Fatalf("request %d finished early", i)
+		case <-closeDone:
+			c.Fatalf("request %d close finished early", i)
+		case <-time.After(0.1e9):
+		}
+		p.startIncoming()
+		for reqDone != nil || closeDone != nil {
+			select {
+			case <-reqDone:
+				reqDone = nil
+			case <-closeDone:
+				closeDone = nil
+			case <-time.After(0.4e9):
+				c.Fatalf("request %d timed out waiting for req (%p) and close(%p)", i, reqDone, closeDone)
+			}
+		}
+		p.close()
+		err = f(conn, "/closetest")
+		c.Check(err, Equals, zk.ZCLOSING)
+	}
+}
+
+type proxy struct {
+	stop, start chan bool
+	listener    net.Listener
+}
+
+// newProxy will listen on proxyAddr and connect its client to dstAddr, and return
+// a proxy instance that can be used to control the connection.
+func newProxy(c *C, dstAddr string) *proxy {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	c.Assert(err, IsNil)
+	p := &proxy{
+		stop:     make(chan bool, 1),
+		start:    make(chan bool, 1),
+		listener: listener,
+	}
+
+	go func() {
+		for {
+			client, err := p.listener.Accept()
+			if err != nil {
+				// Ignore the error, because the connection will fail anyway.
+				return
+			}
+			go func() {
+				defer client.Close()
+				server, err := net.Dial("tcp", dstAddr)
+				if err != nil {
+					log.Printf("cannot dial %q: %v", dstAddr, err)
+					return
+				}
+				defer server.Close()
+				go io.Copy(&haltableWriter{
+					w:     client,
+					stop:  p.stop,
+					start: p.start},
+					server)
+				// When the client is closed, the deferred closes will
+				// take down the other io.Copy too.
+				io.Copy(server, client)
+			}()
+		}
+	}()
+	return p
+}
+
+func (p *proxy) close() error {
+	return p.listener.Close()
+}
+
+func (p *proxy) addr() string {
+	return p.listener.Addr().String()
+}
+
+func (p *proxy) stopIncoming() {
+	if p.stop == nil {
+		panic("cannot stop twice")
+	}
+	p.stop <- true
+	p.stop = nil
+}
+
+func (p *proxy) startIncoming() {
+	if p.start == nil {
+		panic("cannot start twice")
+	}
+	p.start <- true
+	p.start = nil
+}
+
+type haltableWriter struct {
+	w           io.Writer
+	stop, start chan bool
+}
+
+func (w *haltableWriter) Write(buf []byte) (int, error) {
+	select {
+	case <-w.stop:
+		w.stop <- true
+		<-w.start
+		w.start <- true
+	default:
+	}
+	return w.w.Write(buf)
+}
 
=== modified file 'retry_test.go'
--- retry_test.go	2012-02-15 17:18:34 +0000
+++ retry_test.go	2012-02-28 09:18:17 +0000
@@ -25,7 +25,7 @@
 
 	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(acl, DeepEquals, zk.WorldACL(zk.PERM_ALL))
 }
 
 func (s *S) TestRetryChangeSetting(c *C) {
@@ -52,7 +52,7 @@
 	// ACL was unchanged by RetryChange().
 	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(acl, DeepEquals, zk.WorldACL(zk.PERM_ALL))
 }
 
 func (s *S) TestRetryChangeUnchangedValueDoesNothing(c *C) {
@@ -177,7 +177,7 @@
 	// Should be the new ACL.
 	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_READ))
+	c.Assert(acl, DeepEquals, zk.WorldACL(zk.PERM_READ))
 }
 
 func (s *S) TestRetryChangeErrorInCallback(c *C) {
 
=== modified file 'suite_test.go'
--- suite_test.go	2012-02-10 13:33:24 +0000
+++ suite_test.go	2012-02-28 09:18:17 +0000
@@ -31,6 +31,7 @@
 var logLevel = 0 //zk.LOG_ERROR
 
 func (s *S) init(c *C) (*zk.Conn, chan zk.Event) {
+	c.Logf("init dialling %q", s.zkAddr)
 	conn, watch, err := zk.Dial(s.zkAddr, 5e9)
 	c.Assert(err, IsNil)
 	s.handles = append(s.handles, conn)
@@ -71,7 +72,7 @@
 
 func (s *S) SetUpTest(c *C) {
 	c.Assert(zk.CountPendingWatches(), Equals, 0,
-		Bug("Test got a dirty watch state before running!"))
+		Commentf("Test got a dirty watch state before running!"))
 	zk.SetLogLevel(logLevel)
 }
 
@@ -95,7 +96,7 @@
 	s.handles = make([]*zk.Conn, 0)
 
 	c.Assert(zk.CountPendingWatches(), Equals, 0,
-		Bug("Test left live watches behind!"))
+		Commentf("Test left live watches behind!"))
 }
 
 // We use the suite set up and tear down to manage a custom ZooKeeper
@@ -104,7 +105,7 @@
 	var err error
 	s.deadWatches = make(chan bool)
 
-	// N.B. We meed to create a subdirectory because zk.CreateServer
+	// N.B. We need to create a subdirectory because zk.CreateServer
 	// insists on creating its own directory.
 
 	s.zkTestRoot = c.MkDir() + "/zk"
 
=== modified file 'zk.go'
--- zk.go	2012-02-15 17:51:23 +0000
+++ zk.go	2012-02-28 09:18:17 +0000
@@ -32,7 +32,7 @@
 	watchChannels  map[uintptr]chan Event
 	sessionWatchId uintptr
 	handle         *C.zhandle_t
-	mutex          sync.Mutex
+	mutex          sync.RWMutex
 }
 
 // ClientId represents an established ZooKeeper session.  It can be
@@ -429,6 +429,8 @@
 // ClientId returns the client ID for the existing session with ZooKeeper.
 // This is useful to reestablish an existing session via ReInit.
 func (conn *Conn) ClientId() *ClientId {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
 	return &ClientId{*C.zoo_client_id(conn.handle)}
 }
 
@@ -459,6 +461,11 @@
 // unless an error is found. Attempting to retrieve data from a non-existing
 // node is an error.
 func (conn *Conn) Get(path string) (data string, stat *Stat, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return "", nil, ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	cbuffer := (*C.char)(C.malloc(bufferSize))
@@ -481,6 +488,11 @@
 // node changes or when critical session events happen.  See the
 // documentation of the Event type for more details.
 func (conn *Conn) GetW(path string) (data string, stat *Stat, watch <-chan Event, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return "", nil, nil, ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	cbuffer := (*C.char)(C.malloc(bufferSize))
@@ -504,6 +516,11 @@
 // Children returns the children list and status from an existing node.
 // Attempting to retrieve the children list from a non-existent node is an error.
 func (conn *Conn) Children(path string) (children []string, stat *Stat, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return nil, nil, ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
@@ -529,6 +546,11 @@
 // provided path or when critical session events happen.  See the documentation
 // of the Event type for more details.
 func (conn *Conn) ChildrenW(path string) (children []string, stat *Stat, watch <-chan Event, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return nil, nil, nil, ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
@@ -570,6 +592,12 @@
 // stat will contain meta information on the existing node, otherwise
 // it will be nil.
 func (conn *Conn) Exists(path string) (stat *Stat, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return nil, ZCLOSING
+	}
+
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
@@ -593,6 +621,12 @@
 // is removed. It will also receive critical session events. See the
 // documentation of the Event type for more details.
 func (conn *Conn) ExistsW(path string) (stat *Stat, watch <-chan Event, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return nil, nil, ZCLOSING
+	}
+
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
@@ -627,6 +661,12 @@
 // from the requested one, such as when a sequence number is appended
 // to it due to the use of the gozk.SEQUENCE flag.
 func (conn *Conn) Create(path, value string, flags int, aclv []ACL) (pathCreated string, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return "", ZCLOSING
+	}
+
 	cpath := C.CString(path)
 	cvalue := C.CString(value)
 	defer C.free(unsafe.Pointer(cpath))
@@ -658,6 +698,11 @@
 // It is an error to attempt to set the data of a non-existing node with
 // this function. In these cases, use Create instead.
 func (conn *Conn) Set(path, value string, version int) (stat *Stat, err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return nil, ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	cvalue := C.CString(value)
@@ -678,6 +723,12 @@
 // will only succeed if the node is still at this version when the
 // node is deleted as an atomic operation.
 func (conn *Conn) Delete(path string, version int) (err error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return ZCLOSING
+	}
+
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 	rc, cerr := C.zoo_delete(conn.handle, cpath, C.int(version))
@@ -690,6 +741,12 @@
 // identity data itself. For instance, the "digest" scheme requires
 // a pair like "username:password" to be provided as the certificate.
 func (conn *Conn) AddAuth(scheme, cert string) error {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return ZCLOSING
+	}
+
 	cscheme := C.CString(scheme)
 	ccert := C.CString(cert)
 	defer C.free(unsafe.Pointer(cscheme))
@@ -714,6 +771,11 @@
 
 // ACL returns the access control list for path.
 func (conn *Conn) ACL(path string) ([]ACL, *Stat, error) {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return nil, nil, ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
@@ -733,6 +795,11 @@
 
 // SetACL changes the access control list for path.
 func (conn *Conn) SetACL(path string, aclv []ACL, version int) error {
+	conn.mutex.RLock()
+	defer conn.mutex.RUnlock()
+	if conn.handle == nil {
+		return ZCLOSING
+	}
 
 	cpath := C.CString(path)
 	defer C.free(unsafe.Pointer(cpath))
 
=== modified file 'zk_test.go'
--- zk_test.go	2012-02-15 17:51:23 +0000
+++ zk_test.go	2012-02-28 09:18:17 +0000
@@ -312,7 +312,7 @@
 
 	children, stat, err := conn.Children("/")
 	c.Assert(err, IsNil)
-	c.Assert(children, Equals, []string{"zookeeper"})
+	c.Assert(children, DeepEquals, []string{"zookeeper"})
 	c.Assert(stat.NumChildren(), Equals, 1)
 
 	children, stat, err = conn.Children("/non-existent")
@@ -330,7 +330,7 @@
 
 	children, stat, watch, err := conn.ChildrenW("/")
 	c.Assert(err, IsNil)
-	c.Assert(children, Equals, []string{"zookeeper"})
+	c.Assert(children, DeepEquals, []string{"zookeeper"})
 	c.Assert(stat.NumChildren(), Equals, 1)
 
 	select {
@@ -355,7 +355,7 @@
 	c.Assert(stat.NumChildren(), Equals, 2)
 
 	// The ordering is most likely unstable, so this test must be fixed.
-	c.Assert(children, Equals, []string{"test1", "zookeeper"})
+	c.Assert(children, DeepEquals, []string{"test1", "zookeeper"})
 
 	select {
 	case <-watch:
@@ -481,7 +481,7 @@
 	defer zk2.Close()
 	clientId2 := zk2.ClientId()
 
-	c.Assert(clientId1, Equals, clientId2)
+	c.Assert(clientId1, DeepEquals, clientId2)
 }
 
 // Surprisingly for some (including myself, initially), the watch
@@ -512,7 +512,7 @@
 
 	acl, stat, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_ALL))
+	c.Assert(acl, DeepEquals, zk.WorldACL(zk.PERM_ALL))
 	c.Assert(stat, NotNil)
 	c.Assert(stat.Version(), Equals, 0)
 
@@ -538,7 +538,7 @@
 
 	acl, _, err := conn.ACL("/test")
 	c.Assert(err, IsNil)
-	c.Assert(acl, Equals, zk.WorldACL(zk.PERM_READ))
+	c.Assert(acl, DeepEquals, zk.WorldACL(zk.PERM_READ))
 }
 
 func (s *S) TestAddAuth(c *C) {