From afbc80352cd2ce69291939bacdf5bd5ea6d39eb5 Mon Sep 17 00:00:00 2001 From: IWASE Yusuke Date: Tue, 6 Mar 2018 15:52:21 +0900 Subject: [PATCH] controller: Wait for switch to disconnect connection If a sender closes a socket immediately after sending some data, a receiver can fail to receive full data from the sender. This patch fixes to shutdown a socket with "SHUT_WR" (disallows further sends only) and enables "Datapath" to wait for a switch to disconnect the connection. Signed-off-by: IWASE Yusuke Signed-off-by: FUJITA Tomonori --- ryu/controller/controller.py | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/ryu/controller/controller.py b/ryu/controller/controller.py index f46cf009..62135339 100644 --- a/ryu/controller/controller.py +++ b/ryu/controller/controller.py @@ -27,7 +27,7 @@ import logging import random from socket import IPPROTO_TCP from socket import TCP_NODELAY -from socket import SHUT_RDWR +from socket import SHUT_WR from socket import timeout as SocketTimeout import ssl @@ -196,12 +196,10 @@ def _deactivate(method): method(self) finally: try: - self.socket.shutdown(SHUT_RDWR) - except (EOFError, IOError): + self.socket.close() + except IOError: pass - if not self.is_active: - self.socket.close() return deactivate @@ -279,14 +277,24 @@ class Datapath(ofproto_protocol.ProtocolDesc): self._ports = None self.flow_format = ofproto_v1_0.NXFF_OPENFLOW10 self.ofp_brick = ryu.base.app_manager.lookup_service_brick('ofp_event') + self.state = None # for pylint self.set_state(HANDSHAKE_DISPATCHER) - @_deactivate + def _close_write(self): + # Note: Close only further sends in order to wait for the switch to + # disconnect this connection. + try: + self.socket.shutdown(SHUT_WR) + except (EOFError, IOError): + pass + def close(self): - if self.state != DEAD_DISPATCHER: - self.set_state(DEAD_DISPATCHER) + self.set_state(DEAD_DISPATCHER) + self._close_write() def set_state(self, state): + if self.state == state: + return self.state = state ev = ofp_event.EventOFPStateChange(self) ev.state = state @@ -386,8 +394,8 @@ class Datapath(ofproto_protocol.ProtocolDesc): self._send_q_sem.release() except hub.QueueEmpty: pass - # Finally, ensure the _recv_loop terminates. - self.close() + # Finally, disallow further sends. + self._close_write() def send(self, buf, close_socket=False): msg_enqueued = False