使用Dask并行化优美树分解DP图同态计数算法遇TypeError求助

问题背景 #

我在Sage/Python中实现了基于优美树分解的动态规划图同态计数算法，用于计算图G到图H的同态数量。串行版本已完成，现在尝试用Dask并行化（此前使用concurrent.futures时遇到pickle序列化问题）。需要注意的是，该DP算法存在严格依赖：父节点的计算必须在其子节点全部完成后才能执行。

当前并行代码 #

def process_node(self, node):
    node_type = self.dir_labelled_TD.get_vertex(node)
    match node_type:
        case 'intro':
            result = self._add_intro_node_parallel(node)
        case 'forget':
            result = self._add_forget_node_parallel(node)
        case 'join':
            result = self._add_join_node_parallel(node)
        case _:
            result = self._add_leaf_node_parallel(node)

    node_index = get_node_index(node)
    self.DP_table[node_index] = result
    return result

def count_homomorphisms_parallel(self):
    # Dictionary to store all futures/promises
    self.futures = {}
    for node in reversed(self.dir_labelled_TD.vertices()):
        # Delaying each node process and storing in futures
        self.futures[node] = self.process_node(node)

    print("Futures: ", self.futures)

    # Compute all results, respecting the inherent dependencies among them
    results = dask.compute(*self.futures.values())
    print("Results: ", [f.compute() for f in results])
    return self.DP_table[0][0]

运行示例 #

par_counter = ParallelGraphHomomorphismCounter(graph, three_grid)
par_count = par_counter.count_homomorphisms_parallel()
print(par_count)

报错信息 #

Futures:  {(6, {}): Delayed('process_node-0b571dcd-00e5-4871-871a-ef52e16b4ffb'), (5, {2}): Delayed('process_node-0fbc0886-3368-4d0e-8b09-751cce606ffe'), (4, {0, 2}): Delayed('process_node-0187a2da-aba7-42f7-83ab-497f62ea6b1f'), (3, {0}): Delayed('process_node-18729eea-99f4-45ee-af15-0de45395f181'), (2, {0, 1}): Delayed('process_node-8c85c333-301d-4e49-b9d4-c01bc20c05ae'), (1, {0}): Delayed('process_node-7e528db6-4636-4b31-8eb1-6f807ac32627'), (0, {}): Delayed('process_node-6bb90670-11b4-4e40-b7bc-cefb2fef6479')}

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In [3], line 28
     27 par_counter = ParallelGraphHomomorphismCounter(graph, three_grid)
---> 28 par_count = par_counter.count_homomorphisms_parallel()
     29 print(par_count)

File ~/github/local-hom-count/local_hom_count_best_parallel.py:118, in ParallelGraphHomomorphismCounter.count_homomorphisms_parallel(self)
    117 results = dask.compute(*self.futures.values())
--> 118 print("Results: ", [f.compute() for f in results])
    121 return self.DP_table[0][0]

File ~/github/local-hom-count/local_hom_count_best_parallel.py:118, in <listcomp>(.0)
    117 results = dask.compute(*self.futures.values())
--> 118 print("Results: ", [f.compute() for f in results])

File ~/.sage/local/lib/python3.11/site-packages/dask/base.py:375, in DaskMethodsMixin.compute(self, **kwargs)
--> 375     (result,) = compute(self, traverse=False, **kwargs)
    376     return result

File ~/.sage/local/lib/python3.11/site-packages/dask/base.py:661, in compute(traverse, optimize_graph, scheduler, get, *args, **kwargs)
    660 with shorten_traceback():
--> 661     results = schedule(dsk, keys, **kwargs)

File ~/github/local-hom-count/local_hom_count_best_parallel.py:239, in ParallelGraphHomomorphismCounter._add_intro_node_parallel(self, node)
    235 child_DP_entry = self.DP_table[child_node_index]
--> 239 for mapped in range(len(child_DP_entry)):

File ~/.sage/local/lib/python3.11/site-packages/dask/delayed.py:635, in Delayed.__len__(self)
    634     if self._length is None:
--> 635         raise TypeError("Delayed objects of unspecified length have no len()")
    636     return self._length

TypeError: Delayed objects of unspecified length have no len()

串行版本参考代码 #

def count_homomorphisms_best(self):
    r"""
    Return the number of homomorphisms from the graph `G` to the graph `H`.

    A homomorphism from a graph `G` to a graph `H` is a function
    `\varphi : V(G) \mapsto V(H)`, such that for any edge `uv \in E(G)` the
    pair `\varphi(u) \varphi(v)` is an edge of `H`.

    ALGORITHM:

    This is an implementation based on the proof of Prop. 1.6 in [CDM2017]_.

    OUTPUT:

    - an integer, the number of homomorphisms from `graph` to `target_graph`

    EXAMPLES::

        sage: graph = graphs.CompleteBipartiteGraph(1, 4)
        sage: target_graph = graphs.CompleteGraph(4)
        sage: from sage.graphs.hom_count_best import count_homomorphisms_best
        sage: count_homomorphisms_best(graph, target_graph)
        324
    """
    # Whether it's BFS or DFS, every node below join node(s) would be
    # computed first, so we can safely go bottom-up.
    for node in reversed(self.dir_labelled_TD.vertices()):
        node_type = self.dir_labelled_TD.get_vertex(node)

        match node_type:
            case 'intro':
                self._add_intro_node_best(node)
            case 'forget':
                self._add_forget_node_best(node)
            case 'join':
                self._add_join_node_best(node)

            case _: 
                self._add_leaf_node_best(node)

    return self.DP_table[0][0]

解决方案与优化建议 #

核心问题分析 #

报错的根源是：你直接在延迟执行的函数中访问了self.DP_table里的Delayed对象（子节点的计算结果），并尝试调用len()。Delayed对象是Dask的惰性计算容器，只有实际执行后才会变成真实数据，无法直接操作其长度。同时，在Dask任务中修改实例变量（如self.DP_table）是不可靠的——因为任务可能在不同进程中执行，实例状态不会跨进程共享。

修改后的代码实现 #

from dask import delayed

class ParallelGraphHomomorphismCounter:
    def __init__(self, graph, target_graph):
        self.graph = graph
        self.target_graph = target_graph
        self.dir_labelled_TD = ...  # 你的树分解初始化逻辑
        self.actual_target_size = ...  # 目标图顶点数量
        # 其他初始化参数和串行版本一致

    @delayed
    def _add_leaf_node_parallel(self, node):
        # 完全复用串行版本的叶子节点计算逻辑，返回结果而非修改实例变量
        leaf_vtx = ...  # 获取叶子节点对应的原图顶点
        # 计算叶子节点的DP条目，比如初始映射计数
        result = ...
        return result

    @delayed
    def _add_intro_node_parallel(self, node, child_result):
        # 通过参数接收子节点的计算结果，而非从DP_table读取
        intro_vtx = ...  # 获取当前引入的原图顶点
        intro_vtx_nbhs = ...  # 获取该顶点在原图中的邻居（已在bag中的部分）
        result = {}  # 或你使用的其他DP数据结构
        # 直接操作真实的child_result，而非Delayed对象
        for mapped in range(len(child_result)):
            mapped_intro_nbhs = [extract_bag_vertex(mapped, vtx, self.actual_target_size) for vtx in intro_vtx_nbhs]
            # 后续的映射合法性判断、计数累加逻辑和串行版本一致
            ...
        return result

    @delayed
    def _add_forget_node_parallel(self, node, child_result):
        # 同理，接收子节点结果，计算后返回
        forget_vtx = ...
        result = ...
        return result

    @delayed
    def _add_join_node_parallel(self, node, left_child_result, right_child_result):
        # Join节点需要两个子节点的结果，通过参数传入
        result = ...  # 合并两个子节点的DP条目
        return result

    @delayed
    def process_node(self, node, *child_results):
        node_type = self.dir_labelled_TD.get_vertex(node)
        match node_type:
            case 'intro':
                return self._add_intro_node_parallel(node, child_results[0])
            case 'forget':
                return self._add_forget_node_parallel(node, child_results[0])
            case 'join':
                return self._add_join_node_parallel(node, child_results[0], child_results[1])
            case _:
                return self._add_leaf_node_parallel(node)

    def count_homomorphisms_parallel(self):
        self.futures = {}
        # 先构建树的父子关系映射（假设你的树分解提供获取子节点的方法）
        # 例如：get_children(node) 返回当前节点的所有子节点
        for node in reversed(self.dir_labelled_TD.vertices()):
            children = self.dir_labelled_TD.get_children(node)
            # 获取子节点对应的Delayed任务
            child_futures = [self.futures[child] for child in children]
            # 构建当前节点的Delayed任务，依赖子节点的结果
            self.futures[node] = self.process_node(node, *child_futures)
        
        # 计算根节点的结果（根节点标识根据你的树分解调整）
        root_result = dask.compute(self.futures[(0, {})])[0]
        return root_result

关键修改点 #

1. 用@delayed标记延迟函数：所有需要并行执行的节点处理函数都加上该装饰器，调用时会返回Delayed对象，而非立即执行。
2. 显式传递依赖：子节点的计算结果通过参数传递给父节点的处理函数，Dask会自动追踪依赖关系，确保父节点在子节点完成后执行。
3. 避免修改实例变量：延迟函数不再修改self.DP_table，而是直接返回计算结果，最终通过dask.compute()获取根节点的最终结果。
4. 移除重复计算：原代码中dask.compute()已经返回真实结果，无需再调用f.compute()。

额外优化建议 #

• 确保extract_bag_vertex等辅助函数可序列化，如果是计算密集型操作，也可以用@delayed标记。
• 如果DP条目是大型数据结构，可考虑用Dask的数组或bag类型优化存储与计算。
• 测试时可先使用dask.get作为调度器（本地单进程）验证逻辑正确性，再切换到分布式调度器。

内容的提问来源于stack exchange，提问作者hedgehog0