jdk-sandbox: comparison jdk/src/java.rmi/share/classes/sun/rmi/transport/tcp/TCPChannel.java

equal deleted inserted replaced

-:836adbf7a2cd
+:3317bb8137f4
+/*
+* Copyright (c) 1996, 2013, 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 sun.rmi.transport.tcp;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.lang.ref.Reference;
+import java.lang.ref.SoftReference;
+import java.net.Socket;
+import java.rmi.ConnectIOException;
+import java.rmi.RemoteException;
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.WeakHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import sun.rmi.runtime.Log;
+import sun.rmi.runtime.NewThreadAction;
+import sun.rmi.runtime.RuntimeUtil;
+import sun.rmi.transport.Channel;
+import sun.rmi.transport.Connection;
+import sun.rmi.transport.Endpoint;
+import sun.rmi.transport.TransportConstants;
+/**
+* TCPChannel is the socket-based implementation of the RMI Channel
+* abstraction.
+*
+* @author Ann Wollrath
+*/
+public class TCPChannel implements Channel {
+/** endpoint for this channel */
+private final TCPEndpoint ep;
+/** transport for this channel */
+private final TCPTransport tr;
+/** list of cached connections */
+private final List<TCPConnection> freeList =
+new ArrayList<>();
+/** frees cached connections that have expired (guarded by freeList) */
+private Future<?> reaper = null;
+/** using multiplexer (for bi-directional applet communication */
+private boolean usingMultiplexer = false;
+/** connection multiplexer, if used */
+private ConnectionMultiplexer multiplexer = null;
+/** connection acceptor (should be in TCPTransport) */
+private ConnectionAcceptor acceptor;
+/** most recently authorized AccessControlContext */
+private AccessControlContext okContext;
+/** cache of authorized AccessControlContexts */
+private WeakHashMap<AccessControlContext,
+Reference<AccessControlContext>> authcache;
+/** the SecurityManager which authorized okContext and authcache */
+private SecurityManager cacheSecurityManager = null;
+/** client-side connection idle usage timeout */
+private static final long idleTimeout =             // default 15 seconds
+AccessController.doPrivileged((PrivilegedAction<Long>) () ->
+Long.getLong("sun.rmi.transport.connectionTimeout", 15000));
+/** client-side connection handshake read timeout */
+private static final int handshakeTimeout =         // default 1 minute
+AccessController.doPrivileged((PrivilegedAction<Integer>) () ->
+Integer.getInteger("sun.rmi.transport.tcp.handshakeTimeout", 60000));
+/** client-side connection response read timeout (after handshake) */
+private static final int responseTimeout =          // default infinity
+AccessController.doPrivileged((PrivilegedAction<Integer>) () ->
+Integer.getInteger("sun.rmi.transport.tcp.responseTimeout", 0));
+/** thread pool for scheduling delayed tasks */
+private static final ScheduledExecutorService scheduler =
+AccessController.doPrivileged(
+new RuntimeUtil.GetInstanceAction()).getScheduler();
+/**
+* Create channel for endpoint.
+*/
+TCPChannel(TCPTransport tr, TCPEndpoint ep) {
+this.tr = tr;
+this.ep = ep;
+}
+/**
+* Return the endpoint for this channel.
+*/
+public Endpoint getEndpoint() {
+return ep;
+}
+/**
+* Checks if the current caller has sufficient privilege to make
+* a connection to the remote endpoint.
+* @exception SecurityException if caller is not allowed to use this
+* Channel.
+*/
+private void checkConnectPermission() throws SecurityException {
+SecurityManager security = System.getSecurityManager();
+if (security == null)
+return;
+if (security != cacheSecurityManager) {
+// The security manager changed: flush the cache
+okContext = null;
+authcache = new WeakHashMap<AccessControlContext,
+Reference<AccessControlContext>>();
+cacheSecurityManager = security;
+}
+AccessControlContext ctx = AccessController.getContext();
+// If ctx is the same context as last time, or if it
+// appears in the cache, bypass the checkConnect.
+if (okContext == null ||
+!(okContext.equals(ctx) || authcache.containsKey(ctx)))
+{
+security.checkConnect(ep.getHost(), ep.getPort());
+authcache.put(ctx, new SoftReference<AccessControlContext>(ctx));
+// A WeakHashMap is transformed into a SoftHashSet by making
+// each value softly refer to its own key (Peter's idea).
+}
+okContext = ctx;
+}
+/**
+* Supplies a connection to the endpoint of the address space
+* for which this is a channel.  The returned connection may
+* be one retrieved from a cache of idle connections.
+*/
+public Connection newConnection() throws RemoteException {
+TCPConnection conn;
+// loop until we find a free live connection (in which case
+// we return) or until we run out of freelist (in which case
+// the loop exits)
+do {
+conn = null;
+// try to get a free connection
+synchronized (freeList) {
+int elementPos = freeList.size()-1;
+if (elementPos >= 0) {
+// If there is a security manager, make sure
+// the caller is allowed to connect to the
+// requested endpoint.
+checkConnectPermission();
+conn = freeList.get(elementPos);
+freeList.remove(elementPos);
+}
+}
+// at this point, conn is null iff the freelist is empty,
+// and nonnull if a free connection of uncertain vitality
+// has been found.
+if (conn != null) {
+// check to see if the connection has closed since last use
+if (!conn.isDead()) {
+TCPTransport.tcpLog.log(Log.BRIEF, "reuse connection");
+return conn;
+}
+// conn is dead, and cannot be reused (reuse => false)
+this.free(conn, false);
+}
+} while (conn != null);
+// none free, so create a new connection
+return (createConnection());
+}
+/**
+* Create a new connection to the remote endpoint of this channel.
+* The returned connection is new.  The caller must already have
+* passed a security checkConnect or equivalent.
+*/
+private Connection createConnection() throws RemoteException {
+Connection conn;
+TCPTransport.tcpLog.log(Log.BRIEF, "create connection");
+if (!usingMultiplexer) {
+Socket sock = ep.newSocket();
+conn = new TCPConnection(this, sock);
+try {
+DataOutputStream out =
+new DataOutputStream(conn.getOutputStream());
+writeTransportHeader(out);
+// choose protocol (single op if not reusable socket)
+if (!conn.isReusable()) {
+out.writeByte(TransportConstants.SingleOpProtocol);
+} else {
+out.writeByte(TransportConstants.StreamProtocol);
+out.flush();
+/*
+* Set socket read timeout to configured value for JRMP
+* connection handshake; this also serves to guard against
+* non-JRMP servers that do not respond (see 4322806).
+*/
+int originalSoTimeout = 0;
+try {
+originalSoTimeout = sock.getSoTimeout();
+sock.setSoTimeout(handshakeTimeout);
+} catch (Exception e) {
+// if we fail to set this, ignore and proceed anyway
+}
+DataInputStream in =
+new DataInputStream(conn.getInputStream());
+byte ack = in.readByte();
+if (ack != TransportConstants.ProtocolAck) {
+throw new ConnectIOException(
+ack == TransportConstants.ProtocolNack ?
+"JRMP StreamProtocol not supported by server" :
+"non-JRMP server at remote endpoint");
+}
+String suggestedHost = in.readUTF();
+int    suggestedPort = in.readInt();
+if (TCPTransport.tcpLog.isLoggable(Log.VERBOSE)) {
+TCPTransport.tcpLog.log(Log.VERBOSE,
+"server suggested " + suggestedHost + ":" +
+suggestedPort);
+}
+// set local host name, if unknown
+TCPEndpoint.setLocalHost(suggestedHost);
+// do NOT set the default port, because we don't
+// know if we can't listen YET...
+// write out default endpoint to match protocol
+// (but it serves no purpose)
+TCPEndpoint localEp =
+TCPEndpoint.getLocalEndpoint(0, null, null);
+out.writeUTF(localEp.getHost());
+out.writeInt(localEp.getPort());
+if (TCPTransport.tcpLog.isLoggable(Log.VERBOSE)) {
+TCPTransport.tcpLog.log(Log.VERBOSE, "using " +
+localEp.getHost() + ":" + localEp.getPort());
+}
+/*
+* After JRMP handshake, set socket read timeout to value
+* configured for the rest of the lifetime of the
+* connection.  NOTE: this timeout, if configured to a
+* finite duration, places an upper bound on the time
+* that a remote method call is permitted to execute.
+*/
+try {
+/*
+* If socket factory had set a non-zero timeout on its
+* own, then restore it instead of using the property-
+* configured value.
+*/
+sock.setSoTimeout((originalSoTimeout != 0 ?
+originalSoTimeout :
+responseTimeout));
+} catch (Exception e) {
+// if we fail to set this, ignore and proceed anyway
+}
+out.flush();
+}
+} catch (IOException e) {
+if (e instanceof RemoteException)
+throw (RemoteException) e;
+else
+throw new ConnectIOException(
+"error during JRMP connection establishment", e);
+}
+} else {
+try {
+conn = multiplexer.openConnection();
+} catch (IOException e) {
+synchronized (this) {
+usingMultiplexer = false;
+multiplexer = null;
+}
+throw new ConnectIOException(
+"error opening virtual connection " +
+"over multiplexed connection", e);
+}
+}
+return conn;
+}
+/**
+* Free the connection generated by this channel.
+* @param conn The connection
+* @param reuse If true, the connection is in a state in which it
+*        can be reused for another method call.
+*/
+public void free(Connection conn, boolean reuse) {
+if (conn == null) return;
+if (reuse && conn.isReusable()) {
+long lastuse = System.currentTimeMillis();
+TCPConnection tcpConnection = (TCPConnection) conn;
+TCPTransport.tcpLog.log(Log.BRIEF, "reuse connection");
+/*
+* Cache connection; if reaper task for expired
+* connections isn't scheduled, then schedule it.
+*/
+synchronized (freeList) {
+freeList.add(tcpConnection);
+if (reaper == null) {
+TCPTransport.tcpLog.log(Log.BRIEF, "create reaper");
+reaper = scheduler.scheduleWithFixedDelay(
+new Runnable() {
+public void run() {
+TCPTransport.tcpLog.log(Log.VERBOSE,
+"wake up");
+freeCachedConnections();
+}
+}, idleTimeout, idleTimeout, TimeUnit.MILLISECONDS);
+}
+}
+tcpConnection.setLastUseTime(lastuse);
+tcpConnection.setExpiration(lastuse + idleTimeout);
+} else {
+TCPTransport.tcpLog.log(Log.BRIEF, "close connection");
+try {
+conn.close();
+} catch (IOException ignored) {
+}
+}
+}
+/**
+* Send transport header over stream.
+*/
+private void writeTransportHeader(DataOutputStream out)
+throws RemoteException
+{
+try {
+// write out transport header
+DataOutputStream dataOut =
+new DataOutputStream(out);
+dataOut.writeInt(TransportConstants.Magic);
+dataOut.writeShort(TransportConstants.Version);
+} catch (IOException e) {
+throw new ConnectIOException(
+"error writing JRMP transport header", e);
+}
+}
+/**
+* Use given connection multiplexer object to obtain new connections
+* through this channel.
+*/
+synchronized void useMultiplexer(ConnectionMultiplexer newMultiplexer) {
+// for now, always just use the last one given
+multiplexer = newMultiplexer;
+usingMultiplexer = true;
+}
+/**
+* Accept a connection provided over a multiplexed channel.
+*/
+void acceptMultiplexConnection(Connection conn) {
+if (acceptor == null) {
+acceptor = new ConnectionAcceptor(tr);
+acceptor.startNewAcceptor();
+}
+acceptor.accept(conn);
+}
+/**
+* Closes all the connections in the cache, whether timed out or not.
+*/
+public void shedCache() {
+// Build a list of connections, to avoid holding the freeList
+// lock during (potentially long-running) close() calls.
+Connection[] conn;
+synchronized (freeList) {
+conn = freeList.toArray(new Connection[freeList.size()]);
+freeList.clear();
+}
+// Close all the connections that were free
+for (int i = conn.length; --i >= 0; ) {
+Connection c = conn[i];
+conn[i] = null; // help gc
+try {
+c.close();
+} catch (java.io.IOException e) {
+// eat exception
+}
+}
+}
+private void freeCachedConnections() {
+/*
+* Remove each connection whose time out has expired.
+*/
+synchronized (freeList) {
+int size = freeList.size();
+if (size > 0) {
+long time = System.currentTimeMillis();
+ListIterator<TCPConnection> iter = freeList.listIterator(size);
+while (iter.hasPrevious()) {
+TCPConnection conn = iter.previous();
+if (conn.expired(time)) {
+TCPTransport.tcpLog.log(Log.VERBOSE,
+"connection timeout expired");
+try {
+conn.close();
+} catch (java.io.IOException e) {
+// eat exception
+}
+iter.remove();
+}
+}
+}
+if (freeList.isEmpty()) {
+reaper.cancel(false);
+reaper = null;
+}
+}
+}
+}
+/**
+* ConnectionAcceptor manages accepting new connections and giving them
+* to TCPTransport's message handler on new threads.
+*
+* Since this object only needs to know which transport to give new
+* connections to, it doesn't need to be per-channel as currently
+* implemented.
+*/
+class ConnectionAcceptor implements Runnable {
+/** transport that will handle message on accepted connections */
+private TCPTransport transport;
+/** queue of connections to be accepted */
+private List<Connection> queue = new ArrayList<>();
+/** thread ID counter */
+private static int threadNum = 0;
+/**
+* Create a new ConnectionAcceptor that will give connections
+* to the specified transport on a new thread.
+*/
+public ConnectionAcceptor(TCPTransport transport) {
+this.transport = transport;
+}
+/**
+* Start a new thread to accept connections.
+*/
+public void startNewAcceptor() {
+Thread t = AccessController.doPrivileged(
+new NewThreadAction(ConnectionAcceptor.this,
+"Multiplex Accept-" + ++ threadNum,
+true));
+t.start();
+}
+/**
+* Add connection to queue of connections to be accepted.
+*/
+public void accept(Connection conn) {
+synchronized (queue) {
+queue.add(conn);
+queue.notify();
+}
+}
+/**
+* Give transport next accepted connection, when available.
+*/
+public void run() {
+Connection conn;
+synchronized (queue) {
+while (queue.size() == 0) {
+try {
+queue.wait();
+} catch (InterruptedException e) {
+}
+}
+startNewAcceptor();
+conn = queue.remove(0);
+}
+transport.handleMessages(conn, true);
+}
+}

changeset 25859	3317bb8137f4
parent 23333	b0af2c7c8c91
child 45333	fd3fdfe96bb4