5

是否可以创建管道来获取在特定时间段内向下游发送的所有值？我正在实现一个服务器，其中协议允许我连接传出数据包并将它们压缩在一起，所以我想ByteString每 100 毫秒有效地“清空”下游 s的队列，mappend然后它们一起产生到下一个管道进行压缩。

4

3 回答 3

3

这是一个使用pipes-concurrency. 你给它任何Input，它会定期耗尽所有值的输入：

import Control.Applicative ((<|>))
import Control.Concurrent (threadDelay)
import Data.Foldable (forM_)
import Pipes
import Pipes.Concurrent

drainAll :: Input a -> STM (Maybe [a])
drainAll i = do
    ma <- recv i
    case ma of
        Nothing -> return Nothing
        Just a  -> loop (a:)
  where
    loop diffAs = do
        ma <- recv i <|> return Nothing
        case ma of
            Nothing -> return (Just (diffAs []))
            Just a  -> loop (diffAs . (a:))

bucketsEvery :: Int -> Input a -> Producer [a] IO ()
bucketsEvery microseconds i = loop
  where
    loop = do
        lift $ threadDelay microseconds
        ma <- lift $ atomically $ drainAll i
        forM_ ma $ \a -> do
            yield a
            loop

通过选择Buffer用于构建Input.

如果您是. pipes-concurrency_ _ _spawnBufferInput

于 2014-03-09T19:03:44.857 回答
1

这是一个可能的解决方案。它基于 aPipe标签ByteStrings 与 a 一起向下游Bool移动，以识别ByteStrings属于同一个“时间桶”。

首先，一些进口：

import Data.AdditiveGroup
import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as BL
import qualified Data.ByteString.Lazy.Builder as BB
import Data.Thyme.Clock
import Data.Thyme.Clock.POSIX
import Control.Monad.State.Strict
import Control.Lens (view)
import Control.Concurrent (threadDelay)
import Pipes
import Pipes.Lift
import qualified Pipes.Prelude as P
import qualified Pipes.Group as PG

这是标记Pipe。它在StateT内部使用：

tagger :: Pipe B.ByteString (B.ByteString,Bool) IO ()
tagger = do
    startTime <- liftIO getPOSIXTime
    evalStateP (startTime,False) $ forever $ do
        b <- await
        currentTime <- liftIO getPOSIXTime
        -- (POSIXTime,Bool) inner state
        (baseTime,tag) <- get
        if (currentTime ^-^ baseTime > timeLimit)
            then let tag' = not tag in
                 yield (b,tag') >> put (currentTime, tag')
            else yield $ (b,tag)
    where
        timeLimit = fromSeconds 0.1

然后我们可以使用pipes-group包中的函数将ByteString属于同一“时间桶”的 s 分组为惰性ByteStrings：

batch :: Producer B.ByteString IO () -> Producer BL.ByteString IO ()
batch producer =  PG.folds (<>) mempty BB.toLazyByteString
                . PG.maps (flip for $ yield . BB.byteString . fst)
                . view (PG.groupsBy $ \t1 t2-> snd t1 == snd t2)
                $ producer >-> tagger

它似乎正确批处理。这个程序：

main :: IO ()
main = do
    count <- P.length $ batch (yield "boo" >> yield "baa")
    putStrLn $ show count
    count <- P.length $ batch (yield "boo" >> yield "baa" 
                               >> liftIO (threadDelay 200000) >> yield "ddd")
    putStrLn $ show count

有输出：

1
2

请注意，“时间桶”的内容仅yield在下一个桶的第一个元素到达时才被编辑。它们不是yield每 100 毫秒自动编辑一次。这对您来说可能是也可能不是问题。如果您希望yield每 100 毫秒自动执行一次，您将需要一个不同的解决方案，可能基于pipes-concurrency.

此外，您可以考虑直接使用由FreeT提供的基于 - 的“效果列表” pipes-group。这样您就可以在“时间桶”装满之前开始压缩“时间桶”中的数据。

于 2014-03-09T11:02:59.183 回答
0

因此，与丹尼尔的回答不同，我不会在生成数据时对其进行标记。它至少从上游获取元素，然后继续在 monoid 中聚合更多值，直到时间间隔过去。

此代码使用列表进行聚合，但有更好的幺半群来聚合

import Pipes
import qualified Pipes.Prelude as P

import Data.Time.Clock
import Data.Time.Calendar

import Data.Time.Format

import Data.Monoid

import Control.Monad

-- taken from pipes-rt
doubleToNomDiffTime :: Double -> NominalDiffTime
doubleToNomDiffTime x =
  let d0 = ModifiedJulianDay 0
      t0 = UTCTime d0 (picosecondsToDiffTime 0)
      t1 = UTCTime d0 (picosecondsToDiffTime $ floor (x/1e-12))
  in  diffUTCTime t1 t0

-- Adapted from from pipes-parse-1.0 
wrap
  :: Monad m =>
     Producer a m r -> Producer (Maybe a) m r
wrap p = do
  p >-> P.map Just
  forever $ yield Nothing
yieldAggregateOverTime
  :: (Monoid y,  -- monoid dependance so we can do aggregation
      MonadIO m  -- to beable to get the current time the
                 -- base monad must have access to IO
     ) =>
     (t -> y) -- Change element from upstream to monoid
  -> Double -- Time in seconds to aggregate over
  -> Pipe (Maybe t) y m ()
yieldAggregateOverTime wrap period = do
  t0 <- liftIO getCurrentTime
  loop mempty (dtUTC `addUTCTime` t0)
  where
    dtUTC = doubleToNomDiffTime period
    loop m ts = do
      t <- liftIO getCurrentTime
      v0 <- await -- await at least one element
      case v0 of
        Nothing -> yield m
        Just v -> do
          if t > ts
          then do
            yield (m <> wrap v)
            loop mempty (dtUTC `addUTCTime` ts)
          else do
            loop (m <> wrap v) ts


main = do
  runEffect $  wrap (each [1..]) >-> yieldAggregateOverTime (\x -> [x]) (0.0001)
                            >-> P.take 10 >-> P.print

根据您的 CPU 负载，输出数据的聚合方式会有所不同。每个块中至少有一个元素。

$ ghc Main.hs -O2
$ ./Main
[1,2]
[3]
[4]
[5]
[6]
[7]
[8]
[9]
[10]
[11]

$ ./Main
[1,2]
[3]
[4]
[5]
[6,7,8,9,10]
[11,12,13,14,15,16,17,18]
[19,20,21,22,23,24,25,26]
[27,28,29,30,31,32,33,34]
[35,36,37,38,39,40,41,42]
[43,44,45,46,47,48,49,50]

$ ./Main
[1,2,3,4,5,6]
[7]
[8]
[9,10,11,12,13,14,15,16,17,18,19,20]
[21,22,23,24,25,26,27,28,29,30,31,32,33]
[34,35,36,37,38,39,40,41,42,43,44]
[45,46,47,48,49,50,51,52,53,54,55]
[56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72]
[73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88]
[89,90,91,92,93,94,95,96,97,98,99,100,101,102,103]

$ ./Main
[1,2,3,4,5,6,7]
[8]
[9]
[10,11,12,13,14,15,16,17,18]
[19,20,21,22,23,24,25,26,27]
[28,29,30,31,32,33,34,35,36,37]
[38,39,40,41,42,43,44,45,46]
[47,48,49,50]
[51,52,53,54,55,56,57]
[58,59,60,61,62,63,64,65,66]

您可能想查看 pipe-rt的源代码，它显示了一种处理管道中时间的方法。

编辑：感谢 Daniel Díaz Carrete，采用了管道解析 1.0 技术来处理上游终止。管道组解决方案也应该可以使用相同的技术。

于 2014-03-09T11:29:03.533 回答