parallel running

[acse-yw11823]

Hello, I am currently trying the parallel running referred to the tutorial https://www.devitoproject.org/examples/seismic/tutorials/04_dask.html. But it occurs the following error:

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Cell In[14], [line 86](vscode-notebook-cell:?execution_count=14&line=86)
     [83](vscode-notebook-cell:?execution_count=14&line=83)     return shots
     [85](vscode-notebook-cell:?execution_count=14&line=85) # Compute observed data in parallel
---> [86](vscode-notebook-cell:?execution_count=14&line=86) d_obs = forward_modeling_multi_shots(model1, geometry1)
     [88](vscode-notebook-cell:?execution_count=14&line=88) # Print the results
     [89](vscode-notebook-cell:?execution_count=14&line=89) for i, shot in enumerate(d_obs):

Cell In[14], [line 81](vscode-notebook-cell:?execution_count=14&line=81)
     [79](vscode-notebook-cell:?execution_count=14&line=79) shots = []
     [80](vscode-notebook-cell:?execution_count=14&line=80) for future in futures:
---> [81](vscode-notebook-cell:?execution_count=14&line=81)     shots.append(future.result()[0])
     [83](vscode-notebook-cell:?execution_count=14&line=83) return shots

File /opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/distributed/client.py:319, in Future.result(self, timeout)
    [317](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/distributed/client.py:317) if self.status == "error":
    [318](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/distributed/client.py:318)     typ, exc, tb = result
--> [319](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/distributed/client.py:319)     raise exc.with_traceback(tb)
    [320](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/distributed/client.py:320) elif self.status == "cancelled":
    [321](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/distributed/client.py:321)     raise result

Cell In[14], [line 57](vscode-notebook-cell:?execution_count=14&line=57)
     [54](vscode-notebook-cell:?execution_count=14&line=54)     raise ValueError("rec_term is None")
     [56](vscode-notebook-cell:?execution_count=14&line=56) # 创建操作符并执行
...
--> [389](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/devito/passes/clusters/aliases.py:389)     if isinstance(expr, EvalDerivative) and not expr.base.is_Function:
    [390](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/devito/passes/clusters/aliases.py:390)         return expr.args
    [391](https://file+.vscode-resource.vscode-cdn.net/opt/homebrew/Caskroom/miniconda/base/envs/devito/lib/python3.8/site-packages/devito/passes/clusters/aliases.py:391)     else:

AttributeError: 'NoneType' object has no attribute 'is_Function'
Output is truncated. View as a [scrollable element](command:cellOutput.enableScrolling?aa31f896-4d68-4485-b228-cab2725ecc45) or open in a [text editor](command:workbench.action.openLargeOutput?aa31f896-4d68-4485-b228-cab2725ecc45). Adjust cell output [settings](command:workbench.action.openSettings?%5B%22%40tag%3AnotebookOutputLayout%22%5D)...

Here is my code:

# Define true and initial model
shape = (501, 501)  # Number of grid point (nx, nz)
spacing = (2., 2.)  # Grid spacing in m. The domain size is now 1km by 1km (lx,ly)
origin = (0., 0.)  # Need origin to define relative source and receiver locations

#Set the number of boundary layers 
nbl = 200

v = np.full(shape, 2.5, dtype=np.float32)  # 全部初始化为2.5 km/s
# 定义圆的中心和半径
center_x, center_y = shape[0] // 2, shape[1] // 2  # 圆心位置
radius = 150  # 半径为50格

# 遍历每个点，如果点在圆内，则设置速度为3 km/s
for x in range(shape[0]):
    for y in range(shape[1]):
        if (x - center_x) ** 2 + (y - center_y) ** 2 <= radius ** 2:
            v[x, y] = 3.0

#True Vp model
model1 = Model(vp=v, origin=origin, shape=shape, spacing=spacing, space_order=2, nbl=nbl, bcs="damp")

#Smoothed Vp model
v1 = np.full(shape, 2.5, dtype=np.float32)  # 全部初始化为2.5 km/s
model0 = Model(vp=v1, origin=origin, shape=shape, spacing=spacing, space_order=2, nbl=nbl, bcs="damp")

# Set up source geometry, but define 5 sources instead of just one.
nsources = 5
src_coordinates = np.empty((nsources, 2))
src_coordinates[:, 1] = np.linspace(0, model1.domain_size[0], num=nsources)
src_coordinates[:, 0] = 20.  # Source depth is 20m

# Initialize receivers for synthetic and imaging data
nreceivers = 101
rec_coordinates = np.empty((nreceivers, 2))
rec_coordinates[:, 1] = np.linspace(0, model1.domain_size[0], num=nreceivers)
rec_coordinates[:, 0] = 980.

# Create the Geometry
t0 = 0.
tn = 1000. # 1000ms
f0 = 0.010 # 10Hz (=0.01 kHz)
dt = model1.critical_dt
time = TimeAxis(start=t0, stop=tn, step=dt)
# Set up geometry objects for observed and predicted data
geometry1 = AcquisitionGeometry(model1, rec_coordinates, src_coordinates, t0, tn, f0=f0, src_type='Ricker')
geometry0 = AcquisitionGeometry(model0, rec_coordinates, src_coordinates, t0, tn, f0=f0, src_type='Ricker')

# Define the wavefield
u = TimeFunction(name="u", grid=model1.grid, time_order=2, space_order=2, save=time.num)

# PDE equation
pde = u.dt2 - (1/model1.m) * u.laplace + (1/model1.m) * model1.damp * u.dt
stencil = Eq(u.forward, solve(pde, u.forward))

# Serial modeling function
def forward_modeling_single_shot(model, geometry):
    # solver = AcousticWaveSolver(model, geometry, space_order=4)
    # d_obs, u0 = solver.forward(vp=model.vp, save=save)[0:2]
    # 创建 Ricker 源并注入波场
    src = geometry.src
    src_term = src.inject(field=u.forward, expr=src * dt**2 * model.m)

    # 创建接收器并插值波场
    rec = Receiver(name='rec', npoint=nreceivers, grid=model.grid, coordinates=rec_coordinates, time_range=time)
    rec_term = rec.interpolate(expr=u.forward)

    # 创建操作符并执行
    op_fwd = Operator([stencil] + src_term + rec_term)
    op_fwd(dt= dt)
    
    # 获取接收器数据
    rec_data = rec.data * (1/(model.spacing[0]**2))
    # print("Receiver data:")
    # print(rec_data)
    # print("Receiver data shape:", rec.data.shape)
    
    u_data = u.data
    # print("Wavefield data:")
    # print(u_data)
    # print("Wavefield data shape:", u_data.shape)

    return rec_data, u_data

# Parallel modeling function
def forward_modeling_multi_shots(model, geometry):

    futures = []
    for i in range(geometry.nsrc):

        # Geometry for current shot
        geometry_i = AcquisitionGeometry(model, geometry.rec_positions, geometry.src_positions[i,:], 
            geometry.t0, geometry.tn, f0=geometry.f0, src_type=geometry.src_type)
        
        # Call serial modeling function for each index
        futures.append(client.submit(forward_modeling_single_shot, model, geometry_i))

    # Wait for all workers to finish and collect shots
    wait(futures)
    shots = []
    for i in range(geometry.nsrc):
        shots.append(futures[i].result()[0])

    return shots

from distributed import Client, wait
USE_GPU_AWARE_DASK = False

# Start Dask cluster
if USE_GPU_AWARE_DASK:
    from dask_cuda import LocalCUDACluster
    cluster = LocalCUDACluster(threads_per_worker=1, death_timeout=600) 
else:
    from distributed import LocalCluster
    cluster = LocalCluster(n_workers=nsources, death_timeout=600)
    
client = Client(cluster)

# Compute observed data in parallel (inverse crime). In real life we would read the SEG-Y data here.
d_obs = forward_modeling_multi_shots(model1, geometry1)

Does anyone have a clue about it? Many thanks for the helping!