appc-common/src/main/java/org/openecomp/appc/concurrent/Signal.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * APPC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * Copyright (C) 2017 Amdocs
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  21  */
  22
  23 package org.openecomp.appc.concurrent;
  24
  25 import java.util.ArrayList;
  26 import java.util.Arrays;
  27 import java.util.List;
  28 import java.util.concurrent.TimeoutException;
  29
  30 import org.openecomp.appc.util.StringHelper;
  31
  32 /**
  33  * This class is used to synchronize signaling of status between threads.
  34  * <p>
  35  * In complex multi-threaded applications it is often necessary to synchronize operations between threads. This is
  36  * especially true in complex algorithms where processing dependencies exist between different threads and the
  37  * synchronization of the operations of those threads is required. This class is a framework to enable multi-thread
  38  * signaling and wait/post logic that makes the thread synchronization easier.
  39  * </p>
  40  * <p>
  41  * Basically, in thread synchronization, one thread is the "waiter" and one or more other threads are the "notifiers".
  42  * The notifiers send signals to the waiter to inform that thread that certain conditions are true, processing has been
  43  * completed, or to inform the waiter of the state of the other thread(s). In the basic java framework, the waiter and
  44  * notifier are simply using the wait/notify mechanism provided, which does not allow for different conditions, state,
  45  * or "signals" to exist. The wait/notify mechanism, in combination with the object mutex, provides basic blocking and
  46  * releasing of a thread's dispatching state.
  47  * </p>
  48  * <p>
  49  * This class builds upon the java wait/notify mechanism and allows for "signals" to be defined. These signals are
  50  * simply string constants that mean something to the waiter and notifier threads. Any number of signals may be defined,
  51  * and it is possible to wait for more than one signal to be received, wait for any one of a set to be received, or to
  52  * test if a signal has been received without blocking.
  53  * </p>
  54  * <p>
  55  * Some operations are blocking operations. These stop the execution of the calling thread until the specified condition
  56  * is true. These blocking methods are all named "wait...", such as {@link #waitFor(String...)} and
  57  * {@link #waitForAny(String...)}. The thread making the call to these blocking methods MUST be the waiter thread (the
  58  * thread registered with the signal object).
  59  * </p>
  60  * <p>
  61  * Some operations are non-blocking. These operations allow for the testing or setting of signal conditions and do not
  62  * block the caller. When calling these methods ({@link #isSignaled(String)}, {@link #signal(String)}, and
  63  * {@link #setTimeout(long)} the waiter thread mutex will be held and may block the waiter thread for the duration of
  64  * the method call.
  65  * </p>
  66  */
  67 public class Signal {
  68
  69     /**
  70      * The thread must be the thread of the waiter that is waiting for the signals to be received. It is the recipient
  71      * of the signaled condition. This allows any number of other threads to send signals to the recipient and have the
  72      * recipient synchronize its operation with the receipt of the appropriate signal(s).
  73      */
  74     private Thread thread;
  75
  76     /**
  77      * The amount of time to wait for a signal to be receieved. Set to zero to wait forever.
  78      */
  79     private long timeout = 0L;
  80
  81     /**
  82      * The collection of all received signals. Note, this need not be a synchronized collection because it will always
  83      * be accessed while holding the mutex of the thread, therefore it is implicitly synchronized.
  84      */
  85     private List<String> receivedSignals;
  86
  87     /**
  88      * A signal object must access a thread that is waiting for the receipt of the signal(s).
  89      */
  90     public Signal(Thread thread) {
  91         this.thread = thread;
  92         receivedSignals = new ArrayList<String>();
  93     }
  94
  95     /**
  96      * Checks the waiter to see if it has been signaled
  97      *
  98      * @param signal
  99      *            The signal to check for
 100      * @return True if the signal has been received, false otherwise
 101      */
 102     public boolean isSignaled(String signal) {
 103         synchronized (thread) {
 104             return _signaled(signal);
 105         }
 106     }
 107
 108     /**
 109      * Sends the indicated signal to the waiter.
 110      *
 111      * @param signal
 112      *            The signal that is to be sent to the waiting thread and to notify it to process the signal.
 113      */
 114     public void signal(String signal) {
 115         synchronized (thread) {
 116             if (!_signaled(signal)) {
 117                 receivedSignals.add(signal);
 118             }
 119             thread.notify();
 120         }
 121     }
 122
 123     /**
 124      * Blocks the waiting thread until all of the indicated signals have been received, or the wait times out.
 125      *
 126      * @param signals
 127      *            The signals to be received. The waiter is blocked forever or until all of the signals are received.
 128      * @throws TimeoutException
 129      *             If the wait has timed out waiting for a response
 130      */
 131     public void waitFor(String... signals) throws TimeoutException {
 132         long limit = System.currentTimeMillis() + timeout;
 133         synchronized (thread) {
 134             while (true) {
 135                 boolean complete = true;
 136                 for (String signal : signals) {
 137                     if (!_signaled(signal)) {
 138                         complete = false;
 139                     }
 140                 }
 141
 142                 if (complete) {
 143                     receivedSignals.removeAll(Arrays.asList(signals));
 144                     return;
 145                 }
 146
 147                 if (timeout > 0) {
 148                     if (System.currentTimeMillis() > limit) {
 149                         throw new TimeoutException(String.format("Signals %s not received in the allotted timeout.",
 150                             StringHelper.asList(signals)));
 151                     }
 152                 }
 153
 154                 try {
 155                     thread.wait(timeout);
 156                 } catch (InterruptedException e) {
 157                     /*
 158                      * Interrupted exceptions are ignored
 159                      */
 160                 }
 161             }
 162         }
 163     }
 164
 165     /**
 166      * This method blocks the waiter until at least one of the indicated signals have been received.
 167      *
 168      * @param signals
 169      *            A list of signals, any one of which will satisfy the wait condition
 170      * @return The signal that satisfied the wait
 171      * @throws TimeoutException
 172      *             If none of the signals have been received within the allotted time
 173      */
 174     public String waitForAny(String... signals) throws TimeoutException {
 175         long limit = System.currentTimeMillis() + timeout;
 176         synchronized (thread) {
 177             while (true) {
 178                 for (String signal : signals) {
 179                     if (!_signaled(signal)) {
 180                         receivedSignals.remove(signal);
 181                         return signal;
 182                     }
 183                 }
 184
 185                 if (timeout > 0) {
 186                     if (System.currentTimeMillis() > limit) {
 187                         throw new TimeoutException(
 188                             String.format("One of signals \"%s\" not received in the allotted timeout.",
 189                                 StringHelper.asList(signals)));
 190                     }
 191                 }
 192
 193                 try {
 194                     thread.wait(timeout);
 195                 } catch (InterruptedException e) {
 196                     /*
 197                      * Interrupted exceptions are ignored
 198                      */
 199                 }
 200             }
 201         }
 202     }
 203
 204     /**
 205      * This private method is used to handle the check for signaled status. Note that this method assumes the caller
 206      * holds the thread mutex.
 207      *
 208      * @param signals
 209      *            The list of signals to check for
 210      * @return True if any one of the signals has been received.
 211      */
 212     private boolean _signaled(String... signals) {
 213         for (String signal : signals) {
 214             if (receivedSignals.contains(signal)) {
 215                 return true;
 216             }
 217         }
 218         return false;
 219     }
 220
 221     /**
 222      * Sets the timeout value for waiting for signals to be received
 223      *
 224      * @param timeout
 225      */
 226     public void setTimeout(long timeout) {
 227         this.timeout = timeout;
 228     }
 229 }