oss/luanti
oss/luanti
1
0
Fork 0
mirror of https://github.com/luanti-org/luanti.git synced 2025-08-11 17:51:04 +00:00
Code Wiki Activity

use more abstract macros to decide which implementation to compile

Browse source
This commit is contained in:
Desour 2023-06-08 22:54:31 +02:00
parent 4f0085af00
commit 1c76f32d19
3 changed files with 57 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

69
src/threading/ipc_channel.cpp
View file

@ -22,11 +22,12 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include "debug.h" #include "debug.h"
#include "exceptions.h" #include "exceptions.h"
#include "porting.h" #include "porting.h"
#include <errno.h> #include <cerrno>
#include <utility> #include <utility>
#if defined(__linux__)
#if defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
#include <linux/futex.h> #include <linux/futex.h>
#include <string.h> #include <cstring>
#include <sys/syscall.h> #include <sys/syscall.h>
#include <sys/wait.h> #include <sys/wait.h>
#if defined(__i386__) || defined(__x86_64__) #if defined(__i386__) || defined(__x86_64__)
@ -36,7 +37,7 @@ with this program; if not, write to the Free Software Foundation, Inc.,
IPCChannelBuffer::IPCChannelBuffer() IPCChannelBuffer::IPCChannelBuffer()
{ {
#if !defined(__linux__) && !defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
pthread_condattr_t condattr; pthread_condattr_t condattr;
pthread_mutexattr_t mutexattr; pthread_mutexattr_t mutexattr;
if (pthread_condattr_init(&condattr) != 0) if (pthread_condattr_init(&condattr) != 0)
@ -65,18 +66,18 @@ error_mutexattr_init:
pthread_condattr_destroy(&condattr); pthread_condattr_destroy(&condattr);
error_condattr_init: error_condattr_init:
throw BaseException("Unable to initialize IPCChannelBuffer"); throw BaseException("Unable to initialize IPCChannelBuffer");
#endif // !defined(__linux__) && !defined(_WIN32) #endif // defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
} }
IPCChannelBuffer::~IPCChannelBuffer() IPCChannelBuffer::~IPCChannelBuffer()
{ {
#if !defined(__linux__) && !defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
pthread_mutex_destroy(&mutex); pthread_mutex_destroy(&mutex);
pthread_cond_destroy(&cond); pthread_cond_destroy(&cond);
#endif // !defined(__linux__) && !defined(_WIN32) #endif // defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
} }
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
// returns false on timeout // returns false on timeout
static bool wait(HANDLE sem, DWORD timeout) static bool wait(HANDLE sem, DWORD timeout)
@ -92,7 +93,7 @@ static void post(HANDLE sem)
#else #else
#if defined(__linux__) #if defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
#if defined(__i386__) || defined(__x86_64__) #if defined(__i386__) || defined(__x86_64__)
static void busy_wait(int n) noexcept static void busy_wait(int n) noexcept
@ -108,13 +109,13 @@ static int futex(std::atomic<u32> *uaddr, int futex_op, u32 val,
return syscall(SYS_futex, uaddr, futex_op, val, timeout, uaddr2, val3); return syscall(SYS_futex, uaddr, futex_op, val, timeout, uaddr2, val3);
} }
#endif // defined(__linux__) #endif // defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
// timeout: relative on linux, and absolute on other posix // timeout: relative on linux, and absolute on other posix
// returns false on timeout // returns false on timeout
static bool wait(IPCChannelBuffer *buf, const struct timespec *timeout) noexcept static bool wait(IPCChannelBuffer *buf, const struct timespec *timeout) noexcept
{ {
#if defined(__linux__) #if defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
// try busy waiting // try busy waiting
for (int i = 0; i < 100; i++) { for (int i = 0; i < 100; i++) {
// posted? // posted?
@ -145,7 +146,7 @@ static bool wait(IPCChannelBuffer *buf, const struct timespec *timeout) noexcept
} }
} }
} }
#else #elif defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
bool timed_out = false; bool timed_out = false;
pthread_mutex_lock(&buf->mutex); pthread_mutex_lock(&buf->mutex);
if (!buf->posted) { if (!buf->posted) {
@ -157,12 +158,12 @@ static bool wait(IPCChannelBuffer *buf, const struct timespec *timeout) noexcept
buf->posted = false; buf->posted = false;
pthread_mutex_unlock(&buf->mutex); pthread_mutex_unlock(&buf->mutex);
return !timed_out; return !timed_out;
#endif // !defined(__linux__) #endif
} }
static void post(IPCChannelBuffer *buf) noexcept static void post(IPCChannelBuffer *buf) noexcept
{ {
#if defined(__linux__) #if defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
if (buf->futex.exchange(1) == 2) { if (buf->futex.exchange(1) == 2) {
// 2 means reader needs to be notified // 2 means reader needs to be notified
int s = futex(&buf->futex, FUTEX_WAKE, 1, nullptr, nullptr, 0); int s = futex(&buf->futex, FUTEX_WAKE, 1, nullptr, nullptr, 0);
@ -172,35 +173,35 @@ static void post(IPCChannelBuffer *buf) noexcept
FATAL_ERROR(errmsg.c_str()); FATAL_ERROR(errmsg.c_str());
} }
} }
#else #elif defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
pthread_mutex_lock(&buf->mutex); pthread_mutex_lock(&buf->mutex);
buf->posted = true; buf->posted = true;
pthread_cond_broadcast(&buf->cond); pthread_cond_broadcast(&buf->cond);
pthread_mutex_unlock(&buf->mutex); pthread_mutex_unlock(&buf->mutex);
#endif // !defined(__linux__) #endif
} }
#endif // !defined(_WIN32) #endif // !defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
static DWORD get_timeout(int timeout_ms) static DWORD get_timeout(int timeout_ms)
{ {
return timeout_ms < 0 ? INFINITE : (DWORD)timeout_ms; return timeout_ms < 0 ? INFINITE : (DWORD)timeout_ms;
} }
#else #elif defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX) || defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
static struct timespec *set_timespec(struct timespec *ts, int ms) static struct timespec *set_timespec(struct timespec *ts, int ms)
{ {
if (ms < 0) if (ms < 0)
return nullptr; return nullptr;
u64 msu = ms; u64 msu = ms;
#if !defined(__linux__) #if defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
msu += porting::getTimeMs(); // Absolute time msu += porting::getTimeMs(); // Absolute time
#endif #endif
ts->tv_sec = msu / 1000; ts->tv_sec = msu / 1000;
ts->tv_nsec = msu % 1000 * 1000000UL; ts->tv_nsec = msu % 1000 * 1000000UL;
return ts; return ts;
} }
#endif // !defined(_WIN32) #endif
template <typename T> template <typename T>
static inline void write_once(volatile T *var, const T val) static inline void write_once(volatile T *var, const T val)
@ -217,32 +218,32 @@ static inline T read_once(const volatile T *var)
IPCChannelEnd IPCChannelEnd::makeA(std::unique_ptr<IPCChannelStuff> stuff) IPCChannelEnd IPCChannelEnd::makeA(std::unique_ptr<IPCChannelStuff> stuff)
{ {
IPCChannelShared *shared = stuff->getShared(); IPCChannelShared *shared = stuff->getShared();
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
HANDLE sem_a = stuff->getSemA(); HANDLE sem_a = stuff->getSemA();
HANDLE sem_b = stuff->getSemB(); HANDLE sem_b = stuff->getSemB();
return IPCChannelEnd(std::move(stuff), &shared->a, &shared->b, sem_a, sem_b); return IPCChannelEnd(std::move(stuff), &shared->a, &shared->b, sem_a, sem_b);
#else #else
return IPCChannelEnd(std::move(stuff), &shared->a, &shared->b); return IPCChannelEnd(std::move(stuff), &shared->a, &shared->b);
#endif // !defined(_WIN32) #endif // !defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
} }
IPCChannelEnd IPCChannelEnd::makeB(std::unique_ptr<IPCChannelStuff> stuff) IPCChannelEnd IPCChannelEnd::makeB(std::unique_ptr<IPCChannelStuff> stuff)
{ {
IPCChannelShared *shared = stuff->getShared(); IPCChannelShared *shared = stuff->getShared();
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
HANDLE sem_a = stuff->getSemA(); HANDLE sem_a = stuff->getSemA();
HANDLE sem_b = stuff->getSemB(); HANDLE sem_b = stuff->getSemB();
return IPCChannelEnd(std::move(stuff), &shared->b, &shared->a, sem_b, sem_a); return IPCChannelEnd(std::move(stuff), &shared->b, &shared->a, sem_b, sem_a);
#else #else
return IPCChannelEnd(std::move(stuff), &shared->b, &shared->a); return IPCChannelEnd(std::move(stuff), &shared->b, &shared->a);
#endif // !defined(_WIN32) #endif // !defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
} }
void IPCChannelEnd::sendSmall(const void *data, size_t size) noexcept void IPCChannelEnd::sendSmall(const void *data, size_t size) noexcept
{ {
write_once(&m_out->size, size); write_once(&m_out->size, size);
memcpy(m_out->data, data, size); memcpy(m_out->data, data, size);
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
post(m_sem_out); post(m_sem_out);
#else #else
post(m_out); post(m_out);
@ -251,7 +252,7 @@ void IPCChannelEnd::sendSmall(const void *data, size_t size) noexcept
bool IPCChannelEnd::sendLarge(const void *data, size_t size, int timeout_ms) noexcept bool IPCChannelEnd::sendLarge(const void *data, size_t size, int timeout_ms) noexcept
{ {
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
DWORD timeout = get_timeout(timeout_ms); DWORD timeout = get_timeout(timeout_ms);
#else #else
struct timespec timeout; struct timespec timeout;
@ -260,12 +261,12 @@ bool IPCChannelEnd::sendLarge(const void *data, size_t size, int timeout_ms) noe
write_once(&m_out->size, size); write_once(&m_out->size, size);
do { do {
memcpy(m_out->data, data, IPC_CHANNEL_MSG_SIZE); memcpy(m_out->data, data, IPC_CHANNEL_MSG_SIZE);
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
post(m_sem_out); post(m_sem_out);
#else #else
post(m_out); post(m_out);
#endif #endif
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
if (!wait(m_sem_in, timeout)) if (!wait(m_sem_in, timeout))
#else #else
if (!wait(m_in, timeoutp)) if (!wait(m_in, timeoutp))
@ -275,7 +276,7 @@ bool IPCChannelEnd::sendLarge(const void *data, size_t size, int timeout_ms) noe
data = (u8 *)data + IPC_CHANNEL_MSG_SIZE; data = (u8 *)data + IPC_CHANNEL_MSG_SIZE;
} while (size > IPC_CHANNEL_MSG_SIZE); } while (size > IPC_CHANNEL_MSG_SIZE);
memcpy(m_out->data, data, size); memcpy(m_out->data, data, size);
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
post(m_sem_out); post(m_sem_out);
#else #else
post(m_out); post(m_out);
@ -285,13 +286,13 @@ bool IPCChannelEnd::sendLarge(const void *data, size_t size, int timeout_ms) noe
bool IPCChannelEnd::recv(int timeout_ms) noexcept bool IPCChannelEnd::recv(int timeout_ms) noexcept
{ {
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
DWORD timeout = get_timeout(timeout_ms); DWORD timeout = get_timeout(timeout_ms);
#else #else
struct timespec timeout; struct timespec timeout;
struct timespec *timeoutp = set_timespec(&timeout, timeout_ms); struct timespec *timeoutp = set_timespec(&timeout, timeout_ms);
#endif #endif
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
if (!wait(m_sem_in, timeout)) if (!wait(m_sem_in, timeout))
#else #else
if (!wait(m_in, timeoutp)) if (!wait(m_in, timeoutp))
@ -316,12 +317,12 @@ bool IPCChannelEnd::recv(int timeout_ms) noexcept
memcpy(recv_data, m_in->data, IPC_CHANNEL_MSG_SIZE); memcpy(recv_data, m_in->data, IPC_CHANNEL_MSG_SIZE);
size -= IPC_CHANNEL_MSG_SIZE; size -= IPC_CHANNEL_MSG_SIZE;
recv_data += IPC_CHANNEL_MSG_SIZE; recv_data += IPC_CHANNEL_MSG_SIZE;
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
post(m_sem_out); post(m_sem_out);
#else #else
post(m_out); post(m_out);
#endif #endif
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
if (!wait(m_sem_in, timeout)) if (!wait(m_sem_in, timeout))
#else #else
if (!wait(m_in, timeoutp)) if (!wait(m_in, timeoutp))

26
src/threading/ipc_channel.h
View file

@ -27,10 +27,18 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include <vector> #include <vector>
#if defined(_WIN32) #if defined(_WIN32)
#include <windows.h> #define IPC_CHANNEL_IMPLEMENTATION_WIN32
#elif defined(__linux__) #elif defined(__linux__)
#include <atomic> #define IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX
#else #else
#define IPC_CHANNEL_IMPLEMENTATION_POSIX
#endif
#if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
#include <windows.h>
#elif defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
#include <atomic>
#elif defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
#include <pthread.h> #include <pthread.h>
#endif #endif
@ -52,8 +60,7 @@ with this program; if not, write to the Free Software Foundation, Inc.,
struct IPCChannelBuffer struct IPCChannelBuffer
{ {
#if !defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_LINUX_FUTEX)
#if defined(__linux__)
// possible values: // possible values:
// 0: futex is not posted. reader will check value before blocking => no // 0: futex is not posted. reader will check value before blocking => no
// notify needed when posting // notify needed when posting
@ -61,12 +68,13 @@ struct IPCChannelBuffer
// 2: futex is not posted. reader is waiting with futex syscall, and needs // 2: futex is not posted. reader is waiting with futex syscall, and needs
// to be notified // to be notified
std::atomic<u32> futex{0}; std::atomic<u32> futex{0};
#else
#elif defined(IPC_CHANNEL_IMPLEMENTATION_POSIX)
pthread_cond_t cond; pthread_cond_t cond;
pthread_mutex_t mutex; pthread_mutex_t mutex;
bool posted = false; // protected by mutex bool posted = false; // protected by mutex
#endif #endif
#endif // !defined(_WIN32)
// Note: If the other side isn't acting cooperatively, they might write to // Note: If the other side isn't acting cooperatively, they might write to
// this at any times. So we must make sure to copy out the data once, and // this at any times. So we must make sure to copy out the data once, and
// only access that copy. // only access that copy.
@ -89,7 +97,7 @@ struct IPCChannelStuff
{ {
virtual ~IPCChannelStuff() = default; virtual ~IPCChannelStuff() = default;
virtual IPCChannelShared *getShared() = 0; virtual IPCChannelShared *getShared() = 0;
#ifdef _WIN32 #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
virtual HANDLE getSemA() = 0; virtual HANDLE getSemA() = 0;
virtual HANDLE getSemB() = 0; virtual HANDLE getSemB() = 0;
#endif #endif
@ -128,7 +136,7 @@ public:
inline size_t getRecvSize() const noexcept { return m_recv_size; } inline size_t getRecvSize() const noexcept { return m_recv_size; }
private: private:
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
IPCChannelEnd( IPCChannelEnd(
std::unique_ptr<IPCChannelStuff> stuff, std::unique_ptr<IPCChannelStuff> stuff,
IPCChannelBuffer *in, IPCChannelBuffer *out, IPCChannelBuffer *in, IPCChannelBuffer *out,
@ -154,7 +162,7 @@ private:
std::unique_ptr<IPCChannelStuff> m_stuff; std::unique_ptr<IPCChannelStuff> m_stuff;
IPCChannelBuffer *m_in = nullptr; IPCChannelBuffer *m_in = nullptr;
IPCChannelBuffer *m_out = nullptr; IPCChannelBuffer *m_out = nullptr;
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
HANDLE m_sem_in; HANDLE m_sem_in;
HANDLE m_sem_out; HANDLE m_sem_out;
#endif #endif

10
src/unittest/test_threading.cpp
View file

@ -6,7 +6,7 @@
#include <atomic> #include <atomic>
#include <iostream> #include <iostream>
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32) || defined(_WIN32)
#include <windows.h> #include <windows.h>
#endif #endif
#include "threading/ipc_channel.h" #include "threading/ipc_channel.h"
@ -239,13 +239,13 @@ void TestThreading::testIPCChannel()
struct Stuff struct Stuff
{ {
IPCChannelShared shared{}; IPCChannelShared shared{};
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
HANDLE sem_a; HANDLE sem_a;
HANDLE sem_b; HANDLE sem_b;
#endif #endif
Stuff() Stuff()
{ {
#ifdef _WIN32 #ifdef IPC_CHANNEL_IMPLEMENTATION_WIN32
HANDLE sem_a = CreateSemaphoreA(nullptr, 0, 1, nullptr); HANDLE sem_a = CreateSemaphoreA(nullptr, 0, 1, nullptr);
UASSERT(sem_a != INVALID_HANDLE_VALUE); UASSERT(sem_a != INVALID_HANDLE_VALUE);
@ -256,7 +256,7 @@ void TestThreading::testIPCChannel()
~Stuff() ~Stuff()
{ {
#ifdef _WIN32 #ifdef IPC_CHANNEL_IMPLEMENTATION_WIN32
CloseHandle(sem_b); CloseHandle(sem_b);
CloseHandle(sem_a); CloseHandle(sem_a);
#endif #endif
@ -271,7 +271,7 @@ void TestThreading::testIPCChannel()
~IPCChannelStuffSingleProcess() override = default; ~IPCChannelStuffSingleProcess() override = default;
IPCChannelShared *getShared() override { return &stuff->shared; } IPCChannelShared *getShared() override { return &stuff->shared; }
#if defined(_WIN32) #if defined(IPC_CHANNEL_IMPLEMENTATION_WIN32)
HANDLE getSemA() override { return stuff->sem_a; } HANDLE getSemA() override { return stuff->sem_a; }
HANDLE getSemB() override { return stuff->sem_b; } HANDLE getSemB() override { return stuff->sem_b; }
#endif #endif