diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -1,5 +1,9 @@ # Changelog for Hedis +## 0.7.2 + +* Improve speed, rewrite internal logic (PR #56) + ## 0.7.1 * Add NFData instances diff --git a/hedis.cabal b/hedis.cabal --- a/hedis.cabal +++ b/hedis.cabal @@ -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 diff --git a/src/Database/Redis/Core.hs b/src/Database/Redis/Core.hs --- a/src/Database/Redis/Core.hs +++ b/src/Database/Redis/Core.hs @@ -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 diff --git a/src/Database/Redis/ProtocolPipelining.hs b/src/Database/Redis/ProtocolPipelining.hs --- a/src/Database/Redis/ProtocolPipelining.hs +++ b/src/Database/Redis/ProtocolPipelining.hs @@ -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 diff --git a/src/Database/Redis/PubSub.hs b/src/Database/Redis/PubSub.hs --- a/src/Database/Redis/PubSub.hs +++ b/src/Database/Redis/PubSub.hs @@ -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 diff --git a/src/Database/Redis/Types.hs b/src/Database/Redis/Types.hs --- a/src/Database/Redis/Types.hs +++ b/src/Database/Redis/Types.hs @@ -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 diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -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 ------------------------------------------------------------------------------