Merge pull request #10 from GoodManWEN/dev

Performance optimization

Author: GoodManWEN

cx_Oracle_async/AQ.py

@@ -3,6 +3,7 @@
 from asyncio import Lock as aioLock
 from collections.abc import Iterable
 from typing import Union , TYPE_CHECKING
+from collections import deque
 if TYPE_CHECKING:
     from .connections import AsyncConnectionWrapper
     from cx_Oracle import Queue
@@ -60,34 +61,68 @@ def __init__(self , loop : 'ProactorEventLoop' , thread_pool : ThreadPoolExecuto
         self._thread_pool = thread_pool
         self._queue = queue
         self._count = 0
-        self._max = maxMessages if maxMessages > -1 else (1 << 16 - 1) 
-        self._max = self._max if self._max <= (1 << 16 - 1) else (1 << 16 - 1) 
+        self._max_messages = maxMessages
         self._deqlock = deqlock
+        self._buffer = deque()
+        self._soft_max = ((1 << 16) - 1)
+        self._max_limit = maxMessages if maxMessages > -1 else self._soft_max
+        self._max_limit = self._max_limit if self._max_limit <= self._soft_max else self._soft_max 
+        self._deqcount = 0
+        self._closed = False
+
+    @property 
+    def _fetch_num(self):
+        return self._soft_max if self._max_messages < 0 else min(self._max_limit , self._max_messages - self._deqcount)
 
     def __await__(self):
+        if self._closed:
+            raise RuntimeError('Current query has closed , you cannot activate it twice.')
         yield from self._deqlock.acquire().__await__()
         try:
-            ret = yield from self._loop.run_in_executor(self._thread_pool , self._queue.deqMany , self._max).__await__()
+            ret = yield from self._loop.run_in_executor(self._thread_pool , self._queue.deqMany , self._fetch_num).__await__()
         except Exception as exc:
             raise exc
         finally:
             self._deqlock.release()
+        self._closed = True
         return ret
 
     def __aiter__(self):
         return self
 
     async def __anext__(self):
-        self._count += 1
-        if self._count <= self._max:
-            _tmp = self._queue.deqOptions.wait
-            async with self._deqlock:
+        if self._closed:
+            raise RuntimeError('Current query has closed , you cannot activate it twice.')
+
+        if self._max_messages == 0:
+            self._closed = True
+            raise StopAsyncIteration
+
+        # Fetch off
+        if self._max_messages > 0 and self._deqcount >= self._max_messages:
+            if self._buffer:
+                return self._buffer.popleft()
+            self._closed = True
+            raise StopAsyncIteration
+
+        # Fetch on
+        if self._buffer:
+            return self._buffer.popleft()
+        _tmp = self._queue.deqOptions.wait
+        async with self._deqlock:
+            try:
                 self._queue.deqOptions.wait = DEQ_NO_WAIT
-                data = await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne)
+                data = await self._loop.run_in_executor(self._thread_pool , self._queue.deqMany , self._fetch_num)
+            except Exception as exc:
+                raise exc
+            finally:
                 self._queue.deqOptions.wait = _tmp
+
             if data:
-                return data
-            else:
-                raise StopAsyncIteration
-        else:
+                self._buffer.extend(data)
+                self._deqcount += len(data)
+                return self._buffer.popleft()
+
+            # No data return , close iteration.
+            self._closed = True
             raise StopAsyncIteration

docs/temporary_document_of_AQ.md

@@ -258,8 +258,10 @@ async def main():
             assert list(map(queue.unpack , ret)) == list(map(str , range(10)))
 
             # The second way , you can call deqMany as a asynchronous generator.
-            # This is a self implemented method which yield a single queue.deqOne 
-            # each time. The benifits is you will get immediate response. 
+            # This is a self implemented method which yield queue.deqMany() with
+            # queue.deqOptions = DEQ_NO_WAIT until it reaches the message limit or
+            # there's nothing in the queue. The benifits is you will get immediate 
+            # response. 
 
             await queue.enqMany(queue.pack(m) for m in map(str , range(10)))
             await conn.commit()
@@ -283,17 +285,16 @@ async def main():
             # Of course you can change deqOptions into non-blocking mode like
             # `queue.deqOptions.wait = cx_Oracle_async.DEQ_NO_WAIT` to aviod it.
 
-            # On the other hand ,  If you are using the `async with` mode , it will never 
-            # block your main thread , however it's low efficiency , and it will not be affected by 
-            # `Queue.deqOptions` , no matter what setting `Queue.deqOptions` is , it will return 
-            # immediately when there's nothing in the queue.
+            # On the other hand ,  If you are using the `async with` mode , it will  
+            # never block your main thread , however it will not be affected by 
+            # `Queue.deqOptions` , no matter what setting `Queue.deqOptions` is , 
+            # it will return immediately when there's nothing in the queue.
 
-            # So taking into consideration that when argument maxMessages equals to -1 (default value),
-            # it means unlimit fetch untill the queue is empty (whose "unlimit" do has a soft upper bound
-            # of maximum queue length of 65535). It's convenient to clear the whole queue with 
-            # the following code:
+            # So taking into consideration that when argument maxMessages equals to 
+            # -1 (default value), it means unlimit fetch untill the queue is empty. 
+            # It's convenient to clear the whole queue with the following code:
 
-            messages = list(map(str , range(random.randint(0,10))))
+            messages = list(map(str , range(random.randint(0,10000))))
             await queue.enqMany(queue.pack(m) for m in messages)
             await conn.commit()
 

tests/test_aq.py

@@ -238,4 +238,22 @@ async def test_block_behavior():
             ret = await queue.deqMany(-1)
             ed_time = time.time()
             assert queue.unpack(ret) == ["Hello World",]
-            assert (2 - INAQ) <= (ed_time - st_time) <= (2 + INAQ)
+            assert (2 - INAQ) <= (ed_time - st_time) <= (2 + INAQ)
+
+            await queue.enqOne(queue.pack('Hello World'))
+            await conn.commit()
+            obj = queue.deqMany(65535)  
+            ret = await obj 
+            try:
+                ret = await obj 
+                raise TypeError()
+            except Exception as exc:
+                assert isinstance(exc , RuntimeError)
+
+            obj = queue.deqMany(65535) 
+            async for _ in obj:...
+            try:
+                async for _ in obj:...
+                raise TypeError()
+            except Exception as exc:
+                assert isinstance(exc , RuntimeError)