implement blackhole blocking (eager blackholing, no time window for thunk duplication) ; fix RestoreExMask evaluation, it passes the current result to raiseAsyncEx which needs it for correct ApStack construction

Author: csabahruska

external-stg-interpreter/lib/Stg/Interpreter.hs

@@ -184,26 +184,24 @@ builtinStgEval so a@HeapPtr{} = do
   Debugger.checkBreakpoint $ BkpCustom "eval"
   case o of
     ApStack{..} -> do
-      stackPush (Apply []) -- ensure WHNF
+      let HeapPtr l = a
+      store l (BlackHole o [])  -- HINT: prevent duplicate computation
+      stackPush (Update l)      -- HINT: ensure sharing, ApStack is always created from Update frame
       mapM_ stackPush (reverse hoStack)
       pure hoResult
-    RaiseException ex -> mylog (show o) >> PrimExceptions.raiseEx ex
+    RaiseException ex -> PrimExceptions.raiseEx ex
     Con{}       -> pure [a]
-    
-    BlackHole _ t -> do
+
+    BlackHole ho waitingThreads -> do
+      let HeapPtr addr = a
+      tid <- gets ssCurrentThreadId
+      ts <- getThreadState tid
+      updateThreadState tid (ts {tsStatus = ThreadBlocked (BlockedOnBlackHole addr)})
+      store addr (BlackHole ho $ tid : waitingThreads)
       stackPush (Apply []) -- retry evaluation next time also
-      stackPush $ RunScheduler SR_ThreadYield
+      stackPush $ RunScheduler SR_ThreadBlocked
       pure [a]
-    
-    {-
-    -- TODO: check how the cmm stg machine handles this case
-    BlackHole t -> do
-                    Rts{..} <- gets ssRtsSupport
-                    liftIO $ do
-                      hPutStrLn stderr $ takeBaseName rtsProgName ++ ": <<loop>>"
-                      exitWith ExitSuccess
-                    stgErrorM $ "blackhole ; loop in evaluation of : " ++ show t
-    -}
+
     Closure{..}
       | hoCloMissing /= 0
       -> pure [a]
@@ -249,21 +247,20 @@ builtinStgEval so a@HeapPtr{} = do
             -- closure may be entered multiple times, but should not be updated or blackholed.
             evalExpr extendedEnv e
           Updatable -> do
-            tid <- gets ssCurrentThreadId
             -- closure should be updated after evaluation (and may be blackholed during evaluation).
             -- Q: what is eager and lazy blackholing?
             --  read: http://mainisusuallyafunction.blogspot.com/2011/10/thunks-and-lazy-blackholes-introduction.html
             --  read: https://www.microsoft.com/en-us/research/wp-content/uploads/2005/09/2005-haskell.pdf
             stackPush (Update l)
-            --store l (BlackHole tid o)
+            store l (BlackHole o [])
             evalExpr extendedEnv e
           SingleEntry -> do
             tid <- gets ssCurrentThreadId
             -- TODO: investigate how does single-entry blackholing cause problem (estgi does not have racy memops as it is mentioned in GHC Note below)
             -- no backholing, see: GHC Note [Black-holing non-updatable thunks]
             -- closure will only be entered once, and so need not be updated but may safely be blackholed.
             --stackPush (Update l) -- FIX??? Q: what will remove the backhole if there is no update? Q: is the value linear?
-            --store l (BlackHole tid o) -- Q: is this a bug?
+            --store l (BlackHole o) -- Q: is this a bug?
             evalExpr extendedEnv e
     _ -> stgErrorM $ "expected evaluable heap object, got: " ++ show a ++ " heap-object: " ++ show o ++ " static-origin: " ++ show so
 builtinStgEval so a = stgErrorM $ "expected a thunk, got: " ++ show a ++ ", static-origin: " ++ show so
@@ -276,18 +273,24 @@ builtinStgApply so a@HeapPtr{} args = do
   o <- readHeap a
   case o of
     ApStack{..} -> do
+      let HeapPtr l = a
+      store l (BlackHole o [])  -- HINT: prevent duplicate computation
       stackPush (Apply args)
+      stackPush (Update l)      -- HINT: ensure sharing, ApStack is always created from Update frame
       mapM_ stackPush (reverse hoStack)
       pure hoResult
-    RaiseException ex -> mylog (show o) >> PrimExceptions.raiseEx ex
+    RaiseException ex -> PrimExceptions.raiseEx ex
     Con{}             -> stgErrorM $ "unexpected con at apply: " ++ show o ++ ", args: " ++ show args ++ ", static-origin: " ++ show so
-    --BlackHole t       -> stgErrorM $ "blackhole ; loop in application of : " ++ show t
-    {-
-    BlackHole t -> do
-      stackPush (Apply args)
-      stackPush $ RunScheduler SR_ThreadYield
+
+    BlackHole ho waitingThreads -> do
+      tid <- gets ssCurrentThreadId
+      ts <- getThreadState tid
+      updateThreadState tid (ts {tsStatus = ThreadBlocked (BlockedOnBlackHole addr)})
+      store addr (BlackHole ho $ tid : waitingThreads)
+      stackPush (Apply args) -- retry evaluation next time also
+      stackPush $ RunScheduler SR_ThreadBlocked
       pure [a]
-    -}
+
     Closure{..}
       -- under saturation
       | hoCloMissing - argCount > 0
@@ -483,6 +486,7 @@ evalStackContinuation result = \case
   Update dstAddr
     | [src@HeapPtr{}] <- result
     -> do
+      wakeupBlackHoleQueueThreads dstAddr
       o <- readHeap src
       store dstAddr o
       dynamicHeapStartAddr <- gets ssDynamicHeapStart
@@ -557,7 +561,7 @@ evalStackContinuation result = \case
           -- raise exception
           ts <- getCurrentThreadState
           updateThreadState tid ts {tsBlockedExceptions = waitingTids}
-          PrimConcurrency.raiseAsyncEx (tsCurrentResult ts) tid exception
+          PrimConcurrency.raiseAsyncEx result tid exception
       _ -> pure ()
     pure result
 

external-stg-interpreter/lib/Stg/Interpreter/Base.hs

@@ -162,7 +162,8 @@ data HeapObject
     , hoCloArgs     :: [Atom]
     , hoCloMissing  :: Int    -- HINT: this is a Thunk if 0 arg is missing ; if all is missing then Fun ; Pap is some arg is provided
     }
-  | BlackHole Int HeapObject
+  | BlackHole HeapObject [Int] -- original heap object, blocking queue of thread ids
+                               -- NOTE: each blackhole has exactly one corresponding thread and one update frame
   | ApStack                   -- HINT: needed for the async exceptions
     { hoResult      :: [Atom]
     , hoStack       :: [StackContinuation]
@@ -632,6 +633,7 @@ data Rts
   -- closures used by the GC deadlock detection
   , rtsBlockedIndefinitelyOnMVar  :: Atom -- (exception)
   , rtsBlockedIndefinitelyOnSTM   :: Atom -- (exception)
+  , rtsNonTermination             :: Atom -- (exception)
 
   -- rts helper custom closures
   , rtsApplyFun1Arg :: Atom
@@ -1156,7 +1158,7 @@ NOTE:
 data BlockReason
   = BlockedOnMVar         Int (Maybe Atom) -- mvar id, the value that need to put to mvar in case of blocking putMVar#, in case of takeMVar this is Nothing
   | BlockedOnMVarRead     Int       -- mvar id
-  | BlockedOnBlackHole
+  | BlockedOnBlackHole    Int       -- heap address
   | BlockedOnThrowAsyncEx Int       -- target thread id
   | BlockedOnSTM          TLog
   | BlockedOnForeignCall            -- RTS name: BlockedOnCCall
@@ -1445,7 +1447,7 @@ debugPrintHeapObject :: HeapObject -> String
 debugPrintHeapObject  = \case
   Con{..}           -> "Con: " ++ show (dcUniqueName $ unDC hoCon) ++ " " ++ show hoConArgs
   Closure{..}       -> "Clo: " ++ show hoName ++ " args: " ++ show hoCloArgs ++ " env: " ++ show (Map.size hoEnv) ++ " missing: " ++ show hoCloMissing
-  BlackHole t o     -> "BlackHole - tid: " ++ show t ++ " "  ++ debugPrintHeapObject o
+  BlackHole o _q    -> "BlackHole: " ++ debugPrintHeapObject o
   ApStack{}         -> "ApStack"
   RaiseException ex -> "RaiseException: " ++ show ex
 
@@ -1478,4 +1480,16 @@ mylog msg = do
   liftIO $ do
     BS8.putStrLn . BS8.pack $ msg ++ " " ++ show pp ++ " " ++ show ctid
     hFlush stdout
--}
+-}
+
+wakeupBlackHoleQueueThreads :: Int -> M ()
+wakeupBlackHoleQueueThreads addr = readHeap (HeapPtr addr) >>= \case
+  (BlackHole _ waitingThreads) -> do
+    -- wake up blocked threads
+    forM_ waitingThreads $ \waitingTid -> do
+      waitingTS <- getThreadState waitingTid
+      case tsStatus waitingTS of
+        ThreadBlocked (BlockedOnBlackHole dstAddr) -> do
+          updateThreadState waitingTid (waitingTS {tsStatus = ThreadRunning})
+        _ -> error $ "internal error - invalid thread status: " ++ show (tsStatus waitingTS)
+  x -> error $ "internal error - expected BlackHole, got: " ++ show x

external-stg-interpreter/lib/Stg/Interpreter/Debugger/Datalog.hs

@@ -203,7 +203,7 @@ exportStgStateM stgState@StgState{..} = do
       forM_ (zip [0..] (Map.toList hoEnv)) $ \(idx, (n, a)) -> do
         addFact "Heap_ClosureEnv" [I i, I idx, ID n, A $ snd a]
 
-    BlackHole _ o -> do
+    BlackHole o _ -> do
       addFact "Heap_BlackHole" [I i, S (debugPrintHeapObject o)]
 
     ApStack{..} -> do

external-stg-interpreter/lib/Stg/Interpreter/Debugger/UI.hs

@@ -129,7 +129,7 @@ printHeapObject = \case
     printEnv hoEnv
     putStrLn $ "source location: " ++ (ppSrcSpan . binderDefLoc . unId $ hoName)
 
-  BlackHole _ ho -> do
+  BlackHole ho _ -> do
     putStrLn "BlackHole:"
     printHeapObject ho
     putStrLn ""

external-stg-interpreter/lib/Stg/Interpreter/GC/LiveDataAnalysis.hs

@@ -127,7 +127,7 @@ addGCRootFacts prog StgState{..} localGCRoots = do
     ThreadBlocked r -> case r of
       BlockedOnMVar{}         -> pure () -- will be referred by the mvar wait queue
       BlockedOnMVarRead{}     -> pure () -- will be referred by the mvar wait queue
-      BlockedOnBlackHole{}    -> error "not implemented yet"
+      BlockedOnBlackHole{}    -> pure () -- will be referred by the BlockedOnBlackHole ADDR thread status
       BlockedOnThrowAsyncEx{} -> pure () -- will be referred by the target thread's blocked exceptions queue
       BlockedOnSTM{}          -> pure () -- will be referred by the tvar wait queue
       BlockedOnForeignCall{}  -> error "not implemented yet"
@@ -163,6 +163,17 @@ addReferenceFacts prog StgState{..} = do
   addRefs ssStablePointers      NS_StablePointer
   addRefs ssThreads             NS_Thread
 
+  -- references for backhole wait queues
+  let blackholes = [ (tid, addr)
+                   | (tid, ts) <- IntMap.toList ssThreads
+                   , Update addr <- tsStack ts
+                   ]
+  forM_ blackholes $ \(tid, addr) -> case IntMap.lookup addr ssHeap of
+    Just (BlackHole _ waitingThreads) -> do
+      forM_ waitingThreads $ \waitingTid -> do
+        addReference (encodeRef tid NS_Thread) (encodeRef waitingTid NS_Thread)
+    ho -> error $ "internal error - expected Blackhole, got: " ++ show ho
+
   -- stable name references
   let stableNames = Map.toList ssStableNameMap
   forM_ stableNames $ \(v, i) -> visitGCRef (addReference (encodeRef i NS_StableName)) v
@@ -180,7 +191,7 @@ addMaybeDeadlockingThreadFacts prog StgState{..} = do
     ThreadBlocked r -> case r of
       BlockedOnMVar{}         -> addMaybeDeadlockingThread $ encodeRef tid NS_Thread
       BlockedOnMVarRead{}     -> addMaybeDeadlockingThread $ encodeRef tid NS_Thread
-      BlockedOnBlackHole{}    -> error "not implemented yet"
+      BlockedOnBlackHole{}    -> addMaybeDeadlockingThread $ encodeRef tid NS_Thread
       BlockedOnThrowAsyncEx{} -> addMaybeDeadlockingThread $ encodeRef tid NS_Thread
       BlockedOnSTM{}          -> addMaybeDeadlockingThread $ encodeRef tid NS_Thread
       BlockedOnForeignCall{}  -> error "not implemented yet"

external-stg-interpreter/lib/Stg/Interpreter/PrimOp/Concurrency.hs

@@ -183,7 +183,7 @@ evalPrimOp fallback op args t tc = case (op, args) of
           ThreadBlocked r -> case r of
             BlockedOnMVar{}         -> 1
             BlockedOnMVarRead{}     -> 14
-            BlockedOnBlackHole      -> 2
+            BlockedOnBlackHole{}    -> 2
             BlockedOnSTM{}          -> 6
             BlockedOnForeignCall    -> 10
             BlockedOnRead{}         -> 3
@@ -219,10 +219,12 @@ raiseAsyncEx lastResult targetTid exception = do
 
         -- replace Update with ApStack
         Update addr : stackTail -> do
+          when (result == []) $ error "internal error - result should be a [HeapPtr], but it's value is []"
           let apStack = ApStack
                 { hoResult  = result
                 , hoStack   = reverse stackPiece
                 }
+          wakeupBlackHoleQueueThreads addr
           store addr apStack
           let newResult = [HeapPtr addr]
           ctid <- gets ssCurrentThreadId
@@ -291,17 +293,24 @@ removeFromQueues tid = do
   ThreadState{..} <- getThreadState tid
   -- Q: what about the async exception queue?
   case tsStatus of
-    ThreadRunning                       -> pure ()
-    ThreadBlocked (BlockedOnMVar m _)   -> removeFromMVarQueue tid m
-    ThreadBlocked (BlockedOnMVarRead m) -> removeFromMVarQueue tid m
-    ThreadBlocked (BlockedOnSTM tlog)   -> do
-      unsubscribeTVarWaitQueues tid tlog
-    ThreadBlocked BlockedOnDelay{}      -> pure () -- HINT: no queue for delays
-    ThreadBlocked BlockedOnRead{}       -> pure () -- HINT: no queue for file read
-    ThreadBlocked BlockedOnWrite{}      -> pure () -- HINT: no queue for file write
+    ThreadRunning                           -> pure ()
+    ThreadBlocked (BlockedOnMVar m _)       -> removeFromMVarQueue tid m
+    ThreadBlocked (BlockedOnMVarRead m)     -> removeFromMVarQueue tid m
+    ThreadBlocked (BlockedOnSTM tlog)       -> unsubscribeTVarWaitQueues tid tlog
+    ThreadBlocked BlockedOnDelay{}          -> pure () -- HINT: no queue for delays
+    ThreadBlocked BlockedOnRead{}           -> pure () -- HINT: no queue for file read
+    ThreadBlocked BlockedOnWrite{}          -> pure () -- HINT: no queue for file write
+    ThreadBlocked BlockedOnThrowAsyncEx{}   -> pure () -- Q: what to do?
+    ThreadBlocked (BlockedOnBlackHole addr) -> removeFromBlackHoleQueue tid addr
     _ -> error $ "TODO: removeFromQueues " ++ show tsStatus
 
 removeFromMVarQueue :: Int -> Int -> M ()
 removeFromMVarQueue tid m = do
   let filterFun mvd@MVarDescriptor{..} = mvd {mvdQueue = filter (tid /=) mvdQueue}
   modify' $ \s@StgState{..} -> s {ssMVars = IntMap.adjust filterFun m ssMVars}
+
+removeFromBlackHoleQueue :: Int -> Int -> M ()
+removeFromBlackHoleQueue tid addr = do
+  readHeap (HeapPtr addr) >>= \case
+    (BlackHole o queue) -> modify' $ \s@StgState{..} -> s { ssHeap = IntMap.insert addr (BlackHole o $ filter (tid /=) queue) ssHeap }
+    x                   -> error $ "internal error - expected BlackHole, got: " ++ show x