bandwidth-congestion-handling-v6.patch on Ticket #999 – Attachment – Xpra

xpra/server/cystats.pyx

         rw += w
     return int(tv / tw), int(rv / rw)
+def calculate_timesize_weighted_average(data, float sizeunit=1.0):
+def calculate_timesize_weighted_average(data, float unit=1.0):
+    #the value is elapsed time,
+    #so we want to divide by the value:
+    recs = tuple((a,b,unit/c) for a,b,c in data)
+    return calculate_size_weighted_average(recs)
+def calculate_size_weighted_average(data):
     """
         This is a time weighted average where the size
         of each record also gives it a weight boost.
         This is to prevent small packets from skewing the average.
         Data format: (event_time, size, elapsed_time)
+        Data format: (event_time, size, value)
     """
     cdef double size_avg = sum(x for _, x, _ in data)/len(data)
     cdef double now = monotonic_time()              #@DuplicatedSignature
 …
     cdef double event_time                          #@DuplicatedSignature
     cdef double size
     cdef double size_ps
     cdef double elapsed_time
+    cdef double value
     cdef double pw
     cdef double w                                   #@DuplicatedSignature
     cdef double delta                               #@DuplicatedSignature
     for event_time, size, elapsed_time in data:
         if elapsed_time<=0:
+    for event_time, size, value in data:
+        if value<=0:
             continue        #invalid record
         delta = now-event_time
         pw = clogp(size/size_avg)
         size_ps = max(1, size*sizeunit/elapsed_time)
+        size_ps = max(1, size*value)
         w = pw/(1.0+delta)
         tv += w*size_ps
         tw += w
+        tw += w*size
         w = pw/(0.1+delta**2)
         rv += w*size_ps
         rw += w
+        rw += w*size
     return float(tv / tw), float(rv / rw)
 def calculate_for_target(metric, float target_value, float avg_value, float recent_value, float aim=0.5, float div=1.0, float slope=0.1, smoothing=logp, float weight_multiplier=1.0):

xpra/server/source.py

 NEW_STREAM_SOUND = envbool("XPRA_NEW_STREAM_SOUND", True)
 PING_DETAILS = envbool("XPRA_PING_DETAILS", True)
 PING_TIMEOUT = envint("XPRA_PING_TIMEOUT", 60)
+CONGESTION_AVOIDANCE = envbool("XPRA_CONGESTION_AVOIDANCE", True)
 PRINTER_LOCATION_STRING = os.environ.get("XPRA_PRINTER_LOCATION_STRING", "via xpra")
 PROPERTIES_DEBUG = [x.strip() for x in os.environ.get("XPRA_WINDOW_PROPERTIES_DEBUG", "").split(",")]
 …
         self.double_click_time  = -1
         self.double_click_distance = -1, -1
         self.bandwidth_limit = self.server_bandwidth_limit
+        self.soft_bandwidth_limit = self.bandwidth_limit
         #what we send back in hello packet:
         self.ui_client = True
         self.wants_aliases = True
 …
     def update_bandwidth_limits(self):
         if self.bandwidth_limit<=0 or self.mmap_size>0:
+        if self.mmap_size>0:
             return
+        #calculate soft bandwidth limit based on send congestion data:
+        bandwidth_limit = 0
+        if CONGESTION_AVOIDANCE:
+            bandwidth_limit = self.statistics.avg_congestion_send_speed
+            log.warn("avg_congestion_send_speed=%s", bandwidth_limit)
+            if bandwidth_limit>20*1024*1024:
+                #ignore congestion speed if greater 20Mbps
+                bandwidth_limit = 0
+        if self.bandwidth_limit>0:
+            #command line options could overrule what we detect?
+            bandwidth_limit = min(self.bandwidth_limit, bandwidth_limit)
+        self.soft_bandwidth_limit = bandwidth_limit
+        log.warn("update_bandwidth_limits() bandwidth_limit=%s, soft bandwidth limit=%s", self.bandwidth_limit, bandwidth_limit)
+        if self.soft_bandwidth_limit<=0:
+            return
         #figure out how to distribute the bandwidth amongst the windows,
         #we use the window size,
         #(we should actually use the number of bytes actually sent: framerate, compression, etc..)
 …
         for wid, ws in self.window_sources.items():
             weight = window_weight.get(wid)
             if weight is not None:
                 ws.bandwidth_limit = max(1, self.bandwidth_limit*weight/total_weight)
+                ws.bandwidth_limit = max(1, bandwidth_limit*weight/total_weight)
     def recalculate_delays(self):
         """ calls update_averages() on ServerSource.statistics (GlobalStatistics)
 …
         if self.is_closed():
             return
         self.calculate_last_time = monotonic_time()
+        self.statistics.update_averages()
         self.update_bandwidth_limits()
-        self.statistics.update_averages()
         wids = list(self.calculate_window_ids)  #make a copy so we don't clobber new wids
         focus = self.get_focus()
         sources = self.window_sources.items()
 …
                 "suspended"         : self.suspended,
                 "counter"           : self.counter,
                 "hello-sent"        : self.hello_sent,
+                "bandwidth-limit"   : self.bandwidth_limit,
+                "bandwidth-limit"   : {
+                    "setting"       : self.bandwidth_limit,
+                    "actual"        : self.soft_bandwidth_limit,
+                    }
+                }
         if self.desktop_mode_size:
             info["desktop_mode_size"] = self.desktop_mode_size

xpra/server/source_stats.py

 from xpra.log import Logger
 log = Logger("stats")
 from xpra.server.cystats import logp, calculate_time_weighted_average, calculate_for_target, queue_inspect  #@UnresolvedImport
+from xpra.server.cystats import logp, calculate_time_weighted_average, calculate_size_weighted_average, calculate_for_target, queue_inspect  #@UnresolvedImport
 from xpra.simple_stats import get_list_stats
 from xpra.os_util import monotonic_time
 …
                                                             #(event_time, elapsed_time_in_seconds)
         self.server_ping_latency = deque(maxlen=NRECS)      #time it took for the client to get a ping_echo back from us:
                                                             #(event_time, elapsed_time_in_seconds)
+        self.congestion_send_speed = deque(maxlen=4*NRECS)  #when we are being throttled, record what speed we are sending at
+                                                            #last NRECS: (event_time, no of pixels, duration)
+        self.late_packets = set()
         self.client_load = None
         self.damage_events_count = 0
         self.packet_count = 0
         self.decode_errors = 0
+        self.last_congestion_time = 0
         #these values are calculated from the values above (see update_averages)
         self.min_client_latency = self.DEFAULT_LATENCY
         self.avg_client_latency = self.DEFAULT_LATENCY
 …
         self.min_server_ping_latency = self.DEFAULT_LATENCY
         self.avg_server_ping_latency = self.DEFAULT_LATENCY
         self.recent_server_ping_latency = self.DEFAULT_LATENCY
+        self.avg_congestion_send_speed = 0
     def record_latency(self, wid, decode_time, start_send_at, end_send_at, pixels, bytecount):
         now = monotonic_time()
 …
             data = list(self.server_ping_latency)
             self.min_server_ping_latency = min([x for _,x in data])
             self.avg_server_ping_latency, self.recent_server_ping_latency = calculate_time_weighted_average(data)
+        #set to 0 if we have less than 2 events in the last 60 seconds:
+        min_time = monotonic_time()-60
+        css = tuple(x for x in self.congestion_send_speed if x[0]>min_time)
+        if len(css)<=1:
+            self.avg_congestion_send_speed = 0
+        else:
+            self.avg_congestion_send_speed = int(calculate_size_weighted_average(list(self.congestion_send_speed))[0])
     def get_factors(self, pixel_count):
         factors = []
 …
     def get_info(self):
         cwqsizes = [x for _,x in list(self.compression_work_qsizes)]
         pqsizes = [x for _,x in list(self.packet_qsizes)]
+        now = monotonic_time()
+        time_limit = now-60             #ignore old records (60s)
         info = {"damage" : {
                             "events"        : self.damage_events_count,
                             "packets_sent"  : self.packet_count,
 …
                                                },
                             },
                 "encoding" : {"decode_errors"   : self.decode_errors},
+                "congestion" : {
+                    "avg-send-speed"        : self.avg_congestion_send_speed,
+                    "last-congestion-event" : self.last_congestion_time,
+                    "late-packets-per-minute" : sum(1 for x in self.late_packets if x>time_limit),
+                                },
+            }
         #client pixels per second:
-        now = monotonic_time()
-        time_limit = now-30             #ignore old records (30s)
         #pixels per second: decode time and overall
         total_pixels = 0                #total number of pixels processed
         total_time = 0                  #total decoding time

xpra/server/window/batch_delay_calculator.py

     #combine factors: use the highest one:
     target = min(1.0, max(dam_lat_abs, dam_lat_rel, dec_lat, pps, 0.0))
+    #discount for congestion by up to 40%:
+    lp = len(global_statistics.late_packets)
+    target = target*max(40, 100-lp*20)/100
     #scale target between min_speed and 100:
     ms = min(100.0, max(min_speed, 0.0))
 …
                                            "target"   : int(target_decode_speed),
                                            "factor"   : int(100.0*dec_lat),
                                            },
+            "late-packets"              : lp,
+            }
     return info, target_speed
 …
             target = min(1.0, target + (1.0-pctpixdamaged*2))
         if pixl5<pixn5:
             target = sqrt(target)
+    #discount for congestion by up to 40%:
+    lp = len(global_statistics.late_packets)
+    target = target*max(40, 100-lp*20)/100
+    info["late-packets"] = lp
     #apply min-quality:
     mq = min(100.0, max(min_quality, 0.0))
     target_quality = mq + (100.0-mq) * target

xpra/server/window/window_source.py

 import hashlib
 import threading
 from collections import deque
+from math import sqrt
 from xpra.os_util import monotonic_time
 from xpra.util import envint, envbool, csv
 from xpra.log import Logger
 log = Logger("window", "encoding")
+log.enable_debug()
 refreshlog = Logger("window", "refresh")
 compresslog = Logger("window", "compress")
 damagelog = Logger("window", "damage")
 …
 MAX_PIXELS_PREFER_RGB = envint("XPRA_MAX_PIXELS_PREFER_RGB", 4096)
+CONGESTION_AVOIDANCE = envbool("XPRA_CONGESTION_AVOIDANCE", True)
 DELTA = envbool("XPRA_DELTA", True)
 MIN_DELTA_SIZE = envint("XPRA_MIN_DELTA_SIZE", 1024)
 MAX_DELTA_SIZE = envint("XPRA_MAX_DELTA_SIZE", 32768)
 …
 from xpra.server.window.batch_config import DamageBatchConfig
 from xpra.simple_stats import get_list_stats
 from xpra.server.window.batch_delay_calculator import calculate_batch_delay, get_target_speed, get_target_quality
 from xpra.server.cystats import time_weighted_average   #@UnresolvedImport
+from xpra.server.cystats import time_weighted_average, logp #@UnresolvedImport
 from xpra.server.window.region import rectangle, add_rectangle, remove_rectangle, merge_all   #@UnresolvedImport
 from xpra.codecs.xor.cyxor import xor_str           #@UnresolvedImport
 from xpra.server.picture_encode import rgb_encode, mmap_send, argb_swap
 …
         self.suspended = False
         self.strict = STRICT_MODE
+        #
+        self.congestion_timers = {}
         self.may_send_timer = None
         self.auto_refresh_delay = 0
         self.video_helper = None
 …
                                                "pixel_boost"    : self._lossless_threshold_pixel_boost
                                                },
                       })
+        info["congestion"] = {
+            "timers"                : len(self.congestion_timers),
+            }
         try:
             #ie: get_strict_encoding -> "strict_encoding"
             einfo["selection"] = self.get_best_encoding.__name__.replace("get_", "")
 …
         self.cancel_refresh_timer()
         self.cancel_timeout_timer()
         self.cancel_av_sync_timer()
+        self.cancel_congestion_timers()
         #if a region was delayed, we can just drop it now:
         self.refresh_regions = []
         self._damage_delayed = None
 …
         if self.full_frames_only:
             x, y, w, h = 0, 0, ww, wh
+        nlate_packets = len(self.global_statistics.late_packets)
         delayed = self._damage_delayed
         if delayed:
             #use existing delayed region:
 …
                         existing_options[k] = options[k]
             damagelog("damage%-24s wid=%s, using existing delayed %s regions created %.1fms ago",
                 (x, y, w, h, options), self.wid, delayed[3], now-delayed[0])
+            if not self.expire_timer and not self.soft_timer and self.soft_expired==0:
+                log.error("Error: bug, found a delayed region without a timer!")
+            if not self.expire_timer and not self.soft_timer and self.soft_expired==0 and not nlate_packets and not self.congestion_timers:
+                #this can happen after congestion recovery
+                log.warn("Warning: found a delayed region without a timer")
                 self.expire_timer = self.timeout_add(0, self.expire_delayed_region, 0)
             return
         elif self.batch_config.delay <= self.batch_config.min_delay and not self.batch_config.always:
 …
             delay = max(10, min(self.batch_config.min_delay, delay)) * (qsize/4.0)
         delay = max(delay, options.get("min_delay", 0))
         delay = min(delay, options.get("max_delay", self.batch_config.max_delay))
         delay = int(delay)
+        delay = int(delay*sqrt(1+nlate_packets))
         packets_backlog = self.statistics.get_packets_backlog()
         pixels_encoding_backlog, enc_backlog_count = self.statistics.get_pixels_encoding_backlog()
         #only send without batching when things are going well:
 …
         # - no packets backlog from the client
         # - the amount of pixels waiting to be encoded is less than one full frame refresh
         # - no more than 10 regions waiting to be encoded
         if not self.must_batch(delay) and (packets_backlog==0 and pixels_encoding_backlog<=ww*wh and enc_backlog_count<=10):
+        if not self.must_batch(delay) and (packets_backlog==0 and pixels_encoding_backlog<=ww*wh and enc_backlog_count<=10 and nlate_packets==0):
             #send without batching:
             damagelog("damage%-24s wid=%s, sending now with sequence %s", (x, y, w, h, options), self.wid, self._sequence)
             actual_encoding = options.get("encoding")
 …
         delayed = self._damage_delayed
         if not delayed:
             #region has been sent
             return
+            return False
         self.cancel_may_send_timer()
+        self.may_send_delayed()
+        delayed = self._damage_delayed
+        if not delayed:
+            #got sent
+            return
+        #the region has not been sent yet because we are waiting for damage ACKs from the client
+        lp = len(self.global_statistics.late_packets)
+        if lp==0:
+            self.may_send_delayed()
+            delayed = self._damage_delayed
+            if not delayed:
+                #got sent
+                return False
+        #the region has not been sent yet,
+        #because we are waiting for damage ACKs from the client,
+        #or even waiting to send some packets
         if self.soft_expired<self.max_soft_expired:
             #there aren't too many regions soft expired yet
+            #there aren't too many regions soft expired yet,
             #so use the "soft timer":
             self.soft_expired += 1
             #we have already waited for "delay" to get here, wait more as we soft expire more regions:
+            self.soft_timer = self.timeout_add(int(self.soft_expired*delay), self.delayed_region_soft_timeout)
+            soft_delay = int(delay*max(self.soft_expired, sqrt(1+lp)))
+            log("expire_delayed_region() soft delay=%i, soft_expired=%i, late packets=%i", soft_delay, self.soft_expired, lp)
+            self.soft_timer = self.timeout_add(soft_delay, self.delayed_region_soft_timeout, lp)
         else:
             #NOTE: this should never happen...
             #the region should now get sent when we eventually receive the pending ACKs
 …
             #but if somehow they go missing... clean it up from a timeout:
             delayed_region_time = delayed[0]
             self.timeout_timer = self.timeout_add(self.batch_config.timeout_delay, self.delayed_region_timeout, delayed_region_time)
+        return False
+    def delayed_region_soft_timeout(self):
+    def delayed_region_soft_timeout(self, lp):
+        #FIXME: if we have many late packets, re-schedule
+        # but need to know if they belong to this window to know how...
+        # (wait for ack vs timer)
+        new_lp = len(self.global_statistics.late_packets)
+        log("delayed_region_soft_timeout(%i) new lp=%i, late packets=%s", lp, new_lp, self.global_statistics.late_packets)
+        if lp>0 and new_lp>=lp:
+            #not getting better - re-schedule
+            return True
         self.soft_timer = None
         self.do_send_delayed()
         return False
 …
         if dap:
             log.warn(" %i late responses:", len(dap))
             for seq in sorted(dap.keys()):
                 ack_data = dap.get(seq)
                 if ack_data and ack_data[0]>0:
+                ack_data = dap[seq]
+                if ack_data[0]>0:
                     log.warn(" %6i %-5s: %3is", seq, ack_data[1], now-ack_data[3])
         #re-try: cancel anything pending and do a full quality refresh
         self.cancel_damage()
 …
         if not dd:
             log("window %s delayed region already sent", self.wid)
             return
+        stats = self.statistics
+        if not stats:
+            #closing
+            return
         damage_time = dd[0]
-        packets_backlog = self.statistics.get_packets_backlog()
         now = monotonic_time()
         actual_delay = int(1000.0 * (now-damage_time))
+        def check_again(delay=actual_delay/10.0):
+            #schedules a call to check again:
+            delay = int(min(self.batch_config.max_delay, max(10, delay)))
+            self.may_send_timer = self.timeout_add(delay, self._may_send_delayed)
+            return
+        lp = len(self.global_statistics.late_packets)
+        if lp:
+            #reduce the backlog tolerance:
+            pass
+        latency_tolerance_pct = 100//(1+lp)
+        packets_backlog = self.statistics.get_packets_backlog(latency_tolerance_pct)
         if packets_backlog>0:
             if actual_delay>self.batch_config.timeout_delay:
                 log.warn("send_delayed for wid %s, elapsed time %ims is above limit of %.1f", self.wid, actual_delay, self.batch_config.max_delay)
 …
         #if we're here, there is no packet backlog, but there may be damage acks pending or a bandwidth limit to honour,
         #if there are acks pending, may_send_delayed() should be called again from damage_packet_acked,
         #if not, we must either process the region now or set a timer to check again later
-        def check_again(delay=actual_delay/10.0):
-            #schedules a call to check again:
-            delay = int(min(self.batch_config.max_delay, max(10, delay)))
-            self.may_send_timer = self.timeout_add(delay, self._may_send_delayed)
-            return
         #locked means a fixed delay we try to honour,
         #this code ensures that we don't fire too early if called from damage_packet_acked
         if self.batch_config.locked:
 …
             else:
                 self.do_send_delayed()
             return
+        if self.bandwidth_limit>0:
+            used = self.statistics.get_bits_encoded()
+            log("may_send_delayed() bandwidth limit=%i, used=%i : %i%%", self.bandwidth_limit, used, 100*used//self.bandwidth_limit)
+            if used>=self.bandwidth_limit:
+        bwl = self.bandwidth_limit
+        if bwl>0:
+            #log("may_send_delayed() last 5 seconds: bandwidth limit=%i, used=%i : %i%%", bwl, used5, 100*used5//bwl)
+            used = self.statistics.get_bitrate(1) or self.statistics.get_bitrate(5)
+            log("may_send_delayed() last 1 second:  bandwidth limit=%i, used=%i : %i%%", bwl, used, 100*used//bwl)
+            if used>=bwl:
                 check_again(50)
                 return
         pixels_encoding_backlog, enc_backlog_count = self.statistics.get_pixels_encoding_backlog()
 …
         self.do_send_delayed()
     def do_send_delayed(self):
+        import traceback
+        traceback.print_stack(limit=4)
         self.cancel_timeout_timer()
         self.cancel_soft_timer()
         delayed = self._damage_delayed
 …
         def send_full_window_update():
             actual_encoding = get_encoding(ww*wh)
+            if actual_encoding=="png":
+                log.error("send_full_window_update() %s(%i)=%s, get_best_encoding=%s", get_encoding, ww*wh, actual_encoding, get_best_encoding)
+                import traceback
+                traceback.print_stack(limit=5)
             log("send_delayed_regions: using full window update %sx%s with %s", ww, wh, actual_encoding)
             assert actual_encoding is not None
             self.process_damage_region(damage_time, 0, 0, ww, wh, actual_encoding, options)
 …
                 #delay less when speed is high
                 #(the resulting range is 100*100 to 200*200)
                 qsmult = (200-self._current_quality) * (100+self._current_speed)
+                #when there is congestion, the auto-refresh can kill the bandwidth:
+                qsmult *= sqrt(1+len(self.global_statistics.late_packets))
                 #important: must check both, I think:
                 if target_time==0 or not self.refresh_timer:
                     #this means we must schedule the refresh
 …
             We figure out if now is the right time to do the refresh,
             and if not re-schedule.
         """
+        #FIXME: delay some more when we detect congestion
         #timer is running now, clear so we don't try to cancel it somewhere else:
         self.refresh_timer = None
         #re-do some checks that may have changed:
 …
             - damage latency (via a callback once the packet is actually sent)
         """
         #packet = ["draw", wid, x, y, w, h, coding, data, self._damage_packet_sequence, rowstride, client_options]
+        statistics = self.statistics
+        gs = self.global_statistics
+        now = monotonic_time()
         width = packet[4]
         height = packet[5]
         coding = packet[6]
+        ldata = len(packet[7])
         damage_packet_sequence = packet[8]
         actual_batch_delay = process_damage_time-damage_time
+        queue_time = now
         ack_pending = [0, coding, 0, 0, 0, width*height]
+        statistics = self.statistics
+        statistics.damage_ack_pending[damage_packet_sequence] = ack_pending
+        if statistics:
+            statistics.damage_ack_pending[damage_packet_sequence] = ack_pending
+        max_send_delay = int(5*logp(ldata/1024.0))
         def start_send(bytecount):
             ack_pending[0] = monotonic_time()
             ack_pending[2] = bytecount
+            statistics.last_sequence_sending = damage_packet_sequence
+            log.info("queue to send delay (start): %4ims (max=%3i, damage_packet_sequence=%5i)", 1000*(now-queue_time), max_send_delay, damage_packet_sequence)
+            if statistics:
+                statistics.last_sequence_sending = damage_packet_sequence
         def damage_packet_sent(bytecount):
+            if CONGESTION_AVOIDANCE:
+                #use idle_add to prevent race conditions accessing the list of late packets:
+                self.idle_add(self.cancel_congestion_timer, damage_packet_sequence)
             now = monotonic_time()
             ack_pending[3] = now
             ack_pending[4] = bytecount
+            if process_damage_time>0:
+                damage_out_latency = now-process_damage_time
+                statistics.damage_out_latency.append((now, width*height, actual_batch_delay, damage_out_latency))
+        if process_damage_time>0:
+            now = monotonic_time()
+            damage_in_latency = now-process_damage_time
+            statistics.damage_in_latency.append((now, width*height, actual_batch_delay, damage_in_latency))
+        #log.info("queuing %s packet with fail_cb=%s", coding, fail_cb)
+            if process_damage_time and statistics:
+                statistics.damage_out_latency.append((now, width*height, actual_batch_delay, now-process_damage_time))
+            if gs:
+                elapsed = now-ack_pending[0]
+                send_speed = 0
+                if elapsed>0:
+                    send_speed = int(8*ldata/elapsed)
+                log.info("queue to send delay (end)  : %4ims (max=%3i, damage_packet_sequence=%5i) for %4iKB using %6s: %6iMbps, elapsed=%.2fms", 1000*elapsed, max_send_delay, damage_packet_sequence, ldata//1024, coding, send_speed//1024//1024, elapsed*1000)
+                if elapsed*1000>max_send_delay and send_speed<100*1024*1024:
+                    log.warn("recording congestion send speed=%i", send_speed)
+                    gs.congestion_send_speed.append((now, ldata, send_speed))
+        if statistics:
+            statistics.last_sequence_queued = damage_packet_sequence
+            if process_damage_time:
+                statistics.damage_in_latency.append((now, width*height, actual_batch_delay, now-process_damage_time))
+        if CONGESTION_AVOIDANCE:
+            #  1KB -> 5ms, 10KB -> 17ms, 100KB -> 33ms, 1MB -> 50ms, 10MB -> 66ms
+            log.info("queueing %4i, size=%i, max_send_delay=%4i", damage_packet_sequence, ldata, max_send_delay)
+            t = self.timeout_add(max_send_delay, self.congestion_timer_event, damage_packet_sequence)
+            self.congestion_timers[damage_packet_sequence] = t
         self.queue_packet(packet, self.wid, width*height, start_send, damage_packet_sent, self.get_fail_cb(packet))
+    def congestion_timer_event(self, damage_packet_sequence):
+        log.info("congestion_timer_event() damage_packet_sequence=%s, late packets=%s", damage_packet_sequence, self.global_statistics.late_packets)
+        try:
+            del self.congestion_timers[damage_packet_sequence]
+        except:
+            pass
+        self.global_statistics.late_packets.add((self.wid, damage_packet_sequence))
+        self.global_statistics.last_congestion_time = monotonic_time()
+        return False
+    def cancel_congestion_timers(self):
+        timers = list(self.congestion_timers.values())
+        self.congestion_timers = {}
+        for t in timers:
+            self.source_remove(t)
+    def cancel_congestion_timer(self, damage_packet_sequence):
+        t = self.congestion_timers.get(damage_packet_sequence)
+        log("cancel_congestion_timer(%i) timer=%s", damage_packet_sequence, t)
+        if t:
+            self.source_remove(t)
+            try:
+                del self.congestion_timers[damage_packet_sequence]
+            except:
+                pass
+        #record last_congestion_time before removing
+        #what is potentially the last item from late_packets
+        self.global_statistics.last_congestion_time = monotonic_time()
+        late_packets = self.global_statistics.late_packets
+        l = len(late_packets)
+        try:
+            late_packets.remove((self.wid, damage_packet_sequence))
+        except KeyError:
+            #congestion event had not fired yet
+            return
+        if l>0 and len(late_packets)==0:
+            log.info("cancel_congestion_timer(%i) last congestion sequence cleared, trying to send delayed region", damage_packet_sequence)
+            self.may_send_delayed()
+        else:
+            log.info("cancel_congestion_timer(%i) remaining congestion sequences=%s", damage_packet_sequence, late_packets)
     def get_fail_cb(self, packet):
         def resend():
             log("paint packet failure, resending")
 …
         end = monotonic_time()
         compresslog("compress: %5.1fms for %4ix%-4i pixels at %4i,%-4i for wid=%-5i using %6s with ratio %5.1f%%  (%5iKB to %5iKB), sequence %5i, client_options=%s",
                  (end-start)*1000.0, outw, outh, x, y, self.wid, coding, 100.0*csize/psize, psize/1024, csize/1024, self._damage_packet_sequence, client_options)
         self.statistics.encoding_stats.append((end, coding, w*h, bpp, len(data), end-start))
+        self.statistics.encoding_stats.append((end, coding, w*h, bpp, csize, end-start))
         return self.make_draw_packet(x, y, outw, outh, coding, data, outstride, client_options, options)
     def make_draw_packet(self, x, y, outw, outh, coding, data, outstride, client_options={}, options={}):

xpra/server/window/window_stats.py

         self.last_recalculate = 0
         self.damage_events_count = 0
         self.packet_count = 0
+        self.last_sequence_queued = 0
+        self.last_sequence_sending = 0
         self.last_resized = 0
 …
         #client decode speed:
         if len(self.client_decode_time)>0:
             #the elapsed time recorded is in microseconds, so multiply by 1000*1000 to get the real value:
             self.avg_decode_speed, self.recent_decode_speed = calculate_timesize_weighted_average(list(self.client_decode_time), sizeunit=1000*1000)
+            self.avg_decode_speed, self.recent_decode_speed = calculate_timesize_weighted_average(list(self.client_decode_time), unit=1000*1000)
         #network send speed:
         all_l = [0.1,
                  self.avg_damage_in_latency, self.recent_damage_in_latency,
 …
         info = {"damage"    : {"events"         : self.damage_events_count,
                                "packets_sent"   : self.packet_count,
                                "target-latency" : int(1000*self.target_latency),
+                               "last-queued"    : self.last_sequence_queued,
+                               "last-sending"   : self.last_sequence_sending,
+                               }
+                }
         #encoding stats:
 …
     def get_acks_pending(self):
         return sum(1 for x in self.damage_ack_pending.values() if x[0]!=0)
     def get_packets_backlog(self):
+    def get_packets_backlog(self, latency_tolerance_pct=100):
         packets_backlog = 0
         if len(self.damage_ack_pending)>0:
             sent_before = monotonic_time()-(self.target_latency+0.020)
+            sent_before = monotonic_time()-((self.target_latency+0.020)*100.0/latency_tolerance_pct)
             for _, (start_send_at, _, _, end_send_at, _, _) in self.damage_ack_pending.items():
                 if end_send_at>0 and start_send_at<=sent_before:
                     packets_backlog += 1
 …
             count += 1
         return pixels, count
+    def get_bits_encoded(self, elapsed=1):
+        cutoff = monotonic_time()-elapsed
+        return sum(v[4] for v in self.encoding_stats if v[0]>cutoff) * 8
+    def get_bitrate(self, max_elapsed=1):
+        cutoff = monotonic_time()-max_elapsed
+        recs = tuple((v[0], v[4]) for v in self.encoding_stats if v[0]>=cutoff)
+        if len(recs)<2:
+            return 0
+        bits = sum(v[1] for v in recs) * 8
+        elapsed = recs[-1][0]-recs[0][0]
+        return bits/elapsed
     def get_damage_pixels(self, elapsed=1):
         cutoff = monotonic_time()-elapsed

xpra/server/window/window_video_source.py

         #take into account how many pixels need to be encoded:
         #more pixels means we switch to lossless more easily
         lossless_q = min(100, self._lossless_threshold_base + self._lossless_threshold_pixel_boost * pixel_count / (ww*wh))
+        if quality<lossless_q and self.image_depth>16 and "jpeg" in options:
+        lce = monotonic_time()-self.global_statistics.last_congestion_time
+        if (quality<lossless_q or self.global_statistics.late_packets or lce<1.0) and self.image_depth>16 and "jpeg" in options:
             #assume that we have "turbojpeg",
             #which beats everything in terms of efficiency for lossy compression:
             return "jpeg"

Context Navigation

Ticket #999: bandwidth-congestion-handling-v6.patch

xpra/server/cystats.pyx

xpra/server/source.py

xpra/server/source_stats.py

xpra/server/window/batch_delay_calculator.py

xpra/server/window/window_source.py

xpra/server/window/window_stats.py

xpra/server/window/window_video_source.py

Download in other formats: