    001    /*
    002     * Licensed to the Apache Software Foundation (ASF) under one or more
    003     * contributor license agreements.  See the NOTICE file distributed with
    004     * this work for additional information regarding copyright ownership.
    005     * The ASF licenses this file to You under the Apache License, Version 2.0
    006     * (the "License"); you may not use this file except in compliance with
    007     * the License.  You may obtain a copy of the License at
    008     *
    009     *      http://www.apache.org/licenses/LICENSE-2.0
    010     *
    011     * Unless required by applicable law or agreed to in writing, software
    012     * distributed under the License is distributed on an "AS IS" BASIS,
    013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    014     * See the License for the specific language governing permissions and
    015     * limitations under the License.
    016     */
    017    package org.apache.commons.lang3.concurrent;
    018
    019    import java.util.concurrent.ScheduledExecutorService;
    020    import java.util.concurrent.ScheduledFuture;
    021    import java.util.concurrent.ScheduledThreadPoolExecutor;
    022    import java.util.concurrent.TimeUnit;
    023
    024    /**
    025     * <p>
    026     * A specialized <em>semaphore</em> implementation that provides a number of
    027     * permits in a given time frame.
    028     * </p>
    029     * <p>
    030     * This class is similar to the {@code java.util.concurrent.Semaphore} class
    031     * provided by the JDK in that it manages a configurable number of permits.
    032     * Using the {@link #acquire()} method a permit can be requested by a thread.
    033     * However, there is an additional timing dimension: there is no {@code
    034     * release()} method for freeing a permit, but all permits are automatically
    035     * released at the end of a configurable time frame. If a thread calls
    036     * {@link #acquire()} and the available permits are already exhausted for this
    037     * time frame, the thread is blocked. When the time frame ends all permits
    038     * requested so far are restored, and blocking threads are waked up again, so
    039     * that they can try to acquire a new permit. This basically means that in the
    040     * specified time frame only the given number of operations is possible.
    041     * </p>
    042     * <p>
    043     * A use case for this class is to artificially limit the load produced by a
    044     * process. As an example consider an application that issues database queries
    045     * on a production system in a background process to gather statistical
    046     * information. This background processing should not produce so much database
    047     * load that the functionality and the performance of the production system are
    048     * impacted. Here a {@code TimedSemaphore} could be installed to guarantee that
    049     * only a given number of database queries are issued per second.
    050     * </p>
    051     * <p>
    052     * A thread class for performing database queries could look as follows:
    053     *
    054     * <pre>
    055     * public class StatisticsThread extends Thread {
    056     *     // The semaphore for limiting database load.
    057     *     private final TimedSemaphore semaphore;
    058     *     // Create an instance and set the semaphore
    059     *     public StatisticsThread(TimedSemaphore timedSemaphore) {
    060     *         semaphore = timedSemaphore;
    061     *     }
    062     *     // Gather statistics
    063     *     public void run() {
    064     *         try {
    065     *             while(true) {
    066     *                 semaphore.acquire();   // limit database load
    067     *                 performQuery();        // issue a query
    068     *             }
    069     *         } catch(InterruptedException) {
    070     *             // fall through
    071     *         }
    072     *     }
    073     *     ...
    074     * }
    075     * </pre>
    076     *
    077     * The following code fragment shows how a {@code TimedSemaphore} is created
    078     * that allows only 10 operations per second and passed to the statistics
    079     * thread:
    080     *
    081     * <pre>
    082     * TimedSemaphore sem = new TimedSemaphore(1, TimeUnit.SECOND, 10);
    083     * StatisticsThread thread = new StatisticsThread(sem);
    084     * thread.start();
    085     * </pre>
    086     *
    087     * </p>
    088     * <p>
    089     * When creating an instance the time period for the semaphore must be
    090     * specified. {@code TimedSemaphore} uses an executor service with a
    091     * corresponding period to monitor this interval. The {@code
    092     * ScheduledExecutorService} to be used for this purpose can be provided at
    093     * construction time. Alternatively the class creates an internal executor
    094     * service.
    095     * </p>
    096     * <p>
    097     * Client code that uses {@code TimedSemaphore} has to call the
    098     * {@link #acquire()} method in aach processing step. {@code TimedSemaphore}
    099     * keeps track of the number of invocations of the {@link #acquire()} method and
    100     * blocks the calling thread if the counter exceeds the limit specified. When
    101     * the timer signals the end of the time period the counter is reset and all
    102     * waiting threads are released. Then another cycle can start.
    103     * </p>
    104     * <p>
    105     * It is possible to modify the limit at any time using the
    106     * {@link #setLimit(int)} method. This is useful if the load produced by an
    107     * operation has to be adapted dynamically. In the example scenario with the
    108     * thread collecting statistics it may make sense to specify a low limit during
    109     * day time while allowing a higher load in the night time. Reducing the limit
    110     * takes effect immediately by blocking incoming callers. If the limit is
    111     * increased, waiting threads are not released immediately, but wake up when the
    112     * timer runs out. Then, in the next period more processing steps can be
    113     * performed without blocking. By setting the limit to 0 the semaphore can be
    114     * switched off: in this mode the {@link #acquire()} method never blocks, but
    115     * lets all callers pass directly.
    116     * </p>
    117     * <p>
    118     * When the {@code TimedSemaphore} is no more needed its {@link #shutdown()}
    119     * method should be called. This causes the periodic task that monitors the time
    120     * interval to be canceled. If the {@code ScheduledExecutorService} has been
    121     * created by the semaphore at construction time, it is also shut down.
    122     * resources. After that {@link #acquire()} must not be called any more.
    123     * </p>
    124     *
    125     * @version $Id: TimedSemaphore.java 895466 2010-01-03 19:04:08Z oheger $
    126     */
    127    public class TimedSemaphore {
    128        /**
    129         * Constant for a value representing no limit. If the limit is set to a
    130         * value less or equal this constant, the {@code TimedSemaphore} will be
    131         * effectively switched off.
    132         */
    133        public static final int NO_LIMIT = 0;
    134
    135        /** Constant for the thread pool size for the executor. */
    136        private static final int THREAD_POOL_SIZE = 1;
    137
    138        /** The executor service for managing the timer thread. */
    139        private final ScheduledExecutorService executorService;
    140
    141        /** Stores the period for this timed semaphore. */
    142        private final long period;
    143
    144        /** The time unit for the period. */
    145        private final TimeUnit unit;
    146
    147        /** A flag whether the executor service was created by this object. */
    148        private final boolean ownExecutor;
    149
    150        /** A future object representing the timer task. */
    151        private ScheduledFuture<?> task;
    152
    153        /** Stores the total number of invocations of the acquire() method. */
    154        private long totalAcquireCount;
    155
    156        /**
    157         * The counter for the periods. This counter is increased every time a
    158         * period ends.
    159         */
    160        private long periodCount;
    161
    162        /** The limit. */
    163        private int limit;
    164
    165        /** The current counter. */
    166        private int acquireCount;
    167
    168        /** The number of invocations of acquire() in the last period. */
    169        private int lastCallsPerPeriod;
    170
    171        /** A flag whether shutdown() was called. */
    172        private boolean shutdown;
    173
    174        /**
    175         * Creates a new instance of {@link TimedSemaphore} and initializes it with
    176         * the given time period and the limit.
    177         *
    178         * @param timePeriod the time period
    179         * @param timeUnit the unit for the period
    180         * @param limit the limit for the semaphore
    181         * @throws IllegalArgumentException if the period is less or equals 0
    182         */
    183        public TimedSemaphore(long timePeriod, TimeUnit timeUnit, int limit) {
    184            this(null, timePeriod, timeUnit, limit);
    185        }
    186
    187        /**
    188         * Creates a new instance of {@link TimedSemaphore} and initializes it with
    189         * an executor service, the given time period, and the limit. The executor
    190         * service will be used for creating a periodic task for monitoring the time
    191         * period. It can be <b>null</b>, then a default service will be created.
    192         *
    193         * @param service the executor service
    194         * @param timePeriod the time period
    195         * @param timeUnit the unit for the period
    196         * @param limit the limit for the semaphore
    197         * @throws IllegalArgumentException if the period is less or equals 0
    198         */
    199        public TimedSemaphore(ScheduledExecutorService service, long timePeriod,
    200                TimeUnit timeUnit, int limit) {
    201            if (timePeriod <= 0) {
    202                throw new IllegalArgumentException("Time period must be greater 0!");
    203            }
    204
    205            period = timePeriod;
    206            unit = timeUnit;
    207
    208            if (service != null) {
    209                executorService = service;
    210                ownExecutor = false;
    211            } else {
    212                ScheduledThreadPoolExecutor s = new ScheduledThreadPoolExecutor(
    213                        THREAD_POOL_SIZE);
    214                s.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
    215                s.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
    216                executorService = s;
    217                ownExecutor = true;
    218            }
    219
    220            setLimit(limit);
    221        }
    222
    223        /**
    224         * Returns the limit enforced by this semaphore. The limit determines how
    225         * many invocations of {@link #acquire()} are allowed within the monitored
    226         * period.
    227         *
    228         * @return the limit
    229         */
    230        public final synchronized int getLimit() {
    231            return limit;
    232        }
    233
    234        /**
    235         * Sets the limit. This is the number of times the {@link #acquire()} method
    236         * can be called within the time period specified. If this limit is reached,
    237         * further invocations of {@link #acquire()} will block. Setting the limit
    238         * to a value &lt;= {@link #NO_LIMIT} will cause the limit to be disabled,
    239         * i.e. an arbitrary number of{@link #acquire()} invocations is allowed in
    240         * the time period.
    241         *
    242         * @param limit the limit
    243         */
    244        public final synchronized void setLimit(int limit) {
    245            this.limit = limit;
    246        }
    247
    248        /**
    249         * Initializes a shutdown. After that the object cannot be used any more.
    250         * This method can be invoked an arbitrary number of times. All invocations
    251         * after the first one do not have any effect.
    252         */
    253        public synchronized void shutdown() {
    254            if (!shutdown) {
    255
    256                if (ownExecutor) {
    257                    // if the executor was created by this instance, it has
    258                    // to be shutdown
    259                    getExecutorService().shutdownNow();
    260                }
    261                if (task != null) {
    262                    task.cancel(false);
    263                }
    264
    265                shutdown = true;
    266            }
    267        }
    268
    269        /**
    270         * Tests whether the {@link #shutdown()} method has been called on this
    271         * object. If this method returns <b>true</b>, this instance cannot be used
    272         * any longer.
    273         *
    274         * @return a flag whether a shutdown has been performed
    275         */
    276        public synchronized boolean isShutdown() {
    277            return shutdown;
    278        }
    279
    280        /**
    281         * Tries to acquire a permit from this semaphore. This method will block if
    282         * the limit for the current period has already been reached. If
    283         * {@link #shutdown()} has already been invoked, calling this method will
    284         * cause an exception. The very first call of this method starts the timer
    285         * task which monitors the time period set for this {@code TimedSemaphore}.
    286         * From now on the semaphore is active.
    287         *
    288         * @throws InterruptedException if the thread gets interrupted
    289         * @throws IllegalStateException if this semaphore is already shut down
    290         */
    291        public synchronized void acquire() throws InterruptedException {
    292            if (isShutdown()) {
    293                throw new IllegalStateException("TimedSemaphore is shut down!");
    294            }
    295
    296            if (task == null) {
    297                task = startTimer();
    298            }
    299
    300            boolean canPass = false;
    301            do {
    302                canPass = getLimit() <= NO_LIMIT || acquireCount < getLimit();
    303                if (!canPass) {
    304                    wait();
    305                } else {
    306                    acquireCount++;
    307                }
    308            } while (!canPass);
    309        }
    310
    311        /**
    312         * Returns the number of (successful) acquire invocations during the last
    313         * period. This is the number of times the {@link #acquire()} method was
    314         * called without blocking. This can be useful for testing or debugging
    315         * purposes or to determine a meaningful threshold value. If a limit is set,
    316         * the value returned by this method won't be greater than this limit.
    317         *
    318         * @return the number of non-blocking invocations of the {@link #acquire()}
    319         * method
    320         */
    321        public synchronized int getLastAcquiresPerPeriod() {
    322            return lastCallsPerPeriod;
    323        }
    324
    325        /**
    326         * Returns the number of invocations of the {@link #acquire()} method for
    327         * the current period. This may be useful for testing or debugging purposes.
    328         *
    329         * @return the current number of {@link #acquire()} invocations
    330         */
    331        public synchronized int getAcquireCount() {
    332            return acquireCount;
    333        }
    334
    335        /**
    336         * Returns the number of calls to the {@link #acquire()} method that can
    337         * still be performed in the current period without blocking. This method
    338         * can give an indication whether it is safe to call the {@link #acquire()}
    339         * method without risking to be suspended. However, there is no guarantee
    340         * that a subsequent call to {@link #acquire()} actually is not-blocking
    341         * because in the mean time other threads may have invoked the semaphore.
    342         *
    343         * @return the current number of available {@link #acquire()} calls in the
    344         * current period
    345         */
    346        public synchronized int getAvailablePermits() {
    347            return getLimit() - getAcquireCount();
    348        }
    349
    350        /**
    351         * Returns the average number of successful (i.e. non-blocking)
    352         * {@link #acquire()} invocations for the entire life-time of this {@code
    353         * TimedSemaphore}. This method can be used for instance for statistical
    354         * calculations.
    355         *
    356         * @return the average number of {@link #acquire()} invocations per time
    357         * unit
    358         */
    359        public synchronized double getAverageCallsPerPeriod() {
    360            return (periodCount == 0) ? 0 : (double) totalAcquireCount
    361                    / (double) periodCount;
    362        }
    363
    364        /**
    365         * Returns the time period. This is the time monitored by this semaphore.
    366         * Only a given number of invocations of the {@link #acquire()} method is
    367         * possible in this period.
    368         *
    369         * @return the time period
    370         */
    371        public long getPeriod() {
    372            return period;
    373        }
    374
    375        /**
    376         * Returns the time unit. This is the unit used by {@link #getPeriod()}.
    377         *
    378         * @return the time unit
    379         */
    380        public TimeUnit getUnit() {
    381            return unit;
    382        }
    383
    384        /**
    385         * Returns the executor service used by this instance.
    386         *
    387         * @return the executor service
    388         */
    389        protected ScheduledExecutorService getExecutorService() {
    390            return executorService;
    391        }
    392
    393        /**
    394         * Starts the timer. This method is called when {@link #acquire()} is called
    395         * for the first time. It schedules a task to be executed at fixed rate to
    396         * monitor the time period specified.
    397         *
    398         * @return a future object representing the task scheduled
    399         */
    400        protected ScheduledFuture<?> startTimer() {
    401            return getExecutorService().scheduleAtFixedRate(new Runnable() {
    402                public void run() {
    403                    endOfPeriod();
    404                }
    405            }, getPeriod(), getPeriod(), getUnit());
    406        }
    407
    408        /**
    409         * The current time period is finished. This method is called by the timer
    410         * used internally to monitor the time period. It resets the counter and
    411         * releases the threads waiting for this barrier.
    412         */
    413        synchronized void endOfPeriod() {
    414            lastCallsPerPeriod = acquireCount;
    415            totalAcquireCount += acquireCount;
    416            periodCount++;
    417            acquireCount = 0;
    418            notifyAll();
    419        }
    420    }