ms: get rid of std::thread
2fc2b594da6e329577b195cb2543a8dd9e1b9ed0 changed std::thread to pthread
for proper affinity to circumvent startup issues, so just stick to
pthread instead of mixing std::thread and pthread, which made tracking
thread creation difficult due to different functions.
Change-Id: I0ba2fd958530394b9d99ed82111064d428c5870f
diff --git a/Transceiver52M/Makefile.am b/Transceiver52M/Makefile.am
index e82111e..dadfde9 100644
--- a/Transceiver52M/Makefile.am
+++ b/Transceiver52M/Makefile.am
@@ -88,6 +88,7 @@
MS_LOWER_SRC = \
ms/sch.c \
ms/ms.cpp \
+ ms/threadsched.cpp \
ms/ms_rx_lower.cpp \
grgsm_vitac/grgsm_vitac.cpp \
grgsm_vitac/viterbi_detector.cc
@@ -101,6 +102,7 @@
noinst_HEADERS += \
ms/ms.h \
+ ms/threadsched.h \
ms/bladerf_specific.h \
ms/uhd_specific.h \
ms/ms_upper.h \
diff --git a/Transceiver52M/ms/bladerf_specific.h b/Transceiver52M/ms/bladerf_specific.h
index 57aae75..f4abe57 100644
--- a/Transceiver52M/ms/bladerf_specific.h
+++ b/Transceiver52M/ms/bladerf_specific.h
@@ -428,10 +428,12 @@
auto get_rx_burst_handler_fn(bh_fn_t burst_handler)
{
- auto fn = [this] {
+ using thist = decltype(this);
+ auto fn = [](void *args) -> void * {
+ thist t = reinterpret_cast<thist>(args);
int status = 0;
if (!stop_lower_threads_flag)
- status = bladerf_stream(rx_stream, BLADERF_RX_X1);
+ status = bladerf_stream(t->rx_stream, BLADERF_RX_X1);
if (status < 0)
std::cerr << "rx stream error! " << bladerf_strerror(status) << std::endl;
@@ -441,10 +443,12 @@
}
auto get_tx_burst_handler_fn(bh_fn_t burst_handler)
{
- auto fn = [this] {
+ using thist = decltype(this);
+ auto fn = [](void *args) -> void * {
+ thist t = reinterpret_cast<thist>(args);
int status = 0;
if (!stop_lower_threads_flag)
- status = bladerf_stream(tx_stream, BLADERF_TX_X1);
+ status = bladerf_stream(t->tx_stream, BLADERF_TX_X1);
if (status < 0)
std::cerr << "rx stream error! " << bladerf_strerror(status) << std::endl;
diff --git a/Transceiver52M/ms/ms.cpp b/Transceiver52M/ms/ms.cpp
index b8710a6..4ae8668 100644
--- a/Transceiver52M/ms/ms.cpp
+++ b/Transceiver52M/ms/ms.cpp
@@ -35,6 +35,8 @@
#include "sch.h"
}
+#include "threadsched.h"
+
dummylog ms_trx::dummy_log;
#ifdef DBGXX
@@ -83,13 +85,11 @@
if (stop_lower_threads_flag)
return;
auto fn = get_rx_burst_handler_fn(rx_bh());
- lower_rx_task = std::thread(fn);
- set_name_aff_sched(lower_rx_task.native_handle(), sched_params::thread_names::RXRUN);
+ lower_rx_task = spawn_worker_thread(sched_params::thread_names::RXRUN, fn, this);
usleep(1000);
auto fn2 = get_tx_burst_handler_fn(tx_bh());
- lower_tx_task = std::thread(fn2);
- set_name_aff_sched(lower_tx_task.native_handle(), sched_params::thread_names::TXRUN);
+ lower_tx_task = spawn_worker_thread(sched_params::thread_names::TXRUN, fn2, this);
actually_enable_streams();
}
@@ -105,9 +105,9 @@
stop_lower_threads_flag = true;
close_device();
std::cerr << "dev closed..." << std::endl;
- lower_rx_task.join();
+ pthread_join(lower_rx_task, nullptr);
std::cerr << "L rx dead..." << std::endl;
- lower_tx_task.join();
+ pthread_join(lower_tx_task, nullptr);
std::cerr << "L tx dead..." << std::endl;
}
diff --git a/Transceiver52M/ms/ms.h b/Transceiver52M/ms/ms.h
index 18a6954..2ad78de 100644
--- a/Transceiver52M/ms/ms.h
+++ b/Transceiver52M/ms/ms.h
@@ -26,7 +26,7 @@
#include <cstdint>
#include <mutex>
#include <iostream>
-#include <thread>
+// #include <thread>
#if defined(BUILDBLADE)
#include "bladerf_specific.h"
@@ -42,6 +42,7 @@
#include "GSMCommon.h"
#include "itrq.h"
#include "threadpool.h"
+#include "threadsched.h"
const unsigned int ONE_TS_BURST_LEN = (3 + 58 + 26 + 58 + 3 + 8.25) * 4 /*sps*/;
const unsigned int SCH_LEN_SPS = (ONE_TS_BURST_LEN * 8 /*ts*/ * 12 /*frames*/);
@@ -216,43 +217,23 @@
}
};
-static struct sched_params {
- enum thread_names { U_CTL = 0, U_RX, U_TX, SCH_SEARCH, MAIN, LEAKCHECK, RXRUN, TXRUN, _THRD_NAME_COUNT };
- enum target { ODROID = 0, PI4 };
- const char *name;
- int core;
- int schedtype;
- int prio;
-} schdp[][sched_params::_THRD_NAME_COUNT]{
- {
- { "upper_ctrl", 2, SCHED_RR, sched_get_priority_max(SCHED_RR) },
- { "upper_rx", 2, SCHED_RR, sched_get_priority_max(SCHED_RR) - 5 },
- { "upper_tx", 2, SCHED_RR, sched_get_priority_max(SCHED_RR) - 1 },
-
- { "sch_search", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
- { "main", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
- { "leakcheck", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 10 },
-
- { "rxrun", 4, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 2 },
- { "txrun", 5, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 1 },
- },
- {
- { "upper_ctrl", 1, SCHED_RR, sched_get_priority_max(SCHED_RR) },
- { "upper_rx", 1, SCHED_RR, sched_get_priority_max(SCHED_RR) - 5 },
- { "upper_tx", 1, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 1 },
-
- { "sch_search", 1, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
- { "main", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
- { "leakcheck", 1, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 10 },
-
- { "rxrun", 2, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 2 },
- { "txrun", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 1 },
- },
-};
-
using ts_hitter_q_t = spsc_cond<64, GSM::Time, true, false>;
-struct ms_trx : public BASET {
+// used to globally initialize the sched/hw information
+struct sched_hw_info {
+ int hw_cpus;
+ sched_params::target hw_target;
+
+ sched_hw_info()
+ {
+ hw_cpus = std::thread::hardware_concurrency();
+ hw_target = hw_cpus > 4 ? sched_params::target::ODROID : sched_params::target::PI4;
+ set_sched_target(hw_target);
+ std::cerr << "scheduling for: " << (hw_cpus > 4 ? "odroid" : "pi4") << std::endl;
+ }
+};
+
+struct ms_trx : public BASET, public sched_hw_info {
using base = BASET;
static dummylog dummy_log;
unsigned int mTSC;
@@ -260,8 +241,8 @@
int timing_advance;
bool do_auto_gain;
- std::thread lower_rx_task;
- std::thread lower_tx_task;
+ pthread_t lower_rx_task;
+ pthread_t lower_tx_task;
// provides bursts to upper rx thread
rx_queue_t rxqueue;
@@ -277,9 +258,7 @@
int64_t first_sch_ts_start = -1;
time_keeper timekeeper;
- int hw_cpus;
- sched_params::target hw_target;
- single_thread_pool worker_thread;
+ single_thread_pool worker_thread; // uses base class sched target hw info
void start_lower_ms();
std::atomic<bool> upper_is_ready;
@@ -301,12 +280,8 @@
: mTSC(0), mBSIC(0), timing_advance(0), do_auto_gain(false), rxqueue(),
first_sch_buf(new blade_sample_type[SCH_LEN_SPS]),
burst_copy_buffer(new blade_sample_type[ONE_TS_BURST_LEN]), first_sch_buf_rcv_ts(0),
- rcv_done{ false }, sch_thread_done{ false }, hw_cpus(std::thread::hardware_concurrency()),
- hw_target(hw_cpus > 4 ? sched_params::target::ODROID : sched_params::target::PI4),
- upper_is_ready(false)
+ rcv_done{ false }, sch_thread_done{ false }, upper_is_ready(false)
{
- std::cerr << "scheduling for: " << (hw_cpus > 4 ? "odroid" : "pi4") << std::endl;
- set_name_aff_sched(worker_thread.get_handle(), sched_params::thread_names::SCH_SEARCH);
}
virtual ~ms_trx()
@@ -323,73 +298,4 @@
assert(val > -127 && val < 128);
timing_advance = val * 4;
}
-
- void set_name_aff_sched(sched_params::thread_names name)
- {
- set_name_aff_sched(pthread_self(), name);
- }
-
- void set_name_aff_sched(std::thread::native_handle_type h, sched_params::thread_names name)
- {
- auto tgt = schdp[hw_target][name];
- // std::cerr << "scheduling for: " << tgt.name << ":" << tgt.core << std::endl;
- set_name_aff_sched(h, tgt.name, tgt.core, tgt.schedtype, tgt.prio);
- }
-
- using pt_sig = void *(*)(void *);
-
- pthread_t spawn_worker_thread(sched_params::thread_names name, pt_sig fun, void *arg)
- {
- auto tgt = schdp[hw_target][name];
- // std::cerr << "scheduling for: " << tgt.name << ":" << tgt.core << " prio:" << tgt.prio << std::endl;
- return do_spawn_thr(tgt.name, tgt.core, tgt.schedtype, tgt.prio, fun, arg);
- }
-
- private:
- void set_name_aff_sched(std::thread::native_handle_type h, const char *name, int cpunum, int schedtype,
- int prio)
- {
- pthread_setname_np(h, name);
-
- cpu_set_t cpuset;
-
- CPU_ZERO(&cpuset);
- CPU_SET(cpunum, &cpuset);
-
- if (pthread_setaffinity_np(h, sizeof(cpuset), &cpuset) < 0) {
- std::cerr << name << " affinity: errreur! " << std::strerror(errno);
- return exit(0);
- }
-
- sched_param sch_params;
- sch_params.sched_priority = prio;
- if (pthread_setschedparam(h, schedtype, &sch_params) < 0) {
- std::cerr << name << " sched: errreur! " << std::strerror(errno);
- return exit(0);
- }
- }
-
- pthread_t do_spawn_thr(const char *name, int cpunum, int schedtype, int prio, pt_sig fun, void *arg)
- {
- pthread_t thread;
-
- pthread_attr_t attr;
- pthread_attr_init(&attr);
-
- sched_param sch_params;
- sch_params.sched_priority = prio;
- cpu_set_t cpuset;
- CPU_ZERO(&cpuset);
- CPU_SET(cpunum, &cpuset);
- auto a = pthread_attr_setaffinity_np(&attr, sizeof(cpu_set_t), &cpuset);
- a |= pthread_attr_setschedpolicy(&attr, schedtype);
- a |= pthread_attr_setschedparam(&attr, &sch_params);
- a |= pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);
- if(a)
- std::cerr << "thread arg rc:" << a << std::endl;
- pthread_create(&thread, &attr, fun, arg);
- pthread_setname_np(thread, name);
- pthread_attr_destroy(&attr);
- return thread;
- }
};
diff --git a/Transceiver52M/ms/ms_upper.cpp b/Transceiver52M/ms/ms_upper.cpp
index db15226..c5664cd 100644
--- a/Transceiver52M/ms/ms_upper.cpp
+++ b/Transceiver52M/ms/ms_upper.cpp
@@ -50,6 +50,7 @@
#include "ms_trxcon_if.h"
#include "ms_upper.h"
+#include "threadsched.h"
extern bool trxc_l1ctl_init(void *tallctx);
struct trxcon_inst *g_trxcon;
@@ -457,7 +458,7 @@
std::cerr << "Error initializing hardware, quitting.." << std::endl;
return -1;
}
- trx->set_name_aff_sched(sched_params::thread_names::MAIN);
+ set_name_aff_sched(sched_params::thread_names::MAIN);
if (!trxc_l1ctl_init(tall_trxcon_ctx)) {
std::cerr << "Error initializing l1ctl, quitting.." << std::endl;
diff --git a/Transceiver52M/ms/threadpool.h b/Transceiver52M/ms/threadpool.h
index 2180c68..4b1eefd 100644
--- a/Transceiver52M/ms/threadpool.h
+++ b/Transceiver52M/ms/threadpool.h
@@ -20,13 +20,12 @@
*
*/
-#include <functional>
-#include <thread>
#include <atomic>
#include <vector>
#include <future>
#include <mutex>
#include <queue>
+#include "threadsched.h"
struct single_thread_pool {
std::mutex m;
@@ -34,7 +33,7 @@
std::atomic<bool> stop_flag;
std::atomic<bool> is_ready;
std::deque<std::function<void()>> wq;
- std::thread worker_thread;
+ pthread_t worker_thread;
template <class F>
void add_task(F &&f)
@@ -45,19 +44,23 @@
return;
}
- single_thread_pool() : stop_flag(false), is_ready(false), worker_thread(std::thread([this] { thread_loop(); }))
+ single_thread_pool() : stop_flag(false), is_ready(false)
{
+ worker_thread = spawn_worker_thread(
+ sched_params::thread_names::SCH_SEARCH,
+ [](void *args) -> void * {
+ using thist = decltype(this);
+ thist t = reinterpret_cast<thist>(args);
+ t->thread_loop();
+ return 0;
+ },
+ this);
}
~single_thread_pool()
{
stop();
}
- std::thread::native_handle_type get_handle()
- {
- return worker_thread.native_handle();
- }
-
private:
void stop()
{
@@ -67,7 +70,7 @@
stop_flag = true;
cv.notify_one();
}
- worker_thread.join();
+ pthread_join(worker_thread, nullptr);
}
void thread_loop()
diff --git a/Transceiver52M/ms/threadsched.cpp b/Transceiver52M/ms/threadsched.cpp
new file mode 100644
index 0000000..ba5cae7
--- /dev/null
+++ b/Transceiver52M/ms/threadsched.cpp
@@ -0,0 +1,104 @@
+/*
+ * (C) 2023 by sysmocom s.f.m.c. GmbH <info@sysmocom.de>
+ * All Rights Reserved
+ *
+ * Author: Eric Wild <ewild@sysmocom.de>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include <cerrno>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+#include <thread>
+
+extern "C" {
+#include <pthread.h>
+}
+
+#include "threadsched.h"
+
+sched_params::target scheduling_target;
+
+void set_sched_target(sched_params::target t)
+{
+ scheduling_target = t;
+}
+
+void set_name_aff_sched(std::thread::native_handle_type h, const char *name, int cpunum, int schedtype, int prio)
+{
+ pthread_setname_np(h, name);
+
+ cpu_set_t cpuset;
+
+ CPU_ZERO(&cpuset);
+ CPU_SET(cpunum, &cpuset);
+
+ if (pthread_setaffinity_np(h, sizeof(cpuset), &cpuset) < 0) {
+ std::cerr << name << " affinity: errreur! " << std::strerror(errno);
+ return exit(0);
+ }
+
+ sched_param sch_params;
+ sch_params.sched_priority = prio;
+ if (pthread_setschedparam(h, schedtype, &sch_params) < 0) {
+ std::cerr << name << " sched: errreur! " << std::strerror(errno);
+ return exit(0);
+ }
+}
+
+static pthread_t do_spawn_thr(const char *name, int cpunum, int schedtype, int prio, worker_func_sig fun, void *arg)
+{
+ pthread_t thread;
+
+ pthread_attr_t attr;
+ pthread_attr_init(&attr);
+
+ sched_param sch_params;
+ sch_params.sched_priority = prio;
+ cpu_set_t cpuset;
+ CPU_ZERO(&cpuset);
+ CPU_SET(cpunum, &cpuset);
+ auto a = pthread_attr_setaffinity_np(&attr, sizeof(cpu_set_t), &cpuset);
+ a |= pthread_attr_setschedpolicy(&attr, schedtype);
+ a |= pthread_attr_setschedparam(&attr, &sch_params);
+ a |= pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);
+ if (a)
+ std::cerr << "thread arg rc:" << a << std::endl;
+ pthread_create(&thread, &attr, fun, arg);
+ pthread_setname_np(thread, name);
+ pthread_attr_destroy(&attr);
+ return thread;
+}
+
+void set_name_aff_sched(std::thread::native_handle_type h, sched_params::thread_names name)
+{
+ auto tgt = schdp[scheduling_target][name];
+ // std::cerr << "scheduling for: " << tgt.name << ":" << tgt.core << std::endl;
+ set_name_aff_sched(h, tgt.name, tgt.core, tgt.schedtype, tgt.prio);
+}
+
+void set_name_aff_sched(sched_params::thread_names name)
+{
+ set_name_aff_sched(pthread_self(), name);
+}
+
+pthread_t spawn_worker_thread(sched_params::thread_names name, worker_func_sig fun, void *arg)
+{
+ auto tgt = schdp[scheduling_target][name];
+ // std::cerr << "scheduling for: " << tgt.name << ":" << tgt.core << " prio:" << tgt.prio << std::endl;
+ return do_spawn_thr(tgt.name, tgt.core, tgt.schedtype, tgt.prio, fun, arg);
+}
diff --git a/Transceiver52M/ms/threadsched.h b/Transceiver52M/ms/threadsched.h
new file mode 100644
index 0000000..7cc9176
--- /dev/null
+++ b/Transceiver52M/ms/threadsched.h
@@ -0,0 +1,68 @@
+#pragma once
+/*
+ * (C) 2023 by sysmocom s.f.m.c. GmbH <info@sysmocom.de>
+ * All Rights Reserved
+ *
+ * Author: Eric Wild <ewild@sysmocom.de>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+extern "C" {
+#include <pthread.h>
+#include <sched.h>
+}
+
+static struct sched_params {
+ enum thread_names { U_CTL = 0, U_RX, U_TX, SCH_SEARCH, MAIN, LEAKCHECK, RXRUN, TXRUN, _THRD_NAME_COUNT };
+ enum target { ODROID = 0, PI4 };
+ const char *name;
+ int core;
+ int schedtype;
+ int prio;
+} schdp[][sched_params::_THRD_NAME_COUNT]{
+ {
+ { "upper_ctrl", 2, SCHED_RR, sched_get_priority_max(SCHED_RR) },
+ { "upper_rx", 2, SCHED_RR, sched_get_priority_max(SCHED_RR) - 5 },
+ { "upper_tx", 2, SCHED_RR, sched_get_priority_max(SCHED_RR) - 1 },
+
+ { "sch_search", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
+ { "main", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
+ { "leakcheck", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 10 },
+
+ { "rxrun", 4, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 2 },
+ { "txrun", 5, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 1 },
+ },
+ {
+ { "upper_ctrl", 1, SCHED_RR, sched_get_priority_max(SCHED_RR) },
+ { "upper_rx", 1, SCHED_RR, sched_get_priority_max(SCHED_RR) - 5 },
+ { "upper_tx", 1, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 1 },
+
+ { "sch_search", 1, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
+ { "main", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 5 },
+ { "leakcheck", 1, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 10 },
+
+ { "rxrun", 2, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 2 },
+ { "txrun", 3, SCHED_FIFO, sched_get_priority_max(SCHED_FIFO) - 1 },
+ },
+};
+
+void set_sched_target(sched_params::target t);
+
+using worker_func_sig = void *(*)(void *);
+
+void set_name_aff_sched(sched_params::thread_names name);
+
+pthread_t spawn_worker_thread(sched_params::thread_names name, worker_func_sig fun, void *arg);
diff --git a/Transceiver52M/ms/uhd_specific.h b/Transceiver52M/ms/uhd_specific.h
index 5723fd4..151c002 100644
--- a/Transceiver52M/ms/uhd_specific.h
+++ b/Transceiver52M/ms/uhd_specific.h
@@ -231,24 +231,29 @@
auto get_rx_burst_handler_fn(bh_fn_t burst_handler)
{
- auto fn = [this, burst_handler] {
+ // C cb -> ghetto closure capture, which is fine, the args never change.
+ static auto rx_burst_cap_this = this;
+ static auto rx_burst_cap_bh = burst_handler;
+ auto fn = [](void *args) -> void * {
pthread_setname_np(pthread_self(), "rxrun");
uhd::stream_cmd_t stream_cmd(uhd::stream_cmd_t::STREAM_MODE_START_CONTINUOUS);
stream_cmd.stream_now = true;
stream_cmd.time_spec = uhd::time_spec_t();
- rx_stream->issue_stream_cmd(stream_cmd);
+ rx_burst_cap_this->rx_stream->issue_stream_cmd(stream_cmd);
- while (!stop_lower_threads_flag) {
- rx_cb(burst_handler);
+ while (!rx_burst_cap_this->stop_lower_threads_flag) {
+ rx_burst_cap_this->rx_cb(rx_burst_cap_bh);
}
+ return 0;
};
return fn;
}
auto get_tx_burst_handler_fn(bh_fn_t burst_handler)
{
- auto fn = [] {
+ auto fn = [](void *args) -> void * {
// dummy
+ return 0;
};
return fn;
}