Control.java
/*
** Module : Control.java
** Abstract : tracks control flow between client and server
**
** Copyright (c) 2006-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ------------------Description-------------------
** 001 NVS 20060208 @24362 Initial version.
** 002 NVS 20060209 @24466 Added delayed interruption processing.
** 003 NVS 20060210 @24570 Added server side processing.
** 004 NVS 20060213 @24571 Added unconditional push and pop operations.
** 005 NVS 20060220 @24620 Protected against premature calls.
** 006 NVS 20060222 @24624 Protected against duplicate exceptions.
** 007 GES 20060727 @28207 Performance enhancements. This code is
** called for every synchronous message send
** and receive so it needs to be very thin!
** 008 GES 20061019 @30537 Major rewrite to simplify and fix many
** issues. Synchronization was fixed (it was
** mixed use between static and non-static
** "synchronized" methods, which is wrong
** because statics use the Class object's
** monitor and instances use the instance
** object's monitor. Removed the message ID
** usage and the overcomplicated implementation
** where message IDs and thread objects were
** mixed in the stack. Removed some unneeded
** methods. Rewrote interrupt processing to
** operate differently based on the success
** or failure of any remote interrupt
** notification. Added code to honor any
** pending thread interruption that was already
** delivered to a given thread. Added
** documentation. Fixed a re-initialization
** issue that appears at any re-logon.
** 009 GES 20070410 @32914 Provided a global lock to eliminate a
** possible (though unlikely) race condition
** when handling remote interrupt requests on
** the server side.
** 010 ECF 20071106 @35877 Refactored net package. Replaced Queue with
** Session. Implemented generics. Other minor
** cleanup.
** 011 CA 20090831 @43795 Fixed interruption processing when both peers
** are in remote mode.
** 012 CA 20090901 @43810 Now the InterruptHandler implementation is
** retrieved from the SessionManager. Major
** changes to interrupt handling, to fix the
** interruption mechanism when a remote server
** request is active. Also, now virtual controls are
** multiplexed as the virtual sessions are.
** 013 CA 20090902 @43820 Fixed a problem with setRightSideControl, when
** the passed instance is the same as the Control
** instance which receives it (regression introduced
** by H012).
** 014 CA 20091021 @44180 Fixed some server-to-server interruption issues,
** exposed by the changes in SessionManager.endSession
** found @44176. The issues are related to client
** abend while a virtual session is executing.
** 015 CA 20091106 @44319 Added critical section protection against thread
** interruption. This ensures that the Conversation
** thread can't be interrupted while a critical
** section is being executed.
** 016 CA 20140513 Added a weight for the context-local var, to ensure predetermined
** order during context reset.
** 017 CA 20170502 Thread.interrupted() in initControl is called only if this is the
** default Control instance (for the client or current context,
** on server-side).
** 018 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 019 ES 20240513 Added getCurrentContext method to get context local information for
** running thread.
** Added stopAfterInterrupt method for setting context local with the control
** of a Conversation thread. And and performing a interrupt caused a stop-after
** timeout.
** ES 20240620 Added serverInterrupt method to insert Ctrl-C into TypeAhead queue.
** 020 ES 20250221 Added interrupt(Control ct) method to be able to give custom control object
** for interruption.
** 20250224 Remove setting up the serverInstance ContextLocal variable to a custom Control
** object.
** 20250407 Added getTargetThreadStatus and targetThreadStatus to get the current state of
** target thread.
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program 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 Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.net;
import java.util.*;
import java.util.logging.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.ui.client.TypeAhead;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* Tracks the state of the "conversation" between the two sides of the
* networking session and provides a local and remote delivery mechanism
* for notification of interruptions. An interruption that occurs during
* local processing is delivered by marking the given thread as interrupted.
* Any code that checks this flag (e.g. <code>Object.wait()</code>) will
* process accordingly (e.g. throw an <code>InterruptedException</code>).
* Any interruption that occurs during remote processing is delivered
* to the other end of the session asynchronous to the "conversation". In
* other words, this is an "out of band" notification attempt. As long as
* processing is still local to the other side, the interruption will be
* raised there, as a local exception. Any failure to deliver the
* interruption on the remote side (e.g. the remote side called back to this
* local side before the interruption could be delivered) will cause the
* interruption to be honored at the moment that this side next shifts into
* local processing mode.
* <p>
* This class maintains a stack of thread objects so that the current thread
* of execution on the local side is known. This is only used when processing
* is local.
* <p>
* Initially the stack is empty. The first call to the {@link #init} method
* implicitly pushes the first thread into the stack.
* <p>
* The current point of execution for the conversation is tracked between the
* remote side and the local side. Local execution is active in the
* following cases:
* <p>
* <ol>
* <li> Immediately after initialization by <code>init()</code>.
* <li> Before any call to {@link #pushRemoteCall} and after the
* corresponding call to {@link #popRemoteCall}. These calls
* form a "bracket" around remote processing.
* <li> During a remote processing bracket, if the remote side recursively
* calls back into the local side, there will be a bracketed pair of
* notifications. After any call to {@link #pushUnconditionally}
* and before the correspondng call to {@link #popUnconditionally},
* processing is local.
* <li> Because a call to the other side of the conversation can occur
* at any time, there is no built-in limit to recursion. For this
* reason, it is perfectly possible that the local side will
* recursively call back to the remote side before completing an
* "unconditional" bracket. Likewise the remote side may call into
* the local side before returning from a remote processing bracket.
* Thus the state of local processing is directly determined by the
* last call to any one of the above 4 bracketing calls.
* </ol>
* <p>
* To notify this class that an interruption needs to be delivered, use
* {@link #interrupt}.
*/
public final class Control
{
/** Logger */
private static final CentralLogger LOG =
CentralLogger.get(Control.class.getName());
/** The generating side for asynchronous interruptions if non-null. */
private static Control clientInstance = null;
/** The receiving side of asynchronous interruptions if non-null. */
private static ContextLocal<Control> serverInstance = null;
static
{
if (SessionManager.get().isLeaf())
{
clientInstance = new Control();
}
else
{
serverInstance = new ContextLocal<Control>()
{
@Override
public WeightFactor getWeight()
{
return WeightFactor.LAST;
}
protected Control initialValue()
{
return (new Control());
}
};
}
}
/** The currently ongoing operations (requests or replies). */
private final Stack<Operation> operations = new Stack<Operation>();
/** List of remote outstanding reply ids. */
private final Set<Integer> remotePendingReplies = new HashSet<Integer>();
/**
* The operation being processed on the other peer. This is used only
* for interruption processing.
*/
private Operation remoteOperation = null;
/** Initialized flag. */
private boolean ready = false;
/** The connection for this P2J session. */
private Session session = null;
/** Which side owns control now: local (true) or remote (false). */
private Boolean local = false;
/** <code>true</code> if an interruption signal has occurred. */
private boolean interrupted = false;
/** <code>true</code> if currently processing an asynchronous signal. */
private boolean async = false;
/** The control stack. */
private Stack<Thread> control = new Stack<Thread>();
/** Global lock owner or <code>null</code> if not locked. */
private Thread lock = null;
/**
* The Control instance for the session on the left-side of this Control.
* If this is not null, then the {@link #rightSideControl} must be null.
*/
private Control leftSideControl = null;
/**
* The Control instance for the session on the right-side of this Control.
* If this is not null, then the {@link #leftSideControl} must be null.
*/
private Control rightSideControl = null;
/**
* Constructor with restricted access (disallows external construction).
*/
private Control()
{
}
/**
* Initializes the control stack.
*
* @param session
* The connection for this context.
*/
public static synchronized void init(Session session)
{
Control ct = locate(0);
ct.initControl(session);
}
/**
* Initializes the control stack.
*
* @param session
* The connection for this context.
*/
synchronized void initControl(Session session)
{
// avoid duplicate initialization
if (this.ready && this.session == session)
return;
// this may occur more than once (if we are in re-logon mode)
this.local = true;
this.control.clear();
this.control.push(Thread.currentThread());
this.session = session;
if (this == locate(0))
{
Thread.interrupted();
}
this.ready = true;
}
/**
* Clears the pending interrupt flag. Used whenever an
* <code>InterruptedException</code> is caught to avoid duplicate
* exceptions.
*/
public static void clearInterrupted()
{
Control ct = locate(0);
// since this is a static method, the method-level synch would be on
// the CLASS, but we must synch on the INSTANCE before testing or
// modifying its state
synchronized (ct)
{
ct.interrupted = false;
}
}
/**
* Gives the context local information of a currently running thread.
*
* @return Context local information.
*/
public static Control getCurrentContext()
{
return serverInstance.get();
}
/**
* Sets the context local to a control of a Conversation thread.
* Interrupts a thread if it is local execution on server or
* sends a INTERRUPT_SESSION message on client side.
*
* @param control
* Control related to a Conversation thread.
*/
public static void stopAfterInterrupt(Control control)
{
if (control.local)
{
Control.interrupt(control);
}
else
{
control.sentRemoteInterrupt();
}
}
/**
* Method sends a async request to client side with
* SERVER_INTERRUPT message.
*
*/
public void sentRemoteInterrupt()
{
RoutingKey key = new RoutingKey(RoutingKey.SERVER_INTERRUPT);
Message message = new Message(key, new Object[] {});
try
{
// mark this control instance as "in the middle of"
// processing an asynchronous notification
async = true;
message.setType(MessageTypes.REQUEST_ASYNCH);
session.forward(message);
}
catch (Exception exc)
{
// check if the cause is a InterruptedException. in this case,
// do not log it as a warning
Level logLvl = Level.WARNING;
String msg = "Error interrupting virtual session peer.";
Throwable cause = exc;
while (cause != null)
{
if (cause instanceof InterruptedException)
{
logLvl = Level.FINE;
break;
}
if (cause instanceof SilentUnwindException)
{
msg = "Control.interruptImpl(boolean)";
logLvl = Level.FINE;
break;
}
cause = cause.getCause();
}
if (LOG.isLoggable(logLvl))
{
LOG.log(logLvl, msg, exc);
}
}
finally
{
// clear the asynchronous notification mode flag
async = false;
}
}
/**
* Trigger an interrupt cause by a stop-after timeout, by inserting
* a Ctrl-C into TypeAhead queue.
*/
public static void triggerServerInterrupt()
{
TypeAhead typeAhead = TypeAhead.instance();
if (typeAhead != null)
{
typeAhead.triggerServerInterrupt();
}
}
/**
* Interrupts the execution of the conversation as the result of an
* asynchronous interruption notification (usually by an end-user).
* <p>
* Since this call is made on a different thread, the target thread has to
* be interrupted.
* <p>
* If by the time this call is made control is with the remote side AND
* the original interrupt request was generated on this node, a remote
* call is made to interrupt the conversation there.
*
* @return <code>true</code> if the interrupt request was successfully
* delivered.
*/
public static boolean interrupt()
{
Control ct = locate(0);
return interrupt(ct);
}
/**
* Gets the state of the target thread.
*
* @return See above.
*/
public static Thread.State getTargetThreadStatus()
{
Control ct = locate(0);
return ct.targetThreadStatus();
}
/**
* Gets the state of the target thread.
*
* @return See above.
*/
public Thread.State targetThreadStatus()
{
Thread t = getTargetThread();
return t.getState();
}
/**
* Interrupts the execution of the conversation as the result of an
* asynchronous interruption notification (usually by an end-user or
* a stop-after time-out).
* <p>
* Since this call is made on a different thread, the target thread has to
* be interrupted.
* <p>
* If by the time this call is made control is with the remote side AND
* the original interrupt request was generated on this node, a remote
* call is made to interrupt the conversation there.
* @param ct
* Custom Control object for which the interruption is performed.
* @return <code>true</code> if the interrupt request was successfully
* delivered.
*/
static boolean interrupt(Control ct)
{
boolean delivered = false;
synchronized (ct)
{
// this block must be synchronized on the "ct" object, to be sure
// the rightSideControl can't be nullified between the not-null test
// and its usage
if (ct.rightSideControl != null)
{
// first, we dispatch the interrupt to the right-side
// session (i.e. local_server -to-> remote_server, if the session
// wasn't shutdown yet); if this doesn't process it, try to handle
// it locally.
delivered = ct.rightSideControl.session.isRunning() &&
ct.rightSideControl.interruptImpl(true);
}
}
// if the right-side didn't accept the interruption request, we must
// handle it here, on the "central" node.
return delivered || ct.interruptImpl(false);
}
/**
* Interrupts the execution of the conversation as the result of an
* asynchronous interruption notification (usually by an end-user).
* <p>
* Since this call is made on a different thread, the target thread has to
* be interrupted.
* <p>
* If by the time this call is made control is with the remote side AND
* the original interrupt request was generated on this node, a remote
* call is made to interrupt the conversation there.
*
* @param rightSide
* <code>true</code> if the interruption is delivered to the
* right-side relative to the control for the current context.
*
* @return <code>true</code> if the interrupt request was successfully
* delivered.
*/
synchronized boolean interruptImpl(boolean rightSide)
{
boolean delivered = false;
if (ready)
{
interrupted = true;
if (local)
{
// interrupt local execution
Thread t = getTargetThread();
if (t != null)
{
CriticalSectionManager.interrupt(t);
delivered = true;
interrupted = false;
}
}
else
{
// the server side doesn't generate interruptions on the
// client
if (clientInstance == null && !rightSide)
{
// if nothing in the stack or the remote operation is not set,
// it means this interruption didn't come from the remote peer
if (operations.isEmpty() || remoteOperation == null)
{
return false;
}
Operation current = operations.peek();
Operation remote = remoteOperation;
boolean interrupt = false;
if (!remote.request &&
current.request &&
remote.operationId == current.operationId)
{
// case a: CL request == false,
// SRV request == true,
// CL (reply) id == SRV (request) id
interrupt = true;
}
if (remote.request)
{
Operation test = new Operation(false, remote.operationId);
// case b: CL request == true
// SRV request can be true or false
// CL request id is not in the list of
// outstanding SRV reply ids
interrupt = !operations.contains(test);
}
if (remote.request &&
!current.request &&
remote.operationId == current.operationId)
{
// case c: CL request == true
// SRV request == false
// CL (request) id == SRV (reply) id
interrupt = false;
}
if (current.request)
{
// case d: CL request can be true or false
// SRV request == true
// SRV request id is not in the list of
// outstanding CL reply ids
interrupt = !remotePendingReplies
.contains(current.operationId);
}
this.interrupted = interrupt;
return interrupt;
}
// interrupt the other side of the remote connection
RoutingKey key = new RoutingKey(RoutingKey.INTERRUPT_SESSION);
Operation current = operations.peek();
// collect all the unprocessed replies to send them
List<Integer> repliesList = new ArrayList<Integer>();
for (int i = 0; i < operations.size(); i++)
{
Operation operation = operations.get(i);
if (!operation.request)
{
repliesList.add(operation.operationId);
}
}
int[] replies = Utils.integerCollectionToPrimitive(repliesList);
Message message =
new Message(key,
new Object[] { current.request,
current.operationId,
replies });
try
{
// mark this control instance as "in the middle of"
// processing an asynchronous notification
async = true;
Boolean del = (Boolean) session.transact(message, 0);
delivered = (del != null) ? del.booleanValue() : false;
if (delivered)
{
// no further action locally
interrupted = false;
}
}
catch (Exception exc)
{
// check if the cause is a InterruptedException. in this case,
// do not log it as a warning
Level logLvl = Level.WARNING;
String msg = "Error interrupting virtual session peer.";
Throwable cause = exc;
while (cause != null)
{
if (cause instanceof InterruptedException)
{
logLvl = Level.FINE;
break;
}
if (cause instanceof SilentUnwindException)
{
msg = "Control.interruptImpl(boolean)";
logLvl = Level.FINE;
break;
}
cause = cause.getCause();
}
if (LOG.isLoggable(logLvl))
{
LOG.log(logLvl, msg, exc);
}
}
finally
{
// clear the asynchronous notification mode flag
async = false;
}
}
}
return delivered;
}
/**
* Get the session for the specified context.
*
* @param contextID
* The context to which the interruption must be applied.
*
* @return the P2J session.
*/
static Session getSession(int contextID)
{
return locate(contextID).session;
}
/**
* Return the {@link #ready} state of this control.
*
* @return <code>true</code> if this control is {@link #ready}.
*/
boolean isReady()
{
return ready;
}
/**
* Save the state of the peer. This means saving the operation type
* (request or reply), the operation ID and the peer's outstanding replies.
*
* @param request
* The remote operation type. <code>true</code> for request.
* @param operationId
* The remote operation ID.
* @param pendingReplies
* The list of remote outstanding replies.
*/
synchronized void setRemoteOperation(boolean request,
int operationId,
int[] pendingReplies)
{
remoteOperation = new Operation(request, operationId);
remotePendingReplies.clear();
for (int i = 0; i < pendingReplies.length; i++)
{
remotePendingReplies.add(pendingReplies[i]);
}
}
/**
* Clear the data about the remote operation.
*/
synchronized void clearRemoteOperation()
{
remoteOperation = null;
remotePendingReplies.clear();
}
/**
* Push a new operation to the stack, saving its type and its ID.
*
* @param request
* <code>true</code> if this is a requests, <code>false</code> for
* a reply
* @param operationId
* the ID of the current operation being executed.
*/
private void pushOperation(boolean request, int operationId)
{
operations.push(new Operation(request, operationId));
}
/**
* Pop the operation stack. Invoked after the request or reply is finished
* processing and the control is switched to the other side.
*/
private void popOperation()
{
operations.pop();
}
/**
* Obtain the global lock on the session's instance. This will block until
* the lock is obtained. Once obtained, no use of the session's instance
* can be made from other threads until the lock is released with
* {@link #unlock}.
*/
synchronized void lock()
{
lockCore();
if (leftSideControl != null)
{
leftSideControl.lock();
}
}
/**
* Release the global lock on the session's instance. This will allow all
* other threads to use the session's instance as soon as this method
* returns.
*/
synchronized void unlock()
{
unlockCore();
if (leftSideControl != null)
{
leftSideControl.unlock();
}
}
/**
* Switches control to the remote side. If an interrupt is pending
* ({@link #interrupt} has been issued but not yet honored), then this
* method calls <code>handleInterrupted()</code>.
* <p>
* This method will block until it can obtain the global lock on this
* instance. Before returning, that lock will be released.
* <p>
* If this instance is in asynchronous notification mode, then this method
* is a NOP. Otherwise the state of the instance would change during the
* asynchronous notification, which would be improper since such
* notifications are by nature supposed to be "out of band". Thus state
* changes for the normal communication channels must be unaffected.
*
* @param request
* the ID of the current request being executed.
*/
synchronized void pushRemoteCall(int request)
{
lockCore();
try
{
if (!ready || !local || async)
{
// we must push here too, to avoid unbalancing the stack; reason:
// popRemoteCall will call popOperation in its finally clause
pushOperation(true, request);
return;
}
// CASE 1: if the local interrupt processing interrupted the thread
// but the processing on that thread did not honor the thread's
// interrupted status for any reason, we honor it here
// CASE 2: if the remote interrupt processing did not succeed for any
// reason we honor any pending interrupt here (e.g. the interrupt
// request was forwarded at the moment when both nodes were in
// "remote" mode and our node is about to be set back into "local"
// mode), we detect this case using the interrupted member; this case
// can also occur if the local interrupt processing was unable to
// deliver the interruption
if (Thread.interrupted() || interrupted)
{
interrupted = false;
handleInterruptedImpl(new InterruptedException());
}
local = false;
// we switched in "remote" mode, push only if an interruption doesn't
// stop us
pushOperation(true, request);
}
finally
{
unlockCore();
if (leftSideControl != null)
{
leftSideControl.pushRemoteCall(request);
}
}
}
/**
* Switches control back to the local side. If an interrupt is pending
* ({@link #interrupt} has been issued but not yet honored), then this
* method calls <code>handleInterrupted()</code>.
* <p>
* This method will block until it can obtain the global lock on this
* instance. Before returning, that lock will be released.
* <p>
* If this instance is in asynchronous notification mode, then this method
* is a NOP. Otherwise the state of the instance would change during the
* asynchronous notification, which would be improper since such
* notifications are by nature supposed to be "out of band". Thus state
* changes for the normal communication channels must be unaffected.
*/
synchronized void popRemoteCall()
throws InterruptedException
{
lockCore();
try
{
if (!ready || local || async)
return;
local = true;
// CASE 1: if the local interrupt processing interrupted the thread
// but the processing on that thread did not honor the thread's
// interrupted status for any reason, we honor it here
// CASE 2: if the remote interrupt processing did not succeed for any
// reason we honor any pending interrupt here (e.g. the interrupt
// request was forwarded at the moment when both nodes were in
// "remote" mode and our node is about to be set back into "local"
// mode), we detect this case using the interrupted member
if (Thread.interrupted() || interrupted)
{
interrupted = false;
handleInterruptedImpl(new InterruptedException());
}
}
finally
{
popOperation();
unlockCore();
if (leftSideControl != null)
{
leftSideControl.popRemoteCall();
}
}
}
/**
* Unconditionally pushes the current thread onto the stack. Used to
* prepare for asynchronous interrupt requests (e.g. on the server side
* of a conversation, where the processing must be interrupted on an
* asynchronous basis).
* <p>
* This method will block until it can obtain the global lock on this
* instance. Before returning, that lock will be released.
* <p>
* MUST NOT be called from any thread used for an asynchronous
* notification, such as the one generated by {@link #interrupt} to
* trigger a remote interruption.
*
* @param request
* the ID of the request being processed.
*/
synchronized void pushUnconditionally(int request)
{
lockCore();
try
{
control.push(Thread.currentThread());
local = true;
// if the remote interrupt processing did not succeed for any reason
// we honor any pending interrupt here (e.g. the interrupt request
// was forwarded at the moment when both nodes were in "remote" mode
// and our node is about to be set back into "local" mode) we detect
// this case using the interrupted member
if (interrupted)
{
interrupted = false;
handleInterruptedImpl(new InterruptedException());
}
}
finally
{
// we just switched in local mode; this is done regardless if an
// interruption is pending or not - because we do call
// popUnconditionally if an interruption was pending
pushOperation(false, request);
unlockCore();
if (leftSideControl != null)
{
leftSideControl.pushUnconditionally(request);
}
}
}
/**
* Unconditionally pops the current thread off the stack. Used to
* cleanup the server side of a conversation, where the processing must
* be interrupted on an asynchronous basis.
* <p>
* This method will block until it can obtain the global lock on this
* instance. Before returning, that lock will be released.
* <p>
* MUST NOT be called from any thread used for an asynchronous
* notification, such as the one generated by {@link #interrupt} to
* trigger a remote interruption.
*/
synchronized void popUnconditionally()
{
lockCore();
try
{
control.pop();
local = false;
// CASE 1: if the local interrupt processing interrupted the thread
// but the processing on that thread did not honor the thread's
// interrupted status for any reason, we honor it here
// CASE 2: the local interrupt processing was unable to deliver the
// interruption, we detect this with the interrupted member
if (Thread.interrupted() || interrupted)
{
interrupted = false;
handleInterruptedImpl(new InterruptedException());
}
}
finally
{
// we are back in remote mode
popOperation();
unlockCore();
if (leftSideControl != null)
{
leftSideControl.popUnconditionally();
}
}
}
/**
* Gets the currently executing thread without popping it off the stack.
*
* @return Target thread or <code>null</code> if control is on the
* remote side.
*/
synchronized Thread getTargetThread()
{
Thread target = null;
// when the queue is stopped by the Reader/Writer threads, there is a
// possibility for the control thread stack to be empty - this is the
// case for the server-to-server queue, when the Reader thread closes
// the queue after the Dispatcher thread finished processing an incoming
// interrupt request
if (ready && local && !control.isEmpty())
{
target = control.peek();
}
return target;
}
/**
* Set the left-side control for this virtual control, in the connection
* chain.
*
* @param control
* the left-side instance, which transferes control to the virtual
* session
*/
synchronized void setLeftSideControl(Control control)
{
this.leftSideControl = control;
}
/**
* Set the right-side instance for this virtual control.
*
* @param control
* the right-side instance, which transfered control to us
*/
synchronized void setRightSideControl(Control control)
{
if (clientInstance == null && control != this)
{
// the right-side control is saved only if the request
// is being performed via a different control.
rightSideControl = control;
}
}
/**
* Get the right-side instance for this virtual control.
*
* @return the right-side instance, which transfered control to us
*/
synchronized Control getRightSideControl()
{
return clientInstance == null ? rightSideControl : null;
}
/**
* Obtains the lock for the given thread if the lock is unowned, or if
* owned, checks if the given thread owns the lock. Must only be called
* from within a synchronized block.
*
* @param thread
* The potential owner.
*
* @return <code>true</code> if the thread owns the lock (by obtaining
* it within this method or by having previously obtained it).
* <code>false</code> if some other thread owns the lock.
*/
private boolean tryLock(Thread thread)
{
if (lock == null)
{
// obtain the lock
lock = thread;
}
else if (lock == thread)
{
// we already own the lock, just return
}
else
{
// someone else owns the lock
return false;
}
return true;
}
/**
* Obtains the global lock, blocking until it is available. Must only be
* called from within a synchronized block.
*/
private void lockCore()
{
Thread thread = Thread.currentThread();
while (!tryLock(thread))
{
try
{
wait();
}
catch (InterruptedException ie)
{
handleInterruptedImpl(new InterruptedException());
}
}
}
/**
* Unlocks the global lock. Must only be called from within a synchronized
* block.
*/
private void unlockCore()
{
if (lock == Thread.currentThread())
{
lock = null;
notify();
}
}
/**
* Trigger a notification when the thread is interrupted so that the
* handler may take any action deemed appropriate like throwing a
* customized exception.
*
* @param exc
* The original <code>InterruptedException</code>.
*/
static void handleInterrupted(InterruptedException exc)
{
Control ct = locate(0);
synchronized (ct)
{
ct.handleInterruptedImpl(exc);
}
}
/**
* Trigger a notification when the thread is interrupted so that the
* handler may take any action deemed appropriate like throwing a
* customized exception.
*
* @param exc
* The original <code>InterruptedException</code>.
*/
private void handleInterruptedImpl(InterruptedException exc)
{
SessionManager.get().getInterruptHandler().handleInterrupted(exc);
}
/**
* Gets the instance of this class for the current context.
* <p>
* When the passed context ID is 0, it means it must retrieve either the
* control for the client instance or the registered control for the current
* context, if {@link #clientInstance} is null.
*
* @param contextID
* The context for which the control instance must be located.
*
* @return The instance.
*/
static synchronized Control locate(int contextID)
{
// since this is a static method, the method-level synch is on the
// CLASS not on the INSTANCE
if (contextID == 0)
{
return (clientInstance != null ? clientInstance
: serverInstance.get());
}
else
{
SessionManager sessMgr = SessionManager.get();
Control ct = sessMgr.virtualControlByContext.get(contextID);
if (ct == null)
{
ct = new Control();
sessMgr.virtualControlByContext.put(contextID, ct);
}
return ct;
}
}
/**
* Set the server instance for the current context.
*
* @param control
* Object used as the server instance.
*/
static synchronized void setContext(Control control)
{
serverInstance.set(control);
}
/**
* Container for the current executed operation and its type. This may
* represent a reply being processed or a new request being invoked.
*/
private static class Operation
{
/**
* Flag indicating if this operation is a request (when
* <code>true</code>) or a reply (when <code>false</code>).
*/
private boolean request;
/** Field holding the operation ID. */
private int operationId;
/**
* Constructor.
*
* @param request
* if <code>true</code>, this is a request
* @param operationId
* the operation id
*/
private Operation(boolean request, int operationId)
{
this.request = request;
this.operationId = operationId;
}
/**
* Test if the passed object is the same as this operation. Equality
* means both {@link #request} and {@link #operationId} must be the
* same.
*/
public boolean equals(Object obj)
{
if (!(obj instanceof Operation))
{
return false;
}
Operation o = (Operation) obj;
return o.request == this.request &&
o.operationId == this.operationId;
}
}
}