/* * Copyright (c) 2003, 2015, Oracle and/or its affiliates. All rights reserved. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ package com.sun.corba.se.impl.transport; import java.io.IOException; import java.net.ServerSocket; import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectableChannel; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ClosedSelectorException; import java.util.ArrayList; import java.util.HashMap; import java.util.Map; import java.util.Iterator; import java.util.List; import com.sun.corba.se.pept.broker.Broker; import com.sun.corba.se.pept.transport.Acceptor; import com.sun.corba.se.pept.transport.Connection; import com.sun.corba.se.pept.transport.EventHandler; import com.sun.corba.se.pept.transport.ListenerThread; import com.sun.corba.se.pept.transport.ReaderThread; import com.sun.corba.se.spi.logging.CORBALogDomains; import com.sun.corba.se.spi.orb.ORB; import com.sun.corba.se.spi.orbutil.threadpool.Work; import com.sun.corba.se.spi.orbutil.threadpool.NoSuchThreadPoolException; import com.sun.corba.se.spi.orbutil.threadpool.NoSuchWorkQueueException; import com.sun.corba.se.impl.logging.ORBUtilSystemException; import com.sun.corba.se.impl.orbutil.ORBUtility; /** * @author Harold Carr */ class SelectorImpl extends Thread implements com.sun.corba.se.pept.transport.Selector { private ORB orb; private Selector selector; private long timeout; private List deferredRegistrations; private List interestOpsList; private HashMap listenerThreads; private Map readerThreads; private boolean selectorStarted; private volatile boolean closed; private ORBUtilSystemException wrapper; public SelectorImpl(ORB orb) { super(null, null, "PEPT-Transport-Thread", 0, false); this.orb = orb; selector = null; selectorStarted = false; timeout = 60000; deferredRegistrations = new ArrayList(); interestOpsList = new ArrayList(); listenerThreads = new HashMap(); readerThreads = java.util.Collections.synchronizedMap(new HashMap()); closed = false; wrapper = ORBUtilSystemException.get(orb,CORBALogDomains.RPC_TRANSPORT); } public void setTimeout(long timeout) { this.timeout = timeout; } public long getTimeout() { return timeout; } public void registerInterestOps(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".registerInterestOps:-> " + eventHandler); } SelectionKey selectionKey = eventHandler.getSelectionKey(); if (selectionKey.isValid()) { int ehOps = eventHandler.getInterestOps(); SelectionKeyAndOp keyAndOp = new SelectionKeyAndOp(selectionKey, ehOps); synchronized(interestOpsList) { interestOpsList.add(keyAndOp); } // tell Selector Thread there's an update to a SelectorKey's Ops try { if (selector != null) { // wakeup Selector thread to process close request selector.wakeup(); } } catch (Throwable t) { if (orb.transportDebugFlag) { dprint(".registerInterestOps: selector.wakeup: ", t); } } } else { wrapper.selectionKeyInvalid(eventHandler.toString()); if (orb.transportDebugFlag) { dprint(".registerInterestOps: EventHandler SelectionKey not valid " + eventHandler); } } if (orb.transportDebugFlag) { dprint(".registerInterestOps:<- "); } } public void registerForEvent(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".registerForEvent: " + eventHandler); } if (isClosed()) { if (orb.transportDebugFlag) { dprint(".registerForEvent: closed: " + eventHandler); } return; } if (eventHandler.shouldUseSelectThreadToWait()) { synchronized (deferredRegistrations) { deferredRegistrations.add(eventHandler); } if (! selectorStarted) { startSelector(); } selector.wakeup(); return; } switch (eventHandler.getInterestOps()) { case SelectionKey.OP_ACCEPT : createListenerThread(eventHandler); break; case SelectionKey.OP_READ : createReaderThread(eventHandler); break; default: if (orb.transportDebugFlag) { dprint(".registerForEvent: default: " + eventHandler); } throw new RuntimeException( "SelectorImpl.registerForEvent: unknown interest ops"); } } public void unregisterForEvent(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".unregisterForEvent: " + eventHandler); } if (isClosed()) { if (orb.transportDebugFlag) { dprint(".unregisterForEvent: closed: " + eventHandler); } return; } if (eventHandler.shouldUseSelectThreadToWait()) { SelectionKey selectionKey ; synchronized(deferredRegistrations) { selectionKey = eventHandler.getSelectionKey(); } if (selectionKey != null) { selectionKey.cancel(); } if (selector != null) { selector.wakeup(); } return; } switch (eventHandler.getInterestOps()) { case SelectionKey.OP_ACCEPT : destroyListenerThread(eventHandler); break; case SelectionKey.OP_READ : destroyReaderThread(eventHandler); break; default: if (orb.transportDebugFlag) { dprint(".unregisterForEvent: default: " + eventHandler); } throw new RuntimeException( "SelectorImpl.uregisterForEvent: unknown interest ops"); } } public void close() { if (orb.transportDebugFlag) { dprint(".close"); } if (isClosed()) { if (orb.transportDebugFlag) { dprint(".close: already closed"); } return; } setClosed(true); Iterator i; // Kill listeners. i = listenerThreads.values().iterator(); while (i.hasNext()) { ListenerThread listenerThread = (ListenerThread) i.next(); listenerThread.close(); } // Kill readers. i = readerThreads.values().iterator(); while (i.hasNext()) { ReaderThread readerThread = (ReaderThread) i.next(); readerThread.close(); } clearDeferredRegistrations(); // Selector try { if (selector != null) { // wakeup Selector thread to process close request selector.wakeup(); } } catch (Throwable t) { if (orb.transportDebugFlag) { dprint(".close: selector.wakeup: ", t); } } } /////////////////////////////////////////////////// // // Thread methods. // public void run() { while (!closed) { try { int n = 0; if (timeout == 0 && orb.transportDebugFlag) { dprint(".run: Beginning of selection cycle"); } handleDeferredRegistrations(); enableInterestOps(); try { n = selector.select(timeout); } catch (IOException e) { if (orb.transportDebugFlag) { dprint(".run: selector.select: ", e); } } catch (ClosedSelectorException csEx) { if (orb.transportDebugFlag) { dprint(".run: selector.select: ", csEx); } break; } if (closed) { break; } /* if (timeout == 0 && orb.transportDebugFlag) { dprint(".run: selector.select() returned: " + n); } if (n == 0) { continue; } */ Iterator iterator = selector.selectedKeys().iterator(); if (orb.transportDebugFlag) { if (iterator.hasNext()) { dprint(".run: n = " + n); } } while (iterator.hasNext()) { SelectionKey selectionKey = (SelectionKey) iterator.next(); iterator.remove(); EventHandler eventHandler = (EventHandler) selectionKey.attachment(); try { eventHandler.handleEvent(); } catch (Throwable t) { if (orb.transportDebugFlag) { dprint(".run: eventHandler.handleEvent", t); } } } if (timeout == 0 && orb.transportDebugFlag) { dprint(".run: End of selection cycle"); } } catch (Throwable t) { // IMPORTANT: ignore all errors so the select thread keeps running. // Otherwise a guaranteed hang. if (orb.transportDebugFlag) { dprint(".run: ignoring", t); } } } try { if (selector != null) { if (orb.transportDebugFlag) { dprint(".run: selector.close "); } selector.close(); } } catch (Throwable t) { if (orb.transportDebugFlag) { dprint(".run: selector.close: ", t); } } } ///////////////////////////////////////////////////// // // Implementation. // private void clearDeferredRegistrations() { synchronized (deferredRegistrations) { int deferredListSize = deferredRegistrations.size(); if (orb.transportDebugFlag) { dprint(".clearDeferredRegistrations:deferred list size == " + deferredListSize); } for (int i = 0; i < deferredListSize; i++) { EventHandler eventHandler = (EventHandler)deferredRegistrations.get(i); if (orb.transportDebugFlag) { dprint(".clearDeferredRegistrations: " + eventHandler); } SelectableChannel channel = eventHandler.getChannel(); SelectionKey selectionKey = null; try { if (orb.transportDebugFlag) { dprint(".clearDeferredRegistrations:close channel == " + channel); dprint(".clearDeferredRegistrations:close channel class == " + channel.getClass().getName()); } channel.close(); selectionKey = eventHandler.getSelectionKey(); if (selectionKey != null) { selectionKey.cancel(); selectionKey.attach(null); } } catch (IOException ioEx) { if (orb.transportDebugFlag) { dprint(".clearDeferredRegistrations: ", ioEx); } } } deferredRegistrations.clear(); } } private synchronized boolean isClosed () { return closed; } private synchronized void setClosed(boolean closed) { this.closed = closed; } private void startSelector() { try { selector = Selector.open(); } catch (IOException e) { if (orb.transportDebugFlag) { dprint(".startSelector: Selector.open: IOException: ", e); } // REVISIT - better handling/reporting RuntimeException rte = new RuntimeException(".startSelector: Selector.open exception"); rte.initCause(e); throw rte; } setDaemon(true); start(); selectorStarted = true; if (orb.transportDebugFlag) { dprint(".startSelector: selector.start completed."); } } private void handleDeferredRegistrations() { synchronized (deferredRegistrations) { int deferredListSize = deferredRegistrations.size(); for (int i = 0; i < deferredListSize; i++) { EventHandler eventHandler = (EventHandler)deferredRegistrations.get(i); if (orb.transportDebugFlag) { dprint(".handleDeferredRegistrations: " + eventHandler); } SelectableChannel channel = eventHandler.getChannel(); SelectionKey selectionKey = null; try { selectionKey = channel.register(selector, eventHandler.getInterestOps(), (Object)eventHandler); } catch (ClosedChannelException e) { if (orb.transportDebugFlag) { dprint(".handleDeferredRegistrations: ", e); } } eventHandler.setSelectionKey(selectionKey); } deferredRegistrations.clear(); } } private void enableInterestOps() { synchronized (interestOpsList) { int listSize = interestOpsList.size(); if (listSize > 0) { if (orb.transportDebugFlag) { dprint(".enableInterestOps:->"); } SelectionKey selectionKey = null; SelectionKeyAndOp keyAndOp = null; int keyOp, selectionKeyOps = 0; for (int i = 0; i < listSize; i++) { keyAndOp = (SelectionKeyAndOp)interestOpsList.get(i); selectionKey = keyAndOp.selectionKey; // Need to check if the SelectionKey is valid because a // connection's SelectionKey could be put on the list to // have its OP enabled and before it's enabled be reclaimed. // Otherwise, the enabling of the OP will throw an exception // here and exit this method an potentially not enable all // registered ops. // // So, we ignore SelectionKeys that are invalid. They will get // cleaned up on the next Selector.select() call. if (selectionKey.isValid()) { if (orb.transportDebugFlag) { dprint(".enableInterestOps: " + keyAndOp); } keyOp = keyAndOp.keyOp; selectionKeyOps = selectionKey.interestOps(); selectionKey.interestOps(selectionKeyOps | keyOp); } } interestOpsList.clear(); if (orb.transportDebugFlag) { dprint(".enableInterestOps:<-"); } } } } private void createListenerThread(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".createListenerThread: " + eventHandler); } Acceptor acceptor = eventHandler.getAcceptor(); ListenerThread listenerThread = new ListenerThreadImpl(orb, acceptor, this); listenerThreads.put(eventHandler, listenerThread); Throwable throwable = null; try { orb.getThreadPoolManager().getThreadPool(0) .getWorkQueue(0).addWork((Work)listenerThread); } catch (NoSuchThreadPoolException e) { throwable = e; } catch (NoSuchWorkQueueException e) { throwable = e; } if (throwable != null) { RuntimeException rte = new RuntimeException(throwable.toString()); rte.initCause(throwable); throw rte; } } private void destroyListenerThread(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".destroyListenerThread: " + eventHandler); } ListenerThread listenerThread = (ListenerThread) listenerThreads.get(eventHandler); if (listenerThread == null) { if (orb.transportDebugFlag) { dprint(".destroyListenerThread: cannot find ListenerThread - ignoring."); } return; } listenerThreads.remove(eventHandler); listenerThread.close(); } private void createReaderThread(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".createReaderThread: " + eventHandler); } Connection connection = eventHandler.getConnection(); ReaderThread readerThread = new ReaderThreadImpl(orb, connection, this); readerThreads.put(eventHandler, readerThread); Throwable throwable = null; try { orb.getThreadPoolManager().getThreadPool(0) .getWorkQueue(0).addWork((Work)readerThread); } catch (NoSuchThreadPoolException e) { throwable = e; } catch (NoSuchWorkQueueException e) { throwable = e; } if (throwable != null) { RuntimeException rte = new RuntimeException(throwable.toString()); rte.initCause(throwable); throw rte; } } private void destroyReaderThread(EventHandler eventHandler) { if (orb.transportDebugFlag) { dprint(".destroyReaderThread: " + eventHandler); } ReaderThread readerThread = (ReaderThread) readerThreads.get(eventHandler); if (readerThread == null) { if (orb.transportDebugFlag) { dprint(".destroyReaderThread: cannot find ReaderThread - ignoring."); } return; } readerThreads.remove(eventHandler); readerThread.close(); } private void dprint(String msg) { ORBUtility.dprint("SelectorImpl", msg); } protected void dprint(String msg, Throwable t) { dprint(msg); t.printStackTrace(System.out); } // Private class to contain a SelectionKey and a SelectionKey op. // Used only by SelectorImpl to register and enable SelectionKey // Op. // REVISIT - Could do away with this class and use the EventHanlder // directly. private class SelectionKeyAndOp { // A SelectionKey.[OP_READ|OP_WRITE|OP_ACCEPT|OP_CONNECT] public int keyOp; public SelectionKey selectionKey; // constructor public SelectionKeyAndOp(SelectionKey selectionKey, int keyOp) { this.selectionKey = selectionKey; this.keyOp = keyOp; } } // End of file. }