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.

536 lines
13 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
  1. package autofile
  2. import (
  3. "bufio"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "os"
  8. "path/filepath"
  9. "regexp"
  10. "strconv"
  11. "strings"
  12. "sync"
  13. "time"
  14. "github.com/tendermint/tendermint/libs/service"
  15. )
  16. const (
  17. defaultGroupCheckDuration = 5000 * time.Millisecond
  18. defaultHeadSizeLimit = 10 * 1024 * 1024 // 10MB
  19. defaultTotalSizeLimit = 1 * 1024 * 1024 * 1024 // 1GB
  20. maxFilesToRemove = 4 // needs to be greater than 1
  21. )
  22. /*
  23. You can open a Group to keep restrictions on an AutoFile, like
  24. the maximum size of each chunk, and/or the total amount of bytes
  25. stored in the group.
  26. The first file to be written in the Group.Dir is the head file.
  27. Dir/
  28. - <HeadPath>
  29. Once the Head file reaches the size limit, it will be rotated.
  30. Dir/
  31. - <HeadPath>.000 // First rolled file
  32. - <HeadPath> // New head path, starts empty.
  33. // The implicit index is 001.
  34. As more files are written, the index numbers grow...
  35. Dir/
  36. - <HeadPath>.000 // First rolled file
  37. - <HeadPath>.001 // Second rolled file
  38. - ...
  39. - <HeadPath> // New head path
  40. The Group can also be used to binary-search for some line,
  41. assuming that marker lines are written occasionally.
  42. */
  43. type Group struct {
  44. service.BaseService
  45. ID string
  46. Head *AutoFile // The head AutoFile to write to
  47. headBuf *bufio.Writer
  48. Dir string // Directory that contains .Head
  49. ticker *time.Ticker
  50. mtx sync.Mutex
  51. headSizeLimit int64
  52. totalSizeLimit int64
  53. groupCheckDuration time.Duration
  54. minIndex int // Includes head
  55. maxIndex int // Includes head, where Head will move to
  56. // close this when the processTicks routine is done.
  57. // this ensures we can cleanup the dir after calling Stop
  58. // and the routine won't be trying to access it anymore
  59. doneProcessTicks chan struct{}
  60. // TODO: When we start deleting files, we need to start tracking GroupReaders
  61. // and their dependencies.
  62. }
  63. // OpenGroup creates a new Group with head at headPath. It returns an error if
  64. // it fails to open head file.
  65. func OpenGroup(headPath string, groupOptions ...func(*Group)) (*Group, error) {
  66. dir, err := filepath.Abs(filepath.Dir(headPath))
  67. if err != nil {
  68. return nil, err
  69. }
  70. head, err := OpenAutoFile(headPath)
  71. if err != nil {
  72. return nil, err
  73. }
  74. g := &Group{
  75. ID: "group:" + head.ID,
  76. Head: head,
  77. headBuf: bufio.NewWriterSize(head, 4096*10),
  78. Dir: dir,
  79. headSizeLimit: defaultHeadSizeLimit,
  80. totalSizeLimit: defaultTotalSizeLimit,
  81. groupCheckDuration: defaultGroupCheckDuration,
  82. minIndex: 0,
  83. maxIndex: 0,
  84. doneProcessTicks: make(chan struct{}),
  85. }
  86. for _, option := range groupOptions {
  87. option(g)
  88. }
  89. g.BaseService = *service.NewBaseService(nil, "Group", g)
  90. gInfo := g.readGroupInfo()
  91. g.minIndex = gInfo.MinIndex
  92. g.maxIndex = gInfo.MaxIndex
  93. return g, nil
  94. }
  95. // GroupCheckDuration allows you to overwrite default groupCheckDuration.
  96. func GroupCheckDuration(duration time.Duration) func(*Group) {
  97. return func(g *Group) {
  98. g.groupCheckDuration = duration
  99. }
  100. }
  101. // GroupHeadSizeLimit allows you to overwrite default head size limit - 10MB.
  102. func GroupHeadSizeLimit(limit int64) func(*Group) {
  103. return func(g *Group) {
  104. g.headSizeLimit = limit
  105. }
  106. }
  107. // GroupTotalSizeLimit allows you to overwrite default total size limit of the group - 1GB.
  108. func GroupTotalSizeLimit(limit int64) func(*Group) {
  109. return func(g *Group) {
  110. g.totalSizeLimit = limit
  111. }
  112. }
  113. // OnStart implements service.Service by starting the goroutine that checks file
  114. // and group limits.
  115. func (g *Group) OnStart() error {
  116. g.ticker = time.NewTicker(g.groupCheckDuration)
  117. go g.processTicks()
  118. return nil
  119. }
  120. // OnStop implements service.Service by stopping the goroutine described above.
  121. // NOTE: g.Head must be closed separately using Close.
  122. func (g *Group) OnStop() {
  123. g.ticker.Stop()
  124. g.FlushAndSync()
  125. }
  126. // Wait blocks until all internal goroutines are finished. Supposed to be
  127. // called after Stop.
  128. func (g *Group) Wait() {
  129. // wait for processTicks routine to finish
  130. <-g.doneProcessTicks
  131. }
  132. // Close closes the head file. The group must be stopped by this moment.
  133. func (g *Group) Close() {
  134. g.FlushAndSync()
  135. g.mtx.Lock()
  136. _ = g.Head.closeFile()
  137. g.mtx.Unlock()
  138. }
  139. // HeadSizeLimit returns the current head size limit.
  140. func (g *Group) HeadSizeLimit() int64 {
  141. g.mtx.Lock()
  142. defer g.mtx.Unlock()
  143. return g.headSizeLimit
  144. }
  145. // TotalSizeLimit returns total size limit of the group.
  146. func (g *Group) TotalSizeLimit() int64 {
  147. g.mtx.Lock()
  148. defer g.mtx.Unlock()
  149. return g.totalSizeLimit
  150. }
  151. // MaxIndex returns index of the last file in the group.
  152. func (g *Group) MaxIndex() int {
  153. g.mtx.Lock()
  154. defer g.mtx.Unlock()
  155. return g.maxIndex
  156. }
  157. // MinIndex returns index of the first file in the group.
  158. func (g *Group) MinIndex() int {
  159. g.mtx.Lock()
  160. defer g.mtx.Unlock()
  161. return g.minIndex
  162. }
  163. // Write writes the contents of p into the current head of the group. It
  164. // returns the number of bytes written. If nn < len(p), it also returns an
  165. // error explaining why the write is short.
  166. // NOTE: Writes are buffered so they don't write synchronously
  167. // TODO: Make it halt if space is unavailable
  168. func (g *Group) Write(p []byte) (nn int, err error) {
  169. g.mtx.Lock()
  170. defer g.mtx.Unlock()
  171. return g.headBuf.Write(p)
  172. }
  173. // WriteLine writes line into the current head of the group. It also appends "\n".
  174. // NOTE: Writes are buffered so they don't write synchronously
  175. // TODO: Make it halt if space is unavailable
  176. func (g *Group) WriteLine(line string) error {
  177. g.mtx.Lock()
  178. defer g.mtx.Unlock()
  179. _, err := g.headBuf.Write([]byte(line + "\n"))
  180. return err
  181. }
  182. // Buffered returns the size of the currently buffered data.
  183. func (g *Group) Buffered() int {
  184. g.mtx.Lock()
  185. defer g.mtx.Unlock()
  186. return g.headBuf.Buffered()
  187. }
  188. // FlushAndSync writes any buffered data to the underlying file and commits the
  189. // current content of the file to stable storage (fsync).
  190. func (g *Group) FlushAndSync() error {
  191. g.mtx.Lock()
  192. defer g.mtx.Unlock()
  193. err := g.headBuf.Flush()
  194. if err == nil {
  195. err = g.Head.Sync()
  196. }
  197. return err
  198. }
  199. func (g *Group) processTicks() {
  200. defer close(g.doneProcessTicks)
  201. for {
  202. select {
  203. case <-g.ticker.C:
  204. g.checkHeadSizeLimit()
  205. g.checkTotalSizeLimit()
  206. case <-g.Quit():
  207. return
  208. }
  209. }
  210. }
  211. // NOTE: this function is called manually in tests.
  212. func (g *Group) checkHeadSizeLimit() {
  213. limit := g.HeadSizeLimit()
  214. if limit == 0 {
  215. return
  216. }
  217. size, err := g.Head.Size()
  218. if err != nil {
  219. g.Logger.Error("Group's head may grow without bound", "head", g.Head.Path, "err", err)
  220. return
  221. }
  222. if size >= limit {
  223. g.RotateFile()
  224. }
  225. }
  226. func (g *Group) checkTotalSizeLimit() {
  227. limit := g.TotalSizeLimit()
  228. if limit == 0 {
  229. return
  230. }
  231. gInfo := g.readGroupInfo()
  232. totalSize := gInfo.TotalSize
  233. for i := 0; i < maxFilesToRemove; i++ {
  234. index := gInfo.MinIndex + i
  235. if totalSize < limit {
  236. return
  237. }
  238. if index == gInfo.MaxIndex {
  239. // Special degenerate case, just do nothing.
  240. g.Logger.Error("Group's head may grow without bound", "head", g.Head.Path)
  241. return
  242. }
  243. pathToRemove := filePathForIndex(g.Head.Path, index, gInfo.MaxIndex)
  244. fInfo, err := os.Stat(pathToRemove)
  245. if err != nil {
  246. g.Logger.Error("Failed to fetch info for file", "file", pathToRemove)
  247. continue
  248. }
  249. err = os.Remove(pathToRemove)
  250. if err != nil {
  251. g.Logger.Error("Failed to remove path", "path", pathToRemove)
  252. return
  253. }
  254. totalSize -= fInfo.Size()
  255. }
  256. }
  257. // RotateFile causes group to close the current head and assign it some index.
  258. // Note it does not create a new head.
  259. func (g *Group) RotateFile() {
  260. g.mtx.Lock()
  261. defer g.mtx.Unlock()
  262. headPath := g.Head.Path
  263. if err := g.headBuf.Flush(); err != nil {
  264. panic(err)
  265. }
  266. if err := g.Head.Sync(); err != nil {
  267. panic(err)
  268. }
  269. if err := g.Head.closeFile(); err != nil {
  270. panic(err)
  271. }
  272. indexPath := filePathForIndex(headPath, g.maxIndex, g.maxIndex+1)
  273. if err := os.Rename(headPath, indexPath); err != nil {
  274. panic(err)
  275. }
  276. g.maxIndex++
  277. }
  278. // NewReader returns a new group reader.
  279. // CONTRACT: Caller must close the returned GroupReader.
  280. func (g *Group) NewReader(index int) (*GroupReader, error) {
  281. r := newGroupReader(g)
  282. err := r.SetIndex(index)
  283. if err != nil {
  284. return nil, err
  285. }
  286. return r, nil
  287. }
  288. // GroupInfo holds information about the group.
  289. type GroupInfo struct {
  290. MinIndex int // index of the first file in the group, including head
  291. MaxIndex int // index of the last file in the group, including head
  292. TotalSize int64 // total size of the group
  293. HeadSize int64 // size of the head
  294. }
  295. // Returns info after scanning all files in g.Head's dir.
  296. func (g *Group) ReadGroupInfo() GroupInfo {
  297. g.mtx.Lock()
  298. defer g.mtx.Unlock()
  299. return g.readGroupInfo()
  300. }
  301. // Index includes the head.
  302. // CONTRACT: caller should have called g.mtx.Lock
  303. func (g *Group) readGroupInfo() GroupInfo {
  304. groupDir := filepath.Dir(g.Head.Path)
  305. headBase := filepath.Base(g.Head.Path)
  306. var minIndex, maxIndex int = -1, -1
  307. var totalSize, headSize int64 = 0, 0
  308. dir, err := os.Open(groupDir)
  309. if err != nil {
  310. panic(err)
  311. }
  312. defer dir.Close()
  313. fiz, err := dir.Readdir(0)
  314. if err != nil {
  315. panic(err)
  316. }
  317. // For each file in the directory, filter by pattern
  318. for _, fileInfo := range fiz {
  319. if fileInfo.Name() == headBase {
  320. fileSize := fileInfo.Size()
  321. totalSize += fileSize
  322. headSize = fileSize
  323. continue
  324. } else if strings.HasPrefix(fileInfo.Name(), headBase) {
  325. fileSize := fileInfo.Size()
  326. totalSize += fileSize
  327. indexedFilePattern := regexp.MustCompile(`^.+\.([0-9]{3,})$`)
  328. submatch := indexedFilePattern.FindSubmatch([]byte(fileInfo.Name()))
  329. if len(submatch) != 0 {
  330. // Matches
  331. fileIndex, err := strconv.Atoi(string(submatch[1]))
  332. if err != nil {
  333. panic(err)
  334. }
  335. if maxIndex < fileIndex {
  336. maxIndex = fileIndex
  337. }
  338. if minIndex == -1 || fileIndex < minIndex {
  339. minIndex = fileIndex
  340. }
  341. }
  342. }
  343. }
  344. // Now account for the head.
  345. if minIndex == -1 {
  346. // If there were no numbered files,
  347. // then the head is index 0.
  348. minIndex, maxIndex = 0, 0
  349. } else {
  350. // Otherwise, the head file is 1 greater
  351. maxIndex++
  352. }
  353. return GroupInfo{minIndex, maxIndex, totalSize, headSize}
  354. }
  355. func filePathForIndex(headPath string, index int, maxIndex int) string {
  356. if index == maxIndex {
  357. return headPath
  358. }
  359. return fmt.Sprintf("%v.%03d", headPath, index)
  360. }
  361. //--------------------------------------------------------------------------------
  362. // GroupReader provides an interface for reading from a Group.
  363. type GroupReader struct {
  364. *Group
  365. mtx sync.Mutex
  366. curIndex int
  367. curFile *os.File
  368. curReader *bufio.Reader
  369. curLine []byte
  370. }
  371. func newGroupReader(g *Group) *GroupReader {
  372. return &GroupReader{
  373. Group: g,
  374. curIndex: 0,
  375. curFile: nil,
  376. curReader: nil,
  377. curLine: nil,
  378. }
  379. }
  380. // Close closes the GroupReader by closing the cursor file.
  381. func (gr *GroupReader) Close() error {
  382. gr.mtx.Lock()
  383. defer gr.mtx.Unlock()
  384. if gr.curReader != nil {
  385. err := gr.curFile.Close()
  386. gr.curIndex = 0
  387. gr.curReader = nil
  388. gr.curFile = nil
  389. gr.curLine = nil
  390. return err
  391. }
  392. return nil
  393. }
  394. // Read implements io.Reader, reading bytes from the current Reader
  395. // incrementing index until enough bytes are read.
  396. func (gr *GroupReader) Read(p []byte) (n int, err error) {
  397. lenP := len(p)
  398. if lenP == 0 {
  399. return 0, errors.New("given empty slice")
  400. }
  401. gr.mtx.Lock()
  402. defer gr.mtx.Unlock()
  403. // Open file if not open yet
  404. if gr.curReader == nil {
  405. if err = gr.openFile(gr.curIndex); err != nil {
  406. return 0, err
  407. }
  408. }
  409. // Iterate over files until enough bytes are read
  410. var nn int
  411. for {
  412. nn, err = gr.curReader.Read(p[n:])
  413. n += nn
  414. switch {
  415. case err == io.EOF:
  416. if n >= lenP {
  417. return n, nil
  418. }
  419. // Open the next file
  420. if err1 := gr.openFile(gr.curIndex + 1); err1 != nil {
  421. return n, err1
  422. }
  423. case err != nil:
  424. return n, err
  425. case nn == 0: // empty file
  426. return n, err
  427. }
  428. }
  429. }
  430. // IF index > gr.Group.maxIndex, returns io.EOF
  431. // CONTRACT: caller should hold gr.mtx
  432. func (gr *GroupReader) openFile(index int) error {
  433. // Lock on Group to ensure that head doesn't move in the meanwhile.
  434. gr.Group.mtx.Lock()
  435. defer gr.Group.mtx.Unlock()
  436. if index > gr.Group.maxIndex {
  437. return io.EOF
  438. }
  439. curFilePath := filePathForIndex(gr.Head.Path, index, gr.Group.maxIndex)
  440. curFile, err := os.OpenFile(curFilePath, os.O_RDONLY|os.O_CREATE, autoFilePerms)
  441. if err != nil {
  442. return err
  443. }
  444. curReader := bufio.NewReader(curFile)
  445. // Update gr.cur*
  446. if gr.curFile != nil {
  447. gr.curFile.Close() // TODO return error?
  448. }
  449. gr.curIndex = index
  450. gr.curFile = curFile
  451. gr.curReader = curReader
  452. gr.curLine = nil
  453. return nil
  454. }
  455. // CurIndex returns cursor's file index.
  456. func (gr *GroupReader) CurIndex() int {
  457. gr.mtx.Lock()
  458. defer gr.mtx.Unlock()
  459. return gr.curIndex
  460. }
  461. // SetIndex sets the cursor's file index to index by opening a file at this
  462. // position.
  463. func (gr *GroupReader) SetIndex(index int) error {
  464. gr.mtx.Lock()
  465. defer gr.mtx.Unlock()
  466. return gr.openFile(index)
  467. }