IGNITE-14851 Enable partition awareness by default, fix unnecessary double connect - Fixes #42.
diff --git a/docs/async_examples.rst b/docs/async_examples.rst
index 5e60c70..af61a75 100644
--- a/docs/async_examples.rst
+++ b/docs/async_examples.rst
@@ -84,6 +84,7 @@
Client transactions are supported for caches with
:py:attr:`~pyignite.datatypes.cache_config.CacheAtomicityMode.TRANSACTIONAL` mode.
+**Supported only python 3.7+**
Let's create transactional cache:
@@ -97,28 +98,28 @@
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 36-41
+ :lines: 36-42
Let's check that the transaction was committed successfully:
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 44-45
+ :lines: 45-46
Let's check that raising exception inside `async with` block leads to transaction's rollback
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 48-59
+ :lines: 49-61
Let's check that timed out transaction is successfully rolled back
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 62-73
+ :lines: 64-75
See more info about transaction's parameters in a documentation of :py:meth:`~pyignite.aio_client.AioClient.tx_start`
diff --git a/docs/examples.rst b/docs/examples.rst
index b3bc60e..e01f112 100644
--- a/docs/examples.rst
+++ b/docs/examples.rst
@@ -219,35 +219,35 @@
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 82-85
+ :lines: 84-87
Let's start a transaction and commit it:
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 88-92
+ :lines: 90-96
Let's check that the transaction was committed successfully:
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 94-95
+ :lines: 98-99
Let's check that raising exception inside `with` block leads to transaction's rollback
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 98-108
+ :lines: 102-113
Let's check that timed out transaction is successfully rolled back
.. literalinclude:: ../examples/transactions.py
:language: python
:dedent: 8
- :lines: 111-121
+ :lines: 116-126
See more info about transaction's parameters in a documentation of :py:meth:`~pyignite.client.Client.tx_start`
diff --git a/docs/images/partitionawareness01.png b/docs/images/partitionawareness01.png
new file mode 100644
index 0000000..51c11a7
--- /dev/null
+++ b/docs/images/partitionawareness01.png
Binary files differ
diff --git a/docs/images/partitionawareness02.png b/docs/images/partitionawareness02.png
new file mode 100644
index 0000000..d6698be
--- /dev/null
+++ b/docs/images/partitionawareness02.png
Binary files differ
diff --git a/docs/index.rst b/docs/index.rst
index 7c28b6c..5a6f8d3 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -22,6 +22,7 @@
readme
modules
+ partition_awareness
examples
async_examples
diff --git a/docs/partition_awareness.rst b/docs/partition_awareness.rst
new file mode 100644
index 0000000..5382dfc
--- /dev/null
+++ b/docs/partition_awareness.rst
@@ -0,0 +1,63 @@
+.. Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. 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.
+
+===================
+Partition Awareness
+===================
+
+Partition awareness allows the thin client to send query requests directly to the node that owns the queried data.
+
+Without partition awareness, an application that is connected to the cluster via a thin client executes all queries and operations via a single server node that acts as a proxy for the incoming requests. These operations are then re-routed to the node that stores the data that is being requested. This results in a bottleneck that could prevent the application from scaling linearly.
+
+.. image:: images/partitionawareness01.png
+ :alt: Without partition awareness
+
+Notice how queries must pass through the proxy server node, where they are routed to the correct node.
+
+With partition awareness in place, the thin client can directly route queries and operations to the primary nodes that own the data required for the queries. This eliminates the bottleneck, allowing the application to scale more easily.
+
+.. image:: images/partitionawareness02.png
+ :alt: With partition awareness
+
+Partition awareness can be enabled or disabled by setting `partition_aware` parameter in
+:meth:`pyignite.client.Client.__init__` or :meth:`pyignite.aio_client.AioClient.__init__` to `True` (by default)
+or `False`.
+
+Also, it is recommended to pass list of address and port pairs of all server nodes
+to :meth:`pyignite.client.Client.connect` or to :meth:`pyignite.aio_client.AioClient.connect`.
+
+For example:
+
+.. code-block:: python3
+
+ from pyignite import Client
+
+ client = Client(
+ partition_awareness=True
+ )
+ nodes = [('10.128.0.1', 10800), ('10.128.0.2', 10800),...]
+ with client.connect(nodes):
+ ....
+
+.. code-block:: python3
+
+ from pyignite import AioClient
+
+ client = AioClient(
+ partition_awareness=True
+ )
+ nodes = [('10.128.0.1', 10800), ('10.128.0.2', 10800),...]
+ async with client.connect(nodes):
+ ....
\ No newline at end of file
diff --git a/docs/source/pyignite.transaction.rst b/docs/source/pyignite.transaction.rst
index 7c6b016..b0301f4 100644
--- a/docs/source/pyignite.transaction.rst
+++ b/docs/source/pyignite.transaction.rst
@@ -14,7 +14,7 @@
limitations under the License.
pyignite.transaction module
-=========================
+===========================
.. automodule:: pyignite.transaction
:members:
diff --git a/examples/transactions.py b/examples/transactions.py
index a0c90ba..ef9b08c 100644
--- a/examples/transactions.py
+++ b/examples/transactions.py
@@ -35,19 +35,21 @@
# starting transaction
key = 1
async with client.tx_start(
- isolation=TransactionIsolation.REPEATABLE_READ, concurrency=TransactionConcurrency.PESSIMISTIC
+ isolation=TransactionIsolation.REPEATABLE_READ,
+ concurrency=TransactionConcurrency.PESSIMISTIC
) as tx:
await cache.put(key, 'success')
await tx.commit()
# key=1 value=success
val = await cache.get(key)
- print(f"key=1 value={val}")
+ print(f"key={key} value={val}")
# rollback transaction.
try:
async with client.tx_start(
- isolation=TransactionIsolation.REPEATABLE_READ, concurrency=TransactionConcurrency.PESSIMISTIC
+ isolation=TransactionIsolation.REPEATABLE_READ,
+ concurrency=TransactionConcurrency.PESSIMISTIC
):
await cache.put(key, 'fail')
raise RuntimeError('test')
@@ -56,7 +58,7 @@
# key=1 value=success
val = await cache.get(key)
- print(f"key=1 value={val}")
+ print(f"key={key} value={val}")
# rollback transaction on timeout.
try:
@@ -70,7 +72,7 @@
# key=1 value=success
val = await cache.get(key)
- print(f"key=1 value={val}")
+ print(f"key={key} value={val}")
# destroy cache
await cache.destroy()
@@ -85,32 +87,35 @@
})
# starting transaction
+ key = 1
with client.tx_start(
- isolation=TransactionIsolation.REPEATABLE_READ, concurrency=TransactionConcurrency.PESSIMISTIC
+ isolation=TransactionIsolation.REPEATABLE_READ,
+ concurrency=TransactionConcurrency.PESSIMISTIC
) as tx:
- cache.put(1, 'success')
+ cache.put(key, 'success')
tx.commit()
# key=1 value=success
- print(f"key=1 value={cache.get(1)}")
+ print(f"key={key} value={cache.get(key)}")
# rollback transaction.
try:
with client.tx_start(
- isolation=TransactionIsolation.REPEATABLE_READ, concurrency=TransactionConcurrency.PESSIMISTIC
+ isolation=TransactionIsolation.REPEATABLE_READ,
+ concurrency=TransactionConcurrency.PESSIMISTIC
):
- cache.put(1, 'fail')
+ cache.put(key, 'fail')
raise RuntimeError('test')
except RuntimeError:
pass
# key=1 value=success
- print(f"key=1 value={cache.get(1)}")
+ print(f"key={key} value={cache.get(key)}")
# rollback transaction on timeout.
try:
with client.tx_start(timeout=1.0, label='long-tx') as tx:
- cache.put(1, 'fail')
+ cache.put(key, 'fail')
time.sleep(2.0)
tx.commit()
except CacheError as e:
@@ -118,7 +123,7 @@
print(e)
# key=1 value=success
- print(f"key=1 value={cache.get(1)}")
+ print(f"key={key} value={cache.get(key)}")
# destroy cache
cache.destroy()
diff --git a/pyignite/aio_client.py b/pyignite/aio_client.py
index 26d243d..2bc850b 100644
--- a/pyignite/aio_client.py
+++ b/pyignite/aio_client.py
@@ -60,7 +60,7 @@
Asynchronous Client implementation.
"""
- def __init__(self, compact_footer: bool = None, partition_aware: bool = False, **kwargs):
+ def __init__(self, compact_footer: bool = None, partition_aware: bool = True, **kwargs):
"""
Initialize client.
@@ -68,13 +68,10 @@
full (False) schema approach when serializing Complex objects.
Default is to use the same approach the server is using (None).
Apache Ignite binary protocol documentation on this topic:
- https://apacheignite.readme.io/docs/binary-client-protocol-data-format#section-schema
+ https://ignite.apache.org/docs/latest/binary-client-protocol/data-format#schema
:param partition_aware: (optional) try to calculate the exact data
placement from the key before to issue the key operation to the
- server node:
- https://cwiki.apache.org/confluence/display/IGNITE/IEP-23%3A+Best+Effort+Affinity+for+thin+clients
- The feature is in experimental status, so the parameter is `False`
- by default. This will be changed later.
+ server node, `True` by default.
"""
super().__init__(compact_footer, partition_aware, **kwargs)
self._registry_mux = asyncio.Lock()
@@ -494,7 +491,7 @@
isolation: TransactionIsolation = TransactionIsolation.REPEATABLE_READ,
timeout: Union[int, float] = 0, label: Optional[str] = None) -> 'AioTransaction':
"""
- Start async thin client transaction.
+ Start async thin client transaction. **Supported only python 3.7+**
:param concurrency: (optional) transaction concurrency, see
:py:class:`~pyignite.datatypes.transactions.TransactionConcurrency`
diff --git a/pyignite/api/__init__.py b/pyignite/api/__init__.py
index 7deed8c..19a7036 100644
--- a/pyignite/api/__init__.py
+++ b/pyignite/api/__init__.py
@@ -17,7 +17,7 @@
This module contains functions, that are (more or less) directly mapped to
Apache Ignite binary protocol operations. Read more:
-https://apacheignite.readme.io/docs/binary-client-protocol#section-client-operations
+https://ignite.apache.org/docs/latest/binary-client-protocol/binary-client-protocol#client-operations
When the binary client protocol changes, these functions also change. For
stable end user API see :mod:`pyignite.client` module.
diff --git a/pyignite/client.py b/pyignite/client.py
index b411a2b..f848bcc 100644
--- a/pyignite/client.py
+++ b/pyignite/client.py
@@ -335,17 +335,10 @@
class Client(BaseClient):
"""
- This is a main `pyignite` class, that is build upon the
- :class:`~pyignite.connection.Connection`. In addition to the attributes,
- properties and methods of its parent class, `Client` implements
- the following features:
-
- * cache factory. Cache objects are used for key-value operations,
- * Ignite SQL endpoint,
- * binary types registration endpoint.
+ Synchronous Client implementation.
"""
- def __init__(self, compact_footer: bool = None, partition_aware: bool = False, **kwargs):
+ def __init__(self, compact_footer: bool = None, partition_aware: bool = True, **kwargs):
"""
Initialize client.
@@ -353,13 +346,10 @@
full (False) schema approach when serializing Complex objects.
Default is to use the same approach the server is using (None).
Apache Ignite binary protocol documentation on this topic:
- https://apacheignite.readme.io/docs/binary-client-protocol-data-format#section-schema
+ https://ignite.apache.org/docs/latest/binary-client-protocol/data-format#schema
:param partition_aware: (optional) try to calculate the exact data
placement from the key before to issue the key operation to the
- server node:
- https://cwiki.apache.org/confluence/display/IGNITE/IEP-23%3A+Best+Effort+Affinity+for+thin+clients
- The feature is in experimental status, so the parameter is `False`
- by default. This will be changed later.
+ server node, `True` by default.
"""
super().__init__(compact_footer, partition_aware, **kwargs)
diff --git a/pyignite/connection/aio_connection.py b/pyignite/connection/aio_connection.py
index 86993ba..c6ecbc6 100644
--- a/pyignite/connection/aio_connection.py
+++ b/pyignite/connection/aio_connection.py
@@ -37,7 +37,7 @@
from .bitmask_feature import BitmaskFeature
from .connection import BaseConnection
-from .handshake import HandshakeRequest, HandshakeResponse, OP_HANDSHAKE
+from .handshake import HandshakeRequest, HandshakeResponse
from .protocol_context import ProtocolContext
from .ssl import create_ssl_context
from ..stream.binary_stream import BinaryStreamBase
@@ -152,20 +152,23 @@
self._transport = None
self._loop = asyncio.get_event_loop()
self._closed = False
+ self._transport_closed_fut = None
@property
def closed(self) -> bool:
""" Tells if socket is closed. """
return self._closed or not self._transport or self._transport.is_closing()
- async def connect(self) -> Union[dict, OrderedDict]:
+ async def connect(self):
"""
Connect to the given server node with protocol version fallback.
"""
+ if self.alive:
+ return
self._closed = False
- return await self._connect()
+ await self._connect()
- async def _connect(self) -> Union[dict, OrderedDict]:
+ async def _connect(self):
detecting_protocol = False
# choose highest version first
@@ -192,13 +195,16 @@
self.client.protocol_context.features = features
self.uuid = result.get('node_uuid', None) # version-specific (1.4+)
self.failed = False
- return result
def on_connection_lost(self, error, reconnect=False):
self.failed = True
for _, fut in self._pending_reqs.items():
fut.set_exception(error)
self._pending_reqs.clear()
+
+ if self._transport_closed_fut and not self._transport_closed_fut.done():
+ self._transport_closed_fut.set_result(None)
+
if reconnect and not self._closed:
self._loop.create_task(self._reconnect())
@@ -221,7 +227,7 @@
hs_response = await handshake_fut
if hs_response.op_code == 0:
- self._close_transport()
+ await self._close_transport()
self._process_handshake_error(hs_response)
return hs_response
@@ -233,7 +239,7 @@
if self.alive:
return
- self._close_transport()
+ await self._close_transport()
# connect and silence the connection errors
try:
await self._connect()
@@ -259,12 +265,20 @@
async def close(self):
self._closed = True
- self._close_transport()
+ await self._close_transport()
- def _close_transport(self):
+ async def _close_transport(self):
"""
Close connection.
"""
- if self._transport:
+ if self._transport and not self._transport.is_closing():
+ self._transport_closed_fut = self._loop.create_future()
+
self._transport.close()
self._transport = None
+ try:
+ await asyncio.wait_for(self._transport_closed_fut, 1.0)
+ except asyncio.TimeoutError:
+ pass
+ finally:
+ self._transport_closed_fut = None
diff --git a/pyignite/connection/connection.py b/pyignite/connection/connection.py
index e8437dc..2e6d6aa 100644
--- a/pyignite/connection/connection.py
+++ b/pyignite/connection/connection.py
@@ -156,12 +156,9 @@
def closed(self) -> bool:
return self._socket is None
- def connect(self) -> Union[dict, OrderedDict]:
+ def connect(self):
"""
Connect to the given server node with protocol version fallback.
-
- :param host: Ignite server node's host name or IP,
- :param port: Ignite server node's port number.
"""
detecting_protocol = False
@@ -189,7 +186,6 @@
self.client.protocol_context.features = features
self.uuid = result.get('node_uuid', None) # version-specific (1.4+)
self.failed = False
- return result
def _connect_version(self) -> Union[dict, OrderedDict]:
"""
diff --git a/pyignite/exceptions.py b/pyignite/exceptions.py
index fdf1261..7419512 100644
--- a/pyignite/exceptions.py
+++ b/pyignite/exceptions.py
@@ -38,7 +38,7 @@
"""
This exception is raised on Ignite binary protocol handshake failure,
as defined in
- https://apacheignite.readme.io/docs/binary-client-protocol#section-handshake
+ https://ignite.apache.org/docs/latest/binary-client-protocol/binary-client-protocol#connection-handshake
"""
def __init__(self, expected_version: Tuple[int, int, int], message: str):
diff --git a/tests/common/test_transactions.py b/tests/common/test_transactions.py
index f4efba5..57874b6 100644
--- a/tests/common/test_transactions.py
+++ b/tests/common/test_transactions.py
@@ -131,7 +131,7 @@
assert await async_tx_cache.get(1) == 1
async with async_client.tx_start(isolation=iso_level, concurrency=concurrency) as tx:
- async_tx_cache.put(1, 10)
+ await async_tx_cache.put(1, 10)
assert await async_tx_cache.get(1) == 1
