diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,373 @@ +Mozilla Public License Version 2.0 +================================== + +1. Definitions +-------------- + +1.1. "Contributor" + means each individual or legal entity that creates, contributes to + the creation of, or owns Covered Software. + +1.2. "Contributor Version" + means the combination of the Contributions of others (if any) used + by a Contributor and that particular Contributor's Contribution. + +1.3. "Contribution" + means Covered Software of a particular Contributor. + +1.4. "Covered Software" + means Source Code Form to which the initial Contributor has attached + the notice in Exhibit A, the Executable Form of such Source Code + Form, and Modifications of such Source Code Form, in each case + including portions thereof. + +1.5. "Incompatible With Secondary Licenses" + means + + (a) that the initial Contributor has attached the notice described + in Exhibit B to the Covered Software; or + + (b) that the Covered Software was made available under the terms of + version 1.1 or earlier of the License, but not also under the + terms of a Secondary License. + +1.6. "Executable Form" + means any form of the work other than Source Code Form. + +1.7. "Larger Work" + means a work that combines Covered Software with other material, in + a separate file or files, that is not Covered Software. + +1.8. "License" + means this document. + +1.9. "Licensable" + means having the right to grant, to the maximum extent possible, + whether at the time of the initial grant or subsequently, any and + all of the rights conveyed by this License. + +1.10. "Modifications" + means any of the following: + + (a) any file in Source Code Form that results from an addition to, + deletion from, or modification of the contents of Covered + Software; or + + (b) any new file in Source Code Form that contains any Covered + Software. + +1.11. "Patent Claims" of a Contributor + means any patent claim(s), including without limitation, method, + process, and apparatus claims, in any patent Licensable by such + Contributor that would be infringed, but for the grant of the + License, by the making, using, selling, offering for sale, having + made, import, or transfer of either its Contributions or its + Contributor Version. + +1.12. "Secondary License" + means either the GNU General Public License, Version 2.0, the GNU + Lesser General Public License, Version 2.1, the GNU Affero General + Public License, Version 3.0, or any later versions of those + licenses. + +1.13. "Source Code Form" + means the form of the work preferred for making modifications. + +1.14. "You" (or "Your") + means an individual or a legal entity exercising rights under this + License. For legal entities, "You" includes any entity that + controls, is controlled by, or is under common control with You. For + purposes of this definition, "control" means (a) the power, direct + or indirect, to cause the direction or management of such entity, + whether by contract or otherwise, or (b) ownership of more than + fifty percent (50%) of the outstanding shares or beneficial + ownership of such entity. + +2. License Grants and Conditions +-------------------------------- + +2.1. Grants + +Each Contributor hereby grants You a world-wide, royalty-free, +non-exclusive license: + +(a) under intellectual property rights (other than patent or trademark) + Licensable by such Contributor to use, reproduce, make available, + modify, display, perform, distribute, and otherwise exploit its + Contributions, either on an unmodified basis, with Modifications, or + as part of a Larger Work; and + +(b) under Patent Claims of such Contributor to make, use, sell, offer + for sale, have made, import, and otherwise transfer either its + Contributions or its Contributor Version. + +2.2. Effective Date + +The licenses granted in Section 2.1 with respect to any Contribution +become effective for each Contribution on the date the Contributor first +distributes such Contribution. + +2.3. Limitations on Grant Scope + +The licenses granted in this Section 2 are the only rights granted under +this License. No additional rights or licenses will be implied from the +distribution or licensing of Covered Software under this License. +Notwithstanding Section 2.1(b) above, no patent license is granted by a +Contributor: + +(a) for any code that a Contributor has removed from Covered Software; + or + +(b) for infringements caused by: (i) Your and any other third party's + modifications of Covered Software, or (ii) the combination of its + Contributions with other software (except as part of its Contributor + Version); or + +(c) under Patent Claims infringed by Covered Software in the absence of + its Contributions. + +This License does not grant any rights in the trademarks, service marks, +or logos of any Contributor (except as may be necessary to comply with +the notice requirements in Section 3.4). + +2.4. Subsequent Licenses + +No Contributor makes additional grants as a result of Your choice to +distribute the Covered Software under a subsequent version of this +License (see Section 10.2) or under the terms of a Secondary License (if +permitted under the terms of Section 3.3). + +2.5. Representation + +Each Contributor represents that the Contributor believes its +Contributions are its original creation(s) or it has sufficient rights +to grant the rights to its Contributions conveyed by this License. + +2.6. Fair Use + +This License is not intended to limit any rights You have under +applicable copyright doctrines of fair use, fair dealing, or other +equivalents. + +2.7. Conditions + +Sections 3.1, 3.2, 3.3, and 3.4 are conditions of the licenses granted +in Section 2.1. + +3. Responsibilities +------------------- + +3.1. Distribution of Source Form + +All distribution of Covered Software in Source Code Form, including any +Modifications that You create or to which You contribute, must be under +the terms of this License. You must inform recipients that the Source +Code Form of the Covered Software is governed by the terms of this +License, and how they can obtain a copy of this License. You may not +attempt to alter or restrict the recipients' rights in the Source Code +Form. + +3.2. Distribution of Executable Form + +If You distribute Covered Software in Executable Form then: + +(a) such Covered Software must also be made available in Source Code + Form, as described in Section 3.1, and You must inform recipients of + the Executable Form how they can obtain a copy of such Source Code + Form by reasonable means in a timely manner, at a charge no more + than the cost of distribution to the recipient; and + +(b) You may distribute such Executable Form under the terms of this + License, or sublicense it under different terms, provided that the + license for the Executable Form does not attempt to limit or alter + the recipients' rights in the Source Code Form under this License. + +3.3. Distribution of a Larger Work + +You may create and distribute a Larger Work under terms of Your choice, +provided that You also comply with the requirements of this License for +the Covered Software. If the Larger Work is a combination of Covered +Software with a work governed by one or more Secondary Licenses, and the +Covered Software is not Incompatible With Secondary Licenses, this +License permits You to additionally distribute such Covered Software +under the terms of such Secondary License(s), so that the recipient of +the Larger Work may, at their option, further distribute the Covered +Software under the terms of either this License or such Secondary +License(s). + +3.4. Notices + +You may not remove or alter the substance of any license notices +(including copyright notices, patent notices, disclaimers of warranty, +or limitations of liability) contained within the Source Code Form of +the Covered Software, except that You may alter any license notices to +the extent required to remedy known factual inaccuracies. + +3.5. Application of Additional Terms + +You may choose to offer, and to charge a fee for, warranty, support, +indemnity or liability obligations to one or more recipients of Covered +Software. However, You may do so only on Your own behalf, and not on +behalf of any Contributor. You must make it absolutely clear that any +such warranty, support, indemnity, or liability obligation is offered by +You alone, and You hereby agree to indemnify every Contributor for any +liability incurred by such Contributor as a result of warranty, support, +indemnity or liability terms You offer. You may include additional +disclaimers of warranty and limitations of liability specific to any +jurisdiction. + +4. Inability to Comply Due to Statute or Regulation +--------------------------------------------------- + +If it is impossible for You to comply with any of the terms of this +License with respect to some or all of the Covered Software due to +statute, judicial order, or regulation then You must: (a) comply with +the terms of this License to the maximum extent possible; and (b) +describe the limitations and the code they affect. Such description must +be placed in a text file included with all distributions of the Covered +Software under this License. Except to the extent prohibited by statute +or regulation, such description must be sufficiently detailed for a +recipient of ordinary skill to be able to understand it. + +5. Termination +-------------- + +5.1. The rights granted under this License will terminate automatically +if You fail to comply with any of its terms. However, if You become +compliant, then the rights granted under this License from a particular +Contributor are reinstated (a) provisionally, unless and until such +Contributor explicitly and finally terminates Your grants, and (b) on an +ongoing basis, if such Contributor fails to notify You of the +non-compliance by some reasonable means prior to 60 days after You have +come back into compliance. Moreover, Your grants from a particular +Contributor are reinstated on an ongoing basis if such Contributor +notifies You of the non-compliance by some reasonable means, this is the +first time You have received notice of non-compliance with this License +from such Contributor, and You become compliant prior to 30 days after +Your receipt of the notice. + +5.2. If You initiate litigation against any entity by asserting a patent +infringement claim (excluding declaratory judgment actions, +counter-claims, and cross-claims) alleging that a Contributor Version +directly or indirectly infringes any patent, then the rights granted to +You by any and all Contributors for the Covered Software under Section +2.1 of this License shall terminate. + +5.3. In the event of termination under Sections 5.1 or 5.2 above, all +end user license agreements (excluding distributors and resellers) which +have been validly granted by You or Your distributors under this License +prior to termination shall survive termination. + +************************************************************************ +* * +* 6. Disclaimer of Warranty * +* ------------------------- * +* * +* Covered Software is provided under this License on an "as is" * +* basis, without warranty of any kind, either expressed, implied, or * +* statutory, including, without limitation, warranties that the * +* Covered Software is free of defects, merchantable, fit for a * +* particular purpose or non-infringing. The entire risk as to the * +* quality and performance of the Covered Software is with You. * +* Should any Covered Software prove defective in any respect, You * +* (not any Contributor) assume the cost of any necessary servicing, * +* repair, or correction. This disclaimer of warranty constitutes an * +* essential part of this License. No use of any Covered Software is * +* authorized under this License except under this disclaimer. * +* * +************************************************************************ + +************************************************************************ +* * +* 7. Limitation of Liability * +* -------------------------- * +* * +* Under no circumstances and under no legal theory, whether tort * +* (including negligence), contract, or otherwise, shall any * +* Contributor, or anyone who distributes Covered Software as * +* permitted above, be liable to You for any direct, indirect, * +* special, incidental, or consequential damages of any character * +* including, without limitation, damages for lost profits, loss of * +* goodwill, work stoppage, computer failure or malfunction, or any * +* and all other commercial damages or losses, even if such party * +* shall have been informed of the possibility of such damages. This * +* limitation of liability shall not apply to liability for death or * +* personal injury resulting from such party's negligence to the * +* extent applicable law prohibits such limitation. Some * +* jurisdictions do not allow the exclusion or limitation of * +* incidental or consequential damages, so this exclusion and * +* limitation may not apply to You. * +* * +************************************************************************ + +8. Litigation +------------- + +Any litigation relating to this License may be brought only in the +courts of a jurisdiction where the defendant maintains its principal +place of business and such litigation shall be governed by laws of that +jurisdiction, without reference to its conflict-of-law provisions. +Nothing in this Section shall prevent a party's ability to bring +cross-claims or counter-claims. + +9. Miscellaneous +---------------- + +This License represents the complete agreement concerning the subject +matter hereof. If any provision of this License is held to be +unenforceable, such provision shall be reformed only to the extent +necessary to make it enforceable. Any law or regulation which provides +that the language of a contract shall be construed against the drafter +shall not be used to construe this License against a Contributor. + +10. Versions of the License +--------------------------- + +10.1. New Versions + +Mozilla Foundation is the license steward. Except as provided in Section +10.3, no one other than the license steward has the right to modify or +publish new versions of this License. Each version will be given a +distinguishing version number. + +10.2. Effect of New Versions + +You may distribute the Covered Software under the terms of the version +of the License under which You originally received the Covered Software, +or under the terms of any subsequent version published by the license +steward. + +10.3. Modified Versions + +If you create software not governed by this License, and you want to +create a new license for such software, you may create and use a +modified version of this License if you rename the license and remove +any references to the name of the license steward (except to note that +such modified license differs from this License). + +10.4. Distributing Source Code Form that is Incompatible With Secondary +Licenses + +If You choose to distribute Source Code Form that is Incompatible With +Secondary Licenses under the terms of this version of the License, the +notice described in Exhibit B of this License must be attached. + +Exhibit A - Source Code Form License Notice +------------------------------------------- + + This Source Code Form is subject to the terms of the Mozilla Public + 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/. + +If it is not possible or desirable to put the notice in a particular +file, then You may include the notice in a location (such as a LICENSE +file in a relevant directory) where a recipient would be likely to look +for such a notice. + +You may add additional accurate notices of copyright ownership. + +Exhibit B - "Incompatible With Secondary Licenses" Notice +--------------------------------------------------------- + + This Source Code Form is "Incompatible With Secondary Licenses", as + defined by the Mozilla Public License, v. 2.0. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,16 @@ +CQL Cassandra driver +-------------------- + +This library uses the [cql](http://hackage.haskell.org/package/cql) library +which implements Cassandra's CQL protocol and complements it with the +neccessary I/O operations. The feature-set includes: + +**Node discovery**. The driver discovers nodes automatically from a small +set of bootstrap nodes. + +**Customisable load-balancing policies**. In addition to pre-built LB +policies such as round-robin, users of this library can provide their +own policies if desired. + +**Support for connection streams**. Requests can be multiplexed over a +few connections. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/cql-io.cabal b/cql-io.cabal new file mode 100644 --- /dev/null +++ b/cql-io.cabal @@ -0,0 +1,86 @@ +name: cql-io +version: 0.9.6 +synopsis: Cassandra CQL client. +stability: experimental +license: OtherLicense +license-file: LICENSE +author: Toralf Wittner +maintainer: Toralf Wittner +copyright: (c) 2014 Toralf Wittner +homepage: https://github.com/twittner/cql-io/ +bug-reports: https://github.com/twittner/cql-io/issues +category: Database +build-type: Simple +cabal-version: >= 1.10 +extra-source-files: README.md + +description: + CQL Cassandra driver supporting native protocol versions 2 and 3. + . + This library uses the library + which implements Cassandra's CQL protocol and complements it with the + neccessary I/O operations. The feature-set includes: + . + * __Node discovery__. The driver discovers nodes automatically from a small + set of bootstrap nodes. + . + * __Customisable load-balancing policies__. In addition to pre-built LB + policies such as round-robin, users of this library can provide their + own policies if desired. + . + * __Support for connection streams__. Requests can be multiplexed over a + few connections. + +source-repository head + type: git + location: git@github.com:twittner/cql-io.git + +library + default-language: Haskell2010 + hs-source-dirs: src + ghc-options: -Wall -O2 -fwarn-tabs + ghc-prof-options: -prof -auto-all + + exposed-modules: + Database.CQL.IO + + other-modules: + Database.CQL.IO.Client + Database.CQL.IO.Cluster.Discovery + Database.CQL.IO.Cluster.Host + Database.CQL.IO.Cluster.Policies + Database.CQL.IO.Connection + Database.CQL.IO.Hexdump + Database.CQL.IO.Jobs + Database.CQL.IO.Pool + Database.CQL.IO.Protocol + Database.CQL.IO.Settings + Database.CQL.IO.Signal + Database.CQL.IO.Sync + Database.CQL.IO.Tickets + Database.CQL.IO.Timeouts + Database.CQL.IO.Types + + build-depends: + async == 2.0.* + , auto-update == 0.1.* + , base >= 4.7 && < 5.0 + , bytestring >= 0.10 && < 1.0 + , containers >= 0.5 && < 1.0 + , cql == 3.0.* + , exceptions >= 0.4 && < 1.0 + , hashable >= 1.2 && < 2.0 + , iproute >= 1.3 && < 1.4 + , lens >= 4.4 && < 4.5 + , mtl >= 2.1 && < 2.3 + , mwc-random >= 0.13 && < 0.14 + , network >= 2.4 && < 3.0 + , retry >= 0.5 && < 0.6 + , semigroups >= 0.15 && < 0.16 + , stm >= 2.4 && < 3.0 + , text >= 0.11 && < 2.0 + , tinylog >= 0.8 && < 0.11 + , time >= 1.4 && < 2.0 + , transformers >= 0.3 && < 0.5 + , 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 new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO.hs @@ -0,0 +1,171 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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/. + +-- | This driver operates on some state which must be initialised prior to +-- executing client operations and terminated eventually. The library uses +-- for its logging +-- output and expects a 'Logger'. +-- +-- For example: +-- +-- @ +-- > import Data.Text (Text) +-- > import Data.Functor.Identity +-- > import Database.CQL.IO as Client +-- > import Database.CQL.Protocol +-- > import qualified System.Logger as Logger +-- > +-- > g <- Logger.new Logger.defSettings +-- > c <- Client.init g defSettings +-- > let p = QueryParams One False () Nothing Nothing Nothing +-- > runClient c $ query ("SELECT cql_version from system.local" :: QueryString R () (Identity Text)) p +-- [Identity "3.2.0"] +-- > shutdown c +-- @ +-- + +{-# LANGUAGE DeriveFunctor #-} + +module Database.CQL.IO + ( -- * Client settings + Settings + , defSettings + , addContact + , setCompression + , setConnectTimeout + , setContacts + , setIdleTimeout + , setKeyspace + , setMaxConnections + , setMaxStreams + , setMaxTimeouts + , setMaxWaitQueue + , setPolicy + , setPoolStripes + , setPortNumber + , setProtocolVersion + , setResponseTimeout + , setSendTimeout + + -- * Client monad + , Client + , ClientState + , DebugInfo (..) + , init + , runClient + , shutdown + , debugInfo + + , query + , write + , schema + , batch + + , Page (..) + , emptyPage + , paginate + + -- ** low-level + , request + , command + + -- * Policies + , Policy (..) + , random + , roundRobin + + -- ** Hosts + , Host + , HostEvent (..) + , InetAddr (..) + , hostAddr + , dataCentre + , rack + + -- * Exceptions + , InvalidSettings (..) + , InternalError (..) + , HostError (..) + , ConnectionError (..) + , UnexpectedResponse (..) + , Timeout (..) + ) where + +import Control.Applicative +import Control.Monad.Catch +import Control.Monad (void) +import Data.Maybe (isJust) +import Database.CQL.Protocol +import Database.CQL.IO.Client +import Database.CQL.IO.Cluster.Host +import Database.CQL.IO.Cluster.Policies +import Database.CQL.IO.Settings +import Database.CQL.IO.Types +import Prelude hiding (init) + +runQuery :: (Tuple a, Tuple b) => QueryString k a b -> QueryParams a -> Client (Response k a b) +runQuery q p = do + r <- request (RqQuery (Query q p)) + case r of + RsError _ e -> throwM e + _ -> return r + +-- | Run a CQL read-only query against a Cassandra node. +query :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client [b] +query q p = do + r <- runQuery q p + case r of + RsResult _ (RowsResult _ b) -> return b + _ -> throwM UnexpectedResponse + +-- | Run a CQL insert/update query against a Cassandra node. +write :: Tuple a => QueryString W a () -> QueryParams a -> Client () +write q p = void $ runQuery q p + +-- | Run a CQL schema query against a Cassandra node. +schema :: Tuple a => QueryString S a () -> QueryParams a -> Client (Maybe SchemaChange) +schema x y = do + r <- runQuery x y + case r of + RsResult _ (SchemaChangeResult s) -> return $ Just s + RsResult _ VoidResult -> return Nothing + _ -> throwM UnexpectedResponse + +-- | Run a batch query against a Cassandra node. +batch :: Batch -> Client () +batch b = command (RqBatch b) + +-- | Return value of 'paginate'. Contains the actual result values as well +-- as an indication of whether there is more data available and the actual +-- action to fetch the next page. +data Page a = Page + { hasMore :: !Bool + , result :: [a] + , nextPage :: Client (Page a) + } deriving (Functor) + +-- | A page with an empty result list. +emptyPage :: Page a +emptyPage = Page False [] (return emptyPage) + +-- | Run a CQL read-only query against a Cassandra node. +-- +-- This function is like 'query', but limits the result size to 10000 +-- (default) unless there is an explicit size restriction given in +-- 'QueryParams'. The returned 'Page' can be used to continue the query. +-- +-- Please note that -- as of Cassandra 2.1.0 -- if your requested page size +-- is equal to the result size, 'hasMore' might be true and a subsequent +-- 'nextPage' will return an empty list in 'result'. +paginate :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client (Page b) +paginate q p = do + let p' = p { pageSize = pageSize p <|> Just 10000 } + r <- runQuery q p' + case r of + RsResult _ (RowsResult m b) -> + if isJust (pagingState m) then + return $ Page True b (paginate q p' { queryPagingState = pagingState m }) + else + return $ Page False b (return emptyPage) + _ -> throwM UnexpectedResponse diff --git a/src/Database/CQL/IO/Client.hs b/src/Database/CQL/IO/Client.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Client.hs @@ -0,0 +1,460 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} + +module Database.CQL.IO.Client + ( Client + , ClientState + , DebugInfo (..) + , runClient + , Database.CQL.IO.Client.init + , shutdown + , request + , command + , debugInfo + ) where + +import Control.Applicative +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async) +import Control.Concurrent.STM hiding (retry) +import Control.Exception (IOException) +import Control.Lens hiding ((.=), Context) +import Control.Monad (unless, void, when) +import Control.Monad.Catch +import Control.Monad.IO.Class +import Control.Monad.Reader (ReaderT, runReaderT, MonadReader, ask) +import Control.Retry +import Data.Foldable (for_, foldrM) +import Data.List (find) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Map.Strict (Map) +import Data.Maybe (fromMaybe, listToMaybe) +import Data.Text (Text) +import Data.Word +import Database.CQL.IO.Cluster.Discovery as Discovery +import Database.CQL.IO.Cluster.Host +import Database.CQL.IO.Cluster.Policies +import Database.CQL.IO.Connection hiding (request) +import Database.CQL.IO.Jobs (Jobs) +import Database.CQL.IO.Pool +import Database.CQL.IO.Protocol +import Database.CQL.IO.Settings +import Database.CQL.IO.Signal +import Database.CQL.IO.Timeouts (TimeoutManager) +import Database.CQL.IO.Types +import Database.CQL.Protocol hiding (Map) +import Network.Socket (SockAddr (..), PortNumber) +import System.Logger.Class hiding (Settings, new, settings, create) + +import qualified Data.List.NonEmpty as NE +import qualified Data.Map.Strict as Map +import qualified Database.CQL.IO.Connection as C +import qualified Database.CQL.IO.Jobs as Jobs +import qualified Database.CQL.IO.Timeouts as TM +import qualified System.Logger as Logger + +data ControlState + = Disconnected + | Connected + | Reconnecting + deriving (Eq, Ord, Show) + +data Control = Control + { _state :: ControlState + , _connection :: Connection + } + +data Context = Context + { _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 + } + +makeLenses ''Control +makeLenses ''Context +makeLenses ''ClientState + +-- | The Client monad. +-- +-- A simple reader monad around some internal state. Prior to executing +-- this monad via 'runClient', its state must be initialised through +-- 'Database.CQL.IO.Client.init' and after finishing operation it should be +-- terminated with 'shutdown'. +-- +-- Actual CQL queries are handled by invoking 'request' oder 'command'. +-- Additionally 'debugInfo' returns an internal cluster view. +newtype Client a = Client + { client :: ReaderT ClientState IO a + } deriving ( Functor + , Applicative + , Monad + , MonadIO + , MonadThrow + , MonadMask + , MonadCatch + , MonadReader ClientState + ) + +instance MonadLogger Client where + log l m = view (context.logger) >>= \g -> Logger.log g l m + +----------------------------------------------------------------------------- +-- API + +-- | Execute the client monad. +runClient :: MonadIO m => ClientState -> Client a -> m a +runClient p a = liftIO $ runReaderT (client a) p + +-- | Send a CQL 'Request' to the server and return a 'Response'. +-- +-- This function will first ask the clients load-balancing 'Policy' for +-- some host and use its connection pool to acquire a connection for +-- request transmission. +-- +-- If all available hosts are busy (i.e. their connection pools are fully +-- utilised), the function will block until a connection becomes available +-- or the maximum wait-queue length has been reached. +request :: (Tuple a, Tuple b) => Request k a b -> Client (Response k a b) +request a = do + s <- ask + n <- liftIO $ hostCount (s^.policy) + start s n + where + start s n = do + h <- pickHost (s^.policy) + p <- Map.lookup h <$> readTVarIO' (s^.hostmap) + case p of + Just x -> action s n x `catches` handlers h + Nothing -> do + err $ msg (val "no pool for host " +++ h) + p' <- mkPool (s^.context) (h^.hostAddr) + atomically' $ modifyTVar (s^.hostmap) (Map.alter (maybe (Just p') Just) h) + start s n + + action s n p = do + res <- tryWith p (go s) + case res of + Just r -> return r + Nothing -> + if n > 0 then + start s (n - 1) + else case s^.context.settings.maxWaitQueue of + Nothing -> with p (transaction s) + Just q -> retry s q p + + go s h = do + atomically $ modifyTVar (s^.failures) $ \n -> if n > 0 then n - 1 else 0 + transaction s h + + retry s q p = do + n <- atomically' $ do + n <- readTVar (s^.failures) + unless (n == q) $ + writeTVar (s^.failures) (n + 1) + return n + unless (n < q) $ + throwM HostsBusy + with p (go s) + + transaction s c = do + let x = s^.context.settings.connSettings.compression + let v = s^.context.settings.protoVersion + r <- parse x <$> C.request c (serialise v x a) + r `seq` return r + + handlers h = + [ Handler $ \(e :: ConnectionError) -> onConnectionError h e >> throwM e + , Handler $ \(e :: IOException) -> onConnectionError h e >> throwM e + ] + + pickHost p = maybe (throwM NoHostAvailable) return =<< liftIO (select p) + +-- | Like 'request' but not returning any result. +command :: Request k () () -> Client () +command = void . request + +data DebugInfo = DebugInfo + { policyInfo :: String -- ^ 'Policy' string representation + , jobInfo :: [InetAddr] -- ^ hosts currently checked for reachability + , hostInfo :: [Host] -- ^ all known hosts + } + +instance Show DebugInfo where + show dbg = showString "running jobs: " + . shows (jobInfo dbg) + . showString "\nknown hosts: " + . shows (hostInfo dbg) + . showString "\npolicy info: " + . shows (policyInfo dbg) + $ "" + +debugInfo :: Client DebugInfo +debugInfo = do + hosts <- Map.keys <$> (readTVarIO' =<< view hostmap) + pols <- liftIO . display =<< view policy + jbs <- Jobs.showJobs =<< view jobs + return $ DebugInfo pols jbs hosts + +----------------------------------------------------------------------------- +-- Initialisation + +-- | Initialise client state with the given 'Settings' using the provided +-- 'Logger' for all it's logging output. +init :: MonadIO m => Logger -> Settings -> m ClientState +init g s = liftIO $ do + t <- TM.create 250 + c <- tryAll (s^.contacts) (mkConnection t) + e <- Context s g t <$> signal + p <- s^.policyMaker + x <- ClientState e + <$> pure p + <*> newTVarIO (Control Connected c) + <*> newTVarIO 0 + <*> newTVarIO Map.empty + <*> Jobs.new + e^.sigMonit |-> onEvent p + runClient x (initialise c) + return x + where + mkConnection t h = do + a <- C.resolve h (s^.portnumber) + Logger.debug g $ msg (val "connecting to " +++ a) + c <- C.connect (s^.connSettings) t (s^.protoVersion) g a + Logger.info g $ msg (val "control connection: " +++ c) + return c + +initialise :: Connection -> Client () +initialise c = do + startup c + env <- ask + pol <- view policy + ctx <- view context + l <- local2Host (c^.address) . listToMaybe <$> query c One Discovery.local () + r <- discoverPeers ctx c + (u, d) <- mkHostMap ctx pol (l:r) + m <- view hostmap + let h = Map.union u d + atomically' $ writeTVar m h + liftIO $ setup pol (Map.keys u) (Map.keys d) + register c allEventTypes (runClient env . onCqlEvent) + info $ msg (val "known hosts: " +++ show (Map.keys h)) + j <- view jobs + for_ (Map.keys d) $ \down -> + Jobs.add j (down^.hostAddr) $ monitor ctx 1000000 60000000 down + +discoverPeers :: MonadIO m => Context -> Connection -> m [Host] +discoverPeers ctx c = liftIO $ do + let p = ctx^.settings.portnumber + map (peer2Host p . asRecord) <$> query c One peers () + +mkHostMap :: Context -> Policy -> [Host] -> Client (Map Host Pool, Map Host Pool) +mkHostMap c p = liftIO . foldrM checkHost (Map.empty, Map.empty) + where + checkHost h (up, down) = do + okay <- acceptable p h + if okay then do + isUp <- C.ping (h^.hostAddr) + if isUp then do + up' <- Map.insert h <$> mkPool c (h^.hostAddr) <*> pure up + return (up', down) + else do + down' <- Map.insert h <$> mkPool c (h^.hostAddr) <*> pure down + return (up, down') + else + return (up, down) + +mkPool :: MonadIO m => Context -> InetAddr -> m Pool +mkPool ctx i = liftIO $ do + let s = ctx^.settings + let m = s^.connSettings.maxStreams + create (connOpen s) connClose (ctx^.logger) (s^.poolSettings) m + where + connOpen s = do + let g = ctx^.logger + c <- C.connect (s^.connSettings) (ctx^.timeouts) (s^.protoVersion) g i + Logger.debug g $ "client.connect" .= c + connInit c `onException` connClose c + return c + + connInit con = do + C.startup con + for_ (ctx^.settings.connSettings.defKeyspace) $ + C.useKeyspace con + + connClose con = do + Logger.debug (ctx^.logger) $ "client.close" .= con + C.close con + +----------------------------------------------------------------------------- +-- Termination + +-- | Terminate client state, i.e. end all running background checks and +-- shutdown all connection pools. +shutdown :: MonadIO m => ClientState -> m () +shutdown s = liftIO $ do + TM.destroy (s^.context.timeouts) True + Jobs.destroy (s^.jobs) + ignore $ C.close . view connection =<< readTVarIO (s^.control) + mapM_ destroy . Map.elems =<< readTVarIO (s^.hostmap) + +----------------------------------------------------------------------------- +-- Monitoring + +monitor :: Context -> Int -> Int -> Host -> IO () +monitor ctx initial upperBound h = do + threadDelay initial + Logger.info (ctx^.logger) $ msg (val "monitoring: " +++ h) + hostCheck 0 maxN + where + hostCheck :: Int -> Int -> IO () + hostCheck n mx = do + threadDelay (2^(min n mx) * 50000) + isUp <- C.ping (h^.hostAddr) + if isUp then do + ctx^.sigMonit $$ HostUp (h^.hostAddr) + Logger.info (ctx^.logger) $ msg (val "reachable: " +++ h) + else do + Logger.info (ctx^.logger) $ msg (val "unreachable: " +++ h) + hostCheck (n + 1) mx + + maxN :: Int + maxN = floor . logBase 2 $ (fromIntegral (upperBound `div` 50000) :: Double) + +----------------------------------------------------------------------------- +-- Exception handling + +onConnectionError :: Exception e => Host -> e -> Client () +onConnectionError h exc = do + warn $ "exception" .= show exc + e <- ask + c <- atomically' $ do + ctrl <- readTVar (e^.control) + let a = ctrl^.connection.address + if ctrl^.state == Connected && a == h^.hostAddr then do + writeTVar (e^.control) (set state Reconnecting ctrl) + return $ Just (ctrl^.connection) + else + return Nothing + maybe (liftIO . ignore . onEvent (e^.policy) $ HostDown (h^.hostAddr)) + (liftIO . void . async . recovering reconnectPolicy reconnectHandlers . continue e) + c + Jobs.add (e^.jobs) (h^.hostAddr) $ + monitor (e^.context) 0 30000000 h + where + continue e conn = do + Jobs.destroy (e^.jobs) + ignore $ C.close conn + ignore $ onEvent (e^.policy) (HostDown (h^.hostAddr)) + x <- NE.nonEmpty . map (view hostAddr) . Map.keys <$> readTVarIO (e^.hostmap) + case x of + Just a -> a `tryAll` (runClient e . replaceControl) `onException` reconnect e a + Nothing -> do + atomically $ modifyTVar (e^.control) (set state Disconnected) + Logger.fatal (e^.context.logger) $ "error-handler" .= val "no host available" + + reconnect e a = do + Logger.info (e^.context.logger) $ msg (val "reconnecting control ...") + a `tryAll` (runClient e . replaceControl) + + reconnectPolicy = capDelay 5000000 (exponentialBackoff 5000) + + reconnectHandlers = + [ const (Handler $ \(_ :: IOException) -> return True) + , const (Handler $ \(_ :: ConnectionError) -> return True) + , const (Handler $ \(_ :: HostError) -> return True) + ] + +replaceControl :: InetAddr -> Client () +replaceControl a = do + ctx <- view context + ctl <- view control + let s = ctx^.settings + c <- C.connect (s^.connSettings) (ctx^.timeouts) (s^.protoVersion) (ctx^.logger) a + initialise c + atomically' $ writeTVar ctl (Control Connected c) + info $ msg (val "new control connection: " +++ c) + +onCqlEvent :: Event -> Client () +onCqlEvent x = do + info $ "client.event" .= show x + pol <- view policy + prt <- view (context.settings.portnumber) + case x of + StatusEvent Down sa -> do + liftIO $ onEvent pol $ HostDown (InetAddr $ mapPort prt sa) + TopologyEvent RemovedNode sa -> do + let a = InetAddr $ mapPort prt sa + hmap <- view hostmap + atomically' $ + modifyTVar hmap (Map.filterWithKey (\h _ -> h^.hostAddr /= a)) + liftIO $ onEvent pol $ HostGone a + StatusEvent Up sa -> + startMonitor $ (InetAddr $ mapPort prt sa) + TopologyEvent NewNode sa -> do + ctx <- view context + hmap <- view hostmap + ctrl <- readTVarIO' =<< view control + let a = InetAddr $ mapPort prt sa + let c = ctrl^.connection + h <- fromMaybe (Host a "" "") . find ((a == ) . view hostAddr) <$> discoverPeers' ctx c + okay <- liftIO $ acceptable pol h + when okay $ do + p <- mkPool ctx (h^.hostAddr) + atomically' $ modifyTVar hmap (Map.alter (maybe (Just p) Just) h) + liftIO $ onEvent pol $ HostNew h + SchemaEvent _ -> return () + where + mapPort i (SockAddrInet _ a) = SockAddrInet i a + mapPort i (SockAddrInet6 _ f a b) = SockAddrInet6 i f a b + mapPort _ unix = unix + + discoverPeers' ctx c = discoverPeers ctx c `catchAll` (const $ return []) + + startMonitor a = do + hmp <- readTVarIO' =<< view hostmap + case find ((a ==) . view hostAddr) (Map.keys hmp) of + Just h -> do + ctx <- view context + jbs <- view jobs + Jobs.add jbs a $ monitor ctx 3000000 60000000 h + Nothing -> return () + +----------------------------------------------------------------------------- +-- Utilities + +peer2Host :: PortNumber -> Peer -> Host +peer2Host i p = Host (ip2inet i (peerRPC p)) (peerDC p) (peerRack p) + +local2Host :: InetAddr -> Maybe (Text, Text) -> Host +local2Host i (Just (dc, rk)) = Host i dc rk +local2Host i Nothing = Host i "" "" + +allEventTypes :: [EventType] +allEventTypes = [TopologyChangeEvent, StatusChangeEvent, SchemaChangeEvent] + +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) + +atomically' :: STM a -> Client a +atomically' = liftIO . atomically + +readTVarIO' :: TVar a -> Client a +readTVarIO' = liftIO . readTVarIO diff --git a/src/Database/CQL/IO/Cluster/Discovery.hs b/src/Database/CQL/IO/Cluster/Discovery.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Cluster/Discovery.hs @@ -0,0 +1,32 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeFamilies #-} + +module Database.CQL.IO.Cluster.Discovery where + +import Data.Functor.Identity (Identity) +import Data.IP +import Data.Text (Text) +import Database.CQL.Protocol + +data Peer = Peer + { peerAddr :: !IP + , peerRPC :: !IP + , peerDC :: !Text + , peerRack :: !Text + } deriving Show + +recordInstance ''Peer + +peers :: QueryString R () (IP, IP, Text, Text) +peers = "SELECT peer, rpc_address, data_center, rack FROM system.peers" + +peer :: QueryString R (Identity IP) (IP, IP, Text, Text) +peer = "SELECT peer, rpc_address, data_center, rack FROM system.peers where peer = ?" + +local :: QueryString R () (Text, Text) +local = "SELECT data_center, rack FROM system.local WHERE key='local'" diff --git a/src/Database/CQL/IO/Cluster/Host.hs b/src/Database/CQL/IO/Cluster/Host.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Cluster/Host.hs @@ -0,0 +1,49 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 OverloadedStrings #-} + +module Database.CQL.IO.Cluster.Host where + +import Control.Lens ((^.), Lens') +import Data.ByteString.Lazy.Char8 (unpack) +import Data.Text (Text) +import Database.CQL.IO.Types (InetAddr) +import System.Logger.Message + +-- | This event will be passed to a 'Policy' to inform it about +-- cluster changes. +data HostEvent + = HostNew !Host -- ^ a new host has been added to the cluster + | HostGone !InetAddr -- ^ a host has been removed from the cluster + | HostUp !InetAddr -- ^ a host has been started + | HostDown !InetAddr -- ^ a host has been stopped + +-- | Host representation. +data Host = Host + { _hostAddr :: !InetAddr + , _dataCentre :: !Text + , _rack :: !Text + } deriving (Eq, Ord) + +-- | The IP address and port number of this host. +hostAddr :: Lens' Host InetAddr +hostAddr f ~(Host a c r) = fmap (\x -> Host x c r) (f a) +{-# INLINE hostAddr #-} + +-- | The data centre name (may be an empty string). +dataCentre :: Lens' Host Text +dataCentre f ~(Host a c r) = fmap (\x -> Host a x r) (f c) +{-# INLINE dataCentre #-} + +-- | The rack name (may be an empty string). +rack :: Lens' Host Text +rack f ~(Host a c r) = fmap (\x -> Host a c x) (f r) +{-# INLINE rack #-} + +instance Show Host where + show = unpack . eval . bytes + +instance ToBytes Host where + bytes h = h^.dataCentre +++ val ":" +++ h^.rack +++ val ":" +++ h^.hostAddr diff --git a/src/Database/CQL/IO/Cluster/Policies.hs b/src/Database/CQL/IO/Cluster/Policies.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Cluster/Policies.hs @@ -0,0 +1,146 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 TemplateHaskell #-} + +module Database.CQL.IO.Cluster.Policies + ( Policy (..) + , random + , roundRobin + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Lens ((^.), view, over, makeLenses) +import Control.Monad +import Data.Map.Strict (Map) +import Data.Word +import Database.CQL.IO.Cluster.Host +import Database.CQL.IO.Types (InetAddr) +import System.Random.MWC + +import qualified Data.Map.Strict as Map + +-- | A policy defines a load-balancing strategy and generally +-- handles host visibility. +data Policy = Policy + { setup :: [Host] -> [Host] -> IO () + -- ^ Initialise the policy with two sets of hosts. The first + -- parameter are hosts known to be available, the second are other + -- nodes. + -- Please note that a policy may be re-initialised at any point + -- through this method. + , onEvent :: HostEvent -> IO () + -- ^ Event handler. Policies will be informed about cluster changes + -- through this function. + , select :: IO (Maybe Host) + -- ^ Host selection. The driver will ask for a host to use in a query + -- through this function. A policy which has no available nodes my + -- return Nothing. + , acceptable :: Host -> IO Bool + -- ^ During startup and node discovery, the driver will ask the + -- policy if a dicovered host should be ignored. + , hostCount :: IO Word + -- ^ During query processing, the driver will ask the policy for + -- a rough esitimate of alive hosts. The number is used to repeatedly + -- invoke 'select' (with the underlying assumption that the policy + -- returns mostly different hosts). + , display :: IO String + -- ^ Like having an effectful 'Show' instance for this policy. + } + +type HostMap = TVar Hosts + +data Hosts = Hosts + { _alive :: Map InetAddr Host + , _other :: Map InetAddr Host + } deriving Show + +makeLenses ''Hosts + +-- | Iterate over hosts one by one. +roundRobin :: IO Policy +roundRobin = do + h <- newTVarIO emptyHosts + c <- newTVarIO 0 + return $ Policy (defSetup h) (defOnEvent h) (pickHost h c) + defAcceptable (defHostCount h) (defDisplay h) + where + pickHost h c = atomically $ do + m <- view alive <$> readTVar h + if Map.null m then + return Nothing + else do + k <- readTVar c + writeTVar c $ succ k `mod` Map.size m + return . Just . snd $ Map.elemAt (k `mod` Map.size m) m + +-- | Return hosts in random order. +random :: IO Policy +random = do + h <- newTVarIO emptyHosts + g <- createSystemRandom + return $ Policy (defSetup h) (defOnEvent h) (pickHost h g) + defAcceptable (defHostCount h) (defDisplay h) + where + pickHost h g = do + m <- view alive <$> readTVarIO h + if Map.null m then + return Nothing + else do + let i = uniformR (0, Map.size m - 1) g + Just . snd . flip Map.elemAt m <$> i + +----------------------------------------------------------------------------- +-- Defaults + +emptyHosts :: Hosts +emptyHosts = Hosts Map.empty Map.empty + +defDisplay :: HostMap -> IO String +defDisplay h = show <$> readTVarIO h + +defAcceptable :: Host -> IO Bool +defAcceptable = const $ return True + +defSetup :: HostMap -> [Host] -> [Host] -> IO () +defSetup r a b = do + let ha = Map.fromList $ zip (map (view hostAddr) a) a + let hb = Map.fromList $ zip (map (view hostAddr) b) b + let hosts = Hosts ha hb + atomically $ writeTVar r hosts + +defHostCount :: HostMap -> IO Word +defHostCount r = fromIntegral . Map.size . view alive <$> readTVarIO r + +defOnEvent :: HostMap -> HostEvent -> IO () +defOnEvent r (HostNew h) = atomically $ do + m <- readTVar r + when (Nothing == get (h^.hostAddr) m) $ + writeTVar r (over alive (Map.insert (h^.hostAddr) h) m) +defOnEvent r (HostGone a) = atomically $ do + m <- readTVar r + if Map.member a (m^.alive) then + writeTVar r (over alive (Map.delete a) m) + else + writeTVar r (over other (Map.delete a) m) +defOnEvent r (HostUp a) = atomically $ do + m <- readTVar r + case get a m of + Nothing -> return () + Just h -> writeTVar r + $ over alive (Map.insert a h) + . over other (Map.delete a) + $ m +defOnEvent r (HostDown a) = atomically $ do + m <- readTVar r + case get a m of + Nothing -> return () + Just h -> writeTVar r + $ over other (Map.insert a h) + . over alive (Map.delete a) + $ m + +get :: InetAddr -> Hosts -> Maybe Host +get a m = Map.lookup a (m^.alive) <|> Map.lookup a (m^.other) diff --git a/src/Database/CQL/IO/Connection.hs b/src/Database/CQL/IO/Connection.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Connection.hs @@ -0,0 +1,311 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Database.CQL.IO.Connection + ( Connection + , resolve + , ping + , connect + , close + , request + , startup + , register + , query + , useKeyspace + , address + , protocol + , eventSig + + , ConnectionSettings + , defSettings + , connectTimeout + , sendTimeout + , responseTimeout + , maxStreams + , compression + , defKeyspace + ) where + +import Control.Applicative +import Control.Concurrent (myThreadId) +import Control.Concurrent.Async +import Control.Concurrent.MVar +import Control.Exception (throwTo, AsyncException (ThreadKilled)) +import Control.Lens ((^.), makeLenses, view) +import Control.Monad +import Control.Monad.Catch +import Control.Monad.IO.Class +import Data.ByteString.Builder +import Data.ByteString.Lazy (ByteString) +import Data.Int +import Data.Maybe (isJust, fromMaybe) +import Data.Monoid +import Data.Text.Lazy (fromStrict) +import Data.Unique +import Data.Vector (Vector, (!)) +import Database.CQL.Protocol +import Database.CQL.IO.Hexdump +import Database.CQL.IO.Protocol +import Database.CQL.IO.Signal hiding (connect) +import Database.CQL.IO.Sync (Sync) +import Database.CQL.IO.Types +import Database.CQL.IO.Tickets (Pool, toInt, markAvailable) +import Database.CQL.IO.Timeouts (TimeoutManager, withTimeout) +import Foreign.C.Types (CInt (..)) +import Network +import Network.Socket hiding (connect, close, recv, send) +import Network.Socket.ByteString.Lazy (sendAll) +import System.IO (nativeNewline, Newline (..)) +import System.Logger hiding (Settings, close, defSettings, settings) +import System.Timeout + +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as L +import qualified Data.ByteString.Lazy.Char8 as Char8 +import qualified Data.Vector as Vector +import qualified Database.CQL.IO.Sync as Sync +import qualified Database.CQL.IO.Tickets as Tickets +import qualified Network.Socket as S +import qualified Network.Socket.ByteString as NB + +data ConnectionSettings = ConnectionSettings + { _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 + } + +makeLenses ''ConnectionSettings +makeLenses ''Connection + +instance Eq Connection where + a == b = a^.ident == b^.ident + +instance Show Connection where + show = Char8.unpack . eval . bytes + +instance ToBytes Connection where + bytes c = bytes (c^.address) +++ val "#" +++ fd (c^.sock) + +defSettings :: ConnectionSettings +defSettings = + ConnectionSettings 5000 -- connect timeout + 3000 -- send timeout + 10000 -- response timeout + 128 -- max streams per connection + noCompression -- compression + Nothing -- keyspace + +resolve :: String -> PortNumber -> IO InetAddr +resolve host port = + InetAddr . addrAddress . head <$> getAddrInfo (Just hints) (Just host) (Just (show port)) + where + hints = defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream } + +connect :: MonadIO m => ConnectionSettings -> TimeoutManager -> Version -> Logger -> InetAddr -> m Connection +connect t m v g a = liftIO $ + bracketOnError (mkSock a) S.close $ \s -> do + ok <- timeout (ms (t^.connectTimeout) * 1000) (S.connect s (sockAddr a)) + unless (isJust ok) $ + throwM (ConnectTimeout a) + c <- open s + validateSettings c + return c + where + open s = do + tck <- Tickets.pool (t^.maxStreams) + syn <- Vector.replicateM (t^.maxStreams) Sync.create + lck <- newMVar () + sig <- signal + rdr <- async (readLoop v g (t^.compression) tck a s syn sig) + Connection t a m v s syn lck rdr tck g sig <$> newUnique + +mkSock :: InetAddr -> IO Socket +mkSock (InetAddr a) = socket (familyOf a) Stream defaultProtocol + where + familyOf (SockAddrInet {..}) = AF_INET + familyOf (SockAddrInet6 {..}) = AF_INET6 + familyOf (SockAddrUnix {..}) = AF_UNIX + +ping :: MonadIO m => InetAddr -> m Bool +ping a = liftIO $ bracket (mkSock a) S.close $ \s -> + fromMaybe False <$> timeout 5000000 + ((S.connect s (sockAddr a) >> return True) `catchAll` const (return False)) + +readLoop :: Version -> Logger -> Compression -> Pool -> InetAddr -> Socket -> Streams -> Signal Event -> IO () +readLoop v g cmp tck i sck syn s = run `catch` logException `finally` cleanup + where + run = forever $ do + x <- readSocket v g i sck + case fromStreamId $ streamId (fst x) of + -1 -> + case parse cmp x :: Response () () () of + RsError _ e -> throwM e + RsEvent _ e -> emit s e + r -> throwM (UnexpectedResponse' r) + sid -> do + ok <- Sync.put x (syn ! sid) + unless ok $ + markAvailable tck sid + + cleanup = do + Tickets.close (ConnectionClosed i) tck + Vector.mapM_ (Sync.close (ConnectionClosed i)) syn + S.close sck + + logException :: SomeException -> IO () + logException e = case fromException e of + Just ThreadKilled -> return () + _ -> warn g $ msg i ~~ msg (val "read-loop: " +++ show e) + +close :: Connection -> IO () +close = cancel . view reader + +request :: Connection -> (Int -> ByteString) -> IO (Header, ByteString) +request c f = send >>= receive + where + send = withTimeout (c^.tmanager) (c^.settings.sendTimeout) (close c) $ do + i <- toInt <$> Tickets.get (c^.tickets) + let req = f i + trace (c^.logger) $ msg c + ~~ "stream" .= i + ~~ "type" .= val "request" + ~~ msg' (hexdump (L.take 160 req)) + withMVar (c^.wLock) $ + const $ sendAll (c^.sock) req + return i + + receive i = do + let e = TimeoutRead (show c ++ ":" ++ show i) + tid <- myThreadId + withTimeout (c^.tmanager) (c^.settings.responseTimeout) (throwTo tid e) $ do + x <- Sync.get (view streams c ! i) `onException` (Sync.kill e) (view streams c ! i) + markAvailable (c^.tickets) i + return x + +readSocket :: Version -> Logger -> InetAddr -> Socket -> IO (Header, ByteString) +readSocket v g i s = do + b <- recv i (if v == V3 then 9 else 8) s + h <- case header v b of + Left e -> throwM $ InternalError ("response header reading: " ++ e) + Right h -> return h + case headerType h of + RqHeader -> throwM $ InternalError "unexpected request header" + RsHeader -> do + let len = lengthRepr (bodyLength h) + x <- recv i (fromIntegral len) s + trace g $ msg (i +++ val "#" +++ fd s) + ~~ "stream" .= fromStreamId (streamId h) + ~~ "type" .= val "response" + ~~ msg' (hexdump $ L.take 160 (b <> x)) + return (h, x) + +recv :: InetAddr -> Int -> Socket -> IO ByteString +recv _ 0 _ = return L.empty +recv i n c = toLazyByteString <$> go 0 mempty + where + go !k !bb = do + a <- NB.recv c (n - k) + when (B.null a) $ + throwM (ConnectionClosed i) + let b = bb <> byteString a + m = B.length a + k + if m < n then go m b else return b + +----------------------------------------------------------------------------- +-- Operations + +startup :: MonadIO m => Connection -> m () +startup c = liftIO $ do + let cmp = c^.settings.compression + let req = RqStartup (Startup Cqlv300 (algorithm cmp)) + let enc = serialise (c^.protocol) cmp (req :: Request () () ()) + res <- request c enc + (parse cmp res :: Response () () ()) `seq` return () + +register :: MonadIO m => Connection -> [EventType] -> EventHandler -> m () +register c e f = liftIO $ do + let req = RqRegister (Register e) :: Request () () () + let enc = serialise (c^.protocol) (c^.settings.compression) req + res <- request c enc + case parse (c^.settings.compression) res :: Response () () () of + RsReady _ Ready -> c^.eventSig |-> f + other -> throwM (UnexpectedResponse' other) + +validateSettings :: MonadIO m => Connection -> m () +validateSettings c = liftIO $ do + Supported ca _ <- supportedOptions c + let x = algorithm (c^.settings.compression) + unless (x == None || x `elem` ca) $ + throwM $ UnsupportedCompression ca + +supportedOptions :: MonadIO m => Connection -> m Supported +supportedOptions c = liftIO $ do + let options = RqOptions Options :: Request () () () + res <- request c (serialise (c^.protocol) noCompression options) + case parse noCompression res :: Response () () () of + RsSupported _ x -> return x + other -> throwM (UnexpectedResponse' other) + +useKeyspace :: MonadIO m => Connection -> Keyspace -> m () +useKeyspace c ks = liftIO $ do + let cmp = c^.settings.compression + params = QueryParams One False () Nothing Nothing Nothing + kspace = quoted (fromStrict $ unKeyspace ks) + req = RqQuery (Query (QueryString $ "use " <> kspace) params) + res <- request c (serialise (c^.protocol) cmp req) + case parse cmp res :: Response () () () of + RsResult _ (SetKeyspaceResult _) -> return () + other -> throwM (UnexpectedResponse' other) + +query :: forall k a b m. (Tuple a, Tuple b, Show b, MonadIO m) + => Connection + -> Consistency + -> QueryString k a b + -> a + -> m [b] +query c cons q p = liftIO $ do + let req = RqQuery (Query q params) :: Request k a b + let enc = serialise (c^.protocol) (c^.settings.compression) req + res <- request c enc + case parse (c^.settings.compression) res :: Response k a b of + RsResult _ (RowsResult _ b) -> return b + other -> throwM (UnexpectedResponse' other) + where + params = QueryParams cons False p Nothing Nothing Nothing + +-- logging helpers: + +fd :: Socket -> Int32 +fd !s = let CInt !n = fdSocket s in n + +msg' :: ByteString -> Msg -> Msg +msg' x = msg $ case nativeNewline of + LF -> val "\n" +++ x + CRLF -> val "\r\n" +++ x diff --git a/src/Database/CQL/IO/Hexdump.hs b/src/Database/CQL/IO/Hexdump.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Hexdump.hs @@ -0,0 +1,63 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 OverloadedStrings #-} + +module Database.CQL.IO.Hexdump (hexdump, hexdumpBuilder) where + +import Data.ByteString.Builder +import Data.ByteString.Lazy (ByteString) +import Data.Int +import Data.Monoid +import Data.Word +import System.IO (nativeNewline, Newline (..)) + +import qualified Data.ByteString.Lazy as L +import qualified Data.List as List + +width, groups :: Int64 +width = 16 +groups = 4 + +hexdump :: ByteString -> ByteString +hexdump = toLazyByteString . hexdumpBuilder + +hexdumpBuilder :: ByteString -> Builder +hexdumpBuilder = mconcat + . List.intersperse newline + . map toLine + . zipWith (,) [1 ..] + . chunks width + +chunks :: Int64 -> ByteString -> [ByteString] +chunks n b = List.unfoldr step b + where + step "" = Nothing + step c = Just $! L.splitAt n c + +toLine :: (Word16, ByteString) -> Builder +toLine (n, b) = let k = L.length b in + word16HexFixed n + <> ": " + <> mconcat (List.intersperse space (map toGroup (chunks groups b))) + <> spaces ((width - k) * (groups - 1) + pad k + 2) + <> lazyByteString (toAscii b) + +toGroup :: ByteString -> Builder +toGroup = L.foldr (\x y -> word8HexFixed x <> space <> y) mempty + +toAscii :: ByteString -> ByteString +toAscii = L.map (\w -> if w > 0x1F && w < 0x7F then w else 0x2E) + +space, newline :: Builder +space = byteString " " +newline = case nativeNewline of + LF -> byteString "\n" + CRLF -> byteString "\r\n" + +spaces :: Int64 -> Builder +spaces n = lazyByteString $ L.replicate n 0x20 + +pad :: Int64 -> Int64 +pad n = (width - n) `div` groups diff --git a/src/Database/CQL/IO/Jobs.hs b/src/Database/CQL/IO/Jobs.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Jobs.hs @@ -0,0 +1,68 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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/. + +module Database.CQL.IO.Jobs + ( Jobs + , new + , add + , destroy + , showJobs + ) where + +import Control.Applicative +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Monad +import Control.Monad.Catch +import Control.Monad.IO.Class +import Database.CQL.IO.Types +import Data.Map.Strict (Map) +import Data.Unique + +import qualified Data.Map.Strict as Map + +data Job + = Reserved + | Running !Unique (Async ()) + +newtype Jobs k = Jobs (TVar (Map k Job)) + +new :: MonadIO m => m (Jobs k) +new = liftIO $ Jobs <$> newTVarIO Map.empty + +add :: (MonadIO m, Ord k) => Jobs k -> k -> IO () -> m () +add (Jobs d) k j = liftIO $ do + (ok, prev) <- atomically $ do + m <- readTVar d + case Map.lookup k m of + Just Reserved -> return (False, Nothing) + Just (Running _ a) -> do + modifyTVar d (Map.insert k Reserved) + return (True, Just a) + Nothing -> do + modifyTVar d (Map.insert k Reserved) + return (True, Nothing) + when ok $ do + maybe (return ()) (ignore . cancel) prev + u <- newUnique + a <- async $ j `finally` remove u + atomically $ + modifyTVar d (Map.insert k (Running u a)) + where + remove u = atomically $ + modifyTVar d $ flip Map.update k $ \a -> + case a of + Running u' _ | u == u' -> Nothing + _ -> Just a + +destroy :: MonadIO m => Jobs k -> m () +destroy (Jobs d) = liftIO $ do + items <- Map.elems <$> atomically (swapTVar d Map.empty) + mapM_ f items + where + f (Running _ a) = ignore (cancel a) + f _ = return () + +showJobs :: MonadIO m => Jobs k -> m [k] +showJobs (Jobs d) = liftIO $ Map.keys <$> readTVarIO d diff --git a/src/Database/CQL/IO/Pool.hs b/src/Database/CQL/IO/Pool.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Pool.hs @@ -0,0 +1,239 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 MultiWayIf #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} + +module Database.CQL.IO.Pool + ( Pool + , create + , destroy + , purge + , with + , tryWith + + , PoolSettings + , defSettings + , idleTimeout + , maxConnections + , maxTimeouts + , poolStripes + ) where + +import Control.Applicative +import Control.AutoUpdate +import Control.Concurrent +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Exception +import Control.Lens ((^.), makeLenses, view) +import Control.Monad.IO.Class +import Control.Monad hiding (forM_, mapM_) +import Data.Foldable (forM_, mapM_, find) +import Data.Function (on) +import Data.Hashable +import Data.IORef +import Prelude hiding (mapM_) +import Data.Sequence (Seq, ViewL (..), (|>), (><)) +import Data.Time.Clock (UTCTime, NominalDiffTime, getCurrentTime, diffUTCTime) +import Data.Vector (Vector, (!)) +import Database.CQL.IO.Connection (Connection) +import Database.CQL.IO.Types (Timeout, ignore) +import System.Logger hiding (create, defSettings, settings) + +import qualified Data.Sequence as Seq +import qualified Data.Vector as Vec + +----------------------------------------------------------------------------- +-- API + +data PoolSettings = PoolSettings + { _idleTimeout :: !NominalDiffTime + , _maxConnections :: !Int + , _maxTimeouts :: !Int + , _poolStripes :: !Int + } + +data Pool = Pool + { _createFn :: IO Connection + , _destroyFn :: Connection -> IO () + , _logger :: !Logger + , _settings :: !PoolSettings + , _maxRefs :: !Int + , _currentTime :: IO UTCTime + , _stripes :: Vector Stripe + , _finaliser :: IORef () + } + +data Resource = Resource + { tstamp :: !UTCTime + , refcnt :: !Int + , timeouts :: !Int + , value :: !Connection + } deriving Show + +data Stripe = Stripe + { conns :: TVar (Seq Resource) + , inUse :: TVar Int + } + +makeLenses ''PoolSettings +makeLenses ''Pool + +defSettings :: PoolSettings +defSettings = PoolSettings + 60 -- idle timeout + 2 -- max connections per stripe + 16 -- max timeouts per connection + 4 -- max stripes + +create :: IO Connection -> (Connection -> IO ()) -> Logger -> PoolSettings -> Int -> IO Pool +create mk del g s k = do + p <- Pool mk del g s k + <$> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime } + <*> Vec.replicateM (s^.poolStripes) (Stripe <$> newTVarIO Seq.empty <*> newTVarIO 0) + <*> newIORef () + r <- async $ reaper p + void $ mkWeakIORef (p^.finaliser) (cancel r >> destroy p) + return p + +destroy :: Pool -> IO () +destroy = purge + +with :: MonadIO m => Pool -> (Connection -> IO a) -> m 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 + put p s v id + return (Just x) + Nothing -> return Nothing + +purge :: Pool -> IO () +purge p = Vec.forM_ (p^.stripes) $ \s -> + atomically (swapTVar (conns s) Seq.empty) >>= mapM_ (ignore . view destroyFn p . value) + +----------------------------------------------------------------------------- +-- Internal + +handlers :: Pool -> Stripe -> Resource -> [Handler a] +handlers p s r = + [ Handler $ \(x :: Timeout) -> onTimeout >> throwIO x + , Handler $ \(x :: SomeException) -> destroyR p s r >> throwIO x + ] + where + onTimeout = + if timeouts r > p^.settings.maxTimeouts + then do + info (p^.logger) $ msg (show (value r) +++ val " has too many timeouts.") + destroyR p s r + else put p s r incrTimeouts + +take1 :: Pool -> Stripe -> IO Resource +take1 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 -> 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) + case r of + Just (Left x) -> do + atomically (modifyTVar (conns s) (|> x)) + return (Just x) + Just (Right x) -> return (Just x) + Nothing -> return Nothing + +use :: Stripe -> Resource -> Seq Resource -> STM (IO (Either Resource Resource)) +use s r rr = do + writeTVar (conns s) $! rr |> r { refcnt = refcnt r + 1 } + return (return (Right r)) +{-# INLINE use #-} + +mkNew :: Pool -> Stripe -> Int -> STM (IO (Either Resource Resource)) +mkNew p s u = do + writeTVar (inUse s) $! u + 1 + return $ Left <$> onException + (Resource <$> p^.currentTime <*> pure 1 <*> pure 0 <*> p^.createFn) + (atomically (modifyTVar (inUse s) (subtract 1))) +{-# INLINE mkNew #-} + +put :: Pool -> Stripe -> Resource -> (Resource -> Resource) -> IO () +put p s r f = do + now <- p^.currentTime + let updated x = f x { tstamp = now, refcnt = refcnt x - 1 } + atomically $ do + rs <- readTVar (conns s) + let (xs, rr) = Seq.breakl ((value r ==) . value) rs + case Seq.viewl rr of + EmptyL -> writeTVar (conns s) $! xs |> updated r + y :< ys -> writeTVar (conns s) $! (xs >< ys) |> updated y + +destroyR :: Pool -> Stripe -> Resource -> IO () +destroyR p s r = do + atomically $ do + rs <- readTVar (conns s) + case find ((value r ==) . value) rs of + Nothing -> return () + Just _ -> do + modifyTVar (inUse s) (subtract 1) + writeTVar (conns s) $! Seq.filter ((value r /=) . value) rs + ignore $ p^.destroyFn $ value r + +reaper :: Pool -> IO () +reaper p = forever $ do + threadDelay 1000000 + now <- p^.currentTime + let isStale r = refcnt r == 0 && now `diffUTCTime` tstamp r > p^.settings.idleTimeout + Vec.forM_ (p^.stripes) $ \s -> do + x <- atomically $ do + (stale, okay) <- Seq.partition isStale <$> readTVar (conns s) + unless (Seq.null stale) $ do + writeTVar (conns s) okay + modifyTVar (inUse s) (subtract (Seq.length stale)) + return stale + forM_ x $ \v -> ignore $ do + trace (p^.logger) $ "reap" .= show (value v) + p^.destroyFn $ (value v) + +stripe :: Pool -> IO Stripe +stripe p = ((p^.stripes) !) <$> ((`mod` (p^.settings.poolStripes)) . hash) <$> myThreadId +{-# INLINE stripe #-} + +incrTimeouts :: Resource -> Resource +incrTimeouts r = r { timeouts = timeouts r + 1 } +{-# INLINE incrTimeouts #-} + diff --git a/src/Database/CQL/IO/Protocol.hs b/src/Database/CQL/IO/Protocol.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Protocol.hs @@ -0,0 +1,33 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 OverloadedStrings #-} + +module Database.CQL.IO.Protocol where + +import Control.Exception (throw) +import Data.ByteString.Lazy (ByteString) +import Data.Monoid ((<>)) +import Database.CQL.Protocol +import Database.CQL.IO.Types + +import qualified Data.Text.Lazy as LT + +parse :: (Tuple a, Tuple b) => Compression -> (Header, ByteString) -> Response k a b +parse x (h, a) = + case unpack x h a of + Left e -> throw $ InternalError ("response body reading: " ++ e) + Right r -> r + +serialise :: Tuple a => Version -> Compression -> Request k a b -> Int -> ByteString +serialise v f r i = + let c = case getOpCode r of + OcStartup -> noCompression + OcOptions -> noCompression + _ -> f + s = mkStreamId i + in either (throw $ InternalError "request creation") id (pack v c False s r) + +quoted :: LT.Text -> LT.Text +quoted s = "\"" <> LT.replace "\"" "\"\"" s <> "\"" diff --git a/src/Database/CQL/IO/Settings.hs b/src/Database/CQL/IO/Settings.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Settings.hs @@ -0,0 +1,156 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} + +module Database.CQL.IO.Settings where + +import Control.Lens hiding ((<|)) +import Data.List.NonEmpty (NonEmpty (..), (<|)) +import Data.Time +import Data.Word +import Database.CQL.Protocol +import Database.CQL.IO.Connection +import Database.CQL.IO.Cluster.Policies (Policy, random) +import Database.CQL.IO.Connection as C +import Database.CQL.IO.Pool as P +import Database.CQL.IO.Types (Milliseconds (..)) +import Network.Socket (PortNumber (..)) + +data Settings = Settings + { _poolSettings :: PoolSettings + , _connSettings :: ConnectionSettings + , _protoVersion :: Version + , _portnumber :: PortNumber + , _contacts :: NonEmpty String + , _maxWaitQueue :: Maybe Word64 + , _policyMaker :: IO Policy + } + +makeLenses ''Settings + +-- | Default settings: +-- +-- * contact point is \"localhost\" port 9042 +-- +-- * load-balancing policy is 'random' +-- +-- * binary protocol version is 3 (supported by Cassandra >= 2.1.0) +-- +-- * connection idle timeout is 60s +-- +-- * the connection pool uses 4 stripes to mitigate thread contention +-- +-- * connections use a connect timeout of 5s, a send timeout of 3s and +-- a receive timeout of 10s +-- +-- * 128 streams per connection are used +-- +-- * no compression is applied to frame bodies +-- +-- * no default keyspace is used. +defSettings :: Settings +defSettings = Settings + P.defSettings + C.defSettings + V3 + (fromInteger 9042) + ("localhost" :| []) + Nothing + random + +----------------------------------------------------------------------------- +-- Settings + +-- | Set the binary protocol version to use. +setProtocolVersion :: Version -> Settings -> Settings +setProtocolVersion v = set protoVersion v + +-- | Set the initial contact points (hosts) from which node discovery will +-- start. +setContacts :: String -> [String] -> Settings -> Settings +setContacts v vv = set contacts (v :| vv) + +-- | Add an additional host to the contact list. +addContact :: String -> Settings -> Settings +addContact v = over contacts (v <|) + +-- | Set the portnumber to use to connect on /every/ node of the cluster. +setPortNumber :: PortNumber -> Settings -> Settings +setPortNumber v = set portnumber v + +-- | 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 + +-- | Set the connection idle timeout. Connections in a pool will be closed +-- if not in use for longer than this timeout. +setIdleTimeout :: NominalDiffTime -> Settings -> Settings +setIdleTimeout v = set (poolSettings.idleTimeout) v + +-- | Maximum connections per pool /stripe/. +setMaxConnections :: Int -> Settings -> Settings +setMaxConnections v = set (poolSettings.maxConnections) v + +-- | Set the number of pool stripes to use. A good setting is equal to the +-- number of CPU cores this codes is running on. +setPoolStripes :: Int -> Settings -> Settings +setPoolStripes v s + | v < 1 = error "Database.CQL.IO.Settings: stripes must be greater than 0" + | otherwise = set (poolSettings.poolStripes) v s + +-- | When receiving a response times out, we can no longer use the stream of the +-- connection that was used to make the request as it is uncertain if +-- a response will arrive later. Thus the bandwith of a connection will be +-- decreased. This settings defines a threshold after which we close the +-- connection to get a new one with all streams available. +setMaxTimeouts :: Int -> Settings -> Settings +setMaxTimeouts v = set (poolSettings.maxTimeouts) v + +----------------------------------------------------------------------------- +-- Connection Settings + +-- | Set the compression to use for frame body compression. +setCompression :: Compression -> Settings -> Settings +setCompression v = set (connSettings.compression) v + +-- | Set the maximum number of streams per connection. In version 2 of the +-- binary protocol at most 128 streams can be used. Version 3 supports up +-- to 32768 streams. +setMaxStreams :: Int -> Settings -> Settings +setMaxStreams v s = case s^.protoVersion of + V2 | v < 1 || v > 128 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 128]" + V3 | v < 1 || v > 32768 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 32768]" + _ -> set (connSettings.maxStreams) v s + +-- | Set the connect timeout of a connection. +setConnectTimeout :: NominalDiffTime -> Settings -> Settings +setConnectTimeout v = set (connSettings.connectTimeout) (Ms $ round (1000 * v)) + +-- | Set the send timeout of a connection. Request exceeding the send will +-- cause the connection to be closed and fail with 'ConnectionClosed' +-- exception. +setSendTimeout :: NominalDiffTime -> Settings -> Settings +setSendTimeout v = set (connSettings.sendTimeout) (Ms $ round (1000 * v)) + +-- | Set the receive timeout of a connection. Requests exceeding the +-- receive timeout will fail with a 'Timeout' exception. +setResponseTimeout :: NominalDiffTime -> Settings -> Settings +setResponseTimeout v = set (connSettings.responseTimeout) (Ms $ round (1000 * v)) + +-- | Set the default keyspace to use. Every new connection will be +-- initialised to use this keyspace. +setKeyspace :: Keyspace -> Settings -> Settings +setKeyspace v = set (connSettings.defKeyspace) (Just v) diff --git a/src/Database/CQL/IO/Signal.hs b/src/Database/CQL/IO/Signal.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Signal.hs @@ -0,0 +1,35 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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/. + +module Database.CQL.IO.Signal + ( Signal + , signal + , connect + , emit + , (|->) + , ($$) + ) where + +import Control.Applicative +import Control.Concurrent (forkIO) +import Data.IORef + +newtype Signal a = Sig (IORef [a -> IO ()]) + +signal :: IO (Signal a) +signal = Sig <$> newIORef [] + +connect :: Signal a -> (a -> IO ()) -> IO () +connect (Sig s) f = modifyIORef s (f:) + +infixl 2 |-> +(|->) :: Signal a -> (a -> IO ()) -> IO () +(|->) = connect + +emit :: Signal a -> a -> IO () +emit (Sig s) a = readIORef s >>= mapM_ (forkIO . ($ a)) + +infixr 1 $$ +($$) :: Signal a -> a -> IO () +($$) = emit diff --git a/src/Database/CQL/IO/Sync.hs b/src/Database/CQL/IO/Sync.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Sync.hs @@ -0,0 +1,55 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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/. + +module Database.CQL.IO.Sync + ( Sync + , create + , get + , put + , kill + , close + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Exception (SomeException, Exception, toException) + +data State a + = Empty + | Value !a + | Killed !SomeException + | Closed !SomeException + +newtype Sync a = Sync (TVar (State a)) + +create :: IO (Sync a) +create = Sync <$> newTVarIO Empty + +get :: Sync a -> IO a +get (Sync s) = atomically $ do + v <- readTVar s + case v of + Empty -> retry + Value a -> writeTVar s Empty >> return a + Closed x -> throwSTM x + Killed x -> throwSTM x + +put :: a -> Sync a -> IO Bool +put a (Sync s) = atomically $ do + v <- readTVar s + case v of + Empty -> writeTVar s (Value a) >> return True + Closed _ -> return True + _ -> writeTVar s Empty >> return False + +kill :: Exception e => e -> Sync a -> IO () +kill x (Sync s) = atomically $ do + v <- readTVar s + case v of + Closed _ -> return () + _ -> writeTVar s (Killed $ toException x) + +close :: Exception e => e -> Sync a -> IO () +close x (Sync s) = atomically $ writeTVar s (Closed $ toException x) + diff --git a/src/Database/CQL/IO/Tickets.hs b/src/Database/CQL/IO/Tickets.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Tickets.hs @@ -0,0 +1,46 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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/. + +module Database.CQL.IO.Tickets + ( Ticket + , toInt + , Pool + , pool + , close + , get + , markAvailable + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Exception (SomeException, Exception, toException) +import Data.Set (Set) + +import qualified Data.Set as Set + +newtype Ticket = Ticket { toInt :: Int } deriving (Eq, Ord, Show) + +newtype Pool = Pool (TVar (Either SomeException (Set Ticket))) + +pool :: Int -> IO Pool +pool n = Pool <$> newTVarIO (Right . Set.fromList $ map Ticket [0 .. n-1]) + +close :: Exception e => e -> Pool -> IO () +close x (Pool p) = atomically $ writeTVar p (Left $ toException x) + +get :: Pool -> IO Ticket +get (Pool p) = atomically $ readTVar p >>= popHead + where + popHead (Left x) = throwSTM x + popHead (Right x) + | Set.null x = retry + | otherwise = do + let (t, tt) = Set.deleteFindMin x + writeTVar p (Right tt) + return t + +markAvailable :: Pool -> Int -> IO () +markAvailable (Pool p) t = + atomically $ modifyTVar' p (fmap (Set.insert (Ticket t))) + diff --git a/src/Database/CQL/IO/Timeouts.hs b/src/Database/CQL/IO/Timeouts.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Timeouts.hs @@ -0,0 +1,72 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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/. + +module Database.CQL.IO.Timeouts + ( TimeoutManager + , create + , destroy + , Action + , Milliseconds (..) + , add + , cancel + , withTimeout + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Exception (mask_, bracket) +import Control.Reaper +import Control.Monad +import Database.CQL.IO.Types (Milliseconds (..), ignore) + +data TimeoutManager = TimeoutManager + { roundtrip :: !Int + , reaper :: Reaper [Action] Action + } + +data Action = Action + { action :: IO () + , state :: TVar State + } + +data State = Running !Int | Canceled + +create :: Milliseconds -> IO TimeoutManager +create (Ms n) = TimeoutManager n <$> mkReaper defaultReaperSettings + { reaperAction = mkListAction prune + , reaperDelay = n * 1000 + } + where + prune a = do + s <- atomically $ do + x <- readTVar (state a) + writeTVar (state a) (newState x) + return x + case s of + Running 0 -> do + ignore (action a) + return Nothing + Canceled -> return Nothing + _ -> return $ Just a + + newState (Running k) = Running (k - 1) + newState s = s + +destroy :: TimeoutManager -> Bool -> IO () +destroy tm exec = mask_ $ do + a <- reaperStop (reaper tm) + when exec $ + mapM_ (ignore . action) a + +add :: TimeoutManager -> Milliseconds -> IO () -> IO Action +add tm (Ms n) a = do + r <- Action a <$> newTVarIO (Running $ n `div` roundtrip tm) + reaperAdd (reaper tm) r + return r + +cancel :: Action -> IO () +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/CQL/IO/Types.hs b/src/Database/CQL/IO/Types.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Types.hs @@ -0,0 +1,136 @@ +-- This Source Code Form is subject to the terms of the Mozilla Public +-- 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 DeriveDataTypeable #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StandaloneDeriving #-} + +module Database.CQL.IO.Types where + +import Control.Monad.Catch +import Data.IP +import Data.Typeable +import Database.CQL.Protocol (Event, Response, CompressionAlgorithm) +import Network.Socket (SockAddr (..), PortNumber) +import System.Logger.Message + +type EventHandler = Event -> IO () + +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 + +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 + +ip2inet :: PortNumber -> IP -> InetAddr +ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a) +ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0 + +inet2ip :: InetAddr -> IP +inet2ip (InetAddr (SockAddrInet _ a)) = IPv4 (fromHostAddress a) +inet2ip (InetAddr (SockAddrInet6 _ _ a _)) = IPv6 (fromHostAddress6 a) +inet2ip _ = error "inet2Ip: not IP4/IP6 address" + +----------------------------------------------------------------------------- +-- InvalidSettings + +data InvalidSettings + = UnsupportedCompression [CompressionAlgorithm] + | InvalidCacheSize + deriving Typeable + +instance Exception InvalidSettings + +instance Show InvalidSettings where + show (UnsupportedCompression cc) = "cql-io: unsupported compression: " ++ show cc + show InvalidCacheSize = "cql-io: invalid cache size" + +----------------------------------------------------------------------------- +-- InternalError + +data InternalError = InternalError String + deriving Typeable + +instance Exception InternalError + +instance Show InternalError where + show (InternalError e) = "cql-io: internal error: " ++ show e + +----------------------------------------------------------------------------- +-- HostError + +data HostError + = NoHostAvailable + | HostsBusy + deriving Typeable + +instance Exception HostError + +instance Show HostError where + show NoHostAvailable = "cql-io: no host available" + show HostsBusy = "cql-io: hosts busy" + +----------------------------------------------------------------------------- +-- ConnectionError + +data ConnectionError + = ConnectionClosed !InetAddr + | ConnectTimeout !InetAddr + deriving Typeable + +instance Exception ConnectionError + +instance Show ConnectionError where + show (ConnectionClosed i) = "cql-io: connection closed: " ++ show i + show (ConnectTimeout i) = "cql-io: connect timeout: " ++ show i + +----------------------------------------------------------------------------- +-- Timeout + +data Timeout = TimeoutRead !String + deriving Typeable + +instance Exception Timeout + +instance Show Timeout where + show (TimeoutRead e) = "cql-io: read timeout: " ++ e + +----------------------------------------------------------------------------- +-- UnexpectedResponse + +data UnexpectedResponse where + UnexpectedResponse :: UnexpectedResponse + UnexpectedResponse' :: Show b => Response k a b -> UnexpectedResponse + +deriving instance Typeable UnexpectedResponse +instance Exception UnexpectedResponse + +instance Show UnexpectedResponse where + show UnexpectedResponse = "cql-io: unexpected response" + show (UnexpectedResponse' r) = "cql-io: unexpected response: " ++ show r + +ignore :: IO () -> IO () +ignore a = catchAll a (const $ return ()) +{-# INLINE ignore #-}