You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

784 lines
18 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
fix non deterministic test failures and race in privval socket (#3258) * node: decrease retry conn timeout in test Should fix #3256 The retry timeout was set to the default, which is the same as the accept timeout, so it's no wonder this would fail. Here we decrease the retry timeout so we can try many times before the accept timeout. * p2p: increase handshake timeout in test This fails sometimes, presumably because the handshake timeout is so low (only 50ms). So increase it to 1s. Should fix #3187 * privval: fix race with ping. closes #3237 Pings happen in a go-routine and can happen concurrently with other messages. Since we use a request/response protocol, we expect to send a request and get back the corresponding response. But with pings happening concurrently, this assumption could be violated. We were using a mutex, but only a RWMutex, where the RLock was being held for sending messages - this was to allow the underlying connection to be replaced if it fails. Turns out we actually need to use a full lock (not just a read lock) to prevent multiple requests from happening concurrently. * node: fix test name. DelayedStop -> DelayedStart * autofile: Wait() method In the TestWALTruncate in consensus/wal_test.go we remove the WAL directory at the end of the test. However the wal.Stop() does not properly wait for the autofile group to finish shutting down. Hence it was possible that the group's go-routine is still running when the cleanup happens, which causes a panic since the directory disappeared. Here we add a Wait() method to properly wait until the go-routine exits so we can safely clean up. This fixes #2852.
6 years ago
8 years ago
8 years ago
fix non deterministic test failures and race in privval socket (#3258) * node: decrease retry conn timeout in test Should fix #3256 The retry timeout was set to the default, which is the same as the accept timeout, so it's no wonder this would fail. Here we decrease the retry timeout so we can try many times before the accept timeout. * p2p: increase handshake timeout in test This fails sometimes, presumably because the handshake timeout is so low (only 50ms). So increase it to 1s. Should fix #3187 * privval: fix race with ping. closes #3237 Pings happen in a go-routine and can happen concurrently with other messages. Since we use a request/response protocol, we expect to send a request and get back the corresponding response. But with pings happening concurrently, this assumption could be violated. We were using a mutex, but only a RWMutex, where the RLock was being held for sending messages - this was to allow the underlying connection to be replaced if it fails. Turns out we actually need to use a full lock (not just a read lock) to prevent multiple requests from happening concurrently. * node: fix test name. DelayedStop -> DelayedStart * autofile: Wait() method In the TestWALTruncate in consensus/wal_test.go we remove the WAL directory at the end of the test. However the wal.Stop() does not properly wait for the autofile group to finish shutting down. Hence it was possible that the group's go-routine is still running when the cleanup happens, which causes a panic since the directory disappeared. Here we add a Wait() method to properly wait until the go-routine exits so we can safely clean up. This fixes #2852.
6 years ago
8 years ago
8 years ago
fix non deterministic test failures and race in privval socket (#3258) * node: decrease retry conn timeout in test Should fix #3256 The retry timeout was set to the default, which is the same as the accept timeout, so it's no wonder this would fail. Here we decrease the retry timeout so we can try many times before the accept timeout. * p2p: increase handshake timeout in test This fails sometimes, presumably because the handshake timeout is so low (only 50ms). So increase it to 1s. Should fix #3187 * privval: fix race with ping. closes #3237 Pings happen in a go-routine and can happen concurrently with other messages. Since we use a request/response protocol, we expect to send a request and get back the corresponding response. But with pings happening concurrently, this assumption could be violated. We were using a mutex, but only a RWMutex, where the RLock was being held for sending messages - this was to allow the underlying connection to be replaced if it fails. Turns out we actually need to use a full lock (not just a read lock) to prevent multiple requests from happening concurrently. * node: fix test name. DelayedStop -> DelayedStart * autofile: Wait() method In the TestWALTruncate in consensus/wal_test.go we remove the WAL directory at the end of the test. However the wal.Stop() does not properly wait for the autofile group to finish shutting down. Hence it was possible that the group's go-routine is still running when the cleanup happens, which causes a panic since the directory disappeared. Here we add a Wait() method to properly wait until the go-routine exits so we can safely clean up. This fixes #2852.
6 years ago
8 years ago
8 years ago
8 years ago
cs: sync WAL more frequently (#3300) As per #3043, this adds a ticker to sync the WAL every 2s while the WAL is running. * Flush WAL every 2s This adds a ticker that flushes the WAL every 2s while the WAL is running. This is related to #3043. * Fix spelling * Increase timeout to 2mins for slower build environments * Make WAL sync interval configurable * Add TODO to replace testChan with more comprehensive testBus * Remove extraneous debug statement * Remove testChan in favour of using system time As per https://github.com/tendermint/tendermint/pull/3300#discussion_r255886586, this removes the `testChan` WAL member and replaces the approach with a system time-oriented one. In this new approach, we keep track of the system time at which each flush and periodic flush successfully occurred. The naming of the various functions is also updated here to be more consistent with "flushing" as opposed to "sync'ing". * Update naming convention and ensure lock for timestamp update * Add Flush method as part of WAL interface Adds a `Flush` method as part of the WAL interface to enforce the idea that we can manually trigger a WAL flush from outside of the WAL. This is employed in the consensus state management to flush the WAL prior to signing votes/proposals, as per https://github.com/tendermint/tendermint/issues/3043#issuecomment-453853630 * Update CHANGELOG_PENDING * Remove mutex approach and replace with DI The dependency injection approach to dealing with testing concerns could allow similar effects to some kind of "testing bus"-based approach. This commit introduces an example of this, where instead of relying on (potentially fragile) timing of things between the code and the test, we inject code into the function under test that can signal the test through a channel. This allows us to avoid the `time.Sleep()`-based approach previously employed. * Update comment on WAL flushing during vote signing Co-Authored-By: thanethomson <connect@thanethomson.com> * Simplify flush interval definition Co-Authored-By: thanethomson <connect@thanethomson.com> * Expand commentary on WAL disk flushing Co-Authored-By: thanethomson <connect@thanethomson.com> * Add broken test to illustrate WAL sync test problem Removes test-related state (dependency injection code) from the WAL data structure and adds test code to illustrate the problem with using `WALGenerateNBlocks` and `wal.SearchForEndHeight` to test periodic sync'ing. * Fix test error messages * Use WAL group buffer size to check for flush A function is added to `libs/autofile/group.go#Group` in order to return the size of the buffered data (i.e. data that has not yet been flushed to disk). The test now checks that, prior to a `time.Sleep`, the group buffer has data in it. After the `time.Sleep` (during which time the periodic flush should have been called), the buffer should be empty. * Remove config root dir removal from #3291 * Add godoc for NewWAL mentioning periodic sync
6 years ago
8 years ago
fix non deterministic test failures and race in privval socket (#3258) * node: decrease retry conn timeout in test Should fix #3256 The retry timeout was set to the default, which is the same as the accept timeout, so it's no wonder this would fail. Here we decrease the retry timeout so we can try many times before the accept timeout. * p2p: increase handshake timeout in test This fails sometimes, presumably because the handshake timeout is so low (only 50ms). So increase it to 1s. Should fix #3187 * privval: fix race with ping. closes #3237 Pings happen in a go-routine and can happen concurrently with other messages. Since we use a request/response protocol, we expect to send a request and get back the corresponding response. But with pings happening concurrently, this assumption could be violated. We were using a mutex, but only a RWMutex, where the RLock was being held for sending messages - this was to allow the underlying connection to be replaced if it fails. Turns out we actually need to use a full lock (not just a read lock) to prevent multiple requests from happening concurrently. * node: fix test name. DelayedStop -> DelayedStart * autofile: Wait() method In the TestWALTruncate in consensus/wal_test.go we remove the WAL directory at the end of the test. However the wal.Stop() does not properly wait for the autofile group to finish shutting down. Hence it was possible that the group's go-routine is still running when the cleanup happens, which causes a panic since the directory disappeared. Here we add a Wait() method to properly wait until the go-routine exits so we can safely clean up. This fixes #2852.
6 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package autofile
  2. import (
  3. "bufio"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "os"
  8. "path"
  9. "path/filepath"
  10. "regexp"
  11. "strconv"
  12. "strings"
  13. "sync"
  14. "time"
  15. cmn "github.com/tendermint/tendermint/libs/common"
  16. )
  17. const (
  18. defaultGroupCheckDuration = 5000 * time.Millisecond
  19. defaultHeadSizeLimit = 10 * 1024 * 1024 // 10MB
  20. defaultTotalSizeLimit = 1 * 1024 * 1024 * 1024 // 1GB
  21. maxFilesToRemove = 4 // needs to be greater than 1
  22. )
  23. /*
  24. You can open a Group to keep restrictions on an AutoFile, like
  25. the maximum size of each chunk, and/or the total amount of bytes
  26. stored in the group.
  27. The first file to be written in the Group.Dir is the head file.
  28. Dir/
  29. - <HeadPath>
  30. Once the Head file reaches the size limit, it will be rotated.
  31. Dir/
  32. - <HeadPath>.000 // First rolled file
  33. - <HeadPath> // New head path, starts empty.
  34. // The implicit index is 001.
  35. As more files are written, the index numbers grow...
  36. Dir/
  37. - <HeadPath>.000 // First rolled file
  38. - <HeadPath>.001 // Second rolled file
  39. - ...
  40. - <HeadPath> // New head path
  41. The Group can also be used to binary-search for some line,
  42. assuming that marker lines are written occasionally.
  43. */
  44. type Group struct {
  45. cmn.BaseService
  46. ID string
  47. Head *AutoFile // The head AutoFile to write to
  48. headBuf *bufio.Writer
  49. Dir string // Directory that contains .Head
  50. ticker *time.Ticker
  51. mtx sync.Mutex
  52. headSizeLimit int64
  53. totalSizeLimit int64
  54. groupCheckDuration time.Duration
  55. minIndex int // Includes head
  56. maxIndex int // Includes head, where Head will move to
  57. // close this when the processTicks routine is done.
  58. // this ensures we can cleanup the dir after calling Stop
  59. // and the routine won't be trying to access it anymore
  60. doneProcessTicks chan struct{}
  61. // TODO: When we start deleting files, we need to start tracking GroupReaders
  62. // and their dependencies.
  63. }
  64. // OpenGroup creates a new Group with head at headPath. It returns an error if
  65. // it fails to open head file.
  66. func OpenGroup(headPath string, groupOptions ...func(*Group)) (g *Group, err error) {
  67. dir := path.Dir(headPath)
  68. head, err := OpenAutoFile(headPath)
  69. if err != nil {
  70. return nil, err
  71. }
  72. g = &Group{
  73. ID: "group:" + head.ID,
  74. Head: head,
  75. headBuf: bufio.NewWriterSize(head, 4096*10),
  76. Dir: dir,
  77. headSizeLimit: defaultHeadSizeLimit,
  78. totalSizeLimit: defaultTotalSizeLimit,
  79. groupCheckDuration: defaultGroupCheckDuration,
  80. minIndex: 0,
  81. maxIndex: 0,
  82. doneProcessTicks: make(chan struct{}),
  83. }
  84. for _, option := range groupOptions {
  85. option(g)
  86. }
  87. g.BaseService = *cmn.NewBaseService(nil, "Group", g)
  88. gInfo := g.readGroupInfo()
  89. g.minIndex = gInfo.MinIndex
  90. g.maxIndex = gInfo.MaxIndex
  91. return
  92. }
  93. // GroupCheckDuration allows you to overwrite default groupCheckDuration.
  94. func GroupCheckDuration(duration time.Duration) func(*Group) {
  95. return func(g *Group) {
  96. g.groupCheckDuration = duration
  97. }
  98. }
  99. // GroupHeadSizeLimit allows you to overwrite default head size limit - 10MB.
  100. func GroupHeadSizeLimit(limit int64) func(*Group) {
  101. return func(g *Group) {
  102. g.headSizeLimit = limit
  103. }
  104. }
  105. // GroupTotalSizeLimit allows you to overwrite default total size limit of the group - 1GB.
  106. func GroupTotalSizeLimit(limit int64) func(*Group) {
  107. return func(g *Group) {
  108. g.totalSizeLimit = limit
  109. }
  110. }
  111. // OnStart implements cmn.Service by starting the goroutine that checks file
  112. // and group limits.
  113. func (g *Group) OnStart() error {
  114. g.ticker = time.NewTicker(g.groupCheckDuration)
  115. go g.processTicks()
  116. return nil
  117. }
  118. // OnStop implements cmn.Service by stopping the goroutine described above.
  119. // NOTE: g.Head must be closed separately using Close.
  120. func (g *Group) OnStop() {
  121. g.ticker.Stop()
  122. g.FlushAndSync()
  123. }
  124. // Wait blocks until all internal goroutines are finished. Supposed to be
  125. // called after Stop.
  126. func (g *Group) Wait() {
  127. // wait for processTicks routine to finish
  128. <-g.doneProcessTicks
  129. }
  130. // Close closes the head file. The group must be stopped by this moment.
  131. func (g *Group) Close() {
  132. g.FlushAndSync()
  133. g.mtx.Lock()
  134. _ = g.Head.closeFile()
  135. g.mtx.Unlock()
  136. }
  137. // HeadSizeLimit returns the current head size limit.
  138. func (g *Group) HeadSizeLimit() int64 {
  139. g.mtx.Lock()
  140. defer g.mtx.Unlock()
  141. return g.headSizeLimit
  142. }
  143. // TotalSizeLimit returns total size limit of the group.
  144. func (g *Group) TotalSizeLimit() int64 {
  145. g.mtx.Lock()
  146. defer g.mtx.Unlock()
  147. return g.totalSizeLimit
  148. }
  149. // MaxIndex returns index of the last file in the group.
  150. func (g *Group) MaxIndex() int {
  151. g.mtx.Lock()
  152. defer g.mtx.Unlock()
  153. return g.maxIndex
  154. }
  155. // MinIndex returns index of the first file in the group.
  156. func (g *Group) MinIndex() int {
  157. g.mtx.Lock()
  158. defer g.mtx.Unlock()
  159. return g.minIndex
  160. }
  161. // Write writes the contents of p into the current head of the group. It
  162. // returns the number of bytes written. If nn < len(p), it also returns an
  163. // error explaining why the write is short.
  164. // NOTE: Writes are buffered so they don't write synchronously
  165. // TODO: Make it halt if space is unavailable
  166. func (g *Group) Write(p []byte) (nn int, err error) {
  167. g.mtx.Lock()
  168. defer g.mtx.Unlock()
  169. return g.headBuf.Write(p)
  170. }
  171. // WriteLine writes line into the current head of the group. It also appends "\n".
  172. // NOTE: Writes are buffered so they don't write synchronously
  173. // TODO: Make it halt if space is unavailable
  174. func (g *Group) WriteLine(line string) error {
  175. g.mtx.Lock()
  176. defer g.mtx.Unlock()
  177. _, err := g.headBuf.Write([]byte(line + "\n"))
  178. return err
  179. }
  180. // Buffered returns the size of the currently buffered data.
  181. func (g *Group) Buffered() int {
  182. g.mtx.Lock()
  183. defer g.mtx.Unlock()
  184. return g.headBuf.Buffered()
  185. }
  186. // FlushAndSync writes any buffered data to the underlying file and commits the
  187. // current content of the file to stable storage (fsync).
  188. func (g *Group) FlushAndSync() error {
  189. g.mtx.Lock()
  190. defer g.mtx.Unlock()
  191. err := g.headBuf.Flush()
  192. if err == nil {
  193. err = g.Head.Sync()
  194. }
  195. return err
  196. }
  197. func (g *Group) processTicks() {
  198. defer close(g.doneProcessTicks)
  199. for {
  200. select {
  201. case <-g.ticker.C:
  202. g.checkHeadSizeLimit()
  203. g.checkTotalSizeLimit()
  204. case <-g.Quit():
  205. return
  206. }
  207. }
  208. }
  209. // NOTE: this function is called manually in tests.
  210. func (g *Group) checkHeadSizeLimit() {
  211. limit := g.HeadSizeLimit()
  212. if limit == 0 {
  213. return
  214. }
  215. size, err := g.Head.Size()
  216. if err != nil {
  217. g.Logger.Error("Group's head may grow without bound", "head", g.Head.Path, "err", err)
  218. return
  219. }
  220. if size >= limit {
  221. g.RotateFile()
  222. }
  223. }
  224. func (g *Group) checkTotalSizeLimit() {
  225. limit := g.TotalSizeLimit()
  226. if limit == 0 {
  227. return
  228. }
  229. gInfo := g.readGroupInfo()
  230. totalSize := gInfo.TotalSize
  231. for i := 0; i < maxFilesToRemove; i++ {
  232. index := gInfo.MinIndex + i
  233. if totalSize < limit {
  234. return
  235. }
  236. if index == gInfo.MaxIndex {
  237. // Special degenerate case, just do nothing.
  238. g.Logger.Error("Group's head may grow without bound", "head", g.Head.Path)
  239. return
  240. }
  241. pathToRemove := filePathForIndex(g.Head.Path, index, gInfo.MaxIndex)
  242. fInfo, err := os.Stat(pathToRemove)
  243. if err != nil {
  244. g.Logger.Error("Failed to fetch info for file", "file", pathToRemove)
  245. continue
  246. }
  247. err = os.Remove(pathToRemove)
  248. if err != nil {
  249. g.Logger.Error("Failed to remove path", "path", pathToRemove)
  250. return
  251. }
  252. totalSize -= fInfo.Size()
  253. }
  254. }
  255. // RotateFile causes group to close the current head and assign it some index.
  256. // Note it does not create a new head.
  257. func (g *Group) RotateFile() {
  258. g.mtx.Lock()
  259. defer g.mtx.Unlock()
  260. headPath := g.Head.Path
  261. if err := g.headBuf.Flush(); err != nil {
  262. panic(err)
  263. }
  264. if err := g.Head.Sync(); err != nil {
  265. panic(err)
  266. }
  267. if err := g.Head.closeFile(); err != nil {
  268. panic(err)
  269. }
  270. indexPath := filePathForIndex(headPath, g.maxIndex, g.maxIndex+1)
  271. if err := os.Rename(headPath, indexPath); err != nil {
  272. panic(err)
  273. }
  274. g.maxIndex++
  275. }
  276. // NewReader returns a new group reader.
  277. // CONTRACT: Caller must close the returned GroupReader.
  278. func (g *Group) NewReader(index int) (*GroupReader, error) {
  279. r := newGroupReader(g)
  280. err := r.SetIndex(index)
  281. if err != nil {
  282. return nil, err
  283. }
  284. return r, nil
  285. }
  286. // Returns -1 if line comes after, 0 if found, 1 if line comes before.
  287. type SearchFunc func(line string) (int, error)
  288. // Searches for the right file in Group, then returns a GroupReader to start
  289. // streaming lines.
  290. // Returns true if an exact match was found, otherwise returns the next greater
  291. // line that starts with prefix.
  292. // CONTRACT: Caller must close the returned GroupReader
  293. func (g *Group) Search(prefix string, cmp SearchFunc) (*GroupReader, bool, error) {
  294. g.mtx.Lock()
  295. minIndex, maxIndex := g.minIndex, g.maxIndex
  296. g.mtx.Unlock()
  297. // Now minIndex/maxIndex may change meanwhile,
  298. // but it shouldn't be a big deal
  299. // (maybe we'll want to limit scanUntil though)
  300. for {
  301. curIndex := (minIndex + maxIndex + 1) / 2
  302. // Base case, when there's only 1 choice left.
  303. if minIndex == maxIndex {
  304. r, err := g.NewReader(maxIndex)
  305. if err != nil {
  306. return nil, false, err
  307. }
  308. match, err := scanUntil(r, prefix, cmp)
  309. if err != nil {
  310. r.Close()
  311. return nil, false, err
  312. }
  313. return r, match, err
  314. }
  315. // Read starting roughly at the middle file,
  316. // until we find line that has prefix.
  317. r, err := g.NewReader(curIndex)
  318. if err != nil {
  319. return nil, false, err
  320. }
  321. foundIndex, line, err := scanNext(r, prefix)
  322. r.Close()
  323. if err != nil {
  324. return nil, false, err
  325. }
  326. // Compare this line to our search query.
  327. val, err := cmp(line)
  328. if err != nil {
  329. return nil, false, err
  330. }
  331. if val < 0 {
  332. // Line will come later
  333. minIndex = foundIndex
  334. } else if val == 0 {
  335. // Stroke of luck, found the line
  336. r, err := g.NewReader(foundIndex)
  337. if err != nil {
  338. return nil, false, err
  339. }
  340. match, err := scanUntil(r, prefix, cmp)
  341. if !match {
  342. panic("Expected match to be true")
  343. }
  344. if err != nil {
  345. r.Close()
  346. return nil, false, err
  347. }
  348. return r, true, err
  349. } else {
  350. // We passed it
  351. maxIndex = curIndex - 1
  352. }
  353. }
  354. }
  355. // Scans and returns the first line that starts with 'prefix'
  356. // Consumes line and returns it.
  357. func scanNext(r *GroupReader, prefix string) (int, string, error) {
  358. for {
  359. line, err := r.ReadLine()
  360. if err != nil {
  361. return 0, "", err
  362. }
  363. if !strings.HasPrefix(line, prefix) {
  364. continue
  365. }
  366. index := r.CurIndex()
  367. return index, line, nil
  368. }
  369. }
  370. // Returns true iff an exact match was found.
  371. // Pushes line, does not consume it.
  372. func scanUntil(r *GroupReader, prefix string, cmp SearchFunc) (bool, error) {
  373. for {
  374. line, err := r.ReadLine()
  375. if err != nil {
  376. return false, err
  377. }
  378. if !strings.HasPrefix(line, prefix) {
  379. continue
  380. }
  381. val, err := cmp(line)
  382. if err != nil {
  383. return false, err
  384. }
  385. if val < 0 {
  386. continue
  387. } else if val == 0 {
  388. r.PushLine(line)
  389. return true, nil
  390. } else {
  391. r.PushLine(line)
  392. return false, nil
  393. }
  394. }
  395. }
  396. // Searches backwards for the last line in Group with prefix.
  397. // Scans each file forward until the end to find the last match.
  398. func (g *Group) FindLast(prefix string) (match string, found bool, err error) {
  399. g.mtx.Lock()
  400. minIndex, maxIndex := g.minIndex, g.maxIndex
  401. g.mtx.Unlock()
  402. r, err := g.NewReader(maxIndex)
  403. if err != nil {
  404. return "", false, err
  405. }
  406. defer r.Close()
  407. // Open files from the back and read
  408. GROUP_LOOP:
  409. for i := maxIndex; i >= minIndex; i-- {
  410. err := r.SetIndex(i)
  411. if err != nil {
  412. return "", false, err
  413. }
  414. // Scan each line and test whether line matches
  415. for {
  416. line, err := r.ReadLine()
  417. if err == io.EOF {
  418. if found {
  419. return match, found, nil
  420. }
  421. continue GROUP_LOOP
  422. } else if err != nil {
  423. return "", false, err
  424. }
  425. if strings.HasPrefix(line, prefix) {
  426. match = line
  427. found = true
  428. }
  429. if r.CurIndex() > i {
  430. if found {
  431. return match, found, nil
  432. }
  433. continue GROUP_LOOP
  434. }
  435. }
  436. }
  437. return
  438. }
  439. // GroupInfo holds information about the group.
  440. type GroupInfo struct {
  441. MinIndex int // index of the first file in the group, including head
  442. MaxIndex int // index of the last file in the group, including head
  443. TotalSize int64 // total size of the group
  444. HeadSize int64 // size of the head
  445. }
  446. // Returns info after scanning all files in g.Head's dir.
  447. func (g *Group) ReadGroupInfo() GroupInfo {
  448. g.mtx.Lock()
  449. defer g.mtx.Unlock()
  450. return g.readGroupInfo()
  451. }
  452. // Index includes the head.
  453. // CONTRACT: caller should have called g.mtx.Lock
  454. func (g *Group) readGroupInfo() GroupInfo {
  455. groupDir := filepath.Dir(g.Head.Path)
  456. headBase := filepath.Base(g.Head.Path)
  457. var minIndex, maxIndex int = -1, -1
  458. var totalSize, headSize int64 = 0, 0
  459. dir, err := os.Open(groupDir)
  460. if err != nil {
  461. panic(err)
  462. }
  463. defer dir.Close()
  464. fiz, err := dir.Readdir(0)
  465. if err != nil {
  466. panic(err)
  467. }
  468. // For each file in the directory, filter by pattern
  469. for _, fileInfo := range fiz {
  470. if fileInfo.Name() == headBase {
  471. fileSize := fileInfo.Size()
  472. totalSize += fileSize
  473. headSize = fileSize
  474. continue
  475. } else if strings.HasPrefix(fileInfo.Name(), headBase) {
  476. fileSize := fileInfo.Size()
  477. totalSize += fileSize
  478. indexedFilePattern := regexp.MustCompile(`^.+\.([0-9]{3,})$`)
  479. submatch := indexedFilePattern.FindSubmatch([]byte(fileInfo.Name()))
  480. if len(submatch) != 0 {
  481. // Matches
  482. fileIndex, err := strconv.Atoi(string(submatch[1]))
  483. if err != nil {
  484. panic(err)
  485. }
  486. if maxIndex < fileIndex {
  487. maxIndex = fileIndex
  488. }
  489. if minIndex == -1 || fileIndex < minIndex {
  490. minIndex = fileIndex
  491. }
  492. }
  493. }
  494. }
  495. // Now account for the head.
  496. if minIndex == -1 {
  497. // If there were no numbered files,
  498. // then the head is index 0.
  499. minIndex, maxIndex = 0, 0
  500. } else {
  501. // Otherwise, the head file is 1 greater
  502. maxIndex++
  503. }
  504. return GroupInfo{minIndex, maxIndex, totalSize, headSize}
  505. }
  506. func filePathForIndex(headPath string, index int, maxIndex int) string {
  507. if index == maxIndex {
  508. return headPath
  509. }
  510. return fmt.Sprintf("%v.%03d", headPath, index)
  511. }
  512. //--------------------------------------------------------------------------------
  513. // GroupReader provides an interface for reading from a Group.
  514. type GroupReader struct {
  515. *Group
  516. mtx sync.Mutex
  517. curIndex int
  518. curFile *os.File
  519. curReader *bufio.Reader
  520. curLine []byte
  521. }
  522. func newGroupReader(g *Group) *GroupReader {
  523. return &GroupReader{
  524. Group: g,
  525. curIndex: 0,
  526. curFile: nil,
  527. curReader: nil,
  528. curLine: nil,
  529. }
  530. }
  531. // Close closes the GroupReader by closing the cursor file.
  532. func (gr *GroupReader) Close() error {
  533. gr.mtx.Lock()
  534. defer gr.mtx.Unlock()
  535. if gr.curReader != nil {
  536. err := gr.curFile.Close()
  537. gr.curIndex = 0
  538. gr.curReader = nil
  539. gr.curFile = nil
  540. gr.curLine = nil
  541. return err
  542. }
  543. return nil
  544. }
  545. // Read implements io.Reader, reading bytes from the current Reader
  546. // incrementing index until enough bytes are read.
  547. func (gr *GroupReader) Read(p []byte) (n int, err error) {
  548. lenP := len(p)
  549. if lenP == 0 {
  550. return 0, errors.New("given empty slice")
  551. }
  552. gr.mtx.Lock()
  553. defer gr.mtx.Unlock()
  554. // Open file if not open yet
  555. if gr.curReader == nil {
  556. if err = gr.openFile(gr.curIndex); err != nil {
  557. return 0, err
  558. }
  559. }
  560. // Iterate over files until enough bytes are read
  561. var nn int
  562. for {
  563. nn, err = gr.curReader.Read(p[n:])
  564. n += nn
  565. if err == io.EOF {
  566. if n >= lenP {
  567. return n, nil
  568. }
  569. // Open the next file
  570. if err1 := gr.openFile(gr.curIndex + 1); err1 != nil {
  571. return n, err1
  572. }
  573. } else if err != nil {
  574. return n, err
  575. } else if nn == 0 { // empty file
  576. return n, err
  577. }
  578. }
  579. }
  580. // ReadLine reads a line (without delimiter).
  581. // just return io.EOF if no new lines found.
  582. func (gr *GroupReader) ReadLine() (string, error) {
  583. gr.mtx.Lock()
  584. defer gr.mtx.Unlock()
  585. // From PushLine
  586. if gr.curLine != nil {
  587. line := string(gr.curLine)
  588. gr.curLine = nil
  589. return line, nil
  590. }
  591. // Open file if not open yet
  592. if gr.curReader == nil {
  593. err := gr.openFile(gr.curIndex)
  594. if err != nil {
  595. return "", err
  596. }
  597. }
  598. // Iterate over files until line is found
  599. var linePrefix string
  600. for {
  601. bytesRead, err := gr.curReader.ReadBytes('\n')
  602. if err == io.EOF {
  603. // Open the next file
  604. if err1 := gr.openFile(gr.curIndex + 1); err1 != nil {
  605. return "", err1
  606. }
  607. if len(bytesRead) > 0 && bytesRead[len(bytesRead)-1] == byte('\n') {
  608. return linePrefix + string(bytesRead[:len(bytesRead)-1]), nil
  609. }
  610. linePrefix += string(bytesRead)
  611. continue
  612. } else if err != nil {
  613. return "", err
  614. }
  615. return linePrefix + string(bytesRead[:len(bytesRead)-1]), nil
  616. }
  617. }
  618. // IF index > gr.Group.maxIndex, returns io.EOF
  619. // CONTRACT: caller should hold gr.mtx
  620. func (gr *GroupReader) openFile(index int) error {
  621. // Lock on Group to ensure that head doesn't move in the meanwhile.
  622. gr.Group.mtx.Lock()
  623. defer gr.Group.mtx.Unlock()
  624. if index > gr.Group.maxIndex {
  625. return io.EOF
  626. }
  627. curFilePath := filePathForIndex(gr.Head.Path, index, gr.Group.maxIndex)
  628. curFile, err := os.OpenFile(curFilePath, os.O_RDONLY|os.O_CREATE, autoFilePerms)
  629. if err != nil {
  630. return err
  631. }
  632. curReader := bufio.NewReader(curFile)
  633. // Update gr.cur*
  634. if gr.curFile != nil {
  635. gr.curFile.Close() // TODO return error?
  636. }
  637. gr.curIndex = index
  638. gr.curFile = curFile
  639. gr.curReader = curReader
  640. gr.curLine = nil
  641. return nil
  642. }
  643. // PushLine makes the given line the current one, so the next time somebody
  644. // calls ReadLine, this line will be returned.
  645. // panics if called twice without calling ReadLine.
  646. func (gr *GroupReader) PushLine(line string) {
  647. gr.mtx.Lock()
  648. defer gr.mtx.Unlock()
  649. if gr.curLine == nil {
  650. gr.curLine = []byte(line)
  651. } else {
  652. panic("PushLine failed, already have line")
  653. }
  654. }
  655. // CurIndex returns cursor's file index.
  656. func (gr *GroupReader) CurIndex() int {
  657. gr.mtx.Lock()
  658. defer gr.mtx.Unlock()
  659. return gr.curIndex
  660. }
  661. // SetIndex sets the cursor's file index to index by opening a file at this
  662. // position.
  663. func (gr *GroupReader) SetIndex(index int) error {
  664. gr.mtx.Lock()
  665. defer gr.mtx.Unlock()
  666. return gr.openFile(index)
  667. }
  668. //--------------------------------------------------------------------------------
  669. // A simple SearchFunc that assumes that the marker is of form
  670. // <prefix><number>.
  671. // For example, if prefix is '#HEIGHT:', the markers of expected to be of the form:
  672. //
  673. // #HEIGHT:1
  674. // ...
  675. // #HEIGHT:2
  676. // ...
  677. func MakeSimpleSearchFunc(prefix string, target int) SearchFunc {
  678. return func(line string) (int, error) {
  679. if !strings.HasPrefix(line, prefix) {
  680. return -1, fmt.Errorf("Marker line did not have prefix: %v", prefix)
  681. }
  682. i, err := strconv.Atoi(line[len(prefix):])
  683. if err != nil {
  684. return -1, fmt.Errorf("Failed to parse marker line: %v", err.Error())
  685. }
  686. if target < i {
  687. return 1, nil
  688. } else if target == i {
  689. return 0, nil
  690. } else {
  691. return -1, nil
  692. }
  693. }
  694. }