Team2890/allwpilib
4
0
Fork 0
mirror of https://github.com/wpilibsuite/allwpilib synced 2026-06-26 01:51:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add SafeThread to fix thread JNI shutdown races.

Browse Source 
During JVM shutdown, some JNI calls may not return, so it's not possible to
reliably perform a join() during static variable destruction (which occurs
as the JVM unloads the JNI module).

Also, due to static variable destruction, it's not safe to use any members
of a static class instance from a separate thread of execution.

SafeThread is a templated thread class and a related owner class that's
designed for safe operation and shutdown of threads in the presence of
callbacks that may not return.  It also passes ownership of variables from
the static instance to the thread, so the thread can safely operate until
it exits (the last operation of the thread being to destroy its instance).

Notifiers, RpcServer, and Logger now use SafeThread to ensure race-free
destruction in both C++ and Java.

All Java callback threads are now marked as Java daemon threads so they
don't keep the JVM running after main() terminates.

All Java callback threads are now named so their purpose is more easily
identified in a debugger.

Add SetRpcServerOnStart and SetRpcServerOnExit (similar to Listener).
This commit is contained in:
Peter Johnson
2015-12-28 08:28:24 -08:00
parent d8de5e4f19
commit fef8f933d9
13 changed files with 343 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
include/ntcore_c.h
View File

@@ -293,6 +293,9 @@ int NT_NotifierDestroyed();
* Remote Procedure Call Functions
*/
void NT_SetRpcServerOnStart(void (*on_start)(void *data), void *data);
void NT_SetRpcServerOnExit(void (*on_exit)(void *data), void *data);
typedef char *(*NT_RpcCallback)(void *data, const char *name, size_t name_len,
const char *params, size_t params_len,
size_t *results_len);

3
include/ntcore_cpp.h
View File

@@ -204,6 +204,9 @@ bool NotifierDestroyed();
* Remote Procedure Call Functions
*/
void SetRpcServerOnStart(std::function<void()> on_start);
void SetRpcServerOnExit(std::function<void()> on_exit);
typedef std::function<std::string(StringRef name, StringRef params)>
RpcCallback;

153
java/lib/NetworkTablesJNI.cpp
View File

@@ -10,6 +10,7 @@
#include "edu_wpi_first_wpilibj_networktables_NetworkTablesJNI.h"
#include "ntcore.h"
#include "atomic_static.h"
#include "SafeThread.h"
//
// Globals and load/unload
@@ -30,8 +31,12 @@ static JNIEnv *listenerEnv = nullptr;
static void ListenerOnStart() {
if (!jvm) return;
JNIEnv *env;
if (jvm->AttachCurrentThread(reinterpret_cast<void **>(&env),
nullptr) != JNI_OK)
JavaVMAttachArgs args;
args.version = JNI_VERSION_1_2;
args.name = const_cast<char*>("NTListener");
args.group = nullptr;
if (jvm->AttachCurrentThreadAsDaemon(reinterpret_cast<void **>(&env),
&args) != JNI_OK)
return;
if (!env || !env->functions) return;
listenerEnv = env;
@@ -1281,28 +1286,10 @@ JNIEXPORT jlong JNICALL Java_edu_wpi_first_wpilibj_networktables_NetworkTablesJN
// Instead, this class attaches just once. When a hardware notification
// occurs, a condition variable wakes up this thread and this thread actually
// makes the call into Java.
class LoggerThreadJNI {
class LoggerThreadJNI : public nt::SafeThread {
public:
static LoggerThreadJNI& GetInstance() {
ATOMIC_STATIC(LoggerThreadJNI, instance);
return instance;
}
LoggerThreadJNI();
~LoggerThreadJNI();
void SetFunc(JNIEnv* env, jobject func, jmethodID mid);
void Start();
void Stop();
void Main();
void Log(unsigned int level, const char* file, unsigned int line,
const char* msg);
private:
void ThreadMain();
std::thread m_thread;
std::mutex m_mutex;
std::condition_variable m_cond;
std::atomic_bool m_active;
struct LogMessage {
LogMessage(unsigned int level_, const char* file_, unsigned int line_,
const char* msg_)
@@ -1313,84 +1300,56 @@ class LoggerThreadJNI {
std::string msg;
};
std::queue<LogMessage> m_queue;
std::mutex m_shutdown_mutex;
std::condition_variable m_shutdown_cv;
bool m_shutdown = false;
jobject m_func = nullptr;
jmethodID m_mid;
ATOMIC_STATIC_DECL(LoggerThreadJNI)
};
ATOMIC_STATIC_INIT(LoggerThreadJNI)
class LoggerJNI : public nt::SafeThreadOwner<LoggerThreadJNI> {
public:
static LoggerJNI& GetInstance() {
ATOMIC_STATIC(LoggerJNI, instance);
return instance;
}
void SetFunc(JNIEnv* env, jobject func, jmethodID mid);
void Log(unsigned int level, const char* file, unsigned int line,
const char* msg);
LoggerThreadJNI::LoggerThreadJNI() {
m_active = false;
}
private:
ATOMIC_STATIC_DECL(LoggerJNI)
};
LoggerThreadJNI::~LoggerThreadJNI() {
Stop();
}
ATOMIC_STATIC_INIT(LoggerJNI)
void LoggerThreadJNI::SetFunc(JNIEnv* env, jobject func, jmethodID mid) {
std::lock_guard<std::mutex> lock(m_mutex);
void LoggerJNI::SetFunc(JNIEnv* env, jobject func, jmethodID mid) {
auto thr = GetThread();
if (!thr) return;
// free global reference
if (m_func) env->DeleteGlobalRef(m_func);
if (thr->m_func) env->DeleteGlobalRef(thr->m_func);
// create global reference
m_func = env->NewGlobalRef(func);
m_mid = mid;
thr->m_func = env->NewGlobalRef(func);
thr->m_mid = mid;
}
void LoggerThreadJNI::Start() {
{
std::lock_guard<std::mutex> lock(m_mutex);
if (m_active) return;
m_active = true;
}
{
std::lock_guard<std::mutex> lock(m_shutdown_mutex);
m_shutdown = false;
}
m_thread = std::thread(&LoggerThreadJNI::ThreadMain, this);
void LoggerJNI::Log(unsigned int level, const char *file, unsigned int line,
const char *msg) {
auto thr = GetThread();
if (!thr) return;
thr->m_queue.emplace(level, file, line, msg);
thr->m_cond.notify_one();
}
void LoggerThreadJNI::Stop() {
{
std::lock_guard<std::mutex> lock(m_mutex);
if (!m_active) return;
m_active = false;
}
m_cond.notify_one(); // wake up thread
// join threads, with timeout
if (m_thread.joinable()) {
std::unique_lock<std::mutex> lock(m_shutdown_mutex);
auto timeout_time =
std::chrono::steady_clock::now() + std::chrono::seconds(1);
if (m_shutdown_cv.wait_until(lock, timeout_time,
[&] { return m_shutdown; }))
m_thread.join();
else
m_thread.detach(); // timed out, detach it
}
}
void LoggerThreadJNI::Log(unsigned int level, const char *file,
unsigned int line, const char *msg) {
std::lock_guard<std::mutex> lock(m_mutex);
if (!m_active) return;
m_queue.emplace(level, file, line, msg);
m_cond.notify_one();
}
void LoggerThreadJNI::ThreadMain() {
void LoggerThreadJNI::Main() {
JNIEnv *env;
jint rs = jvm->AttachCurrentThread((void**)&env, NULL);
JavaVMAttachArgs args;
args.version = JNI_VERSION_1_2;
args.name = const_cast<char*>("NTLogger");
args.group = nullptr;
jint rs = jvm->AttachCurrentThreadAsDaemon((void**)&env, &args);
if (rs != JNI_OK) return;
std::unique_lock<std::mutex> lock(m_mutex);
while (m_active) {
m_cond.wait(lock, [&] { return !m_active || !m_queue.empty(); });
m_cond.wait(lock, [&] { return !(m_active && m_queue.empty()); });
if (!m_active) break;
while (!m_queue.empty()) {
if (!m_active) break;
@@ -1399,25 +1358,21 @@ void LoggerThreadJNI::ThreadMain() {
auto func = m_func;
auto mid = m_mid;
lock.unlock(); // don't hold mutex during callback execution
env->CallVoidMethod(func, mid, (jint)item.level,
ToJavaString(env, item.file), (jint)item.line,
ToJavaString(env, item.msg));
if (env->ExceptionCheck()) {
env->ExceptionDescribe();
env->ExceptionClear();
{
JavaLocal<jstring> file(env, ToJavaString(env, item.file));
JavaLocal<jstring> msg(env, ToJavaString(env, item.msg));
env->CallVoidMethod(func, mid, (jint)item.level, file.obj(),
(jint)item.line, msg.obj());
if (env->ExceptionCheck()) {
env->ExceptionDescribe();
env->ExceptionClear();
}
}
lock.lock();
}
}
if (jvm) jvm->DetachCurrentThread();
// use condition variable to signal thread shutdown
{
std::lock_guard<std::mutex> lock(m_shutdown_mutex);
m_shutdown = true;
m_shutdown_cv.notify_one();
}
}
extern "C" {
@@ -1439,14 +1394,14 @@ JNIEXPORT void JNICALL Java_edu_wpi_first_wpilibj_networktables_NetworkTablesJNI
cls, "apply", "(ILjava/lang/String;ILjava/lang/String;)V");
if (!mid) return;
auto& thread = LoggerThreadJNI::GetInstance();
thread.SetFunc(env, func, mid);
thread.Start();
auto& logger = LoggerJNI::GetInstance();
logger.Start();
logger.SetFunc(env, func, mid);
nt::SetLogger(
[](unsigned int level, const char *file, unsigned int line,
const char *msg) {
LoggerThreadJNI::GetInstance().Log(level, file, line, msg);
LoggerJNI::GetInstance().Log(level, file, line, msg);
},
minLevel);
}

4
ntcore.def
View File

@@ -76,6 +76,10 @@ NT_FreeCharArray @78
NT_NotifierDestroyed @79
NT_StopRpcServer @80
NT_StopNotifier @81
NT_SetListenerOnStart @82
NT_SetListenerOnExit @83
NT_SetRpcServerOnStart @84
NT_SetRpcServerOnExit @85
; JNI functions
JNI_OnLoad

149
src/Notifier.cpp
View File

@@ -7,53 +7,75 @@
#include "Notifier.h"
#include <queue>
#include <vector>
using namespace nt;
ATOMIC_STATIC_INIT(Notifier)
bool Notifier::s_destroyed = false;
class Notifier::Thread : public SafeThread {
public:
Thread(std::function<void()> on_start, std::function<void()> on_exit)
: m_on_start(on_start), m_on_exit(on_exit) {}
void Main();
struct EntryListener {
EntryListener(StringRef prefix_, EntryListenerCallback callback_,
unsigned int flags_)
: prefix(prefix_), callback(callback_), flags(flags_) {}
std::string prefix;
EntryListenerCallback callback;
unsigned int flags;
};
std::vector<EntryListener> m_entry_listeners;
std::vector<ConnectionListenerCallback> m_conn_listeners;
struct EntryNotification {
EntryNotification(StringRef name_, std::shared_ptr<Value> value_,
unsigned int flags_, EntryListenerCallback only_)
: name(name_),
value(value_),
flags(flags_),
only(only_) {}
std::string name;
std::shared_ptr<Value> value;
unsigned int flags;
EntryListenerCallback only;
};
std::queue<EntryNotification> m_entry_notifications;
struct ConnectionNotification {
ConnectionNotification(bool connected_, const ConnectionInfo& conn_info_,
ConnectionListenerCallback only_)
: connected(connected_), conn_info(conn_info_), only(only_) {}
bool connected;
ConnectionInfo conn_info;
ConnectionListenerCallback only;
};
std::queue<ConnectionNotification> m_conn_notifications;
std::function<void()> m_on_start;
std::function<void()> m_on_exit;
};
Notifier::Notifier() {
m_active = false;
m_local_notifiers = false;
s_destroyed = false;
}
Notifier::~Notifier() {
s_destroyed = true;
Stop();
}
Notifier::~Notifier() { s_destroyed = true; }
void Notifier::Start() {
{
std::lock_guard<std::mutex> lock(m_mutex);
if (m_active) return;
m_active = true;
}
{
std::lock_guard<std::mutex> lock(m_shutdown_mutex);
m_shutdown = false;
}
m_thread = std::thread(&Notifier::ThreadMain, this);
}
void Notifier::Start() { m_owner.Start(new Thread(m_on_start, m_on_exit)); }
void Notifier::Stop() {
m_active = false;
// send notification so the thread terminates
m_cond.notify_one();
if (m_thread.joinable()) {
// join with timeout
std::unique_lock<std::mutex> lock(m_shutdown_mutex);
auto timeout_time =
std::chrono::steady_clock::now() + std::chrono::seconds(1);
if (m_shutdown_cv.wait_until(lock, timeout_time,
[&] { return m_shutdown; }))
m_thread.join();
else
m_thread.detach(); // timed out, detach it
}
}
void Notifier::Stop() { m_owner.Stop(); }
void Notifier::ThreadMain() {
void Notifier::Thread::Main() {
if (m_on_start) m_on_start();
std::unique_lock<std::mutex> lock(m_mutex);
@@ -138,65 +160,66 @@ void Notifier::ThreadMain() {
done:
if (m_on_exit) m_on_exit();
// use condition variable to signal thread shutdown
{
std::lock_guard<std::mutex> lock(m_shutdown_mutex);
m_shutdown = true;
m_shutdown_cv.notify_one();
}
}
unsigned int Notifier::AddEntryListener(StringRef prefix,
EntryListenerCallback callback,
unsigned int flags) {
std::lock_guard<std::mutex> lock(m_mutex);
unsigned int uid = m_entry_listeners.size();
m_entry_listeners.emplace_back(prefix, callback, flags);
auto thr = m_owner.GetThread();
if (!thr) {
Start();
thr = m_owner.GetThread();
}
unsigned int uid = thr->m_entry_listeners.size();
thr->m_entry_listeners.emplace_back(prefix, callback, flags);
if ((flags & NT_NOTIFY_LOCAL) != 0) m_local_notifiers = true;
return uid + 1;
}
void Notifier::RemoveEntryListener(unsigned int entry_listener_uid) {
auto thr = m_owner.GetThread();
if (!thr) return;
--entry_listener_uid;
std::lock_guard<std::mutex> lock(m_mutex);
if (entry_listener_uid < m_entry_listeners.size())
m_entry_listeners[entry_listener_uid].callback = nullptr;
if (entry_listener_uid < thr->m_entry_listeners.size())
thr->m_entry_listeners[entry_listener_uid].callback = nullptr;
}
void Notifier::NotifyEntry(StringRef name, std::shared_ptr<Value> value,
unsigned int flags, EntryListenerCallback only) {
if (!m_active) return;
// optimization: don't generate needless local queue entries if we have
// no local listeners (as this is a common case on the server side)
if ((flags & NT_NOTIFY_LOCAL) != 0 && !m_local_notifiers) return;
std::unique_lock<std::mutex> lock(m_mutex);
m_entry_notifications.emplace(name, value, flags, only);
lock.unlock();
m_cond.notify_one();
auto thr = m_owner.GetThread();
if (!thr) return;
thr->m_entry_notifications.emplace(name, value, flags, only);
thr->m_cond.notify_one();
}
unsigned int Notifier::AddConnectionListener(
ConnectionListenerCallback callback) {
std::lock_guard<std::mutex> lock(m_mutex);
unsigned int uid = m_entry_listeners.size();
m_conn_listeners.emplace_back(callback);
auto thr = m_owner.GetThread();
if (!thr) {
Start();
thr = m_owner.GetThread();
}
unsigned int uid = thr->m_entry_listeners.size();
thr->m_conn_listeners.emplace_back(callback);
return uid + 1;
}
void Notifier::RemoveConnectionListener(unsigned int conn_listener_uid) {
auto thr = m_owner.GetThread();
if (!thr) return;
--conn_listener_uid;
std::lock_guard<std::mutex> lock(m_mutex);
if (conn_listener_uid < m_conn_listeners.size())
m_conn_listeners[conn_listener_uid] = nullptr;
if (conn_listener_uid < thr->m_conn_listeners.size())
thr->m_conn_listeners[conn_listener_uid] = nullptr;
}
void Notifier::NotifyConnection(bool connected,
const ConnectionInfo& conn_info,
ConnectionListenerCallback only) {
if (!m_active) return;
std::unique_lock<std::mutex> lock(m_mutex);
m_conn_notifications.emplace(connected, conn_info, only);
lock.unlock();
m_cond.notify_one();
auto thr = m_owner.GetThread();
if (!thr) return;
thr->m_conn_notifications.emplace(connected, conn_info, only);
thr->m_cond.notify_one();
}

60
src/Notifier.h
View File

@@ -8,16 +8,11 @@
#ifndef NT_NOTIFIER_H_
#define NT_NOTIFIER_H_
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <thread>
#include <utility>
#include <vector>
#include <functional>
#include "atomic_static.h"
#include "ntcore_cpp.h"
#include "SafeThread.h"
namespace nt {
@@ -33,7 +28,6 @@ class Notifier {
void Start();
void Stop();
bool active() const { return m_active; }
bool local_notifiers() const { return m_local_notifiers; }
static bool destroyed() { return s_destroyed; }
@@ -57,57 +51,11 @@ class Notifier {
private:
Notifier();
void ThreadMain();
class Thread;
SafeThreadOwner<Thread> m_owner;
std::atomic_bool m_active;
std::atomic_bool m_local_notifiers;
std::mutex m_mutex;
std::condition_variable m_cond;
struct EntryListener {
EntryListener(StringRef prefix_, EntryListenerCallback callback_,
unsigned int flags_)
: prefix(prefix_), callback(callback_), flags(flags_) {}
std::string prefix;
EntryListenerCallback callback;
unsigned int flags;
};
std::vector<EntryListener> m_entry_listeners;
std::vector<ConnectionListenerCallback> m_conn_listeners;
struct EntryNotification {
EntryNotification(StringRef name_, std::shared_ptr<Value> value_,
unsigned int flags_, EntryListenerCallback only_)
: name(name_),
value(value_),
flags(flags_),
only(only_) {}
std::string name;
std::shared_ptr<Value> value;
unsigned int flags;
EntryListenerCallback only;
};
std::queue<EntryNotification> m_entry_notifications;
struct ConnectionNotification {
ConnectionNotification(bool connected_, const ConnectionInfo& conn_info_,
ConnectionListenerCallback only_)
: connected(connected_), conn_info(conn_info_), only(only_) {}
bool connected;
ConnectionInfo conn_info;
ConnectionListenerCallback only;
};
std::queue<ConnectionNotification> m_conn_notifications;
std::thread m_thread;
std::mutex m_shutdown_mutex;
std::condition_variable m_shutdown_cv;
bool m_shutdown = false;
std::function<void()> m_on_start;
std::function<void()> m_on_exit;

78
src/RpcServer.cpp
View File

@@ -7,70 +7,57 @@
#include "RpcServer.h"
#include <queue>
#include "Log.h"
using namespace nt;
ATOMIC_STATIC_INIT(RpcServer)
class RpcServer::Thread : public SafeThread {
public:
Thread(std::function<void()> on_start, std::function<void()> on_exit)
: m_on_start(on_start), m_on_exit(on_exit) {}
void Main();
std::queue<RpcCall> m_call_queue;
std::function<void()> m_on_start;
std::function<void()> m_on_exit;
};
RpcServer::RpcServer() {
m_active = false;
m_terminating = false;
}
RpcServer::~RpcServer() {
Logger::GetInstance().SetLogger(nullptr);
Stop();
m_terminating = true;
m_poll_cond.notify_all();
}
void RpcServer::Start() {
{
std::lock_guard<std::mutex> lock(m_mutex);
if (m_active) return;
m_active = true;
}
{
std::lock_guard<std::mutex> lock(m_shutdown_mutex);
m_shutdown = false;
}
m_thread = std::thread(&RpcServer::ThreadMain, this);
auto thr = m_owner.GetThread();
if (!thr) m_owner.Start(new Thread(m_on_start, m_on_exit));
}
void RpcServer::Stop() {
m_active = false;
if (m_thread.joinable()) {
// send notification so the thread terminates
m_call_cond.notify_one();
// join with timeout
std::unique_lock<std::mutex> lock(m_shutdown_mutex);
auto timeout_time =
std::chrono::steady_clock::now() + std::chrono::seconds(1);
if (m_shutdown_cv.wait_until(lock, timeout_time,
[&] { return m_shutdown; }))
m_thread.join();
else
m_thread.detach(); // timed out, detach it
}
}
void RpcServer::Stop() { m_owner.Stop(); }
void RpcServer::ProcessRpc(StringRef name, std::shared_ptr<Message> msg,
RpcCallback func, unsigned int conn_id,
SendMsgFunc send_response) {
std::unique_lock<std::mutex> lock(m_mutex);
if (func)
m_call_queue.emplace(name, msg, func, conn_id, send_response);
else
if (func) {
auto thr = m_owner.GetThread();
if (!thr) return;
thr->m_call_queue.emplace(name, msg, func, conn_id, send_response);
thr->m_cond.notify_one();
} else {
std::lock_guard<std::mutex> lock(m_mutex);
m_poll_queue.emplace(name, msg, func, conn_id, send_response);
lock.unlock();
if (func)
m_call_cond.notify_one();
else
m_poll_cond.notify_one();
}
}
bool RpcServer::PollRpc(bool blocking, RpcCallInfo* call_info) {
@@ -103,12 +90,14 @@ void RpcServer::PostRpcResponse(unsigned int rpc_id, unsigned int call_uid,
m_response_map.erase(i);
}
void RpcServer::ThreadMain() {
void RpcServer::Thread::Main() {
if (m_on_start) m_on_start();
std::unique_lock<std::mutex> lock(m_mutex);
std::string tmp;
while (m_active) {
while (m_call_queue.empty()) {
m_call_cond.wait(lock);
m_cond.wait(lock);
if (!m_active) goto done;
}
@@ -132,10 +121,5 @@ void RpcServer::ThreadMain() {
}
done:
// use condition variable to signal thread shutdown
{
std::lock_guard<std::mutex> lock(m_shutdown_mutex);
m_shutdown = true;
m_shutdown_cv.notify_one();
}
if (m_on_exit) m_on_exit();
}

29
src/RpcServer.h
View File

@@ -12,14 +12,13 @@
#include <condition_variable>
#include <mutex>
#include <queue>
#include <thread>
#include <utility>
#include <vector>
#include "llvm/DenseMap.h"
#include "atomic_static.h"
#include "Message.h"
#include "ntcore_cpp.h"
#include "SafeThread.h"
namespace nt {
@@ -37,7 +36,8 @@ class RpcServer {
void Start();
void Stop();
bool active() const { return m_active; }
void SetOnStart(std::function<void()> on_start) { m_on_start = on_start; }
void SetOnExit(std::function<void()> on_exit) { m_on_exit = on_exit; }
void ProcessRpc(StringRef name, std::shared_ptr<Message> msg,
RpcCallback func, unsigned int conn_id,
@@ -50,13 +50,8 @@ class RpcServer {
private:
RpcServer();
void ThreadMain();
std::atomic_bool m_active;
std::atomic_bool m_terminating;
std::mutex m_mutex;
std::condition_variable m_call_cond, m_poll_cond;
class Thread;
SafeThreadOwner<Thread> m_owner;
struct RpcCall {
RpcCall(StringRef name_, std::shared_ptr<Message> msg_, RpcCallback func_,
@@ -73,15 +68,19 @@ class RpcServer {
unsigned int conn_id;
SendMsgFunc send_response;
};
std::queue<RpcCall> m_call_queue, m_poll_queue;
std::mutex m_mutex;
std::queue<RpcCall> m_poll_queue;
llvm::DenseMap<std::pair<unsigned int, unsigned int>, SendMsgFunc>
m_response_map;
std::thread m_thread;
std::mutex m_shutdown_mutex;
std::condition_variable m_shutdown_cv;
bool m_shutdown = false;
std::condition_variable m_poll_cond;
std::atomic_bool m_terminating;
std::function<void()> m_on_start;
std::function<void()> m_on_exit;
ATOMIC_STATIC_DECL(RpcServer)
};

31
src/SafeThread.cpp Normal file
View File

@@ -0,0 +1,31 @@
/*----------------------------------------------------------------------------*/
/* Copyright (c) FIRST 2015. All Rights Reserved. */
/* Open Source Software - may be modified and shared by FRC teams. The code */
/* must be accompanied by the FIRST BSD license file in the root directory of */
/* the project. */
/*----------------------------------------------------------------------------*/
#include "SafeThread.h"
using namespace nt;
void detail::SafeThreadOwnerBase::Start(SafeThread* thr) {
SafeThread* curthr = nullptr;
SafeThread* newthr = thr;
if (!m_thread.compare_exchange_strong(curthr, newthr)) {
delete newthr;
return;
}
std::thread([=]() {
newthr->Main();
delete newthr;
}).detach();
}
void detail::SafeThreadOwnerBase::Stop() {
SafeThread* thr = m_thread.exchange(nullptr);
if (!thr) return;
std::lock_guard<std::mutex> lock(thr->m_mutex);
thr->m_active = false;
thr->m_cond.notify_one();
}

93
src/SafeThread.h Normal file
View File

@@ -0,0 +1,93 @@
/*----------------------------------------------------------------------------*/
/* Copyright (c) FIRST 2015. All Rights Reserved. */
/* Open Source Software - may be modified and shared by FRC teams. The code */
/* must be accompanied by the FIRST BSD license file in the root directory of */
/* the project. */
/*----------------------------------------------------------------------------*/
#ifndef NT_SAFETHREAD_H_
#define NT_SAFETHREAD_H_
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <thread>
namespace nt {
// Base class for SafeThreadOwner threads.
class SafeThread {
public:
virtual ~SafeThread() = default;
virtual void Main() = 0;
std::mutex m_mutex;
bool m_active = true;
std::condition_variable m_cond;
};
namespace detail {
// Non-template proxy base class for common proxy code.
class SafeThreadProxyBase {
public:
SafeThreadProxyBase(SafeThread* thr) : m_thread(thr) {
if (!m_thread) return;
std::unique_lock<std::mutex>(m_thread->m_mutex).swap(m_lock);
if (!m_thread->m_active) {
m_lock.unlock();
m_thread = nullptr;
return;
}
}
explicit operator bool() const { return m_thread != nullptr; }
std::unique_lock<std::mutex>& GetLock() { return m_lock; }
protected:
SafeThread* m_thread;
std::unique_lock<std::mutex> m_lock;
};
// A proxy for SafeThread.
// Also serves as a scoped lock on SafeThread::m_mutex.
template <typename T>
class SafeThreadProxy : public SafeThreadProxyBase {
public:
SafeThreadProxy(SafeThread* thr) : SafeThreadProxyBase(thr) {}
T& operator*() const { return *static_cast<T*>(m_thread); }
T* operator->() const { return static_cast<T*>(m_thread); }
};
// Non-template owner base class for common owner code.
class SafeThreadOwnerBase {
public:
void Stop();
protected:
SafeThreadOwnerBase() { m_thread = nullptr; }
SafeThreadOwnerBase(const SafeThreadOwnerBase&) = delete;
SafeThreadOwnerBase& operator=(const SafeThreadOwnerBase&) = delete;
~SafeThreadOwnerBase() { Stop(); }
void Start(SafeThread* thr);
SafeThread* GetThread() { return m_thread.load(); }
private:
std::atomic<SafeThread*> m_thread;
};
} // namespace detail
template <typename T>
class SafeThreadOwner : public detail::SafeThreadOwnerBase {
public:
void Start() { Start(new T); }
void Start(T* thr) { detail::SafeThreadOwnerBase::Start(thr); }
using Proxy = typename detail::SafeThreadProxy<T>;
Proxy GetThread() { return Proxy(detail::SafeThreadOwnerBase::GetThread()); }
};
} // namespace nt
#endif // NT_SAFETHREAD_H_

2
src/Storage.cpp
View File

@@ -1262,7 +1262,7 @@ void Storage::CreateRpc(StringRef name, StringRef def, RpcCallback callback) {
entry->rpc_callback = callback;
// start the RPC server
if (!m_rpc_server.active()) m_rpc_server.Start();
m_rpc_server.Start();
if (old_value && *old_value == *value) return;

8
src/ntcore_c.cpp
View File

@@ -215,6 +215,14 @@ int NT_NotifierDestroyed() { return nt::NotifierDestroyed(); }
* Remote Procedure Call Functions
*/
void NT_SetRpcServerOnStart(void (*on_start)(void *data), void *data) {
nt::SetRpcServerOnStart([=]() { on_start(data); });
}
void NT_SetRpcServerOnExit(void (*on_exit)(void *data), void *data) {
nt::SetRpcServerOnExit([=]() { on_exit(data); });
}
void NT_CreateRpc(const char *name, size_t name_len, const char *def,
size_t def_len, void *data, NT_RpcCallback callback) {
nt::CreateRpc(

17
src/ntcore_cpp.cpp
View File

@@ -75,9 +75,8 @@ void SetListenerOnExit(std::function<void()> on_exit) {
unsigned int AddEntryListener(StringRef prefix, EntryListenerCallback callback,
unsigned int flags) {
Notifier& notifier = Notifier::GetInstance();
unsigned int uid = notifier.AddEntryListener(prefix, callback, flags);
notifier.Start();
unsigned int uid =
Notifier::GetInstance().AddEntryListener(prefix, callback, flags);
if ((flags & NT_NOTIFY_IMMEDIATE) != 0)
Storage::GetInstance().NotifyEntries(prefix, callback);
return uid;
@@ -89,9 +88,7 @@ void RemoveEntryListener(unsigned int entry_listener_uid) {
unsigned int AddConnectionListener(ConnectionListenerCallback callback,
bool immediate_notify) {
Notifier& notifier = Notifier::GetInstance();
unsigned int uid = notifier.AddConnectionListener(callback);
Notifier::GetInstance().Start();
unsigned int uid = Notifier::GetInstance().AddConnectionListener(callback);
if (immediate_notify) Dispatcher::GetInstance().NotifyConnections(callback);
return uid;
}
@@ -106,6 +103,14 @@ bool NotifierDestroyed() { return Notifier::destroyed(); }
* Remote Procedure Call Functions
*/
void SetRpcServerOnStart(std::function<void()> on_start) {
RpcServer::GetInstance().SetOnStart(on_start);
}
void SetRpcServerOnExit(std::function<void()> on_exit) {
RpcServer::GetInstance().SetOnExit(on_exit);
}
void CreateRpc(StringRef name, StringRef def, RpcCallback callback) {
Storage::GetInstance().CreateRpc(name, def, callback);
}