summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/bimu/client.c107
-rw-r--r--src/bimu/client.h19
-rw-r--r--src/bimu/handler.h12
-rw-r--r--src/bimu/handlers/codec_client.c4
-rw-r--r--src/bimu/meson.build1
-rw-r--r--src/bimu/server.c60
-rw-r--r--src/bimu/server.h9
-rw-r--r--src/bimu/vcr.c166
-rw-r--r--src/bimu/vcr.h45
-rw-r--r--src/buffer/audio.c79
-rw-r--r--src/buffer/audio.h4
-rw-r--r--src/buffer/clock.c43
-rw-r--r--src/buffer/clock.h7
-rw-r--r--src/buffer/video.c16
-rw-r--r--src/codec/libav/decoder.c2
-rw-r--r--src/codec/libav/demuxer.c4
-rw-r--r--src/codec/libav/resampler.c53
-rw-r--r--src/codec/libav/resampler.h6
-rw-r--r--src/codec/meson.build23
-rw-r--r--src/fruits/cmc/cmc.c4
-rw-r--r--src/fruits/sink/sink.c14
-rw-r--r--src/libsink/sink.c170
-rw-r--r--src/libsink/sink.h7
-rw-r--r--src/mixer/audio_miniaudio.c106
-rw-r--r--src/mixer/meson.build1
-rw-r--r--src/mixer/mixer.c15
-rw-r--r--src/tree/common.h11
-rw-r--r--src/tree/list.c3
-rw-r--r--src/tree/tree.c3
-rw-r--r--src/tree/tree.h1
30 files changed, 669 insertions, 326 deletions
diff --git a/src/bimu/client.c b/src/bimu/client.c
index c592906..03910ee 100644
--- a/src/bimu/client.c
+++ b/src/bimu/client.c
@@ -15,6 +15,7 @@ static void data_connection_callback(void *userdata, struct aki_packet_stream *s
aki_packet_write_u8(packet, BIMU_CONNECTION_DATA);
aki_packet_write_u16(packet, client->connection_id);
aki_packet_write_u16(packet, client->node_id);
+ aki_packet_write_u64(packet, client->seek_pos);
aki_packet_stream_send_packet(&client->data, packet);
}
@@ -22,62 +23,35 @@ static void send_reconnect_packet(struct bmu_client *client)
{
struct aki_packet *packet = aki_packet_create();
aki_packet_write_u8(packet, BIMU_CONTROL_RECONNECT);
+ aki_packet_write_u32(packet, client->level);
aki_packet_stream_send_packet(&client->control, packet);
}
-static void flush_streams(struct bmu_client *client)
+static void cleanup_connection_internal(struct bmu_client *client)
{
- struct bmu_client_stream *stream;
- al_array_foreach(client->streams, i, stream) {
- stream->client->flush(stream->client);
- }
+ client->callback(client->userdata, BIMU_CLIENT_CLOSED, NULL, NULL);
}
static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
{
struct bmu_client *client = (struct bmu_client *)userdata;
(void)stream;
+ if (client->closed) {
+ if (!client->control.connected) cleanup_connection_internal(client);
+ return;
+ }
aki_packet_stream_free(&client->data);
client->callback(client->userdata, BIMU_CLIENT_REMOVE_BUFFERS, NULL, NULL);
- flush_streams(client);
+ bmu_vcr_flush(&client->vcr);
send_reconnect_packet(client);
}
-static struct bmu_client_stream *stream_from_index(struct bmu_client *client, s32 index)
-{
- struct bmu_client_stream *stream;
- al_array_foreach(client->streams, i, stream) {
- if (stream->index == index) return stream;
- }
- return NULL;
-}
-
-static void handle_data_packet_internal(struct bmu_client *client, struct aki_packet *packet)
-{
- struct bmu_client_stream *stream;
- switch (aki_packet_read_u8(packet)) {
- case BIMU_PACKET_DATA:
- stream = stream_from_index(client, aki_packet_read_s32(packet));
- stream->client->handle_data_packet(stream->client, packet);
- break;
- case BIMU_PACKET_EOF:
- al_array_foreach(client->streams, i, stream) {
- stream->client->handle_eof(stream->client);
- }
- break;
- case BIMU_PACKET_ERROR:
- al_log_warn("bimu", "Unhandled error packet.");
- break;
- }
-}
-
static void data_packet_callback(void *userdata, struct aki_packet_stream *stream,
struct aki_packet *packet)
{
struct bmu_client *client = (struct bmu_client *)userdata;
(void)stream;
- handle_data_packet_internal(client, packet);
- aki_packet_free(packet);
+ bmu_vcr_push_packet(&client->vcr, packet);
}
static void packet_sent_callback(void *userdata, struct aki_packet *packet)
@@ -106,8 +80,14 @@ static void control_connection_callback(void *userdata, struct aki_packet_stream
static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
{
- (void)userdata;
+ struct bmu_client *client = (struct bmu_client *)userdata;
(void)stream;
+ if (client->closed) {
+ if (!client->data.connected) cleanup_connection_internal(client);
+ return;
+ }
+ aki_packet_stream_free(&client->control);
+ // TODO: Reconnect.
}
static void parse_info_packet(struct bmu_client *client, struct aki_packet *packet)
@@ -115,7 +95,7 @@ static void parse_info_packet(struct bmu_client *client, struct aki_packet *pack
client->duration = aki_packet_read_u64(packet);
u32 count = aki_packet_read_u32(packet);
for (u32 i = 0; i < count; i++) {
- struct bmu_client_stream *stream = al_alloc_object(struct bmu_client_stream);
+ struct bmu_vcr_stream *stream = al_alloc_object(struct bmu_vcr_stream);
stream->client = bmu_codec_client_create();
stream->client->callback = client->callback;
stream->client->userdata = client->userdata;
@@ -154,7 +134,7 @@ static void parse_info_packet(struct bmu_client *client, struct aki_packet *pack
}
if (!stream->client->init(stream->client, NULL, stream)) {
}
- al_array_push(client->streams, stream);
+ bmu_vcr_add_stream(&client->vcr, stream);
}
}
@@ -172,26 +152,12 @@ static void send_subscribe_packet(struct bmu_client *client)
static void parse_start_packet(struct bmu_client *client, struct aki_packet *packet)
{
- u64 seek_pos = aki_packet_read_u64(packet);
u64 ts = aki_packet_read_u64(packet);
- u8 paused = aki_packet_read_u8(packet);
- (void)paused;
- struct bmu_seek_req req = {
- .base = seek_pos,
- .start = ts
- };
- client->callback(client->userdata, BIMU_CLIENT_SET, NULL, &req);
-}
-
-static void parse_resume_packet(struct bmu_client *client, struct aki_packet *packet)
-{
+ u64 delay = aki_packet_read_u64(packet);
u64 seek_pos = aki_packet_read_u64(packet);
- u64 ts = aki_packet_read_u64(packet);
- struct bmu_seek_req req = {
- .base = seek_pos,
- .start = ts
- };
+ struct bmu_seek_req req = { .base = seek_pos, .ts = ts, .delay = delay };
client->callback(client->userdata, BIMU_CLIENT_SET, NULL, &req);
+ client->seek_pos = req.base;
}
static void control_packet_callback(void *userdata, struct aki_packet_stream *stream,
@@ -213,9 +179,9 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
case BIMU_CONTROL_PAUSE:
break;
case BIMU_CONTROL_RESUME:
- //parse_resume_packet(client, packet);
break;
case BIMU_CONTROL_SEEK:
+ client->level = 0;
client->corked = false;
aki_packet_stream_disconnect(&client->data);
break;
@@ -232,19 +198,22 @@ void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop,
client->node_id = node_id;
al_str_clone(&client->addr, addr);
client->port = port;
+ client->level = 0;
client->corked = false;
al_array_init(client->subscribed);
- al_array_init(client->streams);
+ bmu_vcr_init(&client->vcr, client->loop, &client->data);
aki_packet_stream_init(&client->control, AKI_SOCKET_TCP, control_connection_callback,
control_connection_closed_callback, control_packet_callback, packet_sent_callback, client);
+ aki_packet_stream_set_no_delay(&client->control, 1);
aki_packet_stream_connect(&client->control, client->loop, &client->addr, client->port);
}
-void bmu_client_cork(struct bmu_client *client, bool cork)
+void bmu_client_reseek(struct bmu_client *client)
{
- if (cork == client->corked) return;
- aki_packet_stream_cork(&client->data, cork);
- client->corked = cork;
+ client->level++;
+ al_log_debug("bimu", "Attempting re-seek at level %i.", client->level);
+ client->corked = false;
+ aki_packet_stream_disconnect(&client->data);
}
void bmu_client_seek(struct bmu_client *client, f64 percent)
@@ -256,8 +225,22 @@ void bmu_client_seek(struct bmu_client *client, f64 percent)
aki_packet_stream_send_packet(&client->control, packet);
}
+void bmu_client_close(struct bmu_client *client)
+{
+ struct bmu_vcr_stream *stream;
+ al_array_foreach(client->vcr.streams, i, stream) {
+ bmu_vcr_stream_close(stream);
+ }
+ client->closed = true;
+ aki_packet_stream_disconnect(&client->control);
+ aki_packet_stream_disconnect(&client->data);
+}
+
void bmu_client_free(struct bmu_client *client)
{
aki_packet_stream_free(&client->control);
aki_packet_stream_free(&client->data);
+ al_str_free(&client->addr);
+ al_array_free(client->subscribed);
+ bmu_vcr_free(&client->vcr);
}
diff --git a/src/bimu/client.h b/src/bimu/client.h
index 8c9803b..db1f9e5 100644
--- a/src/bimu/client.h
+++ b/src/bimu/client.h
@@ -3,14 +3,7 @@
#include <al/types.h>
#include <aki/packet_stream.h>
-#include "../codec/codec.h"
-
-struct bmu_client_stream {
- u8 type;
- s32 index;
- struct camu_stream stream;
- struct bmu_client_handler *client;
-};
+#include "vcr.h"
struct bmu_client {
struct aki_event_loop *loop;
@@ -20,15 +13,19 @@ struct bmu_client {
s32 port;
struct aki_packet_stream control;
struct aki_packet_stream data;
+ bool closed;
+ u32 level;
bool corked;
u64 duration;
+ u64 seek_pos;
array(s32) subscribed;
- array(struct bmu_client_stream *) streams;
- void (*callback)(void *, u8, struct bmu_client_stream *stream, void *);
+ struct bmu_vcr vcr;
+ void (*callback)(void *, u8, struct bmu_vcr_stream *stream, void *);
void *userdata;
};
void bmu_client_connect(struct bmu_client *client, struct aki_event_loop *loop, str *addr, s32 port, u16 node_id);
-void bmu_client_cork(struct bmu_client *client, bool cork);
+void bmu_client_reseek(struct bmu_client *client);
void bmu_client_seek(struct bmu_client *client, f64 percent);
+void bmu_client_close(struct bmu_client *client);
void bmu_client_free(struct bmu_client *client);
diff --git a/src/bimu/handler.h b/src/bimu/handler.h
index 3646dd6..c6b482d 100644
--- a/src/bimu/handler.h
+++ b/src/bimu/handler.h
@@ -5,6 +5,8 @@
#include "../codec/codec.h"
#include "../cache/handle.h"
+#include "vcr.h"
+
struct bmu_server_handler {
bool (*init)(struct bmu_server_handler *, struct cch_handle *, struct aki_packet_pool *);
void (*write_info)(struct bmu_server_handler *, struct aki_packet *);
@@ -32,17 +34,19 @@ enum {
};
struct bmu_seek_req {
- u64 base, start;
+ u64 base;
+ u64 ts;
+ u64 delay;
};
struct bmu_client_stream;
struct bmu_client_handler {
- bool (*init)(struct bmu_client_handler *, struct camu_renderer *, struct bmu_client_stream *);
+ bool (*init)(struct bmu_client_handler *, struct camu_renderer *, struct bmu_vcr_stream *);
void (*handle_data_packet)(struct bmu_client_handler *, struct aki_packet *);
void (*handle_eof)(struct bmu_client_handler *);
void (*flush)(struct bmu_client_handler *);
void (*free)(struct bmu_client_handler **);
- struct bmu_client_stream *stream;
- void (*callback)(void *, u8, struct bmu_client_stream *stream, void *);
+ struct bmu_vcr_stream *stream;
+ void (*callback)(void *, u8, struct bmu_vcr_stream *stream, void *);
void *userdata;
};
diff --git a/src/bimu/handlers/codec_client.c b/src/bimu/handlers/codec_client.c
index 570ef97..79ef27d 100644
--- a/src/bimu/handlers/codec_client.c
+++ b/src/bimu/handlers/codec_client.c
@@ -2,7 +2,7 @@
#include "../../codec/libav/decoder.h"
#include "../../codec/libav/packet_ext.h"
-#include "../../bimu/client.h"
+#include "../../bimu/vcr.h"
#include "codec.h"
@@ -13,7 +13,7 @@ static void data_callback(void *userdata, struct camu_frame *frame)
}
static bool codec_client_init(struct bmu_client_handler *handler, struct camu_renderer *renderer,
- struct bmu_client_stream *stream)
+ struct bmu_vcr_stream *stream)
{
struct bmu_codec_client *codec = (struct bmu_codec_client *)handler;
codec->dec = camu_lav_decoder_create();
diff --git a/src/bimu/meson.build b/src/bimu/meson.build
index dce7f7b..c7de266 100644
--- a/src/bimu/meson.build
+++ b/src/bimu/meson.build
@@ -5,6 +5,7 @@ bimu_server_src = [
bimu_client_src = [
'client.c',
+ 'vcr.c',
'handlers/codec_client.c'
]
diff --git a/src/bimu/server.c b/src/bimu/server.c
index e1aedce..6db0dd2 100644
--- a/src/bimu/server.c
+++ b/src/bimu/server.c
@@ -4,9 +4,7 @@
#include "handlers/codec.h"
#include "server.h"
-
-#define START_DELAY 3000000 // 3s
-#define DELAY_GIVE 200000 // 200ms
+#include "common.h"
static void control_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
{
@@ -21,6 +19,9 @@ static void control_connection_closed_callback(void *userdata, struct aki_packet
}
aki_packet_stream_free(connection->control);
connection->control = NULL;
+ //cch_entry_return_handle(connection->node->entry, &connection->handle);
+ //connection->handler->free(&connection->handler);
+ //aki_packet_pool_free(&connection->pool);
}
static void data_connection_closed_callback(void *userdata, struct aki_packet_stream *stream)
@@ -81,31 +82,33 @@ static void send_pause_packet(struct bmu_node_connection *connection, u64 seek_p
aki_packet_stream_send_packet(connection->control, packet);
}
-static void send_start_packet(struct bmu_node_connection *connection)
+#define DELAY 275000Lu // 275ms
+#define DELAY_GIVE 30000Lu // 30ms
+
+static void send_start_packet(struct bmu_node_connection *connection, u32 level)
{
struct aki_packet *packet = aki_packet_create();
aki_packet_write_u8(packet, BIMU_CONTROL_START);
- u64 ts = aki_get_timestamp() - START_DELAY;
+ u64 ts = aki_get_timestamp();
+ u64 delay = DELAY * powl(3, level);
u64 seek_pos = connection->node->seek_pos;
- if (!connection->node->paused) {
- if (connection->node->start >= 0) {
+ if (!(connection->node->flags & BIMU_NODE_PAUSED)) {
+ if (connection->node->flags & BIMU_NODE_STARTED) {
if ((ts - DELAY_GIVE) < (u64)connection->node->start) {
+ // Current time is close enough to node start.
ts = connection->node->start;
} else {
- seek_pos += ts - connection->node->start;
+ seek_pos += (ts - connection->node->start) + (delay - DELAY);
}
} else {
+ // Must be set at level 0, this applies for seeks too.
connection->node->start = ts;
+ connection->node->flags |= BIMU_NODE_STARTED;
}
}
- u64 duration = connection->handler->get_duration(connection->handler);
- if (seek_pos > duration) seek_pos = duration;
- // -1 signals to not call seek. We have to distinguish between a seek to 0
- // and the start of a stream.
- connection->seek_pos = seek_pos > 0 ? seek_pos : -1;
- aki_packet_write_u64(packet, seek_pos);
aki_packet_write_u64(packet, ts);
- aki_packet_write_u8(packet, connection->node->paused);
+ aki_packet_write_u64(packet, delay);
+ aki_packet_write_u64(packet, seek_pos);
aki_packet_stream_send_packet(connection->control, packet);
}
@@ -131,11 +134,11 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
for (u32 i = 0; i < count; i++) {
connection->handler->subscribe(connection->handler, aki_packet_read_s32(packet), true);
}
- send_start_packet(connection);
+ send_start_packet(connection, 0);
break;
}
case BIMU_CONTROL_RECONNECT: {
- send_start_packet(connection);
+ send_start_packet(connection, aki_packet_read_u32(packet));
break;
}
case BIMU_CONTROL_TOGGLE_PAUSE: {
@@ -144,13 +147,10 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
u8 paused = connection->paused;
if (paused == BIMU_PLAYING) {
node->seek_pos = pos;
- node->paused = true;
- // If node is paused, start will not be used for new connections.
- // So, set it to -1 and update it on the first resume from any connection.
- node->start = -1;
+ node->flags |= BIMU_NODE_PAUSED;
+ // Update node start?
} else if (paused == BIMU_PAUSED_AND_RUNNING) {
- node->paused = false;
- node->start = aki_get_timestamp() - START_DELAY;
+ node->flags &= ~BIMU_NODE_PAUSED;
}
struct bmu_node_connection *rconnection;
al_array_foreach(node->connections, i, rconnection) {
@@ -180,8 +180,8 @@ static void control_packet_callback(void *userdata, struct aki_packet_stream *st
case BIMU_CONTROL_SEEK: {
u64 pos = aki_packet_read_u64(packet);
connection->node->seek_pos = pos;
- connection->node->start = -1;
- connection->seek_pos = pos;
+ connection->node->flags |= BIMU_NODE_SEEKED;
+ connection->node->flags &= ~BIMU_NODE_STARTED;
send_seek_packet(connection->node);
break;
}
@@ -228,7 +228,8 @@ static aki_thread_result AKI_THREADCALL handler_thread(void *userdata)
static void maybe_start_handler(struct bmu_node_connection *connection)
{
al_assert(!connection->running);
- if (connection->seek_pos >= 0) {
+ // Don't seek to 0 unless NODE_SEEKED is set.
+ if (connection->seek_pos > 0 || (connection->node->flags & BIMU_NODE_SEEKED)) {
connection->handler->seek(connection->handler, connection->seek_pos);
}
connection->running = 1;
@@ -279,7 +280,7 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
switch (type) {
case BIMU_CONNECTION_CONTROL:
- aki_packet_stream_set_no_delay(stream, true);
+ aki_packet_stream_set_no_delay(stream, 1);
connection->control = stream;
stream->packet_callback = control_packet_callback;
stream->packet_sent_callback = default_packet_sent_callback;
@@ -293,6 +294,7 @@ static void identify_packet_callback(void *userdata, struct aki_packet_stream *s
// (disconnect and reconnect very fast?).
// - Call maybe_cleanup_data in this function?
al_assert(!connection->data);
+ connection->seek_pos = aki_packet_read_u64(packet);
connection->data = stream;
stream->packet_callback = data_packet_callback;
stream->packet_sent_callback = packet_sent_callback;
@@ -339,8 +341,8 @@ u16 bmu_server_create_node(struct bmu_server *server, struct cch_entry *entry)
al_array_push(server->nodes, node);
node->id = al_rand_u16();
node->entry = entry;
- node->paused = false;
- node->start = -1;
+ node->flags = 0;
+ node->start = 0;
node->seek_pos = 0;
node->connection_id = 1;
al_array_init(node->connections);
diff --git a/src/bimu/server.h b/src/bimu/server.h
index 09fed99..20f73c6 100644
--- a/src/bimu/server.h
+++ b/src/bimu/server.h
@@ -7,7 +7,6 @@
#include "../cache/entry.h"
#include "../cache/handle.h"
-#include "common.h"
#include "handler.h"
enum {
@@ -31,10 +30,16 @@ struct bmu_node_connection {
struct aki_thread thread;
};
+enum {
+ BIMU_NODE_STARTED = 1,
+ BIMU_NODE_SEEKED = 1 << 1,
+ BIMU_NODE_PAUSED = 1 << 2
+};
+
struct bmu_node {
u16 id;
struct cch_entry *entry;
- bool paused;
+ u8 flags;
u64 start;
u64 seek_pos;
u16 connection_id;
diff --git a/src/bimu/vcr.c b/src/bimu/vcr.c
new file mode 100644
index 0000000..6c39d86
--- /dev/null
+++ b/src/bimu/vcr.c
@@ -0,0 +1,166 @@
+#include <al/log.h>
+
+#include "vcr.h"
+#include "handler.h"
+
+static void signal_callback(void *userdata)
+{
+ struct bmu_vcr *vcr = (struct bmu_vcr *)userdata;
+ aki_packet_stream_cork(vcr->data, false);
+}
+
+void bmu_vcr_init(struct bmu_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data)
+{
+ al_array_init(vcr->streams);
+ al_atomic_u64_store(&vcr->count, 0, AL_ATOMIC_RELAXED);
+ vcr->data = data;
+ aki_signal_init(&vcr->signal, signal_callback, vcr);
+ aki_signal_start(&vcr->signal, loop);
+}
+
+static aki_thread_result AKI_THREADCALL vcr_stream_thread(void *userdata)
+{
+ struct bmu_vcr_stream *stream = (struct bmu_vcr_stream *)userdata;
+ struct bmu_vcr *vcr = stream->vcr;
+ do {
+ if (!aki_packet_cache_wait(&stream->cache)) break;
+ struct aki_packet *packet = aki_packet_cache_pop(&stream->cache);
+ if (!packet) {
+ stream->client->handle_eof(stream->client);
+ break;
+ }
+ if (al_atomic_u64_sub(&vcr->count, 1, AL_ATOMIC_RELAXED) < 128 && stream->buffered) {
+ aki_signal_send(&vcr->signal);
+ }
+ if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == BIMU_STREAM_STOPPED) {
+ aki_mutex_lock(&stream->mutex);
+ aki_cond_wait(&stream->cond, &stream->mutex);
+ aki_mutex_unlock(&stream->mutex);
+ }
+ if (al_atomic_s32_load(&stream->state, AL_ATOMIC_RELAXED) == BIMU_STREAM_CLOSED) {
+ aki_packet_free(packet);
+ break;
+ }
+ stream->client->handle_data_packet(stream->client, packet);
+ aki_packet_free(packet);
+ } while (1);
+ return 0;
+}
+
+void bmu_vcr_add_stream(struct bmu_vcr *vcr, struct bmu_vcr_stream *stream)
+{
+ stream->vcr = vcr;
+ aki_cond_init(&stream->cond);
+ aki_mutex_init(&stream->mutex);
+ aki_packet_cache_init(&stream->cache, 256);
+ stream->buffered = false;
+ al_array_push(vcr->streams, stream);
+ al_atomic_s32_store(&stream->state, BIMU_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+}
+
+static struct bmu_vcr_stream *get_stream_from_index(struct bmu_vcr *vcr, s32 index)
+{
+ struct bmu_vcr_stream *stream;
+ al_array_foreach(vcr->streams, i, stream) {
+ if (stream->index == index) return stream;
+ }
+ return NULL;
+}
+
+bool bmu_vcr_push_packet(struct bmu_vcr *vcr, struct aki_packet *packet)
+{
+ struct bmu_vcr_stream *stream;
+ switch (aki_packet_read_u8(packet)) {
+ case BIMU_PACKET_DATA:
+ stream = get_stream_from_index(vcr, aki_packet_read_s32(packet));
+ al_assert(stream);
+ if (!stream->running) {
+ aki_thread_create(&stream->thread, vcr_stream_thread, stream);
+ stream->running = true;
+ }
+ if (!aki_packet_cache_send_packet(&stream->cache, packet)) {
+ aki_packet_free(packet);
+ } else if (al_atomic_u64_add(&vcr->count, 1, AL_ATOMIC_RELAXED) >= 196) {
+ aki_packet_stream_cork(vcr->data, true);
+ al_array_foreach(vcr->streams, i, stream) {
+ stream->buffered = true;
+ }
+ }
+ break;
+ case BIMU_PACKET_EOF:
+ al_array_foreach(vcr->streams, i, stream) {
+ aki_packet_cache_send_packet(&stream->cache, NULL);
+ }
+ aki_packet_free(packet);
+ break;
+ case BIMU_PACKET_ERROR:
+ aki_packet_free(packet);
+ al_log_warn("bimu", "Unhandled error packet.");
+ break;
+ }
+ return true;
+}
+
+static void vcr_stream_close_internal(struct bmu_vcr_stream *stream)
+{
+ al_atomic_s32_store(&stream->state, BIMU_STREAM_CLOSED, AL_ATOMIC_RELAXED);
+ aki_packet_cache_disable(&stream->cache);
+ aki_mutex_lock(&stream->mutex);
+ if (aki_cond_is_waiting(&stream->cond)) {
+ aki_cond_signal(&stream->cond);
+ }
+ aki_mutex_unlock(&stream->mutex);
+ aki_thread_join(&stream->thread);
+ struct aki_packet *packet;
+ while ((packet = aki_packet_cache_pop(&stream->cache))) {
+ aki_packet_free(packet);
+ }
+}
+
+void bmu_vcr_stream_close(struct bmu_vcr_stream *stream)
+{
+ vcr_stream_close_internal(stream);
+}
+
+void bmu_vcr_flush(struct bmu_vcr *vcr)
+{
+ struct bmu_vcr_stream *stream;
+ al_array_foreach(vcr->streams, i, stream) {
+ vcr_stream_close_internal(stream);
+ stream->client->flush(stream->client);
+ aki_packet_cache_enable(&stream->cache);
+ stream->running = false;
+ al_atomic_s32_store(&stream->state, BIMU_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+ }
+ al_atomic_u64_store(&vcr->count, 0, AL_ATOMIC_RELAXED);
+}
+
+void bmu_vcr_stream_cork(struct bmu_vcr_stream *stream)
+{
+ al_atomic_s32_store(&stream->state, BIMU_STREAM_STOPPED, AL_ATOMIC_RELAXED);
+}
+
+void bmu_vcr_stream_uncork(struct bmu_vcr_stream *stream)
+{
+ al_atomic_s32_store(&stream->state, BIMU_STREAM_RUNNING, AL_ATOMIC_RELAXED);
+ aki_mutex_lock(&stream->mutex);
+ if (aki_cond_is_waiting(&stream->cond)) {
+ aki_cond_signal(&stream->cond);
+ }
+ aki_mutex_unlock(&stream->mutex);
+}
+
+void bmu_vcr_free(struct bmu_vcr *vcr)
+{
+ aki_signal_stop(&vcr->signal);
+ struct bmu_vcr_stream *stream;
+ al_array_foreach(vcr->streams, i, stream) {
+ aki_packet_cache_free(&stream->cache);
+ stream->client->free(&stream->client);
+ if (stream->stream.type == CAMU_FFMPEG_COMPAT) {
+ avformat_free_context(stream->stream.av.format_context);
+ }
+ al_free(stream);
+ }
+ al_array_free(vcr->streams);
+}
diff --git a/src/bimu/vcr.h b/src/bimu/vcr.h
new file mode 100644
index 0000000..5a6d8d5
--- /dev/null
+++ b/src/bimu/vcr.h
@@ -0,0 +1,45 @@
+#pragma once
+
+#include <aki/packet_cache.h>
+#include <al/atomic.h>
+#include <aki/packet_stream.h>
+#include <aki/signal.h>
+
+#include "../codec/codec.h"
+
+enum {
+ BIMU_STREAM_RUNNING = 0,
+ BIMU_STREAM_STOPPED,
+ BIMU_STREAM_CLOSED
+};
+
+struct bmu_vcr_stream {
+ u8 type;
+ s32 index;
+ struct camu_stream stream;
+ struct bmu_client_handler *client;
+ atomic_s32 state;
+ struct aki_packet_cache cache;
+ bool buffered;
+ struct aki_cond cond;
+ struct aki_mutex mutex;
+ bool running;
+ struct aki_thread thread;
+ struct bmu_vcr *vcr;
+};
+
+struct bmu_vcr {
+ array(struct bmu_vcr_stream *) streams;
+ atomic_u64 count;
+ struct aki_packet_stream *data;
+ struct aki_signal signal;
+};
+
+void bmu_vcr_init(struct bmu_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data);
+void bmu_vcr_add_stream(struct bmu_vcr *vcr, struct bmu_vcr_stream *stream);
+bool bmu_vcr_push_packet(struct bmu_vcr *vcr, struct aki_packet *packet);
+void bmu_vcr_stream_close(struct bmu_vcr_stream *stream);
+void bmu_vcr_flush(struct bmu_vcr *vcr);
+void bmu_vcr_stream_cork(struct bmu_vcr_stream *stream);
+void bmu_vcr_stream_uncork(struct bmu_vcr_stream *stream);
+void bmu_vcr_free(struct bmu_vcr *vcr);
diff --git a/src/buffer/audio.c b/src/buffer/audio.c
index ffb10dc..3daeee2 100644
--- a/src/buffer/audio.c
+++ b/src/buffer/audio.c
@@ -1,17 +1,19 @@
#include <al/log.h>
#include "audio.h"
+#include "common.h"
#define BUFFER_USEC (12 * 1000000L)
#define BUFFER_WATERMARK_LOW (3 * 1000000L) // Must be a most half of the buffer size.
-#define BUFFER_WATERMARK_HIGH (0.75 * 1000000L)
+#define BUFFER_WATERMARK_HIGH (4 * 1000000L)
#define DESYNC_PTS 0.022
#ifdef CAMU_AUDIO_BUFFER_FADE
#define FADE 1.0
-#define FADE_LENGTH 6
-#define FADE_STEP(rate) ((FADE / ((f64)rate)) / (FADE_LENGTH - 1.0))
+#define FADE_LENGTH 18000
+#define FADE_STEP(rate) (FADE / FADE_LENGTH)
+//#define FADE_STEP(rate) ((FADE / FADE_LENGTH) / (FADE_LENGTH - 1.0))
#endif
enum {
@@ -22,7 +24,7 @@ enum {
enum {
PAUSE_PRE = 0,
- PAUSE_UNPAUSED,
+ PAUSE_PAUSED,
#ifdef CAMU_AUDIO_BUFFER_FADE
PAUSE_FADING,
#endif
@@ -33,7 +35,8 @@ static bool setup_optimal_resampler(struct camu_audio_buffer *buf, struct camu_l
struct camu_lav_resample_fmt *fmt)
{
/*
- switch ((enum AVSampleFormat)fmt->in_format) {
+ sw
+ // Update node start?itch ((enum AVSampleFormat)fmt->in_format) {
case AV_SAMPLE_FMT_S16P:
fmt->req_format = AV_SAMPLE_FMT_S16;
break;
@@ -56,9 +59,11 @@ static bool setup_optimal_resampler(struct camu_audio_buffer *buf, struct camu_l
}
fmt->req_format = fmt->in_format;
+ fmt->req_channel_count = fmt->in_channel_count;
fmt->req_sample_rate = fmt->in_sample_rate;
*/
+ //buf->fade_rate = buf->fmt.req_sample_rate / 10;
buf->fade_rate = buf->fmt.req_sample_rate / 10;
buf->bytes_per_sample = (s32)camu_lav_resample_fmt_bytes_per_sample(fmt);
@@ -70,9 +75,9 @@ bool camu_audio_buffer_init(struct camu_audio_buffer *buf, struct camu_clock *cl
{
buf->mixer = mixer;
buf->clock = clock;
- buf->buffered = false;
buf->pause = PAUSE_PRE;
buf->ignore_desync = false;
+ buf->buffered = false;
#ifdef CAMU_AUDIO_BUFFER_FADE
buf->fade_period = 0;
buf->fade_offset = 0;
@@ -93,7 +98,6 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_stre
buf->fmt.in_sample_rate = stream->av.stream->codecpar->sample_rate;
camu_mixer_pick_format(buf->mixer, &buf->fmt);
- av_channel_layout_default(&buf->fmt.req_channel_layout, buf->fmt.req_channel_count);
if (!setup_optimal_resampler(buf, &buf->resamp, &buf->fmt)) {
return false;
}
@@ -142,8 +146,7 @@ static bool push_internal(struct camu_audio_buffer *buf, u8 **data, s32 sample_c
camu_peak_buffer_push(&buf->peak, data[0], have);
if (peak >= buf->watermark.low) {
// Discard the peak buffer. If this happens the user of this buffer is
- // taking _way_ too long to stop. We could attempt to recover by
- // re-seeking the whole stream.
+ // taking _way_ too long to stop.
camu_peak_buffer_flush(&buf->peak, &size);
}
// Continuing based on an outdated continue_mark value is safe as long as the
@@ -167,12 +170,13 @@ static bool push_internal(struct camu_audio_buffer *buf, u8 **data, s32 sample_c
#ifdef HAVE_FFMPEG
static void push_av_frame_internal(struct camu_audio_buffer *buf, AVFrame *frame)
{
+ // It doesn't matter if there's a race here with flush(), as long
+ // as any frame is from the same stream.
+ // A reset() must finish before any data is pushed.
if (al_atomic_u8_load(&buf->flow, AL_ATOMIC_RELAXED) == FLOWING) {
- u8 **data = frame->data;
- s32 sample_count = frame->nb_samples;
AVStream *stream = buf->stream->av.stream;
f64 pts = (frame->best_effort_timestamp - stream->start_time) * av_q2d(stream->time_base);
- push_internal(buf, data, sample_count, pts);
+ push_internal(buf, frame->data, frame->nb_samples, pts);
}
av_frame_free(&frame);
}
@@ -198,8 +202,8 @@ void camu_audio_buffer_reset(struct camu_audio_buffer *buf)
al_atomic_size_t_store(&buf->continue_mark, 0, AL_ATOMIC_RELAXED);
al_atomic_u8_store(&buf->flow, FLOWING, AL_ATOMIC_RELAXED);
buf->pts = camu_clock_get_base_pts(buf->clock) + camu_mixer_get_latency(buf->mixer);
- buf->buffered = false;
buf->pause = PAUSE_PRE;
+ buf->buffered = false;
al_ring_buffer_reset(&buf->rb);
}
@@ -248,45 +252,48 @@ static void handle_fade(struct camu_audio_buffer *buf, u8 *data, size_t size, bo
if ((i + 1) % channel_count == 0) {
if (out) buf->volume = AL_CLAMP((buf->volume - FADE_STEP(buf->fade_rate)), 0.0, 1.0);
else buf->volume = AL_CLAMP((buf->volume + FADE_STEP(buf->fade_rate)), 0.0, 1.0);
+ if (buf->fade_period > 0) buf->fade_period--;
}
data += buf->bytes_per_sample;
}
}
#endif
-#define NOT_UNPAUSED(pause) (pause != PAUSE_UNPAUSED && pause != PAUSE_PRE)
+#define NOT_PAUSED(pause) (pause != PAUSE_PAUSED && pause != PAUSE_PRE)
+// PAUSE_PAUSED means the last read was silence, meaning it's safe to skip.
+// The sole purpose of PAUSE_PRE is to avoid a fade in at the start of the stream.
size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t req)
{
- f64 pts = camu_clock_get_pts(buf->clock);
+ f64 pts = camu_clock_get_pts(buf->clock, camu_mixer_get_latency(buf->mixer));
if (pts < 0.0) {
#ifdef CAMU_AUDIO_BUFFER_FADE
if (buf->pause == PAUSE_PLAYING) {
- // Fade out. Signified by >0 fade_period and PAUSE_FADING.
+ // Fade out. Signified by >0 fade_period and pause = PAUSE_FADING.
buf->volume = FADE;
buf->fade_period = FADE_LENGTH;
buf->fade_offset = 0;
buf->pause = PAUSE_FADING;
- } else if (buf->fade_period == 0) {
+ } else if (buf->fade_period == 0) { // Fade out done.
al_memset(data, 0, req);
- if (NOT_UNPAUSED(buf->pause)) {
- buf->pause = PAUSE_UNPAUSED;
+ if (NOT_PAUSED(buf->pause)) {
+ buf->pause = PAUSE_PAUSED;
buf->callback(buf->userdata, CAMU_BUFFER_PAUSED);
}
return req;
}
- } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_UNPAUSED) {
- // Fade in. Signified by >0 fade_period and PAUSE_PLAYING.
+ } else if (buf->pause == PAUSE_FADING || buf->pause == PAUSE_PAUSED) {
+ // Fade in. Signified by >0 fade_period and pause = PAUSE_PLAYING.
buf->volume = 1.0 - FADE;
buf->fade_period = FADE_LENGTH;
buf->fade_offset = 0;
// We don't want to consider sync if reversing an active
- // fade but we do if PAUSE_UNPAUSED.
+ // fade but we do if pause = PAUSE_PAUSED.
if (buf->pause == PAUSE_FADING) buf->pause = PAUSE_PLAYING;
#else
al_memset(data, 0, req);
- if (NOT_UNPAUSED(buf->pause)) {
- buf->pause = PAUSE_UNPAUSED;
+ if (NOT_PAUSED(buf->pause)) {
+ buf->pause = PAUSE_PAUSED;
buf->callback(buf->userdata, CAMU_BUFFER_PAUSED);
}
return req;
@@ -294,15 +301,16 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
}
size_t ret, signal = req;
size_t size = al_ring_buffer_occupied(&buf->rb);
+ // Cut off fade if it's reaching too far.
if (size < buf->fade_offset) size = 0;
else size -= buf->fade_offset;
#ifdef CAMU_AUDIO_BUFFER_FADE
- if (buf->pause != PAUSE_FADING) {
+ if (buf->pause != PAUSE_FADING && buf->pause != PAUSE_PLAYING) {
#endif
pts -= buf->pts;
- // We can assume a call to audio_buffer_read will happen before
- // the user NEEDS the data, so, we can't accurately assess the sync here.
- if (UNLIKELY(!buf->ignore_desync && (buf->pause == PAUSE_PRE || buf->pause == PAUSE_UNPAUSED))) {
+ // Possibly attempt syncing to the clock. For this to work the mixer must
+ // report a reasonably accurate value for latency.
+ if (UNLIKELY(!buf->ignore_desync && (buf->pause == PAUSE_PRE || buf->pause == PAUSE_PAUSED))) {
if (fabs(pts) >= 0.322) {
al_log_warn("audio_buffer", "Abnormally large audio desync of %.5fs", pts);
}
@@ -330,31 +338,42 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
}
#endif
u8 flow = al_atomic_u8_load(&buf->flow, AL_ATOMIC_RELAXED);
- if (size < req) {
+ if (size < req) { // We don't have enough data to fulfill our request.
if (flow == FLUSHED) {
+ // If flow = FLUSHED, we know we've recieved all the data
+ // this stream wants to send. Signal that we are done.
signal = size;
al_atomic_u8_store(&buf->flow, SIGNALED, AL_ATOMIC_RELAXED);
al_log_debug("audio_buffer", "Flushed (signal: %zu).", signal);
buf->callback(buf->userdata, CAMU_BUFFER_EOF);
} else {
+ // Put silence into the remainder of the request buffer.
al_memset(data + size, 0, req - size);
if (flow == FLOWING) {
+ // Our stream hit an underrun because data wasn't coming in fast
+ // enough. It's also possible to underrun if this function takes too
+ // long which isn't checked here. That is unlikely though.
al_log_warn("audio_buffer", "Underrun (req: %zu, have: %zu).", req, size);
}
+ // If flow = SIGNALED, we are just waiting to be removed or reset.
}
req = size;
}
if (req > 0) {
#ifdef CAMU_AUDIO_BUFFER_FADE
if (buf->pause == PAUSE_FADING) {
+ // Peek into the ring buffer. This won't consume the data, so when
+ // we resume it will fade in from the same position the fade out started.
ret = al_ring_buffer_peek(&buf->rb, data, buf->fade_offset, req);
buf->fade_offset += ret;
} else {
#endif
+ // Read as much as we can from the ring buffer.
ret = al_ring_buffer_read(&buf->rb, data, req);
buf->pts += camu_lav_resample_fmt_bytes_to_sec(&buf->fmt, ret);
if (flow == FLOWING) {
ret = al_atomic_size_t_load(&buf->continue_mark, AL_ATOMIC_RELAXED);
+ // Request to uncork if we are below the mark.
if (ret && ((buf->size - size) - req) >= ret) {
buf->callback(buf->userdata, CAMU_BUFFER_UNCORK);
}
@@ -362,11 +381,11 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
#ifdef CAMU_AUDIO_BUFFER_FADE
}
if (buf->fade_period > 0) {
- buf->fade_period--;
handle_fade(buf, data, req, buf->pause == PAUSE_FADING);
}
#endif
}
+ // Never return less than req, unless we hit the end of the stream.
return signal;
}
diff --git a/src/buffer/audio.h b/src/buffer/audio.h
index 7bd30a2..1a8baf0 100644
--- a/src/buffer/audio.h
+++ b/src/buffer/audio.h
@@ -6,13 +6,11 @@
#include "../codec/codec.h"
#include "../codec/libav/resampler.h"
#include "../mixer/mixer.h"
-#include "../mixer/audio.h"
#include "clock.h"
-#include "common.h"
#include "peak_buffer.h"
-#define CAMU_AUDIO_BUFFER_FADE
+//#define CAMU_AUDIO_BUFFER_FADE
struct camu_audio_buffer {
struct camu_stream *stream;
diff --git a/src/buffer/clock.c b/src/buffer/clock.c
index a1c3e83..1e7883c 100644
--- a/src/buffer/clock.c
+++ b/src/buffer/clock.c
@@ -2,26 +2,38 @@
#include "clock.h"
-void camu_clock_set(struct camu_clock *clock, u64 base, u64 start)
+void camu_clock_set(struct camu_clock *clock, u64 base, u64 ts, u64 delay)
{
al_atomic_f64_store(&clock->base, base / 1000000.f, AL_ATOMIC_RELAXED);
al_atomic_f64_store(&clock->tick, 0.0, AL_ATOMIC_RELAXED);
- al_atomic_u64_store(&clock->start, start, AL_ATOMIC_RELAXED);
- camu_clock_resume(clock);
+ al_atomic_u64_store(&clock->start, ts, AL_ATOMIC_RELAXED);
+ clock->delay = delay;
}
-void camu_clock_resume(struct camu_clock *clock)
+bool camu_clock_calc_tick(struct camu_clock *clock)
{
- u64 start = al_atomic_u64_load(&clock->start, AL_ATOMIC_ACQUIRE);
- if (start == 0L) return;
- u64 ts = aki_get_timestamp();
- al_assert(start < ts);
- f64 diff = (ts - start) / 1000000.f;
+ u64 start = al_atomic_u64_load(&clock->start, AL_ATOMIC_RELAXED);
+ if (start == 0L) return false;
+ u64 ts = clock->delay == 0 ? start : aki_get_timestamp();
+ al_assert(start <= ts);
+ u64 diff = ts - start;
f64 tick = al_atomic_f64_load(&clock->tick, AL_ATOMIC_RELAXED);
if (tick == 0.0) {
+ if (!(diff >= clock->delay && clock->delay >= (diff - clock->delay))) {
+ // We are past the requested start time.
+ return false;
+ }
+ diff = clock->delay - (diff - clock->delay);
tick = aki_get_tick() - al_atomic_f64_load(&clock->base, AL_ATOMIC_RELAXED);
}
- al_atomic_f64_store(&clock->tick, tick + diff, AL_ATOMIC_RELAXED);
+ al_atomic_f64_store(&clock->tick, tick + (diff / 1000000.f), AL_ATOMIC_RELAXED);
+ return true;
+}
+
+void camu_clock_resume(struct camu_clock *clock)
+{
+ u64 start = al_atomic_u64_load(&clock->start, AL_ATOMIC_ACQUIRE);
+ if (start == 0L) return;
al_atomic_u64_store(&clock->start, 0L, AL_ATOMIC_RELEASE);
}
@@ -39,13 +51,6 @@ void camu_clock_pause(struct camu_clock *clock)
al_atomic_u64_store(&clock->start, aki_get_timestamp(), AL_ATOMIC_RELEASE);
}
-void camu_clock_seek(struct camu_clock *clock, u64 base, u64 start)
-{
- (void)clock;
- (void)base;
- (void)start;
-}
-
bool camu_clock_is_paused(struct camu_clock *clock)
{
u64 start = al_atomic_u64_load(&clock->start, AL_ATOMIC_RELAXED);
@@ -57,12 +62,12 @@ f64 camu_clock_get_base_pts(struct camu_clock *clock)
return al_atomic_f64_load(&clock->base, AL_ATOMIC_RELAXED);
}
-f64 camu_clock_get_pts(struct camu_clock *clock)
+f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset)
{
u64 start = al_atomic_u64_load(&clock->start, AL_ATOMIC_RELAXED);
if (start != 0L) return -1.0;
f64 base = al_atomic_f64_load(&clock->base, AL_ATOMIC_RELAXED);
- f64 pts = get_pts_internal(clock);
+ f64 pts = get_pts_internal(clock) + offset;
if (pts < base) return -1.0;
return pts;
}
diff --git a/src/buffer/clock.h b/src/buffer/clock.h
index e39a963..a7399b5 100644
--- a/src/buffer/clock.h
+++ b/src/buffer/clock.h
@@ -6,12 +6,13 @@
struct camu_clock {
atomic_f64 base, tick;
atomic_u64 start;
+ u64 delay;
};
-void camu_clock_set(struct camu_clock *clock, u64 base, u64 start);
+void camu_clock_set(struct camu_clock *clock, u64 base, u64 ts, u64 delay);
+bool camu_clock_calc_tick(struct camu_clock *clock);
void camu_clock_resume(struct camu_clock *clock);
void camu_clock_pause(struct camu_clock *clock);
-void camu_clock_seek(struct camu_clock *clock, u64 base, u64 start);
bool camu_clock_is_paused(struct camu_clock *clock);
f64 camu_clock_get_base_pts(struct camu_clock *clock);
-f64 camu_clock_get_pts(struct camu_clock *clock);
+f64 camu_clock_get_pts(struct camu_clock *clock, f64 offset);
diff --git a/src/buffer/video.c b/src/buffer/video.c
index 26e84b0..9a04933 100644
--- a/src/buffer/video.c
+++ b/src/buffer/video.c
@@ -4,17 +4,17 @@
//#define CAMU_VIDEO_BUFFER_FORCE_SCALER
-#define BUFFER_WATERMARK_LOW 16 // frames.
-#define BUFFER_WATERMARK_BUFFERED 12
-#define BUFFER_WATERMARK_HIGH 28
-#define BUFFER_WATERMARK_RESET BUFFER_WATERMARK_HIGH + 16.
+#define BUFFER_WATERMARK_LOW 8 // frames.
+#define BUFFER_WATERMARK_BUFFERED 10
+#define BUFFER_WATERMARK_HIGH 14
+#define BUFFER_WATERMARK_RESET BUFFER_WATERMARK_HIGH + 10.
bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *clock,
struct camu_renderer *renderer)
{
+ buf->clock = clock;
buf->queue = renderer->create_queue(renderer);
buf->buffered = false;
- buf->clock = clock;
#ifdef CAMU_SCREEN_THREADED
al_atomic_bool_store(&buf->ref, false, AL_ATOMIC_RELAXED);
#endif
@@ -132,11 +132,15 @@ bool camu_video_buffer_is_single_frame(struct camu_video_buffer *buf)
void camu_video_buffer_flush(struct camu_video_buffer *buf)
{
buf->queue->flush(buf->queue);
+ if (!buf->buffered) {
+ buf->callback(buf->userdata, CAMU_BUFFER_BUFFERED);
+ buf->buffered = true;
+ }
}
bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out)
{
- f64 pts = buf->single_frame ? 0.0 : camu_clock_get_pts(buf->clock);
+ f64 pts = buf->single_frame ? 0.0 : camu_clock_get_pts(buf->clock, 0);
if (pts < 0.0) pts = camu_clock_get_base_pts(buf->clock);
u8 ret = buf->queue->read(buf->queue, pts, out);
if (ret == CAMU_QUEUE_EOF || (ret == CAMU_QUEUE_OK && buf->single_frame)) {
diff --git a/src/codec/libav/decoder.c b/src/codec/libav/decoder.c
index ebdcb03..3a6ac06 100644
--- a/src/codec/libav/decoder.c
+++ b/src/codec/libav/decoder.c
@@ -38,11 +38,13 @@ static bool lav_decoder_init(struct camu_decoder *dec, struct camu_renderer *ren
goto err;
}
+ /*
s32 cpus = av_cpu_count();
cpus = (cpus > 16) ? 4 : cpus / 4;
if (cpus == 0) cpus = 1;
av->codec_context->thread_count = cpus;
al_log_debug("lav_decoder", "Using %i threads for decoder.", cpus);
+ */
(void)renderer;
/*
diff --git a/src/codec/libav/demuxer.c b/src/codec/libav/demuxer.c
index 18b9020..fb05ec8 100644
--- a/src/codec/libav/demuxer.c
+++ b/src/codec/libav/demuxer.c
@@ -83,9 +83,7 @@ static bool lav_demuxer_init(struct camu_demuxer *demux, struct cch_handle *hand
}
al_log_info("lav_demux", "Setting stream #%i to a duration of %.3fs.", i, stream->duration * av_q2d(stream->time_base));
}
- if (stream->start_time < 0) {
- al_log_warn("lav_demux", "Stream has a negative start_time (%ld).", stream->start_time);
- }
+ if (stream->start_time < 0) stream->start_time = 0;
s64 duration = av_rescale_q(stream->duration, stream->time_base, AV_TIME_BASE_Q);
al_assert(duration >= 0);
if ((u64)duration > av->duration) av->duration = (u64)duration;
diff --git a/src/codec/libav/resampler.c b/src/codec/libav/resampler.c
index ba700f4..229c68c 100644
--- a/src/codec/libav/resampler.c
+++ b/src/codec/libav/resampler.c
@@ -8,40 +8,38 @@
#include "resampler.h"
-static bool in_req_match(struct camu_lav_resample_fmt *fmt)
+static bool in_req_matches(struct camu_lav_resample_fmt *fmt)
{
- return (av_channel_layout_compare(&fmt->in_channel_layout, &fmt->req_channel_layout) == 0) &&
- (fmt->in_format == fmt->req_format) && (fmt->in_sample_rate == fmt->req_sample_rate);
+ return (fmt->in_format == fmt->req_format) &&
+ (av_channel_layout_compare(&fmt->in_channel_layout, &fmt->req_channel_layout) == 0) &&
+ (fmt->in_sample_rate == fmt->req_sample_rate);
}
bool camu_lav_resampler_init(struct camu_lav_resampler *resamp, struct camu_lav_resample_fmt *fmt)
{
al_memset(resamp, 0, sizeof(struct camu_lav_resampler));
- fmt->resampler_needed = !in_req_match(fmt);
+ fmt->resampler_needed = !in_req_matches(fmt);
+ if (!fmt->resampler_needed) return true;
camu_lav_resample_fmt_copy(&resamp->fmt, fmt);
- if (!fmt->resampler_needed) {
- return true;
- }
-
- resamp->resample_context = swr_alloc();
+ SwrContext *ctx = NULL;
+ s32 ret = swr_alloc_set_opts2(&ctx,
+ &fmt->req_channel_layout,
+ fmt->req_format,
+ fmt->req_sample_rate,
+ &fmt->in_channel_layout,
+ fmt->in_format,
+ fmt->in_sample_rate,
+ 0,
+ NULL);
- SwrContext *ctx = resamp->resample_context;
-
- if (!ctx) {
- al_log_error("lav_resampler", "Failed to create resampler context.");
+ if (ret != 0) {
+ al_log_error("lav_resampler", "Failed to configure resampler context.");
return false;
}
- av_opt_set_chlayout(ctx, "in_chlayout", &fmt->in_channel_layout, 0);
- av_opt_set_chlayout(ctx, "out_chlayout", &fmt->req_channel_layout, 0);
- av_opt_set_int(ctx, "in_sample_rate", fmt->in_sample_rate, 0);
- av_opt_set_int(ctx, "out_sample_rate", fmt->req_sample_rate, 0);
- av_opt_set_sample_fmt(ctx, "in_sample_fmt", fmt->in_format, 0);
- av_opt_set_sample_fmt(ctx, "out_sample_fmt", fmt->req_format, 0);
-
// Slower but more accurate resampler.
//av_opt_set_int(ctx, "resampler", SWR_ENGINE_SOXR, 0);
@@ -51,13 +49,16 @@ bool camu_lav_resampler_init(struct camu_lav_resampler *resamp, struct camu_lav_
return false;
}
+ resamp->resample_context = ctx;
+
return true;
}
s32 camu_lav_resampler_sample_count(struct camu_lav_resampler *resamp, s32 in_samples)
{
- al_assert(resamp->fmt.resampler_needed);
return swr_get_out_samples(resamp->resample_context, in_samples);
+ //return av_rescale_rnd(swr_get_delay(resamp->resample_context, resamp->fmt.req_sample_rate) +
+ // in_samples, resamp->fmt.in_sample_rate, resamp->fmt.req_sample_rate, AV_ROUND_UP);
}
static inline void free_sample_data(struct camu_lav_resampler *resamp)
@@ -70,15 +71,14 @@ static inline void alloc_sample_data(struct camu_lav_resampler *resamp, s32 samp
if (resamp->sample_count >= sample_count) return;
free_sample_data(resamp);
s32 nb_channels = resamp->fmt.req_channel_layout.nb_channels;
- av_samples_alloc(resamp->data, NULL, nb_channels, sample_count, resamp->fmt.req_format, 0);
+ av_samples_alloc(resamp->data, NULL, nb_channels, sample_count, resamp->fmt.req_format, 32);
resamp->sample_count = sample_count;
}
s32 camu_lav_resampler_convert(struct camu_lav_resampler *resamp, const u8 **in_data, s32 in_samples)
{
- al_assert(resamp->fmt.resampler_needed);
- s32 max_resample_count = in_samples ?
- camu_lav_resampler_sample_count(resamp, in_samples) : resamp->sample_count;
+ al_assert(in_samples > 0);
+ s32 max_resample_count = camu_lav_resampler_sample_count(resamp, in_samples);
alloc_sample_data(resamp, max_resample_count);
s32 resample_count = swr_convert(resamp->resample_context, resamp->data,
max_resample_count, in_data, in_samples);
@@ -95,7 +95,6 @@ u8 **camu_lav_resampler_get_data(struct camu_lav_resampler *resamp)
void camu_lav_resampler_close(struct camu_lav_resampler *resamp)
{
- al_assert(resamp->fmt.resampler_needed);
if (resamp->resample_context) {
swr_close(resamp->resample_context);
swr_free(&resamp->resample_context);
@@ -105,13 +104,13 @@ void camu_lav_resampler_close(struct camu_lav_resampler *resamp)
void camu_lav_resample_fmt_copy(struct camu_lav_resample_fmt *dest, struct camu_lav_resample_fmt *src)
{
+ dest->resampler_needed = src->resampler_needed;
dest->in_format = src->in_format;
av_channel_layout_copy(&dest->in_channel_layout, &src->in_channel_layout);
dest->in_sample_rate = src->in_sample_rate;
dest->req_format = src->req_format;
av_channel_layout_copy(&dest->req_channel_layout, &src->req_channel_layout);
dest->req_sample_rate = src->req_sample_rate;
- dest->resampler_needed = src->resampler_needed;
}
size_t camu_lav_resample_fmt_bytes_per_sample(struct camu_lav_resample_fmt *fmt)
diff --git a/src/codec/libav/resampler.h b/src/codec/libav/resampler.h
index 703d9a7..92525b2 100644
--- a/src/codec/libav/resampler.h
+++ b/src/codec/libav/resampler.h
@@ -5,15 +5,15 @@
#include <al/types.h>
struct camu_lav_resample_fmt {
+ bool resampler_needed;
+ enum AVSampleFormat in_format;
+ enum AVSampleFormat req_format;
s32 in_channel_count;
s32 req_channel_count;
AVChannelLayout in_channel_layout;
AVChannelLayout req_channel_layout;
s32 in_sample_rate;
s32 req_sample_rate;
- enum AVSampleFormat in_format;
- enum AVSampleFormat req_format;
- bool resampler_needed;
};
struct camu_lav_resampler {
diff --git a/src/codec/meson.build b/src/codec/meson.build
index 0946a17..20524a4 100644
--- a/src/codec/meson.build
+++ b/src/codec/meson.build
@@ -11,13 +11,17 @@ av_server_src = [
av_client_src = [
av_src,
'libav/decoder.c',
- 'libav/resampler.c',
- 'libav/scaler.c'
+ 'libav/resampler.c'
]
-av_inc = []
av_ldflags = []
+av_inc = []
av_server_deps = []
av_client_deps = []
+av_client_use_scaler = false
+
+if av_client_use_scaler
+ av_client_src += ['libav/scaler.c']
+endif
libavutil = dependency('libavutil', required: false)
libavformat = dependency('libavformat', required: false)
@@ -32,9 +36,11 @@ if not (libavutil.found() and libavformat.found() and libavcodec.found() and lib
libavcodec = av_proj.get_variable('avcodec')
libswresample = av_proj.get_variable('swresample')
libswscale = av_proj.get_variable('swscale')
- av_inc += [av_proj.get_variable('inc')]
- av_ldflags += ['-L' + av_proj.get_variable('libdir')]
- av_client_deps += [dependency('zlib')]
+ av_ldflags += ['-L' + av_proj.get_variable('av_lib_dir')]
+ # Only used to set include dirs for libav.
+ av_inc_dep = av_proj.get_variable('av_inc_dep')
+ av_server_deps += [av_inc_dep]
+ av_client_deps += [av_inc_dep, dependency('zlib')]
soxr = compiler.find_library('soxr', required: false)
if soxr.found()
av_client_deps += [soxr]
@@ -45,7 +51,10 @@ endif
if libavutil.found() and libavformat.found() and libavcodec.found() and libswresample.found() and libswscale.found()
av_server_deps += [libavutil, libavformat, libavcodec]
- av_client_deps += [libavutil, libavformat, libavcodec, libswresample, libswscale]
+ av_client_deps += [libavutil, libavformat, libavcodec, libswresample]
+ if av_client_use_scaler
+ av_client_deps += [libswscale]
+ endif
av_server = declare_dependency(sources: av_server_src, dependencies: av_server_deps,
include_directories: av_inc, link_args: av_ldflags)
av_client = declare_dependency(sources: av_client_src, dependencies: av_client_deps,
diff --git a/src/fruits/cmc/cmc.c b/src/fruits/cmc/cmc.c
index 7fd8691..89800f6 100644
--- a/src/fruits/cmc/cmc.c
+++ b/src/fruits/cmc/cmc.c
@@ -764,10 +764,10 @@ s32 main(s32 argc, char *argv[])
aki_event_loop_init(&c.loop);
camu_client_init(&c.client, &c.loop, client_callback, &c);
- camu_client_login(&c.client, al_str_c("andrew"), al_str_c("127.0.0.1"), TREE_PORT);
+ camu_client_login(&c.client, al_str_c("andrew"), TREE_SERVER_IP, TREE_PORT);
camu_resource_client_init(&c.resource_client, &c.loop, resource_client_callback, &c);
- camu_resource_client_connect(&c.resource_client, al_str_c("127.0.0.1"), TREE_RESOURCE_PORT);
+ camu_resource_client_connect(&c.resource_client, TREE_SERVER_IP, TREE_RESOURCE_PORT);
aki_event_loop_run(&c.loop);
diff --git a/src/fruits/sink/sink.c b/src/fruits/sink/sink.c
index 939e0af..d20f7c4 100644
--- a/src/fruits/sink/sink.c
+++ b/src/fruits/sink/sink.c
@@ -1,8 +1,6 @@
#include <al/log.h>
#include <aki/event_loop.h>
-#include "../../codec/libav/common.h"
-
#include "../../mixer/audio_miniaudio.h"
#include "../../screen/screen.h"
@@ -12,8 +10,6 @@
#include "../../libsink/sink.h"
#include "../../tree/common.h"
-#include "../../shoki/src/search.h"
-
struct cmv {
s32 quit;
struct camu_screen scr;
@@ -100,7 +96,6 @@ static u8 sink_callback(void *userdata, u8 op, u8 type, void *opaque)
break;
case CAMU_SINK_EXIT:
camu_sink_close(&c->sink);
- aki_event_loop_break(&c->loop);
break;
}
return CAMU_SINK_OK;
@@ -166,13 +161,16 @@ s32 wmain(s32 argc, wchar_t **argv)
}
c.renderer->render(c.renderer, &c.scr);
- camu_mixer_init(&c.mixer, (struct camu_audio *)&audio_plugin_miniaudio);
- c.mixer.audio->configure_stream(c.mixer.audio, NULL);
+ if (!camu_mixer_init(&c.mixer, (struct camu_audio *)&audio_plugin_miniaudio)) {
+ goto err;
+ }
camu_sink_init(&c.sink, &c.loop, &c.mixer, c.renderer);
c.sink.callback = sink_callback;
c.sink.userdata = &c;
- camu_sink_connect(&c.sink, al_str_c("127.0.0.1"), TREE_PORT);
+ if (!camu_sink_connect(&c.sink, TREE_SERVER_IP, TREE_PORT)) {
+ goto err;
+ }
struct aki_thread thread0;
aki_thread_create(&thread0, event_loop_thread, &c);
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 58e3db3..343275b 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -1,8 +1,11 @@
#include <al/log.h>
#include "../tree/common.h"
+#include "../bimu/common.h"
+#include "../bimu/handler.h"
#include "sink.h"
+#include "common.h"
enum {
SINK_EMPTY = 0,
@@ -12,6 +15,7 @@ enum {
enum {
ENTRY_LOADED = 0,
+ ENTRY_BUFFERED,
ENTRY_DISREGUARDED
};
@@ -27,8 +31,9 @@ enum {
START,
STOP,
TOGGLE_PAUSE,
+ RESEEK,
SEEK,
- ENTRY_BUFFERED, // Currently set entry is buffered.
+ SET_BUFFERED, // Currently set entry is buffered.
CLOSE,
// Internal.
CORK,
@@ -94,6 +99,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
case TOGGLE_PAUSE: {
struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
if (camu_clock_is_paused(&entry->clock)) {
+ camu_clock_calc_tick(&entry->clock);
camu_clock_resume(&entry->clock);
if (sink->audio.state == SINK_PAUSED) {
sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL);
@@ -116,36 +122,59 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
}
break;
}
+ case RESEEK: {
+ struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+ bmu_client_reseek(&entry->client);
+ break;
+ }
case SEEK: {
struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
bmu_client_seek(&entry->client, cmd->value.f);
break;
}
- case ENTRY_BUFFERED: {
+// case SET_BUFFERED: {
+// switch (cmd->value.i) {
+// case CAMU_SINK_AUDIO:
+// sink->callback(sink->userdata, CAMU_SINK_SET_BUFFERED, CAMU_SINK_AUDIO, NULL);
+// break;
+//#ifndef CAMU_SINK_NO_VIDEO
+// case CAMU_SINK_VIDEO:
+// sink->callback(sink->userdata, CAMU_SINK_SET_BUFFERED, CAMU_SINK_VIDEO, NULL);
+// break;
+//#endif
+// }
+// break;
+// }
+ case CLOSE: {
+ sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
+ return;
+ }
+ case CORK: {
+ struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
switch (cmd->value.i) {
case CAMU_SINK_AUDIO:
- sink->callback(sink->userdata, CAMU_SINK_SET_BUFFERED, CAMU_SINK_AUDIO, NULL);
+ bmu_vcr_stream_cork(entry->audio.stream);
break;
#ifndef CAMU_SINK_NO_VIDEO
case CAMU_SINK_VIDEO:
- sink->callback(sink->userdata, CAMU_SINK_SET_BUFFERED, CAMU_SINK_VIDEO, NULL);
+ bmu_vcr_stream_cork(entry->video.stream);
break;
#endif
}
break;
}
- case CLOSE: {
- sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
- return;
- }
- case CORK: {
- struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
- bmu_client_cork(&entry->client, true);
- break;
- }
case UNCORK: {
struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
- bmu_client_cork(&entry->client, false);
+ switch (cmd->value.i) {
+ case CAMU_SINK_AUDIO:
+ bmu_vcr_stream_uncork(entry->audio.stream);
+ break;
+#ifndef CAMU_SINK_NO_VIDEO
+ case CAMU_SINK_VIDEO:
+ bmu_vcr_stream_uncork(entry->video.stream);
+ break;
+#endif
+ }
break;
}
}
@@ -173,18 +202,27 @@ static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
{
u8 state = entry->audio.state;
if (state == BUFFER_SET_OR_BUFFERED) {
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
- queue_cmd(entry->sink, (struct camu_sink_cmd){
- .op = ENTRY_BUFFERED,
- .value.i = CAMU_SINK_AUDIO
- });
+ bool should_resume = ENTRY_VIDEO_EMPTY(entry) || entry->video.state == BUFFER_ADDED;
+ if (should_resume && !camu_clock_calc_tick(&entry->clock)) {
+ queue_cmd(entry->sink, (struct camu_sink_cmd){
+ .op = RESEEK,
+ .opaque = entry
+ });
+ return;
+ }
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = START,
.value.i = CAMU_SINK_AUDIO
});
- if (ENTRY_VIDEO_EMPTY(entry) || entry->video.state == BUFFER_ADDED) {
+ if (should_resume) {
camu_clock_resume(&entry->clock);
+ entry->state = ENTRY_BUFFERED;
}
+ entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+ //queue_cmd(entry->sink, (struct camu_sink_cmd){
+ // .op = SET_BUFFERED,
+ // .value.i = CAMU_SINK_AUDIO
+ //});
state = BUFFER_ADDED;
} else if (state == BUFFER_CONFIGURED) {
state = BUFFER_SET_OR_BUFFERED;
@@ -204,12 +242,14 @@ static void audio_buffer_callback(void *userdata, u8 op)
case CAMU_BUFFER_CORK:
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = CORK,
+ .value.i = CAMU_SINK_AUDIO,
.opaque = entry
});
break;
case CAMU_BUFFER_UNCORK:
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = UNCORK,
+ .value.i = CAMU_SINK_AUDIO,
.opaque = entry
});
break;
@@ -234,10 +274,10 @@ static void audio_buffer_callback(void *userdata, u8 op)
.value.i = CAMU_SINK_AUDIO
});
#ifndef CAMU_SINK_NO_VIDEO
- queue_cmd(entry->sink, (struct camu_sink_cmd){
- .op = ENTRY_BUFFERED,
- .value.i = CAMU_SINK_VIDEO
- });
+ //queue_cmd(entry->sink, (struct camu_sink_cmd){
+ // .op = SET_BUFFERED,
+ // .value.i = CAMU_SINK_VIDEO
+ //});
#endif
}
break;
@@ -250,18 +290,27 @@ static void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
{
u8 state = entry->video.state;
if (state == BUFFER_SET_OR_BUFFERED) {
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
- queue_cmd(entry->sink, (struct camu_sink_cmd){
- .op = ENTRY_BUFFERED,
- .value.i = CAMU_SINK_VIDEO
- });
+ bool should_resume = ENTRY_AUDIO_EMPTY(entry) || entry->audio.state == BUFFER_ADDED;
+ if (should_resume && !camu_clock_calc_tick(&entry->clock)) {
+ queue_cmd(entry->sink, (struct camu_sink_cmd){
+ .op = RESEEK,
+ .opaque = entry
+ });
+ return;
+ }
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = START,
.value.i = CAMU_SINK_VIDEO
});
- if (ENTRY_AUDIO_EMPTY(entry) || entry->audio.state == BUFFER_ADDED) {
+ if (should_resume) {
camu_clock_resume(&entry->clock);
+ entry->state = ENTRY_BUFFERED;
}
+ entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+ //queue_cmd(entry->sink, (struct camu_sink_cmd){
+ // .op = SET_BUFFERED,
+ // .value.i = CAMU_SINK_VIDEO
+ //});
state = BUFFER_ADDED;
} else if (state == BUFFER_CONFIGURED) {
state = BUFFER_SET_OR_BUFFERED;
@@ -281,12 +330,14 @@ static void video_buffer_callback(void *userdata, u8 op)
case CAMU_BUFFER_CORK:
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = CORK,
+ .value.i = CAMU_SINK_VIDEO,
.opaque = entry
});
break;
case CAMU_BUFFER_UNCORK:
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = UNCORK,
+ .value.i = CAMU_SINK_VIDEO,
.opaque = entry
});
break;
@@ -300,22 +351,10 @@ static void video_buffer_callback(void *userdata, u8 op)
}
#endif
-static void client_callback(void *userdata, u8 op, struct bmu_client_stream *stream, void *opaque)
+static void client_callback(void *userdata, u8 op, struct bmu_vcr_stream *stream, void *opaque)
{
struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
switch (op) {
- case BIMU_CLIENT_SET: {
- struct bmu_seek_req *req = (struct bmu_seek_req *)opaque;
- camu_clock_set(&entry->clock, req->base, req->start);
- camu_audio_buffer_reset(&entry->audio.buf);
-#ifndef CAMU_SINK_NO_VIDEO
- bool single_frame = camu_video_buffer_is_single_frame(&entry->video.buf);
- if (!single_frame) {
- camu_video_buffer_reset(&entry->video.buf);
- }
-#endif
- break;
- }
case BIMU_CLIENT_CONFIGURE: {
// No data can be sent until all active streams are configured.
switch (stream->type) {
@@ -346,6 +385,23 @@ static void client_callback(void *userdata, u8 op, struct bmu_client_stream *str
}
break;
}
+ case BIMU_CLIENT_SET: {
+ u64 ts = aki_get_timestamp();
+ struct bmu_seek_req *req = (struct bmu_seek_req *)opaque;
+ //req->delay = 0; // Ignore delay.
+ req->ts -= req->delay;
+ bool late = ts <= req->ts;
+ if (late) {}
+ camu_clock_set(&entry->clock, req->base, req->ts, req->delay);
+ camu_audio_buffer_reset(&entry->audio.buf);
+ if (!late) {
+ // This is a fail-safe for the case where receiving data is extremely slow.
+ // It should not be depended on for checking if an entry buffers in time.
+ aki_timer_set_repeat(&entry->timer, AKI_TS_FROM_USEC(ts - req->ts));
+ aki_timer_again(&entry->timer);
+ }
+ break;
+ }
case BIMU_CLIENT_DATA: {
struct camu_frame *frame = (struct camu_frame *)opaque;
switch (stream->type) {
@@ -395,6 +451,11 @@ static void client_callback(void *userdata, u8 op, struct bmu_client_stream *str
#ifndef CAMU_SINK_NO_VIDEO
}
#endif
+#ifndef CAMU_SINK_NO_VIDEO
+ if (!single_frame) {
+ camu_video_buffer_reset(&entry->video.buf);
+ }
+#endif
break;
}
case BIMU_CLIENT_EOF: {
@@ -417,6 +478,7 @@ static void client_callback(void *userdata, u8 op, struct bmu_client_stream *str
}
case BIMU_CLIENT_CLOSED: {
bmu_client_free(&entry->client);
+ aki_timer_stop(&entry->timer);
camu_audio_buffer_free(&entry->audio.buf);
#ifndef CAMU_SINK_NO_VIDEO
camu_video_buffer_free(&entry->video.buf);
@@ -460,6 +522,15 @@ static struct camu_sink_entry *entry_from_node_id(struct camu_sink *sink, u16 no
return NULL;
}
+static void buffer_timer_callback(void *userdata, struct aki_timer *timer)
+{
+ struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
+ (void)timer;
+ aki_mutex_lock(&entry->sink->mutex);
+ aki_mutex_unlock(&entry->sink->mutex);
+ aki_timer_stop(&entry->timer);
+}
+
static bool buffer_command_callback(void *userdata, struct aki_rpc_connection *conn,
struct aki_packet *packet, struct aki_packet *rpacket)
{
@@ -492,11 +563,14 @@ static bool buffer_command_callback(void *userdata, struct aki_rpc_connection *c
entry->video.buf.userdata = entry;
#endif
+ aki_timer_init(&entry->timer, sink->loop, buffer_timer_callback, entry);
+
entry->client.callback = client_callback;
entry->client.userdata = entry;
bmu_client_connect(&entry->client, sink->loop, &addr, port, node_id);
aki_packet_free(packet);
+
return false;
}
@@ -524,6 +598,7 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
u16 node_id = aki_packet_read_u16(packet);
aki_mutex_lock(&sink->mutex);
struct camu_sink_entry *entry = entry_from_node_id(sink, node_id);
+ if (!entry) goto out;
struct camu_sink_entry *previous = sink->current;
sink->current = entry;
if (entry->audio.state == BUFFER_INIT) {
@@ -542,8 +617,9 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
camu_clock_pause(&previous->clock);
remove_entry_buffers(sink, previous);
}
- aki_mutex_unlock(&sink->mutex);
+out:
+ aki_mutex_unlock(&sink->mutex);
aki_packet_free(packet);
return false;
}
@@ -644,13 +720,13 @@ void camu_sink_stop(struct camu_sink *sink)
void camu_sink_close(struct camu_sink *sink)
{
aki_mutex_lock(&sink->mutex);
- /*
struct camu_sink_entry *entry;
al_array_foreach(sink->entries, i, entry) {
+ bmu_client_close(&entry->client);
}
- */
- al_array_free(sink->entries);
aki_mutex_unlock(&sink->mutex);
+ al_array_free(sink->entries);
+ aki_rpc_disconnect(&sink->client);
aki_signal_stop(&sink->signal);
camu_queue_free(sink->queue);
aki_mutex_destroy(&sink->mutex);
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 95c280f..e3550bb 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -18,8 +18,6 @@
#include "../bimu/client.h"
-#include "common.h"
-
enum {
CAMU_SINK_AUDIO = 0,
#ifndef CAMU_SINK_NO_VIDEO
@@ -41,16 +39,17 @@ struct camu_sink_entry {
u8 state;
struct bmu_client client;
struct camu_clock clock;
+ struct aki_timer timer;
struct {
u8 state;
struct camu_audio_buffer buf;
- struct bmu_client_stream *stream;
+ struct bmu_vcr_stream *stream;
} audio;
#ifndef CAMU_SINK_NO_VIDEO
struct {
u8 state;
struct camu_video_buffer buf;
- struct bmu_client_stream *stream;
+ struct bmu_vcr_stream *stream;
} video;
#endif
struct camu_sink *sink;
diff --git a/src/mixer/audio_miniaudio.c b/src/mixer/audio_miniaudio.c
index 3d6f54b..dde742a 100644
--- a/src/mixer/audio_miniaudio.c
+++ b/src/mixer/audio_miniaudio.c
@@ -7,6 +7,7 @@ static ma_backend backends[] = {
ma_backend_pulseaudio,
#else
ma_backend_wasapi
+// ma_backend_dsound
#endif
};
@@ -38,7 +39,11 @@ static ma_allocation_callbacks alloc_callbacks = {
static void miniaudio_log_callback(void *userdata, u32 level, const char *message)
{
(void)userdata;
- if (level <= MA_LOG_LEVEL_WARNING) al_log_debug("audio_miniaudio", message);
+ if (level < MA_LOG_LEVEL_WARNING) {
+ al_log_debug("audio_miniaudio", message);
+ } else {
+ al_log_warn("audio_miniaudio", message);
+ }
}
static bool audio_miniaudio_init(struct camu_audio *audio, str *name)
@@ -48,53 +53,65 @@ static bool audio_miniaudio_init(struct camu_audio *audio, str *name)
ma_log_register_callback(&ma->log, ma_log_callback_init(miniaudio_log_callback, NULL));
ma->context_config = ma_context_config_init();
ma->context_config.pLog = &ma->log;
- ma->context_config.alsa.useVerboseDeviceEnumeration = true;
+ ma->context_config.alsa.useVerboseDeviceEnumeration = MA_TRUE;
char *c_str = al_str_to_c_str(name);
ma->context_config.pulse.pApplicationName = c_str;
- //ma->context_config.pulse.pServerName = "";
- ma->context_config.pulse.tryAutoSpawn = true;;
+ ma->context_config.pulse.tryAutoSpawn = MA_TRUE;
s32 ret = ma_context_init(backends, AL_ARRAY_SIZE(backends), &ma->context_config, &ma->context);
if (ret != MA_SUCCESS) {
al_log_error("audio_miniaudio", "Failed to initialize context (%d).", ret);
return false;
}
- //if (ma_context_get_devices(&ma->context, &ma->infos, &ma->info_count, NULL, NULL) != MA_SUCCESS) {
- // return false;
- //}
- //ma->selected_device = NULL;
- //for (u32 i = 0; i < ma->info_count; i++) {
- // ma_device_info *info = &ma->infos[i];
- // ma_device_info real_info;
- // ma_context_get_device_info(&ma->context, ma_device_type_playback, &info->id, &real_info);
- // for (u32 j = 0; j < real_info.nativeDataFormatCount; j++) {
- // if (strcmp(real_info.id.alsa, "hw:CARD=D10,DEV=0") == 0) {
- // ma->selected_device = info;
- // break;
- // }
- // /*
- // al_printf("%s\n", real_info.id.alsa);
- // al_printf("%i\n", real_info.nativeDataFormats[j].format);
- // al_printf("%i\n", real_info.nativeDataFormats[j].channels);
- // al_printf("%i\n", real_info.nativeDataFormats[j].sampleRate);
- // */
- // }
- // if (ma->selected_device) break;
- // //al_printf("\n");
- //}
- ////al_printf("%i\n", ma->info_count);
+ /*
+ if (ma_context_get_devices(&ma->context, &ma->infos, &ma->info_count, NULL, NULL) != MA_SUCCESS) {
+ return false;
+ }
+ for (u32 i = 0; i < ma->info_count; i++) {
+ ma_device_info *info = &ma->infos[i];
+ ma_device_info real_info;
+ ma_context_get_device_info(&ma->context, ma_device_type_playback, &info->id, &real_info);
+ for (u32 j = 0; j < real_info.nativeDataFormatCount; j++) {
+ //al_printf("%s\n", real_info.id.alsa);
+ //al_printf("%s\n", real_info.id.pulse);
+ //al_printf("%i\n", real_info.nativeDataFormats[j].format);
+ //al_printf("%i\n", real_info.nativeDataFormats[j].channels);
+ //al_printf("%i\n", real_info.nativeDataFormats[j].sampleRate);
+ //al_printf("\n");
+ }
+ }
+ //al_printf("%i\n", ma->info_count);
+ */
return true;
}
+static enum AVSampleFormat av_sample_fmt_from_miniaudio(s32 fmt)
+{
+ switch (fmt) {
+ case ma_format_u8:
+ return AV_SAMPLE_FMT_U8;
+ case ma_format_s16:
+ return AV_SAMPLE_FMT_S16;
+ case ma_format_s24:
+ case ma_format_s32:
+ return AV_SAMPLE_FMT_S32;
+ case ma_format_f32:
+ return AV_SAMPLE_FMT_FLT;
+ case ma_format_unknown:
+ default:
+ al_assert(false);
+ return AV_SAMPLE_FMT_NONE;
+ }
+}
+
static void audio_miniaudio_pick_format(struct camu_audio *audio, struct camu_lav_resample_fmt *fmt)
{
struct camu_audio_miniaudio *ma = (struct camu_audio_miniaudio *)audio;
- (void)ma;
- fmt->req_channel_count = 2;
- fmt->req_format = AV_SAMPLE_FMT_S32;
- fmt->req_sample_rate = 44100;
+ fmt->req_format = av_sample_fmt_from_miniaudio(ma->device.playback.internalFormat);
+ fmt->req_channel_count = ma->device.playback.internalChannels;
+ fmt->req_sample_rate = ma->device.playback.internalSampleRate;
}
-static void data_callback(ma_device *device, void *output, const void *input, ma_uint32 frame_count)
+static void data_callback(ma_device *device, void *output, const void *input, u32 frame_count)
{
struct camu_audio_miniaudio *ma = (struct camu_audio_miniaudio *)device->pUserData;
(void)input;
@@ -111,22 +128,29 @@ static bool audio_miniaudio_configure_stream(struct camu_audio *audio, void *opa
(void)fmt;
ma->config = ma_device_config_init(ma_device_type_playback);
//ma->config.playback.pDeviceID = &ma->selected_device->id;
+ //ma->config.performanceProfile = ma_performance_profile_low_latency;
ma->config.performanceProfile = ma_performance_profile_conservative;
ma->config.periods = DEFAULT_PERIODS;
ma->config.periodSizeInMilliseconds = DEFAULT_PERIOD_SIZE_IN_MILLISECONDS;
- ma->config.playback.channels = 2;
- ma->config.playback.format = ma_format_s32;
- ma->config.sampleRate = 44100;
+ ma->config.playback.format = ma_format_unknown;
+ ma->config.playback.channels = 0;
+ ma->config.sampleRate = 0;
ma->config.dataCallback = data_callback;
ma->config.pUserData = ma;
- ma->config.noPreSilencedOutputBuffer = true;
+ ma->config.noPreSilencedOutputBuffer = MA_TRUE;
+ ma->config.noFixedSizedCallback = MA_TRUE;
+ ma->config.noClip = MA_TRUE;
+ ma->config.wasapi.noAutoConvertSRC = MA_FALSE;
+ ma->config.wasapi.noDefaultQualitySRC = MA_FALSE;
s32 ret = ma_device_init(&ma->context, &ma->config, &ma->device);
if (ret != MA_SUCCESS) {
al_log_error("audio_miniaudio", "Failed to initialize device (%d).", ret);
return false;
}
- const char *format_name = av_get_sample_fmt_name(AV_SAMPLE_FMT_S32);
- al_log_info("audio_miniaudio", "Output: %s (%dch) %dHz.", format_name, 2, 44100);
+ enum AVSampleFormat avfmt = av_sample_fmt_from_miniaudio(ma->device.playback.internalFormat);
+ const char *format_name = av_get_sample_fmt_name(avfmt);
+ al_log_info("audio_miniaudio", "Output: %s (%dch) %dHz.", format_name,
+ ma->device.playback.internalChannels, ma->device.playback.internalSampleRate);
return true;
}
@@ -177,9 +201,9 @@ struct camu_audio_miniaudio audio_plugin_miniaudio = {
.stop = audio_miniaudio_stop,
.free = audio_miniaudio_free
},
- .log = { 0 },
+ .log = { },
.context_config = { 0 },
- .context = { 0 },
+ .context = { },
.infos = NULL,
.info_count = 0,
.selected_device = NULL,
diff --git a/src/mixer/meson.build b/src/mixer/meson.build
index 71c003b..ab14b6f 100644
--- a/src/mixer/meson.build
+++ b/src/mixer/meson.build
@@ -21,6 +21,7 @@ if not is_windows
miniaudio_args += ['-DMA_ENABLE_PULSEAUDIO', '-DMA_NO_RUNTIME_LINKING']
else
miniaudio_args += ['-DMA_ENABLE_WASAPI']
+ #miniaudio_args += ['-DMA_ENABLE_DSOUND']
endif
mixer = declare_dependency(sources: mixer_src, dependencies: mixer_deps,
diff --git a/src/mixer/mixer.c b/src/mixer/mixer.c
index 663e7aa..a4c0ef5 100644
--- a/src/mixer/mixer.c
+++ b/src/mixer/mixer.c
@@ -1,6 +1,7 @@
#include "../buffer/audio.h"
#include "mixer.h"
+#include "audio.h"
static s32 data_callback(void *userdata, u8 *data, s32 frame_count)
{
@@ -39,8 +40,10 @@ bool camu_mixer_init(struct camu_mixer *mixer, struct camu_audio *audio)
mixer->audio->data_callback = data_callback;
mixer->audio->userdata = mixer;
mixer->audio->init(mixer->audio, al_str_c("camu"));
- mixer->audio->pick_format(mixer->audio, &mixer->fmt);
- av_channel_layout_default(&mixer->fmt.req_channel_layout, mixer->fmt.req_channel_count);
+ if (!mixer->audio->configure_stream(mixer->audio, NULL)) {
+ return false;
+ }
+ camu_mixer_pick_format(mixer, &mixer->fmt);
mixer->paused = true;
al_array_init(mixer->buffers);
#ifdef CAMU_MIXER_THREADED
@@ -55,6 +58,7 @@ bool camu_mixer_init(struct camu_mixer *mixer, struct camu_audio *audio)
void camu_mixer_pick_format(struct camu_mixer *mixer, struct camu_lav_resample_fmt *fmt)
{
mixer->audio->pick_format(mixer->audio, fmt);
+ av_channel_layout_default(&fmt->req_channel_layout, fmt->req_channel_count);
}
f64 camu_mixer_get_latency(struct camu_mixer *mixer)
@@ -90,8 +94,6 @@ void camu_mixer_add_buffer(struct camu_mixer *mixer, struct camu_audio_buffer *b
#ifdef CAMU_MIXER_THREADED
aki_mutex_lock(&mixer->mutex);
al_array_push(mixer->add_queue, buf);
- // Ensure an allocation doesn't happen in data_callback.
- al_array_reserve(mixer->buffers, mixer->buffers.size + mixer->add_queue.size);
al_atomic_bool_store(&mixer->queued, true, AL_ATOMIC_RELAXED);
aki_mutex_unlock(&mixer->mutex);
#else
@@ -102,15 +104,16 @@ void camu_mixer_add_buffer(struct camu_mixer *mixer, struct camu_audio_buffer *b
#ifdef CAMU_MIXER_THREADED
static void run_queue_internal(struct camu_mixer *mixer)
{
+ al_array_reserve(mixer->buffers, mixer->buffers.size + mixer->add_queue.size);
struct camu_audio_buffer *buf;
al_array_foreach(mixer->add_queue, i, buf) {
add_buffer_internal(mixer, buf);
- al_array_remove_at_iter(mixer->add_queue, i);
}
+ mixer->add_queue.size = 0;
al_array_foreach(mixer->rem_queue, i, buf) {
remove_buffer_internal(mixer, buf);
- al_array_remove_at_iter(mixer->rem_queue, i);
}
+ mixer->rem_queue.size = 0;
al_atomic_bool_store(&mixer->queued, false, AL_ATOMIC_RELAXED);
}
#endif
diff --git a/src/tree/common.h b/src/tree/common.h
index 4e3ec3a..d4d4836 100644
--- a/src/tree/common.h
+++ b/src/tree/common.h
@@ -1,8 +1,13 @@
#pragma once
-#define TREE_PORT 4356
-#define TREE_RESOURCE_PORT 4357
-#define TREE_STREAM_PORT 4358
+#include <al/str.h>
+
+#define TREE_PORT 14356
+#define TREE_RESOURCE_PORT 14357
+#define TREE_STREAM_PORT 14358
+
+#define TREE_SERVER_IP al_str_c("127.0.0.1")
+//#define TREE_SERVER_IP al_str_c("108.52.160.112")
enum {
TREE_NODE = 0,
diff --git a/src/tree/list.c b/src/tree/list.c
index 8102e05..c1630b7 100644
--- a/src/tree/list.c
+++ b/src/tree/list.c
@@ -7,6 +7,7 @@
#include "list.h"
#include "tree.h"
+#include "common.h"
void tree_list_init(struct tree_list *list, struct tree_server *tree, str *name)
{
@@ -21,7 +22,7 @@ void tree_list_init(struct tree_list *list, struct tree_server *tree, str *name)
static void send_buffer_cmd(struct tree_list *list, struct tree_sink *sink, struct tree_list_entry *entry)
{
struct aki_packet *packet = aki_rpc_get_packet(&list->tree->server, CAMU_SINK_CMD_BUFFER);
- aki_packet_write_str(packet, al_str_c("127.0.0.1"));
+ aki_packet_write_str(packet, TREE_SERVER_IP);
aki_packet_write_s32(packet, TREE_STREAM_PORT);
aki_packet_write_u16(packet, entry->node_id);
aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
diff --git a/src/tree/tree.c b/src/tree/tree.c
index 8a0972f..367e0ff 100644
--- a/src/tree/tree.c
+++ b/src/tree/tree.c
@@ -2,9 +2,8 @@
#include <aki/file.h>
#include <jansson.h>
-#include "../libclient/commands.h"
-
#include "tree.h"
+#include "common.h"
#define USER_AGENT al_str_c("Mozilla/5.0 (X11; Linux x86_64; rv:96.0) Gecko/20100101 Firefox/96.0")
diff --git a/src/tree/tree.h b/src/tree/tree.h
index 2b61e3b..6339889 100644
--- a/src/tree/tree.h
+++ b/src/tree/tree.h
@@ -12,7 +12,6 @@
#include "../bimu/server.h"
-#include "common.h"
#include "resource_manager.h"
#include "list.h"
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 16:01:50 +0000