conduit-throttle (empty) → 0.2.0.0
raw patch · 6 files changed
+338/−0 lines, 6 filesdep +HUnitdep +asyncdep +basesetup-changed
Dependencies added: HUnit, async, base, conduit, conduit-combinators, conduit-extra, conduit-throttle, resourcet, stm, stm-chans, stm-conduit, test-framework, test-framework-hunit, throttle-io-stream, unliftio, unliftio-core
Files
- LICENSE +30/−0
- README.md +59/−0
- Setup.hs +2/−0
- conduit-throttle.cabal +77/−0
- src/Data/Conduit/Throttle.hs +128/−0
- tests/Data/Conduit/Throttle/Test.hs +42/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Moritz Schulte (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Moritz Schulte nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,59 @@+# throttle-conduit [](https://hackage.haskell.org/package/throttle-conduit) [](https://www.stackage.org/package/throttle-conduit) [](https://travis-ci.org/mtesseract/throttle-conduit)++### About++This is `conduit-throttle`, a compact Haskell package providing+configurable throttling support for Conduit. It is built on top of the+packages `throttle-io-stream` and `unliftio`.++The API (`Data.Conduit.Throttle`) is designed for being imported+qualified, which is why the exported functions and values are not+prefixed by the package name or something similar.++The core function exported by this package is the following:++```haskell+throttleProducer :: (MonadUnliftIO m, MonadResource m)+ => Conf a+ -> Producer m a+ -> Producer m a+```++That is, given a `ThrottleConf` and a Conduit `Producer`, create and+return a new Conduit `Producer`, which yields the same stream of+values like the provided one but throttled according to the provided+throttling configuration.++A throttling configuration is created using `newConf` and then+configured using the functions `setMeasure`, `setMaxThroughput`,+`setBufferSize` and others.++Primarily, the throughput is defined by the provided measure function+and the configured maximum throughput. A measure function is a+function which defines how "big" a Conduit element `a` is. In other+words, it is a function of type `a -> Double`. For instance, it could+simply calculate the length of some deserialized value in bytes or it+could simply be the constant function `const 1` in case it is desired+to throttle the number of elements a Conduit producer produces. The+maximum throughput together with the measure function then defines the+maximum throughput to produce (of course it can be less, if the+original producer provides data with a smaller throughput). The+throughput is considered with respect to a base time interval, which+can be changed using `setInterval`, by default it is `1000` (ms).++The underlying package `throttle-io-stream` uses exponentially+weighted moving averages for smoothing the actual throughput changes.++### Example++```haskell+import qualified Data.Conduit.Throttle as ConduitThrottle++simpleProducerThrottling :: IO ()+simpleProducerThrottling = do+ let conf = ConduitThrottle.newConf+ & ConduitThrottle.setInterval 1000+ & ConduitThrottle.setMaxThroughput 1+ runResourceT . runConduit $+ ConduitThrottle.throttleProducer conf (sourceList [1..5]) .| printC+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ conduit-throttle.cabal view
@@ -0,0 +1,77 @@+-- This file has been generated from package.yaml by hpack version 0.17.0.+--+-- see: https://github.com/sol/hpack++name: conduit-throttle+version: 0.2.0.0+synopsis: Throttle Conduit Producers+description: This packages is based on the throttle-io-stream package and provides functionality for throttling Conduit producers according to a provided configuration.+homepage: https://github.com/mtesseract/conduit-throttle#readme+bug-reports: https://github.com/mtesseract/conduit-throttle/issues+license: BSD3+license-file: LICENSE+author: Moritz Schulte+maintainer: mtesseract@silverratio.net+copyright: (c) 2017 Moritz Schulte+category: Data+build-type: Simple+cabal-version: >= 1.10++extra-source-files:+ README.md++source-repository head+ type: git+ location: https://github.com/mtesseract/conduit-throttle++library+ hs-source-dirs:+ src+ ghc-options: -Wall+ exposed-modules:+ Data.Conduit.Throttle+ other-modules:+ Paths_conduit_throttle+ build-depends:+ base >=4.7 && <5+ , conduit+ , conduit-combinators+ , conduit-extra+ , stm-chans+ , stm+ , resourcet+ , async+ , unliftio+ , unliftio-core+ , throttle-io-stream+ , resourcet+ default-language: Haskell2010++test-suite conduit-test+ type: exitcode-stdio-1.0+ main-is: Test.hs+ hs-source-dirs:+ tests/Data/Conduit/Throttle+ default-extensions: OverloadedStrings+ ghc-options: -Wall -Wall -fno-warn-type-defaults+ build-depends:+ base >=4.7 && <5+ , conduit+ , conduit-combinators+ , conduit-extra+ , stm-chans+ , stm+ , resourcet+ , async+ , unliftio+ , unliftio-core+ , throttle-io-stream+ , resourcet+ , base >=4.7 && <5+ , HUnit+ , test-framework+ , test-framework-hunit+ , conduit-throttle+ , throttle-io-stream+ , stm-conduit+ default-language: Haskell2010
+ src/Data/Conduit/Throttle.hs view
@@ -0,0 +1,128 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}++module Data.Conduit.Throttle+ ( Conf+ , newConf+ , setMeasure+ , setInterval+ , setMaxThroughput+ , setBufferSize+ , setEmaAlpha+ , throttleProducer+ ) where++import Conduit+import Control.Concurrent.STM+import Control.Concurrent.STM.TBMQueue+import qualified Control.Concurrent.Throttle as Throttle+import Control.Monad.Trans.Resource+import Data.Function+import UnliftIO++data Conf a = Conf+ { _measure :: Throttle.Measure a+ , _interval :: Double+ , _maxThroughput :: Double+ , _bufferSize :: Int+ , _emaAlpha :: Double+ }++newConf :: Conf a+newConf = defaultConf++-- | Default 'ThrottleConf'.+defaultConf :: Conf a+defaultConf = Conf+ { _measure = const 1+ , _interval = 1000+ , _maxThroughput = 100+ , _bufferSize = 1024+ , _emaAlpha = defaultEmaAlpha }++-- | Set measure function in configuration.+setMeasure :: Throttle.Measure a+ -> Conf a+ -> Conf a+setMeasure measure conf = conf { _measure = measure }++-- | Set interval in configuration.+setInterval :: Double+ -> Conf a+ -> Conf a+setInterval interval conf = conf { _interval = interval }++-- | Set max throughput in configuration.+setMaxThroughput :: Double+ -> Conf a+ -> Conf a+setMaxThroughput throughput conf =+ conf { _maxThroughput = throughput }++-- | Set buffer size in configuration.+setBufferSize :: Int+ -> Conf a+ -> Conf a+setBufferSize n conf = conf { _bufferSize = n }++-- | Set exponential weight factor used for computing current item+-- size.+setEmaAlpha :: Double+ -> Conf a+ -> Conf a+setEmaAlpha alpha conf = conf { _emaAlpha = alpha }++-- | Default exponential weight factor for computing current item+-- size.+defaultEmaAlpha :: Double+defaultEmaAlpha = 0.5++throttleConfPrepare :: Conf a -> Throttle.ThrottleConf a+throttleConfPrepare Conf { .. } = Throttle.newThrottleConf+ & Throttle.throttleConfThrottleProducer+ & Throttle.throttleConfSetMeasure _measure+ & Throttle.throttleConfSetInterval _interval+ & Throttle.throttleConfSetMaxThroughput _maxThroughput+ & Throttle.throttleConfSetBufferSize _bufferSize+ & Throttle.throttleConfSetEmaAlpha _emaAlpha++-- | Given a 'ThrottleConf' and a 'Producer', create and return a new+-- 'Producer', which yields the same stream of values like the+-- provided producer but throttled according to the provided+-- throttling configuration.+throttleProducer :: (MonadUnliftIO m, MonadResource m)+ => Conf a+ -> Producer m a+ -> Producer m a+throttleProducer conf producer = do+ (UnliftIO unlifter) <- lift askUnliftIO+ queueIn <- liftIO $ newTBMQueueIO 1024+ queueOut <- liftIO $ newTBMQueueIO 1+ (_, _) <- allocate+ (UnliftIO.async (unlifter (runConduit (producer .| drainConduit queueIn))))+ UnliftIO.cancel++ let throttleConf = throttleConfPrepare conf+ readCallback = atomically (readTBMQueue queueIn)+ writeCallback = \case+ Just a -> atomically $ writeTBMQueue queueOut a+ Nothing -> atomically $ closeTBMQueue queueOut++ (_, asyncThrottler) <- allocate+ (Throttle.throttle throttleConf readCallback writeCallback)+ UnliftIO.cancel+ link asyncThrottler+ go queueOut++ where go queue = do+ liftIO (atomically (readTBMQueue queue)) >>= \case+ Just a -> yield a >> go queue+ Nothing -> return ()++ drainConduit queue = do+ await >>= \case+ Just a -> do liftIO (atomically (writeTBMQueue queue a))+ drainConduit queue+ Nothing -> liftIO (atomically (closeTBMQueue queue))
+ tests/Data/Conduit/Throttle/Test.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Conduit+import Data.Conduit.Async+import Data.Conduit.List (sourceList)+import qualified Data.Conduit.Throttle as ConduitThrottle+import Data.Function ((&))+import Test.Framework (Test, defaultMain, testGroup)+import Test.Framework.Providers.HUnit (testCase)++main :: IO ()+main = do+ putStrLn ""+ defaultMain tests++tests :: [Test.Framework.Test]+tests =+ [ testGroup "Test Suite"+ [ testCase "Simple producer throttling"+ simpleProducerThrottling+ , testCase "Simple producer throttling (using stm-conduit)"+ simpleProducerThrottlingStm+ ]+ ]++simpleProducerThrottling :: IO ()+simpleProducerThrottling = do+ let conf = ConduitThrottle.newConf+ & ConduitThrottle.setInterval 1000+ & ConduitThrottle.setMaxThroughput 1+ runResourceT . runConduit $+ ConduitThrottle.throttleProducer conf (sourceList [1..5]) .| printC++simpleProducerThrottlingStm :: IO ()+simpleProducerThrottlingStm = do+ let conf = ConduitThrottle.newConf+ & ConduitThrottle.setInterval 1000+ & ConduitThrottle.setMaxThroughput 1+ runResourceT . runCConduit $+ ConduitThrottle.throttleProducer conf (sourceList [1..5]) =$=& printC