hunt.event.EventLoopGroup source code

1 /*
2  * Hunt - A refined core library for D programming language.
3  *
4  * Copyright (C) 2018-2019 HuntLabs
5  *
6  * Website: https://www.huntlabs.net/
7  *
8  * Licensed under the Apache-2.0 License.
9  *
10  */
11 
12 module hunt.event.EventLoopGroup;
13 
14 import hunt.system.Memory;
15 import hunt.event.EventLoop;
16 import hunt.logging;
17 import hunt.util.Lifecycle;
18 
19 import core.atomic;
20 
21 /**
22 */
23 class EventLoopGroup : Lifecycle {
24 
25     this(size_t size = (totalCPUs - 1)) {
26         size_t _size = size > 0 ? size : 1;
27         eventLoopPool = new EventLoop[_size];
28 
29         foreach (i; 0 .. _size) {
30             eventLoopPool[i] = new EventLoop(i, _size);
31         }
32     }
33 
34     void start() {
35         start(-1);
36     }
37 
38     /**
39         timeout: in millisecond
40     */
41     void start(long timeout) {
42         if (cas(&_isRunning, false, true)) {
43             foreach (EventLoop pool; eventLoopPool) {
44                 pool.runAsync(timeout);
45             }
46         }
47     }
48 
49     void stop() {
50         if (!cas(&_isRunning, true, false))
51             return;
52 
53         version (HUNT_IO_DEBUG)
54             trace("stopping EventLoopGroup...");
55         foreach (EventLoop pool; eventLoopPool) {
56             pool.stop();
57         }
58 
59         version (HUNT_IO_DEBUG)
60             trace("EventLoopGroup stopped.");
61     }
62 
63 	bool isRunning() {
64         return _isRunning;
65     }
66 
67     bool isReady() {
68         
69         foreach (EventLoop pool; eventLoopPool) {
70             if(!pool.isReady()) return false;
71         }
72 
73         return true;
74     }
75 
76     @property size_t size() {
77         return eventLoopPool.length;
78     }
79 
80     EventLoop nextLoop(size_t factor) {
81        return eventLoopPool[factor % eventLoopPool.length];
82     }
83 
84     EventLoop nextLoop() {
85         size_t index = atomicOp!"+="(_loopIndex, 1);
86         if(index > 10000) {
87             index = 0;
88             atomicStore(_loopIndex, 0);
89         }
90         index %= eventLoopPool.length;
91         return eventLoopPool[index];
92     }
93 
94     EventLoop opIndex(size_t index) {
95         auto i = index % eventLoopPool.length;
96         return eventLoopPool[i];
97     }
98 
99     int opApply(scope int delegate(EventLoop) dg) {
100         int ret = 0;
101         foreach (pool; eventLoopPool) {
102             ret = dg(pool);
103             if (ret)
104                 break;
105         }
106         return ret;
107     }
108 
109 private:
110     shared int _loopIndex;
111     shared bool _isRunning;
112     EventLoop[] eventLoopPool;
113 }