[#532,#564,#569] fix stored connections to match desired connection timeout.

Note also: a connection_timeout of value None is now allowed, and
effectively disables the timeout.

Author: d-w-moore

README.md

@@ -184,8 +184,12 @@ session object:
 >>> session.connection_timeout = 300
 ```
 
-This will set the timeout to five minutes for any associated
-connections.
+This will set the timeout to five minutes for any associated connections.
+
+Timeouts are either a positive `int` or `float` with units of seconds, or `None`, all in accordance with their
+meaning in the context of the socket module.  A value of `None` indicates timeouts are effectively
+infinite in value, i.e. turned off.  Setting a session's `connection_timeout` value to 0 is disallowed
+because this would cause the socket to enter non-blocking mode.
 
 Session objects and cleanup
 ---------------------------

irods/pool.py

@@ -22,6 +22,13 @@ def method_(self,*s,**kw):
 
 DEFAULT_APPLICATION_NAME = 'python-irodsclient'
 
+def _adjust_timeout_to_pool_default(conn):
+    set_timeout = conn.socket.gettimeout()
+    desired_value = conn.pool.connection_timeout
+    if desired_value == set_timeout:
+        return
+    conn.socket.settimeout(desired_value)
+
 class Pool(object):
 
     def __init__(self, account, application_name='', connection_refresh_time=-1, session = None):
@@ -57,6 +64,7 @@ def _conn(self, conn_): setattr( self._thread_local, "_conn", conn_)
 
     @attribute_from_return_value("_conn")
     def get_connection(self):
+        new_conn = False
         with self._lock:
             try:
                 conn = self.idle.pop()
@@ -73,9 +81,11 @@ def get_connection(self):
                     # code more predictable as we are not relying on when garbage collector is called
                     conn.disconnect()
                     conn = Connection(self, self.account)
+                    new_conn = True
                     logger.debug("Created new connection with id: {}".format(id(conn)))
             except KeyError:
                 conn = Connection(self, self.account)
+                new_conn = True
                 logger.debug("No connection found in idle set. Created a new connection with id: {}".format(id(conn)))
 
             self.active.add(conn)
@@ -87,6 +97,11 @@ def get_connection(self):
 
             logger.debug("Adding connection with id {} to active set".format(id(conn)))
 
+            # If the connection we're about to make active was cached, it already has a socket object internal to it,
+            # so we potentially have to modify it to have the desired timeout.
+            if not new_conn:
+                _adjust_timeout_to_pool_default(conn)
+
         logger.debug('num active: {}'.format(len(self.active)))
         logger.debug('num idle: {}'.format(len(self.idle)))
 

irods/session.py

@@ -5,6 +5,7 @@
 import errno
 import json
 import logging
+from numbers import Number
 import os
 import threading
 import weakref
@@ -349,9 +350,20 @@ def connection_timeout(self):
 
     @connection_timeout.setter
     def connection_timeout(self, seconds):
+        if seconds == 0:
+            exc = ValueError("Setting an iRODS connection_timeout to 0 seconds would make it non-blocking.")
+            raise exc
+        elif isinstance(seconds, Number):
+            if seconds < 0 or str(seconds) == 'nan' or str(abs(seconds)) == 'inf':
+                exc = ValueError("The iRODS connection_timeout may not be assigned a negative or otherwise rogue value (eg: NaN, Inf).")
+                raise exc
+        elif seconds is None:
+            pass
+        else:
+            exc = ValueError("The iRODS connection_timeout must be assigned a positive int, positive float, or None.")
+            raise exc
         self._cached_connection_timeout = seconds
-        if seconds is not None:
-            self.pool.connection_timeout = seconds
+        self.pool.connection_timeout = seconds
 
     @staticmethod
     def get_irods_password_file():

irods/test/connection_test.py

@@ -6,7 +6,7 @@
 import unittest
 from irods.exception import NetworkException
 import irods.test.helpers as helpers
-
+from irods.test.helpers import (server_side_sleep, temporarily_assign_attribute as temp_setter)
 
 class TestConnections(unittest.TestCase):
 
@@ -105,22 +105,79 @@ def test_that_connection_timeout_works__issue_377(self):
             if local_file:
                 os.unlink(local_file.name)
 
-    def assert_timeout_value_propagated_to_socket(self, session, timeout_value):
+    def _assert_timeout_value_is_propagated_to_all_sockets__issue_569(self, session, expected_timeout_value = 'POOL_TIMEOUT_SETTING'):
+        pool = session.pool
+        new_conn = None
+        if expected_timeout_value == 'POOL_TIMEOUT_SETTING':
+            expected_timeout_value = pool.connection_timeout
+        connections = set()
+        # make sure idle pool is not empty
         session.collections.get(helpers.home_collection(session))
-        conn = next(iter(session.pool.idle))
-        self.assertEqual(conn.socket.gettimeout(), timeout_value)
+        # On any connections thus far created, check that their internal socket objects are set to the expected timeout value.
+        try:
+            # Peel connections off the idle pool and check each for the expected timeout value, but don't release them to that pool yet.
+            while (pool.idle):
+                # Peel a connection (guaranteed newly-allocated for purposes of this test) and check for the proper timeout.
+                conn = pool.get_connection()
+                connections |= {conn}
+                self.assertEqual(conn.socket.gettimeout(), expected_timeout_value)
+
+            # Get an additional connection while idle pool is empty; this way, we know it to be newly-allocated.
+            new_conn = pool.get_connection()
+
+            # Check the expected timeout applies to the newly-allocated connection
+            self.assertEqual(new_conn.socket.gettimeout(), expected_timeout_value)
+
+        finally:
+            # Release and destroy the connection that was newly-allocated for this test.
+            if new_conn:
+                new_conn.release(destroy = True)
+            # Release connections that had been cached, by the same normal mechanism the API endpoints indirectly employ.
+            for conn in connections:
+                pool.release_connection(conn)
 
     def test_connection_timeout_parameter_in_session_init__issue_377(self):
         timeout = 1.0
         sess = helpers.make_session(connection_timeout = timeout)
-        self.assert_timeout_value_propagated_to_socket(sess, timeout)
+        self._assert_timeout_value_is_propagated_to_all_sockets__issue_569(sess, timeout)
 
     def test_assigning_session_connection_timeout__issue_377(self):
-        sess = self.sess
+        sess = helpers.make_session()
         for timeout in (999999, None):
             sess.connection_timeout = timeout
-            sess.cleanup()
-            self.assert_timeout_value_propagated_to_socket(sess, timeout)
+            self._assert_timeout_value_is_propagated_to_all_sockets__issue_569(sess, timeout)
+
+    def test_assigning_session_connection_timeout_to_invalid_values__issue_569(self):
+        sess = helpers.make_session()
+        DESIRED_TIMEOUT = 64.25
+        sess.connection_timeout = DESIRED_TIMEOUT
+        # Test our desired connection pool default timeout has taken hold.
+        self.assertEqual(sess.connection_timeout, DESIRED_TIMEOUT)
+
+        # Test that bad timeout values are met with an exception.
+        for value in (float('NaN'), float('Inf'), -float('Inf'), -1, 0, 0.0, "banana"):
+            with self.assertRaises(ValueError):
+                sess.connection_timeout = value
+
+        # Test previously set value is unaffected
+        self.assertEqual(sess.connection_timeout, DESIRED_TIMEOUT)
+
+    def test_assigning_session_connection_timeout__issue_569(self):
+        sess = helpers.make_session()
+        old_timeout = sess.connection_timeout
+
+        with temp_setter(sess, 'connection_timeout',1.0):
+            # verify we can reproduce a NetworkException from a server timeout
+            with self.assertRaises(NetworkException):
+               server_side_sleep(sess,2.5)
+            # temporarily suspend timeouts on a session
+            with temp_setter(sess, 'connection_timeout',None):
+               server_side_sleep(sess,2.5)
+            # temporarily increase (from 1.0 to 4) the timeout on a session
+            with temp_setter(sess, 'connection_timeout',4):
+               server_side_sleep(sess,2.5)
+        self.assertEqual(old_timeout, sess.connection_timeout)
+        self._assert_timeout_value_is_propagated_to_all_sockets__issue_569(sess, old_timeout)
 
 if __name__ == '__main__':
     # let the tests find the parent irods lib

irods/test/helpers.py

@@ -17,6 +17,7 @@
 import json
 import sys
 import irods.client_configuration as config
+import irods.rule
 from irods.session import iRODSSession
 from irods.message import (iRODSMessage, IRODS_VERSION)
 from irods.password_obfuscation import encode
@@ -377,3 +378,26 @@ def enableLogging(logger, handlerType, args, level_ = logging.INFO):
             logger.removeHandler(h)
 
 
+class _unlikely_value: pass
+
+@contextlib.contextmanager
+def temporarily_assign_attribute(target, attr, value, not_set_indicator = _unlikely_value()):
+    save = not_set_indicator
+    try:
+        save = getattr(target, attr, not_set_indicator)
+        setattr(target, attr, value)
+        yield
+    finally:
+        if save != not_set_indicator:
+            setattr(target, attr, save)
+        else:
+            delattr(target, attr)
+
+# Implement a server-side wait that ensures no TCP communication from server end for a given interval.
+# Useful to test the effect of socket inactivity on a client.  See python-irodsclient issue #569
+def server_side_sleep(session, seconds):
+    # Round floating-point seconds to nearest integer + microseconds figure, required by msiSleep.
+    int_, frac_ = [int(_) for _ in divmod(seconds * 1.0e6 + 0.5, 1.0e6)]
+    rule_code = "msiSleep('{}','{}')".format(int_,frac_)
+    # Call the msiSleep microservice.
+    irods.rule.Rule(session, body = rule_code, instance_name = 'irods_rule_engine_plugin-irods_rule_language-instance').execute()