hunt.event.selector.Kqueue 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.Kqueue;
13 
14 
15 // dfmt off
16 version(HAVE_KQUEUE):
17 // dfmt on
18 import hunt.event.selector.Selector;
19 import hunt.event.timer.Kqueue;
20 import hunt.Exceptions;
21 import hunt.io.channel;
22 import hunt.logging.ConsoleLogger;
23 import hunt.util.Common;
24 
25 import std.exception;
26 import std.socket;
27 import std.string;
28 
29 import core.time;
30 import core.stdc.string;
31 import core.stdc.errno;
32 import core.sys.posix.sys.types; // for ssize_t, size_t
33 import core.sys.posix.signal;
34 import core.sys.posix.netinet.tcp;
35 import core.sys.posix.netinet.in_;
36 import core.sys.posix.unistd;
37 import core.sys.posix.time;
38 
39 /**
40  * 
41  */
42 class AbstractSelector : Selector {
43     // kevent array size
44     enum int NUM_KEVENTS = 128;
45     private bool isDisposed = false;
46     private Kevent[NUM_KEVENTS] events;
47     private int _kqueueFD;
48     private EventChannel _eventChannel;
49 
50     this(size_t number, size_t divider, size_t maxChannels = 1500) {
51         super(number, divider, maxChannels);
52         _kqueueFD = kqueue();
53         _eventChannel = new KqueueEventChannel(this);
54         register(_eventChannel);
55     }
56 
57     ~this() @nogc {
58         // dispose();
59     }
60 
61     override void dispose() {
62         if (isDisposed)
63             return;
64 
65         version (HUNT_IO_DEBUG)
66             tracef("disposing selector[fd=%d]...", _kqueueFD);
67         isDisposed = true;
68         _eventChannel.close();
69         int r = core.sys.posix.unistd.close(_kqueueFD);
70         if(r != 0) {
71             version(HUNT_DEBUG) warningf("error: %d", r);
72         }
73 
74         super.dispose();
75     }
76 
77     override void onStop() {
78         version (HUNT_IO_DEBUG)
79             infof("Selector stopping. fd=%d", _kqueueFD);  
80                
81         if(!_eventChannel.isClosed()) {
82             _eventChannel.trigger();
83             // _eventChannel.onWrite();
84         }
85     }
86 
87     override bool register(AbstractChannel channel) {
88         assert(channel !is null);
89         
90         const int fd = channel.handle;
91         version (HUNT_IO_DEBUG)
92             tracef("register channel: fd=%d, type=%s", fd, channel.type);
93 
94         int err = -1;
95         if (channel.type == ChannelType.Timer) {
96             Kevent ev;
97             AbstractTimer timerChannel = cast(AbstractTimer) channel;
98             if (timerChannel is null)
99                 return false;
100             size_t time = timerChannel.time < 20 ? 20 : timerChannel.time; // in millisecond
101             EV_SET(&ev, timerChannel.handle, EVFILT_TIMER,
102                     EV_ADD | EV_ENABLE | EV_CLEAR, 0, time, cast(void*) channel);
103             err = kevent(_kqueueFD, &ev, 1, null, 0, null);
104         }
105         else {
106             if (fd < 0)
107                 return false;
108             Kevent[2] ev = void;
109             short read = EV_ADD | EV_ENABLE;
110             short write = EV_ADD | EV_ENABLE;
111             if (channel.hasFlag(ChannelFlag.ETMode)) {
112                 read |= EV_CLEAR;
113                 write |= EV_CLEAR;
114             }
115             EV_SET(&(ev[0]), fd, EVFILT_READ, read, 0, 0, cast(void*) channel);
116             EV_SET(&(ev[1]), fd, EVFILT_WRITE, write, 0, 0, cast(void*) channel);
117             if (channel.hasFlag(ChannelFlag.Read) && channel.hasFlag(ChannelFlag.Write))
118                 err = kevent(_kqueueFD, &(ev[0]), 2, null, 0, null);
119             else if (channel.hasFlag(ChannelFlag.Read))
120                 err = kevent(_kqueueFD, &(ev[0]), 1, null, 0, null);
121             else if (channel.hasFlag(ChannelFlag.Write))
122                 err = kevent(_kqueueFD, &(ev[1]), 1, null, 0, null);
123         }
124         if (err < 0) {
125             return false;
126         }
127         return true;
128     }
129 
130     override bool deregister(AbstractChannel channel) {
131         assert(channel !is null);
132         const fd = channel.handle;
133         if (fd < 0)
134             return false;
135 
136         int err = -1;
137         if (channel.type == ChannelType.Timer) {
138             Kevent ev;
139             AbstractTimer timerChannel = cast(AbstractTimer) channel;
140             if (timerChannel is null)
141                 return false;
142             EV_SET(&ev, fd, EVFILT_TIMER, EV_DELETE, 0, 0, cast(void*) channel);
143             err = kevent(_kqueueFD, &ev, 1, null, 0, null);
144         }
145         else {
146             Kevent[2] ev = void;
147             EV_SET(&(ev[0]), fd, EVFILT_READ, EV_DELETE, 0, 0, cast(void*) channel);
148             EV_SET(&(ev[1]), fd, EVFILT_WRITE, EV_DELETE, 0, 0, cast(void*) channel);
149             if (channel.hasFlag(ChannelFlag.Read) && channel.hasFlag(ChannelFlag.Write))
150                 err = kevent(_kqueueFD, &(ev[0]), 2, null, 0, null);
151             else if (channel.hasFlag(ChannelFlag.Read))
152                 err = kevent(_kqueueFD, &(ev[0]), 1, null, 0, null);
153             else if (channel.hasFlag(ChannelFlag.Write))
154                 err = kevent(_kqueueFD, &(ev[1]), 1, null, 0, null);
155         }
156         if (err < 0) {
157             return false;
158         }
159         // channel.currtLoop = null;
160         channel.clear();
161         return true;
162     }
163 
164     protected override int doSelect(long timeout) {
165         timespec ts;
166         timespec *tsp;
167         // timeout is in milliseconds. Convert to struct timespec.
168         // timeout == -1 : wait forever : timespec timeout of NULL
169         // timeout == 0  : return immediately : timespec timeout of zero
170         if (timeout >= 0) {
171             // For some indeterminate reason kevent(2) has been found to fail with
172             // an EINVAL error for timeout values greater than or equal to
173             // 100000001000L. To avoid this problem, clamp the timeout arbitrarily
174             // to the maximum value of a 32-bit signed integer which is
175             // approximately 25 days in milliseconds.
176             const int timeoutMax = int.max;
177             if (timeout > timeoutMax) {
178                 timeout = timeoutMax;
179             }
180             ts.tv_sec = timeout / 1000;
181             ts.tv_nsec = (timeout % 1000) * 1000000; //nanosec = 1 million millisec
182             tsp = &ts;
183         } else {
184             tsp = null;
185         }        
186 
187         // auto tspec = timespec(1, 1000 * 10);
188         int result = kevent(_kqueueFD, null, 0, events.ptr, events.length, tsp);
189 
190         foreach (i; 0 .. result) {
191             AbstractChannel channel = cast(AbstractChannel)(events[i].udata);
192             ushort eventFlags = events[i].flags;
193             version (HUNT_IO_DEBUG)
194             infof("handling event: events=%d, fd=%d", eventFlags, channel.handle);
195 
196             if (eventFlags & EV_ERROR) {
197                 warningf("channel[fd=%d] has a error.", channel.handle);
198                 channel.close();
199                 continue;
200             }
201             if (eventFlags & EV_EOF) {
202                 version (HUNT_IO_DEBUG) infof("channel[fd=%d] closed", channel.handle);
203                 channel.close();
204                 continue;
205             }
206 
207             short filter = events[i].filter;
208             handeChannelEvent(channel, filter);
209         }
210         return result;
211     }
212 
213     private void handeChannelEvent(AbstractChannel channel, uint filter) {
214         version (HUNT_IO_DEBUG)
215         infof("handling event: events=%d, fd=%d", filter, channel.handle);
216         try {
217             if(filter == EVFILT_TIMER) {
218                 channel.onRead();
219             } else if (filter == EVFILT_WRITE) {
220                 channel.onWrite();
221             } else if (filter == EVFILT_READ) {
222                 channel.onRead();
223             } else {
224                 warningf("Unhandled channel fileter: %d", filter);
225             }
226         } catch(Exception e) {
227             errorf("error while handing channel: fd=%s, message=%s",
228                         channel.handle, e.msg);
229         }
230     }
231 }
232 
233 
234 enum : short {
235     EVFILT_READ = -1,
236     EVFILT_WRITE = -2,
237     EVFILT_AIO = -3, /* attached to aio requests */
238     EVFILT_VNODE = -4, /* attached to vnodes */
239     EVFILT_PROC = -5, /* attached to struct proc */
240     EVFILT_SIGNAL = -6, /* attached to struct proc */
241     EVFILT_TIMER = -7, /* timers */
242     EVFILT_MACHPORT = -8, /* Mach portsets */
243     EVFILT_FS = -9, /* filesystem events */
244     EVFILT_USER = -10, /* User events */
245     EVFILT_VM = -12, /* virtual memory events */
246     EVFILT_SYSCOUNT = 11
247 }
248 
249 extern (D) void EV_SET(Kevent* kevp, typeof(Kevent.tupleof) args) @nogc nothrow {
250     *kevp = Kevent(args);
251 }
252 
253 struct Kevent {
254     uintptr_t ident; /* identifier for this event */
255     short filter; /* filter for event */
256     ushort flags;
257     uint fflags;
258     intptr_t data;
259     void* udata; /* opaque user data identifier */
260 }
261 
262 enum {
263     /* actions */
264     EV_ADD = 0x0001, /* add event to kq (implies enable) */
265     EV_DELETE = 0x0002, /* delete event from kq */
266     EV_ENABLE = 0x0004, /* enable event */
267     EV_DISABLE = 0x0008, /* disable event (not reported) */
268 
269     /* flags */
270     EV_ONESHOT = 0x0010, /* only report one occurrence */
271     EV_CLEAR = 0x0020, /* clear event state after reporting */
272     EV_RECEIPT = 0x0040, /* force EV_ERROR on success, data=0 */
273     EV_DISPATCH = 0x0080, /* disable event after reporting */
274 
275     EV_SYSFLAGS = 0xF000, /* reserved by system */
276     EV_FLAG1 = 0x2000, /* filter-specific flag */
277 
278     /* returned values */
279     EV_EOF = 0x8000, /* EOF detected */
280     EV_ERROR = 0x4000, /* error, data contains errno */
281 
282 }
283 
284 enum {
285     /*
286     * data/hint flags/masks for EVFILT_USER, shared with userspace
287     *
288     * On input, the top two bits of fflags specifies how the lower twenty four
289     * bits should be applied to the stored value of fflags.
290     *
291     * On output, the top two bits will always be set to NOTE_FFNOP and the
292     * remaining twenty four bits will contain the stored fflags value.
293     */
294     NOTE_FFNOP = 0x00000000, /* ignore input fflags */
295     NOTE_FFAND = 0x40000000, /* AND fflags */
296     NOTE_FFOR = 0x80000000, /* OR fflags */
297     NOTE_FFCOPY = 0xc0000000, /* copy fflags */
298     NOTE_FFCTRLMASK = 0xc0000000, /* masks for operations */
299     NOTE_FFLAGSMASK = 0x00ffffff,
300 
301     NOTE_TRIGGER = 0x01000000, /* Cause the event to be
302                                     triggered for output. */
303 
304     /*
305     * data/hint flags for EVFILT_{READ|WRITE}, shared with userspace
306     */
307     NOTE_LOWAT = 0x0001, /* low water mark */
308 
309     /*
310     * data/hint flags for EVFILT_VNODE, shared with userspace
311     */
312     NOTE_DELETE = 0x0001, /* vnode was removed */
313     NOTE_WRITE = 0x0002, /* data contents changed */
314     NOTE_EXTEND = 0x0004, /* size increased */
315     NOTE_ATTRIB = 0x0008, /* attributes changed */
316     NOTE_LINK = 0x0010, /* link count changed */
317     NOTE_RENAME = 0x0020, /* vnode was renamed */
318     NOTE_REVOKE = 0x0040, /* vnode access was revoked */
319 
320     /*
321     * data/hint flags for EVFILT_PROC, shared with userspace
322     */
323     NOTE_EXIT = 0x80000000, /* process exited */
324     NOTE_FORK = 0x40000000, /* process forked */
325     NOTE_EXEC = 0x20000000, /* process exec'd */
326     NOTE_PCTRLMASK = 0xf0000000, /* mask for hint bits */
327     NOTE_PDATAMASK = 0x000fffff, /* mask for pid */
328 
329     /* additional flags for EVFILT_PROC */
330     NOTE_TRACK = 0x00000001, /* follow across forks */
331     NOTE_TRACKERR = 0x00000002, /* could not track child */
332     NOTE_CHILD = 0x00000004, /* am a child process */
333 
334 }
335 
336 extern (C) {
337     int kqueue() @nogc nothrow;
338     int kevent(int kq, const Kevent* changelist, int nchanges,
339             Kevent* eventlist, int nevents, const timespec* timeout) @nogc nothrow;
340 }
341 
342 static if (CompilerHelper.isLessThan(2078)) {
343     enum SO_REUSEPORT = 0x0200;
344 }