[#345][#352] Fix connection destructor issue and fix the infinite loop issue when sock.recv() returns an empty buffer

Author: Kaivan Kamali

irods/connection.py

@@ -21,7 +21,9 @@
 
 
 ALLOW_PAM_LONG_TOKENS = True      # True to fix [#279]
-
+# Message to be logged when the connection
+# destructor is called. Used in a unit test
+DESTRUCTOR_MSG = "connection __del__() called"
 
 from irods import (
     MAX_PASSWORD_LENGTH, RESPONSE_LEN,
@@ -80,8 +82,8 @@ def client_signature(self):
         return self._client_signature
 
     def __del__(self):
-        if self.socket and getattr(self,"_disconnected",False):
-            self.disconnect()
+        self.disconnect()
+        logger.debug(DESTRUCTOR_MSG)
 
     def send(self, message):
         string = message.pack()
@@ -107,7 +109,12 @@ def recv(self, into_buffer = None
                 msg = iRODSMessage.recv(self.socket)
             else:
                 msg = iRODSMessage.recv_into(self.socket, into_buffer)
-        except socket.error:
+        except (socket.error, socket.timeout) as e:
+            # If _recv_message_in_len() fails in recv() or recv_into(),
+            # it will throw a socket.error exception. The exception is
+            # caught here, a critical message is logged, and is wrapped
+            # in a NetworkException with a more user friendly message
+            logger.critical(e)
             logger.error("Could not receive server response")
             self.release(True)
             raise NetworkException("Could not receive server response")
@@ -273,17 +280,24 @@ def _connect(self):
         return version_msg.get_main_message(VersionResponse)
 
     def disconnect(self):
-        disconnect_msg = iRODSMessage(msg_type='RODS_DISCONNECT')
-        self.send(disconnect_msg)
-        try:
-            # SSL shutdown handshake
-            self.socket = self.socket.unwrap()
-        except AttributeError:
-            pass
-        self.socket.shutdown(socket.SHUT_RDWR)
-        self.socket.close()
-        self.socket = None
-        self._disconnected = True
+        # Moved the conditions to call disconnect() inside the function.
+        # Added a new criteria for calling disconnect(); Only call
+        # disconnect() if fileno is not -1 (fileno -1 indicates the socket
+        # is already closed). This makes it safe to call disconnect multiple
+        # times on the same connection. The first call cleans up the resources
+        # and next calls are no-ops
+        if self.socket and getattr(self, "_disconnected", False) == False and self.socket.fileno() != -1:
+            disconnect_msg = iRODSMessage(msg_type='RODS_DISCONNECT')
+            self.send(disconnect_msg)
+            try:
+                # SSL shutdown handshake
+                self.socket = self.socket.unwrap()
+            except AttributeError:
+                pass
+            self.socket.shutdown(socket.SHUT_RDWR)
+            self.socket.close()
+            self.socket = None
+            self._disconnected = True
 
     def recvall(self, n):
         # Helper function to recv n bytes or return None if EOF is hit

irods/message/__init__.py

@@ -153,6 +153,17 @@ def ET(xml_type = 0, server_version = None):
 def _recv_message_in_len(sock, size):
     size_left = size
     retbuf = None
+
+    # Get socket properties for debug and exception messages.
+    host, port = sock.getpeername()
+    is_blocking = sock.getblocking()
+    timeout = sock.gettimeout()
+
+    logger.debug(f'host: {host}')
+    logger.debug(f'port: {port}')
+    logger.debug(f'is_blocking: {is_blocking}')
+    logger.debug(f'timeout: {timeout}')
+
     while size_left > 0:
         try:
             buf = sock.recv(size_left, socket.MSG_WAITALL)
@@ -163,11 +174,27 @@ def _recv_message_in_len(sock, size):
             if getattr(e, 'winerror', 0) != 10045:
                 raise
             buf = sock.recv(size_left)
+
+        # This prevents an infinite loop. If the call to recv()
+        # returns an empty buffer, break out of the loop.
+        if len(buf) == 0:
+            break
         size_left -= len(buf)
         if retbuf is None:
             retbuf = buf
         else:
             retbuf += buf
+
+    # This method is supposed to read and return 'size'
+    # bytes from the socket. If it reads no bytes (retbuf
+    # will be None), or if it reads less number of bytes
+    # than 'size', throw a socket.error exception
+    if retbuf is None or len(retbuf) != size:
+        retbuf_size = len(retbuf) if retbuf is not None else 0
+        msg = 'Read {} bytes from the socket (host {}, port {}) instead of expected {} bytes'.format(
+               retbuf_size, host, port, size)
+        raise socket.error(msg)
+
     return retbuf
 
 

irods/pool.py

@@ -64,7 +64,11 @@ def get_connection(self):
                 # release the connection (as its stale) and create a new one
                 if self.refresh_connection and (curr_time - conn.create_time).total_seconds() > self.connection_refresh_time:
                     logger.debug('Connection with id {} was created more than {} seconds ago. Releasing the connection and creating a new one.'.format(id(conn), self.connection_refresh_time))
-                    self.release_connection(conn, True)
+                    # Since calling disconnect() repeatedly is safe, we call disconnect()
+                    # here explicitly, instead of relying on the garbage collector to clean
+                    # up the object and call disconnect(). This makes the behavior of the
+                    # code more predictable as we are not relying on when garbage collector is called
+                    conn.disconnect()
                     conn = Connection(self, self.account)
                     logger.debug("Created new connection with id: {}".format(id(conn)))
             except KeyError:

irods/test/connection_test.py

@@ -24,6 +24,9 @@ def test_connection(self):
     def test_connection_destructor(self):
         conn = self.sess.pool.get_connection()
         conn.__del__()
+        # These asserts confirm that disconnect() in connection destructor is called
+        self.assertIsNone(conn.socket)
+        self.assertTrue(conn._disconnected)
         conn.release(destroy=True)
 
     def test_failed_connection(self):

irods/test/pool_test.py

@@ -1,13 +1,17 @@
 #! /usr/bin/env python
 from __future__ import absolute_import
 import datetime
+import gc
+import logging
 import os
 import re
 import sys
+import tempfile
 import time
 import json
 import unittest
 import irods.test.helpers as helpers
+from irods.connection import DESTRUCTOR_MSG
 
 #  Regular expression to match common synonyms for localhost.
 #
@@ -237,6 +241,76 @@ def test_no_refresh_connection(self):
             self.assertEqual(0, len(self.sess.pool.active))
             self.assertEqual(0, len(self.sess.pool.idle))
 
+    # Test to confirm the connection destructor log message is actually
+    # logged to file, to confirm the destructor is called
+    def test_connection_destructor_called(self):
+        # Set 'irods_connection_refresh_time' to '3' (in seconds) in
+        # ~/.irods/irods_environment.json file. This means any connection
+        # that was created more than 3 seconds ago will be dropped and
+        # a new connection is created/returned. This is to avoid
+        # issue with idle connections that are dropped.
+        conn_obj_id_1 = None
+        conn_obj_id_2 = None
+        create_time_1 = None
+        create_time_2 = None
+        last_used_time_1 = None
+        last_used_time_2 = None
+
+        try:
+
+            # Create a temporary log file
+            my_log_file = tempfile.NamedTemporaryFile()
+
+            logging.getLogger('irods.connection').setLevel(logging.DEBUG)
+            file_handler = logging.FileHandler(my_log_file.name, mode='a')
+            file_handler.setLevel(logging.DEBUG)
+            logging.getLogger('irods.connection').addHandler(file_handler)
+
+            with self.sess.pool.get_connection() as conn:
+                conn_obj_id_1 = id(conn)
+                curr_time = datetime.datetime.now()
+                create_time_1 = conn.create_time
+                last_used_time_1 = conn.last_used_time
+                self.assertTrue(curr_time >= create_time_1)
+                self.assertTrue(curr_time >= last_used_time_1)
+                self.assertEqual(1, len(self.sess.pool.active))
+                self.assertEqual(0, len(self.sess.pool.idle))
+
+                self.sess.pool.release_connection(conn)
+                self.assertEqual(0, len(self.sess.pool.active))
+                self.assertEqual(1, len(self.sess.pool.idle))
+
+            # Wait more than 'irods_connection_refresh_time' seconds,
+            # which is set to 3. Connection object should have a different
+            # object ID (as a new connection is created)
+            time.sleep(5)
+
+            # Call garbage collector, so the unreferenced conn object is garbage collected
+            gc.collect()
+
+            with self.sess.pool.get_connection() as conn:
+                conn_obj_id_2 = id(conn)
+                curr_time = datetime.datetime.now()
+                create_time_2 = conn.create_time
+                last_used_time_2 = conn.last_used_time
+                self.assertTrue(curr_time >= create_time_2)
+                self.assertTrue(curr_time >= last_used_time_2)
+                self.assertNotEqual(conn_obj_id_1, conn_obj_id_2)
+                self.assertTrue(create_time_2 > create_time_1)
+                self.assertEqual(1, len(self.sess.pool.active))
+                self.assertEqual(0, len(self.sess.pool.idle))
+
+                self.sess.pool.release_connection(conn, True)
+                self.assertEqual(0, len(self.sess.pool.active))
+                self.assertEqual(0, len(self.sess.pool.idle))
+
+            # Assert that connection destructor called
+            with open(my_log_file.name, 'r') as fh:
+                lines = fh.read().splitlines()
+                self.assertTrue(DESTRUCTOR_MSG in lines)
+        finally:
+            # Remove irods.connection's file_handler that was added just for this test
+            logging.getLogger('irods.connection').removeHandler(file_handler)
 
     def test_get_connection_refresh_time_no_env_file_input_param(self):
         connection_refresh_time = self.sess.get_connection_refresh_time(first_name="Magic", last_name="Johnson")