Refactor

CMakeLists.txt

@@ -1,9 +1,12 @@
 cmake_minimum_required(VERSION 3.0)
 cmake_policy(SET CMP0020 NEW)
+set (CMAKE_CXX_STANDARD 17)
 
 project(multithread_experiments)
 
+if (DEFINED PATH_TO_SMSDK OR DEFINED ENV{SMSDK_DIR})
     find_package(SM REQUIRED)
+endif()
 
 if (CMAKE_BUILD_TYPE MATCHES Debug)
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g3 -fPIC -std=c++11")
@@ -13,40 +16,5 @@ else()
     set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -O3 -fPIC")
 endif()
 
-include_directories(experiments ${SMBRICKS_INCLUDES} ${PIP_INCLUDES} ${PROJECT_SOURCE_DIR})
-
-if(WIN32)
-    include_directories(can)
-
 add_subdirectory(can)
-
+add_subdirectory(experiments)
-    add_custom_target(copy_dependencies
-            COMMAND ${CMAKE_COMMAND} -E copy ${PCAN_LIB} ${CMAKE_CURRENT_BINARY_DIR}/PCANBasic${CMAKE_SHARED_LIBRARY_SUFFIX}
-            COMMAND ${CMAKE_COMMAND} -E copy ${VSCAN_LIB} ${CMAKE_CURRENT_BINARY_DIR}/vs_can_api${CMAKE_SHARED_LIBRARY_SUFFIX})
-
-    add_executable(can_send_multithread experiments/can_send_multithread.cpp)
-    target_link_libraries(can_send_multithread can)
-    add_dependencies(can_send_multithread copy_dependencies)
-
-    add_executable(can_send experiments/can_send.cpp)
-    target_link_libraries(can_send can)
-    add_dependencies(can_send copy_dependencies)
-endif()
-
-add_executable(mutex experiments/mutex.cpp)
-target_link_libraries(mutex ${PIP_LIBRARY} ${PIP_CONCURRENT_LIBRARY})
-
-add_executable(mutex_multithread experiments/mutex_multithread.cpp)
-target_link_libraries(mutex_multithread ${PIP_LIBRARY} ${PIP_CONCURRENT_LIBRARY})
-
-add_executable(vectors experiments/vectors.cpp)
-target_link_libraries(vectors ${PIP_LIBRARY})
-
-add_executable(block_choice experiments/block_choice.cpp)
-target_link_libraries(block_choice ${PIP_LIBRARY})
-
-add_executable(smbusdata_crash_test experiments/smbusdata_crash_test.cpp)
-target_link_libraries(smbusdata_crash_test SMBricks_shared ${PIP_LIBRARY} ${PIP_CONCURRENT_LIBRARY} ${PIP_CRYPT_LIBRARY})
-
-add_executable(packaged_task experiments/packaged_task.cpp)
-target_link_libraries(packaged_task SMBricks_shared ${PIP_LIBRARY} ${PIP_CONCURRENT_LIBRARY} ${PIP_CRYPT_LIBRARY})

experiments/CMakeLists.txt

@@ -0,0 +1,7 @@
+add_subdirectory(can)
+add_subdirectory(pip)
+#add_subdirectory(concurrent)
+
+if (DEFINED PATH_TO_SMSDK OR DEFINED ENV{SMSDK_DIR})
+    add_subdirectory(sm)
+endif()

experiments/can/CMakeLists.txt

@@ -0,0 +1,20 @@
+cmake_minimum_required(VERSION 3.0)
+cmake_policy(SET CMP0020 NEW)
+
+find_package(PIP REQUIRED)
+
+if(WIN32)
+    add_custom_target(copy_dependencies
+            COMMAND ${CMAKE_COMMAND} -E copy ${PCAN_LIB} ${CMAKE_CURRENT_BINARY_DIR}/PCANBasic${CMAKE_SHARED_LIBRARY_SUFFIX}
+            COMMAND ${CMAKE_COMMAND} -E copy ${VSCAN_LIB} ${CMAKE_CURRENT_BINARY_DIR}/vs_can_api${CMAKE_SHARED_LIBRARY_SUFFIX})
+
+    add_executable(can_send_multithread can_send_multithread.cpp)
+    target_include_directories(can_send_multithread PUBLIC ${PIP_INCLUDES} ${CAN_INCLUDES})
+    target_link_libraries(can_send_multithread can)
+    add_dependencies(can_send_multithread copy_dependencies)
+
+    add_executable(can_send can_send.cpp)
+    target_include_directories(can_send PUBLIC ${PIP_INCLUDES} ${CAN_INCLUDES})
+    target_link_libraries(can_send can)
+    add_dependencies(can_send copy_dependencies)
+endif()

experiments/can/can_send.cpp

@@ -1,16 +1,16 @@
 #include "can_send.h"
 #include <future>
-#include <picout.h>
+#include <iostream>
 
 int main() {
 	auto time1 = std::async(std::launch::deferred, [] { return test_send(PCAN_USBBUS1); });
 	auto time2 = std::async(std::launch::deferred, [] { return test_send(PCAN_USBBUS2); });
 
 	time1.wait();
-	piCout << "measurements for PCAN_USBBUS1:" << time1.get() / 1000.f << "ms";
+	std::cout << "measurements for PCAN_USBBUS1: " << time1.get() / 1000.f << " ms" << std::endl;
 
 	time2.wait();
-	piCout << "measurements for PCAN_USBBUS2:" << time2.get() / 1000.f << "ms";
+	std::cout << "measurements for PCAN_USBBUS2: " << time2.get() / 1000.f << " ms" << std::endl;
 
 	return 0;
 }

experiments/can/can_send_multithread.cpp

@@ -1,13 +1,13 @@
 #include "can_send.h"
 #include <future>
-#include <picout.h>
+#include <iostream>
 
 int main() {
 	auto time1 = std::async(std::launch::async, [] { return test_send(PCAN_USBBUS1); });
 	auto time2 = std::async(std::launch::async, [] { return test_send(PCAN_USBBUS2); });
 
-	piCout << "measurements for PCAN_USBBUS1:" << time1.get() / 1000.f << "ms";
+	std::cout << "measurements for PCAN_USBBUS1: " << time1.get() / 1000.f << " ms" << std::endl;
-	piCout << "measurements for PCAN_USBBUS2:" << time2.get() / 1000.f << "ms";
+	std::cout << "measurements for PCAN_USBBUS2: " << time2.get() / 1000.f << " ms" << std::endl;
 
 	return 0;
 }

experiments/pip/CMakeLists.txt

@@ -0,0 +1,16 @@
+cmake_minimum_required(VERSION 3.0)
+cmake_policy(SET CMP0020 NEW)
+
+find_package(PIP REQUIRED)
+
+add_executable(mutex mutex.cpp)
+target_include_directories(mutex PUBLIC ${PIP_INCLUDES})
+target_link_libraries(mutex ${PIP_LIBRARY} ${PIP_CONCURRENT_LIBRARY})
+
+add_executable(mutex_multithread mutex_multithread.cpp)
+target_include_directories(mutex_multithread PUBLIC ${PIP_INCLUDES})
+target_link_libraries(mutex_multithread ${PIP_LIBRARY} ${PIP_CONCURRENT_LIBRARY})
+
+add_executable(vectors vectors.cpp)
+target_include_directories(vectors PUBLIC ${PIP_INCLUDES})
+target_link_libraries(vectors ${PIP_LIBRARY})

experiments/pip/mutex.cpp

@@ -1,5 +1,4 @@
 #include <pimutex.h>
-#include <piconditionlock.h>
 #include <atomic>
 #include <future>
 #include <picout.h>
@@ -30,14 +29,6 @@ int main() {
 	});
 	piCout << "piMutex:" << piMutexPerformance.get() << "ms";
 
-	PIConditionLock piConditionLock;
-	auto piConditionLockPerformance = check_performance([&piConditionLock](){
-		piConditionLock.lock();
-		int i = 0; while (i < 1000) i++;
-		piConditionLock.unlock();
-	});
-	piCout << "piConditionLock:" << piConditionLockPerformance.get() << "ms";
-
 	std::mutex stdMutex;
 	auto stdMutexPerformance = check_performance([&stdMutex](){
 		stdMutex.lock();

experiments/pip/mutex_multithread.cpp

@@ -1,5 +1,4 @@
 #include <pimutex.h>
-#include <piconditionlock.h>
 #include <atomic>
 #include <future>
 #include <picout.h>
@@ -37,16 +36,6 @@ int main() {
 	});
 	piCout << "piMutex:" << piMutexPerformance << "ms";
 
-	PIConditionLock piConditionLock;
-	auto piConditionLockPerformance = check_performance([&piConditionLock](long& k){
-		piConditionLock.lock();
-		int i = 0; while (i < 1000) { i++; }
-		long res = ++k;
-		piConditionLock.unlock();
-		return res;
-	});
-	piCout << "piConditionLock:" << piConditionLockPerformance << "ms";
-
 	std::mutex stdMutex;
 	auto stdMutexPerformance = check_performance([&stdMutex](long& k){
 		stdMutex.lock();

experiments/sm/CMakeLists.txt

@@ -0,0 +1,12 @@
+cmake_minimum_required(VERSION 3.0)
+cmake_policy(SET CMP0020 NEW)
+
+find_package(SM REQUIRED)
+
+add_executable(block_choice block_choice.cpp)
+target_include_directories(block_choice PUBLIC ${SMBRICKS_INCLUDES} ${PIP_INCLUDES})
+target_link_libraries(block_choice ${PIP_LIBRARY})
+
+add_executable(smbusdata_crash_test smbusdata_crash_test.cpp)
+target_include_directories(smbusdata_crash_test PUBLIC ${SMBRICKS_INCLUDES} ${PIP_INCLUDES})
+target_link_libraries(smbusdata_crash_test SMBricks_shared ${PIP_LIBRARY} ${PIP_CRYPT_LIBRARY})

experiments/sm/block.h

@@ -0,0 +1,63 @@
+#ifndef MULTITHREAD_EXPERIMENTS_BLOCK_H
+#define MULTITHREAD_EXPERIMENTS_BLOCK_H
+
+#include <pivector.h>
+#include <iostream>
+#include <thread>
+#include <cmath>
+
+namespace sm {
+
+	struct block {
+		PIVector<block*> input_blocks;
+		PIVector<block*> output_blocks;
+		std::atomic_flag barrier = ATOMIC_FLAG_INIT;
+		const int is_calc_idx;
+		const std::chrono::microseconds calc_time;
+
+		static std::chrono::microseconds random_time() {
+			float val = powf(rand() % 1000 / 1000.f, 20.f) * 30.f * 1000.f;
+//			std::cout << int(val) << std::endl;
+			return std::chrono::microseconds(int(val));
+		}
+
+		explicit block(const int is_calc_idx) : is_calc_idx(is_calc_idx), calc_time(random_time())  {}
+
+		void calc() {
+			std::this_thread::sleep_for(calc_time);
+		}
+
+		void unlock(int locks_count = -1) {
+			if (locks_count == -1) locks_count = this->input_blocks.size();
+			for (int i = 0; i < locks_count; ++i) {
+				this->input_blocks[i]->barrier.clear(std::memory_order_release);
+			}
+			this->barrier.clear(std::memory_order_release);
+		}
+
+		bool try_lock() {
+			if (this->barrier.test_and_set(std::memory_order_acquire)) return false;
+
+			int locks_count = 0;
+			for (auto & input_block : this->input_blocks) {
+				if (input_block->barrier.test_and_set(std::memory_order_acquire)) break;
+				locks_count++;
+			}
+
+			if (locks_count == this->input_blocks.size()) {
+				return true;
+			} else {
+				unlock(locks_count);
+				return false;
+			}
+		}
+	};
+
+	struct time_report {
+		double calc_time_ms;
+		double sync_time_ms;
+	};
+
+}
+
+#endif //MULTITHREAD_EXPERIMENTS_BLOCK_H

experiments/sm/block_choice.cpp

@@ -1,4 +1,4 @@
-#include "sm/block.h"
+#include "block.h"
 #include <vector>
 #include <pimap.h>
 #include <picout.h>

experiments/sm/smbusdata_crash_test.cpp

@@ -0,0 +1,164 @@
+#include <sm_base.h>
+#include <pithreadpoolexecutor.h>
+#include <future>
+
+struct SomeLargeData {
+	uint8_t data[4]{};
+
+	SomeLargeData() { memset(data, 0xff, sizeof(data)); }
+
+	~SomeLargeData() {
+		memset(data, 0x00, sizeof(data));
+	}
+};
+
+inline PIByteArray & operator <<(PIByteArray & s, const SomeLargeData & v) {
+	s << PIByteArray::RawData(v.data, sizeof(v.data));
+	return s;
+}
+
+inline PIByteArray & operator >>(PIByteArray & s, SomeLargeData & v) {
+	if (s.size() < sizeof(v.data)) {
+		piCout << "Error in operator >> for SomeLargeData";
+		exit(1);
+	}
+	s >> PIByteArray::RawData(v.data, sizeof(v.data));
+	return s;
+}
+
+REGISTER_BUS_TYPE(SomeLargeData)
+
+void test_blocking_queue() {
+	std::atomic_bool is_end(false);
+	PIBlockingQueue<SMBlockData> out_dequeue;
+	PIBlockingQueue<SMBlockData> in_dequeue;
+
+	auto runnable = [&](){
+		while (true) {
+			SMBlockData data;
+			bool is_ok;
+			data = in_dequeue.poll(100, SMBlockData(), &is_ok);
+
+			if (!is_ok) {
+				if (is_end) break;
+				continue;
+			}
+
+			out_dequeue.put(data);
+
+//			if (!is_ok) {
+//				if (is_end) break;
+//				std::this_thread::yield();
+//			}
+		}
+	};
+
+	PIVector<std::future<void>> futures;
+	for (int i = 0; i < 4; ++i) {
+		futures.append(std::async(std::launch::async, runnable));
+	}
+
+	SomeLargeData content;
+	int iteration_count = 100 * 1000;
+	for (int i = 0; i < iteration_count / 20; ++i) {
+		for (int j = 0; j < 20; ++j) {
+			SMBlockData block_data(j+1);
+			for (int k = 0; k < j+1; ++k) {
+				block_data[k].value<SomeLargeData>() = content;
+			}
+//			bus_data << SMBusData::create(content);
+			SMBlockData out_data = block_data.clone();
+			in_dequeue.put(out_data);
+		}
+		for (int j = 0; j < 20; ++j) {
+			auto block_data = out_dequeue.take();
+			if (block_data[0].isInvalid()) {
+				piCout << "Error: bus_data is invalid";
+				exit(1);
+			}
+		}
+//		printf("It's alive! %d\n", i);
+	}
+
+	is_end = true;
+	for (auto& future: futures) future.get();
+}
+
+void test_mutexes() {
+	std::atomic_bool is_end(false);
+	PIDeque<SMBlockData> out_dequeue;
+	PIDeque<SMBlockData> in_dequeue;
+	PIMutex out_mutex;
+	PIMutex in_mutex;
+
+	auto runnable = [&](){
+		while (true) {
+			SMBlockData data;
+			in_mutex.lock();
+			bool is_ok = !in_dequeue.isEmpty();
+			if (is_ok) data = in_dequeue.take_front();
+			in_mutex.unlock();
+
+			if (!is_ok) {
+				if (is_end) break;
+				std::this_thread::yield();
+				continue;
+			}
+
+			out_mutex.lock();
+			out_dequeue.push_back(data);
+			out_mutex.unlock();
+
+//			if (!is_ok) {
+//				if (is_end) break;
+//				std::this_thread::yield();
+//			}
+		}
+	};
+
+	PIVector<std::future<void>> futures;
+	for (int i = 0; i < 4; ++i) {
+		futures.append(std::async(std::launch::async, runnable));
+	}
+
+	SomeLargeData content;
+	int iteration_count = 100 * 1000;
+	for (int i = 0; i < iteration_count / 20; ++i) {
+		for (int j = 0; j < 20; ++j) {
+			SMBlockData block_data(j+1);
+			for (int k = 0; k < j+1; ++k) {
+				block_data[k].value<SomeLargeData>() = content;
+			}
+//			bus_data << SMBusData::create(content);
+			SMBlockData out_data = block_data.clone();
+			in_mutex.lock();
+			in_dequeue.push_back(out_data);
+			in_mutex.unlock();
+		}
+		for (int j = 0; j < 20; ++j) {
+			out_mutex.lock();
+			if (out_dequeue.isEmpty()) {
+				out_mutex.unlock();
+				j--;
+				std::this_thread::yield();
+				continue;
+			}
+			auto block_data = out_dequeue.take_front();
+			out_mutex.unlock();
+			if (block_data[0].isInvalid()) {
+				piCout << "Error: bus_data is invalid";
+				exit(1);
+			}
+		}
+//		printf("It's alive! %d\n", i);
+	}
+
+	is_end = true;
+	for (auto& future: futures) future.get();
+
+}
+
+int main() {
+	test_blocking_queue();
+	return 0;
+}

experiments/smbusdata_crash_test.cpp

@@ -1,77 +0,0 @@
-#include <sm_base.h>
-#include <executor.h>
-
-struct SomeLargeData {
-	uint8_t data[4]{};
-
-	SomeLargeData() { memset(data, 0xff, sizeof(data)); }
-
-	~SomeLargeData() {
-		memset(data, 0x00, sizeof(data));
-	}
-};
-
-inline PIByteArray & operator <<(PIByteArray & s, const SomeLargeData & v) {
-	s << PIByteArray::RawData(v.data, sizeof(v.data));
-	return s;
-}
-
-inline PIByteArray & operator >>(PIByteArray & s, SomeLargeData & v) {
-	if (s.size() < sizeof(v.data)) {
-		piCout << "Error in operator >> for SomeLargeData";
-		exit(1);
-	}
-	s >> PIByteArray::RawData(v.data, sizeof(v.data));
-	return s;
-}
-
-REGISTER_BUS_TYPE(SomeLargeData)
-
-int main() {
-	std::atomic_bool is_end(false);
-	PIThreadPoolExecutor executor(4);
-	PIBlockingDequeue<SMBlockData> out_dequeue;
-	PIBlockingDequeue<SMBlockData> in_dequeue;
-
-	executor.execute([&](){
-		while (true) {
-			bool is_ok;
-			SMBlockData data;
-			data = in_dequeue.poll(100, data, &is_ok);
-			if (!is_ok) {
-				if (is_end) break;
-			}
-
-			out_dequeue.offer(data, 100);
-			if (!is_ok) {
-				if (is_end) break;
-			}
-		}
-	});
-
-	SomeLargeData content;
-	int iteration_count = 100 * 1000;
-	for (int i = 0; i < iteration_count / 20; ++i) {
-		for (int j = 0; j < 20; ++j) {
-			SMBlockData block_data(j+1);
-			for (int k = 0; k < j+1; ++k) {
-				block_data[k].sharedData<SomeLargeData>() = content;
-			}
-//			bus_data << SMBusData::create(content);
-			in_dequeue.offer(block_data.clone());
-		}
-		for (int j = 0; j < 20; ++j) {
-			auto block_data = out_dequeue.take();
-			if (block_data[0].isInvalid()) {
-				piCout << "Error: bus_data is invalid";
-				exit(1);
-			}
-		}
-//		printf("It's alive! %d\n", i);
-	}
-
-	is_end = true;
-	executor.shutdownNow();
-
-	return 0;
-}