appc-core/appc-common-bundle/java/org/onap/appc/concurrent/Signal.java

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