diff --git a/atree/atree.go b/atree/atree.go index ee2dd0f..efce9d8 100644 --- a/atree/atree.go +++ b/atree/atree.go @@ -1,14 +1,31 @@ package atree import ( + "errors" "os" "sync" ) const ( - filePerm = 0770 + filePerm = 0666 ) +type PeriodsWriter interface { + Feed(uint32, float64) + FeedNoSend(uint32, float64) + Close() error +} + +type WorkerMeasureConsumer interface { + FeedNoSend(uint32, float64) +} + +type AtreeMeasureConsumer interface { + Feed(uint32, float64) +} + +// + type _page struct { PageNo uint32 Buf []byte @@ -16,66 +33,35 @@ type _page struct { } type Atree struct { - file *os.File - mutex sync.Mutex - allocatedPagesQty uint32 - pages map[uint32]*_page - pageWaits map[uint32][]chan readResult - pagesToRead []uint32 - readSignalCh chan struct{} - writeSignalCh chan struct{} - //writeTasksQueue []WriteTask + indexFile *os.File + dataFile *os.File + mutex sync.Mutex + pages map[uint32]*_page + pageWaits map[uint32][]chan readResult + pagesToRead []uint32 + readSignalCh chan struct{} } type Options struct { - Dir string - DatabaseName string + IndexFile *os.File + DataFile *os.File } func New(opt Options) (*Atree, error) { - // if opt.Dir == "" { - // return nil, errors.New("Dir option is required") - // } - // if opt.DatabaseName == "" { - // return nil, errors.New("DatabaseName option is required") - // } - // // открываю или создаю dbName.data и dbName.index файлы - // var ( - // fileName = filepath.Join(opt.Dir, opt.DatabaseName+".db") - // file *os.File - // allocatedPagesQty uint32 - // ) - - // // При создании data файла сразу создается индекс, поэтому корректное - // // состояние БД: либо оба файла есть, либо ни одного файла нет. - // isDataExist, err := isFileExist(fileName) - // if err != nil { - // return nil, fmt.Errorf("check data file is exist: %s", err) - // } - // if isDataExist { - // file, allocatedPagesQty, err = openFile(fileName, PageSize) - // if err != nil { - // return nil, fmt.Errorf("open data file: %s", err) - // } - // } else { - // // нет файла - // file, err = os.OpenFile(fileName, os.O_CREATE|os.O_RDWR, filePerm) - // if err != nil { - // return nil, err - // } - // } - - // tree := &Atree{ - // file: file, - // allocatedPagesQty: allocatedPagesQty, - // pages: make(map[uint32]*_page), - // pageWaits: make(map[uint32][]chan readResult), - // readSignalCh: make(chan struct{}, 1), - // writeSignalCh: make(chan struct{}, 1), - // } - - // return tree, nil - return nil, nil + if opt.IndexFile == nil { + return nil, errors.New("IndexFile option is required") + } + if opt.DataFile == nil { + return nil, errors.New("DataFile option is required") + } + s := &Atree{ + indexFile: opt.IndexFile, + dataFile: opt.DataFile, + pages: make(map[uint32]*_page), + pageWaits: make(map[uint32][]chan readResult), + readSignalCh: make(chan struct{}, 1), + } + return s, nil } func (s *Atree) Run() { @@ -236,203 +222,3 @@ func (s *Atree) GetAllPages(rootPageNo uint32) (_ []uint32, err error) { // } return } - -// const ( -// FlagReused byte = 1 // сторінка із FreeList -// FlagNewRoot byte = 2 // новая страница -// ) - -// type PageToWrite struct { -// PageNo uint32 -// Data []byte -// IsReused bool -// } - -//type AppendDataPagesReq struct { -// LastPageNo uint32 -// Legs []PathLeg -// DataPages []NotLinkedDataPage -// } - -// type Report struct { -// NewRootPageNo uint32 -// LastPageNo uint32 -// Pages []PageToWrite -// } - -// // Можливо atree не треба блокувати при читанні. Я можу створити копії index сторінок, -// // а потім під 1 мутексом замінити в pages. -// func (s *Atree) AppendDataPages(req AppendDataPagesReq) Report { -// var ( -// //pagesToRelease []uint32 -// newRootPageNo uint32 -// lastPageNo = req.LastPageNo -// pages []PageToWrite -// legs = req.Legs -// changed = make(map[uint32]PathLeg) -// ) - -// for _, p := range req.DataPages { -// newDataPageNo, isReused := s.allocPageNumber() // alloc only number -// // pagesToRelease = append(pagesToRelease, newDataPage.PageNo) - -// // set prevPageNo -// p.SetPrevPageNo(lastPageNo) - -// pages = append(pages, PageToWrite{ -// PageNo: newDataPageNo, -// Data: p.Data, -// IsReused: isReused, -// }) - -// // FIX - після додавання index page потрібно модифікувати path, -// // або будувати щось типу дерева знизу вверх - -// if len(legs) > 0 { -// newPageNo := newDataPageNo -// lastIdx := len(legs) - 1 - -// for legIdx := lastIdx; legIdx >= 0; legIdx-- { -// leg := req.Legs[legIdx] -// ok := appendPair(leg.Data, p.Since, newPageNo) -// if ok { -// // index FIX -// // потрібно запам'ятати змінені сторінки, але їх можуть змінювати -// // кілька ітерацій, тому додавати в Pages не можна. -// // на індексній сторінці достатньо місця. Запис вставлено. -// changed[leg.PageNo] = leg -// break -// } -// // на індексній сторінці НЕ достатньо місця. Створюю нову. -// newIndexPage := s.allocPage() -// //pagesToRelease = append(pagesToRelease, newIndexPage.PageNo) -// appendPair(newIndexPage.Data, p.Since, newPageNo) -// // ставлю мітку що всі pageNo на сторінці - це data pageNo -// // fix - єдина оптимізація від існування isDataPageNumbersIdx - getAllPages не завантажує data pages -// // if legIdx == lastIdx { -// // newIndexPage.Data[isDataPageNumbersIdx] = 1 -// // } -// pages = append(pages, PageToWrite{ -// PageNo: newIndexPage.PageNo, -// Data: newIndexPage.Data, -// IsReused: newIndexPage.IsReused, -// }) -// // замінюю крок в path на новий -// legs[legIdx] = PathLeg{ -// PageNo: newIndexPage.PageNo, -// Data: newIndexPage.Data, -// } -// // -// newPageNo = newIndexPage.PageNo - -// if legIdx == 0 { -// newRoot := s.allocPage() -// //pagesToRelease = append(pagesToRelease, newRoot.PageNo) -// appendPair(newRoot.Data, getSince(leg.Data), leg.PageNo) // old rootPageNo -// appendPair(newRoot.Data, p.Since, newIndexPage.PageNo) - -// // Фиксирую новый root в REDO логе -// pages = append(pages, PageToWrite{ -// PageNo: newRoot.PageNo, -// Data: newRoot.Data, -// IsReused: newRoot.IsReused, -// }) -// newRootPageNo = newRoot.PageNo -// // додаю в початок списку кроків нову root сторінку -// legs = append([]PathLeg{ -// { -// PageNo: newRoot.PageNo, -// Data: newRoot.Data, -// }, -// }, legs...) -// break -// } -// } -// } else { -// // індексну root сторінку створюю одразу для першої data сторінки, -// // root data сторінки, як в B+Tree не буває. -// newRoot := s.allocPage() -// //pagesToRelease = append(pagesToRelease, newRoot.PageNo) -// //newRoot.Data[isDataPageNumbersIdx] = 1 -// appendPair(newRoot.Data, p.Since, newDataPageNo) - -// pages = append(pages, PageToWrite{ -// PageNo: newRoot.PageNo, -// Data: newRoot.Data, -// IsReused: newRoot.IsReused, -// }) -// newRootPageNo = newRoot.PageNo -// // додаю в початок списку кроків нову root сторінку -// legs = append([]PathLeg{ -// { -// PageNo: newRoot.PageNo, -// Data: newRoot.Data, -// }, -// }, legs...) -// } - -// // На данний момен схема - наступна. Всі сторінки - data та index - зафіксовані в кеші. -// // Отже запис на диск пройде максимально швидко. Після цього ReferenceCount кожної -// // сторінки зменшиться на 1. Оскільки на метрику утримується XLock, сторінки мають -// // ReferenceCount = 1 (немає інших читачів). -// // for _, pageNo := range indexPagesToRelease { -// // s.releasePage(pageNo) -// // } -// lastPageNo = newDataPageNo -// } -// for _, leg := range changed { -// // fix recalc checksum -// pages = append(pages, PageToWrite{ -// PageNo: leg.PageNo, -// Data: leg.Data, -// }) -// } -// return Report{ -// NewRootPageNo: newRootPageNo, -// LastPageNo: lastPageNo, -// Pages: pages, -// } -// } - -// APPEND DATA PAGE - -//MetricID uint32 -//Timestamp uint32 -//Value float64 -//RootPageNo uint32 -//PrevPageNo uint32 - -// type AppendDataPageReq struct { -// Since uint32 -// TimestampsChunks [][]byte -// TimestampsSize uint16 -// ValuesChunks [][]byte -// ValuesSize uint16 -// } - -// type ChangedPage struct { -// PageNo uint32 -// Data []byte -// IsReused bool -// } - -// AppendDataPage - метод не записує дані в data-файл, а лише змінює дані -// в page cache та freeList і повертає звіт що змінено. -// Цей звіт txlog має записати в transaction log і лише потім можна змінювати data файл. -// Є ідея - записати у index файли заглушки 255,255,255,255 замість номерів сторінок і зберегти зміщення. -// А потім одним викликом отримати із FreeList список вільних сторінок. -// Тому що є проблема із відновленням FreeList після збою, якщо з нього будуть паралельно -// забирати та добавляти номери сторінок інші потоки. -// Це буде працювати, якщо додавання в txlog і маніпуляції із freeList будуть відбуватись в одному потоці -// NotLinkedDataPage - це data сторінка із payload, але без встановленого prevPageNo та без розрахованого CRC32 -// type NotLinkedDataPage struct { -// Since uint32 -// Data []byte -// } - -// func (s NotLinkedDataPage) SetPrevPageNo(prevPageNo uint32) { -// bin.PutUint32(s.Data[prevPageIdx:], prevPageNo) -// bin.PutUint32(s.Data[crc32Idx:], util.CalcChecksum(s.Data[:crc32Idx])) -// } - -// diff --git a/atree/cursor.go b/atree/cursor.go index c85b673..ff7e24b 100644 --- a/atree/cursor.go +++ b/atree/cursor.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" + "gordenko.dev/dima/bin" "gordenko.dev/dima/qb" ) @@ -118,69 +119,69 @@ func (s *BackwardCursor) Close() { // HELPER func (s *BackwardCursor) makeDecompressors() error { - // timestampsSize, _ := bin.GetUint16(s.pageData[timestampsSizeIdx:]) - // valuesSize, _ := bin.GetUint16(s.pageData[valuesSizeIdx:]) + timestampsSize, _ := bin.GetUint16(s.pageData[timestampsSizeIdx:]) + valuesSize, _ := bin.GetUint16(s.pageData[valuesSizeIdx:]) - // payloadSize := timestampsSize + valuesSize + payloadSize := timestampsSize + valuesSize - // if payloadSize > dataFooterIdx { - // return fmt.Errorf("corrupted data page %d: timestamps + values size %d gt payload size", - // s.pageNo, payloadSize) - // } + if payloadSize > dataFooterIdx { + return fmt.Errorf("corrupted data page %d: timestamps + values size %d gt payload size", + s.pageNo, payloadSize) + } - // s.timestampDecompressor = enc.NewTimeDeltaDecompressor( - // s.pageData[:timestampsSize], - // ) + s.timestampDecompressor = enc.NewTimeDeltaDecompressor( + s.pageData[:timestampsSize], + ) - // vbuf := s.pageData[timestampsSize : timestampsSize+valuesSize] + vbuf := s.pageData[timestampsSize : timestampsSize+valuesSize] - // switch s.metricType { - // case qb.Instant: - // s.valueDecompressor = enc.NewInstantDeltaDecompressor( - // vbuf, s.fracDigits) + switch s.metricType { + case qb.Instant: + s.valueDecompressor = enc.NewInstantDeltaDecompressor( + vbuf, s.fracDigits) - // case qb.Cumulative: - // s.valueDecompressor = enc.NewCumulativeDeltaDecompressor( - // vbuf, s.fracDigits) + case qb.Cumulative: + s.valueDecompressor = enc.NewCumulativeDeltaDecompressor( + vbuf, s.fracDigits) - // default: - // return fmt.Errorf("bug: wrong metricType %d", s.metricType) - // } + default: + return fmt.Errorf("bug: wrong metricType %d", s.metricType) + } return nil } -func makeDecompressors(pageData []byte, metricType qb.MetricType, fracDigits byte) ( - qb.TimestampDecompressor, qb.ValueDecompressor, error, -) { - // timestampsSize, _ := bin.GetUint16(pageData[timestampsSizeIdx:]) - // valuesSize, _ := bin.GetUint16(pageData[valuesSizeIdx:]) +// func makeDecompressors(pageData []byte, metricType qb.MetricType, fracDigits byte) ( +// qb.TimestampDecompressor, qb.ValueDecompressor, error, +// ) { +// timestampsSize, _ := bin.GetUint16(pageData[timestampsSizeIdx:]) +// valuesSize, _ := bin.GetUint16(pageData[valuesSizeIdx:]) - // payloadSize := timestampsSize + valuesSize +// payloadSize := timestampsSize + valuesSize - // if payloadSize > dataFooterIdx { - // return nil, nil, fmt.Errorf("corrupted: timestamps + values size %d > payload size", - // payloadSize) - // } +// if payloadSize > dataFooterIdx { +// return nil, nil, fmt.Errorf("corrupted: timestamps + values size %d > payload size", +// payloadSize) +// } - // timestampDecompressor := enc.NewTimeDeltaDecompressor( - // pageData[:timestampsSize], - // ) +// timestampDecompressor := enc.NewTimeDeltaDecompressor( +// pageData[:timestampsSize], +// ) - // vbuf := pageData[timestampsSize : timestampsSize+valuesSize] +// vbuf := pageData[timestampsSize : timestampsSize+valuesSize] - // var valueDecompressor qb.ValueDecompressor - // switch metricType { - // case qb.Instant: - // valueDecompressor = enc.NewInstantDeltaDecompressor( - // vbuf, fracDigits) +// var valueDecompressor qb.ValueDecompressor +// switch metricType { +// case qb.Instant: +// valueDecompressor = enc.NewInstantDeltaDecompressor( +// vbuf, fracDigits) - // case qb.Cumulative: - // valueDecompressor = enc.NewCumulativeDeltaDecompressor( - // vbuf, fracDigits) +// case qb.Cumulative: +// valueDecompressor = enc.NewCumulativeDeltaDecompressor( +// vbuf, fracDigits) - // default: - // return nil, nil, fmt.Errorf("bug: wrong metricType %d", metricType) - // } - //return timestampDecompressor, valueDecompressor, nil - return nil, nil, nil -} +// default: +// return nil, nil, fmt.Errorf("bug: wrong metricType %d", metricType) +// } +//return timestampDecompressor, valueDecompressor, nil +// return nil, nil, nil +// } diff --git a/atree/io.go b/atree/io.go index 08ca562..69fa4cc 100644 --- a/atree/io.go +++ b/atree/io.go @@ -1,24 +1,14 @@ package atree import ( - "errors" "fmt" - "io/fs" - "math" - "os" bin "gordenko.dev/dima/bin/little" "gordenko.dev/dima/qb" + "gordenko.dev/dima/qb/storage" "gordenko.dev/dima/qb/util" ) -// fix - додати ID, щоб потім перемістити із тимчасового буфера в pages, або звільнити -// type AllocatedPage struct { -// PageNo uint32 -// Data []byte -// IsReused bool -// } - type readResult struct { Data []byte Err error @@ -47,46 +37,41 @@ func (s *Atree) fetchIndexPage(pageNo uint32) ([]byte, error) { } func (s *Atree) fetchDataPage(pageNo uint32) ([]byte, error) { - // buf, err := s.fetchPage(pageNo) - // if err != nil { - // return nil, err - // } - // if buf[pageTypeIdx] != PageTypeIndex { - // return nil, fmt.Errorf("wrong pageType %d instead of %d", buf[pageTypeIdx], PageTypeIndex) - // } - // return buf, nil - return nil, nil + buf, err := s.fetchPage(pageNo) + if err != nil { + return nil, err + } + return buf, nil } func (s *Atree) fetchPage(pageNo uint32) ([]byte, error) { - // s.mutex.Lock() - // p, ok := s.pages[pageNo] - // if ok { - // p.ReferenceCount++ - // s.mutex.Unlock() - // return p.Buf, nil - // } + s.mutex.Lock() + p, ok := s.pages[pageNo] + if ok { + p.ReferenceCount++ + s.mutex.Unlock() + return p.Buf, nil + } - // resultCh := make(chan readResult, 1) - // s.pageWaits[pageNo] = append(s.pageWaits[pageNo], resultCh) - // if len(s.pageWaits[pageNo]) == 1 { - // s.pagesToRead = append(s.pagesToRead, pageNo) - // s.mutex.Unlock() + resultCh := make(chan readResult, 1) + s.pageWaits[pageNo] = append(s.pageWaits[pageNo], resultCh) + if len(s.pageWaits[pageNo]) == 1 { + s.pagesToRead = append(s.pagesToRead, pageNo) + s.mutex.Unlock() - // select { - // case s.readSignalCh <- struct{}{}: - // default: - // } - // } else { - // s.mutex.Unlock() - // } + select { + case s.readSignalCh <- struct{}{}: + default: + } + } else { + s.mutex.Unlock() + } - // result := <-resultCh - // if result.Err == nil { - // result.Err = s.verifyCRC(result.Data, PageSize) - // } - // return result.Data, result.Err - return nil, nil + result := <-resultCh + if result.Err == nil { + result.Err = s.verifyCRC(result.Data, storage.DataPageSize) + } + return result.Data, result.Err } func (s *Atree) releasePage(pageNo uint32) { @@ -108,98 +93,6 @@ func (s *Atree) releasePage(pageNo uint32) { } } -func (s *Atree) allocPageNumber() (uint32, bool) { - // fix - return 0, true -} - -// func (s *Atree) allocPage() AllocatedPage { -// var ( -// allocated = AllocatedPage{ -// Data: make([]byte, PageSize), -// } -// ) -// //allocated.PageNo = s.freelist.ReservePage() -// s.mutex.Lock() -// // if allocated.PageNo > 0 { -// // allocated.IsReused = true -// // } else { -// // if s.allocatedPagesQty == math.MaxUint32 { -// // qb.Abort(qb.MaxAtreeSizeExceeded, errors.New("no space in Atree index")) -// // } -// // s.allocatedPagesQty++ -// // allocated.PageNo = s.allocatedPagesQty -// // } - -// // s.pages[allocated.PageNo] = &_page{ -// // // fix pageType -// // PageNo: allocated.PageNo, -// // Buf: allocated.Data, -// // ReferenceCount: 1, -// // } -// s.mutex.Unlock() -// return allocated -// } - -// // fix - без freelist -// func (s *Atree) allocIndexPage() AllocatedPage { -// var ( -// allocated = AllocatedPage{ -// Data: make([]byte, PageSize), -// } -// ) - -// allocated.PageNo = s.freelist.ReservePage() -// s.mutex.Lock() -// if allocated.PageNo > 0 { -// allocated.IsReused = true -// } else { -// if s.allocatedPagesQty == math.MaxUint32 { -// qb.Abort(qb.MaxAtreeSizeExceeded, errors.New("no space in Atree index")) -// } -// s.allocatedPagesQty++ -// allocated.PageNo = s.allocatedPagesQty -// } - -// s.pages[allocated.PageNo] = &_page{ -// // fix pageType -// PageNo: allocated.PageNo, -// Buf: allocated.Data, -// ReferenceCount: 1, -// } -// s.mutex.Unlock() -// return allocated -// } - -// func (s *Atree) allocDataPage() AllocatedPage { -// var ( -// allocated = AllocatedPage{ -// Data: make([]byte, PageSize), -// } -// ) - -// allocated.PageNo = s.freelist.ReservePage() -// s.mutex.Lock() -// if allocated.PageNo > 0 { -// allocated.IsReused = true -// } else { -// if s.allocatedPagesQty == math.MaxUint32 { -// qb.Abort(qb.MaxAtreeSizeExceeded, errors.New("no space in Atree index")) -// } -// s.allocatedPagesQty++ -// allocated.PageNo = s.allocatedPagesQty -// } - -// s.pages[allocated.PageNo] = &_page{ -// // fix pageType -// PageNo: allocated.PageNo, -// Buf: allocated.Data, -// ReferenceCount: 1, -// } -// s.mutex.Unlock() -// return allocated -// } - // DATA PAGES // READ @@ -214,155 +107,52 @@ func (s *Atree) pageReader() { } func (s *Atree) readPages() { - // s.mutex.Lock() - // if len(s.pagesToRead) == 0 { - // s.mutex.Unlock() - // return - // } - // pagesToRead := s.pagesToRead - // s.pagesToRead = nil - // s.mutex.Unlock() + s.mutex.Lock() + if len(s.pagesToRead) == 0 { + s.mutex.Unlock() + return + } + pagesToRead := s.pagesToRead + s.pagesToRead = nil + s.mutex.Unlock() - // for _, pageNo := range pagesToRead { - // buf := make([]byte, PageSize) - // off := (pageNo - 1) * PageSize - // n, err := s.file.ReadAt(buf, int64(off)) - // if n != PageSize { - // err = fmt.Errorf("read %d instead of %d", n, PageSize) - // } + for _, pageNo := range pagesToRead { + buf := make([]byte, storage.DataPageSize) + off := int(pageNo-1) * storage.DataPageSize + n, err := s.file.ReadAt(buf, int64(off)) + if n != storage.DataPageSize { + err = fmt.Errorf("read %d instead of %d", n, storage.DataPageSize) + } - // s.mutex.Lock() - // resultChannels := s.pageWaits[pageNo] - // delete(s.pageWaits, pageNo) + s.mutex.Lock() + resultChannels := s.pageWaits[pageNo] + delete(s.pageWaits, pageNo) - // if err != nil { - // s.mutex.Unlock() - // for _, resultCh := range resultChannels { - // resultCh <- readResult{ - // Err: err, - // } - // } - // } else { - // s.pages[pageNo] = &_page{ - // // fix - page type - // PageNo: pageNo, - // Buf: buf, - // ReferenceCount: len(resultChannels), - // } - // s.mutex.Unlock() - // for _, resultCh := range resultChannels { - // resultCh <- readResult{ - // Data: buf, - // } - // } - // } - // } + if err != nil { + s.mutex.Unlock() + for _, resultCh := range resultChannels { + resultCh <- readResult{ + Err: err, + } + } + } else { + s.pages[pageNo] = &_page{ + PageNo: pageNo, + Buf: buf, + ReferenceCount: len(resultChannels), + } + s.mutex.Unlock() + for _, resultCh := range resultChannels { + resultCh <- readResult{ + Data: buf, + } + } + } + } } // WRITE -// func (s *Atree) pageWriter() { -// for { -// select { -// case <-s.writeSignalCh: -// err := s.writeTasks() -// if err != nil { -// qb.Abort(qb.WriteToAtreeFailed, err) -// } -// } -// } -// } - -// type WriteTask struct { -// WaitCh chan struct{} -// Pages []PageToWrite -// } - -// func (s *Atree) appendWriteTaskToQueue(task WriteTask) { -// s.mutex.Lock() -// s.writeTasksQueue = append(s.writeTasksQueue, task) -// s.mutex.Unlock() - -// select { -// case s.writeSignalCh <- struct{}{}: -// default: -// } -// } - -// func (s *Atree) writeTasks() error { -// s.mutex.Lock() -// tasks := s.writeTasksQueue -// s.writeTasksQueue = nil -// s.mutex.Unlock() - -// for _, task := range tasks { -// for _, p := range task.Pages { -// if len(p.Data) != PageSize { -// return fmt.Errorf("wrong page %d size: %d", -// p.PageNo, len(p.Data)) -// } -// bin.PutUint32(p.Data[crc32Idx:], calcChecksum(p.Data[:crc32Idx])) - -// off := (p.PageNo - 1) * PageSize -// n, err := s.file.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)) -// } -// } -// close(task.WaitCh) -// } -// return nil -// } - -// IO - -func isFileExist(fileName string) (bool, error) { - _, err := os.Stat(fileName) - if err != nil { - if errors.Is(err, fs.ErrNotExist) { - return false, nil - } else { - return false, err - } - } else { - return true, nil - } -} - -func openFile(fileName string, pageSize int) (_ *os.File, _ uint32, err error) { - file, err := os.OpenFile(fileName, os.O_RDWR, filePerm) - if err != nil { - return - } - fi, err := file.Stat() - if err != nil { - return - } - fileSize := fi.Size() - - if (fileSize % int64(pageSize)) > 0 { - err = fmt.Errorf("the file size %d is not a multiple of the page size %d", - fileSize, pageSize) - return - } - - allocatedPagesQty := fileSize / int64(pageSize) - if allocatedPagesQty > math.MaxUint32 { - err = fmt.Errorf("allocated pages %d is > max pages %d", - allocatedPagesQty, math.MaxUint32) - return - } - - return file, uint32(allocatedPagesQty), nil -} - -// func (s *Atree) ApplyREDO(task WriteTask) { -// s.appendWriteTaskToQueue(task) -// } - func (s *Atree) verifyCRC(data []byte, pageSize int) error { var ( pos = pageSize - 4 diff --git a/atree/writer.go b/atree/writer.go deleted file mode 100644 index ce98d01..0000000 --- a/atree/writer.go +++ /dev/null @@ -1,158 +0,0 @@ -package atree - -// type Writer struct { -// metricID uint32 -// timestamp uint32 -// value float64 -// tmp []byte -// hasher hash.Hash32 -// isDataPageReused bool -// dataPageNo uint32 -// isRootChanged bool -// newRootPageNo uint32 -// indexPages []uint32 -// reusedIndexPages []uint32 -// indexPagesToWrite []PageToWrite -// } - -// type WriterOptions struct { -// MetricID uint32 -// Value float64 -// Timestamp uint32 -// IsDataPageReused bool -// DataPageNo uint32 -// Page []byte -// } - -// // dataPage можно записати 1 раз. Щоб не заплутувати інтерфейс - передаю data сторінку -// // через Options. Index сторінок може бути від 1 до N, тому виділяю окремий метод -// func NewWriter(opt WriterOptions) (*Writer, error) { -// if opt.MetricID == 0 { -// return nil, errors.New("MetricID option is required") -// } -// if opt.DataPageNo == 0 { -// return nil, errors.New("DataPageNo option is required") -// } -// // if len(opt.Page) != octopus.DataPageSize { -// // return nil, fmt.Errorf("bug: wrong data page size %d", len(opt.Page)) -// // } - -// s := &Writer{ -// metricID: opt.MetricID, -// timestamp: opt.Timestamp, -// value: opt.Value, -// tmp: make([]byte, 21), -// isDataPageReused: opt.IsDataPageReused, -// dataPageNo: opt.DataPageNo, -// hasher: crc32.NewIEEE(), -// } - -// // var err error -// // err = s.init(opt.Page) -// // if err != nil { -// // return nil, err -// // } -// return s, nil -// } - -// /* -// Формат: -// 4b metricID -// 8b value -// 4b timestamp -// 1b flags (reused) -// 4b dataPageNo -// 8KB dataPage -// */ -// // func (s *Writer) init(dataPage []byte) error { -// // bin.PutUint32(s.tmp[0:], s.metricID) -// // bin.PutUint32(s.tmp[4:], s.timestamp) -// // bin.PutFloat64(s.tmp[8:], s.value) -// // if s.isDataPageReused { -// // s.tmp[16] = 1 -// // } -// // bin.PutUint32(s.tmp[17:], s.dataPageNo) - -// // _, err := s.file.Write(s.tmp) -// // if err != nil { -// // return err -// // } - -// // _, err = s.file.Write(dataPage) -// // if err != nil { -// // return err -// // } - -// // s.hasher.Write(s.tmp) -// // s.hasher.Write(dataPage) -// // return nil -// // } - -// /* -// Формат -// 1b index page flags -// 4b indexPageNo -// Nb indexPage -// */ -// // func (s *Writer) AppendIndexPage(indexPageNo uint32, indexPage []byte, flags byte) error { -// // s.tmp[0] = flags -// // bin.PutUint32(s.tmp[1:], indexPageNo) -// // // _, err := s.file.Write(s.tmp[:5]) -// // // if err != nil { -// // // return err -// // // } -// // // _, err = s.file.Write(indexPage) -// // // if err != nil { -// // // return err -// // // } - -// // s.hasher.Write(s.tmp[:5]) -// // s.hasher.Write(indexPage) - -// // s.indexPages = append(s.indexPages, indexPageNo) - -// // if (flags & FlagReused) == FlagReused { -// // s.reusedIndexPages = append(s.reusedIndexPages, indexPageNo) -// // } - -// // if (flags & FlagNewRoot) == FlagNewRoot { -// // s.newRootPageNo = indexPageNo -// // s.isRootChanged = true -// // } - -// // s.indexPagesToWrite = append(s.indexPagesToWrite, -// // PageToWrite{ -// // PageNo: indexPageNo, -// // Data: indexPage, -// // }) -// // return nil -// // } - -// // func (s *Writer) IndexPagesToWrite() []PageToWrite { -// // return s.indexPagesToWrite -// // } - -// // func (s *Writer) Close() (err error) { -// // // финализирую запись -// // bin.PutUint32(s.tmp, s.hasher.Sum32()) -// // _, err = s.file.Write(s.tmp[:4]) -// // if err != nil { -// // return err -// // } -// // err = s.file.Sync() -// // if err != nil { -// // return -// // } -// // return s.file.Close() -// // } - -// // func (s *Writer) GetReport() Report { -// // return Report{ -// // IsDataPageReused: s.isDataPageReused, -// // DataPageNo: s.dataPageNo, -// // //IndexPages: s.indexPages, -// // IsRootChanged: s.isRootChanged, -// // NewRootPageNo: s.newRootPageNo, -// // ReusedIndexPages: s.reusedIndexPages, -// // } -// // } diff --git a/atree/writers.go b/atree/writers.go deleted file mode 100644 index 2fd52e0..0000000 --- a/atree/writers.go +++ /dev/null @@ -1,15 +0,0 @@ -package atree - -type PeriodsWriter interface { - Feed(uint32, float64) - FeedNoSend(uint32, float64) - Close() error -} - -type WorkerMeasureConsumer interface { - FeedNoSend(uint32, float64) -} - -type AtreeMeasureConsumer interface { - Feed(uint32, float64) -} diff --git a/client/client.go b/client/client.go index e631d9a..beb8108 100644 --- a/client/client.go +++ b/client/client.go @@ -389,10 +389,8 @@ func (s *Connection) readCumulativeMeasures() (_ []proto.CumulativeMeasure, err if err != nil { return nil, fmt.Errorf("read response code: %s", err) } - fmt.Println("code", code) switch code { case proto.RespPartOfValue: - fmt.Println("RespPartOfValue") var count int count, err = bin.ReadUint32AsInt(s.src) if err != nil { @@ -415,10 +413,8 @@ func (s *Connection) readCumulativeMeasures() (_ []proto.CumulativeMeasure, err result = append(result, measure) } case proto.RespEndOfValue: - fmt.Println("RespEndOfValue") return result, nil case proto.RespError: - fmt.Println("RespError") return nil, s.onError() default: return nil, fmt.Errorf("unknown reponse code %d", code) diff --git a/database/api.go b/database/api.go index 8a309fc..3a58038 100644 --- a/database/api.go +++ b/database/api.go @@ -278,39 +278,39 @@ func (s *Database) GetMetric(conn io.Writer, req proto.GetMetricReq) error { } func (s *Database) DeleteMetric(req proto.DeleteMetricReq) uint16 { - resultCh := make(chan worker.DeleteMetricResult, 1) + // resultCh := make(chan worker.DeleteMetricResult, 1) - s.workerInbox.Push(worker.DeleteMetricReq{ - MetricID: req.MetricID, - ResultCh: resultCh, - }) + // s.workerInbox.Push(worker.DeleteMetricReq{ + // MetricID: req.MetricID, + // ResultCh: resultCh, + // }) - result := <-resultCh + // result := <-resultCh - switch result.ResultCode { - case worker.Succeed: - // var ( - // //freePageNumbers []uint32 - // ) - // if result.RootPageNo > 0 { - // var err error - // freePageNumbers, err = s.atree.GetAllPages(result.RootPageNo) - // if err != nil { - // qb.Abort(qb.FailedAtreeRequest, err) - // } - // } - // s.storage.Append(storage.DeletedMetric{ - // MetricID: req.MetricID, - // //FreePageNumbers: freePageNumbers, FIX - // }) - //<-waitCh + // switch result.ResultCode { + // case worker.Succeed: + // // var ( + // // //freePageNumbers []uint32 + // // ) + // // if result.RootPageNo > 0 { + // // var err error + // // freePageNumbers, err = s.atree.GetAllPages(result.RootPageNo) + // // if err != nil { + // // qb.Abort(qb.FailedAtreeRequest, err) + // // } + // // } + // // s.storage.Append(storage.DeletedMetric{ + // // MetricID: req.MetricID, + // // //FreePageNumbers: freePageNumbers, FIX + // // }) + // //<-waitCh - case worker.NoMetric: - return proto.ErrNoMetric + // case worker.NoMetric: + // return proto.ErrNoMetric - default: - qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug) - } + // default: + // qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug) + // } return 0 } @@ -363,7 +363,7 @@ func (s *Database) AppendMeasures(conn io.Writer, req proto.AppendMeasuresReq) e } func (s *Database) DeleteMeasures(req proto.DeleteMeasuresReq) uint16 { - resultCh := make(chan worker.DeleteMeasuresResult, 1) + resultCh := make(chan byte, 1) s.workerInbox.Push(worker.DeleteMeasuresReq{ MetricID: req.MetricID, @@ -373,36 +373,38 @@ func (s *Database) DeleteMeasures(req proto.DeleteMeasuresReq) uint16 { result := <-resultCh - switch result.ResultCode { - case worker.NoMeasuresToDelete: - // ok + _ = result - //case worker.DeleteFromAtreeNotNeeded: - // регистрирую удаление в TransactionLog - // s.storage.Append(storage.MeasuresDeleteRecord{ - // MetricID: req.MetricID, - // }) - //<-waitCh + // switch result.ResultCode { + // case worker.NoMeasuresToDelete: + // // ok - //case worker.DeleteFromAtreeRequired: - // собираю номера всех data и index страниц метрики (типа запись REDO лога). - // pageNumbers, err := s.atree.GetAllPages(req.MetricID) - // if err != nil { - // qb.Abort(qb.FailedAtreeRequest, err) - // } + // //case worker.DeleteFromAtreeNotNeeded: // // регистрирую удаление в TransactionLog - // s.storage.Append(storage.DeletedMeasures{ - // MetricID: req.MetricID, - // FreePageNumbers: pageNumbers, - // }) - //<-waitCh + // // s.storage.Append(storage.MeasuresDeleteRecord{ + // // MetricID: req.MetricID, + // // }) + // //<-waitCh - case worker.NoMetric: - return proto.ErrNoMetric + // //case worker.DeleteFromAtreeRequired: + // // собираю номера всех data и index страниц метрики (типа запись REDO лога). + // // pageNumbers, err := s.atree.GetAllPages(req.MetricID) + // // if err != nil { + // // qb.Abort(qb.FailedAtreeRequest, err) + // // } + // // // регистрирую удаление в TransactionLog + // // s.storage.Append(storage.DeletedMeasures{ + // // MetricID: req.MetricID, + // // FreePageNumbers: pageNumbers, + // // }) + // //<-waitCh - default: - qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug) - } + // case worker.NoMetric: + // return proto.ErrNoMetric + + // default: + // qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug) + // } return 0 } @@ -616,17 +618,17 @@ func (s *Database) fullScan(req fullScanReq) error { req.ResponseWriter.Close() case worker.UntilFound: - // err := s.atree.ContinueFullScan(atree.ContinueFullScanReq{ - // FracDigits: result.FracDigits, - // ResponseWriter: req.ResponseWriter, - // LastPageNo: result.LastPageNo, - // }) - // s.worker.AddJobToQueue(req.MetricID) - // if err != nil { - // reply(req.Conn, proto.ErrUnexpected) - // } else { - // req.ResponseWriter.Close() - // } + err := s.atree.ContinueFullScan(atree.ContinueFullScanReq{ + FracDigits: result.FracDigits, + ResponseWriter: req.ResponseWriter, + LastPageNo: result.LastPageNo, + }) + //s.worker.AddJobToQueue(req.MetricID) // FIX release rlock + if err != nil { + reply(req.Conn, proto.ErrUnexpected) + } else { + req.ResponseWriter.Close() + } case worker.NoMetric: reply(req.Conn, proto.ErrNoMetric) diff --git a/database/database.go b/database/database.go index c8c6e51..11d466c 100644 --- a/database/database.go +++ b/database/database.go @@ -10,19 +10,13 @@ import ( "time" "gordenko.dev/dima/qb" + "gordenko.dev/dima/qb/atree" "gordenko.dev/dima/qb/inbox" "gordenko.dev/dima/qb/recovery" "gordenko.dev/dima/qb/storage" "gordenko.dev/dima/qb/worker" ) -// type metricLockEntry struct { -// XLock bool -// RLocks int -// WaitQueue []any -// } -//metricLockEntries map[uint32]*metricLockEntry - type Database struct { mutex sync.Mutex dir string @@ -30,6 +24,7 @@ type Database struct { workerInbox *inbox.Inbox worker *worker.Worker storage *storage.Writer + atree *atree.Atree tcpPort int logfile *os.File logger *log.Logger @@ -122,8 +117,6 @@ func New(opt Options) (_ *Database, err error) { return nil, fmt.Errorf("storage.NewWriter: %s", err) } - fmt.Println("storage created") - s.worker = worker.New(worker.Options{ Inbox: s.workerInbox, StorageInbox: storageInbox, @@ -133,7 +126,14 @@ func New(opt Options) (_ *Database, err error) { ExitCh: opt.ExitCh, WaitGroup: opt.WaitGroup, }) - fmt.Println("worker created") + + s.atree, err = atree.New(atree.Options{ + DataFile: dataFile, + IndexFile: indexFile, + }) + if err != nil { + return nil, fmt.Errorf("atree.New: %s", err) + } return s, nil } @@ -146,13 +146,6 @@ func (s *Database) ListenAndServe() (err error) { //s.waitGroup.Add(1) go s.storage.Run() - // s.atree, err = atree.New(atree.Options{ - // Dir: s.dir, - // DatabaseName: s.databaseName, - // }) - // if err != nil { - // return fmt.Errorf("atree.New: %s", err) - // } // s.atree.Run() s.waitGroup.Add(1) diff --git a/enc/time_delta.go b/enc/time_delta.go index ba442cf..9d83a70 100644 --- a/enc/time_delta.go +++ b/enc/time_delta.go @@ -304,6 +304,12 @@ func (s *TimeDeltaCompressor) CreateDecompressor() qb.TimestampDecompressor { return d } +func (s *TimeDeltaCompressor) FirstTimestamp() uint32 { + pos := len(s.buf) - 4 + timestamp, _ := bin.GetUint32(s.buf[pos:]) + return timestamp +} + func (s *TimeDeltaCompressor) LastTimestamp() uint32 { return s.lastUnixtime } diff --git a/examples/play/main.go b/examples/play/main.go index 488e00e..8681516 100644 --- a/examples/play/main.go +++ b/examples/play/main.go @@ -9,7 +9,6 @@ import ( "time" "gopkg.in/ini.v1" - "gordenko.dev/dima/qb" "gordenko.dev/dima/qb/client" "gordenko.dev/dima/qb/proto" ) @@ -292,46 +291,46 @@ func sendRequests(conn *client.Connection) { // ADD CUMULATIVE METRIC - err = conn.AddMetric(proto.AddMetricReq{ - MetricID: cumulativeMetricID, - MetricType: qb.Cumulative, - FracDigits: fracDigits, - }) - if err != nil { - log.Fatalf("conn.AddMetric: %s\n", err) - } else { - fmt.Printf("\nCumulative metric %d added\n", cumulativeMetricID) - } + // err = conn.AddMetric(proto.AddMetricReq{ + // MetricID: cumulativeMetricID, + // MetricType: qb.Cumulative, + // FracDigits: fracDigits, + // }) + // if err != nil { + // log.Fatalf("conn.AddMetric: %s\n", err) + // } else { + // fmt.Printf("\nCumulative metric %d added\n", cumulativeMetricID) + // } - // GET CUMULATIVE METRIC + // // GET CUMULATIVE METRIC - cMetric, err := conn.GetMetric(cumulativeMetricID) - if err != nil { - log.Fatalf("conn.GetMetric: %s\n", err) - } else { - fmt.Printf(` - GetMetric: - metricID: %d - metricType: %s - fracDigits: %d - `, - cMetric.MetricID, metricTypeToName[cMetric.MetricType], cMetric.FracDigits) - } + // cMetric, err := conn.GetMetric(cumulativeMetricID) + // if err != nil { + // log.Fatalf("conn.GetMetric: %s\n", err) + // } else { + // fmt.Printf(` + // GetMetric: + // metricID: %d + // metricType: %s + // fracDigits: %d + // `, + // cMetric.MetricID, metricTypeToName[cMetric.MetricType], cMetric.FracDigits) + // } - // APPEND MEASURES + // // APPEND MEASURES - cumulativeMeasures := GenerateCumulativeMeasures(62) + // cumulativeMeasures := GenerateCumulativeMeasures(62) - result, err := conn.AppendMeasures(proto.AppendMeasuresReq{ - MetricID: cumulativeMetricID, - Measures: cumulativeMeasures[:1000], - }) - if err != nil { - log.Fatalf("conn.AppendMeasures: %s\n", err) - } else { - fmt.Printf("\nAppended %d measures for the metric %d: count=%d, errorCode=%d\n", - len(cumulativeMeasures), cumulativeMetricID, result.WrittenCount, result.ErrorCode) - } + // result, err := conn.AppendMeasures(proto.AppendMeasuresReq{ + // MetricID: cumulativeMetricID, + // Measures: cumulativeMeasures[:1000], + // }) + // if err != nil { + // log.Fatalf("conn.AppendMeasures: %s\n", err) + // } else { + // fmt.Printf("\nAppended %d measures for the metric %d: count=%d, errorCode=%d\n", + // len(cumulativeMeasures), cumulativeMetricID, result.WrittenCount, result.ErrorCode) + // } // currentValues, err := conn.ListCurrentValues([]uint32{ // cumulativeMetricID, @@ -374,8 +373,8 @@ func sendRequests(conn *client.Connection) { log.Fatalf("conn.ListAllCumulativeMeasures: %s\n", err) } else { fmt.Printf("\nListAllCumulativeMeasures (last 15 items):\n") - fmt.Printf("%#v\n", cumulativeList) - for _, item := range cumulativeList { + //fmt.Printf("%#v\n", cumulativeList) + for _, item := range cumulativeList[:15] { fmt.Printf(" %s => %.2f\n", formatTime(item.Timestamp), item.Value) } } diff --git a/examples/play/play b/examples/play/play index a118ab2..a7dd256 100755 Binary files a/examples/play/play and b/examples/play/play differ diff --git a/qb.go b/qb.go index 0f76350..efc6c36 100644 --- a/qb.go +++ b/qb.go @@ -41,6 +41,7 @@ type TimestampCompressor interface { // Offset() int ReplaceBuffer([]byte) WriteCommitedTo(io.Writer) error + FirstTimestamp() uint32 LastTimestamp() uint32 ReplaceSinceWithUntil() uint32 } diff --git a/storage/misc.go b/storage/misc.go new file mode 100644 index 0000000..c07b28f --- /dev/null +++ b/storage/misc.go @@ -0,0 +1,92 @@ +package storage + +import ( + "fmt" + + bin "gordenko.dev/dima/bin/little" + "gordenko.dev/dima/qb" + "gordenko.dev/dima/qb/enc" +) + +// func (s *BackwardCursor) makeDecompressors() error { +// timestampsSize, _ := bin.GetUint16(s.pageData[timestampsSizeIdx:]) +// valuesSize, _ := bin.GetUint16(s.pageData[valuesSizeIdx:]) + +// payloadSize := timestampsSize + valuesSize + +// if payloadSize > dataFooterIdx { +// return fmt.Errorf("corrupted data page %d: timestamps + values size %d gt payload size", +// s.pageNo, payloadSize) +// } + +// s.timestampDecompressor = enc.NewTimeDeltaDecompressor( +// s.pageData[:timestampsSize], +// ) + +// vbuf := s.pageData[timestampsSize : timestampsSize+valuesSize] + +// switch s.metricType { +// case qb.Instant: +// s.valueDecompressor = enc.NewInstantDeltaDecompressor( +// vbuf, s.fracDigits) + +// case qb.Cumulative: +// s.valueDecompressor = enc.NewCumulativeDeltaDecompressor( +// vbuf, s.fracDigits) + +// default: +// return fmt.Errorf("bug: wrong metricType %d", s.metricType) +// } +// return nil +// } + +func makeDecompressors(pageData []byte, metricType qb.MetricType, fracDigits byte) ( + qb.TimestampDecompressor, qb.ValueDecompressor, error, +) { + + valuesSize, _ := bin.GetUint16(pageData[valuesSizeIdx:]) + + payloadSize := timestampsSize + valuesSize + + if payloadSize > dataFooterIdx { + return nil, nil, fmt.Errorf("corrupted: timestamps + values size %d > payload size", + payloadSize) + } + + timestampDecompressor := enc.NewTimeDeltaDecompressor( + pageData[:timestampsSize], + ) + + vbuf := pageData[timestampsSize : timestampsSize+valuesSize] + + var valueDecompressor qb.ValueDecompressor + switch metricType { + case qb.Instant: + valueDecompressor = enc.NewInstantDeltaDecompressor( + vbuf, fracDigits) + + case qb.Cumulative: + valueDecompressor = enc.NewCumulativeDeltaDecompressor( + vbuf, fracDigits) + + default: + return nil, nil, fmt.Errorf("bug: wrong metricType %d", metricType) + } + return timestampDecompressor, valueDecompressor, nil + return nil, nil, nil +} + +func CreateTimeDeltaDecompressor(page []byte) qb.TimestampDecompressor { + size, _ := bin.GetUint16(page[timestampsSizeIdx:]) + pos := DataPagePayloadSize - int(size) + d := enc.NewTimeDeltaDecompressor() + d.RestoreFromEnd(page[pos:DataPagePayloadSize]) + return d +} + +func CreateValueDeltaDecompressor(page []byte, metricType qb.MetricType, fracDigits byte) qb.ValueDecompressor { + size, _ := bin.GetUint16(page[valuesSizeIdx:]) + d := enc.NewValueDeltaDecompressor(metricType, fracDigits) + d.RestoreFromEnd(page[:size]) + return d +} diff --git a/storage/writer.go b/storage/writer.go index 3f3a90d..5fb8dfa 100644 --- a/storage/writer.go +++ b/storage/writer.go @@ -77,23 +77,20 @@ type Writer struct { isExited bool exitCh chan struct{} waitGroup *sync.WaitGroup - //signalCh chan struct{} } type WriterOptions struct { - Inbox *inbox.Inbox - WorkerInbox *inbox.Inbox - Dir string - DatabaseName string - //SnapshotNumber int // номер журнала + Inbox *inbox.Inbox + WorkerInbox *inbox.Inbox + Dir string + DatabaseName string WAL string DataFreeList *freelist.FreeList IndexFreeList *freelist.FreeList DataFile *os.File IndexFile *os.File - //Atree *atree.Atree - ExitCh chan struct{} - WaitGroup *sync.WaitGroup + ExitCh chan struct{} + WaitGroup *sync.WaitGroup } func NewWriter(opt WriterOptions) (*Writer, error) { @@ -245,17 +242,26 @@ func (s *Writer) packAndWrite() (err error) { fmt.Println("synced to wal") // 4. Пишу в atree сторінки - err = WriteDataPages(s.dataFile, prepared.WriteToData) - if err != nil { - return + if len(prepared.WriteToData) > 0 { + err = WriteDataPages(s.dataFile, prepared.WriteToData) + if err != nil { + return + } + if err = s.dataFile.Sync(); err != nil { + return + } + fmt.Println("written to data") } - fmt.Println("written to data") - err = WriteIndexPages(s.indexFile, prepared.WriteToIndex) - if err != nil { - return + if len(prepared.WriteToIndex) > 0 { + err = WriteIndexPages(s.indexFile, prepared.WriteToIndex) + if err != nil { + return + } + if err = s.indexFile.Sync(); err != nil { + return + } + fmt.Println("written to index") } - fmt.Println("written to index") - // 6. відправляю input - worker-у var ( @@ -413,62 +419,3 @@ func WriteIndexPages(file *os.File, pages []PageToWrite) (err error) { // TimestampsBuf []byte // ValuesBuf []byte // } - -// Якщо (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, -// }) - -// 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 - -// type Metric struct { -// MetricID uint32 -// MetricType qb.MetricType -// FracDigits byte -// LastPageNo uint32 -// Since uint32 -// SinceValue float64 -// Until uint32 -// UntilValue float64 -// Timestamps []byte // payload FIX -// Values []byte // payload FIX add lastDelta + h ? -// } - -// func writeChunks(dst io.Writer, chunks [][]byte, size int) (err error) { -// remaining := size -// for _, buf := range chunks { -// if remaining < len(buf) { -// buf = buf[:remaining] -// } -// _, err = dst.Write(buf) -// if err != nil { -// return -// } -// remaining -= len(buf) -// if remaining == 0 { -// break -// } -// } -// return -// } diff --git a/worker/metric.go b/worker/metric.go index 51c71de..5888c77 100644 --- a/worker/metric.go +++ b/worker/metric.go @@ -33,11 +33,75 @@ type Metric struct { values qb.ValueCompressor xLock bool rLocks int - WaitQueue []any + waitQueue []any indexLevelTails []storage.IndexLevelTail // root - last element capturedState *CapturedState } +// індекси +// Metric encode format: +// metricID - 4b +// metricType - 1b +// fracDigits - 1b +// lastPageNo - 4b +// timestamps size - 2b +// values size - 2b +// timestams payload - Nb +// values payload - Nb +// index levels count - varsize +// [ +// records qty - varsize +// records - Nb +// ] + +func (s *Metric) WriteTo(w io.Writer) (err error) { + _, err = w.Write([]byte{ + byte(s.metricType), + s.fracDigits, + }) + if err != nil { + return + } + err = bin.WriteUint32(w, s.lastPageNo) + if err != nil { + return + } + err = bin.WriteUint16(w, uint16(s.timestamps.CommitedSize())) + if err != nil { + return + } + err = bin.WriteUint16(w, uint16(s.values.CommitedSize())) + if err != nil { + return + } + // timestamps payload + err = s.timestamps.WriteCommitedTo(w) + if err != nil { + return + } + // values payload + err = s.values.WriteCommitedTo(w) + if err != nil { + return + } + // indexes + _, err = bin.WriteVarSize(w, len(s.indexLevelTails)) + if err != nil { + return + } + for _, level := range s.indexLevelTails { + _, err = bin.WriteVarSize(w, level.RecordsCount) + if err != nil { + return + } + _, err = w.Write(level.Buffer[:level.RecordsCount*storage.IndexRecordSize]) + if err != nil { + return + } + } + return +} + func (s *Metric) MetricType() qb.MetricType { return s.metricType } @@ -60,26 +124,12 @@ func (s *Metric) LastTimestamp() uint32 { return s.timestamps.LastTimestamp() } -func (s *Metric) OnMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) { - //s.timestamps.Reset() - //s.values.Reset() - s.xLock = false - // s.Timestamps.Renew() - // s.Values.Renew() - - // s.LastPageNo = 0 - // s.Since = 0 - // s.SinceValue = 0 - // s.Until = 0 - s.indexLevelTails = nil - s.lastPageNo = 0 - s.lastValue = 0 -} +// REQUESTS func (s *Metric) AppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox *inbox.Inbox) { - - if s.capturedState != nil { - s.WaitQueue = append(s.WaitQueue, req) + if s.xLock || s.capturedState != nil { + s.waitQueue = append(s.waitQueue, req) + return } var ( timestamps = s.timestamps @@ -188,7 +238,6 @@ func (s *Metric) AppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox // скопіювати. Причому можна скопіювати зрізи chunks if len(pages) > 0 { - fmt.Println("push pages") // пишу в storage довгим шляхом через redo файл і запис в data файл storageInbox.Push(storage.MeasuresAppendWithGrow{ MetricID: req.MetricID, @@ -206,7 +255,6 @@ func (s *Metric) AppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox ResultCh: req.ResultCh, }) } else { - fmt.Println("push simple") // короткий шлях - запис лише в storage storageInbox.Push(storage.MeasuresAppend{ MetricID: req.MetricID, @@ -221,6 +269,143 @@ func (s *Metric) AppendMeasures(req AppendMeasuresReq, tmp []byte, storageInbox } } +func (s *Metric) DeleteMeasures(req DeleteMeasuresReq) { + if s.xLock || s.capturedState != nil { + s.waitQueue = append(s.waitQueue, req) + return + } + since := s.timestamps.FirstTimestamp() + until := s.timestamps.LastTimestamp() + if since == 0 || (req.Since > 0 && until < req.Since) { + req.ResultCh <- NoMeasuresToDelete + } + // if s.RootPageNo > 0 { + // req.ResultCh <- tryDeleteMeasuresResult{ + // ResultCode: DeleteFromAtreeRequired, + // RootPageNo: metric.RootPageNo, + // } + // } else { + // req.ResultCh <- tryDeleteMeasuresResult{ + // ResultCode: DeleteFromAtreeNotNeeded, + // } + // } +} + +func (s *Metric) StartRangeScan(req RangeScanReq) { + if s.xLock { + s.waitQueue = append(s.waitQueue, req) + return + } + // if s.timestamps.CommitedSize() == 0 { + // req.ResultCh <- RangeScanResult{ + // ResultCode: QueryDone, + // } + // return + // } + + // if req.Since > s.timestamps.LastTimestamp() { + // req.ResultCh <- RangeScanResult{ + // ResultCode: QueryDone, + // } + // return + // } + + // if req.Until < s.Since { + // if s.RootPageNo > 0 { + // req.ResultCh <- RangeScanResult{ + // ResultCode: UntilNotFound, + // RootPageNo: s.RootPageNo, + // FracDigits: s.fracDigits, + // } + // s.rLocks++ + // return + // } else { + // req.ResultCh <- RangeScanResult{ + // ResultCode: QueryDone, + // } + // return + // } + // } + + timestampDecompressor := s.timestamps.CreateDecompressor() + valueDecompressor := s.values.CreateDecompressor(s.metricType, s.fracDigits) + + for { + timestamp, done := timestampDecompressor.NextValue() + if done { + break + } + value, done := valueDecompressor.NextValue() + if done { + qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug) + } + if timestamp <= req.Until { + req.ResponseWriter.FeedNoSend(timestamp, value) + if timestamp < req.Since { + req.ResultCh <- RangeScanResult{ + ResultCode: QueryDone, + } + return + } + } + } + if s.lastPageNo > 0 { + req.ResultCh <- RangeScanResult{ + ResultCode: UntilFound, + LastPageNo: s.lastPageNo, + FracDigits: s.fracDigits, + } + s.rLocks++ + } else { + req.ResultCh <- RangeScanResult{ + ResultCode: QueryDone, + } + } +} + +func (s *Metric) StartFullScan(req FullScanReq) { + if s.xLock { + s.waitQueue = append(s.waitQueue, req) + return + } + if s.timestamps.CommitedSize() == 0 { + req.ResultCh <- FullScanResult{ + ResultCode: QueryDone, + } + return + } + timestampDecompressor := s.timestamps.CreateDecompressor() + valueDecompressor := s.values.CreateDecompressor(s.metricType, s.fracDigits) + for { + timestamp, done := timestampDecompressor.NextValue() + 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) + } + + if s.lastPageNo > 0 { + req.ResultCh <- FullScanResult{ + ResultCode: UntilFound, + LastPageNo: s.lastPageNo, + FracDigits: s.fracDigits, + } + s.rLocks++ + } else { + req.ResultCh <- FullScanResult{ + ResultCode: QueryDone, + } + } +} + +// COMMITS + func (s *Metric) OnMeasuresAppendCommited(rec storage.MeasuresAppendCommited) { // Видаляю state. Оригінальні Timestamps і Values вже мають останню версію s.capturedState = nil @@ -252,175 +437,18 @@ func (s *Metric) OnMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWith } } -// READ +func (s *Metric) OnMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) { + //s.timestamps.Reset() + //s.values.Reset() + s.xLock = false + // s.Timestamps.Renew() + // s.Values.Renew() -func (s *Metric) StartRangeScan(req RangeScanReq) { - // if s.Since == 0 { - // req.ResultCh <- rangeScanResult{ - // ResultCode: QueryDone, - // } - // return - // } - - // if req.Since > s.Until { - // req.ResultCh <- rangeScanResult{ - // ResultCode: QueryDone, - // } - // return - // } - - // if req.Until < s.Since { - // if s.RootPageNo > 0 { - // req.ResultCh <- rangeScanResult{ - // ResultCode: UntilNotFound, - // RootPageNo: s.RootPageNo, - // FracDigits: s.FracDigits, - // } - // s.RLocks++ - // return - // } else { - // req.ResultCh <- rangeScanResult{ - // ResultCode: QueryDone, - // } - // return - // } - // } - - // timestampDecompressor := s.timestamps.CreateDecompressor() - // valueDecompressor := s.values.CreateDecompressor(s.metricType, s.fracDigits) - - // for { - // timestamp, done := timestampDecompressor.NextValue() - // if done { - // break - // } - // value, done := valueDecompressor.NextValue() - // if done { - // qb.Abort(qb.HasTimestampNoValueBug, ErrNoValueBug) - // } - // if timestamp <= req.Until { - // req.ResponseWriter.FeedNoSend(timestamp, value) - // if timestamp < req.Since { - // req.ResultCh <- rangeScanResult{ - // ResultCode: QueryDone, - // } - // return - // } - // } - // } - // if s.lastPageNo > 0 { - // req.ResultCh <- rangeScanResult{ - // ResultCode: UntilFound, - // LastPageNo: s.lastPageNo, - // FracDigits: s.fracDigits, - // } - // s.RLocks++ - // } else { - // req.ResultCh <- rangeScanResult{ - // ResultCode: QueryDone, - // } - // } -} - -func (s *Metric) StartFullScan(req FullScanReq) { - // if s.Since == 0 { - // req.ResultCh <- fullScanResult{ - // ResultCode: QueryDone, - // } - // return - // } - - timestampDecompressor := s.timestamps.CreateDecompressor() - valueDecompressor := s.values.CreateDecompressor(s.metricType, s.fracDigits) - - for { - timestamp, done := timestampDecompressor.NextValue() - 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) - } - - // if s.lastPageNo > 0 { - // req.ResultCh <- FullScanResult{ - // ResultCode: UntilFound, - // LastPageNo: s.lastPageNo, - // FracDigits: s.fracDigits, - // } - // s.rLocks++ - // } else { - req.ResultCh <- FullScanResult{ - ResultCode: QueryDone, - } - //} -} - -// індекси -// Metric encode format: -// metricID - 4b -// metricType - 1b -// fracDigits - 1b -// lastPageNo - 4b -// timestamps size - 2b -// values size - 2b -// timestams payload - Nb -// values payload - Nb -// index levels count - varsize -// [ -// records qty - varsize -// records - Nb -// ] - -func (s *Metric) WriteTo(w io.Writer) (err error) { - _, err = w.Write([]byte{ - byte(s.metricType), - s.fracDigits, - }) - if err != nil { - return - } - err = bin.WriteUint32(w, s.lastPageNo) - if err != nil { - return - } - err = bin.WriteUint16(w, uint16(s.timestamps.CommitedSize())) - if err != nil { - return - } - err = bin.WriteUint16(w, uint16(s.values.CommitedSize())) - if err != nil { - return - } - // timestamps payload - err = s.timestamps.WriteCommitedTo(w) - if err != nil { - return - } - // values payload - err = s.values.WriteCommitedTo(w) - if err != nil { - return - } - // indexes - _, err = bin.WriteVarSize(w, len(s.indexLevelTails)) - if err != nil { - return - } - for _, level := range s.indexLevelTails { - _, err = bin.WriteVarSize(w, level.RecordsCount) - if err != nil { - return - } - _, err = w.Write(level.Buffer[:level.RecordsCount*storage.IndexRecordSize]) - if err != nil { - return - } - } - return + // s.LastPageNo = 0 + // s.Since = 0 + // s.SinceValue = 0 + // s.Until = 0 + s.indexLevelTails = nil + s.lastPageNo = 0 + s.lastValue = 0 } diff --git a/worker/worker.go b/worker/worker.go index 5114354..abb863d 100644 --- a/worker/worker.go +++ b/worker/worker.go @@ -147,31 +147,22 @@ func (s *Worker) doWork() { switch req := untyped.(type) { case AppendMeasuresReq: s.AppendMeasures(req) - case storage.Changes: s.applyCommits(req) // all metrics only - case ListCurrentValuesReq: s.ListCurrentValues(req) // all metrics only - case RangeScanReq: s.RangeScan(req) - case FullScanReq: s.FullScan(req) - case AddMetricReq: s.AddMetric(req) - case DeleteMetricReq: s.DeleteMetric(req) - case DeleteMeasuresReq: s.DeleteMeasures(req) - case GetMetricReq: s.GetMetric(req) - default: qb.Abort(qb.UnknownWorkerQueueItemBug, fmt.Errorf("bug: unknown worker queue item type %T", req)) @@ -181,30 +172,23 @@ func (s *Worker) doWork() { // суть у тому що треба запускати запити, пока не зустріну XLock func (s *Worker) processMetricQueue(metricID uint32, metric *Metric, tmp []byte) { - if len(metric.WaitQueue) == 0 { + if len(metric.waitQueue) == 0 { return } - - for _, untyped := range metric.WaitQueue { + for _, untyped := range metric.waitQueue { switch req := untyped.(type) { case RangeScanReq: metric.StartRangeScan(req) - case FullScanReq: metric.StartFullScan(req) - case GetMetricReq: s.GetMetric(req) - case AppendMeasuresReq: metric.AppendMeasures(req, tmp, s.storageInbox) - case DeleteMetricReq: - s.startDeleteMetric(metric, req) - + s.DeleteMetric(req) case DeleteMeasuresReq: - s.startDeleteMeasures(metric, req) - + metric.DeleteMeasures(req) default: qb.Abort(qb.UnknownMetricWaitQueueItemBug, fmt.Errorf("bug: unknown metric wait queue item type %T", req)) @@ -222,7 +206,6 @@ type AddMetricReq struct { func (s *Worker) AddMetric(req AddMetricReq) { _, ok := s.metrics[req.MetricID] if ok { - fmt.Println("add metric duplicate") req.ResultCh <- MetricDuplicate return } @@ -246,27 +229,6 @@ func (s *Worker) AddMetric(req AddMetricReq) { }) } -func (s *Worker) processTryAddMetricReqsImmediatelyAfterDelete(reqs []AddMetricReq) { - if len(reqs) == 0 { - return - } - var ( - req = reqs[0] - waitQueue []any - ) - if len(reqs) > 1 { - for _, req := range reqs[1:] { - waitQueue = append(waitQueue, req) - } - } - // FIX - // s.metricLockEntries[req.MetricID] = &metricLockEntry{ - // XLock: true, - // WaitQueue: waitQueue, - // } - req.ResultCh <- Succeed -} - type GetMetricResult struct { MetricType qb.MetricType FracDigits byte @@ -293,116 +255,43 @@ func (s *Worker) GetMetric(req GetMetricReq) { } } -type DeleteMetricResult struct { - ResultCode byte - RootPageNo uint32 -} - type DeleteMetricReq struct { MetricID uint32 - ResultCh chan DeleteMetricResult + ResultCh chan byte } func (s *Worker) DeleteMetric(req DeleteMetricReq) { - // FIX - // metric, ok := s.metrics[req.MetricID] - // if !ok { - // req.ResultCh <- tryDeleteMetricResult{ - // ResultCode: NoMetric, - // } - // return - // } - - // lockEntry, ok := s.metricLockEntries[req.MetricID] - // if ok { - // lockEntry.WaitQueue = append(lockEntry.WaitQueue, req) - // } else { - // s.startDeleteMetric(metric, req) - // } -} - -func (s *Worker) startDeleteMetric(metric *Metric, req DeleteMetricReq) { - // FIX - // s.metricLockEntries[req.MetricID] = &metricLockEntry{ - // XLock: true, - // } - // req.ResultCh <- tryDeleteMetricResult{ - // ResultCode: Succeed, - // RootPageNo: metric.RootPageNo, - // } -} - -type DeleteMeasuresResult struct { - ResultCode byte - RootPageNo uint32 + metric, ok := s.metrics[req.MetricID] + if !ok { + req.ResultCh <- NoMetric + return + } + if metric.xLock { + metric.waitQueue = append(metric.waitQueue, req) + } else { + // collect all pages, than + s.storageInbox.Push(storage.MetricDelete{ + MetricID: req.MetricID, + FreeIndexPages: nil, + FreeDataPages: nil, + ResultCh: req.ResultCh, + }) + } } type DeleteMeasuresReq struct { MetricID uint32 Since uint32 - ResultCh chan DeleteMeasuresResult + ResultCh chan byte } func (s *Worker) DeleteMeasures(req DeleteMeasuresReq) { - // FIX - // metric, ok := s.metrics[req.MetricID] - // if !ok { - // req.ResultCh <- tryDeleteMeasuresResult{ - // ResultCode: NoMetric, - // } - // return - // } - - // if metric.Since == 0 || (req.Since > 0 && metric.Until < req.Since) { - // req.ResultCh <- tryDeleteMeasuresResult{ - // ResultCode: NoMeasuresToDelete, - // } - // } - - // lockEntry, ok := s.metricLockEntries[req.MetricID] - // if ok { - // lockEntry.WaitQueue = append(lockEntry.WaitQueue, req) - // } else { - // s.startDeleteMeasures(metric, req) - // } -} - -func (s *Worker) startDeleteMeasures(metric *Metric, req DeleteMeasuresReq) { - // FIX - // s.metricLockEntries[req.MetricID] = &metricLockEntry{ - // XLock: true, - // } - - // if metric.RootPageNo > 0 { - // req.ResultCh <- tryDeleteMeasuresResult{ - // ResultCode: DeleteFromAtreeRequired, - // RootPageNo: metric.RootPageNo, - // } - // } else { - // req.ResultCh <- tryDeleteMeasuresResult{ - // ResultCode: DeleteFromAtreeNotNeeded, - // } - // } -} - -func (s *Worker) onMeasuresAppendCommited(rec storage.MeasuresAppendCommited) { - metric, ok := s.metrics[rec.MetricID] + metric, ok := s.metrics[req.MetricID] if !ok { - qb.Abort(qb.NoMetricBug, - fmt.Errorf("onMeasuresAppendCommited: metric %d not found", - rec.MetricID)) + req.ResultCh <- NoMetric + return } - metric.OnMeasuresAppendCommited(rec) -} - -func (s *Worker) onMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWithGrowCommited) { - metric, ok := s.metrics[rec.MetricID] - if !ok { - qb.Abort(qb.NoMetricBug, - fmt.Errorf("finAppendMeasures: metric %d not found", - rec.MetricID)) - } - metric.OnMeasuresAppendWithGrowCommited(rec) + metric.DeleteMeasures(req) } type AppendMeasuresReq struct { @@ -420,10 +309,10 @@ func (s *Worker) AppendMeasures(req AppendMeasuresReq) { return } if metric.xLock { - metric.WaitQueue = append(metric.WaitQueue, req) - return + metric.waitQueue = append(metric.waitQueue, req) + } else { + metric.AppendMeasures(req, s.tmp, s.storageInbox) } - metric.AppendMeasures(req, s.tmp, s.storageInbox) } type RangeScanResult struct { @@ -456,13 +345,11 @@ func (s *Worker) RangeScan(req RangeScanReq) { } return } - if metric.xLock { - metric.WaitQueue = append(metric.WaitQueue, req) - return + metric.waitQueue = append(metric.waitQueue, req) + } else { + metric.StartRangeScan(req) } - - metric.StartRangeScan(req) } type FullScanResult struct { @@ -492,12 +379,11 @@ func (s *Worker) FullScan(req FullScanReq) { } return } - if metric.xLock { - metric.WaitQueue = append(metric.WaitQueue, req) - return + metric.waitQueue = append(metric.waitQueue, req) + } else { + metric.StartFullScan(req) } - metric.StartFullScan(req) } type ListCurrentValuesReq struct { @@ -566,65 +452,43 @@ func (s *Worker) onMetricAddCommited(rec storage.MetricAddCommited) { metric, ok := s.metrics[rec.MetricID] if !ok { qb.Abort(qb.MetricAddedBug, - fmt.Errorf("metric %d not found after commit", rec.MetricID)) + fmt.Errorf("onMetricAddCommited: metric %d not found", rec.MetricID)) } metric.xLock = false - rec.ResultCh <- Succeed // new + rec.ResultCh <- Succeed } func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) { metric, ok := s.metrics[rec.MetricID] if !ok { qb.Abort(qb.NoMetricBug, - fmt.Errorf("deleteMetric: metric %d not found", - rec.MetricID)) + fmt.Errorf("onMetricDeleteCommited: metric %d not found", rec.MetricID)) } - if !metric.xLock { - qb.Abort(qb.NoXLockBug, - fmt.Errorf("deleteMetric: xlock not set for the metric %d", - rec.MetricID)) - } - - var addMetricReqs []AddMetricReq - - if len(metric.WaitQueue) > 0 { - for _, untyped := range metric.WaitQueue { + if len(metric.waitQueue) > 0 { + for _, untyped := range metric.waitQueue { switch req := untyped.(type) { - // case tryAppendMeasureReq: - // req.ResultCh <- tryAppendMeasureResult{ - // MetricID: req.MetricID, - // ResultCode: NoMetric, - // } - + case AppendMeasuresReq: + req.ResultCh <- storage.MeasuresAppendResult{ + ResultCode: NoMetric, + WrittenCount: 0, + } case RangeScanReq: req.ResultCh <- RangeScanResult{ ResultCode: NoMetric, } - case FullScanReq: req.ResultCh <- FullScanResult{ ResultCode: NoMetric, } - - case AddMetricReq: - addMetricReqs = append(addMetricReqs, req) - case DeleteMetricReq: - req.ResultCh <- DeleteMetricResult{ - ResultCode: NoMetric, - } - + req.ResultCh <- NoMetric case DeleteMeasuresReq: - req.ResultCh <- DeleteMeasuresResult{ - ResultCode: NoMetric, - } - + req.ResultCh <- NoMetric case GetMetricReq: req.ResultCh <- GetMetricResult{ ResultCode: NoMetric, } - default: qb.Abort(qb.UnknownMetricWaitQueueItemBug, fmt.Errorf("bug: unknown metric wait queue item type %T", req)) @@ -632,36 +496,39 @@ func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) { } } delete(s.metrics, rec.MetricID) + // + rec.ResultCh <- Succeed +} - // ADD in storage - // if len(rec.FreePageNumbers) > 0 { - // s.freeList.AddPages(rec.FreePageNumbers) - // } - - if len(addMetricReqs) > 0 { - s.processTryAddMetricReqsImmediatelyAfterDelete(addMetricReqs) +func (s *Worker) onMeasuresAppendCommited(rec storage.MeasuresAppendCommited) { + metric, ok := s.metrics[rec.MetricID] + if !ok { + qb.Abort(qb.NoMetricBug, + fmt.Errorf("onMeasuresAppendCommited: metric %d not found", + rec.MetricID)) } + metric.OnMeasuresAppendCommited(rec) +} + +func (s *Worker) onMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWithGrowCommited) { + metric, ok := s.metrics[rec.MetricID] + if !ok { + qb.Abort(qb.NoMetricBug, + fmt.Errorf("onMeasuresAppendWithGrowCommited: metric %d not found", + rec.MetricID)) + } + metric.OnMeasuresAppendWithGrowCommited(rec) } func (s *Worker) onMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) { metric, ok := s.metrics[rec.MetricID] if !ok { qb.Abort(qb.NoMetricBug, - fmt.Errorf("deleteMeasures: metric %d not found", - rec.MetricID)) - } - - if !metric.xLock { - qb.Abort(qb.NoXLockBug, - fmt.Errorf("deleteMeasures: xlock not set for the metric %d", - rec.MetricID)) + fmt.Errorf("onMeasuresDeleteCommited: metric %d not found", rec.MetricID)) } metric.OnMeasuresDeleteCommited(rec) metric.xLock = false - // FIX add in storage - // if len(rec.FreePageNumbers) > 0 { - // s.freeList.AddPages(rec.FreePageNumbers) - // } + //s.doAfterReleaseXLock(rec.MetricID, metric) }