threadpool_test.go 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415
  1. /*
  2. * Public Domain Software
  3. *
  4. * I (Matthias Ladkau) am the author of the source code in this file.
  5. * I have placed the source code in this file in the public domain.
  6. *
  7. * For further information see: http://creativecommons.org/publicdomain/zero/1.0/
  8. */
  9. package pool
  10. import (
  11. "bytes"
  12. "errors"
  13. "strings"
  14. "sync"
  15. "testing"
  16. "time"
  17. )
  18. type testTask struct {
  19. task func() error
  20. errorHandler func(e error)
  21. }
  22. func (t *testTask) Run() error {
  23. return t.task()
  24. }
  25. func (t *testTask) HandleError(e error) {
  26. t.errorHandler(e)
  27. }
  28. func TestDefaultTaskQueue(t *testing.T) {
  29. var taskFinishCounter int
  30. var tq DefaultTaskQueue
  31. if res := tq.Size(); res != 0 {
  32. t.Error("Initial size should be empty not: ", res)
  33. return
  34. }
  35. if res := tq.Pop(); res != nil {
  36. t.Error("Unexpected result: ", res)
  37. return
  38. }
  39. tq.Clear()
  40. if res := tq.Size(); res != 0 {
  41. t.Error("Initial size should be empty not: ", res)
  42. return
  43. }
  44. if res := tq.Pop(); res != nil {
  45. t.Error("Unexpected result: ", res)
  46. return
  47. }
  48. tq.Push(&testTask{func() error {
  49. taskFinishCounter++
  50. return nil
  51. }, nil})
  52. tq.Push(&testTask{func() error {
  53. taskFinishCounter++
  54. return nil
  55. }, nil})
  56. tq.Push(&testTask{func() error {
  57. taskFinishCounter++
  58. return nil
  59. }, nil})
  60. if res := tq.Size(); res != 3 {
  61. t.Error("Unexpected result: ", res)
  62. return
  63. }
  64. // Execute the functions
  65. tq.Pop().Run()
  66. if res := tq.Size(); res != 2 {
  67. t.Error("Unexpected result: ", res)
  68. return
  69. }
  70. tq.Pop().Run()
  71. if res := tq.Size(); res != 1 {
  72. t.Error("Unexpected result: ", res)
  73. return
  74. }
  75. tq.Pop().Run()
  76. if res := tq.Size(); res != 0 {
  77. t.Error("Unexpected result: ", res)
  78. return
  79. }
  80. if res := tq.Pop(); res != nil {
  81. t.Error("Unexpected result: ", res)
  82. return
  83. }
  84. if taskFinishCounter != 3 {
  85. t.Error("Unexpected result: ", taskFinishCounter)
  86. return
  87. }
  88. }
  89. func TestThreadPool(t *testing.T) {
  90. var taskFinishCounter int
  91. taskFinishCounterLock := &sync.Mutex{}
  92. tp := NewThreadPool()
  93. tp.SetWorkerCount(-10, true)
  94. tp.TooManyThreshold = 1
  95. if status := tp.Status(); status != StatusStopped {
  96. t.Error("Unexpected status:", status)
  97. return
  98. }
  99. tp.SetWorkerCount(3, true)
  100. if status := tp.Status(); status != StatusRunning {
  101. t.Error("Unexpected status:", status)
  102. return
  103. }
  104. if workers := len(tp.workerMap); workers != 3 {
  105. t.Error("Unepxected state:", workers)
  106. return
  107. }
  108. tp.AddTask(&testTask{func() error {
  109. taskFinishCounterLock.Lock()
  110. taskFinishCounter++
  111. taskFinishCounterLock.Unlock()
  112. return nil
  113. }, nil})
  114. tp.AddTask(&testTask{func() error {
  115. taskFinishCounterLock.Lock()
  116. taskFinishCounter++
  117. taskFinishCounterLock.Unlock()
  118. return nil
  119. }, nil})
  120. tp.AddTask(&testTask{func() error {
  121. taskFinishCounterLock.Lock()
  122. taskFinishCounter++
  123. taskFinishCounterLock.Unlock()
  124. return nil
  125. }, nil})
  126. tp.JoinAll()
  127. if workers := len(tp.workerMap); workers != 0 {
  128. t.Error("Unepxected state:", workers)
  129. return
  130. }
  131. if taskFinishCounter != 3 {
  132. t.Error("Unexpected result: ", taskFinishCounter)
  133. return
  134. }
  135. if status := tp.Status(); status != StatusStopped {
  136. t.Error("Unexpected status:", status)
  137. return
  138. }
  139. tp.AddTask(&testTask{func() error {
  140. taskFinishCounterLock.Lock()
  141. taskFinishCounter++
  142. taskFinishCounterLock.Unlock()
  143. return nil
  144. }, nil})
  145. tp.AddTask(&testTask{func() error {
  146. taskFinishCounterLock.Lock()
  147. taskFinishCounter++
  148. taskFinishCounterLock.Unlock()
  149. return nil
  150. }, nil})
  151. tp.AddTask(&testTask{func() error {
  152. taskFinishCounterLock.Lock()
  153. taskFinishCounter++
  154. taskFinishCounterLock.Unlock()
  155. return nil
  156. }, nil})
  157. tp.AddTask(&testTask{func() error {
  158. taskFinishCounterLock.Lock()
  159. taskFinishCounter++
  160. taskFinishCounterLock.Unlock()
  161. time.Sleep(10 * time.Millisecond)
  162. return nil
  163. }, nil})
  164. if status := tp.Status(); status != StatusStopped {
  165. t.Error("Unexpected status:", status)
  166. return
  167. }
  168. tp.SetWorkerCount(3, false)
  169. if workers := len(tp.workerMap); workers != 3 {
  170. t.Error("Unepxected state:", workers)
  171. return
  172. }
  173. // Let the workers go into the idle state
  174. time.Sleep(20 * time.Millisecond)
  175. // Reduce the number of workers
  176. tp.SetWorkerCount(1, true)
  177. if workers := len(tp.workerMap); workers != 1 {
  178. t.Error("Unepxected state:", workers)
  179. return
  180. }
  181. tp.AddTask(&testTask{func() error {
  182. taskFinishCounterLock.Lock()
  183. taskFinishCounter++
  184. taskFinishCounterLock.Unlock()
  185. return nil
  186. }, nil})
  187. tp.AddTask(&testTask{func() error {
  188. taskFinishCounterLock.Lock()
  189. taskFinishCounter++
  190. taskFinishCounterLock.Unlock()
  191. time.Sleep(10 * time.Millisecond)
  192. return nil
  193. }, nil})
  194. // Set the kill value
  195. tp.workerKill = -1
  196. if status := tp.Status(); status != StatusStopping {
  197. t.Error("Unexpected status:", status)
  198. return
  199. }
  200. tp.WaitAll()
  201. tp.SetWorkerCount(-5, true)
  202. if workers := len(tp.workerMap); workers != 0 {
  203. t.Error("Unepxected state:", workers)
  204. return
  205. }
  206. tp.AddTask(&testTask{func() error {
  207. taskFinishCounterLock.Lock()
  208. taskFinishCounter++
  209. taskFinishCounterLock.Unlock()
  210. return nil
  211. }, nil})
  212. tp.WaitAll()
  213. if taskFinishCounter != 9 {
  214. t.Error("Unexpected result: ", taskFinishCounter)
  215. return
  216. }
  217. tp.SetWorkerCount(1, false)
  218. tp.WaitAll()
  219. if taskFinishCounter != 10 {
  220. t.Error("Unexpected result: ", taskFinishCounter)
  221. return
  222. }
  223. tp.SetWorkerCount(0, true)
  224. if status := tp.Status(); status != StatusStopped {
  225. t.Error("Unexpected status:", status)
  226. return
  227. }
  228. }
  229. func TestThreadPoolThresholds(t *testing.T) {
  230. var taskFinishCounter int
  231. taskFinishCounterLock := &sync.Mutex{}
  232. task := &testTask{func() error {
  233. time.Sleep(time.Millisecond * 5)
  234. taskFinishCounterLock.Lock()
  235. taskFinishCounter++
  236. taskFinishCounterLock.Unlock()
  237. return nil
  238. }, nil}
  239. var buf bytes.Buffer
  240. tp := NewThreadPool()
  241. tp.TooFewThreshold = 1
  242. tp.TooManyThreshold = 5
  243. tp.TooFewCallback = func() {
  244. taskFinishCounterLock.Lock()
  245. buf.WriteString("low")
  246. taskFinishCounterLock.Unlock()
  247. }
  248. tp.TooManyCallback = func() {
  249. taskFinishCounterLock.Lock()
  250. buf.WriteString("high")
  251. taskFinishCounterLock.Unlock()
  252. }
  253. tp.SetWorkerCount(10, false)
  254. for i := 0; i < 10; i++ {
  255. tp.AddTask(task)
  256. }
  257. if wc := tp.WorkerCount(); wc != 10 {
  258. t.Error("Unexpected result:", wc)
  259. return
  260. }
  261. tp.SetWorkerCount(10, false)
  262. tp.WaitAll()
  263. if wc := tp.WorkerCount(); wc != 10 {
  264. t.Error("Unexpected result:", wc)
  265. return
  266. }
  267. tp.SetWorkerCount(10, false)
  268. for i := 0; i < 10; i++ {
  269. tp.AddTask(task)
  270. }
  271. tp.WaitAll()
  272. if wc := tp.WorkerCount(); wc != 10 {
  273. t.Error("Unexpected result:", wc)
  274. return
  275. }
  276. if taskFinishCounter != 20 {
  277. t.Error("Unexpected result:", taskFinishCounter)
  278. return
  279. }
  280. tp.JoinAll()
  281. if wc := tp.WorkerCount(); wc != 0 {
  282. t.Error("Unexpected result:", wc)
  283. return
  284. }
  285. // Check that the callbacks where triggered twice each
  286. if !strings.Contains(buf.String(), "high") {
  287. t.Error("Unexpected result:", buf.String())
  288. return
  289. }
  290. if !strings.Contains(buf.String(), "low") {
  291. t.Error("Unexpected result:", buf.String())
  292. return
  293. }
  294. }
  295. func TestThreadPoolIdleTaskPanic(t *testing.T) {
  296. defer func() {
  297. if r := recover(); r == nil {
  298. t.Error("Error handling on the idle task did not cause a panic")
  299. }
  300. }()
  301. // Run error handling function of idle task
  302. idleTask := &idleTask{}
  303. idleTask.HandleError(nil)
  304. }
  305. func TestThreadPoolErrorHandling(t *testing.T) {
  306. // Test error normal task handling
  307. var buf bytes.Buffer
  308. task := &testTask{func() error {
  309. return errors.New("testerror")
  310. }, func(e error) {
  311. buf.WriteString(e.Error())
  312. }}
  313. tp := NewThreadPool()
  314. tp.AddTask(task)
  315. if buf.String() != "" {
  316. t.Error("Unexpected result:", buf.String())
  317. }
  318. tp.SetWorkerCount(1, false)
  319. tp.JoinAll()
  320. if buf.String() != "testerror" {
  321. t.Error("Unexpected result:", buf.String())
  322. }
  323. }