node_driver.go

package paxos

import (
	"context"
	"time"
)

// nodeDriver is a thin wrapper around the deterministic protocol logic in protocol_node.go
// this wrapper is responsible for running the protocol logic in a single goroutine, and for performing the side effects that the protocol logic decides on (sending messages, setting timers, etc)

// to reason better about the flow:
/*
	- nodeDriver is a loop waiting for events that concern the node (proposals, messages, retry timer firing)
	- when an event is received (say, by the 'network'), it is passed to the node protocol logic (node.Handle(event)), which (potentially) returns actions (for the driver to act upon and perform side affects) and traces (to be displayed in CLI, e.g. when a proposer gives up on its round).

	Example:
	 - nodeDriver receives a messageReceived event that is a Prepare request from a proposer
	 - node.Handle(messageReceived) returns actions, including sendMessage to the proposer (a promise in response to the prepare)
	 - nodeDriver performs the sendMessage action by calling transport.Send(...)

*/
type nodeDriver struct {
	protocol *node // the brain that will result in nodeActions, pure and deterministic

	ctx       context.Context
	events    chan nodeEvent // messageReceived, proposalTriggered, retryTimerFired, etc - events that the node protocol logic needs to react to
	transport network

	traces     chan<- TraceEntry // a channel to report protocol traces to the outside world
	retryTimer retryTimer
}

type TraceEntry struct {
	NodeID int
	Event  TraceEvent
}

func newNodeDriver(protocol *node, transport network, traces chan<- TraceEntry) *nodeDriver {
	return &nodeDriver{
		protocol:  protocol,
		events:    make(chan nodeEvent, 64),
		transport: transport,
		traces:    traces,
	}
}

func (d *nodeDriver) startLoop(ctx context.Context) {
	d.ctx = ctx
	go func() {
		defer d.retryTimer.stop()

		for {
			select {
			case <-d.ctx.Done():
				return
			case event := <-d.events:
				output := d.protocol.Handle(event)
				d.applyActions(output.actions)
				d.reportTraces(output.traces)
			}
		}
	}()
}

func (d *nodeDriver) applyActions(actions []nodeAction) {
	for _, action := range actions {
		switch action := action.(type) {
		case sendMessage:
			d.transport.Send(d.ctx, d.protocol.id, action.to, action.msg)
		case resetRetryTimer:
			d.resetRetryTimer(action.ballot, action.after)
		case stopRetryTimer:
			d.retryTimer.stop()
		}
	}
}

func (d *nodeDriver) reportTraces(traces []TraceEvent) {
	for _, event := range traces {
		select {
		case <-d.ctx.Done():
			return
		case d.traces <- TraceEntry{
			NodeID: d.protocol.id,
			Event:  event,
		}:
		}
	}
}

// this is called by the outside world to trigger a new proposal on this node (if it's a proposer)
func (d *nodeDriver) propose(value string) {
	d.enqueue(proposalTriggered{value: value})
}

// called by the transport, this is the transport's entry point for message delivery
func (d *nodeDriver) receive(from int, msg Message) {
	d.enqueue(messageReceived{from: from, msg: msg})
}

func (d *nodeDriver) enqueue(event nodeEvent) {
	select {
	case <-d.ctx.Done():
	case d.events <- event:
	}
}

func (d *nodeDriver) resetRetryTimer(ballot Ballot, after time.Duration) {
	// note: in case of races, stale callbacks are harmless because the protocol logic is inherently designed to handle them correctly (late promises etc)
	d.retryTimer.reset(after, func() {
		d.enqueue(retryTimerFired{ballot: ballot})
	})
}

// thin wrapper for readability, could be deleted
type retryTimer struct {
	timer *time.Timer
}

func (t *retryTimer) reset(after time.Duration, fire func()) {
	t.stop()
	t.timer = time.AfterFunc(after, fire)
}

func (t *retryTimer) stop() {
	if t.timer != nil {
		t.timer.Stop()
		t.timer = nil
	}
}