mirrors/ryu
1
0
Fork 0
mirror of https://github.com/faucetsdn/ryu.git synced 2026-05-08 22:06:10 +02:00
Code Issues Projects Releases Wiki Activity

BGPSpeaker/net_ctrl: Close activity for each session

Browse Source 
Currently, a thread for processing outgoing message will be remained in
"RpcSession" activity even if RPC session is closed by the remote peer,
and garbages on memory will grow.

This patch fixes to close "RpcSession" activity when RPC session
closing.

Signed-off-by: IWASE Yusuke <iwase.yusuke0@gmail.com>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>
This commit is contained in:
IWASE Yusuke 2017-08-22 11:38:55 +09:00 committed by FUJITA Tomonori
parent 98a838e487
commit 80fedbfeaf
1 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
ryu/services/protocols/bgp/net_ctrl.py
View File

@ -107,6 +107,8 @@ class RpcSession(Activity):
self._socket = sock
self._outgoing_msg_sink_iter = outgoing_msg_sink_iter
self.is_connected = True
self.green_in = None
self.green_out = None
def stop(self):
super(RpcSession, self).stop()
@ -115,15 +117,15 @@ class RpcSession(Activity):
def _run(self):
# Process outgoing messages in new thread.
green_out = self._spawn('net_ctrl._process_outgoing',
self._process_outgoing_msg,
self._outgoing_msg_sink_iter)
self.green_out = self._spawn('net_ctrl._process_outgoing',
self._process_outgoing_msg,
self._outgoing_msg_sink_iter)
# Process incoming messages in new thread.
green_in = self._spawn('net_ctrl._process_incoming',
self._process_incoming_msgs)
self.green_in = self._spawn('net_ctrl._process_incoming',
self._process_incoming_msgs)
LOG.info('RPC Session to %s started', self.peer_name)
green_in.wait()
green_out.wait()
self.green_in.wait()
self.green_out.wait()
def _next_msg_id(self):
this_id = self._next_msgid
@ -202,6 +204,10 @@ class RpcSession(Activity):
LOG.error('Invalid message type: %r', msg)
self.pause(0)
# Stop outgoing connection.
if self.green_out:
self.green_out.kill()
def _process_outgoing_msg(self, sink_iter):
"""For every message we construct a corresponding RPC message to be
sent over the given socket inside given RPC session.
@ -231,6 +237,10 @@ class RpcSession(Activity):
self._sendall(rpc_msg)
self.pause(0)
# Stop incoming connection.
if self.green_in:
self.green_in.kill()
def _recv(self):
return self._sock_wrap(self._socket.recv)(RPC_SOCK_BUFF_SIZE)