solve lab1a in cpp

lab1a/cpp/.gitignore

@@ -0,0 +1 @@
+build

lab1a/cpp/CMakeLists.txt

@@ -0,0 +1,8 @@
+cmake_minimum_required(VERSION 3.12)
+project(lab1a)
+
+set(CMAKE_CXX_STANDARD 20)
+
+add_executable(lab1a main.cpp)
+
+TARGET_LINK_LIBRARIES(lab1a pthread)

lab1a/cpp/README.md

@@ -0,0 +1,9 @@
+# Lab1a
+
+```cpp
+mkdir -p build
+cd build
+cmake ..
+make
+./lab1a
+```

lab1a/cpp/common.h

@@ -0,0 +1,8 @@
+#pragma once
+#include <string>
+
+struct DataEntry {
+	std::string name;
+	float sugar;
+	int criteria;
+};

lab1a/cpp/limited_buffer.cpp

@@ -0,0 +1,76 @@
+#include <vector>
+#include <mutex>
+#include <condition_variable>
+
+#include "common.h"
+
+class LimitedBuffer {
+	std::vector<DataEntry> items;
+	int capacity;
+	std::mutex items_mutex;
+	std::condition_variable items_cv;
+
+public:
+	int estimated_size;
+	int consumed_count;
+	int processed_count;
+	std::mutex processed_mutex;
+
+	LimitedBuffer(int capacity) {
+		this->capacity = capacity;
+		estimated_size = 0;
+		consumed_count = 0;
+		processed_count = 0;
+	}
+
+	void insert(DataEntry entry);
+	DataEntry remove();
+
+	void update_estimated(int change);
+	bool consume_estimated();
+	void mark_processed();
+	bool is_done();
+};
+
+void LimitedBuffer::insert(DataEntry entry) {
+	std::unique_lock<std::mutex> guard(items_mutex);
+	items_cv.wait(guard, [&](){ return items.size() < this->capacity; });
+
+	items.push_back(entry);
+	items_cv.notify_all();
+}
+
+DataEntry LimitedBuffer::remove() {
+	std::unique_lock<std::mutex> guard(items_mutex);
+	items_cv.wait(guard, [&](){ return items.size() > 0; });
+
+	DataEntry entry = items[items.size()-1];
+	items.pop_back();
+	items_cv.notify_all();
+	return entry;
+}
+
+void LimitedBuffer::update_estimated(int change) {
+	std::unique_lock<std::mutex> guard(processed_mutex);
+	estimated_size += change;
+}
+
+bool LimitedBuffer::consume_estimated() {
+	std::unique_lock<std::mutex> guard(processed_mutex);
+
+	if (consumed_count == estimated_size) {
+		return false;
+	} else {
+		consumed_count++;
+		return true;
+	}
+}
+
+void LimitedBuffer::mark_processed() {
+	std::unique_lock<std::mutex> guard(processed_mutex);
+	processed_count++;
+}
+
+bool LimitedBuffer::is_done() {
+	return estimated_size == processed_count;
+}

lab1a/cpp/main.cpp

@@ -0,0 +1,138 @@
+#include <algorithm>
+#include <chrono>
+#include <thread>
+#include <iostream>
+#include <fstream>
+#include <string>
+#include <vector>
+
+#include "json.hpp"
+
+using namespace std;
+using json = nlohmann::json;
+
+const int inputBufferSize = 10;
+const int inputMonitorCount = 3;
+
+const int outputBufferSize = 10;
+const int outputMonitorCount = 3;
+
+#include "common.h"
+#include "limited_buffer.cpp"
+
+bool filterEntry(DataEntry *entry) {
+	this_thread::sleep_for(chrono::milliseconds(100 + entry->criteria));
+	return entry->sugar > entry->criteria;
+}
+
+void outputEntry(vector<DataEntry> *entries, DataEntry *entry) {
+	this_thread::sleep_for(chrono::milliseconds(200));
+	cout << "Output: " << entry->name << endl;
+	entries->push_back(*entry);
+	sort(entries->begin(), entries->end(), [](DataEntry &a, DataEntry &b) {
+			return a.sugar < b.sugar;
+	});
+}
+
+void filterEntries(LimitedBuffer *inputBuffer, LimitedBuffer *outputBuffer) {
+	while (!inputBuffer->is_done()) {
+		if (inputBuffer->consume_estimated()) {
+			DataEntry entry = inputBuffer->remove();
+			cout << "Started to filter: " << entry.name << endl;
+			bool is_filtered = filterEntry(&entry);
+			cout << "Finished to filter: " << entry.name << endl;
+
+			if (is_filtered) {
+				outputBuffer->update_estimated(+1);
+				outputBuffer->insert(entry);
+			}
+
+			inputBuffer->mark_processed();
+		}
+	}
+}
+
+void outputEntries(LimitedBuffer *inputBuffer, LimitedBuffer *outputBuffer, vector<DataEntry> *outputList, mutex *outputMutex) {
+	while (!inputBuffer->is_done() || !outputBuffer->is_done()) {
+		if (outputBuffer->consume_estimated()) {
+			DataEntry entry = outputBuffer->remove();
+			cout << "Started to output: " << entry.name << endl;
+			outputEntry(outputList, &entry);
+			cout << "Finished to output: " << entry.name << endl;
+
+			outputBuffer->mark_processed();
+		}
+	}
+}
+
+int main(int argc, char **argv) {
+	if (argc != 3) {
+		cout << "Usage: " << argv[0] << " <data-file> <output-file>" << endl;
+		return -1;
+	}
+
+	char *inputPath = argv[1];
+	char *outputPath = argv[2];
+
+	std::ifstream f(inputPath);
+	json data = json::parse(f);
+
+	vector<DataEntry> entries;
+	for (auto it : data) {
+		entries.push_back({
+				.name = it["name"],
+				.sugar = it["sugar"],
+				.criteria = it["criteria"],
+		});
+	}
+
+	vector<DataEntry> outputList;
+	mutex outputListMutex;
+
+	LimitedBuffer inputBuffer(inputBufferSize);
+	LimitedBuffer outputBuffer(outputBufferSize);
+	inputBuffer.update_estimated(entries.size());
+
+	vector<thread> threads;
+	for (int i = 0; i < inputMonitorCount; i++) {
+		threads.push_back(thread(filterEntries, &inputBuffer, &outputBuffer));
+	}
+	for (int i = 0; i < outputMonitorCount; i++) {
+		threads.push_back(thread(outputEntries, &inputBuffer, &outputBuffer, &outputList, &outputListMutex));
+	}
+
+	for (int i = 0; i < entries.size(); i++) {
+		inputBuffer.insert(entries[i]);
+	}
+
+	for (int i = 0; i < threads.size(); i++) {
+		threads[i].join();
+	}
+
+	std::ofstream outputFile(outputPath);
+	outputFile << "[" << endl;
+	for (int i = 0; i < outputList.size(); i++) {
+		DataEntry *entry = &outputList[i];
+		json json_entry;
+		json_entry["name"] = entry->name;
+		json_entry["sugar"] = entry->sugar;
+		json_entry["criteria"] = entry->criteria;
+
+		std::string json_str = json_entry.dump();
+		outputFile << "\t" << json_str;
+		if (i < outputList.size()-1) {
+			outputFile << ",";
+		}
+		outputFile << endl;
+	}
+	outputFile << "]" << endl;
+
+	cout << "Finished" << endl;
+	return 0;
+}
+
+void execute(const string &name) {
+	cout << name << ": one" << endl;
+	cout << name << ": two" << endl;
+	cout << name << ": three" << endl;
+}

lab1a/cpp/run.sh

@@ -0,0 +1,6 @@
+#!/bin/sh
+mkdir -p build
+cd build
+cmake ..
+make
+./lab1a ../$1 ../$2