summaryrefslogtreecommitdiffstats
path: root/demux
diff options
context:
space:
mode:
Diffstat (limited to 'demux')
-rw-r--r--demux/demux.c33
-rw-r--r--demux/demux.h4
2 files changed, 36 insertions, 1 deletions
diff --git a/demux/demux.c b/demux/demux.c
index 33f3eec231..b3b18257cf 100644
--- a/demux/demux.c
+++ b/demux/demux.c
@@ -37,6 +37,7 @@
#include "common/global.h"
#include "misc/thread_tools.h"
#include "osdep/atomic.h"
+#include "osdep/timer.h"
#include "osdep/threads.h"
#include "stream/stream.h"
@@ -223,6 +224,9 @@ struct demux_internal {
double duration;
// Cached state.
int64_t stream_size;
+ int64_t last_speed_query;
+ uint64_t bytes_per_second;
+ int64_t next_cache_update;
// Updated during init only.
char *stream_base_filename;
};
@@ -1748,6 +1752,12 @@ static bool thread_work(struct demux_internal *in)
if (read_packet(in))
return true; // read_packet unlocked, so recheck conditions
}
+ if (mp_time_us() >= in->next_cache_update) {
+ pthread_mutex_unlock(&in->lock);
+ update_cache(in);
+ pthread_mutex_lock(&in->lock);
+ return true;
+ }
return false;
}
@@ -1761,7 +1771,8 @@ static void *demux_thread(void *pctx)
if (thread_work(in))
continue;
pthread_cond_signal(&in->wakeup);
- pthread_cond_wait(&in->wakeup, &in->lock);
+ struct timespec until = mp_time_us_to_timespec(in->next_cache_update);
+ pthread_cond_timedwait(&in->wakeup, &in->lock, &until);
}
if (in->shutdown_async) {
@@ -3049,7 +3060,11 @@ static void update_cache(struct demux_internal *in)
int64_t stream_size = stream_get_size(stream);
stream_control(stream, STREAM_CTRL_GET_METADATA, &stream_metadata);
+ demuxer->total_unbuffered_read_bytes += stream->total_unbuffered_read_bytes;
+ stream->total_unbuffered_read_bytes = 0;
+
pthread_mutex_lock(&in->lock);
+
in->stream_size = stream_size;
if (stream_metadata) {
for (int n = 0; n < in->num_streams; n++) {
@@ -3059,6 +3074,21 @@ static void update_cache(struct demux_internal *in)
}
talloc_free(stream_metadata);
}
+
+ in->next_cache_update = INT64_MAX;
+
+ int64_t now = mp_time_us();
+ int64_t diff = now - in->last_speed_query;
+ if (diff >= MP_SECOND_US) {
+ uint64_t bytes = demuxer->total_unbuffered_read_bytes;
+ demuxer->total_unbuffered_read_bytes = 0;
+ in->last_speed_query = now;
+ in->bytes_per_second = bytes / (diff / (double)MP_SECOND_US);
+ }
+ // The idea is to update as long as there is "activity".
+ if (in->bytes_per_second)
+ in->next_cache_update = now + MP_SECOND_US + 1;
+
pthread_mutex_unlock(&in->lock);
}
@@ -3115,6 +3145,7 @@ static int cached_demux_control(struct demux_internal *in, int cmd, void *arg)
.seeking = in->seeking_in_progress,
.low_level_seeks = in->low_level_seeks,
.ts_last = in->demux_ts,
+ .bytes_per_second = in->bytes_per_second,
};
bool any_packets = false;
for (int n = 0; n < in->num_streams; n++) {
diff --git a/demux/demux.h b/demux/demux.h
index 5b92e97f49..956548d90b 100644
--- a/demux/demux.h
+++ b/demux/demux.h
@@ -56,6 +56,7 @@ struct demux_ctrl_reader_state {
double seeking; // current low level seek target, or NOPTS
int low_level_seeks; // number of started low level seeks
double ts_last; // approx. timestamp of demuxer position
+ uint64_t bytes_per_second; // low level statistics
// Positions that can be seeked to without incurring the latency of a low
// level seek.
int num_seek_ranges;
@@ -236,6 +237,9 @@ typedef struct demuxer {
// Triggered when ending demuxing forcefully. Usually bound to the stream too.
struct mp_cancel *cancel;
+ // Demuxer thread only.
+ uint64_t total_unbuffered_read_bytes;
+
// Since the demuxer can run in its own thread, and the stream is not
// thread-safe, only the demuxer is allowed to access the stream directly.
// You can freely use demux_stream_control() to send STREAM_CTRLs.