1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
{-# LANGUAGE RecordWildCards #-}
module Network.HTTP2.H2.Sync (
LoopCheck (..),
newLoopCheck,
syncWithSender,
syncWithSender',
makeOutput,
makeOutputIO,
enqueueOutputSIO,
) where
import Control.Concurrent
import Control.Concurrent.STM
import Control.Monad
import Network.Control
import Network.HTTP.Semantics.IO
import Network.HTTP2.H2.Context
import Network.HTTP2.H2.Queue
import Network.HTTP2.H2.Types
syncWithSender
:: Context
-> Stream
-> OutputType
-> LoopCheck
-> IO ()
syncWithSender ctx@Context{..} strm otyp lc = do
(pop, out) <- makeOutput strm otyp
enqueueOutput outputQ out
syncWithSender' ctx pop lc
makeOutput :: Stream -> OutputType -> IO (IO Sync, Output)
makeOutput strm otyp = do
var <- newEmptyMVar
let push mout = case mout of
Nothing -> putMVar var Done
Just ot -> putMVar var $ Cont ot
pop = takeMVar var
out =
Output
{ outputStream = strm
, outputType = otyp
, outputSync = push
}
return (pop, out)
makeOutputIO :: Context -> Stream -> OutputType -> Output
makeOutputIO Context{..} strm otyp = out
where
push mout = case mout of
Nothing -> return ()
-- Sender enqueues output again ignoring
-- the stream TX window.
Just ot -> enqueueOutput outputQ ot
out =
Output
{ outputStream = strm
, outputType = otyp
, outputSync = push
}
enqueueOutputSIO :: Context -> Stream -> OutputType -> IO ()
enqueueOutputSIO ctx@Context{..} strm otyp = do
let out = makeOutputIO ctx strm otyp
enqueueOutput outputQ out
syncWithSender' :: Context -> IO Sync -> LoopCheck -> IO ()
syncWithSender' Context{..} pop lc = loop
where
loop = do
s <- pop
case s of
Done -> return ()
Cont newout -> do
cont <- checkLoop lc
when cont $ do
-- This is justified by the precondition above
enqueueOutput outputQ newout
loop
newLoopCheck :: Stream -> Maybe (TBQueue StreamingChunk) -> IO LoopCheck
newLoopCheck strm mtbq = do
tovar <- newTVarIO False
return $
LoopCheck
{ lcTBQ = mtbq
, lcTimeout = tovar
, lcWindow = streamTxFlow strm
}
data LoopCheck = LoopCheck
{ lcTBQ :: Maybe (TBQueue StreamingChunk)
, lcTimeout :: TVar Bool
, lcWindow :: TVar TxFlow
}
checkLoop :: LoopCheck -> IO Bool
checkLoop LoopCheck{..} = atomically $ do
tout <- readTVar lcTimeout
if tout
then return False
else do
waitStreaming' lcTBQ
waitStreamWindowSizeSTM lcWindow
return True
waitStreaming' :: Maybe (TBQueue a) -> STM ()
waitStreaming' Nothing = return ()
waitStreaming' (Just tbq) = do
isEmpty <- isEmptyTBQueue tbq
check (not isEmpty)
waitStreamWindowSizeSTM :: TVar TxFlow -> STM ()
waitStreamWindowSizeSTM txf = do
w <- txWindowSize <$> readTVar txf
check (w > 0)
|
