ThreadPoolSchedulerTest.cs 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321
  1. // Licensed to the .NET Foundation under one or more agreements.
  2. // The .NET Foundation licenses this file to you under the Apache 2.0 License.
  3. // See the LICENSE file in the project root for more information.
  4. #if !NO_THREAD
  5. using System;
  6. using System.Collections.Generic;
  7. using System.Linq;
  8. using System.Reactive.Concurrency;
  9. using System.Reactive.Disposables;
  10. using System.Threading;
  11. using Microsoft.Reactive.Testing;
  12. using Xunit;
  13. namespace ReactiveTests.Tests
  14. {
  15. public class ThreadPoolSchedulerTest
  16. {
  17. [Fact]
  18. public void Schedule_ArgumentChecking()
  19. {
  20. ReactiveAssert.Throws<ArgumentNullException>(() => ThreadPoolScheduler.Instance.Schedule<int>(42, default(Func<IScheduler, int, IDisposable>)));
  21. ReactiveAssert.Throws<ArgumentNullException>(() => ThreadPoolScheduler.Instance.Schedule<int>(42, DateTimeOffset.Now, default(Func<IScheduler, int, IDisposable>)));
  22. ReactiveAssert.Throws<ArgumentNullException>(() => ThreadPoolScheduler.Instance.Schedule<int>(42, TimeSpan.Zero, default(Func<IScheduler, int, IDisposable>)));
  23. }
  24. [Fact]
  25. public void Get_Now()
  26. {
  27. var res = ThreadPoolScheduler.Instance.Now - DateTime.Now;
  28. Assert.True(res.Seconds < 1);
  29. }
  30. [Fact]
  31. public void ScheduleAction()
  32. {
  33. var id = Thread.CurrentThread.ManagedThreadId;
  34. var nt = ThreadPoolScheduler.Instance;
  35. var evt = new ManualResetEvent(false);
  36. nt.Schedule(() => { Assert.NotEqual(id, Thread.CurrentThread.ManagedThreadId); evt.Set(); });
  37. evt.WaitOne();
  38. }
  39. [Fact]
  40. public void ProperRooting_NoGC_SingleShot()
  41. {
  42. var cts = new CancellationTokenSource();
  43. new Thread(() =>
  44. {
  45. while (!cts.IsCancellationRequested)
  46. {
  47. Thread.Sleep(50);
  48. GC.Collect();
  49. GC.WaitForPendingFinalizers();
  50. }
  51. }).Start();
  52. var tp = ThreadPoolScheduler.Instance;
  53. var N = 100;
  54. var cd = new CountdownEvent(N);
  55. for (int i = 0; i < N; i++)
  56. {
  57. tp.Schedule(TimeSpan.FromMilliseconds(100 + i), () => { cd.Signal(); });
  58. }
  59. Assert.True(cd.Wait(TimeSpan.FromMinutes(1)));
  60. cts.Cancel();
  61. }
  62. [Fact]
  63. public void ProperRooting_NoGC_Periodic()
  64. {
  65. var cts = new CancellationTokenSource();
  66. new Thread(() =>
  67. {
  68. while (!cts.IsCancellationRequested)
  69. {
  70. Thread.Sleep(50);
  71. GC.Collect();
  72. GC.WaitForPendingFinalizers();
  73. }
  74. }).Start();
  75. var tp = ThreadPoolScheduler.Instance;
  76. var N = 5;
  77. var e = new ManualResetEvent(false);
  78. var n = 0;
  79. var d = tp.SchedulePeriodic(TimeSpan.FromMilliseconds(80), () => { if (Interlocked.Increment(ref n) == N) e.Set(); });
  80. Assert.True(e.WaitOne(TimeSpan.FromMinutes(1)));
  81. d.Dispose();
  82. cts.Cancel();
  83. }
  84. [Fact]
  85. public void ScheduleActionDueRelative()
  86. {
  87. var id = Thread.CurrentThread.ManagedThreadId;
  88. var nt = ThreadPoolScheduler.Instance;
  89. var evt = new ManualResetEvent(false);
  90. nt.Schedule(TimeSpan.FromSeconds(0.2), () => { Assert.NotEqual(id, Thread.CurrentThread.ManagedThreadId); evt.Set(); });
  91. evt.WaitOne();
  92. }
  93. [Fact]
  94. public void ScheduleActionDue0()
  95. {
  96. var id = Thread.CurrentThread.ManagedThreadId;
  97. var nt = ThreadPoolScheduler.Instance;
  98. var evt = new ManualResetEvent(false);
  99. nt.Schedule(TimeSpan.FromTicks(0), () => { Assert.NotEqual(id, Thread.CurrentThread.ManagedThreadId); evt.Set(); });
  100. evt.WaitOne();
  101. }
  102. [Fact]
  103. public void ScheduleActionDueAbsolute()
  104. {
  105. var id = Thread.CurrentThread.ManagedThreadId;
  106. var nt = ThreadPoolScheduler.Instance;
  107. var evt = new ManualResetEvent(false);
  108. nt.Schedule(DateTimeOffset.UtcNow + TimeSpan.FromSeconds(0.2), () => { Assert.NotEqual(id, Thread.CurrentThread.ManagedThreadId); evt.Set(); });
  109. evt.WaitOne();
  110. }
  111. [Fact]
  112. public void ScheduleActionCancel()
  113. {
  114. var id = Thread.CurrentThread.ManagedThreadId;
  115. var nt = ThreadPoolScheduler.Instance;
  116. var set = false;
  117. var d = nt.Schedule(TimeSpan.FromSeconds(0.2), () => { Assert.True(false); set = true; });
  118. d.Dispose();
  119. Thread.Sleep(400);
  120. Assert.False(set);
  121. }
  122. #if !NO_PERF
  123. [Fact]
  124. public void ScheduleLongRunning_ArgumentChecking()
  125. {
  126. ReactiveAssert.Throws<ArgumentNullException>(() => ThreadPoolScheduler.Instance.ScheduleLongRunning<int>(42, default(Action<int, ICancelable>)));
  127. }
  128. [Fact]
  129. public void ScheduleLongRunning()
  130. {
  131. var id = Thread.CurrentThread.ManagedThreadId;
  132. var nt = ThreadPoolScheduler.Instance;
  133. var evt = new ManualResetEvent(false);
  134. nt.ScheduleLongRunning(42, (x, cancel) => { Assert.NotEqual(id, Thread.CurrentThread.ManagedThreadId); evt.Set(); });
  135. evt.WaitOne();
  136. }
  137. [Fact]
  138. public void ScheduleLongRunningCancel()
  139. {
  140. var nt = ThreadPoolScheduler.Instance;
  141. var started = new ManualResetEvent(false);
  142. var stopped = new ManualResetEvent(false);
  143. var n = 0;
  144. var d = nt.ScheduleLongRunning(42, (x, cancel) =>
  145. {
  146. for (n = 0; !cancel.IsDisposed; n++)
  147. {
  148. if (n == 10)
  149. started.Set();
  150. }
  151. stopped.Set();
  152. });
  153. started.WaitOne();
  154. d.Dispose();
  155. stopped.WaitOne();
  156. Assert.True(n >= 10);
  157. }
  158. [Fact]
  159. public void Stopwatch()
  160. {
  161. var nt = ThreadPoolScheduler.Instance;
  162. var sw = nt.StartStopwatch();
  163. var s0 = sw.Elapsed.Ticks;
  164. Thread.Sleep(10);
  165. var s1 = sw.Elapsed.Ticks;
  166. Assert.True(s1 > s0);
  167. }
  168. [Fact]
  169. public void Periodic_ArgumentChecking()
  170. {
  171. ReactiveAssert.Throws<ArgumentNullException>(() => ThreadPoolScheduler.Instance.SchedulePeriodic(0, TimeSpan.FromSeconds(1), null));
  172. ReactiveAssert.Throws<ArgumentOutOfRangeException>(() => ThreadPoolScheduler.Instance.SchedulePeriodic(0, TimeSpan.FromSeconds(-1), _ => _));
  173. }
  174. [Fact]
  175. public void Periodic_Regular()
  176. {
  177. Periodic_Impl(TimeSpan.FromMilliseconds(25));
  178. }
  179. [Fact]
  180. public void Periodic_Zero()
  181. {
  182. Periodic_Impl(TimeSpan.Zero);
  183. }
  184. private void Periodic_Impl(TimeSpan period)
  185. {
  186. var gate = new object();
  187. var n = 0;
  188. var e = new ManualResetEvent(false);
  189. var lst = new List<int>();
  190. var d = ThreadPoolScheduler.Instance.SchedulePeriodic(0, period, x =>
  191. {
  192. lock (gate)
  193. {
  194. if (n++ == 10)
  195. e.Set();
  196. }
  197. lst.Add(x);
  198. return x + 1;
  199. });
  200. e.WaitOne();
  201. d.Dispose();
  202. var m = default(int);
  203. var k = default(int);
  204. var i = 0;
  205. do
  206. {
  207. lock (gate)
  208. m = n;
  209. Thread.Sleep(50);
  210. lock (gate)
  211. k = n;
  212. } while (m != k && i++ < 10); // Wait for Dispose to reach the timer; should be almost instantaneous due to nop'ing out of the action.
  213. Assert.NotEqual(10, i);
  214. var res = lst.ToArray();
  215. Assert.True(res.Length >= 10);
  216. Assert.True(res.Take(10).SequenceEqual(Enumerable.Range(0, 10)));
  217. }
  218. [Fact]
  219. public void Periodic_NonReentrant()
  220. {
  221. var n = 0;
  222. var fail = false;
  223. var d = ThreadPoolScheduler.Instance.SchedulePeriodic(0, TimeSpan.FromMilliseconds(50), x =>
  224. {
  225. try
  226. {
  227. if (Interlocked.Increment(ref n) > 1) // Without an AsyncLock this would fail.
  228. fail = true;
  229. Thread.Sleep(100);
  230. return x + 1;
  231. }
  232. finally
  233. {
  234. Interlocked.Decrement(ref n);
  235. }
  236. });
  237. Thread.Sleep(500);
  238. d.Dispose();
  239. Assert.False(fail);
  240. }
  241. #endif
  242. #if DESKTOPCLR
  243. [Fact]
  244. public void No_ThreadPool_Starvation_Dispose()
  245. {
  246. ThreadPool.GetAvailableThreads(out var bwt, out var bio);
  247. var N = Environment.ProcessorCount * 2;
  248. for (int i = 0; i < N; i++)
  249. {
  250. var e = new ManualResetEvent(false);
  251. var f = new ManualResetEvent(false);
  252. var d = ThreadPoolScheduler.Instance.Schedule(TimeSpan.FromMilliseconds(1), () => { e.Set(); f.WaitOne(); });
  253. e.WaitOne();
  254. d.Dispose();
  255. f.Set();
  256. }
  257. ThreadPool.GetAvailableThreads(out var ewt, out var eio);
  258. Assert.False(bwt - ewt >= N);
  259. }
  260. #endif
  261. }
  262. }
  263. #endif