001/** 002 * 003 * Copyright 2018-2020 Florian Schmaus 004 * 005 * Licensed under the Apache License, Version 2.0 (the "License"); 006 * you may not use this file except in compliance with the License. 007 * You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.jivesoftware.smack; 018 019import java.io.IOException; 020import java.nio.channels.CancelledKeyException; 021import java.nio.channels.ClosedChannelException; 022import java.nio.channels.SelectableChannel; 023import java.nio.channels.SelectionKey; 024import java.nio.channels.Selector; 025import java.util.ArrayList; 026import java.util.Collection; 027import java.util.Collections; 028import java.util.Date; 029import java.util.Iterator; 030import java.util.List; 031import java.util.Queue; 032import java.util.Set; 033import java.util.concurrent.ConcurrentLinkedQueue; 034import java.util.concurrent.DelayQueue; 035import java.util.concurrent.Semaphore; 036import java.util.concurrent.TimeUnit; 037import java.util.concurrent.atomic.AtomicBoolean; 038import java.util.concurrent.locks.Lock; 039import java.util.concurrent.locks.ReentrantLock; 040import java.util.logging.Level; 041import java.util.logging.Logger; 042 043/** 044 * The SmackReactor for non-blocking I/O. 045 * <p> 046 * Highlights include: 047 * <ul> 048 * <li>Multiple reactor threads</li> 049 * <li>Scheduled actions</li> 050 * </ul> 051 * 052 * <pre> 053 * 054 * ) ) ) 055 * ( ( ( 056 * ) ) ) 057 * (~~~~~~~~~) 058 * | Smack | 059 * |Reactor| 060 * I _._ 061 * I /' `\ 062 * I | | 063 * f | |~~~~~~~~~~~~~~| 064 * .' | | # # # # | 065 * '______|___|___________###| 066 * </pre> 067 */ 068public class SmackReactor { 069 070 private static final Logger LOGGER = Logger.getLogger(SmackReactor.class.getName()); 071 072 private static final int DEFAULT_REACTOR_THREAD_COUNT = 2; 073 074 private static final int PENDING_SET_INTEREST_OPS_MAX_BATCH_SIZE = 1024; 075 076 private static SmackReactor INSTANCE; 077 078 static synchronized SmackReactor getInstance() { 079 if (INSTANCE == null) { 080 INSTANCE = new SmackReactor("DefaultReactor"); 081 } 082 return INSTANCE; 083 } 084 085 private final Selector selector; 086 private final String reactorName; 087 088 private final List<Reactor> reactorThreads = Collections.synchronizedList(new ArrayList<>()); 089 090 private final DelayQueue<ScheduledAction> scheduledActions = new DelayQueue<>(); 091 092 private final Lock registrationLock = new ReentrantLock(); 093 094 /** 095 * The semaphore protecting the handling of the actions. Note that it is 096 * initialized with -1, which basically means that one thread will always do I/O using 097 * select(). 098 */ 099 private final Semaphore actionsSemaphore = new Semaphore(-1, false); 100 101 private final Queue<SelectionKey> pendingSelectionKeys = new ConcurrentLinkedQueue<>(); 102 103 private final Queue<SetInterestOps> pendingSetInterestOps = new ConcurrentLinkedQueue<>(); 104 105 SmackReactor(String reactorName) { 106 this.reactorName = reactorName; 107 108 try { 109 selector = Selector.open(); 110 } 111 catch (IOException e) { 112 throw new IllegalStateException(e); 113 } 114 115 setReactorThreadCount(DEFAULT_REACTOR_THREAD_COUNT); 116 } 117 118 public SelectionKey registerWithSelector(SelectableChannel channel, int ops, ChannelSelectedCallback callback) 119 throws ClosedChannelException { 120 SelectionKeyAttachment selectionKeyAttachment = new SelectionKeyAttachment(callback); 121 122 registrationLock.lock(); 123 try { 124 selector.wakeup(); 125 return channel.register(selector, ops, selectionKeyAttachment); 126 } finally { 127 registrationLock.unlock(); 128 } 129 } 130 131 public void setInterestOps(SelectionKey selectionKey, int interestOps) { 132 SetInterestOps setInterestOps = new SetInterestOps(selectionKey, interestOps); 133 pendingSetInterestOps.add(setInterestOps); 134 selector.wakeup(); 135 } 136 137 private static final class SetInterestOps { 138 private final SelectionKey selectionKey; 139 private final int interestOps; 140 141 private SetInterestOps(SelectionKey selectionKey, int interestOps) { 142 this.selectionKey = selectionKey; 143 this.interestOps = interestOps; 144 } 145 } 146 147 ScheduledAction schedule(Runnable runnable, long delay, TimeUnit unit, ScheduledAction.Kind scheduledActionKind) { 148 long releaseTimeEpoch = System.currentTimeMillis() + unit.toMillis(delay); 149 Date releaseTimeDate = new Date(releaseTimeEpoch); 150 ScheduledAction scheduledAction = new ScheduledAction(runnable, releaseTimeDate, this, scheduledActionKind); 151 scheduledActions.add(scheduledAction); 152 selector.wakeup(); 153 return scheduledAction; 154 } 155 156 /** 157 * Cancels the scheduled action. 158 * 159 * @param scheduledAction the scheduled action to cancel. 160 * @return <code>true</code> if the scheduled action was still pending and got removed, <code>false</code> otherwise. 161 */ 162 boolean cancel(ScheduledAction scheduledAction) { 163 return scheduledActions.remove(scheduledAction); 164 } 165 166 private class Reactor extends Thread { 167 168 private volatile long shutdownRequestTimestamp = -1; 169 170 @Override 171 public void run() { 172 try { 173 reactorLoop(); 174 } finally { 175 if (shutdownRequestTimestamp > 0) { 176 long shutDownDelay = System.currentTimeMillis() - shutdownRequestTimestamp; 177 LOGGER.info(this + " shut down after " + shutDownDelay + "ms"); 178 } else { 179 boolean contained = reactorThreads.remove(this); 180 assert contained; 181 } 182 } 183 } 184 185 private void reactorLoop() { 186 // Loop until reactor shutdown was requested. 187 while (shutdownRequestTimestamp < 0) { 188 handleScheduledActionsOrPerformSelect(); 189 190 handlePendingSelectionKeys(); 191 } 192 } 193 194 @SuppressWarnings("LockNotBeforeTry") 195 private void handleScheduledActionsOrPerformSelect() { 196 ScheduledAction dueScheduledAction = null; 197 198 boolean permitToHandleScheduledActions = actionsSemaphore.tryAcquire(); 199 if (permitToHandleScheduledActions) { 200 try { 201 dueScheduledAction = scheduledActions.poll(); 202 } finally { 203 actionsSemaphore.release(); 204 } 205 } 206 207 if (dueScheduledAction != null) { 208 dueScheduledAction.run(); 209 return; 210 } 211 212 int newSelectedKeysCount = 0; 213 List<SelectionKey> selectedKeys; 214 synchronized (selector) { 215 ScheduledAction nextScheduledAction = scheduledActions.peek(); 216 217 long selectWait; 218 if (nextScheduledAction == null) { 219 // There is no next scheduled action, wait indefinitely in select() or until another thread invokes 220 // selector.wakeup(). 221 selectWait = 0; 222 } else { 223 selectWait = nextScheduledAction.getTimeToDueMillis(); 224 } 225 226 if (selectWait < 0) { 227 // A scheduled action was just released and became ready to execute. 228 return; 229 } 230 231 // Before we call select, we handle the pending the interest Ops. This will not block since no other 232 // thread is currently in select() at this time. 233 // Note: This was put deliberately before the registration lock. It may cause more synchronization but 234 // allows for more parallelism. 235 // Hopefully that assumption is right. 236 int myHandledPendingSetInterestOps = 0; 237 for (SetInterestOps setInterestOps; (setInterestOps = pendingSetInterestOps.poll()) != null;) { 238 setInterestOpsCancelledKeySafe(setInterestOps.selectionKey, setInterestOps.interestOps); 239 240 if (myHandledPendingSetInterestOps++ >= PENDING_SET_INTEREST_OPS_MAX_BATCH_SIZE) { 241 // This thread has handled enough "set pending interest ops" requests. Wakeup another one to 242 // handle the remaining (if any). 243 selector.wakeup(); 244 break; 245 } 246 } 247 248 // Ensure that a wakeup() in registerWithSelector() gives the corresponding 249 // register() in the same method the chance to actually register the channel. In 250 // other words: This construct ensures that there is never another select() 251 // between a corresponding wakeup() and register() calls. 252 // See also https://stackoverflow.com/a/1112809/194894 253 registrationLock.lock(); 254 registrationLock.unlock(); 255 256 try { 257 newSelectedKeysCount = selector.select(selectWait); 258 } catch (IOException e) { 259 LOGGER.log(Level.SEVERE, "IOException while using select()", e); 260 return; 261 } 262 263 if (newSelectedKeysCount == 0) { 264 return; 265 } 266 267 // Copy the selected-key set over to selectedKeys, remove the keys from the 268 // selected key set and loose interest of the key OPs for the time being. 269 // Note that we perform this operation in two steps in order to maximize the 270 // timespan setRacing() is set. 271 Set<SelectionKey> selectedKeySet = selector.selectedKeys(); 272 for (SelectionKey selectionKey : selectedKeySet) { 273 SelectionKeyAttachment selectionKeyAttachment = (SelectionKeyAttachment) selectionKey.attachment(); 274 selectionKeyAttachment.setRacing(); 275 } 276 for (SelectionKey selectionKey : selectedKeySet) { 277 setInterestOpsCancelledKeySafe(selectionKey, 0); 278 } 279 280 selectedKeys = new ArrayList<>(selectedKeySet.size()); 281 selectedKeys.addAll(selectedKeySet); 282 selectedKeySet.clear(); 283 } 284 285 int selectedKeysCount = selectedKeys.size(); 286 int currentReactorThreadCount = reactorThreads.size(); 287 int myKeyCount; 288 if (selectedKeysCount > currentReactorThreadCount) { 289 myKeyCount = selectedKeysCount / currentReactorThreadCount; 290 } else { 291 myKeyCount = selectedKeysCount; 292 } 293 294 final Level reactorSelectStatsLogLevel = Level.FINE; 295 if (LOGGER.isLoggable(reactorSelectStatsLogLevel)) { 296 LOGGER.log(reactorSelectStatsLogLevel, 297 "New selected key count: " + newSelectedKeysCount 298 + ". Total selected key count " + selectedKeysCount 299 + ". My key count: " + myKeyCount 300 + ". Current reactor thread count: " + currentReactorThreadCount); 301 } 302 303 Collection<SelectionKey> mySelectedKeys = new ArrayList<>(myKeyCount); 304 Iterator<SelectionKey> it = selectedKeys.iterator(); 305 for (int i = 0; i < myKeyCount; i++) { 306 SelectionKey selectionKey = it.next(); 307 mySelectedKeys.add(selectionKey); 308 } 309 while (it.hasNext()) { 310 // Drain to pendingSelectionKeys. 311 SelectionKey selectionKey = it.next(); 312 pendingSelectionKeys.add(selectionKey); 313 } 314 315 if (selectedKeysCount - myKeyCount > 0) { 316 // There where pending selection keys: Wakeup another reactor thread to handle them. 317 selector.wakeup(); 318 } 319 320 handleSelectedKeys(mySelectedKeys); 321 } 322 323 private void handlePendingSelectionKeys() { 324 final int pendingSelectionKeysSize = pendingSelectionKeys.size(); 325 if (pendingSelectionKeysSize == 0) { 326 return; 327 } 328 329 int currentReactorThreadCount = reactorThreads.size(); 330 int myKeyCount = pendingSelectionKeysSize / currentReactorThreadCount; 331 Collection<SelectionKey> selectedKeys = new ArrayList<>(myKeyCount); 332 for (int i = 0; i < myKeyCount; i++) { 333 SelectionKey selectionKey = pendingSelectionKeys.poll(); 334 if (selectionKey == null) { 335 // We lost a race and can abort here since the pendingSelectionKeys queue is empty. 336 break; 337 } 338 selectedKeys.add(selectionKey); 339 } 340 341 if (!pendingSelectionKeys.isEmpty()) { 342 // There are more pending selection keys, wakeup a thread blocked in select() to handle them. 343 selector.wakeup(); 344 } 345 346 handleSelectedKeys(selectedKeys); 347 } 348 349 private void setInterestOpsCancelledKeySafe(SelectionKey selectionKey, int interestOps) { 350 try { 351 selectionKey.interestOps(interestOps); 352 } 353 catch (CancelledKeyException e) { 354 final Level keyCancelledLogLevel = Level.FINER; 355 if (LOGGER.isLoggable(keyCancelledLogLevel)) { 356 LOGGER.log(keyCancelledLogLevel, "Key '" + selectionKey + "' has been cancelled", e); 357 } 358 } 359 } 360 361 void requestShutdown() { 362 shutdownRequestTimestamp = System.currentTimeMillis(); 363 } 364 } 365 366 private static void handleSelectedKeys(Collection<SelectionKey> selectedKeys) { 367 for (SelectionKey selectionKey : selectedKeys) { 368 SelectableChannel channel = selectionKey.channel(); 369 SelectionKeyAttachment selectionKeyAttachment = (SelectionKeyAttachment) selectionKey.attachment(); 370 ChannelSelectedCallback channelSelectedCallback = selectionKeyAttachment.channelSelectedCallback; 371 channelSelectedCallback.onChannelSelected(channel, selectionKey); 372 } 373 } 374 375 public interface ChannelSelectedCallback { 376 void onChannelSelected(SelectableChannel channel, SelectionKey selectionKey); 377 } 378 379 public void setReactorThreadCount(int reactorThreadCount) { 380 if (reactorThreadCount < 2) { 381 throw new IllegalArgumentException("Must have at least two reactor threads, but you requested " + reactorThreadCount); 382 } 383 384 synchronized (reactorThreads) { 385 int deltaThreads = reactorThreadCount - reactorThreads.size(); 386 if (deltaThreads > 0) { 387 // Start new reactor thread. Note that we start the threads before we increase the permits of the 388 // actionsSemaphore. 389 for (int i = 0; i < deltaThreads; i++) { 390 Reactor reactor = new Reactor(); 391 reactor.setDaemon(true); 392 reactor.setName("Smack " + reactorName + " Thread #" + i); 393 reactorThreads.add(reactor); 394 reactor.start(); 395 } 396 397 actionsSemaphore.release(deltaThreads); 398 } else { 399 // Stop existing reactor threads. First we change the sign of deltaThreads, then we decrease the permits 400 // of the actionsSemaphore *before* we signal the selected reactor threads that they should shut down. 401 deltaThreads -= deltaThreads; 402 403 for (int i = deltaThreads - 1; i > 0; i--) { 404 // Note that this could potentially block forever, starving on the unfair semaphore. 405 actionsSemaphore.acquireUninterruptibly(); 406 } 407 408 for (int i = deltaThreads - 1; i > 0; i--) { 409 Reactor reactor = reactorThreads.remove(i); 410 reactor.requestShutdown(); 411 } 412 413 selector.wakeup(); 414 } 415 } 416 } 417 418 public static final class SelectionKeyAttachment { 419 private final ChannelSelectedCallback channelSelectedCallback; 420 private final AtomicBoolean reactorThreadRacing = new AtomicBoolean(); 421 422 private SelectionKeyAttachment(ChannelSelectedCallback channelSelectedCallback) { 423 this.channelSelectedCallback = channelSelectedCallback; 424 } 425 426 private void setRacing() { 427 // We use lazySet here since it is sufficient if the value does not become visible immediately. 428 reactorThreadRacing.lazySet(true); 429 } 430 431 public void resetReactorThreadRacing() { 432 reactorThreadRacing.set(false); 433 } 434 435 public boolean isReactorThreadRacing() { 436 return reactorThreadRacing.get(); 437 } 438 439 } 440}