mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-03 09:01:54 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions
pipewire/spa/plugins/alsa/alsa-compress-offload-sink.c
Diego Viola 7410755c03 Fix typos 
found them with codespell.

Signed-off-by: Diego Viola <diego.viola@gmail.com>
2024-05-22 09:19:34 +02:00

2024 lines
59 KiB
C
Raw Blame History

/* Spa ALSA Compress-Offload sink */
/* SPDX-FileCopyrightText: Copyright © 2022 Wim Taymans */
/* SPDX-FileCopyrightText: Copyright © 2022 Asymptotic Inc. */
/* SPDX-FileCopyrightText: Copyright @ 2023 Carlos Rafael Giani */
/* SPDX-License-Identifier: MIT */
#include <assert.h>
#include <stddef.h>
#include <limits.h>
#include <linux/version.h>
#include <spa/monitor/device.h>
#include <spa/debug/types.h>
#include <spa/support/loop.h>
#include <spa/support/plugin.h>
#include <spa/support/system.h>
#include <spa/node/keys.h>
#include <spa/node/node.h>
#include <spa/node/utils.h>
#include <spa/param/audio/format.h>
#include <spa/param/audio/format-utils.h>
#include <spa/pod/filter.h>
#include <spa/pod/parser.h>
#include <spa/utils/defs.h>
#include <spa/utils/keys.h>
#include <spa/utils/names.h>
#include <spa/utils/result.h>
#include <spa/utils/string.h>
#include "alsa.h"
#include "compress-offload-api.h"
/*
* This creates a PipeWire sink node which uses the ALSA Compress-Offload API
* for writing compressed data ike MP3, FLAC etc. to an DSP that can handle
* such data directly.
*
* These show up under /dev/snd like "comprCxDx", as opposed to regular
* ALSA PCM devices. This sink node still refers to those devices in
* regular ALSA fashion as "hw:x,y" devices, where x = card number and
* y = device number. For example, "hw:4,7" maps to /dev/snd/comprC4D7.
*
* ## Example configuration
*\code{.unparsed}
* context.objects = [
* { factory = adapter
* args = {
* factory.name = "api.alsa.compress.offload.sink"
* node.name = "Compress-Offload-Sink"
* node.description = "Audio sink for compressed audio"
* media.class = "Audio/Sink"
* api.alsa.path = "hw:0,3"
* node.param.PortConfig = {
* direction = Input
* mode = passthrough
* }
* }
* }
*]
*\endcode
*
* TODO:
* - DLL for adjusting driver timer intervals to match the device timestamps in on_driver_timeout()
* - Automatic loading using alsa-udev
*/
/* FLAC support has been present in kernel headers older than 5.5.
* However, those older versions don't support FLAC decoding params. */
#if LINUX_VERSION_CODE >= KERNEL_VERSION(5, 5, 0)
#define COMPRESS_OFFLOAD_HAS_FLAC_DEC_PARAMS
#endif
/* Prior to kernel 5.7, WMA9 Pro/Lossless and WMA10 Lossless
* codec profiles were missing.
* As for ALAC and Monkey's Audio (APE), those are new in 5.7. */
#if LINUX_VERSION_CODE >= KERNEL_VERSION(5, 7, 0)
#define COMPRESS_OFFLOAD_SUPPORTS_WMA9_PRO
#define COMPRESS_OFFLOAD_SUPPORTS_WMA9_LOSSLESS
#define COMPRESS_OFFLOAD_SUPPORTS_WMA10_LOSSLESS
#define COMPRESS_OFFLOAD_SUPPORTS_ALAC
#define COMPRESS_OFFLOAD_SUPPORTS_APE
#endif
#define CHECK_PORT(this, d, p) (((d) == SPA_DIRECTION_INPUT) && ((p) == 0))
#define MAX_BUFFERS (32)
#define BUFFER_FLAG_AVAILABLE_FOR_NEW_DATA (1 << 0)
/* Information about a buffer that got allocated by the PW graph. */
struct buffer {
uint32_t id;
uint32_t flags;
struct spa_buffer *buf;
struct spa_list link;
};
/* Node properties. These are accessible through SPA_PARAM_Props. */
struct props {
/* The'"hw:<cardnr>:<devicenr>" device. */
char device[128];
/* These are the card and device numbers from the
* from the "hw:<cardnr>:<devicenr>" device.*/
int card_nr;
int device_nr;
bool device_name_set;
};
/* Main sink node structure. */
struct impl {
/* Basic states */
struct spa_handle handle;
struct spa_node node;
struct spa_log *log;
struct spa_loop *data_loop;
struct spa_system *data_system;
struct spa_hook_list hooks;
struct spa_callbacks callbacks;
struct props props;
bool have_format;
struct spa_audio_info current_audio_info;
/* This is set to true when the SPA_NODE_COMMAND_Start is
* received, and set back to false when SPA_NODE_COMMAND_Pause
* or SPA_NODE_COMMAND_Suspend is received. */
bool started;
bool freewheel;
/* SPA buffer states */
struct buffer buffers[MAX_BUFFERS];
unsigned int n_buffers;
struct spa_list queued_output_buffers;
size_t offset_within_oldest_output_buffer;
/* Driver and cycle specific states */
int driver_timerfd;
struct spa_source driver_timer_source;
uint64_t next_driver_time;
bool following;
/* Duration and rate of one graph cycle.
* The duration equals the quantum size. */
uint32_t cycle_duration;
int cycle_rate;
/* Node specific states */
uint64_t node_info_all;
struct spa_node_info node_info;
#define NODE_PropInfo 0
#define NODE_Props 1
#define NODE_IO 2
#define NODE_EnumPortConfig 3
#define N_NODE_PARAMS 4
struct spa_param_info node_params[N_NODE_PARAMS];
struct spa_io_clock *node_clock_io;
struct spa_io_position *node_position_io;
/* Port specific states */
uint64_t port_info_all;
struct spa_port_info port_info;
#define PORT_EnumFormat 0
#define PORT_Format 1
#define PORT_IO 2
#define PORT_Buffers 3
#define N_PORT_PARAMS 4
struct spa_param_info port_params[N_PORT_PARAMS];
struct spa_io_buffers *port_buffers_io;
/* Compress-Offload specific states */
struct compress_offload_api_context *device_context;
struct snd_codec audio_codec_info;
bool device_started;
uint32_t min_fragment_size;
uint32_t max_fragment_size;
uint32_t min_num_fragments;
uint32_t max_num_fragments;
uint32_t configured_fragment_size;
uint32_t configured_num_fragments;
bool device_is_paused;
};
/* Compress-Offload device and audio codec functions */
static int init_audio_codec_info(struct impl *this, struct spa_audio_info *info, uint32_t *out_rate);
static int device_open(struct impl *this);
static void device_close(struct impl *this);
static int device_start(struct impl *this);
static int device_pause(struct impl *this);
static int device_resume(struct impl *this);
static int device_write(struct impl *this, const void *data, uint32_t size);
/* Driver timer functions */
static int set_driver_timeout(struct impl *this, uint64_t time);
static int configure_driver_timer(struct impl *this);
static int start_driver_timer(struct impl *this);
static void stop_driver_timer(struct impl *this);
static void on_driver_timeout(struct spa_source *source);
static inline void check_position_and_clock_config(struct impl *this);
static void reevaluate_following_state(struct impl *this);
static void reevaluate_freewheel_state(struct impl *this);
/* Miscellaneous functions */
static int parse_device(struct impl *this);
static void reset_props(struct props *props);
static void clear_buffers(struct impl *this);
static inline bool is_following(struct impl *this);
static int do_start(struct impl *this);
static void do_stop(struct impl *this);
static int write_queued_output_buffers(struct impl *this);
/* Node and port functions */
static void emit_node_info(struct impl *this, bool full);
static void emit_port_info(struct impl *this, bool full);
static int impl_node_add_listener(void *object,
struct spa_hook *listener,
const struct spa_node_events *events,
void *data);
static int impl_node_set_callbacks(void *object,
const struct spa_node_callbacks *callbacks,
void *data);
static int impl_node_sync(void *object, int seq);
static int impl_node_enum_params(void *object, int seq,
uint32_t id, uint32_t start, uint32_t num,
const struct spa_pod *filter);
static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
const struct spa_pod *param);
static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size);
static int impl_node_send_command(void *object, const struct spa_command *command);
static int impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
const struct spa_dict *props);
static int impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_id);
static int port_enum_formats(struct impl *this, int seq, uint32_t start, uint32_t num,
const struct spa_pod *filter, struct spa_pod_builder *b);
static int impl_port_enum_params(void *object, int seq,
enum spa_direction direction, uint32_t port_id,
uint32_t id, uint32_t start, uint32_t num,
const struct spa_pod *filter);
static int port_set_format(void *object,
enum spa_direction direction, uint32_t port_id,
uint32_t flags,
const struct spa_pod *format);
static int impl_port_set_param(void *object,
enum spa_direction direction, uint32_t port_id,
uint32_t id, uint32_t flags,
const struct spa_pod *param);
static int impl_port_use_buffers(void *object,
enum spa_direction direction, uint32_t port_id,
uint32_t flags,
struct spa_buffer **buffers, uint32_t n_buffers);
static int impl_port_set_io(void *object,
enum spa_direction direction,
uint32_t port_id,
uint32_t id,
void *data, size_t size);
static int impl_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id);
static int impl_node_process(void *object);
/* Compress-Offload device and audio codec functions */
struct known_codec_info {
uint32_t codec_id;
uint32_t media_subtype;
const char *name;
};
static struct known_codec_info known_codecs[] = {
{ SND_AUDIOCODEC_VORBIS, SPA_MEDIA_SUBTYPE_vorbis, "Ogg Vorbis" },
{ SND_AUDIOCODEC_MP3, SPA_MEDIA_SUBTYPE_mp3, "MP3" },
{ SND_AUDIOCODEC_AAC, SPA_MEDIA_SUBTYPE_aac, "AAC" },
{ SND_AUDIOCODEC_FLAC, SPA_MEDIA_SUBTYPE_flac, "FLAC" },
{ SND_AUDIOCODEC_WMA, SPA_MEDIA_SUBTYPE_wma, "WMA" },
#ifdef COMPRESS_OFFLOAD_SUPPORTS_ALAC
{ SND_AUDIOCODEC_ALAC, SPA_MEDIA_SUBTYPE_alac, "ALAC" },
#endif
#ifdef COMPRESS_OFFLOAD_SUPPORTS_APE
{ SND_AUDIOCODEC_APE, SPA_MEDIA_SUBTYPE_ape, "Monkey's Audio (APE)" },
#endif
{ SND_AUDIOCODEC_REAL, SPA_MEDIA_SUBTYPE_ra, "Real Audio" },
{ SND_AUDIOCODEC_AMRWB, SPA_MEDIA_SUBTYPE_amr, "AMR wideband" },
{ SND_AUDIOCODEC_AMR, SPA_MEDIA_SUBTYPE_amr, "AMR" },
};
static int init_audio_codec_info(struct impl *this, struct spa_audio_info *info, uint32_t *out_rate)
{
struct snd_codec *codec;
uint32_t channels, rate;
const struct spa_type_info *media_subtype_info;
media_subtype_info = spa_debug_type_find(spa_type_media_subtype, info->media_subtype);
if (media_subtype_info == NULL) {
spa_log_error(this->log, "%p: media subtype %d is unknown", this, info->media_subtype);
return -ENOTSUP;
}
memset(&this->audio_codec_info, 0, sizeof(this->audio_codec_info));
codec = &this->audio_codec_info;
switch (info->media_subtype) {
case SPA_MEDIA_SUBTYPE_vorbis:
codec->id = SND_AUDIOCODEC_VORBIS;
rate = info->info.vorbis.rate;
channels = info->info.vorbis.channels;
spa_log_info(this->log, "%p: initialized codec info to Vorbis; rate: %"
PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
case SPA_MEDIA_SUBTYPE_mp3:
codec->id = SND_AUDIOCODEC_MP3;
rate = info->info.mp3.rate;
channels = info->info.mp3.channels;
spa_log_info(this->log, "%p: initialized codec info to MP3; rate: %"
PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
case SPA_MEDIA_SUBTYPE_aac:
codec->id = SND_AUDIOCODEC_AAC;
rate = info->info.aac.rate;
channels = info->info.aac.channels;
spa_log_info(this->log, "%p: initialized codec info to AAC; rate: %"
PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
case SPA_MEDIA_SUBTYPE_flac:
codec->id = SND_AUDIOCODEC_FLAC;
#ifdef COMPRESS_OFFLOAD_HAS_FLAC_DEC_PARAMS
/* The min/max block sizes are from the FLAC specification:
* https://xiph.org/flac/format.html#blocking
*
* The smallest valid frame possible is 11, which
* is why min_frame_size is set to this quantity.
*
* FFmpeg's flac.h specifies 8192 as the average frame size.
* tinycompress' fcplay uses 4x that amount as the max frame
* size to have enough headroom to be safe.
* We do the same here.
*
* sample_size is set to 0. According to the FLAC spec, this
* is OK to do if a STREAMINFO block was sent into the device
* (see: https://xiph.org/flac/format.html#frame_header), and
* we deal with full FLAC streams here, not just single frames. */
codec->options.flac_d.min_blk_size = 16;
codec->options.flac_d.max_blk_size = 65535;
codec->options.flac_d.min_frame_size = 11;
codec->options.flac_d.max_frame_size = 8192 * 4;
codec->options.flac_d.sample_size = 0;
#endif
rate = info->info.flac.rate;
channels = info->info.flac.channels;
spa_log_info(this->log, "%p: initialized codec info to FLAC; rate: %"
PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
case SPA_MEDIA_SUBTYPE_wma: {
const char *profile_name;
codec->id = SND_AUDIOCODEC_WMA;
/* WMA does not work with Compress-Offload
* if codec profile is not set. */
switch (info->info.wma.profile) {
case SPA_AUDIO_WMA_PROFILE_WMA9:
codec->profile = SND_AUDIOPROFILE_WMA9;
profile_name = "WMA9";
break;
case SPA_AUDIO_WMA_PROFILE_WMA10:
codec->profile = SND_AUDIOPROFILE_WMA10;
profile_name = "WMA10";
break;
#ifdef COMPRESS_OFFLOAD_SUPPORTS_WMA9_PRO
case SPA_AUDIO_WMA_PROFILE_WMA9_PRO:
codec->profile = SND_AUDIOPROFILE_WMA9_PRO;
profile_name = "WMA9 Pro";
break;
#endif
#ifdef COMPRESS_OFFLOAD_SUPPORTS_WMA9_LOSSLESS
case SPA_AUDIO_WMA_PROFILE_WMA9_LOSSLESS:
codec->profile = SND_AUDIOPROFILE_WMA9_LOSSLESS;
profile_name = "WMA9 Lossless";
break;
#endif
#ifdef COMPRESS_OFFLOAD_SUPPORTS_WMA10_LOSSLESS
case SPA_AUDIO_WMA_PROFILE_WMA10_LOSSLESS:
codec->profile = SND_AUDIOPROFILE_WMA10_LOSSLESS;
profile_name = "WMA10 Lossless";
break;
#endif
default:
spa_log_error(this->log, "%p: Invalid WMA profile", this);
return -EINVAL;
}
codec->bit_rate = info->info.wma.bitrate;
codec->align = info->info.wma.block_align;
rate = info->info.wma.rate;
channels = info->info.wma.channels;
spa_log_info(this->log, "%p: initialized codec info to WMA; rate: %"
PRIu32 "; channels: %" PRIu32 "; profile %s", this,
rate, channels, profile_name);
break;
}
#ifdef COMPRESS_OFFLOAD_SUPPORTS_ALAC
case SPA_MEDIA_SUBTYPE_alac:
codec->id = SND_AUDIOCODEC_ALAC;
rate = info->info.alac.rate;
channels = info->info.alac.channels;
spa_log_info(this->log, "%p: initialized codec info to ALAC; rate: %"
PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
#endif
#ifdef COMPRESS_OFFLOAD_SUPPORTS_APE
case SPA_MEDIA_SUBTYPE_ape:
codec->id = SND_AUDIOCODEC_APE;
rate = info->info.ape.rate;
channels = info->info.ape.channels;
spa_log_info(this->log, "%p: initialized codec info to APE (Monkey's Audio);"
" rate: %" PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
#endif
case SPA_MEDIA_SUBTYPE_ra:
codec->id = SND_AUDIOCODEC_REAL;
rate = info->info.ra.rate;
channels = info->info.ra.channels;
spa_log_info(this->log, "%p: initialized codec info to Real Audio; rate: %"
PRIu32 "; channels: %" PRIu32, this, rate, channels);
break;
case SPA_MEDIA_SUBTYPE_amr:
if (info->info.amr.band_mode == SPA_AUDIO_AMR_BAND_MODE_WB)
codec->id = SND_AUDIOCODEC_AMRWB;
else
codec->id = SND_AUDIOCODEC_AMR;
rate = info->info.amr.rate;
channels = info->info.amr.channels;
spa_log_info(this->log, "%p: initialized codec info to %s; rate: %"
PRIu32 "; channels: %" PRIu32, this,
(codec->id == SND_AUDIOCODEC_AMRWB) ? "AMR wideband" : "AMR",
rate, channels);
break;
default:
spa_log_error(this->log, "%p: media subtype %s is not supported", this, media_subtype_info->name);
return -ENOTSUP;
}
codec->ch_in = channels;
codec->ch_out = channels;
codec->sample_rate = rate;
codec->rate_control = 0;
codec->level = 0;
codec->ch_mode = 0;
codec->format = 0;
*out_rate = rate;
return 0;
}
static int device_open(struct impl *this)
{
assert(this->device_context == NULL);
spa_log_info(this->log, "%p: opening Compress-Offload device, card #%d device #%d",
this, this->props.card_nr, this->props.device_nr);
this->device_context = compress_offload_api_open(this->props.card_nr, this->props.device_nr, this->log);
if (this->device_context == NULL)
return -errno;
return 0;
}
static void device_close(struct impl *this)
{
if (this->device_context == NULL)
return;
spa_log_info(this->log, "%p: closing Compress-Offload device, card #%d device #%d",
this, this->props.card_nr, this->props.device_nr);
if (this->device_started)
compress_offload_api_stop(this->device_context);
compress_offload_api_close(this->device_context);
this->device_context = NULL;
this->device_started = false;
this->device_is_paused = false;
this->have_format = false;
}
static int device_start(struct impl *this)
{
assert(this->device_context != NULL);
if (compress_offload_api_start(this->device_context) < 0)
return -errno;
this->device_started = true;
return 0;
}
static int device_pause(struct impl *this)
{
/* device_pause() can sometimes be called when the device context is already
* gone. In particular, this can happen when the suspend command is received
* after the pause command. */
if (this->device_context == NULL)
return 0;
if (this->device_is_paused)
return 0;
if (compress_offload_api_pause(this->device_context) < 0)
return -errno;
this->device_is_paused = true;
return 0;
}
static int device_resume(struct impl *this)
{
assert(this->device_context != NULL);
if (!this->device_is_paused)
return 0;
if (compress_offload_api_resume(this->device_context) < 0)
return -errno;
this->device_is_paused = false;
return 0;
}
static int device_write(struct impl *this, const void *data, uint32_t size)
{
int res;
uint32_t num_bytes_to_write;
struct snd_compr_avail available_space;
/* In here, try to write out as much data as possible,
* in a non-blocking manner, retaining the unwritten
* data for the next write call. */
if (SPA_UNLIKELY((res = compress_offload_api_get_available_space(
this->device_context, &available_space)) < 0))
return res;
/* We can only write data if there is at least enough space for one
* fragment's worth of data, or if the data we want to write is
* small (smaller than a fragment). The latter can happen when we
* are writing the last few bits of the compressed audio medium.
* When the former happens, we try to write as much data as we
* can, limited by the amount of space available in the device. */
if ((available_space.avail < this->min_fragment_size) && (available_space.avail < size))
return 0;
num_bytes_to_write = SPA_MIN(size, available_space.avail);
res = compress_offload_api_write(this->device_context, data, num_bytes_to_write);
if (SPA_UNLIKELY(res < 0)) {
if (res == -EBADFD)
spa_log_debug(this->log, "%p: device is paused", this);
else
spa_log_error(this->log, "%p: write error: %s", this, spa_strerror(res));
return res;
}
spa_log_trace_fp(this->log, "%p: wrote %d bytes; original request: %" PRIu32 "; adjusted "
"for available space in device: %" PRIu32, this, res, size, num_bytes_to_write);
if (SPA_UNLIKELY(((uint32_t)res) > num_bytes_to_write)) {
spa_log_error(this->log, "%p: wrote more bytes than what was requested; "
"requested: %" PRId32 " wrote: %d", this, num_bytes_to_write, res);
return -EIO;
}
return res;
}
/* Driver timer functions */
static int set_driver_timeout(struct impl *this, uint64_t time)
{
struct itimerspec ts;
ts.it_value.tv_sec = time / SPA_NSEC_PER_SEC;
ts.it_value.tv_nsec = time % SPA_NSEC_PER_SEC;
ts.it_interval.tv_sec = 0;
ts.it_interval.tv_nsec = 0;
spa_system_timerfd_settime(this->data_system, this->driver_timerfd,
SPA_FD_TIMER_ABSTIME, &ts, NULL);
return 0;
}
static int configure_driver_timer(struct impl *this)
{
struct timespec now;
int res;
if ((res = spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &now)) < 0) {
spa_log_error(this->log, "%p: could not get time from monotonic sysclock: %s",
this, spa_strerror(res));
return res;
}
this->next_driver_time = SPA_TIMESPEC_TO_NSEC(&now);
if (this->following)
set_driver_timeout(this, 0);
else
set_driver_timeout(this, this->next_driver_time);
return 0;
}
static int start_driver_timer(struct impl *this)
{
int res;
spa_log_debug(this->log, "%p: starting driver timer", this);
this->driver_timer_source.func = on_driver_timeout;
this->driver_timer_source.data = this;
this->driver_timer_source.fd = this->driver_timerfd;
this->driver_timer_source.mask = SPA_IO_IN;
this->driver_timer_source.rmask = 0;
spa_loop_add_source(this->data_loop, &this->driver_timer_source);
if (SPA_UNLIKELY((res = configure_driver_timer(this)) < 0))
return res;
return 0;
}
static int do_remove_driver_timer_source(struct spa_loop *loop,
bool async,
uint32_t seq,
const void *data,
size_t size,
void *user_data)
{
struct impl *this = user_data;
spa_loop_remove_source(this->data_loop, &this->driver_timer_source);
set_driver_timeout(this, 0);
return 0;
}
static void stop_driver_timer(struct impl *this)
{
spa_log_debug(this->log, "%p: stopping driver timer", this);
/* Perform the actual stop within
* the dataloop to avoid data races. */
spa_loop_invoke(this->data_loop, do_remove_driver_timer_source, 0, NULL, 0, true, this);
}
static void on_driver_timeout(struct spa_source *source)
{
struct impl *this = source->data;
uint64_t expire, current_time;
int res;
if (SPA_LIKELY(this->started)) {
if (SPA_UNLIKELY((res = spa_system_timerfd_read(this->data_system,
this->driver_timerfd, &expire)) < 0)) {
if (res != -EAGAIN)
spa_log_warn(this->log, "%p: error reading from timerfd: %s",
this, spa_strerror(res));
return;
}
}
if (SPA_LIKELY(this->node_position_io != NULL)) {
this->cycle_duration = this->node_position_io->clock.target_duration;
this->cycle_rate = this->node_position_io->clock.target_rate.denom;
} else {
/* This can happen at the very beginning if node_position_io
* isn't passed to this node in time. */
this->cycle_duration = 1024;
this->cycle_rate = 48000;
}
current_time = this->next_driver_time;
this->next_driver_time += ((uint64_t)(this->cycle_duration)) * 1000000000ULL / this->cycle_rate;
if (this->node_clock_io != NULL) {
this->node_clock_io->nsec = current_time;
this->node_clock_io->rate = this->node_clock_io->target_rate;
this->node_clock_io->position += this->node_clock_io->duration;
this->node_clock_io->duration = this->cycle_duration;
this->node_clock_io->delay = 0;
this->node_clock_io->rate_diff = 1.0;
this->node_clock_io->next_nsec = this->next_driver_time;
spa_log_trace_fp(this->log, "%p: clock IO updated to: nsec %" PRIu64
" pos %" PRIu64 " dur %" PRIu64 " next-nsec %" PRIu64, this,
this->node_clock_io->nsec, this->node_clock_io->position,
this->node_clock_io->duration, this->node_clock_io->next_nsec);
}
/* Adapt the graph cycle progression to the needs of the sink.
* If the sink still has data to output, don't advance. */
if (spa_list_is_empty(&this->queued_output_buffers)) {
struct spa_io_buffers *io = this->port_buffers_io;
if (SPA_LIKELY(io != NULL)) {
spa_log_trace_fp(this->log, "%p: ran out of buffers to output, "
"need more; IO status: %d", this, io->status);
io->status = SPA_STATUS_NEED_DATA;
spa_node_call_ready(&this->callbacks, SPA_STATUS_NEED_DATA);
} else {
/* This should not happen. If it does, then there may be
* an error in when the timer is stopped. When it happens,
* do not schedule a next timeout. */
spa_log_warn(this->log, "%p: buffers IO was set to NULL before "
"the driver timer was stopped", this);
set_driver_timeout(this, 0);
return;
}
} else {
write_queued_output_buffers(this);
}
// TODO check for impossible timeouts (only relevant when taking device timestamps into account)
set_driver_timeout(this, this->next_driver_time);
}
static inline void check_position_and_clock_config(struct impl *this)
{
if (SPA_LIKELY(this->node_position_io != NULL)) {
this->cycle_duration = this->node_position_io->clock.duration;
this->cycle_rate = this->node_position_io->clock.rate.denom;
} else {
/* This can happen at the very beginning if node_position_io
* isn't passed to this node in time. */
this->cycle_duration = 1024;
this->cycle_rate = 48000;
}
}
static int do_reevaluate_following_state(struct spa_loop *loop,
bool async,
uint32_t seq,
const void *data,
size_t size,
void *user_data)
{
struct impl *this = user_data;
configure_driver_timer(this);
return 0;
}
static void reevaluate_following_state(struct impl *this)
{
bool following;
if (!this->started)
return;
following = is_following(this);
if (following != this->following) {
spa_log_debug(this->log, "%p: following state changed: %d->%d", this, this->following, following);
this->following = following;
spa_loop_invoke(this->data_loop, do_reevaluate_following_state, 0, NULL, 0, true, this);
}
}
static void reevaluate_freewheel_state(struct impl *this)
{
bool freewheel;
if (!this->started)
return;
freewheel = (this->node_position_io != NULL) &&
SPA_FLAG_IS_SET(this->node_position_io->clock.flags, SPA_IO_CLOCK_FLAG_FREEWHEEL);
if (this->freewheel != freewheel) {
spa_log_debug(this->log, "%p: freewheel state changed: %d->%d", this, this->freewheel, freewheel);
this->freewheel = freewheel;
if (freewheel)
device_pause(this);
else
device_resume(this);
}
}
/* Miscellaneous functions */
static int parse_device(struct impl *this)
{
char *device;
char *nextptr;
#define NUM_DEVICE_VALUES (2)
long values[NUM_DEVICE_VALUES];
int value_index;
device = this->props.device;
/* Valid devices always match the "hw:<cardnr>,<devicenr>" pattern. */
if (strncmp(device, "hw:", 3) != 0) {
spa_log_error(this->log, "%p: device \"%s\" does not begin with \"hw:\"", this, device);
return -EINVAL;
}
nextptr = device + 3;
for (value_index = 0; ; ++value_index) {
const char *value_label;
switch (value_index) {
case 0: value_label = "card"; break;
case 1: value_label = "device"; break;
default: spa_assert_not_reached();
}
errno = 0;
values[value_index] = strtol(nextptr, &nextptr, 10);
if (errno != 0) {
spa_log_error(this->log, "%p: device \"%s\" has invalid %s value",
this, device, value_label);
return -EINVAL;
}
if (values[value_index] < 0) {
spa_log_error(this->log, "%p: device \"%s\" has negative %s value",
this, device, value_label);
return -EINVAL;
}
if (values[value_index] > INT_MAX) {
spa_log_error(this->log, "%p: device \"%s\" has %s value larger than %d",
this, device, value_label, INT_MAX);
return -EINVAL;
}
if (value_index >= (NUM_DEVICE_VALUES - 1))
break;
if ((*nextptr) != ',') {
spa_log_error(this->log, "%p: expected ',' separator between numbers in "
"device \"%s\", got '%c'", this, device, *nextptr);
return -EINVAL;
}
/* Skip the comma between the values. */
nextptr++;
}
this->props.card_nr = values[0];
this->props.device_nr = values[1];
return 0;
}
static void reset_props(struct props *props)
{
memset(props->device, 0, sizeof(props->device));
props->card_nr = 0;
props->device_nr = 0;
props->device_name_set = false;
}
static void clear_buffers(struct impl *this)
{
if (this->n_buffers > 0) {
spa_log_debug(this->log, "%p: clearing buffers", this);
spa_list_init(&this->queued_output_buffers);
this->n_buffers = 0;
}
}
static inline bool is_following(struct impl *this)
{
return (this->node_position_io != NULL) &&
(this->node_clock_io != NULL) &&
(this->node_position_io->clock.id != this->node_clock_io->id);
}
static int do_start(struct impl *this)
{
int res;
if (this->started)
return 0;
this->following = is_following(this);
spa_log_debug(this->log, "%p: starting output; starting as follower: %d",
this, this->following);
if (SPA_UNLIKELY((res = start_driver_timer(this)) < 0))
return res;
this->started = true;
/* Not starting the compress-offload device here right away.
* That's because we first need to give it at least one
* fragment's worth of data. Starting the device prior to
* that results in buffer underflows inside the device. */
return 0;
}
static void do_stop(struct impl *this)
{
if (!this->started)
return;
spa_log_debug(this->log, "%p: stopping output", this);
device_pause(this);
this->started = false;
stop_driver_timer(this);
}
static int write_queued_output_buffers(struct impl *this)
{
int res;
uint32_t total_num_written_bytes;
bool wrote_data = false;
check_position_and_clock_config(this);
/* In here, we write as much data as possible. The device may
* initially not have sufficient space, but it is possible
* that due to ongoing data consumption, it can accommodate
* for more data in a next attempt, hence the "again" label.
*
* If during the write attempts, only a portion of a chunk
* is written, we must keep track of the portion that hasn't
* been consumed yet. offset_within_oldest_output_buffer
* exists for this purpose. In this sink node, each SPA
* buffer has exactly one chunk, so when a chunk is fully
* consumed, the corresponding buffer is removed from the
* queued_output_buffers list, marked as available, and
* returned to the pool through spa_node_call_reuse_buffer(). */
again:
total_num_written_bytes = 0;
while (!spa_list_is_empty(&this->queued_output_buffers)) {
struct buffer *b;
struct spa_data *d;
uint32_t chunk_start_offset, chunk_size, pending_data_size;
bool reuse_buffer = false;
b = spa_list_first(&this->queued_output_buffers, struct buffer, link);
d = b->buf->datas;
assert(b->buf->n_datas >= 1);
chunk_start_offset = d[0].chunk->offset + this->offset_within_oldest_output_buffer;
chunk_size = d[0].chunk->size;
/* An empty chunk signals that the source is skipping this cycle. This
* is normal and necessary in cases when the compressed data frames are
* longer than the quantum size. The source then has to keep track of
* the excess lengths, and if these sum up to the length of a quantum,
* it sends a buffer with an empty chunk to compensate. If this is not
* done, there will eventually be an overflow, this sink will miss
* cycles, and audible errors will occur. */
if (chunk_size != 0) {
int num_written_bytes;
pending_data_size = chunk_size - this->offset_within_oldest_output_buffer;
chunk_start_offset = SPA_MIN(chunk_start_offset, d[0].maxsize);
pending_data_size = SPA_MIN(pending_data_size, d[0].maxsize - chunk_start_offset);
num_written_bytes = device_write(this, SPA_PTROFF(d[0].data, chunk_start_offset, void), pending_data_size);
if (SPA_UNLIKELY(num_written_bytes < 0))
return num_written_bytes;
if (num_written_bytes == 0)
break;
this->offset_within_oldest_output_buffer += num_written_bytes;
total_num_written_bytes += num_written_bytes;
wrote_data = wrote_data || (num_written_bytes > 0);
if (this->offset_within_oldest_output_buffer >= chunk_size) {
spa_log_trace_fp(this->log, "%p: buffer with ID %u was fully written; reusing this buffer", this, b->id);
reuse_buffer = true;
this->offset_within_oldest_output_buffer = 0;
}
} else {
spa_log_trace_fp(this->log, "%p: buffer with ID %u has empty chunk; reusing this buffer", this, b->id);
reuse_buffer = true;
}
if (reuse_buffer) {
spa_list_remove(&b->link);
SPA_FLAG_SET(b->flags, BUFFER_FLAG_AVAILABLE_FOR_NEW_DATA);
this->port_buffers_io->buffer_id = b->id;
spa_node_call_reuse_buffer(&this->callbacks, 0, b->id);
}
}
if (!spa_list_is_empty(&this->queued_output_buffers) && (total_num_written_bytes > 0))
goto again;
/* We start the device only after having written data to avoid
* underruns due to an under-populated device ringbuffer. */
if (wrote_data && !this->device_started) {
spa_log_debug(this->log, "%p: starting device", this);
if ((res = device_start(this)) < 0) {
spa_log_error(this->log, "%p: starting device failed: %s", this, spa_strerror(res));
return res;
}
this->device_started = true;
}
return 0;
}
/* Node and port functions */
static const struct spa_dict_item node_info_items[] = {
{ SPA_KEY_DEVICE_API, "alsa" },
{ SPA_KEY_MEDIA_CLASS, "Audio/Sink" },
{ SPA_KEY_NODE_DRIVER, "true" },
{ SPA_KEY_NODE_PAUSE_ON_IDLE, "true" },
};
static void emit_node_info(struct impl *this, bool full)
{
uint64_t old = full ? this->node_info.change_mask : 0;
if (full)
this->node_info.change_mask = this->node_info_all;
if (this->node_info.change_mask) {
this->node_info.props = &SPA_DICT_INIT_ARRAY(node_info_items);
spa_node_emit_info(&this->hooks, &this->node_info);
this->node_info.change_mask = old;
}
}
static void emit_port_info(struct impl *this, bool full)
{
uint64_t old = full ? this->port_info.change_mask : 0;
if (full)
this->port_info.change_mask = this->port_info_all;
if (this->port_info.change_mask) {
spa_node_emit_port_info(&this->hooks,
SPA_DIRECTION_INPUT, 0, &this->port_info);
this->port_info.change_mask = old;
}
}
static int impl_node_add_listener(void *object,
struct spa_hook *listener,
const struct spa_node_events *events,
void *data)
{
struct impl *this = object;
struct spa_hook_list save;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_hook_list_isolate(&this->hooks, &save, listener, events, data);
emit_node_info(this, true);
emit_port_info(this, true);
spa_hook_list_join(&this->hooks, &save);
return 0;
}
static int impl_node_set_callbacks(void *object,
const struct spa_node_callbacks *callbacks,
void *data)
{
struct impl *this = object;
spa_return_val_if_fail(this != NULL, -EINVAL);
this->callbacks = SPA_CALLBACKS_INIT(callbacks, data);
return 0;
}
static int impl_node_sync(void *object, int seq)
{
struct impl *this = object;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_node_emit_result(&this->hooks, seq, 0, 0, NULL);
return 0;
}
static int impl_node_enum_params(void *object, int seq,
uint32_t id, uint32_t start, uint32_t num,
const struct spa_pod *filter)
{
struct impl *this = object;
struct spa_pod *param;
struct spa_pod_builder b = { 0 };
uint8_t buffer[4096];
struct spa_result_node_params result;
uint32_t count = 0;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(num != 0, -EINVAL);
result.id = id;
result.next = start;
next:
result.index = result.next++;
spa_pod_builder_init(&b, buffer, sizeof(buffer));
switch (id) {
case SPA_PARAM_PropInfo:
{
struct props *p = &this->props;
switch (result.index) {
case 0:
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_PropInfo, id,
SPA_PROP_INFO_id, SPA_POD_Id(SPA_PROP_device),
SPA_PROP_INFO_name, SPA_POD_String(SPA_KEY_API_ALSA_PATH),
SPA_PROP_INFO_description, SPA_POD_String("The ALSA Compress-Offload device"),
SPA_PROP_INFO_type, SPA_POD_Stringn(p->device, sizeof(p->device)));
break;
default:
return 0;
}
break;
}
case SPA_PARAM_Props:
{
struct props *p = &this->props;
switch (result.index) {
case 0:
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_Props, id,
SPA_PROP_device, SPA_POD_Stringn(p->device, sizeof(p->device))
);
break;
default:
return 0;
}
break;
}
case SPA_PARAM_IO:
switch (result.index) {
case 0:
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_ParamIO, id,
SPA_PARAM_IO_id, SPA_POD_Id(SPA_IO_Clock),
SPA_PARAM_IO_size, SPA_POD_Int(sizeof(struct spa_io_clock)));
break;
case 1:
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_ParamIO, id,
SPA_PARAM_IO_id, SPA_POD_Id(SPA_IO_Position),
SPA_PARAM_IO_size, SPA_POD_Int(sizeof(struct spa_io_position)));
break;
default:
return 0;
}
break;
case SPA_PARAM_EnumPortConfig:
{
switch (result.index) {
case 0:
/* Force ports to be configured to run in passthrough mode.
* This is essential when dealing with compressed data. */
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_ParamPortConfig, id,
SPA_PARAM_PORT_CONFIG_direction, SPA_POD_Id(SPA_DIRECTION_INPUT),
SPA_PARAM_PORT_CONFIG_mode, SPA_POD_Id(SPA_PARAM_PORT_CONFIG_MODE_passthrough)
);
break;
default:
return 0;
}
break;
}
default:
return -ENOENT;
}
if (spa_pod_filter(&b, &result.param, param, filter) < 0)
goto next;
spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
if (++count != num)
goto next;
return 0;
}
static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
const struct spa_pod *param)
{
struct impl *this = object;
int res;
spa_return_val_if_fail(this != NULL, -EINVAL);
switch (id) {
case SPA_PARAM_Props:
{
struct props *p = &this->props;
if (param == NULL) {
reset_props(p);
return 0;
}
spa_pod_parse_object(param,
SPA_TYPE_OBJECT_Props, NULL,
SPA_PROP_device, SPA_POD_OPT_Stringn(p->device, sizeof(p->device))
);
spa_log_debug(this->log, "%p: setting device name to \"%s\"", this, p->device);
p->device_name_set = true;
if ((res = parse_device(this)) < 0) {
p->device_name_set = false;
return res;
}
emit_node_info(this, false);
break;
}
default:
res = -ENOENT;
break;
}
return res;
}
static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
{
struct impl *this = object;
spa_return_val_if_fail(this != NULL, -EINVAL);
switch (id) {
case SPA_IO_Clock:
spa_log_debug(this->log, "%p: got clock IO", this);
this->node_clock_io = data;
break;
case SPA_IO_Position:
spa_log_debug(this->log, "%p: got position IO", this);
this->node_position_io = data;
break;
default:
return -ENOENT;
}
reevaluate_following_state(this);
reevaluate_freewheel_state(this);
return 0;
}
static int impl_node_send_command(void *object, const struct spa_command *command)
{
struct impl *this = object;
int res;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(command != NULL, -EINVAL);
spa_log_debug(this->log, "%p: got new command: %s", this,
spa_debug_type_find_name(spa_type_node_command_id, SPA_NODE_COMMAND_ID(command)));
switch (SPA_NODE_COMMAND_ID(command)) {
case SPA_NODE_COMMAND_ParamBegin:
if (SPA_UNLIKELY((res = device_open(this)) < 0))
return res;
break;
case SPA_NODE_COMMAND_ParamEnd:
device_close(this);
break;
case SPA_NODE_COMMAND_Start:
if (!this->have_format)
return -EIO;
if (this->n_buffers == 0)
return -EIO;
if (SPA_UNLIKELY((res = do_start(this)) < 0))
return res;
break;
case SPA_NODE_COMMAND_Suspend:
case SPA_NODE_COMMAND_Pause:
do_stop(this);
break;
default:
return -ENOTSUP;
}
return 0;
}
static int impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
const struct spa_dict *props)
{
return -ENOTSUP;
}
static int impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_id)
{
return -ENOTSUP;
}
static int port_enum_formats(struct impl *this, int seq, uint32_t start, uint32_t num,
const struct spa_pod *filter, struct spa_pod_builder *b)
{
bool device_started, device_opened;
struct spa_result_node_params result;
struct spa_pod *fmt;
const struct known_codec_info *codec_info;
uint32_t count = 0;
int res;
bool codec_supported;
struct spa_audio_info info;
device_opened = (this->device_context != NULL);
device_started = this->device_started;
spa_log_debug(this->log, "%p: about to enumerate supported codecs: "
"device opened: %d have configured format: %d device started: %d",
this, device_opened, this->have_format, device_started);
if (!this->started && this->have_format) {
spa_log_debug(this->log, "%p: closing device to reset configured format", this);
device_close(this);
device_opened = false;
}
if (!device_opened) {
if ((res = device_open(this)) < 0)
return res;
}
spa_zero(result);
result.id = SPA_PARAM_EnumFormat;
result.next = start;
next:
result.index = result.next++;
if (result.index >= SPA_N_ELEMENTS(known_codecs))
goto enum_end;
codec_info = &(known_codecs[result.index]);
codec_supported = compress_offload_api_supports_codec(this->device_context, codec_info->codec_id);
spa_log_debug(this->log, "%p: codec %s supported: %s", this,
codec_info->name, codec_supported ? "yes" : "no");
if (!codec_supported)
goto next;
spa_zero(info);
info.media_type = SPA_MEDIA_TYPE_audio;
info.media_subtype = codec_info->media_subtype;
if ((fmt = spa_format_audio_build(b, SPA_PARAM_EnumFormat, &info)) == NULL) {
res = -errno;
spa_log_error(this->log, "%p: error while building enumerated audio info: %s",
this, spa_strerror(res));
return res;
}
if (spa_pod_filter(b, &result.param, fmt, filter) < 0)
goto next;
spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
if (++count != num)
goto next;
enum_end:
res = 0;
if (!device_opened)
device_close(this);
spa_log_debug(this->log, "%p: done enumerating supported codecs", this);
return res;
}
static int impl_port_enum_params(void *object, int seq,
enum spa_direction direction, uint32_t port_id,
uint32_t id, uint32_t start, uint32_t num,
const struct spa_pod *filter)
{
struct impl *this = object;
struct spa_pod *param = NULL;
struct spa_pod_builder b = { 0 };
uint8_t buffer[4096];
struct spa_result_node_params result;
uint32_t count = 0;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(num != 0, -EINVAL);
spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
result.id = id;
result.next = start;
next:
result.index = result.next++;
spa_pod_builder_init(&b, buffer, sizeof(buffer));
switch (id) {
case SPA_PARAM_EnumFormat:
return port_enum_formats(this, seq, start, num, filter, &b);
case SPA_PARAM_Format:
if (!this->have_format) {
spa_log_debug(this->log, "%p: attempted to enumerate current "
"format, but no current audio info set", this);
return -EIO;
}
if (result.index > 0)
return 0;
spa_log_debug(this->log, "%p: current audio info is set; "
"enumerating currently set format", this);
param = spa_format_audio_build(&b, id, &this->current_audio_info);
break;
case SPA_PARAM_IO:
switch (result.index) {
case 0:
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_ParamIO, id,
SPA_PARAM_IO_id, SPA_POD_Id(SPA_IO_Buffers),
SPA_PARAM_IO_size, SPA_POD_Int(sizeof(struct spa_io_buffers)));
break;
default:
return 0;
}
break;
case SPA_PARAM_Buffers:
if (!this->have_format)
return -EIO;
if (result.index > 0)
return 0;
param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_ParamBuffers, id,
SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(1, 1, MAX_BUFFERS),
/* blocks is set to 1 since we don't have planar data */
SPA_PARAM_BUFFERS_blocks, SPA_POD_Int(1),
SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(
this->configured_fragment_size * this->configured_num_fragments,
this->configured_fragment_size * this->configured_num_fragments,
this->max_fragment_size * this->max_num_fragments),
/* "stride" has no meaning when dealing with compressed data */
SPA_PARAM_BUFFERS_stride, SPA_POD_Int(0));
break;
default:
return -ENOENT;
}
if (spa_pod_filter(&b, &result.param, param, filter) < 0)
goto next;
spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
if (++count != num)
goto next;
return 0;
}
static int port_set_format(void *object,
enum spa_direction direction, uint32_t port_id,
uint32_t flags,
const struct spa_pod *format)
{
struct impl *this = object;
int res;
if (format == NULL) {
if (!this->have_format)
return 0;
spa_log_debug(this->log, "%p: clearing format and closing device", this);
device_close(this);
clear_buffers(this);
} else {
struct spa_audio_info info = { 0 };
uint32_t rate;
const struct snd_compr_caps *compress_offload_caps;
spa_log_debug(this->log, "%p: about to set format", this);
if ((res = spa_format_audio_parse(format, &info)) < 0) {
spa_log_error(this->log, "%p: error while parsing audio format: %s",
this, spa_strerror(res));
return res;
}
if (this->device_context != NULL) {
spa_log_debug(this->log, "%p: need to close device to be able to reopen it with new format", this);
device_close(this);
}
if ((res = init_audio_codec_info(this, &info, &rate)) < 0)
return res;
if ((res = device_open(this)) < 0)
return res;
if (!compress_offload_api_supports_codec(this->device_context, this->audio_codec_info.id)) {
spa_log_error(this->log, "%p: codec is not supported by the device", this);
device_close(this);
return -ENOTSUP;
}
if ((res = compress_offload_api_set_params(this->device_context, &(this->audio_codec_info), 0, 0)) < 0)
return res;
compress_offload_caps = compress_offload_api_get_caps(this->device_context);
this->min_fragment_size = compress_offload_caps->min_fragment_size;
this->max_fragment_size = compress_offload_caps->max_fragment_size;
this->min_num_fragments = compress_offload_caps->min_fragments;
this->max_num_fragments = compress_offload_caps->max_fragments;
spa_log_debug(
this->log,
"%p: min/max fragment size: %" PRIu32 "/%" PRIu32 " min/max num fragments: %" PRIu32 "/%" PRIu32,
this,
this->min_fragment_size, this->max_fragment_size,
this->min_num_fragments, this->max_num_fragments
);
compress_offload_api_get_fragment_config(this->device_context,
&(this->configured_fragment_size),
&(this->configured_num_fragments));
spa_log_debug(
this->log, "%p: configured fragment size: %" PRIu32 " configured num fragments: %" PRIu32,
this,
this->configured_fragment_size, this->configured_num_fragments
);
this->current_audio_info = info;
this->have_format = true;
this->port_info.rate = SPA_FRACTION(1, rate);
}
this->node_info.change_mask |= SPA_NODE_CHANGE_MASK_FLAGS;
this->node_info.flags &= ~SPA_NODE_FLAG_NEED_CONFIGURE;
emit_node_info(this, false);
this->port_info.change_mask |= SPA_PORT_CHANGE_MASK_RATE;
this->port_info.change_mask |= SPA_PORT_CHANGE_MASK_PARAMS;
if (this->have_format) {
this->port_params[PORT_Format] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_READWRITE);
this->port_params[PORT_Buffers] = SPA_PARAM_INFO(SPA_PARAM_Buffers, SPA_PARAM_INFO_READ);
} else {
this->port_params[PORT_Format] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE);
this->port_params[PORT_Buffers] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0);
}
emit_port_info(this, false);
return 0;
}
static int impl_port_set_param(void *object,
enum spa_direction direction, uint32_t port_id,
uint32_t id, uint32_t flags,
const struct spa_pod *param)
{
struct impl *this = object;
int res;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
switch (id) {
case SPA_PARAM_Format:
res = port_set_format(this, direction, port_id, flags, param);
break;
default:
res = -ENOENT;
break;
}
return res;
}
static int impl_port_use_buffers(void *object,
enum spa_direction direction, uint32_t port_id,
uint32_t flags,
struct spa_buffer **buffers, uint32_t n_buffers)
{
struct impl *this = object;
uint32_t i;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
if (this->n_buffers > 0) {
spa_log_debug(this->log, "%p: %u buffers currently already in use; stopping device "
"to remove them before using new ones", this, this->n_buffers);
do_stop(this);
clear_buffers(this);
}
spa_log_debug(this->log, "%p: using a pool with %d buffer(s)", this, n_buffers);
if (n_buffers > 0 && !this->have_format)
return -EIO;
if (n_buffers > MAX_BUFFERS)
return -ENOSPC;
for (i = 0; i < n_buffers; i++) {
struct buffer *b = &this->buffers[i];
struct spa_data *d = buffers[i]->datas;
b->id = i;
b->flags = BUFFER_FLAG_AVAILABLE_FOR_NEW_DATA;
b->buf = buffers[i];
if (d[0].data == NULL) {
spa_log_error(this->log, "%p: need mapped memory", this);
return -EINVAL;
}
spa_log_debug(this->log, "%p: got buffer with ID %d bufptr %p data %p", this, i, b->buf, d[0].data);
}
this->n_buffers = n_buffers;
return 0;
}
static int impl_port_set_io(void *object,
enum spa_direction direction,
uint32_t port_id,
uint32_t id,
void *data, size_t size)
{
struct impl *this = object;
spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
switch (id) {
case SPA_IO_Buffers:
spa_log_debug(this->log, "%p: got buffers IO with data %p", this, data);
this->port_buffers_io = data;
break;
default:
return -ENOENT;
}
return 0;
}
static int impl_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id)
{
return -ENOTSUP;
}
static int impl_node_process(void *object)
{
struct impl *this = object;
struct spa_io_buffers *io;
struct buffer *b;
int res;
spa_return_val_if_fail(this != NULL, -EINVAL);
io = this->port_buffers_io;
spa_return_val_if_fail(io != NULL, -EIO);
/* Sinks aren't supposed to actually consume anything
* when the graph runs in freewheel mode. */
if (this->node_position_io && this->node_position_io->clock.flags & SPA_IO_CLOCK_FLAG_FREEWHEEL) {
io->status = SPA_STATUS_NEED_DATA;
return SPA_STATUS_HAVE_DATA;
}
/* Add the incoming data if there is some. We place the data in
* a queue instead of just consuming it directly. This allows for
* adjusting driver cycles to the needs of the sink - if the sink
* already has data queued, it does not yet need to schedule a next
* cycle. See on_driver_timeout() for details. This is only relevnt
* if the sink is running as the graph's driver. */
if ((io->status == SPA_STATUS_HAVE_DATA) && (io->buffer_id < this->n_buffers)) {
b = &this->buffers[io->buffer_id];
if (!SPA_FLAG_IS_SET(b->flags, BUFFER_FLAG_AVAILABLE_FOR_NEW_DATA)) {
spa_log_warn(this->log, "%p: buffer %u in use", this, io->buffer_id);
io->status = -EINVAL;
return -EINVAL;
}
if (this->device_is_paused) {
spa_log_debug(this->log, "%p: resuming paused device", this);
if ((res = device_resume(this)) < 0) {
io->status = res;
return SPA_STATUS_STOPPED;
}
}
spa_log_trace_fp(this->log, "%p: queuing buffer %u", this, io->buffer_id);
spa_list_append(&this->queued_output_buffers, &b->link);
SPA_FLAG_CLEAR(b->flags, BUFFER_FLAG_AVAILABLE_FOR_NEW_DATA);
/* This is essential to be able to hold back this buffer
* (which is because we queued it in a custom list for late
* consumption). By setting buffer_id to SPA_ID_INVALID,
* we essentially inform the graph that it must not attempt
* to return this buffer to the buffer pool. */
io->buffer_id = SPA_ID_INVALID;
if (SPA_UNLIKELY((res = write_queued_output_buffers(this)) < 0)) {
io->status = res;
return SPA_STATUS_STOPPED;
}
io->status = SPA_STATUS_OK;
}
return SPA_STATUS_HAVE_DATA;
}
/* SPA node information and init / clear procedures */
static const struct spa_node_methods impl_node = {
SPA_VERSION_NODE_METHODS,
.add_listener = impl_node_add_listener,
.set_callbacks = impl_node_set_callbacks,
.sync = impl_node_sync,
.enum_params = impl_node_enum_params,
.set_param = impl_node_set_param,
.set_io = impl_node_set_io,
.send_command = impl_node_send_command,
.add_port = impl_node_add_port,
.remove_port = impl_node_remove_port,
.port_enum_params = impl_port_enum_params,
.port_set_param = impl_port_set_param,
.port_use_buffers = impl_port_use_buffers,
.port_set_io = impl_port_set_io,
.port_reuse_buffer = impl_port_reuse_buffer,
.process = impl_node_process,
};
static int impl_get_interface(struct spa_handle *handle, const char *type, void **interface)
{
struct impl *this;
spa_return_val_if_fail(handle != NULL, -EINVAL);
spa_return_val_if_fail(interface != NULL, -EINVAL);
this = (struct impl *) handle;
if (spa_streq(type, SPA_TYPE_INTERFACE_Node))
*interface = &this->node;
else
return -ENOENT;
return 0;
}
static int impl_clear(struct spa_handle *handle)
{
struct impl *this;
spa_return_val_if_fail(handle != NULL, -EINVAL);
this = (struct impl *) handle;
device_close(this);
if (this->driver_timerfd > 0) {
spa_system_close(this->data_system, this->driver_timerfd);
this->driver_timerfd = -1;
}
spa_log_info(this->log, "%p: created Compress-Offload sink", this);
return 0;
}
static size_t impl_get_size(const struct spa_handle_factory *factory,
const struct spa_dict *params)
{
return sizeof(struct impl);
}
static int
impl_init(const struct spa_handle_factory *factory, struct spa_handle *handle,
const struct spa_dict *info, const struct spa_support *support, uint32_t n_support)
{
struct impl *this;
uint32_t i;
int res = 0;
spa_return_val_if_fail(factory != NULL, -EINVAL);
spa_return_val_if_fail(handle != NULL, -EINVAL);
handle->get_interface = impl_get_interface;
handle->clear = impl_clear;
this = (struct impl *) handle;
this->log = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_Log);
/* A logger must always exist, otherwise something is very wrong. */
assert(this->log != NULL);
alsa_log_topic_init(this->log);
this->data_loop = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataLoop);
if (this->data_loop == NULL) {
spa_log_error(this->log, "%p: could not find a loop", this);
res = -EINVAL;
goto error;
}
this->data_system = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataSystem);
if (this->data_system == NULL) {
spa_log_error(this->log, "%p: could not find a data system", this);
res = -EINVAL;
goto error;
}
this->node.iface = SPA_INTERFACE_INIT(
SPA_TYPE_INTERFACE_Node,
SPA_VERSION_NODE,
&impl_node, this);
spa_hook_list_init(&this->hooks);
reset_props(&this->props);
this->have_format = false;
this->started = false;
this->freewheel = false;
this->n_buffers = 0;
spa_list_init(&this->queued_output_buffers);
this->offset_within_oldest_output_buffer = 0;
res = this->driver_timerfd = spa_system_timerfd_create(this->data_system, CLOCK_MONOTONIC,
SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
if (SPA_UNLIKELY(res < 0)) {
spa_log_error(this->log, "%p: could not create driver timerfd: %s", this, spa_strerror(res));
goto error;
}
this->next_driver_time = 0;
this->following = false;
this->node_info_all = SPA_NODE_CHANGE_MASK_FLAGS |
SPA_NODE_CHANGE_MASK_PROPS |
SPA_NODE_CHANGE_MASK_PARAMS;
this->node_info = SPA_NODE_INFO_INIT();
this->node_info.max_input_ports = 1;
this->node_info.flags = SPA_NODE_FLAG_RT |
SPA_NODE_FLAG_IN_PORT_CONFIG |
SPA_NODE_FLAG_NEED_CONFIGURE;
this->node_params[NODE_PropInfo] = SPA_PARAM_INFO(SPA_PARAM_PropInfo, SPA_PARAM_INFO_READ);
this->node_params[NODE_Props] = SPA_PARAM_INFO(SPA_PARAM_Props, SPA_PARAM_INFO_READWRITE);
this->node_params[NODE_IO] = SPA_PARAM_INFO(SPA_PARAM_IO, SPA_PARAM_INFO_READ);
this->node_params[NODE_EnumPortConfig] = SPA_PARAM_INFO(SPA_PARAM_EnumPortConfig, SPA_PARAM_INFO_READ);
this->node_info.params = this->node_params;
this->node_info.n_params = N_NODE_PARAMS;
this->node_clock_io = NULL;
this->node_position_io = NULL;
this->port_info_all = SPA_PORT_CHANGE_MASK_FLAGS |
SPA_PORT_CHANGE_MASK_PARAMS;
this->port_info = SPA_PORT_INFO_INIT();
this->port_info.flags = SPA_PORT_FLAG_LIVE |
SPA_PORT_FLAG_PHYSICAL |
SPA_PORT_FLAG_TERMINAL;
this->port_params[PORT_EnumFormat] = SPA_PARAM_INFO(SPA_PARAM_EnumFormat, SPA_PARAM_INFO_READ);
this->port_params[PORT_Format] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE);
this->port_params[PORT_IO] = SPA_PARAM_INFO(SPA_PARAM_IO, SPA_PARAM_INFO_READ);
this->port_params[PORT_Buffers] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0);
this->port_info.params = this->port_params;
this->port_info.n_params = N_PORT_PARAMS;
this->port_buffers_io = NULL;
this->device_context = NULL;
this->device_started = false;
memset(&this->audio_codec_info, 0, sizeof(this->audio_codec_info));
this->device_is_paused = false;
spa_log_info(this->log, "%p: initialized Compress-Offload sink", this);
for (i = 0; info && i < info->n_items; i++) {
const char *k = info->items[i].key;
const char *s = info->items[i].value;
if (spa_streq(k, SPA_KEY_API_ALSA_PATH)) {
snprintf(this->props.device, sizeof(this->props.device), "%s", s);
if ((res = parse_device(this)) < 0)
return res;
}
}
finish:
return res;
error:
impl_clear((struct spa_handle *)this);
goto finish;
}
static const struct spa_interface_info impl_interfaces[] = {
{SPA_TYPE_INTERFACE_Node,},
};
static int
impl_enum_interface_info(const struct spa_handle_factory *factory,
const struct spa_interface_info **info, uint32_t *index)
{
spa_return_val_if_fail(factory != NULL, -EINVAL);
spa_return_val_if_fail(info != NULL, -EINVAL);
spa_return_val_if_fail(index != NULL, -EINVAL);
switch (*index) {
case 0:
*info = &impl_interfaces[*index];
break;
default:
return 0;
}
(*index)++;
return 1;
}
/* Factory info */
static const struct spa_dict_item info_items[] = {
{ SPA_KEY_FACTORY_AUTHOR, "Sanchayan Maity <sanchayan@asymptotic.io>, Carlos Rafael Giani <crg7475@mailbox.org>" },
{ SPA_KEY_FACTORY_DESCRIPTION, "Play compressed audio (like MP3 or AAC) with the ALSA Compress-Offload API" },
{ SPA_KEY_FACTORY_USAGE, "["SPA_KEY_API_ALSA_PATH"=<path>]" },
};
static const struct spa_dict info = SPA_DICT_INIT_ARRAY(info_items);
const struct spa_handle_factory spa_alsa_compress_offload_sink_factory = {
SPA_VERSION_HANDLE_FACTORY,
SPA_NAME_API_ALSA_COMPRESS_OFFLOAD_SINK,
&info,
impl_get_size,
impl_init,
impl_enum_interface_info,
};
Reference in a new issue View git blame Copy permalink