d73041c · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Pass WorkerInfo only when needed

d73041cHarendra Kumar8 years ago

src/Streamly/SVar.hs+49−33

compare with master

unifiedsplit

@@ -347,7 +347,7 @@ data SVar t m a = SVar

347 , enqueue :: t m a -> IO ()

348 , isWorkDone :: IO Bool

349 , needDoorBell :: IORef Bool

350− , workLoop :: WorkerInfo -> m ()

350+ , workLoop :: Maybe WorkerInfo -> m ()

351

352 -- Shared, thread tracking

353 , workerThreads :: IORef (Set ThreadId)

@@ -905,13 +905,16 @@ workerRateControl sv yinfo winfo = do

905-- streams. latency update must be done when we yield directly to outputQueue

906-- or when we yield to heap.

907{-# INLINE sendYield #-}

908−sendYield :: SVar t m a -> WorkerInfo -> ChildEvent a -> IO Bool

908+sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool

909−sendYield sv winfo msg = do

909+sendYield sv mwinfo msg = do

910 r <- send sv msg

911 rateLimitOk <-

912+ case mwinfo of

913+ Just winfo ->

912− case yieldRateInfo sv of

914+ case yieldRateInfo sv of

915+ Nothing -> return True

916+ Just yinfo -> workerRateControl sv yinfo winfo

913 Nothing -> return True

917 Nothing -> return True

914− Just yinfo -> workerRateControl sv yinfo winfo

915 return $ r && rateLimitOk

918 return $ r && rateLimitOk

916

919

917{-# INLINE workerStopUpdate #-}

920{-# INLINE workerStopUpdate #-}

@@ -921,12 +924,15 @@ workerStopUpdate winfo info = do

921 when (i /= 0) $ workerUpdateLatency info winfo

924 when (i /= 0) $ workerUpdateLatency info winfo

922

925

923{-# INLINABLE sendStop #-}

926{-# INLINABLE sendStop #-}

924−sendStop :: SVar t m a -> WorkerInfo -> IO ()

927+sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()

925−sendStop sv winfo = do

928+sendStop sv mwinfo = do

926 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1

929 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1

930+ case mwinfo of

931+ Just winfo ->

927− case yieldRateInfo sv of

932+ case yieldRateInfo sv of

933+ Nothing -> return ()

934+ Just info -> workerStopUpdate winfo info

928 Nothing -> return ()

935 Nothing -> return ()

929− Just info -> workerStopUpdate winfo info

930 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

936 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

931

937

932-------------------------------------------------------------------------------

938-------------------------------------------------------------------------------

@@ -1193,17 +1199,20 @@ pushWorker yieldMax sv = do

1193#ifdef DIAGNOSTICS

1199#ifdef DIAGNOSTICS

1194 recordMaxWorkers sv

1200 recordMaxWorkers sv

1195#endif

1201#endif

1196− -- XXX we can make this allocation conditional, it might matter when

1197− -- significant number of workers are being sent.

1202+ -- This allocation matters when significant number of workers are being

1203+ -- sent. We allocate it only when needed.

1198− winfo <- do

1204+ winfo <-

1205+ case yieldRateInfo sv of

1206+ Nothing -> return Nothing

1207+ Just _ -> liftIO $ do

1199− cntRef <- liftIO $ newIORef 0

1208+ cntRef <- newIORef 0

1200− t <- liftIO $ getTime Monotonic

1209+ t <- getTime Monotonic

1201− lat <- liftIO $ newIORef (0, t)

1210+ lat <- newIORef (0, t)

1202− return $ WorkerInfo

1211+ return $ Just $ WorkerInfo

1203− { workerYieldMax = yieldMax

1212+ { workerYieldMax = yieldMax

1204− , workerYieldCount = cntRef

1213+ , workerYieldCount = cntRef

1205− , workerLatencyStart = lat

1214+ , workerLatencyStart = lat

1206− }

1215+ }

1207 doFork (workLoop sv winfo) (handleChildException sv) >>= addThread sv

1216 doFork (workLoop sv winfo) (handleChildException sv) >>= addThread sv

1208

1217

1209-- XXX we can push the workerCount modification in accountThread and use the

1218-- XXX we can push the workerCount modification in accountThread and use the

@@ -1215,25 +1224,32 @@ pushWorker yieldMax sv = do

1215-- workerThreads. Alternatively, we can use a CreateThread event to avoid

1224-- workerThreads. Alternatively, we can use a CreateThread event to avoid

1216-- using a CAS based modification.

1225-- using a CAS based modification.

1217{-# NOINLINE pushWorkerPar #-}

1226{-# NOINLINE pushWorkerPar #-}

1218−pushWorkerPar :: MonadAsync m => SVar t m a -> (WorkerInfo -> m ()) -> m ()

1227+pushWorkerPar :: MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()

1219pushWorkerPar sv wloop = do

1228pushWorkerPar sv wloop = do

1220 -- We do not use workerCount in case of ParallelVar but still there is no

1229 -- We do not use workerCount in case of ParallelVar but still there is no

1221 -- harm in maintaining it correctly.

1230 -- harm in maintaining it correctly.

1222#ifdef DIAGNOSTICS

1231#ifdef DIAGNOSTICS

1223 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1

1232 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1

1224 recordMaxWorkers sv

1233 recordMaxWorkers sv

1225−#endif

1234+ -- This allocation matters when significant number of workers are being

1235+ -- sent. We allocate it only when needed. The overhead increases by 4x.

1226− winfo <- do

1236+ winfo <-

1237+ case yieldRateInfo sv of

1238+ Nothing -> return Nothing

1239+ Just _ -> liftIO $ do

1227− cntRef <- liftIO $ newIORef 0

1240+ cntRef <- newIORef 0

1228− t <- liftIO $ getTime Monotonic

1241+ t <- getTime Monotonic

1229− lat <- liftIO $ newIORef (0, t)

1242+ lat <- newIORef (0, t)

1230− return $ WorkerInfo

1243+ return $ Just $ WorkerInfo

1231− { workerYieldMax = 0

1244+ { workerYieldMax = 0

1232− , workerYieldCount = cntRef

1245+ , workerYieldCount = cntRef

1233− , workerLatencyStart = lat

1246+ , workerLatencyStart = lat

1234− }

1247+ }

1235

1248

1236 doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv

1249 doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv

1250+#else

1251+ doFork (wloop Nothing) (handleChildException sv) >>= modifyThread sv

1252+#endif

1237

1253

1238-- Returns:

1254-- Returns:

1239-- True: can dispatch more

1255-- True: can dispatch more

@@ -1860,7 +1876,7 @@ getAheadSVar :: MonadAsync m

1860 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1876 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1861 -> State t m a

1877 -> State t m a

1862 -> SVar t m a

1878 -> SVar t m a

1863− -> WorkerInfo

1879+ -> Maybe WorkerInfo

1864 -> m ())

1880 -> m ())

1865 -> IO (SVar t m a)

1881 -> IO (SVar t m a)

1866getAheadSVar st f = do

1882getAheadSVar st f = do

@@ -2050,7 +2066,7 @@ newAheadVar :: MonadAsync m

2050 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

2066 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

2051 -> State t m a

2067 -> State t m a

2052 -> SVar t m a

2068 -> SVar t m a

2053− -> WorkerInfo

2069+ -> Maybe WorkerInfo

2054 -> m ())

2070 -> m ())

2055 -> m (SVar t m a)

2071 -> m (SVar t m a)

2056newAheadVar st m wloop = do

2072newAheadVar st m wloop = do

src/Streamly/Streams/Ahead.hs+16−11

@@ -192,7 +192,7 @@ processHeap :: MonadIO m

192 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

193 -> State Stream m a

194 -> SVar Stream m a

195− -> WorkerInfo

195+ -> Maybe WorkerInfo

196 -> AheadHeapEntry Stream m a

197 -> Int

198 -> Bool -- we are draining the heap before we stop

@@ -306,7 +306,7 @@ drainHeap :: MonadIO m

306 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

307 -> State Stream m a

308 -> SVar Stream m a

309− -> WorkerInfo

309+ -> Maybe WorkerInfo

310 -> m ()

311drainHeap q heap st sv winfo = do

312 ent <- liftIO $ dequeueFromHeap heap

@@ -320,7 +320,7 @@ processWithoutToken :: MonadIO m

320 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

321 -> State Stream m a

322 -> SVar Stream m a

323− -> WorkerInfo

323+ -> Maybe WorkerInfo

324 -> Stream m a

325 -> Int

326 -> m ()

@@ -344,16 +344,21 @@ processWithoutToken q heap st sv winfo m sno = do

344 ((H.insert (Entry seqNo ent) h, snum), h)

345

346 heapOk <- liftIO $ underMaxHeap sv hp

347+ let keepDraining = drainHeap q heap st sv winfo

348+ mainLoop = workLoopAhead q heap st sv winfo

347 if heapOk

349 if heapOk

348 then

350 then

349 case yieldRateInfo sv of

351 case yieldRateInfo sv of

350− Nothing -> workLoopAhead q heap st sv winfo

352+ Nothing -> mainLoop

351 Just yinfo -> do

353 Just yinfo -> do

354+ case winfo of

355+ Just info -> do

352− rateOk <- liftIO $ workerRateControl sv yinfo winfo

356+ rateOk <- liftIO $ workerRateControl sv yinfo info

353− if rateOk

357+ if rateOk

354− then workLoopAhead q heap st sv winfo

358+ then mainLoop

355− else drainHeap q heap st sv winfo

359+ else keepDraining

356− else drainHeap q heap st sv winfo

360+ Nothing -> mainLoop

361+ else keepDraining

357

362

358 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

363 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

359 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

364 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

@@ -363,7 +368,7 @@ processWithToken :: MonadIO m

363 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

368 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

364 -> State Stream m a

369 -> State Stream m a

365 -> SVar Stream m a

370 -> SVar Stream m a

366− -> WorkerInfo

371+ -> Maybe WorkerInfo

367 -> Stream m a

372 -> Stream m a

368 -> Int

373 -> Int

369 -> m ()

374 -> m ()

@@ -461,7 +466,7 @@ workLoopAhead :: MonadIO m

461 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

466 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

462 -> State Stream m a

467 -> State Stream m a

463 -> SVar Stream m a

468 -> SVar Stream m a

464− -> WorkerInfo

469+ -> Maybe WorkerInfo

465 -> m ()

470 -> m ()

466workLoopAhead q heap st sv winfo = do

471workLoopAhead q heap st sv winfo = do

467#ifdef DIAGNOSTICS

472#ifdef DIAGNOSTICS

src/Streamly/Streams/Async.hs+4−4

@@ -79,7 +79,7 @@ workLoopLIFO

79 => IORef [Stream m a]

80 -> State Stream m a

81 -> SVar Stream m a

82− -> WorkerInfo

82+ -> Maybe WorkerInfo

83 -> m ()

84workLoopLIFO q st sv winfo = run

@@ -118,7 +118,7 @@ workLoopLIFOLimited

118 => IORef [Stream m a]

119 -> State Stream m a

120 -> SVar Stream m a

121− -> WorkerInfo

121+ -> Maybe WorkerInfo

122 -> m ()

123workLoopLIFOLimited q st sv winfo = run

124

@@ -178,7 +178,7 @@ workLoopFIFO

178 => LinkedQueue (Stream m a)

179 -> State Stream m a

180 -> SVar Stream m a

181− -> WorkerInfo

181+ -> Maybe WorkerInfo

182 -> m ()

183workLoopFIFO q st sv winfo = run

184

@@ -208,7 +208,7 @@ workLoopFIFOLimited

208 => LinkedQueue (Stream m a)

209 -> State Stream m a

210 -> SVar Stream m a

211− -> WorkerInfo

211+ -> Maybe WorkerInfo

212 -> m ()

213workLoopFIFOLimited q st sv winfo = run

214

src/Streamly/Streams/Parallel.hs+3−1

@@ -60,7 +60,9 @@ import qualified Streamly.Streams.StreamK as K

60-------------------------------------------------------------------------------

62{-# NOINLINE runOne #-}

63+runOne

64+ :: MonadIO m

63−runOne :: MonadIO m => State Stream m a -> Stream m a -> WorkerInfo -> m ()

65+ => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()

64runOne st m winfo = unStream m st stop single yieldk

66runOne st m winfo = unStream m st stop single yieldk

66 where

68 where