Repository: dask/dask
File: dask/blockwise.py (Python)

Code before:
                    continue
                if (
                    sum(k == dep for k, ind in layers[layer].indices if ind is not None)
                    > 1
                ):
                    stack.append(dep)
                    continue
                if blockwise_layers and not _can_fuse_annotations(
                    layers[next(iter(blockwise_layers))].annotations,
                    layers[dep].annotations,
                ):
                    stack.append(dep)
                    continue

                # passed everything, proceed
                blockwise_layers.add(dep)

                # traverse further to this child's children
                output_indices = set(layers[dep].output_indices)
                input_indices = {
                    i for _, ind in layers[dep].indices if ind for i in ind
                }
                is_io_superset = output_indices.issuperset(input_indices)
                for d in full_graph.dependencies.get(dep, ()):
                    # Don't allow reductions to proceed
                    if len(dependents[d]) <= 1 and is_io_superset:
                        deps.add(d)
                    else:
                        stack.append(d)

            # Merge these Blockwise layers into one
            new_layer = rewrite_blockwise([layers[l] for l in blockwise_layers])
            out[layer] = new_layer

            # Get the new (external) dependencies for this layer.
            # This corresponds to the dependencies defined in
            # full_graph.dependencies and are not in blockwise_layers
            new_deps = set()
            for l in blockwise_layers:
                new_deps |= set(
                    {
                        d
                        for d in full_graph.dependencies[l]
                        if d not in blockwise_layers and d in full_graph.dependencies
                    }
                )
            for k, v in new_layer.indices:
                if v is None:
                    new_deps |= keys_in_tasks(full_graph.dependencies, [k])
                elif k not in io_names:
                    new_deps.add(k)

Review comment (suggestion):
perhaps invert the order?
```suggestion
                    if is_io_superset and len(dependents[d]) <= 1:
```


Code after:
                    continue
                if (
                    sum(k == dep for k, ind in layers[layer].indices if ind is not None)
                    > 1
                ):
                    stack.append(dep)
                    continue
                if blockwise_layers and not _can_fuse_annotations(
                    layers[next(iter(blockwise_layers))].annotations,
                    layers[dep].annotations,
                ):
                    stack.append(dep)
                    continue

                # passed everything, proceed
                blockwise_layers.add(dep)

                # traverse further to this child's children
                output_indices = set(layers[dep].output_indices)
                input_indices = {
                    i for _, ind in layers[dep].indices if ind for i in ind
                }
                is_io_superset = output_indices.issuperset(input_indices)
                for d in full_graph.dependencies.get(dep, ()):
                    # Don't allow reductions to proceed
                    if is_io_superset and len(dependents[d]) <= 1:
                        deps.add(d)
                    else:
                        stack.append(d)

            # Merge these Blockwise layers into one
            new_layer = rewrite_blockwise([layers[l] for l in blockwise_layers])
            out[layer] = new_layer

            # Get the new (external) dependencies for this layer.
            # This corresponds to the dependencies defined in
            # full_graph.dependencies and are not in blockwise_layers
            new_deps = set()
            for l in blockwise_layers:
                new_deps |= set(
                    {
                        d
                        for d in full_graph.dependencies[l]
                        if d not in blockwise_layers and d in full_graph.dependencies
                    }
                )
            for k, v in new_layer.indices:
                if v is None:
                    new_deps |= keys_in_tasks(full_graph.dependencies, [k])
                elif k not in io_names:
                    new_deps.add(k)