jdk8-mips64-public/jaxws: src/share/jaxws_classes/com/sun/xml/internal/ws/api/pipe/Fiber.java@0989ad8c0860 (annotated)

src/share/jaxws_classes/com/sun/xml/internal/ws/api/pipe/Fiber.java@0989ad8c0860 (annotated)

src/share/jaxws_classes/com/sun/xml/internal/ws/api/pipe/Fiber.java

Tue, 09 Apr 2013 14:51:13 +0100

author: alanb
date: Tue, 09 Apr 2013 14:51:13 +0100
changeset 368: 0989ad8c0860
parent 286: f50545b5e2f1
child 384: 8f2986ff0235
permissions: -rw-r--r--

8010393: Update JAX-WS RI to 2.2.9-b12941
Reviewed-by: alanb, erikj
Contributed-by: miroslav.kos@oracle.com, martin.grebac@oracle.com

 /*
  * 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&#x2E; 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} &mdash; 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 = false;
                 do {
                     // if interceptors are set, go through the interceptors.
                     if (ints == null) {
                         this.next = next;
                         if (__doRun(isRequireUnlock, 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
     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;
     }
 }

Mercurial > jdk8-mips64-public > jaxws / annotate

src/share/jaxws_classes/com/sun/xml/internal/ws/api/pipe/Fiber.java@0989ad8c0860 (annotated)

src/share/jaxws_classes/com/sun/xml/internal/ws/api/pipe/Fiber.java