/*
* @(#)SocketOrChannelConnectionImpl.java 1.91 04/06/21
*
* Copyright 2004 Sun Microsystems, Inc. All rights reserved.
* SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*/
package com.sun.corba.se.impl.transport;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Collections;
import java.util.Hashtable;
import java.util.HashMap;
import java.util.Map;
import org.omg.CORBA.COMM_FAILURE;
import org.omg.CORBA.CompletionStatus;
import org.omg.CORBA.DATA_CONVERSION;
import org.omg.CORBA.INTERNAL;
import org.omg.CORBA.MARSHAL;
import org.omg.CORBA.OBJECT_NOT_EXIST;
import org.omg.CORBA.SystemException;
import com.sun.org.omg.SendingContext.CodeBase;
import com.sun.corba.se.pept.broker.Broker;
import com.sun.corba.se.pept.encoding.InputObject;
import com.sun.corba.se.pept.encoding.OutputObject;
import com.sun.corba.se.pept.protocol.MessageMediator;
import com.sun.corba.se.pept.transport.Acceptor;
import com.sun.corba.se.pept.transport.Connection;
import com.sun.corba.se.pept.transport.ConnectionCache;
import com.sun.corba.se.pept.transport.ContactInfo;
import com.sun.corba.se.pept.transport.EventHandler;
import com.sun.corba.se.pept.transport.InboundConnectionCache;
import com.sun.corba.se.pept.transport.OutboundConnectionCache;
import com.sun.corba.se.pept.transport.ResponseWaitingRoom;
import com.sun.corba.se.pept.transport.Selector;
import com.sun.corba.se.spi.ior.IOR;
import com.sun.corba.se.spi.ior.iiop.GIOPVersion;
import com.sun.corba.se.spi.logging.CORBALogDomains;
import com.sun.corba.se.spi.orb.ORB ;
import com.sun.corba.se.spi.orbutil.threadpool.NoSuchThreadPoolException;
import com.sun.corba.se.spi.orbutil.threadpool.NoSuchWorkQueueException;
import com.sun.corba.se.spi.orbutil.threadpool.Work;
import com.sun.corba.se.spi.protocol.CorbaMessageMediator;
import com.sun.corba.se.spi.transport.CorbaContactInfo;
import com.sun.corba.se.spi.transport.CorbaConnection;
import com.sun.corba.se.spi.transport.CorbaResponseWaitingRoom;
import com.sun.corba.se.spi.transport.ReadTimeouts;
import com.sun.corba.se.impl.encoding.CachedCodeBase;
import com.sun.corba.se.impl.encoding.CDRInputStream_1_0;
import com.sun.corba.se.impl.encoding.CDROutputObject;
import com.sun.corba.se.impl.encoding.CDROutputStream_1_0;
import com.sun.corba.se.impl.encoding.CodeSetComponentInfo;
import com.sun.corba.se.impl.encoding.OSFCodeSetRegistry;
import com.sun.corba.se.impl.logging.ORBUtilSystemException;
import com.sun.corba.se.impl.orbutil.ORBConstants;
import com.sun.corba.se.impl.orbutil.ORBUtility;
import com.sun.corba.se.impl.protocol.giopmsgheaders.Message;
import com.sun.corba.se.impl.protocol.giopmsgheaders.MessageBase;
import com.sun.corba.se.impl.transport.CorbaResponseWaitingRoomImpl;
/**
* @author Harold Carr
*/
public class SocketOrChannelConnectionImpl
extends
EventHandlerBase
implements
CorbaConnection,
Work
{
public static boolean dprintWriteLocks = false;
//
// New transport.
//
protected long enqueueTime;
protected SocketChannel socketChannel;
public SocketChannel getSocketChannel()
{
return socketChannel;
}
// REVISIT:
// protected for test: genericRPCMSGFramework.IIOPConnection constructor.
protected CorbaContactInfo contactInfo;
protected Acceptor acceptor;
protected ConnectionCache connectionCache;
//
// From iiop.Connection.java
//
protected Socket socket; // The socket used for this connection.
protected long timeStamp = 0;
protected boolean isServer = false;
// Start at some value other than zero since this is a magic
// value in some protocols.
protected int requestId = 5;
protected CorbaResponseWaitingRoom responseWaitingRoom;
protected int state;
protected java.lang.Object stateEvent = new java.lang.Object();
protected java.lang.Object writeEvent = new java.lang.Object();
protected boolean writeLocked;
protected int serverRequestCount = 0;
// Server request map: used on the server side of Connection
// Maps request ID to IIOPInputStream.
Map serverRequestMap = null;
// This is a flag associated per connection telling us if the
// initial set of sending contexts were sent to the receiver
// already...
protected boolean postInitialContexts = false;
// Remote reference to CodeBase server (supplies
// FullValueDescription, among other things)
protected IOR codeBaseServerIOR;
// CodeBase cache for this connection. This will cache remote operations,
// handle connecting, and ensure we don't do any remote operations until
// necessary.
protected CachedCodeBase cachedCodeBase = new CachedCodeBase(this);
protected ORBUtilSystemException wrapper ;
// transport read timeout values
protected ReadTimeouts readTimeouts;
protected boolean shouldReadGiopHeaderOnly;
// A message mediator used when shouldReadGiopHeaderOnly is
// true to maintain request message state across execution in a
// SelectorThread and WorkerThread.
protected CorbaMessageMediator partialMessageMediator = null;
// Used in genericRPCMSGFramework test.
protected SocketOrChannelConnectionImpl(ORB orb)
{
this.orb = orb;
wrapper = ORBUtilSystemException.get( orb,
CORBALogDomains.RPC_TRANSPORT ) ;
setWork(this);
responseWaitingRoom = new CorbaResponseWaitingRoomImpl(orb, this);
setReadTimeouts(orb.getORBData().getTransportTCPReadTimeouts());
}
// Both client and servers.
protected SocketOrChannelConnectionImpl(ORB orb,
boolean useSelectThreadToWait,
boolean useWorkerThread)
{
this(orb) ;
setUseSelectThreadToWait(useSelectThreadToWait);
setUseWorkerThreadForEvent(useWorkerThread);
}
// Client constructor.
public SocketOrChannelConnectionImpl(ORB orb,
CorbaContactInfo contactInfo,
boolean useSelectThreadToWait,
boolean useWorkerThread,
String socketType,
String hostname,
int port)
{
this(orb, useSelectThreadToWait, useWorkerThread);
this.contactInfo = contactInfo;
try {
socket = orb.getORBData().getSocketFactory()
.createSocket(socketType,
new InetSocketAddress(hostname, port));
socketChannel = socket.getChannel();
if (socketChannel != null) {
boolean isBlocking = !useSelectThreadToWait;
socketChannel.configureBlocking(isBlocking);
} else {
// IMPORTANT: non-channel-backed sockets must use
// dedicated reader threads.
setUseSelectThreadToWait(false);
}
if (orb.transportDebugFlag) {
dprint(".initialize: connection created: " + socket);
}
} catch (Throwable t) {
throw wrapper.connectFailure(t, socketType, hostname,
Integer.toString(port));
}
state = OPENING;
}
// Client-side convenience.
public SocketOrChannelConnectionImpl(ORB orb,
CorbaContactInfo contactInfo,
String socketType,
String hostname,
int port)
{
this(orb, contactInfo,
orb.getORBData().connectionSocketUseSelectThreadToWait(),
orb.getORBData().connectionSocketUseWorkerThreadForEvent(),
socketType, hostname, port);
}
// Server-side constructor.
public SocketOrChannelConnectionImpl(ORB orb,
Acceptor acceptor,
Socket socket,
boolean useSelectThreadToWait,
boolean useWorkerThread)
{
this(orb, useSelectThreadToWait, useWorkerThread);
this.socket = socket;
socketChannel = socket.getChannel();
if (socketChannel != null) {
// REVISIT
try {
boolean isBlocking = !useSelectThreadToWait;
socketChannel.configureBlocking(isBlocking);
} catch (IOException e) {
RuntimeException rte = new RuntimeException();
rte.initCause(e);
throw rte;
}
}
this.acceptor = acceptor;
serverRequestMap = Collections.synchronizedMap(new HashMap());
isServer = true;
state = ESTABLISHED;
}
// Server-side convenience
public SocketOrChannelConnectionImpl(ORB orb,
Acceptor acceptor,
Socket socket)
{
this(orb, acceptor, socket,
(socket.getChannel() == null
? false
: orb.getORBData().connectionSocketUseSelectThreadToWait()),
(socket.getChannel() == null
? false
: orb.getORBData().connectionSocketUseWorkerThreadForEvent()));
}
////////////////////////////////////////////////////
//
// framework.transport.Connection
//
public boolean shouldRegisterReadEvent()
{
return true;
}
public boolean shouldRegisterServerReadEvent()
{
return true;
}
public boolean read()
{
try {
if (orb.transportDebugFlag) {
dprint(".read->: " + this);
}
CorbaMessageMediator messageMediator = readBits();
if (messageMediator != null) {
// Null can happen when client closes stream
// causing purgecalls.
return dispatch(messageMediator);
}
return true;
} finally {
if (orb.transportDebugFlag) {
dprint(".read<-: " + this);
}
}
}
protected CorbaMessageMediator readBits()
{
try {
if (orb.transportDebugFlag) {
dprint(".readBits->: " + this);
}
MessageMediator messageMediator;
// REVISIT - use common factory base class.
if (contactInfo != null) {
messageMediator =
contactInfo.createMessageMediator(orb, this);
} else if (acceptor != null) {
messageMediator = acceptor.createMessageMediator(orb, this);
} else {
throw
new RuntimeException("SocketOrChannelConnectionImpl.readBits");
}
return (CorbaMessageMediator) messageMediator;
} catch (ThreadDeath td) {
if (orb.transportDebugFlag) {
dprint(".readBits: " + this + ": ThreadDeath: " + td, td);
}
try {
purgeCalls(wrapper.connectionAbort(td), false, false);
} catch (Throwable t) {
if (orb.transportDebugFlag) {
dprint(".readBits: " + this + ": purgeCalls: Throwable: " + t, t);
}
}
throw td;
} catch (Throwable ex) {
if (orb.transportDebugFlag) {
dprint(".readBits: " + this + ": Throwable: " + ex, ex);
}
try {
if (ex instanceof INTERNAL) {
sendMessageError(GIOPVersion.DEFAULT_VERSION);
}
} catch (IOException e) {
if (orb.transportDebugFlag) {
dprint(".readBits: " + this +
": sendMessageError: IOException: " + e, e);
}
}
// REVISIT - make sure reader thread is killed.
orb.getTransportManager().getSelector(0).unregisterForEvent(this);
// Notify anyone waiting.
purgeCalls(wrapper.connectionAbort(ex), true, false);
// REVISIT
//keepRunning = false;
// REVISIT - if this is called after purgeCalls then
// the state of the socket is ABORT so the writeLock
// in close throws an exception. It is ignored but
// causes IBM (screen scraping) tests to fail.
//close();
} finally {
if (orb.transportDebugFlag) {
dprint(".readBits<-: " + this);
}
}
return null;
}
protected CorbaMessageMediator finishReadingBits(MessageMediator messageMediator)
{
try {
if (orb.transportDebugFlag) {
dprint(".finishReadingBits->: " + this);
}
// REVISIT - use common factory base class.
if (contactInfo != null) {
messageMediator =
contactInfo.finishCreatingMessageMediator(orb, this, messageMediator);
} else if (acceptor != null) {
messageMediator =
acceptor.finishCreatingMessageMediator(orb, this, messageMediator);
} else {
throw
new RuntimeException("SocketOrChannelConnectionImpl.finishReadingBits");
}
return (CorbaMessageMediator) messageMediator;
} catch (ThreadDeath td) {
if (orb.transportDebugFlag) {
dprint(".finishReadingBits: " + this + ": ThreadDeath: " + td, td);
}
try {
purgeCalls(wrapper.connectionAbort(td), false, false);
} catch (Throwable t) {
if (orb.transportDebugFlag) {
dprint(".finishReadingBits: " + this + ": purgeCalls: Throwable: " + t, t);
}
}
throw td;
} catch (Throwable ex) {
if (orb.transportDebugFlag) {
dprint(".finishReadingBits: " + this + ": Throwable: " + ex, ex);
}
try {
if (ex instanceof INTERNAL) {
sendMessageError(GIOPVersion.DEFAULT_VERSION);
}
} catch (IOException e) {
if (orb.transportDebugFlag) {
dprint(".finishReadingBits: " + this +
": sendMessageError: IOException: " + e, e);
}
}
// REVISIT - make sure reader thread is killed.
orb.getTransportManager().getSelector(0).unregisterForEvent(this);
// Notify anyone waiting.
purgeCalls(wrapper.connectionAbort(ex), true, false);
// REVISIT
//keepRunning = false;
// REVISIT - if this is called after purgeCalls then
// the state of the socket is ABORT so the writeLock
// in close throws an exception. It is ignored but
// causes IBM (screen scraping) tests to fail.
//close();
} finally {
if (orb.transportDebugFlag) {
dprint(".finishReadingBits<-: " + this);
}
}
return null;
}
protected boolean dispatch(CorbaMessageMediator messageMediator)
{
try {
if (orb.transportDebugFlag) {
dprint(".dispatch->: " + this);
}
//
// NOTE:
//
// This call is the transition from the tranport block
// to the protocol block.
//
boolean result =
messageMediator.getProtocolHandler()
.handleRequest(messageMediator);
return result;
} catch (ThreadDeath td) {
if (orb.transportDebugFlag) {
dprint(".dispatch: ThreadDeath", td );
}
try {
purgeCalls(wrapper.connectionAbort(td), false, false);
} catch (Throwable t) {
if (orb.transportDebugFlag) {
dprint(".dispatch: purgeCalls: Throwable", t);
}
}
throw td;
} catch (Throwable ex) {
if (orb.transportDebugFlag) {
dprint(".dispatch: Throwable", ex ) ;
}
try {
if (ex instanceof INTERNAL) {
sendMessageError(GIOPVersion.DEFAULT_VERSION);
}
} catch (IOException e) {
if (orb.transportDebugFlag) {
dprint(".dispatch: sendMessageError: IOException", e);
}
}
purgeCalls(wrapper.connectionAbort(ex), false, false);
// REVISIT
//keepRunning = false;
} finally {
if (orb.transportDebugFlag) {
dprint(".dispatch<-: " + this);
}
}
return true;
}
public boolean shouldUseDirectByteBuffers()
{
return getSocketChannel() != null;
}
public ByteBuffer read(int size, int offset, int length, long max_wait_time)
throws IOException
{
if (shouldUseDirectByteBuffers()) {
ByteBuffer byteBuffer =
orb.getByteBufferPool().getByteBuffer(size);
if (orb.transportDebugFlag) {
// print address of ByteBuffer gotten from pool
int bbAddress = System.identityHashCode(byteBuffer);
StringBuffer sb = new StringBuffer(80);
sb.append(".read: got ByteBuffer id (");
sb.append(bbAddress).append(") from ByteBufferPool.");
String msgStr = sb.toString();
dprint(msgStr);
}
byteBuffer.position(offset);
byteBuffer.limit(size);
readFully(byteBuffer, length, max_wait_time);
return byteBuffer;
}
byte[] buf = new byte[size];
readFully(getSocket().getInputStream(), buf,
offset, length, max_wait_time);
ByteBuffer byteBuffer = ByteBuffer.wrap(buf);
byteBuffer.limit(size);
return byteBuffer;
}
public ByteBuffer read(ByteBuffer byteBuffer, int offset,
int length, long max_wait_time)
throws IOException
{
int size = offset + length;
if (shouldUseDirectByteBuffers()) {
if (! byteBuffer.isDirect()) {
throw wrapper.unexpectedNonDirectByteBufferWithChannelSocket();
}
if (size > byteBuffer.capacity()) {
if (orb.transportDebugFlag) {
// print address of ByteBuffer being released
int bbAddress = System.identityHashCode(byteBuffer);
StringBuffer bbsb = new StringBuffer(80);
bbsb.append(".read: releasing ByteBuffer id (")
.append(bbAddress).append(") to ByteBufferPool.");
String bbmsg = bbsb.toString();
dprint(bbmsg);
}
orb.getByteBufferPool().releaseByteBuffer(byteBuffer);
byteBuffer = orb.getByteBufferPool().getByteBuffer(size);
}
byteBuffer.position(offset);
byteBuffer.limit(size);
readFully(byteBuffer, length, max_wait_time);
byteBuffer.position(0);
byteBuffer.limit(size);
return byteBuffer;
}
if (byteBuffer.isDirect()) {
throw wrapper.unexpectedDirectByteBufferWithNonChannelSocket();
}
byte[] buf = new byte[size];
readFully(getSocket().getInputStream(), buf,
offset, length, max_wait_time);
return ByteBuffer.wrap(buf);
}
public void readFully(ByteBuffer byteBuffer, int size, long max_wait_time)
throws IOException
{
int n = 0;
int bytecount = 0;
long time_to_wait = readTimeouts.get_initial_time_to_wait();
long total_time_in_wait = 0;
// The reading of data incorporates a strategy to detect a
// rogue client. The strategy is implemented as follows. As
// long as data is being read, at least 1 byte or more, we
// assume we have a well behaved client. If no data is read,
// then we sleep for a time to wait, re-calculate a new time to
// wait which is lengthier than the previous time spent waiting.
// Then, if the total time spent waiting does not exceed a
// maximum time we are willing to wait, we attempt another
// read. If the maximum amount of time we are willing to
// spend waiting for more data is exceeded, we throw an
// IOException.
// NOTE: Reading of GIOP headers are treated with a smaller
// maximum time to wait threshold. Based on extensive
// performance testing, all GIOP headers are being
// read in 1 read access.
do {
bytecount = getSocketChannel().read(byteBuffer);
if (bytecount < 0) {
throw new IOException("End-of-stream");
}
else if (bytecount == 0) {
try {
Thread.sleep(time_to_wait);
total_time_in_wait += time_to_wait;
time_to_wait =
(long)(time_to_wait*readTimeouts.get_backoff_factor());
}
catch (InterruptedException ie) {
// ignore exception
if (orb.transportDebugFlag) {
dprint("readFully(): unexpected exception "
+ ie.toString());
}
}
}
else {
n += bytecount;
}
}
while (n < size && total_time_in_wait < max_wait_time);
if (n < size && total_time_in_wait >= max_wait_time)
{
// failed to read entire message
throw wrapper.transportReadTimeoutExceeded(new Integer(size),
new Integer(n), new Long(max_wait_time),
new Long(total_time_in_wait));
}
getConnectionCache().stampTime(this);
}
// To support non-channel connections.
public void readFully(java.io.InputStream is, byte[] buf,
int offset, int size, long max_wait_time)
throws IOException
{
int n = 0;
int bytecount = 0;
long time_to_wait = readTimeouts.get_initial_time_to_wait();
long total_time_in_wait = 0;
// The reading of data incorporates a strategy to detect a
// rogue client. The strategy is implemented as follows. As
// long as data is being read, at least 1 byte or more, we
// assume we have a well behaved client. If no data is read,
// then we sleep for a time to wait, re-calculate a new time to
// wait which is lengthier than the previous time spent waiting.
// Then, if the total time spent waiting does not exceed a
// maximum time we are willing to wait, we attempt another
// read. If the maximum amount of time we are willing to
// spend waiting for more data is exceeded, we throw an
// IOException.
// NOTE: Reading of GIOP headers are treated with a smaller
// maximum time to wait threshold. Based on extensive
// performance testing, all GIOP headers are being
// read in 1 read access.
do {
bytecount = is.read(buf, offset + n, size - n);
if (bytecount < 0) {
throw new IOException("End-of-stream");
}
else if (bytecount == 0) {
try {
Thread.sleep(time_to_wait);
total_time_in_wait += time_to_wait;
time_to_wait =
(long)(time_to_wait*readTimeouts.get_backoff_factor());
}
catch (InterruptedException ie) {
// ignore exception
if (orb.transportDebugFlag) {
dprint("readFully(): unexpected exception "
+ ie.toString());
}
}
}
else {
n += bytecount;
}
}
while (n < size && total_time_in_wait < max_wait_time);
if (n < size && total_time_in_wait >= max_wait_time)
{
// failed to read entire message
throw wrapper.transportReadTimeoutExceeded(new Integer(size),
new Integer(n), new Long(max_wait_time),
new Long(total_time_in_wait));
}
getConnectionCache().stampTime(this);
}
public void write(ByteBuffer byteBuffer)
throws IOException
{
if (shouldUseDirectByteBuffers()) {
/* NOTE: cannot perform this test. If one ask for a
ByteBuffer from the pool which is bigger than the size
of ByteBuffers managed by the pool, then the pool will
return a HeapByteBuffer.
if (byteBuffer.hasArray()) {
throw wrapper.unexpectedNonDirectByteBufferWithChannelSocket();
}
*/
// IMPORTANT: For non-blocking SocketChannels, there's no guarantee
// all bytes are written on first write attempt.
do {
getSocketChannel().write(byteBuffer);
}
while (byteBuffer.hasRemaining());
} else {
if (! byteBuffer.hasArray()) {
throw wrapper.unexpectedDirectByteBufferWithNonChannelSocket();
}
byte[] tmpBuf = byteBuffer.array();
getSocket().getOutputStream().write(tmpBuf, 0, byteBuffer.limit());
getSocket().getOutputStream().flush();
}
// TimeStamp connection to indicate it has been used
// Note granularity of connection usage is assumed for
// now to be that of a IIOP packet.
getConnectionCache().stampTime(this);
}
/**
* Note:it is possible for this to be called more than once
*/
public synchronized void close()
{
try {
if (orb.transportDebugFlag) {
dprint(".close->: " + this);
}
writeLock();
// REVISIT It will be good to have a read lock on the reader thread
// before we proceed further, to avoid the reader thread (server side)
// from processing requests. This avoids the risk that a new request
// will be accepted by ReaderThread while the ListenerThread is
// attempting to close this connection.
if (isBusy()) { // we are busy!
writeUnlock();
if (orb.transportDebugFlag) {
dprint(".close: isBusy so no close: " + this);
}
return;
}
try {
try {
sendCloseConnection(GIOPVersion.V1_0);
} catch (Throwable t) {
wrapper.exceptionWhenSendingCloseConnection(t);
}
synchronized ( stateEvent ){
state = CLOSE_SENT;
stateEvent.notifyAll();
}
// stop the reader without causing it to do purgeCalls
//Exception ex = new Exception();
//reader.stop(ex); // REVISIT
// NOTE: !!!!!!
// This does writeUnlock().
purgeCalls(wrapper.connectionRebind(), false, true);
} catch (Exception ex) {
if (orb.transportDebugFlag) {
dprint(".close: exception: " + this, ex);
}
}
try {
Selector selector = orb.getTransportManager().getSelector(0);
selector.unregisterForEvent(this);
if (socketChannel != null) {
socketChannel.close();
}
socket.close();
} catch (IOException e) {
if (orb.transportDebugFlag) {
dprint(".close: " + this, e);
}
}
} finally {
if (orb.transportDebugFlag) {
dprint(".close<-: " + this);
}
}
}
public Acceptor getAcceptor()
{
return acceptor;
}
public ContactInfo getContactInfo()
{
return contactInfo;
}
public EventHandler getEventHandler()
{
return this;
}
public OutputObject createOutputObject(MessageMediator messageMediator)
{
// REVISIT - remove this method from Connection and all it subclasses.
throw new RuntimeException("*****SocketOrChannelConnectionImpl.createOutputObject - should not be called.");
}
// This is used by the GIOPOutputObject in order to
// throw the correct error when handling code sets.
// Can we determine if we are on the server side by
// other means? XREVISIT
public boolean isServer()
{
return isServer;
}
public boolean isBusy()
{
if (serverRequestCount > 0 ||
getResponseWaitingRoom().numberRegistered() > 0)
{
return true;
} else {
return false;
}
}
public long getTimeStamp()
{
return timeStamp;
}
public void setTimeStamp(long time)
{
timeStamp = time;
}
public void setState(String stateString)
{
synchronized (stateEvent) {
if (stateString.equals("ESTABLISHED")) {
state = ESTABLISHED;
stateEvent.notifyAll();
} else {
// REVISIT: ASSERT
}
}
}
/**
* Sets the writeLock for this connection.
* If the writeLock is already set by someone else, block till the
* writeLock is released and can set by us.
* IMPORTANT: this connection's lock must be acquired before
* setting the writeLock and must be unlocked after setting the writeLock.
*/
public void writeLock()
{
try {
if (dprintWriteLocks && orb.transportDebugFlag) {
dprint(".writeLock->: " + this);
}
// Keep looping till we can set the writeLock.
while ( true ) {
int localState = state;
switch ( localState ) {
case OPENING:
synchronized (stateEvent) {
if (state != OPENING) {
// somebody has changed 'state' so be careful
break;
}
try {
stateEvent.wait();
} catch (InterruptedException ie) {
if (orb.transportDebugFlag) {
dprint(".writeLock: OPENING InterruptedException: " + this);
}
}
}
// Loop back
break;
case ESTABLISHED:
synchronized (writeEvent) {
if (!writeLocked) {
writeLocked = true;
return;
}
try {
// do not stay here too long if state != ESTABLISHED
// Bug 4752117
while (state == ESTABLISHED && writeLocked) {
writeEvent.wait(100);
}
} catch (InterruptedException ie) {
if (orb.transportDebugFlag) {
dprint(".writeLock: ESTABLISHED InterruptedException: " + this);
}
}
}
// Loop back
break;
//
// XXX
// Need to distinguish between client and server roles
// here probably.
//
case ABORT:
synchronized ( stateEvent ){
if (state != ABORT) {
break;
}
throw wrapper.writeErrorSend() ;
}
case CLOSE_RECVD:
// the connection has been closed or closing
// ==> throw rebind exception
synchronized ( stateEvent ){
if (state != CLOSE_RECVD) {
break;
}
throw wrapper.connectionCloseRebind() ;
}
default:
if (orb.transportDebugFlag) {
dprint(".writeLock: default: " + this);
}
// REVISIT
throw new RuntimeException(".writeLock: bad state");
}
}
} finally {
if (dprintWriteLocks && orb.transportDebugFlag) {
dprint(".writeLock<-: " + this);
}
}
}
public void writeUnlock()
{
try {
if (dprintWriteLocks && orb.transportDebugFlag) {
dprint(".writeUnlock->: " + this);
}
synchronized (writeEvent) {
writeLocked = false;
writeEvent.notify(); // wake up one guy waiting to write
}
} finally {
if (dprintWriteLocks && orb.transportDebugFlag) {
dprint(".writeUnlock<-: " + this);
}
}
}
// Assumes the caller handles writeLock and writeUnlock
public void sendWithoutLock(OutputObject outputObject)
{
// Don't we need to check for CloseConnection
// here? REVISIT
// XREVISIT - Shouldn't the MessageMediator
// be the one to handle writing the data here?
try {
// Write the fragment/message
CDROutputObject cdrOutputObject = (CDROutputObject) outputObject;
cdrOutputObject.writeTo(this);
// REVISIT - no flush?
//socket.getOutputStream().flush();
} catch (IOException e1) {
/*
* ADDED(Ram J) 10/13/2000 In the event of an IOException, try
* sending a CancelRequest for regular requests / locate requests
*/
// Since IIOPOutputStream's msgheader is set only once, and not
// altered during sending multiple fragments, the original
// msgheader will always have the requestId.
// REVISIT This could be optimized to send a CancelRequest only
// if any fragments had been sent already.
/* REVISIT: MOVE TO SUBCONTRACT
Message msg = os.getMessage();
if (msg.getType() == Message.GIOPRequest ||
msg.getType() == Message.GIOPLocateRequest) {
GIOPVersion requestVersion = msg.getGIOPVersion();
int requestId = MessageBase.getRequestId(msg);
try {
sendCancelRequest(requestVersion, requestId);
} catch (IOException e2) {
// most likely an abortive connection closure.
// ignore, since nothing more can be done.
if (orb.transportDebugFlag) {
}
}
*/
// REVISIT When a send failure happens, purgeCalls() need to be
// called to ensure that the connection is properly removed from
// further usage (ie., cancelling pending requests with COMM_FAILURE
// with an appropriate minor_code CompletionStatus.MAY_BE).
// Relying on the IIOPOutputStream (as noted below) is not
// sufficient as it handles COMM_FAILURE only for the final
// fragment (during invoke processing). Note that COMM_FAILURE could
// happen while sending the initial fragments.
// Also the IIOPOutputStream does not properly close the connection.
// It simply removes the connection from the table. An orderly
// closure is needed (ie., cancel pending requests on the connection
// COMM_FAILURE as well.
// IIOPOutputStream will cleanup the connection info when it
// sees this exception.
throw wrapper.writeErrorSend(e1) ;
}
}
public void registerWaiter(MessageMediator messageMediator)
{
responseWaitingRoom.registerWaiter(messageMediator);
}
public void unregisterWaiter(MessageMediator messageMediator)
{
responseWaitingRoom.unregisterWaiter(messageMediator);
}
public InputObject waitForResponse(MessageMediator messageMediator)
{
return responseWaitingRoom.waitForResponse(messageMediator);
}
public void setConnectionCache(ConnectionCache connectionCache)
{
this.connectionCache = connectionCache;
}
public ConnectionCache getConnectionCache()
{
return connectionCache;
}
////////////////////////////////////////////////////
//
// EventHandler methods
//
public void setUseSelectThreadToWait(boolean x)
{
useSelectThreadToWait = x;
// REVISIT - Reading of a GIOP header only is information
// that should be passed into the constructor
// from the SocketOrChannelConnection factory.
setReadGiopHeaderOnly(shouldUseSelectThreadToWait());
}
public void handleEvent()
{
if (orb.transportDebugFlag) {
dprint(".handleEvent->: " + this);
}
getSelectionKey().interestOps(getSelectionKey().interestOps() &
(~ getInterestOps()));
if (shouldUseWorkerThreadForEvent()) {
Throwable throwable = null;
try {
int poolToUse = 0;
if (shouldReadGiopHeaderOnly()) {
partialMessageMediator = readBits();
poolToUse =
partialMessageMediator.getThreadPoolToUse();
}
if (orb.transportDebugFlag) {
dprint(".handleEvent: addWork to pool: " + poolToUse);
}
orb.getThreadPoolManager().getThreadPool(poolToUse)
.getWorkQueue(0).addWork(getWork());
} catch (NoSuchThreadPoolException e) {
throwable = e;
} catch (NoSuchWorkQueueException e) {
throwable = e;
}
// REVISIT: need to close connection.
if (throwable != null) {
if (orb.transportDebugFlag) {
dprint(".handleEvent: " + throwable);
}
INTERNAL i = new INTERNAL("NoSuchThreadPoolException");
i.initCause(throwable);
throw i;
}
} else {
if (orb.transportDebugFlag) {
dprint(".handleEvent: doWork");
}
getWork().doWork();
}
if (orb.transportDebugFlag) {
dprint(".handleEvent<-: " + this);
}
}
public SelectableChannel getChannel()
{
return socketChannel;
}
public int getInterestOps()
{
return SelectionKey.OP_READ;
}
// public Acceptor getAcceptor() - already defined above.
public Connection getConnection()
{
return this;
}
////////////////////////////////////////////////////
//
// Work methods.
//
public String getName()
{
return this.toString();
}
public void doWork()
{
try {
if (orb.transportDebugFlag) {
dprint(".doWork->: " + this);
}
// IMPORTANT: Sanity checks on SelectionKeys such as
// SelectorKey.isValid() should not be done
// here.
//
if (!shouldReadGiopHeaderOnly()) {
read();
}
else {
// get the partialMessageMediator
// created by SelectorThread
CorbaMessageMediator messageMediator =
this.getPartialMessageMediator();
// read remaining info needed in a MessageMediator
messageMediator = finishReadingBits(messageMediator);
if (messageMediator != null) {
// Null can happen when client closes stream
// causing purgecalls.
dispatch(messageMediator);
}
}
} catch (Throwable t) {
if (orb.transportDebugFlag) {
dprint(".doWork: ignoring Throwable: "
+ t
+ " " + this);
}
} finally {
if (orb.transportDebugFlag) {
dprint(".doWork<-: " + this);
}
}
}
public void setEnqueueTime(long timeInMillis)
{
enqueueTime = timeInMillis;
}
public long getEnqueueTime()
{
return enqueueTime;
}
////////////////////////////////////////////////////
//
// spi.transport.CorbaConnection.
//
// IMPORTANT: Reader Threads must NOT read Giop header only.
public boolean shouldReadGiopHeaderOnly() {
return shouldReadGiopHeaderOnly;
}
protected void setReadGiopHeaderOnly(boolean shouldReadHeaderOnly) {
shouldReadGiopHeaderOnly = shouldReadHeaderOnly;
}
public ResponseWaitingRoom getResponseWaitingRoom()
{
return responseWaitingRoom;
}
// REVISIT - inteface defines isServer but already defined in
// higher interface.
public void serverRequestMapPut(int requestId,
CorbaMessageMediator messageMediator)
{
serverRequestMap.put(new Integer(requestId), messageMediator);
}
public CorbaMessageMediator serverRequestMapGet(int requestId)
{
return (CorbaMessageMediator)
serverRequestMap.get(new Integer(requestId));
}
public void serverRequestMapRemove(int requestId)
{
serverRequestMap.remove(new Integer(requestId));
}
// REVISIT: this is also defined in:
// com.sun.corba.se.spi.legacy.connection.Connection
public java.net.Socket getSocket()
{
return socket;
}
/** It is possible for a Close Connection to have been
** sent here, but we will not check for this. A "lazy"
** Exception will be thrown in the Worker thread after the
** incoming request has been processed even though the connection
** is closed before the request is processed. This is o.k because
** it is a boundary condition. To prevent it we would have to add
** more locks which would reduce performance in the normal case.
**/
public synchronized void serverRequestProcessingBegins()
{
serverRequestCount++;
}
public synchronized void serverRequestProcessingEnds()
{
serverRequestCount--;
}
//
//
//
public synchronized int getNextRequestId()
{
return requestId++;
}
// Negotiated code sets for char and wchar data
protected CodeSetComponentInfo.CodeSetContext codeSetContext = null;
public ORB getBroker()
{
return orb;
}
public CodeSetComponentInfo.CodeSetContext getCodeSetContext() {
// Needs to be synchronized for the following case when the client
// doesn't send the code set context twice, and we have two threads
// in ServerRequestDispatcher processCodeSetContext.
//
// Thread A checks to see if there is a context, there is none, so
// it calls setCodeSetContext, getting the synch lock.
// Thread B checks to see if there is a context. If we didn't synch,
// it might decide to outlaw wchar/wstring.
if (codeSetContext == null) {
synchronized(this) {
return codeSetContext;
}
}
return codeSetContext;
}
public synchronized void setCodeSetContext(CodeSetComponentInfo.CodeSetContext csc) {
// Double check whether or not we need to do this
if (codeSetContext == null) {
if (OSFCodeSetRegistry.lookupEntry(csc.getCharCodeSet()) == null ||
OSFCodeSetRegistry.lookupEntry(csc.getWCharCodeSet()) == null) {
// If the client says it's negotiated a code set that
// isn't a fallback and we never said we support, then
// it has a bug.
throw wrapper.badCodesetsFromClient() ;
}
codeSetContext = csc;
}
}
//
// from iiop.IIOPConnection.java
//
// Map request ID to an InputObject.
// This is so the client thread can start unmarshaling
// the reply and remove it from the out_calls map while the
// ReaderThread can still obtain the input stream to give
// new fragments. Only the ReaderThread touches the clientReplyMap,
// so it doesn't incur synchronization overhead.
public MessageMediator clientRequestMapGet(int requestId)
{
return responseWaitingRoom.getMessageMediator(requestId);
}
protected MessageMediator clientReply_1_1;
public void clientReply_1_1_Put(MessageMediator x)
{
clientReply_1_1 = x;
}
public MessageMediator clientReply_1_1_Get()
{
return clientReply_1_1;
}
public void clientReply_1_1_Remove()
{
clientReply_1_1 = null;
}
protected MessageMediator serverRequest_1_1;
public void serverRequest_1_1_Put(MessageMediator x)
{
serverRequest_1_1 = x;
}
public MessageMediator serverRequest_1_1_Get()
{
return serverRequest_1_1;
}
public void serverRequest_1_1_Remove()
{
serverRequest_1_1 = null;
}
protected String getStateString( int state )
{
synchronized ( stateEvent ){
switch (state) {
case OPENING : return "OPENING" ;
case ESTABLISHED : return "ESTABLISHED" ;
case CLOSE_SENT : return "CLOSE_SENT" ;
case CLOSE_RECVD : return "CLOSE_RECVD" ;
case ABORT : return "ABORT" ;
default : return "???" ;
}
}
}
public synchronized boolean isPostInitialContexts() {
return postInitialContexts;
}
// Can never be unset...
public synchronized void setPostInitialContexts(){
postInitialContexts = true;
}
/**
* Wake up the outstanding requests on the connection, and hand them
* COMM_FAILURE exception with a given minor code.
*
* Also, delete connection from connection table and
* stop the reader thread.
* Note that this should only ever be called by the Reader thread for
* this connection.
*
* @param minor_code The minor code for the COMM_FAILURE major code.
* @param die Kill the reader thread (this thread) before exiting.
*/
public void purgeCalls(SystemException systemException,
boolean die, boolean lockHeld)
{
int minor_code = systemException.minor;
try{
if (orb.transportDebugFlag) {
dprint(".purgeCalls->: "
+ minor_code + "/" + die + "/" + lockHeld
+ " " + this);
}
// If this invocation is a result of ThreadDeath caused
// by a previous execution of this routine, just exit.
synchronized ( stateEvent ){
if ((state == ABORT) || (state == CLOSE_RECVD)) {
if (orb.transportDebugFlag) {
dprint(".purgeCalls: exiting since state is: "
+ getStateString(state)
+ " " + this);
}
return;
}
}
// Grab the writeLock (freeze the calls)
try {
if (!lockHeld) {
writeLock();
}
} catch (SystemException ex) {
if (orb.transportDebugFlag)
dprint(".purgeCalls: SystemException" + ex
+ "; continuing " + this);
}
// Mark the state of the connection
// and determine the request status
org.omg.CORBA.CompletionStatus completion_status;
synchronized ( stateEvent ){
if (minor_code == ORBUtilSystemException.CONNECTION_REBIND) {
state = CLOSE_RECVD;
systemException.completed = CompletionStatus.COMPLETED_NO;
} else {
state = ABORT;
systemException.completed = CompletionStatus.COMPLETED_MAYBE;
}
stateEvent.notifyAll();
}
try {
socket.getInputStream().close();
socket.getOutputStream().close();
socket.close();
} catch (Exception ex) {
if (orb.transportDebugFlag) {
dprint(".purgeCalls: Exception closing socket: " + ex
+ " " + this);
}
}
// Signal all threads with outstanding requests on this
// connection and give them the SystemException;
responseWaitingRoom.signalExceptionToAllWaiters(systemException);
if (contactInfo != null) {
((OutboundConnectionCache)getConnectionCache()).remove(contactInfo);
} else if (acceptor != null) {
((InboundConnectionCache)getConnectionCache()).remove(this);
}
//
// REVISIT: Stop the reader thread
//
// Signal all the waiters of the writeLock.
// There are 4 types of writeLock waiters:
// 1. Send waiters:
// 2. SendReply waiters:
// 3. cleanUp waiters:
// 4. purge_call waiters:
//
writeUnlock();
} finally {
if (orb.transportDebugFlag) {
dprint(".purgeCalls<-: "
+ minor_code + "/" + die + "/" + lockHeld
+ " " + this);
}
}
}
/*************************************************************************
* The following methods are for dealing with Connection cleaning for
* better scalability of servers in high network load conditions.
**************************************************************************/
public void sendCloseConnection(GIOPVersion giopVersion)
throws IOException
{
Message msg = MessageBase.createCloseConnection(giopVersion);
sendHelper(giopVersion, msg);
}
public void sendMessageError(GIOPVersion giopVersion)
throws IOException
{
Message msg = MessageBase.createMessageError(giopVersion);
sendHelper(giopVersion, msg);
}
/**
* Send a CancelRequest message. This does not lock the connection, so the
* caller needs to ensure this method is called appropriately.
* @exception IOException - could be due to abortive connection closure.
*/
public void sendCancelRequest(GIOPVersion giopVersion, int requestId)
throws IOException
{
Message msg = MessageBase.createCancelRequest(giopVersion, requestId);
sendHelper(giopVersion, msg);
}
protected void sendHelper(GIOPVersion giopVersion, Message msg)
throws IOException
{
// REVISIT: See comments in CDROutputObject constructor.
CDROutputObject outputObject =
new CDROutputObject((ORB)orb, null, giopVersion, this, msg,
ORBConstants.STREAM_FORMAT_VERSION_1);
msg.write(outputObject);
outputObject.writeTo(this);
}
public void sendCancelRequestWithLock(GIOPVersion giopVersion,
int requestId)
throws IOException
{
writeLock();
try {
sendCancelRequest(giopVersion, requestId);
} finally {
writeUnlock();
}
}
// Begin Code Base methods ---------------------------------------
//
// Set this connection's code base IOR. The IOR comes from the
// SendingContext. This is an optional service context, but all
// JavaSoft ORBs send it.
//
// The set and get methods don't need to be synchronized since the
// first possible get would occur during reading a valuetype, and
// that would be after the set.
// Sets this connection's code base IOR. This is done after
// getting the IOR out of the SendingContext service context.
// Our ORBs always send this, but it's optional in CORBA.
public final void setCodeBaseIOR(IOR ior) {
codeBaseServerIOR = ior;
}
public final IOR getCodeBaseIOR() {
return codeBaseServerIOR;
}
// Get a CodeBase stub to use in unmarshaling. The CachedCodeBase
// won't connect to the remote codebase unless it's necessary.
public final CodeBase getCodeBase() {
return cachedCodeBase;
}
// End Code Base methods -----------------------------------------
// set transport read thresholds
protected void setReadTimeouts(ReadTimeouts readTimeouts) {
this.readTimeouts = readTimeouts;
}
protected void setPartialMessageMediator(CorbaMessageMediator messageMediator) {
partialMessageMediator = messageMediator;
}
protected CorbaMessageMediator getPartialMessageMediator() {
return partialMessageMediator;
}
public String toString()
{
synchronized ( stateEvent ){
return
"SocketOrChannelConnectionImpl[" + " "
+ (socketChannel == null ?
socket.toString() : socketChannel.toString()) + " "
+ getStateString( state ) + " "
+ shouldUseSelectThreadToWait() + " "
+ shouldUseWorkerThreadForEvent() + " "
+ shouldReadGiopHeaderOnly()
+ "]" ;
}
}
// Must be public - used in encoding.
public void dprint(String msg)
{
ORBUtility.dprint("SocketOrChannelConnectionImpl", msg);
}
protected void dprint(String msg, Throwable t)
{
dprint(msg);
t.printStackTrace(System.out);
}
}
// End of file.
|
