Repository: XKNX/xknx

Issue:
Tunneling: L_Data.con Frames are not awaited properly before sending next L_Data.req
I have noticed that, for tunnelling connections, every sent frame is processed twice. Once as OUTGOING and once as INCOMING frame. 
When we send a TunnelingRequest the frame is sent as L_Data.req. Transmission shall be confirmed in L_Data.con. We don't handle incoming L_Data.req, L_Data.con and L_Data.ind differently so the confirmation (.con) is treated like an incoming telegram.

I'm not really sure what to do about this. As far as I understand this we should only process (as in fire callbacks) L_Data.ind frames. L_Data.con are just confirmations of our own frames and L_Data.req are only sent by the client (xknx), not by the tunneling server. 

Shall we just ignore the L_Data.con frame and hope for the best? Handling this would probably require a second RequestResponse layer for tunneling...

---------

From 3/3/3 Data Link Layer General §2.2:

> There are 3 service primitives:
> 1. L_Data.req    shall be used to transmit a Frame. 
> 2. L_Data.ind    shall be used to receive a Frame.
> 3. L_Data.con    shall be a local primitive generated by the local Layer-2 for its own client to indicate that the it is satisfied with the transmission.
> 
> If the local user of Layer-2 prepares an LSDU for the remote user it shall apply the L_Data.req primitive to pass the LSDU to the local Layer-2. The local Layer-2 shall accept the service request and shall try to send the LSDU to the remote Layer-2 with the relevant Frame format.
> The local Layer-2 shall pass an L_Data.con primitive to the local user that shall indicate either a correct or erroneous data transfer. Depending if an L2-acknowledgement is requested or not, this confirmation shall be related to the reception of the L2-acknowledgement, or only to the transmission of the Frame on the medium.


From 3/6/3 EMI_IMI §4.1.5 Data Link Layer messages

>4.1.5.1 Flow Control cEMI Client:
>
> To keep the flow control for Data Link Layer services as simple as possible (this allows a simple flow control state machine in the cEMI client), it is recommended that:
> - a cEMI client sends a new Data Link Layer request only when the confirmation of the preceding request is received, or
> - a request-to-confirmation timeout is recognised; the recommended time-out for the cEMI client is 3 seconds.
>
> A cEMI client shall at any time be able to accept an indication message from the cEMI Server.

--------

Possibly the underlying issue of https://github.com/home-assistant/core/issues/38130


Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/822?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#822](https://codecov.io/gh/XKNX/xknx/pull/822?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (04fe346) into [main](https://codecov.io/gh/XKNX/xknx/commit/cb608c187d97c28ac94a21d279f06a3c0ae15ec0?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (cb608c1) will **decrease** coverage by `0.17%`.
> The diff coverage is `22.22%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/822/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/822?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main     #822      +/-   ##
==========================================
- Coverage   94.77%   94.59%   -0.18%     
==========================================
  Files         109      109              
  Lines        6831     6847      +16     
==========================================
+ Hits         6474     6477       +3     
- Misses        357      370      +13     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/822?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/822/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `36.55% <22.22%> (-1.68%)` | :arrow_down: |

marvin-w: Works with my test script and on my local HA without any issue.

Patch:
diff --git a/changelog.md b/changelog.md
index 2c7802de..575b7045 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased changes
+
+### Internals
+
+- Tunnel: Implement flow control according to KNX spec recommendations: wait for L_DATA.con frame before sending next L_DATA.req with 3 second timeout
+
 ## 0.18.13 Hold your colour 2021-11-13
 
 ### Internals
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index e2ff3107..bdc55c1a 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -34,6 +34,9 @@ TelegramCallbackType = Callable[[Telegram], None]
 
 logger = logging.getLogger("xknx.log")
 
+# See 3/6/3 EMI_IMI §4.1.5 Data Link Layer messages
+REQUEST_TO_CONFIRMATION_TIMEOUT = 3
+
 
 class Tunnel(Interface):
     """Class for handling KNX/IP tunnels."""
@@ -59,6 +62,8 @@ class Tunnel(Interface):
         self.route_back = route_back
         self.telegram_received_callback = telegram_received_callback
 
+        self._tunnelling_request_confirmation_event = asyncio.Event()
+
         self.udp_client: UDPClient
         self.init_udp_client()
 
@@ -118,10 +123,10 @@ class Tunnel(Interface):
             await self.udp_client.stop()
             raise ex
         else:
+            self._tunnel_established()
             await self.xknx.connection_manager.connection_state_changed(
                 XknxConnectionState.CONNECTED
             )
-            self._tunnel_established()
             return True
 
     def _tunnel_established(self) -> None:
@@ -273,7 +278,20 @@ class Tunnel(Interface):
             self.sequence_number,
             self.communication_channel,
         )
-        await tunnelling.start()
+        self._tunnelling_request_confirmation_event.clear()
+        send_and_wait_for_confirmation = asyncio.gather(
+            tunnelling.start(), self._tunnelling_request_confirmation_event.wait()
+        )
+        try:
+            await asyncio.wait_for(
+                send_and_wait_for_confirmation, timeout=REQUEST_TO_CONFIRMATION_TIMEOUT
+            )
+        except asyncio.TimeoutError:
+            # REQUEST_TO_CONFIRMATION_TIMEOUT is longer than tunnelling timeout of 1 second
+            # so exception should always be from self._tunnelling_request_confirmation_event
+            logger.warning(
+                "L_DATA_CON Data Link Layer confirmation timed out for %s", telegram
+            )
         return tunnelling.success
 
     def _increase_sequence_number(self) -> None:
@@ -301,20 +319,29 @@ class Tunnel(Interface):
         self, tunneling_request: TunnellingRequest
     ) -> None:
         """Handle incoming tunnel request."""
+        # we should only ACK if the request matches the expected sequence number or one less
+        # we should not ACK and discard the request if the sequence number is higher than the expected sequence number
+        #   or if the sequence number lower thatn (expected -1)
         self._send_tunnelling_ack(
             tunneling_request.communication_channel_id,
             tunneling_request.sequence_counter,
         )
-        # Don't handle invalid cemi frames (None) and only handle incoming L_DATA_IND frames.
-        # Ignore L_DATA_CON confirmation frames. L_DATA_REQ frames should only be outgoing.
-        if (
-            tunneling_request.cemi is not None
-            and tunneling_request.cemi.code is CEMIMessageCode.L_DATA_IND
-        ):
+        if tunneling_request.cemi is None:
+            # Don't handle invalid cemi frames (None)
+            return
+        if tunneling_request.cemi.code is CEMIMessageCode.L_DATA_IND:
             telegram = tunneling_request.cemi.telegram
             telegram.direction = TelegramDirection.INCOMING
             if self.telegram_received_callback is not None:
                 self.telegram_received_callback(telegram)
+        elif tunneling_request.cemi.code is CEMIMessageCode.L_DATA_CON:
+            # L_DATA_CON confirmation frame signals ready to send next telegram
+            self._tunnelling_request_confirmation_event.set()
+        elif tunneling_request.cemi.code is CEMIMessageCode.L_DATA_REQ:
+            # L_DATA_REQ frames should only be outgoing.
+            logger.warning(
+                "Tunnel received unexpected L_DATA_REQ frame: %s", tunneling_request
+            )
 
     def _send_tunnelling_ack(
         self, communication_channel_id: int, sequence_counter: int


Test patch:
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index d958169b..bce28aeb 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -1,14 +1,20 @@
 """Unit test for KNX/IP Tunnelling Request/Response."""
+import asyncio
 from unittest.mock import Mock, patch
 
+import pytest
 from xknx import XKNX
+from xknx.dpt import DPTArray
 from xknx.io import Tunnel
-from xknx.knxip import CEMIFrame
-from xknx.telegram import TelegramDirection
+from xknx.knxip import CEMIFrame, KNXIPFrame, TunnellingAck, TunnellingRequest
+from xknx.knxip.knxip_enum import CEMIMessageCode
+from xknx.telegram import Telegram, TelegramDirection
+from xknx.telegram.apci import GroupValueWrite
 
 
-class TestTunnelling:
-    """Test class for xknx/io/Tunnelling objects."""
+@pytest.mark.asyncio
+class TestTunnel:
+    """Test class for xknx/io/Tunnel objects."""
 
     def setup_method(self):
         """Set up test class."""
@@ -65,3 +71,35 @@ class TestTunnelling:
         self.tunnel.udp_client.data_received_callback(raw)
         self.tg_received_mock.assert_not_called()
         send_ack_mock.assert_called_once_with(0x02, 0x4F)
+
+    async def test_tunnel_wait_for_l2_confirmation(self, time_travel):
+        """Test tunnel waits for L_DATA.con before sending another L_DATA.req."""
+        self.tunnel.udp_client.send = Mock()
+        self.tunnel.communication_channel = 1
+
+        test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        test_ack = KNXIPFrame.init_from_body(
+            TunnellingAck(self.xknx, sequence_counter=23)
+        )
+        confirmation = KNXIPFrame.init_from_body(
+            TunnellingRequest(
+                self.xknx,
+                communication_channel_id=1,
+                sequence_counter=23,
+                cemi=CEMIFrame.init_from_telegram(
+                    self.xknx, test_telegram, code=CEMIMessageCode.L_DATA_CON
+                ),
+            )
+        )
+        task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
+        await time_travel(0)
+        self.tunnel.udp_client.handle_knxipframe(test_ack)
+        await time_travel(0)
+        assert not task.done()
+        assert self.tunnel.udp_client.send.call_count == 1
+        self.tunnel.udp_client.handle_knxipframe(confirmation)
+        await time_travel(0)
+        assert task.done()
+        # one call for the outgoing request and one for the ACK for the confirmation
+        assert self.tunnel.udp_client.send.call_count == 2
+        await task