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-05-14 16:06:37 +03:00
parent ef60296027
commit 5754bd1aa7
5 changed files with 231 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

303
database/proc.go
View File

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