add presence.ChildrenWatcher

state/presence/presence.go

 // The presence package is intended as a replacement for zookeeper ephemeral
 // nodes; the primary difference is that node timeout is unrelated to session
 // timeout, and this allows us to restart a presence-enabled process "silently"
 // (from the perspective of the rest of the system) without dealing with the
 // complication of session re-establishment.
 
 package presence
 
 import (
         "fmt"
         zk "launchpad.net/gozk/zookeeper"
+        "launchpad.net/juju-core/state/watcher"
         "launchpad.net/tomb"
+        "sync"
         "time"
 )
 
 // changeNode wraps a zookeeper node and can induce watches on that node to fire.
 type changeNode struct {
         conn    *zk.Conn
         path    string
         content string
 }
 
@@ skipping 249 matching lines @@
                         return fmt.Errorf("presence: channel closed while waiting")
                 }
                 if !alive {
                         return fmt.Errorf("presence: alive watch misbehaved while waiting")
                 }
         case <-time.After(timeout):
                 return fmt.Errorf("presence: still not alive after timeout")
         }
         return nil
 }
+
+// ChildrenWatcher mimics state.watcher.ChildrenWatcher, but treats nodes that

[niemeyer, 2012/06/27 05:41:57]

state.watcher.ChildrenWatcher => state/watcher's ChildrenWatcher

That dotted notation doesn't exist in Go.

[fwereade, 2012/06/27 14:52:51]

Good point :). Done.

+// do not have active Pingers as nonexistent.
+type ChildrenWatcher struct {
+        conn    *zk.Conn
+        tomb    tomb.Tomb
+        path    string
+        alive   map[string]bool
+        stops   map[string]chan struct{}
+        watcher *watcher.ChildrenWatcher
+        updates chan aliveChange
+        changes chan watcher.ChildrenChange
+        wg      sync.WaitGroup
+}
+
+// aliveChange is used internally by ChildrenWatcher to communicate node
+// status changes.
+type aliveChange struct {
+        key   string
+        alive bool
+}
+
+// NewChildrenWatcher returns a ChildrenWatcher that notifies of the
+// presence and absence of Pingers in the direct child nodes of path.
+func NewChildrenWatcher(conn *zk.Conn, path string) *ChildrenWatcher {
+        w := &ChildrenWatcher{
+                conn:    conn,
+                path:    path,
+                alive:   make(map[string]bool),
+                stops:   make(map[string]chan struct{}),
+                watcher: watcher.NewChildrenWatcher(conn, path),
+                updates: make(chan aliveChange),
+                changes: make(chan watcher.ChildrenChange),
+        }
+        go w.loop()
+        return w
+}
+
+// Changes returns a channel on which presence changes can be received.
+// The first event returns the current set of children on which Pingers
+// are active.
+func (w *ChildrenWatcher) Changes() <-chan watcher.ChildrenChange {
+        return w.changes
+}
+
+// Stop terminates the watcher and returns any error encountered while watching.
+func (w *ChildrenWatcher) Stop() error {
+        w.tomb.Kill(nil)
+        return w.tomb.Wait()
+}
+
+func (w *ChildrenWatcher) loop() {
+        defer func() {

[niemeyer, 2012/06/27 05:41:57]

There's no reason for this to be a closure. defer w.finish() or similar would
work.

[fwereade, 2012/06/27 14:52:51]

Changed enough that Doneness is moot :).

+                if err := w.watcher.Stop(); err != nil {
+                        w.tomb.Kill(err)
+                }
+                for _, stop := range w.stops {
+                        close(stop)
+                }
+                // Wait for the child loops to terminate before
+                // closing the channel on which they send.
+                w.wg.Wait()
+                close(w.updates)
+                close(w.changes)

[niemeyer, 2012/06/27 05:41:57]

Closing these channels seems to just create extra overhead for handling below. I
suspect keeping them unclosed will make the logic simpler without introducing
issues.

[fwereade, 2012/06/27 14:52:51]

Done.

+                w.tomb.Done()
+        }()
+        emittedValue := false
+        for {
+                var change watcher.ChildrenChange
+                select {
+                case <-w.tomb.Dying():
+                        return
+                case ch, ok := <-w.watcher.Changes():
+                        var err error
+                        if !ok {
+                                if err = w.watcher.Err(); err == nil {
+                                        panic("candidates channel closed unexpectedly")
+                                }
+                        }
+                        if err == nil {
+                                change, err = w.updateWatches(ch)
+                        }
+                        if err != nil {
+                                w.tomb.Kill(err)
+                                return
+                        }
+                case ch, ok := <-w.updates:
+                        if !ok {
+                                panic("updates channel closed unexpectedly")

[niemeyer, 2012/06/27 05:41:57]

If closing it can only lead to panics, we don't need to close it.

[fwereade, 2012/06/27 14:52:51]

Done.

+                        }
+                        change = w.updatePresence(ch)
+                }
+                if emittedValue && len(change.Added) == 0 && len(change.Deleted) == 0 {
+                        continue
+                }
+                select {
+                case <-w.tomb.Dying():
+                        return
+                case w.changes <- change:
+                        emittedValue = true
+                }
+        }
+}
+
+// updateWatches stops existing presence watches on deleted candidates, and
+// starts new presence watches on newly-added candidates. It returns a
+// ChildrenChange representing only those changes that correspond to the
+// presence or absence of a Pinger on a candidate node.
+func (w *ChildrenWatcher) updateWatches(ch watcher.ChildrenChange) (watcher.ChildrenChange, error) {
+        change := watcher.ChildrenChange{}
+        for _, key := range ch.Deleted {
+                stop := w.stops[key]
+                delete(w.stops, key)
+                close(stop)

[niemeyer, 2012/06/27 05:41:57]

What if we discarded delete notification for nodes coming from the children
watcher, and allowed the presence watcher itself to handle all deletions?
Wouldn't that suppress the need for all those stops, and avoid the race as well?

[fwereade, 2012/06/27 14:52:51]

Very neat.

/me removes hat.

Done.

+                if w.alive[key] {
+                        // Race: a node deletion might have already been noted by
+                        // the child loop, and already passed on in a previous change.
+                        delete(w.alive, key)
+                        change.Deleted = append(change.Deleted, key)
+                }
+        }
+        for _, key := range ch.Added {
+                alive, aliveW, err := AliveW(w.conn, w.path+"/"+key)
+                if err != nil {
+                        return watcher.ChildrenChange{}, err
+                }
+                if alive {
+                        w.alive[key] = true
+                        change.Added = append(change.Added, key)
+                }
+                w.stops[key] = make(chan struct{})
+                w.wg.Add(1)
+                go w.childLoop(key, aliveW, w.stops[key])
+        }
+        return change, nil
+}
+
+// childLoop sends aliveChange events to w.updates, in response to presence
+// changes received from watch, until it is stopped.
+func (w *ChildrenWatcher) childLoop(key string, watch <-chan bool, stop <-chan struct{}) {
+        defer w.wg.Done()
+        for {
+                select {
+                case <-stop:

[niemeyer, 2012/06/27 05:41:57]

If the above suggestion works, it doesn't feel like we need this. We need to
track whether they're running with wg so we can wait, but this can also monitor
the same tomb as a termination flag.

[fwereade, 2012/06/27 14:52:51]

I don't think we even need wg any more :).

+                        return
+                case alive, ok := <-watch:
+                        if !ok {
+                                w.tomb.Killf("presence watch closed unexpectedly")

[niemeyer, 2012/06/27 05:41:57]

Watchers can close channels due to a number of reasons, including network
errors. This shouldn't be assuming it's unexpected to get an error from the
watcher.

Also, it's not returning.

[fwereade, 2012/06/27 14:52:51]

Tweaked wording. The proper solution, I think, is to have AliveW watches return
events instead of just bools; definitely one for another CL.
Whoops. Done.

+                        }
+                        select {
+                        case <-w.tomb.Dying():
+                                return
+                        case w.updates <- aliveChange{key, alive}:
+                        }
+                }
+        }
+}
+
+// updatePresence interprets an aliveChange event and converts it to
+// a ChildrenChange event.
+func (w *ChildrenWatcher) updatePresence(ch aliveChange) watcher.ChildrenChange {
+        if ch.alive {
+                w.alive[ch.key] = true
+                return watcher.ChildrenChange{Added: []string{ch.key}}
+        }
+        if w.alive[ch.key] {
+                // Race: w.watcher might have processed the node deletion *after*
+                // the childLoop detected the change, but *before* it sent it to
+                // the updates channel (which send we are currently handling).
+                delete(w.alive, ch.key)
+                return watcher.ChildrenChange{Deleted: []string{ch.key}}
+        }
+        return watcher.ChildrenChange{}
+}