qb/worker/metric.go

package worker

import (
	"errors"
	"io"
	"time"

	bin "gordenko.dev/dima/bin/little"
	"gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/inbox"
	"gordenko.dev/dima/qb/proto"
	"gordenko.dev/dima/qb/storage"
)

// METRIC

var ErrNoValueBug = errors.New("has timestamp but no value")

type CapturedState struct {
	LastValue float64
}

type Metric struct {
	metricType      qb.MetricType
	fracDigits      byte
	lastPageNo      uint32
	lastValue       float64
	buffer          []byte
	timestamps      qb.TimestampCompressor
	values          qb.ValueCompressor
	xLock           bool
	rLocks          int
	waitQueue       []any
	indexLevelTails []storage.IndexLevelTail // root - last element
	capturedState   *CapturedState
}

// індекси
// Metric encode format:
// metricID - 4b
// metricType - 1b
// fracDigits - 1b
// lastPageNo - 4b
// timestamps size - 2b
// values size - 2b
// timestams payload - Nb
// values payload - Nb
// index levels count - varsize
// [
//    records qty - varsize
//    records - Nb
// ]

func (s *Metric) WriteTo(w io.Writer) (err error) {
	_, err = w.Write([]byte{
		byte(s.metricType),
		s.fracDigits,
	})
	if err != nil {
		return
	}
	err = bin.WriteUint32(w, s.lastPageNo)
	if err != nil {
		return
	}
	err = bin.WriteUint16(w, uint16(s.timestamps.CommitedSize()))
	if err != nil {
		return
	}
	err = bin.WriteUint16(w, uint16(s.values.CommitedSize()))
	if err != nil {
		return
	}
	// timestamps payload
	err = s.timestamps.WriteCommitedTo(w)
	if err != nil {
		return
	}
	// values payload
	err = s.values.WriteCommitedTo(w)
	if err != nil {
		return
	}
	// indexes
	_, err = bin.WriteVarSize(w, len(s.indexLevelTails))
	if err != nil {
		return
	}
	for _, level := range s.indexLevelTails {
		_, err = bin.WriteVarSize(w, level.RecordsCount)
		if err != nil {
			return
		}
		_, err = w.Write(level.Buffer[:level.RecordsCount*storage.IndexRecordSize])
		if err != nil {
			return
		}
	}
	return
}

func (s *Metric) MetricType() qb.MetricType {
	return s.metricType
}

func (s *Metric) FracDigits() byte {
	return s.fracDigits
}

func (s *Metric) LastValue() float64 {
	if s.capturedState != nil {
		return s.capturedState.LastValue
	}
	return s.lastValue
}

// REQUESTS

func (s *Metric) ReleaseRLock() {
	s.rLocks--
	if s.rLocks == 0 {
		if len(s.waitQueue) > 0 {
			//s.ProcessQueue()
		}
	}
}

type GetMetricResult struct {
	MetricType qb.MetricType
	FracDigits byte
	ResultCode byte
}

type GetMetricReq struct {
	MetricID uint32
	ResultCh chan GetMetricResult
}

func (s *Metric) GetMetric(req GetMetricReq) {
	req.ResultCh <- GetMetricResult{
		ResultCode: Succeed,
		MetricType: s.metricType,
		FracDigits: s.fracDigits,
	}
}

func (s *Metric) DeleteMetric(req DeleteMetricReq) {
	if s.xLock || s.rLocks > 0 || s.capturedState != nil {
		s.waitQueue = append(s.waitQueue, req)
	} else {
		s.xLock = true
		// fix - has pages -> do query from goroutine
		// else push to storage
		// collect all pages, than
		// s.storageInbox.Push(storage.MetricDelete{
		// 	MetricID:       req.MetricID,
		// 	FreeIndexPages: nil,
		// 	FreeDataPages:  nil,
		// 	ResultCh:       req.ResultCh,
		// })
	}
}

type AppendMeasuresReq struct {
	MetricID uint32
	Measures []proto.Measure
	ResultCh chan storage.MeasuresAppendResult
}

func (s *Metric) AppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox *inbox.Inbox) {
	if s.xLock || s.capturedState != nil {
		s.waitQueue = append(s.waitQueue, req)
		return
	}
	var (
		timestamps = s.timestamps
		values     = s.values

		// офсети на head сторінці
		timestampsOffset       int
		timestampsRewindOffset int
		valuesOffset           int
		valuesRewindOffset     int

		headTimestamps []byte
		headValues     []byte

		pages []storage.DataPayload

		written    int
		resultCode byte = Succeed
	)

	s.capturedState = &CapturedState{
		LastValue: s.lastValue,
	}

	s.timestamps.CaptureState()
	s.values.CaptureState()

	for idx, measure := range req.Measures {
		//fmt.Println(idx, measure.Timestamp)
		if measure.Timestamp <= s.timestamps.Until() {
			//fmt.Printf("timestamp %d <= until %d\n", measure.Timestamp, s.timestamps.Until())
			resultCode = ExpiredMeasure
			break
		}
		if s.metricType == qb.Cumulative && measure.Value < s.lastValue {
			resultCode = NonMonotonicValue
			break
		}

		//fmt.Printf("  %d: %s => %.2f\n", idx, formatTime(measure.Timestamp), measure.Value)

		tReport := timestamps.Evaluate(tmp[:7], measure.Timestamp)
		//fmt.Printf("tReport: %#v\n", tReport)
		vReport := values.Evaluate(tmp[7:], measure.Value)
		//fmt.Printf("vReport: %#v\n", vReport)

		totalRequiredSpace := tReport.TotalSpace + vReport.TotalSpace

		if totalRequiredSpace <= storage.DataPagePayloadSize {
			// якщо на сторінці є місце
			timestamps.Append(tReport.RewindOffset, tmp[:tReport.ChangeSize], measure.Timestamp)
			values.Append(vReport.RewindOffset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta)

			if idx == 0 {
				timestampsOffset = tReport.Offset
				timestampsRewindOffset = tReport.RewindOffset
				valuesOffset = vReport.Offset
				valuesRewindOffset = vReport.RewindOffset
			}
		} else {
			//fmt.Println("PAGE FILLED")
			//fmt.Println(timestamps.Size() + values.Size())
			// сторінка заповнена
			since := s.timestamps.ReplaceSinceWithUntil()

			if len(pages) == 0 && idx > 0 {
				// idx > 0 required because page may overflows without append any data
				headTimestamps = timestamps.Tail(timestampsOffset)
				headValues = values.Tail(valuesOffset)
			}

			pages = append(pages, storage.DataPayload{
				Since:          since,
				Content:        s.buffer,
				TimestampsSize: timestamps.Size(),
				ValuesSize:     values.Size(),
			})

			//xxx := timestamps.Tail(0)
			//fmt.Printf("PAGE timestamps %d:\n% x\n", len(xxx), xxx)

			buf := make([]byte, storage.DataPageSize)
			databuf := buf[:storage.DataPagePayloadSize]

			timestamps.ReplaceBuffer(databuf)
			values.ReplaceBuffer(databuf)

			// renew
			s.buffer = buf

			tReport = timestamps.Evaluate(tmp[:7], measure.Timestamp)
			//fmt.Printf("tReport: %#v\n", tReport)
			vReport = values.Evaluate(tmp[7:], measure.Value)
			//fmt.Printf("vReport: %#v\n", vReport)

			timestamps.Append(tReport.RewindOffset, tmp[:tReport.ChangeSize], measure.Timestamp)
			values.Append(vReport.RewindOffset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta)
		}
		//
		s.lastValue = measure.Value
		written++
	}

	if written == 0 {
		s.capturedState = nil
		req.ResultCh <- storage.MeasuresAppendResult{
			ResultCode: resultCode,
		}
		return
	}

	//xxx := timestamps.Tail(0)
	//fmt.Printf("PAGE TAIL timestamps %d:\n% x\n", len(xxx), xxx)

	// виділити змінені байти.
	// скопіювати. Причому можна скопіювати зрізи chunks

	if len(pages) > 0 {
		//fmt.Printf("WITH GROW")
		// пишу в storage довгим шляхом через redo файл і запис в data файл
		storageInbox.Push(storage.MeasuresAppendWithGrow{
			MetricID:               req.MetricID,
			LastPageNo:             s.lastPageNo,
			TimestampsRewindOffset: timestampsRewindOffset,
			Timestamps:             headTimestamps,
			ValuesRewindOffset:     valuesRewindOffset,
			Values:                 headValues,
			IndexLevelTails:        s.indexLevelTails,
			DataPages:              pages,
			TailTimestamps:         timestamps.Tail(0), // payload
			TailValues:             values.Tail(0),
			ResultCode:             resultCode,
			WrittenCount:           written,
			ResultCh:               req.ResultCh,
		})
	} else {
		//fmt.Printf("SIMPLE")
		// короткий шлях - запис лише в storage
		storageInbox.Push(storage.MeasuresAppend{
			MetricID:               req.MetricID,
			TimestampsRewindOffset: timestampsRewindOffset,
			ValuesRewindOffset:     valuesRewindOffset,
			Timestamps:             timestamps.Tail(timestampsOffset),
			Values:                 values.Tail(valuesOffset),
			ResultCode:             resultCode,
			WrittenCount:           written,
			ResultCh:               req.ResultCh,
		})
	}
}

type DeleteMeasuresReq struct {
	MetricID uint32
	Since    uint32
	ResultCh chan byte
}

func (s *Metric) DeleteMeasures(req DeleteMeasuresReq) {
	// if s.xLock || s.capturedState != nil {
	// 	s.waitQueue = append(s.waitQueue, req)
	// 	return
	// }
	// since := s.timestamps.FirstTimestamp()
	// until := s.timestamps.LastTimestamp()
	// if since == 0 || (req.Since > 0 && until < req.Since) {
	// 	req.ResultCh <- NoMeasuresToDelete
	// }
	// if s.RootPageNo > 0 {
	// 	req.ResultCh <- tryDeleteMeasuresResult{
	// 		ResultCode: DeleteFromAtreeRequired,
	// 		RootPageNo: metric.RootPageNo,
	// 	}
	// } else {
	// 	req.ResultCh <- tryDeleteMeasuresResult{
	// 		ResultCode: DeleteFromAtreeNotNeeded,
	// 	}
	// }
}

type RangeScanResult struct {
	ResultCode byte
	FracDigits byte
	LastPageNo uint32
	IsDataPage bool // for UntilNotFound only
}

type RangeScanReq struct {
	MetricID       uint32
	Since          uint32
	Until          uint32
	MetricType     qb.MetricType
	ResponseWriter qb.WorkerMeasureConsumer
	ResultCh       chan RangeScanResult
}

func (s *Metric) RangeScan(req RangeScanReq) {
	if s.xLock {
		s.waitQueue = append(s.waitQueue, req)
		return
	}
	since := s.timestamps.CommitedSince()
	if since == 0 {
		req.ResultCh <- RangeScanResult{
			ResultCode: QueryDone,
		}
		return
	}
	//  range after
	if req.Since > s.timestamps.CommitedUntil() {
		req.ResultCh <- RangeScanResult{
			ResultCode: QueryDone,
		}
		return
	}
	// range before
	if req.Until < since {
		if len(s.indexLevelTails) > 0 {
			pageNo, isDataPage := storage.FindPageOnIndexLevelTails(s.indexLevelTails, req.Until)
			if pageNo > 0 {
				req.ResultCh <- RangeScanResult{
					ResultCode: UntilNotFound,
					FracDigits: s.fracDigits,
					LastPageNo: pageNo,
					IsDataPage: isDataPage,
				}
				s.rLocks++
				return
			} else {
				// range until before 1st measure timestamp
				req.ResultCh <- RangeScanResult{
					ResultCode: QueryDone,
				}
				return
			}
		} else {
			req.ResultCh <- RangeScanResult{
				ResultCode: QueryDone,
			}
			return
		}
	}

	timestampDecompressor := s.timestamps.CreateDecompressor()
	valueDecompressor := s.values.CreateDecompressor(s.metricType, s.fracDigits)

	for {
		timestamp, done := timestampDecompressor.NextValue()
		if done {
			break
		}
		value, done := valueDecompressor.NextValue()
		if done {
			qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug)
		}
		if timestamp <= req.Until {
			req.ResponseWriter.FeedNoSend(timestamp, value)
			if timestamp < req.Since {
				req.ResultCh <- RangeScanResult{
					ResultCode: QueryDone,
				}
				return
			}
		}
	}
	if s.lastPageNo > 0 {
		req.ResultCh <- RangeScanResult{
			ResultCode: UntilFound,
			LastPageNo: s.lastPageNo,
			FracDigits: s.fracDigits,
		}
		s.rLocks++
	} else {
		req.ResultCh <- RangeScanResult{
			ResultCode: QueryDone,
		}
	}
}

type FullScanResult struct {
	ResultCode byte
	FracDigits byte
	LastPageNo uint32
}

type FullScanReq struct {
	MetricID       uint32
	MetricType     qb.MetricType
	ResponseWriter qb.WorkerMeasureConsumer
	ResultCh       chan FullScanResult
}

func (s *Metric) FullScan(req FullScanReq) {
	//fmt.Println("lastPageNo:", s.lastPageNo)
	//fmt.Printf("index: %#v\n", s.indexLevelTails)

	if s.xLock {
		s.waitQueue = append(s.waitQueue, req)
		return
	}
	if s.timestamps.CommitedSize() == 0 {
		req.ResultCh <- FullScanResult{
			ResultCode: QueryDone,
		}
		return
	}
	timestampDecompressor := s.timestamps.CreateDecompressor()
	valueDecompressor := s.values.CreateDecompressor(s.metricType, s.fracDigits)
	idx := 0
	for {
		timestamp, done := timestampDecompressor.NextValue()
		if done {
			//fmt.Println("DONE")
			break
		}
		value, done := valueDecompressor.NextValue()
		if done {
			qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug)
		}
		req.ResponseWriter.FeedNoSend(timestamp, value)
		//fmt.Printf("  %d: %s => %.2f\n", idx, formatTime(timestamp), value)
		idx++
	}

	if s.lastPageNo > 0 {
		req.ResultCh <- FullScanResult{
			ResultCode: UntilFound,
			LastPageNo: s.lastPageNo,
			FracDigits: s.fracDigits,
		}
		s.rLocks++
	} else {
		req.ResultCh <- FullScanResult{
			ResultCode: QueryDone,
		}
	}
}

// COMMITS

func (s *Metric) OnMeasuresAppendCommited(rec storage.MeasuresAppendCommited) {
	//fmt.Println("OnMeasuresAppendCommited: written=", rec.WrittenCount)
	// Видаляю state. Оригінальні Timestamps і Values вже мають останню версію
	s.capturedState = nil
	s.timestamps.ForgetCapturedState()
	s.values.ForgetCapturedState()

	rec.ResultCh <- storage.MeasuresAppendResult{
		ResultCode:   rec.ResultCode,
		WrittenCount: rec.WrittenCount,
	}
}

func (s *Metric) OnMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWithGrowCommited) {
	//fmt.Println("OnMeasuresAppendWithGrowCommited: written=", rec.WrittenCount, ", lastPageNo=", rec.LastPageNo)
	//fmt.Printf("index: %#v\n", rec.Index)
	// Видаляю state. Оригінальні Timestamps і Values вже мають останню версію
	s.capturedState = nil
	s.timestamps.ForgetCapturedState()
	s.values.ForgetCapturedState()

	if rec.LastPageNo > 0 {
		s.lastPageNo = rec.LastPageNo
	}
	// В storage я передав повний індекс. У нього додали елементи (можливо нові рівні).
	// Тому проста заміна
	s.indexLevelTails = rec.Index

	rec.ResultCh <- storage.MeasuresAppendResult{
		ResultCode:   rec.ResultCode,
		WrittenCount: rec.WrittenCount,
	}
}

func (s *Metric) OnMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
	//s.timestamps.Reset()
	//s.values.Reset()
	s.xLock = false
	// s.Timestamps.Renew()
	// s.Values.Renew()
	s.indexLevelTails = nil
	s.lastPageNo = 0
	s.lastValue = 0

}

// const (
// 	free
// )

const datetimeLayout = "2006-01-02 15:04:05"

func formatTime(timestamp uint32) string {
	tm := time.Unix(int64(timestamp), 0)
	return tm.Format(datetimeLayout)
}