queue.go 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. // Copyright (C) 2014 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at https://mozilla.org/MPL/2.0/.
  6. package model
  7. import (
  8. "sort"
  9. "time"
  10. "github.com/syncthing/syncthing/lib/rand"
  11. "github.com/syncthing/syncthing/lib/sync"
  12. )
  13. type jobQueue struct {
  14. progress []string
  15. queued []jobQueueEntry
  16. mut sync.Mutex
  17. }
  18. type jobQueueEntry struct {
  19. name string
  20. size int64
  21. modified int64
  22. }
  23. func newJobQueue() *jobQueue {
  24. return &jobQueue{
  25. mut: sync.NewMutex(),
  26. }
  27. }
  28. func (q *jobQueue) Push(file string, size int64, modified time.Time) {
  29. q.mut.Lock()
  30. // The range of UnixNano covers a range of reasonable timestamps.
  31. q.queued = append(q.queued, jobQueueEntry{file, size, modified.UnixNano()})
  32. q.mut.Unlock()
  33. }
  34. func (q *jobQueue) Pop() (string, bool) {
  35. q.mut.Lock()
  36. defer q.mut.Unlock()
  37. if len(q.queued) == 0 {
  38. return "", false
  39. }
  40. f := q.queued[0].name
  41. q.queued = q.queued[1:]
  42. q.progress = append(q.progress, f)
  43. return f, true
  44. }
  45. func (q *jobQueue) BringToFront(filename string) {
  46. q.mut.Lock()
  47. defer q.mut.Unlock()
  48. for i, cur := range q.queued {
  49. if cur.name == filename {
  50. if i > 0 {
  51. // Shift the elements before the selected element one step to
  52. // the right, overwriting the selected element
  53. copy(q.queued[1:i+1], q.queued[0:])
  54. // Put the selected element at the front
  55. q.queued[0] = cur
  56. }
  57. return
  58. }
  59. }
  60. }
  61. func (q *jobQueue) Done(file string) {
  62. q.mut.Lock()
  63. defer q.mut.Unlock()
  64. for i := range q.progress {
  65. if q.progress[i] == file {
  66. copy(q.progress[i:], q.progress[i+1:])
  67. q.progress = q.progress[:len(q.progress)-1]
  68. return
  69. }
  70. }
  71. }
  72. // Jobs returns a paginated list of file currently being pulled and files queued
  73. // to be pulled. It also returns how many items were skipped.
  74. func (q *jobQueue) Jobs(page, perpage int) ([]string, []string, int) {
  75. q.mut.Lock()
  76. defer q.mut.Unlock()
  77. toSkip := (page - 1) * perpage
  78. plen := len(q.progress)
  79. qlen := len(q.queued)
  80. if tot := plen + qlen; tot <= toSkip {
  81. return nil, nil, tot
  82. }
  83. if plen >= toSkip+perpage {
  84. progress := make([]string, perpage)
  85. copy(progress, q.progress[toSkip:toSkip+perpage])
  86. return progress, nil, toSkip
  87. }
  88. var progress []string
  89. if plen > toSkip {
  90. progress = make([]string, plen-toSkip)
  91. copy(progress, q.progress[toSkip:plen])
  92. toSkip = 0
  93. } else {
  94. toSkip -= plen
  95. }
  96. var queued []string
  97. if qlen-toSkip < perpage-len(progress) {
  98. queued = make([]string, qlen-toSkip)
  99. } else {
  100. queued = make([]string, perpage-len(progress))
  101. }
  102. for i := range queued {
  103. queued[i] = q.queued[i+toSkip].name
  104. }
  105. return progress, queued, (page - 1) * perpage
  106. }
  107. func (q *jobQueue) Shuffle() {
  108. q.mut.Lock()
  109. defer q.mut.Unlock()
  110. rand.Shuffle(q.queued)
  111. }
  112. func (q *jobQueue) Reset() {
  113. q.mut.Lock()
  114. defer q.mut.Unlock()
  115. q.progress = nil
  116. q.queued = nil
  117. }
  118. func (q *jobQueue) lenQueued() int {
  119. q.mut.Lock()
  120. defer q.mut.Unlock()
  121. return len(q.queued)
  122. }
  123. func (q *jobQueue) lenProgress() int {
  124. q.mut.Lock()
  125. defer q.mut.Unlock()
  126. return len(q.progress)
  127. }
  128. func (q *jobQueue) SortSmallestFirst() {
  129. q.mut.Lock()
  130. defer q.mut.Unlock()
  131. sort.Sort(smallestFirst(q.queued))
  132. }
  133. func (q *jobQueue) SortLargestFirst() {
  134. q.mut.Lock()
  135. defer q.mut.Unlock()
  136. sort.Sort(sort.Reverse(smallestFirst(q.queued)))
  137. }
  138. func (q *jobQueue) SortOldestFirst() {
  139. q.mut.Lock()
  140. defer q.mut.Unlock()
  141. sort.Sort(oldestFirst(q.queued))
  142. }
  143. func (q *jobQueue) SortNewestFirst() {
  144. q.mut.Lock()
  145. defer q.mut.Unlock()
  146. sort.Sort(sort.Reverse(oldestFirst(q.queued)))
  147. }
  148. // The usual sort.Interface boilerplate
  149. type smallestFirst []jobQueueEntry
  150. func (q smallestFirst) Len() int { return len(q) }
  151. func (q smallestFirst) Less(a, b int) bool { return q[a].size < q[b].size }
  152. func (q smallestFirst) Swap(a, b int) { q[a], q[b] = q[b], q[a] }
  153. type oldestFirst []jobQueueEntry
  154. func (q oldestFirst) Len() int { return len(q) }
  155. func (q oldestFirst) Less(a, b int) bool { return q[a].modified < q[b].modified }
  156. func (q oldestFirst) Swap(a, b int) { q[a], q[b] = q[b], q[a] }