dast.async.selector.epoll source code

1 module dast.async.selector.epoll;
2 
3 version (linux)  : import dast.async.core,
4 core.time,
5 core.stdc.string,
6 core.stdc.errno,
7 core.sys.linux.epoll,
8 core.sys.linux.sys.eventfd,
9 core.sys.posix.netinet.tcp,
10 core.sys.posix.netinet.in_,
11 core.sys.posix.unistd;
12 
13 version (HaveTimer) import dast.async.timer;
14 
15 alias Selector = Epoll;
16 
17 template Epoll() {
18 	this() {
19 		_eventHandle = epoll_create1(0);
20 		flags |= WF.Read;
21 		handle = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
22 		register(this);
23 	}
24 
25 	~this() {
26 		close();
27 		dispose();
28 	}
29 
30 	void dispose() {
31 		if (!_eventHandle)
32 			return;
33 		unregister(this);
34 		close(_eventHandle);
35 		_eventHandle = 0;
36 	}
37 
38 	bool register(SocketChannel watcher)
39 	in (watcher) {
40 		// debug (Log) info("register, watcher fd=", watcher.handle);
41 		const fd = watcher.handle;
42 		assert(fd >= 0, "The watcher.handle is not initilized");
43 
44 		// if (fd < 0) return false;
45 		epoll_event ev = buildEvent(watcher);
46 		if (epoll_ctl(_eventHandle, EPOLL_CTL_ADD, fd, &ev) != 0) {
47 			if (errno != EEXIST)
48 				return false;
49 		}
50 		return true;
51 	}
52 	/+
53 	bool reregister(SocketChannel watcher)
54 	in (watcher) {
55 		const fd = watcher.handle;
56 		if (fd < 0)
57 			return false;
58 		auto ev = buildEvent(watcher);
59 		return epoll_ctl(_eventHandle, EPOLL_CTL_MOD, fd, &ev) == 0;
60 	}
61 +/
62 
63 	bool unregister(SocketChannel watcher)
64 	in (watcher) {
65 		debug (Log)
66 			info("unregister watcher fd=", watcher.handle);
67 
68 		const fd = watcher.handle;
69 		if (fd < 0)
70 			return false;
71 
72 		if (epoll_ctl(_eventHandle, EPOLL_CTL_DEL, fd, null)) {
73 			error("unregister failed, watcher.handle=", watcher.handle);
74 			return false;
75 		}
76 		// TODO: check this
77 		// watcher.clear();
78 		return true;
79 	}
80 
81 	void onRead() @system {
82 		ulong value = void;
83 		read(handle, &value, value.sizeof);
84 		_rBuf.data = value;
85 	}
86 
87 	void onWeakUp() @system {
88 		epoll_event[64] events;
89 		const len = epoll_wait(_eventHandle, events.ptr, events.length, 10);
90 		foreach (i; 0 .. len) {
91 			auto watch = cast(SocketChannel)events[i].data.ptr;
92 			assert(watch);
93 
94 			if (events[i].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP)) {
95 				debug (Log)
96 					warning("close event: ", watch.handle);
97 				watch.close();
98 				continue;
99 			}
100 
101 			if (watch.isRegistered && (events[i].events & EPOLLIN)) {
102 				watch.onRead();
103 			}
104 
105 			if (watch.isRegistered && (events[i].events & EPOLLOUT)) {
106 				assert(cast(SocketChannel)watch);
107 				// watch.onWrite();
108 			}
109 		}
110 	}
111 
112 	private int _eventHandle;
113 }
114 
115 epoll_event buildEvent(SocketChannel watch) {
116 	const flags = watch.flags;
117 	uint events = EPOLLRDHUP | EPOLLERR | EPOLLHUP | (flags & WF.ReadWrite);
118 	events |= (flags & WF.OneShotET) << 26;
119 	return epoll_event(events, epoll_data_t(watch));
120 }