<commit_before>// 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/log"
	"launchpad.net/tomb"
	"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
}

// change sets the zookeeper node's content (creating it if it doesn't exist) and
// returns the node's new MTime.
func (n *changeNode) change() (mtime time.Time, err error) {
	stat, err := n.conn.Set(n.path, n.content, -1)
	if zk.IsError(err, zk.ZNONODE) {
		_, err = n.conn.Create(n.path, n.content, 0, zk.WorldACL(zk.PERM_ALL))
		if err == nil || zk.IsError(err, zk.ZNODEEXISTS) {
			// *Someone* created the node anyway; just try again.
			return n.change()
		}
	}
	if err != nil {
		return
	}
	return stat.MTime(), nil
}

// Pinger periodically updates a node in zookeeper while it is running.
type Pinger struct {
	conn      *zk.Conn
	target    changeNode
	period    time.Duration
	tomb      tomb.Tomb
	stayAlive chan struct{}
}

// StartPinger returns a new Pinger that refreshes the content of path periodically.
func StartPinger(conn *zk.Conn, path string, period time.Duration) (*Pinger, error) {
	target := changeNode{conn, path, period.String()}
	_, err := target.change()
	if err != nil {
		return nil, err
	}
	p := &Pinger{
		conn:      conn,
		target:    target,
		period:    period,
		stayAlive: make(chan struct{}),
	}
	go p.loop()
	return p, nil
}

// loop calls change on p.target every p.period nanoseconds until p is stopped.
func (p *Pinger) loop() {
	defer p.finish()
	var wait time.Duration
	var then time.Time
	for {
		select {
		case <-p.tomb.Dying():
			log.Debugf("presence: %s pinger died after %s wait", p.target.path, time.Now().Sub(then))
			return
		case now := <-time.After(wait):
			mtime, err := p.target.change()
			if err != nil {
				p.tomb.Kill(err)
				return
			}
			next := then.Add(p.period)
			wait = next.Sub(now)
			log.Debugf("presence: pinged %s at %s; waiting %s", p.target.path, mtime, wait)
			then = now
		}
	}
}

// finish marks the Pinger as finally dead; it may also trigger a final write
// to the target node, if the Pinger has been cleanly shut down via the Stop
// method.
func (p *Pinger) finish() {
	if p.tomb.Err() == nil {
		select {
		case <-p.stayAlive:
			if _, err := p.target.change(); err != nil {
				p.tomb.Kill(err)
			}
		default:
		}
	}
	p.tomb.Done()
}

// Dying returns a channel that will be closed when the Pinger is no longer
// operating.
func (p *Pinger) Dying() <-chan struct{} {
	return p.tomb.Dying()
}

// Stop stops updating the node; AliveW watches will not notice any change
// until they time out. If the pinger is shut down without errors, a final
// write to the target node will occur, to ensure that remote watchers take
// as long as possible to detect the Pinger's absence.
func (p *Pinger) Stop() error {
	close(p.stayAlive)
	p.tomb.Kill(nil)
	return p.tomb.Wait()
}

// Kill stops updating and deletes the node, causing any AliveW watches
// to observe its departure (almost) immediately.
func (p *Pinger) Kill() (err error) {
	p.tomb.Kill(nil)
	if err = p.tomb.Wait(); err == nil {
		if err = p.conn.Delete(p.target.path, -1); zk.IsError(err, zk.ZNONODE) {
			return nil
		}
	}
	return
}

// node represents the state of a presence node from a watcher's perspective.
type node struct {
	conn    *zk.Conn
	path    string
	alive   bool
	timeout time.Duration
}

// setState sets the current values of n.alive and n.timeout, given the
// content and stat of the zookeeper node (which must exist). firstTime
// should be true iff n.setState has not already been called.
func (n *node) setState(content string, stat *zk.Stat, firstTime bool) error {
	// Always check and reset timeout; it could change.
	period, err := time.ParseDuration(content)
	if err != nil {
		return fmt.Errorf("%s presence node has bad data: %q", n.path, content)
	}
	n.timeout = period * 3
	if firstTime {
		clock := changeNode{n.conn, "/clock", ""}
		now, err := clock.change()
		if err != nil {
			return err
		}<commit_msg>more clarity, better docs, in presence.go<commit_after>// 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/log"
	"launchpad.net/tomb"
	"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
}

// change sets the zookeeper node's content (creating it if it doesn't exist) and
// returns the node's new MTime.
func (n *changeNode) change() (mtime time.Time, err error) {
	stat, err := n.conn.Set(n.path, n.content, -1)
	if zk.IsError(err, zk.ZNONODE) {
		_, err = n.conn.Create(n.path, n.content, 0, zk.WorldACL(zk.PERM_ALL))
		if err == nil || zk.IsError(err, zk.ZNODEEXISTS) {
			// *Someone* created the node anyway; just try again.
			return n.change()
		}
	}
	if err != nil {
		return
	}
	return stat.MTime(), nil
}

// Pinger periodically updates a node in zookeeper while it is running.
type Pinger struct {
	conn      *zk.Conn
	tomb      tomb.Tomb
	target    changeNode
	period    time.Duration
	lastPing  time.Time
	stayAlive chan struct{}
}

// StartPinger returns a new Pinger that refreshes the content of path periodically.
func StartPinger(conn *zk.Conn, path string, period time.Duration) (*Pinger, error) {
	target := changeNode{conn, path, period.String()}
	_, err := target.change()
	if err != nil {
		return nil, err
	}
	p := &Pinger{
		conn:      conn,
		target:    target,
		period:    period,
		lastPing:  time.Now(),
		stayAlive: make(chan struct{}),
	}
	go p.loop()
	return p, nil
}

// loop calls change on p.target every p.period nanoseconds until p is stopped.
func (p *Pinger) loop() {
	defer p.finish()
	tick := p.getTick()
	for {
		select {
		case <-p.tomb.Dying():
			log.Debugf("presence: %s pinger died after %s wait", p.target.path, time.Now().Sub(p.lastPing))
			return
		case now := <-tick:
			log.Debugf("presence: %s pinger awakened after %s wait", p.target.path, time.Now().Sub(p.lastPing))
			p.lastPing = now
			mtime, err := p.target.change()
			if err != nil {
				p.tomb.Kill(err)
				return
			}
			log.Debugf("presence: wrote to %s at (zk) %s", p.target.path, mtime)
			tick = p.getTick()
		}
	}
}

// getTick returns a channel that will receive an event when the pinger is next
// due to fire.
func (p *Pinger) getTick() <-chan time.Time {
	next := p.lastPing.Add(p.period)
	wait := next.Sub(time.Now())
	if wait <= 0 {
		wait = 0
	}
	log.Debugf("presence: anticipating ping of %s in %s", p.target.path, wait)
	return time.After(wait)
}

// finish marks the Pinger as finally dead; it may also trigger a final write
// to the target node, if the Pinger has been cleanly shut down via the Stop
// method.
func (p *Pinger) finish() {
	if p.tomb.Err() == nil {
		select {
		case <-p.stayAlive:
			if _, err := p.target.change(); err != nil {
				p.tomb.Kill(err)
			}
		default:
		}
	}
	p.tomb.Done()
}

// Dying returns a channel that will be closed when the Pinger is no longer
// operating.
func (p *Pinger) Dying() <-chan struct{} {
	return p.tomb.Dying()
}

// Stop stops updating the node; AliveW watches will not notice any change
// until they time out. If the pinger is shut down without errors, a final
// write to the target node will occur, to ensure that remote watchers take
// as long as possible to detect the Pinger's absence.
func (p *Pinger) Stop() error {
	close(p.stayAlive)
	p.tomb.Kill(nil)
	return p.tomb.Wait()
}

// Kill stops updating and deletes the node, causing any AliveW watches
// to observe its departure (almost) immediately.
func (p *Pinger) Kill() (err error) {
	p.tomb.Kill(nil)
	if err = p.tomb.Wait(); err == nil {
		if err = p.conn.Delete(p.target.path, -1); zk.IsError(err, zk.ZNONODE) {
			return nil
		}
	}
	return
}

// node represents the state of a presence node from a watcher's perspective.
type node struct {
	conn    *zk.Conn
	path    string
	alive   bool
	timeout time.Duration
}

// setState sets the current values of n.alive and n.timeout, given the
// content and stat of the zookeeper node (which must exist). firstTime
// should be true iff n.setState has not already been called.
func (n *node) setState(content string, stat *zk.Stat, firstTime bool) error {
	// Always check and reset timeout; it could change.
	period, err := time.ParseDuration(content)
	if err != nil {
		return fmt.Errorf("%s presence node has bad data: %q", n.path, content)
	}
	// Worst observed combination of GC pauses and scheduler weirdness led to
	// a gap between "500ms" pings of >1000ms; timeout should comfortably beat
	// that. This is largely an artifact of the notably short periods used in
	// tests; actual distributed Pingers should use longer periods in order to
	// compensate for network flakiness... and to allow a comfortable buffer
	// for Pinger re-establishment by a replacement process when (eg when
	// upgrading agent binaries).
	n.timeout = period * 3
	if firstTime {
		clock := changeNode{n.conn, "/clock", ""}
		now, err := clock.change()
		if err != nil {
			return err
		}
{'fraction_non_alphanumeric': 0.11795756775961831, 'fraction_numerical': 0.001725713125571008, 'mean_word_length': 3.257562662057044, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7527652', 'n_tokens_mistral': 3515, 'n_tokens_neox': 3291, 'n_words': 1233}