git.sesse.net Git - casparcg/blob - common/semaphore.h

   1 /*
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>
   3 *
   4 * This file is part of CasparCG (www.casparcg.com).
   5 *
   6 * CasparCG is free software: you can redistribute it and/or modify
   7 * it under the terms of the GNU General Public License as published by
   8 * the Free Software Foundation, either version 3 of the License, or
   9 * (at your option) any later version.
  10 *
  11 * CasparCG is distributed in the hope that it will be useful,
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 * GNU General Public License for more details.
  15 *
  16 * You should have received a copy of the GNU General Public License
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.
  18 *
  19 * Author: Helge Norberg, helge.norberg@svt.se
  20 */
  21
  22 #pragma once
  23
  24 #include <cmath>
  25
  26 #include <boost/noncopyable.hpp>
  27 #include <boost/thread/mutex.hpp>
  28 #include <boost/thread/condition_variable.hpp>
  29
  30 namespace caspar {
  31
  32 template <class N, class Func>
  33 void repeat_n(N times_to_repeat_block, const Func& func)
  34 {
  35         for (N i = 0; i < times_to_repeat_block; ++i)
  36         {
  37                 func();
  38         }
  39 }
  40
  41 /**
  42  * Counting semaphore modelled after java.util.concurrent.Semaphore
  43  */
  44 class semaphore : boost::noncopyable
  45 {
  46         mutable boost::mutex mutex_;
  47         unsigned int permits_;
  48         boost::condition_variable_any permits_available_;
  49 public:
  50         /**
  51          * Constructor.
  52          *
  53          * @param permits The initial number of permits.
  54          */
  55         explicit semaphore(unsigned int permits)
  56                 : permits_(permits)
  57         {
  58         }
  59
  60         /**
  61          * Release a permit.
  62          */
  63         void release()
  64         {
  65                 boost::unique_lock<boost::mutex> lock(mutex_);
  66
  67                 ++permits_;
  68
  69                 permits_available_.notify_one();
  70         }
  71
  72         /**
  73          * Release a permit.
  74          *
  75          * @param permits The number of permits to release.
  76          */
  77         void release(unsigned int permits)
  78         {
  79                 boost::unique_lock<boost::mutex> lock(mutex_);
  80
  81                 permits_ += permits;
  82
  83                 repeat_n(permits, [this] { permits_available_.notify_one(); });
  84         }
  85
  86         /**
  87          * Acquire a permit. Will block until one becomes available if no permit is
  88          * currently available.
  89          */
  90         void acquire()
  91         {
  92                 boost::unique_lock<boost::mutex> lock(mutex_);
  93
  94                 while (permits_ == 0u)
  95                 {
  96                         permits_available_.wait(lock);
  97                 }
  98
  99                 --permits_;
 100         }
 101
 102         /**
 103          * Acquire a number of permits. Will block until the given number of
 104          * permits has been acquired if not enough permits are currently available.
 105          *
 106          * @param permits The number of permits to acquire.
 107          */
 108         void acquire(unsigned int permits)
 109         {
 110                 boost::unique_lock<boost::mutex> lock(mutex_);
 111                 auto num_acquired = 0u;
 112
 113                 while (true)
 114                 {
 115                         auto num_wanted = permits - num_acquired;
 116                         auto to_drain = std::min(num_wanted, permits_);
 117
 118                         permits_ -= to_drain;
 119                         num_acquired += to_drain;
 120
 121                         if (num_acquired == permits)
 122                                 break;
 123
 124                         permits_available_.wait(lock);
 125                 }
 126         }
 127
 128         /**
 129          * Acquire a number of permits. Will block until the given number of
 130          * permits has been acquired if not enough permits are currently available
 131          * or the timeout has passed.
 132          *
 133          * @param permits The number of permits to acquire.
 134          * @param timeout The timeout (will be used for each permit).
 135          *
 136          * @return whether successfully acquired within timeout or not.
 137          */
 138         template <typename Rep, typename Period>
 139         bool try_acquire(unsigned int permits, const boost::chrono::duration<Rep, Period>& timeout)
 140         {
 141                 boost::unique_lock<boost::mutex> lock(mutex_);
 142                 auto num_acquired = 0u;
 143
 144                 while (true)
 145                 {
 146                         auto num_wanted = permits - num_acquired;
 147                         auto to_drain = std::min(num_wanted, permits_);
 148
 149                         permits_ -= to_drain;
 150                         num_acquired += to_drain;
 151
 152                         if (num_acquired == permits)
 153                                 break;
 154
 155                         if (permits_available_.wait_for(lock, timeout) == boost::cv_status::timeout)
 156                         {
 157                                 lock.unlock();
 158                                 release(num_acquired);
 159                                 return false;
 160                         }
 161                 }
 162
 163                 return true;
 164         }
 165
 166         /**
 167          * Acquire one permits if permits are currently available. Does not block
 168          * until one is available, but returns immediately if unavailable.
 169          *
 170          * @return true if a permit was acquired or false if no permits where
 171          *         currently available.
 172          */
 173         bool try_acquire()
 174         {
 175                 boost::unique_lock<boost::mutex> lock(mutex_);
 176
 177                 if (permits_ == 0u)
 178                         return false;
 179                 else
 180                 {
 181                         --permits_;
 182
 183                         return true;
 184                 }
 185         }
 186
 187         /**
 188          * @return the current number of permits (may have changed at the time of
 189          *         return).
 190          */
 191         unsigned int permits() const
 192         {
 193                 boost::unique_lock<boost::mutex> lock(mutex_);
 194
 195                 return permits_;
 196         }
 197 };
 198
 199 /**
 200  * Enables RAII-style acquire/release on scope exit unless committed.
 201  */
 202 class acquire_transaction : boost::noncopyable
 203 {
 204         semaphore& semaphore_;
 205         bool committed_;
 206 public:
 207         /**
 208          * Constructor.
 209          *
 210          * @param semaphore        The semaphore to acquire one permit from.
 211          * @param already_acquired Whether a permit has already been acquired or not.
 212          */
 213         acquire_transaction(semaphore& semaphore, bool already_acquired = false)
 214                 : semaphore_(semaphore)
 215                 , committed_(false)
 216         {
 217                 if (!already_acquired)
 218                         semaphore_.acquire();
 219         }
 220
 221         /**
 222          * Destructor that will release one permit if commit() has not been called.
 223          */
 224         ~acquire_transaction()
 225         {
 226                 if (!committed_)
 227                         semaphore_.release();
 228         }
 229
 230         /**
 231          * Ensure that the acquired permit is kept on destruction.
 232          */
 233         void commit()
 234         {
 235                 committed_ = true;
 236         }
 237 };
 238
 239 }