diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 9e9adccf3..8399a1049 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -13,9 +13,17 @@ jobs: strategy: fail-fast: false matrix: - os: ['macos-latest', 'ubuntu-latest', 'ubuntu-24.04-arm'] + os: ['macos-latest', 'ubuntu-latest', 'ubuntu-24.04-arm', 'windows-latest'] + defaults: + run: + shell: ${{ contains(matrix.os, 'windows') && 'msys2 {0}' || 'bash -e {0}' }} timeout-minutes: 10 steps: - uses: actions/checkout@v7 + - uses: msys2/setup-msys2@v2 + if: contains(matrix.os, 'windows') + with: + msystem: CLANG64 + install: mingw-w64-clang-x86_64-clang mingw-w64-clang-x86_64-git mingw-w64-clang-x86_64-uv # clang for scons, native git for lefthook - run: ./test.sh - timeout-minutes: 1 + timeout-minutes: ${{ contains(matrix.os, 'windows') && 5 || 1 }} # the 4-core runner builds and installs slower diff --git a/.gitignore b/.gitignore index 9294b1bb5..6c486631e 100644 --- a/.gitignore +++ b/.gitignore @@ -14,9 +14,10 @@ catch2/ /build/ /dist/ -test_runner +test_runner* libmessaging.* libmessaging_shared.* .sconsign.dblite .mypy_cache/ +*.pyd diff --git a/SConstruct b/SConstruct index b7e1a4c76..12036b7aa 100644 --- a/SConstruct +++ b/SConstruct @@ -1,8 +1,10 @@ import os import platform import subprocess +import sys import sysconfig +WINDOWS = platform.system() == "Windows" arch = subprocess.check_output(["uname", "-m"], encoding='utf8').rstrip() if platform.system() == "Darwin": arch = "Darwin" @@ -63,8 +65,12 @@ env = Environment( CXXFLAGS="-std=c++1z", CPPPATH=cpppath, CYTHONCFILESUFFIX=".cpp", - tools=["default", "cython"] + tools=["mingw" if WINDOWS else "default", "cython"], # the default tool picks MSVC on Windows ) +if WINDOWS: + env["CC"], env["CXX"] = "clang", "clang++" # the mingw tool assumes gcc + env.Append(LINKFLAGS=["-static"]) # libc++ into the binaries so they run outside the MSYS2 shell + common = ["ws2_32"] # visionipc's sockets Export('env', 'arch', 'common') @@ -73,6 +79,10 @@ envCython["CCFLAGS"] += ["-Wno-#warnings", "-Wno-cpp", "-Wno-shadow", "-Wno-depr envCython["CCFLAGS"].remove('-Werror') if arch == "Darwin": envCython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"] +elif WINDOWS: + envCython["LINKFLAGS"] = ["-shared", "-static"] + envCython.Append(LIBPATH=[os.path.join(sys.base_prefix, "libs")]) + envCython["LIBS"] = [f"python{sys.version_info.major}{sys.version_info.minor}"] else: envCython["LINKFLAGS"] = ["-pthread", "-shared"] diff --git a/msgq/event.cc b/msgq/event.cc index ac3153a32..4e8ebc3b9 100644 --- a/msgq/event.cc +++ b/msgq/event.cc @@ -4,14 +4,25 @@ #include #include #include +#include #include +#include #include +#include +#ifdef _WIN32 +#define NOMINMAX +#include +#define EVENT_SHM_PREFIX "Local\\msgq_" +#define EVENT_PATH_PREFIX "Local\\msgq_event_" +#else +#define EVENT_SHM_PREFIX "/msgq_" +#define EVENT_PATH_PREFIX "/tmp/msgq_event_" #include #include #include #include -#include +#endif #include "msgq/event.h" @@ -23,13 +34,19 @@ size_t event_fifo_counter = 0; throw std::runtime_error(msg + ", errno: " + std::to_string(errno) + " pid: " + std::to_string(getpid())); } -int open_event_fifo(const char* path) { - if (path[0] == '\0') return -1; - int fd = open(path, O_RDWR | O_NONBLOCK); - if (fd < 0 && errno != ENOENT) throw_errno("Could not open event fifo"); - return fd; -} +#ifdef _WIN32 +// Kernel handles fit in 32 bits, so an event "fd" is just the HANDLE value +HANDLE fd_to_handle(int fd) { return reinterpret_cast(static_cast(fd)); } +int handle_to_fd(HANDLE h) { return static_cast(reinterpret_cast(h)); } +DWORD to_wait_ms(int timeout_sec) { return timeout_sec < 0 ? INFINITE : static_cast(timeout_sec) * 1000; } + +int setenv(const char *name, const char *value, int) { return _putenv_s(name, value); } +int unsetenv(const char *name) { return _putenv_s(name, ""); } +// A section keeps its name only while a handle to it is open, so hold one per view +std::mutex sections_mutex; +std::unordered_map sections; +#else // poll() that retries on EINTR with a monotonic deadline so signal storms // don't extend the effective timeout. int poll_events(pollfd *fds, nfds_t nfds, int timeout_sec) { @@ -46,6 +63,7 @@ int poll_events(pollfd *fds, nfds_t nfds, int timeout_sec) { if (errno != EINTR) throw_errno("Event poll failed"); } } +#endif // macOS limits shm_open names to ~31 chars, so hash the (prefix, identifier, endpoint) // tuple into a fixed-length name. @@ -60,15 +78,50 @@ std::string event_shm_name(const std::string& endpoint, const std::string& ident } char buf[32]; - std::snprintf(buf, sizeof(buf), "/msgq_%016llx", static_cast(h)); + std::snprintf(buf, sizeof(buf), EVENT_SHM_PREFIX "%016llx", static_cast(h)); return buf; } } // namespace +int event_open(const char *path) { + if (path[0] == '\0') return -1; +#ifdef _WIN32 + // manual reset: stays signaled until clear(), like unread bytes in a FIFO + HANDLE h = CreateEventA(NULL, TRUE, FALSE, path); + if (h == NULL) throw_errno("Could not open event"); + return handle_to_fd(h); +#else + int fd = open(path, O_RDWR | O_NONBLOCK); + if (fd < 0 && errno != ENOENT) throw_errno("Could not open event fifo"); + return fd; +#endif +} + +void event_close(int fd) { + if (fd < 0) return; +#ifdef _WIN32 + CloseHandle(fd_to_handle(fd)); +#else + close(fd); +#endif +} + void event_state_shm_mmap(std::string endpoint, std::string identifier, char **shm_mem, std::string *shm_name_out) { std::string name = event_shm_name(endpoint, identifier); +#ifdef _WIN32 + HANDLE section = CreateFileMappingA(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE, 0, sizeof(EventState), name.c_str()); + if (section == NULL) throw_errno("Could not open shared memory"); + + char *mem = reinterpret_cast(MapViewOfFile(section, FILE_MAP_ALL_ACCESS, 0, 0, 0)); + if (mem == nullptr) { + CloseHandle(section); + throw_errno("Could not map shared memory"); + } + std::lock_guard lock(sections_mutex); + sections[mem] = section; +#else int shm_fd = shm_open(name.c_str(), O_RDWR | O_CREAT, 0664); if (shm_fd < 0) throw_errno("Could not open shared memory"); @@ -88,11 +141,23 @@ void event_state_shm_mmap(std::string endpoint, std::string identifier, char **s char *mem = reinterpret_cast(mmap(NULL, sizeof(EventState), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0)); close(shm_fd); if (mem == MAP_FAILED) throw_errno("Could not map shared memory"); +#endif if (shm_mem != nullptr) *shm_mem = mem; if (shm_name_out != nullptr) *shm_name_out = name; } +void event_state_shm_munmap(void *mem) { +#ifdef _WIN32 + std::lock_guard lock(sections_mutex); // held across the unmap so a remap of the address cannot slip in + UnmapViewOfFile(mem); + CloseHandle(sections.at(mem)); + sections.erase(mem); +#else + munmap(mem, sizeof(EventState)); +#endif +} + SocketEventHandle::SocketEventHandle(std::string endpoint, std::string identifier, bool override) { char *mem; event_state_shm_mmap(endpoint, identifier, &mem, &this->shm_name); @@ -101,35 +166,39 @@ SocketEventHandle::SocketEventHandle(std::string endpoint, std::string identifie this->owns_fifos = override; if (override) { - std::string base = "/tmp/msgq_event_" + std::to_string(getpid()) + "_" + std::to_string(event_fifo_counter++); + std::string base = EVENT_PATH_PREFIX + std::to_string(getpid()) + "_" + std::to_string(event_fifo_counter++); for (size_t i = 0; i < 2; i++) { std::string p = base + "." + std::to_string(i); if (p.size() >= EVENT_PATH_MAX) { throw std::runtime_error("Event path too long: " + p); } +#ifndef _WIN32 unlink(p.c_str()); if (mkfifo(p.c_str(), 0664) < 0) throw_errno("Could not create event fifo"); +#endif std::memcpy(this->state->paths[i], p.c_str(), p.size() + 1); } this->state->enabled = false; } for (size_t i = 0; i < 2; i++) { - this->fds[i] = open_event_fifo(this->state->paths[i]); + this->fds[i] = event_open(this->state->paths[i]); } } SocketEventHandle::~SocketEventHandle() { if (this->state == nullptr) return; for (int fd : this->fds) { - if (fd >= 0) close(fd); + event_close(fd); } +#ifndef _WIN32 // named events and sections go away with their last user if (this->owns_fifos) { unlink(this->state->paths[RECV_CALLED]); unlink(this->state->paths[RECV_READY]); shm_unlink(this->shm_name.c_str()); } - munmap(this->state, sizeof(EventState)); +#endif + event_state_shm_munmap(this->state); } bool SocketEventHandle::is_enabled() { @@ -172,7 +241,9 @@ Event::Event(int fd): event_fd(fd) {} void Event::set() const { throw_if_invalid(); - +#ifdef _WIN32 + if (!SetEvent(fd_to_handle(this->event_fd))) throw_errno("Event set failed"); +#else char val = 1; while (true) { ssize_t count = write(this->event_fd, &val, sizeof(val)); @@ -181,11 +252,17 @@ void Event::set() const { if (errno == EAGAIN || errno == EWOULDBLOCK) return; throw_errno("Event write failed"); } +#endif } int Event::clear() const { throw_if_invalid(); - +#ifdef _WIN32 + HANDLE h = fd_to_handle(this->event_fd); + int was_set = WaitForSingleObject(h, 0) == WAIT_OBJECT_0; + ResetEvent(h); + return was_set; +#else int total = 0; char buf[64]; while (true) { @@ -198,22 +275,31 @@ int Event::clear() const { if (errno == EINTR) continue; throw_errno("Event read failed"); } +#endif } void Event::wait(int timeout_sec) const { throw_if_invalid(); - +#ifdef _WIN32 + if (WaitForSingleObject(fd_to_handle(this->event_fd), to_wait_ms(timeout_sec)) != WAIT_OBJECT_0) { + throw std::runtime_error("Event timed out pid: " + std::to_string(getpid())); + } +#else pollfd fds = {this->event_fd, POLLIN, 0}; if (poll_events(&fds, 1, timeout_sec) == 0) { throw std::runtime_error("Event timed out pid: " + std::to_string(getpid())); } +#endif } bool Event::peek() const { throw_if_invalid(); - +#ifdef _WIN32 + return WaitForSingleObject(fd_to_handle(this->event_fd), 0) == WAIT_OBJECT_0; +#else pollfd fds = {this->event_fd, POLLIN, 0}; return poll_events(&fds, 1, 0) > 0; +#endif } bool Event::is_valid() const { @@ -225,6 +311,16 @@ int Event::fd() const { } int Event::wait_for_one(const std::vector& events, int timeout_sec) { +#ifdef _WIN32 + std::vector handles; + for (const Event &e : events) handles.push_back(fd_to_handle(e.fd())); + DWORD ret = WaitForMultipleObjects(static_cast(handles.size()), handles.data(), FALSE, to_wait_ms(timeout_sec)); + if (ret == WAIT_TIMEOUT) { + throw std::runtime_error("Event timed out pid: " + std::to_string(getpid())); + } + if (ret < handles.size()) return static_cast(ret); + throw std::runtime_error("Event poll failed, no events ready"); +#else pollfd fds[events.size()]; for (size_t i = 0; i < events.size(); i++) { fds[i] = {events[i].fd(), POLLIN, 0}; @@ -241,4 +337,5 @@ int Event::wait_for_one(const std::vector& events, int timeout_sec) { } throw std::runtime_error("Event poll failed, no events ready"); +#endif } diff --git a/msgq/event.h b/msgq/event.h index 6989b053f..cbea1deea 100644 --- a/msgq/event.h +++ b/msgq/event.h @@ -10,6 +10,11 @@ constexpr size_t EVENT_PATH_MAX = 128; void event_state_shm_mmap(std::string endpoint, std::string identifier, char **shm_mem, std::string *shm_name); +void event_state_shm_munmap(void *mem); + +// Open/close the OS object behind an event path (a FIFO on POSIX, a named event on Windows) +int event_open(const char *path); +void event_close(int fd); enum EventPurpose { RECV_CALLED, diff --git a/msgq/impl_fake.h b/msgq/impl_fake.h index 3884bf6ea..6d8820b06 100644 --- a/msgq/impl_fake.h +++ b/msgq/impl_fake.h @@ -6,7 +6,6 @@ #include #include -#include #include #include #include @@ -23,7 +22,7 @@ class FakeSubSocket: public TSubSocket { void ensure_fifos_open() { for (size_t i = 0; i < 2; i++) { if (fds[i] < 0 && state->paths[i][0] != '\0') { - fds[i] = open(state->paths[i], O_RDWR | O_NONBLOCK); + fds[i] = event_open(state->paths[i]); } } } @@ -32,10 +31,10 @@ class FakeSubSocket: public TSubSocket { FakeSubSocket(): TSubSocket() {} ~FakeSubSocket() { for (int fd : fds) { - if (fd >= 0) close(fd); + event_close(fd); } if (state != nullptr) { - munmap(state, sizeof(EventState)); + event_state_shm_munmap(state); } } diff --git a/msgq/ipc.h b/msgq/ipc.h index 62a13e442..23a6b169f 100644 --- a/msgq/ipc.h +++ b/msgq/ipc.h @@ -9,7 +9,7 @@ -#ifdef __APPLE__ +#if defined(__APPLE__) || defined(_WIN32) #define CLOCK_BOOTTIME CLOCK_MONOTONIC #endif diff --git a/msgq/msgq.cc b/msgq/msgq.cc index d65fc9ebc..5063ad4c1 100644 --- a/msgq/msgq.cc +++ b/msgq/msgq.cc @@ -12,6 +12,10 @@ #include #include +#ifdef _WIN32 +#define NOMINMAX +#include +#else #include #include #include @@ -20,14 +24,39 @@ #include #include #include +#endif #include #include "msgq/msgq.h" +#ifdef _WIN32 +// Readers wait on a per-thread named event; publishers signal it by thread id, the low 32 bits of a reader uid +static std::string msgq_event_name(uint32_t tid) { + return "Local\\msgq_tid_" + std::to_string(tid); +} + +static HANDLE msgq_thread_event() { + struct ThreadEvent { + HANDLE h = CreateEventA(NULL, FALSE, FALSE, msgq_event_name(GetCurrentThreadId()).c_str()); + ~ThreadEvent() { if (h != NULL) CloseHandle(h); } + }; + thread_local ThreadEvent ev; + return ev.h; +} +#else void sigusr2_handler(int signal) { assert(signal == SIGUSR2); } +#endif + +std::string msgq_shm_dir() { +#ifdef __APPLE__ + return "/tmp"; +#else + return "/dev/shm"; +#endif +} uint64_t msgq_get_uid(void){ std::random_device rd("/dev/urandom"); @@ -36,6 +65,8 @@ uint64_t msgq_get_uid(void){ #ifdef __APPLE__ // TODO: this doesn't work uint64_t uid = distribution(rd) << 32 | getpid(); + #elif defined(_WIN32) + uint64_t uid = distribution(rd) << 32 | GetCurrentThreadId(); #else uint64_t uid = distribution(rd) << 32 | syscall(SYS_gettid); #endif @@ -83,12 +114,14 @@ void msgq_wait_for_subscriber(msgq_queue_t *q){ int msgq_new_queue(msgq_queue_t * q, const char * path, size_t size){ assert(size < 0xFFFFFFFF); // Buffer must be smaller than 2^32 bytes +#ifndef _WIN32 std::signal(SIGUSR2, sigusr2_handler); +#endif -#ifdef __APPLE__ - std::string base_path = "/tmp/msgq_"; +#ifdef _WIN32 + std::string base_path = "Local\\msgq_"; // a named section, kept by the kernel while anyone holds it #else - std::string base_path = "/dev/shm/msgq_"; + std::string base_path = msgq_shm_dir() + "/msgq_"; #endif const char* prefix = std::getenv("OPENPILOT_PREFIX"); if (prefix) { @@ -96,6 +129,20 @@ int msgq_new_queue(msgq_queue_t * q, const char * path, size_t size){ } std::string full_path = base_path + path; +#ifdef _WIN32 + HANDLE section = CreateFileMappingA(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE, 0, (DWORD)(size + sizeof(msgq_header_t)), full_path.c_str()); + if (section == NULL) { + std::cout << "Warning, could not open: " << full_path << std::endl; + return -1; + } + + char * mem = (char*)MapViewOfFile(section, FILE_MAP_ALL_ACCESS, 0, 0, 0); + if (mem == NULL){ + CloseHandle(section); + return -1; + } + q->section = section; // the name only lives while a handle does, so keep it until the queue closes +#else auto fd = open(full_path.c_str(), O_RDWR | O_CREAT, 0664); if (fd < 0) { std::cout << "Warning, could not open: " << full_path << std::endl; @@ -121,6 +168,7 @@ int msgq_new_queue(msgq_queue_t * q, const char * path, size_t size){ if (mem == MAP_FAILED){ return -1; } +#endif q->mmap_p = mem; @@ -149,7 +197,12 @@ int msgq_new_queue(msgq_queue_t * q, const char * path, size_t size){ void msgq_close_queue(msgq_queue_t *q){ if (q->mmap_p != NULL){ +#ifdef _WIN32 + UnmapViewOfFile(q->mmap_p); + CloseHandle(q->section); +#else munmap(q->mmap_p, q->size + sizeof(msgq_header_t)); +#endif } } @@ -173,6 +226,12 @@ static void thread_signal(uint32_t tid) { #ifdef __APPLE__ // macOS doesn't have tkill, rely on polling instead (void)tid; + #elif defined(_WIN32) + HANDLE ev = OpenEventA(EVENT_MODIFY_STATE, FALSE, msgq_event_name(tid).c_str()); + if (ev != NULL) { + SetEvent(ev); + CloseHandle(ev); + } #elif !defined(SYS_tkill) // fallback for systems without tkill kill(tid, SIGUSR2); @@ -444,24 +503,18 @@ int msgq_poll(msgq_pollitem_t * items, size_t nitems, int timeout){ } int ms = (timeout == -1) ? 100 : timeout; - + auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); #ifdef __APPLE__ - // On macOS, signals can't interrupt nanosleep, so poll more frequently - int poll_ms = std::min(ms, 10); - int remaining_ms = ms; -#else - int poll_ms = ms; + ms = std::min(ms, 10); // signals can't interrupt nanosleep on macOS, so poll more frequently #endif - struct timespec ts; - ts.tv_sec = poll_ms / 1000; - ts.tv_nsec = (poll_ms % 1000) * 1000 * 1000; - - while (num == 0) { - int ret; - - ret = nanosleep(&ts, &ts); +#ifdef _WIN32 + WaitForSingleObject(msgq_thread_event(), ms); +#else + struct timespec ts = {ms / 1000, (ms % 1000) * 1000 * 1000}; + nanosleep(&ts, NULL); +#endif // Check if messages ready for (size_t i = 0; i < nitems; i++) { @@ -471,23 +524,11 @@ int msgq_poll(msgq_pollitem_t * items, size_t nitems, int timeout){ } } -#ifdef __APPLE__ - // exit if we had a timeout and we've exhausted it - if (timeout != -1 && ret == 0){ - remaining_ms -= poll_ms; - if (remaining_ms <= 0){ - break; - } - poll_ms = std::min(remaining_ms, 10); - ts.tv_sec = poll_ms / 1000; - ts.tv_nsec = (poll_ms % 1000) * 1000 * 1000; + if (timeout != -1) { + auto left = std::chrono::ceil(deadline - std::chrono::steady_clock::now()).count(); + if (left <= 0) break; + ms = std::min(ms, left); } -#else - // exit if we had a timeout and the sleep finished - if (timeout != -1 && ret == 0){ - break; - } -#endif } return num; diff --git a/msgq/msgq.h b/msgq/msgq.h index 9d4ae53f9..a3c287eba 100644 --- a/msgq/msgq.h +++ b/msgq/msgq.h @@ -5,6 +5,8 @@ #include #include +std::string msgq_shm_dir(); // directory holding the queue and buffer files + #define DEFAULT_SEGMENT_SIZE (1 * 1024 * 1024) #define NUM_READERS 25 #define ALIGN(n) ((n + (8 - 1)) & -8) @@ -30,6 +32,7 @@ struct msgq_queue_t { std::atomic *read_valids[NUM_READERS]; std::atomic *read_uids[NUM_READERS]; char * mmap_p; + void * section; // Windows: the handle that keeps the section's name alive char * data; size_t size; int reader_id; diff --git a/msgq/msgq_tests.cc b/msgq/msgq_tests.cc index b8d6824da..f64155284 100644 --- a/msgq/msgq_tests.cc +++ b/msgq/msgq_tests.cc @@ -2,11 +2,7 @@ #include "msgq/msgq.h" static void cleanup_test_queue() { -#ifdef __APPLE__ - remove("/tmp/msgq_test_queue"); -#else - remove("/dev/shm/msgq_test_queue"); -#endif + remove((msgq_shm_dir() + "/msgq_test_queue").c_str()); } TEST_CASE("ALIGN") @@ -67,6 +63,8 @@ TEST_CASE("msgq_init_subscriber") REQUIRE(*q.read_valids[0] == true); REQUIRE((*q.read_pointers[0] >> 32) == 0); REQUIRE((*q.read_pointers[0] & 0xFFFFFFFF) == 255); + + msgq_close_queue(&q); } TEST_CASE("msgq_msg_send first message") @@ -96,6 +94,7 @@ TEST_CASE("msgq_msg_send first message") delete[] data; msgq_msg_close(&msg); + msgq_close_queue(&q); } } @@ -129,6 +128,7 @@ TEST_CASE("msgq_msg_send test wraparound") REQUIRE(*(int64_t *)tag_location == -1); msgq_msg_close(&msg); + msgq_close_queue(&q); } TEST_CASE("msgq_msg_recv test wraparound") @@ -173,6 +173,8 @@ TEST_CASE("msgq_msg_recv test wraparound") REQUIRE((*q_sub.read_pointers[0] >> 32) == 1); msgq_msg_close(&msg1); + msgq_close_queue(&q_pub); + msgq_close_queue(&q_sub); } } @@ -211,6 +213,8 @@ TEST_CASE("msgq_msg_send test invalidation") REQUIRE(*q_sub.read_valids[0] == false); msgq_msg_close(&msg); + msgq_close_queue(&q_pub); + msgq_close_queue(&q_sub); } } @@ -235,6 +239,9 @@ TEST_CASE("msgq_init_subscriber init 2 subscribers") REQUIRE(*q1.num_readers == 2); REQUIRE(*q2.num_readers == 2); REQUIRE(q2.reader_id == 1); + + msgq_close_queue(&q1); + msgq_close_queue(&q2); } TEST_CASE("Write 1 msg, read 1 msg") @@ -271,6 +278,8 @@ TEST_CASE("Write 1 msg, read 1 msg") msgq_msg_close(&outgoing_msg); msgq_msg_close(&incoming_msg1); msgq_msg_close(&incoming_msg2); + msgq_close_queue(&writer); + msgq_close_queue(&reader); } TEST_CASE("Write 2 msg, read 2 msg - conflate = false") @@ -308,6 +317,8 @@ TEST_CASE("Write 2 msg, read 2 msg - conflate = false") msgq_msg_close(&outgoing_msg); msgq_msg_close(&incoming_msg1); msgq_msg_close(&incoming_msg2); + msgq_close_queue(&writer); + msgq_close_queue(&reader); } TEST_CASE("Write 2 msg, read 2 msg - conflate = true") @@ -346,6 +357,8 @@ TEST_CASE("Write 2 msg, read 2 msg - conflate = true") msgq_msg_close(&outgoing_msg); msgq_msg_close(&incoming_msg1); msgq_msg_close(&incoming_msg2); + msgq_close_queue(&writer); + msgq_close_queue(&reader); } TEST_CASE("1 publisher, 1 slow subscriber") @@ -389,6 +402,9 @@ TEST_CASE("1 publisher, 1 slow subscriber") // TODO: verify these numbers by hand REQUIRE(n_received == 8572); REQUIRE(n_skipped == 1428); + + msgq_close_queue(&writer); + msgq_close_queue(&reader); } TEST_CASE("1 publisher, 2 subscribers") @@ -423,4 +439,8 @@ TEST_CASE("1 publisher, 2 subscribers") msgq_msg_close(&msg1); msgq_msg_close(&msg2); } + + msgq_close_queue(&writer); + msgq_close_queue(&reader1); + msgq_close_queue(&reader2); } diff --git a/msgq/tests/test_messaging.py b/msgq/tests/test_messaging.py index 470159cf4..886f55a68 100644 --- a/msgq/tests/test_messaging.py +++ b/msgq/tests/test_messaging.py @@ -49,9 +49,9 @@ def test_receive_timeout(self): timeout_ms = 5 sub_sock = msgq.sub_sock(sock, timeout=timeout_ms) - start_time = time.monotonic() + start_time = time.perf_counter() recvd = sub_sock.receive() - elapsed = time.monotonic() - start_time + elapsed = time.perf_counter() - start_time assert recvd is None assert elapsed >= timeout_ms / 1000 assert elapsed < 5 # this can be noisy due to other load on the system diff --git a/msgq/visionipc/visionbuf.cc b/msgq/visionipc/visionbuf.cc index ad2399754..e92478dc7 100644 --- a/msgq/visionipc/visionbuf.cc +++ b/msgq/visionipc/visionbuf.cc @@ -6,19 +6,32 @@ #include #include #include -#include #include +#include "msgq/msgq.h" + +#ifdef _WIN32 +#include + +// An anonymous section; the fd field carries its handle, which fits in 32 bits +static void *malloc_with_fd(size_t len, int *fd) { + HANDLE section = CreateFileMappingA(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE, (DWORD)((uint64_t)len >> 32), (DWORD)len, NULL); + assert(section != NULL); + void *addr = MapViewOfFile(section, FILE_MAP_ALL_ACCESS, 0, 0, len); + assert(addr != NULL); + + *fd = (int)(intptr_t)section; + return addr; +} +#else +#include + std::atomic offset = 0; static void *malloc_with_fd(size_t len, int *fd) { char full_path[0x100]; -#ifdef __APPLE__ - snprintf(full_path, sizeof(full_path)-1, "/tmp/visionbuf_%d_%d", getpid(), offset++); -#else - snprintf(full_path, sizeof(full_path)-1, "/dev/shm/msgq_visionbuf_%d_%d", getpid(), offset++); -#endif + snprintf(full_path, sizeof(full_path)-1, "%s/msgq_visionbuf_%d_%d", msgq_shm_dir().c_str(), getpid(), offset++); *fd = open(full_path, O_RDWR | O_CREAT, 0664); assert(*fd >= 0); @@ -32,6 +45,7 @@ static void *malloc_with_fd(size_t len, int *fd) { return addr; } +#endif void VisionBuf::allocate(size_t length) { this->len = length; @@ -42,8 +56,13 @@ void VisionBuf::allocate(size_t length) { void VisionBuf::import(){ assert(this->fd >= 0); +#ifdef _WIN32 + this->addr = MapViewOfFile((HANDLE)(intptr_t)this->fd, FILE_MAP_ALL_ACCESS, 0, 0, this->mmap_len); + assert(this->addr != NULL); +#else this->addr = mmap(NULL, this->mmap_len, PROT_READ | PROT_WRITE, MAP_SHARED, this->fd, 0); assert(this->addr != MAP_FAILED); +#endif this->frame_id = (uint64_t*)((uint8_t*)this->addr + this->len); } @@ -63,11 +82,16 @@ int VisionBuf::sync(int dir) { } int VisionBuf::free() { +#ifdef _WIN32 + if (!UnmapViewOfFile(this->addr)) return -1; + return CloseHandle((HANDLE)(intptr_t)this->fd) ? 0 : -1; +#else int err = munmap(this->addr, this->mmap_len); if (err != 0) return err; err = close(this->fd); return err; +#endif } uint64_t VisionBuf::get_frame_id() { diff --git a/msgq/visionipc/visionipc.cc b/msgq/visionipc/visionipc.cc index 48e13c27d..7090728b0 100644 --- a/msgq/visionipc/visionipc.cc +++ b/msgq/visionipc/visionipc.cc @@ -5,11 +5,18 @@ #include #include +#ifdef _WIN32 +#include +#include +#include +static struct WsaInit { WsaInit() { WSADATA data; WSAStartup(MAKEWORD(2, 2), &data); } } wsa_init; +#else #include #include #include +#endif -#ifdef __APPLE__ +#if defined(__APPLE__) || defined(_WIN32) #define getsocket() socket(AF_UNIX, SOCK_STREAM, 0) #else #define getsocket() socket(AF_UNIX, SOCK_SEQPACKET, 0) @@ -29,7 +36,7 @@ int ipc_connect(const char* socket_path) { snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", socket_path); err = connect(sock, (struct sockaddr*)&addr, sizeof(addr)); if (err != 0) { - close(sock); + ipc_close(sock); return -1; } @@ -56,7 +63,63 @@ int ipc_bind(const char* socket_path) { return sock; } +void ipc_close(int fd) { +#ifdef _WIN32 + closesocket(fd); +#else + close(fd); +#endif +} + + +#ifdef _WIN32 +// No SCM_RIGHTS on Windows: the sender duplicates the buffers' section handles into the peer process, whose pid +// the socket reports, and sends the handle values after the length-prefixed message (a stream has no record boundaries) +int ipc_sendrecv_with_fds(bool send, int fd, void *buf, size_t buf_size, int* fds, int num_fds, + int *out_num_fds) { + uint32_t len = buf_size, n = num_fds; + if (send) { + if (::send(fd, (char*)&len, sizeof(len), 0) < 0 || ::send(fd, (char*)buf, len, 0) < 0 || ::send(fd, (char*)&n, sizeof(n), 0) < 0) return -1; + if (n > 0) { + DWORD peer_pid = 0, bytes = 0; + if (WSAIoctl(fd, SIO_AF_UNIX_GETPEERPID, NULL, 0, &peer_pid, sizeof(peer_pid), &bytes, NULL, NULL) != 0) return -1; + HANDLE peer = OpenProcess(PROCESS_DUP_HANDLE, FALSE, peer_pid); + if (peer == NULL) return -1; + for (uint32_t i = 0; i < n; i++) { + HANDLE dup = NULL; + BOOL ok = DuplicateHandle(GetCurrentProcess(), (HANDLE)(intptr_t)fds[i], peer, &dup, 0, FALSE, DUPLICATE_SAME_ACCESS); + uint32_t handle = (uint32_t)(uintptr_t)dup; // kernel handles fit in 32 bits + if (!ok || ::send(fd, (char*)&handle, sizeof(handle), 0) < 0) { + CloseHandle(peer); + return -1; + } + } + CloseHandle(peer); + } + return len; + } + if (recv(fd, (char*)&len, sizeof(len), MSG_WAITALL) != sizeof(len) || len > buf_size || + recv(fd, (char*)buf, len, MSG_WAITALL) != (int)len || recv(fd, (char*)&n, sizeof(n), MSG_WAITALL) != sizeof(n)) { + errno = ECONNRESET; + return -1; + } + assert(n == 0 || (fds && (int)n <= num_fds)); + for (uint32_t i = 0; i < n; i++) { + uint32_t handle = 0; + if (recv(fd, (char*)&handle, sizeof(handle), MSG_WAITALL) != sizeof(handle)) { + errno = ECONNRESET; + return -1; + } + fds[i] = handle; + } + if (fds) { + assert(out_num_fds); + *out_num_fds = n; + } + return len; +} +#else int ipc_sendrecv_with_fds(bool send, int fd, void *buf, size_t buf_size, int* fds, int num_fds, int *out_num_fds) { char control_buf[CMSG_SPACE(sizeof(int) * num_fds)]; @@ -119,3 +182,4 @@ int ipc_sendrecv_with_fds(bool send, int fd, void *buf, size_t buf_size, int* fd return r; } } +#endif diff --git a/msgq/visionipc/visionipc.h b/msgq/visionipc/visionipc.h index 224f129c9..d24b25518 100644 --- a/msgq/visionipc/visionipc.h +++ b/msgq/visionipc/visionipc.h @@ -6,6 +6,8 @@ int ipc_connect(const char* socket_path); int ipc_bind(const char* socket_path); +// Close a socket from ipc_connect()/ipc_bind() or accepted on one +void ipc_close(int fd); int ipc_sendrecv_with_fds(bool send, int fd, void *buf, size_t buf_size, int* fds, int num_fds, int *out_num_fds); diff --git a/msgq/visionipc/visionipc_client.cc b/msgq/visionipc/visionipc_client.cc index aa9145223..c8484ebce 100644 --- a/msgq/visionipc/visionipc_client.cc +++ b/msgq/visionipc/visionipc_client.cc @@ -62,7 +62,7 @@ bool VisionIpcClient::connect(bool blocking) { if (r < 0) { // only expected error is server shutting down assert(errno == ECONNRESET); - close(socket_fd); + ipc_close(socket_fd); return false; } @@ -77,7 +77,7 @@ bool VisionIpcClient::connect(bool blocking) { buffers[i].init_yuv(buffers[i].width, buffers[i].height, buffers[i].stride, buffers[i].uv_offset); } - close(socket_fd); + ipc_close(socket_fd); connected = true; return true; } @@ -140,12 +140,12 @@ std::set VisionIpcClient::getAvailableStreams(const std::strin if (r < 0) { // only expected error is server shutting down assert(errno == ECONNRESET); - close(socket_fd); + ipc_close(socket_fd); return {}; } assert(r % sizeof(VisionStreamType) == 0); - close(socket_fd); + ipc_close(socket_fd); return std::set(available_streams, available_streams + r / sizeof(VisionStreamType)); } diff --git a/msgq/visionipc/visionipc_server.cc b/msgq/visionipc/visionipc_server.cc index 0b521fd2d..83b0813ad 100644 --- a/msgq/visionipc/visionipc_server.cc +++ b/msgq/visionipc/visionipc_server.cc @@ -1,14 +1,20 @@ #include #include #include +#include #include #include #include #include +#ifdef _WIN32 +#include +#define poll WSAPoll +#else #include #include #include +#endif #include "msgq/visionipc/visionipc.h" #include "msgq/visionipc/visionipc_server.h" @@ -19,7 +25,11 @@ std::string get_endpoint_name(std::string name, VisionStreamType type){ } std::string get_ipc_path(const std::string& name) { +#ifdef _WIN32 + std::string path = std::filesystem::temp_directory_path().string() + "/"; // "/tmp" would be the current drive's root +#else std::string path = "/tmp/"; +#endif if (char* prefix = std::getenv("OPENPILOT_PREFIX")) { path += std::string(prefix) + "_"; } @@ -102,7 +112,7 @@ void VisionIpcServer::listener(){ VisionStreamType type = VISION_STREAM_LIST; int r = ipc_sendrecv_with_fds(false, fd, &type, sizeof(type), nullptr, 0, nullptr); if (r != sizeof(type)) { - close(fd); + ipc_close(fd); if (should_exit) break; continue; } @@ -115,13 +125,13 @@ void VisionIpcServer::listener(){ } r = ipc_sendrecv_with_fds(true, fd, available_stream_types.data(), available_stream_types.size() * sizeof(VisionStreamType), nullptr, 0, nullptr); assert(r == available_stream_types.size() * sizeof(VisionStreamType)); - close(fd); + ipc_close(fd); continue; } if (buffers.count(type) <= 0) { std::cout << "got request for invalid buffer type: " << type << std::endl; - close(fd); + ipc_close(fd); continue; } @@ -141,11 +151,11 @@ void VisionIpcServer::listener(){ r = ipc_sendrecv_with_fds(true, fd, &bufs, sizeof(VisionBuf) * num_fds, fds, num_fds, nullptr); - close(fd); + ipc_close(fd); } LOGD("Stopping listener for: %s", name.c_str()); - close(sock); + ipc_close(sock); unlink(ipc_path.c_str()); } @@ -187,7 +197,7 @@ VisionIpcServer::~VisionIpcServer(){ if (listener_thread.joinable()) { int sock = ipc_connect(get_ipc_path(name).c_str()); if (sock >= 0) { - close(sock); + ipc_close(sock); } listener_thread.join(); } diff --git a/setup.py b/setup.py index 45e2c172d..1bb741ec6 100644 --- a/setup.py +++ b/setup.py @@ -11,7 +11,7 @@ class SConsBuildExt(build_ext): def build_extensions(self): subprocess.check_call([sys.executable, "-m", "SCons", "--minimal", "-j", str(self.parallel or os.cpu_count() or 1)]) for ext in self.extensions: - source = Path(*ext.name.split(".")).with_suffix(".so") + source = Path(*ext.name.split(".")).with_suffix(".pyd" if sys.platform == "win32" else ".so") target = Path(self.get_ext_fullpath(ext.name)) target.parent.mkdir(parents=True, exist_ok=True) self.copy_file(str(source), str(target)) diff --git a/setup.sh b/setup.sh index 09e3af82c..3de23775e 100755 --- a/setup.sh +++ b/setup.sh @@ -2,7 +2,7 @@ set -e DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" >/dev/null && pwd)" -cd $DIR +cd "$DIR" if ! command -v uv &>/dev/null; then echo "'uv' is not installed. Installing 'uv'..." @@ -10,10 +10,12 @@ if ! command -v uv &>/dev/null; then # doesn't require sourcing on all platforms set +e - source $HOME/.local/bin/env + source "$HOME/.local/bin/env" set -e fi +case "$(uname -s)" in MINGW*|MSYS*) export UV_PYTHON="${UV_PYTHON:-3.12}";; esac # not MSYS2's own python, whose wheels are incompatible + export UV_PROJECT_ENVIRONMENT="$DIR/.venv" uv sync --all-extras -source "$DIR/.venv/bin/activate" +source "$DIR"/.venv/*/activate # bin/ on POSIX, Scripts/ on Windows diff --git a/site_scons/site_tools/cython.py b/site_scons/site_tools/cython.py index c29147553..b7d709b79 100644 --- a/site_scons/site_tools/cython.py +++ b/site_scons/site_tools/cython.py @@ -68,5 +68,11 @@ def generate(env): create_builder(env) + # Python imports extension modules as .pyd on Windows; the SConscripts name them .so + if env["PLATFORM"] == "win32": + def pyd_emitter(target, source, env): + return [env.File(str(t)[:-3] + ".pyd") if str(t).endswith(".so") else t for t in target], source + env.Append(PROGEMITTER=[pyd_emitter]) + def exists(env): return True diff --git a/test.sh b/test.sh index bf3a0e0ef..55deb46f8 100755 --- a/test.sh +++ b/test.sh @@ -2,7 +2,7 @@ set -e DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" >/dev/null && pwd)" -cd $DIR +cd "$DIR" source ./setup.sh @@ -16,10 +16,10 @@ lefthook run test ( TEST_DIR=$(mktemp -d) trap 'rm -rf "$TEST_DIR"' EXIT - uv venv --python "$DIR/.venv/bin/python" "$TEST_DIR/.venv" - uv pip install --python "$TEST_DIR/.venv/bin/python" "$DIR" + uv venv --python "$DIR/.venv" "$TEST_DIR/.venv" + uv pip install --python "$TEST_DIR/.venv" "$DIR" cd "$TEST_DIR" - "$TEST_DIR/.venv/bin/python" -m unittest msgq.tests.test_messaging + uv run --no-project python -m unittest msgq.tests.test_messaging ) # *** all done ***