pulse-server: add a pulse.idle.timeout option

When a client is not sending any data when it should be and causes an
underrun, mark it as idle and record the timestamp.

When a client is idle for pulse.idle.timeout seconds, set the stream
as inactive. When more data is received, set it back to active.

Add a pulse.idle.timeout option to set a global server default or
a per-stream value. Set the server default to 5 seconds. A value of 0
can be used to disable this feature.

With this change, badly behaving clients that are not sending any data
will be paused so that the sinks can suspend to save battery power.

Fixes #2839

src/daemon/pipewire-pulse.conf.in

@@ -90,6 +90,7 @@ pulse.properties = {
     #pulse.default.frag     = 96000/48000   # 2 seconds
     #pulse.default.tlength  = 96000/48000   # 2 seconds
     #pulse.min.quantum      = 256/48000     # 5ms
+    #pulse.idle.timeout     = 5             # pause after 5s of underruns
     #pulse.default.format   = F32
     #pulse.default.position = [ FL FR ]
     # These overrides are only applied when running in a vm.
@@ -141,6 +142,7 @@ pulse.rules = [
             update-props = {
                 pulse.min.req          = 1024/48000     # 21ms
                 pulse.min.quantum      = 1024/48000     # 21ms
+                #pulse.idle.timeout    = 0
             }
         }
     }

src/modules/module-protocol-pulse/internal.h

@@ -53,6 +53,7 @@ struct defs {
 	struct sample_spec sample_spec;
 	struct channel_map channel_map;
 	uint32_t quantum_limit;
+	uint32_t idle_timeout;
 };
 
 struct stats {

src/modules/module-protocol-pulse/pulse-server.c

@@ -85,6 +85,7 @@
 #define DEFAULT_MIN_QUANTUM	"256/48000"
 #define DEFAULT_FORMAT		"F32"
 #define DEFAULT_POSITION	"[ FL FR ]"
+#define DEFAULT_IDLE_TIMEOUT	"5"
 
 #define MAX_FORMATS	32
 /* The max amount of data we send in one block when capturing. In PulseAudio this
@@ -1276,6 +1277,7 @@ struct process_data {
 	uint32_t minreq;
 	uint32_t quantum;
 	unsigned int underrun:1;
+	unsigned int idle:1;
 };
 
 static int
@@ -1315,6 +1317,17 @@ do_process_done(struct spa_loop *loop,
 			else
 				stream_send_started(stream);
 		}
+		if (pd->idle) {
+			if (!stream->is_idle) {
+				stream->idle_time = stream->timestamp;
+			} else if (!stream->is_paused &&
+			    stream->idle_timeout_sec > 0 &&
+			    stream->timestamp - stream->idle_time >
+					(stream->idle_timeout_sec * SPA_NSEC_PER_SEC)) {
+				stream_set_paused(stream, true, "long underrun");
+			}
+		}
+		stream->is_idle = pd->idle;
 		stream->playing_for += pd->playing_for;
 		if (stream->underrun_for != (uint64_t)-1)
 			stream->underrun_for += pd->underrun_for;
@@ -1440,6 +1453,7 @@ static void stream_process(void *data)
 
 				pd.playing_for = size;
 			}
+			pd.idle = true;
 			pw_log_debug("%p: [%s] underrun read:%u avail:%d max:%u",
 					stream, client->name, index, avail, minreq);
 		} else {
@@ -5548,6 +5562,20 @@ static int parse_format(struct pw_properties *props, const char *key, const char
 	pw_log_info(": defaults: %s = %s", key, format_id2name(res->format));
 	return 0;
 }
+static int parse_uint32(struct pw_properties *props, const char *key, const char *def,
+		uint32_t *res)
+{
+	const char *str;
+	if (props == NULL ||
+	    (str = pw_properties_get(props, key)) == NULL)
+		str = def;
+	if (!spa_atou32(str, res, 0)) {
+		pw_log_warn(": invalid uint32_t %s, default to %s", str, def);
+		spa_atou32(def, res, 0);
+	}
+	pw_log_info(": defaults: %s = %u", key, *res);
+	return 0;
+}
 
 static void load_defaults(struct defs *def, struct pw_properties *props)
 {
@@ -5559,6 +5587,7 @@ static void load_defaults(struct defs *def, struct pw_properties *props)
 	parse_frac(props, "pulse.min.quantum", DEFAULT_MIN_QUANTUM, &def->min_quantum);
 	parse_format(props, "pulse.default.format", DEFAULT_FORMAT, &def->sample_spec);
 	parse_position(props, "pulse.default.position", DEFAULT_POSITION, &def->channel_map);
+	parse_uint32(props, "pulse.idle.timeout", DEFAULT_IDLE_TIMEOUT, &def->idle_timeout);
 	def->sample_spec.channels = def->channel_map.channels;
 	def->quantum_limit = 8192;
 }

src/modules/module-protocol-pulse/server.c

@@ -191,6 +191,9 @@ static int handle_memblock(struct client *client, struct message *msg)
 
 	stream_send_request(stream);
 
+	if (stream->is_paused && !stream->corked)
+		stream_set_paused(stream, false, "new data");
+
 finish:
 	message_free(msg, false, false);
 	return res;

src/modules/module-protocol-pulse/stream.c

@@ -64,6 +64,7 @@ struct stream *stream_new(struct client *client, enum stream_type type, uint32_t
 {
 	int res;
 	struct defs *defs = &client->impl->defs;
+	const char *str;
 
 	struct stream *stream = calloc(1, sizeof(*stream));
 	if (stream == NULL)
@@ -90,6 +91,9 @@ struct stream *stream_new(struct client *client, enum stream_type type, uint32_t
 	parse_frac(client->props, "pulse.default.req", &defs->default_req, &stream->default_req);
 	parse_frac(client->props, "pulse.default.frag", &defs->default_frag, &stream->default_frag);
 	parse_frac(client->props, "pulse.default.tlength", &defs->default_tlength, &stream->default_tlength);
+	stream->idle_timeout_sec = defs->idle_timeout;
+	if ((str = pw_properties_get(client->props, "pulse.idle.timeout")) != NULL)
+		spa_atou32(str, &stream->idle_timeout_sec, 0);
 
 	switch (type) {
 	case STREAM_TYPE_RECORD:

src/modules/module-protocol-pulse/stream.h

@@ -82,6 +82,7 @@ struct stream {
 	uint64_t playing_for;
 	uint64_t ticks_base;
 	uint64_t timestamp;
+	uint64_t idle_time;
 	int64_t delay;
 
 	uint32_t last_quantum;
@@ -93,6 +94,7 @@ struct stream {
 	struct spa_fraction default_frag;
 	struct spa_fraction default_tlength;
 	struct spa_fraction min_quantum;
+	uint32_t idle_timeout_sec;
 
 	struct sample_spec ss;
 	struct channel_map map;
@@ -115,6 +117,7 @@ struct stream {
 	unsigned int in_prebuf:1;
 	unsigned int killed:1;
 	unsigned int pending:1;
+	unsigned int is_idle:1;
 	unsigned int is_paused:1;
 };