133 lines
2.9 KiB
C++
133 lines
2.9 KiB
C++
// Copyright (c) Charles J. Cliffe
|
|
// SPDX-License-Identifier: GPL-2.0+
|
|
|
|
#include "IOThread.h"
|
|
#include <memory>
|
|
|
|
#define SPIN_WAIT_SLEEP_MS 5
|
|
|
|
IOThread::IOThread() {
|
|
terminated.store(false);
|
|
stopping.store(false);
|
|
}
|
|
|
|
IOThread::~IOThread() {
|
|
terminated.store(true);
|
|
stopping.store(true);
|
|
}
|
|
|
|
#ifdef __APPLE__
|
|
void *IOThread::threadMain() {
|
|
terminated.store(false);
|
|
stopping.store(false);
|
|
try {
|
|
run();
|
|
}
|
|
catch (...) {
|
|
terminated.store(true);
|
|
stopping.store(true);
|
|
throw;
|
|
}
|
|
|
|
terminated.store(true);
|
|
stopping.store(true);
|
|
return this;
|
|
};
|
|
|
|
void *IOThread::pthread_helper(void *context) {
|
|
return ((IOThread *) context)->threadMain();
|
|
};
|
|
#else
|
|
void IOThread::threadMain() {
|
|
terminated.store(false);
|
|
stopping.store(false);
|
|
try {
|
|
run();
|
|
}
|
|
catch (...) {
|
|
terminated.store(true);
|
|
stopping.store(true);
|
|
throw;
|
|
}
|
|
|
|
terminated.store(true);
|
|
stopping.store(true);
|
|
}
|
|
#endif
|
|
|
|
void IOThread::setup() {
|
|
//redefined in subclasses
|
|
}
|
|
|
|
void IOThread::run() {
|
|
//redefined in subclasses
|
|
}
|
|
|
|
|
|
void IOThread::terminate() {
|
|
stopping.store(true);
|
|
}
|
|
|
|
void IOThread::onBindOutput(std::string /* name */, ThreadQueueBasePtr /* threadQueue */) {
|
|
|
|
}
|
|
|
|
void IOThread::onBindInput(std::string /* name */, ThreadQueueBasePtr /* threadQueue */) {
|
|
|
|
}
|
|
|
|
void IOThread::setInputQueue(const std::string& qname, const ThreadQueueBasePtr& threadQueue) {
|
|
std::lock_guard < std::mutex > lock(m_queue_bindings_mutex);
|
|
input_queues[qname] = threadQueue;
|
|
this->onBindInput(qname, threadQueue);
|
|
}
|
|
|
|
ThreadQueueBasePtr IOThread::getInputQueue(const std::string& qname) {
|
|
std::lock_guard < std::mutex > lock(m_queue_bindings_mutex);
|
|
return input_queues[qname];
|
|
}
|
|
|
|
void IOThread::setOutputQueue(const std::string& qname, const ThreadQueueBasePtr& threadQueue) {
|
|
std::lock_guard < std::mutex > lock(m_queue_bindings_mutex);
|
|
output_queues[qname] = threadQueue;
|
|
this->onBindOutput(qname, threadQueue);
|
|
}
|
|
|
|
ThreadQueueBasePtr IOThread::getOutputQueue(const std::string& qname) {
|
|
std::lock_guard < std::mutex > lock(m_queue_bindings_mutex);
|
|
return output_queues[qname];
|
|
}
|
|
|
|
bool IOThread::isTerminated(int waitMs) {
|
|
|
|
if (terminated.load()) {
|
|
return true;
|
|
}
|
|
else if (waitMs == 0) {
|
|
return false;
|
|
}
|
|
|
|
//this is a stupid busy plus sleep loop
|
|
int nbCyclesToWait;
|
|
|
|
if (waitMs < 0) {
|
|
nbCyclesToWait = std::numeric_limits<int>::max();
|
|
}
|
|
else {
|
|
nbCyclesToWait = (waitMs / SPIN_WAIT_SLEEP_MS) + 1;
|
|
}
|
|
|
|
for ( int i = 0; i < nbCyclesToWait; i++) {
|
|
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(SPIN_WAIT_SLEEP_MS));
|
|
|
|
if (terminated.load()) {
|
|
return true;
|
|
}
|
|
}
|
|
|
|
std::cout << "ERROR: thread '" << typeid(*this).name() << "' has not terminated in time ! (> " << waitMs << " ms)" << std::endl << std::flush;
|
|
|
|
return terminated.load();
|
|
}
|