support gather

Author: shayne-fletcher

app/Main.hs

@@ -12,7 +12,7 @@ main = do
       scheduler = BFSScheduler
 
       broadcast = buildSchedule scheduler tiling members shape
-      reduce = reverseSchedule broadcast
+      converge = reverseSchedule broadcast
 
       full = rootTile shape
       occE = Occlusion (== "E")
@@ -109,7 +109,7 @@ main = do
 
   putStrLn "\nrunning full-mesh reduce:"
   runReduce
-    reduce
+    converge
     [ ("A", 1),
       ("B", 2),
       ("C", 3),
@@ -122,6 +122,20 @@ main = do
     (+)
     "A"
 
+  putStrLn "\nrunning full-mesh gather:"
+  runGather
+    converge
+    [ ("A", "value-a"),
+      ("B", "value-b"),
+      ("C", "value-c"),
+      ("D", "value-d"),
+      ("E", "value-e"),
+      ("F", "value-f"),
+      ("G", "value-g"),
+      ("H", "value-h")
+    ]
+    "A"
+
 expectTile :: String -> Maybe Tile -> Tile
 expectTile _ (Just tile) = tile
 expectTile label Nothing = error ("expected " ++ label)

src/Tile/Execution.hs

@@ -1,6 +1,7 @@
 module Tile.Execution
   ( adjacencyList,
     runBroadcast,
+    runGather,
     runReduce,
   )
 where
@@ -92,3 +93,47 @@ runReduce schedule initialValues combine root = do
       writeChan (chanMap Map.! m) (valueMap Map.! m)
 
   threadDelay 1000000
+
+runGather ::
+  (Show a) =>
+  Schedule String ->
+  [(String, a)] ->
+  String ->
+  IO ()
+runGather schedule initialValues root = do
+  let graph = adjacencyList schedule
+      incoming = incomingCounts schedule
+      members =
+        Set.toList $
+          Set.fromList $
+            Map.keys graph ++ concat (Map.elems graph) ++ map fst initialValues
+
+  chanPairs <- forM members $ \m -> do
+    ch <- newChan
+    pure (m, ch)
+
+  let chanMap = Map.fromList chanPairs
+      valueMap = Map.fromList initialValues
+
+  forM_ members $ \m -> do
+    let inbox = chanMap Map.! m
+        children = Map.findWithDefault [] m graph
+        childChans = [(c, chanMap Map.! c) | c <- children]
+        expected = Map.findWithDefault 0 m incoming
+        localValue = [(m, valueMap Map.! m)]
+
+    _ <- forkIO $ do
+      received <- concat <$> replicateM expected (readChan inbox)
+      let gathered = localValue ++ received
+      if m == root
+        then putStrLn $ m ++ " gathered result: " ++ show gathered
+        else forM_ childChans $ \(childName, childInbox) -> do
+          putStrLn $ m ++ " sending gathered values " ++ show gathered ++ " to " ++ childName
+          writeChan childInbox gathered
+    pure ()
+
+  forM_ members $ \m ->
+    when (Map.findWithDefault 0 m incoming == 0) $
+      writeChan (chanMap Map.! m) [(m, valueMap Map.! m)]
+
+  threadDelay 1000000