diff --git a/Dockerfile b/Dockerfile index a30097e1e..a0950ffed 100644 --- a/Dockerfile +++ b/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 diff --git a/rbf/wal.go b/rbf/wal.go index 0f8461e16..aee4b07a3 100644 --- a/rbf/wal.go +++ b/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() } diff --git a/rbf/wal_test.go b/rbf/wal_test.go index 4511578c8..b41e7e792 100644 --- a/rbf/wal_test.go +++ b/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()