Skip to content

Commit cbfa4a5

Browse files
committed
added starmap; bump version.
1 parent d820d9b commit cbfa4a5

File tree

9 files changed

+3469
-3240
lines changed

9 files changed

+3469
-3240
lines changed

README.Rmd

Lines changed: 56 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,23 @@ import ezpq
2929

3030
# `ezpq`: an easy parallel queueing system.
3131

32-
Read this on [GitHub](https://github.com/dm3ll3n/ezpq) or [my site](https://www.donaldmellenbruch.com/project/ezpq/).
32+
> Read this on [GitHub](https://github.com/dm3ll3n/ezpq) or [my site](https://www.donaldmellenbruch.com/project/ezpq/).
33+
34+
## How to get it
35+
36+
Install from [PyPI](https://pypi.org/project/ezpq/) with:
37+
38+
```python
39+
pip install ezpq
40+
```
41+
42+
Optional packages:
43+
44+
```python
45+
pip install pandas # required for plots
46+
pip install plotnine # required for plots
47+
pip install tqdm # required for progress bars
48+
```
3349

3450
## Overview
3551

@@ -59,22 +75,6 @@ The queueing system uses `multiprocessing.Process` by default and can also run j
5975
* Built-in logging to CSV.
6076
* Customizable visualizations of queue operations.
6177

62-
## How to get it
63-
64-
Install from [PyPI](https://pypi.org/project/ezpq/) with:
65-
66-
```python
67-
pip install ezpq
68-
```
69-
70-
Optional packages:
71-
72-
```python
73-
pip install pandas # required for plots
74-
pip install plotnine # required for plots
75-
pip install tqdm # required for progress bars
76-
```
77-
7878
## Quickstart
7979

8080
Suppose you wanted to speed up the following code, which runs 60 operations that take anywhere from 0s to 2s. With an average job time of ~1s, this operation should take ~60s.
@@ -311,6 +311,44 @@ with ezpq.Queue(6) as Q:
311311

312312
![](docs/imgs/tqdm_map.gif)
313313

314+
### starmap
315+
316+
`starmap` is similar to `map`, but operates on a list of lists, with each nested list being unpacked as arguments to the function.
317+
318+
```{python, echo=TRUE}
319+
def my_pow(x, k):
320+
return '{}^{} = {}'.format(x, k, x**k)
321+
322+
# list of lists to iterate over.
323+
args_list = [[x, x%4] # (x, k)
324+
for x in range(100)]
325+
326+
# starmap
327+
with ezpq.Queue(10) as Q:
328+
output = Q.starmap(my_pow, iterable=args_list)
329+
330+
[x['output'] for x in output[:10]]
331+
```
332+
333+
### startmapkw
334+
335+
Same as `starmap`, but operations on a list of *dicts* to be expanded as kwargs to the function.
336+
337+
```{python, echo=TRUE}
338+
def my_pow(x, k):
339+
return '{}^{} = {}'.format(x, k, x**k)
340+
341+
# list of dicts to iterate over.
342+
kwargs_list = [{ 'x':x, 'k':x%4 } # (x, k)
343+
for x in range(100)]
344+
345+
# starmapkw
346+
with ezpq.Queue(10) as Q:
347+
output = Q.starmapkw(my_pow, iterable=kwargs_list)
348+
349+
[x['output'] for x in output[:10]]
350+
```
351+
314352
### dispose
315353

316354
The queueing operations performed by `ezpq.Queue` are performed on a periodic basis. By default, the `poll` parameter for a Queue is `0.1` seconds. This "pulse" thread will continue firing until the Queue is disposed of.
@@ -336,7 +374,7 @@ In the above graphic, notice how same-colored bars never overlap. These bars rep
336374

337375
### Lane Error Handling
338376

339-
You may want to short-circuit a synchronous lane if a job in the lane fails. You can do this by specifying `skip_on_lane_error=True` when putting a job in the queue. If specified and the preceding job has a non-zero exit code, this job will not be run.
377+
You may want to short-circuit a synchronous lane if a job in the lane fails. You can do this by specifying `stop_on_lane_error=True` when putting a job in the queue. If specified and the preceding job has a non-zero exit code, this job will not be run.
340378

341379
```{python, echo=TRUE}
342380
def reciprocal(x):

README.md

Lines changed: 87 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,42 @@
11
# `ezpq`: an easy parallel queueing system.
22

3-
* [`ezpq`: an easy parallel queueing system.](#ezpq-an-easy-parallel-queueing-system)
4-
* [Overview](#overview)
5-
* [Features](#features)
6-
* [How to get it](#how-to-get-it)
7-
* [Quickstart](#quickstart)
8-
* [ezpq.Queue](#ezpqqueue)
9-
* [ezpq.Job](#ezpqjob)
10-
* [put](#put)
11-
* [size](#size)
12-
* [wait](#wait)
13-
* [get](#get)
14-
* [collect](#collect)
15-
* [map](#map)
16-
* [dispose](#dispose)
17-
* [Synchronous Lanes](#synchronous-lanes)
18-
* [Lane Error Handling](#lane-error-handling)
19-
* [ezpq.Plot](#ezpqplot)
20-
* [More Examples](#more-examples)
21-
22-
Read this on [GitHub](https://github.com/dm3ll3n/ezpq) or [my
23-
site](https://www.donaldmellenbruch.com/project/ezpq/).
3+
> Read this on [GitHub](https://github.com/dm3ll3n/ezpq) or [my site](https://www.donaldmellenbruch.com/project/ezpq/).
4+
5+
- [How to get it](#how-to-get-it)
6+
- [Overview](#overview)
7+
- [Features](#features)
8+
- [Quickstart](#quickstart)
9+
- [ezpq.Queue](#ezpq.queue)
10+
- [ezpq.Job](#ezpq.job)
11+
- [put](#put)
12+
- [size](#size)
13+
- [wait](#wait)
14+
- [get](#get)
15+
- [collect](#collect)
16+
- [map](#map)
17+
- [starmap](#starmap)
18+
- [startmapkw](#startmapkw)
19+
- [dispose](#dispose)
20+
- [Synchronous Lanes](#synchronous-lanes)
21+
- [Lane Error Handling](#lane-error-handling)
22+
- [ezpq.Plot](#ezpq.plot)
23+
- [More Examples](#more-examples)
24+
25+
## How to get it
26+
27+
Install from [PyPI](https://pypi.org/project/ezpq/) with:
28+
29+
``` python
30+
pip install ezpq
31+
```
32+
33+
Optional packages:
34+
35+
``` python
36+
pip install pandas # required for plots
37+
pip install plotnine # required for plots
38+
pip install tqdm # required for progress bars
39+
```
2440

2541
## Overview
2642

@@ -53,22 +69,6 @@ also run jobs with `threading.Thread`.
5369
- Built-in logging to CSV.
5470
- Customizable visualizations of queue operations.
5571

56-
## How to get it
57-
58-
Install from [PyPI](https://pypi.org/project/ezpq/) with:
59-
60-
``` python
61-
pip install ezpq
62-
```
63-
64-
Optional packages:
65-
66-
``` python
67-
pip install pandas # required for plots
68-
pip install plotnine # required for plots
69-
pip install tqdm # required for progress bars
70-
```
71-
7272
## Quickstart
7373

7474
Suppose you wanted to speed up the following code, which runs 60
@@ -143,7 +143,7 @@ print( output[0] )
143143
## {'args': [0],
144144
## 'callback': None,
145145
## 'cancelled': False,
146-
## 'ended': datetime.datetime(2019, 2, 18, 20, 21, 0, 902915),
146+
## 'ended': datetime.datetime(2019, 3, 13, 0, 48, 52, 811248),
147147
## 'exception': None,
148148
## 'exitcode': 0,
149149
## 'function': 'random_sleep',
@@ -153,11 +153,11 @@ print( output[0] )
153153
## 'name': 1,
154154
## 'output': 1.3444218515250481,
155155
## 'priority': 100,
156-
## 'processed': datetime.datetime(2019, 2, 18, 20, 21, 0, 955396),
157-
## 'qid': 'f4717edb',
158-
## 'runtime': 1.3515939712524414,
159-
## 'started': datetime.datetime(2019, 2, 18, 20, 20, 59, 551321),
160-
## 'submitted': datetime.datetime(2019, 2, 18, 20, 20, 59, 446199),
156+
## 'processed': datetime.datetime(2019, 3, 13, 0, 48, 52, 867387),
157+
## 'qid': '13318d36',
158+
## 'runtime': 1.3500409126281738,
159+
## 'started': datetime.datetime(2019, 3, 13, 0, 48, 51, 461207),
160+
## 'submitted': datetime.datetime(2019, 3, 13, 0, 48, 51, 357405),
161161
## 'timeout': 0}
162162

163163
Easily convert output to a `pandas` dataframe:
@@ -169,11 +169,11 @@ print( df.head()[['id', 'output', 'runtime', 'exitcode']] )
169169
```
170170

171171
## id output runtime exitcode
172-
## 0 1 1.344422 1.351594 0
173-
## 1 2 0.634364 0.640723 0
174-
## 2 3 1.456034 1.461620 0
175-
## 3 4 0.737965 0.743645 0
176-
## 4 5 0.736048 0.742260 0
172+
## 0 1 1.344422 1.350041 0
173+
## 1 2 0.634364 0.638938 0
174+
## 2 3 1.456034 1.459830 0
175+
## 3 4 0.737965 0.741742 0
176+
## 4 5 0.736048 0.739848 0
177177

178178
Use `ezpq.Plot` to generate a Gannt chart of the job timings.
179179

@@ -335,9 +335,9 @@ with ezpq.Queue(6) as Q:
335335
## 'Total: 60; Waiting: 31; Working: 6; Completed: 23'
336336
## 'Total: 60; Waiting: 24; Working: 6; Completed: 30'
337337
## 'Total: 60; Waiting: 17; Working: 6; Completed: 37'
338-
## 'Total: 60; Waiting: 12; Working: 6; Completed: 42'
338+
## 'Total: 60; Waiting: 11; Working: 6; Completed: 43'
339339
## 'Total: 60; Waiting: 6; Working: 6; Completed: 48'
340-
## 'Total: 60; Waiting: 1; Working: 6; Completed: 53'
340+
## 'Total: 60; Waiting: 0; Working: 5; Completed: 55'
341341
## 'Total: 60; Waiting: 0; Working: 1; Completed: 59'
342342
## 'Total: 60; Waiting: 0; Working: 0; Completed: 60'
343343

@@ -407,6 +407,42 @@ call. Include `show_progress=True` to get output `tqdm` progress bar.
407407

408408
![](docs/imgs/tqdm_map.gif)
409409

410+
### starmap
411+
412+
`starmap` is similar to `map`, but operates on a list of lists, with
413+
each nested list being unpacked as arguments to the function.
414+
415+
``` python
416+
def my_pow(x, k):
417+
return '{}^{} = {}'.format(x, k, x**k)
418+
# list of lists to iterate over.
419+
args_list = [[x, x%4] # (x, k)
420+
for x in range(100)]
421+
# starmap
422+
with ezpq.Queue(10) as Q:
423+
output = Q.starmap(my_pow, iterable=args_list)
424+
425+
[x['output'] for x in output[:10]]
426+
```
427+
428+
### startmapkw
429+
430+
Same as `starmap`, but operations on a list of *dicts* to be expanded as
431+
kwargs to the function.
432+
433+
``` python
434+
def my_pow(x, k):
435+
return '{}^{} = {}'.format(x, k, x**k)
436+
# list of dicts to iterate over.
437+
kwargs_list = [{ 'x':x, 'k':x%4 } # (x, k)
438+
for x in range(100)]
439+
# starmapkw
440+
with ezpq.Queue(10) as Q:
441+
output = Q.starmapkw(my_pow, iterable=kwargs_list)
442+
443+
[x['output'] for x in output[:10]]
444+
```
445+
410446
### dispose
411447

412448
The queueing operations performed by `ezpq.Queue` are performed on a
@@ -435,7 +471,7 @@ synchronously.
435471
### Lane Error Handling
436472

437473
You may want to short-circuit a synchronous lane if a job in the lane
438-
fails. You can do this by specifying `skip_on_lane_error=True` when
474+
fails. You can do this by specifying `stop_on_lane_error=True` when
439475
putting a job in the queue. If specified and the preceding job has a
440476
non-zero exit code, this job will not be run.
441477

0 commit comments

Comments
 (0)