| /* |
| * Copyright (c) 1997, 2013, Oracle and/or its affiliates. All rights reserved. |
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. |
| * |
| * This code is free software; you can redistribute it and/or modify it |
| * under the terms of the GNU General Public License version 2 only, as |
| * published by the Free Software Foundation. Oracle designates this |
| * particular file as subject to the "Classpath" exception as provided |
| * by Oracle in the LICENSE file that accompanied this code. |
| * |
| * This code is distributed in the hope that it will be useful, but WITHOUT |
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or |
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License |
| * version 2 for more details (a copy is included in the LICENSE file that |
| * accompanied this code). |
| * |
| * You should have received a copy of the GNU General Public License version |
| * 2 along with this work; if not, write to the Free Software Foundation, |
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. |
| * |
| * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA |
| * or visit www.oracle.com if you need additional information or have any |
| * questions. |
| */ |
| |
| package com.sun.xml.internal.ws.api.pipe; |
| |
| import com.sun.istack.internal.NotNull; |
| import com.sun.istack.internal.Nullable; |
| import com.sun.xml.internal.ws.api.Cancelable; |
| import com.sun.xml.internal.ws.api.Component; |
| import com.sun.xml.internal.ws.api.ComponentRegistry; |
| import com.sun.xml.internal.ws.api.SOAPVersion; |
| import com.sun.xml.internal.ws.api.addressing.AddressingVersion; |
| import com.sun.xml.internal.ws.api.message.AddressingUtils; |
| import com.sun.xml.internal.ws.api.message.Packet; |
| import com.sun.xml.internal.ws.api.pipe.helper.AbstractFilterTubeImpl; |
| import com.sun.xml.internal.ws.api.pipe.helper.AbstractTubeImpl; |
| import com.sun.xml.internal.ws.api.server.Adapter; |
| import com.sun.xml.internal.ws.api.server.Container; |
| import com.sun.xml.internal.ws.api.server.ContainerResolver; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Set; |
| import java.util.concurrent.CopyOnWriteArraySet; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.locks.Condition; |
| import java.util.concurrent.locks.ReentrantLock; |
| import java.util.logging.Level; |
| import java.util.logging.Logger; |
| |
| import javax.xml.ws.Holder; |
| import javax.xml.ws.WebServiceException; |
| |
| /** |
| * User-level thread. Represents the execution of one request/response processing. |
| * <p/> |
| * <p/> |
| * JAX-WS RI is capable of running a large number of request/response concurrently by |
| * using a relatively small number of threads. This is made possible by utilizing |
| * a {@link Fiber} — a user-level thread that gets created for each request/response |
| * processing. |
| * <p/> |
| * <p/> |
| * A fiber remembers where in the pipeline the processing is at, what needs to be |
| * executed on the way out (when processing response), and other additional information |
| * specific to the execution of a particular request/response. |
| * <p/> |
| * <h2>Suspend/Resume</h2> |
| * <p/> |
| * Fiber can be {@link NextAction#suspend() suspended} by a {@link Tube}. |
| * When a fiber is suspended, it will be kept on the side until it is |
| * {@link #resume(Packet) resumed}. This allows threads to go execute |
| * other runnable fibers, allowing efficient utilization of smaller number of |
| * threads. |
| * <p/> |
| * <h2>Context-switch Interception</h2> |
| * <p/> |
| * {@link FiberContextSwitchInterceptor} allows {@link Tube}s and {@link Adapter}s |
| * to perform additional processing every time a thread starts running a fiber |
| * and stops running it. |
| * <p/> |
| * <h2>Context ClassLoader</h2> |
| * <p/> |
| * Just like thread, a fiber has a context class loader (CCL.) A fiber's CCL |
| * becomes the thread's CCL when it's executing the fiber. The original CCL |
| * of the thread will be restored when the thread leaves the fiber execution. |
| * <p/> |
| * <p/> |
| * <h2>Debugging Aid</h2> |
| * <p/> |
| * Because {@link Fiber} doesn't keep much in the call stack, and instead use |
| * {@link #conts} to store the continuation, debugging fiber related activities |
| * could be harder. |
| * <p/> |
| * <p/> |
| * Setting the {@link #LOGGER} for FINE would give you basic start/stop/resume/suspend |
| * level logging. Using FINER would cause more detailed logging, which includes |
| * what tubes are executed in what order and how they behaved. |
| * <p/> |
| * <p/> |
| * When you debug the server side, consider setting {@link Fiber#serializeExecution} |
| * to true, so that execution of fibers are serialized. Debugging a server |
| * with more than one running threads is very tricky, and this switch will |
| * prevent that. This can be also enabled by setting the system property on. |
| * See the source code. |
| * |
| * @author Kohsuke Kawaguchi |
| * @author Jitendra Kotamraju |
| */ |
| public final class Fiber implements Runnable, Cancelable, ComponentRegistry { |
| |
| /** |
| * Callback interface for notification of suspend and resume. |
| * |
| * @since 2.2.6 |
| * @deprecated Use {@link NextAction#suspend(Runnable)} |
| */ |
| public interface Listener { |
| /** |
| * Fiber has been suspended. Implementations of this callback may resume the Fiber. |
| * @param fiber Fiber |
| */ |
| public void fiberSuspended(Fiber fiber); |
| |
| /** |
| * Fiber has been resumed. Behavior is undefined if implementations of this callback attempt to suspend the Fiber. |
| * @param fiber Fiber |
| */ |
| public void fiberResumed(Fiber fiber); |
| } |
| |
| private final List<Listener> _listeners = new ArrayList<Listener>(); |
| |
| /** |
| * Adds suspend/resume callback listener |
| * @param listener Listener |
| * @since 2.2.6 |
| * @deprecated |
| */ |
| public void addListener(Listener listener) { |
| synchronized(_listeners) { |
| if (!_listeners.contains(listener)) { |
| _listeners.add(listener); |
| } |
| } |
| } |
| |
| /** |
| * Removes suspend/resume callback listener |
| * @param listener Listener |
| * @since 2.2.6 |
| * @deprecated |
| */ |
| public void removeListener(Listener listener) { |
| synchronized(_listeners) { |
| _listeners.remove(listener); |
| } |
| } |
| |
| List<Listener> getCurrentListeners() { |
| synchronized(_listeners) { |
| return new ArrayList<Listener>(_listeners); |
| } |
| } |
| |
| private void clearListeners() { |
| synchronized(_listeners) { |
| _listeners.clear(); |
| } |
| } |
| |
| /** |
| * {@link Tube}s whose {@link Tube#processResponse(Packet)} method needs |
| * to be invoked on the way back. |
| */ |
| private Tube[] conts = new Tube[16]; |
| private int contsSize; |
| |
| /** |
| * If this field is non-null, the next instruction to execute is |
| * to call its {@link Tube#processRequest(Packet)}. Otherwise |
| * the instruction is to call {@link #conts}. |
| */ |
| private Tube next; |
| |
| private Packet packet; |
| |
| private Throwable/*but really it's either RuntimeException or Error*/ throwable; |
| |
| public final Engine owner; |
| |
| /** |
| * Is this thread suspended? 0=not suspended, 1=suspended. |
| * <p/> |
| * <p/> |
| * Logically this is just a boolean, but we need to prepare for the case |
| * where the thread is {@link #resume(Packet) resumed} before we get to the {@link #suspend()}. |
| * This happens when things happen in the following order: |
| * <p/> |
| * <ol> |
| * <li>Tube decides that the fiber needs to be suspended to wait for the external event. |
| * <li>Tube hooks up fiber with some external mechanism (like NIO channel selector) |
| * <li>Tube returns with {@link NextAction#suspend()}. |
| * <li>"External mechanism" becomes signal state and invokes {@link Fiber#resume(Packet)} |
| * to wake up fiber |
| * <li>{@link Fiber#doRun} invokes {@link Fiber#suspend()}. |
| * </ol> |
| * <p/> |
| * <p/> |
| * Using int, this will work OK because {@link #suspendedCount} becomes -1 when |
| * {@link #resume(Packet)} occurs before {@link #suspend()}. |
| * <p/> |
| * <p/> |
| * Increment and decrement is guarded by 'this' object. |
| */ |
| private volatile int suspendedCount = 0; |
| |
| private volatile boolean isInsideSuspendCallbacks = false; |
| |
| /** |
| * Is this {@link Fiber} currently running in the synchronous mode? |
| */ |
| private boolean synchronous; |
| |
| private boolean interrupted; |
| |
| private final int id; |
| |
| /** |
| * Active {@link FiberContextSwitchInterceptor}s for this fiber. |
| */ |
| private List<FiberContextSwitchInterceptor> interceptors; |
| |
| /** |
| * Fiber's context {@link ClassLoader}. |
| */ |
| private |
| @Nullable |
| ClassLoader contextClassLoader; |
| |
| private |
| @Nullable |
| CompletionCallback completionCallback; |
| |
| private boolean isDeliverThrowableInPacket = false; |
| |
| public void setDeliverThrowableInPacket(boolean isDeliverThrowableInPacket) { |
| this.isDeliverThrowableInPacket = isDeliverThrowableInPacket; |
| } |
| |
| /** |
| * The thread on which this Fiber is currently executing, if applicable. |
| */ |
| private Thread currentThread; |
| |
| /** |
| * Replace uses of synchronized(this) with this lock so that we can control |
| * unlocking for resume use cases |
| */ |
| private final ReentrantLock lock = new ReentrantLock(); |
| private final Condition condition = lock.newCondition(); |
| |
| private volatile boolean isCanceled; |
| |
| /** |
| * Set to true if this fiber is started asynchronously, to avoid |
| * doubly-invoking completion code. |
| */ |
| private boolean started; |
| |
| /** |
| * Set to true if this fiber is started sync but allowed to run async. |
| * This property exists for use cases where the processing model is fundamentally async |
| * but some requirement or feature mandates that part of the tubeline run synchronously. For |
| * instance, WS-ReliableMessaging with non-anonymous addressing is compatible with running |
| * asynchronously, but if in-order message delivery is used then message processing must assign |
| * a message number before the remainder of the processing can be asynchronous. |
| */ |
| private boolean startedSync; |
| |
| /** |
| * Callback to be invoked when a {@link Fiber} finishes execution. |
| */ |
| public interface CompletionCallback { |
| /** |
| * Indicates that the fiber has finished its execution. |
| * <p/> |
| * <p/> |
| * Since the JAX-WS RI runs asynchronously, |
| * this method maybe invoked by a different thread |
| * than any of the threads that started it or run a part of tubeline. |
| */ |
| void onCompletion(@NotNull Packet response); |
| |
| /** |
| * Indicates that the fiber has finished abnormally, by throwing a given {@link Throwable}. |
| */ |
| void onCompletion(@NotNull Throwable error); |
| } |
| |
| Fiber(Engine engine) { |
| this.owner = engine; |
| id = iotaGen.incrementAndGet(); |
| if (isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "{0} created", getName()); |
| } |
| |
| // if this is run from another fiber, then we naturally inherit its context classloader, |
| // so this code works for fiber->fiber inheritance just fine. |
| contextClassLoader = Thread.currentThread().getContextClassLoader(); |
| } |
| |
| /** |
| * Starts the execution of this fiber asynchronously. |
| * <p/> |
| * <p/> |
| * This method works like {@link Thread#start()}. |
| * |
| * @param tubeline The first tube of the tubeline that will act on the packet. |
| * @param request The request packet to be passed to <tt>startPoint.processRequest()</tt>. |
| * @param completionCallback The callback to be invoked when the processing is finished and the |
| * final response packet is available. |
| * @see #runSync(Tube, Packet) |
| */ |
| public void start(@NotNull Tube tubeline, @NotNull Packet request, @Nullable CompletionCallback completionCallback) { |
| start(tubeline, request, completionCallback, false); |
| } |
| |
| private void dumpFiberContext(String desc) { |
| if(isTraceEnabled()) { |
| String action = null; |
| String msgId = null; |
| if (packet != null) { |
| for (SOAPVersion sv: SOAPVersion.values()) { |
| for (AddressingVersion av: AddressingVersion.values()) { |
| action = packet.getMessage() != null ? AddressingUtils.getAction(packet.getMessage().getHeaders(), av, sv) : null; |
| msgId = packet.getMessage() != null ? AddressingUtils.getMessageID(packet.getMessage().getHeaders(), av, sv) : null; |
| if (action != null || msgId != null) { |
| break; |
| } |
| } |
| if (action != null || msgId != null) { |
| break; |
| } |
| } |
| } |
| String actionAndMsgDesc; |
| if (action == null && msgId == null) { |
| actionAndMsgDesc = "NO ACTION or MSG ID"; |
| } else { |
| actionAndMsgDesc = "'" + action + "' and msgId '" + msgId + "'"; |
| } |
| |
| String tubeDesc; |
| if (next != null) { |
| tubeDesc = next.toString() + ".processRequest()"; |
| } else { |
| tubeDesc = peekCont() + ".processResponse()"; |
| } |
| |
| LOGGER.log(Level.FINE, "{0} {1} with {2} and ''current'' tube {3} from thread {4} with Packet: {5}", new Object[]{getName(), desc, actionAndMsgDesc, tubeDesc, Thread.currentThread().getName(), packet != null ? packet.toShortString() : null}); |
| } |
| } |
| |
| /** |
| * Starts the execution of this fiber. |
| * |
| * If forceSync is true, then the fiber is started for an ostensibly async invocation, |
| * but allows for some portion of the tubeline to run sync with the calling |
| * client instance (Port/Dispatch instance). This allows tubes that enforce |
| * ordering to see requests in the order they were sent at the point the |
| * client invoked them. |
| * <p> |
| * The forceSync parameter will be true only when the caller (e.g. AsyncInvoker or |
| * SEIStub) knows one or more tubes need to enforce ordering and thus need |
| * to run sync with the client. Such tubes can return |
| * NextAction.INVOKE_ASYNC to indicate that the next tube in the tubeline |
| * should be invoked async to the current thread. |
| * |
| * <p> |
| * This method works like {@link Thread#start()}. |
| * |
| * @param tubeline |
| * The first tube of the tubeline that will act on the packet. |
| * @param request |
| * The request packet to be passed to <tt>startPoint.processRequest()</tt>. |
| * @param completionCallback |
| * The callback to be invoked when the processing is finished and the |
| * final response packet is available. |
| * |
| * @see #start(Tube,Packet,CompletionCallback) |
| * @see #runSync(Tube,Packet) |
| * @since 2.2.6 |
| */ |
| public void start(@NotNull Tube tubeline, @NotNull Packet request, @Nullable CompletionCallback completionCallback, boolean forceSync) { |
| next = tubeline; |
| this.packet = request; |
| this.completionCallback = completionCallback; |
| |
| if (forceSync) { |
| this.startedSync = true; |
| dumpFiberContext("starting (sync)"); |
| run(); |
| } else { |
| this.started = true; |
| dumpFiberContext("starting (async)"); |
| owner.addRunnable(this); |
| } |
| } |
| |
| /** |
| * Wakes up a suspended fiber. |
| * <p/> |
| * <p/> |
| * If a fiber was suspended without specifying the next {@link Tube}, |
| * then the execution will be resumed in the response processing direction, |
| * by calling the {@link Tube#processResponse(Packet)} method on the next/first |
| * {@link Tube} in the {@link Fiber}'s processing stack with the specified resume |
| * packet as the parameter. |
| * <p/> |
| * <p/> |
| * If a fiber was suspended with specifying the next {@link Tube}, |
| * then the execution will be resumed in the request processing direction, |
| * by calling the next tube's {@link Tube#processRequest(Packet)} method with the |
| * specified resume packet as the parameter. |
| * <p/> |
| * <p/> |
| * This method is implemented in a race-free way. Another thread can invoke |
| * this method even before this fiber goes into the suspension mode. So the caller |
| * need not worry about synchronizing {@link NextAction#suspend()} and this method. |
| * |
| * @param resumePacket packet used in the resumed processing |
| */ |
| public void resume(@NotNull Packet resumePacket) { |
| resume(resumePacket, false); |
| } |
| |
| /** |
| * Similar to resume(Packet) but allowing the Fiber to be resumed |
| * synchronously (in the current Thread). If you want to know when the |
| * fiber completes (not when this method returns) then add/wrap a |
| * CompletionCallback on this Fiber. |
| * For example, an asynchronous response endpoint that supports WS-ReliableMessaging |
| * including in-order message delivery may need to resume the Fiber synchronously |
| * until message order is confirmed prior to returning to asynchronous processing. |
| * @since 2.2.6 |
| */ |
| public void resume(@NotNull Packet resumePacket, |
| boolean forceSync) { |
| resume(resumePacket, forceSync, null); |
| } |
| |
| /** |
| * Similar to resume(Packet, boolean) but allowing the Fiber to be resumed |
| * and at the same time atomically assign a new CompletionCallback to it. |
| * @since 2.2.6 |
| */ |
| public void resume(@NotNull Packet resumePacket, |
| boolean forceSync, |
| CompletionCallback callback) { |
| lock.lock(); |
| try { |
| if (callback != null) { |
| setCompletionCallback(callback); |
| } |
| if(isTraceEnabled()) |
| LOGGER.log(Level.FINE, "{0} resuming. Will have suspendedCount={1}", new Object[]{getName(), suspendedCount-1}); |
| packet = resumePacket; |
| if( --suspendedCount == 0 ) { |
| if (!isInsideSuspendCallbacks) { |
| List<Listener> listeners = getCurrentListeners(); |
| for (Listener listener: listeners) { |
| try { |
| listener.fiberResumed(this); |
| } catch (Throwable e) { |
| if (isTraceEnabled()) |
| LOGGER.log(Level.FINE, "Listener {0} threw exception: {1}", new Object[]{listener, e.getMessage()}); |
| } |
| } |
| |
| if(synchronous) { |
| condition.signalAll(); |
| } else if (forceSync || startedSync) { |
| run(); |
| } else { |
| dumpFiberContext("resuming (async)"); |
| owner.addRunnable(this); |
| } |
| } |
| } else { |
| if (isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "{0} taking no action on resume because suspendedCount != 0: {1}", new Object[]{getName(), suspendedCount}); |
| } |
| } |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| /** |
| * Wakes up a suspended fiber and begins response processing. |
| * @since 2.2.6 |
| */ |
| public void resumeAndReturn(@NotNull Packet resumePacket, |
| boolean forceSync) { |
| if(isTraceEnabled()) |
| LOGGER.log(Level.FINE, "{0} resumed with Return Packet", getName()); |
| next = null; |
| resume(resumePacket, forceSync); |
| } |
| |
| /** |
| * Wakes up a suspended fiber with an exception. |
| * <p/> |
| * <p/> |
| * The execution of the suspended fiber will be resumed in the response |
| * processing direction, by calling the {@link Tube#processException(Throwable)} method |
| * on the next/first {@link Tube} in the {@link Fiber}'s processing stack with |
| * the specified exception as the parameter. |
| * <p/> |
| * <p/> |
| * This method is implemented in a race-free way. Another thread can invoke |
| * this method even before this fiber goes into the suspension mode. So the caller |
| * need not worry about synchronizing {@link NextAction#suspend()} and this method. |
| * |
| * @param throwable exception that is used in the resumed processing |
| */ |
| public void resume(@NotNull Throwable throwable) { |
| resume(throwable, packet, false); |
| } |
| |
| /** |
| * Wakes up a suspended fiber with an exception. |
| * <p/> |
| * <p/> |
| * The execution of the suspended fiber will be resumed in the response |
| * processing direction, by calling the {@link Tube#processException(Throwable)} method |
| * on the next/first {@link Tube} in the {@link Fiber}'s processing stack with |
| * the specified exception as the parameter. |
| * <p/> |
| * <p/> |
| * This method is implemented in a race-free way. Another thread can invoke |
| * this method even before this fiber goes into the suspension mode. So the caller |
| * need not worry about synchronizing {@link NextAction#suspend()} and this method. |
| * |
| * @param throwable exception that is used in the resumed processing |
| * @param packet Packet that will be visible on the Fiber after the resume |
| * @since 2.2.8 |
| */ |
| public void resume(@NotNull Throwable throwable, @NotNull Packet packet) { |
| resume(throwable, packet, false); |
| } |
| |
| /** |
| * Wakes up a suspend fiber with an exception. |
| * |
| * If forceSync is true, then the suspended fiber will resume with |
| * synchronous processing on the current thread. This will continue |
| * until some Tube indicates that it is safe to switch to asynchronous |
| * processing. |
| * |
| * @param error exception that is used in the resumed processing |
| * @param forceSync if processing begins synchronously |
| * @since 2.2.6 |
| */ |
| public void resume(@NotNull Throwable error, |
| boolean forceSync) { |
| resume(error, packet, forceSync); |
| } |
| |
| /** |
| * Wakes up a suspend fiber with an exception. |
| * |
| * If forceSync is true, then the suspended fiber will resume with |
| * synchronous processing on the current thread. This will continue |
| * until some Tube indicates that it is safe to switch to asynchronous |
| * processing. |
| * |
| * @param error exception that is used in the resumed processing |
| * @param packet Packet that will be visible on the Fiber after the resume |
| * @param forceSync if processing begins synchronously |
| * @since 2.2.8 |
| */ |
| public void resume(@NotNull Throwable error, |
| @NotNull Packet packet, |
| boolean forceSync) { |
| if(isTraceEnabled()) |
| LOGGER.log(Level.FINE, "{0} resumed with Return Throwable", getName()); |
| next = null; |
| throwable = error; |
| resume(packet, forceSync); |
| } |
| |
| /** |
| * Marks this Fiber as cancelled. A cancelled Fiber will never invoke its completion callback |
| * @param mayInterrupt if cancel should use {@link Thread#interrupt()} |
| * @see java.util.concurrent.Future#cancel(boolean) |
| * @since 2.2.6 |
| */ |
| @Override |
| public void cancel(boolean mayInterrupt) { |
| isCanceled = true; |
| if (mayInterrupt) { |
| // synchronized(this) is used as Thread running Fiber will be holding lock |
| synchronized(this) { |
| if (currentThread != null) |
| currentThread.interrupt(); |
| } |
| } |
| } |
| |
| /** |
| * Suspends this fiber's execution until the resume method is invoked. |
| * <p/> |
| * The call returns immediately, and when the fiber is resumed |
| * the execution picks up from the last scheduled continuation. |
| * @param onExitRunnable runnable to be invoked after fiber is marked for suspension |
| * @return if control loop must exit |
| */ |
| private boolean suspend(Holder<Boolean> isRequireUnlock, Runnable onExitRunnable) { |
| if(isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "{0} suspending. Will have suspendedCount={1}", new Object[]{getName(), suspendedCount+1}); |
| if (suspendedCount > 0) { |
| LOGGER.log(Level.FINE, "WARNING - {0} suspended more than resumed. Will require more than one resume to actually resume this fiber.", getName()); |
| } |
| } |
| |
| List<Listener> listeners = getCurrentListeners(); |
| if (++suspendedCount == 1) { |
| isInsideSuspendCallbacks = true; |
| try { |
| for (Listener listener: listeners) { |
| try { |
| listener.fiberSuspended(this); |
| } catch (Throwable e) { |
| if(isTraceEnabled()) |
| LOGGER.log(Level.FINE, "Listener {0} threw exception: {1}", new Object[]{listener, e.getMessage()}); |
| } |
| } |
| } finally { |
| isInsideSuspendCallbacks = false; |
| } |
| } |
| |
| if (suspendedCount <= 0) { |
| // suspend callback caused fiber to resume |
| for (Listener listener: listeners) { |
| try { |
| listener.fiberResumed(this); |
| } catch (Throwable e) { |
| if(isTraceEnabled()) |
| LOGGER.log(Level.FINE, "Listener {0} threw exception: {1}", new Object[]{listener, e.getMessage()}); |
| } |
| } |
| |
| } else if (onExitRunnable != null) { |
| // synchronous use cases cannot disconnect from the current thread |
| if (!synchronous) { |
| /* INTENTIONALLY UNLOCKING EARLY */ |
| synchronized(this) { |
| // currentThread is protected by the monitor for this fiber so |
| // that it is accessible to cancel() even when the lock is held |
| currentThread = null; |
| } |
| lock.unlock(); |
| assert(!lock.isHeldByCurrentThread()); |
| isRequireUnlock.value = Boolean.FALSE; |
| |
| try { |
| onExitRunnable.run(); |
| } catch(Throwable t) { |
| throw new OnExitRunnableException(t); |
| } |
| |
| return true; |
| |
| } else { |
| // for synchronous we will stay with current thread, so do not disconnect |
| if (isTraceEnabled()) |
| LOGGER.fine("onExitRunnable used with synchronous Fiber execution -- not exiting current thread"); |
| onExitRunnable.run(); |
| } |
| } |
| |
| return false; |
| } |
| |
| private static final class OnExitRunnableException extends RuntimeException { |
| private static final long serialVersionUID = 1L; |
| |
| Throwable target; |
| |
| public OnExitRunnableException(Throwable target) { |
| super((Throwable)null); // see pattern for InvocationTargetException |
| this.target = target; |
| } |
| } |
| |
| /** |
| * Adds a new {@link FiberContextSwitchInterceptor} to this fiber. |
| * <p/> |
| * <p/> |
| * The newly installed fiber will take effect immediately after the current |
| * tube returns from its {@link Tube#processRequest(Packet)} or |
| * {@link Tube#processResponse(Packet)}, before the next tube begins processing. |
| * <p/> |
| * <p/> |
| * So when the tubeline consists of X and Y, and when X installs an interceptor, |
| * the order of execution will be as follows: |
| * <p/> |
| * <ol> |
| * <li>X.processRequest() |
| * <li>interceptor gets installed |
| * <li>interceptor.execute() is invoked |
| * <li>Y.processRequest() |
| * </ol> |
| */ |
| public synchronized void addInterceptor(@NotNull FiberContextSwitchInterceptor interceptor) { |
| if (interceptors == null) { |
| interceptors = new ArrayList<FiberContextSwitchInterceptor>(); |
| } else { |
| List<FiberContextSwitchInterceptor> l = new ArrayList<FiberContextSwitchInterceptor>(); |
| l.addAll(interceptors); |
| interceptors = l; |
| } |
| interceptors.add(interceptor); |
| } |
| |
| /** |
| * Removes a {@link FiberContextSwitchInterceptor} from this fiber. |
| * <p/> |
| * <p/> |
| * The removal of the interceptor takes effect immediately after the current |
| * tube returns from its {@link Tube#processRequest(Packet)} or |
| * {@link Tube#processResponse(Packet)}, before the next tube begins processing. |
| * <p/> |
| * <p/> |
| * <p/> |
| * So when the tubeline consists of X and Y, and when Y uninstalls an interceptor |
| * on the way out, then the order of execution will be as follows: |
| * <p/> |
| * <ol> |
| * <li>Y.processResponse() (notice that this happens with interceptor.execute() in the callstack) |
| * <li>interceptor gets uninstalled |
| * <li>interceptor.execute() returns |
| * <li>X.processResponse() |
| * </ol> |
| * |
| * @return true if the specified interceptor was removed. False if |
| * the specified interceptor was not registered with this fiber to begin with. |
| */ |
| public synchronized boolean removeInterceptor(@NotNull FiberContextSwitchInterceptor interceptor) { |
| if (interceptors != null) { |
| boolean result = interceptors.remove(interceptor); |
| if (interceptors.isEmpty()) |
| interceptors = null; |
| else { |
| List<FiberContextSwitchInterceptor> l = new ArrayList<FiberContextSwitchInterceptor>(); |
| l.addAll(interceptors); |
| interceptors = l; |
| } |
| return result; |
| } |
| return false; |
| } |
| |
| /** |
| * Gets the context {@link ClassLoader} of this fiber. |
| */ |
| public |
| @Nullable |
| ClassLoader getContextClassLoader() { |
| return contextClassLoader; |
| } |
| |
| /** |
| * Sets the context {@link ClassLoader} of this fiber. |
| */ |
| public ClassLoader setContextClassLoader(@Nullable ClassLoader contextClassLoader) { |
| ClassLoader r = this.contextClassLoader; |
| this.contextClassLoader = contextClassLoader; |
| return r; |
| } |
| |
| /** |
| * DO NOT CALL THIS METHOD. This is an implementation detail |
| * of {@link Fiber}. |
| */ |
| @Deprecated |
| @Override |
| public void run() { |
| Container old = ContainerResolver.getDefault().enterContainer(owner.getContainer()); |
| try { |
| assert !synchronous; |
| // doRun returns true to indicate an early exit from fiber processing |
| if (!doRun()) { |
| if (startedSync && suspendedCount == 0 && |
| (next != null || contsSize > 0)) { |
| // We bailed out of running this fiber we started as sync, and now |
| // want to finish running it async |
| startedSync = false; |
| // Start back up as an async fiber |
| dumpFiberContext("restarting (async) after startSync"); |
| owner.addRunnable(this); |
| } else { |
| completionCheck(); |
| } |
| } |
| } finally { |
| ContainerResolver.getDefault().exitContainer(old); |
| } |
| } |
| |
| /** |
| * Runs a given {@link Tube} (and everything thereafter) synchronously. |
| * <p/> |
| * <p/> |
| * This method blocks and returns only when all the successive {@link Tube}s |
| * complete their request/response processing. This method can be used |
| * if a {@link Tube} needs to fallback to synchronous processing. |
| * <p/> |
| * <h3>Example:</h3> |
| * <pre> |
| * class FooTube extends {@link AbstractFilterTubeImpl} { |
| * NextAction processRequest(Packet request) { |
| * // run everything synchronously and return with the response packet |
| * return doReturnWith(Fiber.current().runSync(next,request)); |
| * } |
| * NextAction processResponse(Packet response) { |
| * // never be invoked |
| * } |
| * } |
| * </pre> |
| * |
| * @param tubeline The first tube of the tubeline that will act on the packet. |
| * @param request The request packet to be passed to <tt>startPoint.processRequest()</tt>. |
| * @return The response packet to the <tt>request</tt>. |
| * @see #start(Tube, Packet, CompletionCallback) |
| */ |
| public |
| @NotNull |
| Packet runSync(@NotNull Tube tubeline, @NotNull Packet request) { |
| lock.lock(); |
| try { |
| // save the current continuation, so that we return runSync() without executing them. |
| final Tube[] oldCont = conts; |
| final int oldContSize = contsSize; |
| final boolean oldSynchronous = synchronous; |
| final Tube oldNext = next; |
| |
| if (oldContSize > 0) { |
| conts = new Tube[16]; |
| contsSize = 0; |
| } |
| |
| try { |
| synchronous = true; |
| this.packet = request; |
| next = tubeline; |
| doRun(); |
| if (throwable != null) { |
| if (isDeliverThrowableInPacket) { |
| packet.addSatellite(new ThrowableContainerPropertySet(throwable)); |
| } else { |
| if (throwable instanceof RuntimeException) { |
| throw (RuntimeException) throwable; |
| } |
| if (throwable instanceof Error) { |
| throw (Error) throwable; |
| } |
| // our system is supposed to only accept Error or RuntimeException |
| throw new AssertionError(throwable); |
| } |
| } |
| return this.packet; |
| } finally { |
| conts = oldCont; |
| contsSize = oldContSize; |
| synchronous = oldSynchronous; |
| next = oldNext; |
| if(interrupted) { |
| Thread.currentThread().interrupt(); |
| interrupted = false; |
| } |
| if(!started && !startedSync) |
| completionCheck(); |
| } |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| private void completionCheck() { |
| lock.lock(); |
| try { |
| // Don't trigger completion and callbacks if fiber is suspended |
| if(!isCanceled && contsSize==0 && suspendedCount == 0) { |
| if(isTraceEnabled()) |
| LOGGER.log(Level.FINE, "{0} completed", getName()); |
| clearListeners(); |
| condition.signalAll(); |
| if (completionCallback != null) { |
| if (throwable != null) { |
| if (isDeliverThrowableInPacket) { |
| packet.addSatellite(new ThrowableContainerPropertySet(throwable)); |
| completionCallback.onCompletion(packet); |
| } else |
| completionCallback.onCompletion(throwable); |
| } else |
| completionCallback.onCompletion(packet); |
| } |
| } |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| /** |
| * Invokes all registered {@link InterceptorHandler}s and then call into |
| * {@link Fiber#__doRun()}. |
| */ |
| private class InterceptorHandler implements FiberContextSwitchInterceptor.Work<Tube, Tube> { |
| private final Holder<Boolean> isUnlockRequired; |
| private final List<FiberContextSwitchInterceptor> ints; |
| |
| /** |
| * Index in {@link Fiber#interceptors} to invoke next. |
| */ |
| private int idx; |
| |
| public InterceptorHandler(Holder<Boolean> isUnlockRequired, List<FiberContextSwitchInterceptor> ints) { |
| this.isUnlockRequired = isUnlockRequired; |
| this.ints = ints; |
| } |
| |
| /** |
| * Initiate the interception, and eventually invokes {@link Fiber#__doRun()}. |
| */ |
| Tube invoke(Tube next) { |
| idx = 0; |
| return execute(next); |
| } |
| |
| @Override |
| public Tube execute(Tube next) { |
| if (idx == ints.size()) { |
| Fiber.this.next = next; |
| if (__doRun(isUnlockRequired, ints)) |
| return PLACEHOLDER; |
| } else { |
| FiberContextSwitchInterceptor interceptor = ints.get(idx++); |
| return interceptor.execute(Fiber.this, next, this); |
| } |
| return Fiber.this.next; |
| } |
| } |
| |
| private static final PlaceholderTube PLACEHOLDER = new PlaceholderTube(); |
| |
| private static class PlaceholderTube extends AbstractTubeImpl { |
| |
| @Override |
| public NextAction processRequest(Packet request) { |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public NextAction processResponse(Packet response) { |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public NextAction processException(Throwable t) { |
| return doThrow(t); |
| } |
| |
| @Override |
| public void preDestroy() { |
| } |
| |
| @Override |
| public PlaceholderTube copy(TubeCloner cloner) { |
| throw new UnsupportedOperationException(); |
| } |
| } |
| |
| /** |
| * Executes the fiber as much as possible. |
| * |
| */ |
| private boolean doRun() { |
| dumpFiberContext("running"); |
| |
| if (serializeExecution) { |
| serializedExecutionLock.lock(); |
| try { |
| return _doRun(next); |
| } finally { |
| serializedExecutionLock.unlock(); |
| } |
| } else { |
| return _doRun(next); |
| } |
| } |
| |
| private boolean _doRun(Tube next) { |
| // isRequireUnlock will contain Boolean.FALSE when lock has already been released in suspend |
| Holder<Boolean> isRequireUnlock = new Holder<Boolean>(Boolean.TRUE); |
| lock.lock(); |
| try { |
| List<FiberContextSwitchInterceptor> ints; |
| ClassLoader old; |
| synchronized(this) { |
| ints = interceptors; |
| |
| // currentThread is protected by the monitor for this fiber so |
| // that it is accessible to cancel() even when the lock is held |
| currentThread = Thread.currentThread(); |
| if (isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "Thread entering _doRun(): {0}", currentThread); |
| } |
| |
| old = currentThread.getContextClassLoader(); |
| currentThread.setContextClassLoader(contextClassLoader); |
| } |
| |
| try { |
| boolean needsToReenter; |
| do { |
| // if interceptors are set, go through the interceptors. |
| if (ints == null) { |
| this.next = next; |
| if (__doRun(isRequireUnlock, null /*ints*/)) { |
| return true; |
| } |
| } else { |
| next = new InterceptorHandler(isRequireUnlock, ints).invoke(next); |
| if (next == PLACEHOLDER) { |
| return true; |
| } |
| } |
| |
| synchronized(this) { |
| needsToReenter = (ints != interceptors); |
| if (needsToReenter) |
| ints = interceptors; |
| } |
| } while (needsToReenter); |
| } catch(OnExitRunnableException o) { |
| // catching this exception indicates onExitRunnable in suspend() threw. |
| // we must still avoid double unlock |
| Throwable t = o.target; |
| if (t instanceof WebServiceException) |
| throw (WebServiceException) t; |
| throw new WebServiceException(t); |
| } finally { |
| // don't reference currentThread here because fiber processing |
| // may already be running on a different thread (Note: isAlreadyExited |
| // tracks this state |
| Thread thread = Thread.currentThread(); |
| thread.setContextClassLoader(old); |
| if (isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "Thread leaving _doRun(): {0}", thread); |
| } |
| } |
| |
| return false; |
| } finally { |
| if (isRequireUnlock.value) { |
| synchronized(this) { |
| currentThread = null; |
| } |
| lock.unlock(); |
| } |
| } |
| } |
| |
| /** |
| * To be invoked from {@link #doRun()}. |
| * |
| * @see #doRun() |
| */ |
| private boolean __doRun(Holder<Boolean> isRequireUnlock, List<FiberContextSwitchInterceptor> originalInterceptors) { |
| assert(lock.isHeldByCurrentThread()); |
| |
| final Fiber old = CURRENT_FIBER.get(); |
| CURRENT_FIBER.set(this); |
| |
| // if true, lots of debug messages to show what's being executed |
| final boolean traceEnabled = LOGGER.isLoggable(Level.FINER); |
| |
| try { |
| boolean abortResponse = false; |
| while(isReady(originalInterceptors)) { |
| if (isCanceled) { |
| next = null; |
| throwable = null; |
| contsSize = 0; |
| break; |
| } |
| |
| try { |
| NextAction na; |
| Tube last; |
| if(throwable!=null) { |
| if(contsSize==0 || abortResponse) { |
| contsSize = 0; // abortResponse case |
| // nothing else to execute. we are done. |
| return false; |
| } |
| last = popCont(); |
| if (traceEnabled) |
| LOGGER.log(Level.FINER, "{0} {1}.processException({2})", new Object[]{getName(), last, throwable}); |
| na = last.processException(throwable); |
| } else { |
| if(next!=null) { |
| if(traceEnabled) |
| LOGGER.log(Level.FINER, "{0} {1}.processRequest({2})", new Object[]{getName(), next, packet != null ? "Packet@"+Integer.toHexString(packet.hashCode()) : "null"}); |
| na = next.processRequest(packet); |
| last = next; |
| } else { |
| if(contsSize==0 || abortResponse) { |
| // nothing else to execute. we are done. |
| contsSize = 0; |
| return false; |
| } |
| last = popCont(); |
| if(traceEnabled) |
| LOGGER.log(Level.FINER, "{0} {1}.processResponse({2})", new Object[]{getName(), last, packet != null ? "Packet@"+Integer.toHexString(packet.hashCode()) : "null"}); |
| na = last.processResponse(packet); |
| } |
| } |
| |
| if (traceEnabled) |
| LOGGER.log(Level.FINER, "{0} {1} returned with {2}", new Object[]{getName(), last, na}); |
| |
| // If resume is called before suspend, then make sure |
| // resume(Packet) is not lost |
| if (na.kind != NextAction.SUSPEND) { |
| // preserve in-flight packet so that processException may inspect |
| if (na.kind != NextAction.THROW && |
| na.kind != NextAction.THROW_ABORT_RESPONSE) |
| packet = na.packet; |
| throwable = na.throwable; |
| } |
| |
| switch(na.kind) { |
| case NextAction.INVOKE: |
| case NextAction.INVOKE_ASYNC: |
| pushCont(last); |
| // fall through next |
| case NextAction.INVOKE_AND_FORGET: |
| next = na.next; |
| if (na.kind == NextAction.INVOKE_ASYNC |
| && startedSync) { |
| // Break out here |
| return false; |
| } |
| break; |
| case NextAction.THROW_ABORT_RESPONSE: |
| case NextAction.ABORT_RESPONSE: |
| abortResponse = true; |
| if (isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "Fiber {0} is aborting a response due to exception: {1}", new Object[]{this, na.throwable}); |
| } |
| case NextAction.RETURN: |
| case NextAction.THROW: |
| next = null; |
| break; |
| case NextAction.SUSPEND: |
| if (next != null) { |
| // Only store the 'last' tube when we're processing |
| // a request, since conts array is for processResponse |
| pushCont(last); |
| } |
| next = na.next; |
| if(suspend(isRequireUnlock, na.onExitRunnable)) |
| return true; // explicitly exiting control loop |
| break; |
| default: |
| throw new AssertionError(); |
| } |
| } catch (RuntimeException t) { |
| if (traceEnabled) |
| LOGGER.log(Level.FINER, getName() + " Caught " + t + ". Start stack unwinding", t); |
| throwable = t; |
| } catch (Error t) { |
| if (traceEnabled) |
| LOGGER.log(Level.FINER, getName() + " Caught " + t + ". Start stack unwinding", t); |
| throwable = t; |
| } |
| |
| dumpFiberContext("After tube execution"); |
| } |
| |
| // there's nothing we can execute right away. |
| // we'll be back when this fiber is resumed. |
| |
| } finally { |
| CURRENT_FIBER.set(old); |
| } |
| |
| return false; |
| } |
| |
| private void pushCont(Tube tube) { |
| conts[contsSize++] = tube; |
| |
| // expand if needed |
| int len = conts.length; |
| if (contsSize == len) { |
| Tube[] newBuf = new Tube[len * 2]; |
| System.arraycopy(conts, 0, newBuf, 0, len); |
| conts = newBuf; |
| } |
| } |
| |
| private Tube popCont() { |
| return conts[--contsSize]; |
| } |
| |
| private Tube peekCont() { |
| int index = contsSize - 1; |
| if (index >= 0 && index < conts.length) { |
| return conts[index]; |
| } else { |
| return null; |
| } |
| } |
| |
| /** |
| * Only to be used by Tubes that manipulate the Fiber to create alternate flows |
| * @since 2.2.6 |
| */ |
| public void resetCont(Tube[] conts, int contsSize) { |
| this.conts = conts; |
| this.contsSize = contsSize; |
| } |
| |
| /** |
| * Returns true if the fiber is ready to execute. |
| */ |
| private boolean isReady(List<FiberContextSwitchInterceptor> originalInterceptors) { |
| if (synchronous) { |
| while (suspendedCount == 1) |
| try { |
| if (isTraceEnabled()) { |
| LOGGER.log(Level.FINE, "{0} is blocking thread {1}", new Object[]{getName(), Thread.currentThread().getName()}); |
| } |
| condition.await(); // the synchronized block is the whole runSync method. |
| } catch (InterruptedException e) { |
| // remember that we are interrupted, but don't respond to it |
| // right away. This behavior is in line with what happens |
| // when you are actually running the whole thing synchronously. |
| interrupted = true; |
| } |
| |
| synchronized(this) { |
| return interceptors == originalInterceptors; |
| } |
| } |
| else { |
| if (suspendedCount>0) |
| return false; |
| synchronized(this) { |
| return interceptors == originalInterceptors; |
| } |
| } |
| } |
| |
| private String getName() { |
| return "engine-" + owner.id + "fiber-" + id; |
| } |
| |
| @Override |
| public String toString() { |
| return getName(); |
| } |
| |
| /** |
| * Gets the current {@link Packet} associated with this fiber. |
| * <p/> |
| * <p/> |
| * This method returns null if no packet has been associated with the fiber yet. |
| */ |
| public |
| @Nullable |
| Packet getPacket() { |
| return packet; |
| } |
| |
| /** |
| * Returns completion callback associated with this Fiber |
| * @return Completion callback |
| * @since 2.2.6 |
| */ |
| public CompletionCallback getCompletionCallback() { |
| return completionCallback; |
| } |
| |
| /** |
| * Updates completion callback associated with this Fiber |
| * @param completionCallback Completion callback |
| * @since 2.2.6 |
| */ |
| public void setCompletionCallback(CompletionCallback completionCallback) { |
| this.completionCallback = completionCallback; |
| } |
| |
| /** |
| * (ADVANCED) Returns true if the current fiber is being executed synchronously. |
| * <p/> |
| * <p/> |
| * Fiber may run synchronously for various reasons. Perhaps this is |
| * on client side and application has invoked a synchronous method call. |
| * Perhaps this is on server side and we have deployed on a synchronous |
| * transport (like servlet.) |
| * <p/> |
| * <p/> |
| * When a fiber is run synchronously (IOW by {@link #runSync(Tube, Packet)}), |
| * further invocations to {@link #runSync(Tube, Packet)} can be done |
| * without degrading the performance. |
| * <p/> |
| * <p/> |
| * So this value can be used as a further optimization hint for |
| * advanced {@link Tube}s to choose the best strategy to invoke |
| * the next {@link Tube}. For example, a tube may want to install |
| * a {@link FiberContextSwitchInterceptor} if running async, yet |
| * it might find it faster to do {@link #runSync(Tube, Packet)} |
| * if it's already running synchronously. |
| */ |
| public static boolean isSynchronous() { |
| return current().synchronous; |
| } |
| |
| /** |
| * Returns true if the current Fiber on the current thread was started |
| * synchronously. Note, this is not strictly the same as being synchronous |
| * because the assumption is that the Fiber will ultimately be dispatched |
| * asynchronously, possibly have a completion callback associated with it, etc. |
| * Note, the 'startedSync' flag is cleared once the current Fiber is |
| * converted to running asynchronously. |
| * @since 2.2.6 |
| */ |
| public boolean isStartedSync() { |
| return startedSync; |
| } |
| |
| /** |
| * Gets the current fiber that's running. |
| * <p/> |
| * <p/> |
| * This works like {@link Thread#currentThread()}. |
| * This method only works when invoked from {@link Tube}. |
| */ |
| public static |
| @NotNull |
| @SuppressWarnings({"null", "ConstantConditions"}) |
| Fiber current() { |
| Fiber fiber = CURRENT_FIBER.get(); |
| if (fiber == null) |
| throw new IllegalStateException("Can be only used from fibers"); |
| return fiber; |
| } |
| |
| /** |
| * Gets the current fiber that's running, if set. |
| */ |
| public static Fiber getCurrentIfSet() { |
| return CURRENT_FIBER.get(); |
| } |
| |
| private static final ThreadLocal<Fiber> CURRENT_FIBER = new ThreadLocal<Fiber>(); |
| |
| /** |
| * Used to allocate unique number for each fiber. |
| */ |
| private static final AtomicInteger iotaGen = new AtomicInteger(); |
| |
| private static boolean isTraceEnabled() { |
| return LOGGER.isLoggable(Level.FINE); |
| } |
| |
| private static final Logger LOGGER = Logger.getLogger(Fiber.class.getName()); |
| |
| |
| private static final ReentrantLock serializedExecutionLock = new ReentrantLock(); |
| |
| /** |
| * Set this boolean to true to execute fibers sequentially one by one. |
| * See class javadoc. |
| */ |
| public static volatile boolean serializeExecution = Boolean.getBoolean(Fiber.class.getName() + ".serialize"); |
| |
| private final Set<Component> components = new CopyOnWriteArraySet<Component>(); |
| |
| @Override |
| public <S> S getSPI(Class<S> spiType) { |
| for (Component c : components) { |
| S spi = c.getSPI(spiType); |
| if (spi != null) { |
| return spi; |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Set<Component> getComponents() { |
| return components; |
| } |
| } |
