hedis 0.7.1 → 0.7.2
raw patch · 7 files changed
+179/−140 lines, 7 filesdep +slave-threaddep −BoundedChandep ~bytestringdep ~mtlPVP ok
version bump matches the API change (PVP)
Dependencies added: slave-thread
Dependencies removed: BoundedChan
Dependency ranges changed: bytestring, mtl
API changes (from Hackage documentation)
Files
- CHANGELOG +4/−0
- hedis.cabal +23/−10
- src/Database/Redis/Core.hs +30/−27
- src/Database/Redis/ProtocolPipelining.hs +84/−76
- src/Database/Redis/PubSub.hs +1/−1
- src/Database/Redis/Types.hs +11/−3
- test/Test.hs +26/−23
CHANGELOG view
@@ -1,5 +1,9 @@ # Changelog for Hedis +## 0.7.2++* Improve speed, rewrite internal logic (PR #56)+ ## 0.7.1 * Add NFData instances
hedis.cabal view
@@ -1,7 +1,7 @@ name: hedis-version: 0.7.1+version: 0.7.2 synopsis:- Client library for the Redis datastore: supports full command set, + Client library for the Redis datastore: supports full command set, pipelining. Description: Redis is an open source, advanced key-value store. It is often referred to@@ -50,13 +50,20 @@ type: git location: https://github.com/informatikr/hedis +flag manual+ description: enable this for local development -Werror and profiling options+ default: False+ manual: True+ library hs-source-dirs: src ghc-options: -Wall -fwarn-tabs- ghc-prof-options: -auto-all+ if flag(manual)+ ghc-options: -Werror+ if flag(manual)+ ghc-prof-options: -auto-all exposed-modules: Database.Redis- build-depends: BoundedChan >= 1.0,- attoparsec >= 0.12,+ build-depends: attoparsec >= 0.12, base >= 4.6 && < 5, bytestring >= 0.9, bytestring-lexing >= 0.5,@@ -81,25 +88,31 @@ main-is: benchmark/Benchmark.hs build-depends: base == 4.*,- mtl == 2.*,+ mtl >= 2.0, hedis, time >= 1.2 ghc-options: -O2 -Wall -rtsopts- ghc-prof-options: -auto-all+ if flag(manual)+ ghc-options: -Werror+ if flag(manual)+ ghc-prof-options: -auto-all test-suite hedis-test type: exitcode-stdio-1.0 main-is: test/Test.hs build-depends: base == 4.*,- bytestring >= 0.9 && < 0.11,+ bytestring >= 0.10, hedis, HUnit, mtl == 2.*,+ slave-thread, test-framework, test-framework-hunit, time -- We use -O0 here, since GHC takes *very* long to compile so many constants ghc-options: -O0 -Wall -rtsopts -fno-warn-unused-do-bind- ghc-prof-options: -auto-all- + if flag(manual)+ ghc-options: -Werror+ if flag(manual)+ ghc-prof-options: -auto-all
src/Database/Redis/Core.hs view
@@ -14,15 +14,12 @@ #if __GLASGOW_HASKELL__ < 710 import Control.Applicative #endif-import Control.Exception (evaluate)-import Control.Monad.State+import Control.Monad.Reader import qualified Data.ByteString as B+import Data.IORef import Data.Pool import Data.Time import Network-#if __GLASGOW_HASKELL__ < 710-import Data.Traversable (traverse)-#endif import Database.Redis.Protocol import qualified Database.Redis.ProtocolPipelining as PP@@ -38,9 +35,12 @@ -- -- In this context, each result is wrapped in an 'Either' to account for the -- possibility of Redis returning an 'Error' reply.-newtype Redis a = Redis (StateT (PP.Connection Reply, Maybe Reply) IO a)+newtype Redis a = Redis (ReaderT RedisEnv IO a) deriving (Monad, MonadIO, Functor, Applicative) +data RedisEnv = Env { envConn :: PP.Connection, envLastReply :: IORef Reply }++ -- |This class captures the following behaviour: In a context @m@, a command -- will return it's result wrapped in a \"container\" of type @f@. --@@ -65,31 +65,34 @@ -- while all connections from the pool are in use. runRedis :: Connection -> Redis a -> IO a runRedis (Conn pool) redis =- withResource pool $ \conn -> runRedisInternal conn redis+ withResource pool $ \conn -> runRedisInternal conn redis -- |Internal version of 'runRedis' that does not depend on the 'Connection'--- abstraction. Used to run the AUTH command when connecting. -runRedisInternal :: PP.Connection Reply -> Redis a -> IO a-runRedisInternal env (Redis redis) = do- (r, (_, lastReply)) <- runStateT redis (env, Nothing)- void $ traverse evaluate lastReply+-- abstraction. Used to run the AUTH command when connecting.+runRedisInternal :: PP.Connection -> Redis a -> IO a+runRedisInternal conn (Redis redis) = do+ -- Dummy reply in case no request is sent.+ ref <- newIORef (SingleLine "nobody will ever see this")+ r <- runReaderT redis (Env conn ref)+ -- Evaluate last reply to keep lazy IO inside runRedis.+ readIORef ref >>= (`seq` return ()) return r --getConn :: StateT (PP.Connection Reply, Maybe Reply) IO (PP.Connection Reply)-getConn = fst <$> get---putReply :: Reply -> StateT (PP.Connection Reply, Maybe Reply) IO ()-putReply r = modify $ \(c, _) -> (c, Just r)-+setLastReply :: Reply -> ReaderT RedisEnv IO ()+setLastReply r = do+ ref <- asks envLastReply+ lift (writeIORef ref r) recv :: (MonadRedis m) => m Reply-recv = liftRedis $ Redis getConn >>= liftIO . PP.recv+recv = liftRedis $ Redis $ do+ conn <- asks envConn+ r <- liftIO (PP.recv conn)+ setLastReply r+ return r send :: (MonadRedis m) => [B.ByteString] -> m () send req = liftRedis $ Redis $ do- conn <- getConn+ conn <- asks envConn liftIO $ PP.send conn (renderRequest req) -- |'sendRequest' can be used to implement commands from experimental@@ -106,9 +109,9 @@ => [B.ByteString] -> m (f a) sendRequest req = do r' <- liftRedis $ Redis $ do- conn <- getConn+ conn <- asks envConn r <- liftIO $ PP.request conn (renderRequest req)- putReply r+ setLastReply r return r returnDecode r' @@ -119,7 +122,7 @@ -- |A threadsafe pool of network connections to a Redis server. Use the -- 'connect' function to create one.-newtype Connection = Conn (Pool (PP.Connection Reply))+newtype Connection = Conn (Pool PP.Connection) -- |Information for connnecting to a Redis server. --@@ -127,7 +130,7 @@ -- Instead use 'defaultConnectInfo' and update it with record syntax. For -- example to connect to a password protected Redis server running on localhost -- and listening to the default port:--- +-- -- @ -- myConnectInfo :: ConnectInfo -- myConnectInfo = defaultConnectInfo {connectAuth = Just \"secret\"}@@ -180,7 +183,7 @@ createPool create destroy 1 connectMaxIdleTime connectMaxConnections where create = do- conn <- PP.connect connectHost connectPort reply+ conn <- PP.connect connectHost connectPort runRedisInternal conn $ do -- AUTH case connectAuth of
src/Database/Redis/ProtocolPipelining.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE RecordWildCards, DeriveDataTypeable #-}+{-# LANGUAGE RecordWildCards, DeriveDataTypeable, OverloadedStrings #-} -- |A module for automatic, optimal protocol pipelining. --@@ -13,27 +13,14 @@ -- as possible, i.e. as long as a request's response is not used before any -- subsequent requests. ----- We use a BoundedChan to make sure the evaluator thread can only start to--- evaluate a reply after the request is written to the output buffer.--- Otherwise we will flush the output buffer (in hGetReplies) before a command--- is written by the user thread, creating a deadlock.--------- # Notes------ [Eval thread synchronization]--- * BoundedChan performs better than Control.Concurrent.STM.TBQueue--- module Database.Redis.ProtocolPipelining (- Connection,- connect, disconnect, request, send, recv,- ConnectionLostException(..),- HostName, PortID(..)+ Connection,+ connect, disconnect, request, send, recv,+ ConnectionLostException(..),+ HostName, PortID(..) ) where import Prelude-import Control.Concurrent (ThreadId, forkIO, killThread)-import Control.Concurrent.BoundedChan import Control.Exception import Control.Monad import Data.Attoparsec.ByteString@@ -45,88 +32,109 @@ import System.IO.Error import System.IO.Unsafe +import Database.Redis.Protocol -data Connection a = Conn- { connHandle :: Handle -- ^ Connection socket-handle.- , connReplies :: IORef [a] -- ^ Reply thunks.- , connThunks :: BoundedChan a -- ^ See note [Eval thread synchronization].- , connEvalTId :: ThreadId -- ^ 'ThreadID' of the eval thread.- } +data Connection = Conn+ { connHandle :: Handle -- ^ Connection socket-handle.+ , connReplies :: IORef [Reply] -- ^ Reply thunks for unsent requests.+ , connPending :: IORef [Reply]+ -- ^ Reply thunks for requests "in the pipeline". Refers to the same list as+ -- 'connReplies', but can have an offset.+ , connPendingCnt :: IORef Int+ -- ^ Number of pending replies and thus the difference length between+ -- 'connReplies' and 'connPending'.+ -- length connPending - pendingCount = length connReplies+ }+ data ConnectionLostException = ConnectionLost- deriving (Show, Typeable)+ deriving (Show, Typeable) instance Exception ConnectionLostException -connect- :: HostName- -> PortID- -> Parser a- -> IO (Connection a)-connect host port parser = do- connHandle <- connectTo host port- hSetBinaryMode connHandle True- rs <- hGetReplies connHandle parser- connReplies <- newIORef rs- connThunks <- newBoundedChan 1000- connEvalTId <- forkIO $ forever $ readChan connThunks >>= evaluate- return Conn{..}+connect :: HostName -> PortID -> IO Connection+connect host port = do+ connHandle <- connectTo host port+ hSetBinaryMode connHandle True+ connReplies <- newIORef []+ connPending <- newIORef []+ connPendingCnt <- newIORef 0+ let conn = Conn{..}+ rs <- connGetReplies conn+ writeIORef connReplies rs+ writeIORef connPending rs+ return conn -disconnect :: Connection a -> IO ()+disconnect :: Connection -> IO () disconnect Conn{..} = do- open <- hIsOpen connHandle- when open (hClose connHandle)- killThread connEvalTId+ open <- hIsOpen connHandle+ when open (hClose connHandle) --- |Write the request to the socket output buffer.------ The 'Handle' is 'hFlush'ed when reading replies.-send :: Connection a -> S.ByteString -> IO ()-send Conn{..} = ioErrorToConnLost . S.hPut connHandle+-- |Write the request to the socket output buffer, without actually sending.+-- The 'Handle' is 'hFlush'ed when reading replies from the 'connHandle'.+send :: Connection -> S.ByteString -> IO ()+send Conn{..} s = do+ ioErrorToConnLost (S.hPut connHandle s)+ -- Signal that we expect one more reply from Redis.+ n <- atomicModifyIORef' connPendingCnt $ \n -> let n' = n+1 in (n', n')+ -- Limit the "pipeline length". This is necessary in long pipelines, to avoid+ -- thunk build-up, and thus space-leaks.+ -- TODO find smallest max pending with good-enough performance.+ when (n >= 1000) $ do+ -- Force oldest pending reply.+ r:_ <- readIORef connPending+ r `seq` return () --- |Take a reply from the list of future replies.------ The list of thunks must be deconstructed lazily, i.e. strictly matching (:)--- would block until a reply can be read. Using 'head' and 'tail' achieves ~2%--- more req/s in pipelined code than a lazy pattern match @~(r:rs)@.-recv :: Connection a -> IO a+-- |Take a reply-thunk from the list of future replies.+recv :: Connection -> IO Reply recv Conn{..} = do- rs <- readIORef connReplies- writeIORef connReplies (tail rs)- let r = head rs- writeChan connThunks r- return r+ (r:rs) <- readIORef connReplies+ writeIORef connReplies rs+ return r -request :: Connection a -> S.ByteString -> IO a+-- |Send a request and receive the corresponding reply+request :: Connection -> S.ByteString -> IO Reply request conn req = send conn req >> recv conn --- |Read all the replies from the Handle and return them as a lazy list.+-- |A list of all future 'Reply's of the 'Connection'. ----- The actual reading and parsing of each 'Reply' is deferred until the spine--- of the list is evaluated up to that 'Reply'. Each 'Reply' is cons'd in front--- of the (unevaluated) list of all remaining replies.+-- The spine of the list can be evaluated without forcing the replies. ----- 'unsafeInterleaveIO' only evaluates it's result once, making this function +-- Evaluating/forcing a 'Reply' from the list will 'unsafeInterleaveIO' the+-- reading and parsing from the 'connHandle'. To ensure correct ordering, each+-- Reply first evaluates (and thus reads from the network) the previous one.+--+-- 'unsafeInterleaveIO' only evaluates it's result once, making this function -- thread-safe. 'Handle' as implemented by GHC is also threadsafe, it is safe -- to call 'hFlush' here. The list constructor '(:)' must be called from -- /within/ unsafeInterleaveIO, to keep the replies in correct order.-hGetReplies :: Handle -> Parser a -> IO [a]-hGetReplies h parser = go S.empty+connGetReplies :: Connection -> IO [Reply]+connGetReplies Conn{..} = go S.empty (SingleLine "previous of first") where- go rest = unsafeInterleaveIO $ do - parseResult <- parseWith readMore parser rest+ go rest previous = do+ -- lazy pattern match to actually delay the receiving+ ~(r, rest') <- unsafeInterleaveIO $ do+ -- Force previous reply for correct order.+ previous `seq` return ()+ parseResult <- parseWith readMore reply rest case parseResult of- Fail{} -> errConnClosed- Partial{} -> error "Hedis: parseWith returned Partial"- Done rest' r -> do- rs <- go rest'- return (r:rs)+ Fail{} -> errConnClosed+ Partial{} -> error "Hedis: parseWith returned Partial"+ Done rest' r -> do+ -- r is the same as 'head' of 'connPending'. Since we just+ -- received r, we remove it from the pending list.+ atomicModifyIORef' connPending $ \(_:rs) -> (rs, ())+ -- We now expect one less reply from Redis. We don't count to+ -- negative, which would otherwise occur during pubsub.+ atomicModifyIORef' connPendingCnt $ \n -> (max 0 (n-1), ())+ return (r, rest')+ rs <- unsafeInterleaveIO (go rest' r)+ return (r:rs) readMore = ioErrorToConnLost $ do- hFlush h -- send any pending requests- S.hGetSome h maxRead+ hFlush connHandle -- send any pending requests+ S.hGetSome connHandle 4096 - maxRead = 4*1024 ioErrorToConnLost :: IO a -> IO a ioErrorToConnLost a = a `catchIOError` const errConnClosed
src/Database/Redis/PubSub.hs view
@@ -11,11 +11,11 @@ #if __GLASGOW_HASKELL__ < 710 import Control.Applicative+import Data.Monoid #endif import Control.Monad import Control.Monad.State import Data.ByteString.Char8 (ByteString)-import Data.Monoid import qualified Database.Redis.Core as Core import Database.Redis.Protocol (Reply(..)) import Database.Redis.Types
src/Database/Redis/Types.hs view
@@ -1,7 +1,11 @@ {-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE CPP, FlexibleInstances, OverlappingInstances, TypeSynonymInstances,+{-# LANGUAGE CPP, FlexibleInstances, TypeSynonymInstances, OverloadedStrings #-} +#if __GLASGOW_HASKELL__ < 710+{-# LANGUAGE OverlappingInstances #-}+#endif+ module Database.Redis.Types where #if __GLASGOW_HASKELL__ < 710@@ -45,7 +49,7 @@ instance NFData Status -data RedisType = None | String | Hash | List | Set | ZSet+data RedisType = None | String | Hash | List | Set | ZSet deriving (Show, Eq) instance RedisResult Reply where@@ -93,7 +97,11 @@ decode (MultiBulk Nothing) = Right Nothing decode r = Just <$> decode r -instance (RedisResult a) => RedisResult [a] where+instance+#if __GLASGOW_HASKELL__ >= 710+ {-# OVERLAPPABLE #-}+#endif+ (RedisResult a) => RedisResult [a] where decode (MultiBulk (Just rs)) = mapM decode rs decode r = Left r
test/Test.hs view
@@ -1,14 +1,16 @@-{-# LANGUAGE OverloadedStrings, RecordWildCards #-}+{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards #-} module Main (main) where -import Prelude hiding (catch)+#if __GLASGOW_HASKELL__ < 710 import Control.Applicative+import Data.Monoid (mappend)+#endif import Control.Concurrent import Control.Monad import Control.Monad.Trans-import Data.Monoid (mappend) import Data.Time import Data.Time.Clock.POSIX+import SlaveThread (fork) import qualified Test.Framework as Test (Test, defaultMain) import qualified Test.Framework.Providers.HUnit as Test (testCase) import qualified Test.HUnit as HUnit@@ -32,7 +34,7 @@ where withTimeLimit limit act = do start <- getCurrentTime- act+ _ <- act deltaT <-fmap (`diffUTCTime` start) getCurrentTime when (deltaT > limit) $ putStrLn $ name ++ ": " ++ show deltaT@@ -76,7 +78,7 @@ testForceErrorReply :: Test testForceErrorReply = testCase "force error reply" $ do- set "key" "value"+ Right _ <- set "key" "value" -- key is not a hash -> wrong kind of value reply <- hkeys "key" assert $ case reply of@@ -85,7 +87,7 @@ testPipelining :: Test testPipelining = testCase "pipelining" $ do- let n = 10+ let n = 100 tPipe <- deltaT $ do pongs <- replicateM n ping assert $ pongs == replicate n (Right Pong)@@ -96,7 +98,7 @@ where deltaT redis = do start <- liftIO $ getCurrentTime- redis+ _ <- redis liftIO $ fmap (`diffUTCTime` start) getCurrentTime testEvalReplies :: Test@@ -107,7 +109,7 @@ (liftIO $ do threadDelay $ 10^(5::Int) mvar <- newEmptyMVar- forkIO $ runRedis conn (get "key") >>= putMVar mvar+ _ <- fork $ runRedis conn (get "key") >>= putMVar mvar takeMVar mvar) >>=? Just "value" @@ -153,7 +155,8 @@ lpush "ids" ["1","2","3"] >>=? 3 sort "ids" defaultSortOpts >>=? ["1","2","3"] sortStore "ids" "anotherKey" defaultSortOpts >>=? 3- mset [("weight_1","1")+ Right _ <- mset+ [("weight_1","1") ,("weight_2","2") ,("weight_3","3") ,("object_1","foo")@@ -344,8 +347,8 @@ testZStore :: Test testZStore = testCase "zunionstore/zinterstore" $ do- zadd "k1" [(1, "v1"), (2, "v2")]- zadd "k2" [(2, "v2"), (3, "v3")]+ Right _ <- zadd "k1" [(1, "v1"), (2, "v2")]+ Right _ <- zadd "k2" [(2, "v2"), (3, "v3")] zinterstore "newkey" ["k1","k2"] Sum >>=? 1 zinterstoreWeights "newkey" [("k1",1),("k2",2)] Max >>=? 1 zunionstore "newkey" ["k1","k2"] Sum >>=? 3@@ -358,17 +361,17 @@ testHyperLogLog :: Test testHyperLogLog = testCase "hyperloglog" $ do -- test creation- pfadd "hll1" ["a"]+ Right _ <- pfadd "hll1" ["a"] pfcount ["hll1"] >>=? 1 -- test cardinality- pfadd "hll1" ["a"]+ Right _ <- pfadd "hll1" ["a"] pfcount ["hll1"] >>=? 1- pfadd "hll1" ["b", "c", "foo", "bar"]+ Right _ <- pfadd "hll1" ["b", "c", "foo", "bar"] pfcount ["hll1"] >>=? 5 -- test merge- pfadd "hll2" ["1", "2", "3"]- pfadd "hll3" ["4", "5", "6"]- pfmerge "hll4" ["hll2", "hll3"]+ Right _ <- pfadd "hll2" ["1", "2", "3"]+ Right _ <- pfadd "hll3" ["4", "5", "6"]+ Right _ <- pfmerge "hll4" ["hll2", "hll3"] pfcount ["hll4"] >>=? 6 -- test union cardinality pfcount ["hll2", "hll3"] >>=? 6@@ -381,7 +384,7 @@ where go = do -- producer- liftIO $ forkIO $ do+ _ <- liftIO $ fork $ do runRedis conn $ do let t = 10^(5 :: Int) liftIO $ threadDelay t@@ -409,8 +412,8 @@ testTransaction = testCase "transaction" $ do watch ["k1", "k2"] >>=? Ok unwatch >>=? Ok- set "foo" "foo"- set "bar" "bar"+ Right _ <- set "foo" "foo"+ Right _ <- set "bar" "bar" foobar <- multiExec $ do foo <- get "foo" bar <- get "bar"@@ -428,19 +431,19 @@ let script = "return {false, 42}" scriptRes = (False, 42 :: Integer) Right scriptHash <- scriptLoad script- eval script [] [] >>=? scriptRes+ eval script [] [] >>=? scriptRes evalsha scriptHash [] [] >>=? scriptRes scriptExists [scriptHash, "notAScript"] >>=? [True, False] scriptFlush >>=? Ok -- start long running script from another client configSet "lua-time-limit" "100" >>=? Ok liftIO $ do- forkIO $ runRedis conn $ do+ _ <- fork $ runRedis conn $ do -- we must pattern match to block the thread Left _ <- eval "while true do end" [] [] :: Redis (Either Reply Integer) return ()- threadDelay $ 10^(5 :: Int)+ threadDelay 500000 -- 0.5s scriptKill >>=? Ok ------------------------------------------------------------------------------