diff --git a/chunkenc/cumdelta.go b/chunkenc/cumdelta.go index 3df8889..4befc22 100644 --- a/chunkenc/cumdelta.go +++ b/chunkenc/cumdelta.go @@ -202,6 +202,20 @@ func (s *ReverseCumulativeDeltaCompressor) Unlock() { s.state = nil } +func (s *ReverseCumulativeDeltaCompressor) Renew() { + s.buf = conbuf.New(nil) + s.pos = 0 + // + s.baseValue = 0 + s.lastDelta = 0 + s.lastDeltaSize = 0 + s.h = 0 +} + +func (s *ReverseCumulativeDeltaCompressor) Chunks() [][]byte { + return s.buf.Chunks() +} + // DECOMPRESSOR type ReverseCumulativeDeltaDecompressor struct { diff --git a/database/api.go b/database/api.go index ca0991e..590d109 100644 --- a/database/api.go +++ b/database/api.go @@ -10,8 +10,6 @@ import ( "gordenko.dev/dima/qb/atree" "gordenko.dev/dima/qb/bin" "gordenko.dev/dima/qb/bufreader" - "gordenko.dev/dima/qb/chunkenc" - "gordenko.dev/dima/qb/conbuf" "gordenko.dev/dima/qb/proto" "gordenko.dev/dima/qb/transform" "gordenko.dev/dima/qb/txlog" @@ -426,6 +424,7 @@ type FilledPage struct { type tryAppendMeasuresResult struct { ResultCode byte + Written int // MetricType diploma.MetricType // FracDigits byte // Since uint32 @@ -452,150 +451,6 @@ func (s *Database) AppendMeasures(req proto.AppendMeasuresReq) uint16 { switch result.ResultCode { case CanAppend: - var ( - rootPageNo = result.RootPageNo - prevPageNo = result.PrevPageNo - timestampsBuf = result.TimestampsBuf - valuesBuf = result.ValuesBuf - timestamps = result.Timestamps - values = result.Values - since = result.Since - until = result.Until - untilValue = result.UntilValue - // - //toAppendMeasures []proto.Measure - - dataPages []atree.NotLinkedDataPage - ) - - for idx, measure := range req.Measures { - _ = idx - if since == 0 { - since = measure.Timestamp - } else { - if measure.Timestamp <= until { - // пишу в БД що вже встиг закодувати FIX - // if len(toAppendMeasures) > 0 { - // waitCh := s.txlog.WriteAppendMeasures( - // txlog.AppendedMeasures{ - // MetricID: req.MetricID, - // Measures: toAppendMeasures, - // }, - // false, - // ) - // <-waitCh - // } - return proto.ErrExpiredMeasure - } - - if result.MetricType == diploma.Cumulative && measure.Value < untilValue { - // пишу в БД що вже встиг закодувати FIX - // if len(toAppendMeasures) > 0 { - // waitCh := s.txlog.WriteAppendMeasures( - // txlog.AppendedMeasures{ - // MetricID: req.MetricID, - // Measures: toAppendMeasures, - // }, - // false, - // ) - // <-waitCh - // } - return proto.ErrNonMonotonicValue - } - } - - extraSpace := timestamps.CalcRequiredSpace(measure.Timestamp) + - values.CalcRequiredSpace(measure.Value) - - totalSpace := timestamps.Size() + values.Size() + extraSpace - - if totalSpace <= atree.DataPagePayloadSize { - // накопичую - timestamps.Append(measure.Timestamp) - values.Append(measure.Value) - //toAppendMeasures = append(toAppendMeasures, measure) - } else { - // сторінка заповнена - // if len(toAppendMeasures) > 0 { - // waitCh := s.txlog.WriteAppendMeasures( - // txlog.AppendedMeasures{ - // MetricID: req.MetricID, - // Measures: toAppendMeasures, - // }, - // true, - // ) - // <-waitCh - - // toAppendMeasures = nil - // } - - pageData := make([]byte, atree.PageSize) - - atree.ChunksToNotLinkedDataPage(pageData, atree.ChunksToNotLinkedDataPageReq{ - //PrevPageNo: path.LastPageNo, - TimestampsChunks: timestampsBuf.Chunks(), - TimestampsSize: uint16(timestamps.Size()), - ValuesChunks: valuesBuf.Chunks(), - ValuesSize: uint16(values.Size()), - }) - - dataPages = append(dataPages, atree.NotLinkedDataPage{ - Since: since, - Data: pageData, - }) - - // FIX - // prevPageNo = report.DataPageNo - // if report.IsRootChanged { - // rootPageNo = report.NewRootPageNo - // } - // waitCh := s.txlog.WriteAppendedMeasureWithOverflow( - // txlog.AppendedMeasureWithOverflow{ - // MetricID: req.MetricID, - // Timestamp: measure.Timestamp, - // Value: measure.Value, - // IsDataPageReused: report.IsDataPageReused, - // DataPageNo: report.DataPageNo, - // IsRootChanged: report.IsRootChanged, - // RootPageNo: report.NewRootPageNo, - // ReusedIndexPages: report.ReusedIndexPages, - // }, - // (idx+1) < len(req.Measures), - // ) - // <-waitCh - - timestampsBuf = conbuf.New(nil) - valuesBuf = conbuf.New(nil) - - timestamps = chunkenc.NewReverseTimeDeltaCompressor( - timestampsBuf, 0) - if result.MetricType == diploma.Cumulative { - values = chunkenc.NewReverseCumulativeDeltaCompressor( - valuesBuf, 0, result.FracDigits) - } else { - values = chunkenc.NewReverseInstantDeltaCompressor( - valuesBuf, 0, result.FracDigits) - } - - timestamps.Append(measure.Timestamp) - values.Append(measure.Value) - - since = measure.Timestamp - } - - until = measure.Timestamp - untilValue = measure.Value - } - - waitCh := s.txlog.Append(txlog.AppendedPagesReq{ - MetricID: req.MetricID, - Timestamp: until, - Value: untilValue, - LastPageNo: prevPageNo, - //Measures: toAppendMeasures, - }, - ) - <-waitCh // report, err := s.atree.AppendDataPage(atree.AppendDataPageReq{}) // if err != nil { diff --git a/database/metric.go b/database/metric.go index b40ad79..67c0019 100644 --- a/database/metric.go +++ b/database/metric.go @@ -1,25 +1,25 @@ package database import ( - octopus "gordenko.dev/dima/qb" + "gordenko.dev/dima/qb" "gordenko.dev/dima/qb/conbuf" ) // METRIC type _metric struct { - MetricType octopus.MetricType - FracDigits byte - RootPageNo uint32 - LastPageNo uint32 - SinceValue float64 - Since uint32 - UntilValue float64 - Until uint32 - TimestampsBuf *conbuf.ContinuousBuffer - ValuesBuf *conbuf.ContinuousBuffer - Timestamps octopus.TimestampCompressor - Values octopus.ValueCompressor + MetricType qb.MetricType + FracDigits byte + RootPageNo uint32 + LastPageNo uint32 + SinceValue float64 + Since uint32 + UntilValue float64 + Until uint32 + //TimestampsBuf *conbuf.ContinuousBuffer + //ValuesBuf *conbuf.ContinuousBuffer + Timestamps qb.TimestampCompressor + Values qb.ValueCompressor } func (s *_metric) ReinitBy(timestamp uint32, value float64) { diff --git a/database/proc.go b/database/proc.go index ec758ce..3a58b52 100644 --- a/database/proc.go +++ b/database/proc.go @@ -3,7 +3,7 @@ package database import ( "fmt" - diploma "gordenko.dev/dima/qb" + qb "gordenko.dev/dima/qb" "gordenko.dev/dima/qb/atree" "gordenko.dev/dima/qb/chunkenc" "gordenko.dev/dima/qb/conbuf" @@ -51,19 +51,19 @@ func (s *Database) DoWork() { for _, metricID := range rLocksToRelease { lockEntry, ok := s.metricLockEntries[metricID] if !ok { - diploma.Abort(diploma.NoLockEntryBug, + qb.Abort(qb.NoLockEntryBug, fmt.Errorf("drainQueues: lockEntry not found for the metric %d", metricID)) } if lockEntry.XLock { - diploma.Abort(diploma.XLockBug, + qb.Abort(qb.XLockBug, fmt.Errorf("drainQueues: xlock is set for the metric %d", metricID)) } if lockEntry.RLocks <= 0 { - diploma.Abort(diploma.NoRLockBug, + qb.Abort(qb.NoRLockBug, fmt.Errorf("drainQueues: rlock not set for the metric %d", metricID)) } @@ -73,7 +73,7 @@ func (s *Database) DoWork() { if len(lockEntry.WaitQueue) > 0 { metric, ok := s.metrics[metricID] if !ok { - diploma.Abort(diploma.NoMetricBug, + qb.Abort(qb.NoMetricBug, fmt.Errorf("drainQueues: metric %d not found", metricID)) } s.processMetricQueue(metricID, metric, lockEntry) @@ -117,7 +117,7 @@ func (s *Database) DoWork() { s.tryGetMetric(req) default: - diploma.Abort(diploma.UnknownWorkerQueueItemBug, + qb.Abort(qb.UnknownWorkerQueueItemBug, fmt.Errorf("bug: unknown worker queue item type %T", req)) } } @@ -169,7 +169,7 @@ func (s *Database) processMetricQueue(metricID uint32, metric *_metric, lockEntr s.startDeleteMeasures(metric, req) default: - diploma.Abort(diploma.UnknownMetricWaitQueueItemBug, + qb.Abort(qb.UnknownMetricWaitQueueItemBug, fmt.Errorf("bug: unknown metric wait queue item type %T", req)) } @@ -373,7 +373,7 @@ type ReadBound struct { // return // } -// if metric.MetricType == diploma.Cumulative && req.Value < metric.UntilValue { +// if metric.MetricType == qb.Cumulative && req.Value < metric.UntilValue { // req.ResultCh <- tryAppendMeasureResult{ // MetricID: req.MetricID, // ResultCode: NonMonotonicValue, @@ -434,26 +434,26 @@ type ReadBound struct { // func (s *Database) appendMeasure(rec txlog.AppendedMeasure) { // metric, ok := s.metrics[rec.MetricID] // if !ok { -// diploma.Abort(diploma.NoMetricBug, +// qb.Abort(qb.NoMetricBug, // fmt.Errorf("appendMeasure: metric %d not found", // rec.MetricID)) // } // lockEntry, ok := s.metricLockEntries[rec.MetricID] // if !ok { -// diploma.Abort(diploma.NoLockEntryBug, +// qb.Abort(qb.NoLockEntryBug, // fmt.Errorf("appendMeasure: lockEntry not found for the metric %d", // rec.MetricID)) // } // if lockEntry.XLock { -// diploma.Abort(diploma.XLockBug, +// qb.Abort(qb.XLockBug, // fmt.Errorf("appendMeasure: xlock is set for the metric %d", // rec.MetricID)) // } // if lockEntry.RLocks <= 0 { -// diploma.Abort(diploma.NoRLockBug, +// qb.Abort(qb.NoRLockBug, // fmt.Errorf("appendMeasure: rlock not set for the metric %d", // rec.MetricID)) // } @@ -481,58 +481,49 @@ type ReadBound struct { func (s *Database) appendMeasures(rec txlog.AppendedMeasures) { metric, ok := s.metrics[rec.MetricID] if !ok { - diploma.Abort(diploma.NoMetricBug, + qb.Abort(qb.NoMetricBug, fmt.Errorf("appendMeasureAfterOverflow: metric %d not found", rec.MetricID)) } - lockEntry, ok := s.metricLockEntries[rec.MetricID] - if !ok { - diploma.Abort(diploma.NoLockEntryBug, - fmt.Errorf("appendMeasureAfterOverflow: lockEntry not found for the metric %d", - rec.MetricID)) - } + metric.Values.Unlock() + metric.Timestamps.Unlock() - if !lockEntry.XLock { - diploma.Abort(diploma.NoXLockBug, - fmt.Errorf("appendMeasureAfterOverflow: xlock not set for the metric %d", - rec.MetricID)) - } + // lockEntry, ok := s.metricLockEntries[rec.MetricID] + // if !ok { + // qb.Abort(qb.NoLockEntryBug, + // fmt.Errorf("appendMeasureAfterOverflow: lockEntry not found for the metric %d", + // rec.MetricID)) + // } - for _, measure := range rec.Measures { - if metric.Since == 0 { - metric.Since = measure.Timestamp - metric.SinceValue = measure.Value - } + // if !lockEntry.XLock { + // qb.Abort(qb.NoXLockBug, + // fmt.Errorf("appendMeasureAfterOverflow: xlock not set for the metric %d", + // rec.MetricID)) + // } - metric.Timestamps.Append(measure.Timestamp) - metric.Values.Append(measure.Value) - metric.Until = measure.Timestamp - metric.UntilValue = measure.Value - } - - lockEntry.XLock = false - s.doAfterReleaseXLock(rec.MetricID, metric, lockEntry) + // lockEntry.XLock = false + // s.doAfterReleaseXLock(rec.MetricID, metric, lockEntry) } // func (s *Database) appendMeasureAfterOverflow(extended txlog.AppendedMeasureWithOverflowExtended) { // rec := extended.Record // metric, ok := s.metrics[rec.MetricID] // if !ok { -// diploma.Abort(diploma.NoMetricBug, +// qb.Abort(qb.NoMetricBug, // fmt.Errorf("appendMeasureAfterOverflow: metric %d not found", // rec.MetricID)) // } // lockEntry, ok := s.metricLockEntries[rec.MetricID] // if !ok { -// diploma.Abort(diploma.NoLockEntryBug, +// qb.Abort(qb.NoLockEntryBug, // fmt.Errorf("appendMeasureAfterOverflow: lockEntry not found for the metric %d", // rec.MetricID)) // } // if !lockEntry.XLock { -// diploma.Abort(diploma.NoXLockBug, +// qb.Abort(qb.NoXLockBug, // fmt.Errorf("appendMeasureAfterOverflow: xlock not set for the metric %d", // rec.MetricID)) // } @@ -583,74 +574,170 @@ func (s *Database) tryAppendMeasures(req tryAppendMeasuresReq) { } func (s *Database) startAppendMeasures(metric *_metric, req tryAppendMeasuresReq, lockEntry *metricLockEntry) { - if lockEntry != nil { - if lockEntry.RLocks > 0 { - lockEntry.WaitQueue = append(lockEntry.WaitQueue, req) - return - } - lockEntry.XLock = true - } else { - s.metricLockEntries[req.MetricID] = &metricLockEntry{ - XLock: true, - } - } - var ( - timestampsBuf *conbuf.ContinuousBuffer - valuesBuf *conbuf.ContinuousBuffer - timestamps diploma.TimestampCompressor - values diploma.ValueCompressor + timestamps = metric.Timestamps + values = metric.Values + + dataPages []atree.NotLinkedDataPage + + resultCode byte + written int ) - if metric.Since > 0 { - timestampsBuf = metric.TimestampsBuf.Copy() - valuesBuf = metric.ValuesBuf.Copy() + metric.Values.Lock() + metric.Timestamps.Lock() - timestamps = chunkenc.NewReverseTimeDeltaOfDeltaCompressor( - timestampsBuf, metric.Timestamps.Size()) - - if metric.MetricType == diploma.Cumulative { - values = chunkenc.NewReverseCumulativeDeltaCompressor( - valuesBuf, metric.Values.Size(), metric.FracDigits) + for idx, measure := range req.Measures { + if metric.Since == 0 { + metric.Since = measure.Timestamp } else { - values = chunkenc.NewReverseInstantDeltaCompressor( - valuesBuf, metric.Values.Size(), metric.FracDigits) + // FIX - у випадку помилки треба в транзакції зберегти що помилка, але також зафіксувати скільки елементів збережено. + // якщо idx == 0 - одразу знімаю блокування і нічого не відправляю в txlog + if measure.Timestamp <= metric.Until { + if idx == 0 { + metric.Values.Unlock() + metric.Timestamps.Unlock() + + req.ResultCh <- tryAppendMeasuresResult{ + ResultCode: ExpiredMeasure, + } + return + } + + resultCode = ExpiredMeasure + written = idx + break + } + + if metric.MetricType == qb.Cumulative && measure.Value < metric.UntilValue { + if idx == 0 { + metric.Values.Unlock() + metric.Timestamps.Unlock() + + req.ResultCh <- tryAppendMeasuresResult{ + ResultCode: NonMonotonicValue, + } + return + } + resultCode = NonMonotonicValue + written = idx + break + } } + + // fix - 1 + 8 bytes + extraSpace := timestamps.CalcRequiredSpace(measure.Timestamp) + + values.CalcRequiredSpace(measure.Value) + + totalSpace := timestamps.Size() + values.Size() + extraSpace + + if totalSpace <= atree.DataPagePayloadSize { + // накопичую + timestamps.Append(measure.Timestamp) + values.Append(measure.Value) + } else { + // сторінка заповнена + pageData := make([]byte, atree.PageSize) + + // prevPageNo - виставляю в txlog, коли забираю номер сторінки із freeList або генерую новий + atree.ChunksToNotLinkedDataPage(pageData, atree.ChunksToNotLinkedDataPageReq{ + TimestampsChunks: timestamps.Chunks(), + TimestampsSize: uint16(timestamps.Size()), + ValuesChunks: values.Chunks(), + ValuesSize: uint16(values.Size()), + }) + + dataPages = append(dataPages, atree.NotLinkedDataPage{ + Since: metric.Since, + Data: pageData, + }) + + // FIX + // prevPageNo = report.DataPageNo + // if report.IsRootChanged { + // rootPageNo = report.NewRootPageNo + // } + // waitCh := s.txlog.WriteAppendedMeasureWithOverflow( + // txlog.AppendedMeasureWithOverflow{ + // MetricID: req.MetricID, + // Timestamp: measure.Timestamp, + // Value: measure.Value, + // IsDataPageReused: report.IsDataPageReused, + // DataPageNo: report.DataPageNo, + // IsRootChanged: report.IsRootChanged, + // RootPageNo: report.NewRootPageNo, + // ReusedIndexPages: report.ReusedIndexPages, + // }, + // (idx+1) < len(req.Measures), + // ) + // <-waitCh + + timestamps.Renew() + values.Renew() + + timestamps.Append(measure.Timestamp) + values.Append(measure.Value) + + metric.Since = measure.Timestamp + } + + metric.Until = measure.Timestamp + metric.UntilValue = measure.Value + } + + // виділити змінені байти. + // скопіювати. Причому можна скопіювати зрізи chunks + + if len(dataPages) > 0 { + // пишу в txlog довгим шляхом через redo файл і запис в data файл } else { - timestampsBuf = conbuf.New(nil) - valuesBuf = conbuf.New(nil) - timestamps = chunkenc.NewReverseTimeDeltaOfDeltaCompressor( - timestampsBuf, 0) - if metric.MetricType == diploma.Cumulative { - values = chunkenc.NewReverseCumulativeDeltaCompressor( - valuesBuf, 0, metric.FracDigits) - } else { - values = chunkenc.NewReverseInstantDeltaCompressor( - valuesBuf, 0, metric.FracDigits) - } + // короткий шлях - запис лише в txlog } - req.ResultCh <- tryAppendMeasuresResult{ - ResultCode: CanAppend, - // MetricType: metric.MetricType, - // FracDigits: metric.FracDigits, - // Since: metric.Since, - // Until: metric.Until, - // UntilValue: metric.UntilValue, - // RootPageNo: metric.RootPageNo, - // PrevPageNo: metric.LastPageNo, - // TimestampsBuf: timestampsBuf, - // ValuesBuf: valuesBuf, - // Timestamps: timestamps, - // Values: values, - } + // waitCh := s.txlog.Append(txlog.AppendedPagesReq{ + // MetricID: req.MetricID, + // Timestamp: until, + // Value: untilValue, + // LastPageNo: prevPageNo, + // //Measures: toAppendMeasures, + // }, + // ) + // <-waitCh + + // if lockEntry != nil { + // if lockEntry.RLocks > 0 { + // lockEntry.WaitQueue = append(lockEntry.WaitQueue, req) + // return + // } + // lockEntry.XLock = true + // } else { + // s.metricLockEntries[req.MetricID] = &metricLockEntry{ + // XLock: true, + // } + // } + +} + +type PartialChange struct { + Data [][]byte + Size int + Offset int +} + +type AppendedMeasures struct { + RootPageNo uint32 + LastPageNo uint32 + DataPages []atree.ChunksToNotLinkedDataPageReq + Values PartialChange + Timestamps PartialChange + ResultCh chan tryAppendMeasuresResult } type tryRangeScanReq struct { MetricID uint32 Since uint32 Until uint32 - MetricType diploma.MetricType + MetricType qb.MetricType ResponseWriter atree.WorkerMeasureConsumer ResultCh chan rangeScanResult } @@ -739,7 +826,7 @@ func (*Database) startRangeScan(metric *_metric, req tryRangeScanReq) bool { value, done := valueDecompressor.NextValue() if done { - diploma.Abort(diploma.HasTimestampNoValueBug, ErrNoValueBug) + qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug) } if timestamp <= req.Until { @@ -770,7 +857,7 @@ func (*Database) startRangeScan(metric *_metric, req tryRangeScanReq) bool { type tryFullScanReq struct { MetricID uint32 - MetricType diploma.MetricType + MetricType qb.MetricType ResponseWriter atree.WorkerMeasureConsumer ResultCh chan fullScanResult } @@ -834,7 +921,7 @@ func (*Database) startFullScan(metric *_metric, req tryFullScanReq) bool { } value, done := valueDecompressor.NextValue() if done { - diploma.Abort(diploma.HasTimestampNoValueBug, ErrNoValueBug) + qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug) } req.ResponseWriter.FeedNoSend(timestamp, value) } @@ -913,31 +1000,31 @@ func (s *Database) applyChanges(req txlog.Changes) { func (s *Database) addMetric(rec txlog.AddedMetric) { _, ok := s.metrics[rec.MetricID] if ok { - diploma.Abort(diploma.MetricAddedBug, + qb.Abort(qb.MetricAddedBug, fmt.Errorf("addMetric: metric %d already added", rec.MetricID)) } lockEntry, ok := s.metricLockEntries[rec.MetricID] if !ok { - diploma.Abort(diploma.NoLockEntryBug, + qb.Abort(qb.NoLockEntryBug, fmt.Errorf("addMetric: lockEntry not found for the metric %d", rec.MetricID)) } if !lockEntry.XLock { - diploma.Abort(diploma.NoXLockBug, + qb.Abort(qb.NoXLockBug, fmt.Errorf("addMetric: xlock not set for the metric %d", rec.MetricID)) } var ( - values diploma.ValueCompressor + values qb.ValueCompressor timestampsBuf = conbuf.New(nil) valuesBuf = conbuf.New(nil) ) - if rec.MetricType == diploma.Cumulative { + if rec.MetricType == qb.Cumulative { values = chunkenc.NewReverseCumulativeDeltaCompressor( valuesBuf, 0, byte(rec.FracDigits)) } else { @@ -961,20 +1048,20 @@ func (s *Database) addMetric(rec txlog.AddedMetric) { func (s *Database) deleteMetric(rec txlog.DeletedMetric) { _, ok := s.metrics[rec.MetricID] if !ok { - diploma.Abort(diploma.NoMetricBug, + qb.Abort(qb.NoMetricBug, fmt.Errorf("deleteMetric: metric %d not found", rec.MetricID)) } lockEntry, ok := s.metricLockEntries[rec.MetricID] if !ok { - diploma.Abort(diploma.NoLockEntryBug, + qb.Abort(qb.NoLockEntryBug, fmt.Errorf("deleteMetric: lockEntry not found for the metric %d", rec.MetricID)) } if !lockEntry.XLock { - diploma.Abort(diploma.NoXLockBug, + qb.Abort(qb.NoXLockBug, fmt.Errorf("deleteMetric: xlock not set for the metric %d", rec.MetricID)) } @@ -1019,7 +1106,7 @@ func (s *Database) deleteMetric(rec txlog.DeletedMetric) { } default: - diploma.Abort(diploma.UnknownMetricWaitQueueItemBug, + qb.Abort(qb.UnknownMetricWaitQueueItemBug, fmt.Errorf("bug: unknown metric wait queue item type %T", req)) } } @@ -1039,20 +1126,20 @@ func (s *Database) deleteMetric(rec txlog.DeletedMetric) { func (s *Database) deleteMeasures(rec txlog.DeletedMeasures) { metric, ok := s.metrics[rec.MetricID] if !ok { - diploma.Abort(diploma.NoMetricBug, + qb.Abort(qb.NoMetricBug, fmt.Errorf("deleteMeasures: metric %d not found", rec.MetricID)) } lockEntry, ok := s.metricLockEntries[rec.MetricID] if !ok { - diploma.Abort(diploma.NoLockEntryBug, + qb.Abort(qb.NoLockEntryBug, fmt.Errorf("deleteMeasures: lockEntry not found for the metric %d", rec.MetricID)) } if !lockEntry.XLock { - diploma.Abort(diploma.NoXLockBug, + qb.Abort(qb.NoXLockBug, fmt.Errorf("deleteMeasures: xlock not set for the metric %d", rec.MetricID)) } diff --git a/qb.go b/qb.go index 1be7f8d..448a3df 100644 --- a/qb.go +++ b/qb.go @@ -26,7 +26,11 @@ type TimestampCompressor interface { CalcRequiredSpace(uint32) int Append(uint32) Size() int + Chunks() [][]byte DeleteLast() + Renew() + Lock() + Unlock() //LastTimestamp() uint32 } @@ -34,7 +38,11 @@ type ValueCompressor interface { CalcRequiredSpace(float64) int Append(float64) Size() int + Chunks() [][]byte DeleteLast() + Renew() // створює новий conbuf, але не чіпає state + Lock() + Unlock() //LastValue() float64 }