hunt.event.selector.Selector source code

1 module hunt.event.selector.Selector;
2 
3 import hunt.Exceptions;
4 import hunt.Functions;
5 import hunt.io.channel.AbstractChannel;
6 import hunt.io.channel.Common;
7 import hunt.logging.ConsoleLogger;
8 
9 import core.atomic;
10 import core.thread;
11 
12 
13 /**
14 http://tutorials.jenkov.com/java-nio/selectors.html
15 */
16 abstract class Selector {
17 
18     private shared bool _running = false;
19     private shared bool _isStopping = false;
20     private bool _isReady;
21     protected size_t number;
22     protected size_t divider;
23     protected AbstractChannel[] channels;
24     protected long idleTime = -1; // in millisecond
25     protected int fd;
26 
27     private long timeout = -1; // in millisecond
28     private Thread _thread;
29 
30     private SimpleEventHandler _startedHandler;
31     private SimpleEventHandler _stoppeddHandler;
32 
33     this(size_t number, size_t divider, size_t maxChannels = 1500) {
34         this.number = number;
35         this.divider = divider;
36         channels = new AbstractChannel[maxChannels];
37     }
38 
39     size_t getId() {
40         return number;
41     }
42 
43     bool isReady() {
44         return _isReady;
45     }
46 
47 
48     /**
49      * Tells whether or not this selector is running.
50      *
51      * @return <tt>true</tt> if, and only if, this selector is running
52      */
53     bool isRuning() {
54         return _running;
55     }
56 
57     alias isOpen = isRuning;
58 
59     bool isStopping() {
60         return _isStopping;
61     }
62 
63 
64     abstract bool register(AbstractChannel channel);
65 
66     abstract bool deregister(AbstractChannel channel);
67 
68     bool update(AbstractChannel channel) { return true; }
69 
70     protected abstract int doSelect(long timeout);
71 
72     /**
73         timeout: in millisecond
74     */
75     void run(long timeout = -1) {
76         this.timeout = timeout;
77         doRun();
78     }
79 
80     /**
81         timeout: in millisecond
82     */
83     void runAsync(long timeout = -1, SimpleEventHandler handler = null) {
84         if(_running) {
85             version (HUNT_IO_DEBUG) warningf("The current selector %d has being running already!", number);
86             return;
87         }
88         this.timeout = timeout;
89         version (HUNT_IO_DEBUG) trace("runAsync ...");
90         Thread th = new Thread(() { 
91             try {
92                 doRun(handler); 
93             } catch (Throwable t) {
94                 warning(t.msg);
95                 version(HUNT_DEBUG) warning(t.toString());
96             }
97         });
98         // th.isDaemon = true; // unstable
99         th.start();
100     }
101     
102     private void doRun(SimpleEventHandler handler=null) {
103         if(cas(&_running, false, true)) {
104             version (HUNT_IO_DEBUG) trace("running selector...");
105             _thread = Thread.getThis();
106             if(handler !is null) {
107                 handler();
108             }
109             onLoop(timeout);
110         } else {
111             version (HUNT_DEBUG) warningf("The current selector %d has being running already!", number);
112         }  
113     }
114 
115     void stop() {
116         version (HUNT_IO_DEBUG)
117             tracef("Stopping selector %d. _running=%s, _isStopping=%s", number, _running, _isStopping); 
118         if(cas(&_isStopping, false, true)) {
119             try {
120                 onStop();
121             } catch(Throwable t) {
122                 warning(t.msg);
123                 version(HUNT_DEBUG) warning(t);
124             }
125         }
126     }
127 
128     protected void onStop() {
129         version (HUNT_IO_DEBUG) 
130             tracef("stopping.");
131     }
132 
133     /**
134         timeout: in millisecond
135     */
136     protected void onLoop(long timeout = -1) {
137         _isReady = true;
138         idleTime = timeout;
139 
140         do {
141             // version(HUNT_THREAD_DEBUG) warningf("Threads: %d", Thread.getAll().length);
142             doSelect(timeout);
143             // infof("Selector rolled once. isRuning: %s", isRuning);
144         } while (!_isStopping);
145 
146         _isReady = false;
147         _running = false;
148         version(HUNT_DEBUG_MORE) infof("Selector %d exited.", number);
149         dispose();
150     }
151 
152     /**
153         timeout: in millisecond
154     */
155     int select(long timeout) {
156         if (timeout < 0)
157             throw new IllegalArgumentException("Negative timeout");
158         return doSelect((timeout == 0) ? -1 : timeout);
159     }
160 
161     int select() {
162         return doSelect(0);
163     }
164 
165     int selectNow() {
166         return doSelect(0);
167     }
168 
169     void dispose() {
170         _thread = null;
171         _startedHandler = null;
172         _stoppeddHandler = null;
173     }
174     
175     bool isSelfThread() {
176         return _thread is Thread.getThis();
177     }
178 }