View Javadoc
1   /*
2    * junixsocket
3    *
4    * Copyright 2009-2026 Christian Kohlschütter
5    *
6    * Licensed under the Apache License, Version 2.0 (the "License");
7    * you may not use this file except in compliance with the License.
8    * You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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); // updates keysSelected and numKeysSelected
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       // consume all pending bytes
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           // try again
216           park = true;
217           continue virtualThreadLoop;
218         }
219       } catch (SocketTimeoutException e) {
220         if (virtualBlocking) {
221           // try again
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; // NOPMD.AvoidBranchingStatementAsLastInLoop virtualThreadLoop
233     } while (true); // NOPMD.WhileLoopWithLiteralBoolean
234     return count;
235   }
236 
237   private int updateSelectCount() {
238     int selectId = selectCount.incrementAndGet();
239     if (selectId == 0) {
240       // overflow (unlikely)
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           // empty key slots should never return a ready op
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()) { // NOPMD
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               // ignore (Broken pipe, etc)
355             }
356           }
357         }
358       } catch (IOException e) { // NOPMD.ExceptionAsFlowControl
359         // FIXME throw as runtimeexception?
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     // super.deregister unnecessarily casts SelectionKey to AbstractSelectionKey, and
374     // ((AbstractSelectableChannel)key.channel()).removeKey(key); is not visible.
375     // so we have to resort to some JNI trickery...
376     try {
377       NativeUnixSocket.deregisterSelectionKey((AbstractSelectableChannel) key.channel(), key);
378     } catch (ClassCastException e) {
379       // because our key isn't an AbstractSelectableKey, internal invalidation fails
380       // but at that point, the key is deregistered
381     }
382   }
383 
384   static final class PollFd {
385     // accessed from native code
386     final FileDescriptor[] fds;
387     // accessed from native code
388     final int[] ops;
389     // accessed from native code
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 }