<commit_before>// Package publisher provides a type to publish names to a mounttable.
package publisher

// TODO(toddw): Add unittests.

import (
	"fmt"
	"strings"
	"time"

	"veyron.io/veyron/veyron2/context"
	"veyron.io/veyron/veyron2/naming"
	"veyron.io/veyron/veyron2/vlog"
)

// Publisher manages the publishing of servers in mounttable.
type Publisher interface {
	// AddServer adds a new server to be mounted.
	AddServer(server string)
	// RemoveServer removes a server from the list of mounts.
	RemoveServer(server string)
	// AddName adds a new name for all servers to be mounted as.
	AddName(name string)
	// Published returns the published names rooted at the mounttable.
	Published() []string
	// DebugString returns a string representation of the publisher
	// meant solely for debugging.
	DebugString() string
	// Stop causes the publishing to stop and initiates unmounting of the
	// mounted names.  Stop performs the unmounting asynchronously, and
	// WaitForStop should be used to wait until it is done.
	// Once Stop is called Add/RemoveServer and AddName become noops.
	Stop()
	// WaitForStop waits until all unmounting initiated by Stop is finished.
	WaitForStop()
}

// The publisher adds this much slack to each TTL.
const mountTTLSlack = 20 * time.Second

// publisher maintains the name->server associations in the mounttable.  It
// spawns its own goroutine that does the actual work; the publisher itself
// simply coordinates concurrent access by sending and receiving on the
// appropriate channels.
type publisher struct {
	cmdchan  chan interface{} // value is one of {server,name,debug}Cmd
	donechan chan struct{}    // closed when the publisher is done
}

type addServerCmd struct {
	server string        // server to add
	done   chan struct{} // closed when the cmd is done
}

type removeServerCmd struct {
	server string        // server to remove
	done   chan struct{} // closed when the cmd is done
}

type nameCmd struct {
	name string        // name to add
	done chan struct{} // closed when the cmd is done
}

type debugCmd chan string // debug string is sent when the cmd is done

type publishedCmd chan []string // published names are sent when cmd is done

type stopCmd struct{} // sent to the runloop when we want it to exit.

// New returns a new publisher that updates mounts on ns every period.
func New(ctx context.T, ns naming.Namespace, period time.Duration) Publisher {
	p := &publisher{
		cmdchan:  make(chan interface{}),
		donechan: make(chan struct{}),
	}
	go runLoop(ctx, p.cmdchan, p.donechan, ns, period)
	return p
}

func (p *publisher) sendCmd(cmd interface{}) bool {
	select {
	case p.cmdchan <- cmd:
		return true
	case <-p.donechan:
		return false
	}
}

func (p *publisher) AddServer(server string) {
	done := make(chan struct{})
	if p.sendCmd(addServerCmd{server, done}) {
		<-done
	}
}

func (p *publisher) RemoveServer(server string) {
	done := make(chan struct{})
	if p.sendCmd(removeServerCmd{server, done}) {
		<-done
	}
}

func (p *publisher) AddName(name string) {
	done := make(chan struct{})
	if p.sendCmd(nameCmd{name, done}) {
		<-done
	}
}

// Published returns the published name(s) for this publisher, where each name
// is rooted at the mount table(s) where the name has been mounted.
// The names are returned grouped by published name, where all the names
// corresponding the the mount table replicas are grouped together.
func (p *publisher) Published() []string {
	published := make(publishedCmd)
	if p.sendCmd(published) {
		return <-published
	}
	return []string{}
}

func (p *publisher) DebugString() (dbg string) {
	debug := make(debugCmd)
	if p.sendCmd(debug) {
		dbg = <-debug
	} else {
		dbg = "stopped"
	}
	return
}

// Stop stops the publisher, which in practical terms means un-mounting
// everything and preventing any further publish operations.  The caller can
// be confident that no new names or servers will get published once Stop
// returns.  To wait for existing mounts to be cleaned up, use WaitForStop.
//
// Stopping the publisher is irreversible.
//
// Once the publisher is stopped, any further calls on its public methods
// (including Stop) are no-ops.
func (p *publisher) Stop() {
	p.sendCmd(stopCmd{})
}

func (p *publisher) WaitForStop() {
	<-p.donechan
}

func runLoop(ctx context.T, cmdchan chan interface{}, donechan chan struct{}, ns naming.Namespace, period time.Duration) {
	vlog.VI(2).Info("ipc pub: start runLoop")
	state := newPubState(ctx, ns, period)

	for {
		select {
		case cmd := <-cmdchan:
			switch tcmd := cmd.(type) {
			case stopCmd:
				close(donechan)
				state.unmountAll()
				vlog.VI(2).Info("ipc pub: exit runLoop")
				return
			case addServerCmd:
				state.addServer(tcmd.server)
				close(tcmd.done)
			case removeServerCmd:
				state.removeServer(tcmd.server)
				close(tcmd.done)
			case nameCmd:
				state.addName(tcmd.name)
				close(tcmd.done)
			case publishedCmd:
				tcmd <- state.published()
				close(tcmd)
			case debugCmd:
				tcmd <- state.debugString()
				close(tcmd)
			}
		case <-state.timeout():
			// Sync everything once every period, to refresh the ttls.
			state.sync()
		}
	}
}<commit_msg>veyron/runtimes/google/lib/publisher: We should finish unmounting before claiming we are done.<commit_after>// Package publisher provides a type to publish names to a mounttable.
package publisher

// TODO(toddw): Add unittests.

import (
	"fmt"
	"strings"
	"time"

	"veyron.io/veyron/veyron2/context"
	"veyron.io/veyron/veyron2/naming"
	"veyron.io/veyron/veyron2/vlog"
)

// Publisher manages the publishing of servers in mounttable.
type Publisher interface {
	// AddServer adds a new server to be mounted.
	AddServer(server string)
	// RemoveServer removes a server from the list of mounts.
	RemoveServer(server string)
	// AddName adds a new name for all servers to be mounted as.
	AddName(name string)
	// Published returns the published names rooted at the mounttable.
	Published() []string
	// DebugString returns a string representation of the publisher
	// meant solely for debugging.
	DebugString() string
	// Stop causes the publishing to stop and initiates unmounting of the
	// mounted names.  Stop performs the unmounting asynchronously, and
	// WaitForStop should be used to wait until it is done.
	// Once Stop is called Add/RemoveServer and AddName become noops.
	Stop()
	// WaitForStop waits until all unmounting initiated by Stop is finished.
	WaitForStop()
}

// The publisher adds this much slack to each TTL.
const mountTTLSlack = 20 * time.Second

// publisher maintains the name->server associations in the mounttable.  It
// spawns its own goroutine that does the actual work; the publisher itself
// simply coordinates concurrent access by sending and receiving on the
// appropriate channels.
type publisher struct {
	cmdchan  chan interface{} // value is one of {server,name,debug}Cmd
	donechan chan struct{}    // closed when the publisher is done
}

type addServerCmd struct {
	server string        // server to add
	done   chan struct{} // closed when the cmd is done
}

type removeServerCmd struct {
	server string        // server to remove
	done   chan struct{} // closed when the cmd is done
}

type nameCmd struct {
	name string        // name to add
	done chan struct{} // closed when the cmd is done
}

type debugCmd chan string // debug string is sent when the cmd is done

type publishedCmd chan []string // published names are sent when cmd is done

type stopCmd struct{} // sent to the runloop when we want it to exit.

// New returns a new publisher that updates mounts on ns every period.
func New(ctx context.T, ns naming.Namespace, period time.Duration) Publisher {
	p := &publisher{
		cmdchan:  make(chan interface{}),
		donechan: make(chan struct{}),
	}
	go runLoop(ctx, p.cmdchan, p.donechan, ns, period)
	return p
}

func (p *publisher) sendCmd(cmd interface{}) bool {
	select {
	case p.cmdchan <- cmd:
		return true
	case <-p.donechan:
		return false
	}
}

func (p *publisher) AddServer(server string) {
	done := make(chan struct{})
	if p.sendCmd(addServerCmd{server, done}) {
		<-done
	}
}

func (p *publisher) RemoveServer(server string) {
	done := make(chan struct{})
	if p.sendCmd(removeServerCmd{server, done}) {
		<-done
	}
}

func (p *publisher) AddName(name string) {
	done := make(chan struct{})
	if p.sendCmd(nameCmd{name, done}) {
		<-done
	}
}

// Published returns the published name(s) for this publisher, where each name
// is rooted at the mount table(s) where the name has been mounted.
// The names are returned grouped by published name, where all the names
// corresponding the the mount table replicas are grouped together.
func (p *publisher) Published() []string {
	published := make(publishedCmd)
	if p.sendCmd(published) {
		return <-published
	}
	return []string{}
}

func (p *publisher) DebugString() (dbg string) {
	debug := make(debugCmd)
	if p.sendCmd(debug) {
		dbg = <-debug
	} else {
		dbg = "stopped"
	}
	return
}

// Stop stops the publisher, which in practical terms means un-mounting
// everything and preventing any further publish operations.  The caller can
// be confident that no new names or servers will get published once Stop
// returns.  To wait for existing mounts to be cleaned up, use WaitForStop.
//
// Stopping the publisher is irreversible.
//
// Once the publisher is stopped, any further calls on its public methods
// (including Stop) are no-ops.
func (p *publisher) Stop() {
	p.sendCmd(stopCmd{})
}

func (p *publisher) WaitForStop() {
	<-p.donechan
}

func runLoop(ctx context.T, cmdchan chan interface{}, donechan chan struct{}, ns naming.Namespace, period time.Duration) {
	vlog.VI(2).Info("ipc pub: start runLoop")
	state := newPubState(ctx, ns, period)

	for {
		select {
		case cmd := <-cmdchan:
			switch tcmd := cmd.(type) {
			case stopCmd:
				state.unmountAll()
				close(donechan)
				vlog.VI(2).Info("ipc pub: exit runLoop")
				return
			case addServerCmd:
				state.addServer(tcmd.server)
				close(tcmd.done)
			case removeServerCmd:
				state.removeServer(tcmd.server)
				close(tcmd.done)
			case nameCmd:
				state.addName(tcmd.name)
				close(tcmd.done)
			case publishedCmd:
				tcmd <- state.published()
				close(tcmd)
			case debugCmd:
				tcmd <- state.debugString()
				close(tcmd)
			}
		case <-state.timeout():
			// Sync everything once every period, to refresh the ttls.
			state.sync()
		}
	}
}
{'fraction_non_alphanumeric': 0.10014443909484834, 'fraction_numerical': 0.001348098218584497, 'mean_word_length': 3.541320507214692, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 5, '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': '2223434', 'n_tokens_mistral': 3381, 'n_tokens_neox': 3135, 'n_words': 1227}