c9fdf94 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add more diagnostics to study the adaptation

c9fdf94Harendra Kumar8 years ago

benchmark/Adaptive.hs+8−1

compare with master

unifiedsplit

@@ -13,8 +13,15 @@ import Streamly

13import Streamly.Prelude as S

14import System.Random (randomRIO)

16+-- Note that we should also compare the cpuTime especially when threaded

17+-- runtime is used with this benchmark because thread scheduling is not

18+-- predictable and can add non-deterministic delay to the total time measured.

19+--

20+-- Also, the worker dispatch depends on the worker dispatch latency which is

21+-- set to fixed 200 us. We need to keep that in mind when designing tests.

22+

16value :: Int

23value :: Int

17−value = 10000

24+value = 1000

19{-# INLINE source #-}

26{-# INLINE source #-}

20source :: IsStream t => (Int, Int) -> t IO Int

27source :: IsStream t => (Int, Int) -> t IO Int

src/Streamly/Core.hs+111−12

@@ -82,7 +82,11 @@ import Data.Functor (void)

82import Data.Heap (Heap, Entry(..))

83import qualified Data.Heap as H

84import Data.IORef (IORef, modifyIORef, newIORef,

85− readIORef, atomicModifyIORef)

85+ readIORef, atomicModifyIORef

86+#ifdef DIAGNOSTICS

87+ , writeIORef

88+#endif

89+ )

86import Data.Maybe (isNothing, fromJust)

90import Data.Maybe (isNothing, fromJust)

87import Data.Semigroup (Semigroup(..))

91import Data.Semigroup (Semigroup(..))

88import Data.Set (Set)

92import Data.Set (Set)

@@ -187,6 +191,12 @@ data SVar m a =

187 -- Shared, thread tracking

191 -- Shared, thread tracking

188 , runningThreads :: IORef (Set ThreadId)

192 , runningThreads :: IORef (Set ThreadId)

189 , activeWorkers :: IORef Int

193 , activeWorkers :: IORef Int

194+#ifdef DIAGNOSTICS

195+ , maxWorkers :: IORef Int

196+ , maxOutQSize :: IORef Int

197+ , maxHeapSize :: IORef Int

198+ , maxWorkQSize :: IORef Int

199+#endif

190 }

200 }

191

201

192#ifdef DIAGNOSTICS

202#ifdef DIAGNOSTICS

@@ -200,21 +210,25 @@ dumpSVar sv = do

200 aheadDump <-

210 aheadDump <-

201 if svarStyle sv == AheadVar

211 if svarStyle sv == AheadVar

202 then do

212 then do

203− (oheap, oheapSize) <- readIORef $ outputHeap sv

213+ (oheap, oheapSeq) <- readIORef $ outputHeap sv

204 hsrn <- tryTakeMVar $ heapSiren sv

214 hsrn <- tryTakeMVar $ heapSiren sv

205− (wq, wqSize) <- readIORef $ workQueue sv

215+ (wq, wqSeq) <- readIORef $ workQueue sv

216+ maxHp <- readIORef $ maxHeapSize sv

206 return $ unlines

217 return $ unlines

207− [ "heap length computed = " ++ show (H.size oheap)

218+ [ "heap length = " ++ show (H.size oheap)

208− , "heap length maintained = " ++ show oheapSize

219+ , "heap seqeunce = " ++ show oheapSeq

209 , "heap siren = " ++ show hsrn

220 , "heap siren = " ++ show hsrn

210− , "work queue length computed = " ++ show (length wq)

221+ , "work queue length = " ++ show (length wq)

211− , "work queue length maintained = " ++ show wqSize

222+ , "work queue sequence = " ++ show wqSeq

223+ , "heap max size = " ++ show maxHp

212 ]

224 ]

213 else return []

225 else return []

214

226

215 waiting <- readIORef $ waitingForWork sv

227 waiting <- readIORef $ waitingForWork sv

216 rthread <- readIORef $ runningThreads sv

228 rthread <- readIORef $ runningThreads sv

217 workers <- readIORef $ activeWorkers sv

229 workers <- readIORef $ activeWorkers sv

230+ maxWrk <- readIORef $ maxWorkers sv

231+ maxOq <- readIORef $ maxOutQSize sv

218 -- XXX queueEmpty should be made IO return type

232 -- XXX queueEmpty should be made IO return type

219

233

220 return $ unlines

234 return $ unlines

@@ -224,6 +238,8 @@ dumpSVar sv = do

224 , "outputQueue length maintained = " ++ show oqLen

238 , "outputQueue length maintained = " ++ show oqLen

225 , "output doorBell = " ++ show db

239 , "output doorBell = " ++ show db

226 , "worker siren = " ++ show srn

240 , "worker siren = " ++ show srn

241+ , "max workers = " ++ show maxWrk

242+ , "max outQSize = " ++ show maxOq

227 ]

243 ]

228 ++ aheadDump ++ unlines

244 ++ aheadDump ++ unlines

229 [ "waitingForWork = " ++ show waiting

245 [ "waitingForWork = " ++ show waiting

@@ -584,7 +600,7 @@ runOne sv m = (runStream m) (Just sv) stop single yield

584{-# INLINE sendToHeap #-}

600{-# INLINE sendToHeap #-}

585sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool

601sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool

586sendToHeap sv msg seqNo = do

602sendToHeap sv msg seqNo = do

587− let maxHeapSize = 1500

603+ let maxHeap = 1500

588

604

589 -- Maybe we can use a lock instead of using CAS

605 -- Maybe we can use a lock instead of using CAS

590 -- insertions in the heap may be expensive if the heap is big

606 -- insertions in the heap may be expensive if the heap is big

@@ -597,15 +613,15 @@ sendToHeap sv msg seqNo = do

597

613

598 -- this is only for the case when the consumer generated siren can get

614 -- this is only for the case when the consumer generated siren can get

599 -- missed due to race.

615 -- missed due to race.

600− when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()

616+ when (len < maxHeap) $ void $ tryPutMVar (heapSiren sv) ()

601

617

602 if added

618 if added

603 then do

619 then do

604− when (len + 1 >= maxHeapSize) $ do

620+ when (len + 1 >= maxHeap) $ do

605 withDBGMVar sv "sendToHeap: wait for heapSiren"

621 withDBGMVar sv "sendToHeap: wait for heapSiren"

606 $ takeMVar (heapSiren sv)

622 $ takeMVar (heapSiren sv)

607 (h, _) <- readIORef (outputHeap sv)

623 (h, _) <- readIORef (outputHeap sv)

608− when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()

624+ when (H.size h <= maxHeap) $ void $ tryPutMVar (siren sv) ()

609 return False

625 return False

610 else do

626 else do

611 case msg of

627 case msg of

@@ -659,6 +675,11 @@ drainHeap sv nextSeqNo = do

659-- heap.

675-- heap.

660releaseToken :: SVar m a -> Int -> IO ()

676releaseToken :: SVar m a -> Int -> IO ()

661releaseToken sv nextSeqNo = do

677releaseToken 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

662 drainHeap sv nextSeqNo

683 drainHeap sv nextSeqNo

663 void $ tryPutMVar (heapSiren sv) ()

684 void $ tryPutMVar (heapSiren sv) ()

664

685

@@ -806,10 +827,21 @@ handleChildException sv e = do

806 tid <- myThreadId

827 tid <- myThreadId

807 send sv (ChildStop tid (Just e))

828 send sv (ChildStop tid (Just e))

808

829

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+

809{-# NOINLINE pushWorker #-}

838{-# NOINLINE pushWorker #-}

810pushWorker :: MonadAsync m => SVar m a -> m ()

839pushWorker :: MonadAsync m => SVar m a -> m ()

811pushWorker sv = do

840pushWorker sv = do

812 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

841 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

842+#ifdef DIAGNOSTICS

843+ recordMaxWorkers sv

844+#endif

813 doFork (runqueue sv) (handleChildException sv) >>= addThread sv

845 doFork (runqueue sv) (handleChildException sv) >>= addThread sv

814

846

815-- | In contrast to pushWorker which always happens only from the consumer

847-- | 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 ()

822pushWorkerPar sv m = do

854pushWorkerPar sv m = do

823 -- We do not use activeWorkers in case of ParallelVar but still there is no

855 -- We do not use activeWorkers in case of ParallelVar but still there is no

824 -- harm in maintaining it correctly.

856 -- harm in maintaining it correctly.

857+#ifdef DIAGNOSTICS

825− -- liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

858+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

859+ recordMaxWorkers sv

860+#endif

826 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv

861 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv

827

862

828-- XXX When the queue is LIFO we can put a limit on the number of dispatches.

863-- 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

912 res <- liftIO $ tryTakeMVar (doorBell sv)

947 res <- liftIO $ tryTakeMVar (doorBell sv)

913 when (isNothing res) $ sendWorkerWait sv

948 when (isNothing res) $ sendWorkerWait sv

914

949

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

915 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)

955 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)

956+#endif

916 void $ liftIO $ tryPutMVar (siren sv) ()

957 void $ liftIO $ tryPutMVar (siren sv) ()

917 -- Reversing the output is important to guarantee that we process the

958 -- Reversing the output is important to guarantee that we process the

918 -- outputs in the same order as they were generated by the constituent

959 -- outputs in the same order as they were generated by the constituent

@@ -933,7 +974,14 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do

933 done <- allThreadsDone sv

974 done <- allThreadsDone sv

934 if not done

975 if not done

935 then (runStream (fromStreamVar sv)) Nothing stp sng yld

976 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

936 else stp

983 else stp

984+#endif

937

985

938 processEvents (ev : es) = Stream $ \_ stp sng yld -> do

986 processEvents (ev : es) = Stream $ \_ stp sng yld -> do

939 let continue = (runStream (processEvents es)) Nothing stp sng yld

987 let continue = (runStream (processEvents es)) Nothing stp sng yld

@@ -958,6 +1006,12 @@ getFifoSVar ctype = do

958 wfw <- newIORef False

1006 wfw <- newIORef False

959 running <- newIORef S.empty

1007 running <- newIORef S.empty

960 q <- newQ

1008 q <- newQ

1009+#ifdef DIAGNOSTICS

1010+ maxWrk <- newIORef 0

1011+ maxOq <- newIORef 0

1012+ maxHs <- newIORef 0

1013+ maxWq <- newIORef 0

1014+#endif

961 let sv =

1015 let sv =

962 SVar { outputQueue = outQ

1016 SVar { outputQueue = outQ

963 , doorBell = outQMv

1017 , doorBell = outQMv

@@ -972,6 +1026,12 @@ getFifoSVar ctype = do

972 , waitingForWork = wfw

1026 , waitingForWork = wfw

973 , svarStyle = ctype

1027 , svarStyle = ctype

974 , activeWorkers = active

1028 , activeWorkers = active

1029+#ifdef DIAGNOSTICS

1030+ , maxWorkers = maxWrk

1031+ , maxOutQSize = maxOq

1032+ , maxHeapSize = maxHs

1033+ , maxWorkQSize = maxWq

1034+#endif

975 }

1035 }

976 in return sv

1036 in return sv

977

1037

@@ -984,6 +1044,12 @@ getLifoSVar ctype = do

984 wfw <- newIORef False

1044 wfw <- newIORef False

985 running <- newIORef S.empty

1045 running <- newIORef S.empty

986 q <- newIORef []

1046 q <- newIORef []

1047+#ifdef DIAGNOSTICS

1048+ maxWrk <- newIORef 0

1049+ maxOq <- newIORef 0

1050+ maxHs <- newIORef 0

1051+ maxWq <- newIORef 0

1052+#endif

987 let checkEmpty = null <$> liftIO (readIORef q)

1053 let checkEmpty = null <$> liftIO (readIORef q)

988 let sv =

1054 let sv =

989 SVar { outputQueue = outQ

1055 SVar { outputQueue = outQ

@@ -999,6 +1065,12 @@ getLifoSVar ctype = do

999 , waitingForWork = wfw

1065 , waitingForWork = wfw

1000 , svarStyle = ctype

1066 , svarStyle = ctype

1001 , activeWorkers = active

1067 , activeWorkers = active

1068+#ifdef DIAGNOSTICS

1069+ , maxWorkers = maxWrk

1070+ , maxOutQSize = maxOq

1071+ , maxHeapSize = maxHs

1072+ , maxWorkQSize = maxWq

1073+#endif

1002 }

1074 }

1003 in return sv

1075 in return sv

1004

1076

@@ -1010,6 +1082,12 @@ getParSVar style = do

1010 active <- newIORef 0

1082 active <- newIORef 0

1011 wfw <- newIORef False

1083 wfw <- newIORef False

1012 running <- newIORef S.empty

1084 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

1013 let sv =

1091 let sv =

1014 SVar { outputQueue = outQ

1092 SVar { outputQueue = outQ

1015 , doorBell = outQMv

1093 , doorBell = outQMv

@@ -1024,6 +1102,12 @@ getParSVar style = do

1024 , waitingForWork = wfw

1102 , waitingForWork = wfw

1025 , svarStyle = style

1103 , svarStyle = style

1026 , activeWorkers = active

1104 , activeWorkers = active

1105+#ifdef DIAGNOSTICS

1106+ , maxWorkers = maxWrk

1107+ , maxOutQSize = maxOq

1108+ , maxHeapSize = maxHs

1109+ , maxWorkQSize = maxWq

1110+#endif

1027 }

1111 }

1028 in return sv

1112 in return sv

1029

1113

@@ -1038,6 +1122,14 @@ getAheadSVar style = do

1038 wfw <- newIORef False

1122 wfw <- newIORef False

1039 running <- newIORef S.empty

1123 running <- newIORef S.empty

1040 q <- newIORef ([], -1)

1124 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+

1041 let checkEmpty = liftIO $ do

1133 let checkEmpty = liftIO $ do

1042 (xs, _) <- readIORef q

1134 (xs, _) <- readIORef q

1043 return $ null xs

1135 return $ null xs

@@ -1055,6 +1147,13 @@ getAheadSVar style = do

1055 , waitingForWork = wfw

1147 , waitingForWork = wfw

1056 , svarStyle = style

1148 , svarStyle = style

1057 , activeWorkers = active

1149 , activeWorkers = active

1150+

1151+#ifdef DIAGNOSTICS

1152+ , maxWorkers = maxWrk

1153+ , maxOutQSize = maxOq

1154+ , maxHeapSize = maxHs

1155+ , maxWorkQSize = maxWq

1156+#endif

1058 }

1157 }

1059 in return sv

1158 in return sv

1060

1159