adolforeyna/tactility
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Various services improved (#110)

Browse Source
This commit is contained in:
Ken Van Hoeylandt
2024-12-06 23:16:29 +01:00
committed by GitHub
parent 36bb25deba
commit d52fe52d96
32 changed files with 771 additions and 636 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TactilityCore/Source/Dispatcher.cpp
+31 -20
View File
@@ -1,39 +1,50 @@
#include "Dispatcher.h"
#include "Check.h"
namespace tt {
Dispatcher::Dispatcher(size_t queueLimit) :
queue(queueLimit, sizeof(DispatcherMessage)),
mutex(MutexTypeNormal),
buffer({ .callback = nullptr, .context = nullptr }) { }
#define TAG "Dispatcher"
#define BACKPRESSURE_WARNING_COUNT 100
Dispatcher::Dispatcher() :
mutex(MutexTypeNormal)
{}
Dispatcher::~Dispatcher() {
queue.reset();
// Wait for Mutex usage
mutex.acquire(TtWaitForever);
mutex.release();
}
void Dispatcher::dispatch(Callback callback, void* context) {
DispatcherMessage message = {
.callback = callback,
.context = context
};
void Dispatcher::dispatch(Callback callback, std::shared_ptr<void> context) {
auto message = std::make_shared<DispatcherMessage>(callback, std::move(context));
// Mutate
mutex.acquire(TtWaitForever);
queue.put(&message, TtWaitForever);
queue.push(std::move(message));
if (queue.size() == BACKPRESSURE_WARNING_COUNT) {
TT_LOG_W(TAG, "Backpressure: You're not consuming fast enough (100 queued)");
}
mutex.release();
// Signal
eventFlag.set(1);
}
bool Dispatcher::consume(uint32_t timeout_ticks) {
mutex.acquire(TtWaitForever);
if (queue.get(&buffer, timeout_ticks) == TtStatusOk) {
buffer.callback(buffer.context);
mutex.release();
return true;
} else {
mutex.release();
return false;
uint32_t Dispatcher::consume(uint32_t timeout_ticks) {
// Wait for signal and clear
eventFlag.wait(1, TtFlagWaitAny, timeout_ticks);
eventFlag.clear(1);
// Mutate
if (mutex.acquire(1 / portTICK_PERIOD_MS) == TtStatusOk) {
auto item = queue.front();
queue.pop();
// Don't keep lock as callback might be slow
tt_check(mutex.release() == TtStatusOk);
item->callback(item->context);
}
return true;
}
} // namespace
TactilityCore/Source/Dispatcher.h
+19 -9
View File
@@ -7,29 +7,39 @@
#include "MessageQueue.h"
#include "Mutex.h"
#include "EventFlag.h"
#include <memory>
#include <queue>
namespace tt {
typedef void (*Callback)(void* data);
typedef void (*Callback)(std::shared_ptr<void> data);
class Dispatcher {
private:
typedef struct {
struct DispatcherMessage {
Callback callback;
void* context;
} DispatcherMessage;
std::shared_ptr<void> context; // Can't use unique_ptr with void, so we use shared_ptr
DispatcherMessage(Callback callback, std::shared_ptr<void> context) :
callback(callback),
context(std::move(context))
{}
~DispatcherMessage() = default;
};
MessageQueue queue;
Mutex mutex;
DispatcherMessage buffer; // Buffer for consuming a message
std::queue<std::shared_ptr<DispatcherMessage>> queue;
EventFlag eventFlag;
public:
explicit Dispatcher(size_t queueLimit = 8);
explicit Dispatcher();
~Dispatcher();
void dispatch(Callback callback, void* context);
bool consume(uint32_t timeout_ticks);
void dispatch(Callback callback, std::shared_ptr<void> context);
uint32_t consume(uint32_t timeout_ticks);
};
} // namespace
TactilityCore/Source/Mutex.cpp
+4 -1
View File
@@ -111,6 +111,10 @@ ThreadId Mutex::getOwner() const {
}
std::unique_ptr<ScopedMutexUsage> Mutex::scoped() const {
return std::move(std::make_unique<ScopedMutexUsage>(*this));
}
Mutex* tt_mutex_alloc(MutexType type) {
return new Mutex(type);
}
@@ -125,7 +129,6 @@ TtStatus tt_mutex_acquire(Mutex* mutex, uint32_t timeout) {
TtStatus tt_mutex_release(Mutex* mutex) {
return mutex->release();
}
ThreadId tt_mutex_get_owner(Mutex* mutex) {
TactilityCore/Source/Mutex.h
+28
View File
@@ -7,9 +7,13 @@
#include "CoreTypes.h"
#include "Thread.h"
#include "RtosCompatSemaphore.h"
#include "Check.h"
#include <memory>
namespace tt {
class ScopedMutexUsage;
typedef enum {
MutexTypeNormal,
MutexTypeRecursive,
@@ -30,6 +34,30 @@ public:
TtStatus acquire(uint32_t timeout) const;
TtStatus release() const;
ThreadId getOwner() const;
std::unique_ptr<ScopedMutexUsage> scoped() const;
};
class ScopedMutexUsage {
const Mutex& mutex;
bool acquired = false;
public:
ScopedMutexUsage(const Mutex& mutex) : mutex(mutex) {}
~ScopedMutexUsage() {
if (acquired) {
tt_check(mutex.release() == TtStatusOk);
}
}
bool acquire(uint32_t timeout) {
TtStatus result = mutex.acquire(timeout);
acquired = (result == TtStatusOk);
return acquired;
}
};
/** Allocate Mutex
TactilityCore/Source/Pubsub.cpp
+9 -40
View File
@@ -1,42 +1,11 @@
#include "Pubsub.h"
#include "Check.h"
#include "Mutex.h"
#include <list>
namespace tt {
struct PubSubSubscription {
uint64_t id;
PubSubCallback callback;
void* callback_context;
};
typedef std::list<PubSubSubscription> Subscriptions;
struct PubSub {
uint64_t last_id = 0;
Subscriptions items;
Mutex* mutex;
};
PubSub* tt_pubsub_alloc() {
auto* pubsub = new PubSub();
pubsub->mutex = tt_mutex_alloc(MutexTypeNormal);
tt_assert(pubsub->mutex);
return pubsub;
}
void tt_pubsub_free(PubSub* pubsub) {
tt_assert(pubsub);
tt_check(pubsub->items.empty());
tt_mutex_free(pubsub->mutex);
delete pubsub;
}
PubSubSubscription* tt_pubsub_subscribe(PubSub* pubsub, PubSubCallback callback, void* callback_context) {
tt_check(tt_mutex_acquire(pubsub->mutex, TtWaitForever) == TtStatusOk);
PubSubSubscription* tt_pubsub_subscribe(std::shared_ptr<PubSub> pubsub, PubSubCallback callback, void* callback_context) {
tt_check(pubsub->mutex.acquire(TtWaitForever) == TtStatusOk);
PubSubSubscription subscription = {
.id = (++pubsub->last_id),
.callback = callback,
@@ -46,16 +15,16 @@ PubSubSubscription* tt_pubsub_subscribe(PubSub* pubsub, PubSubCallback callback,
subscription
);
tt_check(tt_mutex_release(pubsub->mutex) == TtStatusOk);
tt_check(pubsub->mutex.release() == TtStatusOk);
return (PubSubSubscription*)pubsub->last_id;
}
void tt_pubsub_unsubscribe(PubSub* pubsub, PubSubSubscription* pubsub_subscription) {
void tt_pubsub_unsubscribe(std::shared_ptr<PubSub> pubsub, PubSubSubscription* pubsub_subscription) {
tt_assert(pubsub);
tt_assert(pubsub_subscription);
tt_check(tt_mutex_acquire(pubsub->mutex, TtWaitForever) == TtStatusOk);
tt_check(pubsub->mutex.acquire(TtWaitForever) == TtStatusOk);
bool result = false;
auto id = (uint64_t)pubsub_subscription;
for (auto it = pubsub->items.begin(); it != pubsub->items.end(); it++) {
@@ -66,19 +35,19 @@ void tt_pubsub_unsubscribe(PubSub* pubsub, PubSubSubscription* pubsub_subscripti
}
}
tt_check(tt_mutex_release(pubsub->mutex) == TtStatusOk);
tt_check(pubsub->mutex.release() == TtStatusOk);
tt_check(result);
}
void tt_pubsub_publish(PubSub* pubsub, void* message) {
tt_check(tt_mutex_acquire(pubsub->mutex, TtWaitForever) == TtStatusOk);
void tt_pubsub_publish(std::shared_ptr<PubSub> pubsub, void* message) {
tt_check(pubsub->mutex.acquire(TtWaitForever) == TtStatusOk);
// Iterate over subscribers
for (auto& it : pubsub->items) {
it.callback(message, it.callback_context);
}
tt_check(tt_mutex_release(pubsub->mutex) == TtStatusOk);
tt_check(pubsub->mutex.release() == TtStatusOk);
}
} // namespace
TactilityCore/Source/Pubsub.h
+20 -20
View File
@@ -4,30 +4,30 @@
*/
#pragma once
#include "Mutex.h"
#include <list>
namespace tt {
/** PubSub Callback type */
typedef void (*PubSubCallback)(const void* message, void* context);
/** PubSub type */
typedef struct PubSub PubSub;
struct PubSubSubscription {
uint64_t id;
PubSubCallback callback;
void* callback_context;
};
/** PubSubSubscription type */
typedef struct PubSubSubscription PubSubSubscription;
struct PubSub {
typedef std::list<PubSubSubscription> Subscriptions;
uint64_t last_id = 0;
Subscriptions items;
Mutex mutex;
/** Allocate PubSub
*
* Reentrable, Not threadsafe, one owner
*
* @return pointer to PubSub instance
*/
PubSub* tt_pubsub_alloc();
/** Free PubSub
*
* @param pubsub PubSub instance
*/
void tt_pubsub_free(PubSub* pubsub);
~PubSub() {
tt_check(items.empty());
}
};
/** Subscribe to PubSub
*
@@ -40,7 +40,7 @@ void tt_pubsub_free(PubSub* pubsub);
* @return pointer to PubSubSubscription instance
*/
PubSubSubscription*
tt_pubsub_subscribe(PubSub* pubsub, PubSubCallback callback, void* callback_context);
tt_pubsub_subscribe(std::shared_ptr<PubSub> pubsub, PubSubCallback callback, void* callback_context);
/** Unsubscribe from PubSub
*
@@ -50,7 +50,7 @@ tt_pubsub_subscribe(PubSub* pubsub, PubSubCallback callback, void* callback_cont
* @param pubsub pointer to PubSub instance
* @param pubsub_subscription pointer to PubSubSubscription instance
*/
void tt_pubsub_unsubscribe(PubSub* pubsub, PubSubSubscription* pubsub_subscription);
void tt_pubsub_unsubscribe(std::shared_ptr<PubSub> pubsub, PubSubSubscription* pubsub_subscription);
/** Publish message to PubSub
*
@@ -59,6 +59,6 @@ void tt_pubsub_unsubscribe(PubSub* pubsub, PubSubSubscription* pubsub_subscripti
* @param pubsub pointer to PubSub instance
* @param message message pointer to publish
*/
void tt_pubsub_publish(PubSub* pubsub, void* message);
void tt_pubsub_publish(std::shared_ptr<PubSub> pubsub, void* message);
} // namespace
TactilityCore/Source/Timer.cpp
+4 -2
View File
@@ -1,4 +1,6 @@
#include "Timer.h"
#include <utility>
#include "Check.h"
#include "Kernel.h"
#include "RtosCompat.h"
@@ -13,11 +15,11 @@ static void timer_callback(TimerHandle_t hTimer) {
}
}
Timer::Timer(Type type, Callback callback, void* callbackContext) {
Timer::Timer(Type type, Callback callback, std::shared_ptr<void> callbackContext) {
tt_assert((kernel_is_irq() == 0U) && (callback != nullptr));
this->callback = callback;
this->callbackContext = callbackContext;
this->callbackContext = std::move(callbackContext);
UBaseType_t reload;
if (type == TypeOnce) {
TactilityCore/Source/Timer.h
+4 -3
View File
@@ -3,6 +3,7 @@
#include "CoreTypes.h"
#include "RtosCompatTimers.h"
#include <memory>
namespace tt {
@@ -11,12 +12,12 @@ private:
TimerHandle_t timerHandle;
public:
typedef void (*Callback)(void* context);
typedef void (*Callback)(std::shared_ptr<void> context);
typedef void (*PendingCallback)(void* context, uint32_t arg);
Callback callback;
void* callbackContext;
std::shared_ptr<void> callbackContext;
typedef enum {
TypeOnce = 0, ///< One-shot timer.
@@ -28,7 +29,7 @@ public:
* @param[in] callback The callback function
* @param callbackContext The callback context
*/
Timer(Type type, Callback callback, void* callbackContext);
Timer(Type type, Callback callback, std::shared_ptr<void> callbackContext);
~Timer();