execqueue.go 2.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. // Copyright 2017 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. package les
  17. import "sync"
  18. // execQueue implements a queue that executes function calls in a single thread,
  19. // in the same order as they have been queued.
  20. type execQueue struct {
  21. mu sync.Mutex
  22. cond *sync.Cond
  23. funcs []func()
  24. closeWait chan struct{}
  25. }
  26. // newExecQueue creates a new execution queue.
  27. func newExecQueue(capacity int) *execQueue {
  28. q := &execQueue{funcs: make([]func(), 0, capacity)}
  29. q.cond = sync.NewCond(&q.mu)
  30. go q.loop()
  31. return q
  32. }
  33. func (q *execQueue) loop() {
  34. for f := q.waitNext(false); f != nil; f = q.waitNext(true) {
  35. f()
  36. }
  37. close(q.closeWait)
  38. }
  39. func (q *execQueue) waitNext(drop bool) (f func()) {
  40. q.mu.Lock()
  41. if drop {
  42. // Remove the function that just executed. We do this here instead of when
  43. // dequeuing so len(q.funcs) includes the function that is running.
  44. q.funcs = append(q.funcs[:0], q.funcs[1:]...)
  45. }
  46. for !q.isClosed() {
  47. if len(q.funcs) > 0 {
  48. f = q.funcs[0]
  49. break
  50. }
  51. q.cond.Wait()
  52. }
  53. q.mu.Unlock()
  54. return f
  55. }
  56. func (q *execQueue) isClosed() bool {
  57. return q.closeWait != nil
  58. }
  59. // canQueue returns true if more function calls can be added to the execution queue.
  60. func (q *execQueue) canQueue() bool {
  61. q.mu.Lock()
  62. ok := !q.isClosed() && len(q.funcs) < cap(q.funcs)
  63. q.mu.Unlock()
  64. return ok
  65. }
  66. // queue adds a function call to the execution queue. Returns true if successful.
  67. func (q *execQueue) queue(f func()) bool {
  68. q.mu.Lock()
  69. ok := !q.isClosed() && len(q.funcs) < cap(q.funcs)
  70. if ok {
  71. q.funcs = append(q.funcs, f)
  72. q.cond.Signal()
  73. }
  74. q.mu.Unlock()
  75. return ok
  76. }
  77. // quit stops the exec queue.
  78. // quit waits for the current execution to finish before returning.
  79. func (q *execQueue) quit() {
  80. q.mu.Lock()
  81. if !q.isClosed() {
  82. q.closeWait = make(chan struct{})
  83. q.cond.Signal()
  84. }
  85. q.mu.Unlock()
  86. <-q.closeWait
  87. }