cql-io 0.10.1 → 0.12.0
raw patch · 11 files changed
+127/−126 lines, 11 filesdep +monad-controldep +transformers-base
Dependencies added: monad-control, transformers-base
Files
- CHANGELOG.md +5/−0
- cql-io.cabal +3/−1
- src/Database/CQL/IO.hs +0/−1
- src/Database/CQL/IO/Client.hs +59/−33
- src/Database/CQL/IO/Cluster/Policies.hs +2/−2
- src/Database/CQL/IO/Connection.hs +18/−18
- src/Database/CQL/IO/Jobs.hs +1/−1
- src/Database/CQL/IO/Pool.hs +22/−44
- src/Database/CQL/IO/Settings.hs +11/−20
- src/Database/CQL/IO/Timeouts.hs +3/−3
- src/Database/CQL/IO/Types.hs +3/−3
CHANGELOG.md view
@@ -1,3 +1,8 @@+0.12.0+-----------------------------------------------------------------------------+- Remove wait-queue.+- Add more type-class instances.+ 0.10.0 ----------------------------------------------------------------------------- - Add `MonadClient` type-class.
cql-io.cabal view
@@ -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
src/Database/CQL/IO.hs view
@@ -40,7 +40,6 @@ , setMaxConnections , setMaxStreams , setMaxTimeouts- , setMaxWaitQueue , setPolicy , setPoolStripes , setPortNumber
src/Database/CQL/IO/Client.hs view
@@ -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
src/Database/CQL/IO/Cluster/Policies.hs view
@@ -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
src/Database/CQL/IO/Connection.hs view
@@ -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
src/Database/CQL/IO/Jobs.hs view
@@ -24,7 +24,7 @@ data Job = Reserved- | Running !Unique (Async ())+ | Running !Unique !(Async ()) newtype Jobs k = Jobs (TVar (Map k Job))
src/Database/CQL/IO/Pool.hs view
@@ -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 #-}-
src/Database/CQL/IO/Settings.hs view
@@ -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
src/Database/CQL/IO/Timeouts.hs view
@@ -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
src/Database/CQL/IO/Types.hs view
@@ -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