From c26b857900009ac81c78c1bc03f85e0c8e47818a Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 26 Oct 2016 21:50:07 -0700 Subject: [PATCH] Fix Search and add test --- group.go | 181 +++++++++++++++++++++++++++++++++---------------- group_test.go | 183 ++++++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 284 insertions(+), 80 deletions(-) diff --git a/group.go b/group.go index c0d199e1d..84aa8a228 100644 --- a/group.go +++ b/group.go @@ -24,6 +24,7 @@ with respect to the Group's Head (the AutoFile being appended to) */ const groupCheckDuration = 1000 * time.Millisecond +const defaultHeadSizeLimit = 10 * 1024 * 1024 // 10MB type Group struct { ID string @@ -33,17 +34,25 @@ type Group struct { mtx sync.Mutex headSizeLimit int64 totalSizeLimit int64 + minIndex int // Includes head + maxIndex int // Includes head, where Head will move to } func OpenGroup(head *AutoFile) (g *Group, err error) { dir := path.Dir(head.Path) g = &Group{ - ID: "group:" + head.ID, - Head: head, - Dir: dir, - ticker: time.NewTicker(groupCheckDuration), + ID: "group:" + head.ID, + Head: head, + Dir: dir, + ticker: time.NewTicker(groupCheckDuration), + headSizeLimit: defaultHeadSizeLimit, + minIndex: 0, + maxIndex: 0, } + gInfo := g.readGroupInfo() + g.minIndex = gInfo.MinIndex + g.maxIndex = gInfo.MaxIndex go g.processTicks() return } @@ -72,6 +81,12 @@ func (g *Group) TotalSizeLimit() int64 { return g.totalSizeLimit } +func (g *Group) MaxIndex() int { + g.mtx.Lock() + defer g.mtx.Unlock() + return g.maxIndex +} + func (g *Group) Close() error { g.ticker.Stop() return nil @@ -83,8 +98,8 @@ func (g *Group) processTicks() { if !ok { return // Done. } - // TODO Check head size limit - // TODO check total size limit + g.checkHeadSizeLimit() + g.checkTotalSizeLimit() } } @@ -112,8 +127,7 @@ func (g *Group) RotateFile() { g.mtx.Lock() defer g.mtx.Unlock() - gInfo := g.readGroupInfo() - dstPath := filePathForIndex(g.Head.Path, gInfo.MaxIndex+1) + dstPath := filePathForIndex(g.Head.Path, g.maxIndex) err := os.Rename(g.Head.Path, dstPath) if err != nil { panic(err) @@ -122,6 +136,7 @@ func (g *Group) RotateFile() { if err != nil { panic(err) } + g.maxIndex += 1 } func (g *Group) NewReader(index int) *GroupReader { @@ -135,23 +150,29 @@ type SearchFunc func(line string) (int, error) // Searches for the right file in Group, // then returns a GroupReader to start streaming lines +// Returns true if an exact match was found, otherwise returns +// the next greater line that starts with prefix. // CONTRACT: caller is responsible for closing GroupReader. -func (g *Group) Search(prefix string, cmp SearchFunc) (*GroupReader, error) { - gInfo := g.ReadGroupInfo() - minIndex, maxIndex := gInfo.MinIndex, gInfo.MaxIndex - curIndex := (minIndex + maxIndex + 1) / 2 +func (g *Group) Search(prefix string, cmp SearchFunc) (*GroupReader, bool, error) { + g.mtx.Lock() + minIndex, maxIndex := g.minIndex, g.maxIndex + g.mtx.Unlock() + // Now minIndex/maxIndex may change meanwhile, + // but it shouldn't be a big deal + // (maybe we'll want to limit scanUntil though) for { + curIndex := (minIndex + maxIndex + 1) / 2 // Base case, when there's only 1 choice left. if minIndex == maxIndex { r := g.NewReader(maxIndex) - err := scanUntil(r, prefix, cmp) + match, err := scanUntil(r, prefix, cmp) if err != nil { r.Close() - return nil, err + return nil, false, err } else { - return r, err + return r, match, err } } @@ -161,13 +182,13 @@ func (g *Group) Search(prefix string, cmp SearchFunc) (*GroupReader, error) { foundIndex, line, err := scanFirst(r, prefix) r.Close() if err != nil { - return nil, err + return nil, false, err } // Compare this line to our search query. val, err := cmp(line) if err != nil { - return nil, err + return nil, false, err } if val < 0 { // Line will come later @@ -175,12 +196,15 @@ func (g *Group) Search(prefix string, cmp SearchFunc) (*GroupReader, error) { } else if val == 0 { // Stroke of luck, found the line r := g.NewReader(foundIndex) - err := scanUntil(r, prefix, cmp) + match, err := scanUntil(r, prefix, cmp) + if !match { + panic("Expected match to be true") + } if err != nil { r.Close() - return nil, err + return nil, false, err } else { - return r, err + return r, true, err } } else { // We passed it @@ -205,24 +229,28 @@ func scanFirst(r *GroupReader, prefix string) (int, string, error) { } } -func scanUntil(r *GroupReader, prefix string, cmp SearchFunc) error { +// Returns true iff an exact match was found. +func scanUntil(r *GroupReader, prefix string, cmp SearchFunc) (bool, error) { for { line, err := r.ReadLine() if err != nil { - return err + return false, err } if !strings.HasPrefix(line, prefix) { continue } val, err := cmp(line) if err != nil { - return err + return false, err } if val < 0 { continue + } else if val == 0 { + r.PushLine(line) + return true, nil } else { r.PushLine(line) - return nil + return false, nil } } } @@ -241,6 +269,7 @@ func (g *Group) ReadGroupInfo() GroupInfo { return g.readGroupInfo() } +// Index includes the head. // CONTRACT: caller should have called g.mtx.Lock func (g *Group) readGroupInfo() GroupInfo { groupDir := filepath.Dir(g.Head.Path) @@ -285,6 +314,15 @@ func (g *Group) readGroupInfo() GroupInfo { } } + // Now account for the head. + if minIndex == -1 { + // If there were no numbered files, + // then the head is index 0. + minIndex, maxIndex = 0, 0 + } else { + // Otherwise, the head file is 1 greater + maxIndex += 1 + } return GroupInfo{minIndex, maxIndex, totalSize, headSize} } @@ -306,27 +344,43 @@ type GroupReader struct { func newGroupReader(g *Group) *GroupReader { return &GroupReader{ Group: g, - curIndex: -1, + curIndex: 0, curFile: nil, curReader: nil, curLine: nil, } } -func (g *GroupReader) ReadLine() (string, error) { - g.mtx.Lock() - defer g.mtx.Unlock() +func (gr *GroupReader) Close() error { + gr.mtx.Lock() + defer gr.mtx.Unlock() + + if gr.curReader != nil { + err := gr.curFile.Close() + gr.curIndex = 0 + gr.curReader = nil + gr.curFile = nil + gr.curLine = nil + return err + } else { + return nil + } +} + +func (gr *GroupReader) ReadLine() (string, error) { + gr.mtx.Lock() + defer gr.mtx.Unlock() // From PushLine - if g.curLine != nil { - line := string(g.curLine) - g.curLine = nil + if gr.curLine != nil { + line := string(gr.curLine) + gr.curLine = nil return line, nil } // Open file if not open yet - if g.curReader == nil { - err := g.openFile(0) + if gr.curReader == nil { + err := gr.openFile(gr.curIndex) if err != nil { return "", err } @@ -334,63 +388,74 @@ func (g *GroupReader) ReadLine() (string, error) { // Iterate over files until line is found for { - bytes, err := g.curReader.ReadBytes('\n') + bytes, err := gr.curReader.ReadBytes('\n') if err != nil { if err != io.EOF { return string(bytes), err } else { // Open the next file - err := g.openFile(g.curIndex + 1) + err := gr.openFile(gr.curIndex + 1) if err != nil { return "", err } + continue } } + return string(bytes), nil } } -// CONTRACT: caller should hold g.mtx -func (g *GroupReader) openFile(index int) error { +// IF index > gr.Group.maxIndex, returns io.EOF +// CONTRACT: caller should hold gr.mtx +func (gr *GroupReader) openFile(index int) error { // Lock on Group to ensure that head doesn't move in the meanwhile. - g.Group.mtx.Lock() - defer g.Group.mtx.Unlock() + gr.Group.mtx.Lock() + defer gr.Group.mtx.Unlock() + + var curFilePath string + if index == gr.Group.maxIndex { + curFilePath = gr.Head.Path + } else if index > gr.Group.maxIndex { + return io.EOF + } else { + curFilePath = filePathForIndex(gr.Head.Path, index) + } - curFilePath := filePathForIndex(g.Head.Path, index) curFile, err := os.Open(curFilePath) if err != nil { return err } curReader := bufio.NewReader(curFile) - // Update g.cur* - g.curIndex = index - g.curFile = curFile - g.curReader = curReader - g.curLine = nil + // Update gr.cur* + gr.curIndex = index + gr.curFile = curFile + gr.curReader = curReader + gr.curLine = nil return nil } -func (g *GroupReader) PushLine(line string) { - g.mtx.Lock() - defer g.mtx.Unlock() +func (gr *GroupReader) PushLine(line string) { + gr.mtx.Lock() + defer gr.mtx.Unlock() - if g.curLine == nil { - g.curLine = []byte(line) + if gr.curLine == nil { + gr.curLine = []byte(line) } else { panic("PushLine failed, already have line") } } // Cursor's file index. -func (g *GroupReader) CurIndex() int { - g.mtx.Lock() - defer g.mtx.Unlock() - return g.curIndex +func (gr *GroupReader) CurIndex() int { + gr.mtx.Lock() + defer gr.mtx.Unlock() + return gr.curIndex } -func (g *GroupReader) SetIndex(index int) { - g.mtx.Lock() - defer g.mtx.Unlock() - g.openFile(index) +func (gr *GroupReader) SetIndex(index int) { + gr.mtx.Lock() + defer gr.mtx.Unlock() + gr.openFile(index) } diff --git a/group_test.go b/group_test.go index 8c1b7b6a8..ced88c8bb 100644 --- a/group_test.go +++ b/group_test.go @@ -1,12 +1,25 @@ package autofile import ( + "errors" + "io" + "os" + "strconv" + "strings" "testing" . "github.com/tendermint/go-common" ) -func createTestGroup(t *testing.T, headPath string) *Group { +// NOTE: Returned group has ticker stopped +func createTestGroup(t *testing.T, headSizeLimit int64) *Group { + testID := RandStr(12) + testDir := "_test_" + testID + err := EnsureDir(testDir, 0700) + if err != nil { + t.Fatal("Error creating dir", err) + } + headPath := testDir + "/myfile" autofile, err := OpenAutoFile(headPath) if err != nil { t.Fatal("Error opening AutoFile", headPath, err) @@ -15,9 +28,18 @@ func createTestGroup(t *testing.T, headPath string) *Group { if err != nil { t.Fatal("Error opening Group", err) } + g.SetHeadSizeLimit(headSizeLimit) + g.stopTicker() return g } +func destroyTestGroup(t *testing.T, g *Group) { + err := os.RemoveAll(g.Dir) + if err != nil { + t.Fatal("Error removing test Group directory", err) + } +} + func assertGroupInfo(t *testing.T, gInfo GroupInfo, minIndex, maxIndex int, totalSize, headSize int64) { if gInfo.MinIndex != minIndex { t.Errorf("GroupInfo MinIndex expected %v, got %v", minIndex, gInfo.MinIndex) @@ -33,23 +55,14 @@ func assertGroupInfo(t *testing.T, gInfo GroupInfo, minIndex, maxIndex int, tota } } -func TestCreateGroup(t *testing.T) { - testID := RandStr(12) - testDir := "_test_" + testID - err := EnsureDir(testDir, 0700) - if err != nil { - t.Fatal("Error creating dir", err) - } - - g := createTestGroup(t, testDir+"/myfile") +func TestCheckHeadSizeLimit(t *testing.T) { + g := createTestGroup(t, 1000*1000) if g == nil { t.Error("Failed to create Group") } - g.SetHeadSizeLimit(1000 * 1000) - g.stopTicker() // At first, there are no files. - assertGroupInfo(t, g.ReadGroupInfo(), -1, -1, 0, 0) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 0, 0) // Write 1000 bytes 999 times. for i := 0; i < 999; i++ { @@ -58,21 +71,21 @@ func TestCreateGroup(t *testing.T) { t.Fatal("Error appending to head", err) } } - assertGroupInfo(t, g.ReadGroupInfo(), -1, -1, 999000, 999000) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 999000, 999000) // Even calling checkHeadSizeLimit manually won't rotate it. g.checkHeadSizeLimit() - assertGroupInfo(t, g.ReadGroupInfo(), -1, -1, 999000, 999000) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 999000, 999000) // Write 1000 more bytes. - _, err = g.Head.Write([]byte(RandStr(999) + "\n")) + _, err := g.Head.Write([]byte(RandStr(999) + "\n")) if err != nil { t.Fatal("Error appending to head", err) } // Calling checkHeadSizeLimit this time rolls it. g.checkHeadSizeLimit() - assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 1000000, 0) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 1000000, 0) // Write 1000 more bytes. _, err = g.Head.Write([]byte(RandStr(999) + "\n")) @@ -82,7 +95,7 @@ func TestCreateGroup(t *testing.T) { // Calling checkHeadSizeLimit does nothing. g.checkHeadSizeLimit() - assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 1001000, 1000) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 1001000, 1000) // Write 1000 bytes 999 times. for i := 0; i < 999; i++ { @@ -91,20 +104,146 @@ func TestCreateGroup(t *testing.T) { t.Fatal("Error appending to head", err) } } - assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 2000000, 1000000) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 2000000, 1000000) // Calling checkHeadSizeLimit rolls it again. g.checkHeadSizeLimit() - assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 2000000, 0) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2000000, 0) // Write 1000 more bytes. _, err = g.Head.Write([]byte(RandStr(999) + "\n")) if err != nil { t.Fatal("Error appending to head", err) } - assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 2001000, 1000) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2001000, 1000) // Calling checkHeadSizeLimit does nothing. g.checkHeadSizeLimit() - assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 2001000, 1000) + assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2001000, 1000) + + // Cleanup + destroyTestGroup(t, g) +} + +func TestSearch(t *testing.T) { + g := createTestGroup(t, 10*1000) + if g == nil { + t.Error("Failed to create Group") + } + + // Create some files in the group that have several INFO lines in them. + // Try to put the INFO lines in various spots. + for i := 0; i < 100; i++ { + // The random junk at the end ensures that this INFO linen + // is equally likely to show up at the end. + _, err := g.Head.Write([]byte(Fmt("INFO %v %v\n", i, RandStr(123)))) + if err != nil { + t.Error("Failed to write to head") + } + g.checkHeadSizeLimit() + for j := 0; j < 10; j++ { + _, err := g.Head.Write([]byte(RandStr(123) + "\n")) + if err != nil { + t.Error("Failed to write to head") + } + g.checkHeadSizeLimit() + } + } + + // Create a search func that searches for line + makeSearchFunc := func(target int) SearchFunc { + return func(line string) (int, error) { + parts := strings.Split(line, " ") + if len(parts) != 3 { + return -1, errors.New("Line did not have 3 parts") + } + i, err := strconv.Atoi(parts[1]) + if err != nil { + return -1, errors.New("Failed to parse INFO: " + err.Error()) + } + if target < i { + return 1, nil + } else if target == i { + return 0, nil + } else { + return -1, nil + } + } + } + + // Now search for each number + for i := 0; i < 100; i++ { + t.Log("Testing for i", i) + gr, match, err := g.Search("INFO", makeSearchFunc(i)) + if err != nil { + t.Fatal("Failed to search for line:", err) + } + if !match { + t.Error("Expected Search to return exact match") + } + line, err := gr.ReadLine() + if err != nil { + t.Fatal("Failed to read line after search", err) + } + if !strings.HasPrefix(line, Fmt("INFO %v ", i)) { + t.Fatal("Failed to get correct line") + } + // Make sure we can continue to read from there. + cur := i + 1 + for { + line, err := gr.ReadLine() + if err == io.EOF { + if cur == 99+1 { + // OK! + break + } else { + t.Fatal("Got EOF after the wrong INFO #") + } + } else if err != nil { + t.Fatal("Error reading line", err) + } + if !strings.HasPrefix(line, "INFO ") { + continue + } + if !strings.HasPrefix(line, Fmt("INFO %v ", cur)) { + t.Fatalf("Unexpected INFO #. Expected %v got:\n%v", cur, line) + } + cur += 1 + } + gr.Close() + } + + // Now search for something that is too small. + // We should get the first available line. + { + gr, match, err := g.Search("INFO", makeSearchFunc(-999)) + if err != nil { + t.Fatal("Failed to search for line:", err) + } + if match { + t.Error("Expected Search to not return exact match") + } + line, err := gr.ReadLine() + if err != nil { + t.Fatal("Failed to read line after search", err) + } + if !strings.HasPrefix(line, "INFO 0 ") { + t.Error("Failed to fetch correct line, which is the earliest INFO") + } + } + + // Now search for something that is too large. + // We should get an EOF error. + { + gr, _, err := g.Search("INFO", makeSearchFunc(999)) + if err != io.EOF { + t.Error("Expected to get an EOF error") + } + if gr != nil { + t.Error("Expected to get nil GroupReader") + } + } + + // Cleanup + destroyTestGroup(t, g) }