add directio for wal writing (#20)

Co-authored-by: Thomas Jungblut <[email protected]>

benchmark/simpledb_test.go

@@ -2,7 +2,9 @@ package benchmark
 
 import (
 	"fmt"
+	"io"
 	"io/ioutil"
+	"log"
 	"math/rand"
 	"os"
 	"runtime"
@@ -18,6 +20,7 @@ import (
 )
 
 func BenchmarkSimpleDBReadLatency(b *testing.B) {
+	log.SetOutput(io.Discard)
 	dbSizes := []int{100, 1000, 10000, 100000}
 
 	for _, n := range dbSizes {
@@ -51,6 +54,7 @@ func BenchmarkSimpleDBReadLatency(b *testing.B) {
 }
 
 func BenchmarkSimpleDBWriteLatency(b *testing.B) {
+	log.SetOutput(io.Discard)
 	dbSizes := []int{100, 1000, 10000, 100000, 1000000}
 
 	for _, n := range dbSizes {

simpledb/db.go

@@ -74,6 +74,7 @@ type DB struct {
 	compactedMaxSizeBytes uint64
 	enableCompactions     bool
 	enableAsyncWAL        bool
+	enableDirectIOWAL     bool
 	open                  bool
 	closed                bool
 
@@ -331,6 +332,7 @@ func NewSimpleDB(basePath string, extraOptions ...ExtraOption) (*DB, error) {
 		MemStoreMaxSizeBytes,
 		true,
 		false,
+		false,
 		NumSSTablesToTriggerCompaction,
 		DefaultCompactionMaxSizeBytes,
 		DefaultCompactionInterval,
@@ -360,6 +362,7 @@ func NewSimpleDB(basePath string, extraOptions ...ExtraOption) (*DB, error) {
 		compactedMaxSizeBytes:       extraOpts.compactionMaxSizeBytes,
 		enableCompactions:           extraOpts.enableCompactions,
 		enableAsyncWAL:              extraOpts.enableAsyncWAL,
+		enableDirectIOWAL:           extraOpts.enableDirectIOWAL,
 		compactionInterval:          extraOpts.compactionRunInterval,
 		closed:                      false,
 		rwLock:                      rwLock,
@@ -379,6 +382,7 @@ type ExtraOptions struct {
 	memstoreSizeBytes       uint64
 	enableCompactions       bool
 	enableAsyncWAL          bool
+	enableDirectIOWAL       bool
 	compactionFileThreshold int
 	compactionMaxSizeBytes  uint64
 	compactionRunInterval   time.Duration
@@ -408,6 +412,13 @@ func EnableAsyncWAL() ExtraOption {
 	}
 }
 
+// EnableDirectIOWAL will turn on the WAL writes using DirectIO, which should give faster aligned block writes and less cache churn.
+func EnableDirectIOWAL() ExtraOption {
+	return func(args *ExtraOptions) {
+		args.enableDirectIOWAL = true
+	}
+}
+
 // CompactionRunInterval configures how often the compaction ticker tries to compact sstables.
 // By default, it's every DefaultCompactionInterval.
 func CompactionRunInterval(interval time.Duration) ExtraOption {

simpledb/recovery.go

@@ -173,14 +173,29 @@ func (db *DB) replayAndSetupWriteAheadLog() error {
 	walBasePath := filepath.Join(db.basePath, WriteAheadFolder)
 	err := os.MkdirAll(walBasePath, 0700)
 	if err != nil {
-		return err
+		return fmt.Errorf("could not mkdir WAL dir at %s: %w", walBasePath, err)
+	}
+
+	writerOpts := []recordio.FileWriterOption{
+		recordio.CompressionType(recordio.CompressionTypeSnappy),
+	}
+	if db.enableDirectIOWAL {
+		ok, err := recordio.IsDirectIOAvailable()
+		if err != nil {
+			return fmt.Errorf("could not detected directIO status: %w", err)
+		}
+		if ok {
+			writerOpts = append(writerOpts, recordio.DirectIO())
+		} else {
+			log.Printf("directIO requested, but not available\n")
+		}
 	}
 
 	walOpts, err := wal.NewWriteAheadLogOptions(wal.BasePath(walBasePath),
-		// we do manual rotation in lockstep with the memstore flushes, thus just set this super high
-		wal.MaximumWalFileSizeBytes(db.memstoreMaxSize*10),
+		// we do manual rotation in lockstep with the memstore flushes, thus just set this super high to not trigger
+		wal.MaximumWalFileSizeBytes(db.memstoreMaxSize*100),
 		wal.WriterFactory(func(path string) (recordio.WriterI, error) {
-			return recordio.NewFileWriter(recordio.Path(path), recordio.CompressionType(recordio.CompressionTypeSnappy))
+			return recordio.NewFileWriter(append(writerOpts, recordio.Path(path))...)
 		}),
 		wal.ReaderFactory(func(path string) (recordio.ReaderI, error) {
 			return recordio.NewFileReaderWithPath(path)