include/simgrid/plugins/ProducerConsumer.hpp

   1 /* Copyright (c) 2021. The SimGrid Team. All rights reserved.          */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 #ifndef SIMGRID_PLUGIN_PRODUCERCONSUMER_HPP
   7 #define SIMGRID_PLUGIN_PRODUCERCONSUMER_HPP
   8
   9 #include <simgrid/s4u/ConditionVariable.hpp>
  10 #include <simgrid/s4u/Mailbox.hpp>
  11 #include <simgrid/s4u/Mutex.hpp>
  12 #include <xbt/asserts.h>
  13 #include <xbt/log.h>
  14
  15 #include <atomic>
  16 #include <climits>
  17 #include <queue>
  18 #include <string>
  19
  20 XBT_LOG_EXTERNAL_CATEGORY(producer_consumer);
  21
  22 /** Stock implementation of a generic monitored queue to solve the producer-consumer problem */
  23
  24 namespace simgrid {
  25 namespace plugin {
  26
  27 template <typename T> class ProducerConsumer;
  28 template <typename T> using ProducerConsumerPtr = boost::intrusive_ptr<ProducerConsumer<T>>;
  29
  30 XBT_PUBLIC_DATA unsigned long pc_id;
  31
  32 template <typename T> class ProducerConsumer {
  33 public:
  34   /** This ProducerConsumer plugin can use two different transfer modes:
  35    *   - TransferMode::MAILBOX: this mode induces a s4u::Comm between the actors doing the calls to put() and get().
  36    *     If these actors are on the same host, this communication goes through the host's loopback and can thus be
  37    *     seen as a memory copy. Otherwise, data goes over the network.
  38    *   - TransferMode::QUEUE: data is internally stored in a std::queue. Putting and getting data to and from this
  39    *     data structure has a zero-cost in terms of simulated time.
  40    *  Both modes guarantee that the data is consumed in the order it has been produced. However, when data goes
  41    *  through the network, s4u::Comm are started in the right order, but may complete in a different order depending
  42    *  the characteristics of the different interconnections between host pairs.
  43    */
  44   enum class TransferMode { MAILBOX = 0, QUEUE };
  45
  46 private:
  47   std::string id;
  48
  49   /* Implementation of a Monitor to handle the data exchanges */
  50   s4u::MutexPtr mutex_;
  51   s4u::ConditionVariablePtr can_put_;
  52   s4u::ConditionVariablePtr can_get_;
  53
  54   /* data containers for each of the transfer modes */
  55   s4u::Mailbox* mbox_ = nullptr;
  56   std::queue<T*> queue_;
  57
  58   unsigned int max_queue_size_ = 1;
  59   TransferMode tmode_          = TransferMode::MAILBOX;
  60
  61   /* Refcounting management */
  62   std::atomic_int_fast32_t refcount_{0};
  63   friend void intrusive_ptr_add_ref(ProducerConsumer* pc) { pc->refcount_.fetch_add(1, std::memory_order_acq_rel); }
  64
  65   friend void intrusive_ptr_release(ProducerConsumer* pc)
  66   {
  67     if (pc->refcount_.fetch_sub(1, std::memory_order_release) == 1) {
  68       std::atomic_thread_fence(std::memory_order_acquire);
  69       delete pc;
  70     }
  71   }
  72
  73   explicit ProducerConsumer(unsigned int max_queue_size) : max_queue_size_(max_queue_size)
  74   {
  75     xbt_assert(max_queue_size > 0, "Max queue size of 0 is not allowed");
  76
  77     id = std::string("ProducerConsumer") + std::to_string(pc_id);
  78     pc_id++;
  79
  80     mutex_   = s4u::Mutex::create();
  81     can_put_ = s4u::ConditionVariable::create();
  82     can_get_ = s4u::ConditionVariable::create();
  83
  84     if (tmode_ == TransferMode::MAILBOX)
  85       mbox_ = s4u::Mailbox::by_name(id);
  86   }
  87   ~ProducerConsumer() = default;
  88
  89 public:
  90   /** Creation of the monitored queue. Its size can be bounded by passing a strictly positive value to 'max_queue_size'
  91    *  as parameter. Calling 'create()' means that the queue size is (virtually) infinite.
  92    */
  93   static ProducerConsumerPtr<T> create(unsigned int max_queue_size = UINT_MAX)
  94   {
  95     return ProducerConsumerPtr<T>(new ProducerConsumer<T>(max_queue_size));
  96   }
  97
  98   /** This method is intended more to set the maximum queue size in a fluent way than changing the size during the
  99    *  utilization of the ProducerConsumer. Hence, the modification occurs in a critical section to prevent
 100    *  inconsistencies.
 101    */
 102   ProducerConsumer* set_max_queue_size(unsigned int max_queue_size)
 103   {
 104     std::unique_lock<s4u::Mutex> lock(*mutex_);
 105     max_queue_size_ = max_queue_size;
 106     return this;
 107   }
 108
 109   unsigned int get_max_queue_size() const { return max_queue_size_; }
 110
 111   /** The underlying data container (and transfer mode) can only be modified when the queue is empty.*/
 112   ProducerConsumer* set_transfer_mode(TransferMode new_mode)
 113   {
 114     if (tmode_ == new_mode) /* No change, do nothing */
 115       return this;
 116
 117     xbt_assert(empty(), "cannot change transfer mode when some data is in queue");
 118     if (new_mode == TransferMode::MAILBOX) {
 119       mbox_ = s4u::Mailbox::by_name(id);
 120     } else {
 121       mbox_ = nullptr;
 122     }
 123     tmode_ = new_mode;
 124     return this;
 125   }
 126   std::string get_transfer_mode() const { return tmode_ == TransferMode::MAILBOX ? "mailbox" : "queue"; }
 127
 128   /** Container-agnostic size() method */
 129   unsigned int size() { return tmode_ == TransferMode::MAILBOX ? mbox_->size() : queue_.size(); }
 130
 131   /** Container-agnostic empty() method */
 132   bool empty() { return tmode_ == TransferMode::MAILBOX ? mbox_->empty() : queue_.empty(); }
 133
 134   /** Asynchronous put() of a data item of a given size
 135    *  - TransferMode::MAILBOX: if put_async is called directly from user code, it can be considered to be done in a
 136    *    fire-and-forget mode. No need to save the s4u::CommPtr.
 137    *  - TransferMode::QUEUE: the data is simply pushed into the queue.
 138    */
 139   s4u::CommPtr put_async(T* data, size_t simulated_size_in_bytes)
 140   {
 141     std::unique_lock<s4u::Mutex> lock(*mutex_);
 142     s4u::CommPtr comm = nullptr;
 143     XBT_CVERB(producer_consumer, (size() < max_queue_size_) ? "can put" : "must wait");
 144
 145     while (size() >= max_queue_size_)
 146       can_put_->wait(lock);
 147     if (tmode_ == TransferMode::MAILBOX) {
 148       comm = mbox_->put_async(data, simulated_size_in_bytes);
 149     } else
 150       queue_.push(data);
 151     can_get_->notify_all();
 152     return comm;
 153   }
 154
 155   /** Synchronous put() of a data item of a given size
 156    *  - TransferMode::MAILBOX: the caller must wait for the induced communication with the getter of the data to be
 157    *    complete to continue with its execution. This wait is done outside of the monitor to prevent serialization.
 158    *  - TransferMode::QUEUE: the behavior is exactly the same as put_async: data is simply pushed into the queue.
 159    */
 160   void put(T* data, size_t simulated_size_in_bytes)
 161   {
 162     s4u::CommPtr comm = put_async(data, simulated_size_in_bytes);
 163     if (comm) {
 164       XBT_CDEBUG(producer_consumer, "Waiting for the data to be consumed");
 165       comm->wait();
 166     }
 167   }
 168
 169   /** Asynchronous get() of a 'data'
 170    *  - TransferMode::MAILBOX: the caller is returned a s4u::CommPtr onto which it can wait when the data is really
 171    *    needed.
 172    *  - TransferMode::QUEUE: the data is simply popped from the queue and directly available. Better to call get() in
 173    *    this transfer mode.
 174    */
 175   s4u::CommPtr get_async(T** data)
 176   {
 177     std::unique_lock<s4u::Mutex> lock(*mutex_);
 178     s4u::CommPtr comm = nullptr;
 179     XBT_CVERB(producer_consumer, empty() ? "must wait" : "can get");
 180     while (empty())
 181       can_get_->wait(lock);
 182     if (tmode_ == TransferMode::MAILBOX)
 183       comm = mbox_->get_async<T>(data);
 184     else {
 185       *data = queue_.front();
 186       queue_.pop();
 187     }
 188     can_put_->notify_all();
 189
 190     return comm;
 191   }
 192
 193   /** Synchronous get() of a 'data'
 194    *  - TransferMode::MAILBOX: the caller waits (outside the monitor to prevent serialization) for the induced
 195    *    communication to be complete to continue with its execution.
 196    *  - TransferMode::QUEUE: the behavior is exactly the same as get_async: data is simply popped from the queue and
 197    *    directly available to the caller.
 198    */
 199   T* get()
 200   {
 201     T* data;
 202     s4u::CommPtr comm = get_async(&data);
 203     if (comm) {
 204       XBT_CDEBUG(producer_consumer, "Waiting for the data to arrive");
 205       comm->wait();
 206     }
 207     XBT_CDEBUG(producer_consumer, "data is available");
 208     return data;
 209   }
 210 };
 211
 212 } // namespace plugin
 213 } // namespace simgrid
 214
 215 #endif // SIMGRID_PLUGIN_PRODUCERCONSUMER_HPP