qb/worker/worker.go

package worker

import (
	"fmt"
	"sync"

	qb "gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/atree"
	"gordenko.dev/dima/qb/enc"
	"gordenko.dev/dima/qb/inbox"
	"gordenko.dev/dima/qb/proto"
	"gordenko.dev/dima/qb/storage"
	"gordenko.dev/dima/qb/transform"
)

// AddMetric - create metric in map and set Xlock, rwad - append to queue
// DeleteMetric - set Xlock
// DeleteSince - xLock (тупо редкая операция)
// Якийсь прапор поставити для pendingDelete, щоб read операції вставали в чергу

const (
	QueryDone          = 1
	UntilFound         = 2
	UntilNotFound      = 3
	RangeFound         = 15
	NoMeasures         = 16
	NoMetric           = 4
	MetricDuplicate    = 5
	Succeed            = 6
	ExpiredMeasure     = 8
	NonMonotonicValue  = 9
	WrongMetricType    = 11
	NoMeasuresToDelete = 12
	//DeleteFromAtreeNotNeeded = 13
	//DeleteFromAtreeRequired = 14
)

type Worker struct {
	inbox          *inbox.Inbox
	storageInbox   *inbox.Inbox
	dir            string
	databaseName   string
	snapshotNumber int
	tmp            []byte
	metrics        map[uint32]*Metric
	exitCh         chan struct{}
	waitGroup      *sync.WaitGroup
}

type Options struct {
	Inbox         *inbox.Inbox
	StorageInbox  *inbox.Inbox
	ReplayMetrics map[uint32]*storage.ReplayMetric
	Dir           string
	DatabaseName  string
	ExitCh        chan struct{}
	WaitGroup     *sync.WaitGroup
}

func New(opt Options) *Worker {
	s := &Worker{
		inbox:        opt.Inbox,
		storageInbox: opt.StorageInbox,
		dir:          opt.Dir,
		databaseName: opt.DatabaseName,
		tmp:          make([]byte, 26),
		metrics:      make(map[uint32]*Metric),
		exitCh:       opt.ExitCh,
		waitGroup:    opt.WaitGroup,
	}

	for metricID, x := range opt.ReplayMetrics {
		databuf := x.Buf[:storage.DataPagePayloadSize]
		values := enc.NewValueDeltaCompressor(x.MetricType, x.FracDigits, databuf, x.ValuesSize)
		s.metrics[metricID] = &Metric{
			metricType:      x.MetricType,
			fracDigits:      x.FracDigits,
			lastPageNo:      x.LastPageNo,
			lastValue:       values.LastValue(),
			buffer:          x.Buf,
			timestamps:      enc.NewTimeDeltaCompressor(databuf, x.TimestampsSize),
			values:          values,
			indexLevelTails: x.IndexLevelTails,
		}
	}
	return s
}

func (s *Worker) ReleaseRLock(metricID uint32) {
	// s.mutex.Lock()
	// //s.rLocksToRelease = append(s.rLocksToRelease, metricID)
	// s.mutex.Unlock()

	// select {
	// case s.signalCh <- struct{}{}:
	// default:
	// }
}

func (s *Worker) Run() {
	for {
		select {
		case <-s.inbox.Ready():
			s.doWork()

		case <-s.exitCh:
			fmt.Println("worker done")
			s.waitGroup.Done()
			return
		}
	}
}

func (s *Worker) doWork() {
	queue := s.inbox.Drain()

	//rLocksToRelease := s.rLocksToRelease
	//s.rLocksToRelease = nil

	// for _, metricID := range rLocksToRelease {
	// 	metric, ok := s.metrics[metricID]
	// 	if !ok {
	// 		qb.Abort(qb.NoMetricBug,
	// 			fmt.Errorf("drainQueues: metric %d not found", metricID))
	// 	}

	// 	if metric.XLock {
	// 		qb.Abort(qb.XLockBug,
	// 			fmt.Errorf("drainQueues: xlock is set for the metric %d",
	// 				metricID))
	// 	}

	// 	if metric.RLocks <= 0 {
	// 		qb.Abort(qb.NoRLockBug,
	// 			fmt.Errorf("drainQueues: rlock not set for the metric %d",
	// 				metricID))
	// 	}

	// 	metric.RLocks--

	// 	if len(metric.WaitQueue) > 0 {
	// 		s.processMetricQueue(metricID, metric)
	// 	}
	// }

	for _, untyped := range queue {
		switch req := untyped.(type) {
		case AppendMeasuresReq:
			s.AppendMeasures(req)
		case storage.Changes:
			s.applyCommits(req) // all metrics only
		case ListCurrentValuesReq:
			s.ListCurrentValues(req) // all metrics only
		case RangeScanReq:
			s.RangeScan(req)
		case FullScanReq:
			s.FullScan(req)
		case AddMetricReq:
			s.AddMetric(req)
		case DeleteMetricReq:
			s.DeleteMetric(req)
		case DeleteMeasuresReq:
			s.DeleteMeasures(req)
		case GetMetricReq:
			s.GetMetric(req)
		default:
			qb.Abort(qb.UnknownWorkerQueueItemBug,
				fmt.Errorf("bug: unknown worker queue item type %T", req))
		}
	}
}

// суть у тому що треба запускати запити, пока не зустріну XLock
func (s *Worker) processMetricQueue(metricID uint32, metric *Metric, tmp []byte) {
	if len(metric.waitQueue) == 0 {
		return
	}
	for _, untyped := range metric.waitQueue {
		switch req := untyped.(type) {
		case RangeScanReq:
			metric.StartRangeScan(req)
		case FullScanReq:
			metric.StartFullScan(req)
		case GetMetricReq:
			s.GetMetric(req)
		case AppendMeasuresReq:
			metric.AppendMeasures(req, tmp, s.storageInbox)
		case DeleteMetricReq:
			s.DeleteMetric(req)
		case DeleteMeasuresReq:
			metric.DeleteMeasures(req)
		default:
			qb.Abort(qb.UnknownMetricWaitQueueItemBug,
				fmt.Errorf("bug: unknown metric wait queue item type %T", req))
		}
	}
}

type AddMetricReq struct {
	MetricID   uint32
	MetricType qb.MetricType
	FracDigits byte
	ResultCh   chan byte
}

func (s *Worker) AddMetric(req AddMetricReq) {
	_, ok := s.metrics[req.MetricID]
	if ok {
		req.ResultCh <- MetricDuplicate
		return
	}
	var (
		buf     = make([]byte, storage.DataPageSize)
		databuf = buf[:storage.DataPagePayloadSize]
	)
	s.metrics[req.MetricID] = &Metric{
		metricType: req.MetricType,
		fracDigits: req.FracDigits,
		buffer:     buf,
		timestamps: enc.NewTimeDeltaCompressor(databuf, 0),
		values:     enc.NewValueDeltaCompressor(req.MetricType, req.FracDigits, databuf, 0),
		xLock:      true,
	}
	s.storageInbox.Push(storage.MetricAdd{
		MetricID:   req.MetricID,
		MetricType: req.MetricType,
		FracDigits: req.FracDigits,
		ResultCh:   req.ResultCh,
	})
}

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

type GetMetricReq struct {
	MetricID uint32
	ResultCh chan GetMetricResult
}

func (s *Worker) GetMetric(req GetMetricReq) {
	metric, ok := s.metrics[req.MetricID]
	if ok {
		req.ResultCh <- GetMetricResult{
			ResultCode: Succeed,
			MetricType: metric.MetricType(),
			FracDigits: metric.FracDigits(),
		}
	} else {
		req.ResultCh <- GetMetricResult{
			ResultCode: NoMetric,
		}
	}
}

type DeleteMetricReq struct {
	MetricID uint32
	ResultCh chan byte
}

func (s *Worker) DeleteMetric(req DeleteMetricReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- NoMetric
		return
	}
	if metric.xLock {
		metric.waitQueue = append(metric.waitQueue, req)
	} else {
		// collect all pages, than
		s.storageInbox.Push(storage.MetricDelete{
			MetricID:       req.MetricID,
			FreeIndexPages: nil,
			FreeDataPages:  nil,
			ResultCh:       req.ResultCh,
		})
	}
}

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

func (s *Worker) DeleteMeasures(req DeleteMeasuresReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- NoMetric
		return
	}
	metric.DeleteMeasures(req)
}

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

func (s *Worker) AppendMeasures(req AppendMeasuresReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- storage.MeasuresAppendResult{
			ResultCode: NoMetric,
		}
		return
	}
	if metric.xLock {
		metric.waitQueue = append(metric.waitQueue, req)
	} else {
		metric.AppendMeasures(req, s.tmp, s.storageInbox)
	}
}

type RangeScanResult struct {
	ResultCode byte
	FracDigits byte
	RootPageNo uint32
	LastPageNo uint32
}

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

func (s *Worker) RangeScan(req RangeScanReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- RangeScanResult{
			ResultCode: NoMetric,
		}
		return
	}
	if metric.MetricType() != req.MetricType {
		req.ResultCh <- RangeScanResult{
			ResultCode: WrongMetricType,
		}
		return
	}
	if metric.xLock {
		metric.waitQueue = append(metric.waitQueue, req)
	} else {
		metric.StartRangeScan(req)
	}
}

type FullScanResult struct {
	ResultCode byte
	FracDigits byte
	LastPageNo uint32
}

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

func (s *Worker) FullScan(req FullScanReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- FullScanResult{
			ResultCode: NoMetric,
		}
		return
	}
	if metric.MetricType() != req.MetricType {
		req.ResultCh <- FullScanResult{
			ResultCode: WrongMetricType,
		}
		return
	}
	if metric.xLock {
		metric.waitQueue = append(metric.waitQueue, req)
	} else {
		metric.StartFullScan(req)
	}
}

type ListCurrentValuesReq struct {
	MetricIDs      []uint32
	ResponseWriter *transform.CurrentValueWriter
	ResultCh       chan struct{}
}

func (s *Worker) ListCurrentValues(req ListCurrentValuesReq) {
	for _, metricID := range req.MetricIDs {
		metric, ok := s.metrics[metricID]
		if ok {
			req.ResponseWriter.BufferValue(transform.CurrentValue{
				MetricID:  metricID,
				Timestamp: metric.LastTimestamp(),
				Value:     metric.LastValue(),
			})
		}
	}
	req.ResultCh <- struct{}{}
}

///////////////////////////////////////////////////////

func (s *Worker) applyCommits(req storage.Changes) {
	for _, untyped := range req.Commits {
		switch rec := untyped.(type) {
		case storage.MetricAddCommited:
			s.onMetricAddCommited(rec)

		case storage.MetricDeleteCommited:
			s.onMetricDeleteCommited(rec)

		case storage.MeasuresAppendCommited:
			s.onMeasuresAppendCommited(rec)

		case storage.MeasuresAppendWithGrowCommited:
			s.onMeasuresAppendWithGrowCommited(rec)

		case storage.MeasuresDeleteCommited:
			s.onMeasuresDeleteCommited(rec)
		}
	}

	if req.SnapshotNumberCh != nil {
		s.snapshotNumber++
		err := writeSnapshot(
			qb.GetSnapshotFilePath(s.dir, s.databaseName, s.snapshotNumber),
			WriteSnapshotIn{
				SnapshotNumber:        s.snapshotNumber,
				WriteBufferSize:       4 * 1024 * 1024, // 4mb
				Metrics:               s.metrics,
				FrozenIndexPagesCount: req.FrozenIndexPagesCount,
				IndexPageNumbers:      req.IndexPageNumbers,
				FrozenDataPagesCount:  req.FrozenDataPagesCount,
				DataPageNumbers:       req.DataPageNumbers,
			})
		if err != nil {
			qb.Abort(qb.WriteSnapshotFailed, err)
		}
		req.SnapshotNumberCh <- s.snapshotNumber
	}
}

func (s *Worker) onMetricAddCommited(rec storage.MetricAddCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.MetricAddedBug,
			fmt.Errorf("onMetricAddCommited: metric %d not found", rec.MetricID))
	}
	metric.xLock = false
	rec.ResultCh <- Succeed
}

func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("onMetricDeleteCommited: metric %d not found", rec.MetricID))
	}

	if len(metric.waitQueue) > 0 {
		for _, untyped := range metric.waitQueue {
			switch req := untyped.(type) {
			case AppendMeasuresReq:
				req.ResultCh <- storage.MeasuresAppendResult{
					ResultCode:   NoMetric,
					WrittenCount: 0,
				}
			case RangeScanReq:
				req.ResultCh <- RangeScanResult{
					ResultCode: NoMetric,
				}
			case FullScanReq:
				req.ResultCh <- FullScanResult{
					ResultCode: NoMetric,
				}
			case DeleteMetricReq:
				req.ResultCh <- NoMetric
			case DeleteMeasuresReq:
				req.ResultCh <- NoMetric
			case GetMetricReq:
				req.ResultCh <- GetMetricResult{
					ResultCode: NoMetric,
				}
			default:
				qb.Abort(qb.UnknownMetricWaitQueueItemBug,
					fmt.Errorf("bug: unknown metric wait queue item type %T", req))
			}
		}
	}
	delete(s.metrics, rec.MetricID)
	//
	rec.ResultCh <- Succeed
}

func (s *Worker) onMeasuresAppendCommited(rec storage.MeasuresAppendCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("onMeasuresAppendCommited: metric %d not found",
				rec.MetricID))
	}
	metric.OnMeasuresAppendCommited(rec)
}

func (s *Worker) onMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWithGrowCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("onMeasuresAppendWithGrowCommited: metric %d not found",
				rec.MetricID))
	}
	metric.OnMeasuresAppendWithGrowCommited(rec)
}

func (s *Worker) onMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("onMeasuresDeleteCommited: metric %d not found", rec.MetricID))
	}
	metric.OnMeasuresDeleteCommited(rec)
	metric.xLock = false

	//s.doAfterReleaseXLock(rec.MetricID, metric)
}

// func (s *Database) doAfterReleaseXLock(metricID uint32, metric *_metric) {
// 	if len(metric.WaitQueue) > 0 {
// 		s.processMetricQueue(metricID, metric)
// 	}
// }