Added multithreading

- Mixer creates producer consumer relations: picks a random producer,
and then a random consumer to free the memory allocated in the producer.

Author: tyleretzel

CMakeLists.txt

@@ -7,6 +7,6 @@ add_executable(memsetVsMadvise memsetVsMadvise.cpp)
 set_property(TARGET memsetVsMadvise PROPERTY CXX_STANDARD 11)
 target_link_libraries(memsetVsMadvise PRIVATE ${CMAKE_THREAD_LIBS_INIT} gflags)
 
-add_executable(stress stress_test/Main.cpp stress_test/Producers.cpp stress_test/Mixer.cpp)
+add_executable(stress stress_test/Main.cpp stress_test/Producers.cpp stress_test/Mixer.cpp stress_test/ToFreeQueue.cpp)
 set_property(TARGET stress PROPERTY CXX_STANDARD 14)
 target_link_libraries(stress PRIVATE gflags)

stress_test/Main.cpp

@@ -1,28 +1,58 @@
 #include <chrono>
 #include <iostream>
+#include <thread>
 #include <vector>
 
 #include <gflags/gflags.h>
 
 #include "Mixer.h"
 
 DEFINE_int32(num_producers, 100, "number of producers to run");
+DEFINE_int32(num_threads, 1, "number of threads to run");
 
-int main(int argc, char **argv) {
+using std::make_shared;
+using std::shared_ptr;
+using std::vector;
+
+void createAndRunMixer(vector<shared_ptr<Producer>> producers, int me,
+                       vector<shared_ptr<ToFreeQueue>> toFreeQueues) {
+  Mixer m(producers, FLAGS_num_producers, me, toFreeQueues);
+  m.run();
+}
 
+int main(int argc, char **argv) {
   gflags::ParseCommandLineFlags(&argc, &argv, true);
 
-  vector<std::unique_ptr<Producer>> producers;
-  producers.push_back(std::move(std::make_unique<SimpleProducer>(8, 100000)));
-  producers.push_back(std::move(std::make_unique<VectorProducer>(
-      100000, std::chrono::duration<double>(1.0))));
+  // Initialize producers
+  vector<shared_ptr<Producer>> producers;
+  producers.push_back(make_shared<SimpleProducer>(8, 100000));
+  producers.push_back(
+      make_shared<VectorProducer>(100000, std::chrono::duration<double>(1.0)));
+
+  // Set up a work queue for each thread
+  vector<std::thread> threads;
+  vector<shared_ptr<ToFreeQueue>> toFreeQueues;
+  for (int i = 0; i < FLAGS_num_threads; i++) {
+    shared_ptr<ToFreeQueue> toFreeQ = make_shared<ToFreeQueue>();
+    toFreeQueues.push_back(toFreeQ);
+  }
+
+  for (int i = 0; i < FLAGS_num_threads; i++) {
+    // each thread gets an arbitrary id given by [i]
+    threads.push_back(
+        std::thread(createAndRunMixer, producers, i, toFreeQueues));
+  }
 
   using namespace std::chrono;
 
-  Mixer m(std::move(producers), FLAGS_num_producers);
-
   high_resolution_clock::time_point beginTime = high_resolution_clock::now();
-  m.run();
+  for (auto it = begin(threads); it != end(threads); ++it) {
+    it->join();
+  }
+  // Cleanup any remaining memory
+  for (int i = 0; i < FLAGS_num_threads; i++) {
+    toFreeQueues[i]->freeIgnoreLifetime();
+  }
   high_resolution_clock::time_point endTime = high_resolution_clock::now();
 
   duration<double> span = duration_cast<duration<double>>(endTime - beginTime);

stress_test/Mixer.cpp

@@ -1,34 +1,33 @@
 #include "Mixer.h"
 
-Mixer::Mixer(vector<unique_ptr<Producer>> producers, int numProducers)
-    : producers_(move(producers)), producersRemaining_(numProducers),
-      producerIndexPicker_(0, this->producers_.size() - 1) {}
+Mixer::Mixer(vector<shared_ptr<Producer>> producers, int numProducers, int me,
+             vector<shared_ptr<ToFreeQueue>> toFreeQueues)
+    : producers_(producers), producersRemaining_(numProducers),
+      toFreeQueues_(toFreeQueues), me_(me),
+      producerIdPicker_(0, producers.size() - 1),
+      consumerIdPicker_(0, toFreeQueues.size() - 1) {}
 
 // Picks next producer for the mixer to run. Currently uniform random choice
-const Producer &Mixer::pick() {
-  int producerIndex = this->producerIndexPicker_(this->generator_);
+const Producer &Mixer::pickProducer() {
+  int producerIndex = this->producerIdPicker_(this->generator_);
   return *(this->producers_[producerIndex]);
 }
 
+// Picks next producer for the mixer to run. Currently uniform random choice
+ToFreeQueue &Mixer::pickConsumer() {
+  int consumerIndex = this->consumerIdPicker_(this->generator_);
+  return *(this->toFreeQueues_[consumerIndex]);
+}
+
 void Mixer::run() {
   while (this->producersRemaining_ > 0) {
-    if (this->allocated_.size() > 0 &&
-        this->allocated_.back().freeAfter() <
-            std::chrono::high_resolution_clock::now()) {
-			// deallocate something if it's lifetime has expired
-      this->allocated_.pop_back();
-    } else {
-			// otherwise run a random producer
-      Allocation a = this->pick().run();
-      if (!a.isEmpty()) {
-        this->allocated_.push_back(std::move(a));
-        sort(rbegin(this->allocated_), rend(this->allocated_));
-      }
-      producersRemaining_--;
+    this->toFreeQueues_[this->me_]->free();
+    // otherwise run a random producer
+    Allocation a = this->pickProducer().run();
+    if (!a.isEmpty()) {
+      this->pickConsumer().addToFree(std::move(a));
     }
+    producersRemaining_--;
   }
-  // cleanup remaining allocated things immediately, regardless of lifetime
-  while (!this->allocated_.empty()) {
-    this->allocated_.pop_back();
-  }
+  // Main loop will eventually cleanup memory
 }

stress_test/Mixer.h

@@ -4,22 +4,33 @@
 #include <vector>
 
 #include "Producers.h"
+#include "ToFreeQueue.h"
 
-using std::unique_ptr;
+using std::shared_ptr;
 using std::vector;
 
 class Mixer {
 public:
   void run();
-  Mixer(vector<unique_ptr<Producer>> producers, int numProducers);
+  Mixer(vector<shared_ptr<Producer>> producers, int numProducers, int me,
+        vector<shared_ptr<ToFreeQueue>> toFreeQueues);
 
 private:
   /* maintains reverse-sorted order by lifetime; i.e. [push_back] yields the
    * allocation that should be deallocated the soonest */
   vector<Allocation> allocated_;
-  vector<unique_ptr<Producer>> producers_;
+  vector<shared_ptr<Producer>> producers_;
   int producersRemaining_;
-  const Producer &pick();
-  std::uniform_int_distribution<int> producerIndexPicker_;
+  // the thread id that this mixer is running on
+  int me_;
+  // work queues for each thread indexed by thread number
+  vector<shared_ptr<ToFreeQueue>> toFreeQueues_;
+  // Picks next producer for the mixer to run. Currently uniform random choice.
+  const Producer &pickProducer();
+  // Picks a consumer to free memory allocated by a producer. Currently uniform
+  // random choice.
+  ToFreeQueue &pickConsumer();
+  std::uniform_int_distribution<int> producerIdPicker_;
+  std::uniform_int_distribution<int> consumerIdPicker_;
   std::default_random_engine generator_;
 };

stress_test/Producers.cpp

@@ -8,6 +8,10 @@ bool Allocation::operator<(const Allocation &that) const {
   return this->toFree_ < that.toFree_;
 }
 
+bool Allocation::operator>(const Allocation &that) const {
+  return !(*this < that);
+}
+
 bool Allocation::isEmpty() const { return this->toFree_.size() == 0; }
 
 std::chrono::high_resolution_clock::time_point Allocation::freeAfter() const {
@@ -41,8 +45,8 @@ Allocation SimpleProducer::run() const {
 }
 
 void swap(Allocation &a1, Allocation &a2) {
-	a1.toFree_.swap(a2.toFree_);
-	std::swap(a1.freeAfter_, a2.freeAfter_);
+  a1.toFree_.swap(a2.toFree_);
+  std::swap(a1.freeAfter_, a2.freeAfter_);
 }
 
 // Vector Producer

stress_test/Producers.h

@@ -7,6 +7,7 @@ class Allocation {
 public:
   // sorts based on [freeAfter] field
   bool operator<(const Allocation &that) const;
+  bool operator>(const Allocation &that) const;
   // true iff [this->toFree_] is empty
   bool isEmpty() const;
   std::chrono::high_resolution_clock::time_point freeAfter() const;
@@ -19,15 +20,15 @@ class Allocation {
   Allocation(Allocation const &) = delete;
   Allocation &operator=(Allocation const &) = delete;
 
-	// must define a move constructor since we deleted the copy constructor
-	Allocation(Allocation&&) = default;
-	Allocation& operator=(Allocation&&) = default;
+  // must define a move constructor since we deleted the copy constructor
+  Allocation(Allocation &&) = default;
+  Allocation &operator=(Allocation &&) = default;
 
-	// The destructor deallocates the memory in [toFree_]
+  // The destructor deallocates the memory in [toFree_]
   ~Allocation();
 
-	// needed to sort
-	friend void swap(Allocation &a1, Allocation &a2);
+  // needed to sort
+  friend void swap(Allocation &a1, Allocation &a2);
 
 private:
   std::vector<void *> toFree_;

stress_test/ToFreeQueue.cpp

@@ -0,0 +1,27 @@
+#include "ToFreeQueue.h"
+
+#include <chrono>
+#include <mutex>
+#include <thread>
+
+void ToFreeQueue::free() {
+  std::lock_guard<std::mutex> guard(this->lock_);
+
+  while (!this->q_.empty() && this->q_.top().freeAfter() <
+                                  std::chrono::high_resolution_clock::now()) {
+    this->q_.pop();
+  }
+}
+
+void ToFreeQueue::freeIgnoreLifetime() {
+  std::lock_guard<std::mutex> guard(this->lock_);
+
+  while (!this->q_.empty()) {
+    this->q_.pop();
+  }
+}
+
+void ToFreeQueue::addToFree(Allocation a) {
+  std::lock_guard<std::mutex> guard(this->lock_);
+  this->q_.push(std::move(a));
+}

stress_test/ToFreeQueue.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <queue>
+#include <thread>
+
+#include "Producers.h"
+
+class ToFreeQueue {
+public:
+  // frees all allocations whose lifetime has elapsed
+  void free();
+  // free all allocations, even if the lifetime hasn't expired
+  void freeIgnoreLifetime();
+  // Add an allocation to be freed after a particular time
+  void addToFree(Allocation a);
+
+private:
+  std::mutex lock_;
+  std::priority_queue<Allocation, std::vector<Allocation>,
+                      std::greater<Allocation>>
+      q_;
+};