hedis 0.3.1 → 0.3.2

raw patch · 7 files changed

+235/−200 lines, 7 filesdep +stmdep ~bytestring-lexingPVP ok

version bump matches the API change (PVP)

Dependencies added: stm

Dependency ranges changed: bytestring-lexing

API changes (from Hackage documentation)

Files

• hedis.cabal +3/−3
• src/Database/Redis.hs +15/−1
• src/Database/Redis/Commands.hs +0/−5
• src/Database/Redis/Connection.hs +0/−137
• src/Database/Redis/Core.hs +192/−28
• src/Database/Redis/Reply.hs +11/−24
• test/Test.hs +14/−2

@@ -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,

@@ -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

@@ -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

@@ -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

@@ -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]

@@ -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

@@ -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 ()