Dispatcher.java
/*
** Module : Dispatcher.java
** Abstract : inbound message processor
**
** Copyright (c) 2005-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ----------------Description------------------------------------------------------
** 001 SIY 20050121 @19494 Created version 1.0.
** 002 SIY 20050210 @19770 Extract original exception if it is
** available.
** 003 NVS 20050228 @20109 Enabled SecurityManager calls in Dispatcher.
** They are about security context switching.
** Many of these are restricted use methods, so
** the proper call/method information is
** gathered and built into SecurityManager.java.
** 004 SIY 20050308 @20241 Changed a lot of small typos.
** 005 NVS 20050314 @20319 Security context switch methods now can throw
** RestrictedUseException. It is handled now.
** 006 SIY 20050329 @20542 Fixed comments and added stop() method.
** 007 NVS 20050331 @20559 stop() method that is used to shutdown the
** server is protected with access rights check.
** The calling subject has to have enough rights
** to access "shutdown" instance of "system"
** abstract resource in SYSTEM_SHUTDOWN_ACCESS
** mode.
** 008 NVS 20050414 @20722 Secured generateRoutingKey and getRoutingKey
** methods with net ACL checks for WRITE and
** READ access respectively. Secured
** addMethodImpl() for WRITE access and
** dispatch() for EXECUTE access.
** 009 NVS 20050415 @20739 Implemented mapping of special internal
** negative EIR indices into "system" group API
** names so that they can be protected with ACL.
** 010 NVS 20050505 @21112 Permanent dispatcher threads now repeatedly
** call setInitialSecurityContext() in their
** main loop to help refresh the security cache.
** 011 NVS 20060208 @24473 Added control flow calls and INTERRUPT call
** handling.
** 012 NVS 20060213 @24572 Fixed asynchronous interrupt request handler.
** 013 NVS 20060614 @27194 Added new method isUp() that checks to see if
** it's safe to call other Dispatcher methods.
** 014 GES 20060721 @28151 Added logging of calls to exported methods
** at FINER level and parameter values/return
** value and elapsed millis at FINEST level.
** Many formatting and code standards cleanups.
** 015 GES 20060728 @28208 Reworked dispatch() and code called from
** there to eliminate any processing that is
** not needed, including extra calls to obtain
** context local data.
** 016 NVS 20060817 @28643 Suppressed unneeded stack trace output when
** dispatcher threads are stopped due to the
** server shutdown.
** 017 NVS 20061017 @30463 Implemenented ability to restart the session
** without exiting the process.
** 018 GES 20061019 @30538 Added a return value to interruptSession().
** Reworked dispatch() to allow the use of
** push/popUnconditionally() to honor any
** pending interrupt. Also we now avoid these
** push/pop calls in the case of an asynch
** interruption request (INTERRUPT_SESSION).
** 019 GES 20061212 @31665 Added some safety code to ensure that we
** never try to forward a response using a
** dead queue.
** 020 GES 20070111 @31785 Cleanup and removal of dead code. Fixed
** some security issues and matched interface
** changes as needed.
** 021 GES 20070115 @31831 Refactored core inbound request processing
** into a separate method so that it can be
** used in conversation mode.
** 022 GES 20070410 @32915 Fix for a case where honoring an interrupt
** in pushUnconditionally() would corrupt the
** state of the control stack. Added global
** locking support during interrupt requests
** so that a potential race condition can't
** occur (where the interrupted thread calls
** over to the peer before the peer can get
** the reply that tells it whether the interrupt
** was delivered or not).
** 023 NVS 20070411 @32928 Dispatcher threads are created as part of a
** thread group named "dispatcher".
** 024 ECF 20071101 @35879 Refactored net package. Removed all context
** management to SessionManager/Session classes.
** Implemented generics. Moved remote
** authentication/termination to SessionManager.
** Other minor cleanup.
** 025 ECF 20071127 @36048 Removed remaining SecurityManager references.
** Replaced with SessionManager. Ensured an
** initial security context is set before any
** security plugin instances are retrieved.
** 026 SVL 20090715 @43196 Added OOME protection into message processing
** thread.
** 027 GES 20090722 @43333 Better thread names to enhance debugging.
** 028 NVS 20090816 @43642 Hardening the dispatch() method to errors and
** exceptions (it shouldn't exit).
** 029 CA 20090831 @43794 Fixed interruption processing when both peers
** are in remote mode. When sending a request, the
** ID of the request is passed to
** ctrl.pushUnconditionally, to be pushed on the
** Control.operations stack. interruptSession was
** modified to receive data about the remote
** operation executed when the interruption was
** encountered. Fixed dispatcher thread close on
** client restart.
** 030 CA 20090901 @43811 Changes related to virtual control multiplexing.
** 031 CA 20100318 @44761 processInbound must unlock control if it acquired
** it, regardless of the queue's state (running or
** not).
** 032 GES 20111025 Reworked method export processing to allow the
** secure replacement of exports by correlating the
** original export caller with the subsequent caller
** using a unique token for "authentication". This
** feature is needed for client-side exports when
** the client code is running in the server process
** and a CTRL-C causes the client to restart. The
** exported APIs must be remapped to new instances
** of the client objects.
** 033 CA 20111209 Added methods to remove a registered API. Must be
** used only for static APIs.
** 034 CA 20130822 Fixed the net protocol to support async requests.
** 035 CA 20130913 A dispatcher thread handling an async request must be registered as
** async thread, for the duration of this request.
** 036 CA 20131121 The sync'ed state processing is moved from Reader/Writer threads to
** the thread which actually sends/receives the message (i.e.
** Conversation thread or Dispatcher thread).
** 037 GES 20150923 Added logging for unexpected throwables during inbound processing.
** 038 GES 20151002 Added another excluded exception to the criteria list.
** 039 IAS 20160805 getMethod(RoutingKey) optimization.
** 040 VVT 20200203 Extra items removed from the method 'throws' list.
** 041 CA 20201202 State-sync via applyChanges() must be performed just before the method is invoked.
** This will allow any exception thrown by this to be properly caught and sent to
** the caller. Also, the 'catch' block was changed to allow catching any Throwable.
** In case of a non-Exception, this will be wrapped in a RuntimeException so that
** the caller can handle it like an Exception.
** CA 20201203 Moved the 'applyChanges' to the previous place, while not allowing the message
** to execute in case of errors (and report this back to the caller).
** TJD 20220504 Java 11 compatibility minor changes
** 042 VVT 20230318 Cosmetic changes, unnecessary casts removed.
** 043 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 044 GBB 20240607 Tracing logging flag removed, replaced by isLoggable.
** 045 ES 20240619 Added serverInterrupt method to insert Ctrl-C into TypeAhead queue.
** ES 20240619 Added serverInterrupt method to trigger a interrupt caused by stop-after timeout.
** 046 ICP 20240704 Fixed processInbound: Moved trackRequest to occur after context setting.
* ICP 20240705 Fixed method of obtaining the class loader.
*/
/*
** 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.io.*;
import java.lang.reflect.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.*;
import com.goldencode.p2j.classloader.MultiClassLoader;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* This class represents an incoming request processor. It contains pool of
* processing threads, Exported Interfaces Registry (EIR) and Security Context
* Map (SCM).
*/
public final class Dispatcher
implements MessageTypes
{
/** Thread group for the dispatcher threads. */
static final ThreadGroup dispThreads = new ThreadGroup(
Thread.currentThread().getThreadGroup(), "dispatcher");
/** Logger (safe as a JVM-wide value rather than as context-local). */
private static final CentralLogger LOG =
CentralLogger.get(Dispatcher.class.getName());
/** List of throwables to exclude from logging. */
private static final Class[] criteria = new Class[]
{
ConditionException.class,
InterruptedException.class,
UnstoppableExitException.class,
EOFException.class
};
/** Session manager (safe as a JVM-wide value). */
private static SessionManager sessMgr = null;
/** Dispatcher shared instance. */
private static Dispatcher dispatcher = null;
/**
* This variable is a part of EIR (Exported Interface Registry) and
* contains names of groups and methods.
*/
private final Vector<KeyMapEntry> keyMap;
/** Map routing key to method */
private final ConcurrentMap<RoutingKey, Method> methodsByKey = new ConcurrentHashMap<>();
/**
* This variable is a part of EIR (Exported Interface Registry) and
* contains mapping between instances of RoutingKey and methods.
*/
private final HashMap<RoutingKey, MethodInvoker> methodMap;
/** Token which authenticates the caller as allowed to modify the EIR. */
private final HashMap<Object, Set<RoutingKey>> modTokens;
/** A map containing semaphores for each static exported method. */
private final Map<Method, Semaphore> staticMethodLocks =
new HashMap<Method, Semaphore>();
/** Dispatcher thread pool array. */
private Thread[] threads;
/** Flag indicating the dispatcher is closing down. */
private boolean closing = false;
/**
* Constructs an instance of <code>Dispatcher</code>, including
* initialization of thread pool.
* <p>
* Enables tracing of method invocations at logging level
* <code>Level.FINER</code>. Enables tracing of parameters/result values
* and elapsed time in milliseconds at logging level
* <code>Level.FINEST</code>.
*
* @param num
* Number of threads in the thread pool.
*/
private Dispatcher(int num)
{
keyMap = new Vector<KeyMapEntry>();
methodMap = new HashMap<RoutingKey, MethodInvoker>();
modTokens = new HashMap<Object, Set<RoutingKey>>();
// at least two threads should be running
if (num <= 0)
num = 2;
threads = new Thread[num];
int i;
for (i = 0; i < threads.length; i++)
{
threads[i] = new Thread(dispThreads, new DispatcherStub());
threads[i].setName(String.format("Dispatcher %d", i));
threads[i].setDaemon(true);
}
// start thread pool
for (i = 0; i < threads.length; i++)
{
threads[i].start();
}
}
/**
* Add a method to the Exported Interface Registry (EIR). The instance of
* <code>Method</code> is provided by the application. If a method with
* the specified key is already present in the EIR it is only replaced if
* the previously returned authentication token is given here, otherwise a
* <code>SecurityException</code> is thrown.
*
* @param key
* RoutingKey instance which is used as a key for adding method.
* @param obj
* Reference to the instance of the object which will handle the
* requests. Should be <code>null</code> for static methods.
* @param method
* Instance of <code>java.lang.reflect.Method</code> which will
* handle requests.
* @param modToken
* The modification token retured from the original registration
* call (to this same method). May be <code>null</code> on the
* first call for this routing key (a new token will be created
* and returned in this case). If <code>non-null</code> on the
* first call, that object will be used as the modification token
* from then on.
*
* @return The token used to authenticate the caller when a modification
* is requested.
*
* @throws SecurityException
* is forwarded from <code>addMethodImpl</code> method.
*/
public static Object addMethod(RoutingKey key,
Object obj,
Method method,
Object modToken)
throws SecurityException
{
return dispatcher.addMethodImpl(key, obj, method, modToken);
}
/**
* Add a method to the Exported Interface Registry (EIR). The instance of
* <code>Method</code> is provided by the application. If a method with
* the specified key is already present in the EIR it is only replaced if
* the previously returned authentication token is given here, otherwise a
* <code>SecurityException</code> is thrown.
*
* @param key
* RoutingKey instance which is used as a key for adding method.
* @param obj
* Reference to the instance of the object which will handle the
* requests. Should be <code>null</code> for static methods.
* @param name
* Name of the method in the target class
* @param signature
* List of types of parameters of the method in the target class.
* @param cls
* Target class
* @param modToken
* The modification token retured from the original registration
* call (to this same method). May be <code>null</code> on the
* first call for this routing key (a new token will be created
* and returned in this case). If <code>non-null</code> on the
* first call, that object will be used as the modification token
* from then on.
*
* @return The token used to authenticate the caller when a modification
* is requested.
*
* @throws NoSuchMethodException
* is forwarded from <code>Class.getMethod</code> method.
* @throws SecurityException
* is forwarded from <code>Class.getMethod</code> or
* <code>addMethodImpl</code> methods.
*/
public static Object addMethod(RoutingKey key,
Object obj,
String name,
Class[] signature,
Class cls,
Object modToken)
throws NoSuchMethodException,
SecurityException
{
Method method = cls.getMethod(name, signature);
return dispatcher.addMethodImpl(key, obj, method, modToken);
}
/**
* Add a method to the Exported Interface Registry (EIR). The instance of
* <code>Method</code> is provided by the application. If a method with
* the specified key is already present in the EIR it is only replaced if
* the previously returned authentication token is given here, otherwise a
* <code>SecurityException</code> is thrown.
*
* @param key
* RoutingKey instance which is used as a key for adding method.
* @param obj
* Reference to the instance of the object which will handle the
* requests. Should be <code>null</code> for static methods.
* @param name
* Name of the method in the target class
* @param signature
* List of types of parameters of the method in the target class.
* @param className
* Target class name.
* @param modToken
* The modification token retured from the original registration
* call (to this same method). May be <code>null</code> on the
* first call for this routing key (a new token will be created
* and returned in this case). If <code>non-null</code> on the
* first call, that object will be used as the modification token
* from then on.
*
* @return The token used to authenticate the caller when a modification
* is requested.
*
* @throws NoSuchMethodException
* is forwarded from <code>Class.getMethod</code> method.
* @throws SecurityException
* is forwarded from <code>Class.getMethod</code> or
* <code>addMethodImpl</code> methods.
* @throws ClassNotFoundException
* is forwarded from
* <code>ClassLoader.getSystemClassLoader().loadClass</code>
* method.
*/
public static Object addMethod(RoutingKey key,
Object obj,
String name,
Class[] signature,
String className,
Object modToken)
throws NoSuchMethodException,
SecurityException,
ClassNotFoundException
{
Class cls = MultiClassLoader.getClassLoader().loadClass(className);
return addMethod(key, obj, name, signature, cls, modToken);
}
/**
* Returns <code>RoutingKey</code> instance using provided method and
* method group names. If group or method names are not present in the EIR,
* then appropriate entries added to EIR.
*
* @param group
* Method group name.
* @param method
* Method name.
* @return Instance of RoutingKey which corresponds to method and method
* group names.
*/
public static RoutingKey generateRoutingKey(String group,
String method)
{
return dispatcher.getRoutingKey(group, method, true);
}
/**
* Returns <code>RoutingKey</code> instance using provided method and
* method group names. If group or method are not registered in the EIR
* then <code>null</code> is returned.
*
* @param group
* Method group name.
* @param method
* Method name.
* @return Instance of RoutingKey which corresponds to method and method
* group names.
*/
public static RoutingKey getRoutingKey(String group, String method)
{
return dispatcher.getRoutingKey(group, method, false);
}
/**
* Start or re-start the dispatcher.
*
* @param num
* Number of threads in the thread pool.
*/
static synchronized void initialize(int num)
{
if (dispatcher != null)
{
// stop all running dispatcher's threads from previous session
dispatcher.closing = true;
for (int i = 0; i < dispatcher.threads.length; i++)
{
dispatcher.threads[i].interrupt();
}
}
dispatcher = new Dispatcher(num);
sessMgr = SessionManager.get();
Class<?> sessMgrClass = sessMgr.getClass();
try
{
Method method = Dispatcher.class.getMethod("getRoutingKey",
String.class,
String.class);
dispatcher.addMethodImpl(new RoutingKey(RoutingKey.GET_ROUTING_KEY),
null,
method,
null);
method = sessMgrClass.getMethod("shutdown");
dispatcher.addMethodImpl(new RoutingKey(RoutingKey.SHUTDOWN_PACKAGE),
sessMgr,
method,
null);
method = Dispatcher.class.getMethod("interruptSession",
boolean.class,
int.class,
int[].class);
dispatcher.addMethodImpl(new RoutingKey(RoutingKey.INTERRUPT_SESSION),
null,
method,
null);
method = Dispatcher.class.getMethod("serverInterrupt");
dispatcher.addMethodImpl(new RoutingKey(RoutingKey.SERVER_INTERRUPT),
null,
method,
null);
// export router node methods if necessary
if (sessMgr.isRouter())
{
method = sessMgrClass.getDeclaredMethod("authenticateRemote",
String.class,
int.class,
Queue.class);
dispatcher.addMethodImpl(new RoutingKey(RoutingKey.
AUTHENTICATE_REMOTE),
sessMgr,
method,
null);
method = sessMgrClass.getDeclaredMethod("terminateRemote",
int.class);
dispatcher.addMethodImpl(new RoutingKey(RoutingKey.
TERMINATE_REMOTE),
sessMgr,
method,
null);
}
}
catch (Exception exc)
{
// Should not happen because exported entry points are local to net
// package and therefore always exist and access to them is allowed,
// unless a programming error exists.
LOG.severe("", exc);
}
}
/**
* Services interrupt request from the remote end.
*
* @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.
*
* @return <code>true</code> if the interrupt request was successfully
* delivered.
*/
public static boolean interruptSession(boolean request,
int operationId,
int[] pendingReplies)
{
Control ct = Control.locate(0);
ct.setRemoteOperation(request, operationId, pendingReplies);
try
{
return Control.interrupt();
}
finally
{
ct.clearRemoteOperation();
}
}
/**
* Trigger an interrupt cause by a stop-after timeout, by inserting
* a Ctrl-C into TypeAhead queue.
*/
public static void serverInterrupt()
{
Control.triggerServerInterrupt();
}
/**
* Checks whether this class is initialized.
*
* @return <code>true</code> if this class is running.
*/
static synchronized boolean isRunning()
{
return dispatcher != null;
}
/**
* Returns the singleton instance of this class.
*
* @return The singleton instance or <code>null</code> if this class
* has not yet been initialized.
*/
static synchronized Dispatcher getInstance()
{
return dispatcher;
}
/**
* Core worker for processing of inbound requests. Reads the request,
* finds the specified method and invokes it if allowed by the security
* settings. For synchronous requests, any result or exception is sent
* back to the peer node.
*
* @param request
* The inbound request to process. Cannot be <code>null</code>.
* @param set
* Override the current thread's security context based on the
* context associated with the queue.
* @param nr
* The network resource security plugin to check for access
* rights. Can be <code>null</code> on the client or if
* network entry point protection is disabled.
*/
void processInbound(InboundRequest request,
boolean set,
NetResource nr)
throws InterruptedException
{
// read the message
Message message = request.getMessage();
Queue queue = request.getQueue();
Control ctrl = null;
RoutingKey key = message.getKey();
MethodInvoker method = null;
synchronized (methodMap)
{
method = methodMap.get(key);
}
Throwable exception = null;
Object result = null;
boolean intr = false;
if (method == null)
{
// invalid routing key
exception = new NoSuchMethodException("No such exported method, " +
key.getGroupID() + " / " +
key.getMethodID());
}
else
{
// lock this method and track the request. only static exports can
// be locked and tracked at this time.
Method ifaceMthd = getMethod(key);
Class<?> iface = getInterface(key);
SessionManager sessMgr = SessionManager.get();
boolean authRemote =
(RoutingKey.AUTHENTICATE_REMOTE == key.getGroupID() &&
RoutingKey.AUTHENTICATE_REMOTE == key.getMethodID());
set = (set && !authRemote);
if (set)
{
// set security context
sessMgr.setContext(key, queue);
}
if (ifaceMthd != null && lockMethod(ifaceMthd))
{
// after the context has been set
try
{
Session session = SessionManager.get().getSession();
ExportTracker.trackRequest(iface, ifaceMthd, session);
}
finally
{
releaseMethod(ifaceMthd);
}
}
// post-processing of the received message
Throwable failedState = null;
try
{
message = queue.getProtocol().applyChanges(message);
}
catch (Throwable t)
{
failedState = t;
exception = new RuntimeException(failedState);
if (LOG.isLoggable(Level.SEVERE) && !Utils.causeChainContains(t, criteria))
{
LOG.logp(Level.SEVERE,
"Dispatcher.processInbound()",
"",
"Unexpected throwable.",
t);
}
}
boolean asyncRequest = message.isAsyncRequest();
try
{
if (asyncRequest)
{
// if we are processing an async request, then mark this thread as async (to allow
// it to send async requests)
SessionManager.get().registerAsyncThread(Thread.currentThread());
}
// access control check (server only)
if (nr != null)
{
String grp = getGroupName(key);
String mth = getMethodName(key);
if (!nr.checkExecuteAccess(grp, mth))
{
throw new NoSuchMethodException("No such exported method " +
grp + ":" + mth);
}
}
intr = (key.getGroupID() == RoutingKey.INTERRUPT_SESSION ||
key.getGroupID() == RoutingKey.SERVER_INTERRUPT);
try
{
ctrl = queue.getControl(key.getContextID());
// bypass interrupt processing state changes if we are
// actually processing an asynchronous interrupt request
if (!intr)
{
// setup interrupt processing state
// this may cause an un-acknowledged interrupt request on
// the local side to be honored, if so we must ensure that
// the finally block handles cleanup so this must be inside
// the try block
ctrl.pushUnconditionally(message.getRequestID());
}
else
{
// obtain the global lock (this won't be released until
// after we send our reply so that this interrupt request
// will have its processing completed and the reply rec'd
// before the any further communication to the peer can
// occur)
ctrl.lock();
}
if (failedState == null)
{
// do not allow the message to execute, if there were problems during sync state processing.
Object[] payload = (Object[]) message.getPayload();
// special handling for AUTHENTICATE_REMOTE payload; method
// call requires that the *local* queue be passed in as the
// final parameter
if (authRemote)
{
int len = payload.length;
Object[] augmented = new Object[len + 1];
System.arraycopy(payload, 0, augmented, 0, len);
augmented[len] = queue;
payload = augmented;
}
// execute the requested method
result = LOG.isLoggable(Level.FINER) ? trace(method, payload) : method.invoke(payload);
}
}
finally
{
// clear interrupt processing state (if it was set
// previously), this must be inside the main try to
// ensure that any generated interruption will be
// honored as the return value
if (!intr)
{
ctrl.popUnconditionally();
}
}
}
catch (Throwable t)
{
if (exception == null)
{
if (LOG.isLoggable(Level.SEVERE) && !Utils.causeChainContains(t, criteria))
{
LOG.logp(Level.SEVERE,
"Dispatcher.processInbound()",
"",
"Unexpected throwable.",
t);
}
if (t instanceof Exception)
{
exception = t;
}
else
{
exception = new RuntimeException(t);
}
// try to extract original exception
if (t instanceof InvocationTargetException)
{
InvocationTargetException ite = (InvocationTargetException) t;
Throwable cause = ite.getCause();
if (cause instanceof Exception)
{
exception = cause;
}
}
}
}
finally
{
if (asyncRequest)
{
// if we are processing an async request, deregister the dispatcher
SessionManager.get().deregisterAsyncThread(Thread.currentThread());
}
if (set)
{
// drop security context
sessMgr.restoreContext();
}
}
}
// it is possible that this thread is returning based on a
// session abend/disconnection, in that case the session is dead
// and we need to abort further processing (we certainly must not
// try to forward any response back...)
if (queue.isRunning())
{
int type = message.getType();
if (exception != null)
{
message.setType(type == REQUEST_SYNCH ? REPLY_EXCEPTION_SYNC : REPLY_EXCEPTION_ASYNC);
message.setPayload(exception);
}
else
{
message.setType(type == REQUEST_SYNCH ? REPLY_SYNC : REPLY_ASYNC);
message.setPayload((Serializable) result);
}
queue.enqueueOutbound(message);
}
// even if the queue has been stopped while the request was being
// processed, the control must unlock its core - as other threads may
// still have pending operations which need to be finished
if (intr)
{
// release the global lock, further communication to the peer
// is OK now (on other threads)
ctrl.unlock();
}
}
/**
* Main message processing method, core of the <code>Dispatcher</code>
* class. It retrieves requests one by one from InboundQueue and processes
* them: locates requested method in EIR, calls it and if necessary -
* prepares reply message with result or exception.
*/
void dispatch()
{
SessionManager sessMgr = SessionManager.get();
sessMgr.setInitialContext();
NetResource nr = (NetResource) sessMgr.getSecurityPlugin("net");
Thread thread = Thread.currentThread();
while (true)
{
try
{
// (re)establish the initial security context (this is only
// needed here to allow the security cache to migrate each
// thread to any newer initial context that may be set over
// time
sessMgr.setInitialContext();
// clear traces of interrupted status, if any
Thread.interrupted();
// get the next request (this will block until a request is
// available) and then process it
processInbound(Queue.dequeueInbound(), true, nr);
}
catch (Throwable thr)
{
try
{
// deal with the server being shutdown
if (thr instanceof InterruptedException)
{
if (Queue.isInboundTerminated() || closing)
{
// no more dispatching to do because the server is stopping
break;
}
}
// other than shutdown, we ignore errors or exceptions that
// occur during this loop because to exit would destabilize
// the server
// log that there was an unexpected failure
if (LOG.isLoggable(Level.SEVERE))
{
LOG.logp(Level.SEVERE,
"Dispatcher.dispatch()",
"",
"failure in dispatching loop",
thr);
}
if (thr instanceof OutOfMemoryError)
{
// delay to try to recover from the OOME
try
{
Thread.sleep(5000);
}
catch (InterruptedException e)
{
// ignore
}
}
}
catch (Throwable t)
{
// ignore any problems that may occur during error processing
// and continue the loop
}
}
}
}
/**
* Lock all the given methods. This means no other deregistration or
* registration can be done on the given methods.
* <p>
* This is the same as calling
* {@link #lockMethods(Class[], Method[][], Object, boolean, boolean)
* lockMethods(classList, methods, modToken, register, true)}.
* <p>
* Locking a method is possible if there is no other locking in effect and
* there is no active session which has invoked the method.
*
* @param list
* The interfaces to export. Each one will be exported using
* the fully qualified interface name as the group and each
* method will be exported by its name.
* @param methods
* The exported methods, grouped by defining interfaces.
* @param modToken
* When not null, it is the modification token from the original
* registration call.
* @param register
* flag indicating that this is method is called when registrating
* a new API, so semaphores need to be created.
*
* @return <code>true</code> if all methods were locked.
*/
static boolean lockMethods(Class<?>[] list,
Method[][] methods,
Object modToken,
boolean register)
{
return dispatcher.lockMethodsImpl(list, methods,
modToken, register, true);
}
/**
* Lock all the given methods. This means no other deregistration or
* registration can be done on the given methods.
*
* @param list
* The interfaces to export. Each one will be exported using
* the fully qualified interface name as the group and each
* method will be exported by its name.
* @param methods
* The exported methods, grouped by defining interfaces.
* @param modToken
* When not null, it is the modification token from the original
* registration call.
* @param register
* flag indicating that this is method is called when registrating
* a new API, so semaphores need to be created.
* @param checkActive
* flag indicating no lock is possible if there are active
* sessions which have invoked this API.
*
* @return <code>true</code> if all methods were locked.
*/
static boolean lockMethods(Class<?>[] list,
Method[][] methods,
Object modToken,
boolean register,
boolean checkActive)
{
return dispatcher.lockMethodsImpl(list, methods,
modToken, register, checkActive);
}
/**
* Release all locks for the given methods.
*
* @param methods
* The exported methods, grouped by defining interfaces.
*/
static void releaseMethods(Method[][] methods)
{
dispatcher.releaseMethodsImpl(methods);
}
/**
* Remove the exported API represented by the given key. Only
* static exported APIs should be removed.
*
* @param key
* The routing key for exported API.
*/
static void removeMethod(RoutingKey key)
{
dispatcher.removeMethodImpl(key);
}
/**
* Tracing worker that handles the local method call to the exported
* interface and logs when the call is made (at FINER level) and logs
* parameters, return values and an elapsed time in millis (at FINEST
* level).
*
* @param method
* Method to be invoked through reflection.
* @param args
* Arguments, if any, to <code>method</code>.
*
* @throws Throwable
* if an exception/error occurs upon the invocation target,
* or if any problem occurs during pre- or post-processing.
*/
private Object trace(MethodInvoker method, Object[] args)
throws Throwable
{
return TraceHelper.trace(method, args, LOG, "Dispatcher.dispatch()");
}
/**
* Add a method to the Exported Interface Registry (EIR). The instance of
* <code>Method</code> is provided by the application. If a method with
* the specified key is already present in the EIR it is only replaced if
* the previously returned authentication token is given here, otherwise a
* <code>SecurityException</code> is thrown.
* <p>
* This method is an "end-point" of all other addMethod(...) methods.
*
* @param key
* Routing data which is used as a key for adding the method.
* @param obj
* Reference to the instance of the object which will handle the
* requests. Should be <code>null</code> for static methods.
* @param method
* Instance of <code>java.lang.reflect.Method</code> which will
* handle requests.
* @param modToken
* The modification token retured from the original registration
* call (to this same method). May be <code>null</code> on the
* first call for this routing key (a new token will be created
* and returned in this case). If <code>non-null</code> on the
* first call, that object will be used as the modification token
* from then on.
*
* @return The token used to authenticate the caller when a modification
* is requested.
*
* @throws SecurityException
* If the caller is does not have the rights to register this
* method or group OR if the caller does not have the proper
* authentication token when trying to replace an existing method.
*/
private synchronized Object addMethodImpl(RoutingKey key,
Object obj,
Method method,
Object modToken)
throws SecurityException
{
if (sessMgr.isRouter())
{
NetResource ar = (NetResource) sessMgr.getSecurityPlugin("net");
if (ar != null)
{
String groupName = getGroupName(key);
String methodName = getMethodName(key);
if (!ar.checkWriteAccess(groupName, methodName))
{
throw new SecurityException("Exporting " + groupName + ":" +
methodName + " is not allowed");
}
}
}
Set<RoutingKey> mappedKeySet = null;
boolean allowMod = false;
boolean newToken = true;
if (modToken != null)
{
// find the routing key associated with the given modification token
mappedKeySet = modTokens.get(modToken);
if (mappedKeySet != null)
{
newToken = false;
if (mappedKeySet.contains(key))
{
// the caller has the right mod token and the key matches, so
// the modification is allowed
allowMod = true;
}
}
}
synchronized (methodMap)
{
if (methodMap.get(key) != null && !allowMod)
{
// the key is already mapped and the caller is not allowed to modify
throw new SecurityException(
"Replacing existing methods in EIR is not allowed");
}
methodMap.put(key, new MethodInvoker(obj, method));
}
if (newToken)
{
// create a unique object reference as the modification token (if
// needed)
if (modToken == null)
{
modToken = new Object();
}
// create the set of keys
mappedKeySet = new HashSet<RoutingKey>();
// save off the mapping of the token to the key that can be modified
modTokens.put(modToken, mappedKeySet);
}
// add the routing key to the mapped list
mappedKeySet.add(key);
return modToken;
}
/**
* Get <code>RoutingKey</code> instance using provided method and group
* names. If group or method are not registered in the EIR and creation
* is not allowed then <code>null</code> is returned. Otherwise the
* appropriate entries are added into EIR and a valid key is generated.
* <p>
* This method implements proper access rights checks.
* <p>
* Must be synchronized as calls to this method can happen from any thread.
*
* @param group
* Group name.
* @param method
* Method name.
* @param canAdd
* Is creation of new instances is allowed or not.
* <code>true</code> if allowed.
*
* @return Key which corresponds to method and group names.
*/
private synchronized RoutingKey getRoutingKey(String group,
String method,
boolean canAdd)
{
// access rights check
if (sessMgr.isRouter())
{
NetResource ar = (NetResource) sessMgr.getSecurityPlugin("net");
if (ar != null)
{
boolean allowed = canAdd ? ar.checkWriteAccess(group, method)
: ar.checkReadAccess(group, method);
if (!allowed)
{
return null;
}
}
}
synchronized (keyMap)
{
KeyMapEntry grp = new KeyMapEntry(group);
int grpPos = keyMap.indexOf(grp);
if (grpPos < 0)
{
// no such group
if (!canAdd)
{
return null;
}
// add method into group
grp.getVector().add(method);
grpPos = keyMap.size();
keyMap.add(grp);
}
else
{
grp = keyMap.elementAt(grpPos);
}
int mtdPos = grp.getVector().indexOf(method);
if (mtdPos < 0)
{
// no such method
if (!canAdd)
{
return null;
}
// add method into group
mtdPos = grp.getVector().size();
grp.getVector().add(method);
}
return new RoutingKey(grpPos, mtdPos);
}
}
/**
* Get the interface used to export the method with the given
* {@link RoutingKey key}, assuming that it is an interface. If the
* {@link RoutingKey key} does not belong to an interface, then return
* <code>null</code>.
*
* @param key
* The {@link RoutingKey key} for which its exported method needs
* to be resolved.
*
* @return See above.
*/
private Class<?> getInterface(RoutingKey key)
{
if (key.getGroupID() < 0)
{
// it's a system API
return null;
}
String group = getGroupName(key);
try
{
Class<?> iface = MultiClassLoader.getClassLoader().loadClass(group);
if (!iface.isInterface())
{
return null;
}
return iface;
}
catch (ClassNotFoundException e)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.severe("Could not find interface for group " + group);
}
// if interface can not be found, then null is returned
return null;
}
}
/**
* Get the method exported with the given {@link RoutingKey key}, assuming
* that it is defined by an interface. If the {@link RoutingKey key} does
* not belong to an interface or the method can not be found in the key's
* interface hierarchy, then return <code>null</code>
*
* @param key
* The {@link RoutingKey key} for which its exported method needs
* to be resolved.
*
* @return See above.
*/
private Method getMethod(RoutingKey key)
{
Method m = methodsByKey.get(key);
if (m != null)
{
return m;
}
String name = getMethodName(key);
String group = getGroupName(key);
Class<?> iface = getInterface(key);
if (iface == null)
{
return null;
}
// go through all methods and find the one with our name
Method[] methods = iface.getMethods();
for (int i = 0; i < methods.length; i++)
{
Method mthd = methods[i];
if (mthd.toString().equals(name))
{
m = methodsByKey.putIfAbsent(key, mthd);
return m != null ? m : mthd;
}
}
// method not found in the interface (maybe something got out of sync)
if (LOG.isLoggable(Level.WARNING))
{
LOG.warning("Could not find method " + name + " in group " + group);
}
return null;
}
/**
* Gets the group name corresponding to the specified routing key.
* Routing keys with negative gropiId are reported as having "system"
* group.
* @param key
* routing key
* @return group name where the key belongs
*/
private String getGroupName(RoutingKey key)
{
if (key.getGroupID() < 0)
return "system";
synchronized (keyMap)
{
KeyMapEntry kme = keyMap.elementAt(key.getGroupID());
return kme.getName();
}
}
/**
* Gets the method name corresponding to the specified routing key.
* Routing keys with negative methodIds are reported as having special
* builtin names within "system" group.
*
* @param key
* routing key
* @return method name
*/
private String getMethodName(RoutingKey key)
{
int methodId = key.getMethodID();
if (methodId < 0)
{
switch (methodId)
{
case RoutingKey.GET_ROUTING_KEY:
return "route";
case RoutingKey.AUTHENTICATE_REMOTE:
return "authenticate";
case RoutingKey.TERMINATE_REMOTE:
return "terminate";
case RoutingKey.SHUTDOWN_PACKAGE:
return "shutdown";
case RoutingKey.INTERRUPT_SESSION:
return "interrupt";
default:
return "undefined";
}
}
else
{
synchronized (keyMap)
{
KeyMapEntry kme = keyMap.elementAt(key.getGroupID());
Vector<String> v = kme.getVector();
return v.elementAt(methodId);
}
}
}
/**
* Lock all the given methods. This means no other deregistration or
* registration can be done on the given methods.
*
* @param list
* The interfaces to export. Each one will be exported using
* the fully qualified interface name as the group and each
* method will be exported by its name.
* @param methods
* The exported methods, grouped by defining interfaces.
* @param modToken
* When not null, it is the modification token from the original
* registration call.
* @param register
* flag indicating that this is method is called when registrating
* a new API, so semaphores need to be created.
* @param checkActive
* flag indicating no lock is possible if there are active
* sessions which have invoked this API.
*
* @return <code>true</code> if all methods were locked.
*/
private boolean lockMethodsImpl(Class<?>[] list,
Method[][] methods,
Object modToken,
boolean register,
boolean checkActive)
{
// TODO: make this to return "error codes" ?
InterruptedException ie = null;
boolean allLocked = true;
Set<Method> locked = new HashSet<Method>();
for (int i = 0; allLocked && i < methods.length; i++)
{
for (int j = 0; j < methods[i].length; j++)
{
// must lock this method only if is not used by active sessions
if (checkActive &&
ExportTracker.hasActiveSessions(list[i], methods[i][j]))
{
allLocked = false;
break;
}
boolean allowMod = true;
// TODO: what if is already registered ??
if (modToken != null && !register)
{
allowMod = false;
// find the routing key associated with the given modification
// token
Set<RoutingKey> mappedKeySet = null;
synchronized (this)
{
mappedKeySet = modTokens.get(modToken);
}
Class<?> clazz = methods[i][j].getDeclaringClass();
String group = RemoteObject.calcGroupName(clazz);
String method = RemoteObject.calcKeyName(methods[i][j]);
RoutingKey key = getRoutingKey(group, method, false);
if (key != null &&
mappedKeySet != null &&
mappedKeySet.contains(key))
{
// the caller has the right mod token and the key matches,
// so the modification is allowed
allowMod = true;
}
}
if (!allowMod)
{
allLocked = false;
break;
}
Semaphore sem = null;
synchronized (staticMethodLocks)
{
sem = staticMethodLocks.get(methods[i][j]);
if (register)
{
if (sem == null)
{
staticMethodLocks.put(methods[i][j],
sem = new Semaphore(1));
}
else
{
// this must be a replace, so leave the semaphore intact
}
}
}
try
{
sem.acquire();
}
catch (InterruptedException ex)
{
allLocked = false;
ie = ex;
break;
}
locked.add(methods[i][j]);
}
}
if (!allLocked)
{
// if couldn't lock all of them, then release all those locked
for (Method m : locked)
{
releaseMethod(m);
}
}
if (ie != null)
{
throw new RuntimeException(ie);
}
return allLocked;
}
/**
* Release all locks for the given methods.
*
* @param methods
* The exported methods, grouped by defining interfaces.
*/
private void releaseMethodsImpl(Method[][] methods)
{
for (int i = 0; i < methods.length; i++)
{
for (int j = 0; j < methods[i].length; j++)
{
releaseMethod(methods[i][j]);
}
}
}
/**
* Release the lock for the given method.
*
* @param method
* The exported method.
*/
private void releaseMethod(Method method)
{
Semaphore sem = null;
synchronized (staticMethodLocks)
{
sem = staticMethodLocks.get(method);
}
if (sem != null)
{
sem.release();
}
}
/**
* Locks the given method, without checking if there are active sessions.
*
* @param method
* The exported method.
*
* @throws InterruptedException
* If thread is interrupted during semaphore acquire.
*/
private boolean lockMethod(Method method)
throws InterruptedException
{
boolean locked = false;
Semaphore sem = null;
// this does not and must not check for active sessions which have
// invoked this exported method. this is needed only to ensure there is
// an exclusive lock on this method.
synchronized (staticMethodLocks)
{
sem = staticMethodLocks.get(method);
}
if (sem == null)
{
return false;
}
try
{
sem.acquire();
locked = true;
}
catch (InterruptedException e)
{
// thread was interrupted, nothing else to do
throw e;
}
return locked;
}
/**
* Remove the exported API represented by the given key. Only
* static exported APIs should be removed.
*
* @param key
* The routing key for exported API.
*/
private synchronized void removeMethodImpl(RoutingKey key)
{
// if I got here, all security checks are in place.
// only static methods can be removed.
Method method = getMethod(key);
if (method == null)
{
return;
}
synchronized (methodMap)
{
methodMap.remove(key);
}
// remove this routing key from all modToken(s)
Iterator<Set<RoutingKey>> iter = modTokens.values().iterator();
while (iter.hasNext())
{
Set<RoutingKey> set = iter.next();
set.remove(key);
if (set.isEmpty())
{
iter.remove();
}
}
// remove the semaphore for this method, but I need the interface method
// and not the real method (the one in methodinvoker)
synchronized (staticMethodLocks)
{
staticMethodLocks.remove(method);
}
int methodId = key.getMethodID();
int groupId = key.getGroupID();
synchronized (keyMap)
{
KeyMapEntry kme = keyMap.elementAt(groupId);
Vector<String> v = kme.getVector();
v.set(methodId, null);
// remove all trailing nulls
while (v.size() > 0 && v.elementAt(v.size() - 1) == null)
{
v.remove(v.size() - 1);
}
if (v.isEmpty())
{
keyMap.set(groupId, null);
}
// remove all trailing nulls
while (keyMap.size() > 0 && keyMap.elementAt(keyMap.size() - 1) == null)
{
keyMap.remove(keyMap.size() - 1);
}
}
}
/**
* This class is used to start threads running <code>dispatch()</code>
* method.
*/
private class DispatcherStub
implements Runnable
{
/**
* Implementation of <code>Runnable</code> interface. It simply calls
* <code>Dispatcher.dispatch()</code>.
*/
public void run()
{
dispatch();
}
}
/**
* This class is used as an entry in the Key Map. It corresponds to named
* group of methods and contains names of the methods which belong to
* particular group.
*/
private static class KeyMapEntry
{
/** name of the entry (corresponds to group name) */
private final String name;
/** Vector of method names associated with this group */
private final Vector<String> vector = new Vector<String>();
/**
* Constructs a <code>KeyMapEntry</code> using given group name.
*
* @param group
* Group name.
*/
KeyMapEntry(String group)
{
name = group;
}
/**
* This method is used by <code>Vector.indexOf()</code> method.
*
* @param obj
* An object to compare with.
*
* @return <code>true</code> if object, passed as a parameter, is
* equal to this instance and <code>false</code> otherwise.
*/
public boolean equals(Object obj)
{
if (this == obj)
return true;
if (obj instanceof KeyMapEntry)
return name.equals(((KeyMapEntry) obj).getName());
return false;
}
/**
* Provides access to group name.
*
* @return The group name.
*/
String getName()
{
return name;
}
/**
* Provides access to the vector of method names.
*
* @return The vector of names associated with this group.
*/
Vector<String> getVector()
{
return vector;
}
}
}