diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +0.12.0 +----------------------------------------------------------------------------- +- Remove wait-queue. +- Add more type-class instances. + 0.10.0 ----------------------------------------------------------------------------- - Add `MonadClient` type-class. diff --git a/cql-io.cabal b/cql-io.cabal --- a/cql-io.cabal +++ b/cql-io.cabal @@ -1,5 +1,5 @@ name: cql-io -version: 0.10.1 +version: 0.12.0 synopsis: Cassandra CQL client. stability: experimental license: OtherLicense @@ -77,6 +77,7 @@ , hashable >= 1.2 && < 2.0 , iproute >= 1.3 && < 1.4 , lens >= 4.4 && < 4.5 + , monad-control >= 0.3 && < 1.0 , mtl >= 2.1 && < 2.3 , mwc-random >= 0.13 && < 0.14 , network >= 2.4 && < 3.0 @@ -86,5 +87,6 @@ , tinylog >= 0.8 && < 0.11 , time >= 1.4 && < 2.0 , transformers >= 0.3 && < 0.5 + , transformers-base >= 0.4 && < 1.0 , uuid >= 1.2.6 && < 2.0 , vector >= 0.10 && < 1.0 diff --git a/src/Database/CQL/IO.hs b/src/Database/CQL/IO.hs --- a/src/Database/CQL/IO.hs +++ b/src/Database/CQL/IO.hs @@ -40,7 +40,6 @@ , setMaxConnections , setMaxStreams , setMaxTimeouts - , setMaxWaitQueue , setPolicy , setPoolStripes , setPortNumber diff --git a/src/Database/CQL/IO/Client.hs b/src/Database/CQL/IO/Client.hs --- a/src/Database/CQL/IO/Client.hs +++ b/src/Database/CQL/IO/Client.hs @@ -2,10 +2,14 @@ -- 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 FlexibleInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeFamilies #-} module Database.CQL.IO.Client ( Client @@ -27,10 +31,16 @@ import Control.Concurrent.STM hiding (retry) import Control.Exception (IOException) import Control.Lens hiding ((.=), Context) -import Control.Monad (unless, void, when) +import Control.Monad (void, when) +import Control.Monad.Base (MonadBase (..)) import Control.Monad.Catch import Control.Monad.IO.Class -import Control.Monad.Reader (ReaderT, runReaderT, MonadReader, ask) +import Control.Monad.Reader (ReaderT (..), runReaderT, MonadReader, ask) +import Control.Monad.Trans.Class +import Control.Monad.Trans.Control (MonadBaseControl (..)) +#if MIN_VERSION_transformers(0,4,0) +import Control.Monad.Trans.Except +#endif import Control.Retry import Data.Foldable (for_, foldrM) import Data.List (find) @@ -55,6 +65,8 @@ import System.Logger.Class hiding (Settings, new, settings, create) import qualified Control.Monad.Reader as Reader +import qualified Control.Monad.State.Strict as S +import qualified Control.Monad.State.Lazy as LS import qualified Data.List.NonEmpty as NE import qualified Data.Map.Strict as Map import qualified Database.CQL.IO.Connection as C @@ -69,25 +81,24 @@ deriving (Eq, Ord, Show) data Control = Control - { _state :: ControlState - , _connection :: Connection + { _state :: !ControlState + , _connection :: !Connection } data Context = Context - { _settings :: Settings - , _logger :: Logger - , _timeouts :: TimeoutManager - , _sigMonit :: Signal HostEvent + { _settings :: !Settings + , _logger :: !Logger + , _timeouts :: !TimeoutManager + , _sigMonit :: !(Signal HostEvent) } -- | Opaque client state/environment. data ClientState = ClientState - { _context :: Context - , _policy :: Policy - , _control :: TVar Control - , _failures :: TVar Word64 - , _hostmap :: TVar (Map Host Pool) - , _jobs :: Jobs InetAddr + { _context :: !Context + , _policy :: !Policy + , _control :: !(TVar Control) + , _hostmap :: !(TVar (Map Host Pool)) + , _jobs :: !(Jobs InetAddr) } makeLenses ''Control @@ -118,6 +129,19 @@ instance MonadLogger Client where log l m = view (context.logger) >>= \g -> Logger.log g l m +instance MonadBase IO Client where + liftBase = liftIO + +instance MonadBaseControl IO Client where + newtype StM Client a = ClientStM + { unClientStM :: StM (ReaderT ClientState IO) a + } + + liftBaseWith f = + Client . liftBaseWith $ \run -> f (fmap ClientStM . run . client) + + restoreM = Client . restoreM . unClientStM + -- | Monads in which 'Client' actions may be embedded. class (Functor m, Applicative m, Monad m, MonadIO m, MonadCatch m) => MonadClient m where @@ -130,6 +154,24 @@ liftClient = id localState = Reader.local +instance MonadClient m => MonadClient (ReaderT r m) where + liftClient = lift . liftClient + localState f m = ReaderT (localState f . runReaderT m) + +instance MonadClient m => MonadClient (S.StateT s m) where + liftClient = lift . liftClient + localState f m = S.StateT (localState f . S.runStateT m) + +instance MonadClient m => MonadClient (LS.StateT s m) where + liftClient = lift . liftClient + localState f m = LS.StateT (localState f . LS.runStateT m) + +#if MIN_VERSION_transformers(0,4,0) +instance MonadClient m => MonadClient (ExceptT e m) where + liftClient = lift . liftClient + localState f m = ExceptT $ localState f (runExceptT m) +#endif + ----------------------------------------------------------------------------- -- API @@ -200,29 +242,14 @@ getResponse a s n where action p = do - res <- tryWith p go + res <- with p transaction case res of Just r -> return r Nothing -> if n > 0 then getResponse a s (n - 1) - else case s^.context.settings.maxWaitQueue of - Nothing -> with p transaction - Just q -> again q p - - go h = do - atomically $ modifyTVar' (s^.failures) $ \x -> if x > 0 then x - 1 else 0 - transaction h - - again q p = do - k <- atomically' $ do - k <- readTVar (s^.failures) - unless (k == q) $ - writeTVar (s^.failures) (k + 1) - return k - unless (k < q) $ - throwM HostsBusy - with p go + else + throwM HostsBusy transaction c = do let x = s^.context.settings.connSettings.compression @@ -277,7 +304,6 @@ x <- ClientState e <$> pure p <*> newTVarIO (Control Connected c) - <*> newTVarIO 0 <*> newTVarIO Map.empty <*> Jobs.new e^.sigMonit |-> onEvent p diff --git a/src/Database/CQL/IO/Cluster/Policies.hs b/src/Database/CQL/IO/Cluster/Policies.hs --- a/src/Database/CQL/IO/Cluster/Policies.hs +++ b/src/Database/CQL/IO/Cluster/Policies.hs @@ -53,8 +53,8 @@ type HostMap = TVar Hosts data Hosts = Hosts - { _alive :: Map InetAddr Host - , _other :: Map InetAddr Host + { _alive :: !(Map InetAddr Host) + , _other :: !(Map InetAddr Host) } deriving Show makeLenses ''Hosts diff --git a/src/Database/CQL/IO/Connection.hs b/src/Database/CQL/IO/Connection.hs --- a/src/Database/CQL/IO/Connection.hs +++ b/src/Database/CQL/IO/Connection.hs @@ -76,29 +76,29 @@ import qualified Network.Socket.ByteString as NB data ConnectionSettings = ConnectionSettings - { _connectTimeout :: Milliseconds - , _sendTimeout :: Milliseconds - , _responseTimeout :: Milliseconds - , _maxStreams :: Int - , _compression :: Compression - , _defKeyspace :: Maybe Keyspace + { _connectTimeout :: !Milliseconds + , _sendTimeout :: !Milliseconds + , _responseTimeout :: !Milliseconds + , _maxStreams :: !Int + , _compression :: !Compression + , _defKeyspace :: !(Maybe Keyspace) } type Streams = Vector (Sync (Header, ByteString)) data Connection = Connection - { _settings :: ConnectionSettings - , _address :: InetAddr - , _tmanager :: TimeoutManager - , _protocol :: Version - , _sock :: Socket - , _streams :: Streams - , _wLock :: MVar () - , _reader :: Async () - , _tickets :: Pool - , _logger :: Logger - , _eventSig :: Signal Event - , _ident :: Unique + { _settings :: !ConnectionSettings + , _address :: !InetAddr + , _tmanager :: !TimeoutManager + , _protocol :: !Version + , _sock :: !Socket + , _streams :: !Streams + , _wLock :: !(MVar ()) + , _reader :: !(Async ()) + , _tickets :: !Pool + , _logger :: !Logger + , _eventSig :: !(Signal Event) + , _ident :: !Unique } makeLenses ''ConnectionSettings diff --git a/src/Database/CQL/IO/Jobs.hs b/src/Database/CQL/IO/Jobs.hs --- a/src/Database/CQL/IO/Jobs.hs +++ b/src/Database/CQL/IO/Jobs.hs @@ -24,7 +24,7 @@ data Job = Reserved - | Running !Unique (Async ()) + | Running !Unique !(Async ()) newtype Jobs k = Jobs (TVar (Map k Job)) diff --git a/src/Database/CQL/IO/Pool.hs b/src/Database/CQL/IO/Pool.hs --- a/src/Database/CQL/IO/Pool.hs +++ b/src/Database/CQL/IO/Pool.hs @@ -13,7 +13,6 @@ , destroy , purge , with - , tryWith , PoolSettings , defSettings @@ -58,14 +57,14 @@ } data Pool = Pool - { _createFn :: IO Connection - , _destroyFn :: Connection -> IO () + { _createFn :: !(IO Connection) + , _destroyFn :: !(Connection -> IO ()) , _logger :: !Logger , _settings :: !PoolSettings , _maxRefs :: !Int - , _currentTime :: IO UTCTime - , _stripes :: Vector Stripe - , _finaliser :: IORef () + , _currentTime :: !(IO UTCTime) + , _stripes :: !(Vector Stripe) + , _finaliser :: !(IORef ()) } data Resource = Resource @@ -75,9 +74,14 @@ , value :: !Connection } deriving Show +data Box + = New !Resource + | Used !Resource + | Empty + data Stripe = Stripe - { conns :: TVar (Seq Resource) - , inUse :: TVar Int + { conns :: !(TVar (Seq Resource)) + , inUse :: !(TVar Int) } makeLenses ''PoolSettings @@ -103,20 +107,11 @@ destroy :: Pool -> IO () destroy = purge -with :: MonadIO m => Pool -> (Connection -> IO a) -> m a +with :: MonadIO m => Pool -> (Connection -> IO a) -> m (Maybe a) with p f = liftIO $ do s <- stripe p mask $ \restore -> do r <- take1 p s - x <- restore (f (value r)) `catches` handlers p s r - put p s r id - return x - -tryWith :: MonadIO m => Pool -> (Connection -> IO a) -> m (Maybe a) -tryWith p f = liftIO $ do - s <- stripe p - mask $ \restore -> do - r <- tryTake1 p s case r of Just v -> do x <- restore (f (value v)) `catches` handlers p s v @@ -144,7 +139,7 @@ destroyR p s r else put p s r incrTimeouts -take1 :: Pool -> Stripe -> IO Resource +take1 :: Pool -> Stripe -> IO (Maybe Resource) take1 p s = do r <- join . atomically $ do c <- readTVar (conns s) @@ -153,40 +148,24 @@ let r :< rr = Seq.viewl $ Seq.unstableSortBy (compare `on` refcnt) c if | u < p^.settings.maxConnections -> mkNew p s u | n > 0 && refcnt r < p^.maxRefs -> use s r rr - | otherwise -> retry - case r of - Left x -> do - atomically (modifyTVar' (conns s) (|> x)) - return x - Right x -> return x - -tryTake1 :: Pool -> Stripe -> IO (Maybe Resource) -tryTake1 p s = do - r <- join . atomically $ do - c <- readTVar (conns s) - u <- readTVar (inUse s) - let n = Seq.length c - let r :< rr = Seq.viewl $ Seq.unstableSortBy (compare `on` refcnt) c - if | u < p^.settings.maxConnections -> fmap Just <$> mkNew p s u - | n > 0 && refcnt r < p^.maxRefs -> fmap Just <$> use s r rr - | otherwise -> return (return Nothing) + | otherwise -> return (return Empty) case r of - Just (Left x) -> do + New x -> do atomically (modifyTVar' (conns s) (|> x)) return (Just x) - Just (Right x) -> return (Just x) - Nothing -> return Nothing + Used x -> return (Just x) + Empty -> return Nothing -use :: Stripe -> Resource -> Seq Resource -> STM (IO (Either Resource Resource)) +use :: Stripe -> Resource -> Seq Resource -> STM (IO Box) use s r rr = do writeTVar (conns s) $! rr |> r { refcnt = refcnt r + 1 } - return (return (Right r)) + return (return (Used r)) {-# INLINE use #-} -mkNew :: Pool -> Stripe -> Int -> STM (IO (Either Resource Resource)) +mkNew :: Pool -> Stripe -> Int -> STM (IO Box) mkNew p s u = do writeTVar (inUse s) $! u + 1 - return $ Left <$> onException + return $ New <$> onException (Resource <$> p^.currentTime <*> pure 1 <*> pure 0 <*> p^.createFn) (atomically (modifyTVar' (inUse s) (subtract 1))) {-# INLINE mkNew #-} @@ -236,4 +215,3 @@ incrTimeouts :: Resource -> Resource incrTimeouts r = r { timeouts = timeouts r + 1 } {-# INLINE incrTimeouts #-} - diff --git a/src/Database/CQL/IO/Settings.hs b/src/Database/CQL/IO/Settings.hs --- a/src/Database/CQL/IO/Settings.hs +++ b/src/Database/CQL/IO/Settings.hs @@ -22,21 +22,20 @@ import Network.Socket (PortNumber (..)) data RetrySettings = RetrySettings - { _retryPolicy :: RetryPolicy - , _reducedConsistency :: Maybe Consistency - , _sendTimeoutChange :: Milliseconds - , _recvTimeoutChange :: Milliseconds + { _retryPolicy :: !RetryPolicy + , _reducedConsistency :: !(Maybe Consistency) + , _sendTimeoutChange :: !Milliseconds + , _recvTimeoutChange :: !Milliseconds } data Settings = Settings - { _poolSettings :: PoolSettings - , _connSettings :: ConnectionSettings - , _retrySettings :: RetrySettings - , _protoVersion :: Version - , _portnumber :: PortNumber - , _contacts :: NonEmpty String - , _maxWaitQueue :: Maybe Word64 - , _policyMaker :: IO Policy + { _poolSettings :: !PoolSettings + , _connSettings :: !ConnectionSettings + , _retrySettings :: !RetrySettings + , _protoVersion :: !Version + , _portnumber :: !PortNumber + , _contacts :: !(NonEmpty String) + , _policyMaker :: !(IO Policy) } makeLenses ''RetrySettings @@ -72,7 +71,6 @@ V3 (fromInteger 9042) ("localhost" :| []) - Nothing random ----------------------------------------------------------------------------- @@ -98,13 +96,6 @@ -- | Set the load-balancing policy. setPolicy :: IO Policy -> Settings -> Settings setPolicy v = set policyMaker v - --- | Set the maximum length of the wait queue which is used if --- connection pools of all nodes in a cluster are busy. Once the maximum --- queue size has been reached, queries will throw a 'HostsBusy' exception --- immediatly. -setMaxWaitQueue :: Word64 -> Settings -> Settings -setMaxWaitQueue v = set maxWaitQueue (Just v) ----------------------------------------------------------------------------- -- Pool Settings diff --git a/src/Database/CQL/IO/Timeouts.hs b/src/Database/CQL/IO/Timeouts.hs --- a/src/Database/CQL/IO/Timeouts.hs +++ b/src/Database/CQL/IO/Timeouts.hs @@ -22,12 +22,12 @@ data TimeoutManager = TimeoutManager { roundtrip :: !Int - , reaper :: Reaper [Action] Action + , reaper :: !(Reaper [Action] Action) } data Action = Action - { action :: IO () - , state :: TVar State + { action :: !(IO ()) + , state :: !(TVar State) } data State = Running !Int | Canceled diff --git a/src/Database/CQL/IO/Types.hs b/src/Database/CQL/IO/Types.hs --- a/src/Database/CQL/IO/Types.hs +++ b/src/Database/CQL/IO/Types.hs @@ -70,7 +70,7 @@ ----------------------------------------------------------------------------- -- InternalError -data InternalError = InternalError String +newtype InternalError = InternalError String deriving Typeable instance Exception InternalError @@ -109,7 +109,7 @@ ----------------------------------------------------------------------------- -- Timeout -data Timeout = TimeoutRead !String +newtype Timeout = TimeoutRead String deriving Typeable instance Exception Timeout @@ -122,7 +122,7 @@ data UnexpectedResponse where UnexpectedResponse :: UnexpectedResponse - UnexpectedResponse' :: Show b => Response k a b -> UnexpectedResponse + UnexpectedResponse' :: Show b => !(Response k a b) -> UnexpectedResponse deriving instance Typeable UnexpectedResponse instance Exception UnexpectedResponse