wp
This commit is contained in:
394
worker/metric.go
Normal file
394
worker/metric.go
Normal file
@@ -0,0 +1,394 @@
|
||||
package worker
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"io"
|
||||
|
||||
bin "gordenko.dev/dima/bin/little"
|
||||
"gordenko.dev/dima/qb"
|
||||
"gordenko.dev/dima/qb/storage"
|
||||
)
|
||||
|
||||
// METRIC
|
||||
|
||||
var ErrNoValueBug = errors.New("has timestamp but no value")
|
||||
|
||||
type CapturedState struct {
|
||||
LastTimestamp uint32
|
||||
LastValue float64
|
||||
}
|
||||
|
||||
type Metric struct {
|
||||
metricType qb.MetricType
|
||||
fracDigits byte
|
||||
lastPageNo uint32
|
||||
//SinceValue float64
|
||||
//Since uint32
|
||||
lastValue float64
|
||||
//Until uint32
|
||||
buffer []byte
|
||||
timestamps qb.TimestampCompressor
|
||||
values qb.ValueCompressor
|
||||
XLock bool
|
||||
RLocks int
|
||||
WaitQueue []any
|
||||
indexLevelTails []storage.IndexLevelTail // root - last element
|
||||
capturedState *CapturedState
|
||||
}
|
||||
|
||||
func (s *Metric) LastValue() float64 {
|
||||
if s.capturedState != nil {
|
||||
return s.capturedState.LastValue
|
||||
}
|
||||
return s.lastValue
|
||||
}
|
||||
|
||||
func (s *Metric) LastTimestamp() uint32 {
|
||||
if s.capturedState != nil {
|
||||
return s.capturedState.LastTimestamp
|
||||
}
|
||||
return s.timestamps.LastTimestamp()
|
||||
}
|
||||
|
||||
func (s *Metric) OnMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
|
||||
//s.timestamps.Reset()
|
||||
//s.values.Reset()
|
||||
s.XLock = false
|
||||
// s.Timestamps.Renew()
|
||||
// s.Values.Renew()
|
||||
|
||||
// s.LastPageNo = 0
|
||||
// s.Since = 0
|
||||
// s.SinceValue = 0
|
||||
// s.Until = 0
|
||||
s.indexLevelTails = nil
|
||||
s.lastPageNo = 0
|
||||
s.lastValue = 0
|
||||
}
|
||||
|
||||
// func (s *_metric) startAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(storage.AppendedMeasures)) {
|
||||
func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, saveToStorage func(any)) {
|
||||
if s.capturedState != nil {
|
||||
s.WaitQueue = append(s.WaitQueue, req)
|
||||
}
|
||||
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
|
||||
)
|
||||
|
||||
s.capturedState = &CapturedState{
|
||||
LastTimestamp: timestamps.LastTimestamp(),
|
||||
LastValue: s.lastValue,
|
||||
}
|
||||
|
||||
s.timestamps.CaptureState()
|
||||
s.values.CaptureState()
|
||||
|
||||
for idx, measure := range req.Measures {
|
||||
if measure.Timestamp <= s.timestamps.LastTimestamp() {
|
||||
resultCode = ExpiredMeasure
|
||||
break
|
||||
}
|
||||
if s.metricType == qb.Cumulative && measure.Value < s.lastValue {
|
||||
resultCode = NonMonotonicValue
|
||||
break
|
||||
}
|
||||
|
||||
tReport := timestamps.Evaluate(tmp, measure.Timestamp)
|
||||
vReport := values.Evaluate(tmp, measure.Value)
|
||||
|
||||
totalRequiredSpace := tReport.TotalSpace + vReport.TotalSpace
|
||||
|
||||
if totalRequiredSpace <= len(s.buffer) {
|
||||
// якщо на сторінці є місце
|
||||
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 {
|
||||
// сторінка заповнена
|
||||
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(),
|
||||
})
|
||||
|
||||
buf := make([]byte, storage.DataPageSize)
|
||||
databuf := buf[:storage.DataPagePayloadSize]
|
||||
|
||||
timestamps.ReplaceBuffer(databuf)
|
||||
values.ReplaceBuffer(databuf)
|
||||
|
||||
// renew
|
||||
s.buffer = buf
|
||||
|
||||
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 <- AppendMeasuresResult{
|
||||
ResultCode: resultCode,
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// виділити змінені байти.
|
||||
// скопіювати. Причому можна скопіювати зрізи chunks
|
||||
|
||||
if len(pages) > 0 {
|
||||
// пишу в storage довгим шляхом через redo файл і запис в data файл
|
||||
saveToStorage(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 {
|
||||
// короткий шлях - запис лише в storage
|
||||
saveToStorage(storage.MeasuresAppend{
|
||||
MetricID: req.MetricID,
|
||||
TimestampsRewindOffset: timestampsRewindOffset,
|
||||
ValuesRewindOffset: valuesRewindOffset,
|
||||
Timestamps: timestamps.Tail(timestampsOffset),
|
||||
Values: values.Tail(valuesOffset),
|
||||
ResultCode: resultCode,
|
||||
WrittenCount: written,
|
||||
//ResultCh: req.ResultCh,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Metric) OnMeasuresAppendCommited(rec storage.MeasuresAppendCommited) {
|
||||
// Видаляю state. Оригінальні Timestamps і Values вже мають останню версію
|
||||
s.capturedState = nil
|
||||
s.timestamps.ForgetCapturedState()
|
||||
s.values.ForgetCapturedState()
|
||||
}
|
||||
|
||||
func (s *Metric) OnMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWithGrowCommited) {
|
||||
// Видаляю 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
|
||||
}
|
||||
|
||||
// READ
|
||||
|
||||
func (s *Metric) StartRangeScan(req RangeScanReq) {
|
||||
// if s.Since == 0 {
|
||||
// req.ResultCh <- rangeScanResult{
|
||||
// ResultCode: QueryDone,
|
||||
// }
|
||||
// return
|
||||
// }
|
||||
|
||||
// if req.Since > s.Until {
|
||||
// req.ResultCh <- rangeScanResult{
|
||||
// ResultCode: QueryDone,
|
||||
// }
|
||||
// return
|
||||
// }
|
||||
|
||||
// if req.Until < s.Since {
|
||||
// if s.RootPageNo > 0 {
|
||||
// req.ResultCh <- rangeScanResult{
|
||||
// ResultCode: UntilNotFound,
|
||||
// RootPageNo: s.RootPageNo,
|
||||
// FracDigits: s.FracDigits,
|
||||
// }
|
||||
// s.RLocks++
|
||||
// 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,
|
||||
// }
|
||||
// }
|
||||
}
|
||||
|
||||
func (s *Metric) StartFullScan(req FullScanReq) {
|
||||
// if s.Since == 0 {
|
||||
// req.ResultCh <- fullScanResult{
|
||||
// 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)
|
||||
}
|
||||
req.ResponseWriter.FeedNoSend(timestamp, value)
|
||||
}
|
||||
|
||||
if s.lastPageNo > 0 {
|
||||
req.ResultCh <- FullScanResult{
|
||||
ResultCode: UntilFound,
|
||||
LastPageNo: s.lastPageNo,
|
||||
FracDigits: s.fracDigits,
|
||||
}
|
||||
s.RLocks++
|
||||
} else {
|
||||
req.ResultCh <- FullScanResult{
|
||||
ResultCode: QueryDone,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// індекси
|
||||
// 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.StoredSize()))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = bin.WriteUint16(w, uint16(s.values.StoredSize()))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// timestamps payload
|
||||
err = s.timestamps.WriteStoredTo(w)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// values payload
|
||||
err = s.values.WriteStoredTo(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)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
676
worker/worker.go
Normal file
676
worker/worker.go
Normal file
@@ -0,0 +1,676 @@
|
||||
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
|
||||
NewPage = 7
|
||||
ExpiredMeasure = 8
|
||||
NonMonotonicValue = 9
|
||||
CanAppend = 10
|
||||
WrongMetricType = 11
|
||||
NoMeasuresToDelete = 12
|
||||
DeleteFromAtreeNotNeeded = 13
|
||||
DeleteFromAtreeRequired = 14
|
||||
)
|
||||
|
||||
type Worker struct {
|
||||
//mutex sync.Mutex
|
||||
inbox *inbox.Inbox
|
||||
storageInbox *inbox.Inbox
|
||||
dir string
|
||||
snapshotNumber int
|
||||
tmp []byte
|
||||
metrics map[uint32]*Metric
|
||||
saveToStorage func(any)
|
||||
exitCh chan struct{}
|
||||
waitGroup *sync.WaitGroup
|
||||
}
|
||||
|
||||
type WorkerOptions struct {
|
||||
Inbox *inbox.Inbox
|
||||
StorageInbox *inbox.Inbox
|
||||
Metrics map[uint32]*Metric
|
||||
Dir string
|
||||
ExitCh chan struct{}
|
||||
WaitGroup *sync.WaitGroup
|
||||
}
|
||||
|
||||
func NewWorker(opt WorkerOptions) *Worker {
|
||||
return &Worker{
|
||||
inbox: opt.Inbox,
|
||||
storageInbox: opt.StorageInbox,
|
||||
dir: opt.Dir,
|
||||
tmp: make([]byte, 26),
|
||||
metrics: opt.Metrics,
|
||||
exitCh: opt.ExitCh,
|
||||
waitGroup: opt.WaitGroup,
|
||||
}
|
||||
}
|
||||
|
||||
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.tryListCurrentValues(req) // all metrics only
|
||||
|
||||
case RangeScanReq:
|
||||
s.RangeScan(req)
|
||||
|
||||
case FullScanReq:
|
||||
s.tryFullScan(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.StartAppendMeasures(req, tmp, s.saveToStorage)
|
||||
|
||||
case DeleteMetricReq:
|
||||
s.startDeleteMetric(metric, req)
|
||||
|
||||
case DeleteMeasuresReq:
|
||||
s.startDeleteMeasures(metric, req)
|
||||
|
||||
default:
|
||||
qb.Abort(qb.UnknownMetricWaitQueueItemBug,
|
||||
fmt.Errorf("bug: unknown metric wait queue item type %T", req))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type AddMetricReq struct {
|
||||
MetricID uint32
|
||||
ResultCh chan byte
|
||||
}
|
||||
|
||||
func (s *Worker) AddMetric(req AddMetricReq) {
|
||||
_, ok := s.metrics[req.MetricID]
|
||||
if ok {
|
||||
req.ResultCh <- MetricDuplicate
|
||||
return
|
||||
}
|
||||
fmt.Println("metric not found. Success")
|
||||
req.ResultCh <- Succeed // new
|
||||
|
||||
// lockEntry, ok := s.metricLockEntries[req.MetricID]
|
||||
// if ok {
|
||||
// lockEntry.WaitQueue = append(lockEntry.WaitQueue, req)
|
||||
// } else {
|
||||
// s.metricLockEntries[req.MetricID] = &metricLockEntry{
|
||||
// XLock: true,
|
||||
// }
|
||||
// req.ResultCh <- Succeed
|
||||
// }
|
||||
}
|
||||
|
||||
func (s *Worker) processTryAddMetricReqsImmediatelyAfterDelete(reqs []AddMetricReq) {
|
||||
if len(reqs) == 0 {
|
||||
return
|
||||
}
|
||||
var (
|
||||
req = reqs[0]
|
||||
waitQueue []any
|
||||
)
|
||||
if len(reqs) > 1 {
|
||||
for _, req := range reqs[1:] {
|
||||
waitQueue = append(waitQueue, req)
|
||||
}
|
||||
}
|
||||
// FIX
|
||||
// s.metricLockEntries[req.MetricID] = &metricLockEntry{
|
||||
// XLock: true,
|
||||
// WaitQueue: waitQueue,
|
||||
// }
|
||||
req.ResultCh <- Succeed
|
||||
}
|
||||
|
||||
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 DeleteMetricResult struct {
|
||||
ResultCode byte
|
||||
RootPageNo uint32
|
||||
}
|
||||
|
||||
type DeleteMetricReq struct {
|
||||
MetricID uint32
|
||||
ResultCh chan DeleteMetricResult
|
||||
}
|
||||
|
||||
func (s *Worker) DeleteMetric(req DeleteMetricReq) {
|
||||
// FIX
|
||||
// metric, ok := s.metrics[req.MetricID]
|
||||
// if !ok {
|
||||
// req.ResultCh <- tryDeleteMetricResult{
|
||||
// ResultCode: NoMetric,
|
||||
// }
|
||||
// return
|
||||
// }
|
||||
|
||||
// lockEntry, ok := s.metricLockEntries[req.MetricID]
|
||||
// if ok {
|
||||
// lockEntry.WaitQueue = append(lockEntry.WaitQueue, req)
|
||||
// } else {
|
||||
// s.startDeleteMetric(metric, req)
|
||||
// }
|
||||
}
|
||||
|
||||
func (s *Worker) startDeleteMetric(metric *Metric, req DeleteMetricReq) {
|
||||
// FIX
|
||||
// s.metricLockEntries[req.MetricID] = &metricLockEntry{
|
||||
// XLock: true,
|
||||
// }
|
||||
// req.ResultCh <- tryDeleteMetricResult{
|
||||
// ResultCode: Succeed,
|
||||
// RootPageNo: metric.RootPageNo,
|
||||
// }
|
||||
}
|
||||
|
||||
type DeleteMeasuresResult struct {
|
||||
ResultCode byte
|
||||
RootPageNo uint32
|
||||
}
|
||||
|
||||
type DeleteMeasuresReq struct {
|
||||
MetricID uint32
|
||||
Since uint32
|
||||
ResultCh chan DeleteMeasuresResult
|
||||
}
|
||||
|
||||
func (s *Worker) DeleteMeasures(req DeleteMeasuresReq) {
|
||||
// FIX
|
||||
// metric, ok := s.metrics[req.MetricID]
|
||||
// if !ok {
|
||||
// req.ResultCh <- tryDeleteMeasuresResult{
|
||||
// ResultCode: NoMetric,
|
||||
// }
|
||||
// return
|
||||
// }
|
||||
|
||||
// if metric.Since == 0 || (req.Since > 0 && metric.Until < req.Since) {
|
||||
// req.ResultCh <- tryDeleteMeasuresResult{
|
||||
// ResultCode: NoMeasuresToDelete,
|
||||
// }
|
||||
// }
|
||||
|
||||
// lockEntry, ok := s.metricLockEntries[req.MetricID]
|
||||
// if ok {
|
||||
// lockEntry.WaitQueue = append(lockEntry.WaitQueue, req)
|
||||
// } else {
|
||||
// s.startDeleteMeasures(metric, req)
|
||||
// }
|
||||
}
|
||||
|
||||
func (s *Worker) startDeleteMeasures(metric *Metric, req DeleteMeasuresReq) {
|
||||
// FIX
|
||||
// s.metricLockEntries[req.MetricID] = &metricLockEntry{
|
||||
// XLock: true,
|
||||
// }
|
||||
|
||||
// if metric.RootPageNo > 0 {
|
||||
// req.ResultCh <- tryDeleteMeasuresResult{
|
||||
// ResultCode: DeleteFromAtreeRequired,
|
||||
// RootPageNo: metric.RootPageNo,
|
||||
// }
|
||||
// } else {
|
||||
// req.ResultCh <- tryDeleteMeasuresResult{
|
||||
// ResultCode: DeleteFromAtreeNotNeeded,
|
||||
// }
|
||||
// }
|
||||
}
|
||||
|
||||
func (s *Worker) onMeasuresAppendCommited(rec storage.MeasuresAppendCommited) {
|
||||
metric, ok := s.metrics[rec.MetricID]
|
||||
if !ok {
|
||||
qb.Abort(qb.NoMetricBug,
|
||||
fmt.Errorf("finAppendMeasures: 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("finAppendMeasures: metric %d not found",
|
||||
rec.MetricID))
|
||||
}
|
||||
metric.OnMeasuresAppendWithGrowCommited(rec)
|
||||
}
|
||||
|
||||
type AppendMeasuresResult struct {
|
||||
ResultCode byte
|
||||
Written int
|
||||
}
|
||||
|
||||
type AppendMeasuresReq struct {
|
||||
MetricID uint32
|
||||
Measures []proto.Measure
|
||||
ResultCh chan AppendMeasuresResult
|
||||
}
|
||||
|
||||
func (s *Worker) AppendMeasures(req AppendMeasuresReq) {
|
||||
metric, ok := s.metrics[req.MetricID]
|
||||
if !ok {
|
||||
req.ResultCh <- AppendMeasuresResult{
|
||||
ResultCode: NoMetric,
|
||||
}
|
||||
return
|
||||
}
|
||||
if metric.XLock {
|
||||
metric.WaitQueue = append(metric.WaitQueue, req)
|
||||
return
|
||||
}
|
||||
metric.StartAppendMeasures(req, s.tmp, s.saveToStorage)
|
||||
}
|
||||
|
||||
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)
|
||||
return
|
||||
}
|
||||
|
||||
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) tryFullScan(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)
|
||||
return
|
||||
}
|
||||
metric.StartFullScan(req)
|
||||
}
|
||||
|
||||
type ListCurrentValuesReq struct {
|
||||
MetricIDs []uint32
|
||||
ResponseWriter *transform.CurrentValueWriter
|
||||
ResultCh chan struct{}
|
||||
}
|
||||
|
||||
func (s *Worker) tryListCurrentValues(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 := storage.WriteSnapshot(storage.WriteSnapshotIn{
|
||||
SnapshotNumber: s.snapshotNumber,
|
||||
Dir: s.dir,
|
||||
WriteBufferSize: 4 * 1024 * 1024, // 1mb
|
||||
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) {
|
||||
// fix lock
|
||||
_, ok := s.metrics[rec.MetricID]
|
||||
if ok {
|
||||
qb.Abort(qb.MetricAddedBug,
|
||||
fmt.Errorf("addMetric: metric %d already added",
|
||||
rec.MetricID))
|
||||
}
|
||||
|
||||
// if !lockEntry.XLock {
|
||||
// qb.Abort(qb.NoXLockBug,
|
||||
// fmt.Errorf("addMetric: xlock not set for the metric %d",
|
||||
// rec.MetricID))
|
||||
// }
|
||||
|
||||
// lockEntry.XLock = false
|
||||
// delete(s.metricLockEntries, rec.MetricID)
|
||||
}
|
||||
|
||||
func (s *Worker) addMetric(rec storage.MetricAddRecord) {
|
||||
var (
|
||||
buf = make([]byte, storage.DataPageSize)
|
||||
databuf = buf[:storage.DataPagePayloadSize]
|
||||
)
|
||||
s.metrics[rec.MetricID] = &Metric{
|
||||
metricType: rec.MetricType,
|
||||
fracDigits: rec.FracDigits,
|
||||
buffer: buf,
|
||||
timestamps: enc.NewTimeDeltaCompressor(databuf, 0),
|
||||
values: enc.NewValueDeltaCompressor(rec.MetricType, rec.FracDigits, databuf, 0),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) {
|
||||
metric, ok := s.metrics[rec.MetricID]
|
||||
if !ok {
|
||||
qb.Abort(qb.NoMetricBug,
|
||||
fmt.Errorf("deleteMetric: metric %d not found",
|
||||
rec.MetricID))
|
||||
}
|
||||
|
||||
if !metric.XLock {
|
||||
qb.Abort(qb.NoXLockBug,
|
||||
fmt.Errorf("deleteMetric: xlock not set for the metric %d",
|
||||
rec.MetricID))
|
||||
}
|
||||
|
||||
var addMetricReqs []AddMetricReq
|
||||
|
||||
if len(metric.WaitQueue) > 0 {
|
||||
for _, untyped := range metric.WaitQueue {
|
||||
switch req := untyped.(type) {
|
||||
// case tryAppendMeasureReq:
|
||||
// req.ResultCh <- tryAppendMeasureResult{
|
||||
// MetricID: req.MetricID,
|
||||
// ResultCode: NoMetric,
|
||||
// }
|
||||
|
||||
case RangeScanReq:
|
||||
req.ResultCh <- RangeScanResult{
|
||||
ResultCode: NoMetric,
|
||||
}
|
||||
|
||||
case FullScanReq:
|
||||
req.ResultCh <- FullScanResult{
|
||||
ResultCode: NoMetric,
|
||||
}
|
||||
|
||||
case AddMetricReq:
|
||||
addMetricReqs = append(addMetricReqs, req)
|
||||
|
||||
case DeleteMetricReq:
|
||||
req.ResultCh <- DeleteMetricResult{
|
||||
ResultCode: NoMetric,
|
||||
}
|
||||
|
||||
case DeleteMeasuresReq:
|
||||
req.ResultCh <- DeleteMeasuresResult{
|
||||
ResultCode: 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)
|
||||
|
||||
// ADD in storage
|
||||
// if len(rec.FreePageNumbers) > 0 {
|
||||
// s.freeList.AddPages(rec.FreePageNumbers)
|
||||
// }
|
||||
|
||||
if len(addMetricReqs) > 0 {
|
||||
s.processTryAddMetricReqsImmediatelyAfterDelete(addMetricReqs)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Worker) onMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
|
||||
metric, ok := s.metrics[rec.MetricID]
|
||||
if !ok {
|
||||
qb.Abort(qb.NoMetricBug,
|
||||
fmt.Errorf("deleteMeasures: metric %d not found",
|
||||
rec.MetricID))
|
||||
}
|
||||
|
||||
if !metric.XLock {
|
||||
qb.Abort(qb.NoXLockBug,
|
||||
fmt.Errorf("deleteMeasures: xlock not set for the metric %d",
|
||||
rec.MetricID))
|
||||
}
|
||||
metric.OnMeasuresDeleteCommited(rec)
|
||||
metric.XLock = false
|
||||
// FIX add in storage
|
||||
// if len(rec.FreePageNumbers) > 0 {
|
||||
// s.freeList.AddPages(rec.FreePageNumbers)
|
||||
// }
|
||||
//s.doAfterReleaseXLock(rec.MetricID, metric)
|
||||
}
|
||||
|
||||
// func (s *Database) doAfterReleaseXLock(metricID uint32, metric *_metric) {
|
||||
// if len(metric.WaitQueue) > 0 {
|
||||
// s.processMetricQueue(metricID, metric)
|
||||
// }
|
||||
// }
|
||||
Reference in New Issue
Block a user