Merge branch 'master' into sql-mapper

Dockerfile

@@ -7,11 +7,13 @@ COPY . pilosa
 
 RUN cd pilosa && make install FLAGS="-a -mod=vendor ${BUILD_FLAGS}" ${MAKE_FLAGS}
 
-FROM alpine:3.9.4
+FROM ubuntu:20.10
 
 LABEL maintainer "dev@pilosa.com"
 
-RUN apk add --no-cache curl jq
+RUN apt-get update
+## debug image: RUN apt-get install -y curl htop vim golang tree jq netcat
+RUN apt-get install -y curl jq
 
 COPY --from=builder /go/bin/pilosa /pilosa
 

rbf/wal.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"syscall"
 
 	"github.com/pilosa/pilosa/v2/syswrap"
@@ -25,11 +26,13 @@ import (
 
 // WALSegment represents a single file in the WAL.
 type WALSegment struct {
-	minWALID int64    // base WALID; calculated from path
-	path     string   // path to file
-	w        *os.File // write handle
-	data     []byte   // read-only mmap data
-	pageN    int      // number of written pages
+	mu         sync.RWMutex
+	minWALID   int64    // base WALID; calculated from path
+	path       string   // path to file
+	w          *os.File // write handle
+	data       []byte   // read-only mmap data
+	writeCache []byte   // write buffer
+	pageN      int      // number of written pages
 }
 
 // NewWALSegment returns a new instance of WALSegment for a given path.
@@ -43,20 +46,37 @@ func NewWALSegment(path string) *WALSegment {
 func (s *WALSegment) Path() string { return s.path }
 
 // MinWALID returns the initial WAL ID of the segment. Only available after Open().
-func (s *WALSegment) MinWALID() int64 { return s.minWALID }
+func (s *WALSegment) MinWALID() int64 {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.minWALID
+}
 
 // MaxWALID returns the maximum WAL ID of the segment. Only available after Open().
 func (s *WALSegment) MaxWALID() int64 {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.minWALID + int64(s.pageN) - 1
 }
 
 // PageN returns the number of pages in the segment.
-func (s *WALSegment) PageN() int { return s.pageN }
+func (s *WALSegment) PageN() int {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.pageN
+}
 
 // Size returns the current size of the segment, in bytes.
-func (s *WALSegment) Size() int64 { return int64(s.pageN) * PageSize }
+func (s *WALSegment) Size() int64 {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return int64(s.pageN) * PageSize
+}
 
 func (s *WALSegment) Open() (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	// Extract base WAL ID and validate path.
 	if s.minWALID, err = ParseWALSegmentPath(s.path); err != nil {
 		return err
@@ -107,7 +127,10 @@ func (s *WALSegment) Open() (err error) {
 
 // Close closes the write handle and the read-only mmap.
 func (s *WALSegment) Close() error {
-	if err := s.CloseForWrite(); err != nil {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if err := s.closeForWrite(); err != nil {
 		return err
 	}
 	if s.data != nil {
@@ -121,6 +144,18 @@ func (s *WALSegment) Close() error {
 
 // CloseForWrite closes the write handle, if initialized.
 func (s *WALSegment) CloseForWrite() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.closeForWrite()
+}
+
+func (s *WALSegment) closeForWrite() error {
+	// Ensure write buffer is flushed out.
+	if err := s.sync(); err != nil {
+		return err
+	}
+
+	// Close underlying file writer.
 	if s.w != nil {
 		if err := s.w.Close(); err != nil {
 			return err
@@ -132,12 +167,24 @@ func (s *WALSegment) CloseForWrite() error {
 
 // ReadWALPage reads a single page at the given WAL ID.
 func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
 	// Ensure requested ID is contained in this file.
 	if walID < s.minWALID || walID > s.minWALID+int64(s.pageN) {
 		return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.minWALID, s.pageN)
 	}
 
 	offset := (walID - s.minWALID) * PageSize
+
+	// If offset is within write buffer, return from write buffer.
+	writeBufferOffset := int64((s.pageN * PageSize) - len(s.writeCache))
+	if offset >= writeBufferOffset {
+		buf := s.writeCache[offset-writeBufferOffset:]
+		return buf[:PageSize:PageSize], nil
+	}
+
+	// Otherwise return from on-disk mmap.
 	return s.data[offset : offset+PageSize], nil
 }
 
@@ -145,6 +192,9 @@ func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
 func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) {
 	assert(len(page) == PageSize, "invalid page size: %d", len(page))
 
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	// Initialize write file handle if not yet initialized.
 	if s.w == nil {
 		if s.w, err = os.OpenFile(s.path, os.O_WRONLY, 0666); err != nil {
@@ -161,20 +211,42 @@ func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err er
 		// TODO: Write meta page checksum
 	}
 
-	// Write page at position & increment page count.
-	if _, err := s.w.WriteAt(page, int64(s.pageN*PageSize)); err != nil {
-		return 0, fmt.Errorf("wal segment write: %w", err)
-	}
+	// Append write to write buffer & increment page count.
+	s.writeCache = append(s.writeCache, page...)
 	s.pageN++
 
 	return walID, nil
 }
 
-// Sync flushes all changes to disk.
+// Flush flushes the write buffer to the OS cache.
+func (s *WALSegment) Flush() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.flush()
+}
+
+func (s *WALSegment) flush() error {
+	if _, err := s.w.WriteAt(s.writeCache, int64((s.pageN*PageSize)-len(s.writeCache))); err != nil {
+		return fmt.Errorf("wal segment write: %w", err)
+	}
+	s.writeCache = nil
+	return nil
+}
+
+// Sync flushes the write buffer and invokes a file sync to flush data to disk.
 func (s *WALSegment) Sync() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.sync()
+}
+
+func (s *WALSegment) sync() error {
 	if s.w == nil {
 		return nil
 	}
+	if err := s.flush(); err != nil {
+		return err
+	}
 	return s.w.Sync()
 }
 

rbf/wal_test.go

@@ -16,7 +16,7 @@ package rbf_test
 
 import (
 	"bytes"
-        "encoding/hex"
+	"encoding/hex"
 	"io/ioutil"
 	"math/rand"
 	"os"
@@ -26,7 +26,6 @@ import (
 	"github.com/pilosa/pilosa/v2/rbf"
 )
 
-
 func TestWALSegment_Open(t *testing.T) {
 	t.Run("OK", func(t *testing.T) {
 		s := MustOpenWALSegment(t, 10)
@@ -108,6 +107,47 @@ func TestParseWALSegmentPath(t *testing.T) {
 	})
 }
 
+func BenchmarkWALSegment_WriteWALPage(b *testing.B) {
+	b.Run("8KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 8*(1<<10)) })
+	b.Run("16KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 16*(1<<10)) })
+	b.Run("64KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 64*(1<<10)) })
+	b.Run("256KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 256*(1<<10)) })
+	b.Run("1MB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, (1 << 20)) })
+	b.Run("10MB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 10*(1<<20)) })
+}
+
+func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) {
+	page := make([]byte, rbf.PageSize)
+
+	for i := 0; i < b.N; i++ {
+		func() {
+			s := MustOpenWALSegment(b, 0)
+			defer MustCloseWALSegment(b, s)
+
+			// Fill the segment but stop after each flush interval to flush the write buffer.
+			for j := 0; j < rbf.MaxWALSegmentFileSize; j += rbf.PageSize {
+				if _, err := s.WriteWALPage(page, false); err != nil {
+					b.Fatal(err)
+				}
+
+				// Flush write buffer.
+				if j != 0 && j%flushSize == 0 {
+					if err := s.Flush(); err != nil {
+						b.Fatal(err)
+					}
+				}
+			}
+
+			// Fsync to disk at the end.
+			if err := s.Sync(); err != nil {
+				b.Fatal(err)
+			}
+		}()
+	}
+
+	b.SetBytes(rbf.MaxWALSegmentFileSize)
+}
+
 // MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
 func MustOpenWALSegment(tb testing.TB, walID int64) *rbf.WALSegment {
 	tb.Helper()