e0ad726 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix generic Map support for classifySessionsBy

e0ad726Harendra Kumar5 years ago

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

compare with master

unifiedsplit

@@ -226,7 +226,7 @@ classifySessionsOfHash :: (S.MonadAsync m) => SerialT m Int -> m ()

226classifySessionsOfHash =

227 S.drain

228 . Internal.classifySessionsByGeneric

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

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

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

231 . Internal.timestamped

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

src/Streamly/Internal/Data/Stream/IsStream/Reduce.hs+21−22

@@ -179,7 +179,6 @@ import Streamly.Internal.Data.Time.Units

179 , toAbsTime)

180

181import qualified Data.Heap as H

182−import qualified Data.Map.Strict as Map

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

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

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

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

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

184import qualified Streamly.Internal.Data.IsMap as IsMap

@@ -1142,18 +1141,18 @@ classifyKeepAliveChunks

1142classifyKeepAliveChunks spanout = classifyChunksBy spanout True

1141classifyKeepAliveChunks spanout = classifyChunksBy spanout True

1143-}

1142-}

1144

1143

1145−data SessionState t m k a b = SessionState

1144+data SessionState t m f s b = SessionState

1146 { sessionCurTime :: !AbsTime -- ^ time since last event

1145 { sessionCurTime :: !AbsTime -- ^ time since last event

1147 , sessionEventTime :: !AbsTime -- ^ time as per last event

1146 , sessionEventTime :: !AbsTime -- ^ time as per last event

1148 -- We can use the Map size instead of maintaining a count, but if we have

1147 -- We can use the Map size instead of maintaining a count, but if we have

1149 -- to switch to HashMap then it can be useful.

1148 -- to switch to HashMap then it can be useful.

1150 , sessionCount :: !Int -- ^ total number of sessions in progress

1149 , sessionCount :: !Int -- ^ total number of sessions in progress

1151− , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts

1150+ , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts

1152− , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys

1151+ , sessionKeyValueMap :: f s -- ^ Stored sessions for keys

1153− , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions

1152+ , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions

1154 }

1153 }

1155

1154

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

1155+data SessionEntry s = LiveSession !AbsTime !s | ZombieSession

1157

1156

1158-- delete from map and output the fold accumulator

1157-- delete from map and output the fold accumulator

1159ejectEntry :: (Monad m, IsStream t, IsMap f) =>

1158ejectEntry :: (Monad m, IsStream t, IsMap f) =>

@@ -1172,10 +1171,10 @@ ejectEntry extract hp mp out cnt acc key = do

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

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

1173

1172

1174{-# NOINLINE flush #-}

1173{-# NOINLINE flush #-}

1175−flush :: (Monad m, Ord k, IsStream t) =>

1174+flush :: (Monad m, IsMap f, IsStream t) =>

1176 (s -> m b)

1175 (s -> m b)

1177− -> SessionState t m k (SessionEntry a s) b

1176+ -> SessionState t m f (SessionEntry s) b

1178− -> m (SessionState t m k (SessionEntry a s) b)

1177+ -> m (SessionState t m f (SessionEntry s) b)

1179flush extract session@SessionState{..} = do

1178flush extract session@SessionState{..} = do

1180 (hp', mp', out, count) <-

1179 (hp', mp', out, count) <-

1181 ejectAll

1180 ejectAll

@@ -1205,20 +1204,20 @@ flush extract session@SessionState{..} = do

1205 ejectEntry extract hp1 mp out cnt acc key

1204 ejectEntry extract hp1 mp out cnt acc key

1206 ejectAll r

1205 ejectAll r

1207 Nothing -> do

1206 Nothing -> do

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

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

1209 return (hp, mp, out, cnt)

1208 return (hp, mp, out, cnt)

1210

1209

1211{-# NOINLINE ejectOne #-}

1210{-# NOINLINE ejectOne #-}

1212−ejectOne :: (IsMap f, IsStream t, Ord k, Monad m) =>

1211+ejectOne :: (IsMap f, IsStream t, Monad m) =>

1213 Bool

1212 Bool

1214 -> (acc -> m b)

1213 -> (acc -> m b)

1215− -> ( H.Heap (Entry k (Key f))

1214+ -> ( H.Heap (Entry AbsTime (Key f))

1216− , f (SessionEntry k acc)

1215+ , f (SessionEntry acc)

1217 , t m (Key f, b)

1216 , t m (Key f, b)

1218 , Int

1217 , Int

1219 )

1218 )

1220− -> m ( H.Heap (Entry k (Key f))

1219+ -> m ( H.Heap (Entry AbsTime (Key f))

1221− , f (SessionEntry k acc)

1220+ , f (SessionEntry acc)

1222 , t m (Key f, b), Int

1221 , t m (Key f, b), Int

1223 )

1222 )

1224ejectOne reset extract = go

1223ejectOne reset extract = go

@@ -1245,13 +1244,13 @@ ejectOne reset extract = go

1245 return (hp, mp, out, cnt)

1244 return (hp, mp, out, cnt)

1246

1245

1247{-# NOINLINE ejectExpired #-}

1246{-# NOINLINE ejectExpired #-}

1248−ejectExpired :: (Ord k, IsStream t, Monad m) =>

1247+ejectExpired :: (IsMap f, IsStream t, Monad m) =>

1249 Bool

1248 Bool

1250 -> (Int -> m Bool)

1249 -> (Int -> m Bool)

1251 -> (acc -> m b)

1250 -> (acc -> m b)

1252− -> SessionState t m k (SessionEntry AbsTime acc) b

1251+ -> SessionState t m f (SessionEntry acc) b

1253 -> AbsTime

1252 -> AbsTime

1254− -> m (SessionState t m k (SessionEntry AbsTime acc) b)

1253+ -> m (SessionState t m f (SessionEntry acc) b)

1255ejectExpired reset ejectPred extract session@SessionState{..} curTime = do

1254ejectExpired reset ejectPred extract session@SessionState{..} curTime = do

1256 (hp', mp', out, count) <-

1255 (hp', mp', out, count) <-

1257 ejectLoop

1256 ejectLoop

@@ -1300,8 +1299,8 @@ ejectExpired reset ejectPred extract session@SessionState{..} curTime = do

1300-- XXX Use mutable IORef in accumulator

1299-- XXX Use mutable IORef in accumulator

1301{-# INLINE classifySessionsByGeneric #-}

1300{-# INLINE classifySessionsByGeneric #-}

1302classifySessionsByGeneric

1301classifySessionsByGeneric

1303− :: (IsStream t, MonadAsync m, Ord (Key f))

1302+ :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)

1304− => Proxy (f s)

1303+ => Proxy (f :: (Type -> Type))

1305 -> Double -- ^ timer tick in seconds

1304 -> Double -- ^ timer tick in seconds

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

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

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

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

@@ -1325,7 +1324,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout

1325 , sessionEventTime = toAbsTime (0 :: MilliSecond64)

1324 , sessionEventTime = toAbsTime (0 :: MilliSecond64)

1326 , sessionCount = 0

1325 , sessionCount = 0

1327 , sessionTimerHeap = H.empty

1326 , sessionTimerHeap = H.empty

1328− , sessionKeyValueMap = IsMap.mapEmpty

1327+ , sessionKeyValueMap = IsMap.mapEmpty :: f s

1329 , sessionOutputStream = IsStream.nil

1328 , sessionOutputStream = IsStream.nil

1330 }

1329 }

1331

1330

@@ -1505,7 +1504,7 @@ classifySessionsBy

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

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

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

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

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

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

1508−classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k s))

1507+classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))

1509

1508

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

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

1511-- option set to 'True'.

1510-- option set to 'True'.