From 5be13d35420ee7f94612c2193ed43c0e49383579 Mon Sep 17 00:00:00 2001
From: Jean-Baptiste Keck <Jean-Baptiste.Keck@imag.fr>
Date: Wed, 15 Apr 2020 21:09:39 +0200
Subject: [PATCH] remove graph_tool dependency in favor of networkx
---
hysop/core/graph/computational_graph.py | 158 ++-----
hysop/core/graph/graph.py | 129 +++++-
hysop/core/graph/graph_builder.py | 384 +++++-------------
hysop/operator/tests/test_absorption.py | 2 +-
.../tests/test_spectral_derivative.py | 2 +-
hysop/problem.py | 16 +-
6 files changed, 261 insertions(+), 430 deletions(-)
diff --git a/hysop/core/graph/computational_graph.py b/hysop/core/graph/computational_graph.py
index 233f4e307..6aaffae39 100644
--- a/hysop/core/graph/computational_graph.py
+++ b/hysop/core/graph/computational_graph.py
@@ -7,7 +7,7 @@ from hysop.tools.string_utils import framed_str, strlen, multiline_split
from hysop.tools.numpywrappers import npw
from hysop.core.graph.graph import not_implemented, initialized, discretized, \
ready, graph_built, not_initialized
-from hysop.core.graph.graph import Graph, ComputationalGraphNodeData, gt
+from hysop.core.graph.graph import ComputationalGraphNodeData
from hysop.core.graph.computational_node import ComputationalGraphNode
from hysop.core.graph.computational_operator import ComputationalGraphOperator
from hysop.core.graph.node_generator import ComputationalGraphNodeGenerator
@@ -353,16 +353,12 @@ class ComputationalGraph(ComputationalGraphNode):
return u'\n{}\n'.format(framed_str(title=title, msg=ss[1:]))
def variable_report(self):
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
fields = self.fields
topologies = {}
for field in self.fields:
field_topologies = {}
- for (i, vid) in enumerate(self.sorted_nodes):
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
+ for (i, node) in enumerate(self.nodes):
if field in op.input_fields:
topo = op.input_fields[field]
field_topologies.setdefault(topo, []).append(op)
@@ -420,43 +416,39 @@ class ComputationalGraph(ComputationalGraphNode):
replace = ('--', '', '', '-', '', '')
reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
ops = []
- for (i, vid) in enumerate(self.sorted_nodes):
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
-
+ for (i, node) in enumerate(self.nodes):
handled_inputs, handled_outputs = (), ()
finputs, foutputs = [], []
- for f in op.input_tensor_fields:
+ for f in node.input_tensor_fields:
f0 = f.fields[0]
- t0 = op.input_fields[f0]
- if all((op.input_fields[fi] is t0) for fi in f.fields):
+ t0 = node.input_fields[f0]
+ if all((node.input_fields[fi] is t0) for fi in f.fields):
finputs.append(u'{}.{}'.format(f.pretty_name.decode('utf-8'),
t0.pretty_tag.decode('utf-8')))
handled_inputs += f.fields
- for f in op.output_tensor_fields:
+ for f in node.output_tensor_fields:
f0 = f.fields[0]
- t0 = op.output_fields[f0]
- if all((op.output_fields[fi] is t0) for fi in f.fields):
+ t0 = node.output_fields[f0]
+ if all((node.output_fields[fi] is t0) for fi in f.fields):
foutputs.append(u'{}.{}'.format(f.pretty_name.decode('utf-8'),
t0.pretty_tag.decode('utf-8')))
handled_outputs += f.fields
finputs += [u'{}.{}'.format(f.pretty_name.decode('utf-8'),
t.pretty_tag.decode('utf-8'))
- for (f, t) in op.input_fields.iteritems()
+ for (f, t) in node.input_fields.iteritems()
if f not in handled_inputs]
foutputs += [u'{}.{}'.format(f.pretty_name.decode('utf-8'),
t.pretty_tag.decode('utf-8'))
- for (f, t) in op.output_fields.iteritems()
+ for (f, t) in node.output_fields.iteritems()
if f not in handled_outputs]
finputs = u','.join(sorted(finputs))
foutputs = u','.join(sorted(foutputs))
pinputs = u','.join(sorted([p.pretty_name.decode('utf-8')
- for p in op.input_params.values()]))
+ for p in node.input_params.values()]))
poutputs = u','.join(sorted([p.pretty_name.decode('utf-8')
- for p in op.output_params.values()]))
+ for p in node.output_params.values()]))
infields = u'[{}]'.format(finputs) if finputs else u''
outfields = u'[{}]'.format(foutputs) if foutputs else u''
@@ -471,8 +463,8 @@ class ComputationalGraph(ComputationalGraphNode):
if outputs == '':
outputs = u'no outputs'
- opname = op.pretty_name.decode('utf-8')
- optype = type(op).__name__
+ opname = node.pretty_name.decode('utf-8')
+ optype = type(node).__name__
strdata = (str(i), opname, inputs, '->', outputs, optype)
ops += multiline_split(strdata, maxlen, split_sep, replace, newline_prefix)
@@ -626,11 +618,8 @@ class ComputationalGraph(ComputationalGraphNode):
def check(self):
super(ComputationalGraph, self).check()
reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
- op.check()
+ for node in self.nodes:
+ node.check()
@debug
def get_field_requirements(self):
@@ -722,66 +711,16 @@ class ComputationalGraph(ComputationalGraphNode):
from hysop import main_rank
if (visu_rank is None) or (main_rank != visu_rank):
return
-
- graph = self.reduced_graph
- edge_text = graph.edge_properties['var_names']
- vertex_text = graph.vertex_properties['op_pnames']
- vertex_info = graph.vertex_properties['op_info']
- if 'command_queues' in graph.vp:
- command_queues = graph.vertex_properties['command_queues']
- active_ops = graph.vertex_properties['active_ops']
- else:
- command_queues = None
- active_ops = None
-
- def draw():
- import time
- from gi.repository import Gtk, GObject
- from graph_tool.draw import GraphWindow, sfdp_layout
-
- pos_layout = sfdp_layout(graph)
-
- win = GraphWindow(graph, pos_layout, geometry=(800, 600),
- vertex_text=vertex_text,
- edge_text=edge_text,
- vertex_font_size=vertex_font_size,
- edge_font_size=edge_font_size,
- vertex_color=active_ops,
- vertex_fill_color=command_queues,
- display_props=vertex_info,
- display_props_size=14,
- max_render_time=50)
-
- def update_window():
- win.graph.regenerate_surface()
- win.graph.queue_draw()
- time.sleep(0.01)
- return True
-
- GObject.idle_add(update_window)
- win.connect("delete_event", Gtk.main_quit)
- win.show_all()
- Gtk.main()
- self.graph_is_rendering = False
-
- self.graph_is_rendering = True
-
- from threading import Thread
- display_thread = Thread(target=draw)
- display_thread.start()
+ raise NotImplementedError('This feature has not been implemented yet.')
@debug
@graph_built
def discretize(self):
if self.discretized:
return
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
- if not op.discretized:
- op.discretize()
+ for node in self.nodes:
+ if not node.discretized:
+ node.discretize()
if self.is_root:
input_discrete_fields = {}
@@ -846,14 +785,10 @@ class ComputationalGraph(ComputationalGraphNode):
def get_work_properties(self):
requests = MultipleOperatorMemoryRequests()
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
- if op not in requests.operators():
- wp = op.get_work_properties()
- requests += op.get_work_properties()
+ for node in self.nodes:
+ if node not in requests.operators():
+ wp = node.get_work_properties()
+ requests += node.get_work_properties()
if __DEBUG__ or (__VERBOSE__ and self.level == 0) or self.__FORCE_REPORTS__:
srequests = requests.sreport()
ss = (srequests if (srequests != u'') else u' *no extra work requested*')
@@ -870,13 +805,9 @@ class ComputationalGraph(ComputationalGraphNode):
if (work is None):
work = self.get_work_properties()
work.allocate(allow_subbuffers=allow_subbuffers)
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
- if not op.ready:
- op.setup(work=work)
+ for node in self.nodes:
+ if not node.ready:
+ node.setup(work=work)
self.ready = True
def build(self, outputs_are_inputs=True, method=None, allow_subbuffers=False):
@@ -896,38 +827,17 @@ class ComputationalGraph(ComputationalGraphNode):
@debug
@ready
def apply(self, **kwds):
- drawing = self.graph_is_rendering
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
-
- if drawing:
- active_ops = reduced_graph.vertex_properties['active_ops']
- old_color = None
- for vid in self.sorted_nodes:
- if old_color:
- active_ops[vertex] = old_color
- vertex = reduced_graph.vertex(vid)
- old_color = active_ops[vertex]
- active_ops[vertex] = 'red'
- op = operators[vertex]
- dprint('{}.apply()'.format(op.name))
- op.apply(**kwds)
- active_ops[vertex] = old_color
- else:
- for op in self.nodes:
- dprint('{}.apply()'.format(op.name))
- op.apply(**kwds)
+ for node in self.nodes:
+ dprint('{}.apply()'.format(node.name))
+ node.apply(**kwds)
@debug
@ready
def finalize(self, **kwds):
reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
- if op.ready:
- op.finalize(**kwds)
+ for node in self.nodes:
+ if node.ready:
+ node.finalize(**kwds)
self.ready = False
@classmethod
diff --git a/hysop/core/graph/graph.py b/hysop/core/graph/graph.py
index 09e5a7f57..6c2f37aff 100644
--- a/hysop/core/graph/graph.py
+++ b/hysop/core/graph/graph.py
@@ -1,11 +1,79 @@
-import inspect
-import graph_tool as gt
-from graph_tool import Graph, GraphView
-from graph_tool import topology, stats, search
-from hysop.tools.decorators import not_implemented, debug, wraps, profile
+import inspect, networkx
from hysop import dprint
+from hysop.tools.types import check_instance, first_not_None
+from hysop.tools.decorators import not_implemented, debug, wraps, profile
+
+is_directed_acyclic_graph = networkx.algorithms.dag.is_directed_acyclic_graph
+transitive_reduction = networkx.algorithms.dag.transitive_reduction
+
+def all_simple_paths(graph, src, dst):
+ return tuple(networkx.algorithms.simple_paths.all_simple_paths(graph, src, dst))
+
+def topological_sort(graph):
+ return tuple(networkx.algorithms.dag.topological_sort(graph))
+
+def new_directed_graph():
+ return networkx.DiGraph()
+
+def new_vertex(graph, *args, **kwds):
+ # /!\ We have to use networkx 2.2 which has a different interface for attributes
+ node = VertexAttributes(graph, *args, **kwds)
+ graph.add_node(node)
+ return node
+def new_edge(graph, u, v, *args, **kwds):
+ # /!\ We have to use networkx 2.2 which has a different interface for attributes
+ graph.add_edge(u, v, object=EdgeAttributes(*args, **kwds))
+ return (u,v)
+
+class VertexAttributes(object):
+ """Simple class to hold vertex data."""
+ def __init__(self, graph, operator=None):
+ if not hasattr(graph, '_hysop_node_counter'):
+ graph._hysop_node_counter = 0
+ node_id = graph._hysop_node_counter
+ graph._hysop_node_counter += 1
+
+ self.node_id = node_id
+ self.operator = operator
+
+ self.input_states = None
+ self.output_states = None
+ self.op_ordering = None
+ self.command_queue = None
+ def copy_attributes(self, other):
+ if (other is None):
+ return self
+ check_instance(other, VertexAttributes)
+ for vname in ('operator',
+ 'input_states', 'output_states',
+ 'op_ordering', 'command_queue'):
+ setattr(self, vname, first_not_None(getattr(self, vname),
+ getattr(other, vname)))
+ return self
+
+ def set_op_info(self, operator, input_states, output_states):
+ assert (self.operator is not None)
+ assert self.operator is operator
+ self.operator = operator
+ self.input_states = input_states
+ self.output_states = output_states
+ return self
+
+ def hash(self):
+ return self.node_id
+ def __eq__(self, other):
+ return self.node_id == other.node_id
+ def __int__(self):
+ return self.node_id
+
+class EdgeAttributes(object):
+ """Simple class to hold edge data."""
+ def __init__(self, variable=None, topology=None):
+ self.variable = variable
+ self.topology = topology
+
class ComputationalGraphNodeData(object):
"""
Simple class to hold some node data.
@@ -127,7 +195,6 @@ def op_apply(f):
@profile
@ready
def apply(*args, **kwds):
- #print u'APPLY {}'.format(args[0].name)
dbg = ('dbg' in kwds)
dbg = dbg and (kwds['dbg'] is not None)
dbg = dbg and (kwds['dbg'].enable_on_op_apply)
@@ -175,3 +242,53 @@ def op_apply(f):
return
return ret
return apply
+
+def _op_info(op, istates=None, ostates=None, jmp=False):
+ ifields = op.input_fields
+ ofields = op.output_fields
+ iparams = op.input_params
+ oparams = op.output_params
+
+ memorder2str = {
+ MemoryOrdering.C_CONTIGUOUS: 'C',
+ MemoryOrdering.F_CONTIGUOUS: 'F',
+ }
+
+ def ifinfo(field, topo):
+ info = (field.name, topo.id)
+ if istates:
+ assert field in istates
+ istate = istates[field]
+ assert (istate is not None)
+ info+=(memorder2str[istate.memory_order],)
+ info+=(str(istate.tstate),)
+ return info
+ def ofinfo(field, topo):
+ info = (field.name, topo.id)
+ if ostates:
+ assert field in ostates
+ ostate = ostates[field]
+ assert (ostate is not None)
+ info+=(memorder2str[ostate.memory_order],)
+ info+=(str(ostate.tstate),)
+ return info
+ def ipinfo(param):
+ return param.name
+ def opinfo(param):
+ return param.name
+
+ ss = 'Operator {} => \n {}{}{}{}\n {}'.format(op.name,
+ 'Pin:{}\n '.format([ ipinfo(param) for param in iparams.values() ])
+ if iparams else '',
+ 'Fin:{}\n '.format([ ifinfo(f,topo) for (f,topo) in ifields.iteritems() ])
+ if ifields else '',
+ 'Pout:{}\n '.format([ opinfo(param) for param in oparams.values() ])
+ if oparams else '',
+ 'Fout:{}\n '.format([ ofinfo(f,topo) for (f,topo) in ofields.iteritems() ])
+ if ofields else '',
+ op.__class__)
+ if jmp:
+ return ss
+ else:
+ return ss.replace('\n',' ')
+
diff --git a/hysop/core/graph/graph_builder.py b/hysop/core/graph/graph_builder.py
index 3abe3a7ef..5cda50bcc 100644
--- a/hysop/core/graph/graph_builder.py
+++ b/hysop/core/graph/graph_builder.py
@@ -1,3 +1,4 @@
+
from hysop import vprint, dprint, Problem
from hysop.deps import np, __builtin__, print_function
from hysop.tools.types import check_instance
@@ -9,16 +10,21 @@ from hysop.constants import MemoryOrdering, Backend
from hysop.parameters.parameter import Parameter
from hysop.topology.cartesian_topology import CartesianTopologyState
+from hysop.core.graph.graph import (new_directed_graph, new_vertex, new_edge,
+ is_directed_acyclic_graph, transitive_reduction,
+ topological_sort, all_simple_paths)
from hysop.core.graph.computational_graph import ComputationalGraph
from hysop.core.graph.computational_node import ComputationalGraphNode
from hysop.core.graph.computational_operator import ComputationalGraphOperator
-from hysop.core.graph.graph import Graph, ComputationalGraphNodeData, gt
-from hysop.fields.field_requirements import DiscreteFieldRequirements, MultiFieldRequirements
+from hysop.fields.field_requirements import (DiscreteFieldRequirements,
+ MultiFieldRequirements)
-from hysop.operator.redistribute import Redistribute, RedistributeNotImplementedError
-from hysop.operator.transpose import Transpose, TranspositionNotImplementedError
-from hysop.operator.memory_reordering import MemoryReordering, MemoryReorderingNotImplementedError
+from hysop.operator.redistribute import (Redistribute,
+ RedistributeNotImplementedError)
+from hysop.operator.transpose import Transpose, TranspositionNotImplementedError
+from hysop.operator.memory_reordering import (MemoryReordering,
+ MemoryReorderingNotImplementedError)
# Debug level for graph building
# 0: no debug logs
@@ -35,57 +41,6 @@ def gprint2(*args, **kwds):
kwds['level'] = 2
gprint(*args, **kwds)
-def _op_info(op,
- istates=None, ostates=None,
- jmp=False):
- ifields = op.input_fields
- ofields = op.output_fields
- iparams = op.input_params
- oparams = op.output_params
-
- memorder2str = {
- MemoryOrdering.C_CONTIGUOUS: 'C',
- MemoryOrdering.F_CONTIGUOUS: 'F',
- }
-
- def ifinfo(field, topo):
- info = (field.name, topo.id)
- if istates:
- assert field in istates
- istate = istates[field]
- assert (istate is not None)
- info+=(memorder2str[istate.memory_order],)
- info+=(str(istate.tstate),)
- return info
- def ofinfo(field, topo):
- info = (field.name, topo.id)
- if ostates:
- assert field in ostates
- ostate = ostates[field]
- assert (ostate is not None)
- info+=(memorder2str[ostate.memory_order],)
- info+=(str(ostate.tstate),)
- return info
- def ipinfo(param):
- return param.name
- def opinfo(param):
- return param.name
-
- ss = 'Operator {} => \n {}{}{}{}\n {}'.format(op.name,
- 'Pin:{}\n '.format([ ipinfo(param) for param in iparams.values() ])
- if iparams else '',
- 'Fin:{}\n '.format([ ifinfo(f,topo) for (f,topo) in ifields.iteritems() ])
- if ifields else '',
- 'Pout:{}\n '.format([ opinfo(param) for param in oparams.values() ])
- if oparams else '',
- 'Fout:{}\n '.format([ ofinfo(f,topo) for (f,topo) in ofields.iteritems() ])
- if ofields else '',
- op.__class__)
- if jmp:
- return ss
- else:
- return ss.replace('\n',' ')
-
class GraphBuilder(object):
"""
@@ -126,21 +81,7 @@ class GraphBuilder(object):
gprint(msg)
def setup_graph(self):
- graph = Graph(directed=True)
-
- vertex_properties = {}
- vertex_properties['op_names'] = graph.new_vertex_property('string')
- vertex_properties['op_pnames'] = graph.new_vertex_property('string')
- vertex_properties['op_info'] = graph.new_vertex_property('string')
- vertex_properties['operators'] = graph.new_vertex_property('python::object')
-
- edge_properties = {}
- edge_properties['var_names'] = graph.new_edge_property('string')
- edge_properties['variables'] = graph.new_edge_property('python::object')
-
- self.graph = graph
- self.vertex_properties = vertex_properties
- self.edge_properties = edge_properties
+ self.graph = new_directed_graph()
def setup_variables(self):
self.input_fields = {}
@@ -160,40 +101,23 @@ class GraphBuilder(object):
self.target_node._input_fields_to_dump,
self.target_node.method)
- def new_node(self, opname, oppname, op, subgraph,
- current_level, node, node_id,
- extra_node_props, opvertex):
- graph = self.graph
- vertex_properties = self.vertex_properties
+ def new_node(self, op, subgraph,
+ current_level, node, node_id, opvertex):
- opnode = graph.add_vertex()
- vertex_properties['op_names'][opnode] = opname
- vertex_properties['op_pnames'][opnode] = oppname
- vertex_properties['operators'][opnode] = op
+ graph = self.graph
+ opnode = new_vertex(graph, op).copy_attributes(opvertex)
gprint(' *Created node is {}.'.format(int(opnode)))
-
- if opvertex:
- assert (extra_node_props is not None)
- assert 'op_info' in extra_node_props
- level = node.level + 1
- for enp in extra_node_props:
- vertex_properties[enp][opnode] = subgraph.vp[enp][opvertex]
- else:
- level = current_level
return opnode
def build_graph(self):
-
target_node = self.target_node
current_level = self.current_level
outputs_are_inputs = self.outputs_are_inputs
graph = self.graph
- vertex_properties = self.vertex_properties
- edge_properties = self.edge_properties
- parameter_handler = self.__ParameterHandler(graph, edge_properties, vertex_properties)
+ parameter_handler = self.__ParameterHandler(graph)
input_fields = self.input_fields
output_fields = self.output_fields
@@ -230,7 +154,7 @@ class GraphBuilder(object):
# build its own local graph and we extract all its operators (graph nodes).
# Else if node is a ComputationalGraphOperator, we just take the
# current node operator.
- subgraph, node_ops, node_vertices, from_subgraph, extra_node_props = \
+ subgraph, node_ops, node_vertices, from_subgraph = \
self.build_subgraph(node, current_level)
# iterate over subgraph operators
@@ -245,9 +169,8 @@ class GraphBuilder(object):
field_requirements = op._field_requirements
# add operator node and fill vertex properties
- opnode = self.new_node(opname, oppname, op, subgraph,
- current_level, node, node_id,
- extra_node_props, opvertex)
+ opnode = self.new_node(op, subgraph,
+ current_level, node, node_id, opvertex)
if not isinstance(op, Problem):
# try to fill in undertermined topologies (experimental feature)
@@ -342,8 +265,7 @@ class GraphBuilder(object):
is_new = False
dstate = cstate.handle_input(opnode, itopo, ifreqs,
- graph, edge_properties, vertex_properties,
- is_new)
+ graph, is_new)
input_states[ifield] = dstate
if is_new:
@@ -370,15 +292,13 @@ class GraphBuilder(object):
self.new_topology_state(ofield))
invalidate_field = (ofield not in op.get_preserved_input_fields())
dstate = cstate.handle_output(opnode, otopo, ofreqs,
- op, istates, invalidate_field,
- graph, edge_properties, vertex_properties)
+ op, istates, invalidate_field, graph)
output_fields[ofield] = otopo
output_states[ofield] = dstate
output_topology_states[ofield] = (None, dstate)
if (current_level==0) and ((op,opnode) not in deferred_operators):
- vertex_properties['op_info'][opnode] = _op_info(op, input_states,
- output_states)
+ opnode.set_op_info(op, input_states, output_states)
op_input_topology_states[op] = input_states
op_output_topology_states[op] = output_states
@@ -408,10 +328,9 @@ class GraphBuilder(object):
ifields[ifield] = itopo
input_states[ifield] = dstate
field_requirements.update_inputs({ifield: ireqs})
- cstate.add_edge(graph, edge_properties, opnode, node, ifield, itopo)
+ cstate.add_edge(graph, opnode, node, ifield, itopo)
if current_level==0:
- vertex_properties['op_info'][opnode] = _op_info(op, input_states,
- output_states)
+ opnode.set_op_info(op, input_states, output_states)
if current_level==0:
msg='\nComputationalGraph {} inputs:\n'.format(target_node.name)
@@ -455,20 +374,18 @@ class GraphBuilder(object):
**op_kwds)
op.initialize(topgraph_method=self.target_node.method)
op.get_and_set_field_requirements()
- opnode = self.new_node(opname, oppname, op, None,
- current_level, None, None, None, None)
+ opnode = self.new_node(op, None, current_level,
+ None, None, None, None)
ifreqs = None if (current_level!=0) \
else field_requirements.get_input_requirement(field)[1]
cstate = self.topology_states[field]
- state = cstate.handle_input(opnode, target_topo, ifreqs,
- graph, edge_properties, vertex_properties, False)
+ state = cstate.handle_input(opnode, target_topo, ifreqs, graph,False)
input_states = {field: state}
output_states = {}
self.op_input_topology_states[op] = input_states
self.op_output_topology_states[op] = output_states
if current_level==0:
- vertex_properties['op_info'][opnode] = _op_info(op, input_states,
- output_states)
+ opnode.set_op_info(op, input_states, output_states)
# Alter states such that output topology states match input topology states
# this is only done if required (outputs_are_inputs) and if we are
@@ -487,118 +404,68 @@ class GraphBuilder(object):
requirements.memory_order = input_topology_state.memory_order
cstate = self.topology_states[field]
- cstate.output_as_input(target_topo, requirements,
- graph, edge_properties, vertex_properties)
+ cstate.output_as_input(target_topo, requirements, graph)
# Check that the generated graph is a directed acyclic graph
- if not gt.topology.is_DAG(graph):
+ if not is_directed_acyclic_graph(graph):
msg='\nGenerated operator graph is not acyclic.'
- print(msg)
-
- #display graph for debug purposes
- gt.stats.remove_parallel_edges(graph)
- for prop_name,edge_property in edge_properties.iteritems():
- graph.edge_properties[prop_name] = edge_property
- for prop_name,vertex_property in vertex_properties.iteritems():
- if prop_name != 'command_queues':
- graph.vertex_properties[prop_name] = vertex_property
- target_node.graph_built = True
- target_node.reduced_graph = graph
- target_node.display()
-
- # and finally raise error
raise RuntimeError(msg)
- # Transitive reduction of graph (remove parallel and unnecessary transitive edges)
+ # Transitive reduction of graph
+ # This removes parallel and unnecessary transitive edges
# ie. remove useless redondant dependencies
- transitive_reduction = gt.stats.label_parallel_edges(graph, mark_only=True,
- eprop = graph.new_edge_property('bool',val=False))
- for vertex in graph.vertices():
- for neighbor_vertex in vertex.out_neighbours():
- accessible_vertices = \
- [v for v in gt.search.dfs_iterator(graph, neighbor_vertex)]
- for edge in accessible_vertices:
- edge = graph.edge(vertex, edge.target())
- if edge is not None:
- transitive_reduction[edge] = True
- transitive_reduction.a = [ not val for val in transitive_reduction.get_array()]
- reduced_graph = gt.GraphView(graph, efilt = transitive_reduction)
+ reduced_graph = transitive_reduction(graph)
# Topological sort
# ie. find out operator order for execution purposes
- sorted_nodes = gt.topology.topological_sort(reduced_graph)
- vertex_properties['op_ordering'] = reduced_graph.new_vertex_property('int')
- for i,node_id in enumerate(sorted_nodes):
- vertex = reduced_graph.vertex(node_id)
- vertex_properties['op_ordering'][vertex] = i
- if current_level==0:
- vertex_properties['op_names'][vertex] += ' ('+str(i)+')'
- vertex_properties['op_pnames'][vertex] += ' ('+str(i)+')'
+ sorted_nodes = topological_sort(reduced_graph)
+ for (i, node) in enumerate(sorted_nodes):
+ node.op_ordering = i
# Command queues (each color represents a command queue)
# ie. try to find out data independent subgraphs
color = 0
queues = {}
- vertex_properties['command_queues'] = graph.new_vertex_property('int',val=-1)
- vertex_properties['active_ops'] = graph.new_vertex_property('string',val='darkgray')
- for vertex_id in sorted_nodes:
- vertex = reduced_graph.vertex(vertex_id)
- if (vertex_properties['command_queues'][vertex] >= 0):
+ for node in sorted_nodes:
+ if (node.command_queue is not None):
continue
- vertices = [vertex]
- uncolored_childs = [ v for v in vertex.out_neighbours() \
- if (vertex_properties['command_queues'][v] == -1) ]
+ nodes = [node]
+ uncolored_childs = tuple(filter(lambda n: n.command_queue is None,
+ reduced_graph.adj[node]))
while len(uncolored_childs)>0:
- vid = np.argmin( [vertex_properties['op_ordering'][v]
- for v in uncolored_childs] )
- vertex = uncolored_childs[vid]
- vertices.append(vertex)
- uncolored_childs = [ v for v in vertex.out_neighbours() \
- if (vertex_properties['command_queues'][v] == -1) ]
+ vid = np.argmin( [n.op_ordering for n in uncolored_childs] )
+ node = uncolored_childs[vid]
+ nodes.append(node)
+ uncolored_childs = tuple(filter(lambda n: n.command_queue is None,
+ reduced_graph.adj[node]))
- idx_range = (vertex_properties['op_ordering'][vertices[0]],
- vertex_properties['op_ordering'][vertices[-1]])
+ idx_range = (nodes[0].op_ordering, nodes[-1].op_ordering)
if queues:
color = queues.keys()[-1]+1
-
for k in queues.keys()[::-1]:
paths = queues[k]
if (paths[-1][1] < idx_range[0]):
- src = reduced_graph.vertex(sorted_nodes[paths[-1][1]])
- dst = reduced_graph.vertex(sorted_nodes[idx_range[0]])
- all_paths = gt.topology.all_paths(reduced_graph,src,dst)
- all_paths = [p for p in all_paths]
+ src = sorted_nodes[paths[-1][1]]
+ dst = sorted_nodes[idx_range[0]]
+ all_paths = all_simple_paths(reduced_graph,src,dst)
if len(all_paths)>0:
color = k
break
queues.setdefault(color,[]).append(idx_range)
- for vertex in vertices:
- vertex_properties['command_queues'][vertex] = color
-
- # bind all original graph properties to reduced graph
- for prop_name,edge_property in edge_properties.iteritems():
- reduced_graph.edge_properties[prop_name] = edge_property
- for prop_name,vertex_property in vertex_properties.iteritems():
- reduced_graph.vertex_properties[prop_name] = vertex_property
+ for node in nodes:
+ node.command_queue = color
self.reduced_graph = reduced_graph
- self.sorted_nodes = sorted_nodes
- self.nodes = self._gather_nodes()
-
- def _gather_nodes(self):
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
- nodes = [ operators[reduced_graph.vertex(vid)] for vid in self.sorted_nodes ]
- return nodes
+ self.sorted_nodes = sorted_nodes
+ self.nodes = tuple(map(lambda x: x.operator, sorted_nodes))
def build_subgraph(self, node, current_level, **kwds):
node_ops = []
node_vertices = []
- extra_node_props = []
subgraph = None
from_subgraph = False
@@ -613,18 +480,11 @@ class GraphBuilder(object):
node_ordering = node.sorted_nodes
subgraph = node.reduced_graph
from_subgraph = True
- subgraph_ops = subgraph.vertex_properties['operators']
- vertex_properties = None
for nid in node_ordering:
- _node = subgraph.vertex(nid)
- op = subgraph_ops[_node]
+ _node = nid
+ op = _node.operator
node_vertices.append(_node)
node_ops.append(op)
- for prop_name,vp in subgraph.vertex_properties.iteritems():
- if prop_name not in self.vertex_properties:
- self.vertex_properties[prop_name] = \
- self.graph.new_vertex_property(vp.value_type())
- extra_node_props.append(prop_name)
elif isinstance(node, ComputationalGraphOperator):
node_operators = node.operators()
node_ops.extend(node_operators)
@@ -633,22 +493,19 @@ class GraphBuilder(object):
msg = 'Unknown node type {}.'
raise NotImplementedError(msg.format(node.__class__.__name__))
- return subgraph, node_ops, node_vertices, from_subgraph, extra_node_props
+ return subgraph, node_ops, node_vertices, from_subgraph
class __ParameterHandler(object):
- def __init__(self, graph, edge_properties, vertex_properties):
+ def __init__(self, graph):
self.graph = graph
- self.edge_properties = edge_properties
- self.vertex_properties = vertex_properties
self.last_write_node = {}
self.reading_nodes = {}
def add_edge(self, src_node, dst_node, parameter):
if (src_node is not None) and (dst_node is not None) \
and (src_node != dst_node):
- edge = self.graph.add_edge(src_node, dst_node)
- self.edge_properties['var_names'][edge] = parameter.pretty_name
+ edge = new_edge(self.graph, src_node, dst_node, parameter)
return edge
else:
return None
@@ -720,30 +577,18 @@ class GraphBuilder(object):
self.method = topgraph_method
- def add_vertex(self, graph, vertex_properties, operator):
- vertex = graph.add_vertex()
- vertex_properties['operators'][vertex] = operator
- vertex_properties['op_names'][vertex] = operator.name
- vertex_properties['op_pnames'][vertex] = operator.pretty_name
- #gprint('Creating vertex {}.'.format(int(vertex)))
- return vertex
-
- def add_edge(self, graph, edge_properties,
- src_node, dst_node,
- field, topology):
+ def add_vertex(self, graph, operator):
+ return new_vertex(graph, operator)
+
+ def add_edge(self, graph, src_node, dst_node, field, topology):
if (src_node is not None) and (dst_node is not None) \
and (src_node != dst_node):
- edge = graph.add_edge(src_node, dst_node)
- edge_properties['var_names'][edge] = \
- '{}.{}'.format(field.pretty_name, topology.pretty_tag)
- #gprint('Adding edge between {} and {}.'.format(int(src_node), int(dst_node)))
- return edge
+ return new_edge(graph, src_node, dst_node, field, topology)
else:
return None
-
def push_generated_operators(self, op_generator, op_name_prefix,
- src_topo, graph, vertex_properties, edge_properties):
+ src_topo, graph):
field = self.field
read_nodes = self.read_nodes
@@ -767,21 +612,17 @@ class GraphBuilder(object):
assert op.input_fields.values()[0] == src_topo
dst_topo = op.output_fields.values()[0]
- op_node = self.add_vertex(graph, vertex_properties, op)
+ op_node = self.add_vertex(graph, op)
# handle input
if (src_topo in write_nodes):
src_node = write_nodes[src_topo]
- self.add_edge(graph, edge_properties,
- src_node, op_node,
- field, src_topo)
+ self.add_edge(graph, src_node, op_node, field, src_topo)
# handle output
ro_nodes = read_nodes.setdefault(dst_topo, [])
for ro_node in ro_nodes:
- self.add_edge(graph, edge_properties,
- ro_node, op_node,
- field, dst_topo)
+ self.add_edge(graph, ro_node, op_node, field, dst_topo)
read_nodes[dst_topo] = []
write_nodes[dst_topo] = op_node
@@ -798,17 +639,14 @@ class GraphBuilder(object):
op_input_topology_states[op] = istate
op_output_topology_states[op] = ostate
- vertex_properties['op_info'][op_node] = _op_info(op,
- istates=istate, ostates=ostate)
+ op_node.set_op_info(op, istate, ostate)
src_node = op_node
src_topo = dst_topo
return dst_topo
- def redistribute(self, target_topo,
- graph, vertex_properties, edge_properties,
- src_topo=None):
+ def redistribute(self, target_topo, graph, src_topo=None):
field = self.field
write_nodes = self.write_nodes
@@ -852,11 +690,10 @@ class GraphBuilder(object):
assert src_topo in src_topos
dst_topo = self.push_generated_operators(redistribute_generator, 'R',
- src_topo, graph, vertex_properties, edge_properties)
+ src_topo, graph)
assert dst_topo == target_topo
- def transpose(self, topo, target_axes,
- graph, vertex_properties, edge_properties):
+ def transpose(self, topo, target_axes, graph):
field = self.field
write_nodes = self.write_nodes
@@ -906,12 +743,11 @@ class GraphBuilder(object):
print('\n{}\n'.format(msg))
raise
- dst_topo = self.push_generated_operators(transpose_generator, 'T', topo,
- graph, vertex_properties, edge_properties)
+ dst_topo = self.push_generated_operators(transpose_generator, 'T',
+ topo, graph)
assert dst_topo == topo
- def reorder(self, topo, target_memory_order,
- graph, vertex_properties, edge_properties):
+ def reorder(self, topo, target_memory_order, graph):
field = self.field
write_nodes = self.write_nodes
@@ -946,12 +782,12 @@ class GraphBuilder(object):
print('\n{}\n'.format(msg))
raise
- dst_topo = self.push_generated_operators(reorder_generator, 'MR', topo,
- graph, vertex_properties, edge_properties)
+ dst_topo = self.push_generated_operators(reorder_generator, 'MR',
+ topo, graph)
assert dst_topo == topo
def handle_input(self, opnode, target_topo, target_dfield_requirements,
- graph, edge_properties, vertex_properties, is_new):
+ graph, is_new):
ifield = self.field
write_nodes = self.write_nodes
@@ -976,10 +812,8 @@ class GraphBuilder(object):
**op_kwds)
writer_op.initialize(topgraph_method=self.method)
writer_op.get_and_set_field_requirements()
- writer_opnode = self.add_vertex(graph, vertex_properties, writer_op)
- self.add_edge(graph, edge_properties,
- writer_opnode, opnode,
- ifield, target_topo)
+ writer_opnode = self.add_vertex(graph, writer_op)
+ self.add_edge(graph, writer_opnode, opnode, ifield, target_topo)
# we only handle input field requirements when we are root graph
# ie. target_dfield_requirements is None
@@ -1039,13 +873,10 @@ class GraphBuilder(object):
if (src_topo is not target_topo):
gprint(' >Redistributing field {} from up to date topologies {} to host topology {}.'.format(
ifield.name, ' ,'.join(t.pretty_tag for t in src_topos), target_topo.pretty_tag))
- self.transpose(src_topo, target_axes,
- graph, vertex_properties, edge_properties)
- self.redistribute(target_topo, graph,
- vertex_properties, edge_properties, src_topo=src_topo)
+ self.transpose(src_topo, target_axes, graph)
+ self.redistribute(target_topo, graph, src_topo=src_topo)
# we can always reorder target because this a host topology
- self.reorder(target_topo, target_memory_order,
- graph, vertex_properties, edge_properties)
+ self.reorder(target_topo, target_memory_order, graph)
elif (target_topo.backend.kind is Backend.OPENCL) and write_nodes:
# give source topo priority according to topology_affinity
src_topos = write_nodes.keys()
@@ -1054,18 +885,13 @@ class GraphBuilder(object):
if (src_topo is not target_topo):
gprint(' >Redistributing field {} from up to date topologies {} to device topology {}.'.format(
ifield.name, ' ,'.join(t.pretty_tag for t in src_topos), target_topo.pretty_tag))
- self.reorder(src_topo, target_memory_order,
- graph, vertex_properties, edge_properties)
- self.redistribute(target_topo, graph,
- vertex_properties, edge_properties, src_topo=src_topo)
+ self.reorder(src_topo, target_memory_order, graph)
+ self.redistribute(target_topo, graph, src_topo=src_topo)
# target is always opencl so we transpose here
- self.transpose(target_topo, target_axes,
- graph, vertex_properties, edge_properties)
+ self.transpose(target_topo, target_axes, graph)
else:
- self.transpose(target_topo, target_axes,
- graph, vertex_properties, edge_properties)
- self.reorder(target_topo, target_memory_order,
- graph, vertex_properties, edge_properties)
+ self.transpose(target_topo, target_axes, graph)
+ self.reorder(target_topo, target_memory_order, graph)
istate = dtopology_states[target_topo]
gprint2(' >Input state is now {}'.format(istate))
@@ -1077,21 +903,16 @@ class GraphBuilder(object):
output_states = {}
self.op_input_topology_states[writer_op] = input_states
self.op_output_topology_states[writer_op] = output_states
- vertex_properties['op_info'][writer_opnode] = _op_info(writer_op,
- input_states, output_states)
+ writer_opnode.set_op_info(writer_op, input_states, output_states)
# add read dependency to last written node before current op
# (so that inputs are modified before actual call to the operator)
if (target_topo in write_nodes):
last_write_node = write_nodes[target_topo]
- self.add_edge(graph, edge_properties,
- last_write_node, opnode,
- ifield, target_topo)
+ self.add_edge(graph, last_write_node, opnode, ifield, target_topo)
elif (not is_root) and write_nodes:
for node in self.write_nodes.values():
- self.add_edge(graph, edge_properties,
- node, opnode,
- ifield, target_topo)
+ self.add_edge(graph, node, opnode, ifield, target_topo)
read_nodes.setdefault(target_topo, []).append(opnode)
if is_new:
@@ -1100,9 +921,8 @@ class GraphBuilder(object):
return istate
- def handle_output(self, opnode, output_topo, oreqs,
- operator, input_topology_states, invalidate_field,
- graph, edge_properties, vertex_properties):
+ def handle_output(self, opnode, output_topo, oreqs, operator,
+ input_topology_states, invalidate_field, graph):
ofield = self.field
write_nodes = self.write_nodes
@@ -1114,9 +934,8 @@ class GraphBuilder(object):
# add dependency to last node written to prevent
# concurent write-writes.
if (output_topo in write_nodes):
- self.add_edge(graph, edge_properties,
- write_nodes[output_topo], opnode,
- ofield, output_topo)
+ self.add_edge(graph, write_nodes[output_topo],
+ opnode, ofield, output_topo)
if invalidate_field:
gprint(' >Invalidating output field {} on all topologies but {} because is has been freshly written.'.format(ofield.name, output_topo.pretty_tag))
@@ -1124,9 +943,7 @@ class GraphBuilder(object):
# to prevent concurent read-writes.
if output_topo in read_nodes:
for ro_node in read_nodes[output_topo]:
- self.add_edge(graph, edge_properties,
- ro_node, opnode,
- ofield, output_topo)
+ self.add_edge(graph, ro_node, opnode, ofield, output_topo)
# remove read/write dependencies and states
write_nodes.clear()
@@ -1155,9 +972,6 @@ class GraphBuilder(object):
return ostate
- def output_as_input(self, target_topo, dstate,
- graph, edge_properties, vertex_properties):
-
- self.handle_input(None, target_topo, dstate,
- graph, edge_properties, vertex_properties, False)
+ def output_as_input(self, target_topo, dstate, graph):
+ self.handle_input(None, target_topo, dstate, graph,False)
diff --git a/hysop/operator/tests/test_absorption.py b/hysop/operator/tests/test_absorption.py
index ce2735ecf..e0bf8d80a 100644
--- a/hysop/operator/tests/test_absorption.py
+++ b/hysop/operator/tests/test_absorption.py
@@ -180,7 +180,7 @@ class TestVorticityAbsorption(object):
dist = npw.abs(fout-fref)
dinf = npw.max(dist)
deps = int(npw.ceil(dinf/eps))
- if (deps < 100):
+ if (deps < 200):
print '{}eps, '.format(deps),
continue
has_nan = npw.any(npw.isnan(fout))
diff --git a/hysop/operator/tests/test_spectral_derivative.py b/hysop/operator/tests/test_spectral_derivative.py
index 5b3d93075..fbed428d1 100644
--- a/hysop/operator/tests/test_spectral_derivative.py
+++ b/hysop/operator/tests/test_spectral_derivative.py
@@ -275,7 +275,7 @@ class TestSpectralDerivative(object):
except OverflowError:
import numpy as np
deps = np.inf
- if (deps <= 10**(nidx+2)):
+ if (deps <= 2*10**(nidx+2)):
if (j == 1):
print '{}eps ({})'.format(deps, dinf),
else:
diff --git a/hysop/problem.py b/hysop/problem.py
index cbd34801c..84e5eaf5c 100644
--- a/hysop/problem.py
+++ b/hysop/problem.py
@@ -81,13 +81,8 @@ class Problem(ComputationalGraph):
self.check_unique_clenv()
def check_unique_clenv(self):
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
-
- cl_env, op = None, None
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
+ cl_env, first_op = None, None
+ for op in self.nodes:
for topo in set(op.input_fields.values() + op.output_fields.values()):
if (topo.backend.kind == Backend.OPENCL):
if (cl_env is None):
@@ -105,13 +100,8 @@ class Problem(ComputationalGraph):
def initialize_field(self, field, **kwds):
"""Initialize a field on all its input and output topologies."""
- reduced_graph = self.reduced_graph
- operators = reduced_graph.vertex_properties['operators']
-
initialized = set()
- for vid in self.sorted_nodes:
- vertex = reduced_graph.vertex(vid)
- op = operators[vertex]
+ for op in self.nodes:
# give priority to tensor field initialization
for op_fields in (self.input_discrete_tensor_fields, self.output_discrete_tensor_fields,
op.input_discrete_tensor_fields, op.output_discrete_tensor_fields,
--
GitLab