wp
This commit is contained in:
445
txlog/writer.go
445
txlog/writer.go
@@ -9,9 +9,9 @@ import (
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
bin "gordenko.dev/dima/bin/little"
|
||||
octopus "gordenko.dev/dima/qb"
|
||||
"gordenko.dev/dima/qb/bin"
|
||||
"gordenko.dev/dima/qb/proto"
|
||||
"gordenko.dev/dima/qb/atree"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -28,14 +28,19 @@ const (
|
||||
)
|
||||
|
||||
const (
|
||||
CodeAddedMetric byte = 1
|
||||
CodeDeletedMetric byte = 2
|
||||
CodeAppendedMeasure byte = 4
|
||||
CodeAppendedMeasures byte = 5
|
||||
CodeAppendedMeasureWithOverflow byte = 6
|
||||
CodeDeletedMeasures byte = 7
|
||||
CodeAddedMetric byte = 1
|
||||
CodeDeletedMetric byte = 2
|
||||
CodeAppendedMeasure byte = 4
|
||||
CodeAppendedMeasures byte = 5
|
||||
CodeAppendedPages byte = 6
|
||||
CodeDeletedMeasures byte = 7
|
||||
)
|
||||
|
||||
type FreeList interface {
|
||||
// використовується в allocPage
|
||||
ReservePage() uint32
|
||||
}
|
||||
|
||||
func JoinChangesFileName(dir string, logNumber int) string {
|
||||
return filepath.Join(dir, fmt.Sprintf("%d.changes", logNumber))
|
||||
}
|
||||
@@ -49,11 +54,16 @@ type Changes struct {
|
||||
}
|
||||
|
||||
type Writer struct {
|
||||
mutex sync.Mutex
|
||||
logNumber int
|
||||
dir string
|
||||
file *os.File
|
||||
buf *bytes.Buffer
|
||||
mutex sync.Mutex
|
||||
freelist FreeList
|
||||
atree *atree.Atree
|
||||
logNumber int
|
||||
dir string
|
||||
dataFile *os.File
|
||||
file *os.File
|
||||
input []any
|
||||
//buf *bytes.Buffer
|
||||
pagesToWrite []atree.PageToWrite
|
||||
workerReqs []any
|
||||
waitCh chan struct{}
|
||||
appendToWorkerQueue func(any)
|
||||
@@ -69,6 +79,8 @@ type WriterOptions struct {
|
||||
Dir string
|
||||
LogNumber int // номер журнала
|
||||
AppendToWorkerQueue func(any)
|
||||
FreeList FreeList
|
||||
Atree *atree.Atree
|
||||
ExitCh chan struct{}
|
||||
WaitGroup *sync.WaitGroup
|
||||
}
|
||||
@@ -80,6 +92,12 @@ func NewWriter(opt WriterOptions) (*Writer, error) {
|
||||
if opt.AppendToWorkerQueue == nil {
|
||||
return nil, errors.New("AppendToWorkerQueue option is required")
|
||||
}
|
||||
if opt.FreeList == nil {
|
||||
return nil, errors.New("FreeList option is required")
|
||||
}
|
||||
if opt.Atree == nil {
|
||||
return nil, errors.New("Atree option is required")
|
||||
}
|
||||
if opt.ExitCh == nil {
|
||||
return nil, errors.New("ExitCh option is required")
|
||||
}
|
||||
@@ -89,12 +107,14 @@ func NewWriter(opt WriterOptions) (*Writer, error) {
|
||||
|
||||
s := &Writer{
|
||||
dir: opt.Dir,
|
||||
buf: bytes.NewBuffer(nil),
|
||||
appendToWorkerQueue: opt.AppendToWorkerQueue,
|
||||
freelist: opt.FreeList,
|
||||
atree: opt.Atree,
|
||||
logNumber: opt.LogNumber,
|
||||
exitCh: opt.ExitCh,
|
||||
waitGroup: opt.WaitGroup,
|
||||
signalCh: make(chan struct{}, 1),
|
||||
waitCh: make(chan struct{}),
|
||||
}
|
||||
|
||||
var err error
|
||||
@@ -119,8 +139,6 @@ func NewWriter(opt WriterOptions) (*Writer, error) {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
s.reset()
|
||||
return s, nil
|
||||
}
|
||||
|
||||
@@ -128,7 +146,7 @@ func (s *Writer) Run() {
|
||||
for {
|
||||
select {
|
||||
case <-s.signalCh:
|
||||
if err := s.flush(); err != nil {
|
||||
if err := s.packAndWrite(); err != nil {
|
||||
octopus.Abort(octopus.FailedWriteToTxLog, err)
|
||||
}
|
||||
|
||||
@@ -139,21 +157,67 @@ func (s *Writer) Run() {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Writer) reset() {
|
||||
s.buf.Reset()
|
||||
s.buf.Write([]byte{
|
||||
0, 0, 0, 0, // packet length
|
||||
0, 0, 0, 0, // crc32
|
||||
0, 0, 0, 0, // lsn
|
||||
|
||||
})
|
||||
s.workerReqs = nil
|
||||
func (s *Writer) packAndWrite() error {
|
||||
s.mutex.Lock()
|
||||
input := s.input
|
||||
waitCh := s.waitCh
|
||||
s.input = nil
|
||||
s.waitCh = make(chan struct{})
|
||||
s.mutex.Unlock()
|
||||
|
||||
w := bytes.NewBuffer(nil)
|
||||
|
||||
for _, untyped := range input {
|
||||
switch x := untyped.(type) {
|
||||
case AppendedPagesReq:
|
||||
s.packWriteAppended(w, x)
|
||||
case AddedMetric:
|
||||
case DeletedMetric:
|
||||
if len(x.FreePageNumbers) > 0 {
|
||||
//s.freeList.AddPages(x.FreePageNumbers)
|
||||
}
|
||||
case AppendedMeasure:
|
||||
case AppendedMeasures:
|
||||
case DeletedMeasures:
|
||||
//case DeletedMeasuresSince:
|
||||
}
|
||||
}
|
||||
// flush
|
||||
close(waitCh)
|
||||
return s.flush(w)
|
||||
}
|
||||
|
||||
func (s *Writer) flush() error {
|
||||
func (s *Writer) Append(req any) chan struct{} {
|
||||
s.mutex.Lock()
|
||||
s.input = append(s.input, req)
|
||||
s.workerReqs = append(s.workerReqs, req)
|
||||
waitCh := s.waitCh
|
||||
s.mutex.Unlock()
|
||||
|
||||
select {
|
||||
case s.signalCh <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
return waitCh
|
||||
}
|
||||
|
||||
// func (s *Writer) reset() {
|
||||
// s.buf.Reset()
|
||||
// s.buf.Write([]byte{
|
||||
// 0, 0, 0, 0, // packet length
|
||||
// 0, 0, 0, 0, // crc32
|
||||
// 0, 0, 0, 0, // lsn
|
||||
|
||||
// })
|
||||
// s.pagesToWrite = nil
|
||||
// s.workerReqs = nil
|
||||
// s.waitCh = make(chan struct{})
|
||||
// }
|
||||
|
||||
func (s *Writer) flush(w *bytes.Buffer) error {
|
||||
s.mutex.Lock()
|
||||
|
||||
pagesToWrite := s.pagesToWrite
|
||||
workerReqs := s.workerReqs
|
||||
waitCh := s.waitCh
|
||||
isExited := s.isExited
|
||||
@@ -163,12 +227,12 @@ func (s *Writer) flush() error {
|
||||
exitWaitGroup = s.waitGroup
|
||||
}
|
||||
|
||||
if s.buf.Len() > packetPrefixSize {
|
||||
if w.Len() > packetPrefixSize {
|
||||
s.lsn++
|
||||
lsn := s.lsn
|
||||
packet := make([]byte, s.buf.Len())
|
||||
copy(packet, s.buf.Bytes())
|
||||
s.reset()
|
||||
packet := make([]byte, w.Len())
|
||||
copy(packet, w.Bytes())
|
||||
//s.reset() fix
|
||||
|
||||
s.written += int64(len(packet)) + 12
|
||||
s.mutex.Unlock()
|
||||
@@ -181,14 +245,17 @@ func (s *Writer) flush() error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("TxLog write: %s", err)
|
||||
}
|
||||
|
||||
if n != len(packet) {
|
||||
return fmt.Errorf("TxLog written %d != packet size %d", n, len(packet))
|
||||
}
|
||||
|
||||
if err := s.file.Sync(); err != nil {
|
||||
return fmt.Errorf("TxLog sync: %s", err)
|
||||
}
|
||||
|
||||
err = s.writePagesToAtree(pagesToWrite)
|
||||
if err != nil {
|
||||
return fmt.Errorf("TxLog writePagesToAtree: %s", err)
|
||||
}
|
||||
} else {
|
||||
s.waitCh = make(chan struct{})
|
||||
s.mutex.Unlock()
|
||||
@@ -232,263 +299,48 @@ func (s *Writer) flush() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Writer) writePagesToAtree(pages []atree.PageToWrite) error {
|
||||
for _, p := range pages {
|
||||
if len(p.Data) != atree.PageSize {
|
||||
return fmt.Errorf("wrong page %d size: %d",
|
||||
p.PageNo, len(p.Data))
|
||||
}
|
||||
|
||||
off := (p.PageNo - 1) * atree.PageSize
|
||||
n, err := s.dataFile.WriteAt(p.Data, int64(off))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n != len(p.Data) {
|
||||
return fmt.Errorf("write %d instead of %d", n, len(p.Data))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Writer) exit() {
|
||||
s.mutex.Lock()
|
||||
s.isExited = true
|
||||
s.mutex.Unlock()
|
||||
|
||||
if err := s.flush(); err != nil {
|
||||
if err := s.packAndWrite(); err != nil {
|
||||
octopus.Abort(octopus.FailedWriteToTxLog, err)
|
||||
}
|
||||
}
|
||||
|
||||
// API
|
||||
|
||||
type AddedMetric struct {
|
||||
MetricID uint32
|
||||
MetricType octopus.MetricType
|
||||
FracDigits int
|
||||
}
|
||||
|
||||
func (s *Writer) WriteAddedMetric(req AddedMetric) chan struct{} {
|
||||
arr := []byte{
|
||||
CodeAddedMetric,
|
||||
0, 0, 0, 0, //
|
||||
byte(req.MetricType),
|
||||
byte(req.FracDigits),
|
||||
}
|
||||
bin.PutUint32(arr[1:], req.MetricID)
|
||||
// пишу в буфер
|
||||
s.mutex.Lock()
|
||||
s.buf.Write(arr)
|
||||
s.workerReqs = append(s.workerReqs, req)
|
||||
s.mutex.Unlock()
|
||||
|
||||
s.sendSignal()
|
||||
return s.waitCh
|
||||
}
|
||||
|
||||
type DeletedMetric struct {
|
||||
MetricID uint32
|
||||
FreeDataPages []uint32
|
||||
FreeIndexPages []uint32
|
||||
}
|
||||
|
||||
func (s *Writer) WriteDeletedMetric(req DeletedMetric) chan struct{} {
|
||||
arr := []byte{
|
||||
CodeDeletedMetric,
|
||||
0, 0, 0, 0, // metricID
|
||||
}
|
||||
bin.PutUint32(arr[1:], req.MetricID)
|
||||
|
||||
// пишу в буфер
|
||||
s.mutex.Lock()
|
||||
defer s.mutex.Unlock()
|
||||
|
||||
s.buf.Write(arr)
|
||||
s.packFreeDataAndIndexPages(req.FreeDataPages, req.FreeIndexPages)
|
||||
s.workerReqs = append(s.workerReqs, req)
|
||||
|
||||
s.sendSignal()
|
||||
return s.waitCh
|
||||
}
|
||||
|
||||
type AppendedMeasure struct {
|
||||
MetricID uint32
|
||||
Timestamp uint32
|
||||
Value float64
|
||||
}
|
||||
|
||||
func (s *Writer) WriteAppendMeasure(req AppendedMeasure) chan struct{} {
|
||||
arr := []byte{
|
||||
CodeAppendedMeasure,
|
||||
0, 0, 0, 0, // metricID
|
||||
0, 0, 0, 0, // timestamp
|
||||
0, 0, 0, 0, 0, 0, 0, 0, // value
|
||||
}
|
||||
bin.PutUint32(arr[1:], req.MetricID)
|
||||
bin.PutUint32(arr[5:], req.Timestamp)
|
||||
bin.PutFloat64(arr[9:], req.Value)
|
||||
//
|
||||
s.mutex.Lock()
|
||||
s.buf.Write(arr)
|
||||
s.workerReqs = append(s.workerReqs, req)
|
||||
s.mutex.Unlock()
|
||||
|
||||
s.sendSignal()
|
||||
return s.waitCh
|
||||
}
|
||||
|
||||
type AppendedMeasures struct {
|
||||
MetricID uint32
|
||||
Measures []proto.Measure
|
||||
}
|
||||
|
||||
type AppendedMeasuresExtended struct {
|
||||
Record AppendedMeasures
|
||||
HoldLock bool
|
||||
}
|
||||
|
||||
func (s *Writer) WriteAppendMeasures(req AppendedMeasures, holdLock bool) chan struct{} {
|
||||
arr := []byte{
|
||||
CodeAppendedMeasures,
|
||||
0, 0, 0, 0, // metricID
|
||||
0, 0, // qty
|
||||
}
|
||||
bin.PutUint32(arr[1:], req.MetricID)
|
||||
bin.PutUint16(arr[5:], uint16(len(req.Measures)))
|
||||
//
|
||||
s.mutex.Lock()
|
||||
s.buf.Write(arr)
|
||||
for _, measure := range req.Measures {
|
||||
bin.WriteUint32(s.buf, measure.Timestamp)
|
||||
bin.WriteFloat64(s.buf, measure.Value)
|
||||
}
|
||||
s.workerReqs = append(s.workerReqs, AppendedMeasuresExtended{
|
||||
Record: req,
|
||||
HoldLock: holdLock,
|
||||
})
|
||||
s.mutex.Unlock()
|
||||
|
||||
s.sendSignal()
|
||||
return s.waitCh
|
||||
}
|
||||
|
||||
type AppendedMeasureWithOverflow struct {
|
||||
MetricID uint32
|
||||
Timestamp uint32
|
||||
Value float64
|
||||
IsDataPageReused bool
|
||||
DataPageNo uint32
|
||||
IsRootChanged bool
|
||||
RootPageNo uint32
|
||||
ReusedIndexPages []uint32
|
||||
}
|
||||
|
||||
type AppendedMeasureWithOverflowExtended struct {
|
||||
Record AppendedMeasureWithOverflow
|
||||
HoldLock bool
|
||||
}
|
||||
|
||||
/*
|
||||
Формат:
|
||||
1b code
|
||||
4b metricID
|
||||
4b timestamp
|
||||
8b value
|
||||
1b isReusedDataPage
|
||||
4b dataPageNo
|
||||
1b isRootChanged
|
||||
[4b] newRootPageNo
|
||||
1b reusedIndexPages length
|
||||
[N * 4b] reusedIndexPages
|
||||
------------------
|
||||
|
||||
4b metricID
|
||||
4b timestamp
|
||||
8b value
|
||||
4b newRootPageNo (0 if no new root page)
|
||||
|
||||
Nb - varsize (N pages)
|
||||
[
|
||||
|
||||
4b pageNo
|
||||
1b isReused
|
||||
Xb page (page size)
|
||||
|
||||
]
|
||||
Nb - varsize (last data page size)
|
||||
Nb - payload
|
||||
*/
|
||||
func (s *Writer) WriteAppendedMeasureWithOverflow(req AppendedMeasureWithOverflow, holdLock bool) chan struct{} {
|
||||
size := 24 + len(req.ReusedIndexPages)*4
|
||||
if req.IsRootChanged {
|
||||
size += 4
|
||||
}
|
||||
|
||||
tmp := make([]byte, size)
|
||||
|
||||
tmp[0] = CodeAppendedMeasureWithOverflow
|
||||
bin.PutUint32(tmp[1:], req.MetricID)
|
||||
bin.PutUint32(tmp[5:], req.Timestamp)
|
||||
bin.PutFloat64(tmp[9:], req.Value)
|
||||
if req.IsDataPageReused {
|
||||
tmp[17] = 1
|
||||
}
|
||||
bin.PutUint32(tmp[18:], req.DataPageNo)
|
||||
|
||||
pos := 22
|
||||
if req.IsRootChanged {
|
||||
tmp[pos] = 1
|
||||
bin.PutUint32(tmp[pos+1:], req.RootPageNo)
|
||||
pos += 5
|
||||
} else {
|
||||
tmp[pos] = 0
|
||||
pos += 1
|
||||
}
|
||||
|
||||
tmp[pos] = byte(len(req.ReusedIndexPages))
|
||||
pos += 1
|
||||
for _, indexPageNo := range req.ReusedIndexPages {
|
||||
bin.PutUint32(tmp[pos:], indexPageNo)
|
||||
pos += 4
|
||||
}
|
||||
|
||||
s.mutex.Lock()
|
||||
s.buf.Write(tmp)
|
||||
s.workerReqs = append(s.workerReqs, AppendedMeasureWithOverflowExtended{
|
||||
Record: req,
|
||||
HoldLock: holdLock,
|
||||
})
|
||||
s.mutex.Unlock()
|
||||
|
||||
s.sendSignal()
|
||||
return s.waitCh
|
||||
}
|
||||
|
||||
type DeletedMeasures struct {
|
||||
MetricID uint32
|
||||
FreeDataPages []uint32
|
||||
FreeIndexPages []uint32
|
||||
}
|
||||
|
||||
/*
|
||||
Формат:
|
||||
1b code
|
||||
4b metricID
|
||||
1b freeDataPages length
|
||||
[N * 4b] freeDataPages
|
||||
1b freeIndexPages length
|
||||
[N * 4b] freeIndexPages
|
||||
*/
|
||||
func (s *Writer) WriteDeletedMeasures(op DeletedMeasures) chan struct{} {
|
||||
tmp := []byte{
|
||||
CodeDeletedMeasures,
|
||||
0, 0, 0, 0,
|
||||
}
|
||||
bin.PutUint32(tmp[1:], op.MetricID)
|
||||
|
||||
// записываю часть фиксированного размера
|
||||
s.mutex.Lock()
|
||||
s.buf.Write(tmp)
|
||||
s.packFreeDataAndIndexPages(op.FreeDataPages, op.FreeIndexPages)
|
||||
s.workerReqs = append(s.workerReqs, op)
|
||||
s.mutex.Unlock()
|
||||
|
||||
s.sendSignal()
|
||||
return s.waitCh
|
||||
}
|
||||
|
||||
type DeletedMeasuresSince struct {
|
||||
MetricID uint32
|
||||
LastPageNo uint32
|
||||
IsRootChanged bool
|
||||
RootPageNo uint32
|
||||
FreeDataPages []uint32
|
||||
FreeIndexPages []uint32
|
||||
TimestampsBuf []byte
|
||||
ValuesBuf []byte
|
||||
}
|
||||
// FIX - add
|
||||
// type DeletedMeasuresSince struct {
|
||||
// MetricID uint32
|
||||
// LastPageNo uint32
|
||||
// IsRootChanged bool
|
||||
// RootPageNo uint32
|
||||
// FreeDataPages []uint32
|
||||
// FreeIndexPages []uint32
|
||||
// TimestampsBuf []byte
|
||||
// ValuesBuf []byte
|
||||
// }
|
||||
|
||||
func (s *Writer) sendSignal() {
|
||||
select {
|
||||
@@ -497,17 +349,30 @@ func (s *Writer) sendSignal() {
|
||||
}
|
||||
}
|
||||
|
||||
// helper
|
||||
// Якщо (Pages) > 0 - незаповнену сторінку кодую в стисненому вигляді.
|
||||
// Якщо (Pages) = 0 - кодую просто пари timestamp / value
|
||||
func (s *Writer) packWriteAppended(w *bytes.Buffer, req AppendedPagesReq) {
|
||||
// завантажений path. Отже просто додаємо в дерево data сторінки, створюємо нові індексні
|
||||
// без звернення до диску.
|
||||
report := s.atree.AppendDataPages(atree.AppendDataPagesReq{
|
||||
LastPageNo: req.LastPageNo,
|
||||
Legs: req.Legs,
|
||||
DataPages: req.Pages,
|
||||
})
|
||||
|
||||
func (s *Writer) packFreeDataAndIndexPages(freeDataPages, freeIndexPages []uint32) {
|
||||
// записываю data pages
|
||||
bin.WriteVarUint64(s.buf, uint64(len(freeDataPages)))
|
||||
for _, dataPageNo := range freeDataPages {
|
||||
bin.WriteUint32(s.buf, dataPageNo)
|
||||
}
|
||||
// записываю index pages
|
||||
bin.WriteVarUint64(s.buf, uint64(len(freeIndexPages)))
|
||||
for _, indexPageNo := range freeIndexPages {
|
||||
bin.WriteUint32(s.buf, indexPageNo)
|
||||
m := AppendedPages{
|
||||
MetricID: req.MetricID,
|
||||
Timestamp: req.Timestamp,
|
||||
Value: req.Value,
|
||||
NewRootPageNo: report.NewRootPageNo,
|
||||
LastPageNo: report.LastPageNo,
|
||||
Pages: report.Pages,
|
||||
TimestampsChunks: req.TimestampsChunks,
|
||||
TimestampsSize: int(req.TimestampsSize),
|
||||
ValuesChunks: req.ValuesChunks,
|
||||
ValuesSize: int(req.ValuesSize),
|
||||
}
|
||||
m.Pack(w)
|
||||
}
|
||||
|
||||
// helpers
|
||||
|
||||
Reference in New Issue
Block a user