hunt.event.selector.IOCP 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.selector.IOCP;
13 
14 // dfmt off
15 version (HAVE_IOCP) : 
16 // dfmt on
17 
18 import hunt.event.selector.Selector;
19 import hunt.io.channel.Common;
20 import hunt.io.channel;
21 import hunt.event.timer;
22 import hunt.logging.ConsoleLogger;
23 import hunt.system.Error;
24 import hunt.io.channel.iocp.AbstractStream;
25 import core.sys.windows.windows;
26 import std.conv;
27 import std.container : DList;
28 /**
29 */
30 class AbstractSelector : Selector {
31 
32     this(size_t number, size_t divider, size_t maxChannels = 1500) {
33         super(number, divider, maxChannels);
34         _iocpHandle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, null, 0, 0);
35         if (_iocpHandle is null)
36             errorf("CreateIoCompletionPort failed: %d\n", GetLastError());
37         _timer.init();
38     }
39 
40     ~this() {
41         // import std.socket;
42         // std.socket.close(_iocpHandle);
43     }
44 
45     override bool register(AbstractChannel channel) {
46         assert(channel !is null);
47         ChannelType ct = channel.type;
48         auto fd = channel.handle;
49         version (HUNT_IO_DEBUG)
50             tracef("register, channel(fd=%d, type=%s)", fd, ct);
51 
52         if (ct == ChannelType.Timer) {
53             AbstractTimer timerChannel = cast(AbstractTimer) channel;
54             assert(timerChannel !is null);
55             if (!timerChannel.setTimerOut())
56                 return false;
57             _timer.timeWheel().addNewTimer(timerChannel.timer, timerChannel.wheelSize());
58         } else if (ct == ChannelType.TCP
59                 || ct == ChannelType.Accept || ct == ChannelType.UDP) {
60             version (HUNT_IO_DEBUG)
61                 trace("Run CreateIoCompletionPort on socket: ", fd);
62 
63             // _event.setNext(channel);
64             CreateIoCompletionPort(cast(HANDLE) fd, _iocpHandle,
65                     cast(size_t)(cast(void*) channel), 0);
66 
67             //cast(AbstractStream)channel)
68         } else {
69             warningf("Can't register a channel: %s", ct);
70         }
71 
72         (cast(AbstractStream)channel).beginRead();
73         return true;
74     }
75 
76     override bool deregister(AbstractChannel channel) {
77         // FIXME: Needing refactor or cleanup -@Administrator at 8/28/2018, 3:28:18 PM
78         // https://stackoverflow.com/questions/6573218/removing-a-handle-from-a-i-o-completion-port-and-other-questions-about-iocp
79         tracef("deregister (fd=%d)", channel.handle);
80 
81         // IocpContext _data;
82         // _data.channel = channel;
83         // _data.operation = IocpOperation.close;
84         // PostQueuedCompletionStatus(_iocpHandle, 0, 0, &_data.overlapped);
85         //(cast(AbstractStream)channel).stopAction();
86         //WaitForSingleObject
87         return true;
88     }
89 
90     // void weakUp() {
91     //     IocpContext _data;
92     //     // _data.channel = _event;
93     //     _data.operation = IocpOperation.event;
94 
95     //     // PostQueuedCompletionStatus(_iocpHandle, 0, 0, &_data.overlapped);
96     //     PostQueuedCompletionStatus(_iocpHandle, 0, 0, null);
97     // }
98 
99     void putTast(AbstractStream stream)
100     {
101         _queue.insertBack(stream);
102     }
103 
104     override void onLoop(long timeout = -1) {
105         _timer.init();
106         super.onLoop(timeout);
107     }
108 
109     protected override int doSelect(long t) {
110         //if(!_queue.empty)
111         //{
112         //    auto task = _queue.front();
113         //    if (task !is null)
114         //    {
115         //        task.beginRead();
116         //    }
117         //    _queue.removeFront();
118         //}
119         //trace("ssssss");
120         auto timeout = _timer.doWheel();
121         OVERLAPPED* overlapped;
122         ULONG_PTR key = 0;
123         DWORD bytes = 0;
124 
125         // const int ret = GetQueuedCompletionStatus(_iocpHandle, &bytes,
126         //         &key, &overlapped, INFINITE);
127         // tracef("GetQueuedCompletionStatus, ret=%d", ret);
128 
129         // trace("timeout=", timeout);
130         const int ret = GetQueuedCompletionStatus(_iocpHandle, &bytes, &key,
131                 &overlapped, INFINITE);
132 
133         IocpContext* ev = cast(IocpContext*) overlapped;
134         if (ret == 0) {
135             if (overlapped is null)
136             {
137                 return ret;
138             }
139             if (key == 0)
140             {
141                 return ret;
142             }
143             const auto erro = GetLastError();
144             //// About ERROR_OPERATION_ABORTED
145             //// https://stackoverflow.com/questions/7228703/the-i-o-operation-has-been-aborted-because-of-either-a-thread-exit-or-an-applica
146             if (erro == WAIT_TIMEOUT || erro == ERROR_OPERATION_ABORTED) //
147                 return ret;
148 
149             debug warningf("error occurred, code=%d, message: %s", erro, getErrorMessage(erro));
150             assert(ev !is null);
151             AbstractChannel channel = ev.channel;
152             if (channel !is null && !channel.isClosed())
153             {
154                 channel.close();
155             }
156 
157         }
158         //else if (ev is null || ev.channel is null)
159         //    warning("ev is null or ev.watche is null");
160         else {
161             handleChannelEvent(ev.operation, ev.channel, bytes);
162         }
163 
164         return ret;
165     }
166 
167     private void handleChannelEvent(IocpOperation op, AbstractChannel channel, DWORD bytes) {
168 
169         version (HUNT_IO_DEBUG)
170             infof("ev.operation: %s, fd=%d", op, channel.handle);
171         try {
172             switch (op) {
173                 case IocpOperation.accept:
174                // channel.onRead();
175                 warningf("accept ............................");
176                 break;
177                 case IocpOperation.connect:
178                 onSocketRead(channel, 0);
179                 warningf("connect ............................");
180                 break;
181                 case IocpOperation.read:
182                 onSocketRead(channel, bytes);
183                 warningf("read ............................");
184                 break;
185                 case IocpOperation.write:
186                 onSocketWrite(channel, bytes);
187                 warningf("write ............................");
188                 break;
189                 case IocpOperation.event:
190                 warningf("event ............................");
191                 channel.onRead();
192                 break;
193                 case IocpOperation.close:
194                 warningf("close -------------------------: %d", channel.handle);
195                 break;
196                 default:
197                 warning("unsupported operation type -------------------------: ", op);
198                 break;
199             }
200         } catch (Exception e)
201         {
202 
203         }
204 
205     }
206 
207     override void stop() {
208         super.stop();
209         // weakUp();
210         PostQueuedCompletionStatus(_iocpHandle, 0, 0, null);
211     }
212 
213     void handleTimer() {
214 
215     }
216 
217     // override void dispose() {
218 
219     // }
220 
221     private void onSocketRead(AbstractChannel channel, size_t len) {
222         debug if (channel is null) {
223             warning("channel is null");
224             return;
225         }
226 
227         if (channel is null)
228         {
229             return;
230         }
231 
232         (cast(AbstractStream)channel).setBusyWrite(false);
233 
234         if (len == 0 || channel.isClosed) {
235             //version (HUNT_DEBUG)
236             //    infof("channel [fd=%d] closed %d  %d", channel.handle, channel.isClosed,len);
237             //(cast(AbstractStream)channel).setFristRead(false);
238             //(cast(AbstractStream)channel).setBusyWrite(true);
239             //channel.close();
240             return;
241         }
242 
243         AbstractSocketChannel socketChannel = cast(AbstractSocketChannel) channel;
244         // assert(socketChannel !is null, "The type of channel is: " ~ typeid(channel).name);
245         if (socketChannel is null) {
246             warning("The channel socket is null: ");
247         } else {
248             socketChannel.setRead(len);
249             if (!(cast(AbstractStream)channel).getFristRead)
250             {
251                 (cast(AbstractStream)channel).setFristRead(true);
252             }
253             channel.onRead();
254         }
255     }
256 
257     private void onSocketWrite(AbstractChannel channel, size_t len) {
258         debug if (channel is null) {
259             warning("channel is null");
260             return;
261         }
262         AbstractStream client = cast(AbstractStream) channel;
263         // assert(client !is null, "The type of channel is: " ~ typeid(channel).name);
264         if (client is null) {
265             warning("The channel socket is null: ");
266             return;
267         }
268         warning("len ------------------------ %d",len);
269         client.onWriteDone(len); // Notify the client about how many bytes actually sent.
270     }
271 
272 private:
273     HANDLE _iocpHandle;
274     CustomTimer _timer;
275     DList!AbstractStream _queue;
276 }