main_concurrent_queue.cpp

#include "../implementation/concurrent_queue.hpp"
#include "parameter_processor.hpp"

#include <chrono>
#include <functional>
#include <iostream>
#include <thread>

// std::atomic<size_t> num_inserted = 0;

size_t Q_SIZE;
size_t NUM_PRODUCERS;
size_t NUM_CONSUMERS;
size_t NUM_ELEMENTS_FOR_EACH;

uint EOW = uint();

void produce(uint id, ConcurrentQueue<uint> &q) {
  const size_t begin = id * NUM_ELEMENTS_FOR_EACH;
  const size_t end   = begin + NUM_ELEMENTS_FOR_EACH;
  for (size_t i = begin; i < end; ++i) {
    // num_inserted++;

    // std::cout << "p: " << i << std::endl;
    q.push_back(id, i + 1);
  }
  // std::cout << "done pushing: " << id << std::endl;
}

void consume(uint id, ConcurrentQueue<uint> &q) {
  bool active = true;
  while (active) {
    uint current = q.pop_front(id);
    if (current == EOW) {
      active = false;
    }
  }
}

template <class T> void print(std::ostream &out, const T &t, size_t w) {
  out.width(w);
  out << t << " " << std::flush;
  // std::cout.width(w);
  // std::cout << t << " " << std::flush;
}

void print_timing(std::ostream &out, size_t s, size_t p, size_t c, size_t w,
                  double construction, double work) {
  print(out, s, 9);
  print(out, p, 3);
  print(out, c, 3);
  print(out, w, 9);
  print(out, construction, 16);
  print(out, work, 13);
  out << std::endl;
}

int main(int argc, char *argv[]) {
  ParameterProcessor param(argc, argv);
  Q_SIZE                = param.getInt("s", 10);
  NUM_PRODUCERS         = param.getInt("p", 2);
  NUM_CONSUMERS         = param.getInt("c", 1);
  NUM_ELEMENTS_FOR_EACH = param.getInt("w", 9);

  // double distance = Dijkstra(id1, id2, graph);

  auto q_construction_start = std::chrono::high_resolution_clock::now();
  ConcurrentQueue<uint> q(Q_SIZE, NUM_PRODUCERS + NUM_CONSUMERS,
                          NUM_ELEMENTS_FOR_EACH * NUM_PRODUCERS);
  auto q_construction_end = std::chrono::high_resolution_clock::now();

  std::vector<std::thread> producers;
  producers.reserve(NUM_PRODUCERS);

  std::vector<std::thread> consumers;
  consumers.reserve(NUM_CONSUMERS);

  auto work_start = std::chrono::high_resolution_clock::now();
  for (size_t i = 0; i < NUM_PRODUCERS; ++i) {
    producers.push_back(std::thread(produce, i, std::ref(q)));
  }

  // for (size_t i = 0; i < NUM_PRODUCERS; ++i) {
  //   producers[i].join();
  // }
  // for (auto e : check) {
  //   if (!e) {
  //     std::cout << "Element was not inserted" << std::endl;
  //   }
  // }

  for (size_t i = 0; i < NUM_CONSUMERS; ++i) {
    consumers.push_back(std::thread(consume, NUM_PRODUCERS + i, std::ref(q)));
  }

  for (size_t i = 0; i < NUM_PRODUCERS; ++i) {
    producers[i].join();
  }

  for (size_t i = 0; i < NUM_CONSUMERS; ++i) {
    consumers[i].join();
  }

  auto work_end = std::chrono::high_resolution_clock::now();

  double q_construction_time =
      std::chrono::duration_cast<std::chrono::microseconds>(
          q_construction_end - q_construction_start)
          .count() /
      1000.;

  double work_time = (std::chrono::duration_cast<std::chrono::microseconds>(
                          work_end - work_start)
                          .count() /
                      1000.) /
                     (NUM_ELEMENTS_FOR_EACH * NUM_PRODUCERS / 1000.0);

  // print_headline(cout);
  print_timing(cout, Q_SIZE, NUM_PRODUCERS, NUM_CONSUMERS,
               NUM_ELEMENTS_FOR_EACH, q_construction_time, work_time);

  // if (num_inserted != NUM_PRODUCERS * NUM_ELEMENTS_FOR_EACH) {
  //   std::cerr << "Error: " << num_inserted << " inserted not "
  //             << NUM_PRODUCERS * NUM_ELEMENTS_FOR_EACH << std::endl;
  // }

  return 0;
}