dima/qb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wp

Browse Source
This commit is contained in:
Dima Gordenko
2026-06-14 23:12:03 +03:00
parent 181031a753
commit b32279deaf
35 changed files with 1395 additions and 1062 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
worker/metric.go
View File

@@ -2,6 +2,7 @@ package worker
import (
"errors"
"fmt"
"io"
bin "gordenko.dev/dima/bin/little"
@@ -30,8 +31,8 @@ type Metric struct {
buffer []byte
timestamps qb.TimestampCompressor
values qb.ValueCompressor
XLock bool
RLocks int
xLock bool
rLocks int
WaitQueue []any
indexLevelTails []storage.IndexLevelTail // root - last element
capturedState *CapturedState
@@ -62,7 +63,7 @@ func (s *Metric) LastTimestamp() uint32 {
func (s *Metric) OnMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
//s.timestamps.Reset()
//s.values.Reset()
s.XLock = false
s.xLock = false
// s.Timestamps.Renew()
// s.Values.Renew()
@@ -75,8 +76,7 @@ func (s *Metric) OnMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
s.lastValue = 0
}
// func (s *_metric) startAppendMeasures(req tryAppendMeasuresReq, sendToStorage func(storage.AppendedMeasures)) {
func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox *inbox.Inbox) {
func (s *Metric) AppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox *inbox.Inbox) {
if s.capturedState != nil {
s.WaitQueue = append(s.WaitQueue, req)
}
@@ -96,7 +96,7 @@ func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, storageI
pages []storage.DataPayload
written int
resultCode byte
resultCode byte = Succeed
)
s.capturedState = &CapturedState{
@@ -117,8 +117,10 @@ func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, storageI
break
}
tReport := timestamps.Evaluate(tmp, measure.Timestamp)
vReport := values.Evaluate(tmp, 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
@@ -169,7 +171,7 @@ func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, storageI
if written == 0 {
s.capturedState = nil
req.ResultCh <- AppendMeasuresResult{
req.ResultCh <- storage.MeasuresAppendResult{
ResultCode: resultCode,
}
return
@@ -193,7 +195,7 @@ func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, storageI
TailValues: values.Tail(0),
ResultCode: resultCode,
WrittenCount: written,
//ResultCh: req.ResultCh,
ResultCh: req.ResultCh,
})
} else {
// короткий шлях - запис лише в storage
@@ -205,7 +207,7 @@ func (s *Metric) StartAppendMeasures(req AppendMeasuresReq, tmp []byte, storageI
Values: values.Tail(valuesOffset),
ResultCode: resultCode,
WrittenCount: written,
//ResultCh: req.ResultCh,
ResultCh: req.ResultCh,
})
}
}
@@ -215,6 +217,11 @@ func (s *Metric) OnMeasuresAppendCommited(rec storage.MeasuresAppendCommited) {
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) {
@@ -229,6 +236,11 @@ func (s *Metric) OnMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWith
// В storage я передав повний індекс. У нього додали елементи (можливо нові рівні).
// Тому проста заміна
s.indexLevelTails = rec.Index
rec.ResultCh <- storage.MeasuresAppendResult{
ResultCode: rec.ResultCode,
WrittenCount: rec.WrittenCount,
}
}
// READ
@@ -317,10 +329,12 @@ func (s *Metric) StartFullScan(req FullScanReq) {
if done {
break
}
fmt.Println("ts:", timestamp)
value, done := valueDecompressor.NextValue()
if done {
qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug)
}
fmt.Println("value:", value)
req.ResponseWriter.FeedNoSend(timestamp, value)
}
@@ -330,7 +344,7 @@ func (s *Metric) StartFullScan(req FullScanReq) {
LastPageNo: s.lastPageNo,
FracDigits: s.fracDigits,
}
s.RLocks++
s.rLocks++
} else {
req.ResultCh <- FullScanResult{
ResultCode: QueryDone,
@@ -366,21 +380,21 @@ func (s *Metric) WriteTo(w io.Writer) (err error) {
if err != nil {
return
}
err = bin.WriteUint16(w, uint16(s.timestamps.StoredSize()))
err = bin.WriteUint16(w, uint16(s.timestamps.CommitedSize()))
if err != nil {
return
}
err = bin.WriteUint16(w, uint16(s.values.StoredSize()))
err = bin.WriteUint16(w, uint16(s.values.CommitedSize()))
if err != nil {
return
}
// timestamps payload
err = s.timestamps.WriteStoredTo(w)
err = s.timestamps.WriteCommitedTo(w)
if err != nil {
return
}
// values payload
err = s.values.WriteStoredTo(w)
err = s.values.WriteCommitedTo(w)
if err != nil {
return
}
@@ -394,7 +408,7 @@ func (s *Metric) WriteTo(w io.Writer) (err error) {
if err != nil {
return
}
_, err = w.Write(level.Buffer)
_, err = w.Write(level.Buffer[:level.RecordsCount*storage.IndexRecordSize])
if err != nil {
return
}

69
worker/snapshot.go
View File

@@ -5,7 +5,6 @@ import (
"fmt"
"io"
"os"
"path/filepath"
bin "gordenko.dev/dima/bin/little"
"gordenko.dev/dima/qb"
@@ -30,11 +29,10 @@ dataFreeList - Nb
CRC32 - 4b
*/
const readBufferSize = 4 * 1024 * 1024 // 8mb
const readBufferSize = 4 * 1024 * 1024
type WriteSnapshotIn struct {
SnapshotNumber int
Dir string
WriteBufferSize int
Metrics map[uint32]*Metric
FrozenIndexPagesCount int
@@ -43,24 +41,21 @@ type WriteSnapshotIn struct {
DataPageNumbers []uint32
}
func WriteSnapshot(in WriteSnapshotIn) (err error) {
var (
fileName = filepath.Join(in.Dir, fmt.Sprintf("%d.snapshot", in.SnapshotNumber))
hasher = util.NewHasher()
)
file, err := os.OpenFile(fileName, os.O_CREATE|os.O_WRONLY, 0666) //0770)
func writeSnapshot(filePath string, in WriteSnapshotIn) (err error) {
file, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
if err != nil {
return
qb.Abort(qb.WriteSnapshotFailed, err)
}
dst := io.MultiWriter(bufio.NewWriterSize(file, in.WriteBufferSize), hasher)
var (
bufferedWriter = bufio.NewWriterSize(file, in.WriteBufferSize)
hasher = util.NewHasher()
)
dst := io.MultiWriter(bufferedWriter, hasher)
//
_, err = bin.WriteVarSize(dst, len(in.Metrics))
if err != nil {
return
}
for metricID, metric := range in.Metrics {
err = bin.WriteUint32(dst, metricID)
if err != nil {
@@ -81,13 +76,16 @@ func WriteSnapshot(in WriteSnapshotIn) (err error) {
if err != nil {
return
}
// CRC32
bin.WriteUint32(file, hasher.Sum32())
err = file.Close()
err = bufferedWriter.Flush()
if err != nil {
return
}
// CRC32
err = bin.WriteUint32(file, hasher.Sum32())
if err != nil {
return
}
return file.Close()
// prevLogNumber := logNumber - 1
// prevChanges := filepath.Join(s.dir, fmt.Sprintf("%d.changes", prevLogNumber))
@@ -116,7 +114,6 @@ func WriteSnapshot(in WriteSnapshotIn) (err error) {
// qb.Abort(qb.DeletePrevSnapshotFileFailed, err)
// }
// }
return
}
type ReadSnapshotOut struct {
@@ -127,7 +124,7 @@ type ReadSnapshotOut struct {
DataPageNumbers []uint32
}
func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
func ReadSnapshot(filePath string) (out ReadSnapshotOut, err error) {
var (
metrics = make(map[uint32]*storage.ReplayMetric)
frozenIndexPagesCount int
@@ -137,7 +134,7 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
hasher = util.NewHasher()
)
file, err := os.Open(fileName)
file, err := os.Open(filePath)
if err != nil {
return
}
@@ -148,25 +145,16 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
return
}
fileSize := stat.Size()
if fileSize == 0 {
return ReadSnapshotOut{
Metrics: metrics,
}, nil
}
if fileSize < 4 {
err = fmt.Errorf("%s is corrupted", fileName)
err = fmt.Errorf("snapshot %s is corrupted", filePath)
return
}
payloadSize := fileSize - 4
// 2. Створюємо великий буфер для NVMe (4 MB)
bufferedReader := bufio.NewReaderSize(file, readBufferSize)
// 3. Обмежуємо читання лише розміром Payload
limitReader := io.LimitReader(bufferedReader, payloadSize)
// 4. Створюємо хеш та обгортку TeeReader, яка рахує CRC32 на льоту
src := io.TeeReader(limitReader, hasher)
@@ -189,6 +177,7 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
if err != nil {
return
}
fmt.Println("metricID", metricID)
metricType, err = bin.ReadByte(src)
if err != nil {
return
@@ -198,10 +187,13 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
if err != nil {
return
}
fmt.Println("m.MetricType", m.MetricType)
fmt.Println("m.FracDigits", m.FracDigits)
m.LastPageNo, err = bin.ReadUint32(src)
if err != nil {
return
}
fmt.Println("m.LastPageNo", m.LastPageNo)
m.TimestampsSize, err = bin.ReadUint16AsInt(src)
if err != nil {
return
@@ -210,7 +202,8 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
if err != nil {
return
}
err = bin.ReadNInto(src, buf[len(buf)-m.TimestampsSize:])
fmt.Println("m.TimestampsSize", m.TimestampsSize, "m.ValuesSize", m.ValuesSize)
err = bin.ReadNInto(src, buf[storage.DataPagePayloadSize-m.TimestampsSize:storage.DataPagePayloadSize])
if err != nil {
return
}
@@ -218,11 +211,13 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
if err != nil {
return
}
fmt.Printf("% x\n", buf)
//
levelsCount, err = bin.ReadVarSize(src)
if err != nil {
return
}
fmt.Println("levels count:", levelsCount)
for range levelsCount {
var (
buf = make([]byte, storage.IndexPageSize)
@@ -232,10 +227,12 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
if err != nil {
return
}
fmt.Println("records count:", count)
err = bin.ReadNInto(src, buf[:count*storage.IndexRecordSize])
if err != nil {
return
}
fmt.Printf("% x\n", buf)
m.IndexLevelTails = append(m.IndexLevelTails, storage.IndexLevelTail{
Buffer: buf,
RecordsCount: count,
@@ -258,12 +255,10 @@ func ReadSnapshot(fileName string) (out ReadSnapshotOut, err error) {
if err != nil {
return
}
calculatedCRC := hasher.Sum32()
if expectedCRC != calculatedCRC {
err = fmt.Errorf("%s is corrupted. Calculated CRC %d not equal expected CRC %d",
fileName, calculatedCRC, expectedCRC)
filePath, calculatedCRC, expectedCRC)
return
}
@@ -283,6 +278,7 @@ func writeFreePages(dst io.Writer, frozenPagesCount int, pageNumbers []uint32) (
if err != nil {
return
}
fmt.Println("write frozenPagesCount:", frozenPagesCount)
_, err = bin.WriteVarSize(dst, len(pageNumbers))
if err != nil {
return
@@ -305,6 +301,7 @@ func readFreePages(src io.Reader) (_ int, _ []uint32, err error) {
if err != nil {
return
}
fmt.Println("read frozenPagesCount:", frozenPagesCount)
pageNumbersCount, err := bin.ReadVarSize(src)
if err != nil {
return

93
worker/snapshot_test.go
View File

@@ -1,25 +1,74 @@
package worker
import (
"bytes"
"fmt"
"os"
"reflect"
"slices"
"testing"
"gordenko.dev/dima/qb"
"gordenko.dev/dima/qb/enc"
"gordenko.dev/dima/qb/storage"
)
func TestWriteReadSnapshot(t *testing.T) {
metrics := make(map[uint32]*Metric)
metrics[1] = &Metric{
metricType: qb.Cumulative,
fracDigits: 3,
lastPageNo: 1000,
buffer: []byte{},
storage.DataPageSize = 32
storage.DataPagePayloadSize = 20
storage.IndexPageSize = 16
buf := []byte{
// values
0x8f, // base value
0x80, // delta 0
0x01, // run (len = 3)
0x81, // delta 1
0x81, // literal (len = 2)
// empty space
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00,
// timestamps
0x80, // h-byte (literal, len=1)
0x94, // delta 20
0x01, // h-byte (run, len=3)
0xbc, // delta 60
0x28, 0x80, 0x24, 0x6a, // since
// footer
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
}
databuf := buf[:storage.DataPagePayloadSize]
timestampsSize := 8
valuesSize := 5
metrics := map[uint32]*Metric{
1: {
metricType: qb.Cumulative,
fracDigits: 3,
lastPageNo: 1000,
buffer: buf,
timestamps: enc.NewTimeDeltaCompressor(databuf, timestampsSize),
values: enc.NewValueDeltaCompressor(qb.Cumulative, 3, databuf, valuesSize),
indexLevelTails: []storage.IndexLevelTail{
{
Buffer: []byte{
0x01, 0x01, 0x01, 0x01,
0x02, 0x02, 0x02, 0x02,
0x00,
// footer
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00,
},
RecordsCount: 1,
},
},
},
}
in := WriteSnapshotIn{
SnapshotNumber: 1,
Dir: ".",
WriteBufferSize: 8 * 1024 * 1024,
Metrics: metrics,
FrozenIndexPagesCount: 7,
@@ -31,14 +80,17 @@ func TestWriteReadSnapshot(t *testing.T) {
6, 7, 8,
},
}
err := WriteSnapshot(in)
fileName := "test.snapshot"
err := writeSnapshot(fileName, in)
if err != nil {
t.Fatalf("writeSnapshot: %s", err)
}
out, err := ReadSnapshot("1.snapshot")
out, err := ReadSnapshot(fileName)
if err != nil {
return
t.Fatal(err)
}
if out.FrozenIndexPagesCount != in.FrozenIndexPagesCount {
@@ -67,6 +119,7 @@ func TestWriteReadSnapshot(t *testing.T) {
t.Fatalf("metric %d: %s", metricID, err)
}
}
os.Remove(fileName)
}
func cmpMetric(replay *storage.ReplayMetric, origin *Metric) error {
@@ -82,9 +135,21 @@ func cmpMetric(replay *storage.ReplayMetric, origin *Metric) error {
return fmt.Errorf("lastPageNo: got %d are not equal expected %d",
replay.LastPageNo, origin.lastPageNo)
}
// if !reflect.DeepEqual() {
// return fmt.Errorf("lastPageNo: got %d are not equal expected %d",
// replay.LastPageNo, origin.lastPageNo)
// }
if !reflect.DeepEqual(replay.IndexLevelTails, origin.indexLevelTails) {
return fmt.Errorf("indexLevelTails: got %#v are not equal expected %#v",
replay.IndexLevelTails, origin.indexLevelTails)
}
if !bytes.Equal(replay.Buf, origin.buffer) {
return fmt.Errorf("buffer: got % x are not equal expected % x",
replay.Buf, origin.buffer)
}
if replay.TimestampsSize != origin.timestamps.Size() {
return fmt.Errorf("timestampsSize: got %d are not equal expected %d",
replay.TimestampsSize, origin.timestamps.Size())
}
if replay.ValuesSize != origin.values.Size() {
return fmt.Errorf("valuesSize: got %d are not equal expected %d",
replay.ValuesSize, origin.values.Size())
}
return nil
}

160
worker/worker.go
View File

@@ -19,28 +19,27 @@ import (
// Якийсь прапор поставити для 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
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
@@ -53,6 +52,7 @@ type Options struct {
StorageInbox *inbox.Inbox
ReplayMetrics map[uint32]*storage.ReplayMetric
Dir string
DatabaseName string
ExitCh chan struct{}
WaitGroup *sync.WaitGroup
}
@@ -62,6 +62,7 @@ func New(opt Options) *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,
@@ -151,13 +152,13 @@ func (s *Worker) doWork() {
s.applyCommits(req) // all metrics only
case ListCurrentValuesReq:
s.tryListCurrentValues(req) // all metrics only
s.ListCurrentValues(req) // all metrics only
case RangeScanReq:
s.RangeScan(req)
case FullScanReq:
s.tryFullScan(req)
s.FullScan(req)
case AddMetricReq:
s.AddMetric(req)
@@ -196,7 +197,7 @@ func (s *Worker) processMetricQueue(metricID uint32, metric *Metric, tmp []byte)
s.GetMetric(req)
case AppendMeasuresReq:
metric.StartAppendMeasures(req, tmp, s.storageInbox)
metric.AppendMeasures(req, tmp, s.storageInbox)
case DeleteMetricReq:
s.startDeleteMetric(metric, req)
@@ -212,28 +213,37 @@ func (s *Worker) processMetricQueue(metricID uint32, metric *Metric, tmp []byte)
}
type AddMetricReq struct {
MetricID uint32
ResultCh chan byte
MetricID uint32
MetricType qb.MetricType
FracDigits byte
ResultCh chan byte
}
func (s *Worker) AddMetric(req AddMetricReq) {
_, ok := s.metrics[req.MetricID]
if ok {
fmt.Println("add metric duplicate")
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
// }
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,
})
}
func (s *Worker) processTryAddMetricReqsImmediatelyAfterDelete(reqs []AddMetricReq) {
@@ -379,7 +389,7 @@ 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",
fmt.Errorf("onMeasuresAppendCommited: metric %d not found",
rec.MetricID))
}
metric.OnMeasuresAppendCommited(rec)
@@ -395,30 +405,25 @@ func (s *Worker) onMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWith
metric.OnMeasuresAppendWithGrowCommited(rec)
}
type AppendMeasuresResult struct {
ResultCode byte
Written int
}
type AppendMeasuresReq struct {
MetricID uint32
Measures []proto.Measure
ResultCh chan AppendMeasuresResult
ResultCh chan storage.MeasuresAppendResult
}
func (s *Worker) AppendMeasures(req AppendMeasuresReq) {
metric, ok := s.metrics[req.MetricID]
if !ok {
req.ResultCh <- AppendMeasuresResult{
req.ResultCh <- storage.MeasuresAppendResult{
ResultCode: NoMetric,
}
return
}
if metric.XLock {
if metric.xLock {
metric.WaitQueue = append(metric.WaitQueue, req)
return
}
metric.StartAppendMeasures(req, s.tmp, s.storageInbox)
metric.AppendMeasures(req, s.tmp, s.storageInbox)
}
type RangeScanResult struct {
@@ -452,7 +457,7 @@ func (s *Worker) RangeScan(req RangeScanReq) {
return
}
if metric.XLock {
if metric.xLock {
metric.WaitQueue = append(metric.WaitQueue, req)
return
}
@@ -473,7 +478,7 @@ type FullScanReq struct {
ResultCh chan FullScanResult
}
func (s *Worker) tryFullScan(req FullScanReq) {
func (s *Worker) FullScan(req FullScanReq) {
metric, ok := s.metrics[req.MetricID]
if !ok {
req.ResultCh <- FullScanResult{
@@ -488,7 +493,7 @@ func (s *Worker) tryFullScan(req FullScanReq) {
return
}
if metric.XLock {
if metric.xLock {
metric.WaitQueue = append(metric.WaitQueue, req)
return
}
@@ -501,7 +506,7 @@ type ListCurrentValuesReq struct {
ResultCh chan struct{}
}
func (s *Worker) tryListCurrentValues(req ListCurrentValuesReq) {
func (s *Worker) ListCurrentValues(req ListCurrentValuesReq) {
for _, metricID := range req.MetricIDs {
metric, ok := s.metrics[metricID]
if ok {
@@ -539,55 +544,32 @@ func (s *Worker) applyCommits(req storage.Changes) {
if req.SnapshotNumberCh != nil {
s.snapshotNumber++
err := WriteSnapshot(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,
})
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) {
// fix lock
_, ok := s.metrics[rec.MetricID]
if ok {
metric, 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),
fmt.Errorf("metric %d not found after commit", rec.MetricID))
}
metric.xLock = false
rec.ResultCh <- Succeed // new
}
func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) {
@@ -598,7 +580,7 @@ func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) {
rec.MetricID))
}
if !metric.XLock {
if !metric.xLock {
qb.Abort(qb.NoXLockBug,
fmt.Errorf("deleteMetric: xlock not set for the metric %d",
rec.MetricID))
@@ -669,13 +651,13 @@ func (s *Worker) onMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
rec.MetricID))
}
if !metric.XLock {
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
metric.xLock = false
// FIX add in storage
// if len(rec.FreePageNumbers) > 0 {
// s.freeList.AddPages(rec.FreePageNumbers)