add producer-consumer project

Author: yuanye

producer-consumer/Makefile

@@ -0,0 +1,24 @@
+
+    ###########################################
+    #Makefile for simple programs
+    ###########################################
+INC=
+LIB= -lpthread
+CC=g++ -std=c++0x
+# display all warnings
+CC_FLAG=-Wall -g
+
+PRG=app
+OBJ=producer_consumer.o
+
+$(PRG):$(OBJ)
+	$(CC) $(INC) $(LIB) -o $@ $(OBJ) $(LIB)
+
+.SUFFIXES: .c .o .cpp
+.cpp.o:
+	$(CC) $(CC_FLAG) $(INC) -c $*.cpp -o $*.o
+
+.PRONY:clean
+clean:
+	@echo "Removing linked and compiled files......"
+	rm -f $(OBJ) $(PRG)

producer-consumer/circular_buf.h

@@ -0,0 +1,90 @@
+#pragma once
+#include <stdint.h>
+#include <vector>
+#include <stdexcept>
+#include <sstream>
+
+template<typename T>
+class CircularBuf {
+public:
+  explicit CircularBuf(size_t n) {
+    buf_ = std::vector<T>(n);
+    item_num_ = before_ = rear_ = 0;
+  }
+
+  inline bool empty() const {
+    return item_num_ == 0;
+  }
+
+  inline bool full() const {
+    return item_num_ == buf_.size();
+  }
+
+  inline size_t capacity() const {
+    return buf_.size();
+  }
+
+  inline size_t size() const {
+    return item_num_;
+  }
+
+  std::string str() const {
+    auto elems = this->items();
+    std::ostringstream oss;
+    oss << "[";
+    for (size_t i = 0; i < elems.size(); i++) {
+      oss << elems[i];
+      if (i != elems.size()-1) {
+        oss << " ";
+      }
+    }
+    oss << "]";
+    return oss.str();
+  }
+
+  inline std::vector<T> items() const {
+    std::vector<T> ret;
+    size_t count = item_num_;
+    size_t curIdx = before_;
+    while(count != 0) {
+      curIdx = next(curIdx);
+      ret.push_back(buf_[curIdx]);
+      --count;
+    }
+    return ret;
+  };
+
+  // return true on success
+  inline void push(const T& t) {
+    if (item_num_ >= buf_.size()) {
+      throw std::logic_error("size exceeded.");
+    }
+    rear_ = next(rear_);
+    buf_[rear_] = t;
+    item_num_ +=1;
+  }
+
+  inline T poll() {
+    if (item_num_ == 0) {
+      throw std::logic_error("container is empty.");
+    }
+    before_ = next(before_);
+    item_num_ -= 1;
+    return buf_[before_];
+  }
+
+  inline T peek() const {
+    if (item_num_ == 0) {
+      throw std::logic_error("container is empty.");
+    }
+    return buf_[next(before_)];
+  }
+private:
+  size_t next(size_t cur) const {
+    return cur+1 >= buf_.size()? 0:cur+1;
+  }
+  std::vector<T> buf_;
+  size_t item_num_;
+  size_t before_;  // pop element from before_+1
+  size_t rear_; // insert element to rear
+};

producer-consumer/producer_consumer.cpp

@@ -0,0 +1,79 @@
+
+#include "circular_buf.h"
+#include <mutex>
+#include <condition_variable>
+#include <thread>
+#include <chrono>
+
+std::mutex mu;
+std::condition_variable cv;
+
+void produce(CircularBuf<int> &buf, int producer_id, int product_id, size_t milli_sec = 10) {
+  std::unique_lock<std::mutex> ul(mu);
+  cv.wait(ul, [&]{
+    printf("Producer %d: buffer(at address %p) %zu / %zu\n",
+            producer_id, &buf, buf.size(), buf.capacity());
+    return !buf.full();
+  });
+  // not full
+  std::this_thread::sleep_for(std::chrono::milliseconds(milli_sec));
+  buf.push(product_id);
+  printf("Producer %d: produced %d...\n", producer_id, product_id);
+  ul.unlock();
+  cv.notify_one();
+}
+
+void consume(CircularBuf<int> &buf, int consumer_id, size_t milli_sec = 7) {
+  std::unique_lock<std::mutex> ul(mu);
+  cv.wait(ul, [&]{
+    printf("Consumer %d: buffer(at address %p) %zu / %zu \n",
+            consumer_id, &buf, buf.size(), buf.capacity());
+    return !buf.empty();
+  });
+  // not empty
+  std::this_thread::sleep_for(std::chrono::milliseconds(milli_sec));
+  int num = buf.poll();
+  printf("Consumer %d: consumed %d...\n", consumer_id, num);
+  ul.unlock();
+  cv.notify_one();
+}
+
+int main(int argc, char const *argv[]) {
+  CircularBuf<int> buf(5);
+
+  std::vector<std::thread> threads;
+  int n = std::thread::hardware_concurrency();
+  int product_num = 10;
+  // half producer, half consumer
+  for (int id = 0; id < n/2; id++) {
+    threads.push_back(std::thread([&](int producer_id){
+      // produce 10 products
+      for (int i = 1; i <= product_num; i++) {
+        produce(buf, producer_id, 500 + i);
+        printf("++++ Producer %d: produced %dth product\n", producer_id, i);
+        printf("Current buf: %s\n", buf.str().c_str());
+      }
+      printf("**** Producer %d quit\n", producer_id);
+    }, id));
+
+    threads.push_back(std::thread([&](int consumer_id){
+      // consume 10 products
+      for (int i = 1; i <= product_num; i++) {
+        consume(buf, consumer_id);
+        printf("---- Consumer %d: consumed %dth product\n", consumer_id, i);
+        printf("Current buf: %s\n", buf.str().c_str());
+      }
+      printf("**** Consumer %d quit.\n", consumer_id);
+    }, id));
+  }
+
+  printf("%lu threads created...\n", threads.size());
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+
+  return 0;
+}
+
+// compiled with: clang++ producer_consumer.cpp -lpthread -o app -std=c++11