5c975de · streamly · git.iko.soy

68

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

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

71− tryPutMVar, takeMVar)

71+ tryPutMVar, takeMVar, readMVar)

72import Control.Exception (SomeException (..))

73import qualified Control.Exception.Lifted as EL

74import Control.Monad (when)

148-- already enqueued computations get evaluated. See 'joinStreamVar2'.

149--

150data SVar m a =

151− SVar { outputQueue :: IORef [ChildEvent a]

151+ SVar { outputQueue :: IORef ([ChildEvent a], Int)

152− , doorBell :: MVar () -- wakeup mechanism for outQ

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

153+ , siren :: MVar () -- hooter for workers to begin work

153 , enqueue :: Stream m a -> IO ()

154 , enqueue :: Stream m a -> IO ()

154 , runqueue :: m ()

155 , runqueue :: m ()

155 , runningThreads :: IORef (Set ThreadId)

156 , runningThreads :: IORef (Set ThreadId)

304

305

305-- XXX exception safety of all atomic/MVar operations

306-- XXX exception safety of all atomic/MVar operations

306

307

308+-- TBD Each worker can have their own queue and the consumer can empty one

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

307{-# INLINE send #-}

310{-# INLINE send #-}

308send :: MonadIO m => SVar m a -> ChildEvent a -> m ()

311send :: MonadIO m => SVar m a -> ChildEvent a -> m ()

309send sv msg = liftIO $ do

312send sv msg = liftIO $ do

310− old <- atomicModifyIORefCAS (outputQueue sv) $ \es -> (msg : es, es)

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

311− if null old

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

315+ if (len <= 0) then do

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

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

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

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

318+ --

319+ -- Since multiple workers can try this at the same time, it is possible

320+ -- that we may put a spurious MVar after the consumer has already seen

321+ -- the output. But that's harmless, at worst it may cause the consumer

322+ -- to read the queue again and find it empty.

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

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

314− then void $ tryPutMVar (doorBell sv) ()

325+ void $ tryPutMVar (doorBell sv) ()

326+ -- The first worker who notices the output queue was emptied puts the

327+ -- siren off.

328+ void $ tryTakeMVar (siren sv)

329+ else if (len + 1 >= 1500) then do

330+ -- We are guaranteed to receive the siren if the consumer reads the

331+ -- queue because the consumer puts the siren on before reading the

332+ -- queue.

333+ --

334+ -- We may get the siren between the siren being swicthed on and the

335+ -- queue getting read but that's harmless, at amost everyone will go

336+ -- back to work and will have to sleep again if queue was still not

337+ -- emptied.

338+ --

339+ -- If even before a worker could read the MVar, the queue gets emptied

340+ -- and another worker queuing on it switches off the siren, then we may

341+ -- sleep here. In that case we are guaranteed to be woken up on the

342+ -- next siren. The next siren is guaranteed as we send a doorbell

343+ -- before switching off the siren, and the consumer switches on the

344+ -- siren after receiving the doorbell.

345+ readMVar (siren sv)

315− else return ()

346+ else return ()

316

347

317{-# INLINE sendStop #-}

348{-# INLINE sendStop #-}

318sendStop :: MonadIO m => SVar m a -> m ()

349sendStop :: MonadIO m => SVar m a -> m ()

451 -- sent then we block, so that we do not keep looping fruitlessly.

482 -- sent then we block, so that we do not keep looping fruitlessly.

452

483

453 liftIO $ threadDelay 200

484 liftIO $ threadDelay 200

454− output <- liftIO $ readIORef (outputQueue sv)

485+ (_, n) <- liftIO $ readIORef (outputQueue sv)

455− when (null output) $ do

486+ when (n <= 0) $ do

456 done <- queueEmpty sv

487 done <- queueEmpty sv

457 if not done

488 if not done

458 then pushWorker sv >> sendWorkerWait sv

489 then pushWorker sv >> sendWorkerWait sv

469 res <- liftIO $ tryTakeMVar (doorBell sv)

500 res <- liftIO $ tryTakeMVar (doorBell sv)

470 when (isNothing res) $ sendWorkerWait sv

501 when (isNothing res) $ sendWorkerWait sv

471

502

503+ void $ liftIO $ tryPutMVar (siren sv) ()

472− list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)

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

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

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

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

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

475 -- streams.

507 -- streams.

507

539

508getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)

540getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)

509getFifoSVar ctype = do

541getFifoSVar ctype = do

510− outQ <- newIORef []

542+ outQ <- newIORef ([], 0)

511 outQMv <- newEmptyMVar

543 outQMv <- newEmptyMVar

544+ hooter <- newEmptyMVar

512 active <- newIORef 0

545 active <- newIORef 0

513 running <- newIORef S.empty

546 running <- newIORef S.empty

514 q <- newQ

547 q <- newQ

515 let sv =

548 let sv =

516 SVar { outputQueue = outQ

549 SVar { outputQueue = outQ

517 , doorBell = outQMv

550 , doorBell = outQMv

551+ , siren = hooter

518 , runningThreads = running

552 , runningThreads = running

519 , runqueue = runqueueFIFO sv q

553 , runqueue = runqueueFIFO sv q

520 , enqueue = pushL q

554 , enqueue = pushL q

526

560

527getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)

561getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)

528getLifoSVar ctype = do

562getLifoSVar ctype = do

529− outQ <- newIORef []

563+ outQ <- newIORef ([], 0)

530 outQMv <- newEmptyMVar

564 outQMv <- newEmptyMVar

565+ hooter <- newEmptyMVar

531 active <- newIORef 0

566 active <- newIORef 0

532 running <- newIORef S.empty

567 running <- newIORef S.empty

533 q <- newIORef []

568 q <- newIORef []

535 let sv =

570 let sv =

536 SVar { outputQueue = outQ

571 SVar { outputQueue = outQ

537 , doorBell = outQMv

572 , doorBell = outQMv

573+ , siren = hooter

538 , runningThreads = running

574 , runningThreads = running

539 , runqueue = runqueueLIFO sv q

575 , runqueue = runqueueLIFO sv q

540 , enqueue = enqueueLIFO q

576 , enqueue = enqueueLIFO q

546

582

547getParSVar :: SVarStyle -> IO (SVar m a)

583getParSVar :: SVarStyle -> IO (SVar m a)

548getParSVar style = do

584getParSVar style = do

549− outQ <- newIORef []

585+ outQ <- newIORef ([], 0)

550 outQMv <- newEmptyMVar

586 outQMv <- newEmptyMVar

587+ hooter <- newEmptyMVar

551 active <- newIORef 0

588 active <- newIORef 0

552 running <- newIORef S.empty

589 running <- newIORef S.empty

553 let sv =

590 let sv =

554 SVar { outputQueue = outQ

591 SVar { outputQueue = outQ

555 , doorBell = outQMv

592 , doorBell = outQMv

593+ , siren = hooter

556 , runningThreads = running

594 , runningThreads = running

557 , runqueue = undefined

595 , runqueue = undefined

558 , enqueue = undefined

596 , enqueue = undefined