dast.async.selector.iocp source code

1 module dast.async.selector.iocp;
2 
3 version (Windows)  : import dast.async.core,
4 dast.async.iocp,
5 dast.async.tcpclient;
6 
7 alias Selector = Iocp;
8 
9 template Iocp() {
10 	import dast.async.core,
11 	dast.async.iocp,
12 	dast.async.tcpclient,
13 	dast.async.thread,
14 	tame.env : totalCPUs;
15 
16 @safe:
17 	ThreadPool* workerPool;
18 	this(uint threads = totalCPUs - 1) @trusted {
19 		_eventHandle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, null, 0, 0);
20 		if (!_eventHandle)
21 			throw new Exception("CreateIoCompletionPort failed");
22 		if (threads)
23 			workerPool = new ThreadPool(threads);
24 	}
25 
26 	/*~this() @nogc nothrow {
27 		CloseHandle(_eventHandle);
28 	}*/
29 
30 	bool register(SocketChannel watcher) @trusted nothrow
31 	in (watcher.type <= WT.Event) {
32 		const fd = watcher.handle;
33 		const type = watcher.type;
34 		if (type == WT.TCP || type == WT.Accept) {
35 			debug (Log)
36 				trace("register on socket: ", fd);
37 			CreateIoCompletionPort(cast(HANDLE)fd, _eventHandle, cast(ulong)cast(void*)watcher, 0);
38 		} else
39 			return false;
40 
41 		debug (Log)
42 			info("watcher(fd=", fd, ", type=", type, ')');
43 		_watchers ~= watcher;
44 		return true;
45 	}
46 
47 	void weakUp() nothrow @trusted {
48 		IocpContext ctx = {operation: IocpOperation.event}; // TODO
49 		PostQueuedCompletionStatus(_eventHandle, 0, 0, &ctx.overlapped);
50 	}
51 
52 	void onWeakUp() @trusted {
53 		enum timeout = 250, N = 32;
54 		OVERLAPPED_ENTRY[N] entries = void;
55 		uint n = void;
56 		const ret = GetQueuedCompletionStatusEx(_eventHandle, entries.ptr, entries.length, &n, timeout, 0);
57 		if (ret == 0) {
58 			debug (Log) {
59 				const err = GetLastError();
60 				if (err != WAIT_TIMEOUT) // && err != ERROR_OPERATION_ABORTED
61 					error("error occurred, code=", err);
62 			}
63 			return;
64 		}
65 		foreach (i; 0 .. n) {
66 			auto entry = entries[i];
67 			if (workerPool)
68 				workerPool.run!handleEvent(entry);
69 			else
70 				handleEvent(entry);
71 		}
72 	}
73 
74 private:
75 	HANDLE _eventHandle;
76 	SocketChannel[] _watchers;
77 }
78 
79 package(dast.async) void handleEvent(OVERLAPPED_ENTRY entry) {
80 	import dast.async.tcpserver;
81 
82 	const len = entry.dwNumberOfBytesTransferred;
83 	auto ev = cast(IocpContext*)entry.lpOverlapped;
84 	auto channel = cast(void*)entry.lpCompletionKey;
85 	switch (ev.operation) with (IocpOperation) {
86 	case accept:
87 		(cast(TCPServer)channel).onRead();
88 		break;
89 	case read:
90 		if (len && (cast(SocketChannel)channel).isRegistered)
91 			(cast(TCPClient)channel).onRead(len);
92 		break;
93 	case write:
94 		debug (Log)
95 			info("finishing writing ", len, " bytes");
96 		// Notify the client about how many bytes actually sent
97 		(cast(TCPClient)channel).onWrite(len);
98 		break;
99 		// TODO
100 		//case event:
101 		//break;
102 	default:
103 	}
104 }