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

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