@@ -146,7 +146,6 @@ where
147#include "inline.hs"
147#include "inline.hs"
149−import Control.Concurrent (threadDelay)
150import Control.Exception (assert)
149import Control.Exception (assert)
151import Control.Monad.Catch (MonadThrow)
150import Control.Monad.Catch (MonadThrow)
152import Control.Monad.IO.Class (MonadIO(..))
151import Control.Monad.IO.Class (MonadIO(..))
@@ -165,8 +164,6 @@ import Streamly.Internal.Data.Stream.IsStream.Common
165 , interjectSuffix
164 , interjectSuffix
166 , intersperseM
165 , intersperseM
170 , scanlMAfter'
167 , scanlMAfter'
171 , splitOnSeq
168 , splitOnSeq
172 , fromPure)
169 , fromPure)
@@ -1188,7 +1185,9 @@ classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1188data SessionState t m k a b = SessionState
1185data SessionState t m k a b = SessionState
1189 { sessionCurTime :: !AbsTime -- ^ time since last event
1186 { sessionCurTime :: !AbsTime -- ^ time since last event
1190 , sessionEventTime :: !AbsTime -- ^ time as per last event
1187 , sessionEventTime :: !AbsTime -- ^ time as per last event
1188+ -- We can use the Map size instead of maintaining a count, but if we have
1189+ -- to switch to HashMap then it can be useful.
1191− , sessionCount :: !Int -- ^ total number sessions in progress
1190+ , sessionCount :: !Int -- ^ total number of sessions in progress
1192 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1191 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1193 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1192 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1194 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1193 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
@@ -1219,6 +1218,11 @@ data SessionEntry a b = LiveSession !a !b | ZombieSession
1219-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1218-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1220-- oldest session is ejected before inserting a new session.
1219-- oldest session is ejected before inserting a new session.
1221+-- When the stream ends any buffered sessions are ejected immediately.
1223+-- If a session key is received even after a session has finished, another
1224+-- session is created for that key.
1222-- >>> :{
1226-- >>> :{
1223-- Stream.mapM_ print
1227-- Stream.mapM_ print
1224-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1228-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
@@ -1243,9 +1247,11 @@ classifySessionsBy
1243 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1247 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1244 -> t m (k, b) -- ^ session key, fold result
1248 -> t m (k, b) -- ^ session key, fold result
1245classifySessionsBy tick reset ejectPred tmout
1249classifySessionsBy tick reset ejectPred tmout
1246− (Fold step initial extract) str =
1250+ (Fold step initial extract) input =
1247− concatMap sessionOutputStream $
1251+ concatMap sessionOutputStream
1248− scanlMAfter' sstep (return szero) flush stream
1252+ $ scanlMAfter' sstep (return szero) flush
1253+ $ interjectSuffix tick (return Nothing)
@@ -1272,12 +1278,14 @@ classifySessionsBy tick reset ejectPred tmout
1273 -- Got a new stream input element
1279 -- Got a new stream input element
1274 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1280 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1275− -- XXX we should use a heap in pinned memory to scale it to a large
1281+ -- XXX instead of a heap we could use a timer wheel.
1278 -- XXX if the key is an Int, we can also use an IntMap for slightly
1283 -- XXX if the key is an Int, we can also use an IntMap for slightly
1279 -- better performance.
1284 -- better performance.
1288+ -- Values for each key are collected in a map using the supplied fold.
1281 -- When we insert a key in the Map we insert an entry into the heap as
1289 -- When we insert a key in the Map we insert an entry into the heap as
1282 -- well with the session expiry as the sort key. The Map entry
1290 -- well with the session expiry as the sort key. The Map entry
1283 -- consists of the fold result, and the expiry time of the session. If
1291 -- consists of the fold result, and the expiry time of the session. If
@@ -1302,6 +1310,14 @@ classifySessionsBy tick reset ejectPred tmout
1302 -- finished we still keep a dummy Map entry (ZombieSession) until the
1310 -- finished we still keep a dummy Map entry (ZombieSession) until the
1303 -- heap entry is removed. That way if we have a Map entry we do not
1311 -- heap entry is removed. That way if we have a Map entry we do not
1304 -- insert a heap entry because we know it is already there.
1312 -- insert a heap entry because we know it is already there.
1313+ -- XXX The ZombieSession mechanism does not work as expected as we
1314+ -- ignore ZombieSession when inserting a new entry. Anyway, we can
1315+ -- remove this mechanism as at most only two heap entries may be
1316+ -- created and they will be ultimately cleaned up.
1318+ -- Heap processing needs the map and map processing needs the heap,
1319+ -- therefore we cannot separate the two for modularity unless we have a
1320+ -- way to achieve mutual recursion.
1306 let curTime = max sessionEventTime timestamp
1322 let curTime = max sessionEventTime timestamp
1307 mOld = Map.lookup key sessionKeyValueMap
1323 mOld = Map.lookup key sessionKeyValueMap
@@ -1476,12 +1492,6 @@ classifySessionsBy tick reset ejectPred tmout
1476 assert (Map.null mp) (return ())
1492 assert (Map.null mp) (return ())
1477 return (hp, mp, out, cnt)
1493 return (hp, mp, out, cnt)
1479− -- merge timer events in the stream
1480− stream = map Just str `parallelFst` repeatM timer
1482− liftIO $ threadDelay (round $ tick * 1000000)
1485-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1495-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1486-- option set to 'True'.
1496-- option set to 'True'.