NBL-5: Affinity Awareness for Python Thin client

This closes #180

Author: dmelnichuk

docs/examples.rst

@@ -445,27 +445,24 @@ When connection to the server is broken or timed out,
 (`OSError` or `SocketError`), but keeps its constructor's parameters intact
 and tries to reconnect transparently.
 
-When there's no way for :class:`~pyignite.client.Client` to reconnect, it
-raises a special :class:`~pyignite.exceptions.ReconnectError` exception.
+When :class:`~pyignite.client.Client` detects that all nodes in the list are
+failed without the possibility of restoring connection, it raises a special
+:class:`~pyignite.exceptions.ReconnectError` exception.
 
-The following example features a simple node list traversal failover mechanism.
 Gather 3 Ignite nodes on `localhost` into one cluster and run:
 
 .. literalinclude:: ../examples/failover.py
   :language: python
-  :lines: 16-49
+  :lines: 16-51
 
 Then try shutting down and restarting nodes, and see what happens.
 
 .. literalinclude:: ../examples/failover.py
   :language: python
-  :lines: 51-61
+  :lines: 53-65
 
 Client reconnection do not require an explicit user action, like calling
-a special method or resetting a parameter. Note, however, that reconnection
-is lazy: it happens only if (and when) it is needed. In this example,
-the automatic reconnection happens, when the script checks upon the last
-saved value:
+a special method or resetting a parameter.
 
 .. literalinclude:: ../examples/failover.py
   :language: python
@@ -475,29 +472,6 @@ It means that instead of checking the connection status it is better for
 `pyignite` user to just try the supposed data operations and catch
 the resulting exception.
 
-:py:meth:`~pyignite.connection.Connection.connect` method accepts any
-iterable, not just list. It means that you can implement any reconnection
-policy (round-robin, nodes prioritization, pause on reconnect or graceful
-backoff) with a generator.
-
-`pyignite` comes with a sample
-:class:`~pyignite.connection.generators.RoundRobin` generator. In the above
-example try to replace
-
-.. literalinclude:: ../examples/failover.py
-  :language: python
-  :lines: 29
-
-with
-
-.. code-block:: python3
-
-    client.connect(RoundRobin(nodes, max_reconnects=20))
-
-The client will try to reconnect to node 1 after node 3 is crashed, then to
-node 2, et c. At least one node should be active for the
-:class:`~pyignite.connection.generators.RoundRobin` to work properly.
-
 SSL/TLS
 -------
 

examples/failover.py

@@ -27,35 +27,39 @@
 
 client = Client(timeout=4.0)
 client.connect(nodes)
-print('Connected to {}'.format(client))
+print('Connected')
 
 my_cache = client.get_or_create_cache({
     PROP_NAME: 'my_cache',
-    PROP_CACHE_MODE: CacheMode.REPLICATED,
+    PROP_CACHE_MODE: CacheMode.PARTITIONED,
+    PROP_BACKUPS_NUMBER: 2,
 })
 my_cache.put('test_key', 0)
+test_value = 0
 
 # abstract main loop
 while True:
     try:
         # do the work
-        test_value = my_cache.get('test_key')
+        test_value = my_cache.get('test_key') or 0
         my_cache.put('test_key', test_value + 1)
     except (OSError, SocketError) as e:
         # recover from error (repeat last command, check data
         # consistency or just continue − depends on the task)
         print('Error: {}'.format(e))
-        print('Last value: {}'.format(my_cache.get('test_key')))
-        print('Reconnected to {}'.format(client))
+        print('Last value: {}'.format(test_value))
+        print('Reconnecting')
 
-# Connected to 127.0.0.1:10800
-# Error: [Errno 104] Connection reset by peer
-# Last value: 6999
-# Reconnected to 127.0.0.1:10801
-# Error: Socket connection broken.
-# Last value: 12302
-# Reconnected to 127.0.0.1:10802
-# Error: [Errno 111] Client refused
+# Connected
+# Error: Connection broken.
+# Last value: 2650
+# Reconnecting
+# Error: Connection broken.
+# Last value: 10204
+# Reconnecting
+# Error: Connection broken.
+# Last value: 18932
+# Reconnecting
 # Traceback (most recent call last):
-#     ...
-# pyignite.exceptions.ReconnectError: Can not reconnect: out of nodes
+#   ...
+# pyignite.exceptions.ReconnectError: Can not reconnect: out of nodes.

pyignite/api/__init__.py

@@ -23,6 +23,9 @@
 stable end user API see :mod:`pyignite.client` module.
 """
 
+from .affinity import (
+    cache_get_node_partitions,
+)
 from .cache_config import (
     cache_create,
     cache_get_names,
@@ -54,6 +57,7 @@
     cache_remove_keys,
     cache_remove_all,
     cache_get_size,
+    cache_local_peek,
 )
 from .sql import (
     scan,

pyignite/api/affinity.py

@@ -0,0 +1,123 @@
+#
+# Copyright 2019 GridGain Systems, Inc. and Contributors.
+#
+# Licensed under the GridGain Community Edition License (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from typing import Iterable, Union
+
+from pyignite.datatypes import Bool, Int, Long, UUIDObject
+from pyignite.datatypes.internal import StructArray
+from pyignite.queries import Query
+from pyignite.queries.op_codes import OP_CACHE_PARTITIONS
+from pyignite.utils import is_iterable
+from .result import APIResult
+
+
+cache_ids = StructArray([
+    ('cache_id', Int),
+])
+
+cache_config = StructArray([
+    ('key_type_id', Int),
+    ('affinity_key_field_id', Int),
+])
+
+node_partitions = StructArray([
+    ('partition_id', Int),
+])
+
+node_mapping = StructArray([
+    ('node_uuid', UUIDObject),
+    ('node_partitions', node_partitions)
+])
+
+cache_mapping = StructArray([
+    ('cache_id', Int),
+    ('cache_config', cache_config),
+])
+
+partition_mapping = StructArray([
+    ('is_applicable', Bool),
+    ('cache_mapping', cache_mapping),
+    ('node_mapping', node_mapping),
+])
+
+
+def cache_get_node_partitions(
+    conn: 'Connection', caches: Union[int, Iterable[int]],
+    query_id: int = None,
+) -> APIResult:
+    """
+    Gets partition mapping for an Ignite cache or a number of caches. See
+    “IEP-23: Best Effort Affinity for thin clients”.
+
+    :param conn: connection to Ignite server,
+    :param caches: cache ID(s) the mapping is provided for,
+    :param query_id: (optional) a value generated by client and returned as-is
+     in response.query_id. When the parameter is omitted, a random value
+     is generated,
+    :return: API result data object.
+    """
+    query_struct = Query(
+        OP_CACHE_PARTITIONS,
+        [
+            ('cache_ids', cache_ids),
+        ],
+        query_id=query_id
+    )
+    if not is_iterable(caches):
+        caches = [caches]
+
+    result = query_struct.perform(
+        conn,
+        query_params={
+            'cache_ids': [{'cache_id': cache} for cache in caches],
+        },
+        response_config=[
+            ('version_major', Long),
+            ('version_minor', Int),
+            ('partition_mapping', partition_mapping),
+        ],
+    )
+    if result.status == 0:
+        # tidying up the result
+        value = {
+            'version': (
+                result.value['version_major'],
+                result.value['version_minor']
+            ),
+            'partition_mapping': [],
+        }
+        for i, partition_map in enumerate(result.value['partition_mapping']):
+            cache_id = partition_map['cache_mapping'][0]['cache_id']
+            value['partition_mapping'].insert(
+                i,
+                {
+                    'cache_id': cache_id,
+                    'is_applicable': partition_map['is_applicable'],
+                }
+            )
+            if partition_map['is_applicable']:
+                value['partition_mapping'][i]['cache_config'] = {
+                    a['key_type_id']: a['affinity_key_field_id']
+                    for a in partition_map['cache_mapping'][0]['cache_config']
+                }
+                value['partition_mapping'][i]['node_mapping'] = {
+                    p['node_uuid']: [
+                        x['partition_id'] for x in p['node_partitions']
+                    ]
+                    for p in partition_map['node_mapping']
+                }
+        result.value = value
+
+    return result

pyignite/api/binary.py

@@ -20,7 +20,7 @@
     body_struct, enum_struct, schema_struct, binary_fields_struct,
 )
 from pyignite.datatypes import String, Int, Bool
-from pyignite.queries import Query, Response
+from pyignite.queries import Query, get_response_class
 from pyignite.queries.op_codes import *
 from pyignite.utils import int_overflow, entity_id
 from .result import APIResult
@@ -53,7 +53,7 @@ def get_binary_type(
     })
     connection.send(send_buffer)
 
-    response_head_struct = Response([
+    response_head_struct = get_response_class(connection)([
         ('type_exists', Bool),
     ])
     response_head_type, recv_buffer = response_head_struct.parse(connection)