Test persistent writer

state/writer.go

@@ -42,16 +42,22 @@ func newWriter(
 	// The memory used by buffers is locked by the kernel which means tha `ulimit -l` command must report at least that
 	// size + margin for other possible locks if needed.
 
-	numOfBuffers := volatileSize / uring.MaxBufferSize
+	numOfBuffers := (volatileSize + uring.MaxBufferSize - 1) / uring.MaxBufferSize
 	if numOfBuffers > uring.MaxNumOfBuffers {
 		return nil, errors.Errorf("number of buffers %d esceeds the maximum allowed number of buffers %d",
 			numOfBuffers, uring.MaxNumOfBuffers)
 	}
 	v := make([]syscall.Iovec, 0, numOfBuffers)
 	for i := range numOfBuffers {
+		bufferSize := uint64(uring.MaxBufferSize)
+		if bufferSize > volatileSize {
+			bufferSize = volatileSize
+		}
+		volatileSize -= bufferSize
+
 		v = append(v, syscall.Iovec{
 			Base: (*byte)(unsafe.Add(volatileOrigin, i*uring.MaxBufferSize)),
-			Len:  uring.MaxBufferSize,
+			Len:  bufferSize,
 		})
 	}
 

state/writer_test.go

@@ -0,0 +1,83 @@
+package state
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/outofforest/quantum/types"
+)
+
+func TestWriter(t *testing.T) {
+	const stateSize = 1000 * types.NodeLength
+
+	requireT := require.New(t)
+
+	s := NewForTest(t, stateSize)
+	volatileAllocator := s.NewVolatileAllocator()
+	persistentAllocator := s.NewPersistentAllocator()
+	writer, err := s.NewPersistentWriter()
+	requireT.NoError(err)
+	t.Cleanup(writer.Close)
+
+	persistentAddresses := make([]types.PersistentAddress, 0, stateSize/types.NodeLength-2)
+	for i := range cap(persistentAddresses) {
+		volatileAddress, err := volatileAllocator.Allocate()
+		requireT.NoError(err)
+		persistentAddress, err := persistentAllocator.Allocate()
+		requireT.NoError(err)
+
+		persistentAddresses = append(persistentAddresses, persistentAddress)
+		b := s.Bytes(volatileAddress)
+		for j := range b {
+			b[j] = byte(i)
+		}
+
+		requireT.NoError(writer.Write(persistentAddress, volatileAddress))
+	}
+	for i := range types.SnapshotID(numOfPersistentSingularityNodes) {
+		singularityNode := s.SingularityNodeRoot(i)
+		b := s.Bytes(singularityNode.VolatileAddress)
+		for j := range b {
+			b[j] = byte(100 + i)
+		}
+
+		requireT.NoError(writer.Write(singularityNode.Pointer.PersistentAddress, singularityNode.VolatileAddress))
+	}
+
+	f := os.NewFile(uintptr(s.persistentFile), "")
+	buf := make([]byte, types.NodeLength)
+	for i, pa := range persistentAddresses {
+		_, err = f.Seek(int64(pa)*types.NodeLength, io.SeekStart)
+		requireT.NoError(err)
+
+		_, err = f.Read(buf)
+		requireT.NoError(err)
+
+		requireT.Equal(bytes.Repeat([]byte{byte(i)}, types.NodeLength), buf)
+	}
+	for i := range types.SnapshotID(numOfPersistentSingularityNodes) {
+		singularityNode := s.SingularityNodeRoot(i)
+		_, err = f.Seek(int64(singularityNode.Pointer.PersistentAddress)*types.NodeLength, io.SeekStart)
+		requireT.NoError(err)
+
+		_, err = f.Read(buf)
+		requireT.NoError(err)
+
+		requireT.Equal(bytes.Repeat([]byte{byte(100 + i)}, types.NodeLength), buf)
+	}
+
+	requireT.NoError(writer.Write(persistentAddresses[0], 2))
+	writer.Close()
+
+	_, err = f.Seek(int64(persistentAddresses[0])*types.NodeLength, io.SeekStart)
+	requireT.NoError(err)
+
+	_, err = f.Read(buf)
+	requireT.NoError(err)
+
+	requireT.Equal(bytes.Repeat([]byte{byte(1)}, types.NodeLength), buf)
+}