Added channel CSMA parameter stats to RNode Interface

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):
+    def __init__(self, stream_id: int = None, data: bytes = None, eof: bool = False, compressed: 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 = False
+        self.compressed = compressed
         self.data = data or bytes()
         self.eof = eof
 
@@ -50,13 +50,6 @@ 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()))
 
@@ -133,7 +126,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 calling RawChannelReader(" + str(self._stream_id) + ") callback: " + str(ex), RNS.LOG_ERROR)
                     return True
         return False
 
@@ -186,6 +179,10 @@ 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.
@@ -199,10 +196,36 @@ class RawChannelWriter(RawIOBase, AbstractContextManager):
 
     def write(self, __b: bytes) -> int | None:
         try:
-            chunk = bytes(__b[:StreamDataMessage.MAX_DATA_LEN])
-            message = StreamDataMessage(self._stream_id, chunk, self._eof)
+            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, comp_success)
+            
             self._channel.send(message)
-            return len(chunk)
+            return processed_length
+
         except RNS.Channel.ChannelException as cex:
             if cex.type != RNS.Channel.CEType.ME_LINK_NOT_READY:
                 raise

RNS/Channel.py

@@ -234,13 +234,16 @@ class Channel(contextlib.AbstractContextManager):
     WINDOW     = 2
 
     # Absolute minimum window size
-    WINDOW_MIN = 1
+    WINDOW_MIN = 2
+    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    = 16
+    WINDOW_MAX_MEDIUM    = 12
 
     # The maximum window size for transfers on fast links
     WINDOW_MAX_FAST      = 48
@@ -255,7 +258,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.25
+    RTT_FAST            = 0.18
     RTT_MEDIUM          = 0.75
     RTT_SLOW            = 1.45
 
@@ -385,9 +388,8 @@ 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 envelope.sequence < window_overflow:
+                if envelope.sequence < existing.sequence and not (self._next_rx_sequence - envelope.sequence) > (Channel.SEQ_MAX//2):
                     ring.insert(i, envelope)
-                    RNS.log("Inserted seq "+str(envelope.sequence)+" at "+str(i), RNS.LOG_DEBUG)
 
                     envelope.tracked = True
                     return True
@@ -396,6 +398,7 @@ class Channel(contextlib.AbstractContextManager):
             
             envelope.tracked = True
             ring.append(envelope)
+
             return True
 
     def _run_callbacks(self, message: MessageBase):
@@ -436,6 +439,11 @@ 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:
@@ -474,6 +482,7 @@ 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:
@@ -481,11 +490,9 @@ 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_EXTREME)
+                        # RNS.log("Increased "+str(self)+" window to "+str(self.window), RNS.LOG_DEBUG)
 
                     if self._outlet.rtt != 0:
                         if self._outlet.rtt > Channel.RTT_FAST:
@@ -498,15 +505,20 @@ 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_EXTREME)
+                                    # RNS.log("Increased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_DEBUG)
+                                    # 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_EXTREME)
+                                # RNS.log("Increased "+str(self)+" max window to "+str(self.window_max), RNS.LOG_DEBUG)
+                                # 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)
@@ -516,8 +528,15 @@ 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:
-        return pow(2, tries - 1) * max(self._outlet.rtt, 0.01) * 5
+        to = pow(1.5, tries - 1) * max(self._outlet.rtt*2.5, 0.025) * (len(self._tx_ring)+1.5)
+        return to
 
     def _packet_timeout(self, packet: TPacket):
         def retry_envelope(envelope: Envelope) -> bool:
@@ -526,17 +545,22 @@ 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
-                if self.window_max > self.window_min:
+                # TODO: Remove at some point
+                # RNS.log("Decreased "+str(self)+" window to "+str(self.window), RNS.LOG_DEBUG)
+
+                if self.window_max > (self.window_min+self.window_flexibility):
                     self.window_max -= 1
-                    if (self.window_max - self.window) > (self.window_flexibility-1):
-                        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)
@@ -573,6 +597,7 @@ 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,6 +53,7 @@ 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
@@ -405,6 +406,10 @@ 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
@@ -1091,6 +1096,33 @@ 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,6 +53,7 @@ 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
@@ -170,6 +171,10 @@ 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
@@ -718,6 +723,33 @@ 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.8"
+__version__ = "0.5.9"

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  # 10 KB
+        data = "01234556789"*1024*5  # 50 KB
         count = 0
         write_finished = False