cc-backend/internal/metricstore/buffer.go

// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
// All rights reserved. This file is part of cc-backend.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package metricstore

import (
	"errors"
	"sync"

	"github.com/ClusterCockpit/cc-lib/v2/schema"
)

// BufferCap is the default buffer capacity.
// buffer.data will only ever grow up to its capacity and a new link
// in the buffer chain will be created if needed so that no copying
// of data or reallocation needs to happen on writes.
const BufferCap int = DefaultBufferCapacity

var bufferPool sync.Pool = sync.Pool{
	New: func() any {
		return &buffer{
			data: make([]schema.Float, 0, BufferCap),
		}
	},
}

var (
	ErrNoData           error = errors.New("[METRICSTORE]> no data for this metric/level")
	ErrDataDoesNotAlign error = errors.New("[METRICSTORE]> data from lower granularities does not align")
)

// Each metric on each level has it's own buffer.
// This is where the actual values go.
// If `cap(data)` is reached, a new buffer is created and
// becomes the new head of a buffer list.
type buffer struct {
	prev      *buffer
	next      *buffer
	data      []schema.Float
	frequency int64
	start     int64
	archived  bool
	closed    bool
}

func newBuffer(ts, freq int64) *buffer {
	b := bufferPool.Get().(*buffer)
	b.frequency = freq
	b.start = ts - (freq / 2)
	b.prev = nil
	b.next = nil
	b.archived = false
	b.closed = false
	b.data = b.data[:0]
	return b
}

// If a new buffer was created, the new head is returnd.
// Otherwise, the existing buffer is returnd.
// Normaly, only "newer" data should be written, but if the value would
// end up in the same buffer anyways it is allowed.
func (b *buffer) write(ts int64, value schema.Float) (*buffer, error) {
	if ts < b.start {
		return nil, errors.New("[METRICSTORE]> cannot write value to buffer from past")
	}

	// idx := int((ts - b.start + (b.frequency / 3)) / b.frequency)
	idx := int((ts - b.start) / b.frequency)
	if idx >= cap(b.data) {
		newbuf := newBuffer(ts, b.frequency)
		newbuf.prev = b
		b.next = newbuf
		b = newbuf
		idx = 0
	}

	// Overwriting value or writing value from past
	if idx < len(b.data) {
		b.data[idx] = value
		return b, nil
	}

	// Fill up unwritten slots with NaN
	for i := len(b.data); i < idx; i++ {
		b.data = append(b.data, schema.NaN)
	}

	b.data = append(b.data, value)
	return b, nil
}

func (b *buffer) end() int64 {
	return b.firstWrite() + int64(len(b.data))*b.frequency
}

func (b *buffer) firstWrite() int64 {
	return b.start + (b.frequency / 2)
}

// Return all known values from `from` to `to`. Gaps of information are represented as NaN.
// Simple linear interpolation is done between the two neighboring cells if possible.
// If values at the start or end are missing, instead of NaN values, the second and thrid
// return values contain the actual `from`/`to`.
// This function goes back the buffer chain if `from` is older than the currents buffer start.
// The loaded values are added to `data` and `data` is returned, possibly with a shorter length.
// If `data` is not long enough to hold all values, this function will panic!
func (b *buffer) read(from, to int64, data []schema.Float) ([]schema.Float, int64, int64, error) {
	if from < b.firstWrite() {
		if b.prev != nil {
			return b.prev.read(from, to, data)
		}
		from = b.firstWrite()
	}

	i := 0
	t := from
	for ; t < to; t += b.frequency {
		idx := int((t - b.start) / b.frequency)
		if idx >= cap(b.data) {
			if b.next == nil {
				break
			}
			b = b.next
			idx = 0
		}

		if idx >= len(b.data) {
			if b.next == nil || to <= b.next.start {
				break
			}
			data[i] += schema.NaN
		} else if t < b.start {
			data[i] += schema.NaN
		} else {
			data[i] += b.data[idx]
		}
		i++
	}

	return data[:i], from, t, nil
}

// Returns true if this buffer needs to be freed.
func (b *buffer) free(t int64) (delme bool, n int) {
	if b.prev != nil {
		delme, m := b.prev.free(t)
		n += m
		if delme {
			b.prev.next = nil
			if cap(b.prev.data) == BufferCap {
				bufferPool.Put(b.prev)
			}
			b.prev = nil
		}
	}

	end := b.end()
	if end < t {
		return true, n + 1
	}

	return false, n
}

// forceFreeOldest recursively finds the end of the linked list (the oldest buffer)
// and removes it.
// Returns:
//
//	delme: true if 'b' itself is the oldest and should be removed by the caller
//	n:     the number of buffers freed (will be 1 or 0)
func (b *buffer) forceFreeOldest() (delme bool, n int) {
	// If there is a previous buffer, recurse down to find the oldest
	if b.prev != nil {
		delPrev, freed := b.prev.forceFreeOldest()

		// If the previous buffer signals it should be deleted:
		if delPrev {
			// Unlink references
			b.prev.next = nil

			// Return to pool if capacity matches
			if cap(b.prev.data) == BufferCap {
				bufferPool.Put(b.prev)
			}

			// Remove the link from the current buffer
			b.prev = nil
		}
		return false, freed
	}

	// If b.prev is nil, THIS buffer is the oldest.
	// We return true so the parent (or the Level loop) knows to delete reference to 'b'.
	return true, 1
}

// Call `callback` on every buffer that contains data in the range from `from` to `to`.
func (b *buffer) iterFromTo(from, to int64, callback func(b *buffer) error) error {
	if b == nil {
		return nil
	}

	if err := b.prev.iterFromTo(from, to, callback); err != nil {
		return err
	}

	if from <= b.end() && b.start <= to {
		return callback(b)
	}

	return nil
}

func (b *buffer) count() int64 {
	res := int64(len(b.data))
	if b.prev != nil {
		res += b.prev.count()
	}
	return res
}