Based on this excellent SO answer I can get multiple tasks working in parallel in RxPy, my problem is how do you wait for them to all complete? I know using threading I can do .join()
but there doesn't seem to be any such option with Rx Schedulers. .to_blocking()
doesn't help either, the MainThread completes before all notifications have fired and the complete handler has been called. Here's an example:
from __future__ import print_function
import os, sys
import time
import random
from rx import Observable
from rx.core import Scheduler
from threading import current_thread
def printthread(val):
print("{}, thread: {}".format(val, current_thread().name))
def intense_calculation(value):
printthread("calc {}".format(value))
time.sleep(random.randint(5, 20) * .1)
return value
if __name__ == "__main__":
Observable.range(1, 3) \
.select_many(lambda i: Observable.start(lambda: intense_calculation(i), scheduler=Scheduler.timeout)) \
.observe_on(Scheduler.event_loop) \
.subscribe(
on_next=lambda x: printthread("on_next: {}".format(x)),
on_completed=lambda: printthread("on_completed"),
on_error=lambda err: printthread("on_error: {}".format(err)))
printthread("\nAll done")
# time.sleep(2)
calc 1, thread: Thread-1
calc 2, thread: Thread-2
calc 3, thread: Thread-3
on_next: 2, thread: Thread-4
on_next: 3, thread: Thread-4
on_next: 1, thread: Thread-4
on_completed, thread: Thread-4
All done, thread: MainThread
calc 1, thread: Thread-1
calc 2, thread: Thread-2
calc 3, thread: Thread-3
All done, thread: MainThread
calc 1, thread: Thread-1
calc 2, thread: Thread-2
calc 3, thread: Thread-3
All done, thread: MainThread
on_next: 2, thread: Thread-4
on_next: 3, thread: Thread-4
on_next: 1, thread: Thread-4
on_completed, thread: Thread-4
Posting complete solution here:
from __future__ import print_function
import os, sys
import time
import random
from rx import Observable
from rx.core import Scheduler
from threading import current_thread
from rx.concurrency import ThreadPoolScheduler
def printthread(val):
print("{}, thread: {}".format(val, current_thread().name))
def intense_calculation(value):
printthread("calc {}".format(value))
time.sleep(random.randint(5, 20) * .1)
return value
if __name__ == "__main__":
scheduler = ThreadPoolScheduler(4)
Observable.range(1, 3) \
.select_many(lambda i: Observable.start(lambda: intense_calculation(i), scheduler=scheduler)) \
.observe_on(Scheduler.event_loop) \
.subscribe(
on_next=lambda x: printthread("on_next: {}".format(x)),
on_completed=lambda: printthread("on_completed"),
on_error=lambda err: printthread("on_error: {}".format(err)))
printthread("\nAll done")
scheduler.executor.shutdown()
# time.sleep(2)
For ThreadPoolScheduler
, you can:
scheduler.executor.shutdown()
then, you can get all results once all are done.
Use run()
to wait for RxPy parallel threads to complete.
BlockingObservables have been removed from RxPY v3.
from threading import current_thread
import rx, random, multiprocessing, time
from rx import operators as ops
def intense_calculation(value):
delay = random.randint(5, 20) * 0.2
time.sleep(delay)
print("From adding_delay: {0} Value : {1} {2}".format(current_thread(), value, delay))
return (value[0], value[1]+ " processed")
thread_pool_scheduler = rx.scheduler.NewThreadScheduler()
my_dict={'A':'url1', 'B':'url2', 'C':'url3'}
new_dict = rx.from_iterable(my_dict.items()).pipe(
ops.flat_map(lambda a: rx.of(a).pipe(
ops.map(lambda a: intense_calculation(a)),
ops.subscribe_on(thread_pool_scheduler)
)),
ops.to_dict(lambda x: x[0], lambda x: x[1])
).run()
print("From main: {0}".format(current_thread()))
print(str(new_dict))
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With