Try to reproduce (PIByteArray << CAN_Raw) operator bug

experiments/can/CMakeLists.txt

@@ -3,18 +3,21 @@ cmake_policy(SET CMP0020 NEW)
 
 find_package(PIP REQUIRED)
 
+add_executable(asyncdevicepool asyncdevicepool.cpp)
+target_link_libraries(asyncdevicepool PIP can)
+
 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_dependencies(asyncdevicepool copy_dependencies)
+
     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/asyncdevicepool.cpp

@@ -0,0 +1,80 @@
+#include <asyncdevicepool.h>
+
+using namespace std::chrono;
+
+class FakeDevice : public CANDevice {
+PIOBJECT_SUBCLASS(FakeDevice, CANDevice)
+public:
+	double duration_ms;
+
+	explicit FakeDevice(int counter) : duration_ms(0.0), counter(counter) {}
+
+	bool send(const CAN_Raw &m) override {
+		counter--;
+		if (counter <= 0) {
+			if (counter == 0) duration_ms = (duration_cast<milliseconds>(high_resolution_clock::now().time_since_epoch()) - start_time).count();
+			counter = -1;
+			throw can::error(0, can_category);
+		}
+		return true;
+	}
+
+	bool open() override {
+		start_time = duration_cast<milliseconds>(high_resolution_clock::now().time_since_epoch());
+		return true;
+	}
+
+	void close() override {
+		if (read_handle != can::event::invalid_handle) can::event::close(read_handle);
+	}
+
+	can::event::handle getReadEventHandle() override {
+		if (read_handle == can::event::invalid_handle) {
+			read_handle = can::event::make();
+		}
+		return read_handle;
+	}
+
+protected:
+	bool nextMsg(CAN_Raw &rawMsg) override {
+		return false;
+	}
+
+private:
+	int counter = 0;
+	milliseconds start_time;
+};
+
+int main() {
+	int device_count = 63;
+	PIBlockingQueue<double> wait_queue(device_count);
+
+	AsyncDevicePool asyncDevicePool;
+	asyncDevicePool.set_can_error_handler([&wait_queue](const can::error& error, CANDevice* device) {
+//		can::stdout_error_handler(error, device);
+		wait_queue.put(dynamic_cast<FakeDevice*>(device)->duration_ms);
+		return false;
+	});
+
+	int it_count = 1'000'000;
+	PIVector<CANDevice*> devices;
+	for (int i = 0; i < device_count; ++i) {
+		devices << new FakeDevice(it_count / device_count);
+		asyncDevicePool.add(devices.back());
+	}
+
+	for (int i = 0; i < it_count; ++i) {
+		CAN_Raw msg = { .Id = 0x11, .Size = 4, .Data = { 0x0, 0x1, 0x2, 0x3 } };
+		asyncDevicePool.send(devices[i % device_count], msg);
+		if (i % 1000 == 999) {
+			std::this_thread::yield();
+//			std::cout << i / 1000 << "/" << counter / 1000 << std::endl;
+//			std::this_thread::sleep_for(milliseconds(1));
+		}
+	}
+
+	double duration_ms;
+	for (int i = 0; i < device_count; ++i) duration_ms = wait_queue.take();
+	std::cout << duration_ms << " ms" << std::endl;
+	return 0;
+}

experiments/can/can_receive_counter.cpp

@@ -0,0 +1,41 @@
+//
+// Created by zuma on 12.10.2020.
+//
+
+#include <asyncdevice.h>
+#include <picandevice.h>
+#include <future>
+
+int main() {
+	AsyncDevice device;
+	auto canDevice = new PICANDevice("can1");
+
+	int counter = 1;
+	std::promise<bool> is_success_promise;
+	auto is_success = is_success_promise.get_future();
+	auto closure = [&is_success_promise, &counter, canDevice](CAN_Raw msg){
+		if (msg.Id != 0x20) return;
+		if (msg.Size != sizeof(int)) {
+			is_success_promise.set_value(false);
+			throw can::error(canDevice, can::error::closed, "");
+		}
+		int recv_counter = reinterpret_cast<int&>(msg.Data);
+		if (counter != recv_counter) {
+			piCout << recv_counter << "<- miss ordering";
+//			is_success_promise.set_value(false);
+//			throw can::error(canDevice, can::error::closed, "");
+		} else {
+			piCout << recv_counter;
+		}
+		counter = recv_counter + 1;
+		if (counter - 1 == 1000) {
+			is_success_promise.set_value(true);
+			throw can::error(canDevice, can::error::closed, "");
+		}
+	};
+	CONNECTL(&device, readEvent, closure);
+
+	device.replace(canDevice);
+	bool result = is_success.get();
+	piCout << (result ? "success" : "failure");
+}

experiments/can/can_send_counter.cpp

@@ -0,0 +1,22 @@
+//
+// Created by zuma on 12.10.2020.
+//
+
+#include <asyncdevice.h>
+#include <picandevice.h>
+
+int main() {
+	AsyncDevice device;
+	device.replace(new PICANDevice("can0"));
+
+	CAN_Raw msg = { .Id = 0x20, .Size = sizeof(int) };
+	int& counter = reinterpret_cast<int&>(msg.Data);
+
+	counter = 0;
+	while (counter++ < 1000) {
+		device.send(msg);
+		piMSleep(1);
+	}
+
+	piSleep(5);
+}

experiments/pip/CMakeLists.txt

@@ -4,13 +4,10 @@ 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})
+target_link_libraries(mutex PIP)
 
 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})
+target_link_libraries(mutex_multithread PIP)
 
 add_executable(vectors vectors.cpp)
-target_include_directories(vectors PUBLIC ${PIP_INCLUDES})
-target_link_libraries(vectors ${PIP_LIBRARY})
+target_link_libraries(vectors PIP)

experiments/sm/CMakeLists.txt

@@ -4,9 +4,10 @@ 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})
+target_link_libraries(block_choice PIP)
 
 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})
+target_link_libraries(smbusdata_crash_test SMBricks_shared PIP::Crypt)
+
+add_executable(pibytearray_template_spec template_specialization.cpp)
+target_link_libraries(pibytearray_template_spec SMBricks_shared PIP::Crypt)

experiments/sm/template_specialization.cpp

@@ -0,0 +1,35 @@
+//
+// Created by Stepan on 16.10.2020.
+//
+
+#include "../../can/can_data.h"
+#include <sm_base.h>
+
+REGISTER_BUS_TYPE(CAN_Raw)
+
+PICout operator <<(PICout s, const CAN_Raw & v) {
+	s.saveControl();
+	s.setControl(0);
+	s << "{id=0x" << PIString::fromNumber(v.Id, 16) << ", size=" << v.Size << ", data=" << PIByteArray(v.Data, v.Size);
+	s.restoreControl();
+	return s;
+}
+
+int main() {
+	CAN_Raw msg = { .Id = 0x22, .Size = 8, .Data = { 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07 } };
+	piCout << "src: " << msg << "\n";
+
+	CAN_Raw newMsg;
+	PIByteArray ar;
+	ar << msg;
+	ar >> newMsg;
+	piCout << "after ByteArray: " << newMsg << "\n";
+
+	SMBusData busData;
+	busData = SMBusData::create(msg);
+	PIByteArray serializedBusData = busData.save();
+
+	SMBusData loadedBusData = SMBusData::fromByteArray(serializedBusData, busData.busType());
+	newMsg = loadedBusData.value<CAN_Raw>();
+	piCout << "after bus data:" << newMsg;
+}

sm/block.h

@@ -1,63 +0,0 @@
-#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