libs: internalize some packages (#6366)

## Description

Internalize some libs. This reduces the amount ot public API tendermint is supporting. The moved libraries are mainly ones that are used within Tendermint-core.

internal/libs/autofile/README.md

@@ -0,0 +1 @@
+# go-autofile

internal/libs/autofile/autofile.go

@@ -0,0 +1,194 @@
+package autofile
+
+import (
+	"os"
+	"os/signal"
+	"path/filepath"
+	"sync"
+	"syscall"
+	"time"
+
+	tmrand "github.com/tendermint/tendermint/libs/rand"
+)
+
+/* AutoFile usage
+
+// Create/Append to ./autofile_test
+af, err := OpenAutoFile("autofile_test")
+if err != nil {
+	panic(err)
+}
+
+// Stream of writes.
+// During this time, the file may be moved e.g. by logRotate.
+for i := 0; i < 60; i++ {
+	af.Write([]byte(Fmt("LOOP(%v)", i)))
+	time.Sleep(time.Second)
+}
+
+// Close the AutoFile
+err = af.Close()
+if err != nil {
+	panic(err)
+}
+*/
+
+const (
+	autoFileClosePeriod = 1000 * time.Millisecond
+	autoFilePerms       = os.FileMode(0600)
+)
+
+// AutoFile automatically closes and re-opens file for writing. The file is
+// automatically setup to close itself every 1s and upon receiving SIGHUP.
+//
+// This is useful for using a log file with the logrotate tool.
+type AutoFile struct {
+	ID   string
+	Path string
+
+	closeTicker      *time.Ticker
+	closeTickerStopc chan struct{} // closed when closeTicker is stopped
+	hupc             chan os.Signal
+
+	mtx  sync.Mutex
+	file *os.File
+}
+
+// OpenAutoFile creates an AutoFile in the path (with random ID). If there is
+// an error, it will be of type *PathError or *ErrPermissionsChanged (if file's
+// permissions got changed (should be 0600)).
+func OpenAutoFile(path string) (*AutoFile, error) {
+	var err error
+	path, err = filepath.Abs(path)
+	if err != nil {
+		return nil, err
+	}
+	af := &AutoFile{
+		ID:               tmrand.Str(12) + ":" + path,
+		Path:             path,
+		closeTicker:      time.NewTicker(autoFileClosePeriod),
+		closeTickerStopc: make(chan struct{}),
+	}
+	if err := af.openFile(); err != nil {
+		af.Close()
+		return nil, err
+	}
+
+	// Close file on SIGHUP.
+	af.hupc = make(chan os.Signal, 1)
+	signal.Notify(af.hupc, syscall.SIGHUP)
+	go func() {
+		for range af.hupc {
+			_ = af.closeFile()
+		}
+	}()
+
+	go af.closeFileRoutine()
+
+	return af, nil
+}
+
+// Close shuts down the closing goroutine, SIGHUP handler and closes the
+// AutoFile.
+func (af *AutoFile) Close() error {
+	af.closeTicker.Stop()
+	close(af.closeTickerStopc)
+	if af.hupc != nil {
+		close(af.hupc)
+	}
+	return af.closeFile()
+}
+
+func (af *AutoFile) closeFileRoutine() {
+	for {
+		select {
+		case <-af.closeTicker.C:
+			_ = af.closeFile()
+		case <-af.closeTickerStopc:
+			return
+		}
+	}
+}
+
+func (af *AutoFile) closeFile() (err error) {
+	af.mtx.Lock()
+	defer af.mtx.Unlock()
+
+	file := af.file
+	if file == nil {
+		return nil
+	}
+
+	af.file = nil
+	return file.Close()
+}
+
+// Write writes len(b) bytes to the AutoFile. It returns the number of bytes
+// written and an error, if any. Write returns a non-nil error when n !=
+// len(b).
+// Opens AutoFile if needed.
+func (af *AutoFile) Write(b []byte) (n int, err error) {
+	af.mtx.Lock()
+	defer af.mtx.Unlock()
+
+	if af.file == nil {
+		if err = af.openFile(); err != nil {
+			return
+		}
+	}
+
+	n, err = af.file.Write(b)
+	return
+}
+
+// Sync commits the current contents of the file to stable storage. Typically,
+// this means flushing the file system's in-memory copy of recently written
+// data to disk.
+// Opens AutoFile if needed.
+func (af *AutoFile) Sync() error {
+	af.mtx.Lock()
+	defer af.mtx.Unlock()
+
+	if af.file == nil {
+		if err := af.openFile(); err != nil {
+			return err
+		}
+	}
+	return af.file.Sync()
+}
+
+func (af *AutoFile) openFile() error {
+	file, err := os.OpenFile(af.Path, os.O_RDWR|os.O_CREATE|os.O_APPEND, autoFilePerms)
+	if err != nil {
+		return err
+	}
+	// fileInfo, err := file.Stat()
+	// if err != nil {
+	// 	return err
+	// }
+	// if fileInfo.Mode() != autoFilePerms {
+	// 	return errors.NewErrPermissionsChanged(file.Name(), fileInfo.Mode(), autoFilePerms)
+	// }
+	af.file = file
+	return nil
+}
+
+// Size returns the size of the AutoFile. It returns -1 and an error if fails
+// get stats or open file.
+// Opens AutoFile if needed.
+func (af *AutoFile) Size() (int64, error) {
+	af.mtx.Lock()
+	defer af.mtx.Unlock()
+
+	if af.file == nil {
+		if err := af.openFile(); err != nil {
+			return -1, err
+		}
+	}
+
+	stat, err := af.file.Stat()
+	if err != nil {
+		return -1, err
+	}
+	return stat.Size(), nil
+}

internal/libs/autofile/autofile_test.go

@@ -0,0 +1,146 @@
+package autofile
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSIGHUP(t *testing.T) {
+	origDir, err := os.Getwd()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := os.Chdir(origDir); err != nil {
+			t.Error(err)
+		}
+	})
+
+	// First, create a temporary directory and move into it
+	dir, err := ioutil.TempDir("", "sighup_test")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = os.RemoveAll(dir)
+	})
+	require.NoError(t, os.Chdir(dir))
+
+	// Create an AutoFile in the temporary directory
+	name := "sighup_test"
+	af, err := OpenAutoFile(name)
+	require.NoError(t, err)
+	require.True(t, filepath.IsAbs(af.Path))
+
+	// Write to the file.
+	_, err = af.Write([]byte("Line 1\n"))
+	require.NoError(t, err)
+	_, err = af.Write([]byte("Line 2\n"))
+	require.NoError(t, err)
+
+	// Move the file over
+	require.NoError(t, os.Rename(name, name+"_old"))
+
+	// Move into a different temporary directory
+	otherDir, err := ioutil.TempDir("", "sighup_test_other")
+	require.NoError(t, err)
+	t.Cleanup(func() { os.RemoveAll(otherDir) })
+	require.NoError(t, os.Chdir(otherDir))
+
+	// Send SIGHUP to self.
+	require.NoError(t, syscall.Kill(syscall.Getpid(), syscall.SIGHUP))
+
+	// Wait a bit... signals are not handled synchronously.
+	time.Sleep(time.Millisecond * 10)
+
+	// Write more to the file.
+	_, err = af.Write([]byte("Line 3\n"))
+	require.NoError(t, err)
+	_, err = af.Write([]byte("Line 4\n"))
+	require.NoError(t, err)
+	require.NoError(t, af.Close())
+
+	// Both files should exist
+	if body := mustReadFile(t, filepath.Join(dir, name+"_old")); string(body) != "Line 1\nLine 2\n" {
+		t.Errorf("unexpected body %s", body)
+	}
+	if body := mustReadFile(t, filepath.Join(dir, name)); string(body) != "Line 3\nLine 4\n" {
+		t.Errorf("unexpected body %s", body)
+	}
+
+	// The current directory should be empty
+	files, err := ioutil.ReadDir(".")
+	require.NoError(t, err)
+	assert.Empty(t, files)
+}
+
+// // Manually modify file permissions, close, and reopen using autofile:
+// // We expect the file permissions to be changed back to the intended perms.
+// func TestOpenAutoFilePerms(t *testing.T) {
+// 	file, err := ioutil.TempFile("", "permission_test")
+// 	require.NoError(t, err)
+// 	err = file.Close()
+// 	require.NoError(t, err)
+// 	name := file.Name()
+
+// 	// open and change permissions
+// 	af, err := OpenAutoFile(name)
+// 	require.NoError(t, err)
+// 	err = af.file.Chmod(0755)
+// 	require.NoError(t, err)
+// 	err = af.Close()
+// 	require.NoError(t, err)
+
+// 	// reopen and expect an ErrPermissionsChanged as Cause
+// 	af, err = OpenAutoFile(name)
+// 	require.Error(t, err)
+// 	if e, ok := err.(*errors.ErrPermissionsChanged); ok {
+// 		t.Logf("%v", e)
+// 	} else {
+// 		t.Errorf("unexpected error %v", e)
+// 	}
+// }
+
+func TestAutoFileSize(t *testing.T) {
+	// First, create an AutoFile writing to a tempfile dir
+	f, err := ioutil.TempFile("", "sighup_test")
+	require.NoError(t, err)
+	require.NoError(t, f.Close())
+
+	// Here is the actual AutoFile.
+	af, err := OpenAutoFile(f.Name())
+	require.NoError(t, err)
+
+	// 1. Empty file
+	size, err := af.Size()
+	require.Zero(t, size)
+	require.NoError(t, err)
+
+	// 2. Not empty file
+	data := []byte("Maniac\n")
+	_, err = af.Write(data)
+	require.NoError(t, err)
+	size, err = af.Size()
+	require.EqualValues(t, len(data), size)
+	require.NoError(t, err)
+
+	// 3. Not existing file
+	require.NoError(t, af.Close())
+	require.NoError(t, os.Remove(f.Name()))
+	size, err = af.Size()
+	require.EqualValues(t, 0, size, "Expected a new file to be empty")
+	require.NoError(t, err)
+
+	// Cleanup
+	t.Cleanup(func() { os.Remove(f.Name()) })
+}
+
+func mustReadFile(t *testing.T, filePath string) []byte {
+	fileBytes, err := ioutil.ReadFile(filePath)
+	require.NoError(t, err)
+
+	return fileBytes
+}

internal/libs/autofile/cmd/logjack.go

@@ -0,0 +1,125 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"strconv"
+	"strings"
+
+	auto "github.com/tendermint/tendermint/internal/libs/autofile"
+	tmos "github.com/tendermint/tendermint/libs/os"
+)
+
+const Version = "0.0.1"
+const readBufferSize = 1024 // 1KB at a time
+
+// Parse command-line options
+func parseFlags() (headPath string, chopSize int64, limitSize int64, version bool) {
+	var flagSet = flag.NewFlagSet(os.Args[0], flag.ExitOnError)
+	var chopSizeStr, limitSizeStr string
+	flagSet.StringVar(&headPath, "head", "logjack.out", "Destination (head) file.")
+	flagSet.StringVar(&chopSizeStr, "chop", "100M", "Move file if greater than this")
+	flagSet.StringVar(&limitSizeStr, "limit", "10G", "Only keep this much (for each specified file). Remove old files.")
+	flagSet.BoolVar(&version, "version", false, "Version")
+	if err := flagSet.Parse(os.Args[1:]); err != nil {
+		fmt.Printf("err parsing flag: %v\n", err)
+		os.Exit(1)
+	}
+	chopSize = parseBytesize(chopSizeStr)
+	limitSize = parseBytesize(limitSizeStr)
+	return
+}
+
+type fmtLogger struct{}
+
+func (fmtLogger) Info(msg string, keyvals ...interface{}) {
+	strs := make([]string, len(keyvals))
+	for i, kv := range keyvals {
+		strs[i] = fmt.Sprintf("%v", kv)
+	}
+	fmt.Printf("%s %s\n", msg, strings.Join(strs, ","))
+}
+
+func main() {
+	// Stop upon receiving SIGTERM or CTRL-C.
+	tmos.TrapSignal(fmtLogger{}, func() {
+		fmt.Println("logjack shutting down")
+	})
+
+	// Read options
+	headPath, chopSize, limitSize, version := parseFlags()
+	if version {
+		fmt.Printf("logjack version %v\n", Version)
+		return
+	}
+
+	// Open Group
+	group, err := auto.OpenGroup(headPath, auto.GroupHeadSizeLimit(chopSize), auto.GroupTotalSizeLimit(limitSize))
+	if err != nil {
+		fmt.Printf("logjack couldn't create output file %v\n", headPath)
+		os.Exit(1)
+	}
+
+	if err = group.Start(); err != nil {
+		fmt.Printf("logjack couldn't start with file %v\n", headPath)
+		os.Exit(1)
+	}
+
+	// Forever read from stdin and write to AutoFile.
+	buf := make([]byte, readBufferSize)
+	for {
+		n, err := os.Stdin.Read(buf)
+		if err != nil {
+			if err := group.Stop(); err != nil {
+				fmt.Fprintf(os.Stderr, "logjack stopped with error %v\n", headPath)
+				os.Exit(1)
+			}
+			if err == io.EOF {
+				os.Exit(0)
+			} else {
+				fmt.Println("logjack errored")
+				os.Exit(1)
+			}
+		}
+		_, err = group.Write(buf[:n])
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "logjack failed write with error %v\n", headPath)
+			os.Exit(1)
+		}
+		if err := group.FlushAndSync(); err != nil {
+			fmt.Fprintf(os.Stderr, "logjack flushsync fail with error %v\n", headPath)
+			os.Exit(1)
+		}
+	}
+}
+
+func parseBytesize(chopSize string) int64 {
+	// Handle suffix multiplier
+	var multiplier int64 = 1
+	if strings.HasSuffix(chopSize, "T") {
+		multiplier = 1042 * 1024 * 1024 * 1024
+		chopSize = chopSize[:len(chopSize)-1]
+	}
+	if strings.HasSuffix(chopSize, "G") {
+		multiplier = 1042 * 1024 * 1024
+		chopSize = chopSize[:len(chopSize)-1]
+	}
+	if strings.HasSuffix(chopSize, "M") {
+		multiplier = 1042 * 1024
+		chopSize = chopSize[:len(chopSize)-1]
+	}
+	if strings.HasSuffix(chopSize, "K") {
+		multiplier = 1042
+		chopSize = chopSize[:len(chopSize)-1]
+	}
+
+	// Parse the numeric part
+	chopSizeInt, err := strconv.Atoi(chopSize)
+	if err != nil {
+		panic(err)
+	}
+
+	return int64(chopSizeInt) * multiplier
+}

internal/libs/autofile/group.go

@@ -0,0 +1,540 @@
+package autofile
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/tendermint/tendermint/libs/service"
+)
+
+const (
+	defaultGroupCheckDuration = 5000 * time.Millisecond
+	defaultHeadSizeLimit      = 10 * 1024 * 1024       // 10MB
+	defaultTotalSizeLimit     = 1 * 1024 * 1024 * 1024 // 1GB
+	maxFilesToRemove          = 4                      // needs to be greater than 1
+)
+
+/*
+You can open a Group to keep restrictions on an AutoFile, like
+the maximum size of each chunk, and/or the total amount of bytes
+stored in the group.
+
+The first file to be written in the Group.Dir is the head file.
+
+	Dir/
+	- <HeadPath>
+
+Once the Head file reaches the size limit, it will be rotated.
+
+	Dir/
+	- <HeadPath>.000   // First rolled file
+	- <HeadPath>       // New head path, starts empty.
+										 // The implicit index is 001.
+
+As more files are written, the index numbers grow...
+
+	Dir/
+	- <HeadPath>.000   // First rolled file
+	- <HeadPath>.001   // Second rolled file
+	- ...
+	- <HeadPath>       // New head path
+
+The Group can also be used to binary-search for some line,
+assuming that marker lines are written occasionally.
+*/
+type Group struct {
+	service.BaseService
+
+	ID                 string
+	Head               *AutoFile // The head AutoFile to write to
+	headBuf            *bufio.Writer
+	Dir                string // Directory that contains .Head
+	ticker             *time.Ticker
+	mtx                sync.Mutex
+	headSizeLimit      int64
+	totalSizeLimit     int64
+	groupCheckDuration time.Duration
+	minIndex           int // Includes head
+	maxIndex           int // Includes head, where Head will move to
+
+	// close this when the processTicks routine is done.
+	// this ensures we can cleanup the dir after calling Stop
+	// and the routine won't be trying to access it anymore
+	doneProcessTicks chan struct{}
+
+	// TODO: When we start deleting files, we need to start tracking GroupReaders
+	// and their dependencies.
+}
+
+// OpenGroup creates a new Group with head at headPath. It returns an error if
+// it fails to open head file.
+func OpenGroup(headPath string, groupOptions ...func(*Group)) (*Group, error) {
+	dir, err := filepath.Abs(filepath.Dir(headPath))
+	if err != nil {
+		return nil, err
+	}
+	head, err := OpenAutoFile(headPath)
+	if err != nil {
+		return nil, err
+	}
+
+	g := &Group{
+		ID:                 "group:" + head.ID,
+		Head:               head,
+		headBuf:            bufio.NewWriterSize(head, 4096*10),
+		Dir:                dir,
+		headSizeLimit:      defaultHeadSizeLimit,
+		totalSizeLimit:     defaultTotalSizeLimit,
+		groupCheckDuration: defaultGroupCheckDuration,
+		minIndex:           0,
+		maxIndex:           0,
+		doneProcessTicks:   make(chan struct{}),
+	}
+
+	for _, option := range groupOptions {
+		option(g)
+	}
+
+	g.BaseService = *service.NewBaseService(nil, "Group", g)
+
+	gInfo := g.readGroupInfo()
+	g.minIndex = gInfo.MinIndex
+	g.maxIndex = gInfo.MaxIndex
+	return g, nil
+}
+
+// GroupCheckDuration allows you to overwrite default groupCheckDuration.
+func GroupCheckDuration(duration time.Duration) func(*Group) {
+	return func(g *Group) {
+		g.groupCheckDuration = duration
+	}
+}
+
+// GroupHeadSizeLimit allows you to overwrite default head size limit - 10MB.
+func GroupHeadSizeLimit(limit int64) func(*Group) {
+	return func(g *Group) {
+		g.headSizeLimit = limit
+	}
+}
+
+// GroupTotalSizeLimit allows you to overwrite default total size limit of the group - 1GB.
+func GroupTotalSizeLimit(limit int64) func(*Group) {
+	return func(g *Group) {
+		g.totalSizeLimit = limit
+	}
+}
+
+// OnStart implements service.Service by starting the goroutine that checks file
+// and group limits.
+func (g *Group) OnStart() error {
+	g.ticker = time.NewTicker(g.groupCheckDuration)
+	go g.processTicks()
+	return nil
+}
+
+// OnStop implements service.Service by stopping the goroutine described above.
+// NOTE: g.Head must be closed separately using Close.
+func (g *Group) OnStop() {
+	g.ticker.Stop()
+	if err := g.FlushAndSync(); err != nil {
+		g.Logger.Error("Error flushin to disk", "err", err)
+	}
+}
+
+// Wait blocks until all internal goroutines are finished. Supposed to be
+// called after Stop.
+func (g *Group) Wait() {
+	// wait for processTicks routine to finish
+	<-g.doneProcessTicks
+}
+
+// Close closes the head file. The group must be stopped by this moment.
+func (g *Group) Close() {
+	if err := g.FlushAndSync(); err != nil {
+		g.Logger.Error("Error flushin to disk", "err", err)
+	}
+
+	g.mtx.Lock()
+	_ = g.Head.closeFile()
+	g.mtx.Unlock()
+}
+
+// HeadSizeLimit returns the current head size limit.
+func (g *Group) HeadSizeLimit() int64 {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.headSizeLimit
+}
+
+// TotalSizeLimit returns total size limit of the group.
+func (g *Group) TotalSizeLimit() int64 {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.totalSizeLimit
+}
+
+// MaxIndex returns index of the last file in the group.
+func (g *Group) MaxIndex() int {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.maxIndex
+}
+
+// MinIndex returns index of the first file in the group.
+func (g *Group) MinIndex() int {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.minIndex
+}
+
+// Write writes the contents of p into the current head of the group. It
+// returns the number of bytes written. If nn < len(p), it also returns an
+// error explaining why the write is short.
+// NOTE: Writes are buffered so they don't write synchronously
+// TODO: Make it halt if space is unavailable
+func (g *Group) Write(p []byte) (nn int, err error) {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.headBuf.Write(p)
+}
+
+// WriteLine writes line into the current head of the group. It also appends "\n".
+// NOTE: Writes are buffered so they don't write synchronously
+// TODO: Make it halt if space is unavailable
+func (g *Group) WriteLine(line string) error {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	_, err := g.headBuf.Write([]byte(line + "\n"))
+	return err
+}
+
+// Buffered returns the size of the currently buffered data.
+func (g *Group) Buffered() int {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.headBuf.Buffered()
+}
+
+// FlushAndSync writes any buffered data to the underlying file and commits the
+// current content of the file to stable storage (fsync).
+func (g *Group) FlushAndSync() error {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	err := g.headBuf.Flush()
+	if err == nil {
+		err = g.Head.Sync()
+	}
+	return err
+}
+
+func (g *Group) processTicks() {
+	defer close(g.doneProcessTicks)
+	for {
+		select {
+		case <-g.ticker.C:
+			g.checkHeadSizeLimit()
+			g.checkTotalSizeLimit()
+		case <-g.Quit():
+			return
+		}
+	}
+}
+
+// NOTE: this function is called manually in tests.
+func (g *Group) checkHeadSizeLimit() {
+	limit := g.HeadSizeLimit()
+	if limit == 0 {
+		return
+	}
+	size, err := g.Head.Size()
+	if err != nil {
+		g.Logger.Error("Group's head may grow without bound", "head", g.Head.Path, "err", err)
+		return
+	}
+	if size >= limit {
+		g.RotateFile()
+	}
+}
+
+func (g *Group) checkTotalSizeLimit() {
+	limit := g.TotalSizeLimit()
+	if limit == 0 {
+		return
+	}
+
+	gInfo := g.readGroupInfo()
+	totalSize := gInfo.TotalSize
+	for i := 0; i < maxFilesToRemove; i++ {
+		index := gInfo.MinIndex + i
+		if totalSize < limit {
+			return
+		}
+		if index == gInfo.MaxIndex {
+			// Special degenerate case, just do nothing.
+			g.Logger.Error("Group's head may grow without bound", "head", g.Head.Path)
+			return
+		}
+		pathToRemove := filePathForIndex(g.Head.Path, index, gInfo.MaxIndex)
+		fInfo, err := os.Stat(pathToRemove)
+		if err != nil {
+			g.Logger.Error("Failed to fetch info for file", "file", pathToRemove)
+			continue
+		}
+		err = os.Remove(pathToRemove)
+		if err != nil {
+			g.Logger.Error("Failed to remove path", "path", pathToRemove)
+			return
+		}
+		totalSize -= fInfo.Size()
+	}
+}
+
+// RotateFile causes group to close the current head and assign it some index.
+// Note it does not create a new head.
+func (g *Group) RotateFile() {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+
+	headPath := g.Head.Path
+
+	if err := g.headBuf.Flush(); err != nil {
+		panic(err)
+	}
+
+	if err := g.Head.Sync(); err != nil {
+		panic(err)
+	}
+
+	if err := g.Head.closeFile(); err != nil {
+		panic(err)
+	}
+
+	indexPath := filePathForIndex(headPath, g.maxIndex, g.maxIndex+1)
+	if err := os.Rename(headPath, indexPath); err != nil {
+		panic(err)
+	}
+
+	g.maxIndex++
+}
+
+// NewReader returns a new group reader.
+// CONTRACT: Caller must close the returned GroupReader.
+func (g *Group) NewReader(index int) (*GroupReader, error) {
+	r := newGroupReader(g)
+	err := r.SetIndex(index)
+	if err != nil {
+		return nil, err
+	}
+	return r, nil
+}
+
+// GroupInfo holds information about the group.
+type GroupInfo struct {
+	MinIndex  int   // index of the first file in the group, including head
+	MaxIndex  int   // index of the last file in the group, including head
+	TotalSize int64 // total size of the group
+	HeadSize  int64 // size of the head
+}
+
+// Returns info after scanning all files in g.Head's dir.
+func (g *Group) ReadGroupInfo() GroupInfo {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.readGroupInfo()
+}
+
+// Index includes the head.
+// CONTRACT: caller should have called g.mtx.Lock
+func (g *Group) readGroupInfo() GroupInfo {
+	groupDir := filepath.Dir(g.Head.Path)
+	headBase := filepath.Base(g.Head.Path)
+	var minIndex, maxIndex int = -1, -1
+	var totalSize, headSize int64 = 0, 0
+
+	dir, err := os.Open(groupDir)
+	if err != nil {
+		panic(err)
+	}
+	defer dir.Close()
+	fiz, err := dir.Readdir(0)
+	if err != nil {
+		panic(err)
+	}
+
+	// For each file in the directory, filter by pattern
+	for _, fileInfo := range fiz {
+		if fileInfo.Name() == headBase {
+			fileSize := fileInfo.Size()
+			totalSize += fileSize
+			headSize = fileSize
+			continue
+		} else if strings.HasPrefix(fileInfo.Name(), headBase) {
+			fileSize := fileInfo.Size()
+			totalSize += fileSize
+			indexedFilePattern := regexp.MustCompile(`^.+\.([0-9]{3,})$`)
+			submatch := indexedFilePattern.FindSubmatch([]byte(fileInfo.Name()))
+			if len(submatch) != 0 {
+				// Matches
+				fileIndex, err := strconv.Atoi(string(submatch[1]))
+				if err != nil {
+					panic(err)
+				}
+				if maxIndex < fileIndex {
+					maxIndex = fileIndex
+				}
+				if minIndex == -1 || fileIndex < minIndex {
+					minIndex = fileIndex
+				}
+			}
+		}
+	}
+
+	// Now account for the head.
+	if minIndex == -1 {
+		// If there were no numbered files,
+		// then the head is index 0.
+		minIndex, maxIndex = 0, 0
+	} else {
+		// Otherwise, the head file is 1 greater
+		maxIndex++
+	}
+	return GroupInfo{minIndex, maxIndex, totalSize, headSize}
+}
+
+func filePathForIndex(headPath string, index int, maxIndex int) string {
+	if index == maxIndex {
+		return headPath
+	}
+	return fmt.Sprintf("%v.%03d", headPath, index)
+}
+
+//--------------------------------------------------------------------------------
+
+// GroupReader provides an interface for reading from a Group.
+type GroupReader struct {
+	*Group
+	mtx       sync.Mutex
+	curIndex  int
+	curFile   *os.File
+	curReader *bufio.Reader
+	curLine   []byte
+}
+
+func newGroupReader(g *Group) *GroupReader {
+	return &GroupReader{
+		Group:     g,
+		curIndex:  0,
+		curFile:   nil,
+		curReader: nil,
+		curLine:   nil,
+	}
+}
+
+// Close closes the GroupReader by closing the cursor file.
+func (gr *GroupReader) Close() error {
+	gr.mtx.Lock()
+	defer gr.mtx.Unlock()
+
+	if gr.curReader != nil {
+		err := gr.curFile.Close()
+		gr.curIndex = 0
+		gr.curReader = nil
+		gr.curFile = nil
+		gr.curLine = nil
+		return err
+	}
+	return nil
+}
+
+// Read implements io.Reader, reading bytes from the current Reader
+// incrementing index until enough bytes are read.
+func (gr *GroupReader) Read(p []byte) (n int, err error) {
+	lenP := len(p)
+	if lenP == 0 {
+		return 0, errors.New("given empty slice")
+	}
+
+	gr.mtx.Lock()
+	defer gr.mtx.Unlock()
+
+	// Open file if not open yet
+	if gr.curReader == nil {
+		if err = gr.openFile(gr.curIndex); err != nil {
+			return 0, err
+		}
+	}
+
+	// Iterate over files until enough bytes are read
+	var nn int
+	for {
+		nn, err = gr.curReader.Read(p[n:])
+		n += nn
+		switch {
+		case err == io.EOF:
+			if n >= lenP {
+				return n, nil
+			}
+			// Open the next file
+			if err1 := gr.openFile(gr.curIndex + 1); err1 != nil {
+				return n, err1
+			}
+		case err != nil:
+			return n, err
+		case nn == 0: // empty file
+			return n, err
+		}
+	}
+}
+
+// IF index > gr.Group.maxIndex, returns io.EOF
+// CONTRACT: caller should hold gr.mtx
+func (gr *GroupReader) openFile(index int) error {
+	// Lock on Group to ensure that head doesn't move in the meanwhile.
+	gr.Group.mtx.Lock()
+	defer gr.Group.mtx.Unlock()
+
+	if index > gr.Group.maxIndex {
+		return io.EOF
+	}
+
+	curFilePath := filePathForIndex(gr.Head.Path, index, gr.Group.maxIndex)
+	curFile, err := os.OpenFile(curFilePath, os.O_RDONLY|os.O_CREATE, autoFilePerms)
+	if err != nil {
+		return err
+	}
+	curReader := bufio.NewReader(curFile)
+
+	// Update gr.cur*
+	if gr.curFile != nil {
+		gr.curFile.Close() // TODO return error?
+	}
+	gr.curIndex = index
+	gr.curFile = curFile
+	gr.curReader = curReader
+	gr.curLine = nil
+	return nil
+}
+
+// CurIndex returns cursor's file index.
+func (gr *GroupReader) CurIndex() int {
+	gr.mtx.Lock()
+	defer gr.mtx.Unlock()
+	return gr.curIndex
+}
+
+// SetIndex sets the cursor's file index to index by opening a file at this
+// position.
+func (gr *GroupReader) SetIndex(index int) error {
+	gr.mtx.Lock()
+	defer gr.mtx.Unlock()
+	return gr.openFile(index)
+}

internal/libs/autofile/group_test.go

@@ -0,0 +1,291 @@
+package autofile
+
+import (
+	"io"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	tmos "github.com/tendermint/tendermint/libs/os"
+	tmrand "github.com/tendermint/tendermint/libs/rand"
+)
+
+func createTestGroupWithHeadSizeLimit(t *testing.T, headSizeLimit int64) *Group {
+	testID := tmrand.Str(12)
+	testDir := "_test_" + testID
+	err := tmos.EnsureDir(testDir, 0700)
+	require.NoError(t, err, "Error creating dir")
+
+	headPath := testDir + "/myfile"
+	g, err := OpenGroup(headPath, GroupHeadSizeLimit(headSizeLimit))
+	require.NoError(t, err, "Error opening Group")
+	require.NotEqual(t, nil, g, "Failed to create Group")
+
+	return g
+}
+
+func destroyTestGroup(t *testing.T, g *Group) {
+	g.Close()
+
+	err := os.RemoveAll(g.Dir)
+	require.NoError(t, err, "Error removing test Group directory")
+}
+
+func assertGroupInfo(t *testing.T, gInfo GroupInfo, minIndex, maxIndex int, totalSize, headSize int64) {
+	assert.Equal(t, minIndex, gInfo.MinIndex)
+	assert.Equal(t, maxIndex, gInfo.MaxIndex)
+	assert.Equal(t, totalSize, gInfo.TotalSize)
+	assert.Equal(t, headSize, gInfo.HeadSize)
+}
+
+func TestCheckHeadSizeLimit(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 1000*1000)
+
+	// At first, there are no files.
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 0, 0)
+
+	// Write 1000 bytes 999 times.
+	for i := 0; i < 999; i++ {
+		err := g.WriteLine(tmrand.Str(999))
+		require.NoError(t, err, "Error appending to head")
+	}
+	err := g.FlushAndSync()
+	require.NoError(t, err)
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 999000, 999000)
+
+	// Even calling checkHeadSizeLimit manually won't rotate it.
+	g.checkHeadSizeLimit()
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 999000, 999000)
+
+	// Write 1000 more bytes.
+	err = g.WriteLine(tmrand.Str(999))
+	require.NoError(t, err, "Error appending to head")
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+
+	// Calling checkHeadSizeLimit this time rolls it.
+	g.checkHeadSizeLimit()
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 1000000, 0)
+
+	// Write 1000 more bytes.
+	err = g.WriteLine(tmrand.Str(999))
+	require.NoError(t, err, "Error appending to head")
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+
+	// Calling checkHeadSizeLimit does nothing.
+	g.checkHeadSizeLimit()
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 1001000, 1000)
+
+	// Write 1000 bytes 999 times.
+	for i := 0; i < 999; i++ {
+		err = g.WriteLine(tmrand.Str(999))
+		require.NoError(t, err, "Error appending to head")
+	}
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 2000000, 1000000)
+
+	// Calling checkHeadSizeLimit rolls it again.
+	g.checkHeadSizeLimit()
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2000000, 0)
+
+	// Write 1000 more bytes.
+	_, err = g.Head.Write([]byte(tmrand.Str(999) + "\n"))
+	require.NoError(t, err, "Error appending to head")
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2001000, 1000)
+
+	// Calling checkHeadSizeLimit does nothing.
+	g.checkHeadSizeLimit()
+	assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2001000, 1000)
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}
+
+func TestRotateFile(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 0)
+
+	// Create a different temporary directory and move into it, to make sure
+	// relative paths are resolved at Group creation
+	origDir, err := os.Getwd()
+	require.NoError(t, err)
+	defer func() {
+		if err := os.Chdir(origDir); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	dir, err := ioutil.TempDir("", "rotate_test")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+	err = os.Chdir(dir)
+	require.NoError(t, err)
+
+	require.True(t, filepath.IsAbs(g.Head.Path))
+	require.True(t, filepath.IsAbs(g.Dir))
+
+	// Create and rotate files
+	err = g.WriteLine("Line 1")
+	require.NoError(t, err)
+	err = g.WriteLine("Line 2")
+	require.NoError(t, err)
+	err = g.WriteLine("Line 3")
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+	g.RotateFile()
+	err = g.WriteLine("Line 4")
+	require.NoError(t, err)
+	err = g.WriteLine("Line 5")
+	require.NoError(t, err)
+	err = g.WriteLine("Line 6")
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+
+	// Read g.Head.Path+"000"
+	body1, err := ioutil.ReadFile(g.Head.Path + ".000")
+	assert.NoError(t, err, "Failed to read first rolled file")
+	if string(body1) != "Line 1\nLine 2\nLine 3\n" {
+		t.Errorf("got unexpected contents: [%v]", string(body1))
+	}
+
+	// Read g.Head.Path
+	body2, err := ioutil.ReadFile(g.Head.Path)
+	assert.NoError(t, err, "Failed to read first rolled file")
+	if string(body2) != "Line 4\nLine 5\nLine 6\n" {
+		t.Errorf("got unexpected contents: [%v]", string(body2))
+	}
+
+	// Make sure there are no files in the current, temporary directory
+	files, err := ioutil.ReadDir(".")
+	require.NoError(t, err)
+	assert.Empty(t, files)
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}
+
+func TestWrite(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 0)
+
+	written := []byte("Medusa")
+	_, err := g.Write(written)
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+
+	read := make([]byte, len(written))
+	gr, err := g.NewReader(0)
+	require.NoError(t, err, "failed to create reader")
+
+	_, err = gr.Read(read)
+	assert.NoError(t, err, "failed to read data")
+	assert.Equal(t, written, read)
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}
+
+// test that Read reads the required amount of bytes from all the files in the
+// group and returns no error if n == size of the given slice.
+func TestGroupReaderRead(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 0)
+
+	professor := []byte("Professor Monster")
+	_, err := g.Write(professor)
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+	g.RotateFile()
+	frankenstein := []byte("Frankenstein's Monster")
+	_, err = g.Write(frankenstein)
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+
+	totalWrittenLength := len(professor) + len(frankenstein)
+	read := make([]byte, totalWrittenLength)
+	gr, err := g.NewReader(0)
+	require.NoError(t, err, "failed to create reader")
+
+	n, err := gr.Read(read)
+	assert.NoError(t, err, "failed to read data")
+	assert.Equal(t, totalWrittenLength, n, "not enough bytes read")
+	professorPlusFrankenstein := professor
+	professorPlusFrankenstein = append(professorPlusFrankenstein, frankenstein...)
+	assert.Equal(t, professorPlusFrankenstein, read)
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}
+
+// test that Read returns an error if number of bytes read < size of
+// the given slice. Subsequent call should return 0, io.EOF.
+func TestGroupReaderRead2(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 0)
+
+	professor := []byte("Professor Monster")
+	_, err := g.Write(professor)
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+	g.RotateFile()
+	frankenstein := []byte("Frankenstein's Monster")
+	frankensteinPart := []byte("Frankenstein")
+	_, err = g.Write(frankensteinPart) // note writing only a part
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+
+	totalLength := len(professor) + len(frankenstein)
+	read := make([]byte, totalLength)
+	gr, err := g.NewReader(0)
+	require.NoError(t, err, "failed to create reader")
+
+	// 1) n < (size of the given slice), io.EOF
+	n, err := gr.Read(read)
+	assert.Equal(t, io.EOF, err)
+	assert.Equal(t, len(professor)+len(frankensteinPart), n, "Read more/less bytes than it is in the group")
+
+	// 2) 0, io.EOF
+	n, err = gr.Read([]byte("0"))
+	assert.Equal(t, io.EOF, err)
+	assert.Equal(t, 0, n)
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}
+
+func TestMinIndex(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 0)
+
+	assert.Zero(t, g.MinIndex(), "MinIndex should be zero at the beginning")
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}
+
+func TestMaxIndex(t *testing.T) {
+	g := createTestGroupWithHeadSizeLimit(t, 0)
+
+	assert.Zero(t, g.MaxIndex(), "MaxIndex should be zero at the beginning")
+
+	err := g.WriteLine("Line 1")
+	require.NoError(t, err)
+	err = g.FlushAndSync()
+	require.NoError(t, err)
+	g.RotateFile()
+
+	assert.Equal(t, 1, g.MaxIndex(), "MaxIndex should point to the last file")
+
+	// Cleanup
+	destroyTestGroup(t, g)
+}

internal/libs/clist/bench_test.go

@@ -0,0 +1,46 @@
+package clist
+
+import "testing"
+
+func BenchmarkDetaching(b *testing.B) {
+	lst := New()
+	for i := 0; i < b.N+1; i++ {
+		lst.PushBack(i)
+	}
+	start := lst.Front()
+	nxt := start.Next()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		start.removed = true
+		start.DetachNext()
+		start.DetachPrev()
+		tmp := nxt
+		nxt = nxt.Next()
+		start = tmp
+	}
+}
+
+// This is used to benchmark the time of RMutex.
+func BenchmarkRemoved(b *testing.B) {
+	lst := New()
+	for i := 0; i < b.N+1; i++ {
+		lst.PushBack(i)
+	}
+	start := lst.Front()
+	nxt := start.Next()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		start.Removed()
+		tmp := nxt
+		nxt = nxt.Next()
+		start = tmp
+	}
+}
+
+func BenchmarkPushBack(b *testing.B) {
+	lst := New()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		lst.PushBack(i)
+	}
+}

internal/libs/clist/clist.go

@@ -0,0 +1,404 @@
+package clist
+
+/*
+
+The purpose of CList is to provide a goroutine-safe linked-list.
+This list can be traversed concurrently by any number of goroutines.
+However, removed CElements cannot be added back.
+NOTE: Not all methods of container/list are (yet) implemented.
+NOTE: Removed elements need to DetachPrev or DetachNext consistently
+to ensure garbage collection of removed elements.
+
+*/
+
+import (
+	"fmt"
+	"sync"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+// MaxLength is the max allowed number of elements a linked list is
+// allowed to contain.
+// If more elements are pushed to the list it will panic.
+const MaxLength = int(^uint(0) >> 1)
+
+/*
+
+CElement is an element of a linked-list
+Traversal from a CElement is goroutine-safe.
+
+We can't avoid using WaitGroups or for-loops given the documentation
+spec without re-implementing the primitives that already exist in
+golang/sync. Notice that WaitGroup allows many go-routines to be
+simultaneously released, which is what we want. Mutex doesn't do
+this. RWMutex does this, but it's clumsy to use in the way that a
+WaitGroup would be used -- and we'd end up having two RWMutex's for
+prev/next each, which is doubly confusing.
+
+sync.Cond would be sort-of useful, but we don't need a write-lock in
+the for-loop. Use sync.Cond when you need serial access to the
+"condition". In our case our condition is if `next != nil || removed`,
+and there's no reason to serialize that condition for goroutines
+waiting on NextWait() (since it's just a read operation).
+
+*/
+type CElement struct {
+	mtx        tmsync.RWMutex
+	prev       *CElement
+	prevWg     *sync.WaitGroup
+	prevWaitCh chan struct{}
+	next       *CElement
+	nextWg     *sync.WaitGroup
+	nextWaitCh chan struct{}
+	removed    bool
+
+	Value interface{} // immutable
+}
+
+// Blocking implementation of Next().
+// May return nil iff CElement was tail and got removed.
+func (e *CElement) NextWait() *CElement {
+	for {
+		e.mtx.RLock()
+		next := e.next
+		nextWg := e.nextWg
+		removed := e.removed
+		e.mtx.RUnlock()
+
+		if next != nil || removed {
+			return next
+		}
+
+		nextWg.Wait()
+		// e.next doesn't necessarily exist here.
+		// That's why we need to continue a for-loop.
+	}
+}
+
+// Blocking implementation of Prev().
+// May return nil iff CElement was head and got removed.
+func (e *CElement) PrevWait() *CElement {
+	for {
+		e.mtx.RLock()
+		prev := e.prev
+		prevWg := e.prevWg
+		removed := e.removed
+		e.mtx.RUnlock()
+
+		if prev != nil || removed {
+			return prev
+		}
+
+		prevWg.Wait()
+	}
+}
+
+// PrevWaitChan can be used to wait until Prev becomes not nil. Once it does,
+// channel will be closed.
+func (e *CElement) PrevWaitChan() <-chan struct{} {
+	e.mtx.RLock()
+	defer e.mtx.RUnlock()
+
+	return e.prevWaitCh
+}
+
+// NextWaitChan can be used to wait until Next becomes not nil. Once it does,
+// channel will be closed.
+func (e *CElement) NextWaitChan() <-chan struct{} {
+	e.mtx.RLock()
+	defer e.mtx.RUnlock()
+
+	return e.nextWaitCh
+}
+
+// Nonblocking, may return nil if at the end.
+func (e *CElement) Next() *CElement {
+	e.mtx.RLock()
+	val := e.next
+	e.mtx.RUnlock()
+	return val
+}
+
+// Nonblocking, may return nil if at the end.
+func (e *CElement) Prev() *CElement {
+	e.mtx.RLock()
+	prev := e.prev
+	e.mtx.RUnlock()
+	return prev
+}
+
+func (e *CElement) Removed() bool {
+	e.mtx.RLock()
+	isRemoved := e.removed
+	e.mtx.RUnlock()
+	return isRemoved
+}
+
+func (e *CElement) DetachNext() {
+	e.mtx.Lock()
+	if !e.removed {
+		e.mtx.Unlock()
+		panic("DetachNext() must be called after Remove(e)")
+	}
+	e.next = nil
+	e.mtx.Unlock()
+}
+
+func (e *CElement) DetachPrev() {
+	e.mtx.Lock()
+	if !e.removed {
+		e.mtx.Unlock()
+		panic("DetachPrev() must be called after Remove(e)")
+	}
+	e.prev = nil
+	e.mtx.Unlock()
+}
+
+// NOTE: This function needs to be safe for
+// concurrent goroutines waiting on nextWg.
+func (e *CElement) SetNext(newNext *CElement) {
+	e.mtx.Lock()
+
+	oldNext := e.next
+	e.next = newNext
+	if oldNext != nil && newNext == nil {
+		// See https://golang.org/pkg/sync/:
+		//
+		// If a WaitGroup is reused to wait for several independent sets of
+		// events, new Add calls must happen after all previous Wait calls have
+		// returned.
+		e.nextWg = waitGroup1() // WaitGroups are difficult to re-use.
+		e.nextWaitCh = make(chan struct{})
+	}
+	if oldNext == nil && newNext != nil {
+		e.nextWg.Done()
+		close(e.nextWaitCh)
+	}
+	e.mtx.Unlock()
+}
+
+// NOTE: This function needs to be safe for
+// concurrent goroutines waiting on prevWg
+func (e *CElement) SetPrev(newPrev *CElement) {
+	e.mtx.Lock()
+
+	oldPrev := e.prev
+	e.prev = newPrev
+	if oldPrev != nil && newPrev == nil {
+		e.prevWg = waitGroup1() // WaitGroups are difficult to re-use.
+		e.prevWaitCh = make(chan struct{})
+	}
+	if oldPrev == nil && newPrev != nil {
+		e.prevWg.Done()
+		close(e.prevWaitCh)
+	}
+	e.mtx.Unlock()
+}
+
+func (e *CElement) SetRemoved() {
+	e.mtx.Lock()
+
+	e.removed = true
+
+	// This wakes up anyone waiting in either direction.
+	if e.prev == nil {
+		e.prevWg.Done()
+		close(e.prevWaitCh)
+	}
+	if e.next == nil {
+		e.nextWg.Done()
+		close(e.nextWaitCh)
+	}
+	e.mtx.Unlock()
+}
+
+//--------------------------------------------------------------------------------
+
+// CList represents a linked list.
+// The zero value for CList is an empty list ready to use.
+// Operations are goroutine-safe.
+// Panics if length grows beyond the max.
+type CList struct {
+	mtx    tmsync.RWMutex
+	wg     *sync.WaitGroup
+	waitCh chan struct{}
+	head   *CElement // first element
+	tail   *CElement // last element
+	len    int       // list length
+	maxLen int       // max list length
+}
+
+// Return CList with MaxLength. CList will panic if it goes beyond MaxLength.
+func New() *CList { return newWithMax(MaxLength) }
+
+// Return CList with given maxLength.
+// Will panic if list exceeds given maxLength.
+func newWithMax(maxLength int) *CList {
+	l := new(CList)
+	l.maxLen = maxLength
+
+	l.wg = waitGroup1()
+	l.waitCh = make(chan struct{})
+	l.head = nil
+	l.tail = nil
+	l.len = 0
+
+	return l
+}
+
+func (l *CList) Len() int {
+	l.mtx.RLock()
+	len := l.len
+	l.mtx.RUnlock()
+	return len
+}
+
+func (l *CList) Front() *CElement {
+	l.mtx.RLock()
+	head := l.head
+	l.mtx.RUnlock()
+	return head
+}
+
+func (l *CList) FrontWait() *CElement {
+	// Loop until the head is non-nil else wait and try again
+	for {
+		l.mtx.RLock()
+		head := l.head
+		wg := l.wg
+		l.mtx.RUnlock()
+
+		if head != nil {
+			return head
+		}
+		wg.Wait()
+		// NOTE: If you think l.head exists here, think harder.
+	}
+}
+
+func (l *CList) Back() *CElement {
+	l.mtx.RLock()
+	back := l.tail
+	l.mtx.RUnlock()
+	return back
+}
+
+func (l *CList) BackWait() *CElement {
+	for {
+		l.mtx.RLock()
+		tail := l.tail
+		wg := l.wg
+		l.mtx.RUnlock()
+
+		if tail != nil {
+			return tail
+		}
+		wg.Wait()
+		// l.tail doesn't necessarily exist here.
+		// That's why we need to continue a for-loop.
+	}
+}
+
+// WaitChan can be used to wait until Front or Back becomes not nil. Once it
+// does, channel will be closed.
+func (l *CList) WaitChan() <-chan struct{} {
+	l.mtx.Lock()
+	defer l.mtx.Unlock()
+
+	return l.waitCh
+}
+
+// Panics if list grows beyond its max length.
+func (l *CList) PushBack(v interface{}) *CElement {
+	l.mtx.Lock()
+
+	// Construct a new element
+	e := &CElement{
+		prev:       nil,
+		prevWg:     waitGroup1(),
+		prevWaitCh: make(chan struct{}),
+		next:       nil,
+		nextWg:     waitGroup1(),
+		nextWaitCh: make(chan struct{}),
+		removed:    false,
+		Value:      v,
+	}
+
+	// Release waiters on FrontWait/BackWait maybe
+	if l.len == 0 {
+		l.wg.Done()
+		close(l.waitCh)
+	}
+	if l.len >= l.maxLen {
+		panic(fmt.Sprintf("clist: maximum length list reached %d", l.maxLen))
+	}
+	l.len++
+
+	// Modify the tail
+	if l.tail == nil {
+		l.head = e
+		l.tail = e
+	} else {
+		e.SetPrev(l.tail) // We must init e first.
+		l.tail.SetNext(e) // This will make e accessible.
+		l.tail = e        // Update the list.
+	}
+	l.mtx.Unlock()
+	return e
+}
+
+// CONTRACT: Caller must call e.DetachPrev() and/or e.DetachNext() to avoid memory leaks.
+// NOTE: As per the contract of CList, removed elements cannot be added back.
+func (l *CList) Remove(e *CElement) interface{} {
+	l.mtx.Lock()
+
+	prev := e.Prev()
+	next := e.Next()
+
+	if l.head == nil || l.tail == nil {
+		l.mtx.Unlock()
+		panic("Remove(e) on empty CList")
+	}
+	if prev == nil && l.head != e {
+		l.mtx.Unlock()
+		panic("Remove(e) with false head")
+	}
+	if next == nil && l.tail != e {
+		l.mtx.Unlock()
+		panic("Remove(e) with false tail")
+	}
+
+	// If we're removing the only item, make CList FrontWait/BackWait wait.
+	if l.len == 1 {
+		l.wg = waitGroup1() // WaitGroups are difficult to re-use.
+		l.waitCh = make(chan struct{})
+	}
+
+	// Update l.len
+	l.len--
+
+	// Connect next/prev and set head/tail
+	if prev == nil {
+		l.head = next
+	} else {
+		prev.SetNext(next)
+	}
+	if next == nil {
+		l.tail = prev
+	} else {
+		next.SetPrev(prev)
+	}
+
+	// Set .Done() on e, otherwise waiters will wait forever.
+	e.SetRemoved()
+
+	l.mtx.Unlock()
+	return e.Value
+}
+
+func waitGroup1() (wg *sync.WaitGroup) {
+	wg = &sync.WaitGroup{}
+	wg.Add(1)
+	return
+}

internal/libs/clist/clist_test.go

@@ -0,0 +1,336 @@
+package clist
+
+import (
+	"fmt"
+	mrand "math/rand"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPanicOnMaxLength(t *testing.T) {
+	maxLength := 1000
+
+	l := newWithMax(maxLength)
+	for i := 0; i < maxLength; i++ {
+		l.PushBack(1)
+	}
+	assert.Panics(t, func() {
+		l.PushBack(1)
+	})
+}
+
+func TestSmall(t *testing.T) {
+	l := New()
+	el1 := l.PushBack(1)
+	el2 := l.PushBack(2)
+	el3 := l.PushBack(3)
+	if l.Len() != 3 {
+		t.Error("Expected len 3, got ", l.Len())
+	}
+
+	// fmt.Printf("%p %v\n", el1, el1)
+	// fmt.Printf("%p %v\n", el2, el2)
+	// fmt.Printf("%p %v\n", el3, el3)
+
+	r1 := l.Remove(el1)
+
+	// fmt.Printf("%p %v\n", el1, el1)
+	// fmt.Printf("%p %v\n", el2, el2)
+	// fmt.Printf("%p %v\n", el3, el3)
+
+	r2 := l.Remove(el2)
+
+	// fmt.Printf("%p %v\n", el1, el1)
+	// fmt.Printf("%p %v\n", el2, el2)
+	// fmt.Printf("%p %v\n", el3, el3)
+
+	r3 := l.Remove(el3)
+
+	if r1 != 1 {
+		t.Error("Expected 1, got ", r1)
+	}
+	if r2 != 2 {
+		t.Error("Expected 2, got ", r2)
+	}
+	if r3 != 3 {
+		t.Error("Expected 3, got ", r3)
+	}
+	if l.Len() != 0 {
+		t.Error("Expected len 0, got ", l.Len())
+	}
+
+}
+
+func TestGCFifo(t *testing.T) {
+
+	const numElements = 1000000
+	l := New()
+	gcCount := 0
+
+	// SetFinalizer doesn't work well with circular structures,
+	// so we construct a trivial non-circular structure to
+	// track.
+	type value struct {
+		Int int
+	}
+
+	gcCh := make(chan struct{})
+	for i := 0; i < numElements; i++ {
+		v := new(value)
+		v.Int = i
+		l.PushBack(v)
+		runtime.SetFinalizer(v, func(v *value) {
+			gcCh <- struct{}{}
+		})
+	}
+
+	for el := l.Front(); el != nil; {
+		l.Remove(el)
+		// oldEl := el
+		el = el.Next()
+		// oldEl.DetachPrev()
+		// oldEl.DetachNext()
+	}
+
+	tickerQuitCh := make(chan struct{})
+	tickerDoneCh := make(chan struct{})
+	go func() {
+		defer close(tickerDoneCh)
+		ticker := time.NewTicker(time.Second)
+		for {
+			select {
+			case <-ticker.C:
+				runtime.GC()
+			case <-tickerQuitCh:
+				return
+			}
+		}
+	}()
+
+	for i := 0; i < numElements; i++ {
+		<-gcCh
+		gcCount++
+	}
+
+	close(tickerQuitCh)
+	<-tickerDoneCh
+
+	if gcCount != numElements {
+		t.Errorf("expected gcCount to be %v, got %v", numElements,
+			gcCount)
+	}
+}
+
+func TestGCRandom(t *testing.T) {
+
+	const numElements = 1000000
+	l := New()
+	gcCount := 0
+
+	// SetFinalizer doesn't work well with circular structures,
+	// so we construct a trivial non-circular structure to
+	// track.
+	type value struct {
+		Int int
+	}
+
+	gcCh := make(chan struct{})
+	for i := 0; i < numElements; i++ {
+		v := new(value)
+		v.Int = i
+		l.PushBack(v)
+		runtime.SetFinalizer(v, func(v *value) {
+			gcCh <- struct{}{}
+		})
+	}
+
+	els := make([]*CElement, 0, numElements)
+	for el := l.Front(); el != nil; el = el.Next() {
+		els = append(els, el)
+	}
+
+	for _, i := range mrand.Perm(numElements) {
+		el := els[i]
+		l.Remove(el)
+		_ = el.Next()
+	}
+
+	tickerQuitCh := make(chan struct{})
+	tickerDoneCh := make(chan struct{})
+	go func() {
+		defer close(tickerDoneCh)
+		ticker := time.NewTicker(time.Second)
+		for {
+			select {
+			case <-ticker.C:
+				runtime.GC()
+			case <-tickerQuitCh:
+				return
+			}
+		}
+	}()
+
+	for i := 0; i < numElements; i++ {
+		<-gcCh
+		gcCount++
+	}
+
+	close(tickerQuitCh)
+	<-tickerDoneCh
+
+	if gcCount != numElements {
+		t.Errorf("expected gcCount to be %v, got %v", numElements,
+			gcCount)
+	}
+}
+
+func TestScanRightDeleteRandom(t *testing.T) {
+
+	const numElements = 1000
+	const numTimes = 100
+	const numScanners = 10
+
+	l := New()
+	stop := make(chan struct{})
+
+	els := make([]*CElement, numElements)
+	for i := 0; i < numElements; i++ {
+		el := l.PushBack(i)
+		els[i] = el
+	}
+
+	// Launch scanner routines that will rapidly iterate over elements.
+	for i := 0; i < numScanners; i++ {
+		go func(scannerID int) {
+			var el *CElement
+			restartCounter := 0
+			counter := 0
+		FOR_LOOP:
+			for {
+				select {
+				case <-stop:
+					fmt.Println("stopped")
+					break FOR_LOOP
+				default:
+				}
+				if el == nil {
+					el = l.FrontWait()
+					restartCounter++
+				}
+				el = el.Next()
+				counter++
+			}
+			fmt.Printf("Scanner %v restartCounter: %v counter: %v\n", scannerID, restartCounter, counter)
+		}(i)
+	}
+
+	// Remove an element, push back an element.
+	for i := 0; i < numTimes; i++ {
+		// Pick an element to remove
+		rmElIdx := mrand.Intn(len(els))
+		rmEl := els[rmElIdx]
+
+		// Remove it
+		l.Remove(rmEl)
+		// fmt.Print(".")
+
+		// Insert a new element
+		newEl := l.PushBack(-1*i - 1)
+		els[rmElIdx] = newEl
+
+		if i%100000 == 0 {
+			fmt.Printf("Pushed %vK elements so far...\n", i/1000)
+		}
+
+	}
+
+	// Stop scanners
+	close(stop)
+	// time.Sleep(time.Second * 1)
+
+	// And remove all the elements.
+	for el := l.Front(); el != nil; el = el.Next() {
+		l.Remove(el)
+	}
+	if l.Len() != 0 {
+		t.Fatal("Failed to remove all elements from CList")
+	}
+}
+
+func TestWaitChan(t *testing.T) {
+	l := New()
+	ch := l.WaitChan()
+
+	// 1) add one element to an empty list
+	go l.PushBack(1)
+	<-ch
+
+	// 2) and remove it
+	el := l.Front()
+	v := l.Remove(el)
+	if v != 1 {
+		t.Fatal("where is 1 coming from?")
+	}
+
+	// 3) test iterating forward and waiting for Next (NextWaitChan and Next)
+	el = l.PushBack(0)
+
+	done := make(chan struct{})
+	pushed := 0
+	go func() {
+		for i := 1; i < 100; i++ {
+			l.PushBack(i)
+			pushed++
+			time.Sleep(time.Duration(mrand.Intn(25)) * time.Millisecond)
+		}
+		// apply a deterministic pause so the counter has time to catch up
+		time.Sleep(25 * time.Millisecond)
+		close(done)
+	}()
+
+	next := el
+	seen := 0
+FOR_LOOP:
+	for {
+		select {
+		case <-next.NextWaitChan():
+			next = next.Next()
+			seen++
+			if next == nil {
+				t.Fatal("Next should not be nil when waiting on NextWaitChan")
+			}
+		case <-done:
+			break FOR_LOOP
+		case <-time.After(10 * time.Second):
+			t.Fatal("max execution time")
+		}
+	}
+
+	if pushed != seen {
+		t.Fatalf("number of pushed items (%d) not equal to number of seen items (%d)", pushed, seen)
+	}
+
+	// 4) test iterating backwards (PrevWaitChan and Prev)
+	prev := next
+	seen = 0
+FOR_LOOP2:
+	for {
+		select {
+		case <-prev.PrevWaitChan():
+			prev = prev.Prev()
+			seen++
+			if prev == nil {
+				t.Fatal("expected PrevWaitChan to block forever on nil when reached first elem")
+			}
+		case <-time.After(3 * time.Second):
+			break FOR_LOOP2
+		}
+	}
+
+	if pushed != seen {
+		t.Fatalf("number of pushed items (%d) not equal to number of seen items (%d)", pushed, seen)
+	}
+}

internal/libs/fail/fail.go

@@ -0,0 +1,40 @@
+package fail
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+)
+
+func envSet() int {
+	callIndexToFailS := os.Getenv("FAIL_TEST_INDEX")
+
+	if callIndexToFailS == "" {
+		return -1
+	}
+
+	var err error
+	callIndexToFail, err := strconv.Atoi(callIndexToFailS)
+	if err != nil {
+		return -1
+	}
+
+	return callIndexToFail
+}
+
+// Fail when FAIL_TEST_INDEX == callIndex
+var callIndex int // indexes Fail calls
+
+func Fail() {
+	callIndexToFail := envSet()
+	if callIndexToFail < 0 {
+		return
+	}
+
+	if callIndex == callIndexToFail {
+		fmt.Printf("*** fail-test %d ***\n", callIndex)
+		os.Exit(1)
+	}
+
+	callIndex++
+}

internal/libs/flowrate/README.md

@@ -0,0 +1,10 @@
+Data Flow Rate Control
+======================
+
+To download and install this package run:
+
+go get github.com/mxk/go-flowrate/flowrate
+
+The documentation is available at:
+
+<http://godoc.org/github.com/mxk/go-flowrate/flowrate>

internal/libs/flowrate/flowrate.go

@@ -0,0 +1,276 @@
+//
+// Written by Maxim Khitrov (November 2012)
+//
+
+// Package flowrate provides the tools for monitoring and limiting the flow rate
+// of an arbitrary data stream.
+package flowrate
+
+import (
+	"math"
+	"time"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+// Monitor monitors and limits the transfer rate of a data stream.
+type Monitor struct {
+	mu      tmsync.Mutex  // Mutex guarding access to all internal fields
+	active  bool          // Flag indicating an active transfer
+	start   time.Duration // Transfer start time (clock() value)
+	bytes   int64         // Total number of bytes transferred
+	samples int64         // Total number of samples taken
+
+	rSample float64 // Most recent transfer rate sample (bytes per second)
+	rEMA    float64 // Exponential moving average of rSample
+	rPeak   float64 // Peak transfer rate (max of all rSamples)
+	rWindow float64 // rEMA window (seconds)
+
+	sBytes int64         // Number of bytes transferred since sLast
+	sLast  time.Duration // Most recent sample time (stop time when inactive)
+	sRate  time.Duration // Sampling rate
+
+	tBytes int64         // Number of bytes expected in the current transfer
+	tLast  time.Duration // Time of the most recent transfer of at least 1 byte
+}
+
+// New creates a new flow control monitor. Instantaneous transfer rate is
+// measured and updated for each sampleRate interval. windowSize determines the
+// weight of each sample in the exponential moving average (EMA) calculation.
+// The exact formulas are:
+//
+// 	sampleTime = currentTime - prevSampleTime
+// 	sampleRate = byteCount / sampleTime
+// 	weight     = 1 - exp(-sampleTime/windowSize)
+// 	newRate    = weight*sampleRate + (1-weight)*oldRate
+//
+// The default values for sampleRate and windowSize (if <= 0) are 100ms and 1s,
+// respectively.
+func New(sampleRate, windowSize time.Duration) *Monitor {
+	if sampleRate = clockRound(sampleRate); sampleRate <= 0 {
+		sampleRate = 5 * clockRate
+	}
+	if windowSize <= 0 {
+		windowSize = 1 * time.Second
+	}
+	now := clock()
+	return &Monitor{
+		active:  true,
+		start:   now,
+		rWindow: windowSize.Seconds(),
+		sLast:   now,
+		sRate:   sampleRate,
+		tLast:   now,
+	}
+}
+
+// Update records the transfer of n bytes and returns n. It should be called
+// after each Read/Write operation, even if n is 0.
+func (m *Monitor) Update(n int) int {
+	m.mu.Lock()
+	m.update(n)
+	m.mu.Unlock()
+	return n
+}
+
+// Hack to set the current rEMA.
+func (m *Monitor) SetREMA(rEMA float64) {
+	m.mu.Lock()
+	m.rEMA = rEMA
+	m.samples++
+	m.mu.Unlock()
+}
+
+// IO is a convenience method intended to wrap io.Reader and io.Writer method
+// execution. It calls m.Update(n) and then returns (n, err) unmodified.
+func (m *Monitor) IO(n int, err error) (int, error) {
+	return m.Update(n), err
+}
+
+// Done marks the transfer as finished and prevents any further updates or
+// limiting. Instantaneous and current transfer rates drop to 0. Update, IO, and
+// Limit methods become NOOPs. It returns the total number of bytes transferred.
+func (m *Monitor) Done() int64 {
+	m.mu.Lock()
+	if now := m.update(0); m.sBytes > 0 {
+		m.reset(now)
+	}
+	m.active = false
+	m.tLast = 0
+	n := m.bytes
+	m.mu.Unlock()
+	return n
+}
+
+// timeRemLimit is the maximum Status.TimeRem value.
+const timeRemLimit = 999*time.Hour + 59*time.Minute + 59*time.Second
+
+// Status represents the current Monitor status. All transfer rates are in bytes
+// per second rounded to the nearest byte.
+type Status struct {
+	Start    time.Time     // Transfer start time
+	Bytes    int64         // Total number of bytes transferred
+	Samples  int64         // Total number of samples taken
+	InstRate int64         // Instantaneous transfer rate
+	CurRate  int64         // Current transfer rate (EMA of InstRate)
+	AvgRate  int64         // Average transfer rate (Bytes / Duration)
+	PeakRate int64         // Maximum instantaneous transfer rate
+	BytesRem int64         // Number of bytes remaining in the transfer
+	Duration time.Duration // Time period covered by the statistics
+	Idle     time.Duration // Time since the last transfer of at least 1 byte
+	TimeRem  time.Duration // Estimated time to completion
+	Progress Percent       // Overall transfer progress
+	Active   bool          // Flag indicating an active transfer
+}
+
+// Status returns current transfer status information. The returned value
+// becomes static after a call to Done.
+func (m *Monitor) Status() Status {
+	m.mu.Lock()
+	now := m.update(0)
+	s := Status{
+		Active:   m.active,
+		Start:    clockToTime(m.start),
+		Duration: m.sLast - m.start,
+		Idle:     now - m.tLast,
+		Bytes:    m.bytes,
+		Samples:  m.samples,
+		PeakRate: round(m.rPeak),
+		BytesRem: m.tBytes - m.bytes,
+		Progress: percentOf(float64(m.bytes), float64(m.tBytes)),
+	}
+	if s.BytesRem < 0 {
+		s.BytesRem = 0
+	}
+	if s.Duration > 0 {
+		rAvg := float64(s.Bytes) / s.Duration.Seconds()
+		s.AvgRate = round(rAvg)
+		if s.Active {
+			s.InstRate = round(m.rSample)
+			s.CurRate = round(m.rEMA)
+			if s.BytesRem > 0 {
+				if tRate := 0.8*m.rEMA + 0.2*rAvg; tRate > 0 {
+					ns := float64(s.BytesRem) / tRate * 1e9
+					if ns > float64(timeRemLimit) {
+						ns = float64(timeRemLimit)
+					}
+					s.TimeRem = clockRound(time.Duration(ns))
+				}
+			}
+		}
+	}
+	m.mu.Unlock()
+	return s
+}
+
+// Limit restricts the instantaneous (per-sample) data flow to rate bytes per
+// second. It returns the maximum number of bytes (0 <= n <= want) that may be
+// transferred immediately without exceeding the limit. If block == true, the
+// call blocks until n > 0. want is returned unmodified if want < 1, rate < 1,
+// or the transfer is inactive (after a call to Done).
+//
+// At least one byte is always allowed to be transferred in any given sampling
+// period. Thus, if the sampling rate is 100ms, the lowest achievable flow rate
+// is 10 bytes per second.
+//
+// For usage examples, see the implementation of Reader and Writer in io.go.
+func (m *Monitor) Limit(want int, rate int64, block bool) (n int) {
+	if want < 1 || rate < 1 {
+		return want
+	}
+	m.mu.Lock()
+
+	// Determine the maximum number of bytes that can be sent in one sample
+	limit := round(float64(rate) * m.sRate.Seconds())
+	if limit <= 0 {
+		limit = 1
+	}
+
+	// If block == true, wait until m.sBytes < limit
+	if now := m.update(0); block {
+		for m.sBytes >= limit && m.active {
+			now = m.waitNextSample(now)
+		}
+	}
+
+	// Make limit <= want (unlimited if the transfer is no longer active)
+	if limit -= m.sBytes; limit > int64(want) || !m.active {
+		limit = int64(want)
+	}
+	m.mu.Unlock()
+
+	if limit < 0 {
+		limit = 0
+	}
+	return int(limit)
+}
+
+// SetTransferSize specifies the total size of the data transfer, which allows
+// the Monitor to calculate the overall progress and time to completion.
+func (m *Monitor) SetTransferSize(bytes int64) {
+	if bytes < 0 {
+		bytes = 0
+	}
+	m.mu.Lock()
+	m.tBytes = bytes
+	m.mu.Unlock()
+}
+
+// update accumulates the transferred byte count for the current sample until
+// clock() - m.sLast >= m.sRate. The monitor status is updated once the current
+// sample is done.
+func (m *Monitor) update(n int) (now time.Duration) {
+	if !m.active {
+		return
+	}
+	if now = clock(); n > 0 {
+		m.tLast = now
+	}
+	m.sBytes += int64(n)
+	if sTime := now - m.sLast; sTime >= m.sRate {
+		t := sTime.Seconds()
+		if m.rSample = float64(m.sBytes) / t; m.rSample > m.rPeak {
+			m.rPeak = m.rSample
+		}
+
+		// Exponential moving average using a method similar to *nix load
+		// average calculation. Longer sampling periods carry greater weight.
+		if m.samples > 0 {
+			w := math.Exp(-t / m.rWindow)
+			m.rEMA = m.rSample + w*(m.rEMA-m.rSample)
+		} else {
+			m.rEMA = m.rSample
+		}
+		m.reset(now)
+	}
+	return
+}
+
+// reset clears the current sample state in preparation for the next sample.
+func (m *Monitor) reset(sampleTime time.Duration) {
+	m.bytes += m.sBytes
+	m.samples++
+	m.sBytes = 0
+	m.sLast = sampleTime
+}
+
+// waitNextSample sleeps for the remainder of the current sample. The lock is
+// released and reacquired during the actual sleep period, so it's possible for
+// the transfer to be inactive when this method returns.
+func (m *Monitor) waitNextSample(now time.Duration) time.Duration {
+	const minWait = 5 * time.Millisecond
+	current := m.sLast
+
+	// sleep until the last sample time changes (ideally, just one iteration)
+	for m.sLast == current && m.active {
+		d := current + m.sRate - now
+		m.mu.Unlock()
+		if d < minWait {
+			d = minWait
+		}
+		time.Sleep(d)
+		m.mu.Lock()
+		now = m.update(0)
+	}
+	return now
+}

internal/libs/flowrate/io.go

@@ -0,0 +1,133 @@
+//
+// Written by Maxim Khitrov (November 2012)
+//
+
+package flowrate
+
+import (
+	"errors"
+	"io"
+)
+
+// ErrLimit is returned by the Writer when a non-blocking write is short due to
+// the transfer rate limit.
+var ErrLimit = errors.New("flowrate: flow rate limit exceeded")
+
+// Limiter is implemented by the Reader and Writer to provide a consistent
+// interface for monitoring and controlling data transfer.
+type Limiter interface {
+	Done() int64
+	Status() Status
+	SetTransferSize(bytes int64)
+	SetLimit(new int64) (old int64)
+	SetBlocking(new bool) (old bool)
+}
+
+// Reader implements io.ReadCloser with a restriction on the rate of data
+// transfer.
+type Reader struct {
+	io.Reader // Data source
+	*Monitor  // Flow control monitor
+
+	limit int64 // Rate limit in bytes per second (unlimited when <= 0)
+	block bool  // What to do when no new bytes can be read due to the limit
+}
+
+// NewReader restricts all Read operations on r to limit bytes per second.
+func NewReader(r io.Reader, limit int64) *Reader {
+	return &Reader{r, New(0, 0), limit, true}
+}
+
+// Read reads up to len(p) bytes into p without exceeding the current transfer
+// rate limit. It returns (0, nil) immediately if r is non-blocking and no new
+// bytes can be read at this time.
+func (r *Reader) Read(p []byte) (n int, err error) {
+	p = p[:r.Limit(len(p), r.limit, r.block)]
+	if len(p) > 0 {
+		n, err = r.IO(r.Reader.Read(p))
+	}
+	return
+}
+
+// SetLimit changes the transfer rate limit to new bytes per second and returns
+// the previous setting.
+func (r *Reader) SetLimit(new int64) (old int64) {
+	old, r.limit = r.limit, new
+	return
+}
+
+// SetBlocking changes the blocking behavior and returns the previous setting. A
+// Read call on a non-blocking reader returns immediately if no additional bytes
+// may be read at this time due to the rate limit.
+func (r *Reader) SetBlocking(new bool) (old bool) {
+	old, r.block = r.block, new
+	return
+}
+
+// Close closes the underlying reader if it implements the io.Closer interface.
+func (r *Reader) Close() error {
+	defer r.Done()
+	if c, ok := r.Reader.(io.Closer); ok {
+		return c.Close()
+	}
+	return nil
+}
+
+// Writer implements io.WriteCloser with a restriction on the rate of data
+// transfer.
+type Writer struct {
+	io.Writer // Data destination
+	*Monitor  // Flow control monitor
+
+	limit int64 // Rate limit in bytes per second (unlimited when <= 0)
+	block bool  // What to do when no new bytes can be written due to the limit
+}
+
+// NewWriter restricts all Write operations on w to limit bytes per second. The
+// transfer rate and the default blocking behavior (true) can be changed
+// directly on the returned *Writer.
+func NewWriter(w io.Writer, limit int64) *Writer {
+	return &Writer{w, New(0, 0), limit, true}
+}
+
+// Write writes len(p) bytes from p to the underlying data stream without
+// exceeding the current transfer rate limit. It returns (n, ErrLimit) if w is
+// non-blocking and no additional bytes can be written at this time.
+func (w *Writer) Write(p []byte) (n int, err error) {
+	var c int
+	for len(p) > 0 && err == nil {
+		s := p[:w.Limit(len(p), w.limit, w.block)]
+		if len(s) > 0 {
+			c, err = w.IO(w.Writer.Write(s))
+		} else {
+			return n, ErrLimit
+		}
+		p = p[c:]
+		n += c
+	}
+	return
+}
+
+// SetLimit changes the transfer rate limit to new bytes per second and returns
+// the previous setting.
+func (w *Writer) SetLimit(new int64) (old int64) {
+	old, w.limit = w.limit, new
+	return
+}
+
+// SetBlocking changes the blocking behavior and returns the previous setting. A
+// Write call on a non-blocking writer returns as soon as no additional bytes
+// may be written at this time due to the rate limit.
+func (w *Writer) SetBlocking(new bool) (old bool) {
+	old, w.block = w.block, new
+	return
+}
+
+// Close closes the underlying writer if it implements the io.Closer interface.
+func (w *Writer) Close() error {
+	defer w.Done()
+	if c, ok := w.Writer.(io.Closer); ok {
+		return c.Close()
+	}
+	return nil
+}

internal/libs/flowrate/io_test.go

@@ -0,0 +1,197 @@
+//
+// Written by Maxim Khitrov (November 2012)
+//
+
+package flowrate
+
+import (
+	"bytes"
+	"testing"
+	"time"
+)
+
+const (
+	_50ms  = 50 * time.Millisecond
+	_100ms = 100 * time.Millisecond
+	_200ms = 200 * time.Millisecond
+	_300ms = 300 * time.Millisecond
+	_400ms = 400 * time.Millisecond
+	_500ms = 500 * time.Millisecond
+)
+
+func nextStatus(m *Monitor) Status {
+	samples := m.samples
+	for i := 0; i < 30; i++ {
+		if s := m.Status(); s.Samples != samples {
+			return s
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	return m.Status()
+}
+
+func TestReader(t *testing.T) {
+	in := make([]byte, 100)
+	for i := range in {
+		in[i] = byte(i)
+	}
+	b := make([]byte, 100)
+	r := NewReader(bytes.NewReader(in), 100)
+	start := time.Now()
+
+	// Make sure r implements Limiter
+	_ = Limiter(r)
+
+	// 1st read of 10 bytes is performed immediately
+	if n, err := r.Read(b); n != 10 || err != nil {
+		t.Fatalf("r.Read(b) expected 10 (<nil>); got %v (%v)", n, err)
+	} else if rt := time.Since(start); rt > _50ms {
+		t.Fatalf("r.Read(b) took too long (%v)", rt)
+	}
+
+	// No new Reads allowed in the current sample
+	r.SetBlocking(false)
+	if n, err := r.Read(b); n != 0 || err != nil {
+		t.Fatalf("r.Read(b) expected 0 (<nil>); got %v (%v)", n, err)
+	} else if rt := time.Since(start); rt > _50ms {
+		t.Fatalf("r.Read(b) took too long (%v)", rt)
+	}
+
+	status := [6]Status{0: r.Status()} // No samples in the first status
+
+	// 2nd read of 10 bytes blocks until the next sample
+	r.SetBlocking(true)
+	if n, err := r.Read(b[10:]); n != 10 || err != nil {
+		t.Fatalf("r.Read(b[10:]) expected 10 (<nil>); got %v (%v)", n, err)
+	} else if rt := time.Since(start); rt < _100ms {
+		t.Fatalf("r.Read(b[10:]) returned ahead of time (%v)", rt)
+	}
+
+	status[1] = r.Status()            // 1st sample
+	status[2] = nextStatus(r.Monitor) // 2nd sample
+	status[3] = nextStatus(r.Monitor) // No activity for the 3rd sample
+
+	if n := r.Done(); n != 20 {
+		t.Fatalf("r.Done() expected 20; got %v", n)
+	}
+
+	status[4] = r.Status()
+	status[5] = nextStatus(r.Monitor) // Timeout
+	start = status[0].Start
+
+	// Active, Bytes, Samples, InstRate, CurRate, AvgRate, PeakRate, BytesRem, Start, Duration, Idle, TimeRem, Progress
+	want := []Status{
+		{start, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, true},
+		{start, 10, 1, 100, 100, 100, 100, 0, _100ms, 0, 0, 0, true},
+		{start, 20, 2, 100, 100, 100, 100, 0, _200ms, _100ms, 0, 0, true},
+		{start, 20, 3, 0, 90, 67, 100, 0, _300ms, _200ms, 0, 0, true},
+		{start, 20, 3, 0, 0, 67, 100, 0, _300ms, 0, 0, 0, false},
+		{start, 20, 3, 0, 0, 67, 100, 0, _300ms, 0, 0, 0, false},
+	}
+	for i, s := range status {
+		s := s
+		if !statusesAreEqual(&s, &want[i]) {
+			t.Errorf("r.Status(%v)\nexpected: %v\ngot     : %v", i, want[i], s)
+		}
+	}
+	if !bytes.Equal(b[:20], in[:20]) {
+		t.Errorf("r.Read() input doesn't match output")
+	}
+}
+
+func TestWriter(t *testing.T) {
+	b := make([]byte, 100)
+	for i := range b {
+		b[i] = byte(i)
+	}
+	w := NewWriter(&bytes.Buffer{}, 200)
+	start := time.Now()
+
+	// Make sure w implements Limiter
+	_ = Limiter(w)
+
+	// Non-blocking 20-byte write for the first sample returns ErrLimit
+	w.SetBlocking(false)
+	if n, err := w.Write(b); n != 20 || err != ErrLimit {
+		t.Fatalf("w.Write(b) expected 20 (ErrLimit); got %v (%v)", n, err)
+	} else if rt := time.Since(start); rt > _50ms {
+		t.Fatalf("w.Write(b) took too long (%v)", rt)
+	}
+
+	// Blocking 80-byte write
+	w.SetBlocking(true)
+	if n, err := w.Write(b[20:]); n != 80 || err != nil {
+		t.Fatalf("w.Write(b[20:]) expected 80 (<nil>); got %v (%v)", n, err)
+	} else if rt := time.Since(start); rt < _300ms {
+		// Explanation for `rt < _300ms` (as opposed to `< _400ms`)
+		//
+		//                 |<-- start        |        |
+		// epochs: -----0ms|---100ms|---200ms|---300ms|---400ms
+		// sends:        20|20      |20      |20      |20#
+		//
+		// NOTE: The '#' symbol can thus happen before 400ms is up.
+		// Thus, we can only panic if rt < _300ms.
+		t.Fatalf("w.Write(b[20:]) returned ahead of time (%v)", rt)
+	}
+
+	w.SetTransferSize(100)
+	status := []Status{w.Status(), nextStatus(w.Monitor)}
+	start = status[0].Start
+
+	// Active, Bytes, Samples, InstRate, CurRate, AvgRate, PeakRate, BytesRem, Start, Duration, Idle, TimeRem, Progress
+	want := []Status{
+		{start, 80, 4, 200, 200, 200, 200, 20, _400ms, 0, _100ms, 80000, true},
+		{start, 100, 5, 200, 200, 200, 200, 0, _500ms, _100ms, 0, 100000, true},
+	}
+
+	for i, s := range status {
+		s := s
+		if !statusesAreEqual(&s, &want[i]) {
+			t.Errorf("w.Status(%v)\nexpected: %v\ngot     : %v\n", i, want[i], s)
+		}
+	}
+	if !bytes.Equal(b, w.Writer.(*bytes.Buffer).Bytes()) {
+		t.Errorf("w.Write() input doesn't match output")
+	}
+}
+
+const maxDeviationForDuration = 50 * time.Millisecond
+const maxDeviationForRate int64 = 50
+
+// statusesAreEqual returns true if s1 is equal to s2. Equality here means
+// general equality of fields except for the duration and rates, which can
+// drift due to unpredictable delays (e.g. thread wakes up 25ms after
+// `time.Sleep` has ended).
+func statusesAreEqual(s1 *Status, s2 *Status) bool {
+	if s1.Active == s2.Active &&
+		s1.Start == s2.Start &&
+		durationsAreEqual(s1.Duration, s2.Duration, maxDeviationForDuration) &&
+		s1.Idle == s2.Idle &&
+		s1.Bytes == s2.Bytes &&
+		s1.Samples == s2.Samples &&
+		ratesAreEqual(s1.InstRate, s2.InstRate, maxDeviationForRate) &&
+		ratesAreEqual(s1.CurRate, s2.CurRate, maxDeviationForRate) &&
+		ratesAreEqual(s1.AvgRate, s2.AvgRate, maxDeviationForRate) &&
+		ratesAreEqual(s1.PeakRate, s2.PeakRate, maxDeviationForRate) &&
+		s1.BytesRem == s2.BytesRem &&
+		durationsAreEqual(s1.TimeRem, s2.TimeRem, maxDeviationForDuration) &&
+		s1.Progress == s2.Progress {
+		return true
+	}
+	return false
+}
+
+func durationsAreEqual(d1 time.Duration, d2 time.Duration, maxDeviation time.Duration) bool {
+	return d2-d1 <= maxDeviation
+}
+
+func ratesAreEqual(r1 int64, r2 int64, maxDeviation int64) bool {
+	sub := r1 - r2
+	if sub < 0 {
+		sub = -sub
+	}
+	if sub <= maxDeviation {
+		return true
+	}
+	return false
+}

internal/libs/flowrate/util.go

@@ -0,0 +1,67 @@
+//
+// Written by Maxim Khitrov (November 2012)
+//
+
+package flowrate
+
+import (
+	"math"
+	"strconv"
+	"time"
+)
+
+// clockRate is the resolution and precision of clock().
+const clockRate = 20 * time.Millisecond
+
+// czero is the process start time rounded down to the nearest clockRate
+// increment.
+var czero = time.Now().Round(clockRate)
+
+// clock returns a low resolution timestamp relative to the process start time.
+func clock() time.Duration {
+	return time.Now().Round(clockRate).Sub(czero)
+}
+
+// clockToTime converts a clock() timestamp to an absolute time.Time value.
+func clockToTime(c time.Duration) time.Time {
+	return czero.Add(c)
+}
+
+// clockRound returns d rounded to the nearest clockRate increment.
+func clockRound(d time.Duration) time.Duration {
+	return (d + clockRate>>1) / clockRate * clockRate
+}
+
+// round returns x rounded to the nearest int64 (non-negative values only).
+func round(x float64) int64 {
+	if _, frac := math.Modf(x); frac >= 0.5 {
+		return int64(math.Ceil(x))
+	}
+	return int64(math.Floor(x))
+}
+
+// Percent represents a percentage in increments of 1/1000th of a percent.
+type Percent uint32
+
+// percentOf calculates what percent of the total is x.
+func percentOf(x, total float64) Percent {
+	if x < 0 || total <= 0 {
+		return 0
+	} else if p := round(x / total * 1e5); p <= math.MaxUint32 {
+		return Percent(p)
+	}
+	return Percent(math.MaxUint32)
+}
+
+func (p Percent) Float() float64 {
+	return float64(p) * 1e-3
+}
+
+func (p Percent) String() string {
+	var buf [12]byte
+	b := strconv.AppendUint(buf[:0], uint64(p)/1000, 10)
+	n := len(b)
+	b = strconv.AppendUint(b, 1000+uint64(p)%1000, 10)
+	b[n] = '.'
+	return string(append(b, '%'))
+}

internal/libs/protoio/io.go

@@ -0,0 +1,99 @@
+// Protocol Buffers for Go with Gadgets
+//
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
+// http://github.com/gogo/protobuf
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+// Modified to return number of bytes written by Writer.WriteMsg(), and added byteReader.
+
+package protoio
+
+import (
+	"io"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+type Writer interface {
+	WriteMsg(proto.Message) (int, error)
+}
+
+type WriteCloser interface {
+	Writer
+	io.Closer
+}
+
+type Reader interface {
+	ReadMsg(msg proto.Message) (int, error)
+}
+
+type ReadCloser interface {
+	Reader
+	io.Closer
+}
+
+type marshaler interface {
+	MarshalTo(data []byte) (n int, err error)
+}
+
+func getSize(v interface{}) (int, bool) {
+	if sz, ok := v.(interface {
+		Size() (n int)
+	}); ok {
+		return sz.Size(), true
+	} else if sz, ok := v.(interface {
+		ProtoSize() (n int)
+	}); ok {
+		return sz.ProtoSize(), true
+	} else {
+		return 0, false
+	}
+}
+
+// byteReader wraps an io.Reader and implements io.ByteReader, required by
+// binary.ReadUvarint(). Reading one byte at a time is extremely slow, but this
+// is what Amino did previously anyway, and the caller can wrap the underlying
+// reader in a bufio.Reader if appropriate.
+type byteReader struct {
+	reader    io.Reader
+	buf       []byte
+	bytesRead int // keeps track of bytes read via ReadByte()
+}
+
+func newByteReader(r io.Reader) *byteReader {
+	return &byteReader{
+		reader: r,
+		buf:    make([]byte, 1),
+	}
+}
+
+func (r *byteReader) ReadByte() (byte, error) {
+	n, err := r.reader.Read(r.buf)
+	r.bytesRead += n
+	if err != nil {
+		return 0x00, err
+	}
+	return r.buf[0], nil
+}

internal/libs/protoio/io_test.go

@@ -0,0 +1,184 @@
+// Protocol Buffers for Go with Gadgets
+//
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
+// http://github.com/gogo/protobuf
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+package protoio_test
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/gogo/protobuf/test"
+	"github.com/stretchr/testify/require"
+
+	"github.com/tendermint/tendermint/internal/libs/protoio"
+)
+
+func iotest(writer protoio.WriteCloser, reader protoio.ReadCloser) error {
+	varint := make([]byte, binary.MaxVarintLen64)
+	size := 1000
+	msgs := make([]*test.NinOptNative, size)
+	lens := make([]int, size)
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	for i := range msgs {
+		msgs[i] = test.NewPopulatedNinOptNative(r, true)
+		// issue 31
+		if i == 5 {
+			msgs[i] = &test.NinOptNative{}
+		}
+		// issue 31
+		if i == 999 {
+			msgs[i] = &test.NinOptNative{}
+		}
+		// FIXME Check size
+		bz, err := proto.Marshal(msgs[i])
+		if err != nil {
+			return err
+		}
+		visize := binary.PutUvarint(varint, uint64(len(bz)))
+		n, err := writer.WriteMsg(msgs[i])
+		if err != nil {
+			return err
+		}
+		if n != len(bz)+visize {
+			return fmt.Errorf("WriteMsg() wrote %v bytes, expected %v", n, len(bz)+visize) // nolint
+		}
+		lens[i] = n
+	}
+	if err := writer.Close(); err != nil {
+		return err
+	}
+	i := 0
+	for {
+		msg := &test.NinOptNative{}
+		if n, err := reader.ReadMsg(msg); err != nil {
+			if err == io.EOF {
+				break
+			}
+			return err
+		} else if n != lens[i] {
+			return fmt.Errorf("read %v bytes, expected %v", n, lens[i])
+		}
+		if err := msg.VerboseEqual(msgs[i]); err != nil {
+			return err
+		}
+		i++
+	}
+	if i != size {
+		panic("not enough messages read")
+	}
+	if err := reader.Close(); err != nil {
+		return err
+	}
+	return nil
+}
+
+type buffer struct {
+	*bytes.Buffer
+	closed bool
+}
+
+func (b *buffer) Close() error {
+	b.closed = true
+	return nil
+}
+
+func newBuffer() *buffer {
+	return &buffer{bytes.NewBuffer(nil), false}
+}
+
+func TestVarintNormal(t *testing.T) {
+	buf := newBuffer()
+	writer := protoio.NewDelimitedWriter(buf)
+	reader := protoio.NewDelimitedReader(buf, 1024*1024)
+	err := iotest(writer, reader)
+	require.NoError(t, err)
+	require.True(t, buf.closed, "did not close buffer")
+}
+
+func TestVarintNoClose(t *testing.T) {
+	buf := bytes.NewBuffer(nil)
+	writer := protoio.NewDelimitedWriter(buf)
+	reader := protoio.NewDelimitedReader(buf, 1024*1024)
+	err := iotest(writer, reader)
+	require.NoError(t, err)
+}
+
+// issue 32
+func TestVarintMaxSize(t *testing.T) {
+	buf := newBuffer()
+	writer := protoio.NewDelimitedWriter(buf)
+	reader := protoio.NewDelimitedReader(buf, 20)
+	err := iotest(writer, reader)
+	require.Error(t, err)
+}
+
+func TestVarintError(t *testing.T) {
+	buf := newBuffer()
+	buf.Write([]byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x7f})
+	reader := protoio.NewDelimitedReader(buf, 1024*1024)
+	msg := &test.NinOptNative{}
+	n, err := reader.ReadMsg(msg)
+	require.Error(t, err)
+	require.Equal(t, 10, n)
+}
+
+func TestVarintTruncated(t *testing.T) {
+	buf := newBuffer()
+	buf.Write([]byte{0xff, 0xff})
+	reader := protoio.NewDelimitedReader(buf, 1024*1024)
+	msg := &test.NinOptNative{}
+	n, err := reader.ReadMsg(msg)
+	require.Error(t, err)
+	require.Equal(t, 2, n)
+}
+
+func TestShort(t *testing.T) {
+	buf := newBuffer()
+
+	varintBuf := make([]byte, binary.MaxVarintLen64)
+	varintLen := binary.PutUvarint(varintBuf, 100)
+	_, err := buf.Write(varintBuf[:varintLen])
+	require.NoError(t, err)
+
+	bz, err := proto.Marshal(&test.NinOptNative{Field15: []byte{0x01, 0x02, 0x03}})
+	require.NoError(t, err)
+	buf.Write(bz)
+
+	reader := protoio.NewDelimitedReader(buf, 1024*1024)
+	require.NoError(t, err)
+	msg := &test.NinOptNative{}
+	n, err := reader.ReadMsg(msg)
+	require.Error(t, err)
+	require.Equal(t, varintLen+len(bz), n)
+}

internal/libs/protoio/reader.go

@@ -0,0 +1,106 @@
+// Protocol Buffers for Go with Gadgets
+//
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
+// http://github.com/gogo/protobuf
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+// Modified from original GoGo Protobuf to not buffer the reader.
+
+package protoio
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+// NewDelimitedReader reads varint-delimited Protobuf messages from a reader.
+// Unlike the gogoproto NewDelimitedReader, this does not buffer the reader,
+// which may cause poor performance but is necessary when only reading single
+// messages (e.g. in the p2p package). It also returns the number of bytes
+// read, which is necessary for the p2p package.
+func NewDelimitedReader(r io.Reader, maxSize int) ReadCloser {
+	var closer io.Closer
+	if c, ok := r.(io.Closer); ok {
+		closer = c
+	}
+	return &varintReader{r, nil, maxSize, closer}
+}
+
+type varintReader struct {
+	r       io.Reader
+	buf     []byte
+	maxSize int
+	closer  io.Closer
+}
+
+func (r *varintReader) ReadMsg(msg proto.Message) (int, error) {
+	// ReadUvarint needs an io.ByteReader, and we also need to keep track of the
+	// number of bytes read, so we use our own byteReader. This can't be
+	// buffered, so the caller should pass a buffered io.Reader to avoid poor
+	// performance.
+	byteReader := newByteReader(r.r)
+	l, err := binary.ReadUvarint(byteReader)
+	n := byteReader.bytesRead
+	if err != nil {
+		return n, err
+	}
+
+	// Make sure length doesn't overflow the native int size (e.g. 32-bit),
+	// and that the returned sum of n+length doesn't overflow either.
+	length := int(l)
+	if l >= uint64(^uint(0)>>1) || length < 0 || n+length < 0 {
+		return n, fmt.Errorf("invalid out-of-range message length %v", l)
+	}
+	if length > r.maxSize {
+		return n, fmt.Errorf("message exceeds max size (%v > %v)", length, r.maxSize)
+	}
+
+	if len(r.buf) < length {
+		r.buf = make([]byte, length)
+	}
+	buf := r.buf[:length]
+	nr, err := io.ReadFull(r.r, buf)
+	n += nr
+	if err != nil {
+		return n, err
+	}
+	return n, proto.Unmarshal(buf, msg)
+}
+
+func (r *varintReader) Close() error {
+	if r.closer != nil {
+		return r.closer.Close()
+	}
+	return nil
+}
+
+func UnmarshalDelimited(data []byte, msg proto.Message) error {
+	_, err := NewDelimitedReader(bytes.NewReader(data), len(data)).ReadMsg(msg)
+	return err
+}

internal/libs/protoio/writer.go

@@ -0,0 +1,100 @@
+// Protocol Buffers for Go with Gadgets
+//
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
+// http://github.com/gogo/protobuf
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+// Modified from original GoGo Protobuf to return number of bytes written.
+
+package protoio
+
+import (
+	"bytes"
+	"encoding/binary"
+	"io"
+
+	"github.com/gogo/protobuf/proto"
+)
+
+// NewDelimitedWriter writes a varint-delimited Protobuf message to a writer. It is
+// equivalent to the gogoproto NewDelimitedWriter, except WriteMsg() also returns the
+// number of bytes written, which is necessary in the p2p package.
+func NewDelimitedWriter(w io.Writer) WriteCloser {
+	return &varintWriter{w, make([]byte, binary.MaxVarintLen64), nil}
+}
+
+type varintWriter struct {
+	w      io.Writer
+	lenBuf []byte
+	buffer []byte
+}
+
+func (w *varintWriter) WriteMsg(msg proto.Message) (int, error) {
+	if m, ok := msg.(marshaler); ok {
+		n, ok := getSize(m)
+		if ok {
+			if n+binary.MaxVarintLen64 >= len(w.buffer) {
+				w.buffer = make([]byte, n+binary.MaxVarintLen64)
+			}
+			lenOff := binary.PutUvarint(w.buffer, uint64(n))
+			_, err := m.MarshalTo(w.buffer[lenOff:])
+			if err != nil {
+				return 0, err
+			}
+			_, err = w.w.Write(w.buffer[:lenOff+n])
+			return lenOff + n, err
+		}
+	}
+
+	// fallback
+	data, err := proto.Marshal(msg)
+	if err != nil {
+		return 0, err
+	}
+	length := uint64(len(data))
+	n := binary.PutUvarint(w.lenBuf, length)
+	_, err = w.w.Write(w.lenBuf[:n])
+	if err != nil {
+		return 0, err
+	}
+	_, err = w.w.Write(data)
+	return len(data) + n, err
+}
+
+func (w *varintWriter) Close() error {
+	if closer, ok := w.w.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
+}
+
+func MarshalDelimited(msg proto.Message) ([]byte, error) {
+	var buf bytes.Buffer
+	_, err := NewDelimitedWriter(&buf).WriteMsg(msg)
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}

internal/libs/sync/closer.go

@@ -0,0 +1,31 @@
+package sync
+
+import "sync"
+
+// Closer implements a primitive to close a channel that signals process
+// termination while allowing a caller to call Close multiple times safely. It
+// should be used in cases where guarantees cannot be made about when and how
+// many times closure is executed.
+type Closer struct {
+	closeOnce sync.Once
+	doneCh    chan struct{}
+}
+
+// NewCloser returns a reference to a new Closer.
+func NewCloser() *Closer {
+	return &Closer{doneCh: make(chan struct{})}
+}
+
+// Done returns the internal done channel allowing the caller either block or wait
+// for the Closer to be terminated/closed.
+func (c *Closer) Done() <-chan struct{} {
+	return c.doneCh
+}
+
+// Close gracefully closes the Closer. A caller should only call Close once, but
+// it is safe to call it successive times.
+func (c *Closer) Close() {
+	c.closeOnce.Do(func() {
+		close(c.doneCh)
+	})
+}

internal/libs/sync/closer_test.go

@@ -0,0 +1,28 @@
+package sync_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+func TestCloser(t *testing.T) {
+	closer := tmsync.NewCloser()
+
+	var timeout bool
+
+	select {
+	case <-closer.Done():
+	case <-time.After(time.Second):
+		timeout = true
+	}
+
+	for i := 0; i < 10; i++ {
+		closer.Close()
+	}
+
+	require.True(t, timeout)
+	<-closer.Done()
+}

internal/libs/sync/deadlock.go

@@ -0,0 +1,17 @@
+// +build deadlock
+
+package sync
+
+import (
+	deadlock "github.com/sasha-s/go-deadlock"
+)
+
+// A Mutex is a mutual exclusion lock.
+type Mutex struct {
+	deadlock.Mutex
+}
+
+// An RWMutex is a reader/writer mutual exclusion lock.
+type RWMutex struct {
+	deadlock.RWMutex
+}

internal/libs/sync/sync.go

@@ -0,0 +1,15 @@
+// +build !deadlock
+
+package sync
+
+import "sync"
+
+// A Mutex is a mutual exclusion lock.
+type Mutex struct {
+	sync.Mutex
+}
+
+// An RWMutex is a reader/writer mutual exclusion lock.
+type RWMutex struct {
+	sync.RWMutex
+}

internal/libs/sync/waker.go

@@ -0,0 +1,30 @@
+package sync
+
+// Waker is used to wake up a sleeper when some event occurs. It debounces
+// multiple wakeup calls occurring between each sleep, and wakeups are
+// non-blocking to avoid having to coordinate goroutines.
+type Waker struct {
+	wakeCh chan struct{}
+}
+
+// NewWaker creates a new Waker.
+func NewWaker() *Waker {
+	return &Waker{
+		wakeCh: make(chan struct{}, 1), // buffer used for debouncing
+	}
+}
+
+// Sleep returns a channel that blocks until Wake() is called.
+func (w *Waker) Sleep() <-chan struct{} {
+	return w.wakeCh
+}
+
+// Wake wakes up the sleeper.
+func (w *Waker) Wake() {
+	// A non-blocking send with a size 1 buffer ensures that we never block, and
+	// that we queue up at most a single wakeup call between each Sleep().
+	select {
+	case w.wakeCh <- struct{}{}:
+	default:
+	}
+}

internal/libs/sync/waker_test.go

@@ -0,0 +1,47 @@
+package sync_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+func TestWaker(t *testing.T) {
+
+	// A new waker should block when sleeping.
+	waker := tmsync.NewWaker()
+
+	select {
+	case <-waker.Sleep():
+		require.Fail(t, "unexpected wakeup")
+	default:
+	}
+
+	// Wakeups should not block, and should cause the next sleeper to awaken.
+	waker.Wake()
+
+	select {
+	case <-waker.Sleep():
+	default:
+		require.Fail(t, "expected wakeup, but sleeping instead")
+	}
+
+	// Multiple wakeups should only wake a single sleeper.
+	waker.Wake()
+	waker.Wake()
+	waker.Wake()
+
+	select {
+	case <-waker.Sleep():
+	default:
+		require.Fail(t, "expected wakeup, but sleeping instead")
+	}
+
+	select {
+	case <-waker.Sleep():
+		require.Fail(t, "unexpected wakeup")
+	default:
+	}
+}

internal/libs/tempfile/tempfile.go

@@ -0,0 +1,129 @@
+package tempfile
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+const (
+	atomicWriteFilePrefix = "write-file-atomic-"
+	// Maximum number of atomic write file conflicts before we start reseeding
+	// (reduced from golang's default 10 due to using an increased randomness space)
+	atomicWriteFileMaxNumConflicts = 5
+	// Maximum number of attempts to make at writing the write file before giving up
+	// (reduced from golang's default 10000 due to using an increased randomness space)
+	atomicWriteFileMaxNumWriteAttempts = 1000
+	// LCG constants from Donald Knuth MMIX
+	// This LCG's has a period equal to 2**64
+	lcgA = 6364136223846793005
+	lcgC = 1442695040888963407
+	// Create in case it doesn't exist and force kernel
+	// flush, which still leaves the potential of lingering disk cache.
+	// Never overwrites files
+	atomicWriteFileFlag = os.O_WRONLY | os.O_CREATE | os.O_SYNC | os.O_TRUNC | os.O_EXCL
+)
+
+var (
+	atomicWriteFileRand   uint64
+	atomicWriteFileRandMu tmsync.Mutex
+)
+
+func writeFileRandReseed() uint64 {
+	// Scale the PID, to minimize the chance that two processes seeded at similar times
+	// don't get the same seed. Note that PID typically ranges in [0, 2**15), but can be
+	// up to 2**22 under certain configurations. We left bit-shift the PID by 20, so that
+	// a PID difference of one corresponds to a time difference of 2048 seconds.
+	// The important thing here is that now for a seed conflict, they would both have to be on
+	// the correct nanosecond offset, and second-based offset, which is much less likely than
+	// just a conflict with the correct nanosecond offset.
+	return uint64(time.Now().UnixNano() + int64(os.Getpid()<<20))
+}
+
+// Use a fast thread safe LCG for atomic write file names.
+// Returns a string corresponding to a 64 bit int.
+// If it was a negative int, the leading number is a 0.
+func randWriteFileSuffix() string {
+	atomicWriteFileRandMu.Lock()
+	r := atomicWriteFileRand
+	if r == 0 {
+		r = writeFileRandReseed()
+	}
+
+	// Update randomness according to lcg
+	r = r*lcgA + lcgC
+
+	atomicWriteFileRand = r
+	atomicWriteFileRandMu.Unlock()
+	// Can have a negative name, replace this in the following
+	suffix := strconv.Itoa(int(r))
+	if string(suffix[0]) == "-" {
+		// Replace first "-" with "0". This is purely for UI clarity,
+		// as otherwhise there would be two `-` in a row.
+		suffix = strings.Replace(suffix, "-", "0", 1)
+	}
+	return suffix
+}
+
+// WriteFileAtomic creates a temporary file with data and provided perm and
+// swaps it atomically with filename if successful.
+func WriteFileAtomic(filename string, data []byte, perm os.FileMode) (err error) {
+	// This implementation is inspired by the golang stdlibs method of creating
+	// tempfiles. Notable differences are that we use different flags, a 64 bit LCG
+	// and handle negatives differently.
+	// The core reason we can't use golang's TempFile is that we must write
+	// to the file synchronously, as we need this to persist to disk.
+	// We also open it in write-only mode, to avoid concerns that arise with read.
+	var (
+		dir = filepath.Dir(filename)
+		f   *os.File
+	)
+
+	nconflict := 0
+	// Limit the number of attempts to create a file. Something is seriously
+	// wrong if it didn't get created after 1000 attempts, and we don't want
+	// an infinite loop
+	i := 0
+	for ; i < atomicWriteFileMaxNumWriteAttempts; i++ {
+		name := filepath.Join(dir, atomicWriteFilePrefix+randWriteFileSuffix())
+		f, err = os.OpenFile(name, atomicWriteFileFlag, perm)
+		// If the file already exists, try a new file
+		if os.IsExist(err) {
+			// If the files exists too many times, start reseeding as we've
+			// likely hit another instances seed.
+			if nconflict++; nconflict > atomicWriteFileMaxNumConflicts {
+				atomicWriteFileRandMu.Lock()
+				atomicWriteFileRand = writeFileRandReseed()
+				atomicWriteFileRandMu.Unlock()
+			}
+			continue
+		} else if err != nil {
+			return err
+		}
+		break
+	}
+	if i == atomicWriteFileMaxNumWriteAttempts {
+		return fmt.Errorf("could not create atomic write file after %d attempts", i)
+	}
+
+	// Clean up in any case. Defer stacking order is last-in-first-out.
+	defer os.Remove(f.Name())
+	defer f.Close()
+
+	if n, err := f.Write(data); err != nil {
+		return err
+	} else if n < len(data) {
+		return io.ErrShortWrite
+	}
+	// Close the file before renaming it, otherwise it will cause "The process
+	// cannot access the file because it is being used by another process." on windows.
+	f.Close()
+
+	return os.Rename(f.Name(), filename)
+}

internal/libs/tempfile/tempfile_test.go

@@ -0,0 +1,145 @@
+package tempfile
+
+// Need access to internal variables, so can't use _test package
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	mrand "math/rand"
+	"os"
+	testing "testing"
+
+	"github.com/stretchr/testify/require"
+
+	tmrand "github.com/tendermint/tendermint/libs/rand"
+)
+
+func TestWriteFileAtomic(t *testing.T) {
+	var (
+		data             = []byte(tmrand.Str(mrand.Intn(2048)))
+		old              = tmrand.Bytes(mrand.Intn(2048))
+		perm os.FileMode = 0600
+	)
+
+	f, err := ioutil.TempFile("/tmp", "write-atomic-test-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(f.Name())
+
+	if err = ioutil.WriteFile(f.Name(), old, 0600); err != nil {
+		t.Fatal(err)
+	}
+
+	if err = WriteFileAtomic(f.Name(), data, perm); err != nil {
+		t.Fatal(err)
+	}
+
+	rData, err := ioutil.ReadFile(f.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !bytes.Equal(data, rData) {
+		t.Fatalf("data mismatch: %v != %v", data, rData)
+	}
+
+	stat, err := os.Stat(f.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if have, want := stat.Mode().Perm(), perm; have != want {
+		t.Errorf("have %v, want %v", have, want)
+	}
+}
+
+// This tests atomic write file when there is a single duplicate file.
+// Expected behavior is for a new file to be created, and the original write file to be unaltered.
+func TestWriteFileAtomicDuplicateFile(t *testing.T) {
+	var (
+		defaultSeed    uint64 = 1
+		testString            = "This is a glorious test string"
+		expectedString        = "Did the test file's string appear here?"
+
+		fileToWrite = "/tmp/TestWriteFileAtomicDuplicateFile-test.txt"
+	)
+	// Create a file at the seed, and reset the seed.
+	atomicWriteFileRand = defaultSeed
+	firstFileRand := randWriteFileSuffix()
+	atomicWriteFileRand = defaultSeed
+	fname := "/tmp/" + atomicWriteFilePrefix + firstFileRand
+	f, err := os.OpenFile(fname, atomicWriteFileFlag, 0777)
+	defer os.Remove(fname)
+	// Defer here, in case there is a panic in WriteFileAtomic.
+	defer os.Remove(fileToWrite)
+
+	require.NoError(t, err)
+	_, err = f.WriteString(testString)
+	require.NoError(t, err)
+	err = WriteFileAtomic(fileToWrite, []byte(expectedString), 0777)
+	require.NoError(t, err)
+	// Check that the first atomic file was untouched
+	firstAtomicFileBytes, err := ioutil.ReadFile(fname)
+	require.NoError(t, err, "Error reading first atomic file")
+	require.Equal(t, []byte(testString), firstAtomicFileBytes, "First atomic file was overwritten")
+	// Check that the resultant file is correct
+	resultantFileBytes, err := ioutil.ReadFile(fileToWrite)
+	require.NoError(t, err, "Error reading resultant file")
+	require.Equal(t, []byte(expectedString), resultantFileBytes, "Written file had incorrect bytes")
+
+	// Check that the intermediate write file was deleted
+	// Get the second write files' randomness
+	atomicWriteFileRand = defaultSeed
+	_ = randWriteFileSuffix()
+	secondFileRand := randWriteFileSuffix()
+	_, err = os.Stat("/tmp/" + atomicWriteFilePrefix + secondFileRand)
+	require.True(t, os.IsNotExist(err), "Intermittent atomic write file not deleted")
+}
+
+// This tests atomic write file when there are many duplicate files.
+// Expected behavior is for a new file to be created under a completely new seed,
+// and the original write files to be unaltered.
+func TestWriteFileAtomicManyDuplicates(t *testing.T) {
+	var (
+		defaultSeed    uint64 = 2
+		testString            = "This is a glorious test string, from file %d"
+		expectedString        = "Did any of the test file's string appear here?"
+
+		fileToWrite = "/tmp/TestWriteFileAtomicDuplicateFile-test.txt"
+	)
+	// Initialize all of the atomic write files
+	atomicWriteFileRand = defaultSeed
+	for i := 0; i < atomicWriteFileMaxNumConflicts+2; i++ {
+		fileRand := randWriteFileSuffix()
+		fname := "/tmp/" + atomicWriteFilePrefix + fileRand
+		f, err := os.OpenFile(fname, atomicWriteFileFlag, 0777)
+		require.Nil(t, err)
+		_, err = f.WriteString(fmt.Sprintf(testString, i))
+		require.NoError(t, err)
+		defer os.Remove(fname)
+	}
+
+	atomicWriteFileRand = defaultSeed
+	// Defer here, in case there is a panic in WriteFileAtomic.
+	defer os.Remove(fileToWrite)
+
+	err := WriteFileAtomic(fileToWrite, []byte(expectedString), 0777)
+	require.NoError(t, err)
+	// Check that all intermittent atomic file were untouched
+	atomicWriteFileRand = defaultSeed
+	for i := 0; i < atomicWriteFileMaxNumConflicts+2; i++ {
+		fileRand := randWriteFileSuffix()
+		fname := "/tmp/" + atomicWriteFilePrefix + fileRand
+		firstAtomicFileBytes, err := ioutil.ReadFile(fname)
+		require.Nil(t, err, "Error reading first atomic file")
+		require.Equal(t, []byte(fmt.Sprintf(testString, i)), firstAtomicFileBytes,
+			"atomic write file %d was overwritten", i)
+	}
+
+	// Check that the resultant file is correct
+	resultantFileBytes, err := ioutil.ReadFile(fileToWrite)
+	require.Nil(t, err, "Error reading resultant file")
+	require.Equal(t, []byte(expectedString), resultantFileBytes, "Written file had incorrect bytes")
+}

internal/libs/test/mutate.go

@@ -0,0 +1,29 @@
+// nolint:gosec // G404: Use of weak random number generator
+package test
+
+import (
+	mrand "math/rand"
+)
+
+// Contract: !bytes.Equal(input, output) && len(input) >= len(output)
+func MutateByteSlice(bytez []byte) []byte {
+	// If bytez is empty, panic
+	if len(bytez) == 0 {
+		panic("Cannot mutate an empty bytez")
+	}
+
+	// Copy bytez
+	mBytez := make([]byte, len(bytez))
+	copy(mBytez, bytez)
+	bytez = mBytez
+
+	// Try a random mutation
+	switch mrand.Int() % 2 {
+	case 0: // Mutate a single byte
+		bytez[mrand.Int()%len(bytez)] += byte(mrand.Int()%255 + 1)
+	case 1: // Remove an arbitrary byte
+		pos := mrand.Int() % len(bytez)
+		bytez = append(bytez[:pos], bytez[pos+1:]...)
+	}
+	return bytez
+}

internal/libs/timer/throttle_timer.go

@@ -0,0 +1,76 @@
+package timer
+
+import (
+	"time"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+/*
+ThrottleTimer fires an event at most "dur" after each .Set() call.
+If a short burst of .Set() calls happens, ThrottleTimer fires once.
+If a long continuous burst of .Set() calls happens, ThrottleTimer fires
+at most once every "dur".
+*/
+type ThrottleTimer struct {
+	Name string
+	Ch   chan struct{}
+	quit chan struct{}
+	dur  time.Duration
+
+	mtx   tmsync.Mutex
+	timer *time.Timer
+	isSet bool
+}
+
+func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer {
+	var ch = make(chan struct{})
+	var quit = make(chan struct{})
+	var t = &ThrottleTimer{Name: name, Ch: ch, dur: dur, quit: quit}
+	t.mtx.Lock()
+	t.timer = time.AfterFunc(dur, t.fireRoutine)
+	t.mtx.Unlock()
+	t.timer.Stop()
+	return t
+}
+
+func (t *ThrottleTimer) fireRoutine() {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+	select {
+	case t.Ch <- struct{}{}:
+		t.isSet = false
+	case <-t.quit:
+		// do nothing
+	default:
+		t.timer.Reset(t.dur)
+	}
+}
+
+func (t *ThrottleTimer) Set() {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+	if !t.isSet {
+		t.isSet = true
+		t.timer.Reset(t.dur)
+	}
+}
+
+func (t *ThrottleTimer) Unset() {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+	t.isSet = false
+	t.timer.Stop()
+}
+
+// For ease of .Stop()'ing services before .Start()'ing them,
+// we ignore .Stop()'s on nil ThrottleTimers
+func (t *ThrottleTimer) Stop() bool {
+	if t == nil {
+		return false
+	}
+	close(t.quit)
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+	return t.timer.Stop()
+}

internal/libs/timer/throttle_timer_test.go

@@ -0,0 +1,82 @@
+package timer
+
+import (
+	"testing"
+	"time"
+
+	// make govet noshadow happy...
+
+	asrt "github.com/stretchr/testify/assert"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+)
+
+type thCounter struct {
+	input chan struct{}
+	mtx   tmsync.Mutex
+	count int
+}
+
+func (c *thCounter) Increment() {
+	c.mtx.Lock()
+	c.count++
+	c.mtx.Unlock()
+}
+
+func (c *thCounter) Count() int {
+	c.mtx.Lock()
+	val := c.count
+	c.mtx.Unlock()
+	return val
+}
+
+// Read should run in a go-routine and
+// updates count by one every time a packet comes in
+func (c *thCounter) Read() {
+	for range c.input {
+		c.Increment()
+	}
+}
+
+func TestThrottle(test *testing.T) {
+	assert := asrt.New(test)
+
+	ms := 50
+	delay := time.Duration(ms) * time.Millisecond
+	longwait := time.Duration(2) * delay
+	t := NewThrottleTimer("foo", delay)
+
+	// start at 0
+	c := &thCounter{input: t.Ch}
+	assert.Equal(0, c.Count())
+	go c.Read()
+
+	// waiting does nothing
+	time.Sleep(longwait)
+	assert.Equal(0, c.Count())
+
+	// send one event adds one
+	t.Set()
+	time.Sleep(longwait)
+	assert.Equal(1, c.Count())
+
+	// send a burst adds one
+	for i := 0; i < 5; i++ {
+		t.Set()
+	}
+	time.Sleep(longwait)
+	assert.Equal(2, c.Count())
+
+	// send 12, over 2 delay sections, adds 3 or more. It
+	// is possible for more to be added if the overhead
+	// in executing the loop is large
+	short := time.Duration(ms/5) * time.Millisecond
+	for i := 0; i < 13; i++ {
+		t.Set()
+		time.Sleep(short)
+	}
+	time.Sleep(longwait)
+	assert.LessOrEqual(5, c.Count())
+
+	close(t.Ch)
+}