scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'sup-146' of github.com:molecula/featurebase into sup-146

Browse source
This commit is contained in:
Todd Gruben 2022-02-14 14:09:13 -06:00
commit d4b7d0cb57
10 changed files with 698 additions and 128 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
disco/disco.go
View file

@ -312,6 +312,14 @@ 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),
}
}
// Schema is an in-memory implementation of the Schemator Schema method.
func (s *inMemSchemator) Schema(ctx context.Context) (Schema, error) {
s.mu.RLock()

107
executor.go
View file

@ -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
}

30
holder.go
View file

@ -9,7 +9,6 @@ import (
"path/filepath"
"runtime"
"sort"
"strings"
"sync"
"time"
@ -70,6 +69,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{}
@ -216,7 +218,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,
@ -332,23 +334,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 +342,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")

2
holder_internal_test.go
View file

@ -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
}

42
index.go
View file

@ -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),
@ -264,43 +264,26 @@ 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
if idx == nil {
return nil
}
fileLoop:
for _, loopFi := range fis {
for fname, fld := range idx.Fields {
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
}
// 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")
}
// 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{}{}
@ -308,9 +291,9 @@ fileLoop:
defer func() {
<-indexQueue
}()
i.holder.Logger.Debugf("open field: %s", fi.Name())
i.holder.Logger.Debugf("open field: %s", fname)
_, err := i.openField(&mu, cfm, fi.Name())
_, err := i.openField(&mu, cfm, fname)
if err != nil {
return errors.Wrap(err, "opening field")
}
@ -319,6 +302,7 @@ fileLoop:
})
}
}
err = eg.Wait()
if err != nil {
// Close any fields which got opened, since the overall

1
server.go
View file

@ -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

42
task/doc.go Normal file
View file

@ -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

151
task/pool.go Normal file
View file

@ -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()
}
}

430
task/pool_test.go Normal file
View file

@ -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)
}
}

13
txfactory.go
View file

@ -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()