2ebed4d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Make Map generic version of classifySessionsBy

2ebed4dHarendra Kumar5 years ago

benchmark/Streamly/Benchmark/Prelude/Serial/NestedFold.hs+17−1

compare with master

unifiedsplit

@@ -18,6 +18,8 @@ import Control.DeepSeq (NFData(..))

18import Control.Monad (when)

19import Control.Monad.IO.Class (MonadIO(..))

20import Data.Monoid (Sum(..))

21+import Data.Proxy (Proxy(..))

22+import Data.HashMap.Strict (HashMap)

21import GHC.Generics (Generic)

23import GHC.Generics (Generic)

23import qualified Streamly.Internal.Data.Refold.Type as Refold

25import qualified Streamly.Internal.Data.Refold.Type as Refold

@@ -219,11 +221,25 @@ classifySessionsOf =

219 . Internal.timestamped

221 . Internal.timestamped

220 . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

222 . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

221

223

224+{-# INLINE classifySessionsOfHash #-}

225+classifySessionsOfHash :: (S.MonadAsync m) => SerialT m Int -> m ()

226+classifySessionsOfHash =

227+ S.drain

228+ . Internal.classifySessionsByGeneric

229+ (Proxy :: Proxy (HashMap k v))

230+ 1 False (const (return False)) 3 FL.drain

231+ . Internal.timestamped

232+ . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

233+

222o_n_space_grouping :: Int -> [Benchmark]

234o_n_space_grouping :: Int -> [Benchmark]

223o_n_space_grouping value =

235o_n_space_grouping value =

224 -- Buffering operations using heap proportional to group/window sizes.

236 -- Buffering operations using heap proportional to group/window sizes.

225 [ bgroup "grouping"

237 [ bgroup "grouping"

238+ -- We use 10 element stream per input, so div by 10 here

226− [ benchIOSink (value `div` 10) "classifySessionsOf" classifySessionsOf

239+ [ benchIOSink (value `div` 10) "classifySessionsOf"

240+ classifySessionsOf

241+ , benchIOSink (value `div` 10) "classifySessionsOfHash"

242+ classifySessionsOfHash

227 ]

243 ]

228 ]

244 ]

229

245

bin/bench-config.sh+1−1

@@ -68,7 +68,7 @@ bench_rts_options () {

68 Prelude.Serial/o-1-space/mixed/sum-product-fold) echo -n "-K64M" ;;

70 # XXX These should be moved to o-n-space?

71− Prelude.Serial/o-n-heap/grouping/classifySessionsOf) echo -n "-K1M -M32M" ;;

71+ Prelude.Serial/o-n-heap/grouping/classifySessionsOf*) echo -n "-K1M -M32M" ;;

72 Prelude.Serial/o-n-heap/Functor/*) echo -n "-K4M -M32M" ;;

73 Prelude.Serial/o-n-heap/transformer/*) echo -n "-K8M -M64M" ;;

core/src/Streamly/Internal/Data/IsMap.hs+4

@@ -29,6 +29,7 @@ class IsMap f where

29 mapInsert :: Key f -> a -> f a -> f a

30 mapDelete :: Key f -> f a -> f a

31 mapUnion :: f a -> f a -> f a

32+ mapNull :: f a -> Bool

33instance Ord k => IsMap (Map k) where

34instance Ord k => IsMap (Map k) where

34 type Key (Map k) = k

35 type Key (Map k) = k

@@ -39,6 +40,7 @@ instance Ord k => IsMap (Map k) where

39 mapInsert = Map.insert

40 mapInsert = Map.insert

40 mapDelete = Map.delete

41 mapDelete = Map.delete

41 mapUnion = Map.union

42 mapUnion = Map.union

43+ mapNull = Map.null

43instance IsMap IntMap.IntMap where

45instance IsMap IntMap.IntMap where

44 type Key IntMap.IntMap = Int

46 type Key IntMap.IntMap = Int

@@ -49,6 +51,7 @@ instance IsMap IntMap.IntMap where

49 mapInsert = IntMap.insert

51 mapInsert = IntMap.insert

50 mapDelete = IntMap.delete

52 mapDelete = IntMap.delete

51 mapUnion = IntMap.union

53 mapUnion = IntMap.union

54+ mapNull = IntMap.null

53instance (Hashable k, Eq k) => IsMap (HashMap.HashMap k) where

56instance (Hashable k, Eq k) => IsMap (HashMap.HashMap k) where

54 type Key (HashMap.HashMap k) = k

57 type Key (HashMap.HashMap k) = k

@@ -59,3 +62,4 @@ instance (Hashable k, Eq k) => IsMap (HashMap.HashMap k) where

59 mapInsert = HashMap.insert

62 mapInsert = HashMap.insert

60 mapDelete = HashMap.delete

63 mapDelete = HashMap.delete

61 mapUnion = HashMap.union

64 mapUnion = HashMap.union

65+ mapNull = HashMap.null

src/Streamly/Internal/Data/Stream/IsStream/Reduce.hs+79−62

@@ -89,6 +89,7 @@ module Streamly.Internal.Data.Stream.IsStream.Reduce

89 -- | A new window starts after the previous window is finished.

91 -- , classifyChunksOf

92+ , classifySessionsByGeneric

92 , classifySessionsBy

93 , classifySessionsBy

93 , classifySessionsOf

94 , classifySessionsOf

@@ -153,10 +154,13 @@ import Control.Monad.Catch (MonadThrow)

153import Control.Monad.IO.Class (MonadIO(..))

154import Control.Monad.IO.Class (MonadIO(..))

154import Data.Heap (Entry(..))

155import Data.Heap (Entry(..))

155import Data.Kind (Type)

156import Data.Kind (Type)

157+import Data.Map (Map)

156import Data.Maybe (isNothing)

158import Data.Maybe (isNothing)

159+import Data.Proxy (Proxy(..))

157import Foreign.Storable (Storable)

160import Foreign.Storable (Storable)

158import Streamly.Internal.Control.Concurrent (MonadAsync)

161import Streamly.Internal.Control.Concurrent (MonadAsync)

159import Streamly.Internal.Data.Fold.Type (Fold (..))

162import Streamly.Internal.Data.Fold.Type (Fold (..))

163+import Streamly.Internal.Data.IsMap (IsMap(..))

160import Streamly.Internal.Data.Refold.Type (Refold (..))

164import Streamly.Internal.Data.Refold.Type (Refold (..))

161import Streamly.Internal.Data.Parser (Parser (..))

165import Streamly.Internal.Data.Parser (Parser (..))

162import Streamly.Internal.Data.Array.Foreign.Type (Array)

166import Streamly.Internal.Data.Array.Foreign.Type (Array)

@@ -178,6 +182,7 @@ import qualified Data.Heap as H

178import qualified Data.Map.Strict as Map

182import qualified Data.Map.Strict as Map

179import qualified Streamly.Internal.Data.Array.Foreign.Type as A

183import qualified Streamly.Internal.Data.Array.Foreign.Type as A

180import qualified Streamly.Internal.Data.Fold as FL

184import qualified Streamly.Internal.Data.Fold as FL

185+import qualified Streamly.Internal.Data.IsMap as IsMap

181import qualified Streamly.Internal.Data.Parser.ParserD as PRD

186import qualified Streamly.Internal.Data.Parser.ParserD as PRD

182import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream

187import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream

183import qualified Streamly.Internal.Data.Stream.StreamD as D

188import qualified Streamly.Internal.Data.Stream.StreamD as D

@@ -1150,64 +1155,23 @@ data SessionState t m k a b = SessionState

1150

1155

1151data SessionEntry a b = LiveSession !a !b | ZombieSession

1156data SessionEntry a b = LiveSession !a !b | ZombieSession

1152

1157

1153−-- | @classifySessionsBy tick keepalive predicate timeout fold stream@

1154−-- classifies an input event @stream@ consisting of @(timestamp, (key,

1155−-- value))@ into sessions based on the @key@, folding all the values

1156−-- corresponding to the same key into a session using the supplied @fold@.

1157−--

1158−-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the

1159−-- session key and the folded value is emitted in the output stream. The

1160−-- timeout is measured from the first event in the session. If the @keepalive@

1161−-- option is set to 'True' the timeout is reset to 0 whenever an event is

1162−-- received.

1163−--

1164−-- The @timestamp@ in the input stream is an absolute time from some epoch,

1165−-- characterizing the time when the input event was generated. The notion of

1166−-- current time is maintained by a monotonic event time clock using the

1167−-- timestamps seen in the input stream. The latest timestamp seen till now is

1168−-- used as the base for the current time. When no new events are seen, a timer

1169−-- is started with a clock resolution of @tick@ seconds. This timer is used to

1170−-- detect session timeouts in the absence of new events.

1171−--

1172−-- To ensure an upper bound on the memory used the number of sessions can be

1173−-- limited to an upper bound. If the ejection @predicate@ returns 'True', the

1174−-- oldest session is ejected before inserting a new session.

1175−--

1176−-- When the stream ends any buffered sessions are ejected immediately.

1177−--

1178−-- If a session key is received even after a session has finished, another

1179−-- session is created for that key.

1180−--

1181−-- >>> :{

1182−-- Stream.mapM_ print

1183−-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)

1184−-- $ Stream.timestamped

1185−-- $ Stream.delay 0.1

1186−-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']

1187−-- :}

1188−-- (1,"abc")

1189−-- (2,"abc")

1190−-- (3,"abc")

1191−--

1192−-- /Pre-release/

1193−

1194−-- XXX Use a proxy type for the map type to use.

1195-- XXX Use mutable IORef in accumulator

1158-- XXX Use mutable IORef in accumulator

1196--

1159--

1197-- XXX this fuses with INLINE but it may be too much work for the compiler.

1160-- XXX this fuses with INLINE but it may be too much work for the compiler.

1198-- Maybe we can NOINLINE the heap processing stuff and then INLINE it to reduce

1161-- Maybe we can NOINLINE the heap processing stuff and then INLINE it to reduce

1199-- the code bloat.

1162-- the code bloat.

1200−{-# INLINABLE classifySessionsBy #-}

1163+{-# INLINABLE classifySessionsByGeneric #-}

1201−classifySessionsBy

1164+classifySessionsByGeneric

1202− :: (IsStream t, MonadAsync m, Ord k)

1165+ :: (IsStream t, MonadAsync m, Ord (Key f))

1166+ => Proxy (f s)

1203− => Double -- ^ timer tick in seconds

1167+ -> Double -- ^ timer tick in seconds

1204 -> Bool -- ^ reset the timer when an event is received

1168 -> Bool -- ^ reset the timer when an event is received

1205 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1169 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1206 -> Double -- ^ session timeout in seconds

1170 -> Double -- ^ session timeout in seconds

1207 -> Fold m a b -- ^ Fold to be applied to session data

1171 -> Fold m a b -- ^ Fold to be applied to session data

1208− -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)

1172+ -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)

1209− -> t m (k, b) -- ^ session key, fold result

1173+ -> t m (Key f, b) -- ^ session key, fold result

1210−classifySessionsBy tick reset ejectPred tmout

1174+classifySessionsByGeneric _ tick reset ejectPred tmout

1211 (Fold step initial extract) input =

1175 (Fold step initial extract) input =

1212 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)

1176 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)

1213 $ scanlMAfter' sstep (return szero) flush

1177 $ scanlMAfter' sstep (return szero) flush

@@ -1223,7 +1187,7 @@ classifySessionsBy tick reset ejectPred tmout

1223 , sessionEventTime = toAbsTime (0 :: MilliSecond64)

1187 , sessionEventTime = toAbsTime (0 :: MilliSecond64)

1224 , sessionCount = 0

1188 , sessionCount = 0

1225 , sessionTimerHeap = H.empty

1189 , sessionTimerHeap = H.empty

1226− , sessionKeyValueMap = Map.empty

1190+ , sessionKeyValueMap = IsMap.mapEmpty

1227 , sessionOutputStream = IsStream.nil

1191 , sessionOutputStream = IsStream.nil

1228 }

1192 }

1229

1193

@@ -1281,7 +1245,7 @@ classifySessionsBy tick reset ejectPred tmout

1281 -- way to achieve mutual recursion.

1245 -- way to achieve mutual recursion.

1282 --

1246 --

1283 let curTime = max sessionEventTime timestamp

1247 let curTime = max sessionEventTime timestamp

1284− mOld = Map.lookup key sessionKeyValueMap

1248+ mOld = IsMap.mapLookup key sessionKeyValueMap

1285 let done fb = do

1249 let done fb = do

1286 -- deleting a key from the heap is expensive, so we never

1250 -- deleting a key from the heap is expensive, so we never

1287 -- delete a key from heap, we just purge it from the Map and it

1251 -- delete a key from heap, we just purge it from the Map and it

@@ -1291,7 +1255,8 @@ classifySessionsBy tick reset ejectPred tmout

1291 --

1255 --

1292 let (mp, cnt) = case mOld of

1256 let (mp, cnt) = case mOld of

1293 Just (LiveSession _ _) ->

1257 Just (LiveSession _ _) ->

1258+ ( IsMap.mapInsert

1294− ( Map.insert key ZombieSession sessionKeyValueMap

1259+ key ZombieSession sessionKeyValueMap

1295 , sessionCount - 1

1260 , sessionCount - 1

1296 )

1261 )

1297 _ -> (sessionKeyValueMap, sessionCount)

1262 _ -> (sessionKeyValueMap, sessionCount)

@@ -1324,7 +1289,7 @@ classifySessionsBy tick reset ejectPred tmout

1324 Just _ -> return vars

1289 Just _ -> return vars

1325

1290

1326 let acc = LiveSession expiry fs1

1291 let acc = LiveSession expiry fs1

1327− mp2 = Map.insert key acc mp1

1292+ mp2 = IsMap.mapInsert key acc mp1

1328 return $ SessionState

1293 return $ SessionState

1329 { sessionCurTime = curTime

1294 { sessionCurTime = curTime

1330 , sessionEventTime = curTime

1295 , sessionEventTime = curTime

@@ -1371,17 +1336,17 @@ classifySessionsBy tick reset ejectPred tmout

1371 ejectEntry hp mp out cnt acc key = do

1336 ejectEntry hp mp out cnt acc key = do

1372 sess <- extract acc

1337 sess <- extract acc

1373 let out1 = (key, sess) `cons` out

1338 let out1 = (key, sess) `cons` out

1374− let mp1 = Map.delete key mp

1339+ let mp1 = IsMap.mapDelete key mp

1375 return (hp, mp1, out1, cnt - 1)

1340 return (hp, mp1, out1, cnt - 1)

1376

1341

1377 ejectAll (hp, mp, out, !cnt) = do

1342 ejectAll (hp, mp, out, !cnt) = do

1378 let hres = H.uncons hp

1343 let hres = H.uncons hp

1379 case hres of

1344 case hres of

1380 Just (Entry _ key, hp1) -> do

1345 Just (Entry _ key, hp1) -> do

1381− r <- case Map.lookup key mp of

1346+ r <- case IsMap.mapLookup key mp of

1382 Nothing -> return (hp1, mp, out, cnt)

1347 Nothing -> return (hp1, mp, out, cnt)

1383 Just ZombieSession ->

1348 Just ZombieSession ->

1384− return (hp1, Map.delete key mp, out, cnt)

1349+ return (hp1, IsMap.mapDelete key mp, out, cnt)

1385 Just (LiveSession _ acc) ->

1350 Just (LiveSession _ acc) ->

1386 ejectEntry hp1 mp out cnt acc key

1351 ejectEntry hp1 mp out cnt acc key

1387 ejectAll r

1352 ejectAll r

@@ -1393,10 +1358,10 @@ classifySessionsBy tick reset ejectPred tmout

1393 let hres = H.uncons hp

1358 let hres = H.uncons hp

1394 case hres of

1359 case hres of

1395 Just (Entry expiry key, hp1) ->

1360 Just (Entry expiry key, hp1) ->

1396− case Map.lookup key mp of

1361+ case IsMap.mapLookup key mp of

1397 Nothing -> ejectOne (hp1, mp, out, cnt)

1362 Nothing -> ejectOne (hp1, mp, out, cnt)

1398 Just ZombieSession ->

1363 Just ZombieSession ->

1399− ejectOne (hp1, Map.delete key mp, out, cnt)

1364+ ejectOne (hp1, IsMap.mapDelete key mp, out, cnt)

1400 Just (LiveSession expiry1 acc) -> do

1365 Just (LiveSession expiry1 acc) -> do

1401 if not reset || expiry1 <= expiry

1366 if not reset || expiry1 <= expiry

1402 then ejectEntry hp1 mp out cnt acc key

1367 then ejectEntry hp1 mp out cnt acc key

@@ -1405,7 +1370,7 @@ classifySessionsBy tick reset ejectPred tmout

1405 let hp2 = H.insert (Entry expiry1 key) hp1

1370 let hp2 = H.insert (Entry expiry1 key) hp1

1406 in ejectOne (hp2, mp, out, cnt)

1371 in ejectOne (hp2, mp, out, cnt)

1407 Nothing -> do

1372 Nothing -> do

1408− assert (Map.null mp) (return ())

1373+ assert (IsMap.mapNull mp) (return ())

1409 return (hp, mp, out, cnt)

1374 return (hp, mp, out, cnt)

1410

1375

1411 ejectExpired session@SessionState{..} curTime = do

1376 ejectExpired session@SessionState{..} curTime = do

@@ -1434,10 +1399,10 @@ classifySessionsBy tick reset ejectPred tmout

1434 return (r, r)

1399 return (r, r)

1435 if eject

1400 if eject

1436 then

1401 then

1437− case Map.lookup key mp of

1402+ case IsMap.mapLookup key mp of

1438 Nothing -> ejectLoop hp1 mp out cnt

1403 Nothing -> ejectLoop hp1 mp out cnt

1439 Just ZombieSession ->

1404 Just ZombieSession ->

1440− ejectLoop hp1 (Map.delete key mp) out cnt

1405+ ejectLoop hp1 (IsMap.mapDelete key mp) out cnt

1441 Just (LiveSession expiry1 acc) -> do

1406 Just (LiveSession expiry1 acc) -> do

1442 if expiry1 <= curTime || not reset || force

1407 if expiry1 <= curTime || not reset || force

1443 then do

1408 then do

@@ -1450,9 +1415,61 @@ classifySessionsBy tick reset ejectPred tmout

1450 in ejectLoop hp2 mp out cnt

1415 in ejectLoop hp2 mp out cnt

1451 else return (hp, mp, out, cnt)

1416 else return (hp, mp, out, cnt)

1452 Nothing -> do

1417 Nothing -> do

1453− assert (Map.null mp) (return ())

1418+ assert (IsMap.mapNull mp) (return ())

1454 return (hp, mp, out, cnt)

1419 return (hp, mp, out, cnt)

1455

1420

1421+-- | @classifySessionsBy tick keepalive predicate timeout fold stream@

1422+-- classifies an input event @stream@ consisting of @(timestamp, (key,

1423+-- value))@ into sessions based on the @key@, folding all the values

1424+-- corresponding to the same key into a session using the supplied @fold@.

1425+--

1426+-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the

1427+-- session key and the folded value is emitted in the output stream. The

1428+-- timeout is measured from the first event in the session. If the @keepalive@

1429+-- option is set to 'True' the timeout is reset to 0 whenever an event is

1430+-- received.

1431+--

1432+-- The @timestamp@ in the input stream is an absolute time from some epoch,

1433+-- characterizing the time when the input event was generated. The notion of

1434+-- current time is maintained by a monotonic event time clock using the

1435+-- timestamps seen in the input stream. The latest timestamp seen till now is

1436+-- used as the base for the current time. When no new events are seen, a timer

1437+-- is started with a clock resolution of @tick@ seconds. This timer is used to

1438+-- detect session timeouts in the absence of new events.

1439+--

1440+-- To ensure an upper bound on the memory used the number of sessions can be

1441+-- limited to an upper bound. If the ejection @predicate@ returns 'True', the

1442+-- oldest session is ejected before inserting a new session.

1443+--

1444+-- When the stream ends any buffered sessions are ejected immediately.

1445+--

1446+-- If a session key is received even after a session has finished, another

1447+-- session is created for that key.

1448+--

1449+-- >>> :{

1450+-- Stream.mapM_ print

1451+-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)

1452+-- $ Stream.timestamped

1453+-- $ Stream.delay 0.1

1454+-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']

1455+-- :}

1456+-- (1,"abc")

1457+-- (2,"abc")

1458+-- (3,"abc")

1459+--

1460+-- /Pre-release/

1461+{-# INLINE classifySessionsBy #-}

1462+classifySessionsBy

1463+ :: (IsStream t, MonadAsync m, Ord k)

1464+ => Double -- ^ timer tick in seconds

1465+ -> Bool -- ^ reset the timer when an event is received

1466+ -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1467+ -> Double -- ^ session timeout in seconds

1468+ -> Fold m a b -- ^ Fold to be applied to session data

1469+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)

1470+ -> t m (k, b) -- ^ session key, fold result

1471+classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k s))

1472+

1456-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive

1473-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive

1457-- option set to 'True'.

1474-- option set to 'True'.

1458--

1475--