cellular-infrastructure
/
osmo-trx
mirror of https://gerrit.osmocom.org/osmo-trx
9
0
Fork 0
Code Issues Projects Releases Wiki Activity

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
This commit is contained in:
Eric Wild 2023-10-06 16:06:13 +02:00 committed by Eric Wild
parent 8aea236c56
commit 989fe75038
9 changed files with 234 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Transceiver52M/Makefile.am
View File

@ -88,6 +88,7 @@ TRXCON_LDADD = \
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 @@ MS_UPPER_SRC = \
noinst_HEADERS += \
ms/ms.h \
ms/threadsched.h \
ms/bladerf_specific.h \
ms/uhd_specific.h \
ms/ms_upper.h \

12
Transceiver52M/ms/bladerf_specific.h
View File

@ -428,10 +428,12 @@ struct blade_hw {
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 @@ struct blade_hw {
}
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;

12
Transceiver52M/ms/ms.cpp
View File

@ -35,6 +35,8 @@ extern "C" {
#include "sch.h"
}
#include "threadsched.h"
dummylog ms_trx::dummy_log;
#ifdef DBGXX
@ -83,13 +85,11 @@ void ms_trx::start_lower_ms()
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 @@ void ms_trx::stop_threads()
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;
}

136
Transceiver52M/ms/ms.h
View File

@ -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 @@ class time_keeper {
}
};
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 @@ struct ms_trx : public BASET {
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 @@ struct ms_trx : public BASET {
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 @@ struct ms_trx : public BASET {
: 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 @@ struct ms_trx : public BASET {
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;
}
};

3
Transceiver52M/ms/ms_upper.cpp
View File

@ -50,6 +50,7 @@ void __lsan_do_recoverable_leak_check();
#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 @@ int main(int argc, char *argv[])
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;

23
Transceiver52M/ms/threadpool.h
View File

@ -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 @@ struct single_thread_pool {
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 @@ struct single_thread_pool {
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 @@ struct single_thread_pool {
stop_flag = true;
cv.notify_one();
}
worker_thread.join();
pthread_join(worker_thread, nullptr);
}
void thread_loop()

104
Transceiver52M/ms/threadsched.cpp Normal file
View File

@ -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);
}

68
Transceiver52M/ms/threadsched.h Normal file
View File

@ -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);

15
Transceiver52M/ms/uhd_specific.h
View File

@ -231,24 +231,29 @@ struct uhd_hw {
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;
}