mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-04 13:30:12 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

module-netjack2: clear empty buffers

Browse source
This commit is contained in:
Wim Taymans 2023-06-02 15:09:31 +02:00
parent 287566937e
commit 629e322fac
2 changed files with 26 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

23
src/modules/module-netjack2-driver.c
View file

@ -393,6 +393,7 @@ receive_error:
struct data_info { struct data_info {
void *data; void *data;
uint32_t id; uint32_t id;
bool filled;
}; };
static int netjack2_send_sync(struct stream *s, uint32_t nframes) static int netjack2_send_sync(struct stream *s, uint32_t nframes)
@ -565,8 +566,6 @@ static int netjack2_recv_midi(struct stream *s, struct nj2_packet_header *header
if ((len = recv(impl->socket->fd, impl->recv_buffer, ntohl(header->packet_size), 0)) < 0) if ((len = recv(impl->socket->fd, impl->recv_buffer, ntohl(header->packet_size), 0)) < 0)
return -errno; return -errno;
impl->sync.is_last = ntohl(header->is_last);
impl->sync.cycle = ntohl(header->cycle); impl->sync.cycle = ntohl(header->cycle);
impl->sync.num_packets = ntohl(header->num_packets); impl->sync.num_packets = ntohl(header->num_packets);
@ -586,8 +585,6 @@ static int netjack2_recv_audio(struct stream *s, struct nj2_packet_header *heade
if ((len = recv(impl->socket->fd, impl->recv_buffer, ntohl(header->packet_size), 0)) < 0) if ((len = recv(impl->socket->fd, impl->recv_buffer, ntohl(header->packet_size), 0)) < 0)
return -errno; return -errno;
impl->sync.is_last = ntohl(header->is_last);
sub_cycle = ntohl(header->sub_cycle); sub_cycle = ntohl(header->sub_cycle);
active_ports = ntohl(header->active_ports); active_ports = ntohl(header->active_ports);
@ -616,12 +613,9 @@ static int netjack2_recv_audio(struct stream *s, struct nj2_packet_header *heade
sub_cycle * sub_period_size * sizeof(float), sub_cycle * sub_period_size * sizeof(float),
float); float);
do_volume(dst, (float*)&ap[1], &s->volume, active_port, sub_period_size); do_volume(dst, (float*)&ap[1], &s->volume, active_port, sub_period_size);
info[active_port].filled = impl->sync.is_last;
} }
} }
if (impl->sync.is_last) {
pw_log_trace_fp("got last audio packet");
}
return 0; return 0;
} }
@ -629,7 +623,7 @@ static int netjack2_recv_data(struct stream *s, struct data_info *info, uint32_t
{ {
struct impl *impl = s->impl; struct impl *impl = s->impl;
ssize_t len; ssize_t len;
uint32_t count = 0; uint32_t i, count = 0;
struct nj2_packet_header *header = (struct nj2_packet_header *)impl->recv_buffer; struct nj2_packet_header *header = (struct nj2_packet_header *)impl->recv_buffer;
while (!impl->sync.is_last) { while (!impl->sync.is_last) {
@ -647,6 +641,8 @@ static int netjack2_recv_data(struct stream *s, struct data_info *info, uint32_t
continue; continue;
} }
impl->sync.is_last = ntohl(header->is_last);
switch (ntohl(header->data_type)) { switch (ntohl(header->data_type)) {
case 'm': case 'm':
netjack2_recv_midi(s, header, &count, info, n_info); netjack2_recv_midi(s, header, &count, info, n_info);
@ -656,9 +652,14 @@ static int netjack2_recv_data(struct stream *s, struct data_info *info, uint32_t
break; break;
case 's': case 's':
pw_log_info("missing last data packet"); pw_log_info("missing last data packet");
return 0; impl->sync.is_last = true;
break;
} }
} }
for (i = 0; i < s->n_ports; i++) {
if (!info[i].filled && info[i].data != NULL)
memset(info[i].data, 0, impl->params.period_size * sizeof(float));
}
impl->sync.cycle = ntohl(header->cycle); impl->sync.cycle = ntohl(header->cycle);
return 0; return 0;
@ -685,8 +686,8 @@ static void source_process(void *d, struct spa_io_position *position)
struct port *p = s->ports[i]; struct port *p = s->ports[i];
info[i].data = p ? pw_filter_get_dsp_buffer(p, n_samples) : NULL; info[i].data = p ? pw_filter_get_dsp_buffer(p, n_samples) : NULL;
info[i].id = i; info[i].id = i;
info[i].filled = false;
} }
netjack2_recv_data(s, info, s->n_ports); netjack2_recv_data(s, info, s->n_ports);
} }

23
src/modules/module-netjack2-manager.c
View file

@ -361,6 +361,7 @@ static void stream_state_changed(void *d, enum pw_filter_state old,
struct data_info { struct data_info {
void *data; void *data;
uint32_t id; uint32_t id;
bool filled;
}; };
static int netjack2_send_sync(struct stream *s, uint32_t nframes) static int netjack2_send_sync(struct stream *s, uint32_t nframes)
@ -571,6 +572,7 @@ static int netjack2_recv_audio(struct stream *s, struct nj2_packet_header *heade
sub_cycle * sub_period_size * sizeof(float), sub_cycle * sub_period_size * sizeof(float),
float); float);
do_volume(dst, (float*)&ap[1], &s->volume, active_port, sub_period_size); do_volume(dst, (float*)&ap[1], &s->volume, active_port, sub_period_size);
info[active_port].filled = follower->sync.is_last;
} }
} }
return 0; return 0;
@ -606,6 +608,7 @@ static int32_t netjack2_sync_wait(struct follower *follower)
offset = follower->cycle - follower->sync.cycle; offset = follower->cycle - follower->sync.cycle;
if (offset < (int32_t)follower->params.network_latency) { if (offset < (int32_t)follower->params.network_latency) {
pw_log_info("sync offset %d %d %d", follower->cycle, follower->sync.cycle, offset); pw_log_info("sync offset %d %d %d", follower->cycle, follower->sync.cycle, offset);
follower->sync.is_last = true;
return 0; return 0;
} else { } else {
if ((len = recv(follower->socket->fd, follower->recv_buffer, follower->params.mtu, 0)) < 0) if ((len = recv(follower->socket->fd, follower->recv_buffer, follower->params.mtu, 0)) < 0)
@ -623,10 +626,12 @@ static int netjack2_recv_data(struct stream *s, struct data_info *info, uint32_t
{ {
struct follower *follower = s->follower; struct follower *follower = s->follower;
ssize_t len; ssize_t len;
uint32_t count = 0; uint32_t i, count = 0;
struct nj2_packet_header *header = (struct nj2_packet_header *)follower->recv_buffer; struct nj2_packet_header *header = (struct nj2_packet_header *)follower->recv_buffer;
int res = 0; int res = 0;
netjack2_sync_wait(follower);
while (!follower->sync.is_last) { while (!follower->sync.is_last) {
if ((len = recv(follower->socket->fd, follower->recv_buffer, follower->params.mtu, MSG_PEEK)) < 0) if ((len = recv(follower->socket->fd, follower->recv_buffer, follower->params.mtu, MSG_PEEK)) < 0)
goto receive_error; goto receive_error;
@ -652,13 +657,18 @@ static int netjack2_recv_data(struct stream *s, struct data_info *info, uint32_t
break; break;
case 's': case 's':
pw_log_info("missing last data packet"); pw_log_info("missing last data packet");
return 0; follower->sync.is_last = true;
break;
} }
if (res < 0) { if (res < 0) {
pw_log_warn("recv error: %s", spa_strerror(res)); pw_log_warn("recv error: %s", spa_strerror(res));
break; break;
} }
} }
for (i = 0; i < n_info; i++) {
if (!info[i].filled && info[i].data != NULL)
memset(info[i].data, 0, follower->params.period_size * sizeof(float));
}
follower->sync.cycle = ntohl(header->cycle); follower->sync.cycle = ntohl(header->cycle);
return 0; return 0;
@ -672,19 +682,14 @@ static void source_process(void *d, struct spa_io_position *position)
struct stream *s = d; struct stream *s = d;
uint32_t i, n_samples = position->clock.duration; uint32_t i, n_samples = position->clock.duration;
struct data_info info[s->n_ports]; struct data_info info[s->n_ports];
uint32_t sync;
sync = netjack2_sync_wait(s->follower);
for (i = 0; i < s->n_ports; i++) { for (i = 0; i < s->n_ports; i++) {
struct port *p = s->ports[i]; struct port *p = s->ports[i];
info[i].data = p ? pw_filter_get_dsp_buffer(p, n_samples) : NULL; info[i].data = p ? pw_filter_get_dsp_buffer(p, n_samples) : NULL;
info[i].id = i; info[i].id = i;
if (sync == 0 && info[i].data != NULL) info[i].filled = false;
memset(info[i].data, 0, n_samples * sizeof(float));
} }
if (sync > 0) netjack2_recv_data(s, info, s->n_ports);
netjack2_recv_data(s, info, s->n_ports);
} }
static void follower_free(struct follower *follower) static void follower_free(struct follower *follower)