pipeling-viz.go

package main

import (
	"bytes"
	"flag"
	"fmt"
	"log"
	"strconv"
)

type opts struct {
	RTT        float64
	CommitTime float64
	N          int
	Duration   int

	MaxInFlight int
}

func main() {
	var o opts
	flag.Float64Var(&o.RTT, "rtt", 0.2, "network round-trip time")
	flag.Float64Var(&o.CommitTime, "ct", 3, "disk commit time")
	flag.IntVar(&o.N, "n", 16, "number of requests in duration")
	flag.IntVar(&o.Duration, "d", 16, "duration in time units")
	flag.IntVar(&o.MaxInFlight, "max", 3, "max in-flight")

	flag.Parse()

	tikz, err := drawTikz(o)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Print(tikz)
}

type sim struct {
	o                 opts
	leaderLogTimes    []float64
	inFlight          []batch
	complete          []batch
	queueEvents       []queueEvent
	followerBusyUntil float64
}

type batch struct {
	idxStart, idxEnd       int
	leaderSent, leaderRecv float64
	followerRecv           float64
	procStart, procEnd     float64
}

type queueEvent struct {
	time  float64
	depth int
}

func newSim(o opts) sim {
	s := sim{o: o}

	// Work out when logs come in.
	delta := float64(o.Duration) / float64(o.N)
	for t := float64(0); t < float64(o.Duration); t += delta {
		s.leaderLogTimes = append(s.leaderLogTimes, t)
	}

	return s
}

func (s *sim) Run() error {
	sentIdx := -1

	for triggerIdx, triggerT := range s.leaderLogTimes {
		// First, see if we got an ack since the last trigger which would have
		// unblocked the next send ahead of this trigger. By definition there must
		// be space to send if we just got an ack because we never have more than
		// Max and we just effectively decremented the count of inflights.
		if len(s.inFlight) > 0 && s.inFlight[0].leaderRecv < triggerT {
			// Check if there are any unsent logs
			if sentIdx < triggerIdx-1 {
				// Send a batch at the previous ack time for the previous index
				s.sendBatch(sentIdx+1, triggerIdx-1, s.inFlight[0].leaderRecv)
				sentIdx = triggerIdx - 1
			}
		}

		// Are any in-flight requests now complete?
		for _, b := range s.inFlight {
			if b.leaderRecv <= triggerT {
				s.inFlight = s.inFlight[1:]
				s.complete = append(s.complete, b)
				s.updateQueueDepth(b.leaderRecv)
			} else {
				break
			}
		}

		// Does pipeline have space?
		if len(s.inFlight) >= s.o.MaxInFlight {
			// Move forward to the next triggerTime
			continue
		}
		s.sendBatch(sentIdx+1, triggerIdx, triggerT)
		sentIdx = triggerIdx
	}

	// See if we can complete any still in flight without lines going beyond the
	// duration.
	for _, b := range s.inFlight {
		if b.leaderRecv < float64(s.o.Duration) {
			s.complete = append(s.complete, b)
		}
	}

	return nil
}

func (s *sim) updateQueueDepth(t float64) {
	if len(s.queueEvents) > 0 && s.queueEvents[len(s.queueEvents)-1].time == t {
		// Same time as last event, just update it directly
		s.queueEvents[len(s.queueEvents)-1].depth = len(s.inFlight)
		return
	}
	// Append a new observation
	s.queueEvents = append(s.queueEvents, queueEvent{t, len(s.inFlight)})
}

func (s *sim) sendBatch(startIdx, endIdx int, t float64) {
	// Send logs ready on leader
	b := batch{
		idxStart:     startIdx,
		idxEnd:       endIdx,
		leaderSent:   t,
		followerRecv: t + (s.o.RTT / 2),
	}

	// Work out when follower will start and end processing
	if s.followerBusyUntil <= b.followerRecv {
		// Follower will be idle at receive time and can start right away
		b.procStart = b.followerRecv
		b.procEnd = b.procStart + s.o.CommitTime
	} else {
		b.procStart = s.followerBusyUntil
		b.procEnd = b.procStart + s.o.CommitTime
	}
	s.followerBusyUntil = b.procEnd
	b.leaderRecv = b.procEnd + (s.o.RTT / 2)
	s.inFlight = append(s.inFlight, b)
	s.updateQueueDepth(t)
}

func drawTikz(o opts) (string, error) {

	s := newSim(o)
	if err := s.Run(); err != nil {
		return "", err
	}

	// Draw LeadLog->Replication ticks
	var leaderLogs bytes.Buffer
	for idx, t := range s.leaderLogTimes {
		// indexes are zero based in sim but display as 1-based to match raft indexes
		fmt.Fprintf(&leaderLogs, "  \\draw[append,->]  (%[1]f,3) -- (%[1]f,2) node[above right,near start] {%[2]d};\n", t, idx+1)
	}

	// Draw replication batches
	var batches bytes.Buffer
	for _, b := range s.complete {
		fmt.Fprintf(&leaderLogs,
			`
  \draw[append,->]  (%[2]f,2) -- (%[3]f,0) node[above right,near start] {%[1]d};
  \draw[append,->]  (%[3]f,0) -- (%[4]f,-1) node[above right,near end] {%[1]d};
  \draw[work]  (%[4]f,-1.1) rectangle (%[5]f,-0.9);
  \draw[ack,->]  (%[5]f,-1) -- (%[6]f,2) node[below right,near end] {%[1]d};
	`, b.idxEnd+1, b.leaderSent, b.followerRecv, b.procStart, b.procEnd, b.leaderRecv)
	}

	// Draw in-flight bars
	var inFlight bytes.Buffer
	scaleFactor := 1 / float64(o.MaxInFlight)
	for _, qe := range s.queueEvents {
		fmt.Fprintf(&inFlight,
			"  \\draw[bar]  (%f,-3) rectangle ++(0.1,%f) node[help lines, right] {%d};\n",
			qe.time, float64(qe.depth)*scaleFactor, qe.depth,
		)
	}

	exemplar := ""
	if len(s.complete) > 1 {
		ex := s.complete[len(s.complete)-1]
		start := s.leaderLogTimes[ex.idxStart]
		exemplar = fmt.Sprintf(
			"\\draw[draw=black,|-|] (%f, 3.5) node[above right] {Idx %d Latency  (%.1f units)}-- (%f, 3.5);",
			start, ex.idxStart+1, ex.leaderRecv-start, ex.leaderRecv,
		)
	}

	tikz := fmt.Sprintf(`
\documentclass[tikz,border=2cm]{standalone}
\usepackage[utf8]{inputenc}
\usepackage[T1]{fontenc}
\usepackage{lmodern}

\tikzset{
	host/.style={rectangle,rounded corners,
								thick,draw=#1!60,fill=#1!15},
	work/.style={rectangle,draw=gray,fill=none},
	bar/.style={rectangle,draw=gray,fill=gray},
	host/.default=blue,
	append/.style={thick,draw=#1!60,fill=#1!60},
	append/.default=purple,
	ack/.style={thick,draw=#1!60,fill=#1!60},
	ack/.default=green,
}

\usetikzlibrary{calc,arrows,positioning}
\begin{document}
\begin{tikzpicture}[>=stealth', font=\small\sffamily]

  \node[anchor=south west] at (-2,4.5) {\large Max In-flight %[1]d, RTT %[6]s, Service Time %[7]s};

  \draw[help lines,->] (-0.2,3) node[host,left] {LeaderLog}--(%[2]d,3);
  \draw[help lines,->] (-0.2,2) node[host,left] {Leader Replication}--(%[2]d,2);
  \draw[help lines,->] (-0.2,0) node[host,left] {Network Buffer}--(%[2]d,0);
  \draw[help lines,->] (-0.2,-1) node[host,left] {Follower}--(%[2]d,-1);
  %[3]s
  %[4]s

	%[5]s

  \draw[help lines,->] (-0.2,-3) --(%[2]d,-3);
  \node[anchor=south east] at (-0.5,-2.8) {In-flight};
  \node[help lines] at (-0.4,-2) {%[1]d};
  \node[help lines] at (-0.4,-3) {0};
  \draw[help lines,-] (-0.2,-3) -- (-0.2,-2);

	%[8]s

\end{tikzpicture}
\end{document}
`, o.MaxInFlight, o.Duration, leaderLogs.String(), batches.String(), exemplar,
		// Format floats without trailing zeros
		strconv.FormatFloat(o.RTT, 'f', -1, 64),
		strconv.FormatFloat(o.CommitTime, 'f', -1, 64),
		inFlight.String(),
	)

	return tikz, nil
}