Synchronous and asynchronous UDF execution

This notebook demonstrates the following features that were introduced in release 0.7.0:

  • Execute several UDFs in one pass

  • Obtain intermediate results from each merge step by executing UDFs as an iterator

  • Execute UDFs asynchronously

Please see example live-plotting.ipynb for the related live plotting feature!

[1]:

import os
import pprint
import asyncio
import copy

import numpy as np

import libertem.api as lt
from libertem.udf.sum import SumUDF
from libertem.udf.sumsigudf import SumSigUDF

[2]:

ctx = lt.Context()

/home/weber/miniconda3/envs/libertem39/lib/python3.9/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.
Perhaps you already have a cluster running?
Hosting the HTTP server on port 40445 instead
  warnings.warn(
2023-01-12 16:17:08,313 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=15)
2023-01-12 16:17:08,313 - distributed.utils - INFO - Reload module tmpioryzcko from .py file
2023-01-12 16:17:08,323 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=16)
2023-01-12 16:17:08,323 - distributed.utils - INFO - Reload module tmph8_w2ycc from .py file
2023-01-12 16:17:08,338 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=21)
2023-01-12 16:17:08,339 - distributed.utils - INFO - Reload module tmp7qw173dt from .py file
2023-01-12 16:17:08,354 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=18)
2023-01-12 16:17:08,354 - distributed.utils - INFO - Reload module tmp5qsl_mgf from .py file
2023-01-12 16:17:08,359 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=11)
2023-01-12 16:17:08,360 - distributed.utils - INFO - Reload module tmpg9shv96a from .py file
2023-01-12 16:17:08,374 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=7)
2023-01-12 16:17:08,375 - distributed.utils - INFO - Reload module tmpk38e72ru from .py file
2023-01-12 16:17:08,387 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=19)
2023-01-12 16:17:08,388 - distributed.utils - INFO - Reload module tmp5k8ln3d6 from .py file
2023-01-12 16:17:08,403 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=8)
2023-01-12 16:17:08,403 - distributed.utils - INFO - Reload module tmpciaau8cj from .py file
2023-01-12 16:17:08,416 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=13)
2023-01-12 16:17:08,416 - distributed.utils - INFO - Reload module tmph4ri158_ from .py file
2023-01-12 16:17:08,439 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=9)
2023-01-12 16:17:08,440 - distributed.utils - INFO - Reload module tmp8n6u_f8r from .py file
2023-01-12 16:17:08,454 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=23)
2023-01-12 16:17:08,455 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=4)
2023-01-12 16:17:08,455 - distributed.utils - INFO - Reload module tmpu4b_ap_8 from .py file
2023-01-12 16:17:08,455 - distributed.utils - INFO - Reload module tmp7briwqsy from .py file
2023-01-12 16:17:08,488 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=1)
2023-01-12 16:17:08,488 - distributed.utils - INFO - Reload module tmp28hfv12y from .py file
2023-01-12 16:17:08,492 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=17)
2023-01-12 16:17:08,492 - distributed.utils - INFO - Reload module tmp91gia771 from .py file
2023-01-12 16:17:08,510 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=10)
2023-01-12 16:17:08,511 - distributed.utils - INFO - Reload module tmph1cyy14q from .py file
2023-01-12 16:17:08,531 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=22)
2023-01-12 16:17:08,531 - distributed.utils - INFO - Reload module tmpmp21k6t_ from .py file
2023-01-12 16:17:08,533 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-service-0', service_id='0')
2023-01-12 16:17:08,533 - distributed.utils - INFO - Reload module tmp__q7et61 from .py file
2023-01-12 16:17:08,557 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=6)
2023-01-12 16:17:08,558 - distributed.utils - INFO - Reload module tmpagqz6y9z from .py file
2023-01-12 16:17:08,559 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=12)
2023-01-12 16:17:08,559 - distributed.utils - INFO - Reload module tmpgifm6cp1 from .py file
2023-01-12 16:17:08,569 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=5)
2023-01-12 16:17:08,570 - distributed.utils - INFO - Reload module tmpwzgbq6em from .py file
2023-01-12 16:17:08,597 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=3)
2023-01-12 16:17:08,598 - distributed.utils - INFO - Reload module tmp3cyljleg from .py file
2023-01-12 16:17:08,600 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CUDA", device=0)
2023-01-12 16:17:08,600 - distributed.utils - INFO - Reload module tmp2tfzjdmu from .py file
2023-01-12 16:17:08,606 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=14)
2023-01-12 16:17:08,607 - distributed.utils - INFO - Reload module tmpw32ydyaz from .py file
2023-01-12 16:17:08,613 - distributed.preloading - INFO - Import preload module: /tmp/tmph8_w2ycc.py
2023-01-12 16:17:08,613 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-16', service_id='16')
2023-01-12 16:17:08,613 - distributed.utils - INFO - Reload module tmpw_3r5ys8 from .py file
2023-01-12 16:17:08,614 - distributed.preloading - INFO - Import preload module: /tmp/tmpw_3r5ys8.py
2023-01-12 16:17:08,614 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,617 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=0)
2023-01-12 16:17:08,617 - distributed.utils - INFO - Reload module tmpmf74ytxf from .py file
2023-01-12 16:17:08,639 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=20)
2023-01-12 16:17:08,639 - distributed.utils - INFO - Reload module tmpg8_h6muw from .py file
2023-01-12 16:17:08,652 - distributed.preloading - INFO - Import preload module: /tmp/tmp5qsl_mgf.py
2023-01-12 16:17:08,652 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-18', service_id='18')
2023-01-12 16:17:08,653 - distributed.utils - INFO - Reload module tmps3w38vgu from .py file
2023-01-12 16:17:08,653 - distributed.preloading - INFO - Import preload module: /tmp/tmps3w38vgu.py
2023-01-12 16:17:08,653 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,659 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,666 - distributed.preloading - INFO - Import preload module: /tmp/tmpg9shv96a.py
2023-01-12 16:17:08,666 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-11', service_id='11')
2023-01-12 16:17:08,666 - distributed.utils - INFO - Reload module tmpue1fwqm_ from .py file
2023-01-12 16:17:08,667 - distributed.preloading - INFO - Import preload module: /tmp/tmpue1fwqm_.py
2023-01-12 16:17:08,667 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,668 - distributed.preloading - INFO - Import preload module: /tmp/tmpioryzcko.py
2023-01-12 16:17:08,669 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-15', service_id='15')
2023-01-12 16:17:08,669 - distributed.utils - INFO - Reload module tmp9dg3oypu from .py file
2023-01-12 16:17:08,669 - distributed.preloading - INFO - Import preload module: /tmp/tmp9dg3oypu.py
2023-01-12 16:17:08,669 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,673 - distributed.preloading - INFO - Import preload module: /tmp/tmpk38e72ru.py
2023-01-12 16:17:08,674 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-7', service_id='7')
2023-01-12 16:17:08,674 - distributed.preloading - INFO - Import preload module: /tmp/tmp7qw173dt.py
2023-01-12 16:17:08,674 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-21', service_id='21')
2023-01-12 16:17:08,674 - distributed.utils - INFO - Reload module tmp2qlo6kj1 from .py file
2023-01-12 16:17:08,674 - distributed.utils - INFO - Reload module tmpmrnqf556 from .py file
2023-01-12 16:17:08,674 - distributed.preloading - INFO - Import preload module: /tmp/tmp2qlo6kj1.py
2023-01-12 16:17:08,674 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,674 - distributed.preloading - INFO - Import preload module: /tmp/tmpmrnqf556.py
2023-01-12 16:17:08,675 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,682 - distributed.preloading - INFO - Import preload module: /tmp/tmp5k8ln3d6.py
2023-01-12 16:17:08,682 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-19', service_id='19')
2023-01-12 16:17:08,682 - distributed.utils - INFO - Reload module tmp1jzei7i6 from .py file
2023-01-12 16:17:08,682 - distributed.preloading - INFO - Import preload module: /tmp/tmp1jzei7i6.py
2023-01-12 16:17:08,682 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,690 - distributed.preloading - INFO - Creating preload: from libertem.executor.dask import worker_setup; worker_setup(resource="CPU", device=2)
2023-01-12 16:17:08,691 - distributed.utils - INFO - Reload module tmpcv7bxv_2 from .py file
2023-01-12 16:17:08,701 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,705 - distributed.preloading - INFO - Import preload module: /tmp/tmph4ri158_.py
2023-01-12 16:17:08,705 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-13', service_id='13')
2023-01-12 16:17:08,705 - distributed.utils - INFO - Reload module tmpieybz66v from .py file
2023-01-12 16:17:08,706 - distributed.preloading - INFO - Import preload module: /tmp/tmpieybz66v.py
2023-01-12 16:17:08,706 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,714 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,716 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,719 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,720 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,728 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,734 - distributed.preloading - INFO - Import preload module: /tmp/tmpu4b_ap_8.py
2023-01-12 16:17:08,735 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-23', service_id='23')
2023-01-12 16:17:08,735 - distributed.utils - INFO - Reload module tmpvjz5j24f from .py file
2023-01-12 16:17:08,735 - distributed.preloading - INFO - Import preload module: /tmp/tmpvjz5j24f.py
2023-01-12 16:17:08,735 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,745 - distributed.preloading - INFO - Import preload module: /tmp/tmp8n6u_f8r.py
2023-01-12 16:17:08,745 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-9', service_id='9')
2023-01-12 16:17:08,746 - distributed.utils - INFO - Reload module tmpoplyvbei from .py file
2023-01-12 16:17:08,746 - distributed.preloading - INFO - Import preload module: /tmp/tmpoplyvbei.py
2023-01-12 16:17:08,746 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,750 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,765 - distributed.preloading - INFO - Import preload module: /tmp/tmpciaau8cj.py
2023-01-12 16:17:08,765 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-8', service_id='8')
2023-01-12 16:17:08,766 - distributed.utils - INFO - Reload module tmp1kfvaomt from .py file
2023-01-12 16:17:08,766 - distributed.preloading - INFO - Import preload module: /tmp/tmp1kfvaomt.py
2023-01-12 16:17:08,766 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,767 - distributed.preloading - INFO - Import preload module: /tmp/tmp91gia771.py
2023-01-12 16:17:08,767 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-17', service_id='17')
2023-01-12 16:17:08,767 - distributed.utils - INFO - Reload module tmpd_0j1z0u from .py file
2023-01-12 16:17:08,768 - distributed.preloading - INFO - Import preload module: /tmp/tmpd_0j1z0u.py
2023-01-12 16:17:08,768 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,781 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,788 - distributed.preloading - INFO - Import preload module: /tmp/tmp__q7et61.py
2023-01-12 16:17:08,788 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,791 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,813 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,813 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,820 - distributed.preloading - INFO - Import preload module: /tmp/tmph1cyy14q.py
2023-01-12 16:17:08,820 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-10', service_id='10')
2023-01-12 16:17:08,821 - distributed.utils - INFO - Reload module tmpg69_e1ll from .py file
2023-01-12 16:17:08,821 - distributed.preloading - INFO - Import preload module: /tmp/tmpg69_e1ll.py
2023-01-12 16:17:08,821 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,826 - distributed.preloading - INFO - Import preload module: /tmp/tmp7briwqsy.py
2023-01-12 16:17:08,827 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-4', service_id='4')
2023-01-12 16:17:08,827 - distributed.utils - INFO - Reload module tmps67qqebg from .py file
2023-01-12 16:17:08,828 - distributed.preloading - INFO - Import preload module: /tmp/tmps67qqebg.py
2023-01-12 16:17:08,828 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,841 - distributed.preloading - INFO - Import preload module: /tmp/tmpwzgbq6em.py
2023-01-12 16:17:08,841 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-5', service_id='5')
2023-01-12 16:17:08,841 - distributed.utils - INFO - Reload module tmpgtd36kxl from .py file
2023-01-12 16:17:08,842 - distributed.preloading - INFO - Import preload module: /tmp/tmpgtd36kxl.py
2023-01-12 16:17:08,842 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,845 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,848 - distributed.preloading - INFO - Import preload module: /tmp/tmpmp21k6t_.py
2023-01-12 16:17:08,848 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-22', service_id='22')
2023-01-12 16:17:08,849 - distributed.utils - INFO - Reload module tmpsrlhukl0 from .py file
2023-01-12 16:17:08,849 - distributed.preloading - INFO - Import preload module: /tmp/tmpsrlhukl0.py
2023-01-12 16:17:08,849 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,857 - distributed.preloading - INFO - Import preload module: /tmp/tmpgifm6cp1.py
2023-01-12 16:17:08,857 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-12', service_id='12')
2023-01-12 16:17:08,857 - distributed.utils - INFO - Reload module tmpjguc39bg from .py file
2023-01-12 16:17:08,858 - distributed.preloading - INFO - Import preload module: /tmp/tmpjguc39bg.py
2023-01-12 16:17:08,858 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,860 - distributed.preloading - INFO - Import preload module: /tmp/tmpagqz6y9z.py
2023-01-12 16:17:08,860 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-6', service_id='6')
2023-01-12 16:17:08,861 - distributed.utils - INFO - Reload module tmp5394248q from .py file
2023-01-12 16:17:08,861 - distributed.preloading - INFO - Import preload module: /tmp/tmp5394248q.py
2023-01-12 16:17:08,861 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,866 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,880 - distributed.preloading - INFO - Import preload module: /tmp/tmp3cyljleg.py
2023-01-12 16:17:08,880 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-3', service_id='3')
2023-01-12 16:17:08,881 - distributed.utils - INFO - Reload module tmphha6hli_ from .py file
2023-01-12 16:17:08,881 - distributed.preloading - INFO - Import preload module: /tmp/tmphha6hli_.py
2023-01-12 16:17:08,881 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,886 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,889 - distributed.preloading - INFO - Import preload module: /tmp/tmp2tfzjdmu.py
2023-01-12 16:17:08,889 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cuda-0', service_id='0')
2023-01-12 16:17:08,890 - distributed.utils - INFO - Reload module tmpvs6mftm5 from .py file
2023-01-12 16:17:08,890 - distributed.preloading - INFO - Import preload module: /tmp/tmpvs6mftm5.py
2023-01-12 16:17:08,890 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,891 - distributed.preloading - INFO - Import preload module: /tmp/tmpmf74ytxf.py
2023-01-12 16:17:08,891 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-0', service_id='0')
2023-01-12 16:17:08,891 - distributed.utils - INFO - Reload module tmptqa4lwge from .py file
2023-01-12 16:17:08,892 - distributed.preloading - INFO - Import preload module: /tmp/tmptqa4lwge.py
2023-01-12 16:17:08,892 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,894 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,896 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,897 - distributed.preloading - INFO - Import preload module: /tmp/tmp28hfv12y.py
2023-01-12 16:17:08,897 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-1', service_id='1')
2023-01-12 16:17:08,898 - distributed.utils - INFO - Reload module tmp2dwo7__c from .py file
2023-01-12 16:17:08,898 - distributed.preloading - INFO - Import preload module: /tmp/tmp2dwo7__c.py
2023-01-12 16:17:08,898 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,903 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,907 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,909 - distributed.preloading - INFO - Import preload module: /tmp/tmpg8_h6muw.py
2023-01-12 16:17:08,909 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-20', service_id='20')
2023-01-12 16:17:08,909 - distributed.utils - INFO - Reload module tmpo8kbgqp1 from .py file
2023-01-12 16:17:08,909 - distributed.preloading - INFO - Import preload module: /tmp/tmpo8kbgqp1.py
2023-01-12 16:17:08,909 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,913 - distributed.preloading - INFO - Import preload module: /tmp/tmpw32ydyaz.py
2023-01-12 16:17:08,913 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-14', service_id='14')
2023-01-12 16:17:08,914 - distributed.utils - INFO - Reload module tmpuwbyt0sh from .py file
2023-01-12 16:17:08,914 - distributed.preloading - INFO - Import preload module: /tmp/tmpuwbyt0sh.py
2023-01-12 16:17:08,914 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:08,927 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,937 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,945 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,945 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,953 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,958 - distributed.preloading - INFO - Import preload module: libertem.preload
2023-01-12 16:17:08,964 - distributed.preloading - INFO - Import preload module: /tmp/tmpcv7bxv_2.py
2023-01-12 16:17:08,964 - distributed.preloading - INFO - Creating preload: from libertem.common.tracing import maybe_setup_tracing; maybe_setup_tracing(service_name='default-cpu-2', service_id='2')
2023-01-12 16:17:08,965 - distributed.utils - INFO - Reload module tmpc1f0r09y from .py file
2023-01-12 16:17:08,965 - distributed.preloading - INFO - Import preload module: /tmp/tmpc1f0r09y.py
2023-01-12 16:17:08,965 - distributed.preloading - INFO - Creating preload: libertem.preload
2023-01-12 16:17:09,010 - distributed.preloading - INFO - Import preload module: libertem.preload

[3]:

data_base_path = os.environ.get("TESTDATA_BASE_PATH", "/home/alex/Data/")

[4]:

ds = ctx.load("auto", path=os.path.join(data_base_path, "20200518 165148/default.hdr"))

2023-01-12 16:17:09,425 - distributed.worker - WARNING - Compute Failed
Key:       _do_detect-fa956156-516c-43c3-a192-5484c489a9e2
Function:  _do_detect
args:      ()
kwargs:    {}
Exception: 'DataSetException("OSError(\'Unable to open file (file signature not found)\')")'


[5]:

udfs = [SumUDF(), SumSigUDF()]

Synchronous execution, only result

Note that both UDFs are executed in a single pass!

[6]:

res = ctx.run_udf(dataset=ds, udf=udfs)

The result is a tuple with one entry per UDF:

[7]:

pprint.pprint(res)

({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})

The previous API when passing a single UDF is not changed, i.e. it doesn’t return a tuple but a single UDF result

[8]:

res = ctx.run_udf(dataset=ds, udf=udfs[0])

[9]:

pprint.pprint(res)

{'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>}

Asynchronous execution, only result

By setting sync=False, the result is awaitable:

[10]:

async_res = ctx.run_udf(dataset=ds, udf=udfs, sync=False)
print("Do something else while UDFs are running in the background")
res = await async_res
print("Finished")

Do something else while UDFs are running in the background
Finished

[11]:

pprint.pprint(res)

({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})

Just like in the synchronous case, running a single UDF returns the UDF result directly, not a tuple:

[12]:

async_res = ctx.run_udf(dataset=ds, udf=udfs[0], sync=False)
print("Do something else while UDF is running in the background")
res = await async_res
print("Finished")

Do something else while UDF is running in the background
Finished

[13]:

pprint.pprint(res)

{'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>}

Synchronous execution as an iterator

This returns UDFResults objects with attributes buffers and damage. buffers is a tuple with the results per UDF, and damage is a BufferWrapper with kind='nav' and dtype=bool that indicates which parts of the navigation space have been merged already.

[14]:

# NBVAL_IGNORE_OUTPUT
# (output is ignored in nbval run because the number of nav positions can be different)
for res in ctx.run_udf_iter(dataset=ds, udf=udfs):
    print(np.count_nonzero(res.damage.data), "nav positions processed")
    pprint.pprint(res.buffers)

683 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
1366 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
2048 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
2731 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
3414 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
4096 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
4779 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
5461 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
6143 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
6826 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
7509 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
8191 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
8874 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
9557 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
10240 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
10923 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
11606 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
12289 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
12971 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
13654 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
14337 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
15020 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
15702 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})
16384 nav positions processed
({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
 {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>})

Asynchronous execution as an iterator

This allows several iterators to proceed asynchronously in parallel. This approach is used in the backend for the web GUI of LiberTEM to run several analyses concurrently. It could also be useful to implement live feedback to instrument control from UDF results if the control solution works asynchronously.

Note that the UDFs are copied here so that different instances are executed in parallel. Executing the same UDF instances concurrently can lead to undefined behavior.

[15]:

# NBVAL_IGNORE_OUTPUT
# (output is ignored in nbval run because the number of nav positions can be different)
async def doit(label, udfs):
    async for res in ctx.run_udf_iter(dataset=ds, udf=udfs, sync=False):
        print(label, np.count_nonzero(res.damage.data), "nav positions processed")
        pprint.pprint((label, res.buffers))
    return res

p1 = doit("one", copy.deepcopy(udfs))
p2 = doit("two", copy.deepcopy(udfs))
print("Do something else while UDFs are running in the background")
await asyncio.gather(p1, p2)

Do something else while UDFs are running in the background
one 682 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 1365 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 2048 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 2731 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 3414 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 4097 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 4779 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 5462 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 6145 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 6828 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 7510 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 8192 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 8875 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 9558 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 10240 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 10923 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 11606 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 12289 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 12972 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 13654 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 14336 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 15019 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 682 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 15702 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
one 16384 nav positions processed
('one',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 1365 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 2048 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 2730 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 3413 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 4096 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 4778 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 5461 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 6144 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 6827 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 7509 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 8192 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 8875 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 9557 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 10240 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 10923 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 11606 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 12289 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 12971 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 13654 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 14336 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 15018 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 15701 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))
two 16384 nav positions processed
('two',
 ({'intensity': <BufferWrapper kind=sig dtype=float32 extra_shape=()>},
  {'intensity': <BufferWrapper kind=nav dtype=float32 extra_shape=()>}))

[15]:

[<libertem.udf.base.UDFResults at 0x7f01744ebe50>,
 <libertem.udf.base.UDFResults at 0x7f01770e5f70>]

[ ]: