42e843b463
* Remove version from artifact name * Target C++ 20 and higher * Use cpp string * Better crash implementation * String utils in cpp style * Replace parameter methods with start() method * MutexType to Mutex::Type * Kernel c to cpp style * Cleanup event flag * More cpp conversions * Test fixes * Updated ideas docs
51 lines
1.2 KiB
C++
51 lines
1.2 KiB
C++
#include "Dispatcher.h"
|
|
#include "Check.h"
|
|
|
|
namespace tt {
|
|
|
|
#define TAG "Dispatcher"
|
|
#define BACKPRESSURE_WARNING_COUNT 100
|
|
|
|
Dispatcher::Dispatcher() :
|
|
mutex(Mutex::TypeNormal)
|
|
{}
|
|
|
|
Dispatcher::~Dispatcher() {
|
|
// Wait for Mutex usage
|
|
mutex.acquire(TtWaitForever);
|
|
mutex.release();
|
|
}
|
|
|
|
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.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);
|
|
}
|
|
|
|
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
|