hunt.event.selector.Epoll 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.Epoll;
13 
14 // dfmt off
15 version(HAVE_EPOLL):
16 
17 // dfmt on
18 
19 import std.exception;
20 import std.socket;
21 import std.string;
22 
23 import core.time;
24 import core.stdc.string;
25 import core.stdc.errno;
26 import core.sys.posix.sys.types;
27 import core.sys.posix.netinet.tcp;
28 import core.sys.posix.netinet.in_;
29 import core.sys.posix.unistd;
30 
31 import core.sys.posix.sys.resource;
32 import core.sys.posix.sys.time;
33 import core.sys.linux.epoll;
34 
35 import hunt.event.selector.Selector;
36 import hunt.Exceptions;
37 import hunt.io.channel;
38 import hunt.logging.ConsoleLogger;
39 import hunt.event.timer;
40 import hunt.system.Error;
41 import hunt.concurrency.TaskPool;
42 
43 /* Max. theoretical number of file descriptors on system. */
44 __gshared size_t fdLimit = 0;
45 
46 shared static this() {
47     rlimit fileLimit;
48     getrlimit(RLIMIT_NOFILE, &fileLimit);
49     fdLimit = fileLimit.rlim_max;
50 }
51 
52 
53 /**
54 */
55 class AbstractSelector : Selector {
56     enum int NUM_KEVENTS = 1024;
57     private int _epollFD;
58     private bool isDisposed = false;
59     private epoll_event[NUM_KEVENTS] events;
60     private EventChannel _eventChannel;
61 
62     this(size_t number, size_t divider, size_t maxChannels = 1500) {
63         super(number, divider, maxChannels);
64 
65         // http://man7.org/linux/man-pages/man2/epoll_create.2.html
66         /*
67          * Set the close-on-exec (FD_CLOEXEC) flag on the new file descriptor.
68          * See the description of the O_CLOEXEC flag in open(2) for reasons why
69          * this may be useful.
70          */
71         _epollFD = epoll_create1(EPOLL_CLOEXEC);
72         if (_epollFD < 0)
73             throw new IOException("epoll_create failed");
74 
75         _eventChannel = new EpollEventChannel(this);
76         register(_eventChannel);
77     }
78 
79     ~this() @nogc {
80         // dispose();
81     }
82 
83     override void dispose() {
84         if (isDisposed)
85             return;
86 
87         version (HUNT_IO_DEBUG)
88             tracef("disposing selector[fd=%d]...", _epollFD);
89         isDisposed = true;
90         _eventChannel.close();
91         int r = core.sys.posix.unistd.close(_epollFD);
92         if(r != 0) {
93             version (HUNT_IO_DEBUG) warningf("error: %d", r);
94         }
95 
96         super.dispose();
97     }
98 
99     override void onStop() {
100         version (HUNT_IO_DEBUG)
101             infof("Selector stopping. fd=%d, id: %d", _epollFD, number);
102 
103         if(!_eventChannel.isClosed()) {
104             _eventChannel.trigger();
105             // _eventChannel.onWrite();
106         }
107     }
108 
109     override bool register(AbstractChannel channel) {
110         assert(channel !is null);
111         int infd = cast(int) channel.handle;
112         version (HUNT_IO_DEBUG)
113             tracef("register channel: fd=%d", infd);
114 
115         size_t index = cast(size_t)(infd / divider);
116         if (index >= channels.length) {
117             debug warningf("expanding channels uplimit to %d", index);
118             import std.algorithm : max;
119 
120             size_t length = max(cast(size_t)(index * 3 / 2), 16);
121             AbstractChannel[] arr = new AbstractChannel[length];
122             arr[0 .. channels.length] = channels[0 .. $];
123             channels = arr;
124         }
125         channels[index] = channel;
126 
127         // epoll_event e;
128 
129         // e.data.fd = infd;
130         // e.data.ptr = cast(void*) channel;
131         // e.events = EPOLLIN | EPOLLET | EPOLLERR | EPOLLHUP | EPOLLRDHUP | EPOLLOUT;
132         // int s = epoll_ctl(_epollFD, EPOLL_CTL_ADD, infd, &e);
133         // if (s == -1) {
134         //     debug warningf("failed to register channel: fd=%d", infd);
135         //     return false;
136         // } else {
137         //     return true;
138         // }
139         if (epollCtl(channel, EPOLL_CTL_ADD)) {
140             return true;
141         } else {
142             debug warningf("failed to register channel: fd=%d", infd);
143             return false;
144         }
145     }
146 
147     override bool deregister(AbstractChannel channel) {
148         size_t fd = cast(size_t) channel.handle;
149         size_t index = cast(size_t)(fd / divider);
150         version (HUNT_IO_DEBUG)
151             tracef("deregister channel: fd=%d, index=%d", fd, index);
152         channels[index] = null;
153 
154         if (epollCtl(channel, EPOLL_CTL_DEL)) {
155             return true;
156         } else {
157             warningf("deregister channel failed: fd=%d", fd);
158             return false;
159         }
160     }
161 
162     override bool update(AbstractChannel channel) {
163         if (epollCtl(channel, EPOLL_CTL_MOD)) {
164             return true;
165         } else {
166             warningf("update channel failed: fd=%d", fd);
167             return false;
168         }
169     }
170 
171     /**
172         timeout: in millisecond
173     */
174     protected override int doSelect(long timeout) {
175         int len = 0;
176 
177         if (timeout <= 0) { /* Indefinite or no wait */
178             do {
179                 // http://man7.org/linux/man-pages/man2/epoll_wait.2.html
180                 // https://stackoverflow.com/questions/6870158/epoll-wait-fails-due-to-eintr-how-to-remedy-this/6870391#6870391
181                 len = epoll_wait(_epollFD, events.ptr, events.length, cast(int) timeout);
182             } while ((len == -1) && (errno == EINTR));
183         } else { /* Bounded wait; bounded restarts */
184             len = iepoll(_epollFD, events.ptr, events.length, cast(int) timeout);
185         }
186 
187         if (len > 0) {
188             if(defaultPoolThreads > 0) {  // using worker thread
189                 foreach (i; 0 .. len) {
190                     AbstractChannel channel = cast(AbstractChannel)(events[i].data.ptr);
191                     if (channel is null) {
192                         debug warningf("channel is null");
193                     } else {
194                         uint currentEvents = events[i].events;
195                         workerPool.put(cast(int)channel.handle, makeTask(&handeChannelEvent, channel, currentEvents));
196                     }
197                 }
198             } else {
199                 foreach (i; 0 .. len) {
200                     AbstractChannel channel = cast(AbstractChannel)(events[i].data.ptr);
201                     if (channel is null) {
202                         debug warningf("channel is null");
203                     } else {
204                         handeChannelEvent(channel, events[i].events);
205                     }
206                 }
207             }
208         }
209 
210         return len;
211     }
212 
213     private void handeChannelEvent(AbstractChannel channel, uint event) {
214         version (HUNT_IO_DEBUG)
215         infof("handling event: selector=%d, events=%d, channel=%d", this._epollFD, event, channel.handle);
216 
217         try {
218             if (isClosed(event)) { // && errno != EINTR
219                 /* An error has occured on this fd, or the socket is not
220                     ready for reading (why were we notified then?) */
221                 version (HUNT_IO_DEBUG) {
222                     if (isError(event)) {
223                         warningf("channel error: fd=%s, event=%d, errno=%d, message=%s",
224                                 channel.handle, event, errno, getErrorMessage(errno));
225                     } else {
226                         tracef("channel closed: fd=%d, errno=%d, message=%s",
227                                     channel.handle, errno, getErrorMessage(errno));
228                     }
229                 }
230                 // FIXME: Needing refactor or cleanup -@zxp at 2/28/2019, 3:25:24 PM
231                 // May be triggered twice for a channel, for example:
232                 // events=8197, fd=13
233                 // events=8221, fd=13
234                 // The remote connection broken abnormally, so the channel should be notified.
235                 // channel.onClose(); // More tests are needed
236                 channel.close();
237             } else if (event == EPOLLIN) {
238                 version (HUNT_IO_DEBUG)
239                     tracef("channel read event: fd=%d", channel.handle);
240                 channel.onRead();
241             } else if (event == EPOLLOUT) {
242                 version (HUNT_IO_DEBUG)
243                     tracef("channel write event: fd=%d", channel.handle);
244                 channel.onWrite();
245             } else if (event == (EPOLLIN | EPOLLOUT)) {
246                 version (HUNT_IO_DEBUG)
247                     tracef("channel read and write: fd=%d", channel.handle);
248                 channel.onWrite();
249                 channel.onRead();
250             } else {
251                 debug warningf("this thread only for read/write/close events, event: %d", event);
252             }
253         } catch (Exception e) {
254             debug {
255                 errorf("error while handing channel: fd=%s, exception=%s, message=%s",
256                         channel.handle, typeid(e), e.msg);
257             }
258             version(HUNT_DEBUG) warning(e);
259         }
260     }
261 
262     private int iepoll(int epfd, epoll_event* events, int numfds, int timeout) {
263         long start, now;
264         int remaining = timeout;
265         timeval t;
266         long diff;
267 
268         gettimeofday(&t, null);
269         start = t.tv_sec * 1000 + t.tv_usec / 1000;
270 
271         for (;;) {
272             int res = epoll_wait(epfd, events, numfds, remaining);
273             if (res < 0 && errno == EINTR) {
274                 if (remaining >= 0) {
275                     gettimeofday(&t, null);
276                     now = t.tv_sec * 1000 + t.tv_usec / 1000;
277                     diff = now - start;
278                     remaining -= diff;
279                     if (diff < 0 || remaining <= 0) {
280                         return 0;
281                     }
282                     start = now;
283                 }
284             } else {
285                 return res;
286             }
287         }
288     }
289 
290     // https://blog.csdn.net/ljx0305/article/details/4065058
291     private static bool isError(uint events) nothrow {
292         return (events & EPOLLERR) != 0;
293     }
294 
295     private static bool isClosed(uint e) nothrow {
296         return (e & EPOLLERR) != 0 || (e & EPOLLHUP) != 0 || (e & EPOLLRDHUP) != 0
297                 || (!(e & EPOLLIN) && !(e & EPOLLOUT)) != 0;
298     }
299 
300     private static bool isReadable(uint events) nothrow {
301         return (events & EPOLLIN) != 0;
302     }
303 
304     private static bool isWritable(uint events) nothrow {
305         return (events & EPOLLOUT) != 0;
306     }
307 
308     private static buildEpollEvent(AbstractChannel channel, ref epoll_event ev) {
309         ev.data.ptr = cast(void*) channel;
310         // ev.data.fd = channel.handle;
311         ev.events = EPOLLRDHUP | EPOLLERR | EPOLLHUP;
312         if (channel.hasFlag(ChannelFlag.Read))
313             ev.events |= EPOLLIN;
314         if (channel.hasFlag(ChannelFlag.Write))
315             ev.events |= EPOLLOUT;
316         // if (channel.hasFlag(ChannelFlag.OneShot))
317         //     ev.events |= EPOLLONESHOT;
318         if (channel.hasFlag(ChannelFlag.ETMode))
319             ev.events |= EPOLLET;
320         return ev;
321     }
322 
323     private bool epollCtl(AbstractChannel channel, int opcode) {
324         assert(channel !is null);
325         const fd = channel.handle;
326         assert(fd >= 0, "The channel.handle is not initialized!");
327 
328         epoll_event ev;
329         buildEpollEvent(channel, ev);
330         int res = 0;
331 
332         do {
333             res = epoll_ctl(_epollFD, opcode, fd, &ev);
334         }
335         while ((res == -1) && (errno == EINTR));
336 
337         /*
338          * A channel may be registered with several Selectors. When each Selector
339          * is polled a EPOLL_CTL_DEL op will be inserted into its pending update
340          * list to remove the file descriptor from epoll. The "last" Selector will
341          * close the file descriptor which automatically unregisters it from each
342          * epoll descriptor. To avoid costly synchronization between Selectors we
343          * allow pending updates to be processed, ignoring errors. The errors are
344          * harmless as the last update for the file descriptor is guaranteed to
345          * be EPOLL_CTL_DEL.
346          */
347         if (res < 0 && errno != EBADF && errno != ENOENT && errno != EPERM) {
348             warning("epoll_ctl failed");
349             return false;
350         } else
351             return true;
352     }
353 }