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) }