Add device sharing support

2026-08-05 01:18:36 +02:00 · 2022-12-13 11:46:32 -05:00
parent 78465e0f20
commit a9157c82e5
46 changed files with 1946 additions and 102 deletions
@@ -14,6 +14,10 @@
 #include "icsneo/communication/message/readsettingsmessage.h"
 #include "icsneo/communication/message/versionmessage.h"

+#ifdef ICSNEO_ENABLE_DEVICE_SHARING
+#include "icsneo/communication/socket.h"
+#endif
+
 using namespace icsneo;

 int Communication::messageCallbackIDCounter = 1;
@@ -67,6 +71,11 @@ bool Communication::isDisconnected() {
 	return driver->isDisconnected();
 }

+void Communication::modifyRawCallbacks(std::function<void(std::list<Communication::RawCallback>&)>&& cb) {
+	std::scoped_lock lk(rawCallbacksMutex);
+	cb(rawCallbacks);
+}
+
 bool Communication::sendPacket(std::vector<uint8_t>& bytes) {
 	// This is here so that other communication types (like multichannel) can override it
 	return rawWrite(bytes);
@@ -217,6 +226,15 @@ std::shared_ptr<Message> Communication::waitForMessageSync(std::function<bool(vo
 	std::condition_variable cv;
 	std::shared_ptr<Message> returnedMessage;

+	#ifdef ICSNEO_ENABLE_DEVICE_SHARING
+	auto socket = lockSocket();
+	int64_t ms = timeout.count();
+	if(!(socket.writeTyped(RPC::DEVICE_LOCK) && socket.writeString(driver->device.serial) && socket.writeTyped(ms)))
+		return nullptr;
+	if(bool ret; !(socket.readTyped(ret) && ret))
+		return nullptr;
+	#endif
+	
 	std::unique_lock<std::mutex> fnLk(syncMessageMutex); // Only allow for one sync message at a time
 	std::unique_lock<std::mutex> cvLk(cvMutex); // Don't let the callback fire until we're waiting for it
 	int cb = addMessageCallback(std::make_shared<MessageCallback>([&cvMutex, &returnedMessage, &cv](std::shared_ptr<Message> message) {
@@ -239,6 +257,13 @@ std::shared_ptr<Message> Communication::waitForMessageSync(std::function<bool(vo

 	if(fail) // The caller's function failed, so don't return a message
 		returnedMessage.reset();
+	
+	#ifdef ICSNEO_ENABLE_DEVICE_SHARING
+	if(!(socket.writeTyped(RPC::DEVICE_UNLOCK) && socket.writeString(driver->device.serial)))
+		return nullptr;
+	if(bool ret; !(socket.readTyped(ret) && ret))
+		return nullptr;
+	#endif

 	// Then we either will return the message we got or we will return the empty shared_ptr, caller responsible for checking
 	return returnedMessage;
@@ -274,6 +299,11 @@ void Communication::readTask() {
 }

 void Communication::handleInput(Packetizer& p, std::vector<uint8_t>& readBytes) {
+	{
+		std::lock_guard lk(rawCallbacksMutex);
+		for(auto& cb : rawCallbacks)
+			cb(readBytes);
+	}
 	if(redirectingRead) {
 		// redirectingRead is an atomic so it can be set without acquiring a mutex
 		// However, we do not clear it without the mutex. The idea is that if another
@@ -0,0 +1,60 @@
+#include <cstring>
+#include "icsneo/communication/interprocessmailbox.h"
+
+using namespace icsneo;
+
+bool InterprocessMailbox::open(const std::string& name, bool create)
+{
+	if(!queuedSem.open(name + "-qs", create))
+		return false;
+
+	if(!emptySem.open(name + "-es", create, MESSAGE_COUNT))
+		return false;
+
+	if(!sharedMem.open(name + "-sm", BLOCK_SIZE * MESSAGE_COUNT, create))
+		return false;
+
+	valid = true;
+	return true;
+}
+
+InterprocessMailbox::operator bool() const
+{
+	return valid;
+}
+
+bool InterprocessMailbox::close()
+{
+	valid = false;
+	return queuedSem.close() && emptySem.close() && sharedMem.close();
+}
+
+bool InterprocessMailbox::read(void* data, LengthFieldType& messageLength, const std::chrono::milliseconds& timeout)
+{
+	if(!queuedSem.wait(timeout))
+		return false;
+	auto it = sharedMem.data() + (index * BLOCK_SIZE);
+	messageLength = *(LengthFieldType*)it;
+	it += LENGTH_FIELD_SIZE;
+	std::memcpy(data, it, std::min(messageLength, MAX_DATA_SIZE));
+	if(!emptySem.post())
+		return false;
+	++index;
+	index %= MESSAGE_COUNT;
+	return true;
+}
+
+bool InterprocessMailbox::write(const void* data, LengthFieldType messageLength, const std::chrono::milliseconds& timeout)
+{
+	if(!emptySem.wait(timeout))
+		return false; // the buffer is full and we timed out
+	auto it = sharedMem.data() + (index * BLOCK_SIZE);
+	*(LengthFieldType*)it = messageLength;
+	it += LENGTH_FIELD_SIZE;
+	std::memcpy(it, data, messageLength);
+	if(!queuedSem.post())
+		return false;
+	++index;
+	index %= MESSAGE_COUNT;
+	return true;
+}
@@ -0,0 +1,179 @@
+#include <cstring>
+
+#include "icsneo/communication/sdio.h"
+#include "icsneo/platform/sharedsemaphore.h"
+#include "icsneo/platform/sharedmemory.h"
+#include "icsneo/device/device.h"
+#include "icsneo/communication/socket.h"
+
+using namespace icsneo;
+
+void SDIO::Find(std::vector<FoundDevice>& found) {
+	auto socket = lockSocket();
+	if(!socket.writeTyped(RPC::DEVICE_FINDER_FIND_ALL))
+		return;
+	uint16_t count;
+	if(!socket.readTyped(count))
+		return;
+
+	static constexpr auto serialSize = sizeof(FoundDevice::serial);
+	std::vector<std::array<char, sizeof(FoundDevice::serial)>> serials(count);
+	if(!socket.read(serials.data(), serials.size() * serialSize))
+		return;
+
+	for(const auto& serial : serials) {
+		auto& foundDevice = found.emplace_back();
+		for(std::size_t i = 0; i < serialSize - 1 /* omit '\0' */; i++)
+			foundDevice.serial[i] = static_cast<char>(std::toupper(serial[i]));
+		foundDevice.makeDriver = [](const device_eventhandler_t& r, neodevice_t& d) {
+			return std::unique_ptr<SDIO>(new SDIO(r, d));
+		};
+	}
+}
+
+bool SDIO::open() {
+	{
+		auto socket = lockSocket();
+		if(!(socket.writeTyped(RPC::SDIO_OPEN) && socket.writeString(device.device->getSerial())))
+			return false;
+		if(bool ret; !(socket.readTyped(ret) && ret))
+			return false;
+
+		{
+			std::string mailboxName;
+			if(!socket.readString(mailboxName))
+				return false;
+			if(!inboundIO.open(mailboxName))
+				return false;
+		}
+
+		{
+			std::string mailboxName;
+			if(!socket.readString(mailboxName))
+				return false;
+			if(!outboundIO.open(mailboxName))
+				return false;
+		}
+	}
+
+	readThread = std::thread(&SDIO::readTask, this);
+	writeThread = std::thread(&SDIO::writeTask, this);
+
+	deviceOpen = true;
+
+	return deviceOpen;
+}
+
+bool SDIO::close() {
+	if(!isOpen() && !isDisconnected()) {
+		report(APIEvent::Type::DeviceCurrentlyClosed, APIEvent::Severity::Error);
+		return false;
+	}
+
+	closing = true;
+
+	 // wait for the reader/writer threads to close
+	std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+	// unblocks the reader/writer threads
+	if(!inboundIO.close())
+		return false;
+	if(!outboundIO.close())
+		return false;
+
+	if(readThread.joinable())
+		readThread.join();
+
+	if(writeThread.joinable())
+		writeThread.join();
+
+	{
+		auto socket = lockSocket();
+		if(!socket.writeTyped(RPC::SDIO_CLOSE))
+			return false;
+		if(!socket.writeString(device.device->getSerial()))
+			return false;
+		if(bool ret; !(socket.readTyped(ret) && ret))
+			return false;
+	}
+
+	uint8_t flush;
+	WriteOperation flushop;
+	while(readQueue.try_dequeue(flush)) {}
+	while(writeQueue.try_dequeue(flushop)) {}
+
+	closing = false;
+	disconnected = false;
+	deviceOpen = false;
+	return true;
+}
+
+bool SDIO::isOpen() {
+	return deviceOpen;
+}
+
+void SDIO::readTask() {
+	uint8_t data[MAX_DATA_SIZE];
+	uint16_t messageLength;
+	while(!closing) {
+		if(!inboundIO.read(data, messageLength, std::chrono::milliseconds(100))) {
+			if(!inboundIO)
+				break;
+			continue;
+		}
+
+		if(messageLength > 0) {
+			if(messageLength > MAX_DATA_SIZE) { // split message
+				std::vector<uint8_t> reassembled(messageLength);
+				std::memcpy(reassembled.data(), data, MAX_DATA_SIZE);
+				auto offset = reassembled.data() + MAX_DATA_SIZE;
+				for(auto remaining = messageLength - MAX_DATA_SIZE; remaining > 0; remaining -= messageLength) {
+					if(!inboundIO.read(offset, messageLength, std::chrono::milliseconds(10))) {
+						report(APIEvent::Type::FailedToRead, APIEvent::Severity::Error);
+						break;
+					}
+					offset += messageLength;
+				}
+				readQueue.enqueue_bulk(reassembled.data(), reassembled.size());
+			} else {
+				readQueue.enqueue_bulk(data, messageLength);
+			}
+		}
+	}
+}
+
+void SDIO::writeTask() {
+	WriteOperation writeOp;
+	while(!closing && !isDisconnected()) {
+		if(!writeQueue.wait_dequeue_timed(writeOp, std::chrono::milliseconds(100)))
+			continue;
+
+		const auto dataSize = static_cast<LengthFieldType>(writeOp.bytes.size());
+
+		const auto tryWrite = [&](const void* input, LengthFieldType length) -> bool {
+			for(int i = 0; i < 50; ++i) { // try to write for 5s, making sure we can close if need be
+				if(outboundIO.write(input, length, std::chrono::milliseconds(100)))
+					return true;
+				if(!outboundIO)
+					return false;
+			}
+			disconnected = true;
+			report(APIEvent::Type::DeviceDisconnected, APIEvent::Severity::Error);
+			return false;
+		};
+
+		if(!tryWrite(writeOp.bytes.data(), dataSize))
+			continue;
+
+		if(writeOp.bytes.size() > MAX_DATA_SIZE) {
+			auto offset = writeOp.bytes.data() + MAX_DATA_SIZE;
+			for(LengthFieldType remaining = dataSize - MAX_DATA_SIZE; remaining > 0; ) {
+				const auto toWrite = std::min(MAX_DATA_SIZE, remaining);
+				if(!tryWrite(offset, toWrite))
+					break;
+				remaining -= toWrite;
+				offset += toWrite;
+			}
+		}
+	}
+}
@@ -0,0 +1,257 @@
+#include "icsneo/communication/socket.h"
+#include "icsneo/api/event.h"
+#include "icsneo/api/eventmanager.h"
+
+namespace icsneo {
+
+bool SocketBase::open() {
+	#ifdef _WIN32
+	WSADATA wsaData;
+	if(::WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {
+		EventManager::GetInstance().add(APIEvent::Type::SocketFailedToOpen, APIEvent::Severity::Error);
+		return false;
+	}
+	#endif
+
+	if((sockFileDescriptor = ::socket(AF_INET, SOCK_STREAM, 0)) < 0) {
+		#ifdef _WIN32
+		::WSACleanup();
+		#endif
+		EventManager::GetInstance().add(APIEvent::Type::SocketFailedToOpen, APIEvent::Severity::Error);
+		return false;
+	}
+
+	sockIsOpen = true;
+	return true;
+}
+
+bool SocketBase::close() {
+	#ifdef _WIN32
+	if(::closesocket(sockFileDescriptor) < 0) {
+		// should probably check for WSAEWOULDBLOCK as ::closesocket must be repeated to close in that case
+		#ifdef ICSNEO_ENABLE_DEVICE_SHARING
+		EventManager::GetInstance().add(APIEvent::Type::SocketFailedToClose, APIEvent::Severity::Error);
+		#endif
+		return false;
+	}
+	::WSACleanup();
+	#else
+	// ignore ENOTCONN from ::shutdown as the peer may have already forcibly closed its socket (e.g. a crash)
+	if( ((::shutdown(sockFileDescriptor, SHUT_RDWR) < 0) && (ENOTCONN != errno)) ||
+		((::close(sockFileDescriptor) < 0) && (EBADF == errno)) )
+	{
+		#ifdef ICSNEO_ENABLE_DEVICE_SHARING
+		EventManager::GetInstance().add(APIEvent::Type::SocketFailedToClose, APIEvent::Severity::Error);
+		#endif
+		return false;
+	}
+	#endif
+
+	sockIsOpen = false;
+	sockIsConnected = false;
+	return true;
+}
+
+bool SocketBase::connect() {
+	sockaddr_in addr = {0};
+	addr.sin_family = AF_INET;
+	addr.sin_port = htons(port);
+
+	if( (!isOpen() && !open()) ||
+	    (::inet_pton(addr.sin_family, "127.0.0.1", &addr.sin_addr) <= 0) ||
+	    (::connect(sockFileDescriptor, (sockaddr*)&addr, sizeof(addr)) < 0) )
+	{
+		EventManager::GetInstance().add(APIEvent::Type::SocketFailedToConnect, APIEvent::Severity::Error);
+		return false;
+	}
+
+	#ifdef _WIN32
+	DWORD tv = 5000u; // 5 second receive timeout but in windows
+	#else
+	struct timeval tv;
+	tv.tv_sec = 5u; // 5 second receive timeout
+	tv.tv_usec = 0;
+	setIgnoreSIGPIPE();
+	#endif
+
+	// Set the 5 second timeout from above in the socket options
+	::setsockopt(sockFileDescriptor, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof(tv));
+	::setsockopt(sockFileDescriptor, SOL_SOCKET, SO_SNDTIMEO, (const char*)&tv, sizeof(tv));
+
+	sockIsConnected = true;
+	return true;
+}
+
+bool SocketBase::isOpen() {
+	return sockIsOpen;
+}
+
+bool SocketBase::isConnected() {
+	return sockIsConnected;
+}
+
+bool SocketBase::read(void* output, std::size_t length) {
+	if(!(isOpen() && isConnected()))
+		return false;
+
+	#ifdef _WIN32
+	return ::recv(sockFileDescriptor, (char*)output, (int)length, 0) > 0;
+	#else
+	if(::read(sockFileDescriptor, output, length) <= 0) {
+		EventManager::GetInstance().add(APIEvent::Type::SocketFailedToRead, APIEvent::Severity::Error);
+		return false;
+	}
+	return true;
+	#endif
+}
+
+bool SocketBase::write(const void* input, std::size_t length) {
+	if(!(isOpen() && isConnected()))
+		return false;
+
+	#ifdef _WIN32
+	if(::send(sockFileDescriptor, (char*)input, (int)length, 0) < 0) {
+		switch(WSAGetLastError()) {
+			case WSAETIMEDOUT:
+			case WSAENOTCONN:
+			case WSAESHUTDOWN:
+			case WSAECONNRESET:
+			case WSAECONNABORTED:
+			{
+				sockIsOpen = false;
+				break;
+			}
+			default:
+			{
+				EventManager::GetInstance().add(APIEvent::Type::SocketFailedToWrite, APIEvent::Severity::Error);
+				break;
+			}
+		}
+		return false;
+	}
+	#else
+	if(::write(sockFileDescriptor, input, length) < 0) {
+		switch(errno) {
+			case EPIPE:
+			case ETIMEDOUT:
+			{
+				sockIsOpen = false;
+				break;
+			}
+			default:
+			{
+				EventManager::GetInstance().add(APIEvent::Type::SocketFailedToWrite, APIEvent::Severity::Error);
+				break;
+			}
+		}
+		return false;
+	}
+	#endif
+	return true;
+}
+
+bool SocketBase::readString(std::string& str) {
+	size_t length;
+	if(!read(&length, sizeof(length)))
+		return false;
+	str.resize(length);
+	if(!read(str.data(), length))
+		return false;
+	return true;
+}
+
+bool SocketBase::writeString(const std::string& str) {
+	size_t length = str.size();
+	if(!write(&length, sizeof(length)))
+		return false;
+	if(!write(str.data(), length))
+		return false;
+	return true;
+}
+
+ActiveSocket::ActiveSocket(SocketFileDescriptor sockFD) {
+	sockFileDescriptor = sockFD;
+	sockIsOpen = true;
+	sockIsConnected = true;
+}
+
+ActiveSocket::ActiveSocket(Protocol protocol, uint16_t port) {
+	this->protocol = protocol;
+	this->port = port;
+}
+
+ActiveSocket::~ActiveSocket() {
+	if(isOpen())
+		close();
+}
+
+LockedSocket::LockedSocket(SocketBase& base, std::unique_lock<std::mutex>&& l) :
+	SocketBase(base), lock(std::move(l)) {
+}
+
+LockedSocket lockSocket() {
+	static ActiveSocket socket(SocketBase::Protocol::TCP, RPC_PORT);
+	if(!socket.isOpen())
+		socket.open();
+	if(!socket.isConnected())
+		socket.connect();
+	static std::mutex lock;
+	return LockedSocket(socket, std::unique_lock<std::mutex>(lock));
+}
+
+void SocketBase::setIgnoreSIGPIPE() {
+	#ifndef _WIN32
+	struct sigaction sa{};
+	sa.sa_handler = SIG_IGN;
+	sigemptyset(&sa.sa_mask);
+	::sigaction(SIGPIPE, &sa, NULL);
+	#endif
+}
+
+Acceptor::Acceptor(Protocol protocol, uint16_t port)
+	: ActiveSocket(protocol, port) {
+}
+
+bool Acceptor::initialize() {
+	if(open() && bind() && listen()) {
+		isValid = true;
+		return true;
+	}
+	return false;
+}
+
+std::shared_ptr<ActiveSocket> Acceptor::accept()
+{
+	if(!isValid)
+		return nullptr;
+	const SocketFileDescriptor acceptFd = ::accept(sockFileDescriptor, (sockaddr*)NULL, NULL);
+	if(acceptFd < 0)
+		return nullptr;
+	return std::make_shared<ActiveSocket>(acceptFd);
+}
+
+bool Acceptor::bind()
+{
+	sockaddr_in addr = {0};
+	addr.sin_family = AF_INET;
+	addr.sin_addr.s_addr = htonl(INADDR_ANY);
+	addr.sin_port = htons(port);
+	if( (::inet_pton(addr.sin_family, "127.0.0.1", &addr.sin_addr) <= 0) ||
+	    (::bind(sockFileDescriptor, (sockaddr*)&addr, sizeof(addr)) < 0) )
+	{
+		EventManager::GetInstance().add(APIEvent::Type::SocketAcceptorFailedToBind, APIEvent::Severity::Error);
+		return false;
+	}
+	return true;
+}
+
+bool Acceptor::listen()
+{
+	if(::listen(sockFileDescriptor, UINT8_MAX) < 0) {
+		EventManager::GetInstance().add(APIEvent::Type::SocketAcceptorFailedToListen, APIEvent::Severity::Error);
+		return false;
+	}
+	return true;
+}
+
+} //namespace icsneo