Returns the sub-dispatcher induced by the workflow from sources.

        The induced sub-dispatcher of the dsp contains the reachable nodes and
        edges evaluated with breadth-first-search on the workflow graph from
        source nodes.

        :param sources:
            Source nodes for the breadth-first-search.
            A container of nodes which will be iterated through once.
        :type sources: list[str], iterable

        :param graph:
            A directed graph where evaluate the breadth-first-search.
        :type graph: networkx.DiGraph, optional

        :param reverse:
            If True the workflow graph is assumed as reversed.
        :type reverse: bool, optional

        :param add_missing:
            If True, missing function' inputs are added to the sub-dispatcher.
        :type add_missing: bool, optional

        :param check_inputs:
            If True the missing function' inputs are not checked.
        :type check_inputs: bool, optional

        :param blockers:
            Nodes to not be added to the queue.
        :type blockers: set[str], iterable, optional

        :param wildcard:
            If True, when the data node is used as input and target in the
            ArciDispatch algorithm, the input value will be used as input for
            the connected functions, but not as output.
        :type wildcard: bool, optional

        :param _update_links:
            If True, it updates remote links of the extracted dispatcher.
        :type _update_links: bool, optional

        :return:
            A sub-dispatcher.
        :rtype: Dispatcher

        .. seealso:: :func:`get_sub_dsp`

        .. note::

            The sub-dispatcher edge or node attributes just point to the
            original dispatcher. So changes to the node or edge structure
            will not be reflected in the original dispatcher map while changes
            to the attributes will.

        **--------------------------------------------------------------------**

        **Example**:

        A dispatcher with a function `fun` and a node `a` with a default value:

        .. dispatcher:: dsp
           :opt: graph_attr={'ratio': '1'}

            >>> dsp = Dispatcher(name='Dispatcher')
            >>> dsp.add_data(data_id='a', default_value=1)
            'a'
            >>> dsp.add_function(function_id='fun1', inputs=['a', 'b'],
            ...                  outputs=['c', 'd'])
            'fun1'
            >>> dsp.add_function(function_id='fun2', inputs=['e'],
            ...                  outputs=['c'])
            'fun2'

        Dispatch with no calls in order to have a workflow::

            >>> o = dsp.dispatch(inputs=['a', 'b'], no_call=True)

        Get sub-dispatcher from workflow inputs `a` and `b`::

            >>> sub_dsp = dsp.get_sub_dsp_from_workflow(['a', 'b'])

        .. dispatcher:: sub_dsp
           :opt: graph_attr={'ratio': '1'}

            >>> sub_dsp.name = 'Sub-Dispatcher'

        Get sub-dispatcher from a workflow output `c`::

            >>> sub_dsp = dsp.get_sub_dsp_from_workflow(['c'], reverse=True)

        .. dispatcher:: sub_dsp
           :opt: graph_attr={'ratio': '1'}

            >>> sub_dsp.name = 'Sub-Dispatcher (reverse workflow)'

def get_sub_dsp_from_workflow(
            self, sources, graph=None, reverse=False, add_missing=False,
            check_inputs=True, blockers=None, wildcard=False,
            _update_links=True):
        """
        Returns the sub-dispatcher induced by the workflow from sources.

        The induced sub-dispatcher of the dsp contains the reachable nodes and
        edges evaluated with breadth-first-search on the workflow graph from
        source nodes.

        :param sources:
            Source nodes for the breadth-first-search.
            A container of nodes which will be iterated through once.
        :type sources: list[str], iterable

        :param graph:
            A directed graph where evaluate the breadth-first-search.
        :type graph: networkx.DiGraph, optional

        :param reverse:
            If True the workflow graph is assumed as reversed.
        :type reverse: bool, optional

        :param add_missing:
            If True, missing function' inputs are added to the sub-dispatcher.
        :type add_missing: bool, optional

        :param check_inputs:
            If True the missing function' inputs are not checked.
        :type check_inputs: bool, optional

        :param blockers:
            Nodes to not be added to the queue.
        :type blockers: set[str], iterable, optional

        :param wildcard:
            If True, when the data node is used as input and target in the
            ArciDispatch algorithm, the input value will be used as input for
            the connected functions, but not as output.
        :type wildcard: bool, optional

        :param _update_links:
            If True, it updates remote links of the extracted dispatcher.
        :type _update_links: bool, optional

        :return:
            A sub-dispatcher.
        :rtype: Dispatcher

        .. seealso:: :func:`get_sub_dsp`

        .. note::

            The sub-dispatcher edge or node attributes just point to the
            original dispatcher. So changes to the node or edge structure
            will not be reflected in the original dispatcher map while changes
            to the attributes will.

        **--------------------------------------------------------------------**

        **Example**:

        A dispatcher with a function `fun` and a node `a` with a default value:

        .. dispatcher:: dsp
           :opt: graph_attr={'ratio': '1'}

            >>> dsp = Dispatcher(name='Dispatcher')
            >>> dsp.add_data(data_id='a', default_value=1)
            'a'
            >>> dsp.add_function(function_id='fun1', inputs=['a', 'b'],
            ...                  outputs=['c', 'd'])
            'fun1'
            >>> dsp.add_function(function_id='fun2', inputs=['e'],
            ...                  outputs=['c'])
            'fun2'

        Dispatch with no calls in order to have a workflow::

            >>> o = dsp.dispatch(inputs=['a', 'b'], no_call=True)

        Get sub-dispatcher from workflow inputs `a` and `b`::

            >>> sub_dsp = dsp.get_sub_dsp_from_workflow(['a', 'b'])

        .. dispatcher:: sub_dsp
           :opt: graph_attr={'ratio': '1'}

            >>> sub_dsp.name = 'Sub-Dispatcher'

        Get sub-dispatcher from a workflow output `c`::

            >>> sub_dsp = dsp.get_sub_dsp_from_workflow(['c'], reverse=True)

        .. dispatcher:: sub_dsp
           :opt: graph_attr={'ratio': '1'}

            >>> sub_dsp.name = 'Sub-Dispatcher (reverse workflow)'
        """

        # Define an empty dispatcher map.
        sub_dsp = self.copy_structure()

        if not graph:  # Set default graph.
            graph = self.solution.workflow

        # Visited nodes used as queue.
        family = {}

        # Namespace shortcuts for speed.
        nodes, dmap_nodes = sub_dsp.dmap._node, self.dmap.nodes
        dlt_val, dsp_dlt_val = sub_dsp.default_values, self.default_values

        if not reverse:
            # Namespace shortcuts for speed.
            neighbors, dmap_succ = graph.neighbors, self.dmap.succ
            succ, pred = sub_dsp.dmap._succ, sub_dsp.dmap._pred

            # noinspection PyUnusedLocal
            def _check_node_inputs(c, p):
                if c == START:
                    return True

                node_attr = dmap_nodes[c]

                if node_attr['type'] == 'function':
                    if set(node_attr['inputs']).issubset(family):
                        _set_node_attr(c)

                        # namespace shortcuts for speed
                        s_pred = pred[c]

                        for p in node_attr['inputs']:
                            # add attributes to both representations of edge
                            succ[p][c] = s_pred[p] = dmap_succ[p][c]
                    elif not check_inputs or add_missing:
                        _set_node_attr(c)

                        # namespace shortcuts for speed
                        s_pred = pred[c]

                        if add_missing:
                            for p in node_attr['inputs']:
                                if p not in family:
                                    _set_node_attr(p, add2family=False)
                                    succ[p][c] = s_pred[p] = dmap_succ[p][c]

                        for p in set(node_attr['inputs']).intersection(family):
                            # add attributes to both representations of edge
                            succ[p][c] = s_pred[p] = dmap_succ[p][c]
                        return False

                    return True

                return False

        else:
            # Namespace shortcuts for speed.
            neighbors, dmap_succ = graph.predecessors, self.dmap.pred
            pred, succ = sub_dsp.dmap._succ, sub_dsp.dmap._pred

            def _check_node_inputs(c, p):
                if c == START:
                    try:
                        node_attr = dmap_nodes[p]
                        return node_attr['type'] == 'data'
                    except KeyError:
                        return True
                return False

        from collections import deque
        queue = deque([])

        blockers = set(blockers or ())

        # Function to set node attributes.
        def _set_node_attr(n, add2family=True, block=False):
            # Set node attributes.
            nodes[n] = dmap_nodes[n]

            # Add node in the adjacency matrix.
            succ[n], pred[n] = ({}, {})

            if n in dsp_dlt_val:
                dlt_val[n] = dsp_dlt_val[n]  # Set the default value.

            if add2family:
                # Append a new parent to the family.
                family[n] = () if block and n in blockers else neighbors(n)

                queue.append(n)

        # Set initial node attributes.
        for s in sources:
            if s in dmap_nodes and s in graph.nodes:
                _set_node_attr(s, block=not (wildcard and s in blockers))

        # Start breadth-first-search.
        while queue:
            parent = queue.popleft()

            # Namespace shortcuts for speed.
            nbrs, dmap_nbrs = succ[parent], dmap_succ[parent]

            # Iterate parent's children.
            for child in sorted(family[parent]):

                if _check_node_inputs(child, parent):
                    continue

                if child not in family:
                    _set_node_attr(child, block=True)  # Set node attributes.

                # Add attributes to both representations of edge: u-v and v-u.
                nbrs[child] = pred[child][parent] = dmap_nbrs[child]

        if _update_links:
            from .utils.alg import _update_io, _get_sub_out, _get_sub_inp
            succ, pred = sub_dsp.dmap.succ, sub_dsp.dmap.pred
            for k, a in sub_dsp.sub_dsp_nodes.items():
                nodes[k] = a = a.copy()

                inp, out = _get_sub_inp(a, pred[k]), _get_sub_out(a, succ[k])

                a['function'] = a['function'].get_sub_dsp_from_workflow(
                    sources=out.union(inp), graph=a['function'].dmap,
                    reverse=True, blockers=inp, wildcard=True
                )

                i, o = _update_io(a, pred[k], succ[k])  # Unreachable nodes.
                msg = 'Sub-dsp {} missing: inp {}, out {}'
                assert not i and not o, msg.format(k, i, o)
        return sub_dsp