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     * @since 3.0
126     * @version $Id: TimedSemaphore.java 1082044 2011-03-16 04:26:58Z bayard $
127     */
128    public class TimedSemaphore {
129        /**
130         * Constant for a value representing no limit. If the limit is set to a
131         * value less or equal this constant, the {@code TimedSemaphore} will be
132         * effectively switched off.
133         */
134        public static final int NO_LIMIT = 0;
135    
136        /** Constant for the thread pool size for the executor. */
137        private static final int THREAD_POOL_SIZE = 1;
138    
139        /** The executor service for managing the timer thread. */
140        private final ScheduledExecutorService executorService;
141    
142        /** Stores the period for this timed semaphore. */
143        private final long period;
144    
145        /** The time unit for the period. */
146        private final TimeUnit unit;
147    
148        /** A flag whether the executor service was created by this object. */
149        private final boolean ownExecutor;
150    
151        /** A future object representing the timer task. */
152        private ScheduledFuture<?> task;
153    
154        /** Stores the total number of invocations of the acquire() method. */
155        private long totalAcquireCount;
156    
157        /**
158         * The counter for the periods. This counter is increased every time a
159         * period ends.
160         */
161        private long periodCount;
162    
163        /** The limit. */
164        private int limit;
165    
166        /** The current counter. */
167        private int acquireCount;
168    
169        /** The number of invocations of acquire() in the last period. */
170        private int lastCallsPerPeriod;
171    
172        /** A flag whether shutdown() was called. */
173        private boolean shutdown;
174    
175        /**
176         * Creates a new instance of {@link TimedSemaphore} and initializes it with
177         * the given time period and the limit.
178         *
179         * @param timePeriod the time period
180         * @param timeUnit the unit for the period
181         * @param limit the limit for the semaphore
182         * @throws IllegalArgumentException if the period is less or equals 0
183         */
184        public TimedSemaphore(long timePeriod, TimeUnit timeUnit, int limit) {
185            this(null, timePeriod, timeUnit, limit);
186        }
187    
188        /**
189         * Creates a new instance of {@link TimedSemaphore} and initializes it with
190         * an executor service, the given time period, and the limit. The executor
191         * service will be used for creating a periodic task for monitoring the time
192         * period. It can be <b>null</b>, then a default service will be created.
193         *
194         * @param service the executor service
195         * @param timePeriod the time period
196         * @param timeUnit the unit for the period
197         * @param limit the limit for the semaphore
198         * @throws IllegalArgumentException if the period is less or equals 0
199         */
200        public TimedSemaphore(ScheduledExecutorService service, long timePeriod,
201                TimeUnit timeUnit, int limit) {
202            if (timePeriod <= 0) {
203                throw new IllegalArgumentException("Time period must be greater 0!");
204            }
205    
206            period = timePeriod;
207            unit = timeUnit;
208    
209            if (service != null) {
210                executorService = service;
211                ownExecutor = false;
212            } else {
213                ScheduledThreadPoolExecutor s = new ScheduledThreadPoolExecutor(
214                        THREAD_POOL_SIZE);
215                s.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
216                s.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
217                executorService = s;
218                ownExecutor = true;
219            }
220    
221            setLimit(limit);
222        }
223    
224        /**
225         * Returns the limit enforced by this semaphore. The limit determines how
226         * many invocations of {@link #acquire()} are allowed within the monitored
227         * period.
228         *
229         * @return the limit
230         */
231        public final synchronized int getLimit() {
232            return limit;
233        }
234    
235        /**
236         * Sets the limit. This is the number of times the {@link #acquire()} method
237         * can be called within the time period specified. If this limit is reached,
238         * further invocations of {@link #acquire()} will block. Setting the limit
239         * to a value &lt;= {@link #NO_LIMIT} will cause the limit to be disabled,
240         * i.e. an arbitrary number of{@link #acquire()} invocations is allowed in
241         * the time period.
242         *
243         * @param limit the limit
244         */
245        public final synchronized void setLimit(int limit) {
246            this.limit = limit;
247        }
248    
249        /**
250         * Initializes a shutdown. After that the object cannot be used any more.
251         * This method can be invoked an arbitrary number of times. All invocations
252         * after the first one do not have any effect.
253         */
254        public synchronized void shutdown() {
255            if (!shutdown) {
256    
257                if (ownExecutor) {
258                    // if the executor was created by this instance, it has
259                    // to be shutdown
260                    getExecutorService().shutdownNow();
261                }
262                if (task != null) {
263                    task.cancel(false);
264                }
265    
266                shutdown = true;
267            }
268        }
269    
270        /**
271         * Tests whether the {@link #shutdown()} method has been called on this
272         * object. If this method returns <b>true</b>, this instance cannot be used
273         * any longer.
274         *
275         * @return a flag whether a shutdown has been performed
276         */
277        public synchronized boolean isShutdown() {
278            return shutdown;
279        }
280    
281        /**
282         * Tries to acquire a permit from this semaphore. This method will block if
283         * the limit for the current period has already been reached. If
284         * {@link #shutdown()} has already been invoked, calling this method will
285         * cause an exception. The very first call of this method starts the timer
286         * task which monitors the time period set for this {@code TimedSemaphore}.
287         * From now on the semaphore is active.
288         *
289         * @throws InterruptedException if the thread gets interrupted
290         * @throws IllegalStateException if this semaphore is already shut down
291         */
292        public synchronized void acquire() throws InterruptedException {
293            if (isShutdown()) {
294                throw new IllegalStateException("TimedSemaphore is shut down!");
295            }
296    
297            if (task == null) {
298                task = startTimer();
299            }
300    
301            boolean canPass = false;
302            do {
303                canPass = getLimit() <= NO_LIMIT || acquireCount < getLimit();
304                if (!canPass) {
305                    wait();
306                } else {
307                    acquireCount++;
308                }
309            } while (!canPass);
310        }
311    
312        /**
313         * Returns the number of (successful) acquire invocations during the last
314         * period. This is the number of times the {@link #acquire()} method was
315         * called without blocking. This can be useful for testing or debugging
316         * purposes or to determine a meaningful threshold value. If a limit is set,
317         * the value returned by this method won't be greater than this limit.
318         *
319         * @return the number of non-blocking invocations of the {@link #acquire()}
320         * method
321         */
322        public synchronized int getLastAcquiresPerPeriod() {
323            return lastCallsPerPeriod;
324        }
325    
326        /**
327         * Returns the number of invocations of the {@link #acquire()} method for
328         * the current period. This may be useful for testing or debugging purposes.
329         *
330         * @return the current number of {@link #acquire()} invocations
331         */
332        public synchronized int getAcquireCount() {
333            return acquireCount;
334        }
335    
336        /**
337         * Returns the number of calls to the {@link #acquire()} method that can
338         * still be performed in the current period without blocking. This method
339         * can give an indication whether it is safe to call the {@link #acquire()}
340         * method without risking to be suspended. However, there is no guarantee
341         * that a subsequent call to {@link #acquire()} actually is not-blocking
342         * because in the mean time other threads may have invoked the semaphore.
343         *
344         * @return the current number of available {@link #acquire()} calls in the
345         * current period
346         */
347        public synchronized int getAvailablePermits() {
348            return getLimit() - getAcquireCount();
349        }
350    
351        /**
352         * Returns the average number of successful (i.e. non-blocking)
353         * {@link #acquire()} invocations for the entire life-time of this {@code
354         * TimedSemaphore}. This method can be used for instance for statistical
355         * calculations.
356         *
357         * @return the average number of {@link #acquire()} invocations per time
358         * unit
359         */
360        public synchronized double getAverageCallsPerPeriod() {
361            return (periodCount == 0) ? 0 : (double) totalAcquireCount
362                    / (double) periodCount;
363        }
364    
365        /**
366         * Returns the time period. This is the time monitored by this semaphore.
367         * Only a given number of invocations of the {@link #acquire()} method is
368         * possible in this period.
369         *
370         * @return the time period
371         */
372        public long getPeriod() {
373            return period;
374        }
375    
376        /**
377         * Returns the time unit. This is the unit used by {@link #getPeriod()}.
378         *
379         * @return the time unit
380         */
381        public TimeUnit getUnit() {
382            return unit;
383        }
384    
385        /**
386         * Returns the executor service used by this instance.
387         *
388         * @return the executor service
389         */
390        protected ScheduledExecutorService getExecutorService() {
391            return executorService;
392        }
393    
394        /**
395         * Starts the timer. This method is called when {@link #acquire()} is called
396         * for the first time. It schedules a task to be executed at fixed rate to
397         * monitor the time period specified.
398         *
399         * @return a future object representing the task scheduled
400         */
401        protected ScheduledFuture<?> startTimer() {
402            return getExecutorService().scheduleAtFixedRate(new Runnable() {
403                public void run() {
404                    endOfPeriod();
405                }
406            }, getPeriod(), getPeriod(), getUnit());
407        }
408    
409        /**
410         * The current time period is finished. This method is called by the timer
411         * used internally to monitor the time period. It resets the counter and
412         * releases the threads waiting for this barrier.
413         */
414        synchronized void endOfPeriod() {
415            lastCallsPerPeriod = acquireCount;
416            totalAcquireCount += acquireCount;
417            periodCount++;
418            acquireCount = 0;
419            notifyAll();
420        }
421    }