ut.concurrency.scheduler source code

1 module ut.concurrency.scheduler;
2 
3 import concurrency.operations;
4 import concurrency.sender : DelaySender;
5 import concurrency;
6 import unit_threaded;
7 import concurrency.stoptoken;
8 import core.time : msecs;
9 import concurrency.scheduler;
10 
11 @("scheduleAfter")
12 @safe unittest {
13   DelaySender(10.msecs).syncWait;
14 }
15 
16 @("scheduleAfter.cancel")
17 @safe unittest {
18   race(DelaySender(10.msecs), DelaySender(3.msecs)).syncWait;
19 }
20 
21 @("ManualTimeWorker")
22 @safe unittest {
23   import core.atomic : atomicOp;
24 
25   shared int g, h;
26   auto worker = new shared ManualTimeWorker();
27   worker.addTimer((TimerTrigger trigger) shared { g.atomicOp!"+="(1); }, 10.msecs);
28   worker.addTimer((TimerTrigger trigger) shared { h.atomicOp!"+="(1); }, 5.msecs);
29 
30   worker.timeUntilNextEvent().should == 5.msecs;
31   g.should == 0;
32   h.should == 0;
33 
34   worker.advance(4.msecs);
35   worker.timeUntilNextEvent().should == 1.msecs;
36   h.should == 0;
37   g.should == 0;
38 
39   worker.advance(1.msecs);
40   worker.timeUntilNextEvent().should == 5.msecs;
41   h.should == 1;
42   g.should == 0;
43 
44   worker.advance(5.msecs);
45   h.should == 1;
46   g.should == 1;
47   worker.timeUntilNextEvent().should == null;
48 }
49 
50 @("ManualTimeWorker.cancel")
51 @safe unittest {
52   import core.atomic : atomicOp;
53 
54   shared int g;
55   auto worker = new shared ManualTimeWorker();
56   auto timer = worker.addTimer((TimerTrigger trigger) shared { g.atomicOp!"+="(1 + (trigger == TimerTrigger.cancel)); }, 10.msecs);
57   worker.timeUntilNextEvent().should == 10.msecs;
58   g.should == 0;
59 
60   worker.advance(4.msecs);
61   worker.timeUntilNextEvent().should == 6.msecs;
62   g.should == 0;
63 
64   worker.cancelTimer(timer);
65   worker.timeUntilNextEvent().should == null;
66   g.should == 2;
67 }
68 
69 @("ManualTimeWorker.error")
70 @safe unittest {
71   import core.time;
72   import concurrency.operations : withScheduler, whenAll;
73   import concurrency.sender : justFrom;
74 
75   shared int p = 0;
76   import concurrency.scheduler : ManualTimeWorker;
77 
78   auto worker = new shared ManualTimeWorker();
79 
80   auto sender = DelaySender(10.msecs)
81     .withScheduler(worker.getScheduler);
82 
83   auto driver = justFrom(() shared {
84       worker.advance(7.msecs);
85       throw new Exception("halt");
86     });
87 
88   whenAll(sender, driver).syncWait.assumeOk.shouldThrowWithMessage("halt");
89 }