main/src/main/java/org/onap/policy/pap/main/comm/TimerManager.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP PAP
   4  * ================================================================================
   5  * Copyright (C) 2019-2021 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.pap.main.comm;
  22
  23 import java.util.LinkedHashMap;
  24 import java.util.Map;
  25 import java.util.concurrent.CountDownLatch;
  26 import java.util.concurrent.Semaphore;
  27 import java.util.concurrent.TimeUnit;
  28 import java.util.function.Consumer;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * Manager of timers. All of the timers for a given manager have the same wait time, which
  34  * makes it possible to use a linked hash map to track the timers. As a result, timers can
  35  * be quickly added and removed. In addition, the expiration time of any new timer is
  36  * always greater than or equal to the timers that are already in the map. Consequently,
  37  * the map's iterator will go in ascending order from the minimum expiration time to
  38  * maximum expiration time.
  39  *
  40  * <p>This class has not been tested for multiple threads invoking {@link #run()}
  41  * simultaneously.
  42  */
  43 public class TimerManager implements Runnable {
  44     private static final Logger logger = LoggerFactory.getLogger(TimerManager.class);
  45
  46     /**
  47      * Name of this manager, used for logging purposes.
  48      */
  49     private final String name;
  50
  51     /**
  52      * Time that each new timer should wait.
  53      */
  54     private final long waitTimeMs;
  55
  56     /**
  57      * When the map is empty, the timer thread will block waiting for this semaphore. When
  58      * a new timer is added to the map, the semaphore will be released, thus allowing the
  59      * timer thread to progress.
  60      */
  61     private final Semaphore sem = new Semaphore(0);
  62
  63     /**
  64      * This is decremented to indicate that this manager should be stopped.
  65      */
  66     private final CountDownLatch stopper = new CountDownLatch(1);
  67
  68     /**
  69      * Used to lock updates to the map.
  70      */
  71     private final Object lockit = new Object();
  72
  73     /**
  74      * Maps a timer name to a timer.
  75      */
  76     private final Map<String, Timer> name2timer = new LinkedHashMap<>();
  77
  78     /**
  79      * Constructs the object.
  80      *
  81      * @param name name of this manager, used for logging purposes
  82      * @param waitTimeMs time that each new timer should wait
  83      */
  84     public TimerManager(String name, long waitTimeMs) {
  85         this.name = name;
  86         this.waitTimeMs = waitTimeMs;
  87     }
  88
  89     /**
  90      * Stops the timer thread.
  91      */
  92     public void stop() {
  93         logger.info("timer manager {} stopping", name);
  94
  95         // Note: Must decrement the latch BEFORE releasing the semaphore
  96         stopper.countDown();
  97         sem.release();
  98     }
  99
 100     /**
 101      * Registers a timer with the given name. When the timer expires, it is automatically
 102      * unregistered and then executed.
 103      *
 104      * @param timerName name of the timer to register
 105      * @param action action to take when the timer expires; the "timerName" is passed as
 106      *        the only argument
 107      * @return the timer
 108      */
 109     public Timer register(String timerName, Consumer<String> action) {
 110
 111         synchronized (lockit) {
 112             // always remove existing entry so that new entry goes at the end of the map
 113             var timer = name2timer.remove(timerName);
 114             if (timer != null) {
 115                 logger.info("{} timer replaced {}", name, timer);
 116             }
 117
 118             timer = new Timer(timerName, action);
 119             name2timer.put(timerName, timer);
 120
 121             logger.info("{} timer registered {}", name, timer);
 122
 123             // release the timer thread in case it's waiting
 124             sem.release();
 125
 126             return timer;
 127         }
 128     }
 129
 130     /**
 131      * Continuously processes timers until {@link #stop()} is invoked.
 132      */
 133     @Override
 134     public void run() {
 135         logger.info("timer manager {} started", name);
 136
 137         while (stopper.getCount() > 0) {
 138
 139             try {
 140                 sem.acquire();
 141                 sem.drainPermits();
 142
 143                 processTimers();
 144
 145             } catch (InterruptedException e) {
 146                 logger.warn("timer manager {} stopping due to interrupt", name);
 147                 stopper.countDown();
 148                 Thread.currentThread().interrupt();
 149             }
 150         }
 151
 152         logger.info("timer manager {} stopped", name);
 153     }
 154
 155     /**
 156      * Process all timers, continuously, as long as a timer remains in the map (and
 157      * {@link #stop()} has not been called).
 158      *
 159      * @throws InterruptedException if the thread is interrupted
 160      */
 161     private void processTimers() throws InterruptedException {
 162         Timer timer;
 163         while ((timer = getNextTimer()) != null && stopper.getCount() > 0) {
 164             processTimer(timer);
 165         }
 166     }
 167
 168     /**
 169      * Gets the timer that will expire first.
 170      *
 171      * @return the timer that will expire first, or {@code null} if there are no timers
 172      */
 173     private Timer getNextTimer() {
 174
 175         synchronized (lockit) {
 176             if (name2timer.isEmpty()) {
 177                 return null;
 178             }
 179
 180             // use an iterator to get the first timer in the map
 181             return name2timer.values().iterator().next();
 182         }
 183     }
 184
 185     /**
 186      * Process a timer, waiting until it expires, unregistering it, and then executing its
 187      * action.
 188      *
 189      * @param timer timer to process
 190      * @throws InterruptedException if the thread is interrupted
 191      */
 192     private void processTimer(Timer timer) throws InterruptedException {
 193         timer.await();
 194
 195         if (stopper.getCount() == 0) {
 196             // stop() was called
 197             return;
 198         }
 199
 200         if (!timer.cancel("expired")) {
 201             // timer was cancelled while we were waiting
 202             return;
 203         }
 204
 205
 206         // run the timer
 207         try {
 208             logger.info("{} timer firing {}", TimerManager.this.name, timer);
 209             timer.runner.accept(timer.name);
 210         } catch (RuntimeException e) {
 211             logger.warn("{} timer threw an exception {}", TimerManager.this.name, timer, e);
 212         }
 213     }
 214
 215     /**
 216      * Timer info.
 217      */
 218     public class Timer {
 219         /**
 220          * The timer's name.
 221          */
 222         private String name;
 223
 224         /**
 225          * Time, in milliseconds, when the timer will expire.
 226          */
 227         private long expireMs;
 228
 229         /**
 230          * Action to take when the timer expires.
 231          */
 232         private Consumer<String> runner;
 233
 234
 235         private Timer(String name, Consumer<String> runner2) {
 236             this.name = name;
 237             this.expireMs = waitTimeMs + currentTimeMillis();
 238             this.runner = runner2;
 239         }
 240
 241         private void await() throws InterruptedException {
 242             // wait for it to expire, if necessary
 243             long tleft = expireMs - currentTimeMillis();
 244             if (tleft > 0) {
 245                 logger.info("{} timer waiting {}ms {}", TimerManager.this.name, tleft, this);
 246                 sleep(tleft);
 247             }
 248         }
 249
 250         /**
 251          * Cancels the timer.
 252          *
 253          * @return {@code true} if the timer was cancelled, {@code false} if the timer was
 254          *         not running
 255          */
 256         public boolean cancel() {
 257             return cancel("cancelled");
 258         }
 259
 260         /**
 261          * Cancels the timer.
 262          *
 263          * @param cancelMsg message to log if the timer is successfully
 264          *        cancelled
 265          * @return {@code true} if the timer was cancelled, {@code false} if the timer was
 266          *         not running
 267          */
 268         private boolean cancel(String cancelMsg) {
 269
 270             synchronized (lockit) {
 271                 if (!name2timer.remove(name, this)) {
 272                     // have a new timer in the map - ignore "this" timer
 273                     logger.info("{} timer discarded ({}) {}", TimerManager.this.name, cancelMsg, this);
 274                     return false;
 275                 }
 276
 277                 logger.info("{} timer {} {}", TimerManager.this.name, cancelMsg, this);
 278                 return true;
 279             }
 280         }
 281
 282         @Override
 283         public String toString() {
 284             return "Timer [name=" + name + ", expireMs=" + expireMs + "]";
 285         }
 286     }
 287
 288     // these may be overridden by junit tests
 289
 290     /**
 291      * Gets the current time, in milli-seconds.
 292      *
 293      * @return the current time, in milli-seconds
 294      */
 295     protected long currentTimeMillis() {
 296         return System.currentTimeMillis();
 297     }
 298
 299     /**
 300      * "Sleeps" for a bit, stopping if {@link #stop()} is invoked.
 301      *
 302      * @param timeMs time, in milli-seconds, to sleep
 303      * @throws InterruptedException if this thread is interrupted while sleeping
 304      */
 305     protected void sleep(long timeMs) throws InterruptedException {
 306         if (stopper.await(timeMs, TimeUnit.MILLISECONDS)) {
 307             logger.info("sleep finishing due to stop()");
 308         }
 309     }
 310 }