diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,10 @@ # Revision history for mysql-haskell +## 0.5.1.0 -- 2016-10-20 + +* Add `queryVector`, `queryVector_` and `queryStmtVector`. +* Use binary-parsers to speed up binary parsers. + ## 0.5.0.0 -- 2016-8-22 * Export exception types. diff --git a/Database/MySQL/Base.hs b/Database/MySQL/Base.hs --- a/Database/MySQL/Base.hs +++ b/Database/MySQL/Base.hs @@ -38,12 +38,15 @@ , executeMany , execute_ , query_ + , queryVector_ , query + , queryVector -- * prepared query statement , prepareStmt , prepareStmtDetail , executeStmt , queryStmt + , queryStmtVector , closeStmt , resetStmt -- * helpers @@ -82,6 +85,7 @@ import Database.MySQL.Query import System.IO.Streams (InputStream, OutputStream) import qualified System.IO.Streams as Stream +import qualified Data.Vector as V -------------------------------------------------------------------------------- @@ -123,6 +127,13 @@ query :: MySQLConn -> Query -> [MySQLValue] -> IO ([ColumnDef], InputStream [MySQLValue]) query conn qry params = query_ conn (renderParams qry params) +-- | 'V.Vector' version of 'query'. +-- +-- @since 0.5.1.0 +-- +queryVector :: MySQLConn -> Query -> [MySQLValue] -> IO (V.Vector ColumnDef, InputStream (V.Vector MySQLValue)) +queryVector conn qry params = queryVector_ conn (renderParams qry params) + -- | Execute a MySQL query which return a result-set. -- query_ :: MySQLConn -> Query -> IO ([ColumnDef], InputStream [MySQLValue]) @@ -144,6 +155,29 @@ | otherwise -> Just <$> getFromPacket (getTextRow fields) q return (fields, rows) +-- | 'V.Vector' version of 'query_'. +-- +-- @since 0.5.1.0 +-- +queryVector_ :: MySQLConn -> Query -> IO (V.Vector ColumnDef, InputStream (V.Vector MySQLValue)) +queryVector_ conn@(MySQLConn is os _ consumed) (Query qry) = do + guardUnconsumed conn + writeCommand (COM_QUERY qry) os + p <- readPacket is + if isERR p + then decodeFromPacket p >>= throwIO . ERRException + else do + len <- getFromPacket getLenEncInt p + fields <- V.replicateM len $ (decodeFromPacket <=< readPacket) is + _ <- readPacket is -- eof packet, we don't verify this though + writeIORef consumed False + rows <- Stream.makeInputStream $ do + q <- readPacket is + if | isEOF q -> writeIORef consumed True >> return Nothing + | isERR q -> decodeFromPacket q >>= throwIO . ERRException + | otherwise -> Just <$> getFromPacket (getTextRowVector fields) q + return (fields, rows) + -- | Ask MySQL to prepare a query statement. -- prepareStmt :: MySQLConn -> Query -> IO StmtID @@ -220,7 +254,31 @@ writeIORef consumed False rows <- Stream.makeInputStream $ do q <- readPacket is - if | isOK q -> Just <$> getFromPacket (getBinaryRow fields len) q -- fast path for decode rows + if | isOK q -> Just <$> getFromPacket (getBinaryRow fields len) q + | isEOF q -> writeIORef consumed True >> return Nothing + | isERR q -> decodeFromPacket q >>= throwIO . ERRException + | otherwise -> throwIO (UnexpectedPacket q) + return (fields, rows) + +-- | 'V.Vector' version of 'queryStmt' +-- +-- @since 0.5.1.0 +-- +queryStmtVector :: MySQLConn -> StmtID -> [MySQLValue] -> IO (V.Vector ColumnDef, InputStream (V.Vector MySQLValue)) +queryStmtVector conn@(MySQLConn is os _ consumed) stid params = do + guardUnconsumed conn + writeCommand (COM_STMT_EXECUTE stid params) os + p <- readPacket is + if isERR p + then decodeFromPacket p >>= throwIO . ERRException + else do + len <- getFromPacket getLenEncInt p + fields <- V.replicateM len $ (decodeFromPacket <=< readPacket) is + _ <- readPacket is -- eof packet, we don't verify this though + writeIORef consumed False + rows <- Stream.makeInputStream $ do + q <- readPacket is + if | isOK q -> Just <$> getFromPacket (getBinaryRowVector fields len) q | isEOF q -> writeIORef consumed True >> return Nothing | isERR q -> decodeFromPacket q >>= throwIO . ERRException | otherwise -> throwIO (UnexpectedPacket q) diff --git a/Database/MySQL/BinLogProtocol/BinLogEvent.hs b/Database/MySQL/BinLogProtocol/BinLogEvent.hs --- a/Database/MySQL/BinLogProtocol/BinLogEvent.hs +++ b/Database/MySQL/BinLogProtocol/BinLogEvent.hs @@ -110,14 +110,14 @@ case pushEndOfInput $ pushChunks (runGetIncremental g) body of Done _ _ r -> return r Fail buf offset errmsg -> throwIO (DecodePacketFailed buf offset errmsg) - Partial _ -> throwIO DecodePacketPartial + _ -> error "getFromBinLogPacket: impossible!" getFromBinLogPacket' :: (BinLogEventType -> Get a) -> BinLogPacket -> IO a getFromBinLogPacket' g (BinLogPacket _ typ _ _ _ _ body _ ) = case pushEndOfInput $ pushChunks (runGetIncremental (g typ)) body of Done _ _ r -> return r Fail buf offset errmsg -> throwIO (DecodePacketFailed buf offset errmsg) - Partial _ -> throwIO DecodePacketPartial + _ -> error "getFromBinLogPacket': impossible!" -------------------------------------------------------------------------------- diff --git a/Database/MySQL/Connection.hs b/Database/MySQL/Connection.hs --- a/Database/MySQL/Connection.hs +++ b/Database/MySQL/Connection.hs @@ -80,14 +80,14 @@ -- | Establish a MySQL connection with 'Greeting' back, so you can find server's version .etc. -- connectDetail :: ConnectInfo -> IO (Greeting, MySQLConn) -connectDetail ci@(ConnectInfo host port _ _ _) = +connectDetail (ConnectInfo host port db user pass) = bracketOnError (TCP.connectWithBufferSize host port bUFSIZE) (\(_, _, sock) -> N.close sock) $ \ (is, os, sock) -> do is' <- decodeInputStream is os' <- Binary.encodeOutputStream os p <- readPacket is' greet <- decodeFromPacket p - let auth = mkAuth ci greet + let auth = mkAuth db user pass greet Stream.write (Just (encodeToPacket 1 auth)) os' q <- readPacket is' if isOK q @@ -97,8 +97,8 @@ return (greet, conn) else Stream.write Nothing os' >> decodeFromPacket q >>= throwIO . ERRException -mkAuth :: ConnectInfo -> Greeting -> Auth -mkAuth (ConnectInfo _ _ db user pass) greet = +mkAuth :: ByteString -> ByteString -> ByteString -> Greeting -> Auth +mkAuth db user pass greet = let salt = greetingSalt1 greet `B.append` greetingSalt2 greet scambleBuf = scramble salt pass in Auth clientCap clientMaxPacketSize clientCharset user scambleBuf db diff --git a/Database/MySQL/OpenSSL.hs b/Database/MySQL/OpenSSL.hs --- a/Database/MySQL/OpenSSL.hs +++ b/Database/MySQL/OpenSSL.hs @@ -41,7 +41,7 @@ connect c cp = fmap snd (connectDetail c cp) connectDetail :: ConnectInfo -> (Session.SSLContext, String) -> IO (Greeting, MySQLConn) -connectDetail ci@(ConnectInfo host port _ _ _) (ctx, subname) = +connectDetail (ConnectInfo host port db user pass) (ctx, subname) = bracketOnError (TCP.connectWithBufferSize host port bUFSIZE) (\(_, _, sock) -> N.close sock) $ \ (is, os, sock) -> do is' <- decodeInputStream is @@ -62,7 +62,7 @@ (sslIs, sslOs) <- SSL.sslToStreams ssl sslIs' <- decodeInputStream sslIs sslOs' <- Binary.encodeOutputStream sslOs - let auth = mkAuth ci greet + let auth = mkAuth db user pass greet Stream.write (Just (encodeToPacket 2 auth)) sslOs' q <- readPacket sslIs' if isOK q diff --git a/Database/MySQL/Protocol/ColumnDef.hs b/Database/MySQL/Protocol/ColumnDef.hs --- a/Database/MySQL/Protocol/ColumnDef.hs +++ b/Database/MySQL/Protocol/ColumnDef.hs @@ -57,6 +57,7 @@ <*> getWord16le -- flags <*> getWord8 -- decimals <* skip 2 -- const 0x00 0x00 +{-# INLINE getField #-} putField :: ColumnDef -> Put putField (ColumnDef db tbl otbl name oname charset len typ flags dec) = do @@ -72,6 +73,7 @@ putWord16le flags putWord8 dec putWord16le 0X0000 +{-# INLINE putField #-} instance Binary ColumnDef where get = getField @@ -123,9 +125,11 @@ getFieldType :: Get FieldType getFieldType = FieldType <$> getWord8 +{-# INLINE getFieldType #-} putFieldType :: FieldType -> Put putFieldType (FieldType t) = putWord8 t +{-# INLINE putFieldType #-} instance Binary FieldType where get = getFieldType diff --git a/Database/MySQL/Protocol/Command.hs b/Database/MySQL/Protocol/Command.hs --- a/Database/MySQL/Protocol/Command.hs +++ b/Database/MySQL/Protocol/Command.hs @@ -104,3 +104,4 @@ skip 1 -- reserved wc <- fromIntegral <$> getWord16le return (StmtPrepareOK stmtid cc pc wc) +{-# INLINE getStmtPrepareOK #-} diff --git a/Database/MySQL/Protocol/MySQLValue.hs b/Database/MySQL/Protocol/MySQLValue.hs --- a/Database/MySQL/Protocol/MySQLValue.hs +++ b/Database/MySQL/Protocol/MySQLValue.hs @@ -18,7 +18,7 @@ import qualified Blaze.Text as Textual import Control.Applicative import Control.Monad -import Data.Binary.Get +import Data.Binary.Parser import Data.Binary.Put import Data.Bits import Data.ByteString (ByteString) @@ -47,6 +47,7 @@ import Database.MySQL.Protocol.Escape import Database.MySQL.Protocol.Packet import GHC.Generics (Generic) +import qualified Data.Vector as V -------------------------------------------------------------------------------- -- | Data type mapping between MySQL values and haskell values. @@ -240,12 +241,20 @@ -- | Text row decoder getTextRow :: [ColumnDef] -> Get [MySQLValue] getTextRow fs = forM fs $ \ f -> do - p <- lookAhead getWord8 + p <- peek if p == 0xFB then skip 1 >> return MySQLNull else getTextField f {-# INLINE getTextRow #-} +getTextRowVector :: V.Vector ColumnDef -> Get (V.Vector MySQLValue) +getTextRowVector fs = V.forM fs $ \ f -> do + p <- peek + if p == 0xFB + then skip 1 >> return MySQLNull + else getTextField f +{-# INLINE getTextRowVector #-} + -------------------------------------------------------------------------------- -- | Binary protocol decoder getBinaryField :: ColumnDef -> Get MySQLValue @@ -460,18 +469,28 @@ go :: [ColumnDef] -> ByteString -> Int -> Get [MySQLValue] go [] _ _ = pure [] go (f:fs) nullmap pos = do - r <- if isNull nullmap pos + r <- if isNull_ nullmap pos then return MySQLNull else getBinaryField f let pos' = pos + 1 rest <- pos' `seq` go fs nullmap pos' return (r `seq` (r : rest)) - - isNull nullmap pos = -- This 'isNull' is special for offset = 2 - let (i, j) = (pos + 2) `quotRem` 8 - in (nullmap `B.unsafeIndex` i) `testBit` j - {-# INLINE isNull #-} {-# INLINE getBinaryRow #-} + +getBinaryRowVector :: V.Vector ColumnDef -> Int -> Get (V.Vector MySQLValue) +getBinaryRowVector fields flen = do + skip 1 -- 0x00 + let maplen = (flen + 7 + 2) `shiftR` 3 + nullmap <- getByteString maplen + (`V.imapM` fields) $ \ pos f -> + if isNull_ nullmap pos then return MySQLNull else getBinaryField f +{-# INLINE getBinaryRowVector #-} + +-- This 'isNull_' is special for offset = 2 +isNull_ nullmap pos = + let (i, j) = (pos + 2) `quotRem` 8 + in (nullmap `B.unsafeIndex` i) `testBit` j +{-# INLINE isNull_ #-} -------------------------------------------------------------------------------- -- | Use 'ByteString' to present a bitmap. diff --git a/Database/MySQL/Protocol/Packet.hs b/Database/MySQL/Protocol/Packet.hs --- a/Database/MySQL/Protocol/Packet.hs +++ b/Database/MySQL/Protocol/Packet.hs @@ -17,15 +17,17 @@ import Control.Applicative import Control.Exception (Exception (..), throwIO) -import Data.Binary -import Data.Binary.Get +import Data.Binary.Parser import Data.Binary.Put +import Data.Binary (Binary(..), encode) import Data.Bits import qualified Data.ByteString as B import Data.ByteString.Char8 as BC import qualified Data.ByteString.Lazy as L +import qualified Data.ByteString.Lazy.Internal as L import Data.Int.Int24 import Data.Int +import Data.Word import Data.Typeable import Data.Word.Word24 @@ -76,22 +78,23 @@ -- if we successful parse a packet, then we don't care if there're bytes left. -- decodeFromPacket :: Binary a => Packet -> IO a -decodeFromPacket (Packet _ _ body) = case pushEndOfInput $ pushChunks (runGetIncremental get) body of - Done _ _ r -> return r - Fail buf offset errmsg -> throwIO (DecodePacketFailed buf offset errmsg) - Partial _ -> throwIO DecodePacketPartial +decodeFromPacket = getFromPacket get {-# INLINE decodeFromPacket #-} getFromPacket :: Get a -> Packet -> IO a -getFromPacket g (Packet _ _ body) = case pushEndOfInput $ pushChunks (runGetIncremental g) body of - Done _ _ r -> return r - Fail buf offset errmsg -> throwIO (DecodePacketFailed buf offset errmsg) - Partial _ -> throwIO DecodePacketPartial +getFromPacket g (Packet _ _ body) = + case body of + (L.Chunk bs lbs) -> case pushEndOfInput (pushChunks (parse g bs) lbs) of + Done _ _ r -> return r + Fail buf offset errmsg -> throwIO (DecodePacketFailed buf offset errmsg) + _ -> error "getFromPacket: impossible error!" + L.Empty -> case pushEndOfInput (parse g B.empty) of + Done _ _ r -> return r + Fail buf offset errmsg -> throwIO (DecodePacketFailed buf offset errmsg) + _ -> error "getFromPacket: impossible error!" {-# INLINE getFromPacket #-} -data DecodePacketException - = DecodePacketFailed ByteString ByteOffset String - | DecodePacketPartial +data DecodePacketException = DecodePacketFailed ByteString ByteOffset String deriving (Typeable, Show) instance Exception DecodePacketException @@ -128,6 +131,7 @@ <*> getLenEncInt <*> getWord16le <*> getWord16le +{-# INLINE getOK #-} putOK :: OK -> Put putOK (OK row lid stat wcnt) = do @@ -136,6 +140,7 @@ putLenEncInt lid putWord16le stat putWord16le wcnt +{-# INLINE putOK #-} instance Binary OK where get = getOK @@ -155,6 +160,7 @@ <* skip 1 <*> getByteString 5 <*> getRemainingByteString +{-# INLINE getERR #-} putERR :: ERR -> Put putERR (ERR code stat msg) = do @@ -163,6 +169,7 @@ putWord8 35 -- '#' putByteString stat putByteString msg +{-# INLINE putERR #-} instance Binary ERR where get = getERR @@ -179,12 +186,14 @@ getEOF = EOF <$ skip 1 <*> getWord16le <*> getWord16le +{-# INLINE getEOF #-} putEOF :: EOF -> Put putEOF (EOF wcnt stat) = do putWord8 0xFE putWord16le wcnt putWord16le stat +{-# INLINE putEOF #-} instance Binary EOF where get = getEOF diff --git a/Database/MySQL/TLS.hs b/Database/MySQL/TLS.hs --- a/Database/MySQL/TLS.hs +++ b/Database/MySQL/TLS.hs @@ -39,7 +39,7 @@ connect c cp = fmap snd (connectDetail c cp) connectDetail :: ConnectInfo -> (TLS.ClientParams, String) -> IO (Greeting, MySQLConn) -connectDetail ci@(ConnectInfo host port _ _ _) (cparams, subName) = +connectDetail (ConnectInfo host port db user pass) (cparams, subName) = bracketOnError (TCP.connectWithBufferSize host port bUFSIZE) (\(_, _, sock) -> N.close sock) $ \ (is, os, sock) -> do is' <- decodeInputStream is @@ -58,7 +58,7 @@ (tlsIs, tlsOs) <- TLS.tlsToStreams ctx tlsIs' <- decodeInputStream tlsIs tlsOs' <- Binary.encodeOutputStream tlsOs - let auth = mkAuth ci greet + let auth = mkAuth db user pass greet Stream.write (Just (encodeToPacket 2 auth)) tlsOs' q <- readPacket tlsIs' if isOK q diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2016, winterland1989 +Copyright (c) 2016, Winterland All rights reserved. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -83,7 +83,11 @@ main :: IO () main = do conn <- MySQL.connect - MySQL.defaultConnectInfo {ciUser = "username", ciPassword = "password", ciDatabase = "dbname"} + MySQL.defaultConnectInfo + { MySQL.ciUser = "username" + , MySQL.ciPassword = "password" + , MySQL.ciDatabase = "dbname" + } MySQL.getLastBinLogTracker conn >>= \ case Just tracker -> do es <- MySQL.decodeRowBinLogEvent =<< MySQL.dumpBinLog conn 1024 tracker False @@ -139,10 +143,10 @@ Enter benchmark directory and run `./bench.sh` to benchmark 1) c++ version 2) mysql-haskell 3) FFI version mysql, you may need to: -+ modify `bench.sh`(change the include path) to get c++ version compiled. -+ modify `mysql-haskell-bench.cabal`(change the openssl's lib path) to get haskell version compiled. -+ setup MySQL's TLS support, modify `MySQLHaskellOpenSSL.hs/MySQLHaskellTLS.hs` to change the CA file's path, and certificate's subject name. -+ adjust rts options `-N` to get best results. ++ Modify `bench.sh`(change the include path) to get c++ version compiled. ++ Modify `mysql-haskell-bench.cabal`(change the openssl's lib path) to get haskell version compiled. ++ Setup MySQL's TLS support, modify `MySQLHaskellOpenSSL.hs/MySQLHaskellTLS.hs` to change the CA file's path, and certificate's subject name. ++ Adjust rts options `-N` to get best results. With `-N10` on my company's 24-core machine, binary protocol performs almost identical to c version! diff --git a/mysql-haskell.cabal b/mysql-haskell.cabal --- a/mysql-haskell.cabal +++ b/mysql-haskell.cabal @@ -1,5 +1,5 @@ name: mysql-haskell -version: 0.5.0.0 +version: 0.5.1.0 synopsis: pure haskell MySQL driver description: pure haskell MySQL driver license: BSD3 @@ -45,9 +45,10 @@ , monad-loops == 0.4.* , network >= 2.3 && < 3.0 , io-streams >= 1.2 && < 2.0 - , tcp-streams == 0.4.* - , wire-streams >= 0.0.2 && < 0.1 + , tcp-streams >= 0.4 && < 0.6 + , wire-streams >= 0.1 , binary >= 0.8.4 + , binary-parsers >= 0.2.1 , bytestring >= 0.10.2.0 , text >= 1.1 && < 1.3 , cryptonite == 0.* @@ -56,20 +57,12 @@ , scientific == 0.3.* , bytestring-lexing == 0.5.* , blaze-textual == 0.2.* - , word24 == 1.* - , tls >=1.3.5 && < 1.4 + , word24 >= 1.0 && <= 3.0 + , tls >= 1.3.5 && < 1.4 + , vector >= 0.8 if flag(openssl) build-depends: HsOpenSSL >=0.10.3 && <0.12 - if os(mingw32) || os(windows) - extra-libraries: eay32, ssl32 - else - if os(osx) - extra-libraries: crypto - extra-lib-dirs: /usr/local/lib - include-dirs: /usr/local/include - else - extra-libraries: crypto default-language: Haskell2010 default-extensions: DeriveDataTypeable @@ -89,16 +82,16 @@ , TextRowNew , ExecuteMany hs-source-dirs: test - build-depends: - mysql-haskell - , base - , bytestring - , optparse-applicative < 0.13 - , tasty == 0.11.* - , tasty-hunit - , text - , io-streams - , time + build-depends: mysql-haskell + , base + , bytestring + , optparse-applicative < 0.13 + , tasty == 0.11.* + , tasty-hunit + , text + , io-streams + , time + , vector default-extensions: MultiWayIf , OverloadedStrings diff --git a/test/BinaryRow.hs b/test/BinaryRow.hs --- a/test/BinaryRow.hs +++ b/test/BinaryRow.hs @@ -13,6 +13,7 @@ import Test.Tasty.HUnit import qualified Data.Text as T import qualified Data.ByteString as B +import qualified Data.Vector as V tests :: MySQLConn -> Assertion tests c = do @@ -155,8 +156,14 @@ , MySQLText "韩冬真赞" , MySQLText "foo" , MySQLText "foo,bar"] - Stream.skipToEof is + + (_, is') <- queryStmtVector c selStmt [] + Just v' <- Stream.read is' + Stream.skipToEof is' + + assertEqual "decode binary protocol(queryStmtVector)" v (V.toList v') + -------------------------------------------------------------------------------- updStmt <- prepareStmt c "UPDATE test SET \ diff --git a/test/TextRow.hs b/test/TextRow.hs --- a/test/TextRow.hs +++ b/test/TextRow.hs @@ -9,6 +9,7 @@ import Database.MySQL.Base import qualified System.IO.Streams as Stream import Test.Tasty.HUnit +import qualified Data.Vector as V tests :: MySQLConn -> Assertion tests c = do @@ -153,6 +154,12 @@ ] Stream.skipToEof is + + (_, is') <- queryVector_ c "SELECT * FROM test" + Just v' <- Stream.read is' + Stream.skipToEof is' + + assertEqual "decode text protocol(queryVector_)" v (V.toList v') execute c "UPDATE test SET \ \__bit = ? ,\