1d13085 · streamly · git.iko.soy

67

68import Control.Concurrent (ThreadId, myThreadId, threadDelay)

69import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

70− tryPutMVar, takeMVar, putMVar)

70+ tryPutMVar, takeMVar)

71import Control.Exception (SomeException (..), catch, mask)

72import Control.Monad (when)

73import Control.Monad.Catch (MonadThrow, throwM)

118 ChildYield a

119 | ChildStop ThreadId (Maybe SomeException)

120

121−-- | Sorting out of turn outputs in a heap for Ahead style streams

121+-- | Sorting out-of-turn outputs in a heap for Ahead style streams

122−data AheadHeapEntry a =

122+data AheadHeapEntry m a =

123− AheadYieldSingle a

123+ AheadEntryPure a

124− | AheadYieldMany a (MVar ()) -- value, MVar to wakeup for more values

124+ | AheadEntryStream (Stream m a)

125

126------------------------------------------------------------------------------

127-- State threaded around the monad for thread management

169 -- Shared output queue (events, length)

170 , outputQueue :: IORef ([ChildEvent a], Int)

171 , doorBell :: MVar () -- signal the consumer about output

172− , siren :: MVar () -- hooter for workers to begin work

173

172

174 -- Output synchronization mechanism for Ahead streams (Ahead and

173 -- Output synchronization mechanism for Ahead streams (Ahead and

175 -- wAhead). We maintain a heap of out of sequence ahead of time

174 -- wAhead). We maintain a heap of out of sequence ahead of time

177 -- currently at the head of the stream. Concurrent execute ahead

176 -- currently at the head of the stream. Concurrent execute ahead

178 -- tasks that have a sequence number greater than the task at the

177 -- tasks that have a sequence number greater than the task at the

179 -- head should add their output to the heap.

178 -- head should add their output to the heap.

180− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)

179+ , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry m a))

181− , heapSiren :: MVar () -- hooter for workers to continue

180+ , Int

182− -- adding to the heap

181+ )

183

182

184 -- Shared work queue

183 -- Shared work queue

185 , workQueue :: IORef ([Stream m a], Int)

184 , workQueue :: IORef ([Stream m a], Int)

192 , runningThreads :: IORef (Set ThreadId)

191 , runningThreads :: IORef (Set ThreadId)

193 , activeWorkers :: IORef Int

192 , activeWorkers :: IORef Int

194#ifdef DIAGNOSTICS

193#ifdef DIAGNOSTICS

194+ , totalDispatches :: IORef Int

195 , maxWorkers :: IORef Int

196 , maxOutQSize :: IORef Int

197 , maxHeapSize :: IORef Int

206 tid <- myThreadId

207 (oqList, oqLen) <- readIORef $ outputQueue sv

208 db <- tryTakeMVar $ doorBell sv

209− srn <- tryTakeMVar $ siren sv

210 aheadDump <-

209 aheadDump <-

211 if svarStyle sv == AheadVar

210 if svarStyle sv == AheadVar

212 then do

211 then do

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

212 (oheap, oheapSeq) <- readIORef $ outputHeap sv

214− hsrn <- tryTakeMVar $ heapSiren sv

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

213 (wq, wqSeq) <- readIORef $ workQueue sv

216 maxHp <- readIORef $ maxHeapSize sv

214 maxHp <- readIORef $ maxHeapSize sv

217 return $ unlines

215 return $ unlines

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

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

219 , "heap seqeunce = " ++ show oheapSeq

217 , "heap seqeunce = " ++ show oheapSeq

220− , "heap siren = " ++ show hsrn

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

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

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

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

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

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

228 rthread <- readIORef $ runningThreads sv

225 rthread <- readIORef $ runningThreads sv

229 workers <- readIORef $ activeWorkers sv

226 workers <- readIORef $ activeWorkers sv

230 maxWrk <- readIORef $ maxWorkers sv

227 maxWrk <- readIORef $ maxWorkers sv

228+ dispatches <- readIORef $ totalDispatches sv

231 maxOq <- readIORef $ maxOutQSize sv

229 maxOq <- readIORef $ maxOutQSize sv

232 -- XXX queueEmpty should be made IO return type

230 -- XXX queueEmpty should be made IO return type

233

231

237 , "outputQueue length computed = " ++ show (length oqList)

235 , "outputQueue length computed = " ++ show (length oqList)

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

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

239 , "output doorBell = " ++ show db

237 , "output doorBell = " ++ show db

240− , "worker siren = " ++ show srn

238+ , "total dispatches = " ++ show dispatches

241 , "max workers = " ++ show maxWrk

239 , "max workers = " ++ show maxWrk

242 , "max outQSize = " ++ show maxOq

240 , "max outQSize = " ++ show maxOq

243 ]

241 ]

419-- queue at a time, that way contention can be reduced.

417-- queue at a time, that way contention can be reduced.

420

418

421-- | This function is used by the producer threads to queue output for the

419-- | This function is used by the producer threads to queue output for the

422−-- consumer thread to consume.

420+-- consumer thread to consume. Returns whether the queue has more space.

423{-# NOINLINE send #-}

421{-# NOINLINE send #-}

424−send :: SVar m a -> ChildEvent a -> IO ()

422+send :: SVar m a -> ChildEvent a -> IO Bool

425send sv msg = do

423send sv msg = do

426 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

424 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

427 ((msg : es, n + 1), n)

425 ((msg : es, n + 1), n)

428− if (len <= 0) then do

426+ when (len <= 0) $ do

429 -- XXX need a memory barrier? The wake up must happen only after the

427 -- XXX need a memory barrier? The wake up must happen only after the

430 -- store has finished otherwise we can have lost wakeup problems.

428 -- store has finished otherwise we can have lost wakeup problems.

431 --

429 --

436 -- The important point is that the consumer is guaranteed to receive a

434 -- The important point is that the consumer is guaranteed to receive a

437 -- doorbell if something was added to the queue after it empties it.

435 -- doorbell if something was added to the queue after it empties it.

438 void $ tryPutMVar (doorBell sv) ()

436 void $ tryPutMVar (doorBell sv) ()

439− -- those who were waiting because of buffer limit should now go ahead

437+ return (len < 1500)

440− void $ tryPutMVar (siren sv) ()

441− -- The only thread that reduces the length of the queue is the consumer

442− -- thread. It is possible that the queue may have been emptied in the

443− -- window when we acquired "len" and now when we are using it. In that case

444− -- we will be woken up by the consumer blowing the siren.

445− -- Note: current length after adding our output is len + 1

446− else when (len + 1 >= 1500) $ do

447− withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)

448− (_, n) <- readIORef (outputQueue sv)

449− when (n <= 1500) $ void $ tryPutMVar (siren sv) ()

450

438

451{-# NOINLINE sendStop #-}

439{-# NOINLINE sendStop #-}

452sendStop :: SVar m a -> IO ()

440sendStop :: SVar m a -> IO ()

453−sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)

441+sendStop sv = myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

454

442

455-------------------------------------------------------------------------------

443-------------------------------------------------------------------------------

456-- Async

444-- Async

489 Nothing -> liftIO $ sendStop sv

477 Nothing -> liftIO $ sendStop sv

490 Just m -> (runStream m) (Just sv) run single yield

478 Just m -> (runStream m) (Just sv) run single yield

491

479

480+ single a = do

492− sendit a = liftIO $ send sv (ChildYield a)

481+ res <- liftIO $ send sv (ChildYield a)

493− single a = sendit a >> run

482+ if res then run else liftIO $ sendStop sv

483+ yield a r = do

484+ res <- liftIO $ send sv (ChildYield a)

485+ if res

494− yield a r = sendit a >> (runStream r) (Just sv) run single yield

486+ then (runStream r) (Just sv) run single yield

487+ else liftIO $ enqueueLIFO sv q r >> sendStop sv

495

488

496 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

489 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

497 [] -> ([], Nothing)

490 [] -> ([], Nothing)

528 Just m -> (runStream m) (Just sv) run single yield

521 Just m -> (runStream m) (Just sv) run single yield

529

522

530 dequeue = liftIO $ tryPopR q

523 dequeue = liftIO $ tryPopR q

531− sendit a = liftIO $ send sv (ChildYield a)

532− single a = sendit a >> run

524+ single a = do

525+ res <- liftIO $ send sv (ChildYield a)

526+ if res then run else liftIO $ sendStop sv

527+ yield a r = do

528+ res <- liftIO $ send sv (ChildYield a)

533− yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run

529+ liftIO (enqueueFIFO sv q r)

530+ if res then run else liftIO $ sendStop sv

534

531

535-------------------------------------------------------------------------------

532-------------------------------------------------------------------------------

536-- Parallel

533-- Parallel

545 stop = liftIO $ sendStop sv

542 stop = liftIO $ sendStop sv

546 sendit a = liftIO $ send sv (ChildYield a)

543 sendit a = liftIO $ send sv (ChildYield a)

547 single a = sendit a >> stop

544 single a = sendit a >> stop

545+ -- XXX there is no flow control in parallel case. We should perhaps use a

546+ -- queue and queue it back on that and exit the thread when the outputQueue

547+ -- overflows. Parallel is dangerous because it can accumulate unbounded

548+ -- output in the buffer.

548− yield a r = sendit a >> runOne sv r

549+ yield a r = void (sendit a) >> runOne sv r

549

550

550-------------------------------------------------------------------------------

551-------------------------------------------------------------------------------

551-- Ahead

552-- Ahead

552-------------------------------------------------------------------------------

553-------------------------------------------------------------------------------

553

554

554−-- Lookahead streams can execute multiple tasks concurrently, ahead of time but

555+-- Lookahead streams can execute multiple tasks concurrently, ahead of time,

555−-- always serves them in the same order as they appear in the stream. To

556+-- but always serve them in the same order as they appear in the stream. To

556-- implement lookahead streams efficiently we assign a sequence number to each

557-- implement lookahead streams efficiently we assign a sequence number to each

557−-- task when the task is picked up for execution. When the task finishes the

558+-- task when the task is picked up for execution. When the task finishes, the

558-- output is tagged with the same sequence number and we rearrange the outputs

559-- output is tagged with the same sequence number and we rearrange the outputs

559-- in sequence based on that number.

560-- in sequence based on that number.

560--

561--

561-- To explain the mechanism imagine that the current task at the head of the

562-- To explain the mechanism imagine that the current task at the head of the

562−-- stream has a "token" to yield to the output. The ownership of the token is

563+-- stream has a "token" to yield to the outputQueue. The ownership of the token

563−-- determined by the current sequence number maintained in outputHeap. Sequence

564+-- is determined by the current sequence number is maintained in outputHeap.

564−-- number is assigned when a task is queued. When a thread dequeues a task it

565+-- Sequence number is assigned when a task is queued. When a thread dequeues a

565−-- picks up the sequence number as well and when the output is ready it uses

566+-- task it picks up the sequence number as well and when the output is ready it

566−-- the sequence number to queue the output to the outputQueue.

567+-- uses the sequence number to queue the output to the outputQueue.

567--

568--

568-- The thread with current sequence number sends the output directly to the

569-- The thread with current sequence number sends the output directly to the

569-- outputQueue. Other threads push the output to the outputHeap. When the task

570-- outputQueue. Other threads push the output to the outputHeap. When the task

570−-- being queued on the heap is a stream of many elements we yield the first

571+-- being queued on the heap is a stream of many elements we evaluate only the

571−-- element on the heap and an MVar to wakeup for more elements. When such a

572+-- first element and keep the rest of the unevaluated computation in the heap.

572−-- task gets the "token" for outputQueue it can directly keep yielding all the

573+-- When such a task gets the "token" for outputQueue it evaluates and directly

573−-- elements to the outputQueue without checking for the "token".

574+-- yields all the elements to the outputQueue without checking for the

575+-- "token".

574--

576--

575-- Note that no two outputs in the heap can have the same sequence numbers and

577-- Note that no two outputs in the heap can have the same sequence numbers and

576-- therefore we do not need a stable heap. We have also separated the buffer

578-- therefore we do not need a stable heap. We have also separated the buffer

587-- reason we have a single element buffer limitation for the streams being

589-- reason we have a single element buffer limitation for the streams being

588-- executed in advance.

590-- executed in advance.

589--

591--

590−-- This scheme works pretty efficiently with only 50% more overhead compared to

592+-- This scheme works pretty efficiently with less than 40% extra overhead

591−-- the Async streams where we do not have any kind of sequencing of the

593+-- compared to the Async streams where we do not have any kind of sequencing of

592−-- outputs. It is especially devised so that we are most efficient when we have

594+-- the outputs. It is especially devised so that we are most efficient when we

593−-- short tasks and need just a single thread. Also when a thread yields many

595+-- have short tasks and need just a single thread. Also when a thread yields

594−-- items it can hold lockfree access to the outputQueue and do it efficiently.

596+-- many items it can hold lockfree access to the outputQueue and do it

597+-- efficiently.

595--

598--

596-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so

599-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so

597-- that they do not hog the resources and hinder the progress of the threads in

600-- that they do not hog the resources and hinder the progress of the threads in

598-- front of them.

601-- front of them.

599

602

600−{-# INLINE sendToHeap #-}

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

602−sendToHeap sv msg seqNo = do

603− let maxHeap = 1500

604−

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

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

607− -- Note: len is valid only when added is True

608− (len, added) <- atomicModifyIORefCAS (outputHeap sv) $

609− \ref@(h, snum) ->

610− if seqNo > snum

611− then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))

612− else (ref, (0, False))

613−

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

615− -- missed due to race.

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

617−

618− if added

619− then do

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

621− withDBGMVar sv "sendToHeap: wait for heapSiren"

622− $ takeMVar (heapSiren sv)

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

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

625− return False

626− else do

627− case msg of

628− AheadYieldSingle a -> send sv (ChildYield a)

629− AheadYieldMany a mv -> do

630− send sv (ChildYield a)

631− withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()

632− return True

633−

634−popFromHeap

635− :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)

636− -> Int

637− -> IO (Maybe (AheadHeapEntry a))

638−popFromHeap hp nextSeqNo =

639− atomicModifyIORefCAS hp $ \(h, snum) -> do

640− let r = H.uncons h

641− let n = max nextSeqNo snum

642− case r of

643− Nothing -> ((h, n), Nothing)

644− Just (Entry seqNo ev, hp') ->

645− if (seqNo <= n)

646− then ((hp', seqNo), Just ev)

647− else ((h, n), Nothing)

648−

649−-- | As soon as we set the nextSeqNo in outputHeap we have transferred the

650−-- output "token" to the thread executing the task with that sequence number.

651−-- We have to make sure that there is no entry in the heap that has nextSeqNo

652−-- before we set nextSeqNo as the expected sequence number in the heap. It is

653−-- possible that when we are clearing the heap entry of the current sequence

654−-- number the next sequence number gets added and then we have to clear that,

655−-- this can keep going on and on.

656−drainHeap :: SVar m a -> Int -> IO ()

657−drainHeap sv nextSeqNo = do

658− r <- popFromHeap (outputHeap sv) nextSeqNo

659− case r of

660− Nothing -> return ()

661− Just ev -> do

662− case ev of

663− AheadYieldSingle a -> do

664− send sv (ChildYield a)

665− atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->

666− (h, snum + 1)

667− drainHeap sv nextSeqNo

668− AheadYieldMany a mv -> do

669− send sv (ChildYield a)

670− withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()

671−

672−-- | The thread that has the token has the responsibility to move any pending

673−-- entries in the heap to the outputQueue. When the token is assigned to a

674−-- sequence number we must not have any entries for that sequence number in the

675−-- heap.

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

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

683− drainHeap sv nextSeqNo

684− void $ tryPutMVar (heapSiren sv) ()

685−

686−-- | When we have the output "token" we are allowed to append to the

687−-- outputQueue otherwise we have to insert the output in the outputHeap.

688−-- Returns whether we still have the token.

689−handleOutput :: SVar m a

690− -> Bool

691− -> Int

692− -> Int

693− -> a

694− -> AheadHeapEntry a

695− -> IO Bool

696−handleOutput sv haveToken prevSeqNo seqNo a ev = do

697− if haveToken

698− then do

699− let nextSeqNo = prevSeqNo + 1

700− if seqNo == nextSeqNo

701− then send sv (ChildYield a) >> return True

702− else do

703− releaseToken sv nextSeqNo

704− toHeap

705− else toHeap

706− where toHeap = sendToHeap sv ev seqNo

707−

708-- Left associated ahead expressions are expensive. We start a new SVar for

603-- Left associated ahead expressions are expensive. We start a new SVar for

709-- each left associative expression. The queue is used only for right

604-- each left associative expression. The queue is used only for right

710-- associated expression, we queue the right expression and execute the left.

605-- associated expression, we queue the right expression and execute the left.

720 void $ tryPutMVar (doorBell sv) ()

615 void $ tryPutMVar (doorBell sv) ()

721 atomicModifyIORefCAS_ (waitingForWork sv) (const False)

616 atomicModifyIORefCAS_ (waitingForWork sv) (const False)

722

617

618+-- Normally the thread that has the token should never go away. The token gets

619+-- handed over to another thread, but someone or the other has the token at any

620+-- point of time. But if the task that has the token finds that the outputQueue

621+-- is full, in that case it can go away without even handing over the token to

622+-- another thread. In that case it sets the nextSequence number in the heap its

623+-- own sequence number before going away. To handle this case, any task that

624+-- does not have the token tries to dequeue from the heap first before

625+-- dequeuing from the work queue. If it finds that the task at the top of the

626+-- heap is the one that owns the current sequence number then it grabs the

627+-- token and starts with that.

723runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()

628runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()

724−runqueueAhead sv q = run False (-1)

629+runqueueAhead sv q = runHeap

725

630

726 where

631 where

727

632

633+ maxHeap = 1500

634+

635+ toHeap seqNo ent = do

636+ hp <- liftIO $ atomicModifyIORefCAS (outputHeap sv) $ \(h, snum) ->

637+ ((H.insert (Entry seqNo ent) h, snum), h)

638+ if H.size hp <= maxHeap

639+ then runHeap

640+ else liftIO $ sendStop sv

641+

642+ singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

643+ yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `cons` r))

644+

645+ singleOutput seqNo a = do

646+ continue <- liftIO $ send sv (ChildYield a)

647+ if continue

648+ then runQueueToken seqNo

649+ else liftIO $ do

650+ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) -> (h, seqNo + 1)

651+ sendStop sv

652+

653+ yieldOutput seqNo a r = do

654+ continue <- liftIO $ send sv (ChildYield a)

655+ if continue

656+ then (runStream r) (Just sv) (runQueueToken seqNo)

657+ (singleOutput seqNo)

658+ (yieldOutput seqNo)

659+ else liftIO $ do

660+ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->

661+ (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)

662+ sendStop sv

663+

664+ {-# INLINE runQueueToken #-}

728− run haveToken prevSeqNo = do

665+ runQueueToken prevSeqNo = do

729 work <- dequeue

666 work <- dequeue

730 case work of

667 case work of

731− Nothing -> liftIO $ do

668+ Nothing -> do

669+ liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->

732− when haveToken $ releaseToken sv (prevSeqNo + 1)

670+ (h, prevSeqNo + 1)

733− sendStop sv

671+ runHeap

734 Just (m, seqNo) -> do

672 Just (m, seqNo) -> do

735− let tStatus = haveToken || seqNo == 0

673+ if seqNo == prevSeqNo + 1

736− let stop = do

674+ then

737− let nextSeqNo = prevSeqNo + 1

675+ (runStream m) (Just sv) (runQueueToken seqNo)

738− if tStatus && seqNo /= nextSeqNo

676+ (singleOutput seqNo)

677+ (yieldOutput seqNo)

678+ else do

679+ liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->

680+ (h, prevSeqNo + 1)

681+ (runStream m) (Just sv) runHeap

682+ (singleToHeap seqNo)

683+ (yieldToHeap seqNo)

684+ runQueueNoToken = do

685+ work <- dequeue

686+ case work of

687+ Nothing -> runHeap

688+ Just (m, seqNo) -> do

689+ if seqNo == 0

739− then do

690+ then

740− liftIO $ releaseToken sv nextSeqNo

691+ (runStream m) (Just sv) (runQueueToken seqNo)

741− run False seqNo

692+ (singleOutput seqNo)

742− else run tStatus seqNo

693+ (yieldOutput seqNo)

743− let single a = do

694+ else

744− tokenStatus <- liftIO $

695+ (runStream m) (Just sv) runHeap

745− handleOutput sv tStatus prevSeqNo seqNo a

696+ (singleToHeap seqNo)

746− (AheadYieldSingle a)

697+ (yieldToHeap seqNo)

747− run tokenStatus seqNo

698+

748−

699+ {-# NOINLINE runHeap #-}

749− -- If we are yielding more than one item and we do not have the

700+ runHeap = do

750− -- token then we insert the current item in the pending

701+#ifdef DIAGNOSTICS

751− -- outputHeap and wait for our turn.

702+ liftIO $ do

752− let

703+ maxHp <- readIORef (maxHeapSize sv)

753− stopNext = run True seqNo

704+ (hp, _) <- readIORef (outputHeap sv)

754− singleNext a = do

705+ when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)

755− liftIO $ send sv (ChildYield a)

706+#endif

756− run True seqNo

707+ ent <- liftIO $ dequeueFromHeap (outputHeap sv)

757− yieldNext a r = do

708+ case ent of

758− liftIO $ send sv (ChildYield a)

709+ Nothing -> do

759− (runStream r) (Just sv) stopNext singleNext yieldNext

710+ done <- queueEmpty sv

760−

711+ if done

761− yield a r = do

762− -- XXX we can allocate the MVar when actually inserting

763− -- in the heap and then return it.

764− liftIO $ do

712+ then liftIO $ sendStop sv

765− mv <- newEmptyMVar

713+ else runQueueNoToken

766− gotToken <-

714+ Just (Entry seqNo hent) -> do

767− handleOutput sv tStatus prevSeqNo seqNo a

715+ case hent of

768− (AheadYieldMany a mv)

716+ AheadEntryPure a -> singleOutput seqNo a

769− when (not gotToken) $ void $

717+ AheadEntryStream r ->

770− withDBGMVar sv "runqueueAhead: stream waiting"

771− $ takeMVar mv

772− (runStream r) (Just sv) stopNext singleNext yieldNext

718+ (runStream r) (Just sv) (runQueueToken seqNo)

773−

719+ (singleOutput seqNo)

774− (runStream m) (Just sv) stop single yield

720+ (yieldOutput seqNo)

775

721

776 dequeue = liftIO $ do

722 dequeue = liftIO $ do

777 atomicModifyIORefCAS q $ \case

723 atomicModifyIORefCAS q $ \case

779 (x : [], n) -> (([], n), Just (x, n))

725 (x : [], n) -> (([], n), Just (x, n))

780 _ -> error "more than one item on queue"

726 _ -> error "more than one item on queue"

781

727

728+ dequeueFromHeap

729+ :: IORef (Heap (Entry Int (AheadHeapEntry m a)), Int)

730+ -> IO (Maybe (Entry Int (AheadHeapEntry m a)))

731+ dequeueFromHeap hpRef = do

732+ atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do

733+ let r = H.uncons h

734+ case r of

735+ Nothing -> (hp, Nothing)

736+ Just (ent@(Entry seqNo _ev), hp') ->

737+ if (seqNo == snum)

738+ then ((hp', seqNo), Just ent)

739+ else (hp, Nothing)

740+

782-- Thread tracking is needed for two reasons:

741-- Thread tracking is needed for two reasons:

783--

742--

784-- 1) Killing threads on exceptions. Threads may not be left to go away by

743-- 1) Killing threads on exceptions. Threads may not be left to go away by

825handleChildException :: SVar m a -> SomeException -> IO ()

784handleChildException :: SVar m a -> SomeException -> IO ()

826handleChildException sv e = do

785handleChildException sv e = do

827 tid <- myThreadId

786 tid <- myThreadId

828− send sv (ChildStop tid (Just e))

787+ void $ send sv (ChildStop tid (Just e))

829

788

830#ifdef DIAGNOSTICS

789#ifdef DIAGNOSTICS

831recordMaxWorkers :: MonadIO m => SVar m a -> m ()

790recordMaxWorkers :: MonadIO m => SVar m a -> m ()

833 active <- readIORef (activeWorkers sv)

792 active <- readIORef (activeWorkers sv)

834 maxWrk <- readIORef (maxWorkers sv)

793 maxWrk <- readIORef (maxWorkers sv)

835 when (active > maxWrk) $ writeIORef (maxWorkers sv) active

794 when (active > maxWrk) $ writeIORef (maxWorkers sv) active

795+ modifyIORef (totalDispatches sv) (+1)

836#endif

796#endif

837

797

838{-# NOINLINE pushWorker #-}

798{-# NOINLINE pushWorker #-}

905 else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"

865 else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"

906 $ takeMVar (doorBell sv)

866 $ takeMVar (doorBell sv)

907 else do

867 else do

868+ -- XXX Assert here that if the heap is not empty then there is at

869+ -- least one outstanding worker. Otherwise we could be sleeping

870+ -- forever.

871+ --

908 -- We found that the queue was empty, but it may be empty

872 -- We found that the queue was empty, but it may be empty

909 -- temporarily, when we checked it. If that's the case we might

873 -- temporarily, when we checked it. If that's the case we might

910 -- sleep indefinitely unless the active workers produce some

874 -- sleep indefinitely unless the active workers produce some

954#else

918#else

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

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

956#endif

920#endif

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

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

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

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

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

960 -- streams.

923 -- streams.

969 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread

932 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread

970 throwM e

933 throwM e

971

934

972− {-# INLINE processEvents #-}

935+ allDone stp = do

973− processEvents [] = Stream $ \_ stp sng yld -> do

974− done <- allThreadsDone sv

975− if not done

976− then (runStream (fromStreamVar sv)) Nothing stp sng yld

977#ifdef DIAGNOSTICS

936#ifdef DIAGNOSTICS

978− else do

937+#ifdef DIAGNOSTICS_VERBOSE

979 svInfo <- liftIO $ dumpSVar sv

938 svInfo <- liftIO $ dumpSVar sv

980 liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo

939 liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo

981− stp

982−#else

983− else stp

984#endif

940#endif

941+#endif

942+ stp

943+

944+ {-# INLINE processEvents #-}

945+ processEvents [] = Stream $ \_ stp sng yld -> do

946+ workersDone <- allThreadsDone sv

947+ if not workersDone

948+ then (runStream (fromStreamVar sv)) Nothing stp sng yld

949+ else

950+ if svarStyle sv == ParallelVar

951+ then allDone stp

952+ else do

953+ -- All the workers may come back if the outputQueue is full.

954+ -- So check if there is work pending and kick off a worker to

955+ -- finish it.

956+ -- XXX we should perhaps kickoff even before we start

957+ -- processing if the active workers are zero, rather than at

958+ -- the end of processing the outputQueue. But some thread stop

959+ -- events may be in the outputQueue itself, so we may not be

960+ -- able to determine the correct number of workers before we

961+ -- process it.

962+ -- XXX It is necessary to do this before the processing to keep

963+ -- minimal concurrency always.

964+ -- We should have a separate queue for Stop events so that we

965+ -- can process them before the outputs and send workers

966+ -- promptly when needed.

967+ heapDone <-

968+ if (svarStyle sv == AheadVar)

969+ then do

970+ (hp, _) <- liftIO $ readIORef (outputHeap sv)

971+ return (H.size hp <= 0)

972+ else return True

973+ queueDone <- queueEmpty sv

974+ if queueDone && heapDone

975+ then allDone stp

976+ else do

977+ pushWorker sv

978+ (runStream (fromStreamVar sv)) Nothing stp sng yld

985

979

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

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

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

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

1001getFifoSVar ctype = do

995getFifoSVar ctype = do

1002 outQ <- newIORef ([], 0)

996 outQ <- newIORef ([], 0)

1003 outQMv <- newEmptyMVar

997 outQMv <- newEmptyMVar

1004− hooter <- newEmptyMVar

1005 active <- newIORef 0

998 active <- newIORef 0

1006 wfw <- newIORef False

999 wfw <- newIORef False

1007 running <- newIORef S.empty

1000 running <- newIORef S.empty

1008 q <- newQ

1001 q <- newQ

1009#ifdef DIAGNOSTICS

1002#ifdef DIAGNOSTICS

1003+ disp <- newIORef 0

1010 maxWrk <- newIORef 0

1004 maxWrk <- newIORef 0

1011 maxOq <- newIORef 0

1005 maxOq <- newIORef 0

1012 maxHs <- newIORef 0

1006 maxHs <- newIORef 0

1015 let sv =

1009 let sv =

1016 SVar { outputQueue = outQ

1010 SVar { outputQueue = outQ

1017 , doorBell = outQMv

1011 , doorBell = outQMv

1018− , siren = hooter

1019 , outputHeap = undefined

1012 , outputHeap = undefined

1020− , heapSiren = undefined

1021 , runningThreads = running

1013 , runningThreads = running

1022 , workQueue = undefined

1014 , workQueue = undefined

1023 , runqueue = runqueueFIFO sv q

1015 , runqueue = runqueueFIFO sv q

1027 , svarStyle = ctype

1019 , svarStyle = ctype

1028 , activeWorkers = active

1020 , activeWorkers = active

1029#ifdef DIAGNOSTICS

1021#ifdef DIAGNOSTICS

1022+ , totalDispatches = disp

1030 , maxWorkers = maxWrk

1023 , maxWorkers = maxWrk

1031 , maxOutQSize = maxOq

1024 , maxOutQSize = maxOq

1032 , maxHeapSize = maxHs

1025 , maxHeapSize = maxHs

1039getLifoSVar ctype = do

1032getLifoSVar ctype = do

1040 outQ <- newIORef ([], 0)

1033 outQ <- newIORef ([], 0)

1041 outQMv <- newEmptyMVar

1034 outQMv <- newEmptyMVar

1042− hooter <- newEmptyMVar

1043 active <- newIORef 0

1035 active <- newIORef 0

1044 wfw <- newIORef False

1036 wfw <- newIORef False

1045 running <- newIORef S.empty

1037 running <- newIORef S.empty

1046 q <- newIORef []

1038 q <- newIORef []

1047#ifdef DIAGNOSTICS

1039#ifdef DIAGNOSTICS

1040+ disp <- newIORef 0

1048 maxWrk <- newIORef 0

1041 maxWrk <- newIORef 0

1049 maxOq <- newIORef 0

1042 maxOq <- newIORef 0

1050 maxHs <- newIORef 0

1043 maxHs <- newIORef 0

1054 let sv =

1047 let sv =

1055 SVar { outputQueue = outQ

1048 SVar { outputQueue = outQ

1056 , doorBell = outQMv

1049 , doorBell = outQMv

1057− , siren = hooter

1058 , outputHeap = undefined

1050 , outputHeap = undefined

1059− , heapSiren = undefined

1060 , runningThreads = running

1051 , runningThreads = running

1061 , workQueue = undefined

1052 , workQueue = undefined

1062 , runqueue = runqueueLIFO sv q

1053 , runqueue = runqueueLIFO sv q

1067 , activeWorkers = active

1058 , activeWorkers = active

1068#ifdef DIAGNOSTICS

1059#ifdef DIAGNOSTICS

1069 , maxWorkers = maxWrk

1060 , maxWorkers = maxWrk

1061+ , totalDispatches = disp

1070 , maxOutQSize = maxOq

1062 , maxOutQSize = maxOq

1071 , maxHeapSize = maxHs

1063 , maxHeapSize = maxHs

1072 , maxWorkQSize = maxWq

1064 , maxWorkQSize = maxWq

1078getParSVar style = do

1070getParSVar style = do

1079 outQ <- newIORef ([], 0)

1071 outQ <- newIORef ([], 0)

1080 outQMv <- newEmptyMVar

1072 outQMv <- newEmptyMVar

1081− hooter <- newEmptyMVar

1082 active <- newIORef 0

1073 active <- newIORef 0

1083 wfw <- newIORef False

1074 wfw <- newIORef False

1084 running <- newIORef S.empty

1075 running <- newIORef S.empty

1085#ifdef DIAGNOSTICS

1076#ifdef DIAGNOSTICS

1077+ disp <- newIORef 0

1086 maxWrk <- newIORef 0

1078 maxWrk <- newIORef 0

1087 maxOq <- newIORef 0

1079 maxOq <- newIORef 0

1088 maxHs <- newIORef 0

1080 maxHs <- newIORef 0

1091 let sv =

1083 let sv =

1092 SVar { outputQueue = outQ

1084 SVar { outputQueue = outQ

1093 , doorBell = outQMv

1085 , doorBell = outQMv

1094− , siren = hooter

1095 , outputHeap = undefined

1086 , outputHeap = undefined

1096− , heapSiren = undefined

1097 , runningThreads = running

1087 , runningThreads = running

1098 , workQueue = undefined

1088 , workQueue = undefined

1099 , runqueue = undefined

1089 , runqueue = undefined

1103 , svarStyle = style

1093 , svarStyle = style

1104 , activeWorkers = active

1094 , activeWorkers = active

1105#ifdef DIAGNOSTICS

1095#ifdef DIAGNOSTICS

1096+ , totalDispatches = disp

1106 , maxWorkers = maxWrk

1097 , maxWorkers = maxWrk

1107 , maxOutQSize = maxOq

1098 , maxOutQSize = maxOq

1108 , maxHeapSize = maxHs

1099 , maxHeapSize = maxHs

1116 outQ <- newIORef ([], 0)

1107 outQ <- newIORef ([], 0)

1117 outH <- newIORef (H.empty, 0)

1108 outH <- newIORef (H.empty, 0)

1118 outQMv <- newEmptyMVar

1109 outQMv <- newEmptyMVar

1119− hooter <- newEmptyMVar

1120− hsiren <- newEmptyMVar

1121 active <- newIORef 0

1110 active <- newIORef 0

1122 wfw <- newIORef False

1111 wfw <- newIORef False

1123 running <- newIORef S.empty

1112 running <- newIORef S.empty

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

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

1125

1114

1126#ifdef DIAGNOSTICS

1115#ifdef DIAGNOSTICS

1116+ disp <- newIORef 0

1127 maxWrk <- newIORef 0

1117 maxWrk <- newIORef 0

1128 maxOq <- newIORef 0

1118 maxOq <- newIORef 0

1129 maxHs <- newIORef 0

1119 maxHs <- newIORef 0

1136 let sv =

1126 let sv =

1137 SVar { outputQueue = outQ

1127 SVar { outputQueue = outQ

1138 , doorBell = outQMv

1128 , doorBell = outQMv

1139− , siren = hooter

1140 , outputHeap = outH

1129 , outputHeap = outH

1141− , heapSiren = hsiren

1142 , runningThreads = running

1130 , runningThreads = running

1143 , workQueue = q

1131 , workQueue = q

1144 , runqueue = runqueueAhead sv q

1132 , runqueue = runqueueAhead sv q

1149 , activeWorkers = active

1137 , activeWorkers = active

1150

1138

1151#ifdef DIAGNOSTICS

1139#ifdef DIAGNOSTICS

1140+ , totalDispatches = disp

1152 , maxWorkers = maxWrk

1141 , maxWorkers = maxWrk

1153 , maxOutQSize = maxOq

1142 , maxOutQSize = maxOq

1154 , maxHeapSize = maxHs

1143 , maxHeapSize = maxHs