Use mmap for allocations

internal/memstore/allocator.go

@@ -1,7 +1,73 @@
 package memstore
 
-func RequestBytes(len int) []byte {
-	// TODO: Use mmap etc...!
+import (
+	"os"
+	"reflect"
+	"sync"
+	"unsafe"
 
-	return make([]byte, len)
+	"github.com/ClusterCockpit/cc-metric-store/internal/types"
+	"golang.org/x/sys/unix"
+)
+
+const bufferSizeInFloats int = 512
+const bufferSizeInBytes int = bufferSizeInFloats * 8
+
+// The allocator rarely used, so a single big lock should be fine!
+var allocatorLock sync.Mutex
+var allocatorPool [][]byte
+
+func RequestBytes(size int) []byte {
+	size = (size + bufferSizeInBytes - 1) / bufferSizeInBytes * bufferSizeInBytes
+	if size == bufferSizeInBytes {
+		allocatorLock.Lock()
+		if len(allocatorPool) > 0 {
+			bytes := allocatorPool[len(allocatorPool)-1]
+			allocatorPool = allocatorPool[:len(allocatorPool)-1]
+			allocatorLock.Unlock()
+			return bytes
+		}
+		allocatorLock.Unlock()
+	}
+
+	pagesize := os.Getpagesize()
+	if size < pagesize || pagesize%size != 0 {
+		panic("page size and buffer size do not go with each other")
+	}
+
+	bytes, err := unix.Mmap(-1, 0, size, unix.PROT_READ|unix.PROT_WRITE, unix.MAP_ANONYMOUS)
+	if err != nil {
+		panic(err)
+	}
+
+	return bytes
+}
+
+func ReleaseBytes(bytes []byte) {
+	if cap(bytes)%bufferSizeInBytes != 0 {
+		panic("bytes released that do not match the buffer size constraints")
+	}
+
+	allocatorLock.Lock()
+	defer allocatorLock.Unlock()
+
+	n := cap(bytes) / bufferSizeInBytes
+	for i := 0; i < n; i++ {
+		chunk := bytes[i*bufferSizeInBytes : i*bufferSizeInBytes+bufferSizeInBytes]
+		allocatorPool = append(allocatorPool, chunk[0:0:bufferSizeInBytes])
+	}
+}
+
+func ReleaseFloatSlice(slice []types.Float) {
+	var x types.Float
+	sh := (*reflect.SliceHeader)(unsafe.Pointer(&slice))
+	bytes := unsafe.Slice((*byte)(unsafe.Pointer(sh.Data)), sh.Cap*int(unsafe.Sizeof(x)))
+	ReleaseBytes(bytes)
+}
+
+func RequestFloatSlice(size int) []types.Float {
+	var x types.Float
+	bytes := RequestBytes(size * int(unsafe.Sizeof(x)))
+	sh := (*reflect.SliceHeader)(unsafe.Pointer(&bytes))
+	return unsafe.Slice((*types.Float)(unsafe.Pointer(sh.Data)), size)
 }

internal/memstore/checkpoints.go

@@ -0,0 +1,317 @@
+package memstore
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"reflect"
+	"unsafe"
+
+	"github.com/ClusterCockpit/cc-metric-store/internal/types"
+)
+
+const magicValue string = "CCMSv1.0" // MUST be 8 bytes!
+const writeBufferSize int = 8192
+
+// A single checkpoint file is defined by this structure.
+type Checkpoint struct {
+	Reserved []byte           // Reserved for future use/metadata.
+	From, To uint64           // Time covered by this checkpoint.
+	Root     *CheckpointLevel // Root of the internal tree structure.
+}
+
+// A single level in the tree of the in-memory store is defined by this structure.
+type CheckpointLevel struct {
+	Reserved  []byte                      // Reserved for future use/metadata.
+	Metrics   map[string]*Metrics         // Payload...
+	Sublevels map[string]*CheckpointLevel // Child in the internal tree structure.
+}
+
+// Payload/Data
+type Metrics struct {
+	Reserved  []byte        // Reserved for future use/metadata.
+	Frequency uint64        // Frequency of measurements for this metric.
+	From      uint64        // Timestamp of the first measurement of this metric.
+	Data      []types.Float // Payload...
+}
+
+func (c *Checkpoint) Serialize(w io.Writer) error {
+	// The header:
+	buf := make([]byte, 0, writeBufferSize*2)
+	buf = append(buf, magicValue...)
+	buf = encodeBytes(buf, c.Reserved)
+	buf = encodeUint64(buf, c.From)
+	buf = encodeUint64(buf, c.To)
+
+	// The rest:
+	var err error
+	if buf, err = c.Root.serialize(w, buf); err != nil {
+		return err
+	}
+
+	if _, err := w.Write(buf); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *CheckpointLevel) serialize(w io.Writer, buf []byte) ([]byte, error) {
+	// The reserved data:
+	buf = encodeBytes(buf, c.Reserved)
+
+	var err error
+	// metrics:
+	buf = encodeUint32(buf, uint32(len(c.Metrics)))
+	for key, metric := range c.Metrics {
+		buf = encodeString(buf, key)
+		buf, err = metric.serialize(w, buf)
+		if err != nil {
+			return nil, err
+		}
+
+		if len(buf) >= writeBufferSize {
+			if _, err = w.Write(buf); err != nil {
+				return nil, err
+			}
+
+			buf = buf[0:]
+		}
+	}
+
+	// Sublevels:
+	buf = encodeUint32(buf, uint32(len(c.Sublevels)))
+	for key, sublevel := range c.Sublevels {
+		buf = encodeString(buf, key)
+		buf, err = sublevel.serialize(w, buf)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return buf, nil
+}
+
+func (m *Metrics) serialize(w io.Writer, buf []byte) ([]byte, error) {
+	buf = encodeBytes(buf, m.Reserved)           // Reserved
+	buf = encodeUint64(buf, m.Frequency)         // Frequency
+	buf = encodeUint64(buf, m.From)              // First measurmenet timestamp
+	buf = encodeUint32(buf, uint32(len(m.Data))) // Number of elements
+
+	var x types.Float
+	elmsize := unsafe.Sizeof(x)
+	sh := (*reflect.SliceHeader)(unsafe.Pointer(&m.Data))
+	bytes := unsafe.Slice((*byte)(unsafe.Pointer(sh.Data)), sh.Len*int(elmsize))
+
+	buf = append(buf, bytes...)
+	return buf, nil
+}
+
+func (c *Checkpoint) Deserialize(r *bufio.Reader) error {
+	buf := make([]byte, len(magicValue), 16384)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return err
+	}
+
+	if string(buf) != magicValue {
+		return fmt.Errorf("file corrupted: expected the file to start with %#v (got %#v)", magicValue, string(buf))
+	}
+
+	bytes, err := decodeBytes(buf, r)
+	if err != nil {
+		return err
+	}
+	c.Reserved = bytes
+
+	if c.From, err = decodeUint64(buf, r); err != nil {
+		return err
+	}
+	if c.To, err = decodeUint64(buf, r); err != nil {
+		return err
+	}
+
+	c.Root = &CheckpointLevel{}
+	return c.Root.deserialize(buf, r)
+}
+
+func (c *CheckpointLevel) deserialize(buf []byte, r *bufio.Reader) error {
+	var err error
+	if c.Reserved, err = decodeBytes(buf, r); err != nil {
+		return err
+	}
+
+	// Metrics:
+	n, err := decodeUint32(buf, r)
+	if err != nil {
+		return err
+	}
+	if n != 0 {
+		c.Metrics = make(map[string]*Metrics, n)
+		for i := 0; i < int(n); i++ {
+			key, err := decodeString(buf, r)
+			if err != nil {
+				return fmt.Errorf("decoding metric key: %w", err)
+			}
+
+			// log.Printf("decoding metric %#v...\n", key)
+			metrics := &Metrics{}
+			if err := metrics.deserialize(buf, r); err != nil {
+				return fmt.Errorf("decoding metric for key %#v: %w", key, err)
+			}
+
+			c.Metrics[key] = metrics
+		}
+	}
+
+	// Sublevels:
+	n, err = decodeUint32(buf, r)
+	if err != nil {
+		return err
+	}
+	if n != 0 {
+		c.Sublevels = make(map[string]*CheckpointLevel, n)
+		for i := 0; i < int(n); i++ {
+			key, err := decodeString(buf, r)
+			if err != nil {
+				return fmt.Errorf("decoding sublevel key: %w", err)
+			}
+
+			// log.Printf("decoding sublevel %#v...\n", key)
+			sublevel := &CheckpointLevel{}
+			if err := sublevel.deserialize(buf, r); err != nil {
+				return fmt.Errorf("decoding sublevel for key %#v: %w", key, err)
+			}
+
+			c.Sublevels[key] = sublevel
+		}
+	}
+
+	return nil
+}
+
+func (m *Metrics) deserialize(buf []byte, r *bufio.Reader) error {
+	bytes, err := decodeBytes(buf, r)
+	if err != nil {
+		return err
+	}
+	m.Reserved = bytes
+
+	if m.Frequency, err = decodeUint64(buf, r); err != nil {
+		return err
+	}
+	if m.From, err = decodeUint64(buf, r); err != nil {
+		return err
+	}
+
+	n, err := decodeUint32(buf, r)
+	if err != nil {
+		return err
+	}
+
+	var x types.Float
+	elmsize := unsafe.Sizeof(x)
+	bytes = RequestBytes(int(n) * int(elmsize))
+	if _, err := io.ReadFull(r, bytes); err != nil {
+		return fmt.Errorf("reading payload (n=%d, elmsize=%d): %w", n, elmsize, err)
+	}
+
+	sh := (*reflect.SliceHeader)(unsafe.Pointer(&bytes))
+	m.Data = unsafe.Slice((*types.Float)(unsafe.Pointer(sh.Data)), n)
+	return nil
+}
+
+func encodeBytes(buf []byte, bytes []byte) []byte {
+	buf = encodeUint32(buf, uint32(len(bytes)))
+	return append(buf, bytes...)
+}
+
+func encodeString(buf []byte, str string) []byte {
+	buf = encodeUint32(buf, uint32(len(str)))
+	return append(buf, str...)
+}
+
+func encodeUint64(buf []byte, i uint64) []byte {
+	return append(buf,
+		byte((i>>0)&0xff),
+		byte((i>>8)&0xff),
+		byte((i>>16)&0xff),
+		byte((i>>24)&0xff),
+		byte((i>>32)&0xff),
+		byte((i>>40)&0xff),
+		byte((i>>48)&0xff),
+		byte((i>>56)&0xff))
+}
+
+func encodeUint32(buf []byte, i uint32) []byte {
+	return append(buf,
+		byte((i>>0)&0xff),
+		byte((i>>8)&0xff),
+		byte((i>>16)&0xff),
+		byte((i>>24)&0xff))
+}
+
+func decodeBytes(buf []byte, r *bufio.Reader) ([]byte, error) {
+	len, err := decodeUint32(buf, r)
+	if err != nil {
+		return nil, err
+	}
+
+	if len == 0 {
+		return nil, nil
+	}
+
+	bytes := make([]byte, len)
+	if _, err := io.ReadFull(r, bytes); err != nil {
+		return nil, fmt.Errorf("decoding %d bytes: %w", len, err)
+	}
+	return bytes, nil
+}
+
+func decodeString(buf []byte, r *bufio.Reader) (string, error) {
+	len, err := decodeUint32(buf, r)
+	if err != nil {
+		return "", err
+	}
+
+	var bytes []byte
+	if cap(buf) <= int(len) {
+		bytes = buf[0:int(len)]
+	} else {
+		bytes = make([]byte, int(len))
+	}
+
+	if _, err := io.ReadFull(r, bytes); err != nil {
+		return "", err
+	}
+	return string(bytes), nil
+}
+
+func decodeUint64(buf []byte, r *bufio.Reader) (uint64, error) {
+	buf = buf[0:8]
+	_, err := io.ReadFull(r, buf)
+	if err != nil {
+		return 0, err
+	}
+
+	return (uint64(buf[0]) << 0) |
+		(uint64(buf[1]) << 8) |
+		(uint64(buf[2]) << 16) |
+		(uint64(buf[3]) << 24) |
+		(uint64(buf[4]) << 32) |
+		(uint64(buf[5]) << 40) |
+		(uint64(buf[6]) << 48) |
+		(uint64(buf[7]) << 56), nil
+}
+
+func decodeUint32(buf []byte, r *bufio.Reader) (uint32, error) {
+	buf = buf[0:4]
+	_, err := io.ReadFull(r, buf)
+	if err != nil {
+		return 0, err
+	}
+
+	return (uint32(buf[0]) << 0) |
+		(uint32(buf[1]) << 8) |
+		(uint32(buf[2]) << 16) |
+		(uint32(buf[3]) << 24), nil
+}

internal/memstore/checkpoints_test.go

@@ -0,0 +1,109 @@
+package memstore
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"log"
+	"reflect"
+	"testing"
+
+	"github.com/ClusterCockpit/cc-metric-store/internal/types"
+)
+
+func TestIntEncoding(t *testing.T) {
+	buf := make([]byte, 0, 100)
+	x1 := uint64(0x0102030405060708)
+	buf = encodeUint64(buf, x1)
+
+	x2, err := decodeUint64(make([]byte, 0, 10), bufio.NewReader(bytes.NewReader(buf)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if x1 != x2 {
+		t.Fatalf("uint64: x1 != x2: %x != %x", x1, x2)
+	}
+
+	buf = buf[0:0:100]
+	x3 := uint32(0xabcde)
+	buf = encodeUint32(buf, x3)
+
+	x4, err := decodeUint32(make([]byte, 0, 10), bufio.NewReader(bytes.NewReader(buf)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if x3 != x4 {
+		t.Fatalf("uint32: x3 != x4: %x != %x", x3, x4)
+	}
+}
+
+func TestIdentity(t *testing.T) {
+	input := &Checkpoint{
+		Reserved: []byte("Hallo Welt"),
+		From:     1234,
+		To:       5678,
+		Root: &CheckpointLevel{
+			Reserved: nil,
+			Sublevels: map[string]*CheckpointLevel{
+				"host1": {
+					Reserved: []byte("some text..."),
+					Metrics: map[string]*Metrics{
+						"flops": {
+							Reserved:  []byte("blablabla"),
+							Frequency: 60,
+							From:      12345,
+							Data:      []types.Float{1.1, 2.2, 3.3, 4.4, 5.5, 6.6},
+						},
+						"xyz": {
+							Frequency: 10,
+							From:      123,
+							Data:      []types.Float{1.2, 3.4, 5.6, 7.8},
+						},
+					},
+				},
+				"host2": {
+					Sublevels: map[string]*CheckpointLevel{
+						"cpu0": {
+							Metrics: map[string]*Metrics{
+								"abc": {
+									Frequency: 42,
+									From:      420,
+									Data:      []types.Float{1., 2., 3., 4., 5., 6., 7., 8., 9., 10., 11., 12., 13., 14., 15.},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	disk := &bytes.Buffer{}
+	if err := input.Serialize(disk); err != nil {
+		t.Fatal("serialization failed:", err)
+	}
+
+	// fmt.Printf("disk.Len() = %d\n", disk.Len())
+	// fmt.Printf("disk.Bytes() = %#v", disk.Bytes())
+
+	output := &Checkpoint{}
+	if err := output.Deserialize(bufio.NewReader(disk)); err != nil {
+		t.Fatal("deserialization failed:", err)
+	}
+
+	if !reflect.DeepEqual(input, output) {
+		input_json, err := json.Marshal(input)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		output_json, err := json.Marshal(output)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		log.Printf("a: %#v", string(input_json))
+		log.Printf("b: %#v", string(output_json))
+		t.Fatal("x != deserialize(serialize(x))")
+	}
+}