From ce8f2c5d1009f5a8a9043b012faddfcc49d1ec88 Mon Sep 17 00:00:00 2001 From: Wim Taymans Date: Tue, 25 Jul 2023 18:36:48 +0200 Subject: [PATCH] modules-vban: add vban sender and receiver Fixes #3380 --- doc/pipewire-modules.dox | 2 + src/modules/meson.build | 22 ++ src/modules/module-vban-recv.c | 572 +++++++++++++++++++++++++++++++ src/modules/module-vban-send.c | 534 +++++++++++++++++++++++++++++ src/modules/module-vban/audio.c | 281 +++++++++++++++ src/modules/module-vban/stream.c | 479 ++++++++++++++++++++++++++ src/modules/module-vban/stream.h | 54 +++ src/modules/module-vban/vban.h | 60 ++++ 8 files changed, 2004 insertions(+) create mode 100644 src/modules/module-vban-recv.c create mode 100644 src/modules/module-vban-send.c create mode 100644 src/modules/module-vban/audio.c create mode 100644 src/modules/module-vban/stream.c create mode 100644 src/modules/module-vban/stream.h create mode 100644 src/modules/module-vban/vban.h diff --git a/doc/pipewire-modules.dox b/doc/pipewire-modules.dox index db7e3301d..460b0934f 100644 --- a/doc/pipewire-modules.dox +++ b/doc/pipewire-modules.dox @@ -86,6 +86,8 @@ List of known modules: - \subpage page_module_rtp_session - \subpage page_module_rt - \subpage page_module_session_manager +- \subpage page_module_vban_recv +- \subpage page_module_vban_send - \subpage page_module_x11_bell - \subpage page_module_zeroconf_discover diff --git a/src/modules/meson.build b/src/modules/meson.build index 3b9cacf13..9e38742cd 100644 --- a/src/modules/meson.build +++ b/src/modules/meson.build @@ -38,6 +38,8 @@ module_sources = [ 'module-rtp-session.c', 'module-rtp-source.c', 'module-rtp-sink.c', + 'module-vban-recv.c', + 'module-vban-send.c', 'module-session-manager.c', 'module-zeroconf-discover.c', 'module-roc-source.c', @@ -659,6 +661,26 @@ pipewire_module_rtp_sap = shared_library('pipewire-module-rtp-sap', dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep], ) +pipewire_module_vban_send = shared_library('pipewire-module-vban-send', + [ 'module-vban-send.c', + 'module-vban/stream.c' ], + include_directories : [configinc], + install : true, + install_dir : modules_install_dir, + install_rpath: modules_install_dir, + dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep], +) + +pipewire_module_vban_recv = shared_library('pipewire-module-vban-recv', + [ 'module-vban-recv.c', + 'module-vban/stream.c' ], + include_directories : [configinc], + install : true, + install_dir : modules_install_dir, + install_rpath: modules_install_dir, + dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep], +) + build_module_roc = roc_dep.found() if build_module_roc pipewire_module_roc_sink = shared_library('pipewire-module-roc-sink', diff --git a/src/modules/module-vban-recv.c b/src/modules/module-vban-recv.c new file mode 100644 index 000000000..72be2e722 --- /dev/null +++ b/src/modules/module-vban-recv.c @@ -0,0 +1,572 @@ +/* PipeWire */ +/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans */ +/* SPDX-License-Identifier: MIT */ + +#include "config.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include + +#ifdef __FreeBSD__ +#define ifr_ifindex ifr_index +#endif + +/** \page page_module_vban_recv PipeWire Module: VBAN receiver + * + * The `vban-recv` module creates a PipeWire source that receives audio + * and midi [VBAN](https://vb-audio.com) packets. + * + * ## Module Options + * + * Options specific to the behavior of this module + * + * - `local.ifname = `: interface name to use + * - `source.ip = `: the source ip address, default 127.0.0.1 + * - `source.port = `: the source port + * - `node.always-process = `: true to receive even when not running + * - `sess.latency.msec = `: target network latency in milliseconds, default 100 + * - `sess.ignore-ssrc = `: ignore SSRC, default false + * - `sess.media = `: the media type audio|midi|opus, default audio + * - `stream.props = {}`: properties to be passed to the stream + * + * ## General options + * + * Options with well-known behavior: + * + * - \ref PW_KEY_REMOTE_NAME + * - \ref PW_KEY_AUDIO_FORMAT + * - \ref PW_KEY_AUDIO_RATE + * - \ref PW_KEY_AUDIO_CHANNELS + * - \ref SPA_KEY_AUDIO_POSITION + * - \ref PW_KEY_MEDIA_NAME + * - \ref PW_KEY_MEDIA_CLASS + * - \ref PW_KEY_NODE_NAME + * - \ref PW_KEY_NODE_DESCRIPTION + * - \ref PW_KEY_NODE_GROUP + * - \ref PW_KEY_NODE_LATENCY + * - \ref PW_KEY_NODE_VIRTUAL + * + * ## Example configuration + *\code{.unparsed} + * context.modules = [ + * { name = libpipewire-module-vban-recv + * args = { + * #local.ifname = eth0 + * #source.ip = 127.0.0.1 + * #source.port = 6980 + * sess.latency.msec = 100 + * #sess.ignore-ssrc = false + * #node.always-process = false + * #sess.media = "audio" + * #audio.format = "S16LE" + * #audio.rate = 44100 + * #audio.channels = 2 + * #audio.position = [ FL FR ] + * stream.props = { + * #media.class = "Audio/Source" + * node.name = "vban-receiver" + * } + * } + * } + * ] + *\endcode + * + * \since 0.3.76 + */ + +#define NAME "vban-recv" + +PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME); +#define PW_LOG_TOPIC_DEFAULT mod_topic + +#define DEFAULT_CLEANUP_SEC 60 +#define DEFAULT_SOURCE_IP "127.0.0.1" +#define DEFAULT_SOURCE_PORT 6980 + +#define USAGE "( local.ifname= ) " \ + "( source.ip= ) " \ + "( source.port= " \ + "( sess.latency.msec= ) "\ + "( sess.media= ) " \ + "( audio.format= ) " \ + "( audio.rate= ) " \ + "( audio.channels= ) " \ + "( audio.position= ) " \ + "( stream.props= { key=value ... } ) " + +static const struct spa_dict_item module_info[] = { + { PW_KEY_MODULE_AUTHOR, "Wim Taymans " }, + { PW_KEY_MODULE_DESCRIPTION, "VBAN Receiver" }, + { PW_KEY_MODULE_USAGE, USAGE }, + { PW_KEY_MODULE_VERSION, PACKAGE_VERSION }, +}; + +struct impl { + struct pw_impl_module *module; + struct spa_hook module_listener; + struct pw_properties *props; + struct pw_context *context; + + struct pw_loop *loop; + struct pw_loop *data_loop; + + struct pw_core *core; + struct spa_hook core_listener; + struct spa_hook core_proxy_listener; + unsigned int do_disconnect:1; + + char *ifname; + bool always_process; + uint32_t cleanup_interval; + + struct spa_source *timer; + + struct pw_properties *stream_props; + struct vban_stream *stream; + + uint16_t src_port; + struct sockaddr_storage src_addr; + socklen_t src_len; + struct spa_source *source; + + unsigned receiving:1; +}; + +static void +on_vban_io(void *data, int fd, uint32_t mask) +{ + struct impl *impl = data; + ssize_t len; + uint8_t buffer[2048]; + + if (mask & SPA_IO_IN) { + if ((len = recv(fd, buffer, sizeof(buffer), 0)) < 0) + goto receive_error; + + if (len < 12) + goto short_packet; + + if (SPA_LIKELY(impl->stream)) + vban_stream_receive_packet(impl->stream, buffer, len); + + impl->receiving = true; + } + return; + +receive_error: + pw_log_warn("recv error: %m"); + return; +short_packet: + pw_log_warn("short packet received"); + return; +} + +static int parse_address(const char *address, uint16_t port, + struct sockaddr_storage *addr, socklen_t *len) +{ + struct sockaddr_in *sa4 = (struct sockaddr_in*)addr; + struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)addr; + + if (inet_pton(AF_INET, address, &sa4->sin_addr) > 0) { + sa4->sin_family = AF_INET; + sa4->sin_port = htons(port); + *len = sizeof(*sa4); + } else if (inet_pton(AF_INET6, address, &sa6->sin6_addr) > 0) { + sa6->sin6_family = AF_INET6; + sa6->sin6_port = htons(port); + *len = sizeof(*sa6); + } else + return -EINVAL; + + return 0; +} + +static int make_socket(const struct sockaddr* sa, socklen_t salen, char *ifname) +{ + int af, fd, val, res; + struct ifreq req; + + af = sa->sa_family; + if ((fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) { + pw_log_error("socket failed: %m"); + return -errno; + } +#ifdef SO_TIMESTAMP + val = 1; + if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &val, sizeof(val)) < 0) { + res = -errno; + pw_log_error("setsockopt failed: %m"); + goto error; + } +#endif + val = 1; + if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) < 0) { + res = -errno; + pw_log_error("setsockopt failed: %m"); + goto error; + } + + spa_zero(req); + if (ifname) { + snprintf(req.ifr_name, sizeof(req.ifr_name), "%s", ifname); + res = ioctl(fd, SIOCGIFINDEX, &req); + if (res < 0) + pw_log_warn("SIOCGIFINDEX %s failed: %m", ifname); + } + res = 0; + if (af == AF_INET) { + static const uint32_t ipv4_mcast_mask = 0xe0000000; + struct sockaddr_in *sa4 = (struct sockaddr_in*)sa; + if ((ntohl(sa4->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask) { + struct ip_mreqn mr4; + memset(&mr4, 0, sizeof(mr4)); + mr4.imr_multiaddr = sa4->sin_addr; + mr4.imr_ifindex = req.ifr_ifindex; + res = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4)); + } else { + sa4->sin_addr.s_addr = INADDR_ANY; + } + } else if (af == AF_INET6) { + struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)sa; + if (sa6->sin6_addr.s6_addr[0] == 0xff) { + struct ipv6_mreq mr6; + memset(&mr6, 0, sizeof(mr6)); + mr6.ipv6mr_multiaddr = sa6->sin6_addr; + mr6.ipv6mr_interface = req.ifr_ifindex; + res = setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mr6, sizeof(mr6)); + } else { + sa6->sin6_addr = in6addr_any; + } + } else { + res = -EINVAL; + goto error; + } + + if (res < 0) { + res = -errno; + pw_log_error("join mcast failed: %m"); + goto error; + } + + if (bind(fd, sa, salen) < 0) { + res = -errno; + pw_log_error("bind() failed: %m"); + goto error; + } + return fd; +error: + close(fd); + return res; +} + +static int stream_start(struct impl *impl) +{ + int fd; + + if (impl->source != NULL) + return 0; + + pw_log_info("starting VBAN listener"); + + if ((fd = make_socket((const struct sockaddr *)&impl->src_addr, + impl->src_len, impl->ifname)) < 0) { + pw_log_error("failed to create socket: %m"); + return fd; + } + + impl->source = pw_loop_add_io(impl->data_loop, fd, + SPA_IO_IN, true, on_vban_io, impl); + if (impl->source == NULL) { + pw_log_error("can't create io source: %m"); + close(fd); + return -errno; + } + return 0; +} + +static void stream_stop(struct impl *impl) +{ + if (!impl->source) + return; + + pw_log_info("stopping VBAN listener"); + + pw_loop_destroy_source(impl->data_loop, impl->source); + impl->source = NULL; +} + +static void stream_destroy(void *d) +{ + struct impl *impl = d; + impl->stream = NULL; +} + +static void stream_state_changed(void *data, bool started, const char *error) +{ + struct impl *impl = data; + + if (error) { + pw_log_error("stream error: %s", error); + pw_impl_module_schedule_destroy(impl->module); + } else if (started) { + if ((errno = -stream_start(impl)) < 0) + pw_log_error("failed to start VBAN stream: %m"); + } else { + if (!impl->always_process) + stream_stop(impl); + } +} + +static const struct vban_stream_events stream_events = { + VBAN_VERSION_STREAM_EVENTS, + .destroy = stream_destroy, + .state_changed = stream_state_changed, +}; + +static void on_timer_event(void *data, uint64_t expirations) +{ + struct impl *impl = data; + + if (!impl->receiving) { + pw_log_info("timeout, inactive VBAN source"); + //pw_impl_module_schedule_destroy(impl->module); + } else { + pw_log_debug("timeout, keeping active VBAN source"); + } + impl->receiving = false; +} + +static void core_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->core_listener); + impl->core = NULL; + pw_impl_module_schedule_destroy(impl->module); +} + +static const struct pw_proxy_events core_proxy_events = { + .destroy = core_destroy, +}; + +static void impl_destroy(struct impl *impl) +{ + if (impl->stream) + vban_stream_destroy(impl->stream); + if (impl->source) + pw_loop_destroy_source(impl->data_loop, impl->source); + + if (impl->core && impl->do_disconnect) + pw_core_disconnect(impl->core); + + if (impl->timer) + pw_loop_destroy_source(impl->loop, impl->timer); + + pw_properties_free(impl->stream_props); + pw_properties_free(impl->props); + + free(impl->ifname); + free(impl); +} + +static void module_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->module_listener); + impl_destroy(impl); +} + +static const struct pw_impl_module_events module_events = { + PW_VERSION_IMPL_MODULE_EVENTS, + .destroy = module_destroy, +}; + +static void on_core_error(void *d, uint32_t id, int seq, int res, const char *message) +{ + struct impl *impl = d; + + pw_log_error("error id:%u seq:%d res:%d (%s): %s", + id, seq, res, spa_strerror(res), message); + + if (id == PW_ID_CORE && res == -EPIPE) + pw_impl_module_schedule_destroy(impl->module); +} + +static const struct pw_core_events core_events = { + PW_VERSION_CORE_EVENTS, + .error = on_core_error, +}; + +static void copy_props(struct impl *impl, struct pw_properties *props, const char *key) +{ + const char *str; + if ((str = pw_properties_get(props, key)) != NULL) { + if (pw_properties_get(impl->stream_props, key) == NULL) + pw_properties_set(impl->stream_props, key, str); + } +} + +SPA_EXPORT +int pipewire__module_init(struct pw_impl_module *module, const char *args) +{ + struct pw_context *context = pw_impl_module_get_context(module); + struct impl *impl; + const char *str, *sess_name; + struct timespec value, interval; + struct pw_properties *props, *stream_props; + int res = 0; + + PW_LOG_TOPIC_INIT(mod_topic); + + impl = calloc(1, sizeof(struct impl)); + if (impl == NULL) + return -errno; + + if (args == NULL) + args = ""; + + props = impl->props = pw_properties_new_string(args); + stream_props = impl->stream_props = pw_properties_new(NULL, NULL); + if (props == NULL || stream_props == NULL) { + res = -errno; + pw_log_error( "can't create properties: %m"); + goto out; + } + + impl->module = module; + impl->context = context; + impl->loop = pw_context_get_main_loop(context); + impl->data_loop = pw_data_loop_get_loop(pw_context_get_data_loop(context)); + + if ((sess_name = pw_properties_get(props, "sess.name")) == NULL) + sess_name = pw_get_host_name(); + + if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL) + pw_properties_setf(props, PW_KEY_NODE_NAME, "vban_session.%s", sess_name); + if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL) + pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name); + if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL) + pw_properties_setf(props, PW_KEY_MEDIA_NAME, "VBAN Session with %s", + sess_name); + + if ((str = pw_properties_get(props, "stream.props")) != NULL) + pw_properties_update_string(stream_props, str, strlen(str)); + + copy_props(impl, props, PW_KEY_AUDIO_FORMAT); + copy_props(impl, props, PW_KEY_AUDIO_RATE); + copy_props(impl, props, PW_KEY_AUDIO_CHANNELS); + copy_props(impl, props, SPA_KEY_AUDIO_POSITION); + copy_props(impl, props, PW_KEY_NODE_NAME); + copy_props(impl, props, PW_KEY_NODE_DESCRIPTION); + copy_props(impl, props, PW_KEY_NODE_GROUP); + copy_props(impl, props, PW_KEY_NODE_LATENCY); + copy_props(impl, props, PW_KEY_NODE_VIRTUAL); + copy_props(impl, props, PW_KEY_NODE_CHANNELNAMES); + copy_props(impl, props, PW_KEY_MEDIA_NAME); + copy_props(impl, props, PW_KEY_MEDIA_CLASS); + copy_props(impl, props, "net.mtu"); + copy_props(impl, props, "sess.media"); + copy_props(impl, props, "sess.name"); + copy_props(impl, props, "sess.min-ptime"); + copy_props(impl, props, "sess.max-ptime"); + copy_props(impl, props, "sess.latency.msec"); + + str = pw_properties_get(props, "local.ifname"); + impl->ifname = str ? strdup(str) : NULL; + + impl->src_port = pw_properties_get_uint32(props, "source.port", DEFAULT_SOURCE_PORT); + if (impl->src_port == 0) { + pw_log_error("invalid source.port"); + goto out; + } + if ((str = pw_properties_get(props, "source.ip")) == NULL) + str = DEFAULT_SOURCE_IP; + if ((res = parse_address(str, impl->src_port, &impl->src_addr, &impl->src_len)) < 0) { + pw_log_error("invalid source.ip %s: %s", str, spa_strerror(res)); + goto out; + } + + impl->always_process = pw_properties_get_bool(stream_props, + PW_KEY_NODE_ALWAYS_PROCESS, true); + + impl->cleanup_interval = pw_properties_get_uint32(props, + "cleanup.sec", DEFAULT_CLEANUP_SEC); + + impl->core = pw_context_get_object(impl->context, PW_TYPE_INTERFACE_Core); + if (impl->core == NULL) { + str = pw_properties_get(props, PW_KEY_REMOTE_NAME); + impl->core = pw_context_connect(impl->context, + pw_properties_new( + PW_KEY_REMOTE_NAME, str, + NULL), + 0); + impl->do_disconnect = true; + } + if (impl->core == NULL) { + res = -errno; + pw_log_error("can't connect: %m"); + goto out; + } + + pw_proxy_add_listener((struct pw_proxy*)impl->core, + &impl->core_proxy_listener, + &core_proxy_events, impl); + pw_core_add_listener(impl->core, + &impl->core_listener, + &core_events, impl); + + impl->timer = pw_loop_add_timer(impl->loop, on_timer_event, impl); + if (impl->timer == NULL) { + res = -errno; + pw_log_error("can't create timer source: %m"); + goto out; + } + value.tv_sec = impl->cleanup_interval; + value.tv_nsec = 0; + interval.tv_sec = impl->cleanup_interval; + interval.tv_nsec = 0; + pw_loop_update_timer(impl->loop, impl->timer, &value, &interval, false); + + impl->stream = vban_stream_new(impl->core, + PW_DIRECTION_OUTPUT, pw_properties_copy(stream_props), + &stream_events, impl); + if (impl->stream == NULL) { + res = -errno; + pw_log_error("can't create stream: %m"); + goto out; + } + + pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl); + + pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_info)); + + pw_log_info("Successfully loaded module-vban-recv"); + + return 0; +out: + impl_destroy(impl); + return res; +} diff --git a/src/modules/module-vban-send.c b/src/modules/module-vban-send.c new file mode 100644 index 000000000..26961b81b --- /dev/null +++ b/src/modules/module-vban-send.c @@ -0,0 +1,534 @@ +/* PipeWire */ +/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans */ +/* SPDX-License-Identifier: MIT */ + +#include "config.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include +#include + +#include + +#ifndef IPTOS_DSCP +#define IPTOS_DSCP_MASK 0xfc +#define IPTOS_DSCP(x) ((x) & IPTOS_DSCP_MASK) +#endif + +/** \page page_module_vban_send PipeWire Module: VBAN sender + * + * The `vban-send` module creates a PipeWire sink that sends + * audio and midi [VBAN](https://vb-audio.com) packets. + * + * ## Module Options + * + * Options specific to the behavior of this module + * + * - `source.ip =`: source IP address, default "0.0.0.0" + * - `destination.ip =`: destination IP address, default "127.0.0.1" + * - `destination.port =`: destination port, default 6980 + * - `local.ifname = `: interface name to use + * - `net.mtu = `: MTU to use, default 1500 + * - `net.ttl = `: TTL to use, default 1 + * - `net.loop = `: loopback multicast, default false + * - `sess.min-ptime = `: minimum packet time in milliseconds, default 2 + * - `sess.max-ptime = `: maximum packet time in milliseconds, default 20 + * - `sess.name = `: a session name + * - `sess.media = `: the media type audio|midi, default audio + * - `stream.props = {}`: properties to be passed to the stream + * + * ## General options + * + * Options with well-known behavior: + * + * - \ref PW_KEY_REMOTE_NAME + * - \ref PW_KEY_AUDIO_FORMAT + * - \ref PW_KEY_AUDIO_RATE + * - \ref PW_KEY_AUDIO_CHANNELS + * - \ref SPA_KEY_AUDIO_POSITION + * - \ref PW_KEY_NODE_NAME + * - \ref PW_KEY_NODE_DESCRIPTION + * - \ref PW_KEY_MEDIA_NAME + * - \ref PW_KEY_NODE_GROUP + * - \ref PW_KEY_NODE_LATENCY + * - \ref PW_KEY_NODE_VIRTUAL + * - \ref PW_KEY_MEDIA_CLASS + * + * ## Example configuration + *\code{.unparsed} + * context.modules = [ + * { name = libpipewire-module-vban-send + * args = { + * #local.ifname = "eth0" + * #source.ip = "0.0.0.0" + * #destination.ip = "127.0.0.1" + * #destination.port = 6980 + * #net.mtu = 1500 + * #net.ttl = 1 + * #net.loop = false + * #sess.min-ptime = 2 + * #sess.max-ptime = 20 + * #sess.name = "PipeWire VBAN stream" + * #sess.media = "audio" + * #audio.format = "S16LE" + * #audio.rate = 44100 + * #audio.channels = 2 + * #audio.position = [ FL FR ] + * stream.props = { + * node.name = "vban-sender" + * } + * } + *} + *] + *\endcode + * + * \since 0.3.76 + */ + +#define NAME "vban-send" + +PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME); +#define PW_LOG_TOPIC_DEFAULT mod_topic + +#define DEFAULT_PORT 6980 +#define DEFAULT_SOURCE_IP "0.0.0.0" +#define DEFAULT_DESTINATION_IP "127.0.0.1" +#define DEFAULT_TTL 1 +#define DEFAULT_LOOP false +#define DEFAULT_DSCP 34 /* Default to AES-67 AF41 (34) */ + +#define USAGE "( source.ip= ) " \ + "( destination.ip= ) " \ + "( destination.port= ) " \ + "( local.ifname= ) " \ + "( net.mtu= ) " \ + "( net.ttl= ) " \ + "( net.loop= ) " \ + "( net.dscp= ) " \ + "( sess.name= ) " \ + "( sess.min-ptime= ) " \ + "( sess.max-ptime= ) " \ + "( sess.media= ) " \ + "( audio.format= ) " \ + "( audio.rate= ) " \ + "( audio.channels= ) " \ + "( audio.position= ) " \ + "( stream.props= { key=value ... } ) " + +static const struct spa_dict_item module_info[] = { + { PW_KEY_MODULE_AUTHOR, "Wim Taymans " }, + { PW_KEY_MODULE_DESCRIPTION, "VBAN Sender" }, + { PW_KEY_MODULE_USAGE, USAGE }, + { PW_KEY_MODULE_VERSION, PACKAGE_VERSION }, +}; + +struct impl { + struct pw_context *context; + + struct pw_impl_module *module; + struct spa_hook module_listener; + struct pw_properties *props; + + struct pw_loop *loop; + + struct pw_core *core; + struct spa_hook core_listener; + struct spa_hook core_proxy_listener; + + struct pw_properties *stream_props; + struct vban_stream *stream; + + unsigned int do_disconnect:1; + + char *ifname; + char *session_name; + uint32_t ttl; + bool mcast_loop; + uint32_t dscp; + + struct sockaddr_storage src_addr; + socklen_t src_len; + + uint16_t dst_port; + struct sockaddr_storage dst_addr; + socklen_t dst_len; + + int vban_fd; +}; + +static void stream_destroy(void *d) +{ + struct impl *impl = d; + impl->stream = NULL; +} + +static void stream_send_packet(void *data, struct iovec *iov, size_t iovlen) +{ + struct impl *impl = data; + struct msghdr msg; + ssize_t n; + + spa_zero(msg); + msg.msg_iov = iov; + msg.msg_iovlen = iovlen; + msg.msg_control = NULL; + msg.msg_controllen = 0; + msg.msg_flags = 0; + + n = sendmsg(impl->vban_fd, &msg, MSG_NOSIGNAL); + if (n < 0) + pw_log_debug("sendmsg() failed: %m"); +} + +static void stream_state_changed(void *data, bool started, const char *error) +{ + struct impl *impl = data; + + if (error) { + pw_log_error("stream error: %s", error); + pw_impl_module_schedule_destroy(impl->module); + } +} + +static const struct vban_stream_events stream_events = { + VBAN_VERSION_STREAM_EVENTS, + .destroy = stream_destroy, + .state_changed = stream_state_changed, + .send_packet = stream_send_packet, +}; + +static int parse_address(const char *address, uint16_t port, + struct sockaddr_storage *addr, socklen_t *len) +{ + struct sockaddr_in *sa4 = (struct sockaddr_in*)addr; + struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)addr; + + if (inet_pton(AF_INET, address, &sa4->sin_addr) > 0) { + sa4->sin_family = AF_INET; + sa4->sin_port = htons(port); + *len = sizeof(*sa4); + } else if (inet_pton(AF_INET6, address, &sa6->sin6_addr) > 0) { + sa6->sin6_family = AF_INET6; + sa6->sin6_port = htons(port); + *len = sizeof(*sa6); + } else + return -EINVAL; + + return 0; +} + +static bool is_multicast(struct sockaddr *sa, socklen_t salen) +{ + if (sa->sa_family == AF_INET) { + static const uint32_t ipv4_mcast_mask = 0xe0000000; + struct sockaddr_in *sa4 = (struct sockaddr_in*)sa; + return (ntohl(sa4->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask; + } else if (sa->sa_family == AF_INET6) { + struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)sa; + return sa6->sin6_addr.s6_addr[0] == 0xff; + } + return false; +} + +static int make_socket(struct sockaddr_storage *src, socklen_t src_len, + struct sockaddr_storage *dst, socklen_t dst_len, + bool loop, int ttl, int dscp) +{ + int af, fd, val, res; + + af = src->ss_family; + if ((fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) { + pw_log_error("socket failed: %m"); + return -errno; + } + if (bind(fd, (struct sockaddr*)src, src_len) < 0) { + res = -errno; + pw_log_error("bind() failed: %m"); + goto error; + } + if (connect(fd, (struct sockaddr*)dst, dst_len) < 0) { + res = -errno; + pw_log_error("connect() failed: %m"); + goto error; + } + if (is_multicast((struct sockaddr*)dst, dst_len)) { + val = loop; + if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &val, sizeof(val)) < 0) + pw_log_warn("setsockopt(IP_MULTICAST_LOOP) failed: %m"); + + val = ttl; + if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &val, sizeof(val)) < 0) + pw_log_warn("setsockopt(IP_MULTICAST_TTL) failed: %m"); + } +#ifdef SO_PRIORITY + val = 6; + if (setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &val, sizeof(val)) < 0) + pw_log_warn("setsockopt(SO_PRIORITY) failed: %m"); +#endif + if (dscp > 0) { + val = IPTOS_DSCP(dscp << 2); + if (setsockopt(fd, IPPROTO_IP, IP_TOS, &val, sizeof(val)) < 0) + pw_log_warn("setsockopt(IP_TOS) failed: %m"); + } + + + return fd; +error: + close(fd); + return res; +} + +static int get_ip(const struct sockaddr_storage *sa, char *ip, size_t len) +{ + if (sa->ss_family == AF_INET) { + struct sockaddr_in *in = (struct sockaddr_in*)sa; + inet_ntop(sa->ss_family, &in->sin_addr, ip, len); + } else if (sa->ss_family == AF_INET6) { + struct sockaddr_in6 *in = (struct sockaddr_in6*)sa; + inet_ntop(sa->ss_family, &in->sin6_addr, ip, len); + } else + return -EIO; + return 0; +} + +static void core_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->core_listener); + impl->core = NULL; + pw_impl_module_schedule_destroy(impl->module); +} + +static const struct pw_proxy_events core_proxy_events = { + .destroy = core_destroy, +}; + +static void impl_destroy(struct impl *impl) +{ + if (impl->stream) + vban_stream_destroy(impl->stream); + + if (impl->core && impl->do_disconnect) + pw_core_disconnect(impl->core); + + if (impl->vban_fd != -1) + close(impl->vban_fd); + + pw_properties_free(impl->stream_props); + pw_properties_free(impl->props); + + free(impl->ifname); + free(impl->session_name); + free(impl); +} + +static void module_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->module_listener); + impl_destroy(impl); +} + +static const struct pw_impl_module_events module_events = { + PW_VERSION_IMPL_MODULE_EVENTS, + .destroy = module_destroy, +}; + +static void on_core_error(void *d, uint32_t id, int seq, int res, const char *message) +{ + struct impl *impl = d; + + pw_log_error("error id:%u seq:%d res:%d (%s): %s", + id, seq, res, spa_strerror(res), message); + + if (id == PW_ID_CORE && res == -EPIPE) + pw_impl_module_schedule_destroy(impl->module); +} + +static const struct pw_core_events core_events = { + PW_VERSION_CORE_EVENTS, + .error = on_core_error, +}; + +static void copy_props(struct impl *impl, struct pw_properties *props, const char *key) +{ + const char *str; + if ((str = pw_properties_get(props, key)) != NULL) { + if (pw_properties_get(impl->stream_props, key) == NULL) + pw_properties_set(impl->stream_props, key, str); + } +} + +SPA_EXPORT +int pipewire__module_init(struct pw_impl_module *module, const char *args) +{ + struct pw_context *context = pw_impl_module_get_context(module); + struct impl *impl; + struct pw_properties *props = NULL, *stream_props = NULL; + char addr[64]; + const char *str, *sess_name; + int res = 0; + + PW_LOG_TOPIC_INIT(mod_topic); + + impl = calloc(1, sizeof(struct impl)); + if (impl == NULL) + return -errno; + + impl->vban_fd = -1; + + if (args == NULL) + args = ""; + + props = pw_properties_new_string(args); + if (props == NULL) { + res = -errno; + pw_log_error( "can't create properties: %m"); + goto out; + } + impl->props = props; + + stream_props = pw_properties_new(NULL, NULL); + if (stream_props == NULL) { + res = -errno; + pw_log_error( "can't create properties: %m"); + goto out; + } + impl->stream_props = stream_props; + + impl->module = module; + impl->context = context; + impl->loop = pw_context_get_main_loop(context); + + if ((sess_name = pw_properties_get(props, "sess.name")) == NULL) + sess_name = pw_get_host_name(); + + if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL) + pw_properties_setf(props, PW_KEY_NODE_NAME, "vban_session.%s", sess_name); + if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL) + pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name); + if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL) + pw_properties_setf(props, PW_KEY_MEDIA_NAME, "VBAN Session with %s", + sess_name); + + if ((str = pw_properties_get(props, "stream.props")) != NULL) + pw_properties_update_string(stream_props, str, strlen(str)); + + copy_props(impl, props, PW_KEY_AUDIO_FORMAT); + copy_props(impl, props, PW_KEY_AUDIO_RATE); + copy_props(impl, props, PW_KEY_AUDIO_CHANNELS); + copy_props(impl, props, SPA_KEY_AUDIO_POSITION); + copy_props(impl, props, PW_KEY_NODE_NAME); + copy_props(impl, props, PW_KEY_NODE_DESCRIPTION); + copy_props(impl, props, PW_KEY_NODE_GROUP); + copy_props(impl, props, PW_KEY_NODE_LATENCY); + copy_props(impl, props, PW_KEY_NODE_VIRTUAL); + copy_props(impl, props, PW_KEY_NODE_CHANNELNAMES); + copy_props(impl, props, PW_KEY_MEDIA_NAME); + copy_props(impl, props, PW_KEY_MEDIA_CLASS); + copy_props(impl, props, "net.mtu"); + copy_props(impl, props, "sess.media"); + copy_props(impl, props, "sess.name"); + copy_props(impl, props, "sess.min-ptime"); + copy_props(impl, props, "sess.max-ptime"); + copy_props(impl, props, "sess.latency.msec"); + copy_props(impl, props, "sess.ts-refclk"); + + str = pw_properties_get(props, "local.ifname"); + impl->ifname = str ? strdup(str) : NULL; + + if ((str = pw_properties_get(props, "source.ip")) == NULL) + str = DEFAULT_SOURCE_IP; + if ((res = parse_address(str, 0, &impl->src_addr, &impl->src_len)) < 0) { + pw_log_error("invalid source.ip %s: %s", str, spa_strerror(res)); + goto out; + } + + impl->dst_port = pw_properties_get_uint32(props, "destination.port", DEFAULT_PORT); + if ((str = pw_properties_get(props, "destination.ip")) == NULL) + str = DEFAULT_DESTINATION_IP; + if ((res = parse_address(str, impl->dst_port, &impl->dst_addr, &impl->dst_len)) < 0) { + pw_log_error("invalid destination.ip %s: %s", str, spa_strerror(res)); + goto out; + } + + impl->ttl = pw_properties_get_uint32(props, "net.ttl", DEFAULT_TTL); + impl->mcast_loop = pw_properties_get_bool(props, "net.loop", DEFAULT_LOOP); + impl->dscp = pw_properties_get_uint32(props, "net.dscp", DEFAULT_DSCP); + + get_ip(&impl->src_addr, addr, sizeof(addr)); + pw_properties_set(stream_props, "vban.source.ip", addr); + get_ip(&impl->dst_addr, addr, sizeof(addr)); + pw_properties_set(stream_props, "vban.destination.ip", addr); + pw_properties_setf(stream_props, "vban.destination.port", "%u", impl->dst_port); + pw_properties_setf(stream_props, "vban.ttl", "%u", impl->ttl); + pw_properties_setf(stream_props, "vban.dscp", "%u", impl->dscp); + + impl->core = pw_context_get_object(impl->context, PW_TYPE_INTERFACE_Core); + if (impl->core == NULL) { + str = pw_properties_get(props, PW_KEY_REMOTE_NAME); + impl->core = pw_context_connect(impl->context, + pw_properties_new( + PW_KEY_REMOTE_NAME, str, + NULL), + 0); + impl->do_disconnect = true; + } + if (impl->core == NULL) { + res = -errno; + pw_log_error("can't connect: %m"); + goto out; + } + + pw_proxy_add_listener((struct pw_proxy*)impl->core, + &impl->core_proxy_listener, + &core_proxy_events, impl); + pw_core_add_listener(impl->core, + &impl->core_listener, + &core_events, impl); + + if ((res = make_socket(&impl->src_addr, impl->src_len, + &impl->dst_addr, impl->dst_len, + impl->mcast_loop, impl->ttl, impl->dscp)) < 0) { + pw_log_error("can't make socket: %s", spa_strerror(res)); + goto out; + } + impl->vban_fd = res; + + impl->stream = vban_stream_new(impl->core, + PW_DIRECTION_INPUT, pw_properties_copy(stream_props), + &stream_events, impl); + if (impl->stream == NULL) { + res = -errno; + pw_log_error("can't create stream: %m"); + goto out; + } + + pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl); + + pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_info)); + + pw_log_info("Successfully loaded module-vban-send"); + + return 0; +out: + impl_destroy(impl); + return res; +} diff --git a/src/modules/module-vban/audio.c b/src/modules/module-vban/audio.c new file mode 100644 index 000000000..ad32398e0 --- /dev/null +++ b/src/modules/module-vban/audio.c @@ -0,0 +1,281 @@ +/* PipeWire */ +/* SPDX-FileCopyrightText: Copyright © 2022 Wim Taymans */ +/* SPDX-License-Identifier: MIT */ + +static void vban_audio_process_playback(void *data) +{ + struct impl *impl = data; + struct pw_buffer *buf; + struct spa_data *d; + uint32_t wanted, timestamp, target_buffer, stride, maxsize; + int32_t avail; + + if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) { + pw_log_debug("Out of stream buffers: %m"); + return; + } + d = buf->buffer->datas; + + stride = impl->stride; + + maxsize = d[0].maxsize / stride; + wanted = buf->requested ? SPA_MIN(buf->requested, maxsize) : maxsize; + + avail = spa_ringbuffer_get_read_index(&impl->ring, ×tamp); + + target_buffer = impl->target_buffer; + + if (avail < (int32_t)wanted) { + enum spa_log_level level; + memset(d[0].data, 0, wanted * stride); + if (impl->have_sync) { + impl->have_sync = false; + level = SPA_LOG_LEVEL_WARN; + } else { + level = SPA_LOG_LEVEL_DEBUG; + } + pw_log(level, "underrun %d/%u < %u", + avail, target_buffer, wanted); + } else { + float error, corr; + if (impl->first) { + if ((uint32_t)avail > target_buffer) { + uint32_t skip = avail - target_buffer; + pw_log_debug("first: avail:%d skip:%u target:%u", + avail, skip, target_buffer); + timestamp += skip; + avail = target_buffer; + } + impl->first = false; + } else if (avail > (int32_t)SPA_MIN(target_buffer * 8, BUFFER_SIZE / stride)) { + pw_log_warn("overrun %u > %u", avail, target_buffer * 8); + timestamp += avail - target_buffer; + avail = target_buffer; + } + /* try to adjust our playback rate to keep the + * requested target_buffer bytes in the ringbuffer */ + error = (float)target_buffer - (float)avail; + error = SPA_CLAMP(error, -impl->max_error, impl->max_error); + + corr = spa_dll_update(&impl->dll, error); + + pw_log_debug("avail:%u target:%u error:%f corr:%f", avail, + target_buffer, error, corr); + + if (impl->io_rate_match) { + SPA_FLAG_SET(impl->io_rate_match->flags, + SPA_IO_RATE_MATCH_FLAG_ACTIVE); + impl->io_rate_match->rate = 1.0f / corr; + } + spa_ringbuffer_read_data(&impl->ring, + impl->buffer, + BUFFER_SIZE, + (timestamp * stride) & BUFFER_MASK, + d[0].data, wanted * stride); + + timestamp += wanted; + spa_ringbuffer_read_update(&impl->ring, timestamp); + } + d[0].chunk->size = wanted * stride; + d[0].chunk->stride = stride; + d[0].chunk->offset = 0; + buf->size = wanted; + + pw_stream_queue_buffer(impl->stream, buf); +} + +static int vban_audio_receive(struct impl *impl, uint8_t *buffer, ssize_t len) +{ + struct vban_header *hdr; + ssize_t hlen, plen; + uint32_t n_frames, timestamp, samples, write, expected_write; + uint32_t stride = impl->stride; + int32_t filled; + + if (len < VBAN_HEADER_SIZE) + goto short_packet; + + hdr = (struct vban_header*)buffer; + if (strncmp(hdr->vban, "VBAN", 3)) + goto invalid_version; + + impl->receiving = true; + + hlen = VBAN_HEADER_SIZE; + plen = len - hlen; + samples = SPA_MIN(hdr->format_nbs, plen / stride); + + n_frames = hdr->n_frames; + if (impl->have_sync && impl->n_frames != n_frames) { + pw_log_info("unexpected frame (%d != %d)", + n_frames, impl->n_frames); + impl->have_sync = false; + } + impl->n_frames = n_frames + 1; + + timestamp = impl->timestamp; + impl->timestamp += samples; + + filled = spa_ringbuffer_get_write_index(&impl->ring, &expected_write); + + /* we always write to timestamp + delay */ + write = timestamp + impl->target_buffer; + + if (!impl->have_sync) { + pw_log_info("sync to timestamp:%u target:%u", + timestamp, impl->target_buffer); + + /* we read from timestamp, keeping target_buffer of data + * in the ringbuffer. */ + impl->ring.readindex = timestamp; + impl->ring.writeindex = write; + filled = impl->target_buffer; + + spa_dll_init(&impl->dll); + spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate); + memset(impl->buffer, 0, BUFFER_SIZE); + impl->have_sync = true; + } else if (expected_write != write) { + pw_log_debug("unexpected write (%u != %u)", + write, expected_write); + } + + if (filled + samples > BUFFER_SIZE / stride) { + pw_log_debug("capture overrun %u + %u > %u", filled, samples, + BUFFER_SIZE / stride); + impl->have_sync = false; + } else { + pw_log_debug("got samples:%u", samples); + spa_ringbuffer_write_data(&impl->ring, + impl->buffer, + BUFFER_SIZE, + (write * stride) & BUFFER_MASK, + &buffer[hlen], (samples * stride)); + write += samples; + spa_ringbuffer_write_update(&impl->ring, write); + } + return 0; + +short_packet: + pw_log_warn("short packet received"); + return -EINVAL; +invalid_version: + pw_log_warn("invalid VBAN version"); + spa_debug_mem(0, buffer, len); + return -EPROTO; +} + +static inline void +set_iovec(struct spa_ringbuffer *rbuf, void *buffer, uint32_t size, + uint32_t offset, struct iovec *iov, uint32_t len) +{ + iov[0].iov_len = SPA_MIN(len, size - offset); + iov[0].iov_base = SPA_PTROFF(buffer, offset, void); + iov[1].iov_len = len - iov[0].iov_len; + iov[1].iov_base = buffer; +} + +static void vban_audio_flush_packets(struct impl *impl) +{ + int32_t avail, tosend; + uint32_t stride, timestamp; + struct iovec iov[3]; + struct vban_header header; + + avail = spa_ringbuffer_get_read_index(&impl->ring, ×tamp); + tosend = impl->psamples; + + if (avail < tosend) + return; + + stride = impl->stride; + + header = impl->header; + header.format_nbs = tosend - 1; + header.format_nbc = impl->stream_info.info.raw.channels - 1; + + iov[0].iov_base = &header; + iov[0].iov_len = sizeof(header); + + while (avail >= tosend) { + set_iovec(&impl->ring, + impl->buffer, BUFFER_SIZE, + (timestamp * stride) & BUFFER_MASK, + &iov[1], tosend * stride); + + pw_log_trace("sending %d timestamp:%08x", tosend, timestamp); + + vban_stream_emit_send_packet(impl, iov, 3); + + timestamp += tosend; + avail -= tosend; + impl->header.n_frames++; + } + spa_ringbuffer_read_update(&impl->ring, timestamp); +} + +static void vban_audio_process_capture(void *data) +{ + struct impl *impl = data; + struct pw_buffer *buf; + struct spa_data *d; + uint32_t offs, size, timestamp, expected_timestamp, stride; + int32_t filled, wanted; + + if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) { + pw_log_debug("Out of stream buffers: %m"); + return; + } + d = buf->buffer->datas; + + offs = SPA_MIN(d[0].chunk->offset, d[0].maxsize); + size = SPA_MIN(d[0].chunk->size, d[0].maxsize - offs); + stride = impl->stride; + wanted = size / stride; + + filled = spa_ringbuffer_get_write_index(&impl->ring, &expected_timestamp); + + if (SPA_LIKELY(impl->io_position)) { + uint32_t rate = impl->io_position->clock.rate.denom; + timestamp = impl->io_position->clock.position * impl->rate / rate; + } else + timestamp = expected_timestamp; + + if (!impl->have_sync) { + pw_log_info("sync to timestamp:%u", timestamp); + impl->ring.readindex = impl->ring.writeindex = timestamp; + memset(impl->buffer, 0, BUFFER_SIZE); + impl->have_sync = true; + expected_timestamp = timestamp; + } else { + if (SPA_ABS((int32_t)expected_timestamp - (int32_t)timestamp) > 32) { + pw_log_warn("expected %u != timestamp %u", expected_timestamp, timestamp); + impl->have_sync = false; + } else if (filled + wanted > (int32_t)(BUFFER_SIZE / stride)) { + pw_log_warn("overrun %u + %u > %u", filled, wanted, BUFFER_SIZE / stride); + impl->have_sync = false; + } + } + + spa_ringbuffer_write_data(&impl->ring, + impl->buffer, + BUFFER_SIZE, + (expected_timestamp * stride) & BUFFER_MASK, + SPA_PTROFF(d[0].data, offs, void), wanted * stride); + expected_timestamp += wanted; + spa_ringbuffer_write_update(&impl->ring, expected_timestamp); + + pw_stream_queue_buffer(impl->stream, buf); + + vban_audio_flush_packets(impl); +} + +static int vban_audio_init(struct impl *impl, enum spa_direction direction) +{ + if (direction == SPA_DIRECTION_INPUT) + impl->stream_events.process = vban_audio_process_capture; + else + impl->stream_events.process = vban_audio_process_playback; + impl->receive_vban = vban_audio_receive; + return 0; +} diff --git a/src/modules/module-vban/stream.c b/src/modules/module-vban/stream.c new file mode 100644 index 000000000..08a38cc21 --- /dev/null +++ b/src/modules/module-vban/stream.c @@ -0,0 +1,479 @@ +/* PipeWire */ +/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans */ +/* SPDX-License-Identifier: MIT */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "config.h" + +#include +#include + +#include +#include + +#define BUFFER_SIZE (1u<<22) +#define BUFFER_MASK (BUFFER_SIZE-1) +#define BUFFER_SIZE2 (BUFFER_SIZE>>1) +#define BUFFER_MASK2 (BUFFER_SIZE2-1) + +#define vban_stream_emit(s,m,v,...) spa_hook_list_call(&s->listener_list, \ + struct vban_stream_events, m, v, ##__VA_ARGS__) +#define vban_stream_emit_destroy(s) vban_stream_emit(s, destroy, 0) +#define vban_stream_emit_state_changed(s,n,e) vban_stream_emit(s, state_changed,0,n,e) +#define vban_stream_emit_send_packet(s,i,l) vban_stream_emit(s, send_packet,0,i,l) +#define vban_stream_emit_send_feedback(s,seq) vban_stream_emit(s, send_feedback,0,seq) + +struct impl { + struct spa_audio_info info; + struct spa_audio_info stream_info; + + struct pw_stream *stream; + struct spa_hook stream_listener; + struct pw_stream_events stream_events; + + struct spa_hook_list listener_list; + struct spa_hook listener; + + const struct format_info *format_info; + + void *stream_data; + + uint32_t rate; + uint32_t stride; + uint32_t psamples; + uint32_t mtu; + + struct vban_header header; + uint32_t timestamp; + uint32_t n_frames; + + struct spa_ringbuffer ring; + uint8_t buffer[BUFFER_SIZE]; + + struct spa_io_rate_match *io_rate_match; + struct spa_io_position *io_position; + struct spa_dll dll; + double corr; + uint32_t target_buffer; + float max_error; + + float last_timestamp; + float last_time; + + unsigned always_process:1; + unsigned started:1; + unsigned have_sync:1; + unsigned receiving:1; + unsigned first:1; + + int (*receive_vban)(struct impl *impl, uint8_t *buffer, ssize_t len); +}; + +#include "module-vban/audio.c" +//#include "module-vban/midi.c" + +struct format_info { + uint32_t media_subtype; + uint32_t format; + uint32_t size; + uint8_t format_bit; +}; + +static const struct format_info audio_format_info[] = { + { SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_U8, 1, VBAN_DATATYPE_U8, }, + { SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S16_LE, 2, VBAN_DATATYPE_INT16, }, + { SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S24_LE, 3, VBAN_DATATYPE_INT24, }, + { SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S32_LE, 4, VBAN_DATATYPE_INT32, }, + { SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F32_LE, 4, VBAN_DATATYPE_FLOAT32, }, + { SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F64_LE, 8, VBAN_DATATYPE_FLOAT64, }, + { SPA_MEDIA_SUBTYPE_control, 0, 1, }, +}; + +static void stream_io_changed(void *data, uint32_t id, void *area, uint32_t size) +{ + struct impl *impl = data; + switch (id) { + case SPA_IO_RateMatch: + impl->io_rate_match = area; + break; + case SPA_IO_Position: + impl->io_position = area; + break; + } +} + +static void stream_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->stream_listener); + impl->stream = NULL; +} + +static int stream_start(struct impl *impl) +{ + if (impl->started) + return 0; + + vban_stream_emit_state_changed(impl, true, NULL); + + impl->started = true; + return 0; +} + +static int stream_stop(struct impl *impl) +{ + if (!impl->started) + return 0; + + vban_stream_emit_state_changed(impl, false, NULL); + + impl->started = false; + return 0; +} + +static void on_stream_state_changed(void *d, enum pw_stream_state old, + enum pw_stream_state state, const char *error) +{ + struct impl *impl = d; + + switch (state) { + case PW_STREAM_STATE_UNCONNECTED: + pw_log_info("stream disconnected"); + break; + case PW_STREAM_STATE_ERROR: + pw_log_error("stream error: %s", error); + vban_stream_emit_state_changed(impl, false, error); + break; + case PW_STREAM_STATE_STREAMING: + if ((errno = -stream_start(impl)) < 0) + pw_log_error("failed to start RTP stream: %m"); + break; + case PW_STREAM_STATE_PAUSED: + if (!impl->always_process) + stream_stop(impl); + impl->have_sync = false; + break; + default: + break; + } +} + +static const struct pw_stream_events stream_events = { + PW_VERSION_STREAM_EVENTS, + .destroy = stream_destroy, + .state_changed = on_stream_state_changed, + .io_changed = stream_io_changed, +}; + +static const struct format_info *find_audio_format_info(const struct spa_audio_info *info) +{ + SPA_FOR_EACH_ELEMENT_VAR(audio_format_info, f) + if (f->media_subtype == info->media_subtype && + (f->format == 0 || f->format == info->info.raw.format)) + return f; + return NULL; +} + +static inline uint32_t format_from_name(const char *name, size_t len) +{ + int i; + for (i = 0; spa_type_audio_format[i].name; i++) { + if (strncmp(name, spa_debug_type_short_name(spa_type_audio_format[i].name), len) == 0) + return spa_type_audio_format[i].type; + } + return SPA_AUDIO_FORMAT_UNKNOWN; +} + +static uint32_t channel_from_name(const char *name) +{ + int i; + for (i = 0; spa_type_audio_channel[i].name; i++) { + if (spa_streq(name, spa_debug_type_short_name(spa_type_audio_channel[i].name))) + return spa_type_audio_channel[i].type; + } + return SPA_AUDIO_CHANNEL_UNKNOWN; +} + +static void parse_position(struct spa_audio_info_raw *info, const char *val, size_t len) +{ + struct spa_json it[2]; + char v[256]; + + spa_json_init(&it[0], val, len); + if (spa_json_enter_array(&it[0], &it[1]) <= 0) + spa_json_init(&it[1], val, len); + + info->channels = 0; + while (spa_json_get_string(&it[1], v, sizeof(v)) > 0 && + info->channels < SPA_AUDIO_MAX_CHANNELS) { + info->position[info->channels++] = channel_from_name(v); + } +} + +static void parse_audio_info(const struct pw_properties *props, struct spa_audio_info_raw *info) +{ + const char *str; + + spa_zero(*info); + if ((str = pw_properties_get(props, PW_KEY_AUDIO_FORMAT)) == NULL) + str = DEFAULT_FORMAT; + info->format = format_from_name(str, strlen(str)); + + info->rate = pw_properties_get_uint32(props, PW_KEY_AUDIO_RATE, info->rate); + if (info->rate == 0) + info->rate = DEFAULT_RATE; + + info->channels = pw_properties_get_uint32(props, PW_KEY_AUDIO_CHANNELS, info->channels); + info->channels = SPA_MIN(info->channels, SPA_AUDIO_MAX_CHANNELS); + if ((str = pw_properties_get(props, SPA_KEY_AUDIO_POSITION)) != NULL) + parse_position(info, str, strlen(str)); + if (info->channels == 0) + parse_position(info, DEFAULT_POSITION, strlen(DEFAULT_POSITION)); +} + +static uint32_t msec_to_samples(struct impl *impl, uint32_t msec) +{ + return msec * impl->rate / 1000; +} + +struct vban_stream *vban_stream_new(struct pw_core *core, + enum pw_direction direction, struct pw_properties *props, + const struct vban_stream_events *events, void *data) +{ + struct impl *impl; + const char *str; + uint8_t buffer[1024]; + struct spa_pod_builder b; + uint32_t n_params, min_samples, max_samples; + float min_ptime, max_ptime; + const struct spa_pod *params[1]; + enum pw_stream_flags flags; + int latency_msec; + int res; + + impl = calloc(1, sizeof(*impl)); + if (impl == NULL) { + res = -errno; + goto out; + return NULL; + } + impl->first = true; + spa_hook_list_init(&impl->listener_list); + impl->stream_events = stream_events; + + if ((str = pw_properties_get(props, "sess.media")) == NULL) + str = "audio"; + + if (spa_streq(str, "audio")) { + impl->info.media_type = SPA_MEDIA_TYPE_audio; + impl->info.media_subtype = SPA_MEDIA_SUBTYPE_raw; + } + else if (spa_streq(str, "midi")) { + impl->info.media_type = SPA_MEDIA_TYPE_application; + impl->info.media_subtype = SPA_MEDIA_SUBTYPE_control; + } + else { + pw_log_error("unsupported media type:%s", str); + res = -EINVAL; + goto out; + } + memcpy(impl->header.vban, "VBAN", 4); + if ((str = pw_properties_get(props, "sess.name")) == NULL) + str = "Stream1"; + strcpy(impl->header.stream_name, str); + + switch (impl->info.media_subtype) { + case SPA_MEDIA_SUBTYPE_raw: + parse_audio_info(props, &impl->info.info.raw); + impl->stream_info = impl->info; + impl->format_info = find_audio_format_info(&impl->info); + if (impl->format_info == NULL) { + pw_log_error("unsupported audio format:%d channels:%d", + impl->stream_info.info.raw.format, + impl->stream_info.info.raw.channels); + res = -EINVAL; + goto out; + } + impl->stride = impl->format_info->size * impl->stream_info.info.raw.channels; + impl->rate = impl->stream_info.info.raw.rate; + impl->header.format_SR = vban_sr_index(impl->rate); + if (impl->header.format_SR == VBAN_SR_MAXNUMBER) { + pw_log_error("unsupported audio rate:%u", impl->rate); + res = -EINVAL; + goto out; + } + impl->header.format_bit = impl->format_info->format_bit; + break; + case SPA_MEDIA_SUBTYPE_control: + impl->stream_info = impl->info; + impl->format_info = find_audio_format_info(&impl->info); + if (impl->format_info == NULL) { + res = -EINVAL; + goto out; + } + pw_properties_set(props, PW_KEY_FORMAT_DSP, "8 bit raw midi"); + impl->stride = impl->format_info->size; + impl->rate = pw_properties_get_uint32(props, "midi.rate", 10000); + if (impl->rate == 0) + impl->rate = 10000; + break; + default: + spa_assert_not_reached(); + break; + } + + if (pw_properties_get(props, PW_KEY_NODE_VIRTUAL) == NULL) + pw_properties_set(props, PW_KEY_NODE_VIRTUAL, "true"); + if (pw_properties_get(props, PW_KEY_NODE_NETWORK) == NULL) + pw_properties_set(props, PW_KEY_NODE_NETWORK, "true"); + + impl->mtu = pw_properties_get_uint32(props, "net.mtu", DEFAULT_MTU); + + str = pw_properties_get(props, "sess.min-ptime"); + if (!spa_atof(str, &min_ptime)) + min_ptime = DEFAULT_MIN_PTIME; + str = pw_properties_get(props, "sess.max-ptime"); + if (!spa_atof(str, &max_ptime)) + max_ptime = DEFAULT_MAX_PTIME; + + min_samples = min_ptime * impl->rate / 1000; + max_samples = SPA_MIN(256, max_ptime * impl->rate / 1000); + + float ptime = 0; + if ((str = pw_properties_get(props, "vban.ptime")) != NULL) + if (!spa_atof(str, &ptime)) + ptime = 0.0; + + if (ptime) { + impl->psamples = ptime * impl->rate / 1000; + } else { + impl->psamples = impl->mtu / impl->stride; + impl->psamples = SPA_CLAMP(impl->psamples, min_samples, max_samples); + if (direction == PW_DIRECTION_OUTPUT) + pw_properties_setf(props, "vban.ptime", "%f", + impl->psamples * 1000.0 / impl->rate); + } + latency_msec = pw_properties_get_uint32(props, + "sess.latency.msec", DEFAULT_SESS_LATENCY); + impl->target_buffer = msec_to_samples(impl, latency_msec); + impl->max_error = msec_to_samples(impl, ERROR_MSEC); + + pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", impl->rate); + if (direction == PW_DIRECTION_INPUT) { + pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d", + impl->psamples, impl->rate); + } else { + pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d", + impl->target_buffer / 2, impl->rate); + } + + pw_properties_setf(props, "net.mtu", "%u", impl->mtu); + pw_properties_setf(props, "vban.rate", "%u", impl->rate); + if (impl->info.info.raw.channels > 0) + pw_properties_setf(props, "vban.channels", "%u", impl->info.info.raw.channels); + + spa_dll_init(&impl->dll); + spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate); + impl->corr = 1.0; + + impl->stream = pw_stream_new(core, "vban-session", props); + props = NULL; + if (impl->stream == NULL) { + res = -errno; + pw_log_error("can't create stream: %m"); + goto out; + } + + n_params = 0; + spa_pod_builder_init(&b, buffer, sizeof(buffer)); + + flags = PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS; + + + switch (impl->info.media_subtype) { + case SPA_MEDIA_SUBTYPE_raw: + params[n_params++] = spa_format_audio_build(&b, + SPA_PARAM_EnumFormat, &impl->stream_info); + flags |= PW_STREAM_FLAG_AUTOCONNECT; + vban_audio_init(impl, direction); + break; + case SPA_MEDIA_SUBTYPE_control: + params[n_params++] = spa_pod_builder_add_object(&b, + SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat, + SPA_FORMAT_mediaType, SPA_POD_Id(SPA_MEDIA_TYPE_application), + SPA_FORMAT_mediaSubtype, SPA_POD_Id(SPA_MEDIA_SUBTYPE_control)); +// vban_midi_init(impl, direction); + break; + default: + res = -EINVAL; + goto out; + } + + pw_stream_add_listener(impl->stream, + &impl->stream_listener, + &impl->stream_events, impl); + + if ((res = pw_stream_connect(impl->stream, + direction, + PW_ID_ANY, + flags, + params, n_params)) < 0) { + pw_log_error("can't connect stream: %s", spa_strerror(res)); + goto out; + } + + if (impl->always_process && + (res = stream_start(impl)) < 0) + goto out; + + spa_hook_list_append(&impl->listener_list, &impl->listener, events, data); + + return (struct vban_stream*)impl; +out: + pw_properties_free(props); + errno = -res; + return NULL; +} + +void vban_stream_destroy(struct vban_stream *s) +{ + struct impl *impl = (struct impl*)s; + + vban_stream_emit_destroy(impl); + + if (impl->stream) + pw_stream_destroy(impl->stream); + + spa_hook_list_clean(&impl->listener_list); + free(impl); +} + +int vban_stream_receive_packet(struct vban_stream *s, uint8_t *buffer, size_t len) +{ + struct impl *impl = (struct impl*)s; + return impl->receive_vban(impl, buffer, len); +} + +uint64_t vban_stream_get_time(struct vban_stream *s, uint64_t *rate) +{ + struct impl *impl = (struct impl*)s; + struct spa_io_position *pos = impl->io_position; + + if (pos == NULL) + return -EIO; + + *rate = impl->rate; + return pos->clock.position * impl->rate * + pos->clock.rate.num / pos->clock.rate.denom; +} diff --git a/src/modules/module-vban/stream.h b/src/modules/module-vban/stream.h new file mode 100644 index 000000000..36125a8bf --- /dev/null +++ b/src/modules/module-vban/stream.h @@ -0,0 +1,54 @@ +/* PipeWire */ +/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans */ +/* SPDX-License-Identifier: MIT */ + +#ifndef PIPEWIRE_VBAN_STREAM_H +#define PIPEWIRE_VBAN_STREAM_H + +#ifdef __cplusplus +extern "C" { +#endif + +struct vban_stream; + +#define DEFAULT_FORMAT "S16LE" +#define DEFAULT_RATE 44100 +#define DEFAULT_CHANNELS 2 +#define DEFAULT_POSITION "[ FL FR ]" + +#define ERROR_MSEC 2 +#define DEFAULT_SESS_LATENCY 100 + +#define DEFAULT_MTU VBAN_PROTOCOL_MAX_SIZE +#define DEFAULT_MIN_PTIME 2 +#define DEFAULT_MAX_PTIME 20 + +struct vban_stream_events { +#define VBAN_VERSION_STREAM_EVENTS 0 + uint32_t version; + + void (*destroy) (void *data); + + void (*state_changed) (void *data, bool started, const char *error); + + void (*send_packet) (void *data, struct iovec *iov, size_t iovlen); + + void (*send_feedback) (void *data, uint32_t senum); +}; + +struct vban_stream *vban_stream_new(struct pw_core *core, + enum pw_direction direction, struct pw_properties *props, + const struct vban_stream_events *events, void *data); + +void vban_stream_destroy(struct vban_stream *s); + +int vban_stream_receive_packet(struct vban_stream *s, uint8_t *buffer, size_t len); + +uint64_t vban_stream_get_time(struct vban_stream *s, uint64_t *rate); + + +#ifdef __cplusplus +} +#endif + +#endif /* PIPEWIRE_VBAN_STREAM_H */ diff --git a/src/modules/module-vban/vban.h b/src/modules/module-vban/vban.h new file mode 100644 index 000000000..a0d12b1c4 --- /dev/null +++ b/src/modules/module-vban/vban.h @@ -0,0 +1,60 @@ +/* PipeWire */ +/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans */ +/* SPDX-License-Identifier: MIT */ + +#ifndef PIPEWIRE_VBAN_H +#define PIPEWIRE_VBAN_H + +#ifdef __cplusplus +extern "C" { +#endif + +#define VBAN_HEADER_SIZE (4 + 4 + 16 + 4) +#define VBAN_STREAM_NAME_SIZE 16 +#define VBAN_PROTOCOL_MAX_SIZE 1464 +#define VBAN_DATA_MAX_SIZE (VBAN_PROTOCOL_MAX_SIZE - VBAN_HEADER_SIZE) +#define VBAN_CHANNELS_MAX_NB 256 +#define VBAN_SAMPLES_MAX_NB 256 + +struct vban_header { + char vban[4]; /* contains 'V' 'B', 'A', 'N' */ + uint8_t format_SR; /* SR index */ + uint8_t format_nbs; /* nb sample per frame (1 to 256) */ + uint8_t format_nbc; /* nb channel (1 to 256) */ + uint8_t format_bit; /* bit format */ + char stream_name[VBAN_STREAM_NAME_SIZE]; /* stream name */ + uint32_t n_frames; /* growing frame number. */ +} __attribute__ ((packed)); + +#define VBAN_SR_MAXNUMBER 21 + +static uint32_t const vban_SR[VBAN_SR_MAXNUMBER] = { + 6000, 12000, 24000, 48000, 96000, 192000, 384000, + 8000, 16000, 32000, 64000, 128000, 256000, 512000, + 11025, 22050, 44100, 88200, 176400, 352800, 705600 +}; + +static inline uint8_t vban_sr_index(uint32_t rate) +{ + uint8_t i; + for (i = 0; i < SPA_N_ELEMENTS(vban_SR); i++) { + if (vban_SR[i] == rate) + return i; + } + return VBAN_SR_MAXNUMBER; +} + +#define VBAN_DATATYPE_U8 0x00 +#define VBAN_DATATYPE_INT16 0x01 +#define VBAN_DATATYPE_INT24 0x02 +#define VBAN_DATATYPE_INT32 0x03 +#define VBAN_DATATYPE_FLOAT32 0x04 +#define VBAN_DATATYPE_FLOAT64 0x05 +#define VBAN_DATATYPE_12BITS 0x06 +#define VBAN_DATATYPE_10BITS 0x07 + +#ifdef __cplusplus +} +#endif + +#endif /* PIPEWIRE_VBAN_H */