flow: insert splitters

examples/dataflow/arithmetic.py

@@ -29,31 +29,32 @@ class Dumper(SimActor):
 		super().__init__(dumper_gen(),
 			("result", Sink, [("r", BV(32))]))
 
+def draw(g):
+	if len(sys.argv) > 1 and sys.argv[1] == "draw":
+		nx.draw(g)
+		plt.show()
+
 def main():
 	# Create graph
 	g = DataFlowGraph()
-	a1 = ComposableSource(g, NumberGen())
+	gen1 = ComposableSource(g, NumberGen())
-	a2 = ComposableSource(g, NumberGen())
+	gen2 = ComposableSource(g, NumberGen())
-	a3 = ComposableSource(g, NumberGen())
-	c3 = (a1 + a2)*a3
-	g.add_connection(c3.actor_node, Dumper())
 	
-	a1.actor_node.actor.name = "gen1"
+	ps = gen1 + gen2
-	a2.actor_node.actor.name = "gen2"
+	result = ps*gen1 + ps*gen2
-	a3.actor_node.actor.name = "gen3"
+	
-	c3.actor_node.name = "result"
+	g.add_connection(result.actor_node, Dumper())
+
+	gen1.actor_node.actor.name = "gen1"
+	gen2.actor_node.actor.name = "gen2"
+	result.actor_node.name = "result"
 	
 	# Elaborate
-	draw = len(sys.argv) > 1 and sys.argv[1] == "draw"
 	print("is_abstract before elaboration: " + str(g.is_abstract()))
-	if draw:
+	draw(g)
-		nx.draw(g)
-		plt.show()
 	g.elaborate()
 	print("is_abstract after elaboration : " + str(g.is_abstract()))
-	if draw:
+	draw(g)
-		nx.draw(g)
-		plt.show()
 
 	# Simulate
 	c = CompositeActor(g)

migen/flow/network.py

@@ -63,7 +63,11 @@ class DataFlowGraph(MultiDiGraph):
 	
 	def del_connections(self, source_node, sink_node, data_requirements):
 		edges_to_delete = []
-		for key, data in self.get_edge_data(source_node, sink_node).items():
+		edge_data = self.get_edge_data(source_node, sink_node)
+		if edge_data is None:
+			# the two nodes are already completely disconnected
+			return
+		for key, data in edge_data.items():
 			if all(k not in data_requirements or data_requirements[k] == v
 			  for k, v in data.items()):
 				edges_to_delete.append(key)
@@ -115,9 +119,9 @@ class DataFlowGraph(MultiDiGraph):
 		return any(x.is_abstract() for x in self) \
 			or any(d["source_subr"] is not None or d["sink_subr"] is not None
 				for u, v, d in self.edges_iter(data=True)) \
-			or self._list_divergences()
+			or bool(self._list_divergences())
 	
-	def _eliminate_subrecords(self):
+	def _eliminate_subrecords_and_divergences(self):
 		# Insert combinators.
 		for (dst_node, dst_endpoint), sources in self._sink_to_sources().items():
 			if len(sources) > 1 or sources[0][2] is not None:
@@ -135,10 +139,19 @@ class DataFlowGraph(MultiDiGraph):
 				# connect combinator_source -> sink
 				self.add_connection(combinator, dst_node, "source", dst_endpoint)
 		# Insert splitters.
-		# TODO
+		for (src_node, src_endpoint), sinks in self._source_to_sinks().items():
-	
+			if len(sinks) > 1 or sinks[0][2] is not None:
-	def _eliminate_divergences(self):
+				subrecords = [src_subrecord for dst_node, dst_endpoint, src_subrecord in sinks]
-		pass # TODO
+				splitter = ActorNode(plumbing.Splitter, {"subrecords": subrecords})
+				# disconnect source -> sink1 ... source -> sinkn
+				# connect splitter_source1 -> sink1 ... splitter_sourcen -> sinkn
+				for n, (dst_node, dst_endpoint, src_subrecord) in enumerate(sinks):
+					self.del_connections(src_node, dst_node,
+						{"source": src_endpoint, "sink": dst_endpoint})
+					self.add_connection(splitter, dst_node,
+						"source{0}".format(n), dst_endpoint)
+				# connect source -> splitter_sink
+				self.add_connection(src_node, splitter, src_endpoint, "sink")
 	
 	def _infer_plumbing_layout(self):
 		while True:
@@ -182,17 +195,16 @@ class DataFlowGraph(MultiDiGraph):
 				d["sink"] = sink_eps[0]
 	
 	# Elaboration turns an abstract DFG into a concrete one.
-	#   Pass 1: eliminate subrecords by inserting Combinator/Splitter actors
+	#   Pass 1: eliminate subrecords and divergences
-	#   Pass 2: eliminate divergences by inserting Distributor actors
+	#           by inserting Combinator/Splitter actors
-	#   Pass 3: run optimizer (e.g. share and duplicate actors)
+	#   Pass 2: run optimizer (e.g. share and duplicate actors)
-	#   Pass 4: instantiate all abstract actors and explicit "None" endpoints
+	#   Pass 3: instantiate all abstract actors and explicit "None" endpoints
 	def elaborate(self, optimizer=None):
 		if self.elaborated:
 			return
 		self.elaborated = True
 		
-		self._eliminate_subrecords()
+		self._eliminate_subrecords_and_divergences()
-		self._eliminate_divergences()
 		if optimizer is not None:
 			optimizer(self)
 		self._instantiate_actors()

migen/flow/plumbing.py

@@ -35,20 +35,39 @@ class Combinator(CombinatorialActor):
 class Splitter(CombinatorialActor):
 	def __init__(self, layout, subrecords):
 		sink = Record(layout)
-		subrecords = [sink.subrecord(*subr) for subr in subrecords]
+		subr = []
+		for s in subrecords:
+			if s is None:
+				subr.append(sink)
+			else:
+				subr.append(sink.subrecord(*s))
 		eps = [("source{0}".format(n), Source, r)
-			for n, r in enumerate(subrecords)]
+			for n, r in enumerate(subr)]
 		ep_sink = ("sink", Sink, sink)
 		eps.append(ep_sink)
 		super().__init__(*eps)
 		
-	# TODO def get_fragment(self):
+	def get_fragment(self):
+		sources = [self.endpoints[e] for e in self.sources()]
+		sink = self.endpoints[self.sinks()[0]]
 		
-class Distributor:
+		already_acked = Signal(BV(len(sources)))
-	pass # TODO
+		sync = [
+			If(sink.stb,
+				already_acked.eq(already_acked | Cat(*[s.ack for s in sources])),
+				If(sink.ack, already_acked.eq(0))
+			)
+		]
+		comb = [
+			sink.ack.eq(optree("&",
+				[s.ack | already_acked[n] for n, s in enumerate(sources)]))
+		]
+		for n, s in enumerate(sources):
+			comb.append(s.stb.eq(sink.stb & ~already_acked[n]))
+		return Fragment(comb, sync)
 
 # Actors whose layout should be inferred from what their single sink is connected to.
-layout_sink = {Buffer, Splitter, Distributor}
+layout_sink = {Buffer, Splitter}
 # Actors whose layout should be inferred from what their single source is connected to.
 layout_source = {Buffer, Combinator}
 # All actors.