diff --git a/hedis.cabal b/hedis.cabal --- a/hedis.cabal +++ b/hedis.cabal @@ -1,5 +1,5 @@ name: hedis -version: 0.3.1 +version: 0.3.2 synopsis: Client library for the Redis datastore: supports full command set, pipelining. @@ -67,14 +67,14 @@ build-depends: attoparsec == 0.10.*, base == 4.*, bytestring == 0.9.*, - bytestring-lexing == 0.2.*, + bytestring-lexing == 0.3.*, mtl == 2.*, network == 2.*, resource-pool == 0.2.1.*, + stm == 2.2.*, time other-modules: Database.Redis.Core, - Database.Redis.Connection Database.Redis.PubSub, Database.Redis.Reply, Database.Redis.Request, diff --git a/src/Database/Redis.hs b/src/Database/Redis.hs --- a/src/Database/Redis.hs +++ b/src/Database/Redis.hs @@ -21,6 +21,21 @@ -- liftIO $ print (hello,world) -- @ + -- ** Automatic Pipelining + -- |Commands are automatically pipelined as much as possible. For example, + -- in the above \"hello world\" example, all four commands are pipelined. + -- Automatic pipelining makes use of Haskell's laziness. As long as a + -- previous reply is not evaluated, subsequent commands can be pipelined. + -- + -- Automatic pipelining also works across several calls to 'runRedis', as + -- long as replies are only evaluated /outside/ the 'runRedis' block. + -- + -- To keep memory usage low, the number of requests \"in the pipeline\" is + -- limited (per connection) to 1000. After that number, the next command is + -- sent only when at least one reply has been received. That means, command + -- functions may block until there are less than 1000 outstanding replies. + -- + -- ** Error Behavior -- | -- [Operations against keys holding the wrong kind of value:] If the Redis @@ -65,7 +80,6 @@ ) where import Database.Redis.Core -import Database.Redis.Connection import Database.Redis.PubSub import Database.Redis.Reply import Database.Redis.Types diff --git a/src/Database/Redis/Commands.hs b/src/Database/Redis/Commands.hs --- a/src/Database/Redis/Commands.hs +++ b/src/Database/Redis/Commands.hs @@ -588,11 +588,6 @@ -> Redis (Either Reply Bool) hexists key field = sendRequest (["HEXISTS"] ++ [encode key] ++ [encode field] ) -auth - :: ByteString -- ^ password - -> Redis (Either Reply Status) -auth password = sendRequest (["AUTH"] ++ [encode password] ) - sinterstore :: ByteString -- ^ destination -> [ByteString] -- ^ key diff --git a/src/Database/Redis/Connection.hs b/src/Database/Redis/Connection.hs deleted file mode 100644 --- a/src/Database/Redis/Connection.hs +++ /dev/null @@ -1,137 +0,0 @@ -{-# LANGUAGE RecordWildCards, DeriveDataTypeable #-} - -module Database.Redis.Connection ( - HostName,PortID(PortNumber,UnixSocket), - ConnectInfo(..),defaultConnectInfo, - Connection(), connect, - ConnectionLostException(..) -) where - -import Prelude hiding (catch) -import Control.Applicative -import Control.Monad.Reader -import Control.Concurrent -import Control.Exception (Exception, catch, throwIO) -import qualified Data.Attoparsec as P -import qualified Data.ByteString as B -import Data.IORef -import Data.Pool -import Data.Time -import Data.Typeable -import Network (HostName, PortID(..), connectTo) -import System.IO (Handle, hClose, hIsOpen, hSetBinaryMode, hFlush) -import System.IO.Unsafe (unsafeInterleaveIO) - -import Database.Redis.Core -import Database.Redis.Commands (auth) -import Database.Redis.Reply - -data ConnectionLostException = ConnectionLost - deriving (Show, Typeable) - -instance Exception ConnectionLostException - --- |Information for connnecting to a Redis server. --- --- It is recommended to not use the 'ConnInfo' data constructor directly. --- 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\"} --- @ --- -data ConnectInfo = ConnInfo - { connectHost :: HostName - , connectPort :: PortID - , connectAuth :: Maybe B.ByteString - -- ^ When the server is protected by a password, set 'connectAuth' to 'Just' - -- the password. Each connection will then authenticate by the 'auth' - -- command. - , connectMaxConnections :: Int - -- ^ Maximum number of connections to keep open. The smallest acceptable - -- value is 1. - , connectMaxIdleTime :: NominalDiffTime - -- ^ Amount of time for which an unused connection is kept open. The - -- smallest acceptable value is 0.5 seconds. - } - --- |Default information for connecting: --- --- @ --- connectHost = \"localhost\" --- connectPort = PortNumber 6379 -- Redis default port --- connectAuth = Nothing -- No password --- connectMaxConnections = 50 -- Up to 50 connections --- connectMaxIdleTime = 30 -- Keep open for 30 seconds --- @ --- -defaultConnectInfo :: ConnectInfo -defaultConnectInfo = ConnInfo - { connectHost = "localhost" - , connectPort = PortNumber 6379 - , connectAuth = Nothing - , connectMaxConnections = 50 - , connectMaxIdleTime = 30 - } - --- |Opens a 'Connection' to a Redis server designated by the given --- 'ConnectInfo'. -connect :: ConnectInfo -> IO Connection -connect ConnInfo{..} = Conn <$> - createPool create destroy 1 connectMaxIdleTime connectMaxConnections - where - create = do - h <- connectTo connectHost connectPort - rs <- hGetReplies h >>= newIORef - hSetBinaryMode h True - let conn = (h,rs) - maybe (return ()) - (\pass -> runRedisInternal conn (auth pass) >> return ()) - connectAuth - newMVar conn - - destroy conn = withMVar conn $ \(h,_) -> do - open <- hIsOpen h - when open (hClose h) - --- |Read all the 'Reply's from the Handle and return them as a lazy list. --- --- 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. --- --- '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 -> IO [Reply] -hGetReplies h = lazyRead (Right B.empty) - where - lazyRead rest = unsafeInterleaveIO $ do - parseResult <- either continueParse readAndParse rest - case parseResult of - P.Fail _ _ _ -> error "Hedis: reply parse failed" - P.Partial cont -> lazyRead (Left cont) - P.Done rest' r -> do - rs <- lazyRead (Right rest') - return (r:rs) - - continueParse cont = cont <$> B.hGetSome h maxRead - - readAndParse rest = P.parse reply <$> - if B.null rest - then do - hFlush h -- send any pending requests - s <- B.hGetSome h maxRead `catchIOError` const errConnClosed - when (B.null s) errConnClosed - return s - else return rest - - maxRead = 4*1024 - errConnClosed = throwIO ConnectionLost - - catchIOError :: IO a -> (IOError -> IO a) -> IO a - catchIOError = catch 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 @@ -1,41 +1,44 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings, GeneralizedNewtypeDeriving, RecordWildCards, + DeriveDataTypeable #-} + module Database.Redis.Core ( - Connection(..), - Redis(),runRedis,runRedisInternal, - send, - recv, - sendRequest + Connection(..), connect, + ConnectInfo(..), defaultConnectInfo, + Redis(),runRedis, + send, recv, sendRequest, + HostName, PortID(..), + ConnectionLostException(..), + auth ) where +import Prelude hiding (catch) import Control.Applicative -import Control.Arrow import Control.Monad.Reader import Control.Concurrent +import Control.Concurrent.STM +import Control.Exception +import qualified Data.Attoparsec as P import qualified Data.ByteString as B -import Data.IORef import Data.Pool -import System.IO (Handle) +import Data.Time +import Data.Typeable +import Network +import System.IO +import System.IO.Unsafe import Database.Redis.Reply import Database.Redis.Request import Database.Redis.Types --- |A threadsafe pool of network connections to a Redis server. Use the --- 'connect' function to create one. -newtype Connection = Conn (Pool (MVar (Handle, IORef [Reply]))) +-------------------------------------------------------------------------------- +-- The Redis Monad +-- + -- |All Redis commands run in the 'Redis' monad. newtype Redis a = Redis (ReaderT RedisEnv IO a) deriving (Monad, MonadIO, Functor, Applicative) -type RedisEnv = (Handle, IORef [Reply]) - -askHandle :: ReaderT RedisEnv IO Handle -askHandle = asks fst - -askReplies :: ReaderT RedisEnv IO (IORef [Reply]) -askReplies = asks snd - -- |Interact with a Redis datastore specified by the given 'Connection'. -- -- Each call of 'runRedis' takes a network connection from the 'Connection' @@ -51,18 +54,179 @@ runRedisInternal :: RedisEnv -> Redis a -> IO a runRedisInternal env (Redis redis) = runReaderT redis env + +-------------------------------------------------------------------------------- +-- Redis Environment. +-- + +-- |The per-connection environment the 'Redis' monad can read from. +-- +-- Create with 'newEnv'. Modified by 'recv' and 'send'. +data RedisEnv = Env + { envHandle :: Handle -- ^ Connection socket-handle. + , envReplies :: TVar [Reply] -- ^ Reply thunks. + , envThunkCnt :: TVar Integer -- ^ Number of thunks in 'envThunkChan'. + , envEvalTId :: ThreadId -- ^ 'ThreadID' of the evaluator thread. + } + +-- |Create a new 'RedisEnv' +newEnv :: Handle -> IO RedisEnv +newEnv envHandle = do + replies <- lazify <$> hGetReplies envHandle + envReplies <- newTVarIO replies + envThunkCnt <- newTVarIO 0 + envEvalTId <- forkIO $ forceThunks envThunkCnt replies + return Env{..} + where + lazify rs = head rs : lazify (tail rs) + +forceThunks :: TVar Integer -> [Reply] -> IO () +forceThunks thunkCnt = go + where + go [] = return () + go (r:rs) = do + -- wait for a thunk + atomically $ do + cnt <- readTVar thunkCnt + guard (cnt > 0) + writeTVar thunkCnt (cnt-1) + r `seq` go rs + +recv :: Redis Reply +recv = Redis $ do + Env{..} <- ask + liftIO $ atomically $ do + -- limit the amount of reply-thunks per connection. + cnt <- readTVar envThunkCnt + guard $ cnt < 1000 + writeTVar envThunkCnt (cnt+1) + r:rs <- readTVar envReplies + writeTVar envReplies rs + return r + send :: [B.ByteString] -> Redis () send req = Redis $ do - h <- askHandle + h <- asks envHandle -- hFlushing the handle is done while reading replies. liftIO $ {-# SCC "send.hPut" #-} B.hPut h (renderRequest req) -recv :: Redis Reply -recv = Redis $ do - -- head/tail avoids forcing the ':' constructor, enabling automatic - -- pipelining. - rs <- askReplies - liftIO $ atomicModifyIORef rs (tail &&& head) - sendRequest :: (RedisResult a) => [B.ByteString] -> Redis (Either Reply a) sendRequest req = decode <$> (send req >> recv) + + +-------------------------------------------------------------------------------- +-- Connection +-- + +-- |A threadsafe pool of network connections to a Redis server. Use the +-- 'connect' function to create one. +newtype Connection = Conn (Pool (MVar RedisEnv)) + +data ConnectionLostException = ConnectionLost + deriving (Show, Typeable) + +instance Exception ConnectionLostException + +-- |Information for connnecting to a Redis server. +-- +-- It is recommended to not use the 'ConnInfo' data constructor directly. +-- 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\"} +-- @ +-- +data ConnectInfo = ConnInfo + { connectHost :: HostName + , connectPort :: PortID + , connectAuth :: Maybe B.ByteString + -- ^ When the server is protected by a password, set 'connectAuth' to 'Just' + -- the password. Each connection will then authenticate by the 'auth' + -- command. + , connectMaxConnections :: Int + -- ^ Maximum number of connections to keep open. The smallest acceptable + -- value is 1. + , connectMaxIdleTime :: NominalDiffTime + -- ^ Amount of time for which an unused connection is kept open. The + -- smallest acceptable value is 0.5 seconds. + } + +-- |Default information for connecting: +-- +-- @ +-- connectHost = \"localhost\" +-- connectPort = PortNumber 6379 -- Redis default port +-- connectAuth = Nothing -- No password +-- connectMaxConnections = 50 -- Up to 50 connections +-- connectMaxIdleTime = 30 -- Keep open for 30 seconds +-- @ +-- +defaultConnectInfo :: ConnectInfo +defaultConnectInfo = ConnInfo + { connectHost = "localhost" + , connectPort = PortNumber 6379 + , connectAuth = Nothing + , connectMaxConnections = 50 + , connectMaxIdleTime = 30 + } + +-- |Opens a 'Connection' to a Redis server designated by the given +-- 'ConnectInfo'. +connect :: ConnectInfo -> IO Connection +connect ConnInfo{..} = Conn <$> + createPool create destroy 1 connectMaxIdleTime connectMaxConnections + where + create = do + h <- connectTo connectHost connectPort + hSetBinaryMode h True + conn <- newEnv h + maybe (return ()) + (\pass -> runRedisInternal conn (auth pass) >> return ()) + connectAuth + newMVar conn + + destroy conn = withMVar conn $ \Env{..} -> do + open <- hIsOpen envHandle + when open (hClose envHandle) + killThread envEvalTId + +-- |Read all the 'Reply's from the Handle and return them as a lazy list. +-- +-- 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. +-- +-- '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 -> IO [Reply] +hGetReplies h = go B.empty + where + go rest = unsafeInterleaveIO $ do + parseResult <- P.parseWith readMore reply rest + case parseResult of + P.Fail _ _ _ -> errConnClosed + P.Partial _ -> error "Hedis: parseWith returned Partial" + P.Done rest' r -> do + rs <- go rest' + return (r:rs) + + readMore = do + hFlush h -- send any pending requests + B.hGetSome h maxRead `catchIOError` const errConnClosed + + maxRead = 4*1024 + errConnClosed = throwIO ConnectionLost + + catchIOError :: IO a -> (IOError -> IO a) -> IO a + catchIOError = catch + +-- The AUTH command. It has to be here because it is used in 'connect'. +auth + :: B.ByteString -- ^ password + -> Redis (Either Reply Status) +auth password = sendRequest ["AUTH", password] diff --git a/src/Database/Redis/Reply.hs b/src/Database/Redis/Reply.hs --- a/src/Database/Redis/Reply.hs +++ b/src/Database/Redis/Reply.hs @@ -3,9 +3,9 @@ import Prelude hiding (error, take) import Control.Applicative -import Data.Attoparsec.Char8 -import qualified Data.Attoparsec as P -import Data.ByteString.Char8 +import Data.Attoparsec (takeTill) +import Data.Attoparsec.Char8 hiding (takeTill) +import Data.ByteString.Char8 (ByteString) -- |Low-level representation of replies from the Redis server. data Reply = SingleLine ByteString @@ -22,37 +22,24 @@ reply = choice [singleLine, integer, bulk, multiBulk, error] singleLine :: Parser Reply -singleLine = SingleLine <$> '+' `prefixing` line +singleLine = SingleLine <$> (char '+' *> takeTill isEndOfLine <* endOfLine) error :: Parser Reply -error = Error <$> '-' `prefixing` line +error = Error <$> (char '-' *> takeTill isEndOfLine <* endOfLine) integer :: Parser Reply -integer = Integer <$> ':' `prefixing` signed decimal +integer = Integer <$> (char ':' *> signed decimal <* endOfLine) bulk :: Parser Reply -bulk = Bulk <$> do - len <- '$' `prefixing` signed decimal +bulk = Bulk <$> do + len <- char '$' *> signed decimal <* endOfLine if len < 0 then return Nothing - else Just <$> P.take len <* crlf + else Just <$> take len <* endOfLine multiBulk :: Parser Reply multiBulk = MultiBulk <$> do - len <- '*' `prefixing` signed decimal + len <- char '*' *> signed decimal <* endOfLine if len < 0 then return Nothing - else Just <$> P.count len reply - - ------------------------------------------------------------------------------- --- Helpers & Combinators --- -prefixing :: Char -> Parser a -> Parser a -c `prefixing` a = char c *> a <* crlf - -crlf :: Parser ByteString -crlf = string "\r\n" - -line :: Parser ByteString -line = takeTill (=='\r') + else Just <$> count len reply diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -1,6 +1,7 @@ {-# LANGUAGE OverloadedStrings, RecordWildCards #-} module Main (main) where +import Prelude hiding (catch) import Control.Applicative import Control.Concurrent import Control.Monad @@ -52,8 +53,16 @@ -- Miscellaneous -- testsMisc :: [Test] -testsMisc = [testForceErrorReply, testPipelining] +testsMisc = [testConstantSpacePipelining, testForceErrorReply, testPipelining] +testConstantSpacePipelining :: Test +testConstantSpacePipelining = testCase "constant-space pipelining" $ do + -- This testcase should not exceed the maximum heap size, as set in + -- the run-test.sh script. + replicateM_ 10000 ping + -- If the program didn't crash, pipelining takes constant memory. + assert True + testForceErrorReply :: Test testForceErrorReply = testCase "force error reply" $ do set "key" "value" @@ -73,7 +82,7 @@ tNoPipe <- time $ replicateM_ n (ping >>=? Pong) -- pipelining should at least be twice as fast. assert $ tNoPipe / tPipe > 2 - + time :: Redis () -> Redis NominalDiffTime time redis = do start <- liftIO $ getCurrentTime @@ -512,7 +521,10 @@ -- producer liftIO $ forkIO $ do runRedis conn $ do + let t = 10^(5 :: Int) + liftIO $ threadDelay t publish "chan1" "hello" >>=? 1 + liftIO $ threadDelay t publish "chan2" "world" >>=? 1 return ()