Ticket #849: sound-autotune-queue.patch

xpra/sound/sink.py

@@ -4 +4 @@
 # Xpra is released under the terms of the GNU GPL v2, or, at your option, any
 # later version. See the file COPYING for details.
 
-import sys, os, time
+import sys, os
 
 from xpra.sound.sound_pipeline import SoundPipeline, gobject, one_arg_signal
 from xpra.sound.pulseaudio_util import has_pa
@@ -46 +46 @@
     DEFAULT_SINK = SINKS[0]
 QUEUE_SILENT = 0
 QUEUE_TIME = get_queue_time(450)
-QUEUE_MIN_TIME = get_queue_time(QUEUE_TIME//10//MS_TO_NS, "MIN")
-assert QUEUE_MIN_TIME<=QUEUE_TIME
 
-VARIABLE_MIN_QUEUE = os.environ.get("XPRA_VARIABLE_MIN_QUEUE", "1")=="1"
-
-
 GST_FORMAT_BUFFERS = 4
 
 def sink_has_device_attribute(sink):
@@ -94 +89 @@
         pipeline_els.append("volume name=volume volume=%s" % volume)
         queue_el = ["queue",
                     "name=queue",
-                    "min-threshold-time=%s" % QUEUE_MIN_TIME,
+                    "min-threshold-time=0",
                     "max-size-buffers=0",
                     "max-size-bytes=0",
                     "max-size-time=%s" % QUEUE_TIME,
@@ -131 +126 @@
     def queue_pushing(self, *args):
         ltime = self.queue.get_property("current-level-time")/MS_TO_NS
         log("queue pushing: level=%i", ltime)
+        self.check_levels()
         self.queue_state = "pushing"
-        self.emit_info()
+        #self.emit_info()
+        return 0
 
     def queue_running(self, *args):
         ltime = self.queue.get_property("current-level-time")/MS_TO_NS
         log("queue running: level=%s", ltime)
-        if self.queue_state=="underrun" and VARIABLE_MIN_QUEUE:
+        self.check_levels()
+        self.queue_state = "running"
+        return 0
+
+    def check_levels(self):
+        if True:
+            return
+        if self.queue_state=="underrun":
             #lift min time restrictions:
             self.queue.set_property("min-threshold-time", 0)
         elif self.queue_state == "overrun":
             clt = self.queue.get_property("current-level-time")
             qpct = min(QUEUE_TIME, clt)*100//QUEUE_TIME
-            log("resetting max-size-time back to %ims (level=%ims, %i%%)", QUEUE_TIME//MS_TO_NS, clt//MS_TO_NS, qpct)
+            log.info("resetting max-size-time back to %ims (level=%ims, %i%%)", QUEUE_TIME//MS_TO_NS, clt//MS_TO_NS, qpct)
             self.queue.set_property("max-size-time", QUEUE_TIME)
         self.queue_state = "running"
         self.emit_info()
 
+
     def queue_underrun(self, *args):
+        if self.queue_state=="underrun" or self.queue_state=="starting":
+            return
         ltime = self.queue.get_property("current-level-time")/MS_TO_NS
-        log("queue underrun: level=%i", ltime)
-        if self.queue_state!="underrun" and VARIABLE_MIN_QUEUE:
-            #lift min time restrictions:
-            self.queue.set_property("min-threshold-time", QUEUE_MIN_TIME)
+        log.info("queue underrun: level=%i, previous state=%s", ltime, self.queue_state)
         self.queue_state = "underrun"
-        self.emit_info()
+        MIN_QUEUE = QUEUE_TIME//2
+        mts = self.queue.get_property("min-threshold-time")
+        if mts==MIN_QUEUE:
+            return 0
+        #set min time restrictions to fill up queue:
+        log.info("increasing the min-threshold-time to %ims", MIN_QUEUE//MS_TO_NS)
+        self.queue.set_property("min-threshold-time", MIN_QUEUE)
+        def restore():
+            if self.queue.get_property("min-threshold-time")==0:
+                log.info("min-threshold-time already restored!")
+                return False
+            ltime = self.queue.get_property("current-level-time")//MS_TO_NS
+            if ltime==0:
+                log.info("not restored! (still underrun: %ims)", ltime)
+                return True
+            log.info("resetting the min-threshold-time back to %ims", 0)
+            self.queue.set_property("min-threshold-time", 0)
+            self.queue_state = "running"
+            return False
+        import glib
+        glib.timeout_add(1000, restore)
+        #self.emit_info()
+        return 0
 
     def queue_overrun(self, *args):
+        if self.queue_state=="overrun":
+            return
         ltime = self.queue.get_property("current-level-time")//MS_TO_NS
-        pqs = self.queue_state
+        log("queue overrun: level=%i, previous state=%s", ltime, self.queue_state)
         self.queue_state = "overrun"
-        #no overruns for the first 2 seconds:
-        if ltime<QUEUE_TIME//MS_TO_NS//2*75//100:
-            elapsed = time.time()-self.start_time
-            log("queue overrun ignored: level=%i, elapsed time=%.1f", ltime, elapsed)
-            return
-        log("queue overrun: level=%i, previous state=%s", ltime//MS_TO_NS, pqs)
-        if pqs!="overrun":
-            log("halving the max-size-time to %ims", QUEUE_TIME//2//MS_TO_NS)
-            self.queue.set_property("max-size-time", QUEUE_TIME//2)
-            self.overruns += 1
-            self.emit("overrun", ltime)
-            self.emit_info()
+        REDUCED_QT = QUEUE_TIME//2
+        #empty the queue by reducing its max size:
+        log.info("queue overrun: halving the max-size-time to %ims", REDUCED_QT//MS_TO_NS)
+        self.queue.set_property("max-size-time", REDUCED_QT)
+        self.overruns += 1
+        self.emit("overrun", ltime)
+        def restore():
+            if self.queue.get_property("max-size-time")==QUEUE_TIME:
+                log.info("max-size-time already restored!")
+                return False
+            ltime = self.queue.get_property("current-level-time")//MS_TO_NS
+            if ltime>=REDUCED_QT:
+                log.info("max-size-time not restored! (still overrun: %ims)", ltime)
+                return True
+            log.info("raising the max-size-time back to %ims", QUEUE_TIME//MS_TO_NS)
+            self.queue.set_property("max-size-time", QUEUE_TIME)
+            self.queue_state = "running"
+            return False
+        import glib
+        glib.timeout_add(1000, restore)
+        #self.emit_info()
+        return 0
 
     def eos(self):
         log("eos()")
@@ -179 +217 @@
         if self.src:
             self.src.emit('end-of-stream')
         self.cleanup()
+        return 0
 
     def get_info(self):
         info = SoundPipeline.get_info(self)
@@ -186 +225 @@
             clt = self.queue.get_property("current-level-time")
             updict(info, "queue", {
                 "time"          : QUEUE_TIME//MS_TO_NS,
-                "min_time"      : QUEUE_MIN_TIME//MS_TO_NS,
                 "used_pct"      : min(QUEUE_TIME, clt)*100//QUEUE_TIME,
                 "used"          : clt//MS_TO_NS,
                 "overruns"      : self.overruns,
@@ -197 +235 @@
         if not self.src:
             log("add_data(..) dropped")
             return
+        if self.queue_state=="overrun":
+            log.info("add_data(..) queue in overrun, data dropped")
+            return
         #having a timestamp causes problems with the queue and overruns:
         if "timestamp" in metadata:
             del metadata["timestamp"]

xpra/sound/src.py

@@ -16 +16 @@
 from xpra.log import Logger
 log = Logger("sound")
 
-APPSINK = os.environ.get("XPRA_SOURCE_APPSINK", "appsink name=sink emit-signals=true max-buffers=10 drop=true sync=false async=false qos=false")
+APPSINK = os.environ.get("XPRA_SOURCE_APPSINK", "appsink name=sink emit-signals=true max-buffers=10 drop=true sync=true async=false qos=false")
 
 
 class SoundSource(SoundPipeline):