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.

380 lines
8.1 KiB

  1. package clist
  2. /*
  3. The purpose of CList is to provide a goroutine-safe linked-list.
  4. This list can be traversed concurrently by any number of goroutines.
  5. However, removed CElements cannot be added back.
  6. NOTE: Not all methods of container/list are (yet) implemented.
  7. NOTE: Removed elements need to DetachPrev or DetachNext consistently
  8. to ensure garbage collection of removed elements.
  9. */
  10. import (
  11. "fmt"
  12. "sync"
  13. )
  14. // MaxLength is the max allowed number of elements a linked list is
  15. // allowed to contain.
  16. // If more elements are pushed to the list it will panic.
  17. const MaxLength = int(^uint(0) >> 1)
  18. /*
  19. CElement is an element of a linked-list
  20. Traversal from a CElement is goroutine-safe.
  21. We can't avoid using WaitGroups or for-loops given the documentation
  22. spec without re-implementing the primitives that already exist in
  23. golang/sync. Notice that WaitGroup allows many go-routines to be
  24. simultaneously released, which is what we want. Mutex doesn't do
  25. this. RWMutex does this, but it's clumsy to use in the way that a
  26. WaitGroup would be used -- and we'd end up having two RWMutex's for
  27. prev/next each, which is doubly confusing.
  28. sync.Cond would be sort-of useful, but we don't need a write-lock in
  29. the for-loop. Use sync.Cond when you need serial access to the
  30. "condition". In our case our condition is if `next != nil || removed`,
  31. and there's no reason to serialize that condition for goroutines
  32. waiting on NextWait() (since it's just a read operation).
  33. */
  34. type CElement struct {
  35. mtx sync.RWMutex
  36. prev *CElement
  37. prevWaitCh chan struct{}
  38. next *CElement
  39. nextWaitCh chan struct{}
  40. removed bool
  41. Value interface{} // immutable
  42. }
  43. // Blocking implementation of Next().
  44. // May return nil iff CElement was tail and got removed.
  45. func (e *CElement) NextWait() *CElement {
  46. for {
  47. e.mtx.RLock()
  48. next := e.next
  49. removed := e.removed
  50. signal := e.nextWaitCh
  51. e.mtx.RUnlock()
  52. if next != nil || removed {
  53. return next
  54. }
  55. <-signal
  56. // e.next doesn't necessarily exist here.
  57. // That's why we need to continue a for-loop.
  58. }
  59. }
  60. // Blocking implementation of Prev().
  61. // May return nil iff CElement was head and got removed.
  62. func (e *CElement) PrevWait() *CElement {
  63. for {
  64. e.mtx.RLock()
  65. prev := e.prev
  66. removed := e.removed
  67. signal := e.prevWaitCh
  68. e.mtx.RUnlock()
  69. if prev != nil || removed {
  70. return prev
  71. }
  72. <-signal
  73. }
  74. }
  75. // PrevWaitChan can be used to wait until Prev becomes not nil. Once it does,
  76. // channel will be closed.
  77. func (e *CElement) PrevWaitChan() <-chan struct{} {
  78. e.mtx.RLock()
  79. defer e.mtx.RUnlock()
  80. return e.prevWaitCh
  81. }
  82. // NextWaitChan can be used to wait until Next becomes not nil. Once it does,
  83. // channel will be closed.
  84. func (e *CElement) NextWaitChan() <-chan struct{} {
  85. e.mtx.RLock()
  86. defer e.mtx.RUnlock()
  87. return e.nextWaitCh
  88. }
  89. // Nonblocking, may return nil if at the end.
  90. func (e *CElement) Next() *CElement {
  91. e.mtx.RLock()
  92. val := e.next
  93. e.mtx.RUnlock()
  94. return val
  95. }
  96. // Nonblocking, may return nil if at the end.
  97. func (e *CElement) Prev() *CElement {
  98. e.mtx.RLock()
  99. prev := e.prev
  100. e.mtx.RUnlock()
  101. return prev
  102. }
  103. func (e *CElement) Removed() bool {
  104. e.mtx.RLock()
  105. isRemoved := e.removed
  106. e.mtx.RUnlock()
  107. return isRemoved
  108. }
  109. func (e *CElement) DetachNext() {
  110. e.mtx.Lock()
  111. if !e.removed {
  112. e.mtx.Unlock()
  113. panic("DetachNext() must be called after Remove(e)")
  114. }
  115. e.next = nil
  116. e.mtx.Unlock()
  117. }
  118. func (e *CElement) DetachPrev() {
  119. e.mtx.Lock()
  120. if !e.removed {
  121. e.mtx.Unlock()
  122. panic("DetachPrev() must be called after Remove(e)")
  123. }
  124. e.prev = nil
  125. e.mtx.Unlock()
  126. }
  127. // NOTE: This function needs to be safe for
  128. // concurrent goroutines waiting on nextWg.
  129. func (e *CElement) SetNext(newNext *CElement) {
  130. e.mtx.Lock()
  131. oldNext := e.next
  132. e.next = newNext
  133. if oldNext != nil && newNext == nil {
  134. // See https://golang.org/pkg/sync/:
  135. //
  136. // If a WaitGroup is reused to wait for several independent sets of
  137. // events, new Add calls must happen after all previous Wait calls have
  138. // returned.
  139. e.nextWaitCh = make(chan struct{})
  140. }
  141. if oldNext == nil && newNext != nil {
  142. close(e.nextWaitCh)
  143. }
  144. e.mtx.Unlock()
  145. }
  146. // NOTE: This function needs to be safe for
  147. // concurrent goroutines waiting on prevWg
  148. func (e *CElement) SetPrev(newPrev *CElement) {
  149. e.mtx.Lock()
  150. defer e.mtx.Unlock()
  151. oldPrev := e.prev
  152. e.prev = newPrev
  153. if oldPrev != nil && newPrev == nil {
  154. e.prevWaitCh = make(chan struct{})
  155. }
  156. if oldPrev == nil && newPrev != nil {
  157. close(e.prevWaitCh)
  158. }
  159. }
  160. func (e *CElement) SetRemoved() {
  161. e.mtx.Lock()
  162. defer e.mtx.Unlock()
  163. e.removed = true
  164. // This wakes up anyone waiting in either direction.
  165. if e.prev == nil {
  166. close(e.prevWaitCh)
  167. }
  168. if e.next == nil {
  169. close(e.nextWaitCh)
  170. }
  171. }
  172. //--------------------------------------------------------------------------------
  173. // CList represents a linked list.
  174. // The zero value for CList is an empty list ready to use.
  175. // Operations are goroutine-safe.
  176. // Panics if length grows beyond the max.
  177. type CList struct {
  178. mtx sync.RWMutex
  179. wg *sync.WaitGroup
  180. waitCh chan struct{}
  181. head *CElement // first element
  182. tail *CElement // last element
  183. len int // list length
  184. maxLen int // max list length
  185. }
  186. // Return CList with MaxLength. CList will panic if it goes beyond MaxLength.
  187. func New() *CList { return newWithMax(MaxLength) }
  188. // Return CList with given maxLength.
  189. // Will panic if list exceeds given maxLength.
  190. func newWithMax(maxLength int) *CList {
  191. l := new(CList)
  192. l.maxLen = maxLength
  193. l.waitCh = make(chan struct{})
  194. l.head = nil
  195. l.tail = nil
  196. l.len = 0
  197. return l
  198. }
  199. func (l *CList) Len() int {
  200. l.mtx.RLock()
  201. len := l.len
  202. l.mtx.RUnlock()
  203. return len
  204. }
  205. func (l *CList) Front() *CElement {
  206. l.mtx.RLock()
  207. head := l.head
  208. l.mtx.RUnlock()
  209. return head
  210. }
  211. func (l *CList) FrontWait() *CElement {
  212. // Loop until the head is non-nil else wait and try again
  213. for {
  214. l.mtx.RLock()
  215. head := l.head
  216. signal := l.waitCh
  217. l.mtx.RUnlock()
  218. if head != nil {
  219. return head
  220. }
  221. <-signal
  222. // NOTE: If you think l.head exists here, think harder.
  223. }
  224. }
  225. func (l *CList) Back() *CElement {
  226. l.mtx.RLock()
  227. back := l.tail
  228. l.mtx.RUnlock()
  229. return back
  230. }
  231. func (l *CList) BackWait() *CElement {
  232. for {
  233. l.mtx.RLock()
  234. tail := l.tail
  235. wg := l.wg
  236. l.mtx.RUnlock()
  237. if tail != nil {
  238. return tail
  239. }
  240. wg.Wait()
  241. // l.tail doesn't necessarily exist here.
  242. // That's why we need to continue a for-loop.
  243. }
  244. }
  245. // WaitChan can be used to wait until Front or Back becomes not nil. Once it
  246. // does, channel will be closed.
  247. func (l *CList) WaitChan() <-chan struct{} {
  248. l.mtx.Lock()
  249. defer l.mtx.Unlock()
  250. return l.waitCh
  251. }
  252. // Panics if list grows beyond its max length.
  253. func (l *CList) PushBack(v interface{}) *CElement {
  254. l.mtx.Lock()
  255. // Construct a new element
  256. e := &CElement{
  257. prev: nil,
  258. prevWaitCh: make(chan struct{}),
  259. next: nil,
  260. nextWaitCh: make(chan struct{}),
  261. removed: false,
  262. Value: v,
  263. }
  264. // Release waiters on FrontWait/BackWait maybe
  265. if l.len == 0 {
  266. close(l.waitCh)
  267. }
  268. if l.len >= l.maxLen {
  269. panic(fmt.Sprintf("clist: maximum length list reached %d", l.maxLen))
  270. }
  271. l.len++
  272. // Modify the tail
  273. if l.tail == nil {
  274. l.head = e
  275. l.tail = e
  276. } else {
  277. e.SetPrev(l.tail) // We must init e first.
  278. l.tail.SetNext(e) // This will make e accessible.
  279. l.tail = e // Update the list.
  280. }
  281. l.mtx.Unlock()
  282. return e
  283. }
  284. // CONTRACT: Caller must call e.DetachPrev() and/or e.DetachNext() to avoid memory leaks.
  285. // NOTE: As per the contract of CList, removed elements cannot be added back.
  286. func (l *CList) Remove(e *CElement) interface{} {
  287. l.mtx.Lock()
  288. defer l.mtx.Unlock()
  289. prev := e.Prev()
  290. next := e.Next()
  291. if l.head == nil || l.tail == nil {
  292. panic("Remove(e) on empty CList")
  293. }
  294. if prev == nil && l.head != e {
  295. panic("Remove(e) with false head")
  296. }
  297. if next == nil && l.tail != e {
  298. panic("Remove(e) with false tail")
  299. }
  300. // If we're removing the only item, make CList FrontWait/BackWait wait.
  301. if l.len == 1 {
  302. l.waitCh = make(chan struct{})
  303. }
  304. // Update l.len
  305. l.len--
  306. // Connect next/prev and set head/tail
  307. if prev == nil {
  308. l.head = next
  309. } else {
  310. prev.SetNext(next)
  311. }
  312. if next == nil {
  313. l.tail = prev
  314. } else {
  315. next.SetPrev(prev)
  316. }
  317. // Set .Done() on e, otherwise waiters will wait forever.
  318. e.SetRemoved()
  319. return e.Value
  320. }