diff --git a/database/metric.go b/database/metric.go index 4f3b263..c718fc9 100644 --- a/database/metric.go +++ b/database/metric.go @@ -31,6 +31,7 @@ type _metric struct { RLocks int WaitQueue []any indexLevelTails []storage.IndexLevelTail // root - last element + capturedState *CapturedState } //IndexLevels [][]IndexRec // root - last element @@ -59,59 +60,45 @@ func (s *_metric) DeleteMeasures() { // s.UntilValue = 0 } +type CapturedState struct { + Pages []storage.DataPayload + TimeState qb.TimeDeltaCapturedState + ValueState qb.ValueDeltaCapturedState +} + // func (s *_metric) startAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(storage.AppendedMeasures)) { func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, sendToStorage func(any)) { + if s.capturedState != nil { + s.WaitQueue = append(s.WaitQueue, req) + } var ( timestamps = s.timestamps values = s.values - dataPages []storage.DataPayload - // офсети на head сторінці timestampsOffset int valuesOffset int - //written int - //resultCode byte + written int + resultCode byte ) - s.values.CaptureState() - s.timestamps.CaptureState() + s.capturedState = &CapturedState{ + // + TimeState: s.timestamps.CaptureState(), + ValueState: s.values.CaptureState(), + } for idx, measure := range req.Measures { - // FIX - у випадку помилки треба в транзакції зберегти що помилка, але також зафіксувати скільки елементів збережено. - // якщо idx == 0 - одразу знімаю блокування і нічого не відправляю в storage if measure.Timestamp <= s.timestamps.LastTimestamp() { - if idx == 0 { - s.values.ForgetCapturedState() - s.timestamps.ForgetCapturedState() - - req.ResultCh <- tryAppendMeasuresResult{ - ResultCode: ExpiredMeasure, - } - return - } - - //resultCode = ExpiredMeasure - //written = idx + resultCode = ExpiredMeasure break } if s.metricType == qb.Cumulative && measure.Value < s.lastValue { - if idx == 0 { - s.values.ForgetCapturedState() - s.timestamps.ForgetCapturedState() - - req.ResultCh <- tryAppendMeasuresResult{ - ResultCode: NonMonotonicValue, - } - return - } - //resultCode = NonMonotonicValue - //written = idx + resultCode = NonMonotonicValue break } - //} // fix - 1 + 8 bytes @@ -144,13 +131,13 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, send // сторінка заповнена since := s.timestamps.ReplaceSinceWithUntil() - if len(dataPages) == 0 { + if len(s.capturedState.Pages) == 0 { // head page FIX // timestampsPayload = // valuesPayload = } - dataPages = append(dataPages, storage.DataPayload{ + s.capturedState.Pages = append(s.capturedState.Pages, storage.DataPayload{ Since: since, Content: s.buffer, TimestampsSize: timestamps.Size(), @@ -171,38 +158,54 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, send values.Append(vReport.Offset, tmp[7:7+vReport.ChangeSize], measure.Value, vReport.Delta) // s.lastValue = measure.Value + written++ + } + + if written == 0 { + s.capturedState = nil + req.ResultCh <- tryAppendMeasuresResult{ + ResultCode: resultCode, + } + return } // виділити змінені байти. // скопіювати. Причому можна скопіювати зрізи chunks - if len(dataPages) > 0 { + if len(s.capturedState.Pages) > 0 { + head := s.capturedState.Pages[0] + databuf := head.Content[:storage.DataPagePayloadSize] // пишу в storage довгим шляхом через redo файл і запис в data файл sendToStorage(storage.AppendedMeasures{ MetricID: req.MetricID, LastPageNo: s.lastPageNo, TimestampsOffset: timestampsOffset, + Timestamps: databuf[len(databuf)-head.TimestampsSize : len(databuf)-timestampsOffset], ValuesOffset: valuesOffset, - //Payload: s.payload, // fix - timestamps + values ? or timestamps and values (for WAL) - IndexLevelTails: s.indexLevelTails, - DataPages: dataPages, - Timestamps: nil, - Values: nil, - //ResultCode: resultCode, - //WrittenCount: wri, - ResultCh: nil, + Values: databuf[valuesOffset:head.ValuesSize], + IndexLevelTails: s.indexLevelTails, + DataPages: s.capturedState.Pages, + TailTimestamps: nil, // payload + TailValues: s.buffer[:values.Size()], + ResultCode: resultCode, + WrittenCount: written, + ResultCh: nil, }) } else { + databuf := s.buffer + if len(s.buffer) == storage.DataPageSize { + databuf = s.buffer[:storage.DataPagePayloadSize] + } // короткий шлях - запис лише в storage sendToStorage(storage.AppendedMeasures{ MetricID: req.MetricID, TimestampsOffset: timestampsOffset, ValuesOffset: valuesOffset, - Timestamps: nil, // pos - offset - Values: nil, - //ResultCode: resultCode, - //WrittenCount: wri, - ResultCh: nil, + Timestamps: databuf[len(databuf)-timestamps.Size() : len(databuf)-timestampsOffset], + Values: databuf[valuesOffset:values.Size()], + ResultCode: resultCode, + WrittenCount: written, + ResultCh: nil, }) } @@ -210,8 +213,7 @@ func (s *_metric) StartAppendMeasures(req tryAppendMeasuresReq, tmp []byte, send func (s *_metric) FinAppendMeasures(rec storage.AppendMeasuresSummary) { // Видаляю state. Оригінальні Timestamps і Values вже мають останню версію - s.values.ForgetCapturedState() - s.timestamps.ForgetCapturedState() + s.capturedState = nil if rec.LastPageNo > 0 { s.lastPageNo = rec.LastPageNo @@ -367,13 +369,21 @@ func (s *_metric) WriteTo(w io.Writer) (err error) { if err != nil { return } + var ( + timestampsCapturedState *qb.TimeDeltaCapturedState + valuesCapturedState *qb.ValueDeltaCapturedState + ) + if s.capturedState != nil { + timestampsCapturedState = &s.capturedState.TimeState + valuesCapturedState = &s.capturedState.ValueState + } // timestamps payload - err = s.timestamps.WritePayloadTo(w) + err = s.timestamps.WritePayloadTo(w, timestampsCapturedState) if err != nil { return } // values payload - err = s.values.WritePayloadTo(w) + err = s.values.WritePayloadTo(w, valuesCapturedState) if err != nil { return } @@ -398,3 +408,32 @@ func (s *_metric) WriteTo(w io.Writer) (err error) { // since - 4b // sinceValue - 8b - // untilValue - 8b + +// time + +// if s.state != nil { +// h = s.state.H +// lastUnixtime = s.state.LastUnixtime +// lastDelta = s.state.LastDelta +// payload = s.state.Payload +// } +// return NewTimeDeltaDecompressorFromState(TimeDeltaDecompressorFromStateOptions{ +// H: h, +// LastUnixtime: lastUnixtime, +// LastDelta: lastDelta, +// Payload: payload, +// }) + +// value +// if s.state != nil { +// h = s.state.H +// lastDelta = s.state.LastDelta +// payload = s.state.Payload +// } +// return NewValueDeltaDecompressorFromState(ValueDeltaDecompressorFromStateOptions{ +// Coef: s.coef, +// ToFloat64: s.toFloat64, +// H: s.buf[s.pos-1], +// LastDelta: s.lastDelta, +// Payload: s.buf[:s.pos], +// }) diff --git a/database/proc.go b/database/proc.go index 0e5eb7d..b099f2a 100644 --- a/database/proc.go +++ b/database/proc.go @@ -132,7 +132,7 @@ func (s *Database) processMetricQueue(metricID uint32, metric *_metric) { s.tryGetMetric(req) case tryAppendMeasuresReq: - metric.StartAppendMeasures(req, s.storage.Append) + //metric.StartAppendMeasures(req, s.storage.Append) FIX case tryDeleteMetricReq: s.startDeleteMetric(metric, req) @@ -322,7 +322,7 @@ func (s *Database) tryAppendMeasures(req tryAppendMeasuresReq) { metric.WaitQueue = append(metric.WaitQueue, req) return } - metric.StartAppendMeasures(req, s.storage.Append) + //metric.StartAppendMeasures(req, s.storage.Append) FIX } type tryRangeScanReq struct { @@ -472,20 +472,14 @@ func (s *Database) applyAddMetric(rec storage.MetricAddRecord) { func (s *Database) addMetric(rec storage.MetricAddRecord) { var ( - values qb.ValueCompressor buffer = make([]byte, minBufferSize) ) - if rec.MetricType == qb.Cumulative { - values = enc.NewCumulativeDeltaCompressor(byte(rec.FracDigits), buffer, 0) - } else { - values = enc.NewInstantDeltaCompressor(byte(rec.FracDigits), buffer, 0) - } s.metrics[rec.MetricID] = &_metric{ metricType: rec.MetricType, fracDigits: byte(rec.FracDigits), buffer: buffer, timestamps: enc.NewTimeDeltaCompressor(buffer, 0), - values: values, + values: enc.NewValueDeltaCompressor(rec.MetricType, byte(rec.FracDigits), buffer, 0), } } diff --git a/database/replay_metric.go b/database/replay_metric.go index 7d1fb31..2628b05 100644 --- a/database/replay_metric.go +++ b/database/replay_metric.go @@ -272,7 +272,6 @@ func composeHeadDataPage(databuf []byte, head storage.HeadDataPage) storage.Page func (s *ReplayMetric) ToMetric() *_metric { var ( - values qb.ValueCompressor requiredSpace = s.tSize + s.vSize ) bufferSize := calculateDataBufferSize(requiredSpace) @@ -284,11 +283,7 @@ func (s *ReplayMetric) ToMetric() *_metric { NewBufferSize: bufferSize, }) } - if s.metricType == qb.Cumulative { - values = enc.NewCumulativeDeltaCompressor(s.fracDigits, s.buf, s.vSize) - } else { - values = enc.NewInstantDeltaCompressor(s.fracDigits, s.buf, s.vSize) - } + values := enc.NewValueDeltaCompressor(s.metricType, s.fracDigits, s.buf, s.vSize) return &_metric{ metricType: s.metricType, fracDigits: s.fracDigits, diff --git a/enc/enc.go b/enc/enc.go index 80bc136..44f3f6a 100644 --- a/enc/enc.go +++ b/enc/enc.go @@ -5,14 +5,14 @@ const eps = 0.000001 const ( flagLiteral = 128 - hSize = 1 + //hSize = 1 // append scripts - addBaseValue = 0 // cumulative and instant only - addUnixtime = 0 // time only - incrementRun = 1 - incrementLiteral = 2 - endSeries = 3 - startRun = 4 - add1stDelta = 5 // time only + // addBaseValue = 0 // cumulative and instant only + // addUnixtime = 0 // time only + // incrementRun = 1 + // incrementLiteral = 2 + // endSeries = 3 + // startRun = 4 + // add1stDelta = 5 // time only ) diff --git a/enc/time_delta.go b/enc/time_delta.go index 1138d8c..529ca18 100644 --- a/enc/time_delta.go +++ b/enc/time_delta.go @@ -28,7 +28,6 @@ type TimeDeltaCompressor struct { pos int lastUnixtime uint32 lastDelta uint32 - state *TimeDeltaCapturedState } // Початкове створення, коли даних немає @@ -61,11 +60,6 @@ func NewTimeDeltaCompressor(buf []byte, payloadSize int) *TimeDeltaCompressor { return s } -// FIX - restire lastUnixtime -// func (s *TimeDeltaCompressor) RestoreState() { - -// } - func (s *TimeDeltaCompressor) Evaluate(tmp []byte, timestamp uint32) qb.TimeEvaluationReport { var ( delta = timestamp - s.lastUnixtime @@ -160,20 +154,10 @@ func (s *TimeDeltaCompressor) Append(offset int, change []byte, timestamp uint32 // } -type TimeDeltaCapturedState struct { - H byte - LastUnixtime uint32 - LastDelta uint32 - Payload []byte -} - // delta h -func (s *TimeDeltaCompressor) CaptureState() { - if s.state != nil { - qb.Abort(qb.RepeatableLock, nil) - } +func (s *TimeDeltaCompressor) CaptureState() qb.TimeDeltaCapturedState { // позиція посувається вліво, отже може перескочити на попередній chunk - s.state = &TimeDeltaCapturedState{ + return qb.TimeDeltaCapturedState{ H: s.buf[s.pos], LastUnixtime: s.lastUnixtime, LastDelta: s.lastDelta, @@ -181,11 +165,6 @@ func (s *TimeDeltaCompressor) CaptureState() { } } -// fix - повернути в Pool буфери -func (s *TimeDeltaCompressor) ForgetCapturedState() { - s.state = nil -} - // коли сторінка заповнена і відправляється в txlog, since замінюю на until func (s *TimeDeltaCompressor) ReplaceSinceWithUntil() uint32 { pos := len(s.buf) - 4 @@ -218,34 +197,26 @@ func (s *TimeDeltaCompressor) Size() int { // } // } -func (s *TimeDeltaCompressor) WritePayloadTo(w io.Writer) (err error) { - if s.state == nil { +func (s *TimeDeltaCompressor) WritePayloadTo(w io.Writer, state *qb.TimeDeltaCapturedState) (err error) { + if state == nil { _, err = w.Write(s.buf[:s.pos]) return } else { - _, err = w.Write(s.state.Payload) + _, err = w.Write(state.Payload) if err != nil { return } - _, err = bin.WriteVarUint64(w, uint64(s.state.LastDelta)) + _, err = bin.WriteVarUint64(w, uint64(state.LastDelta)) if err != nil { return } _, err = w.Write([]byte{ - s.state.H, + state.H, }) return } } -// func (s *TimeDeltaCompressor) encodeTail(lastUnixtime, lastDelta uint32, h byte) []byte { -// tail := make([]byte, 9) -// bin.PutUint32(tail, lastUnixtime) -// tail[4] = h -// n, _ := bin.PutVarUint64(tail[5:], uint64(lastDelta)) -// return tail[:n+5] -// } - func (s *TimeDeltaCompressor) Rotate(newbuf []byte) { // УВАГА! // state не чіпаємо @@ -259,23 +230,11 @@ func (s *TimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor { if s.pos == len(s.buf) { return nil } - var ( - h = s.buf[s.pos] - lastUnixtime = s.lastUnixtime - lastDelta = s.lastDelta - payload = s.buf[s.pos:] - ) - if s.state != nil { - h = s.state.H - lastUnixtime = s.state.LastUnixtime - lastDelta = s.state.LastDelta - payload = s.state.Payload - } return NewTimeDeltaDecompressorFromState(TimeDeltaDecompressorFromStateOptions{ - H: h, - LastUnixtime: lastUnixtime, - LastDelta: lastDelta, - Payload: payload, + H: s.buf[s.pos], + LastUnixtime: s.lastUnixtime, + LastDelta: s.lastDelta, + Payload: s.buf[s.pos:], }) } @@ -306,7 +265,6 @@ func NewTimeDeltaDecompressor(buf []byte) *TimeDeltaDecompressor { log.Fatalf("bug: get last unixtime: %s", err) } s.pos += 4 - //fmt.Println("restored last unixtime", s.lastUnixtime) } else { s.done = true } @@ -329,15 +287,10 @@ func NewTimeDeltaDecompressorFromState(opt TimeDeltaDecompressorFromStateOptions s.lastDelta = opt.LastDelta s.decodeHeaderByte(opt.H) } - //fmt.Println("-------------------") - //fmt.Printf("restore from bound: isRun=%t, pending=%d, lastUnix=%d, lastDelta=%d\n", - //s.isRun, s.pending, s.lastUnixtime, s.lastDelta) - //fmt.Printf("buf: %d\n", s.buf) return s } func (s *TimeDeltaDecompressor) NextValue() (value uint32, done bool) { - //fmt.Printf("NextValue(): pos: %d, isRegular: %t, pending: %d\n", s.pos, s.isRegular, s.pending) if s.done { return 0, true } @@ -346,7 +299,6 @@ func (s *TimeDeltaDecompressor) NextValue() (value uint32, done bool) { if s.lastDelta > 0 { s.lastUnixtime -= s.lastDelta s.pending-- - //fmt.Printf("lastUnix: %d, s.pending: %d\n", s.lastUnixtime, s.pending) if s.pending > 0 { // якщо в серії залишаються елементи if !s.isRun { @@ -358,7 +310,6 @@ func (s *TimeDeltaDecompressor) NextValue() (value uint32, done bool) { s.readHeader() s.readDelta() } else { - //s.done = true s.lastDelta = 0 } } else { @@ -372,12 +323,6 @@ func (s *TimeDeltaDecompressor) readHeader() { h := s.buf[s.pos] s.pos++ s.decodeHeaderByte(h) - // fmt.Println() - // fmt.Println("read from pos:", s.pos) - // fmt.Println("h:", h) - // fmt.Println("isRun:", s.isRun) - // fmt.Println("pending:", s.pending) - // fmt.Println() } func (s *TimeDeltaDecompressor) decodeHeaderByte(h byte) { @@ -394,11 +339,6 @@ func (s *TimeDeltaDecompressor) readDelta() { if err != nil { log.Fatalln(err) } - // fmt.Println() - // fmt.Println("read from pos:", s.pos) - // fmt.Println("read delta:", u64) - // fmt.Println("read delta n:", n) - // fmt.Println() s.pos += n s.lastDelta = uint32(u64) } diff --git a/enc/value_delta.go b/enc/value_delta.go index 4e64714..ddb596f 100644 --- a/enc/value_delta.go +++ b/enc/value_delta.go @@ -19,7 +19,6 @@ type ValueDeltaCompressor struct { pos int // payload size baseValue float64 lastDelta uint64 - state *ValueDeltaCapturedState // (baseValue, coef, value) => delta calcDelta func(float64, float64, float64) uint64 // (value, coef) => uint64 @@ -75,18 +74,6 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu ) if s.pos > 0 { delta = s.calcDelta(s.baseValue, s.coef, value) - // if false { - // delta = uint64((value-s.baseValue)*s.coef + eps) - - // } else { - // f64 := (value - s.baseValue) * s.coef - // if f64 > 0 { - // f64 += eps - // } else { - // f64 -= eps - // } - // delta = bin.EncodeZigZag(int64(f64)) - // } h := s.buf[s.pos-1] if h < 128 { // run @@ -137,12 +124,6 @@ func (s *ValueDeltaCompressor) Evaluate(tmp []byte, value float64) qb.ValueEvalu } } } else { - // var baseValue uint64 - // if false { - // baseValue = uint64(value * s.coef) - // } else { - // baseValue = bin.EncodeZigZag(int64(value * s.coef)) - // } n, _ := bin.PutVarUint64(tmp, s.toUint64(value, s.coef)) i += n n, _ = bin.ReversePutVarUint64(tmp[i:], 0) // delta @@ -173,30 +154,19 @@ func (s *ValueDeltaCompressor) DeleteLast() { } -type ValueDeltaCapturedState struct { - H byte - LastDelta uint64 - Payload []byte -} - -func (s *ValueDeltaCompressor) CaptureState() { - if s.state != nil { - qb.Abort(qb.RepeatableLock, nil) - } +func (s *ValueDeltaCompressor) CaptureState() qb.ValueDeltaCapturedState { + // if s.state != nil { + // qb.Abort(qb.RepeatableLock, nil) + // } // позиція посувається вліво, отже може перескочити на попередній chunk pos := s.pos - 1 - bin.CountVarUint64(s.lastDelta) - s.state = &ValueDeltaCapturedState{ + return qb.ValueDeltaCapturedState{ H: s.buf[s.pos-1], LastDelta: s.lastDelta, Payload: s.buf[:pos], } } -// fix - повернути в Pool буфери -func (s *ValueDeltaCompressor) ForgetCapturedState() { - s.state = nil -} - // для зростання буфера під час вставки даних. State не цікавить func (s *ValueDeltaCompressor) Size() int { //if s.state == nil { @@ -215,21 +185,21 @@ func (s *ValueDeltaCompressor) Size() int { // } // } -func (s *ValueDeltaCompressor) WritePayloadTo(w io.Writer) (err error) { - if s.state == nil { +func (s *ValueDeltaCompressor) WritePayloadTo(w io.Writer, state *qb.ValueDeltaCapturedState) (err error) { + if state == nil { _, err = w.Write(s.buf[:s.pos]) return } else { - _, err = w.Write(s.state.Payload) + _, err = w.Write(state.Payload) if err != nil { return } - _, err = bin.WriteVarUint64(w, s.state.LastDelta) + _, err = bin.WriteVarUint64(w, state.LastDelta) if err != nil { return } _, err = w.Write([]byte{ - s.state.H, + state.H, }) return } @@ -245,35 +215,27 @@ func (s *ValueDeltaCompressor) Rotate(newbuf []byte) { } func (s *ValueDeltaCompressor) LastValue() float64 { - var delta uint64 - if s.state == nil { - delta = s.lastDelta - } else { - delta = s.state.LastDelta - } - return s.baseValue + s.toFloat64(delta, s.coef) + // FIX + // var delta uint64 + // if s.state == nil { + // delta = s.lastDelta + // } else { + // delta = s.state.LastDelta + // } + // return s.baseValue + s.toFloat64(delta, s.coef) + return 0 } func (s *ValueDeltaCompressor) CreateDecompressor() qb.ValueDecompressor { if s.pos == 0 { return nil } - var ( - h = s.buf[s.pos-1] - lastDelta = s.lastDelta - payload = s.buf[:s.pos] - ) - if s.state != nil { - h = s.state.H - lastDelta = s.state.LastDelta - payload = s.state.Payload - } return NewValueDeltaDecompressorFromState(ValueDeltaDecompressorFromStateOptions{ Coef: s.coef, ToFloat64: s.toFloat64, - H: h, - LastDelta: lastDelta, - Payload: payload, + H: s.buf[s.pos-1], + LastDelta: s.lastDelta, + Payload: s.buf[:s.pos], }) } @@ -302,17 +264,10 @@ func NewValueDeltaDecompressor(fracDigits byte, buf []byte, toFloat64 func(uint6 if err != nil { log.Fatalf("bug: get base value: %s", err) } - // if false { - // baseValue = float64(u64) / coef - // } else { - // baseValue = float64(bin.DecodeZigZag(u64)) / coef - // } - //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/coef, n, size) s := &ValueDeltaDecompressor{ - buf: buf, - coef: coef, - pos: len(buf), // first free - // baseValue: baseValue, + buf: buf, + coef: coef, + pos: len(buf), // first free bound: n, toFloat64: toFloat64, } @@ -337,32 +292,16 @@ func NewValueDeltaDecompressorFromState(opt ValueDeltaDecompressorFromStateOptio if err != nil { log.Fatalf("bug: get base value: %s", err) } - // var baseValue float64 - // if false { - // baseValue = float64(u64) / opt.Coef - // } else { - // baseValue = float64(bin.DecodeZigZag(u64)) / opt.Coef - // } - //fmt.Printf("baseValue: %.2f, bound: %d, pos: %d\n", float64(u64)/opt.Coef, n, size) s := &ValueDeltaDecompressor{ - buf: opt.Payload, - coef: opt.Coef, - pos: len(opt.Payload), // first free - //baseValue: baseValue, + buf: opt.Payload, + coef: opt.Coef, + pos: len(opt.Payload), // first free bound: n, toFloat64: opt.ToFloat64, } s.baseValue = s.toFloat64(u64, s.coef) - // if false { - // s.lastValue = s.baseValue + float64(opt.LastDelta)/s.coef - // } else { - // s.lastValue = s.baseValue + float64(bin.DecodeZigZag(opt.LastDelta))/s.coef - // } s.lastValue = s.baseValue + s.toFloat64(opt.LastDelta, s.coef) s.decodeHeaderByte(opt.H) - //fmt.Printf("payload: %d\n", opt.Payload) - //fmt.Printf("restore from state: bound=%d, isRun=%t, pending=%d, baseValue=%v, lastValue=%v\n", - // n, s.isRun, s.pending, s.baseValue, s.lastValue) return s } @@ -392,33 +331,17 @@ func (s *ValueDeltaDecompressor) NextValue() (value float64, done bool) { } func (s *ValueDeltaDecompressor) readHeader() { - //fmt.Println("read from pos:", s.pos) s.pos-- h := s.buf[s.pos] s.decodeHeaderByte(h) - - // fmt.Println("h:", h) - // fmt.Println("isRun:", s.isRun) - // fmt.Println("pending:", s.pending) } func (s *ValueDeltaDecompressor) readValue() { u64, n, err := bin.ReverseGetVarUint64(s.buf[:s.pos]) if err != nil { log.Fatalln(err) } - // fmt.Println() - // fmt.Println("read from pos:", s.pos) - // fmt.Println("read delta:", u64) - // fmt.Println("read delta n:", n) - // fmt.Println() s.pos -= n - // if false { - // s.lastValue = s.baseValue + float64(u64)/s.coef - // } else { - // s.lastValue = s.baseValue + float64(bin.DecodeZigZag(u64))/s.coef - // } s.lastValue = s.baseValue + s.toFloat64(u64, s.coef) - //fmt.Println(s.baseValue, float64(u64)/s.coef) } func (s *ValueDeltaDecompressor) decodeHeaderByte(h byte) { diff --git a/qb.go b/qb.go index 083b4ba..670dcc0 100644 --- a/qb.go +++ b/qb.go @@ -31,17 +31,23 @@ type TimestampCompressor interface { Size() int //Chunks() [][]byte //DeleteLast() - CaptureState() - ForgetCapturedState() + CaptureState() TimeDeltaCapturedState CreateDecompressor() TimestampDecompressor //Payload() []byte // для снапшота // Offset() int Rotate([]byte) - WritePayloadTo(io.Writer) error + WritePayloadTo(io.Writer, *TimeDeltaCapturedState) error LastTimestamp() uint32 ReplaceSinceWithUntil() uint32 } +type TimeDeltaCapturedState struct { + H byte + LastUnixtime uint32 + LastDelta uint32 + Payload []byte +} + type TimeEvaluationReport struct { TotalSpace int Offset int @@ -56,17 +62,22 @@ type ValueCompressor interface { Size() int //Chunks() [][]byte //DeleteLast() - CaptureState() - ForgetCapturedState() + CaptureState() ValueDeltaCapturedState // fracDigits CreateDecompressor() ValueDecompressor //Payload() []byte // для снапшота //Offset() int Rotate([]byte) - WritePayloadTo(io.Writer) error + WritePayloadTo(io.Writer, *ValueDeltaCapturedState) error LastValue() float64 } +type ValueDeltaCapturedState struct { + H byte + LastDelta uint64 + Payload []byte +} + type ValueEvaluationReport struct { TotalSpace int Offset int diff --git a/storage/data_preparer.go b/storage/data_preparer.go index e009abd..37e724d 100644 --- a/storage/data_preparer.go +++ b/storage/data_preparer.go @@ -189,7 +189,7 @@ type AppendedMeasures struct { DataPages []DataPayload TailTimestamps []byte // data level tail TailValues []byte // data level tail - ResultCode int + ResultCode byte WrittenCount int ResultCh chan struct{} } @@ -201,7 +201,7 @@ type AppendMeasuresSummary struct { MetricID uint32 LastPageNo uint32 Index []IndexLevelTail - ResultCode int + ResultCode byte WrittenCount int ResultCh chan struct{} }