From 4e7c72cc00aaa70d077f0df0744c49ed66c10882 Mon Sep 17 00:00:00 2001
From: kcrodgers24 <kcrodgers@live.com>
Date: Fri, 11 Feb 2022 11:47:49 -0800
Subject: [PATCH 1/5] make etcd schema primary source of truth for indexes and
 fields

---
 holder.go | 25 ++++-----------------
 index.go  | 66 ++++++++++++++++++++-----------------------------------
 2 files changed, 28 insertions(+), 63 deletions(-)

diff --git a/holder.go b/holder.go
index 18cd7154c..db9aa0cd2 100644
--- a/holder.go
+++ b/holder.go
@@ -9,7 +9,6 @@ import (
 	"path/filepath"
 	"runtime"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -332,23 +331,7 @@ func (h *Holder) Open() error {
 	}
 	defer f.Close()
 
-	fis, err := f.Readdir(0)
-	if err != nil {
-		return errors.Wrap(err, "reading directory")
-	}
-
-	for _, fi := range fis {
-		// Skip files or hidden directories.
-		if !fi.IsDir() || strings.HasPrefix(fi.Name(), ".") {
-			continue
-		}
-
-		// Only continue with indexes which are present in schema.
-		idx, ok := schema[fi.Name()]
-		if !ok {
-			continue
-		}
-
+	for idxKey, idx := range schema {
 		// decode the CreateIndexMessage from the schema data in order to
 		// get its metadata, such as CreateAt.
 		cim, err := decodeCreateIndexMessage(h.serializer, idx.Data)
@@ -356,11 +339,11 @@ func (h *Holder) Open() error {
 			return errors.Wrap(err, "decoding create index message")
 		}
 
-		h.Logger.Printf("opening index: %s", filepath.Base(fi.Name()))
+		h.Logger.Printf("opening index: %s", idxKey)
 
-		index, err := h.newIndex(h.IndexPath(filepath.Base(fi.Name())), filepath.Base(fi.Name()))
+		index, err := h.newIndex(h.IndexPath(idxKey), idxKey)
 		if errors.Cause(err) == ErrName {
-			h.Logger.Errorf("opening index: %s, err=%s", fi.Name(), err)
+			h.Logger.Errorf("opening index: %s, err=%s", idxKey, err)
 			continue
 		} else if err != nil {
 			return errors.Wrap(err, "opening index")
diff --git a/index.go b/index.go
index ba3901d67..11c924c08 100644
--- a/index.go
+++ b/index.go
@@ -264,36 +264,18 @@ func (i *Index) openFields(idx *disco.Index) error {
 	}
 	defer f.Close()
 
-	fis, err := f.Readdir(0)
-	if err != nil {
-		return errors.Wrap(err, "reading directory")
-	}
 	eg, ctx := errgroup.WithContext(context.Background())
 	var mu sync.Mutex
 
-fileLoop:
-	for _, loopFi := range fis {
-		select {
-		case <-ctx.Done():
-			break fileLoop
-		default:
-			fi := loopFi
-			if !fi.IsDir() {
-				continue
-			}
-
-			var cfm *CreateFieldMessage = &CreateFieldMessage{}
-			var err error
-
-			// Only continue with fields which are present in the provided,
-			// non-nil index schema. The reason we have to check for idx != nil
-			// here is because there are tests which call index.Open without
-			// having a disco.Index available.
-			if idx != nil {
-				fld, ok := idx.Fields[fi.Name()]
-				if !ok {
-					continue
-				}
+	if idx != nil {
+	fileLoop:
+		for fname, fld := range idx.Fields {
+			select {
+			case <-ctx.Done():
+				break fileLoop
+			default:
+				var cfm *CreateFieldMessage = &CreateFieldMessage{}
+				var err error
 
 				// Decode the CreateFieldMessage from the schema data in order to
 				// get its metadata.
@@ -301,22 +283,22 @@ fileLoop:
 				if err != nil {
 					return errors.Wrap(err, "decoding create field message")
 				}
+
+				indexQueue <- struct{}{}
+				eg.Go(func() error {
+					defer func() {
+						<-indexQueue
+					}()
+					i.holder.Logger.Debugf("open field: %s", fname)
+
+					_, err := i.openField(&mu, cfm, fname)
+					if err != nil {
+						return errors.Wrap(err, "opening field")
+					}
+
+					return nil
+				})
 			}
-
-			indexQueue <- struct{}{}
-			eg.Go(func() error {
-				defer func() {
-					<-indexQueue
-				}()
-				i.holder.Logger.Debugf("open field: %s", fi.Name())
-
-				_, err := i.openField(&mu, cfm, fi.Name())
-				if err != nil {
-					return errors.Wrap(err, "opening field")
-				}
-
-				return nil
-			})
 		}
 	}
 	err = eg.Wait()

From ff091b034615f88e26eb7c7c7461fc6936606339 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@molecula.com>
Date: Wed, 2 Feb 2022 14:20:38 -0600
Subject: [PATCH 2/5] implement a task pool

This implements a task pool which can handle backpressure; the
idea is, you have a target number of workers, but when a worker
blocks, you can tell it that it's blocking, and it can spawn
another worker in the mean time. This reduces the bounding provided
by the worker pool, and can significantly overshoot the intended size
of the pool in some cases, but it provides quick scaling up when
part of a workload gets blocked.

There's also a simulator attached to it. The simulator's job is
to act similarly to the executor's worker pool working on RBF
databases, including the weird semantics of writes and reads;
specifically, that reads aren't blocked by writes, but a write
can't terminate until every read that started before it has exited.
(This is an oversimplification; actually, writes can complete,
but they still hold the write lock until any WAL merge completes,
and the WAL merge can't complete until old reads are done.)

The simulator is significantly more complicated than the pool.
---
 task/doc.go       |  42 +++++
 task/pool.go      | 151 ++++++++++++++++
 task/pool_test.go | 430 ++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 623 insertions(+)
 create mode 100644 task/doc.go
 create mode 100644 task/pool.go
 create mode 100644 task/pool_test.go

diff --git a/task/doc.go b/task/doc.go
new file mode 100644
index 000000000..ccfbea653
--- /dev/null
+++ b/task/doc.go
@@ -0,0 +1,42 @@
+// Copyright 2022 Molecula Corp. All rights reserved.
+
+// Package task provides an interface for indicating when an operation has
+// been blocked, so that a worker pool which wants to be doing N things at
+// a time can start trying new things when some things are blocked.
+//
+// To understand this, you have to start with the original context: We have
+// a worker pool, which can handle up to N tasks at once. Tasks come in
+// in batches, asynchronously. At most one write task can be active on a given
+// database at a time, but many read tasks can be active on the database,
+// with or without a write task. Each read task completes only when its entire
+// containing operation completes. Write tasks can *partially* complete
+// immediately, but in some cases, must wait for read tasks to finish before
+// they can do crucial bookkeeping work.
+//
+// Regardless of the workload, we always have tasks which can progress
+// available, and if we do them, eventually everything will complete. However,
+// for some workloads, it is possible to pick N tasks *all of which are
+// blocked*. In this case, the worker pool becomes useless. Furthermore,
+// even if we don't hit that state, we can hit a state where nearly all worker
+// pool tasks are blocked.
+//
+// To address this, we need a way for a worker pool to recognize that a worker
+// has become blocked, and *start another worker*. This can result in running
+// more than N workers at once. However, it rarely results in running *many*
+// more. The typical case would be that we have a worker pool of N, and M of
+// them are blocked waiting for write access to a given database. If one of them
+// becomes unblocked, we may end up with N+1 active workers, but the other M-1
+// waiting on that database are still blocked.
+//
+// It might seem like the simplest thing to do is use a buffered channel as a
+// semaphore, this being a standard Go idiom for pools. It's a great idiom, but
+// in our case, it runs into a problem. When each worker starts, it writes into
+// a buffered channel. When it becomes blocked, it reads from the channel to
+// free up a slot. When it becomes unblocked, then, it has to write to the
+// channel to indicate that it's taking up a slot again. But writes to the
+// channel are contested, and usually only become possible when something else
+// either blocks or exits... Meaning that, precisely at the moment that we have
+// gained a highly contested lock and are able to proceed, we block for an
+// indeterminate period of time *while holding that lock*. This is the opposite
+// of what we want.
+package task
diff --git a/task/pool.go b/task/pool.go
new file mode 100644
index 000000000..2102b468b
--- /dev/null
+++ b/task/pool.go
@@ -0,0 +1,151 @@
+// Copyright 2022 Molecula Corp. All rights reserved.
+
+package task
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+// Pool represents a worker-pool type thing, which will call a given
+// function in parallel aiming for a given level of concurrency.
+// To use a pool, you create it, passing in a worker function; it
+// then spawns goroutines to run that function in a loop. If the Pool's
+// Block method is called, this marks one instance of the worker goroutine
+// as blocked; the Unblock method marks it as unblocked. When there are
+// insufficient unblocked goroutines, more are spawned. When there are
+// excess goroutines, they exit.
+//
+// The pool can be shut down by calling Close(), setting its target number
+// of workers to 0.
+type Pool struct {
+	mu        sync.Mutex // locker used for cond
+	cond      *sync.Cond // notify of exiting workers
+	step      func()
+	targetN   int32 // desired number
+	unblocked int32 // currently active and unblocked
+	live      int32 // currently active including blocked
+	stats     PoolStats
+}
+
+type PoolStats interface {
+	PoolSize(int) // reports current pool size
+}
+
+// NewPool creates a pool that attempts to keep targetN goroutines
+// active, executing step() repeatedly. It updates poolSize with the
+// current size of the pool when that changes.
+func NewPool(targetN int, step func(), stats PoolStats) *Pool {
+	p := &Pool{targetN: int32(targetN), step: step, stats: stats}
+	p.cond = sync.NewCond(&p.mu)
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for i := 0; i < targetN; i++ {
+		p.addWorker()
+	}
+	return p
+}
+
+// Block marks a worker as blocked, indicating that we may need a new worker
+// spawned because the caller is about to be blocked for an indeterminate
+// period of time. If a new worker is needed, it's spawned immediately before
+// Block returns.
+func (p *Pool) Block() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	unblocked := atomic.AddInt32(&p.unblocked, -1)
+	target := atomic.LoadInt32(&p.targetN)
+	if unblocked < target {
+		p.addWorker()
+	}
+}
+
+// Unblock marks a worker as unblocked, potentially allowing the pool to
+// retire a worker thread at some point in the future.
+func (p *Pool) Unblock() {
+	atomic.AddInt32(&p.unblocked, 1)
+}
+
+// Shutdown tells a pool to terminate by setting its desired pool size
+// to zero, but does not wait for the jobs in it to stop. It is safe to
+// call this before calling Close.
+func (p *Pool) Shutdown() {
+	atomic.StoreInt32(&p.targetN, 0)
+}
+
+// Stats reports on the pool's current state -- total live workers it
+// has, how many it thinks are unblocked, and what its target is.
+// These numbers are sampled individually, and there's no locking, so they
+// are not guaranteed to be consistent. This is useful for approximate
+// monitoring.
+func (p *Pool) Stats() (live, unblocked, target int) {
+	return int(atomic.LoadInt32(&p.live)), int(atomic.LoadInt32(&p.unblocked)), int(atomic.LoadInt32(&p.targetN))
+}
+
+// Close is a Shutdown followed by waiting for all jobs to exit.
+func (p *Pool) Close() {
+	p.mu.Lock()
+	p.Shutdown()
+	live := atomic.LoadInt32(&p.live)
+	for live > 0 {
+		p.cond.Wait()
+		// This line occurs while we hold p.mu. addWorker can't be called
+		// except from inside something that would also hold the lock.
+		// So, if the value can't be stale and increasing, and it can't
+		// increase anyway once targetN is 0.
+		live = atomic.LoadInt32(&p.live)
+	}
+}
+
+// addWorker increments the number of unblocked things, and starts a worker.
+// The unblocked count is technically wrong until the worker gets running, but
+// it's right "soon". The live count maintenance is done inside the worker.
+func (p *Pool) addWorker() {
+	// update worker count. we don't notify the condition variable because
+	// increasing workers can't make us more-closed.
+	live := atomic.AddInt32(&p.live, 1)
+	if p.stats != nil {
+		p.stats.PoolSize(int(live))
+	}
+	atomic.AddInt32(&p.unblocked, 1)
+	go p.work()
+}
+
+// work runs the provided work function in a loop as long as there's not
+// too many unblocked goroutines, otherwise it exits.
+func (p *Pool) work() {
+	defer func() {
+		live := atomic.AddInt32(&p.live, -1)
+		if p.stats != nil {
+			p.stats.PoolSize(int(live))
+		}
+		// notify any waiters that we're done
+		if live == 0 {
+			p.cond.Broadcast()
+		}
+	}()
+	for {
+		unblocked := atomic.LoadInt32(&p.unblocked)
+		target := atomic.LoadInt32(&p.targetN)
+		for unblocked > target {
+			// Might have too many!
+			swapped := atomic.CompareAndSwapInt32(&p.unblocked, unblocked, unblocked-1)
+			if swapped {
+				// we've successfully removed ourselves from the unblocked count.
+				// now return, letting the deferred add above remove us from the live
+				// count as well.
+				return
+			}
+			// If the swap failed, unblocked increased or decreased. We
+			// re-extract it, and try the loop again. If it's no longer higher
+			// than the target, this loop ends and we continue running.
+			// If it's higher than the target, we'll try again with this new
+			// value.
+			// We also reload target because someone could have told us to
+			// terminate.
+			unblocked = atomic.LoadInt32(&p.unblocked)
+			target = atomic.LoadInt32(&p.targetN)
+		}
+		p.step()
+	}
+}
diff --git a/task/pool_test.go b/task/pool_test.go
new file mode 100644
index 000000000..dbdf5f210
--- /dev/null
+++ b/task/pool_test.go
@@ -0,0 +1,430 @@
+// Copyright 2021 Molecula Corp. All rights reserved.
+
+package task
+
+import (
+	"fmt"
+	"golang.org/x/sync/errgroup"
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// db represents a thing which can be locked, and which can
+// perform read and write operations, which are modeled as channels which
+// a workload can wait on writes to, and which embeds a lockable RWMutex.
+// The RWMutex actually makes this slightly stricter than the semantics
+// of RBF, which usually allows writes and reads to coexist, but fairly
+// accurately represents the specific issue that RBF can't *finish* a write
+// while an older read is active. Not the same, but has similar impact.
+type db struct {
+	read, write chan struct{}
+	sync.Mutex
+}
+
+// server represents a set of dbs, which jobs can be run against. They're
+// [26] because they're denoted by lowercase/uppercase letters.
+type server struct {
+	dbs     [26]db
+	mu      sync.Mutex     // mutex to govern access to readers
+	readers [26][]workload // a list of readers associated with each db
+	waiters [26]struct {
+		mu   sync.Mutex
+		cond *sync.Cond
+	}
+	pool *Pool
+	jobs chan *job
+	tb   testing.TB
+}
+
+// a job represents a single operation on a server, and a receiver
+// waiting to hear back when it's done. It also has a reference to the
+// bitmasks of read/write locks so that the parent operation can clean
+// them all up when it's done. This is roughly parallel to the Qcx/Tx
+// locking behavior in featurebase.
+type job struct {
+	descr  workload // the workload that generated this job, used to identify them
+	id     int
+	write  bool
+	locked *uint32 // bitmask of read-locked jobs
+	ch     chan<- struct{}
+}
+
+// workload represents a series of jobs as letters;
+// lowercase letters read from the read channel of a component, uppercase
+// letters read from the write channel of the corresponding lowercase
+// component. each operation locks components as it reaches them for
+// the first time, then unlocks all of them at the end of the string.
+type workload string
+
+// runJob grabs a single job from the server's job queue, does it, and
+// notifies the waiter. To "do" a job is to acquire the appropriate
+// lock (for read or write), mark the appropriate bit in a bitmap of
+// active locks, and then read from either a read or write channel, which
+// then corresponds to values being passed to Satisfy.
+func (s *server) runJob() {
+	j, ok := <-s.jobs
+	if !ok {
+		return
+	}
+	if j.write {
+		s.pool.Block()
+		s.dbs[j.id].Lock()
+		s.pool.Unblock()
+		s.mu.Lock()
+		// obtain list of existing readers
+		waiting := make([]workload, len(s.readers[j.id]))
+		copy(waiting, s.readers[j.id])
+		s.mu.Unlock()
+		<-s.dbs[j.id].write
+		// In RBF, the write lock can't be released until the last outstanding
+		// reader predating this write terminates, but that's asynchronous
+		// from the actual request processing. So, similarly, we launch a thing
+		// that will unlock this slot in the database, once it's done waiting
+		// for any readers. We do that without the pool marked as blocked.
+		go func() {
+			// but the write can't actually complete until any pending readers
+			// that were already in play complete
+			if len(waiting) > 0 {
+				// We might need to wait for things. We need to be sure,
+				// though, that the server's list of readers for this isn't
+				// changing while we're checking it. So, we grab the specific
+				// lock, then check the reader list, and if we think we need
+				// to wait, we wait on a condition variable which then
+				// releases that lock so something else can update the reader
+				// list and notify us.
+				func() {
+					s.waiters[j.id].mu.Lock()
+					defer s.waiters[j.id].mu.Unlock()
+					// we have to check this with the specific lock held, so if
+					// anything were to change the list, it'd have to wait
+					// until we're done or waiting on the cond.
+					stillWaiting := s.stillWaiting(j.id, waiting)
+					for stillWaiting {
+						s.waiters[j.id].cond.Wait()
+						stillWaiting = s.stillWaiting(j.id, waiting)
+					}
+				}()
+			}
+			s.dbs[j.id].Unlock()
+		}()
+	} else {
+		s.pool.Block()
+		// attach us to the list of known readers, which must exit before
+		// any writers starting after them can exit
+		s.mu.Lock()
+		s.readers[j.id] = append(s.readers[j.id], j.descr)
+		s.mu.Unlock()
+		s.pool.Unblock()
+		cur := atomic.LoadUint32(j.locked)
+		// mask this bit in
+		for (cur>>j.id)&1 == 0 {
+			added := cur | (1 << j.id)
+			atomic.CompareAndSwapUint32(j.locked, cur, added)
+			cur = atomic.LoadUint32(j.locked)
+		}
+		<-s.dbs[j.id].read
+	}
+	j.ch <- struct{}{}
+}
+
+// stillWaiting determines whether we're still waiting on anything in
+// a given list terminating.
+func (s *server) stillWaiting(id int, waitingOn []workload) bool {
+	s.mu.Lock()
+	readers := s.readers[id]
+	s.mu.Unlock()
+	for _, waiter := range waitingOn {
+		for _, reader := range readers {
+			if waiter == reader {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// runWorkload runs the tasks within a workload, passing them to the worker
+// queue, and then waiting for them all to complete. When it's done waiting
+// for them, it releases any locks they obtained.
+func (s *server) runWorkload(w workload) {
+	var locked uint32
+	defer func() {
+		// unlock everything marked as locked
+		read := atomic.LoadUint32(&locked)
+		for i := 0; i < 32; i++ {
+			if (read>>i)&1 != 0 {
+				s.waiters[i].mu.Lock()
+				s.mu.Lock()
+				// remove us from readers list
+				for j := range s.readers[i] {
+					if s.readers[i][j] == w {
+						copy(s.readers[i][j:], s.readers[i][j+1:])
+						s.readers[i] = s.readers[i][:len(s.readers[i])-1]
+						break
+					}
+				}
+				s.mu.Unlock()
+				s.waiters[i].mu.Unlock()
+				// and wake up anything that was waiting for this.
+				s.waiters[i].cond.Broadcast()
+			}
+		}
+	}()
+	ch := make(chan struct{})
+	eg := &errgroup.Group{}
+	j := job{ch: ch, locked: &locked, descr: w}
+	for _, c := range w {
+		switch {
+		case c >= 'a' && c <= 'z':
+			j.id = int(c - 'a')
+			j.write = false
+		case c >= 'A' && c <= 'Z':
+			j.id = int(c - 'A')
+			j.write = true
+		default:
+			s.tb.Logf("unhandled character '%c'", c)
+			continue
+		}
+		j := j
+		eg.Go(func() error {
+			s.jobs <- &j
+			<-ch
+			return nil
+		})
+	}
+	_ = eg.Wait()
+}
+
+// newServer creates a server associated with the given testing.TB,
+// allowing us to log things.
+func newServer(tb testing.TB) *server {
+	s := &server{tb: tb, jobs: make(chan *job)}
+	for i := range s.dbs {
+		s.dbs[i].read = make(chan struct{})
+		s.dbs[i].write = make(chan struct{})
+		s.waiters[i].cond = sync.NewCond(&s.waiters[i].mu)
+	}
+	return s
+}
+
+// close shuts the server down by closing all of its channels, and may
+// not really be necessary.
+func (s *server) close() {
+	for i := range s.dbs {
+		db := &s.dbs[i]
+		db.Lock()
+		close(db.read)
+		close(db.write)
+		db.Unlock()
+	}
+	close(s.jobs)
+}
+
+// Satisfy satisfies the given read or write operations asynchronously,
+// but waits for all of them in this batch to complete before returning.
+func (s *server) satisfy(w workload) {
+	var eg errgroup.Group
+	for _, c := range w {
+		var id int
+		var write bool
+		switch {
+		case c >= 'a' && c <= 'z':
+			id = int(c - 'a')
+			write = false
+		case c >= 'A' && c <= 'Z':
+			id = int(c - 'A')
+			write = true
+		default:
+			s.tb.Logf("unhandled character '%c'", c)
+			continue
+		}
+		eg.Go(func() error {
+			if write {
+				s.dbs[id].write <- struct{}{}
+			} else {
+				s.dbs[id].read <- struct{}{}
+			}
+			return nil
+		})
+	}
+	_ = eg.Wait()
+}
+
+// makeWorkload generates a sequence of letters, some of which may be
+// capitalized, in order
+func makeWorkload() workload {
+	var letters [26]byte
+	var n int
+	write := rand.Intn(8) == 0
+	for i := 0; i < 26; i++ {
+		if rand.Intn(4) == 0 {
+			if write {
+				letters[n] = 'A' + byte(i)
+			} else {
+				letters[n] = 'a' + byte(i)
+			}
+			n++
+		}
+	}
+	return workload(letters[:n])
+}
+
+// testRandomWorkload makes up an arbitrary workload and tries to run
+// the server against it.
+func testRandomWorkload(t *testing.T) {
+	s := newServer(t)
+	eg := &errgroup.Group{}
+	p := NewPool(2, s.runJob, nil)
+	s.pool = p
+	defer p.Close()
+	defer s.close()
+	var workloads []workload // the requests we make
+	var quick []workload     // the requests that get satisfied soon
+	var slow []workload      // the requests that don't get satisfied until later
+	for i := 0; i < 10; i++ {
+		w := makeWorkload()
+		if len(w) == 0 {
+			continue
+		}
+		workloads = append(workloads, w)
+		partial := rand.Intn(26)
+		// possibly truncate and postpone some
+		if partial < len(w) {
+			quick = append(quick, w[:partial])
+			slow = append(slow, w[partial:])
+		} else {
+			quick = append(quick, w)
+		}
+	}
+	for _, w := range workloads {
+		w := w
+		eg.Go(func() error {
+			s.runWorkload(w)
+			return nil
+		})
+	}
+	for _, w := range quick {
+		w := w
+		eg.Go(func() error {
+			s.satisfy(w)
+			return nil
+		})
+	}
+	l, u, target := p.Stats()
+	// Only one worker at a time can be invoking the mark-as-blocked logic,
+	// so you can run after it marks that, but before the new worker is spawned,
+	// but the next worker can't invoke the blocked logic until that completes.
+	//
+	// Live count always decreases after unblocked count on the exit path, and
+	// increases before unblocked count on the startup path. So even if the
+	// samples are interrupted, I think it should be impossible for live
+	// to be less than unblocked.
+	if u < target-1 || l < u {
+		t.Fatalf("inconsistent pool stats: %d live, %d unblocked, %d target", l, u, target)
+	}
+	for _, w := range slow {
+		w := w
+		eg.Go(func() error {
+			s.satisfy(w)
+			return nil
+		})
+	}
+	_ = eg.Wait()
+}
+
+// TestRandomWorkloads makes up some arbitrary workloads, then tries to
+// satisfy them out of order.
+// In theory, this should work for any sequence of operations as long as
+// no operation has the same letter for both read and write ops, and
+// ops always occur in order.
+func TestRandomWorkloads(t *testing.T) {
+	for i := 0; i < 10; i++ {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			testRandomWorkload(t)
+		})
+	}
+}
+
+func TestServer(t *testing.T) {
+	s := newServer(t)
+	eg := &errgroup.Group{}
+	p := NewPool(3, s.runJob, nil)
+	s.pool = p
+	defer p.Close()
+	defer s.close()
+	request := func(w workload) {
+		eg.Go(func() error {
+			s.runWorkload(w)
+			return nil
+		})
+	}
+	// requesting "abcd" means that the reader "abcd" will still be active on
+	// a until all the other letters show up.
+	request("abcd")
+	// satisfy won't complete until at least two of the jobs have happened,
+	// so there's a decent chance that we've marked ourselves as a reader on
+	// a.
+	s.satisfy("abc")
+	// so we request a write on A. we get the write lock, but we can't
+	// relinquish it until "d" shows up.
+	request("A")
+	// satisfy that request immediately, but to no avail.
+	s.satisfy("A")
+	// three more requests come in. if they get pool slots, they definitely
+	// block; that request on A can't have finished yet. so they could fully
+	// block our work pool.
+	request("A")
+	request("A")
+	request("A")
+	// spawn something to provide "efg"
+	go s.satisfy("efg")
+	// runWorkload means we actually block waiting for it. if all the worker
+	// pool is blocked waiting on A, we can't do that.
+	s.runWorkload("efg")
+	// now we provide the missing d, which should allow the first request to
+	// finally complete, and then the next three A, which should finish
+	// the rest.
+	s.satisfy("dAAA")
+	// If we spawned new jobs, this should complete. Otherwise it should hang
+	// because the requests can't be satisfied because the queue is full
+	// of blocked operations.
+	_ = eg.Wait()
+}
+
+func TestPoolStartup(t *testing.T) {
+	var counter int32
+	started := make(chan struct{})
+	done := make(chan struct{})
+	addAndWait := func() {
+		<-started
+		atomic.AddInt32(&counter, 1)
+		<-done
+	}
+	// we expect this to spawn three counters
+	p := NewPool(3, addAndWait, nil)
+	time.Sleep(50 * time.Millisecond)
+	v := atomic.LoadInt32(&counter)
+	if v != 0 {
+		t.Fatalf("expected no adds yet, got %d", v)
+	}
+	close(started)
+	time.Sleep(50 * time.Millisecond)
+	v = atomic.LoadInt32(&counter)
+	if v != 3 {
+		t.Fatalf("expected 3 adds, got %d", v)
+	}
+	// Tell the pool to stop processing jobs
+	p.Shutdown()
+	// Allow the jobs to complete. Since this happens after the
+	// shutdown has set desired pool size to zero, they should now all exit.
+	close(done)
+	p.Close()
+	time.Sleep(50 * time.Millisecond)
+	v = atomic.LoadInt32(&counter)
+	if v != 3 {
+		t.Fatalf("expected no more adds, got %d including previous 3", v)
+	}
+}

From 96ab9314d1deb934b3ac17d80401440efc1e823d Mon Sep 17 00:00:00 2001
From: Seebs <seebs@molecula.com>
Date: Thu, 10 Feb 2022 12:09:38 -0600
Subject: [PATCH 3/5] use task pool for executor workers

This adopts the task pool functionality to let us spawn new worker
threads when worker threads are blocked. The underlying reason for
this is the same as the reason for the previous worker-pool-growing
strategy; while our design persistently has at least one thing which
can proceed, it can be the case that there are N things blocked,
where N is the size of our worker pool. Blocked workers shouldn't
count against our desired number of workers.

Originally, the intent was to thread this into RBF, and provide
backpressure from RBF on the pool when blocking on writes. Unfortunately,
that's not good enough, because while a write is blocked, the Qcx
calling it is *also* holding the Qcx's mutex, which means that any other
NewTx on that Qcx will *also* block. So we need to block for the
entire time of the NewTx.

Removing the existing worker spawning code resulted in a subtle
and maybe-harmless change; prior to this, each invocation of `mapperLocal`
would hold a lock, which meant that all the tasks for a given local mapper
would be put in the queue *sequentially*, ensuring that they'd all be
picked up by workers before things from later workers.

With the new pushback, that's not, strictly, necessary. Also, if you
disable it, you can end up with 300,000 goroutines at once, most of them
blocked.

A smallish run does, in fact, eventually complete anyway -- it will
indeed keep making workers until everything gets one. However, while
it's *correct*, it's also noticably *slower*. The same test workload
goes from around 33 seconds to a bit over 40 seconds when that lock
isn't present. (But that's with an extremely small WAL write cap
introduced to make the previous deadlock possible.)

With large numbers of shards, the practical impact is that you can
have quite a lot of things in process, with hundreds of goroutines
each, all blocked waiting for one writer. If we force them to all be
processed at the same time, all the reads that are connected to
each other are much more likely to get all processed at once, before
something new comes along.

In short, that lock isn't strictly necessary but it seems to help
noticably with performance and reduce simultaneous goroutines
significantly.
---
 executor.go  | 107 ++++++++++++++++-----------------------------------
 holder.go    |   3 ++
 server.go    |   1 +
 txfactory.go |  13 ++++++-
 4 files changed, 48 insertions(+), 76 deletions(-)

diff --git a/executor.go b/executor.go
index 3b09cd92c..232c1984c 100644
--- a/executor.go
+++ b/executor.go
@@ -22,6 +22,7 @@ import (
 	"github.com/molecula/featurebase/v3/proto"
 	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/shardwidth"
+	"github.com/molecula/featurebase/v3/task"
 	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
@@ -51,9 +52,6 @@ type executor struct {
 	Node    *topology.Node
 	Cluster *cluster
 
-	// how many jobs the work queue has seen
-	workCounter uint64
-
 	// Client used for remote requests.
 	client *InternalClient
 
@@ -61,10 +59,9 @@ type executor struct {
 	MaxWritesPerRequest int
 
 	shutdown       bool
-	workMu         sync.RWMutex
-	workersWG      sync.WaitGroup
+	workers        *task.Pool
+	workerPoolMu   sync.Mutex
 	workerPoolSize int
-	currentWorkers int64
 	work           chan job
 
 	// Maximum per-request memory usage (Extract() only)
@@ -130,72 +127,11 @@ func newExecutor(opts ...executorOption) *executor {
 	// workloads. Possible that it could be smaller.
 	e.work = make(chan job, e.workerPoolSize)
 	_ = testhook.Opened(NewAuditor(), e, nil)
-	for i := 0; i < e.workerPoolSize; i++ {
-		e.addWorker()
-	}
-	go func() {
-		// background task: every so often, check to see whether we have
-		// work in the queue but none has been taken for a while. if so, we
-		// need more workers.
-		prev := atomic.LoadUint64(&e.workCounter)
-		periodic := time.NewTicker(50 * time.Millisecond)
-		defer periodic.Stop()
-		running := true
-		idle := 0
-		for running {
-			<-periodic.C
-			func() {
-				e.workMu.RLock()
-				defer e.workMu.RUnlock()
-				if e.shutdown {
-					running = false
-					return
-				}
-				if len(e.work) == 0 {
-					idle++
-					if idle > 10 && atomic.LoadInt64(&e.currentWorkers) > int64(e.workerPoolSize*2) {
-						select {
-						case e.work <- job{idleHands: true}:
-							// we closed an excess worker
-						default:
-							// somehow between our test above and now the work
-							// queue FILLED UP and we stoically accept this
-						}
-						idle = 0
-					}
-					return
-				}
-				next := atomic.LoadUint64(&e.workCounter)
-				if next == prev {
-					e.addWorker()
-				}
-				prev = next
-			}()
-		}
-	}()
+	e.workers = task.NewPool(e.workerPoolSize, e.doOneJob, e)
 	return e
 }
 
-func (e *executor) addWorker() {
-	e.workersWG.Add(1)
-	n := atomic.AddInt64(&e.currentWorkers, 1)
-	if e.Holder != nil {
-		e.Holder.Stats.Gauge("worker_total", float64(n), 0)
-	}
-
-	go func() {
-		defer e.workersWG.Done()
-		e.worker(e.work)
-		n := atomic.AddInt64(&e.currentWorkers, -1)
-		if e.Holder != nil {
-			e.Holder.Stats.Gauge("worker_total", float64(n), 0)
-		}
-	}()
-}
-
 func (e *executor) Close() error {
-	e.workMu.Lock()
-	defer e.workMu.Unlock()
 	if e.shutdown {
 		// otherwise close(e.work) can result in
 		// panic: close of closed channel.
@@ -206,15 +142,23 @@ func (e *executor) Close() error {
 	e.shutdown = true
 	_ = testhook.Closed(NewAuditor(), e, nil)
 	close(e.work)
-	e.workersWG.Wait()
+	e.workers.Close()
 	return nil
 }
 
+// PoolSize is exported to let the task pool update us
+func (e *executor) PoolSize(n int) {
+	if e.Holder != nil {
+		e.Holder.Stats.Gauge("worker_total", float64(n), 0)
+	}
+}
+
 // InitStats initializes stats counters. Must be called after Holder set.
 func (e *executor) InitStats() {
 	if e.Holder != nil {
 		e.Holder.Stats.Count("job_total", 0, 0)
-		e.Holder.Stats.Gauge("worker_total", float64(atomic.LoadInt64(&e.currentWorkers)), 0)
+		l, _, _ := e.workers.Stats()
+		e.Holder.Stats.Gauge("worker_total", float64(l), 0)
 	}
 }
 
@@ -6068,9 +6012,25 @@ type job struct {
 	idleHands       bool
 }
 
+// doOneJob had one job. *disappointed sigh*
+func (e *executor) doOneJob() {
+	j, ok := <-e.work
+	if !ok {
+		return
+	}
+	// Skip out early if the context is done, but still send
+	// an ack so mapperLocal can be sure we aren't about to
+	// work on something it sent us.
+	if err := j.ctx.Err(); err != nil {
+		j.resultChan <- mapResponse{result: nil, err: err}
+		return
+	}
+	result, err := j.mapFn(j.ctx, j.shard, &mapOptions{memoryAvailable: j.memoryAvailable})
+	j.resultChan <- mapResponse{result: result, err: err}
+}
+
 func (e *executor) worker(work chan job) {
 	for j := range work {
-		atomic.AddUint64(&e.workCounter, 1)
 		e.Holder.Stats.Count("job_total", 1, 0)
 		if j.idleHands {
 			return
@@ -6096,9 +6056,8 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	done := ctx.Done()
-	e.workMu.RLock()
-	defer e.workMu.RUnlock()
-
+	e.workerPoolMu.Lock()
+	defer e.workerPoolMu.Unlock()
 	if e.shutdown {
 		return nil, errShutdown
 	}
diff --git a/holder.go b/holder.go
index 18cd7154c..4ecb5ec6e 100644
--- a/holder.go
+++ b/holder.go
@@ -70,6 +70,9 @@ type Holder struct {
 	sharder     disco.Sharder
 	serializer  Serializer
 
+	// executor, which we use only to get access to its worker pool
+	executor *executor
+
 	// Close management
 	wg      sync.WaitGroup
 	closing chan struct{}
diff --git a/server.go b/server.go
index 3475b670a..531e56737 100644
--- a/server.go
+++ b/server.go
@@ -494,6 +494,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("node_id:%s", s.nodeID))
 
 	s.executor.Holder = s.holder
+	s.holder.executor = s.executor
 	s.executor.Cluster = s.cluster
 	s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
 	s.cluster.broadcaster = s
diff --git a/txfactory.go b/txfactory.go
index 58134a6bb..54dfa4390 100644
--- a/txfactory.go
+++ b/txfactory.go
@@ -8,6 +8,7 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/molecula/featurebase/v3/task"
 	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/molecula/featurebase/v3/vprint"
 	"github.com/pkg/errors"
@@ -83,8 +84,9 @@ var sep = string(os.PathSeparator)
 // See also the Qcx.GetTx() example and the TxGroup description below.
 //
 type Qcx struct {
-	Grp *TxGroup
-	Txf *TxFactory
+	Grp     *TxGroup
+	Txf     *TxFactory
+	workers *task.Pool
 
 	// if we go back to using Qcx values, this must become a pointer,
 	// or otherwise be dealt with because copies of Mutex are a no-no.
@@ -178,6 +180,9 @@ func (f *TxFactory) NewQcx() (qcx *Qcx) {
 		Grp: f.NewTxGroup(),
 		Txf: f,
 	}
+	if f.holder != nil && f.holder.executor != nil {
+		qcx.workers = f.holder.executor.workers
+	}
 	if f.typeOfTx == "roaring" {
 		qcx.isRoaring = true
 	}
@@ -223,6 +228,10 @@ var ErrQcxDone = fmt.Errorf("Qcx already Aborted or Finished, so must call reset
 // to make it clear we are referring to the first and final error.
 //
 func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
+	if qcx.workers != nil {
+		qcx.workers.Block()
+		defer qcx.workers.Unblock()
+	}
 	qcx.mu.Lock()
 	defer qcx.mu.Unlock()
 

From 7013910158a9b2689afe051742ac19249d54436d Mon Sep 17 00:00:00 2001
From: kcrodgers24 <kcrodgers@live.com>
Date: Mon, 14 Feb 2022 09:15:23 -0800
Subject: [PATCH 4/5] give each test its own InMemSchemator

---
 disco/disco.go          | 6 ++++++
 holder.go               | 2 +-
 holder_internal_test.go | 2 +-
 index.go                | 2 +-
 4 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/disco/disco.go b/disco/disco.go
index 7f4519f13..6ec6641a0 100644
--- a/disco/disco.go
+++ b/disco/disco.go
@@ -312,6 +312,12 @@ type inMemSchemator struct {
 	schema Schema
 }
 
+func NewInMemSchemator() *inMemSchemator {
+	return &inMemSchemator{
+		schema: make(Schema),
+	}
+}
+
 // Schema is an in-memory implementation of the Schemator Schema method.
 func (s *inMemSchemator) Schema(ctx context.Context) (Schema, error) {
 	s.mu.RLock()
diff --git a/holder.go b/holder.go
index db9aa0cd2..e98c4e4bc 100644
--- a/holder.go
+++ b/holder.go
@@ -215,7 +215,7 @@ func DefaultHolderConfig() *HolderConfig {
 		OpenIDAllocator:      func(string, bool) (*idAllocator, error) { return &idAllocator{}, nil },
 		TranslationSyncer:    NopTranslationSyncer,
 		Serializer:           GobSerializer,
-		Schemator:            disco.InMemSchemator,
+		Schemator:            disco.NewInMemSchemator(),
 		Sharder:              disco.InMemSharder,
 		CacheFlushInterval:   defaultCacheFlushInterval,
 		StatsClient:          stats.NopStatsClient,
diff --git a/holder_internal_test.go b/holder_internal_test.go
index e18704ac8..c9e164f27 100644
--- a/holder_internal_test.go
+++ b/holder_internal_test.go
@@ -10,7 +10,7 @@ func mustHolderConfig() *HolderConfig {
 	cfg := DefaultHolderConfig()
 	cfg.StorageConfig.FsyncEnabled = false
 	cfg.RBFConfig.FsyncEnabled = false
-	cfg.Schemator = disco.InMemSchemator
+	cfg.Schemator = disco.NewInMemSchemator()
 	cfg.Sharder = disco.InMemSharder
 	return cfg
 }
diff --git a/index.go b/index.go
index 11c924c08..0901f1a36 100644
--- a/index.go
+++ b/index.go
@@ -77,7 +77,7 @@ func NewIndex(holder *Holder, path, name string) (*Index, error) {
 		holder:         holder,
 		trackExistence: true,
 
-		Schemator:  disco.InMemSchemator,
+		Schemator:  disco.NewInMemSchemator(),
 		serializer: NopSerializer,
 
 		translateStores: make(map[int]TranslateStore),

From d57050d966cba7137e8b04e2725d7ead71b8abaa Mon Sep 17 00:00:00 2001
From: kcrodgers24 <kcrodgers@live.com>
Date: Mon, 14 Feb 2022 10:09:33 -0800
Subject: [PATCH 5/5] requested idx == nil fix; add doc comment

---
 disco/disco.go |  2 ++
 index.go       | 58 ++++++++++++++++++++++++++------------------------
 2 files changed, 32 insertions(+), 28 deletions(-)

diff --git a/disco/disco.go b/disco/disco.go
index 6ec6641a0..8ad815c28 100644
--- a/disco/disco.go
+++ b/disco/disco.go
@@ -312,6 +312,8 @@ type inMemSchemator struct {
 	schema Schema
 }
 
+// NewInMemSchemator instantiates an InMemSchemator
+// this allows new holders to have thier own, and not rely on a shared instance
 func NewInMemSchemator() *inMemSchemator {
 	return &inMemSchemator{
 		schema: make(Schema),
diff --git a/index.go b/index.go
index 0901f1a36..0e900cb05 100644
--- a/index.go
+++ b/index.go
@@ -267,40 +267,42 @@ func (i *Index) openFields(idx *disco.Index) error {
 	eg, ctx := errgroup.WithContext(context.Background())
 	var mu sync.Mutex
 
-	if idx != nil {
-	fileLoop:
-		for fname, fld := range idx.Fields {
-			select {
-			case <-ctx.Done():
-				break fileLoop
-			default:
-				var cfm *CreateFieldMessage = &CreateFieldMessage{}
-				var err error
+	if idx == nil {
+		return nil
+	}
+fileLoop:
+	for fname, fld := range idx.Fields {
+		select {
+		case <-ctx.Done():
+			break fileLoop
+		default:
+			var cfm *CreateFieldMessage = &CreateFieldMessage{}
+			var err error
 
-				// Decode the CreateFieldMessage from the schema data in order to
-				// get its metadata.
-				cfm, err = decodeCreateFieldMessage(i.holder.serializer, fld.Data)
+			// Decode the CreateFieldMessage from the schema data in order to
+			// get its metadata.
+			cfm, err = decodeCreateFieldMessage(i.holder.serializer, fld.Data)
+			if err != nil {
+				return errors.Wrap(err, "decoding create field message")
+			}
+
+			indexQueue <- struct{}{}
+			eg.Go(func() error {
+				defer func() {
+					<-indexQueue
+				}()
+				i.holder.Logger.Debugf("open field: %s", fname)
+
+				_, err := i.openField(&mu, cfm, fname)
 				if err != nil {
-					return errors.Wrap(err, "decoding create field message")
+					return errors.Wrap(err, "opening field")
 				}
 
-				indexQueue <- struct{}{}
-				eg.Go(func() error {
-					defer func() {
-						<-indexQueue
-					}()
-					i.holder.Logger.Debugf("open field: %s", fname)
-
-					_, err := i.openField(&mu, cfm, fname)
-					if err != nil {
-						return errors.Wrap(err, "opening field")
-					}
-
-					return nil
-				})
-			}
+				return nil
+			})
 		}
 	}
+
 	err = eg.Wait()
 	if err != nil {
 		// Close any fields which got opened, since the overall