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

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