Add pending_limit into Scheduler (ref #19)
#23
Conversation
- Add parameter `pending_limit` into `Scheduler` for control pending jobs size - Add property `pending_limit` into `Scheduler` - Add property `waiting_count` into `Scheduler` - Add method `_release_waiter` into `Scheduler`
Codecov Report
@@ Coverage Diff @@
## master #23 +/- ##
=====================================
Coverage 100% 100%
=====================================
Files 4 4
Lines 221 259 +38
=====================================
+ Hits 221 259 +38
Continue to review full report at Codecov.
|
aiojobs/_scheduler.py
Outdated
| @@ -10,20 +10,22 @@ | |||
| from collections.abc import Sized, Iterable, Container | |||
| bases = Sized, Iterable, Container | |||
| else: # pragma: no cover | |||
| bases = (Collection,) | |||
| bases = (Collection, ) | |||
There was a problem hiding this comment.
Please don't touch the line, space is not needed.
aiojobs/_scheduler.py
Outdated
| @@ -71,8 +81,19 @@ def closed(self): | |||
| if self._closed: | |||
| raise RuntimeError("Scheduling a new job after closing") | |||
| job = Job(coro, self, self._loop) | |||
| should_start = (self._limit is None or | |||
| self.active_count < self._limit) | |||
| should_start = (self._limit is None | |||
There was a problem hiding this comment.
Please keep existing formatting for the line.
The same should be applied to should_wait
| @@ -60,6 +66,10 @@ def active_count(self): | |||
| def pending_count(self): | |||
| return len(self._pending) | |||
|
|
|||
| @property | |||
| def waiting_count(self): | |||
There was a problem hiding this comment.
No need to expose the property IMHO
aiojobs/_scheduler.py
Outdated
| and self.pending_count >= self._pending_limit) | ||
| if not should_start and should_wait: | ||
| waiter = self._loop.create_future() | ||
| waiter._job = job |
There was a problem hiding this comment.
No chance to accept this line: adding a custom attribute to asyncio Future has very bad smell.
| self._exception_handler = exception_handler | ||
| self._failed_tasks = asyncio.Queue(loop=loop) | ||
| self._failed_task = loop.create_task(self._wait_failed()) | ||
| self._pending = deque() | ||
| self._waiting = deque() |
There was a problem hiding this comment.
I believe using asyncio.Queue for _pending and not adding _waiting is more robust and simple solution.
There was a problem hiding this comment.
using asyncio.Queue is simply, but can't limit the size of pending jobs , when set pending_limit == 0
There was a problem hiding this comment.
asyncio.Queue(maxsize=pending_limit), infinite size if pending_limit=0.
Do you see any problem with this solution?
There was a problem hiding this comment.
I mean spawn always waits until one active job closes when pending_limit=0. And the _pending has infinite size when pending_limit=None
|
#24 is simpler |
This PR for #19
pending_limitintoSchedulerfor control pending jobs size
pending_limitintoSchedulerwaiting_countintoScheduler