@@ -82,7 +82,11 @@ import Data.Functor (void)
8282import Data.Heap (Heap, Entry(..))
8383import qualified Data.Heap as H
8484import Data.IORef (IORef, modifyIORef, newIORef,
85− readIORef, atomicModifyIORef)
85+ readIORef, atomicModifyIORef
86+#ifdef DIAGNOSTICS
87+ , writeIORef
88+#endif
89+ )
8690import Data.Maybe (isNothing, fromJust)
8791import Data.Semigroup (Semigroup(..))
8892import Data.Set (Set)
@@ -187,6 +191,12 @@ data SVar m a =
187191 -- Shared, thread tracking
188192 , runningThreads :: IORef (Set ThreadId)
189193 , activeWorkers :: IORef Int
194+#ifdef DIAGNOSTICS
195+ , maxWorkers :: IORef Int
196+ , maxOutQSize :: IORef Int
197+ , maxHeapSize :: IORef Int
198+ , maxWorkQSize :: IORef Int
199+#endif
190200 }
191201
192202#ifdef DIAGNOSTICS
@@ -200,21 +210,25 @@ dumpSVar sv = do
200210 aheadDump <-
201211 if svarStyle sv == AheadVar
202212 then do
203− (oheap, oheapSize) <- readIORef $ outputHeap sv
213+ (oheap, oheapSeq) <- readIORef $ outputHeap sv
204214 hsrn <- tryTakeMVar $ heapSiren sv
205− (wq, wqSize) <- readIORef $ workQueue sv
215+ (wq, wqSeq) <- readIORef $ workQueue sv
216+ maxHp <- readIORef $ maxHeapSize sv
206217 return $ unlines
207− [ "heap length computed = " ++ show (H.size oheap)
208− , "heap length maintained = " ++ show oheapSize
218+ [ "heap length = " ++ show (H.size oheap)
219+ , "heap seqeunce = " ++ show oheapSeq
209220 , "heap siren = " ++ show hsrn
210− , "work queue length computed = " ++ show (length wq)
211− , "work queue length maintained = " ++ show wqSize
221+ , "work queue length = " ++ show (length wq)
222+ , "work queue sequence = " ++ show wqSeq
223+ , "heap max size = " ++ show maxHp
212224 ]
213225 else return []
214226
215227 waiting <- readIORef $ waitingForWork sv
216228 rthread <- readIORef $ runningThreads sv
217229 workers <- readIORef $ activeWorkers sv
230+ maxWrk <- readIORef $ maxWorkers sv
231+ maxOq <- readIORef $ maxOutQSize sv
218232 -- XXX queueEmpty should be made IO return type
219233
220234 return $ unlines
@@ -224,6 +238,8 @@ dumpSVar sv = do
224238 , "outputQueue length maintained = " ++ show oqLen
225239 , "output doorBell = " ++ show db
226240 , "worker siren = " ++ show srn
241+ , "max workers = " ++ show maxWrk
242+ , "max outQSize = " ++ show maxOq
227243 ]
228244 ++ aheadDump ++ unlines
229245 [ "waitingForWork = " ++ show waiting
@@ -584,7 +600,7 @@ runOne sv m = (runStream m) (Just sv) stop single yield
584600{-# INLINE sendToHeap #-}
585601sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
586602sendToHeap sv msg seqNo = do
587− let maxHeapSize = 1500
603+ let maxHeap = 1500
588604
589605 -- Maybe we can use a lock instead of using CAS
590606 -- insertions in the heap may be expensive if the heap is big
@@ -597,15 +613,15 @@ sendToHeap sv msg seqNo = do
597613
598614 -- this is only for the case when the consumer generated siren can get
599615 -- missed due to race.
600− when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()
616+ when (len < maxHeap) $ void $ tryPutMVar (heapSiren sv) ()
601617
602618 if added
603619 then do
604− when (len + 1 >= maxHeapSize) $ do
620+ when (len + 1 >= maxHeap) $ do
605621 withDBGMVar sv "sendToHeap: wait for heapSiren"
606622 $ takeMVar (heapSiren sv)
607623 (h, _) <- readIORef (outputHeap sv)
608− when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
624+ when (H.size h <= maxHeap) $ void $ tryPutMVar (siren sv) ()
609625 return False
610626 else do
611627 case msg of
@@ -659,6 +675,11 @@ drainHeap sv nextSeqNo = do
659675-- heap.
660676releaseToken :: SVar m a -> Int -> IO ()
661677releaseToken sv nextSeqNo = do
678+#ifdef DIAGNOSTICS
679+ maxHp <- readIORef (maxHeapSize sv)
680+ (hp, _) <- readIORef (outputHeap sv)
681+ when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
682+#endif
662683 drainHeap sv nextSeqNo
663684 void $ tryPutMVar (heapSiren sv) ()
664685
@@ -806,10 +827,21 @@ handleChildException sv e = do
806827 tid <- myThreadId
807828 send sv (ChildStop tid (Just e))
808829
830+#ifdef DIAGNOSTICS
831+recordMaxWorkers :: MonadIO m => SVar m a -> m ()
832+recordMaxWorkers sv = liftIO $ do
833+ active <- readIORef (activeWorkers sv)
834+ maxWrk <- readIORef (maxWorkers sv)
835+ when (active > maxWrk) $ writeIORef (maxWorkers sv) active
836+#endif
837+
809838{-# NOINLINE pushWorker #-}
810839pushWorker :: MonadAsync m => SVar m a -> m ()
811840pushWorker sv = do
812841 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
842+#ifdef DIAGNOSTICS
843+ recordMaxWorkers sv
844+#endif
813845 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
814846
815847-- | In contrast to pushWorker which always happens only from the consumer
@@ -822,7 +854,10 @@ pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
822854pushWorkerPar sv m = do
823855 -- We do not use activeWorkers in case of ParallelVar but still there is no
824856 -- harm in maintaining it correctly.
825− -- liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
857+#ifdef DIAGNOSTICS
858+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
859+ recordMaxWorkers sv
860+#endif
826861 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
827862
828863-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
@@ -912,7 +947,13 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
912947 res <- liftIO $ tryTakeMVar (doorBell sv)
913948 when (isNothing res) $ sendWorkerWait sv
914949
950+#ifdef DIAGNOSTICS
951+ (list, len) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
952+ oqLen <- liftIO $ readIORef (maxOutQSize sv)
953+ when (len > oqLen) $ liftIO $ writeIORef (maxOutQSize sv) len
954+#else
915955 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
956+#endif
916957 void $ liftIO $ tryPutMVar (siren sv) ()
917958 -- Reversing the output is important to guarantee that we process the
918959 -- outputs in the same order as they were generated by the constituent
@@ -933,7 +974,14 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
933974 done <- allThreadsDone sv
934975 if not done
935976 then (runStream (fromStreamVar sv)) Nothing stp sng yld
977+#ifdef DIAGNOSTICS
978+ else do
979+ svInfo <- liftIO $ dumpSVar sv
980+ liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo
981+ stp
982+#else
936983 else stp
984+#endif
937985
938986 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
939987 let continue = (runStream (processEvents es)) Nothing stp sng yld
@@ -958,6 +1006,12 @@ getFifoSVar ctype = do
9581006 wfw <- newIORef False
9591007 running <- newIORef S.empty
9601008 q <- newQ
1009+#ifdef DIAGNOSTICS
1010+ maxWrk <- newIORef 0
1011+ maxOq <- newIORef 0
1012+ maxHs <- newIORef 0
1013+ maxWq <- newIORef 0
1014+#endif
9611015 let sv =
9621016 SVar { outputQueue = outQ
9631017 , doorBell = outQMv
@@ -972,6 +1026,12 @@ getFifoSVar ctype = do
9721026 , waitingForWork = wfw
9731027 , svarStyle = ctype
9741028 , activeWorkers = active
1029+#ifdef DIAGNOSTICS
1030+ , maxWorkers = maxWrk
1031+ , maxOutQSize = maxOq
1032+ , maxHeapSize = maxHs
1033+ , maxWorkQSize = maxWq
1034+#endif
9751035 }
9761036 in return sv
9771037
@@ -984,6 +1044,12 @@ getLifoSVar ctype = do
9841044 wfw <- newIORef False
9851045 running <- newIORef S.empty
9861046 q <- newIORef []
1047+#ifdef DIAGNOSTICS
1048+ maxWrk <- newIORef 0
1049+ maxOq <- newIORef 0
1050+ maxHs <- newIORef 0
1051+ maxWq <- newIORef 0
1052+#endif
9871053 let checkEmpty = null <$> liftIO (readIORef q)
9881054 let sv =
9891055 SVar { outputQueue = outQ
@@ -999,6 +1065,12 @@ getLifoSVar ctype = do
9991065 , waitingForWork = wfw
10001066 , svarStyle = ctype
10011067 , activeWorkers = active
1068+#ifdef DIAGNOSTICS
1069+ , maxWorkers = maxWrk
1070+ , maxOutQSize = maxOq
1071+ , maxHeapSize = maxHs
1072+ , maxWorkQSize = maxWq
1073+#endif
10021074 }
10031075 in return sv
10041076
@@ -1010,6 +1082,12 @@ getParSVar style = do
10101082 active <- newIORef 0
10111083 wfw <- newIORef False
10121084 running <- newIORef S.empty
1085+#ifdef DIAGNOSTICS
1086+ maxWrk <- newIORef 0
1087+ maxOq <- newIORef 0
1088+ maxHs <- newIORef 0
1089+ maxWq <- newIORef 0
1090+#endif
10131091 let sv =
10141092 SVar { outputQueue = outQ
10151093 , doorBell = outQMv
@@ -1024,6 +1102,12 @@ getParSVar style = do
10241102 , waitingForWork = wfw
10251103 , svarStyle = style
10261104 , activeWorkers = active
1105+#ifdef DIAGNOSTICS
1106+ , maxWorkers = maxWrk
1107+ , maxOutQSize = maxOq
1108+ , maxHeapSize = maxHs
1109+ , maxWorkQSize = maxWq
1110+#endif
10271111 }
10281112 in return sv
10291113
@@ -1038,6 +1122,14 @@ getAheadSVar style = do
10381122 wfw <- newIORef False
10391123 running <- newIORef S.empty
10401124 q <- newIORef ([], -1)
1125+
1126+#ifdef DIAGNOSTICS
1127+ maxWrk <- newIORef 0
1128+ maxOq <- newIORef 0
1129+ maxHs <- newIORef 0
1130+ maxWq <- newIORef 0
1131+#endif
1132+
10411133 let checkEmpty = liftIO $ do
10421134 (xs, _) <- readIORef q
10431135 return $ null xs
@@ -1055,6 +1147,13 @@ getAheadSVar style = do
10551147 , waitingForWork = wfw
10561148 , svarStyle = style
10571149 , activeWorkers = active
1150+
1151+#ifdef DIAGNOSTICS
1152+ , maxWorkers = maxWrk
1153+ , maxOutQSize = maxOq
1154+ , maxHeapSize = maxHs
1155+ , maxWorkQSize = maxWq
1156+#endif
10581157 }
10591158 in return sv
10601159