feat: parameterize WAL (#113)

* feat: parameterize WAL
* perf: optimize flush timer
* fix: wakeup inflight append after close wal
* fix: wakeup typo
* fix: close log file
* test: wal config

Author: ifplusor

internal/store/wal/block/allocator.go

@@ -0,0 +1,73 @@
+// Copyright 2022 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package block
+
+import (
+	// standard libraries.
+	"sync"
+
+	// third-party libraries.
+	"github.com/ncw/directio"
+)
+
+type Allocator struct {
+	size     int
+	next     int64
+	emptyBuf []byte
+	pool     sync.Pool
+}
+
+func NewAllocator(size int, so int64) *Allocator {
+	a := &Allocator{
+		size:     size,
+		next:     so,
+		emptyBuf: make([]byte, size),
+	}
+	a.pool = sync.Pool{
+		New: func() interface{} {
+			return a.rawAlloc()
+		},
+	}
+	return a
+}
+
+func (a *Allocator) BlockSize() int {
+	return a.size
+}
+
+func (a *Allocator) rawAlloc() *Block {
+	buf := directio.AlignedBlock(a.size)
+	return &Block{
+		block: block{
+			buf: buf,
+		},
+	}
+}
+
+func (a *Allocator) Next() *Block {
+	b, _ := a.pool.Get().(*Block)
+	// Reset block.
+	copy(b.buf, a.emptyBuf)
+	b.wp = 0
+	b.fp = 0
+	b.cp = 0
+	b.SO = a.next
+	a.next += int64(a.size)
+	return b
+}
+
+func (a *Allocator) Free(b *Block) {
+	a.pool.Put(b)
+}

internal/store/wal/block/allocator_test.go

@@ -0,0 +1,60 @@
+// Copyright 2022 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package block
+
+import (
+	// standard libraries.
+	"testing"
+
+	// third-party libraries.
+	. "github.com/smartystreets/goconvey/convey"
+	// this project.
+)
+
+var emptyBuf = make([]byte, blockSize)
+
+func TestAllocator(t *testing.T) {
+	Convey("wal block allocator", t, func() {
+		allocator := NewAllocator(blockSize, blockSize)
+		So(allocator, ShouldNotBeNil)
+		So(allocator.BlockSize(), ShouldEqual, blockSize)
+
+		b0 := allocator.Next()
+		So(b0.Capacity(), ShouldEqual, blockSize)
+		So(b0.Size(), ShouldEqual, 0)
+		So(b0.Committed(), ShouldEqual, 0)
+		So(b0.SO, ShouldEqual, blockSize)
+		So(b0.fp, ShouldEqual, 0)
+		So(b0.buf, ShouldResemble, emptyBuf)
+
+		b1 := allocator.Next()
+		So(b1.Capacity(), ShouldEqual, blockSize)
+		So(b1.Size(), ShouldEqual, 0)
+		So(b1.Committed(), ShouldEqual, 0)
+		So(b1.SO, ShouldEqual, 2*blockSize)
+		So(b1.fp, ShouldEqual, 0)
+		So(b1.buf, ShouldResemble, emptyBuf)
+
+		allocator.Free(b0)
+
+		b2 := allocator.Next()
+		So(b2.Capacity(), ShouldEqual, blockSize)
+		So(b2.Size(), ShouldEqual, 0)
+		So(b2.Committed(), ShouldEqual, 0)
+		So(b2.SO, ShouldEqual, 3*blockSize)
+		So(b2.fp, ShouldEqual, 0)
+		So(b2.buf, ShouldResemble, emptyBuf)
+	})
+}

internal/store/wal/block.go renamed to internal/store/wal/block/block.go

@@ -12,9 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package wal
+package block
 
 import (
+	// standard libraries.
+	"os"
+
 	// this project.
 	"github.com/linkall-labs/vanus/internal/store/wal/io"
 	"github.com/linkall-labs/vanus/internal/store/wal/record"
@@ -40,6 +43,10 @@ func (b *block) Size() int {
 	return b.wp
 }
 
+func (b *block) Committed() int {
+	return b.cp
+}
+
 func (b *block) Remaining() int {
 	return b.remaining(b.Size())
 }
@@ -52,7 +59,7 @@ func (b *block) Full() bool {
 	return b.Remaining() < record.HeaderSize
 }
 
-func (b *block) full(off int) bool {
+func (b *block) FullWithOff(off int) bool {
 	return b.remaining(off) < record.HeaderSize
 }
 
@@ -84,3 +91,19 @@ func (b *block) Flush(writer io.WriterAt, offset int, base int64, cb FlushCallba
 		}
 	})
 }
+
+func (b *block) RecoverFromFile(f *os.File, at int64, committed int) error {
+	if _, err := f.ReadAt(b.buf, at); err != nil {
+		return err
+	}
+	b.wp = committed
+	b.fp = committed
+	b.cp = committed
+	return nil
+}
+
+type Block struct {
+	block
+	// SO is start offset
+	SO int64
+}

internal/store/wal/block/block_test.go

@@ -0,0 +1,94 @@
+// Copyright 2022 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package block
+
+import (
+	// standard libraries.
+	"testing"
+
+	// third-party libraries.
+	. "github.com/smartystreets/goconvey/convey"
+
+	// this project.
+	"github.com/linkall-labs/vanus/internal/store/wal/io"
+	"github.com/linkall-labs/vanus/internal/store/wal/record"
+)
+
+const (
+	blockSize = 4 * 1024
+)
+
+func TestBlock(t *testing.T) {
+	Convey("wal block", t, func() {
+		b := Block{
+			block: block{
+				buf: make([]byte, blockSize),
+			},
+		}
+
+		Convey("properties", func() {
+			So(b.Capacity(), ShouldEqual, blockSize)
+			So(b.Size(), ShouldEqual, 0)
+			So(b.Committed(), ShouldEqual, 0)
+			So(b.Remaining(), ShouldEqual, blockSize)
+			So(b.Full(), ShouldBeFalse)
+		})
+
+		Convey("append", func() {
+			dataSize := blockSize - record.HeaderSize*2 + 1
+			records := record.Pack(make([]byte, dataSize), dataSize+record.HeaderSize, 0)
+			r := records[0]
+
+			n, err := b.Append(r)
+
+			So(err, ShouldBeNil)
+			So(n, ShouldEqual, r.Size())
+			So(b.Full(), ShouldBeTrue)
+			So(b.FullWithOff(n-1), ShouldBeFalse)
+			So(b.Committed(), ShouldEqual, 0)
+
+			Convey("flush", func() {
+				b.Flush(io.WriteAtFunc(func(b []byte, base int64,
+					cb io.WriteCallback,
+				) {
+					So(b[:n], ShouldResemble, r.Marshal())
+					So(base, ShouldEqual, 0)
+
+					cb(len(b), nil)
+				}), n, 0, func(commit int64, err error) {
+					So(err, ShouldBeNil)
+					So(commit, ShouldEqual, n)
+				})
+
+				So(b.Committed(), ShouldEqual, n)
+
+				b.Flush(nil, 0, 0, func(commit int64, err error) {
+					So(err, ShouldBeNil)
+					So(commit, ShouldEqual, n)
+				})
+			})
+		})
+
+		Convey("append with too large data", func() {
+			dataSize := blockSize - record.HeaderSize + 1
+			records := record.Pack(make([]byte, dataSize), dataSize+record.HeaderSize, 0)
+			r := records[0]
+
+			_, err := b.Append(r)
+
+			So(err, ShouldNotBeNil)
+		})
+	})
+}