@@ -11,8 +11,7 @@ import Control.Applicative ((<$>),(<*>))
1111import Control.Concurrent (forkIO)
1212import Control.Concurrent.STM
1313import Control.Exception (SomeException)
14−import Control.Monad (void,(<$!>))
15−import Control.Reaper
14+import Control.Monad (void)
1615import Data.ByteString (ByteString)
1716import qualified Data.ByteString as BS
1817import Data.IntMap.Strict (IntMap, IntMap)
@@ -110,7 +109,7 @@ data Context = Context {
110109
111110newContext :: IO Context
112111newContext = Context <$> newIORef defaultSettings
113− <*> initialize 10 -- fixme: hard coding: 10
112+ <*> newStreamTable
114113 <*> newIORef 0
115114 <*> newIORef 0
116115 <*> newIORef Nothing
@@ -122,9 +121,7 @@ newContext = Context <$> newIORef defaultSettings
122121 <*> newTVarIO defaultInitialWindowSize
123122
124123clearContext :: Context -> IO ()
125−clearContext ctx = do
126− !_ <- reaperStop $ streamTable ctx
127− return ()
124+clearContext _ctx = return ()
128125
129126----------------------------------------------------------------
130127
@@ -203,45 +200,30 @@ opened Context{concurrency} Stream{streamState} = do
203200 writeIORef streamState (Open JustOpened)
204201
205202closed :: Context -> Stream -> ClosedCode -> IO ()
206−closed Context{concurrency} Stream{streamState} cc = do
203+closed Context{concurrency,streamTable} Stream{streamState,streamNumber} cc = do
204+ remove streamTable streamNumber
207205 atomicModifyIORef' concurrency (\x -> (x-1,()))
208− writeIORef streamState (Closed cc)
206+ writeIORef streamState (Closed cc) -- anyway
209207
210208----------------------------------------------------------------
211209
212−type StreamTable = Reaper (IntMap Stream) (M.Key, Stream)
210+newtype StreamTable = StreamTable (IORef (IntMap Stream))
213211
214−initialize :: Int -> IO StreamTable
215−initialize duration = mkReaper settings
216− where
217− settings = defaultReaperSettings {
218− reaperAction = clean
219− , reaperDelay = duration * 1000000
220− , reaperCons = uncurry M.insert
221− , reaperNull = M.null
222− , reaperEmpty = M.empty
223− }
224−
225−clean :: IntMap Stream -> IO (IntMap Stream -> IntMap Stream)
226−clean old = do
227− !new <- M.fromAscList <$> prune oldlist []
228− return $! M.union new
229− where
230− !oldlist = M.toDescList old
231− prune [] lst = return lst
232− prune (x@(_,s):xs) lst = do
233− !st <- readIORef (streamState s)
234− if isClosed st then
235− prune xs lst
236− else
237− prune xs (x:lst)
212+newStreamTable :: IO StreamTable
213+newStreamTable = StreamTable <$> newIORef M.empty
238214
239215insert :: StreamTable -> M.Key -> Stream -> IO ()
240−insert strmtbl k v = reaperAdd strmtbl (k,v)
216+insert (StreamTable ref) k v = atomicModifyIORef' ref $ \m ->
217+ let !m' = M.insert k v m
218+ in (m', ())
241219
242−search :: StreamTable -> M.Key -> IO (Maybe Stream)
243−search strmtbl k = M.lookup k <$!> reaperRead strmtbl
220+remove :: StreamTable -> M.Key -> IO ()
221+remove (StreamTable ref) k = atomicModifyIORef' ref $ \m ->
222+ let !m' = M.delete k m
223+ in (m', ())
244224
225+search :: StreamTable -> M.Key -> IO (Maybe Stream)
226+search (StreamTable ref) k = M.lookup k <$> readIORef ref
245227
246228-- INVARIANT: streams in the output queue have non-zero window size.
247229enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO ()