NBL-31: Python thin now does not fails if one node is not avail

This closes #255

Author: isapego

pygridgain/cache.py

@@ -313,7 +313,9 @@ def get_best_node(
                     in self.affinity['node_mapping'].items()
                     if part in p
                 )
-                best_conn = conn.client._nodes[node_uuid]
+                best_conn = next(
+                    n for n in conn.client._nodes if n.uuid == node_uuid
+                )
                 if best_conn.alive:
                     conn = best_conn
             except (StopIteration, KeyError):

pygridgain/client.py

@@ -42,7 +42,7 @@
 
 from collections import defaultdict, OrderedDict
 import random
-from typing import Dict, Iterable, Optional, Tuple, Type, Union
+from typing import Dict, Iterable, List, Optional, Tuple, Type, Union
 
 from .api.binary import get_binary_type, put_binary_type
 from .api.cache_config import cache_get_names
@@ -80,10 +80,10 @@ class Client:
     _compact_footer: bool = None
     _connection_args: Dict = None
     _current_node: int = None
-    _nodes: Union[Iterable[Connection], Dict['UUID', Connection]] = None
+    _nodes: List[Connection] = None
 
-    affinity_version: Tuple = None
-    protocol_version = None
+    affinity_version: Optional[Tuple] = None
+    protocol_version: Optional[Tuple] = None
 
     def __init__(
         self, compact_footer: bool = None, affinity_aware: bool = False,
@@ -106,6 +106,7 @@ def __init__(
         """
         self._compact_footer = compact_footer
         self._connection_args = kwargs
+        self._nodes = []
         self._current_node = 0
         self._affinity_aware = affinity_aware
         self.affinity_version = (0, 0)
@@ -126,43 +127,6 @@ def get_protocol_version(self) -> Optional[Tuple]:
     def affinity_aware(self):
         return self._affinity_aware
 
-    @select_version
-    def _add_node(
-        self, host: str = None, port: int = None,
-        conn: Connection = None, node_uuid: 'UUID' = None,
-    ) -> 'UUID':
-        """
-        Opens a connection to GridGain server and adds it to the nodes'
-        collection (connection pool).
-        """
-        if self._nodes is None:
-            self._nodes = {}
-
-        if conn is None:
-            conn = Connection(self, **self._connection_args)
-            hs_response = conn.connect(host, port)
-            node_uuid = hs_response['node_uuid']
-
-        if node_uuid:
-            self._nodes[node_uuid] = conn
-        return node_uuid
-
-    def _add_node_130(
-        self, host: str = None, port: int = None, conn: Connection = None,
-        *args, **kwargs,
-    ):
-        if self._nodes is None:
-            self._nodes = []
-
-        if conn is None:
-            conn = Connection(self, **self._connection_args)
-            conn.host = host
-            conn.port = port
-
-        self._nodes.append(conn)
-
-    _add_node_120 = _add_node_130
-
     def connect(self, *args):
         """
         Connect to GridGain cluster node(s).
@@ -185,36 +149,51 @@ def connect(self, *args):
         else:
             raise ConnectionError('Connection parameters are not valid.')
 
-        nodes = iter(nodes)
+        # the following code is quite twisted, because the protocol version
+        # is initially unknown
+
         # TODO: open first node in foregroung, others − in background
+        for i, node in enumerate(nodes):
+            host, port = node
+            conn = Connection(self, **self._connection_args)
+            conn.host = host
+            conn.port = port
 
-        first_node = Connection(self, **self._connection_args)
+            try:
+                if (
+                    self.protocol_version is None
+                    or self.protocol_version >= (1, 4, 0)
+                ):
+                    # open connection before adding to the pool
+                    conn.connect(host, port)
+
+                    # now we have the protocol version
+                    if self.protocol_version < (1, 4, 0):
+                        # do not try to open more nodes
+                        self._current_node = i
+                    else:
+                        # take a chance to schedule the reconnection
+                        # for all the failed connections, that was probed
+                        # before this
+                        for failed_node in self._nodes[:i]:
+                            failed_node.reconnect()
 
-        # now we know protocol version
-        self._add_node(
-            conn=first_node,
-            node_uuid=first_node.connect(*next(nodes)).get('node_uuid', None),
-        )
+            except connection_errors:
+                conn._fail()
+                if (
+                    self.protocol_version
+                    and self.protocol_version >= (1, 4, 0)
+                ):
+                    # schedule the reconnection
+                    conn.reconnect()
 
-        for host, port in nodes:
-            self._add_node(host, port)
+            self._nodes.append(conn)
 
-    @select_version
     def close(self):
-        """
-        Close all connections to the server and clean the connection pool.
-        """
-        for conn in self._nodes.values():
-            conn.close()
-        self._nodes.clear()
-
-    def close_130(self):
         for conn in self._nodes:
             conn.close()
         self._nodes.clear()
 
-    close_120 = close_130
-
     @property
     @select_version
     def random_node(self) -> Connection:
@@ -227,7 +206,7 @@ def random_node(self) -> Connection:
         """
         try:
             return random.choice(
-                list(n for n in self._nodes.values() if n.alive)
+                list(n for n in self._nodes if n.alive)
             )
         except IndexError:
             # cannot choose from an empty sequence

pygridgain/connection/__init__.py

@@ -21,7 +21,7 @@
 
 from collections import OrderedDict
 import socket
-from threading import Lock, Timer
+from threading import Lock
 from typing import Union
 
 from pygridgain.constants import *
@@ -30,7 +30,7 @@
 )
 from pygridgain.datatypes import Byte, Int, Short, String, UUIDObject
 from pygridgain.datatypes.internal import Struct
-from pygridgain.utils import select_version
+from pygridgain.utils import DaemonicTimer, select_version
 
 from .handshake import HandshakeRequest
 from .ssl import wrap
@@ -62,6 +62,7 @@ class Connection:
     username = None
     password = None
     ssl_params = {}
+    uuid = None
 
     @staticmethod
     def _check_ssl_params(params):
@@ -260,6 +261,8 @@ def connect(
                 raise e
 
         # connection is ready for end user
+        self.uuid = result.get('node_uuid', None)  # version-specific (1.4+)
+
         self._failed = False
         return result
 
@@ -334,7 +337,7 @@ def reconnect(self, seq_no=0):
         self._reconnect()
 
         if self.failed:
-            Timer(
+            DaemonicTimer(
                 RECONNECT_BACKOFF_SEQUENCE[seq_no],
                 self.reconnect,
                 kwargs={'seq_no': seq_no + 1},
@@ -477,7 +480,10 @@ def close(self, release=True):
         garbage-collected.
         """
         if release:
-            self._in_use.release()
+            try:
+                self._in_use.release()
+            except RuntimeError:
+                pass
 
         if self._socket:
             try:

pygridgain/utils.py

@@ -16,6 +16,7 @@
 import ctypes
 import decimal
 from functools import wraps
+from threading import Event, Thread
 from typing import Any, Callable, Optional, Type, Tuple, Union
 
 from pygridgain.datatypes.base import GridGainDataType
@@ -279,3 +280,27 @@ def get_field_by_id(
 def unsigned(value: int, c_type: ctypes._SimpleCData = ctypes.c_uint) -> int:
     """ Convert signed integer value to unsigned. """
     return c_type(value).value
+
+
+class DaemonicTimer(Thread):
+    """
+    Same as normal `threading.Timer`, but do not delay the program exit.
+    """
+
+    def __init__(self, interval, function, args=None, kwargs=None):
+        Thread.__init__(self, daemon=True)
+        self.interval = interval
+        self.function = function
+        self.args = args if args is not None else []
+        self.kwargs = kwargs if kwargs is not None else {}
+        self.finished = Event()
+
+    def cancel(self):
+        """Stop the timer if it hasn't finished yet."""
+        self.finished.set()
+
+    def run(self):
+        self.finished.wait(self.interval)
+        if not self.finished.is_set():
+            self.function(*self.args, **self.kwargs)
+        self.finished.set()

tests/test_affinity.py

@@ -15,7 +15,6 @@
 #
 from datetime import datetime, timedelta
 import decimal
-import time
 from uuid import UUID, uuid4
 
 import pytest
@@ -30,6 +29,13 @@
 
 def test_get_node_partitions(client):
 
+    if client.protocol_version < (1, 4, 0):
+        pytest.skip(
+            'Best effort affinity is not supported by the protocol {}.'.format(
+                client.protocol_version
+            )
+        )
+
     conn = client.random_node
 
     cache_1 = client.get_or_create_cache('test_cache_1')
@@ -46,8 +52,6 @@ def test_get_node_partitions(client):
     cache_4 = client.get_or_create_cache('test_cache_4')
     cache_5 = client.get_or_create_cache('test_cache_5')
 
-    time.sleep(0.1)
-
     result = cache_get_node_partitions(
         conn,
         [cache_1.cache_id, cache_2.cache_id]
@@ -120,7 +124,12 @@ def test_get_node_partitions(client):
 )
 def test_affinity(client, key, key_hint):
 
-    time.sleep(0.1)
+    if client.protocol_version < (1, 4, 0):
+        pytest.skip(
+            'Best effort affinity is not supported by the protocol {}.'.format(
+                client.protocol_version
+            )
+        )
 
     cache_1 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_1',
@@ -131,7 +140,7 @@ def test_affinity(client, key, key_hint):
 
     best_node = cache_1.get_best_node(key, key_hint=key_hint)
 
-    for node in client._nodes.values():
+    for node in filter(lambda n: n.alive, client._nodes):
         result = cache_local_peek(
             node, cache_1.cache_id, key, key_hint=key_hint,
         )
@@ -149,7 +158,12 @@ def test_affinity(client, key, key_hint):
 
 def test_affinity_for_generic_object(client):
 
-    time.sleep(0.1)
+    if client.protocol_version < (1, 4, 0):
+        pytest.skip(
+            'Best effort affinity is not supported by the protocol {}.'.format(
+                client.protocol_version
+            )
+        )
 
     cache_1 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_1',
@@ -173,7 +187,7 @@ class KeyClass(
 
     best_node = cache_1.get_best_node(key, key_hint=BinaryObject)
 
-    for node in client._nodes.values():
+    for node in filter(lambda n: n.alive, client._nodes):
         result = cache_local_peek(
             node, cache_1.cache_id, key, key_hint=BinaryObject,
         )
@@ -191,7 +205,12 @@ class KeyClass(
 
 def test_affinity_for_generic_object_without_type_hints(client):
 
-    time.sleep(0.1)
+    if client.protocol_version < (1, 4, 0):
+        pytest.skip(
+            'Best effort affinity is not supported by the protocol {}.'.format(
+                client.protocol_version
+            )
+        )
 
     cache_1 = client.get_or_create_cache({
         PROP_NAME: 'test_cache_1',
@@ -215,7 +234,7 @@ class KeyClass(
 
     best_node = cache_1.get_best_node(key)
 
-    for node in client._nodes.values():
+    for node in filter(lambda n: n.alive, client._nodes):
         result = cache_local_peek(
             node, cache_1.cache_id, key
         )