LocalAI/backend/go/timestral-cpp/gotimestralcpp.go

package main

import (
	"fmt"
	"os"
	"path/filepath"
	"runtime"
	"sync"
	"unsafe"

	"github.com/mudler/LocalAI/pkg/grpc/base"
	pb "github.com/mudler/LocalAI/pkg/grpc/proto"
	"github.com/mudler/xlog"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// TimestralCpp implements the Forecast RPC of the Backend gRPC service over
// libtimestral.so. The C handle is not reentrant: the server serializes calls
// through base.SingleThread, and mu repeats that guard for Forecast and Free so
// the handle is safe even when the type is used without the server.
type TimestralCpp struct {
	base.SingleThread

	mu     sync.Mutex
	ctxPtr uintptr
	opts   loadOptions
	levels []float64 // quantile levels of the loaded model, ascending
}

func (t *TimestralCpp) Load(opts *pb.ModelOptions) error {
	model := opts.ModelFile
	if model == "" {
		model = opts.ModelPath
	}
	if !filepath.IsAbs(model) && opts.ModelPath != "" && model != opts.ModelPath {
		model = filepath.Join(opts.ModelPath, model)
	}
	if model == "" {
		return fmt.Errorf("timestral-cpp: ModelFile is required")
	}
	if st, err := os.Stat(model); err != nil || st.IsDir() {
		return fmt.Errorf("timestral-cpp: model file %q is not readable", model)
	}

	if v := CppAbiVersion(); v != abiVersionRequired {
		return fmt.Errorf("timestral-cpp: libtimestral ABI version %d, this backend needs %d", v, abiVersionRequired)
	}

	threads := opts.Threads
	if threads <= 0 {
		threads = defaultThreads
	}

	t.mu.Lock()
	defer t.mu.Unlock()
	t.freeLocked()

	// last_error(NULL) is thread-local, and a goroutine may move to another OS
	// thread between two calls into C. Pin the thread across the load and the
	// error read.
	runtime.LockOSThread()
	ctx := CppLoad(model, threads)
	if ctx == 0 {
		msg := CppLastError(0)
		runtime.UnlockOSThread()
		return fmt.Errorf("timestral-cpp: load %q failed: %s", filepath.Base(model), msg)
	}
	runtime.UnlockOSThread()

	n := int(CppNumQuantiles(ctx))
	if n <= 0 {
		msg := CppLastError(ctx)
		CppFree(ctx)
		return fmt.Errorf("timestral-cpp: model reports no quantile levels: %s", msg)
	}
	levels := make([]float64, n)
	for i := range levels {
		levels[i] = CppQuantileLevel(ctx, int32(i))
	}

	t.ctxPtr = ctx
	t.levels = levels
	t.opts = parseOptions(opts.Options)
	xlog.Info("timestral-cpp: model loaded", "model", filepath.Base(model), "threads", threads,
		"quantile_levels", len(levels), "abi", abiVersionRequired)
	return nil
}

func (t *TimestralCpp) Free() error {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.freeLocked()
	return nil
}

func (t *TimestralCpp) freeLocked() {
	if t.ctxPtr != 0 {
		CppFree(t.ctxPtr)
		t.ctxPtr = 0
	}
	t.levels = nil
}

// Forecast groups the series by shape, runs one ABI call per group and returns
// the results in the order the series were sent.
func (t *TimestralCpp) Forecast(req *pb.ForecastRequest) (pb.ForecastResponse, error) {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.ctxPtr == 0 {
		return pb.ForecastResponse{}, status.Error(codes.FailedPrecondition, "timestral-cpp: no model loaded")
	}
	if req.GetPredictionLength() < 1 {
		return pb.ForecastResponse{}, invalidArgument("prediction_length must be at least 1")
	}
	if req.GetMaxContext() < 0 {
		return pb.ForecastResponse{}, invalidArgument("max_context must not be negative")
	}
	h := req.GetPredictionLength()
	if err := validateSeries(req.GetSeries(), h, t.opts); err != nil {
		return pb.ForecastResponse{}, err
	}
	sel, levels, err := selectQuantiles(req.GetQuantiles(), t.opts.defaultQuantiles, t.levels)
	if err != nil {
		return pb.ForecastResponse{}, err
	}

	maxContext := req.GetMaxContext()
	if maxContext == 0 {
		maxContext = t.opts.maxContext
	}

	results := make([]*pb.ForecastSeriesResult, len(req.GetSeries()))
	for _, g := range planGroups(req.GetSeries()) {
		out, err := t.forecastGroup(req, g, h, maxContext)
		if err != nil {
			return pb.ForecastResponse{}, err
		}
		for i, r := range reshape(out, req.GetSeries(), g, int(h), len(t.levels), sel) {
			results[i] = r
		}
	}

	return pb.ForecastResponse{PredictionLength: h, QuantileLevels: levels, Series: results}, nil
}

// flagValue maps an optional bool to the ABI flag: -1 model default, 0 or 1.
func flagValue(b *bool) int32 {
	switch {
	case b == nil:
		return -1
	case *b:
		return 1
	default:
		return 0
	}
}

func floatPtr(s []float32) unsafe.Pointer {
	if len(s) == 0 {
		return nil
	}
	return unsafe.Pointer(&s[0])
}

func doublePtr(s []float64) unsafe.Pointer {
	if len(s) == 0 {
		return nil
	}
	return unsafe.Pointer(&s[0])
}

// forecastGroup runs one ABI call and returns a Go copy of the [B,V,H,Q]
// output. The C buffer is freed before it returns.
func (t *TimestralCpp) forecastGroup(req *pb.ForecastRequest, g group, h, maxContext int32) ([]float32, error) {
	b := buildBatch(req.GetSeries(), g)
	cr := forecastRequest{
		context:       floatPtr(b.context),
		B:             int32(len(g.idx)),
		V:             g.shape.V,
		T:             g.shape.T,
		predictionLen: h,
		pastCov:       floatPtr(b.pastCov),
		cPast:         g.shape.cPast,
		knownCov:      floatPtr(b.knownCov),
		cKnown:        g.shape.cKnown,
		catPast:       doublePtr(b.catPast),
		ccPast:        g.shape.ccPast,
		catKnown:      doublePtr(b.catKnown),
		ccKnown:       g.shape.ccKnown,
		symmetricAvg:  flagValue(req.SymmetricAveraging),
		clipNonneg:    flagValue(req.ClipNonnegative),
		maxContext:    maxContext,
	}

	var q int32
	ptr := CppForecast(t.ctxPtr, unsafe.Pointer(&cr), unsafe.Pointer(&q))
	// The buffers behind cr must outlive the call: they are only reachable from
	// the C side while it runs.
	runtime.KeepAlive(b)
	runtime.KeepAlive(&cr)
	if ptr == 0 {
		// The ABI rejects what it cannot serve (for example covariates the model
		// has no input for), so the message is about the request.
		return nil, invalidArgument("forecast failed: %s", CppLastError(t.ctxPtr))
	}
	defer CppFreeBuffer(ptr)

	if int(q) != len(t.levels) {
		return nil, status.Errorf(codes.Internal, "timestral-cpp: library returned %d quantiles, model reports %d", q, len(t.levels))
	}
	n := len(g.idx) * int(g.shape.V) * int(h) * int(q)
	src := unsafe.Slice((*float32)(unsafe.Pointer(ptr)), n) //nolint:govet // C-owned malloc'd buffer, not Go-GC memory; freed by CppFreeBuffer above
	out := make([]float32, n)
	copy(out, src)
	return out, nil
}