63ce5d9 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add lchunksOf and lsessionsOf

63ce5d9Harendra Kumar7 years ago

src/Streamly/Fold.hs+95−10

compare with master

unifiedsplit

@@ -221,6 +221,9 @@ module Streamly.Fold

221 , chunksOf

222 , sessionsOf

223

224+ , lchunksOf

225+ , lsessionsOf

226+

224 -- *** By Elements

227 -- *** By Elements

225 , splitBy

228 , splitBy

226 , splitSuffixBy

229 , splitSuffixBy

@@ -346,9 +349,13 @@ module Streamly.Fold

346 )

349 )

347where

350where

348

351

349−import Control.Concurrent (threadDelay)

352+import Control.Concurrent (threadDelay, forkIO, killThread)

353+import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)

354+import Control.Exception (SomeException(..), catch, mask)

350import Control.Monad (void)

355import Control.Monad (void)

356+import Control.Monad.Catch (throwM)

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

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

358+import Control.Monad.Trans.Control (control)

352import Data.Functor.Identity (Identity)

359import Data.Functor.Identity (Identity)

353import Data.Heap (Entry(..))

360import Data.Heap (Entry(..))

354import Data.Map.Strict (Map)

361import Data.Map.Strict (Map)

@@ -2167,26 +2174,104 @@ lconcatMap ::(IsStream t, Monad m) => (a -> t m b)

2167 -> Fold m b c

2174 -> Fold m b c

2168 -> Fold m a c

2175 -> Fold m a c

2169lconcatMap s f1 f2 = undefined

2176lconcatMap s f1 f2 = undefined

2177+-}

2170

2178

2171-- All the grouping transformation that we apply to a stream can also be

2179-- All the grouping transformation that we apply to a stream can also be

2172-- applied to a fold input stream.

2180-- applied to a fold input stream.

2173

2181

2174−-- | Group the input elements of a fold input stream by some criterion and fold

2182+-- | Group the input stream into groups of @n@ elements each and then fold each

2175−-- each group using a given fold before its fed to the fold.

2183+-- group using the provided fold function.

2176--

2184--

2185+-- @

2186+--

2187+-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c

2188+--

2189+-- @

2190+--

2191+lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c

2192+lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =

2193+ Fold step' initial' extract'

2194+

2195+ where

2196+

2197+ initial' = (Tuple3' 0) <$> initial1 <*> initial2

2198+ step' (Tuple3' i r1 r2) a = do

2199+ if i < n

2200+ then do

2201+ res <- step1 r1 a

2202+ return $ Tuple3' (i + 1) res r2

2203+ else do

2204+ i1 <- initial1

2205+ res1 <- extract1 r1

2206+ res <- step2 r2 res1

2207+ return $ Tuple3' 0 i1 res

2208+ extract' (Tuple3' _ _ r) = extract2 r

2209+

2210+-- | Group the input stream into windows of n second each and then fold each

2211+-- group using the provided fold function.

2212+--

2177−-- For example, we can copy and distribute a stream to multiple folds and then

2213+-- For example, we can copy and distribute a stream to multiple folds where

2178−-- in each fold we can group the input differently e.g. by one second, one

2214+-- each fold can group the input differently e.g. by one second, one minute and

2179−-- minute and one hour windows respectively and fold each resulting stream of

2215+-- one hour windows respectively and fold each resulting stream of folds.

2180−-- folds.

2181--

2216--

2182-- @

2217-- @

2183--

2218--

2184-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c

2219-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c

2185--

2220--

2186-- @

2221-- @

2187−lchunksOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c

2222+lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c

2188−lchunksOf n f1 f2 = undefined

2223+lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =

2189−-}

2224+ Fold step' initial' extract'

2225+

2226+ where

2227+

2228+ -- XXX MVar may be expensive we need a cheaper synch mechanism here

2229+ initial' = do

2230+ i1 <- initial1

2231+ i2 <- initial2

2232+ mv1 <- liftIO $ newMVar i1

2233+ mv2 <- liftIO $ newMVar (Right i2)

2234+ t <- control $ \run ->

2235+ mask $ \restore -> do

2236+ tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))

2237+ (handleChildException mv2)

2238+ run (return tid)

2239+ return $ Tuple3' t mv1 mv2

2240+ step' acc@(Tuple3' _ mv1 _) a = do

2241+ r1 <- liftIO $ takeMVar mv1

2242+ res <- step1 r1 a

2243+ liftIO $ putMVar mv1 res

2244+ return acc

2245+ extract' (Tuple3' tid _ mv2) = do

2246+ r2 <- liftIO $ takeMVar mv2

2247+ liftIO $ killThread tid

2248+ case r2 of

2249+ Left e -> throwM e

2250+ Right x -> extract2 x

2251+

2252+ timerThread mv1 mv2 = do

2253+ liftIO $ threadDelay (round $ n * 1000000)

2254+

2255+ r1 <- liftIO $ takeMVar mv1

2256+ i1 <- initial1

2257+ liftIO $ putMVar mv1 i1

2258+

2259+ res1 <- extract1 r1

2260+ r2 <- liftIO $ takeMVar mv2

2261+ res <- case r2 of

2262+ Left _ -> return r2

2263+ Right x -> fmap Right $ step2 x res1

2264+ liftIO $ putMVar mv2 res

2265+ timerThread mv1 mv2

2266+

2267+ handleChildException ::

2268+ MVar (Either SomeException a) -> SomeException -> IO ()

2269+ handleChildException mv2 e = do

2270+ r2 <- takeMVar mv2

2271+ let r = case r2 of

2272+ Left _ -> r2

2273+ Right _ -> Left e

2274+ putMVar mv2 r

2190

2275

2191------------------------------------------------------------------------------

2276------------------------------------------------------------------------------

2192-- Windowed classification

2277-- Windowed classification