demux, stream: readd cache-speed in some other form

it's more like an input speed rather than a cache speed, but who cares.

DOCS/man/input.rst

@@ -1455,6 +1455,11 @@ Property list
     playing at all. In other words, it's only ``no`` if there's actually
     video playing. (Behavior since mpv 0.7.0.)
 
+``cache-speed`` (R)
+    Current I/O read speed between the cache and the lower layer (like network).
+    This gives the number bytes per seconds over a 1 second window (using
+    the type ``MPV_FORMAT_INT64`` for the client API).
+
 ``demuxer-cache-duration``
     Approximate duration of video buffered in the demuxer, in seconds. The
     guess is very unreliable, and often the property will not be available

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++) {

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.

osdep/timer.h

@@ -39,6 +39,9 @@ void mp_sleep_us(int64_t us);
 
 #define MP_START_TIME 10000000
 
+// Duration of a second in mpv time.
+#define MP_SECOND_US (1000 * 1000)
+
 // Add a time in seconds to the given time in microseconds, and return it.
 // Takes care of possible overflows. Never returns a negative or 0 time.
 int64_t mp_add_timeout(int64_t time_us, double timeout_sec);

player/command.c

@@ -1559,6 +1559,26 @@ static int mp_property_playback_abort(void *ctx, struct m_property *prop,
     return m_property_flag_ro(action, arg, !mpctx->playing || mpctx->stop_play);
 }
 
+static int mp_property_cache_speed(void *ctx, struct m_property *prop,
+                                   int action, void *arg)
+{
+    MPContext *mpctx = ctx;
+    if (!mpctx->demuxer)
+        return M_PROPERTY_UNAVAILABLE;
+
+    struct demux_ctrl_reader_state s;
+    if (demux_control(mpctx->demuxer, DEMUXER_CTRL_GET_READER_STATE, &s) < 1)
+        return M_PROPERTY_UNAVAILABLE;
+
+    uint64_t val = s.bytes_per_second;
+
+    if (action == M_PROPERTY_PRINT) {
+        *(char **)arg = talloc_strdup_append(format_file_size(val), "/s");
+        return M_PROPERTY_OK;
+    }
+    return m_property_int64_ro(action, arg, val);
+}
+
 static int mp_property_demuxer_cache_duration(void *ctx, struct m_property *prop,
                                               int action, void *arg)
 {
@@ -3771,6 +3791,7 @@ static const struct m_property mp_properties_base[] = {
     {"eof-reached", mp_property_eof_reached},
     {"seeking", mp_property_seeking},
     {"playback-abort", mp_property_playback_abort},
+    {"cache-speed", mp_property_cache_speed},
     {"demuxer-cache-duration", mp_property_demuxer_cache_duration},
     {"demuxer-cache-time", mp_property_demuxer_cache_time},
     {"demuxer-cache-idle", mp_property_demuxer_cache_idle},

stream/stream.c

@@ -342,6 +342,7 @@ static int stream_read_unbuffered(stream_t *s, void *buf, int len)
     // When reading succeeded we are obviously not at eof.
     s->eof = 0;
     s->pos += res;
+    s->total_unbuffered_read_bytes += res;
     return res;
 }
 

stream/stream.h

@@ -175,6 +175,10 @@ typedef struct stream {
 
     struct mp_cancel *cancel;   // cancellation notification
 
+    // Read statistic for fill_buffer calls. All bytes read by fill_buffer() are
+    // added to this. The user can reset this as needed.
+    uint64_t total_unbuffered_read_bytes;
+
     // Includes additional padding in case sizes get rounded up by sector size.
     unsigned char buffer[];
 } stream_t;