Change in vdsm[ovirt-3.6]: jsonrpc: fd leak

[Piotr Kliczewski]

Piotr Kliczewski has uploaded a new change for review.

Change subject: jsonrpc: fd leak
......................................................................

jsonrpc: fd leak

We need to make sure that there is no reference to objects which holds
a socket because m2c closes socket only when its object is garbage
collected.

This patch fixes two places where we had direct or indirect references
to socket object. In _StompConnection we kept a reference to it but now
we use dispatcher.socket. We passed connection object to JsonRpcServer
in order to get get local address which slowed down collection of socket
object (different thread) which could lead to reaching fd limit allowed
for the process. We removed this issue by passing local address to the
server instead.


Bug-Url: https://bugzilla.redhat.com/1256446
Change-Id: Ib15cca4659553b320babf928b603a0a75013ba99
Signed-off-by: pkliczewski <piotr.kliczewski(a)gmail.com&gt;
---
M lib/yajsonrpc/__init__.py
M lib/yajsonrpc/stompreactor.py
2 files changed, 14 insertions(+), 15 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/14/45714/1

diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py
index 76d97b8..03c49ed 100644
--- a/lib/yajsonrpc/__init__.py
+++ b/lib/yajsonrpc/__init__.py
@@ -245,10 +245,10 @@
 
 
 class _JsonRpcServeRequestContext(object):
-    def __init__(self, client, conn):
+    def __init__(self, client, addr):
         self._requests = []
         self._client = client
-        self._conn = conn
+        self._addr = addr
         self._counter = 0
         self._requests = {}
         self._responses = []
@@ -266,8 +266,8 @@
         return self._counter
 
     @property
-    def connection(self):
-        return self._conn
+    def address(self):
+        return self._addr
 
     def sendReply(self):
         if len(self._requests) > 0:
@@ -514,8 +514,7 @@
 
         try:
             params = req.params
-            server_address = ctx.connection.get_local_address()
-            self._bridge.register_server_address(server_address)
+            self._bridge.register_server_address(ctx.address)
             if isinstance(req.params, list):
                 res = method(*params)
             else:
@@ -541,11 +540,11 @@
             if obj is None:
                 break
 
-            client, conn, msg = obj
-            self._parseMessage(client, conn, msg)
+            client, addr, msg = obj
+            self._parseMessage(client, addr, msg)
 
-    def _parseMessage(self, client, conn, msg):
-        ctx = _JsonRpcServeRequestContext(client, conn)
+    def _parseMessage(self, client, addr, msg):
+        ctx = _JsonRpcServeRequestContext(client, addr)
 
         try:
             rawRequests = json.loads(msg)
diff --git a/lib/yajsonrpc/stompreactor.py b/lib/yajsonrpc/stompreactor.py
index 92d1dca..87cab3f 100644
--- a/lib/yajsonrpc/stompreactor.py
+++ b/lib/yajsonrpc/stompreactor.py
@@ -243,14 +243,13 @@
 class _StompConnection(object):
 
     def __init__(self, server, aclient, sock, reactor):
-        self._socket = sock
         self._reactor = reactor
         self._server = server
         self._messageHandler = None
 
         self._async_client = aclient
-        adisp = self._adisp = stomp.AsyncDispatcher(self, aclient)
-        self._dispatcher = Dispatcher(adisp, sock=sock, map=reactor._map)
+        self._dispatcher = reactor.create_dispatcher(
+            sock, stomp.AsyncDispatcher(self, aclient))
 
     def send_raw(self, msg):
         self._async_client.queue_frame(msg)
@@ -268,7 +267,7 @@
             self._async_client.remove_subscriptions()
 
     def get_local_address(self):
-        return self._socket.getsockname()[0]
+        return self._dispatcher.socket.getsockname()[0]
 
     def set_message_handler(self, msgHandler):
         self._messageHandler = msgHandler
@@ -276,7 +275,8 @@
 
     def handleMessage(self, data):
         if self._messageHandler is not None:
-            self._messageHandler((self._server, self, data))
+            self._messageHandler((self._server, self.get_local_address(),
+                                  data))
 
     def is_closed(self):
         return not self._dispatcher.connected


-- 
To view, visit https://gerrit.ovirt.org/45714
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ib15cca4659553b320babf928b603a0a75013ba99
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: ovirt-3.6
Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski(a)gmail.com&gt;