diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +0.5.0 +----------------------------------------------------------------------------- +- `transactional` has been added to execute redis transactions. +- Bugfixes + 0.4.1 ----------------------------------------------------------------------------- - Support `monad-control` 1.* diff --git a/redis-io.cabal b/redis-io.cabal --- a/redis-io.cabal +++ b/redis-io.cabal @@ -1,11 +1,11 @@ name: redis-io -version: 0.4.1 +version: 0.5.0 synopsis: Yet another redis client. -license: OtherLicense +license: MPL-2.0 license-file: LICENSE author: Toralf Wittner maintainer: Toralf Wittner -copyright: (c) 2014-2015 Toralf Wittner +copyright: (C) 2014-2015 Toralf Wittner homepage: https://github.com/twittner/redis-io/ bug-reports: https://github.com/twittner/redis-io/issues stability: experimental @@ -25,7 +25,6 @@ default-language: Haskell2010 hs-source-dirs: src ghc-options: -Wall -O2 -fwarn-tabs -funbox-strict-fields - ghc-prof-options: -prof -auto-all exposed-modules: Database.Redis.IO @@ -44,12 +43,15 @@ , bytestring >= 0.9 && < 1.0 , containers >= 0.5 && < 1.0 , exceptions >= 0.6 && < 1.0 + , iproute >= 1.3 && < 2.0 , monad-control >= 0.3 && < 2.0 , mtl >= 2.1 && < 3.0 , network >= 2.5 && < 3.0 , operational == 0.2.* , redis-resp >= 0.2 && < 0.4 , resource-pool >= 0.2 && < 0.3 + , semigroups >= 0.16 && < 0.20 + , stm >= 2.4 && < 3.0 , time >= 1.4 && < 2.0 , transformers >= 0.3 && < 0.5 , transformers-base >= 0.4 && < 1.0 diff --git a/src/Database/Redis/IO.hs b/src/Database/Redis/IO.hs --- a/src/Database/Redis/IO.hs +++ b/src/Database/Redis/IO.hs @@ -9,6 +9,7 @@ , runRedis , stepwise , pipelined + , transactional , pubSub -- * Connection pool @@ -28,9 +29,10 @@ , setSendRecvTimeout -- * Exceptions - , ConnectionError (..) - , InternalError (..) - , Timeout (..) + , ConnectionError (..) + , InternalError (..) + , Timeout (..) + , TransactionFailure (..) -- * Re-exports , module Data.Redis.Command diff --git a/src/Database/Redis/IO/Client.hs b/src/Database/Redis/IO/Client.hs --- a/src/Database/Redis/IO/Client.hs +++ b/src/Database/Redis/IO/Client.hs @@ -40,6 +40,7 @@ import qualified Control.Monad.State.Strict as S import qualified Control.Monad.State.Lazy as L +import qualified Data.List.NonEmpty as NE import qualified Data.Pool as P import qualified Database.Redis.IO.Connection as C import qualified System.Logger as Logger @@ -121,7 +122,7 @@ <*> pure g <*> pure t where - connOpen t a = do + connOpen t aa = tryAll (NE.fromList aa) $ \a -> do c <- C.connect s g t a Logger.debug g $ "client.connect" .= sHost s ~~ msg (show c) return c @@ -153,6 +154,11 @@ pipelined :: MonadClient m => Redis IO a -> m a pipelined a = liftClient $ withConnection (flip (eval getLazy) a) + +-- | Execute the given redis commands in a Redis transaction. +transactional :: MonadClient m => Redis IO a -> m a +transactional a = liftClient $ withConnection (flip (eval getTransaction) a) + -- | Execute the given publish\/subscribe commands. The first parameter is -- the callback function which will be invoked with channel and message -- once messages arrive. @@ -368,6 +374,18 @@ return (a, a `seq` return ()) {-# INLINE getLazy #-} +-- | Just like 'getLazy', but executes the 'Connection' buffer in a Redis +-- transaction. +getTransaction :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ()) +getTransaction h x g = do + r <- newIORef (throw $ RedisError "missing response") + C.request x r h + a <- unsafeInterleaveIO $ do + C.transaction h + either throwIO return =<< g <$> readIORef r + return (a, a `seq` return ()) +{-# INLINE getTransaction #-} + getNow :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ()) getNow h x g = do r <- newIORef (throw $ RedisError "missing response") @@ -392,3 +410,8 @@ withTimeout 0 c = c { C.settings = setSendRecvTimeout 0 (C.settings c) } withTimeout t c = c { C.settings = setSendRecvTimeout (10 + fromIntegral t) (C.settings c) } {-# INLINE withTimeout #-} + +tryAll :: NonEmpty a -> (a -> IO b) -> IO b +tryAll (a :| []) f = f a +tryAll (a :| aa) f = f a `catchAll` (const $ tryAll (NE.fromList aa) f) +{-# INLINE tryAll #-} diff --git a/src/Database/Redis/IO/Connection.hs b/src/Database/Redis/IO/Connection.hs --- a/src/Database/Redis/IO/Connection.hs +++ b/src/Database/Redis/IO/Connection.hs @@ -2,6 +2,8 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections #-} @@ -13,6 +15,7 @@ , close , request , sync + , transaction , send , receive ) where @@ -23,12 +26,14 @@ import Data.Attoparsec.ByteString hiding (Result) import Data.ByteString (ByteString) import Data.ByteString.Lazy (toChunks) -import Data.Foldable hiding (concatMap) +import Data.Foldable (for_, foldlM, toList) import Data.IORef import Data.Maybe (isJust) import Data.Redis import Data.Sequence (Seq, (|>)) +import Data.Int import Data.Word +import Foreign.C.Types (CInt (..)) import Database.Redis.IO.Settings import Database.Redis.IO.Types import Database.Redis.IO.Timeouts (TimeoutManager, withTimeout) @@ -37,61 +42,92 @@ import Network.Socket.ByteString (recv, sendMany) import System.Logger hiding (Settings, settings, close) import System.Timeout +import Prelude -import qualified Data.Sequence as Seq +import qualified Data.ByteString.Lazy.Char8 as Char8 +import qualified Data.Sequence as Seq import qualified Network.Socket as S data Connection = Connection { settings :: !Settings , logger :: !Logger , timeouts :: !TimeoutManager + , address :: !InetAddr , sock :: !Socket , leftover :: !(IORef ByteString) , buffer :: !(IORef (Seq (Resp, IORef Resp))) } instance Show Connection where - show c = "Connection" ++ show (sock c) + show = Char8.unpack . eval . bytes -resolve :: String -> Word16 -> IO AddrInfo +instance ToBytes Connection where + bytes c = bytes (address c) +++ val "#" +++ fd (sock c) + +resolve :: String -> Word16 -> IO [InetAddr] resolve host port = - head <$> getAddrInfo (Just hints) (Just host) (Just (show port)) + map (InetAddr . addrAddress) <$> getAddrInfo (Just hints) (Just host) (Just (show port)) where hints = defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream } -connect :: Settings -> Logger -> TimeoutManager -> AddrInfo -> IO Connection +connect :: Settings -> Logger -> TimeoutManager -> InetAddr -> IO Connection connect t g m a = bracketOnError mkSock S.close $ \s -> do - ok <- timeout (ms (sConnectTimeout t) * 1000) (S.connect s (addrAddress a)) + ok <- timeout (ms (sConnectTimeout t) * 1000) (S.connect s (sockAddr a)) unless (isJust ok) $ throwIO ConnectTimeout - Connection t g m s <$> newIORef "" <*> newIORef Seq.empty + Connection t g m a s <$> newIORef "" <*> newIORef Seq.empty where - mkSock = socket (addrFamily a) (addrSocketType a) (addrProtocol a) + mkSock = socket (familyOf $ sockAddr a) Stream defaultProtocol + familyOf (SockAddrInet _ _ ) = AF_INET + familyOf (SockAddrInet6 _ _ _ _) = AF_INET6 + familyOf (SockAddrUnix _ ) = AF_UNIX +#if MIN_VERSION_network(2,6,1) + familyOf (SockAddrCan _ ) = AF_CAN +#endif + close :: Connection -> IO () close = S.close . sock request :: Resp -> IORef Resp -> Connection -> IO () request x y c = modifyIORef' (buffer c) (|> (x, y)) +transaction :: Connection -> IO () +transaction c = do + buf <- readIORef (buffer c) + unless (Seq.null buf) $ do + writeIORef (buffer c) Seq.empty + case sSendRecvTimeout (settings c) of + 0 -> go buf + t -> withTimeout (timeouts c) t (abort c) (go buf) + where + go buf = do + let (reqs, vars) = unzip (toList buf) + send c (cmdMulti : reqs ++ [cmdExecute]) + receive c >>= expect "MULTI" "OK" + for_ vars $ const $ + receive c >>= expect "*" "QUEUED" + (lft, res) <- receiveWith c =<< readIORef (leftover c) + writeIORef (leftover c) lft + case res of + Array n resps + | n == length vars -> mapM_ (uncurry writeIORef) (zip vars resps) + Err e -> throwIO (TransactionFailure $ show e) + _ -> throwIO (TransactionFailure "invalid exec response") + sync :: Connection -> IO () sync c = do - a <- readIORef (buffer c) - unless (Seq.null a) $ do + buf <- readIORef (buffer c) + unless (Seq.null buf) $ do writeIORef (buffer c) Seq.empty case sSendRecvTimeout (settings c) of - 0 -> go a - t -> withTimeout (timeouts c) t abort (go a) + 0 -> go buf + t -> withTimeout (timeouts c) t (abort c) (go buf) where - go a = do - send c (toList $ fmap fst a) + go buf = do + send c (toList $ fmap fst buf) bb <- readIORef (leftover c) - foldlM fetchResult bb (fmap snd a) >>= writeIORef (leftover c) - - abort = do - err (logger c) $ "connection.timeout" .= show c - close c - throwIO $ Timeout (show c) + foldlM fetchResult bb (fmap snd buf) >>= writeIORef (leftover c) fetchResult :: ByteString -> IORef Resp -> IO ByteString fetchResult b r = do @@ -99,6 +135,12 @@ writeIORef r x return b' +abort :: Connection -> IO a +abort c = do + err (logger c) $ "connection.timeout" .= show c + close c + throwIO $ Timeout (show c) + send :: Connection -> [Resp] -> IO () send c = sendMany (sock c) . concatMap (toChunks . encode) @@ -120,3 +162,17 @@ errorCheck :: Resp -> IO Resp errorCheck (Err e) = throwIO $ RedisError e errorCheck r = return r + +-- Helpers: + +fd :: Socket -> Int32 +fd !s = let CInt !n = fdSocket s in n + +cmdMulti :: Resp +cmdMulti = Array 1 [Bulk "MULTI"] + +cmdExecute :: Resp +cmdExecute = Array 1 [Bulk "EXEC"] + +expect :: String -> Char8.ByteString -> Resp -> IO () +expect x y = void . either throwIO return . matchStr x y diff --git a/src/Database/Redis/IO/Timeouts.hs b/src/Database/Redis/IO/Timeouts.hs --- a/src/Database/Redis/IO/Timeouts.hs +++ b/src/Database/Redis/IO/Timeouts.hs @@ -14,11 +14,12 @@ ) where import Control.Applicative +import Control.Concurrent.STM import Control.Exception (mask_, bracket) import Control.Reaper import Control.Monad -import Data.IORef import Database.Redis.IO.Types (Milliseconds (..), ignore) +import Prelude data TimeoutManager = TimeoutManager { roundtrip :: !Int @@ -27,7 +28,7 @@ data Action = Action { action :: !(IO ()) - , state :: !(IORef State) + , state :: !(TVar State) } data State = Running !Int | Canceled @@ -39,7 +40,10 @@ } where prune a = do - s <- atomicModifyIORef' (state a) $ \x -> (newState x, x) + s <- atomically $ do + x <- readTVar (state a) + writeTVar (state a) (newState x) + return x case s of Running 0 -> do ignore (action a) @@ -53,17 +57,20 @@ destroy :: TimeoutManager -> Bool -> IO () destroy tm exec = mask_ $ do a <- reaperStop (reaper tm) - when exec $ - mapM_ (ignore . action) a + when exec $ mapM_ f a + where + f e = readTVarIO (state e) >>= \s -> case s of + Running _ -> ignore (action e) + Canceled -> return () add :: TimeoutManager -> Milliseconds -> IO () -> IO Action add tm (Ms n) a = do - r <- Action a <$> newIORef (Running $ n `div` roundtrip tm) + r <- Action a <$> newTVarIO (Running $ n `div` roundtrip tm) reaperAdd (reaper tm) r return r cancel :: Action -> IO () -cancel a = atomicWriteIORef (state a) Canceled +cancel a = atomically $ writeTVar (state a) Canceled withTimeout :: TimeoutManager -> Milliseconds -> IO () -> IO a -> IO a withTimeout tm m x a = bracket (add tm m x) cancel $ const a diff --git a/src/Database/Redis/IO/Types.hs b/src/Database/Redis/IO/Types.hs --- a/src/Database/Redis/IO/Types.hs +++ b/src/Database/Redis/IO/Types.hs @@ -2,17 +2,55 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. +{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} module Database.Redis.IO.Types where import Control.Exception (Exception, SomeException, catch) +import Data.IP import Data.Typeable +import Network.Socket (SockAddr (..), PortNumber) +import System.Logger.Message newtype Milliseconds = Ms { ms :: Int } deriving (Eq, Show, Num) ----------------------------------------------------------------------------- +-- InetAddr + +newtype InetAddr = InetAddr { sockAddr :: SockAddr } deriving (Eq, Ord) + +instance Show InetAddr where + show (InetAddr (SockAddrInet p a)) = + let i = fromIntegral p :: Int in + shows (fromHostAddress a) . showString ":" . shows i $ "" + show (InetAddr (SockAddrInet6 p _ a _)) = + let i = fromIntegral p :: Int in + shows (fromHostAddress6 a) . showString ":" . shows i $ "" + show (InetAddr (SockAddrUnix unix)) = unix +#if MIN_VERSION_network(2,6,1) + show (InetAddr (SockAddrCan int32)) = show int32 +#endif + +instance ToBytes InetAddr where + bytes (InetAddr (SockAddrInet p a)) = + let i = fromIntegral p :: Int in + show (fromHostAddress a) +++ val ":" +++ i + bytes (InetAddr (SockAddrInet6 p _ a _)) = + let i = fromIntegral p :: Int in + show (fromHostAddress6 a) +++ val ":" +++ i + bytes (InetAddr (SockAddrUnix unix)) = bytes unix +#if MIN_VERSION_network(2,6,1) + bytes (InetAddr (SockAddrCan int32)) = bytes int32 +#endif + +ip2inet :: PortNumber -> IP -> InetAddr +ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a) +ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0 + +----------------------------------------------------------------------------- -- ConnectionError data ConnectionError @@ -24,9 +62,9 @@ instance Exception ConnectionError instance Show ConnectionError where - show ConnectionsBusy = "Network.Redis.IO.ConnectionsBusy" - show ConnectionClosed = "Network.Redis.IO.ConnectionClosed" - show ConnectTimeout = "Network.Redis.IO.ConnectTimeout" + show ConnectionsBusy = "redis-io: connections busy" + show ConnectionClosed = "redis-io: connection closed" + show ConnectTimeout = "redis-io: connect timeout" ----------------------------------------------------------------------------- -- InternalError @@ -38,7 +76,7 @@ instance Exception InternalError instance Show InternalError where - show (InternalError e) = "Network.Redis.IO.InternalError: " ++ show e + show (InternalError e) = "redis-io: internal error: " ++ show e ----------------------------------------------------------------------------- -- Timeout @@ -50,7 +88,19 @@ instance Exception Timeout instance Show Timeout where - show (Timeout e) = "Network.Redis.IO.Timeout: " ++ e + show (Timeout e) = "redis-io: timeout: " ++ e + +----------------------------------------------------------------------------- +-- Transaction failure + +-- | An exception thrown on transaction failures. +newtype TransactionFailure = TransactionFailure String + deriving Typeable + +instance Exception TransactionFailure + +instance Show TransactionFailure where + show (TransactionFailure e) = "redis-io: transaction failed: " ++ e ignore :: IO () -> IO () ignore a = catch a (const $ return () :: SomeException -> IO ())