Repository: petermattis/pebble
Pull request #225 (2019-08-15 23:43:53)

Add TableCreated event
Add an event for table creation. This event fires immediately after the
sstable file is created which means the file will be empty (nothing has
been written to it yet).

Refactor the write stall event tests to use the TableCreated event for
detecting when a table is created for flush vs compaction.


Patch:
diff --git a/compaction.go b/compaction.go
index 3f7e59f..588e3e6 100644
--- a/compaction.go
+++ b/compaction.go
@@ -601,7 +601,7 @@ func (d *DB) getCompactionPacerInfo() compactionPacerInfo {
 	d.mu.Lock()
 	estimatedMaxWAmp := d.mu.versions.picker.estimatedMaxWAmp
 	pacerInfo := compactionPacerInfo{
-		slowdownThreshold: uint64(estimatedMaxWAmp * float64(d.opts.MemTableSize)),
+		slowdownThreshold:   uint64(estimatedMaxWAmp * float64(d.opts.MemTableSize)),
 		totalCompactionDebt: d.mu.versions.picker.estimatedCompactionDebt(bytesFlushed),
 	}
 	d.mu.Unlock()
@@ -685,7 +685,7 @@ func (d *DB) flush1() error {
 		memTableSize: uint64(d.opts.MemTableSize),
 		getInfo:      d.getFlushPacerInfo,
 	})
-	ve, pendingOutputs, err := d.runCompaction(c, flushPacer)
+	ve, pendingOutputs, err := d.runCompaction(jobID, c, flushPacer)
 
 	if d.opts.EventListener.FlushEnd != nil {
 		info := FlushInfo{
@@ -832,7 +832,7 @@ func (d *DB) compact1() (err error) {
 		memTableSize: uint64(d.opts.MemTableSize),
 		getInfo:      d.getCompactionPacerInfo,
 	})
-	ve, pendingOutputs, err := d.runCompaction(c, compactionPacer)
+	ve, pendingOutputs, err := d.runCompaction(jobID, c, compactionPacer)
 
 	if d.opts.EventListener.CompactionEnd != nil {
 		info.Err = err
@@ -869,7 +869,7 @@ func (d *DB) compact1() (err error) {
 //
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
-func (d *DB) runCompaction(c *compaction, pacer pacer) (
+func (d *DB) runCompaction(jobID int, c *compaction, pacer pacer) (
 	ve *versionEdit, pendingOutputs []uint64, retErr error,
 ) {
 	// Check for a trivial move of one table from one level to the next. We avoid
@@ -959,6 +959,18 @@ func (d *DB) runCompaction(c *compaction, pacer pacer) (
 		if err != nil {
 			return err
 		}
+		if d.opts.EventListener.TableCreated != nil {
+			reason := "flushing"
+			if c.flushing == nil {
+				reason = "compacting"
+			}
+			d.opts.EventListener.TableCreated(TableCreateInfo{
+				JobID:   jobID,
+				Reason:  reason,
+				Path:    filename,
+				FileNum: fileNum,
+			})
+		}
 		file = vfs.NewSyncingFile(file, vfs.SyncingFileOptions{
 			BytesPerSync: d.opts.BytesPerSync,
 		})
diff --git a/event.go b/event.go
index dc3fb6c..2b9f064 100644
--- a/event.go
+++ b/event.go
@@ -21,6 +21,9 @@ type ManifestCreateInfo = base.ManifestCreateInfo
 // ManifestDeleteInfo exports the base.ManifestDeleteInfo type.
 type ManifestDeleteInfo = base.ManifestDeleteInfo
 
+// TableCreateInfo exports the base.TableCreateInfo type.
+type TableCreateInfo = base.TableCreateInfo
+
 // TableDeleteInfo exports the base.TableDeleteInfo type.
 type TableDeleteInfo = base.TableDeleteInfo
 
diff --git a/internal/base/event.go b/internal/base/event.go
index 94612e8..46cfa0a 100644
--- a/internal/base/event.go
+++ b/internal/base/event.go
@@ -143,6 +143,19 @@ func (i ManifestDeleteInfo) String() string {
 	return fmt.Sprintf("[JOB %d] MANIFEST deleted %06d", i.JobID, i.FileNum)
 }
 
+// TableCreateInfo contains the info for a table creation event.
+type TableCreateInfo struct {
+	JobID int
+	// Reason is the reason for the table creation (flushing or compacting).
+	Reason  string
+	Path    string
+	FileNum uint64
+}
+
+func (i TableCreateInfo) String() string {
+	return fmt.Sprintf("[JOB %d] %s: sstable created %06d", i.JobID, i.Reason, i.FileNum)
+}
+
 // TableDeleteInfo contains the info for a table deletion event.
 type TableDeleteInfo struct {
 	JobID   int
@@ -230,6 +243,7 @@ func (i WALDeleteInfo) String() string {
 	return fmt.Sprintf("[JOB %d] WAL deleted %06d", i.JobID, i.FileNum)
 }
 
+// WriteStallBeginInfo contains the info for a write stall begin event.
 type WriteStallBeginInfo struct {
 	Reason string
 }
@@ -270,6 +284,9 @@ type EventListener struct {
 	// ManifestDeleted is invoked after a manifest has been deleted.
 	ManifestDeleted func(ManifestDeleteInfo)
 
+	// TableCreated is invoked when a table has been created.
+	TableCreated func(TableCreateInfo)
+
 	// TableDeleted is invoked after a table has been deleted.
 	TableDeleted func(TableDeleteInfo)
 
@@ -330,6 +347,9 @@ func MakeLoggingEventListener(logger Logger) EventListener {
 		ManifestDeleted: func(info ManifestDeleteInfo) {
 			logger.Infof("%s", info.String())
 		},
+		TableCreated: func(info TableCreateInfo) {
+			logger.Infof("%s", info.String())
+		},
 		TableDeleted: func(info TableDeleteInfo) {
 			logger.Infof("%s", info.String())
 		},
diff --git a/open.go b/open.go
index 1776680..5ffff04 100644
--- a/open.go
+++ b/open.go
@@ -204,9 +204,13 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	sort.Slice(logFiles, func(i, j int) bool {
 		return logFiles[i].num < logFiles[j].num
 	})
+
+	jobID := d.mu.nextJobID
+	d.mu.nextJobID++
+
 	var ve versionEdit
 	for _, lf := range logFiles {
-		maxSeqNum, err := d.replayWAL(&ve, opts.FS, filepath.Join(d.walDirname, lf.name), lf.num)
+		maxSeqNum, err := d.replayWAL(jobID, &ve, opts.FS, filepath.Join(d.walDirname, lf.name), lf.num)
 		if err != nil {
 			return nil, err
 		}
@@ -254,8 +258,6 @@ func Open(dirname string, opts *Options) (*DB, error) {
 		return nil, err
 	}
 
-	jobID := d.mu.nextJobID
-	d.mu.nextJobID++
 	d.scanObsoleteFiles(ls)
 	d.deleteObsoleteFiles(jobID)
 	d.maybeScheduleFlush()
@@ -270,6 +272,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
 func (d *DB) replayWAL(
+	jobID int,
 	ve *versionEdit,
 	fs vfs.FS,
 	filename string,
@@ -342,7 +345,7 @@ func (d *DB) replayWAL(
 	if mem != nil && !mem.empty() {
 		c := newFlush(d.opts, d.mu.versions.currentVersion(),
 			1 /* base level */, []flushable{mem}, &d.bytesFlushed)
-		newVE, pendingOutputs, err := d.runCompaction(c, nilPacer)
+		newVE, pendingOutputs, err := d.runCompaction(jobID, c, nilPacer)
 		if err != nil {
 			return 0, err
 		}


Test patch:
diff --git a/data_test.go b/data_test.go
index eae7a87..62356fe 100644
--- a/data_test.go
+++ b/data_test.go
@@ -304,7 +304,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 		c := newFlush(d.opts, d.mu.versions.currentVersion(),
 			d.mu.versions.picker.baseLevel, []flushable{mem}, &d.bytesFlushed)
 		c.disableRangeTombstoneElision = true
-		newVE, _, err := d.runCompaction(c, nilPacer)
+		newVE, _, err := d.runCompaction(0, c, nilPacer)
 		if err != nil {
 			return nil
 		}
diff --git a/event_listener_test.go b/event_listener_test.go
index cfe4a50..31666d8 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -10,7 +10,6 @@ import (
 	"io"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"testing"
 
 	"github.com/petermattis/pebble/internal/base"
@@ -177,129 +176,85 @@ func TestEventListener(t *testing.T) {
 	})
 }
 
-type createDelayingFS struct {
-	vfs.FS
-	index          int32         // index of sstable whose creation will be delayed
-	createReleased chan struct{} // when closed, the delayed sstable creation can proceed
-	createDelayed  chan struct{} // if non-nil, will be closed before delaying sstable creation
-}
-
-func (fs *createDelayingFS) Create(name string) (vfs.File, error) {
-	if strings.HasSuffix(name, ".sst") && atomic.AddInt32(&fs.index, -1) == -1 {
-		if fs.createDelayed != nil {
-			close(fs.createDelayed)
-		}
-		<-fs.createReleased
-	}
-	return fs.FS.Create(name)
-}
-
-// Verify `WriteStallBegin()` and `WriteStallEnd()` events for stalls triggered
-// for reaching memtable count limit.
-func TestWriteStallForMemTableLimit(t *testing.T) {
-	const memTableLimitReason = "memtable count limit reached"
-	const memTableLimit = 2
-
-	stallBegan := make(chan struct{})
-	stallEnded := make(chan struct{})
-	mem := vfs.NewMem()
-	// `customFS` delays the sstable with index 0, which is the first flushed sstable.
-	// Delay until writes stall, which we trigger by calling `AsyncFlush()` again while
-	// the first flush is still delayed.
-	customFS := createDelayingFS{
-		FS:             mem,
-		index:          0,
-		createReleased: stallBegan,
-		createDelayed:  nil,
-	}
-	listener := EventListener{
-		WriteStallBegin: func(info WriteStallBeginInfo) {
-			require.EqualValues(t, memTableLimitReason, info.Reason)
-			close(stallBegan)
-		},
-		WriteStallEnd: func() {
-			close(stallEnded)
-		},
-	}
-	d, err := Open("db", &Options{
-		EventListener:               listener,
-		FS:                          &customFS,
-		MemTableStopWritesThreshold: memTableLimit,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
+func TestWriteStallEvents(t *testing.T) {
+	const flushCount = 10
+	const writeStallEnd = "write stall ending"
 
-	for i := 0; i < memTableLimit; i++ {
-		if err := d.Set([]byte("a"), nil, NoSync); err != nil {
-			t.Fatal(err)
-		}
-		_, err := d.AsyncFlush()
-		if err != nil {
-			t.Fatal(err)
-		}
+	testCases := []struct {
+		delayFlush bool
+		expected   string
+	}{
+		{true, "memtable count limit reached"},
+		{false, "L0 file count limit exceeded"},
 	}
-	<-stallBegan
-	<-stallEnded
-}
 
-// Verify `WriteStallBegin()` and `WriteStallEnd()` events for stalls triggered
-// for reaching L0 file count limit.
-func TestWriteStallForL0FileLimit(t *testing.T) {
-	const l0FileLimitReason = "L0 file count limit exceeded"
-	const (
-		l0FileCompactionTrigger = 2
-		l0FileStopTrigger       = 4
-	)
+	for _, c := range testCases {
+		t.Run("", func(t *testing.T) {
+			stallEnded := make(chan struct{}, 1)
+			createReleased := make(chan struct{}, flushCount)
+			var buf syncedBuffer
+			var delayOnce sync.Once
+			listener := EventListener{
+				TableCreated: func(info TableCreateInfo) {
+					if c.delayFlush == (info.Reason == "flushing") {
+						delayOnce.Do(func() {
+							<-createReleased
+						})
+					}
+				},
+				WriteStallBegin: func(info WriteStallBeginInfo) {
+					fmt.Fprintln(&buf, info.String())
+					createReleased <- struct{}{}
+				},
+				WriteStallEnd: func() {
+					fmt.Fprintln(&buf, writeStallEnd)
+					select {
+					case stallEnded <- struct{}{}:
+					default:
+					}
+				},
+			}
+			d, err := Open("db", &Options{
+				EventListener:               listener,
+				FS:                          vfs.NewMem(),
+				MemTableStopWritesThreshold: 2,
+				L0CompactionThreshold:       2,
+				L0StopWritesThreshold:       2,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer d.Close()
 
-	stallBegan := make(chan struct{})
-	stallEnded := make(chan struct{})
-	createDelayed := make(chan struct{})
-	mem := vfs.NewMem()
-	// The indexes of the first few sstables are as follows:
-	//
-	// * index 0: first flushed sstable
-	// * index 1: second flushed sstable
-	// * index 2: first sstable created by L0->Lbase compaction
-	//
-	// `customFS` delays the sstable with index 2, i.e., an Lbase file created during
-	// compaction. Delay until writes stall, which we trigger by calling `Flush()`
-	// repeatedly while the L0->Lbase compaction is still delayed.
-	customFS := createDelayingFS{mem, l0FileCompactionTrigger /* index */, stallBegan, /* createReleased */
-		createDelayed}
-	listener := EventListener{
-		WriteStallBegin: func(info WriteStallBeginInfo) {
-			require.EqualValues(t, l0FileLimitReason, info.Reason)
-			close(stallBegan)
-		},
-		WriteStallEnd: func() {
-			close(stallEnded)
-		},
-	}
-	d, err := Open("db", &Options{
-		EventListener:         listener,
-		FS:                    &customFS,
-		L0CompactionThreshold: l0FileCompactionTrigger,
-		L0StopWritesThreshold: l0FileStopTrigger,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
+			for i := 0; i < flushCount; i++ {
+				if err := d.Set([]byte("a"), nil, NoSync); err != nil {
+					t.Fatal(err)
+				}
+				ch, err := d.AsyncFlush()
+				if err != nil {
+					t.Fatal(err)
+				}
+				// If we're delaying the flush (because we're testing for memtable
+				// write stalls), we can't wait for the flush to finish as doing so
+				// would deadlock. If we're not delaying the flush (because we're
+				// testing for L0 write stals), we wait for the flush to finish so we
+				// don't create too many memtables which would trigger a memtable write
+				// stall.
+				if !c.delayFlush {
+					<-ch
+				}
+				if strings.Contains(buf.String(), c.expected) {
+					break
+				}
+			}
+			<-stallEnded
 
-	// TODO(ajkr): Are the two files pending L0->Lbase not counted towards the
-	// limit? Shouldn't they be?
-	for i := 0; i < l0FileStopTrigger+l0FileCompactionTrigger; i++ {
-		if err := d.Set([]byte("a"), nil, NoSync); err != nil {
-			t.Fatal(err)
-		}
-		d.Flush()
-		if i == l0FileCompactionTrigger-1 {
-			// make sure compaction gets started on the Lbase file before
-			// we proceed to flushing the next sstable (we wouldn't want the
-			// delay to be applied on an L0 file accidentally).
-			<-createDelayed
-		}
+			events := buf.String()
+			require.Contains(t, events, c.expected)
+			require.Contains(t, events, writeStallEnd)
+			if testing.Verbose() {
+				t.Logf("\n%s", events)
+			}
+		})
 	}
-	<-stallBegan
-	<-stallEnded
 }
diff --git a/testdata/event_listener b/testdata/event_listener
index 55b36ef..672674d 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -29,6 +29,7 @@ sync: wal/000002.log
 [JOB 2] WAL created 000005
 [JOB 3] flushing to L0
 create: db/000006.sst
+[JOB 3] flushing: sstable created 000006
 sync: db/000006.sst
 sync: db
 [JOB 3] flushed to L0: 1 (825 B)
@@ -51,6 +52,7 @@ sync: wal/000005.log
 [JOB 4] WAL created 000008 (recycled 000002)
 [JOB 5] flushing to L0
 create: db/000009.sst
+[JOB 5] flushing: sstable created 000009
 sync: db/000009.sst
 sync: db
 [JOB 5] flushed to L0: 1 (825 B)
@@ -64,6 +66,7 @@ sync: db
 [JOB 5] MANIFEST deleted 000007
 [JOB 6] compacting L0 -> L6: 2+0 (1.6 K + 0 B)
 create: db/000011.sst
+[JOB 6] compacting: sstable created 000011
 sync: db/000011.sst
 sync: db
 [JOB 6] compacted L0 -> L6: 2+0 (1.6 K + 0 B) -> 1 (825 B)