1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.newsclub.net.unix;
19
20 import java.io.FileDescriptor;
21 import java.io.IOException;
22 import java.net.SocketException;
23 import java.net.SocketTimeoutException;
24 import java.nio.ByteBuffer;
25 import java.nio.channels.ClosedSelectorException;
26 import java.nio.channels.SelectableChannel;
27 import java.nio.channels.SelectionKey;
28 import java.nio.channels.Selector;
29 import java.nio.channels.spi.AbstractSelectableChannel;
30 import java.nio.channels.spi.AbstractSelector;
31 import java.util.Collections;
32 import java.util.Iterator;
33 import java.util.Map;
34 import java.util.Set;
35 import java.util.concurrent.ConcurrentHashMap;
36 import java.util.concurrent.atomic.AtomicInteger;
37
38 import com.kohlschutter.annotations.compiletime.SuppressFBWarnings;
39
40 final class AFSelector extends AbstractSelector {
41 private final AFPipe selectorPipe;
42 private final PollFd selectorPipePollFd;
43
44 private final ByteBuffer pipeMsgWakeUp = ByteBuffer.allocate(1);
45 private final ByteBuffer pipeMsgReceiveBuffer = ByteBuffer.allocateDirect(256);
46
47 private final Map<AFSelectionKey, Integer> keysRegistered = new ConcurrentHashMap<>();
48 private final Set<AFSelectionKey> keysRegisteredKeySet = keysRegistered.keySet();
49 private final Set<SelectionKey> keysRegisteredPublic = Collections.unmodifiableSet(
50 keysRegisteredKeySet);
51
52 private final AtomicInteger selectCount = new AtomicInteger(0);
53
54 @SuppressWarnings("PMD.LooseCoupling")
55 private final MapValueSet<SelectionKey, Integer> selectedKeysSet =
56 new MapValueSet<SelectionKey, Integer>(keysRegistered, selectCount::get, 0);
57 private final Set<SelectionKey> selectedKeysPublic = new UngrowableSet<>(selectedKeysSet);
58
59 private PollFd pollFd = null;
60
61 AFSelector(AFSelectorProvider<?> provider) throws IOException {
62 super(provider);
63
64 this.selectorPipe = AFUNIXSelectorProvider.getInstance().openSelectablePipe();
65 this.selectorPipePollFd = new PollFd(selectorPipe.sourceFD());
66 }
67
68 @Override
69 protected SelectionKey register(AbstractSelectableChannel ch, int ops, Object att) {
70 AFSelectionKey key = new AFSelectionKey(this, ch, ops, att);
71 synchronized (this) {
72 pollFd = null;
73 selectedKeysSet.markRemoved(key);
74 }
75 return key;
76 }
77
78 @Override
79 public Set<SelectionKey> keys() {
80 return keysRegisteredPublic;
81 }
82
83 @Override
84 @SuppressFBWarnings("EI_EXPOSE_REP")
85 public Set<SelectionKey> selectedKeys() {
86 return selectedKeysPublic;
87 }
88
89 @Override
90 public int selectNow() throws IOException {
91 return select0(0);
92 }
93
94 @Override
95 public int select(long timeout) throws IOException {
96 if (timeout > Integer.MAX_VALUE) {
97 timeout = Integer.MAX_VALUE;
98 } else if (timeout < 0) {
99 throw new IllegalArgumentException("Timeout must not be negative");
100 }
101
102 return select0((int) timeout);
103 }
104
105 @Override
106 public int select() throws IOException {
107 try {
108 return select0(-1);
109 } catch (SocketTimeoutException e) {
110 return 0;
111 }
112 }
113
114 private int select0(int timeout) throws IOException {
115 PollFd pfd;
116
117 int selectId = updateSelectCount();
118
119 synchronized (this) {
120 if (!isOpen()) {
121 throw new ClosedSelectorException();
122 }
123
124 pfd = pollFd = initPollFd(pollFd);
125 }
126 int num;
127 try {
128 begin();
129 num = NativeUnixSocket.poll(pfd, timeout);
130 } finally {
131 end();
132 }
133 synchronized (this) {
134 pfd = pollFd;
135 if (pfd != null) {
136 AFSelectionKey[] keys = pfd.keys;
137 if (keys != null) {
138 for (AFSelectionKey key : keys) {
139 if (key != null && key.hasOpInvalid()) {
140 SelectableChannel ch = key.channel();
141 if (ch != null && ch.isOpen()) {
142 ch.close();
143 }
144 }
145 }
146 }
147 }
148 if (num > 0) {
149 consumeAllBytesAfterPoll();
150 setOpsReady(pfd, selectId);
151 }
152 return selectedKeysSet.size();
153 }
154 }
155
156 private synchronized void consumeAllBytesAfterPoll() throws IOException {
157 if (pollFd == null) {
158 return;
159 }
160 if ((pollFd.rops[0] & SelectionKey.OP_READ) == 0) {
161 return;
162 }
163 int maxReceive;
164 int bytesReceived;
165
166 int options = selectorPipe.getOptions();
167
168 synchronized (pipeMsgReceiveBuffer) {
169 pipeMsgReceiveBuffer.clear();
170 maxReceive = pipeMsgReceiveBuffer.remaining();
171 bytesReceived = receive(maxReceive, options);
172 }
173
174 if (bytesReceived == maxReceive && maxReceive > 0) {
175
176 int read;
177 do {
178 if ((read = NativeUnixSocket.poll(selectorPipePollFd, 0)) > 0) {
179 synchronized (pipeMsgReceiveBuffer) {
180 pipeMsgReceiveBuffer.clear();
181 read = receive(maxReceive, options);
182 }
183 }
184 } while (read == maxReceive && read > 0);
185 }
186 }
187
188 @SuppressWarnings("PMD.CognitiveComplexity")
189 private int receive(int maxReceive, int options) throws IOException {
190 final boolean virtualBlocking = ThreadUtil.isVirtualThread();
191 final long now;
192 if (virtualBlocking) {
193 now = System.currentTimeMillis();
194 options |= NativeUnixSocket.OPT_NON_BLOCKING;
195 } else {
196 now = 0;
197 }
198
199 FileDescriptor fdesc = selectorPipePollFd.fds[0];
200
201 boolean park = false;
202 int count;
203 virtualThreadLoop : do {
204 if (virtualBlocking) {
205 if (park) {
206 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_WRITE, now,
207 AFPipe.DUMMY_TIMEOUT, this::close);
208 }
209 NativeUnixSocket.configureBlocking(fdesc, false);
210 }
211 try {
212 count = NativeUnixSocket.receive(fdesc, pipeMsgReceiveBuffer, 0, maxReceive, null, options,
213 null, 1);
214 if (count == 0 && virtualBlocking) {
215
216 park = true;
217 continue virtualThreadLoop;
218 }
219 } catch (SocketTimeoutException e) {
220 if (virtualBlocking) {
221
222 park = true;
223 continue virtualThreadLoop;
224 } else {
225 throw e;
226 }
227 } finally {
228 if (virtualBlocking) {
229 NativeUnixSocket.configureBlocking(fdesc, true);
230 }
231 }
232 break;
233 } while (true);
234 return count;
235 }
236
237 private int updateSelectCount() {
238 int selectId = selectCount.incrementAndGet();
239 if (selectId == 0) {
240
241 selectedKeysSet.markAllRemoved();
242 selectId = selectCount.incrementAndGet();
243 }
244 return selectId;
245 }
246
247 private void setOpsReady(PollFd pfd, int selectId) {
248 if (pfd != null) {
249 for (int i = 1; i < pfd.rops.length; i++) {
250 int rops = pfd.rops[i];
251 AFSelectionKey key = pfd.keys[i];
252 if (key == null) {
253
254 assert (rops == 0);
255 continue;
256 }
257 key.setOpsReady(rops);
258 if (rops != 0 && keysRegistered.containsKey(key)) {
259 keysRegistered.put(key, selectId);
260 }
261 }
262 }
263 }
264
265 @SuppressWarnings({"resource", "PMD.CognitiveComplexity"})
266 private PollFd initPollFd(PollFd existingPollFd) throws IOException {
267 synchronized (this) {
268 for (Iterator<AFSelectionKey> it = keysRegisteredKeySet.iterator(); it.hasNext();) {
269 AFSelectionKey key = it.next();
270 if (!key.getAFCore().fd.valid() || !key.isValid()) {
271 key.cancelNoRemove();
272 it.remove();
273 existingPollFd = null;
274 } else {
275 key.setOpsReady(0);
276 }
277 }
278
279 if (existingPollFd != null &&
280 existingPollFd.keys != null &&
281 (existingPollFd.keys.length - 1) == keysRegistered.size()) {
282 boolean needsUpdate = false;
283 int i = 1;
284 for (AFSelectionKey key : keysRegisteredKeySet) {
285 if (existingPollFd.keys[i] != key || !key.isValid()) {
286 needsUpdate = true;
287 break;
288 }
289 existingPollFd.ops[i] = key.interestOps();
290
291 i++;
292 }
293
294 if (!needsUpdate) {
295 return existingPollFd;
296 }
297 }
298
299 int keysToPoll = keysRegistered.size();
300 for (AFSelectionKey key : keysRegisteredKeySet) {
301 if (!key.isValid()) {
302 keysToPoll--;
303 }
304 }
305
306 int size = keysToPoll + 1;
307 FileDescriptor[] fds = new FileDescriptor[size];
308 int[] ops = new int[size];
309
310 AFSelectionKey[] keys = new AFSelectionKey[size];
311 fds[0] = selectorPipe.sourceFD();
312 ops[0] = SelectionKey.OP_READ;
313
314 int i = 1;
315 for (AFSelectionKey key : keysRegisteredKeySet) {
316 if (!key.isValid()) {
317 continue;
318 }
319 keys[i] = key;
320 fds[i] = key.getAFCore().fd;
321 ops[i] = key.interestOps();
322 i++;
323 }
324 return new PollFd(keys, fds, ops);
325 }
326 }
327
328 @Override
329 protected void implCloseSelector() throws IOException {
330 wakeup();
331 Set<SelectionKey> keys;
332 synchronized (this) {
333 keys = keys();
334 keysRegistered.clear();
335 }
336 for (SelectionKey key : keys) {
337 ((AFSelectionKey) key).cancelNoRemove();
338 }
339 selectorPipe.close();
340 }
341
342 @Override
343 public Selector wakeup() {
344 if (isOpen()) {
345 try {
346 synchronized (pipeMsgWakeUp) {
347 pipeMsgWakeUp.clear();
348 try {
349 selectorPipe.sink().write(pipeMsgWakeUp);
350 } catch (SocketException e) {
351 if (selectorPipe.sinkFD().valid()) {
352 throw e;
353 } else {
354
355 }
356 }
357 }
358 } catch (IOException e) {
359
360 StackTraceUtil.printStackTrace(e);
361 }
362 }
363 return this;
364 }
365
366 synchronized void remove(AFSelectionKey key) {
367 selectedKeysSet.remove(key);
368 deregister(key);
369 pollFd = null;
370 }
371
372 private void deregister(AFSelectionKey key) {
373
374
375
376 try {
377 NativeUnixSocket.deregisterSelectionKey((AbstractSelectableChannel) key.channel(), key);
378 } catch (ClassCastException e) {
379
380
381 }
382 }
383
384 static final class PollFd {
385
386 final FileDescriptor[] fds;
387
388 final int[] ops;
389
390 final int[] rops;
391
392 final AFSelectionKey[] keys;
393
394 PollFd(FileDescriptor pipeSourceFd) {
395 this(pipeSourceFd, SelectionKey.OP_READ);
396 }
397
398 PollFd(FileDescriptor pipeSourceFd, int op) {
399 this.fds = new FileDescriptor[] {pipeSourceFd};
400 this.ops = new int[] {op};
401 this.rops = new int[1];
402 this.keys = null;
403 }
404
405 PollFd(FileDescriptor[] fds, int[] ops) {
406 this(null, fds, ops);
407 }
408
409 @SuppressWarnings("PMD.ArrayIsStoredDirectly")
410 PollFd(AFSelectionKey[] keys, FileDescriptor[] fds, int[] ops) {
411 this.keys = keys;
412 if (fds.length != ops.length) {
413 throw new IllegalStateException();
414 }
415 this.fds = fds;
416 this.ops = ops;
417 this.rops = new int[ops.length];
418 }
419 }
420 }