RNS/Buffer.py

@@ -29,7 +29,7 @@ class StreamDataMessage(MessageBase):
     calculcated based on the value of OVERHEAD
     """
 
-    def __init__(self, stream_id: int = None, data: bytes = None, eof: bool = False, compressed: bool = False):
+    def __init__(self, stream_id: int = None, data: bytes = None, eof: bool = False):
         """
         This class is used to encapsulate binary stream
         data to be sent over a ``Channel``.
@@ -42,7 +42,7 @@ class StreamDataMessage(MessageBase):
         if stream_id is not None and stream_id > self.STREAM_ID_MAX:
             raise ValueError("stream_id must be 0-16383")
         self.stream_id = stream_id
-        self.compressed = compressed
+        self.compressed = False
         self.data = data or bytes()
         self.eof = eof
 
@@ -50,6 +50,13 @@ class StreamDataMessage(MessageBase):
         if self.stream_id is None:
             raise ValueError("stream_id")
 
+        compressed_data   = bz2.compress(self.data)
+        saved = len(self.data)-len(compressed_data)
+
+        if saved > 0:
+            self.data = compressed_data
+            self.compressed = True
+
         header_val = (0x3fff & self.stream_id) | (0x8000 if self.eof else 0x0000) | (0x4000 if self.compressed > 0 else 0x0000)
         return bytes(struct.pack(">H", header_val) + (self.data if self.data else bytes()))
 
@@ -126,7 +133,7 @@ class RawChannelReader(RawIOBase, AbstractContextManager):
                         try:
                             threading.Thread(target=listener, name="Message Callback", args=[len(self._buffer)], daemon=True).start()
                         except Exception as ex:
-                            RNS.log("Error calling RawChannelReader(" + str(self._stream_id) + ") callback: " + str(ex), RNS.LOG_ERROR)
+                            RNS.log("Error calling RawChannelReader(" + str(self._stream_id) + ") callback: " + str(ex))
                     return True
         return False
 
@@ -179,10 +186,6 @@ class RawChannelWriter(RawIOBase, AbstractContextManager):
       object, see the Python documentation for
       ``RawIOBase``.
     """
-
-    MAX_CHUNK_LEN     = 1024*16
-    COMPRESSION_TRIES = 4
-
     def __init__(self, stream_id: int, channel: Channel):
         """
         Create a raw channel writer.
@@ -196,36 +199,10 @@ class RawChannelWriter(RawIOBase, AbstractContextManager):
 
     def write(self, __b: bytes) -> int | None:
         try:
-            comp_tries = RawChannelWriter.COMPRESSION_TRIES
-            comp_try = 1
-            comp_success = False
-            chunk_len = len(__b)
-            if chunk_len > RawChannelWriter.MAX_CHUNK_LEN:
-                chunk_len = RawChannelWriter.MAX_CHUNK_LEN
-                __b = __b[:RawChannelWriter.MAX_CHUNK_LEN]
-            chunk_segment = None
-            while chunk_len > 32 and comp_try < comp_tries:
-                chunk_segment_length = int(chunk_len/comp_try)
-                compressed_chunk = bz2.compress(__b[:chunk_segment_length])
-                compressed_length = len(compressed_chunk)
-                if compressed_length < StreamDataMessage.MAX_DATA_LEN and compressed_length < chunk_segment_length:
-                    comp_success = True
-                    break
-                else:
-                    comp_try += 1
-
-            if comp_success:
-                chunk = compressed_chunk
-                processed_length = chunk_segment_length
-            else:
             chunk = bytes(__b[:StreamDataMessage.MAX_DATA_LEN])
-                processed_length = len(chunk)
+            message = StreamDataMessage(self._stream_id, chunk, self._eof)
-
-            message = StreamDataMessage(self._stream_id, chunk, self._eof, comp_success)
-            
             self._channel.send(message)
-            return processed_length
+            return len(chunk)
-
         except RNS.Channel.ChannelException as cex:
             if cex.type != RNS.Channel.CEType.ME_LINK_NOT_READY:
                 raise

RNS/Channel.py

@@ -234,16 +234,13 @@ class Channel(contextlib.AbstractContextManager):
     WINDOW     = 2
 
     # Absolute minimum window size
-    WINDOW_MIN = 2
+    WINDOW_MIN = 1
-    WINDOW_MIN_LIMIT_SLOW    = 2
-    WINDOW_MIN_LIMIT_MEDIUM  = 5
-    WINDOW_MIN_LIMIT_FAST    = 16
 
     # The maximum window size for transfers on slow links
     WINDOW_MAX_SLOW      = 5
 
     # The maximum window size for transfers on mid-speed links
-    WINDOW_MAX_MEDIUM    = 12
+    WINDOW_MAX_MEDIUM    = 16
 
     # The maximum window size for transfers on fast links
     WINDOW_MAX_FAST      = 48
@@ -258,7 +255,7 @@ class Channel(contextlib.AbstractContextManager):
 
     # If the RTT rate is higher than this value,
     # the max window size for fast links will be used.
-    RTT_FAST            = 0.18
+    RTT_FAST            = 0.25
     RTT_MEDIUM          = 0.75
     RTT_SLOW            = 1.45
 
@@ -388,8 +385,9 @@ class Channel(contextlib.AbstractContextManager):
                     RNS.log(f"Envelope: Emplacement of duplicate envelope with sequence "+str(envelope.sequence), RNS.LOG_EXTREME)
                     return False
                 
-                if envelope.sequence < existing.sequence and not (self._next_rx_sequence - envelope.sequence) > (Channel.SEQ_MAX//2):
+                if envelope.sequence < existing.sequence and not envelope.sequence < window_overflow:
                     ring.insert(i, envelope)
+                    RNS.log("Inserted seq "+str(envelope.sequence)+" at "+str(i), RNS.LOG_DEBUG)
 
                     envelope.tracked = True
                     return True
@@ -398,7 +396,6 @@ class Channel(contextlib.AbstractContextManager):
             
             envelope.tracked = True
             ring.append(envelope)
-
             return True
 
     def _run_callbacks(self, message: MessageBase):
@@ -439,11 +436,6 @@ class Channel(contextlib.AbstractContextManager):
                         if e.sequence == self._next_rx_sequence:
                             contigous.append(e)
                             self._next_rx_sequence = (self._next_rx_sequence + 1) % Channel.SEQ_MODULUS
-                            if self._next_rx_sequence == 0:
-                                for e in self._rx_ring:
-                                    if e.sequence == self._next_rx_sequence:
-                                        contigous.append(e)
-                                        self._next_rx_sequence = (self._next_rx_sequence + 1) % Channel.SEQ_MODULUS
 
                     for e in contigous:
                         if not e.unpacked:
@@ -482,7 +474,6 @@ class Channel(contextlib.AbstractContextManager):
         with self._lock:
             envelope = next(filter(lambda e: self._outlet.get_packet_id(e.packet) == self._outlet.get_packet_id(packet),
                                    self._tx_ring), None)
-
             if envelope and op(envelope):
                 envelope.tracked = False
                 if envelope in self._tx_ring:
@@ -490,9 +481,11 @@ class Channel(contextlib.AbstractContextManager):
 
                     if self.window < self.window_max:
                         self.window += 1
+                        if (self.window - self.window_min) > (self.window_flexibility-1):
+                            self.window_min += 1
 
                         # TODO: Remove at some point
-                        # RNS.log("Increased "+str(self)+" window to "+str(self.window), RNS.LOG_DEBUG)
+                        # RNS.log("Increased "+str(self)+" window to "+str(self.window), RNS.LOG_EXTREME)
 
                     if self._outlet.rtt != 0:
                         if self._outlet.rtt > Channel.RTT_FAST:
@@ -505,20 +498,15 @@ class Channel(contextlib.AbstractContextManager):
                                 self.medium_rate_rounds += 1
                                 if self.window_max < Channel.WINDOW_MAX_MEDIUM and self.medium_rate_rounds == Channel.FAST_RATE_THRESHOLD:
                                     self.window_max = Channel.WINDOW_MAX_MEDIUM
-                                    self.window_min = Channel.WINDOW_MIN_LIMIT_MEDIUM
                                     # TODO: Remove at some point
-                                    # RNS.log("Increased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_DEBUG)
+                                    # RNS.log("Increased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_EXTREME)
-                                    # RNS.log("Increased "+str(self)+" min window to "+str(self.window_min), RNS.LOG_DEBUG)
                             
                         else:
                             self.fast_rate_rounds += 1
                             if self.window_max < Channel.WINDOW_MAX_FAST and self.fast_rate_rounds == Channel.FAST_RATE_THRESHOLD:
                                 self.window_max = Channel.WINDOW_MAX_FAST                                
-                                self.window_min = Channel.WINDOW_MIN_LIMIT_FAST
                                 # TODO: Remove at some point
-                                # RNS.log("Increased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_DEBUG)
+                                # RNS.log("Increased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_EXTREME)
-                                # RNS.log("Increased "+str(self)+" min window to "+str(self.window_min), RNS.LOG_DEBUG)
-
 
                 else:
                     RNS.log("Envelope not found in TX ring for "+str(self), RNS.LOG_EXTREME)
@@ -528,15 +516,8 @@ class Channel(contextlib.AbstractContextManager):
     def _packet_delivered(self, packet: TPacket):
         self._packet_tx_op(packet, lambda env: True)
 
-    def _update_packet_timeouts(self):
-        for envelope in self._tx_ring:
-            updated_timeout = self._get_packet_timeout_time(envelope.tries)
-            if updated_timeout > envelope.packet.receipt.timeout:
-                envelope.packet.receipt.set_timeout(updated_timeout)
-
     def _get_packet_timeout_time(self, tries: int) -> float:
-        to = pow(1.5, tries - 1) * max(self._outlet.rtt*2.5, 0.025) * (len(self._tx_ring)+1.5)
+        return pow(2, tries - 1) * max(self._outlet.rtt, 0.01) * 5
-        return to
 
     def _packet_timeout(self, packet: TPacket):
         def retry_envelope(envelope: Envelope) -> bool:
@@ -545,22 +526,17 @@ class Channel(contextlib.AbstractContextManager):
                 self._shutdown()  # start on separate thread?
                 self._outlet.timed_out()
                 return True
-
             envelope.tries += 1
             self._outlet.resend(envelope.packet)
             self._outlet.set_packet_delivered_callback(envelope.packet, self._packet_delivered)
             self._outlet.set_packet_timeout_callback(envelope.packet, self._packet_timeout, self._get_packet_timeout_time(envelope.tries))
-            self._update_packet_timeouts()
 
             if self.window > self.window_min:
                 self.window -= 1
-                # TODO: Remove at some point
+                if self.window_max > self.window_min:
-                # RNS.log("Decreased "+str(self)+" window to "+str(self.window), RNS.LOG_DEBUG)
+                    self.window_max -= 1
-
+                    if (self.window_max - self.window) > (self.window_flexibility-1):
-                if self.window_max > (self.window_min+self.window_flexibility):
                         self.window_max -= 1
-                    # TODO: Remove at some point
-                    # RNS.log("Decreased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_DEBUG)
 
                 # TODO: Remove at some point
                 # RNS.log("Decreased "+str(self)+" window to "+str(self.window), RNS.LOG_EXTREME)
@@ -597,7 +573,6 @@ class Channel(contextlib.AbstractContextManager):
         envelope.tries += 1
         self._outlet.set_packet_delivered_callback(envelope.packet, self._packet_delivered)
         self._outlet.set_packet_timeout_callback(envelope.packet, self._packet_timeout, self._get_packet_timeout_time(envelope.tries))
-        self._update_packet_timeouts()
 
         return envelope
 

RNS/Interfaces/Android/RNodeInterface.py

@@ -53,7 +53,6 @@ class KISS():
     CMD_STAT_RSSI   = 0x23
     CMD_STAT_SNR    = 0x24
     CMD_STAT_CHTM   = 0x25
-    CMD_STAT_PHYPRM = 0x26
     CMD_BLINK       = 0x30
     CMD_RANDOM      = 0x40
     CMD_FB_EXT      = 0x41
@@ -406,10 +405,6 @@ class RNodeInterface(Interface):
         self.r_airtime_long       = 0.0
         self.r_channel_load_short = 0.0
         self.r_channel_load_long  = 0.0
-        self.r_symbol_time_ms = None
-        self.r_symbol_rate = None
-        self.r_preamble_symbols = None
-        self.r_premable_time_ms = None
 
         self.packet_queue    = []
         self.flow_control    = flow_control
@@ -1096,33 +1091,6 @@ class RNodeInterface(Interface):
                                     self.r_airtime_long       = atl/100.0
                                     self.r_channel_load_short = cus/100.0
                                     self.r_channel_load_long  = cul/100.0
-                        elif (command == KISS.CMD_STAT_PHYPRM):
-                            if (byte == KISS.FESC):
-                                escape = True
-                            else:
-                                if (escape):
-                                    if (byte == KISS.TFEND):
-                                        byte = KISS.FEND
-                                    if (byte == KISS.TFESC):
-                                        byte = KISS.FESC
-                                    escape = False
-                                command_buffer = command_buffer+bytes([byte])
-                                if (len(command_buffer) == 10):
-                                    lst = (command_buffer[0] << 8 | command_buffer[1])/1000.0
-                                    lsr = command_buffer[2] << 8 | command_buffer[3]
-                                    prs = command_buffer[4] << 8 | command_buffer[5]
-                                    prt = command_buffer[6] << 8 | command_buffer[7]
-                                    cst = command_buffer[8] << 8 | command_buffer[9]
-
-                                    if lst != self.r_symbol_time_ms or lsr != self.r_symbol_rate or prs != self.r_preamble_symbols or prt != self.r_premable_time_ms or cst != self.r_csma_slot_time_ms:
-                                        self.r_symbol_time_ms    = lst
-                                        self.r_symbol_rate       = lsr
-                                        self.r_preamble_symbols  = prs
-                                        self.r_premable_time_ms  = prt
-                                        self.r_csma_slot_time_ms = cst
-                                        RNS.log(str(self)+" Radio reporting symbol time is "+str(round(self.r_symbol_time_ms,2))+"ms (at "+str(self.r_symbol_rate)+" baud)", RNS.LOG_DEBUG)
-                                        RNS.log(str(self)+" Radio reporting preamble is "+str(self.r_preamble_symbols)+" symbols ("+str(self.r_premable_time_ms)+"ms)", RNS.LOG_DEBUG)
-                                        RNS.log(str(self)+" Radio reporting CSMA slot time is "+str(self.r_csma_slot_time_ms)+"ms", RNS.LOG_DEBUG)
                         elif (command == KISS.CMD_RANDOM):
                             self.r_random = byte
                         elif (command == KISS.CMD_PLATFORM):

RNS/Interfaces/RNodeInterface.py

@@ -53,7 +53,6 @@ class KISS():
     CMD_STAT_RSSI   = 0x23
     CMD_STAT_SNR    = 0x24
     CMD_STAT_CHTM   = 0x25
-    CMD_STAT_PHYPRM = 0x26
     CMD_BLINK       = 0x30
     CMD_RANDOM      = 0x40
     CMD_FB_EXT      = 0x41
@@ -171,10 +170,6 @@ class RNodeInterface(Interface):
         self.r_airtime_long       = 0.0
         self.r_channel_load_short = 0.0
         self.r_channel_load_long  = 0.0
-        self.r_symbol_time_ms = None
-        self.r_symbol_rate = None
-        self.r_preamble_symbols = None
-        self.r_premable_time_ms = None
 
         self.packet_queue    = []
         self.flow_control    = flow_control
@@ -723,33 +718,6 @@ class RNodeInterface(Interface):
                                     self.r_airtime_long       = atl/100.0
                                     self.r_channel_load_short = cus/100.0
                                     self.r_channel_load_long  = cul/100.0
-                        elif (command == KISS.CMD_STAT_PHYPRM):
-                            if (byte == KISS.FESC):
-                                escape = True
-                            else:
-                                if (escape):
-                                    if (byte == KISS.TFEND):
-                                        byte = KISS.FEND
-                                    if (byte == KISS.TFESC):
-                                        byte = KISS.FESC
-                                    escape = False
-                                command_buffer = command_buffer+bytes([byte])
-                                if (len(command_buffer) == 10):
-                                    lst = (command_buffer[0] << 8 | command_buffer[1])/1000.0
-                                    lsr = command_buffer[2] << 8 | command_buffer[3]
-                                    prs = command_buffer[4] << 8 | command_buffer[5]
-                                    prt = command_buffer[6] << 8 | command_buffer[7]
-                                    cst = command_buffer[8] << 8 | command_buffer[9]
-
-                                    if lst != self.r_symbol_time_ms or lsr != self.r_symbol_rate or prs != self.r_preamble_symbols or prt != self.r_premable_time_ms or cst != self.r_csma_slot_time_ms:
-                                        self.r_symbol_time_ms    = lst
-                                        self.r_symbol_rate       = lsr
-                                        self.r_preamble_symbols  = prs
-                                        self.r_premable_time_ms  = prt
-                                        self.r_csma_slot_time_ms = cst
-                                        RNS.log(str(self)+" Radio reporting symbol time is "+str(round(self.r_symbol_time_ms,2))+"ms (at "+str(self.r_symbol_rate)+" baud)", RNS.LOG_DEBUG)
-                                        RNS.log(str(self)+" Radio reporting preamble is "+str(self.r_preamble_symbols)+" symbols ("+str(self.r_premable_time_ms)+"ms)", RNS.LOG_DEBUG)
-                                        RNS.log(str(self)+" Radio reporting CSMA slot time is "+str(self.r_csma_slot_time_ms)+"ms", RNS.LOG_DEBUG)
                         elif (command == KISS.CMD_RANDOM):
                             self.r_random = byte
                         elif (command == KISS.CMD_PLATFORM):

RNS/_version.py

@@ -1 +1 @@
-__version__ = "0.5.9"
+__version__ = "0.5.8"

tests/channel.py

@@ -407,7 +407,7 @@ class TestChannel(unittest.TestCase):
     def test_buffer_big(self):
         writer = RNS.Buffer.create_writer(15, self.h.channel)
         reader = RNS.Buffer.create_reader(15, self.h.channel)
-        data = "01234556789"*1024*5  # 50 KB
+        data = "01234556789"*1024  # 10 KB
         count = 0
         write_finished = False