start porting frame streams
Some checks are pending
ContinuousDelivery / linux-ubuntu (push) Waiting to run
ContinuousDelivery / android (map[ndk_abi:arm64-v8a vcpkg_toolkit:arm64-android]) (push) Waiting to run
ContinuousDelivery / android (map[ndk_abi:x86_64 vcpkg_toolkit:x64-android]) (push) Waiting to run
ContinuousDelivery / windows (push) Waiting to run
ContinuousDelivery / windows-asan (push) Waiting to run
ContinuousDelivery / release (push) Blocked by required conditions
ContinuousIntegration / linux (push) Waiting to run
ContinuousIntegration / android (map[ndk_abi:arm64-v8a vcpkg_toolkit:arm64-android]) (push) Waiting to run
ContinuousIntegration / android (map[ndk_abi:x86_64 vcpkg_toolkit:x64-android]) (push) Waiting to run
ContinuousIntegration / macos (push) Waiting to run
ContinuousIntegration / windows (push) Waiting to run

This commit is contained in:
Green Sky 2024-09-27 13:26:18 +02:00
parent 3bdf262068
commit b899b8131e
No known key found for this signature in database
8 changed files with 446 additions and 1 deletions

View File

@ -102,6 +102,11 @@ target_sources(tomato PUBLIC
./chat_gui4.hpp ./chat_gui4.hpp
./chat_gui4.cpp ./chat_gui4.cpp
./frame_streams/frame_stream2.hpp
./frame_streams/audio_stream2.hpp
./frame_streams/stream_manager.hpp
./frame_streams/stream_manager.cpp
) )
if (TOMATO_TOX_AV) if (TOMATO_TOX_AV)

View File

@ -0,0 +1,39 @@
#pragma once
#include "./frame_stream2.hpp"
#include <solanaceae/util/span.hpp>
#include <cstdint>
#include <variant>
#include <vector>
// raw audio
// channels make samples interleaved,
// planar channels are not supported
// s16 only stopgap audio frame (simplified)
struct AudioFrame2 {
// samples per second
uint32_t sample_rate {48'000};
// only >0 is valid
size_t channels {0};
std::variant<
std::vector<int16_t>, // S16, platform endianess
Span<int16_t> // non owning variant, for direct consumption
> buffer;
// helpers
Span<int16_t> getSpan(void) const {
if (std::holds_alternative<std::vector<int16_t>>(buffer)) {
return Span<int16_t>{std::get<std::vector<int16_t>>(buffer)};
} else {
return std::get<Span<int16_t>>(buffer);
}
return {};
}
};
using AudioFrame2Stream2I = FrameStream2I<AudioFrame2>;

View File

@ -0,0 +1,47 @@
#pragma once
#include <cstdint>
#include <memory>
#include <optional>
#include <vector>
// Frames often consist of:
// - seq id // incremental sequential id, gaps in ids can be used to detect loss
// - or timestamp
// - data // the frame data
// eg:
//struct ExampleFrame {
//int64_t seq_id {0};
//std::vector<uint8_t> data;
//};
template<typename FrameType>
struct FrameStream2I {
virtual ~FrameStream2I(void) {}
// get number of available frames
// returns -1 if unknown
[[nodiscard]] virtual int32_t size(void) = 0;
// get next frame
// data sharing? -> no, data is copied for each fsr, if concurency supported
[[nodiscard]] virtual std::optional<FrameType> pop(void) = 0;
// returns true if there are readers (or we dont know)
virtual bool push(const FrameType& value) = 0;
};
template<typename FrameType>
struct FrameStream2SourceI {
virtual ~FrameStream2SourceI(void) {}
[[nodiscard]] virtual std::shared_ptr<FrameStream2I<FrameType>> subscribe(void) = 0;
virtual bool unsubscribe(const std::shared_ptr<FrameStream2I<FrameType>>& sub) = 0;
};
template<typename FrameType>
struct FrameStream2SinkI {
virtual ~FrameStream2SinkI(void) {}
[[nodiscard]] virtual std::shared_ptr<FrameStream2I<FrameType>> subscribe(void) = 0;
virtual bool unsubscribe(const std::shared_ptr<FrameStream2I<FrameType>>& sub) = 0;
};

View File

@ -0,0 +1,139 @@
#include "./stream_manager.hpp"
StreamManager::Connection::Connection(
ObjectHandle src_,
ObjectHandle sink_,
std::unique_ptr<Data>&& data_,
std::function<void(Connection&)>&& pump_fn_,
std::function<void(Connection&)>&& unsubscribe_fn_,
bool on_main_thread_
) :
src(src_),
sink(sink_),
data(std::move(data_)),
pump_fn(std::move(pump_fn_)),
unsubscribe_fn(std::move(unsubscribe_fn_)),
on_main_thread(on_main_thread_)
{
if (!on_main_thread) {
// start thread
pump_thread = std::thread([this](void) {
while (!stop) {
pump_fn(*this);
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
finished = true;
});
}
}
StreamManager::StreamManager(ObjectStore2& os) : _os(os) {}
StreamManager::~StreamManager(void) {
// stop all connetions
for (const auto& con : _connections) {
con->stop = true;
if (!con->on_main_thread) {
con->pump_thread.join(); // we skip the finished check and wait
}
con->unsubscribe_fn(*con);
}
}
bool StreamManager::connect(Object src, Object sink, bool threaded) {
auto h_src = _os.objectHandle(src);
auto h_sink = _os.objectHandle(sink);
if (!static_cast<bool>(h_src) || !static_cast<bool>(h_sink)) {
// an object does not exist
return false;
}
// get src and sink comps
if (!h_src.all_of<Components::StreamSource>()) {
// src not stream source
return false;
}
if (!h_sink.all_of<Components::StreamSink>()) {
// sink not stream sink
return false;
}
const auto& ssrc = h_src.get<Components::StreamSource>();
const auto& ssink = h_sink.get<Components::StreamSink>();
// compare type
if (ssrc.frame_type_name != ssink.frame_type_name) {
return false;
}
// always fail in debug mode
assert(static_cast<bool>(ssrc.connect_fn));
if (!static_cast<bool>(ssrc.connect_fn)) {
return false;
}
// use connect fn from src
return ssrc.connect_fn(*this, src, sink, threaded);
}
bool StreamManager::disconnect(Object src, Object sink) {
auto res = std::find_if(
_connections.cbegin(), _connections.cend(),
[&](const auto& a) { return a->src == src && a->sink == sink; }
);
if (res == _connections.cend()) {
// not found
return false;
}
// do disconnect
(*res)->stop = true;
return true;
}
bool StreamManager::disconnectAll(Object o) {
bool succ {false};
for (const auto& con : _connections) {
if (con->src == o || con->sink == o) {
con->stop = true;
succ = true;
}
}
return succ;
}
// do we need the time delta?
float StreamManager::tick(float) {
// pump all mainthread connections
for (auto it = _connections.begin(); it != _connections.end();) {
auto& con = **it;
if (!static_cast<bool>(con.src) || !static_cast<bool>(con.sink)) {
// either side disappeard without disconnectAll
// TODO: warn/error log
con.stop = true;
}
if (con.on_main_thread) {
con.pump_fn(con);
}
if (con.stop && (con.finished || con.on_main_thread)) {
if (!con.on_main_thread) {
assert(con.pump_thread.joinable());
con.pump_thread.join();
}
con.unsubscribe_fn(con);
it = _connections.erase(it);
} else {
it++;
}
}
// return min over intervals instead
return 2.f; // TODO: 2sec makes mainthread connections unusable
}

View File

@ -0,0 +1,205 @@
#pragma once
#include <solanaceae/object_store/fwd.hpp>
#include <solanaceae/object_store/object_store.hpp>
#include <entt/core/type_info.hpp>
#include "./frame_stream2.hpp"
#include <vector>
#include <memory>
#include <algorithm>
#include <thread>
#include <chrono>
#include <atomic>
// fwd
class StreamManager;
namespace Components {
struct StreamSource {
std::string name;
std::string frame_type_name;
std::function<bool(StreamManager&, Object, Object, bool)> connect_fn;
template<typename FrameType>
static StreamSource create(const std::string& name);
};
struct StreamSink {
std::string name;
std::string frame_type_name;
template<typename FrameType>
static StreamSink create(const std::string& name);
};
template<typename FrameType>
using FrameStream2Source = std::unique_ptr<FrameStream2SourceI<FrameType>>;
template<typename FrameType>
using FrameStream2Sink = std::unique_ptr<FrameStream2SinkI<FrameType>>;
} // Components
class StreamManager {
friend class StreamManagerUI; // TODO: make this go away
ObjectStore2& _os;
struct Connection {
ObjectHandle src;
ObjectHandle sink;
struct Data {
virtual ~Data(void) {}
};
std::unique_ptr<Data> data; // stores reader writer type erased
std::function<void(Connection&)> pump_fn; // TODO: make it return next interval?
std::function<void(Connection&)> unsubscribe_fn;
bool on_main_thread {true};
std::atomic_bool stop {false}; // disconnect
std::atomic_bool finished {false}; // disconnect
// pump thread
std::thread pump_thread;
// frame interval counters and estimates
// TODO
Connection(void) = default;
Connection(
ObjectHandle src_,
ObjectHandle sink_,
std::unique_ptr<Data>&& data_,
std::function<void(Connection&)>&& pump_fn_,
std::function<void(Connection&)>&& unsubscribe_fn_,
bool on_main_thread_ = true
);
};
std::vector<std::unique_ptr<Connection>> _connections;
public:
StreamManager(ObjectStore2& os);
virtual ~StreamManager(void);
template<typename FrameType>
bool connect(Object src, Object sink, bool threaded = true);
bool connect(Object src, Object sink, bool threaded = true);
bool disconnect(Object src, Object sink);
bool disconnectAll(Object o);
// do we need the time delta?
float tick(float);
};
// template impls
namespace Components {
// we require the complete sm type here
template<typename FrameType>
StreamSource StreamSource::create(const std::string& name) {
return StreamSource{
name,
std::string{entt::type_name<FrameType>::value()},
+[](StreamManager& sm, Object src, Object sink, bool threaded) {
return sm.connect<FrameType>(src, sink, threaded);
},
};
}
template<typename FrameType>
StreamSink StreamSink::create(const std::string& name) {
return StreamSink{
name,
std::string{entt::type_name<FrameType>::value()},
};
}
} // Components
template<typename FrameType>
bool StreamManager::connect(Object src, Object sink, bool threaded) {
auto res = std::find_if(
_connections.cbegin(), _connections.cend(),
[&](const auto& a) { return a->src == src && a->sink == sink; }
);
if (res != _connections.cend()) {
// already exists
return false;
}
auto h_src = _os.objectHandle(src);
auto h_sink = _os.objectHandle(sink);
if (!static_cast<bool>(h_src) || !static_cast<bool>(h_sink)) {
// an object does not exist
return false;
}
if (!h_src.all_of<Components::FrameStream2Source<FrameType>>()) {
// src not stream source
return false;
}
if (!h_sink.all_of<Components::FrameStream2Sink<FrameType>>()) {
// sink not stream sink
return false;
}
auto& src_stream = h_src.get<Components::FrameStream2Source<FrameType>>();
auto& sink_stream = h_sink.get<Components::FrameStream2Sink<FrameType>>();
struct inlineData : public Connection::Data {
virtual ~inlineData(void) {}
std::shared_ptr<FrameStream2I<FrameType>> reader;
std::shared_ptr<FrameStream2I<FrameType>> writer;
};
auto our_data = std::make_unique<inlineData>();
our_data->reader = src_stream->subscribe();
if (!our_data->reader) {
return false;
}
our_data->writer = sink_stream->subscribe();
if (!our_data->writer) {
return false;
}
_connections.push_back(std::make_unique<Connection>(
h_src,
h_sink,
std::move(our_data),
[](Connection& con) -> void {
// there might be more stored
for (size_t i = 0; i < 10; i++) {
auto new_frame_opt = static_cast<inlineData*>(con.data.get())->reader->pop();
// TODO: frame interval estimates
if (new_frame_opt.has_value()) {
static_cast<inlineData*>(con.data.get())->writer->push(new_frame_opt.value());
} else {
break;
}
}
},
[](Connection& con) -> void {
auto* src_stream_ptr = con.src.try_get<Components::FrameStream2Source<FrameType>>();
if (src_stream_ptr != nullptr) {
(*src_stream_ptr)->unsubscribe(static_cast<inlineData*>(con.data.get())->reader);
}
auto* sink_stream_ptr = con.sink.try_get<Components::FrameStream2Sink<FrameType>>();
if (sink_stream_ptr != nullptr) {
(*sink_stream_ptr)->unsubscribe(static_cast<inlineData*>(con.data.get())->writer);
}
},
!threaded
));
return true;
}

View File

@ -35,7 +35,7 @@ int main(int argc, char** argv) {
// setup hints // setup hints
#ifndef __ANDROID__ #ifndef __ANDROID__
if (SDL_SetHint(SDL_HINT_VIDEO_ALLOW_SCREENSAVER, "1") != SDL_TRUE) { if (!SDL_SetHint(SDL_HINT_VIDEO_ALLOW_SCREENSAVER, "1")) {
std::cerr << "Failed to set '" << SDL_HINT_VIDEO_ALLOW_SCREENSAVER << "' to 1\n"; std::cerr << "Failed to set '" << SDL_HINT_VIDEO_ALLOW_SCREENSAVER << "' to 1\n";
} }
#endif #endif

View File

@ -19,6 +19,7 @@ MainScreen::MainScreen(SimpleConfigModel&& conf_, SDL_Renderer* renderer_, Theme
rmm(cr), rmm(cr),
msnj{cr, {}, {}}, msnj{cr, {}, {}},
mts(rmm), mts(rmm),
sm(os),
tc(save_path, save_password), tc(save_path, save_password),
tpi(tc.getTox()), tpi(tc.getTox()),
ad(tc), ad(tc),
@ -474,6 +475,8 @@ Screen* MainScreen::render(float time_delta, bool&) {
} }
Screen* MainScreen::tick(float time_delta, bool& quit) { Screen* MainScreen::tick(float time_delta, bool& quit) {
const float sm_interval = sm.tick(time_delta);
quit = !tc.iterate(time_delta); // compute quit = !tc.iterate(time_delta); // compute
tcm.iterate(time_delta); // compute tcm.iterate(time_delta); // compute
@ -505,6 +508,10 @@ Screen* MainScreen::tick(float time_delta, bool& quit) {
std::pow(tc.toxIterationInterval(), 1.6f)/1000.f, std::pow(tc.toxIterationInterval(), 1.6f)/1000.f,
pm_interval pm_interval
); );
_min_tick_interval = std::min<float>(
_min_tick_interval,
sm_interval
);
_min_tick_interval = std::min<float>( _min_tick_interval = std::min<float>(
_min_tick_interval, _min_tick_interval,
fo_interval fo_interval

View File

@ -11,6 +11,7 @@
#include <solanaceae/plugin/plugin_manager.hpp> #include <solanaceae/plugin/plugin_manager.hpp>
#include <solanaceae/toxcore/tox_event_logger.hpp> #include <solanaceae/toxcore/tox_event_logger.hpp>
#include "./tox_private_impl.hpp" #include "./tox_private_impl.hpp"
#include "./frame_streams/stream_manager.hpp"
#include <solanaceae/tox_contacts/tox_contact_model2.hpp> #include <solanaceae/tox_contacts/tox_contact_model2.hpp>
#include <solanaceae/tox_messages/tox_message_manager.hpp> #include <solanaceae/tox_messages/tox_message_manager.hpp>
@ -58,6 +59,8 @@ struct MainScreen final : public Screen {
MessageSerializerNJ msnj; MessageSerializerNJ msnj;
MessageTimeSort mts; MessageTimeSort mts;
StreamManager sm;
ToxEventLogger tel{std::cout}; ToxEventLogger tel{std::cout};
ToxClient tc; ToxClient tc;
ToxPrivateImpl tpi; ToxPrivateImpl tpi;