response optimizations + ConnectionState.IDLE

Author: igorcoding

README.md

@@ -68,10 +68,10 @@ RPS on running 200k requests in 300 parallel coroutines (no `uvloop`):
 | ------------- |:-------------:| ---------:|
 | ping          | 24961.12      | 28155.32  |
 | call          | 21748.06      | 22103.14  |
-| eval          | 20497.69      | 21456.38  |
-| select        | 19968.26      | 23558.00  |
+| eval          | 20497.69      | 24417.38  |
+| select        | 19968.26      | 26151.00  |
 | insert        | 20604.61      | 22256.69  |
-| update        | 18852.46      | 21988.80  |
+| update        | 18852.46      | 23175.80  |
 
 
 Let's enable uvloop. This is where asynctnt shines.
@@ -130,6 +130,9 @@ async def run():
         await conn.insert('tester', [i, 'hello{}'.format(i)])
 
     values = await conn.select('tester', [])
+    print('Code: {}'.format(values.code))
+    print('Data: {}'.format(values.body))
+    print(values.body2yaml())  # prints as yaml
 
     await conn.disconnect()
 

asynctnt/__init__.py

@@ -1,4 +1,4 @@
 from .connection import Connection, connect
 from .iproto.protocol import Iterator, Response
 
-__version__ = '0.0.7'
+__version__ = '0.0.8'

asynctnt/connection.py

@@ -14,11 +14,12 @@
 
 
 class ConnectionState(enum.IntEnum):
-    CONNECTING = 0
-    CONNECTED = 1
-    RECONNECTING = 2
-    DISCONNECTING = 3
-    DISCONNECTED = 4
+    IDLE = 0
+    CONNECTING = 1
+    CONNECTED = 2
+    RECONNECTING = 3
+    DISCONNECTING = 4
+    DISCONNECTED = 5
 
 
 class Connection:
@@ -109,8 +110,8 @@ def __init__(self, *,
         self._protocol = None
         self._db = DbMock()
 
-        self._state = ConnectionState.DISCONNECTED
-        self._state_prev = ConnectionState.DISCONNECTED
+        self._state = ConnectionState.IDLE
+        self._state_prev = ConnectionState.IDLE
         self._disconnect_waiter = None
         self._reconnect_coro = None
 
@@ -167,6 +168,7 @@ async def _connect(self, return_exceptions=True):
                     ConnectionState.CONNECTING,
                     ConnectionState.CONNECTED,
                     ConnectionState.DISCONNECTING,
+                    ConnectionState.DISCONNECTED,
                 }
                 if self._state in ignore_states:
                     return
@@ -312,6 +314,7 @@ async def reconnect(self):
             Just calls disconnect() and connect()
         """
         await self.disconnect()
+        self._set_state(ConnectionState.IDLE)
         await self.connect()
 
     @property
@@ -463,6 +466,7 @@ def ping(self, *, timeout=-1):
         """
             Ping request coroutine
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.ping(timeout=timeout)
 
@@ -474,6 +478,7 @@ def call16(self, func_name, args=None, *, timeout=-1):
         :param func_name: function name to call
         :param args: arguments to pass to the function (list object)
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.call16(func_name, args,
                                timeout=timeout)
@@ -488,6 +493,7 @@ def call(self, func_name, args=None, *, timeout=-1):
         :param func_name: function name to call
         :param args: arguments to pass to the function (list object)
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.call(func_name, args,
                              timeout=timeout)
@@ -500,6 +506,7 @@ def eval(self, expression, args=None, *, timeout=-1):
         :param args: arguments to pass to the function, that will
                      execute your expression (list object)
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.eval(expression, args,
                              timeout=timeout)
@@ -519,6 +526,7 @@ def select(self, space, key=None, **kwargs):
                     * asynctnt.Iterator object
                     * string with an iterator name
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.select(space, key, **kwargs)
 
@@ -530,6 +538,7 @@ def insert(self, space, t, *, replace=False, timeout=-1):
         :param t: tuple to insert (list object)
         :param replace: performs replace request instead of insert
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.insert(space, t,
                                replace=replace, timeout=timeout)
@@ -541,6 +550,7 @@ def replace(self, space, t, *, timeout=-1):
         :param space: space id or space name.
         :param t: tuple to insert (list object)
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.replace(space, t,
                                 timeout=timeout)
@@ -553,6 +563,7 @@ def delete(self, space, key, **kwargs):
         :param key: key to delete
         :param index: index id or name
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.delete(space, key, **kwargs)
 
@@ -581,6 +592,7 @@ def upsert(self, space, t, operations, **kwargs):
                 Please refer to
                 https://tarantool.org/doc/book/box/box_space.html?highlight=update#lua-function.space_object.update
         :param timeout: Request timeout
+        :rtype: asynctnt.Response
         """
         return self._db.upsert(space, t, operations, **kwargs)
 

asynctnt/iproto/coreproto.pyx

@@ -119,11 +119,11 @@ cdef class CoreProtocol:
                 resp = response_parse(p, packet_len, self.encoding)
                 p = &p[packet_len]  # skip entire packet
 
-                sync_obj = <object>resp.sync
+                sync_obj = <object>resp._sync
 
                 req_p = cpython.dict.PyDict_GetItem(self.reqs, sync_obj)
                 if req_p is NULL:
-                    logger.warning('sync %d not found', resp.sync)
+                    logger.warning('sync %d not found', resp._sync)
                     continue
 
                 req = <Request>req_p
@@ -133,9 +133,9 @@ cdef class CoreProtocol:
                 waiter = req.waiter
                 if waiter is not None \
                         and not waiter.done():
-                    if resp.code != 0:
+                    if resp._code != 0:
                         waiter.set_exception(
-                            TarantoolDatabaseError(resp.code, resp.errmsg))
+                            TarantoolDatabaseError(resp._code, resp._errmsg))
                     else:
                         waiter.set_result(resp)
 

asynctnt/iproto/response.pxd

@@ -3,17 +3,16 @@ from libc.stdint cimport uint64_t, uint32_t, int64_t
 
 cdef class Response:
     cdef:
-        public uint32_t code
-        public uint64_t sync
-        public int64_t schema_id
-        public str errmsg
-        public list body
-        public bytes encoding
+        uint32_t _code
+        uint64_t _sync
+        int64_t _schema_id
+        str _errmsg
+        list _body
+        bytes _encoding
 
     @staticmethod
     cdef inline Response new(bytes encoding)
 
-    cdef inline has_schema_id(self)
     cdef inline is_error(self)
 
 

asynctnt/iproto/response.pyx

@@ -9,29 +9,57 @@ import yaml
 
 
 cdef class Response:
+    def __cinit__(self):
+        self._sync = 0
+        self._code = 0
+        self._schema_id = -1
+        self._errmsg = None
+        self._body = None
+        self._encoding = None
+
     @staticmethod
     cdef inline Response new(bytes encoding):
         cdef Response resp
         resp = Response.__new__(Response)
-        resp.sync = 0
-        resp.code = 0
-        resp.schema_id = -1
-        resp.errmsg = None
-        resp.body = None
-        resp.encoding = encoding
+        resp._encoding = encoding
         return resp
 
-    cdef inline has_schema_id(self):
-        return self.schema_id != -1
-
     cdef inline is_error(self):
-        return self.code != 0
+        return self._code != 0
 
     def __repr__(self):
-        return '<Response: code={}, sync={}>'.format(self.code, self.sync)
+        body_len = None
+        if self._body is not None:
+            body_len = len(self._body)
+        return '<Response: code={}, sync={}, body_len={}>'.format(
+            self._code, self._sync, body_len)
 
     def body2yaml(self):
-        return yaml.dump(self.body, allow_unicode=True)
+        return yaml.dump(self._body, allow_unicode=True)
+
+    @property
+    def sync(self):
+        return self._sync
+
+    @property
+    def code(self):
+        return self._code
+
+    @property
+    def schema_id(self):
+        return self._schema_id
+
+    @property
+    def errmsg(self):
+        return self._errmsg
+
+    @property
+    def body(self):
+        return self._body
+
+    @property
+    def encoding(self):
+        return self._encoding
 
 
 cdef object _decode_obj(const char** p, bytes encoding):
@@ -152,18 +180,18 @@ cdef Response response_parse(const char *buf, uint32_t buf_len,
             if mp_typeof(b[0]) != MP_UINT:
                 raise TypeError('code type must be a MP_UINT')
 
-            resp.code = <uint32_t>mp_decode_uint(&b)
-            resp.code &= 0x7FFF
+            resp._code = <uint32_t>mp_decode_uint(&b)
+            resp._code &= 0x7FFF
         elif key == tnt.TP_SYNC:
             if mp_typeof(b[0]) != MP_UINT:
                 raise TypeError('sync type must be a MP_UINT')
 
-            resp.sync = mp_decode_uint(&b)
+            resp._sync = mp_decode_uint(&b)
         elif key == tnt.TP_SCHEMA_ID:
             if mp_typeof(b[0]) != MP_UINT:
                 raise TypeError('schema_id type must be a MP_UINT')
 
-            resp.schema_id = mp_decode_uint(&b)
+            resp._schema_id = mp_decode_uint(&b)
         else:
             logger.warning('Unknown argument in header. Skipping.')
             mp_next(&b)
@@ -189,10 +217,10 @@ cdef Response response_parse(const char *buf, uint32_t buf_len,
             s = NULL
             s_len = 0
             s = mp_decode_str(&b, &s_len)
-            resp.errmsg = s[:s_len].decode(encoding)
+            resp._errmsg = s[:s_len].decode(encoding)
         elif key == tnt.TP_DATA:
             if mp_typeof(b[0]) != MP_ARRAY:
                 raise TypeError('body data type must be a MP_ARRAY')
-            resp.body = _response_parse_body_data(b, encoding)
+            resp._body = _response_parse_body_data(b, encoding)
 
     return resp

tests/test_connect.py

@@ -15,6 +15,8 @@ async def test__connect(self):
         conn = asynctnt.Connection(host=self.tnt.host, port=self.tnt.port,
                                    reconnect_timeout=0,
                                    loop=self.loop)
+        self.assertEqual(conn.state, ConnectionState.IDLE)
+
         await conn.connect()
         self.assertIsNotNone(conn._transport)
         self.assertIsNotNone(conn._protocol)
@@ -149,3 +151,21 @@ async def test__connect_tnt_restarted(self):
         finally:
             await conn.disconnect()
 
+    async def test__connect_force_disconnect(self):
+        conn = asynctnt.Connection(host=self.tnt.host, port=44444,
+                                   reconnect_timeout=0.3,
+                                   loop=self.loop)
+        self.ensure_future(conn.connect())
+        await self.sleep(1)
+        await conn.disconnect()
+        self.assertEqual(conn.state, ConnectionState.DISCONNECTED)
+
+    async def test__close(self):
+        conn = asynctnt.Connection(host=self.tnt.host, port=self.tnt.port,
+                                   reconnect_timeout=0.3,
+                                   loop=self.loop)
+        await conn.connect()
+        await self.sleep(0.1)
+        conn.close()
+        await self.sleep(0.1)
+        self.assertEqual(conn.state, ConnectionState.DISCONNECTED)