memcache 0.2.0.1 → 0.4.2.0
raw patch · 23 files changed
Files
- CHANGELOG.md +146/−25
- Database/Memcache/Auth.hs +62/−0
- Database/Memcache/Client.hs +242/−218
- Database/Memcache/Cluster.hs +184/−97
- Database/Memcache/ElastiCacheClient.hs +190/−0
- Database/Memcache/Errors.hs +21/−23
- Database/Memcache/SASL.hs +0/−69
- Database/Memcache/Server.hs +101/−74
- Database/Memcache/Socket.hs +354/−381
- Database/Memcache/Types.hs +32/−175
- Database/Memcache/Types/Authentication.hs +32/−0
- Database/Memcache/Types/ServerSpec.hs +89/−0
- Database/Memcache/Util.hs +18/−0
- LICENSE +1/−1
- README.md +174/−33
- Setup.hs +0/−7
- TODO.md +22/−21
- bench/Parser.hs +30/−51
- memcache.cabal +130/−108
- test-integration/Integration.hs +304/−0
- test/Full.hs +0/−104
- test/MockServer.hs +142/−0
- test/Spec.hs +720/−0
CHANGELOG.md view
@@ -1,47 +1,168 @@+# 0.4.2.0 - October 20th, 2026++No API changes from 0.4.0.0. This is the first Hackage release since 0.4.0.0,+so it also includes the 0.4.1.0 changes below.++- Fix `getMany` returning one key's value under another key's name when a+ plain key equals the other's base64 form, such as `"IA=="` and `" "`.+- Requests that carry a value or a number are now encoded into a single+ buffer instead of a chain of appends that copied the value once per piece.+ Encoding a 64KiB `set` takes about 4µs instead of about 57µs.+- Document that a `Version` of `0` skips the CAS check, what `cas`, `replace`,+ `append`, `prepend` and `delete` do for a missing key or a version mismatch,+ and when `increment` and `decrement` create a missing key. Add the missing+ `OverloadedStrings` to the usage examples in `Database.Memcache.Client` and+ the package description, add usage and development instructions to the+ README, and replace the template `SECURITY.md` with a full security policy.++# 0.4.1.0 - October 19th, 2026++Not uploaded to Hackage; these changes first ship there in 0.4.2.0. No API or+behaviour changes from 0.4.0.0.++- Reading a large value no longer allocates memory quadratic in its size. Each+ receive is now capped at 64KiB, so fetching a 32MiB value allocates about+ 67MB rather than about 1.7GB.+- Drop the dependency on `errors` (and, through it, `safe` and+ `transformers-compat`), which was only used for a one-line helper.+- Replace the hpack `package.yaml` with a hand-written `memcache.cabal`+ (`cabal-version: 3.0`) that lists the exposed modules explicitly and gives+ each component only the dependencies it uses. Remove `Setup.hs`, which only+ did what `build-type: Simple` does anyway. The license field is now the SPDX+ identifier `BSD-3-Clause`; the license itself is unchanged.+- Declare support for GHC 9.2 to 9.14 in `tested-with`.+- Fix the `parser` benchmark, which no longer compiled, and extend it to cover+ key encoding, more requests, and reading CAS, flags and numbers out of+ responses.+- Expand the package description. Hackage already shows it, as a revision of+ 0.4.0.0.+- Document the `InvalidAuthentication` and `KeyTooLong` constructors of+ `ClientError`, and fix typos in the Haddock documentation.+- Greatly expand the test suite. Unit tests against mock servers now cover+ every request encoder and client operation, retries and failover,+ authentication, `parseServerSpec` and ElastiCache discovery, and the+ integration tests also run against memcached in `-Y` auth-file mode and+ against a three-server cluster.+- CI builds and tests with cabal on GHC 9.2 to 9.14 and from the source+ distribution, builds the library against the oldest dependency versions the+ bounds allow, and runs weekly, as well as building with Stack and running+ HLint and weeder.++# 0.4.0.0 - October 19th, 2026++- Replace the deprecated binary protocol with buffered Meta Text requests.+- Add base64 binary-key support, `getMany`, hspec tests, and `Auth`.+- Remove binary protocol and builder dependencies.+- Replace request/response wrapper types with direct Meta Text encoders and `Response`.+- Authentication now uses the server's experimental `-Y authfile` mode+ (introduced in memcached 1.5.15) instead of binary `-S` SASL.+- `encodeKey` now returns a validated `WireKey`; request builders and the+ low-level `keyedOp` APIs accept encoded keys to avoid partial validation and+ repeated encoding.+- Requires memcached 1.6.0 or newer for the meta commands.+- `stats` returns `[(Server, StatResults)]` rather than+ `[(Server, Maybe StatResults)]`; the `Nothing` case was unreachable. Stat+ values containing spaces are no longer truncated, and stats arguments may+ now contain spaces (`stats cachedump 1 100`).+- `Status` loses `ErrInvalidArgs` and `ErrUnknownCommand`, which have no Meta+ Text equivalent and now surface as `ProtocolError`. `ErrValueTooLarge`,+ `ErrOutOfMemory` and `ErrValueNonNumeric` are recovered from the server's+ error text so they keep being reported as `OpError`.+- A command the server rejects (`ERROR`, `CLIENT_ERROR`) or answers with an+ operation status is no longer retried, and no longer counts towards marking+ a server dead.+- `getMany` writes each batch in a single bounded send, so a large key list+ cannot deadlock against the server's replies.+- Response lines are capped at 8KiB, bounding memory use on a desynchronised+ connection.+- Authentication first checks that the server is in auth-file mode and+ refuses to send credentials otherwise, so they are never stored as a plain+ `auth` item. Credentials with a forbidden delimiter, or a server that isn't+ in auth-file mode, raise `ClientError InvalidAuthentication`, and+ credentials the server rejects raise `OpError TextAuthFail`. Neither is+ retried or marks the server dead.+- Empty keys, and keys longer than 250 bytes, are rejected with+ `ClientError KeyTooLong` before anything is sent.+- `append` and `prepend` on a missing key return `Nothing` instead of throwing+ `OpError ErrItemNotStored`.+- `stats` accepts replies made of lines other than `STAT`, such as those from+ `stats reset` and `stats cachedump`.+- `Database.Memcache.Socket` sends and receives on a buffered `Connection`+ with `sendRaw` and `recvResponse`, replacing `send` and `recv`.+- ElastiCache auto discovery no longer holds the server list lock while+ waiting for the next refresh.+- Exception helpers now come from `unliftio`, which is a new dependency.+- Library dependencies declare lower bounds, set to the oldest versions CI+ builds against (GHC 9.2, Stackage LTS 20.26).+- Depend on `data-default` instead of the deprecated `data-default-class`,+ which is now a compatibility shim over `data-default` 0.8.+- Add `Database.Memcache.ElastiCacheClient`, which creates a client from an+ Amazon ElastiCache configuration endpoint and refreshes its node list on an+ interval.+- Add `parseServerSpec`, which reads a `ServerSpec` from a `memcached://` URI.+- `Options` gains `optsServerOptions`, whose `ServerOptions` set the size,+ stripes and keep-alive of each server's connection pool, and+ `optsServerSpecsToServers`, which controls how servers are created.+ `Database.Memcache.Server` exports `newServerDefault` in place of+ `newServer`.++# 0.3.0.2 - March 27th, 2024++- Make the key hashing algorithm configurable by clients -- that is, let+ clients provide a function mapping a key to a server.++# 0.3.0.1 - January 17th, 2021++- No changes, needed a new release to fix a Hackage upload issue.++# 0.3.0.0 - January 17th, 2021++- Bump package dependencies for newer GHC/back/network.+- Update code to work with newer dependencies.+ # 0.2.0.1 - November 2nd, 2016 -* Fix compatability with latest `data-default-class`-* Add new ReqRaw type for external clients to implement custom requests. Quite+- Fix compatibility with latest `data-default-class`+- Add new ReqRaw type for external clients to implement custom requests. Quite a hack right now, so behind a WARNING pragma. # 0.2.0.0 - May 27th, 2016 -* Big design change to reduce code duplication (`Protocol` module gone).-* Remove `Options` type - just fixed configuration for now.-* Design change also allows proper retry handling on operation failure - we+- Big design change to reduce code duplication (`Protocol` module gone).+- Remove `Options` type - just fixed configuration for now.+- Design change also allows proper retry handling on operation failure - we retry an operation against the same server, but after N consecutive failures, we mark the server as dead and don't try using it again until M seconds has passed.-* Simplify exception hierachy - just one type `MemcacheError` now for-* Remove `defaultOptions` and `defaultServerSpec`, will revist usefulness.+- Simplify exception hierarchy - just one type `MemcacheError` now for exceptions.-* Remove many `Typeable` instances.-* Support better testing with a mock Memcached server.-* Fix bug in socket handling - detected EOF properly.-* Greatly improve documentation.-* Use `data-default-class` for defaults of servers and options.+- Remove `defaultOptions` and `defaultServerSpec`, will revisit usefulness.+- Remove many `Typeable` instances.+- Support better testing with a mock Memcached server.+- Fix bug in socket handling - detected EOF properly.+- Greatly improve documentation.+- Use `data-default-class` for defaults of servers and options. # 0.1.0.1 - February 26th, 2016 -* Consistent usage of 'memcached' instead of 'memcache'.-* Document `Database.Memcache.Client`.-* Add inline pragmas in appropriate places.-* Fix bug handling fragmented IP packets (Alfredo Di Napoli).+- Consistent usage of 'memcached' instead of 'memcache'.+- Document `Database.Memcache.Client`.+- Add inline pragmas in appropriate places.+- Fix bug handling fragmented IP packets (Alfredo Di Napoli). # 0.1.0.0 - May 18th, 2015 -* First proper release (although still lots of rough edges!).-* Filled out `Data.Memcache.Client` to a complete API.-* Integrated cluster and authentication handling.-* Better error and exception handling.-* Fix compilation under GHC 7.10.+- First proper release (although still lots of rough edges!).+- Filled out `Data.Memcache.Client` to a complete API.+- Integrated cluster and authentication handling.+- Better error and exception handling.+- Fix compilation under GHC 7.10. # 0.0.1 - May 5th, 2015 -* Initial (incomplete) support for a cluster of memcached servers.-* Fixed compilation under GHC 7.4.+- Initial (incomplete) support for a cluster of memcached servers.+- Fixed compilation under GHC 7.4. # 0.0.0 - August 23rd, 2013 -* Initial release. Support for a single server.-+- Initial release. Support for a single server.
+ Database/Memcache/Auth.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE OverloadedStrings #-}++{-|+Module : Database.Memcache.Auth+Description : Meta Text auth-file authentication+Copyright : (c) David Terei, 2016+License : BSD+Maintainer : code@davidterei.com+Stability : stable+Portability : GHC++Authentication for Memcached's experimental text-protocol auth-file mode.+-}+module Database.Memcache.Auth+ ( Authentication(..)+ , Username+ , Password+ , authenticate+ , validateCredentials+ ) where++import Database.Memcache.Errors+import Database.Memcache.Socket+import Database.Memcache.Types++import Control.Monad (unless)+import qualified Data.ByteString.Char8 as B8+import UnliftIO.Exception (catch, throwIO)++-- | Authenticate a connection using text-protocol credentials.+authenticate :: Connection -> Authentication -> IO ()+authenticate _ NoAuth = return ()+authenticate socket (Auth user pass) = do+ case validateCredentials user pass of+ Left message -> throwIO $ ClientError (InvalidAuthentication message)+ Right () -> return ()+ -- Without -Y authfile, "set auth ..." is just a normal store, so credentials+ -- would be readable via "get auth"; probe with a real command first so we+ -- never send them to a server that isn't gating on auth-file mode.+ sendRaw socket versionRequest+ probe <- recvResponse socket+ unless (responseCode probe == "CLIENT_ERROR") $ throwIO $ ClientError+ (InvalidAuthentication "server is not in auth-file mode; refusing to send credentials")+ let credentials = user <> B8.singleton ' ' <> pass+ sendRaw socket (authRequest credentials)+ response <- recvResponse socket+ result <- responseStatus response `catch` handleAuthFailure+ case result of+ NoError -> return ()+ status -> throwIO $ OpError status+ where+ handleAuthFailure (ProtocolError BadCommand{}) = throwIO $ OpError TextAuthFail+ handleAuthFailure err = throwIO (err :: MemcacheError)++-- | Validate that credentials contain no delimiters used by the auth command.+validateCredentials :: Username -> Password -> Either String ()+validateCredentials user pass+ | B8.any forbidden user = Left "username contains a forbidden delimiter"+ | B8.any forbidden pass = Left "password contains a forbidden delimiter"+ | otherwise = Right ()+ where+ forbidden c = c == ' ' || c == '\t' || c == '\r' || c == '\n' || c == ':'
Database/Memcache/Client.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-} {-| Module : Database.Memcache.Client@@ -9,16 +9,18 @@ Stability : stable Portability : GHC -A Memcached client. Memcached is an in-memory key-value store typically used as-a distributed and shared cache. Clients connect to a group of Memcached servers-and perform out-of-band caching for things like SQL results, rendered pages, or-third-party APIs.+A Memcached client using the Meta (text) protocol. Memcached is an in-memory+key-value store typically used as a distributed and shared cache. Clients can+connect to a single Memcached server or a cluster of servers; clusters use+consistent hashing to route requests to the appropriate server. -A client can connect to a single Memcached server or a cluster of them. In the-later case, consistent hashing is used to route requests to the appropriate-server. The /binary/ Memcached protocol is used and /SASL authentication/ is-supported.+This requires memcached 1.6.0 or newer, the first release with the meta+commands; older servers reject every operation with @ERROR@. +Authentication uses Memcached's experimental text-protocol auth-file mechanism+(@-Y authfile@), introduced in memcached 1.5.15. Enabling @-Y@ also disables+binary and UDP protocols on the server.+ Expected return values (like misses) are returned as part of the return type, while unexpected errors are thrown as exceptions. Exceptions are either of type 'MemcacheError' or an 'IO' exception thrown by the network.@@ -27,6 +29,7 @@ * __Timeouts__: we timeout any operation that takes too long and consider it failed.+ * __Retry__: on operation failure (timeout, network error) we close the connection and retry the operation, doing this up to a configurable maximum.@@ -45,6 +48,9 @@ [@version@] Each value has a 'Version' associated with it. This is simply a numeric, monotonically increasing value. The version field allows for a primitive version of 'cas' to be implemented.+ Operations that take a 'Version' only change a value whose+ version matches it, except that a 'Version' of @0@ skips the+ check. [@expiration@] Each value pair has an 'Expiration' associated with it. Once a a value expires, it will no longer be returned from the cache@@ -63,20 +69,23 @@ Usage is roughly as follows: -> module Main where+> {-# LANGUAGE OverloadedStrings #-} >+> module Main (main) where+> > import qualified Database.Memcache.Client as M >+> main :: IO () > main = do-> -- use default values: connects to localhost:11211+> -- use default values: connects to 127.0.0.1:11211 > mc <- M.newClient [M.def] M.def > > -- store and then retrieve a key-value pair-> M.set mc "key" "value" 0 0+> _ <- M.set mc "key" "value" 0 0 > v' <- M.get mc "key" > case v' of > Nothing -> putStrLn "Miss!"-> Just (v, _, _) -> putStrLn $ "Hit: " + show v+> Just (v, _, _) -> putStrLn $ "Hit: " <> show v -} module Database.Memcache.Client ( -- * Client creation@@ -87,7 +96,7 @@ -- * Operations -- ** Get operations- get, gat, touch,+ get, getMany, gat, touch, -- ** Set operations set, cas, add, replace,@@ -105,273 +114,288 @@ MemcacheError(..), Status(..), ClientError(..), ProtocolError(..) ) where -import Database.Memcache.Cluster-import Database.Memcache.Errors-import Database.Memcache.Server-import Database.Memcache.Socket-import Database.Memcache.Types hiding (cas)+import Database.Memcache.Cluster+import Database.Memcache.Errors+import Database.Memcache.Server+import Database.Memcache.Socket+import Database.Memcache.Types -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif-import Control.Exception (handle, throwIO, SomeException)-import Control.Monad (forM_, void, when)-import Data.Default.Class-import Data.Word-import Data.ByteString (ByteString)-import qualified Data.ByteString as B (null)+import Control.Monad (foldM, forM_, void, when)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B (any, empty, length, null)+import qualified Data.ByteString.Char8 as C (unwords)+import Data.Default+import Data.Maybe (fromMaybe, listToMaybe)+import Data.Word+import UnliftIO.Exception (SomeException, handle, throwIO) --- | A Memcached client, connected to a collection of Memcached servers.+-- | A Memcached cluster client. type Client = Cluster+-- | Key/value pairs returned by 'stats'.+type StatResults = [(ByteString, ByteString)] --- | Establish a new connection to a group of Memcached servers.+-- | Create a client for the supplied servers. An empty list uses the default+-- local server specification. newClient :: [ServerSpec] -> Options -> IO Client-newClient = newCluster+newClient scs = newCluster $ if null scs then [def] else scs --- | Gracefully close a connection to a Memcached cluster.+-- | Close all client connections, ignoring individual shutdown failures. quit :: Cluster -> IO ()-quit c = void $ allOp' c serverQuit- where- serverQuit :: Server -> IO ()- serverQuit s = handle consumeError $ do- let msg = emptyReq { reqOp = ReqQuit Quiet }- withSocket s $ \sock -> send sock msg- close s-- consumeError :: SomeException -> IO ()- consumeError _ = return ()+quit c = void $ allOp' c $ \server -> handle ignore $ do+ withSocket server $ \socket -> sendRaw socket quitRequest+ close server+ where ignore :: SomeException -> IO ()+ ignore = const (pure ()) --- | Retrieve the value for the given key from Memcached.+-- | Retrieve the value for the given key from Memcached, along with its+-- flags and CAS version. Returns 'Nothing' on a miss. get :: Cluster -> Key -> IO (Maybe (Value, Flags, Version)) get c k = do- let msg = emptyReq { reqOp = ReqGet Loud NoKey k }- r <- keyedOp c k msg- (v, f) <- case resOp r of- ResGet Loud v f -> return (v, f)- _ -> throwIO $ wrongOp r "GET"- case resStatus r of- NoError -> return $ Just (v, f, resCas r)- ErrKeyNotFound -> return Nothing- rs -> throwStatus rs+ response <- keyedOp c k getRequest+ status <- responseStatus response+ case (responseCode response, status) of+ ("VA", NoError) -> return $ Just (valueResult response)+ (_, ErrKeyNotFound) -> return Nothing+ _ -> failResponse response "GET" +-- | Fetch many keys. Missing keys are omitted from the result.+getMany :: Cluster -> [Key] -> IO [(Key, (Value, Flags, Version))]+getMany c keys = do+ replies <- keyedBatchOp c [(key, getManyRequest) | key <- keys]+ reverse <$> foldM collect [] replies+ where+ collect values (key, response) = do+ status <- responseStatus response+ case (responseCode response, status) of+ ("VA", NoError) ->+ return ((key, valueResult response) : values)+ (_, ErrKeyNotFound) -> return values+ _ -> failResponse response "GETMANY"+ -- | Get-and-touch: Retrieve the value for the given key from Memcached, and -- also update the stored key-value pairs expiration time at the server. Use an -- expiration value of @0@ to store forever. gat :: Cluster -> Key -> Expiration -> IO (Maybe (Value, Flags, Version)) gat c k e = do- let msg = emptyReq { reqOp = ReqGAT Loud NoKey k (SETouch e) }- r <- keyedOp c k msg- (v, f) <- case resOp r of- ResGAT Loud v f -> return (v, f)- _ -> throwIO $ wrongOp r "GAT"- case resStatus r of- NoError -> return $ Just (v, f, resCas r)- ErrKeyNotFound -> return Nothing- rs -> throwStatus rs+ response <- keyedOp c k (`gatRequest` e)+ status <- responseStatus response+ case (responseCode response, status) of+ ("VA", NoError) -> return $ Just (valueResult response)+ (_, ErrKeyNotFound) -> return Nothing+ _ -> failResponse response "GAT" -- | Update the expiration time of a stored key-value pair, returning its--- version identifier. Use an expiration value of @0@ to store forever.+-- new CAS version or 'Nothing'.+-- Use an expiration value of @0@ to store forever. touch :: Cluster -> Key -> Expiration -> IO (Maybe Version) touch c k e = do- let msg = emptyReq { reqOp = ReqTouch k (SETouch e) }- r <- keyedOp c k msg- when (resOp r /= ResTouch) $ throwIO $ wrongOp r "TOUCH"- case resStatus r of- NoError -> return $ Just (resCas r)- ErrKeyNotFound -> return Nothing- rs -> throwStatus rs+ response <- keyedOp c k (`touchRequest` e)+ status <- responseStatus response+ case status of+ NoError | responseCode response == "HD" -> return $ Just (responseCas response)+ ErrKeyNotFound -> return Nothing+ _ -> failResponse response "TOUCH" --- | Store a new (or overwrite exisiting) key-value pair, returning its version+-- | Store a new (or overwrite existing) key-value pair, returning its @Version@ -- identifier. Use an expiration value of @0@ to store forever. set :: Cluster -> Key -> Value -> Flags -> Expiration -> IO Version-set c k v f e = do- let msg = emptyReq { reqOp = ReqSet Loud k v (SESet f e) }- r <- keyedOp c k msg- when (resOp r /= ResSet Loud) $ throwIO $ wrongOp r "SET"- case resStatus r of- NoError -> return $ resCas r- rs -> throwStatus rs+set c k v f e = store c k v f e 0 Set "SET" -- | Store a key-value pair, but only if the version specified by the client--- matches the Version of the key-value pair at the server. The version--- identifier of the stored key-value pair is returned, or if the version match--- fails, @Nothing@ is returned. Use an expiration value of @0@ to store--- forever.+-- matches the @Version@ of the key-value pair at the server. The version+-- identifier of the stored key-value pair is returned, or if the key doesn't+-- exist or the version match fails, @Nothing@ is returned. Use an expiration+-- value of @0@ to store forever.+--+-- A version of @0@ skips the check, so the value is stored unconditionally, as+-- with 'set'. cas :: Cluster -> Key -> Value -> Flags -> Expiration -> Version -> IO (Maybe Version) cas c k v f e ver = do- let msg = emptyReq { reqOp = ReqSet Loud k v (SESet f e), reqCas = ver }- r <- keyedOp c k msg- when (resOp r /= ResSet Loud) $ throwIO $ wrongOp r "SET"- case resStatus r of- NoError -> return $ Just (resCas r)- ErrKeyNotFound -> return Nothing -- CAS: key doesn't exist- ErrKeyExists -> return Nothing -- CAS: version doesn't match- rs -> throwStatus rs+ response <- keyedOp c k (\wire -> storeRequest Set wire v f e ver)+ status <- responseStatus response+ case status of+ NoError -> return $ Just (responseCas response)+ ErrKeyNotFound -> return Nothing+ ErrKeyExists -> return Nothing+ _ -> failResponse response "SET" --- | Store a new key-value pair, returning it's version identifier. If the+-- | Store a new key-value pair, returning its @Version@ identifier. If the -- key-value pair already exists, then fail (return 'Nothing'). Use an -- expiration value of @0@ to store forever. add :: Cluster -> Key -> Value -> Flags -> Expiration -> IO (Maybe Version) add c k v f e = do- let msg = emptyReq { reqOp = ReqAdd Loud k v (SESet f e) }- r <- keyedOp c k msg- when (resOp r /= ResAdd Loud) $ throwIO $ wrongOp r "ADD"- case resStatus r of- NoError -> return $ Just (resCas r)- ErrKeyExists -> return Nothing- rs -> throwStatus rs+ response <- keyedOp c k (\wire -> storeRequest Add wire v f e 0)+ status <- responseStatus response+ case status of+ NoError -> return $ Just (responseCas response)+ ErrItemNotStored -> return Nothing+ _ -> failResponse response "ADD" --- | Update the value of an existing key-value pair, returning it's new version--- identifier. If the key doesn't already exist, the fail and return Nothing.+-- | Update the value of an existing key-value pair, returning its new @Version@+-- identifier. If the key doesn't already exist, then fail and return 'Nothing'. -- Use an expiration value of @0@ to store forever.+--+-- A non-zero version must match the @Version@ of the key-value pair at the+-- server, otherwise 'Nothing' is returned. A version of @0@ skips the check. replace :: Cluster -> Key -> Value -> Flags -> Expiration -> Version -> IO (Maybe Version) replace c k v f e ver = do- let msg = emptyReq { reqOp = ReqReplace Loud k v (SESet f e), reqCas = ver }- r <- keyedOp c k msg- when (resOp r /= ResReplace Loud) $ throwIO $ wrongOp r "REPLACE"- case resStatus r of- NoError -> return $ Just (resCas r)- -- replace only applies to an existing key...- ErrKeyNotFound -> return Nothing- -- version specified and doesn't match key...- ErrKeyExists -> return Nothing- rs -> throwStatus rs+ response <- keyedOp c k (\wire -> storeRequest Replace wire v f e ver)+ status <- responseStatus response+ case status of+ NoError -> return $ Just (responseCas response)+ ErrKeyNotFound -> return Nothing+ ErrKeyExists -> return Nothing+ ErrItemNotStored -> return Nothing+ _ -> failResponse response "REPLACE" +store :: Cluster -> Key -> Value -> Flags -> Expiration -> Version -> StoreMode -> String -> IO Version+store c k v f e ver mode operation = do+ response <- keyedOp c k (\wire -> storeRequest mode wire v f e ver)+ status <- responseStatus response+ case status of+ NoError -> return $ responseCas response+ _ -> failResponse response operation+ -- | Increment a numeric value stored against a key, returning the incremented--- value and the version identifier of the key-value pair. Use an expiration--- value of @0@ to store forever.+-- value and the @Version@ identifier of the key-value pair. Throws+-- 'OpError' 'ErrValueNonNumeric' if the stored value isn't a number.+--+-- If the key doesn't exist, it is created with the 'Initial' value, which is+-- returned without the delta applied. The expiration is only used for a key+-- created this way, and doesn't change the expiration of an existing key; use+-- @0@ for a created key that never expires. To leave a missing key missing+-- instead, pass an expiration of 'maxBound': nothing is created and 'Nothing'+-- is returned.+--+-- A non-zero version must match the @Version@ of an existing key-value pair,+-- otherwise 'Nothing' is returned. A version of @0@ skips the check. increment :: Cluster -> Key -> Initial -> Delta -> Expiration -> Version -> IO (Maybe (Word64, Version))-increment c k i d e ver = do- let msg = emptyReq { reqOp = ReqIncrement Loud k (SEIncr i d e), reqCas = ver }- r <- keyedOp c k msg- n <- case resOp r of- ResIncrement Loud n -> return n- _ -> throwIO $ wrongOp r "INCREMENT"- case resStatus r of- NoError -> return $ Just (n, resCas r)- ErrKeyNotFound -> return Nothing- ErrKeyExists -> return Nothing- rs -> throwStatus rs+increment c k i d e ver = arithmetic c k i d e ver Incr "INCREMENT" -- | Decrement a numeric value stored against a key, returning the decremented--- value and the version identifier of the key-value pair. Use an expiration--- value of @0@ to store forever.+-- value and the @Version@ identifier of the key-value pair. Memcached doesn't+-- decrement below @0@.+--+-- Otherwise this behaves like 'increment': a missing key is created with the+-- 'Initial' value unless the expiration is 'maxBound', a version of @0@ skips+-- the check, and a non-numeric value throws 'OpError' 'ErrValueNonNumeric'. decrement :: Cluster -> Key -> Initial -> Delta -> Expiration -> Version -> IO (Maybe (Word64, Version))-decrement c k i d e ver = do- let msg = emptyReq { reqOp = ReqDecrement Loud k (SEIncr i d e), reqCas = ver }- r <- keyedOp c k msg- n <- case resOp r of- ResDecrement Loud n -> return n- _ -> throwIO $ wrongOp r "DECREMENT"- case resStatus r of- NoError -> return $ Just (n, resCas r)- ErrKeyNotFound -> return Nothing- ErrKeyExists -> return Nothing- rs -> throwStatus rs+decrement c k i d e ver = arithmetic c k i d e ver Decr "DECREMENT" --- | Append a value to an existing key-value pair, returning the new version--- identifier of the key-value pair when successful.+arithmetic :: Cluster -> Key -> Initial -> Delta -> Expiration -> Version -> ArithMode -> String -> IO (Maybe (Word64, Version))+arithmetic c k i d e ver mode operation = do+ response <- keyedOp c k (\wire -> arithmeticRequest mode wire i d e ver)+ status <- responseStatus response+ case status of+ NoError -> return $ Just (responseNumber response, responseCas response)+ ErrKeyNotFound -> return Nothing+ ErrKeyExists -> return Nothing+ ErrItemNotStored -> return Nothing+ _ -> failResponse response operation++-- | Append a value to an existing key-value pair, returning the new @Version@+-- identifier of the key-value pair when successful, or 'Nothing' if the key+-- doesn't exist.+--+-- A non-zero version must match the @Version@ of the key-value pair at the+-- server, otherwise 'OpError' 'ErrKeyExists' is thrown. A version of @0@ skips+-- the check. append :: Cluster -> Key -> Value -> Version -> IO (Maybe Version)-append c k v ver = do- let msg = emptyReq { reqOp = ReqAppend Loud k v, reqCas = ver }- r <- keyedOp c k msg- when (resOp r /= ResAppend Loud) $ throwIO $ wrongOp r "APPEND"- case resStatus r of- NoError -> return $ Just (resCas r)- ErrKeyNotFound -> return Nothing- rs -> throwStatus rs+append c k v ver = modify c k v ver Append "APPEND" --- | Prepend a value to an existing key-value pair, returning the new version--- identifier of the key-value pair when successful.+-- | Prepend a value to an existing key-value pair, returning the new @Version@+-- identifier of the key-value pair when successful, or 'Nothing' if the key+-- doesn't exist.+--+-- A non-zero version must match the @Version@ of the key-value pair at the+-- server, otherwise 'OpError' 'ErrKeyExists' is thrown. A version of @0@ skips+-- the check. prepend :: Cluster -> Key -> Value -> Version -> IO (Maybe Version)-prepend c k v ver = do- let msg = emptyReq { reqOp = ReqPrepend Loud k v, reqCas = ver }- r <- keyedOp c k msg- when (resOp r /= ResPrepend Loud) $ throwIO $ wrongOp r "PREPEND"- case resStatus r of- NoError -> return $ Just (resCas r)- ErrKeyNotFound -> return Nothing- rs -> throwStatus rs+prepend c k v ver = modify c k v ver Prepend "PREPEND" --- | Delete a key-value pair at the server, returning true if successful.+modify :: Cluster -> Key -> Value -> Version -> StoreMode -> String -> IO (Maybe Version)+modify c k v ver mode operation = do+ response <- keyedOp c k (\wire -> modifyRequest mode wire v ver)+ status <- responseStatus response+ case status of+ NoError -> return $ Just (responseCas response)+ ErrKeyNotFound -> return Nothing+ ErrItemNotStored -> return Nothing+ _ -> failResponse response operation++-- | Delete a key, returning whether it was deleted.+--+-- A non-zero version must match the @Version@ of the key-value pair at the+-- server, otherwise the key isn't deleted and 'False' is returned. A version+-- of @0@ deletes the key unconditionally. delete :: Cluster -> Key -> Version -> IO Bool delete c k ver = do- let msg = emptyReq { reqOp = ReqDelete Loud k, reqCas = ver }- r <- keyedOp c k msg- when (resOp r /= ResDelete Loud) $ throwIO $ wrongOp r "DELETE"- case resStatus r of- NoError -> return True- -- delete only applies to an existing key...- ErrKeyNotFound -> return False- -- version specified and doesn't match key...- ErrKeyExists -> return False- rs -> throwStatus rs+ response <- keyedOp c k (`deleteRequest` ver)+ status <- responseStatus response+ case status of+ NoError -> return True+ ErrKeyNotFound -> return False+ ErrKeyExists -> return False+ _ -> failResponse response "DELETE" -- | Remove (delete) all currently stored key-value pairs from the cluster. The -- expiration value can be used to cause this flush to occur in the future -- rather than immediately. flush :: Cluster -> Maybe Expiration -> IO () flush c e = do- let msg = emptyReq { reqOp = ReqFlush Loud (SETouch <$> e) }- results <- allOp c msg- forM_ results $ \(_, r) -> do- when (resOp r /= ResFlush Loud) $ throwIO $ wrongOp r "FLUSH"- case resStatus r of- NoError -> return ()- rs -> throwStatus rs---- | StatResults are a list of key-value pairs.-type StatResults = [(ByteString, ByteString)]+ results <- allOp c (flushRequest e)+ forM_ results $ \(_, response) -> do+ status <- responseStatus response+ when (status /= NoError) $ failResponse response "FLUSH" --- | Return statistics on the stored key-value pairs at each server in the--- cluster. The optional key can be used to select a different set of+-- | Return statistics on the stored key-value pairs at every available server+-- in the cluster. The optional key can be used to select a different set of -- statistics from the server than the default. Most Memcached servers support -- @"items"@, @"slabs"@ or @"settings"@.-stats :: Cluster -> Maybe Key -> IO [(Server, Maybe StatResults)]-stats c key = allOp' c serverStats+stats :: Cluster -> Maybe Key -> IO [(Server, StatResults)]+stats c key = validateStatsArgs key >> allOp' c (runStats key) where- msg :: Request- msg = emptyReq { reqOp = ReqStat key }+ runStats statsArgs server = withSocket server $ \socket -> do+ sendRaw socket (statsRequest statsArgs)+ collect socket [] - serverStats :: Server -> IO (Maybe StatResults)- serverStats s = withSocket s $ \sock -> do- send sock msg- recvAllStats sock []+ -- Arguments may contain spaces (@stats cachedump 1 100@), but control+ -- bytes would let a caller inject a second command line.+ validateStatsArgs Nothing = return ()+ validateStatsArgs (Just args)+ | B.null args ||+ B.length args > 250 ||+ B.any (\byte -> byte < 0x20 || byte == 0x7f) args =+ throwIO $ ClientError (KeyTooLong "invalid stats argument")+ | otherwise = return () - recvAllStats :: Socket -> StatResults -> IO (Maybe StatResults)- recvAllStats s xs = do- r <- recv s- (k, v) <- case resOp r of- ResStat k v -> return (k, v)- _ -> throwIO $ wrongOp r "STATS"- case resStatus r of- NoError | B.null k && B.null v -> return $ Just xs- | otherwise -> recvAllStats s $ (k, v):xs- ErrKeyNotFound -> return Nothing- rs -> throwStatus rs+ collect socket values = do+ response <- recvResponse socket+ case (responseCode response, responseTokens response) of+ (code, _) | code `elem` ["ERROR", "CLIENT_ERROR", "SERVER_ERROR"] ->+ responseStatus response >> failResponse response "STATS"+ (code, _) | code `elem` ["END", "RESET", "OK"] -> return (reverse values)+ ("STAT", field:value) -> collect socket ((field, C.unwords value) : values)+ (code, tokens) -> collect socket ((code, C.unwords tokens) : values) --- | Version returns the version string of the Memcached cluster. We just query--- one server and assume all servers in the cluster are the same version.+-- | Return the version string of the Memcached cluster. Only queries+-- one server and assumes all servers in the cluster run the same version. version :: Cluster -> IO ByteString version c = do- let msg = emptyReq { reqOp = ReqVersion }- r <- anyOp c msg- v <- case resOp r of- ResVersion v -> return v- _ -> throwIO $ wrongOp r "VERSION"- case resStatus r of- NoError -> return v- rs -> throwStatus rs+ response <- anyOp c versionRequest+ status <- responseStatus response+ if responseCode response == "VERSION" && status == NoError+ then return $ fromMaybe B.empty (listToMaybe (responseTokens response))+ else failResponse response "VERSION" --- | Noop sends a Non-operation command to the specified Memcached server. We--- leave it blanked out, but here for documentation purposes of the full--- protocol.--- noop :: Server -> IO ()--- noop c = do--- let msg = emptyReq { reqOp = ReqNoop }--- r <- sendRecv c msg--- when (resOp r /= ResNoop) $ throwIncorrectRes r "NOOP"--- case resStatus r of--- NoError -> return ()--- rs -> throwStatus rs+-- | The value, flags and CAS version carried by a @VA@ response.+valueResult :: Response -> (Value, Flags, Version)+valueResult response = (responseValue response, responseFlags response, responseCas response) +failResponse :: Response -> String -> IO a+failResponse response operation = do+ status <- responseStatus response+ throwIO $ if status /= NoError+ then OpError status+ else ProtocolError UnexpectedResponse+ { protocolError = "Expected " <> operation <> ", got: " <> show (responseCode response) }
Database/Memcache/Cluster.hs view
@@ -1,8 +1,5 @@-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE CPP #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-} {-| Module : Database.Memcache.Cluster@@ -16,52 +13,43 @@ Handles a group of connections to different Memcached servers. We use consistent hashing to choose which server to route a request to. On an-error, we mark the server as failed and remove it temporarialy from the set of+error, we mark the server as failed and remove it temporarily from the set of servers available. -} module Database.Memcache.Cluster ( -- * Cluster- Cluster, ServerSpec(..), Options(..), newCluster,+ -- | A collection of Memcached servers with retry and routing state.+ Cluster, ServerSpec(..), Options(..), newCluster, getServers, setServers, -- * Operations- Retries, keyedOp, anyOp, allOp, allOp'+ Retries, keyedOp, keyedBatchOp, anyOp, allOp, allOp' ) where -import Database.Memcache.Errors-import Database.Memcache.Server-import Database.Memcache.Types+import Database.Memcache.Errors+import Database.Memcache.Server+import Database.Memcache.Types -import Control.Concurrent (threadDelay)-import Control.Exception (handle, throwIO, SomeException)-import Data.Default.Class-import Data.Fixed (Milli)-import Data.Hashable (hash)-import Data.IORef-import Data.Maybe (fromMaybe)-import Data.List (sort)-import Data.Time.Clock (NominalDiffTime)-import Data.Time.Clock.POSIX (getPOSIXTime)-import qualified Data.Vector as V-import Network.Socket (HostName, PortNumber)-import System.Timeout+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, readMVar)+import Control.Monad (foldM, void)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.Default+import Data.Fixed (Milli)+import Data.Hashable (hash)+import Data.IORef+import Data.List (find, sort)+import Data.Maybe (fromMaybe)+import Data.Time.Clock (NominalDiffTime)+import Data.Time.Clock.POSIX (getPOSIXTime)+import qualified Data.Vector as V+import System.Timeout+import UnliftIO.Exception (SomeException, fromException, handle, throwIO)+import Database.Memcache.Socket (Response(..), WireKey(..), encodeKey, noOpRequest, recvResponse, responseStatus, sendRaw) -- | Number of times to retry an operation before considering it failed. type Retries = Int --- | ServerSpec specifies a server configuration for connection.-data ServerSpec = ServerSpec {- -- | Hostname of server to connect to.- ssHost :: HostName,- -- | Port number server is running on.- ssPort :: PortNumber,- -- | Authentication values to use for SASL authentication with this- -- server.- ssAuth :: Authentication- } deriving (Eq, Show)--instance Default ServerSpec where- def = ServerSpec "localhost" 11211 NoAuth- -- | Options specifies how a Memcached cluster should be configured. data Options = Options { -- | Number of times to retry an operation on failure. If consecutive@@ -69,66 +57,88 @@ -- failover to a different server for the next operation. -- -- Default is 2.- optsServerRetries :: Retries,+ optsServerRetries :: Retries, -- | After an operation has failed, how long to wait before retrying it -- while still within the 'optsServerRetries' count? -- -- Default is 200ms.- optsFailRetryDelay :: Milli,+ optsFailRetryDelay :: Milli, -- | How long to wait after a server has been marked down, before -- trying to use it again. -- -- Default is 1500ms.- optsDeadRetryDelay :: Milli,+ optsDeadRetryDelay :: Milli, -- | How long to wait for an operation to complete before considering -- it failed. -- -- Default is 750ms.- optsServerTimeout :: Milli- -- TODO: socket_timeout- -- TODO: failover- -- TODO: expires_in- -- TODO: namespace- -- TODO: compress- -- TODO: compress_min_size- -- TODO: compress_max_size- -- TODO: value_max_bytes- } deriving (Eq, Show)+ optsServerTimeout :: Milli,+ -- | Figure out which server to talk to for a given key.+ --+ -- Default is @getServerForKeyDefault@.+ optsGetServerForKey :: Cluster -> Key -> IO (Maybe Server),+ -- | Convert server specifications into servers.+ --+ -- Default uses 'newServerDefault'.+ optsServerSpecsToServers :: ServerOptions -> [ServerSpec] -> IO [Server], + -- | Connection-pool options for each server.+ optsServerOptions :: ServerOptions+ }+ instance Default Options where def = Options { optsServerRetries = 2, optsFailRetryDelay = 200, optsDeadRetryDelay = 1500,- optsServerTimeout = 750+ optsServerTimeout = 750,+ optsGetServerForKey = getServerForKeyDefault,+ optsServerSpecsToServers = mapM . newServerDefault,+ optsServerOptions = def } -- | Memcached cluster. data Cluster = Cluster {- cServers :: V.Vector Server,+ cServers :: Either (MVar (V.Vector Server)) (V.Vector Server), - -- See 'Options' for description of these values.+ -- See @Options@ for description of these values. - cRetries :: {-# UNPACK #-} !Int,- cFailDelay :: {-# UNPACK #-} !Int, -- ^ microseconds- cDeadDelay :: !NominalDiffTime,- cTimeout :: {-# UNPACK #-} !Int -- ^ microseconds- } deriving (Eq, Show)+ cRetries :: {-# UNPACK #-} !Int,+ cFailDelay :: {-# UNPACK #-} !Int, -- ^ microseconds+ cDeadDelay :: !NominalDiffTime,+ cTimeout :: {-# UNPACK #-} !Int, -- ^ microseconds+ cGetServerForKey :: Cluster -> Key -> IO (Maybe Server)+ } +-- | Get the currently configured servers.+getServers :: Cluster -> IO (V.Vector Server)+getServers = either readMVar pure . cServers++-- | Replace the cluster's server set, preserving its routing and retry options.+setServers :: Cluster -> Either (MVar (V.Vector Server)) (V.Vector Server) -> Cluster+setServers c servers = c { cServers = servers }+ -- | Establish a new connection to a group of Memcached servers. newCluster :: [ServerSpec] -> Options -> IO Cluster newCluster [] _ = throwIO $ ClientError NoServersReady newCluster hosts Options{..} = do- s <- mapM (\ServerSpec{..} -> newServer ssHost ssPort ssAuth) hosts+ s <- optsServerSpecsToServers optsServerOptions hosts return $ Cluster {- cServers = (V.fromList $ sort s),+ cServers = Right $ V.fromList $ sort s, cRetries = optsServerRetries ,- cFailDelay = fromEnum optsFailRetryDelay,+ cFailDelay = toMicroseconds optsFailRetryDelay, cDeadDelay = fromRational $ toRational optsDeadRetryDelay / 1000,- cTimeout = fromEnum optsServerTimeout + cTimeout = toMicroseconds optsServerTimeout,+ cGetServerForKey = optsGetServerForKey } +-- | Convert a duration in milliseconds to the microseconds that 'threadDelay'+-- and 'timeout' take. 'Milli' has three decimal places, so the product is a+-- whole number and 'truncate' loses nothing.+toMicroseconds :: Milli -> Int+toMicroseconds ms = truncate (toRational ms * 1000)+ -- | Check if server is alive. serverAlive :: NominalDiffTime -> Server -> IO Bool {-# INLINE serverAlive #-}@@ -144,55 +154,116 @@ writeIORef (failed s) 0 return True +-- | The servers not currently marked as failed.+aliveServers :: Cluster -> IO (V.Vector Server)+{-# INLINE aliveServers #-}+aliveServers c = getServers c >>= V.filterM (serverAlive $ cDeadDelay c)+ -- | Figure out which server to talk to for this key. I.e., the distribution -- method. We use consistent hashing based on the CHORD approach.-getServerForKey :: Cluster -> Key -> IO (Maybe Server)-{-# INLINE getServerForKey #-}-getServerForKey c k = do+getServerForKeyDefault :: Cluster -> Key -> IO (Maybe Server)+{-# INLINE getServerForKeyDefault #-}+getServerForKeyDefault c k = do let hashedKey = hash k searchF s = sid s < hashedKey- servers' <- V.filterM (serverAlive $ cDeadDelay c) $ cServers c+ servers' <- aliveServers c return $ if V.null servers' then Nothing else Just $ fromMaybe (V.last servers') (V.find searchF servers') -- | Run a Memcached operation against a particular server, handling any -- failures that occur, retrying the specified number of times.-serverOp :: Cluster -> Server -> Request -> IO Response+serverOp :: Cluster -> Server -> ByteString -> IO Response {-# INLINE serverOp #-}-serverOp c s req = retryOp c s $ sendRecv s req+serverOp c s req = retryOp c s $ withSocket s $ \connection -> do+ sendRaw connection req+ response <- recvResponse connection+ rejectProtocolError response+ return response --- | Run a Memcached operation against a particular server, handling any--- failures that occur, retrying the specified number of times.-keyedOp :: Cluster -> Key -> Request -> IO Response+rejectProtocolError :: Response -> IO ()+rejectProtocolError response+ | responseCode response `elem` ["ERROR", "CLIENT_ERROR", "SERVER_ERROR"] =+ void (responseStatus response)+ | otherwise = return ()++-- | Run a keyed request against the server selected for the key.+keyedOp :: Cluster -> Key -> (WireKey -> ByteString) -> IO Response {-# INLINE keyedOp #-}-keyedOp c k req = do- s' <- getServerForKey c k+keyedOp c k makeReq = do+ wire <- either (throwIO . ClientError) return (encodeKey k)+ s' <- cGetServerForKey c c k case s' of- Just s -> serverOp c s req+ Just s -> serverOp c s (makeReq wire) Nothing -> throwIO $ ClientError NoServersReady +-- | Run keyed requests grouped by their selected server.+keyedBatchOp :: Cluster -> [(Key, WireKey -> ByteString)] -> IO [(Key, Response)]+keyedBatchOp c requests = do+ groups <- foldM add [] requests+ concat <$> mapM run groups+ where+ add groups (key, makeRequest) = do+ wire <- either (throwIO . ClientError) return (encodeKey key)+ server <- cGetServerForKey c c key+ case server of+ Nothing -> throwIO $ ClientError NoServersReady+ Just s -> return (insertGroup s (key, wire, makeRequest) groups)++ insertGroup s item [] = [(s, [item])]+ insertGroup s item ((s', items):rest)+ | s == s' = (s', item:items) : rest+ | otherwise = (s', items) : insertGroup s item rest++ run (server, items) = concat <$> mapM (runBatch server) (chunksOf batchSize (reverse items))++ -- Write the whole batch in one go, then read it back. Batches are bounded+ -- so the server's replies cannot fill the socket buffers while we are+ -- still writing, which would deadlock both ends until the op times out.+ runBatch server items = retryOp c server $ withSocket server $ \connection -> do+ sendRaw connection (B.concat (map request items) <> noOpRequest)+ replies <- recvUntilMN connection []+ return [(key, response) | response <- replies, Just key <- [responseKey response items]]++ recvUntilMN connection acc = do+ response <- recvResponse connection+ rejectProtocolError response+ if responseCode response == "MN"+ then return (reverse acc)+ else recvUntilMN connection (response : acc)++ -- Match the echoed key together with its base64 flag: a plain key can+ -- have the same bytes as another key's base64 form.+ responseKey response items = do+ echoed <- find (B.isPrefixOf "k") (responseTokens response)+ let wire = WireKey (B.drop 1 echoed) ("b" `elem` responseTokens response)+ (key, _, _) <- find ((== wire) . snd3) items+ return key++ request (_, wire, makeRequest) = makeRequest wire+ snd3 (_, wire, _) = wire++ batchSize = 256 :: Int++ chunksOf n xs = case splitAt n xs of+ (chunk, []) -> [chunk]+ (chunk, rest) -> chunk : chunksOf n rest+ -- | Run a Memcached operation against any single server in the cluster, -- handling any failures that occur, retrying the specified number of times.-anyOp :: Cluster -> Request -> IO Response+anyOp :: Cluster -> ByteString -> IO Response {-# INLINE anyOp #-} anyOp c req = do- servers' <- V.filterM (serverAlive $ cDeadDelay c) $ cServers c- if V.null servers'- then throwIO $ ClientError NoServersReady- else serverOp c (V.head servers') req+ servers' <- aliveServers c+ case V.uncons servers' of+ Nothing -> throwIO $ ClientError NoServersReady+ Just (s, _) -> serverOp c s req -- | Run a Memcached operation against all servers in the cluster, handling any -- failures that occur, retrying the specified number of times.-allOp :: Cluster -> Request -> IO [(Server, Response)]+allOp :: Cluster -> ByteString -> IO [(Server, Response)] {-# INLINE allOp #-}-allOp c req = do- servers' <- V.filterM (serverAlive $ cDeadDelay c) $ cServers c- if V.null servers'- then throwIO $ ClientError NoServersReady- else do- res <- V.forM servers' $ \s -> serverOp c s req- return $ V.toList $ V.zip servers' res+allOp c req = forAliveServers c $ \s -> serverOp c s req -- | Run a Memcached operation against all servers in the cluster, handling any -- failures that occur, retrying the specified number of times. Similar to@@ -200,12 +271,18 @@ -- request and response. allOp' :: Cluster -> (Server -> IO a) -> IO [(Server, a)] {-# INLINE allOp' #-}-allOp' c op = do- servers' <- V.filterM (serverAlive $ cDeadDelay c) $ cServers c+allOp' c op = forAliveServers c $ \s -> retryOp c s (op s)++-- | Run an action against every server not marked as failed, pairing each+-- result with its server. Throws 'NoServersReady' if there are none.+forAliveServers :: Cluster -> (Server -> IO a) -> IO [(Server, a)]+{-# INLINE forAliveServers #-}+forAliveServers c f = do+ servers' <- aliveServers c if V.null servers' then throwIO $ ClientError NoServersReady else do- res <- V.forM servers' $ \s -> retryOp c s (op s)+ res <- V.mapM f servers' return $ V.toList $ V.zip servers' res -- | Run an IO operation multiple times if an exception is thrown, marking the@@ -224,10 +301,20 @@ handleErrs :: Int -> SomeException -> IO a {-# INLINE handleErrs #-}- handleErrs 0 err = do t <- getPOSIXTime- writeIORef (failed s) t- throwIO err- handleErrs n _ = do- threadDelay cFailDelay- go n+ handleErrs n err+ | clientFault err = throwIO err+ | n <= 0 = do+ writeIORef (failed s) =<< getPOSIXTime+ throwIO err+ | otherwise = threadDelay cFailDelay >> go n +-- | A rejected command or a reported operation status is an answer, not a+-- server failure: retrying it will fail identically and must not count towards+-- marking the server dead.+clientFault :: SomeException -> Bool+clientFault err = case fromException err of+ Just (ProtocolError BadCommand{}) -> True+ Just (ProtocolError UnknownOp{}) -> True+ Just (OpError _) -> True+ Just (ClientError InvalidAuthentication{}) -> True+ _ -> False
+ Database/Memcache/ElastiCacheClient.hs view
@@ -0,0 +1,190 @@+{-# LANGUAGE OverloadedStrings #-}++{-|+Module : Database.Memcache.ElastiCacheClient+Description : Amazon ElastiCache service discovery+Copyright : (c) David Terei, 2016+License : BSD+Maintainer : code@davidterei.com+Stability : stable+Portability : GHC++Client creation and automatic node discovery for Amazon ElastiCache clusters.+-}+module Database.Memcache.ElastiCacheClient+ ( ConfigurationEndpoint+ , parseConfigurationEndpoint,+ newClient,+ )+where++import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar (MVar, modifyMVar_, newMVar)+import Control.Monad (forever, guard, (<=<))+import Data.Bifunctor (first)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as C+import Data.List (sort)+import qualified Data.List.NonEmpty as NE+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.Vector as V+import Data.Version (Version, makeVersion)+import qualified Data.Version as Version+import Database.Memcache.Client (Client, optsServerSpecsToServers, optsServerOptions)+import qualified Database.Memcache.Client as Client+import Database.Memcache.Cluster (Options, getServers, setServers)+import Database.Memcache.Server (Server, withSocket)+import Database.Memcache.Socket (Connection, crlf, recvRawUntil, sendRaw, versionRequest)+import Database.Memcache.Types.ServerSpec (ServerSpec, parseServerSpec)+import Database.Memcache.Util (note)+import Text.ParserCombinators.ReadP (readP_to_S)+import UnliftIO.Exception (bracket, throwString)++-- | Parse an ElastiCache configuration endpoint.+--+-- A /Configuration Endpoint/ will always contain /.cfg/ in its address:+--+-- https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.Using.html+--+parseConfigurationEndpoint :: String -> Either String ConfigurationEndpoint+parseConfigurationEndpoint url = do+ note "URI does not contain '.cfg'" $ guard $ ".cfg" `T.isInfixOf` T.pack url+ ConfigurationEndpoint <$> parseServerSpec url++-- Please see:+-- https://github.com/memcached/memcached/blob/77709d04dd4fc7d59f59425802cb9645b2cfe0f8/doc/protocol.txt#L1805-L1812+-- ByteString format: "VERSION <version>\r\n"+parseVersion :: ByteString -> Either String Version+parseVersion = interpretParseResults . delegateParse . C.unpack . C.dropEnd (C.length crlf) . C.drop (C.length "VERSION ")+ where+ delegateParse = readP_to_S Version.parseVersion+ interpretParseResults parseResults = do+ results <- note "Unable to parse input" $ NE.nonEmpty parseResults+ let tail' = NE.last results+ case tail' of+ (version, "") -> pure version+ (_, rest) -> Left $ "eof expected. Got: " <> rest++-- | A validated ElastiCache configuration endpoint.+--+-- Use the smart constructor @parseConfigurationEndpoint@ to create one.+newtype ConfigurationEndpoint+ = ConfigurationEndpoint+ { unConfigurationEndpoint :: ServerSpec+ }++-- | Create a new client using service discovery.+--+-- This function discovers all the nodes in a cluster and sets up a new thread to run autodiscovery on an interval.+--+-- N.B. The interval is in the same units as 'threadDelay'.+newClient :: Options -> Int -> ConfigurationEndpoint -> IO Client+newClient options interval cfgEndpoint = bracket acquireCfgClient releaseCfgClient resolveCluster'+ where+ acquireCfgClient = Client.newClient [unConfigurationEndpoint cfgEndpoint] options+ releaseCfgClient = Client.quit+ resolveCluster' cfgClient = do+ serverSpecs <- resolveCluster cfgClient++ -- Create client where cServers = Right <Vector Server>+ client <- Client.newClient serverSpecs options++ -- Prepare MVar+ serversMVar <- newMVar =<< getServers client+ _ <- forkIO $ repeatAutoDiscover options interval cfgEndpoint serversMVar++ -- Update client where cServers = Left <MVar <Vector Server>>+ pure $ setServers client $ Left serversMVar++repeatAutoDiscover :: Options -> Int -> ConfigurationEndpoint -> MVar (V.Vector Server) -> IO ()+repeatAutoDiscover opts interval cfgEndpoint mVar = do+ cfgClient <- Client.newClient [unConfigurationEndpoint cfgEndpoint] opts+ forever $ do+ threadDelay interval+ serverSpecs <- resolveCluster cfgClient+ servers <- optsServerSpecsToServers opts (optsServerOptions opts) serverSpecs+ modifyMVar_ mVar $ \_curServers -> pure $ V.fromList $ sort servers++resolveCluster :: Client -> IO [ServerSpec]+resolveCluster cfgClient = do+ servers <- getServers cfgClient++ server <-+ case V.uncons servers of+ Nothing -> throwString "No servers were found"+ Just (s, _) -> pure s++ withSocket server $ \socket -> do+ -- Get version+ sendRaw socket versionRequest+ rawVersion <- recvRawUntil crlf socket+ version <- unTry $ parseVersion rawVersion++ -- Get nodes+ rawResponse <-+ if version >= firstCfgCmdVersion+ then resolveViaConfigCmd socket+ else resolveViaAutoDiscoveryKey socket++ rawNodeInfo <- unTry $ handleConfigGetClusterResponse rawResponse++ unTry $ handleRawNodeInfo rawNodeInfo++-- Handle the raw response from /memcached/.+--+-- See: https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.AddingToYourClientLibrary.html#AutoDiscovery.AddingToYourClientLibrary.OutputFormat+handleConfigGetClusterResponse :: ByteString -> Either String Text+handleConfigGetClusterResponse bs = do+ text <- first show $ T.decodeUtf8' bs++ case T.lines text of+ [_, _, nodeInfo, _, _] -> pure $ T.strip nodeInfo+ xs -> Left $ "Unrecognized number of lines: " <> show (length xs)++-- Handle node information+--+-- Note that we chose to use the CNAME rather than the IP. The IP /may/ be missing. The CNAME will always be present.+--+-- See: https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.AddingToYourClientLibrary.html#AutoDiscovery.AddingToYourClientLibrary.OutputFormat+handleRawNodeInfo :: Text -> Either String [ServerSpec]+handleRawNodeInfo = mapM (parseServerSpec <=< pure . T.unpack <=< convertToUri) . T.splitOn " "+ where+ convertToUri :: Text -> Either String Text+ convertToUri t =+ case T.splitOn "|" t of+ [host, _ip, port] -> pure $ "memcached://" <> host <> ":" <> port+ xs -> Left $ "Unrecognized components in node info: " <> show xs++-- This is the first version within which the /memcache/ engine on /ElastiCache/ uses the custom /config/ command.+--+-- https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.AddingToYourClientLibrary.html+--+firstCfgCmdVersion :: Version+firstCfgCmdVersion = makeVersion [1, 4, 14]++-- Pre-1.4.14 engines don't support meta commands, so use a plain text get.+resolveViaAutoDiscoveryKey :: Connection -> IO ByteString+resolveViaAutoDiscoveryKey socket = do+ sendRaw socket ("get " <> autoDiscoveryKey <> crlf)+ recvRawUntil ("END" <> crlf) socket++-- Before 'firstCfgCmdVersion', /ElastiCache/ stored configuration in a special-purpose key within the /Configuration Endpoint/.+--+-- https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.AddingToYourClientLibrary.html+--+autoDiscoveryKey :: ByteString+autoDiscoveryKey = "AmazonElastiCache:cluster"++-- See: https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.AddingToYourClientLibrary.html#AutoDiscovery.AddingToYourClientLibrary.OutputFormat+resolveViaConfigCmd :: Connection -> IO ByteString+resolveViaConfigCmd socket = do+ sendRaw socket (configGetCluster <> crlf)+ recvRawUntil ("END" <> crlf) socket++configGetCluster :: ByteString+configGetCluster = "config get cluster"++unTry :: Either String a -> IO a+unTry = either throwString pure
Database/Memcache/Errors.hs view
@@ -1,5 +1,3 @@-{-# LANGUAGE DeriveDataTypeable #-}- {-| Module : Database.Memcache.Errors Description : Errors Handling@@ -19,24 +17,22 @@ ProtocolError(..), -- * Error creation- throwStatus,- wrongOp+ throwStatus ) where -import Database.Memcache.Types+import Database.Memcache.Types -import Control.Exception-import Data.Typeable+import UnliftIO.Exception -- | All exceptions that a Memcached client may throw. data MemcacheError -- | Memcached operation error. = OpError Status- -- | Error occuring on client side.+ -- | Error occurring on client side. | ClientError ClientError -- | Errors occurring communicating with Memcached server. | ProtocolError ProtocolError- deriving (Eq, Show, Typeable)+ deriving (Eq, Show) instance Exception MemcacheError @@ -46,32 +42,34 @@ = NoServersReady -- | Timeout occurred sending request to server. | Timeout- deriving (Eq, Show, Typeable)+ -- | Credentials were rejected before being sent: they contain a+ -- forbidden delimiter, or the server is not in auth-file mode.+ | InvalidAuthentication String+ -- | A key or stats argument was rejected before being sent: it is empty,+ -- too long, or contains forbidden bytes.+ | KeyTooLong String+ deriving (Eq, Show) -- | Errors related to Memcached protocol and bytes on the wire. data ProtocolError- -- | Received an unknown response packet.+ -- | Received an unknown response line. = UnknownPkt { protocolError :: String } -- | Unknown Memcached operation. | UnknownOp { protocolError :: String } -- | Unknown Memcached status field value. | UnknownStatus { protocolError :: String }- -- | Unexpected length of a Memcached field (extras, key, or value).+ -- | Unexpected length of a Meta Text value or other field. | BadLength { protocolError :: String }- -- | Response packet is for a different operation than expected.- | WrongOp { protocolError :: String }+ -- | Response is for a different operation than expected.+ | UnexpectedResponse { protocolError :: String }+ -- | Memcached reported a server failure.+ | ServerError { protocolError :: String }+ -- | Memcached rejected the command.+ | BadCommand { protocolError :: String } -- | Network socket closed without receiving enough bytes. | UnexpectedEOF { protocolError :: String }- deriving (Eq, Show, Typeable)+ deriving (Eq, Show) -- | Convert a status to 'MemcacheError' exception. throwStatus :: Status -> IO a throwStatus s = throwIO $ OpError s---- | Create a properly formatted 'WrongOp' protocol error.-wrongOp :: Response -> String -> MemcacheError-wrongOp r msg = ProtocolError $- WrongOp {- protocolError = "Expected " ++ msg ++ "! Got: " ++ show (resOp r)- }-
− Database/Memcache/SASL.hs
@@ -1,69 +0,0 @@-{-# OPTIONS_GHC -fno-warn-unused-binds #-}--{-|-Module : Database.Memcache.SASL-Description : SASL Authentication-Copyright : (c) David Terei, 2016-License : BSD-Maintainer : code@davidterei.com-Stability : stable-Portability : GHC--SASL authentication support for Memcached.--}-module Database.Memcache.SASL (- -- * Types- Authentication(..), Username, Password,-- -- * Operations- authenticate- ) where--import Database.Memcache.Errors-import Database.Memcache.Socket-import Database.Memcache.Types--import Control.Exception (throwIO)-import Control.Monad-import Data.ByteString.Char8 as B8 (ByteString, pack, singleton)-import Data.Monoid---- | Perform SASL authentication with the server.-authenticate :: Socket -> Authentication -> IO ()-{-# INLINE authenticate #-}-authenticate _ NoAuth = return ()-authenticate s (Auth u p) = saslAuthPlain s u p--- NOTE: For correctness really should check that PLAIN auth is supported first--- but we'll just assume it is as that's all mainline and other implementations--- support and one exception is nearly as good as another.---- | Perform SASL PLAIN authentication.-saslAuthPlain :: Socket -> Username -> Password -> IO ()-{-# INLINE saslAuthPlain #-}-saslAuthPlain s u p = do- let credentials = singleton '\0' <> u <> singleton '\0' <> p- msg = emptyReq { reqOp = ReqSASLStart (B8.pack "PLAIN") credentials }- send s msg- r <- recv s- when (resOp r /= ResSASLStart) $- throwIO $ wrongOp r "SASL_START"- case resStatus r of- NoError -> return ()- rs -> throwIO $ OpError rs---- | List available SASL authentication methods. We could call this but as we--- only support PLAIN as does the Memcached server, we simply assume PLAIN--- authentication is supprted and try that.-saslListMechs :: Socket -> IO B8.ByteString-{-# INLINE saslListMechs #-}-saslListMechs s = do- let msg = emptyReq { reqOp = ReqSASLList }- send s msg- r <- recv s- v <- case resOp r of- ResSASLList v -> return v- _ -> throwIO $ wrongOp r "SASL_LIST"- case resStatus r of- NoError -> return v- rs -> throwIO $ OpError rs-
Database/Memcache/Server.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE CPP #-} {-| Module : Database.Memcache.Server@@ -11,58 +12,51 @@ Handles the connections between a Memcached client and a single server. -Memcached expected errors (part of protocol) are returned in the Response,-unexpected errors (e.g., network failure) are thrown as exceptions. While-the Server datatype supports a `failed` and `failedAt` flag for managing-retries, it's up to consumers to use this.+Memcached expected errors (part of the protocol) are returned in the Response;+unexpected errors, such as network failures, are thrown as exceptions. The+server's 'failed' timestamp is used by cluster retry handling. -} module Database.Memcache.Server ( -- * Server- Server(sid, failed), newServer, sendRecv, withSocket, close- ) where+ Server(sid, failed), newServerDefault, withSocket, close, -import Database.Memcache.SASL-import Database.Memcache.Socket+ -- * ServerOptions+ ServerOptions(..)+ ) where -import Control.Exception-import Data.Hashable-import Data.IORef-import Data.Pool-import Data.Time.Clock (NominalDiffTime)-import Data.Time.Clock.POSIX (POSIXTime)+import Database.Memcache.Auth+import Database.Memcache.Socket -import Network.BSD (getProtocolNumber, getHostByName, hostAddress)-import Network.Socket (HostName, PortNumber(..))-import qualified Network.Socket as S+import Data.Default+import Data.Hashable+import Data.IORef+import qualified Data.Pool as P+import Data.Pool (Pool)+import Data.Time.Clock.POSIX (POSIXTime)+import Database.Memcache.Types (ServerSpec (..))+import UnliftIO.Exception --- Connection pool constants.--- TODO: make configurable-sSTRIPES, sCONNECTIONS :: Int-sKEEPALIVE :: NominalDiffTime-sSTRIPES = 1-sCONNECTIONS = 1-sKEEPALIVE = 300+import Network.Socket (HostName, ServiceName, getAddrInfo)+import qualified Network.Socket as S+#if MIN_VERSION_resource_pool(0,4,0)+#else+import Data.Time.Clock (NominalDiffTime)+#endif -- | Memcached server connection. data Server = Server { -- | ID of server for consistent hashing.- sid :: {-# UNPACK #-} !Int,+ sid :: {-# UNPACK #-} !Int, -- | Connection pool to server.- pool :: Pool Socket,+ pool :: Pool Connection, -- | Hostname of server.- addr :: !HostName,+ addr :: !HostName, -- | Port number of server.- port :: !PortNumber,+ port :: !ServiceName, -- | Credentials for server.- auth :: !Authentication,+ auth :: !Authentication, -- | When did the server fail? 0 if it is alive.- failed :: IORef POSIXTime-- -- TODO: - -- weight :: Double- -- tansport :: Transport (UDP vs. TCP)- -- poolLim :: Int (pooled connection limit)- -- cnxnBuf :: IORef ByteString+ failed :: IORef POSIXTime } instance Show Server where@@ -75,55 +69,88 @@ instance Ord Server where compare x y = compare (sid x) (sid y) --- | Create a new Memcached server connection.-newServer :: HostName -> PortNumber -> Authentication -> IO Server-newServer host port auth = do+-- | Configurable options when creating a @Server@.+--+-- At the moment, this only applies to the @Pool@ information. This can be expanded in the future.+--+data ServerOptions+ = ServerOptions+ -- | Maximum number of pooled connections per stripe.+ { soNumResources :: Int+ -- | Number of pool stripes.+ , soNumStripes :: Int+#if MIN_VERSION_resource_pool(0,4,0)+ -- | Connection keep-alive duration.+ , soKeepAlive :: Double+#else+ -- | Connection keep-alive duration.+ , soKeepAlive :: NominalDiffTime+#endif+ }++instance Default ServerOptions where+ def = ServerOptions+ { soNumResources = 1+ , soNumStripes = 1+ , soKeepAlive = 300+ }++-- | Create a server using the default socket-pool implementation.+newServerDefault :: ServerOptions -> ServerSpec -> IO Server+newServerDefault serverOptions ss@ServerSpec{..} = do fat <- newIORef 0- pSock <- createPool connectSocket releaseSocket- sSTRIPES sKEEPALIVE sCONNECTIONS+ pSock <- getNewPool serverOptions ss return Server { sid = serverHash , pool = pSock- , addr = host- , port = port- , auth = auth+ , addr = ssHost+ , port = ssPort+ , auth = ssAuth , failed = fat } where- serverHash = hash (host, fromEnum port)+ serverHash = hash (ssHost, ssPort) - connectSocket = do- proto <- getProtocolNumber "tcp"- bracketOnError- (S.socket S.AF_INET S.Stream proto)- releaseSocket- (\s -> do- h <- getHostByName host- S.connect s (S.SockAddrInet port $ hostAddress h)- S.setSocketOption s S.KeepAlive 1- S.setSocketOption s S.NoDelay 1- authenticate s auth- return s- ) - releaseSocket = S.close---- | Send and receive a single request/response pair to the Memcached server.-sendRecv :: Server -> Request -> IO Response-{-# INLINE sendRecv #-}-sendRecv svr msg = withSocket svr $ \s -> do- send s msg- recv s---- | Run a function with access to an server socket for using 'send' and--- 'recv'.-withSocket :: Server -> (Socket -> IO a) -> IO a+-- | Run a function with access to a pooled server connection.+withSocket :: Server -> (Connection -> IO a) -> IO a {-# INLINE withSocket #-}-withSocket svr = withResource $ pool svr+withSocket svr = P.withResource $ pool svr --- | Close the server connection. If you perform another operation after this,--- the connection will be re-established.+-- | Close all pooled connections. A later operation re-establishes them. close :: Server -> IO () {-# INLINE close #-}-close srv = destroyAllResources $ pool srv+close srv = P.destroyAllResources $ pool srv +#if MIN_VERSION_resource_pool(0,4,0)+getNewPool :: ServerOptions -> ServerSpec -> IO (Pool Connection)+getNewPool serverOptions ss =+ P.newPool+ $ P.setNumStripes (Just $ soNumStripes serverOptions)+ $ P.defaultPoolConfig (connectSocket ss) releaseSocket (soKeepAlive serverOptions) (soNumResources serverOptions)+#else+getNewPool :: ServerOptions -> ServerSpec -> IO (Pool Connection)+getNewPool serverOptions ss =+ P.createPool (connectSocket ss) releaseSocket (soNumStripes serverOptions) (soKeepAlive serverOptions) (soNumResources serverOptions)+#endif++connectSocket :: ServerSpec -> IO Connection+connectSocket ServerSpec{..} = do+ let hints = S.defaultHints {+ S.addrSocketType = S.Stream+ }+ addr:_ <- getAddrInfo (Just hints) (Just ssHost) (Just ssPort)+ bracketOnError+ (S.socket (S.addrFamily addr) (S.addrSocketType addr) (S.addrProtocol addr))+ S.close+ (\s -> do+ S.connect s $ S.addrAddress addr+ S.setSocketOption s S.KeepAlive 1+ S.setSocketOption s S.NoDelay 1+ buffered <- newConnection s+ authenticate buffered ssAuth+ return buffered+ )++releaseSocket :: Connection -> IO ()+releaseSocket = closeConnection
Database/Memcache/Socket.hs view
@@ -1,427 +1,400 @@-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE CPP #-}-{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-}+{-# LANGUAGE OverloadedStrings #-} {-| Module : Database.Memcache.Socket-Description : Connection Handling+Description : Meta Text protocol socket handling Copyright : (c) David Terei, 2016 License : BSD Maintainer : code@davidterei.com Stability : stable Portability : GHC -Handles a single Memcached connection, sending and receiving requests.+Low-level connections, replies, and request encoders for Memcached's Meta Text+protocol. Authentication uses the text auth-file command. -} module Database.Memcache.Socket ( -- * Types- Socket, Request(..), Response(..),+ ArithMode(..), Connection, Response(..), StoreMode(..), WireKey(..),+ crlf, -- * Operations- send, recv,+ newConnection, closeConnection,+ sendRaw,+ recvExact, recvLine, recvRawUntil, recvResponse, - -- * Serialization / Deserialization- szRequest, szResponse, dzHeader, dzResponse- ) where+ -- * Response readers+ responseStatus, responseCas, responseFlags, responseNumber, --- FIXME: Wire works with lazy bytestrings but we receive strict bytestrings--- from the network...+ -- * Constructors+ encodeKey, getRequest, getManyRequest, gatRequest, touchRequest,+ storeRequest, arithmeticRequest, deleteRequest, flushRequest,+ versionRequest, modifyRequest, statsRequest, quitRequest, authRequest,+ noOpRequest+ ) where import Database.Memcache.Errors import Database.Memcache.Types -import Blaze.ByteString.Builder-#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif-import Control.Exception (throw, throwIO)-import Control.Monad-import Data.Binary.Get-import qualified Data.ByteString as B-import qualified Data.ByteString.Lazy as L-import Data.Monoid-import Data.Word-import Network.Socket (Socket, isConnected, isReadable)+import Control.Monad (unless)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import qualified Data.ByteString.Base64 as Base64+import Data.ByteString.Builder (Builder, byteString, intDec, word32Dec, word64Dec)+import Data.ByteString.Builder.Extra (byteStringCopy, smallChunkSize, toLazyByteStringWith,+ untrimmedStrategy)+import qualified Data.ByteString.Char8 as C+import qualified Data.ByteString.Lazy as BL+import Data.IORef+import Data.List (find, isInfixOf)+import Data.Maybe (fromMaybe)+import Data.Word (Word64)+import qualified Network.Socket as N import qualified Network.Socket.ByteString as N+import Text.Read (readMaybe)+import UnliftIO.Exception (throwIO) --- | Send a request to the Memcached server.-send :: Socket -> Request -> IO ()-{-# INLINE send #-}-send s m = N.sendAll s (toByteString $ szRequest m)+-- | A connected Memcached socket and its receive buffer.+data Connection = Connection N.Socket (IORef ByteString) --- | Retrieve a single response from the Memcached server.--- FIXME: read into buffer to minimize read syscalls-recv :: Socket -> IO Response-{-# INLINE recv #-}-recv s = do- header <- recvAll mEMCACHE_HEADER_SIZE mempty- let h = runGet (dzHeader PktResponse) (L.fromChunks [header])- if bodyLen h > 0- then do- let bytesToRead = fromIntegral $ bodyLen h- body <- recvAll bytesToRead mempty- return $ dzResponse h (L.fromChunks [body])- else return $ dzResponse h L.empty- where- recvAll :: Int -> Builder -> IO B.ByteString- recvAll 0 !acc = return $! toByteString acc- recvAll !n !acc = do- canRead <- isSocketActive s- if canRead- then do- buf <- N.recv s n- case B.length buf of- 0 -> throwIO errEOF- bl | bl == n ->- return $! (toByteString $! acc <> fromByteString buf)- bl -> recvAll (n - bl) (acc <> fromByteString buf)- else throwIO errEOF- - errEOF :: MemcacheError- errEOF = ProtocolError UnexpectedEOF { protocolError = "" }+-- | A decoded Meta Text response.+data Response = Response+ -- | The response code, such as @VA@ or @HD@.+ { responseCode :: ByteString+ -- | Tokens following the response code.+ , responseTokens :: [ByteString]+ -- | The response value, when the response contains one.+ , responseValue :: ByteString+ } deriving (Eq, Show) --- | Check whether we can still operate on this socket or not.-isSocketActive :: Socket -> IO Bool-{-# INLINE isSocketActive #-}-isSocketActive s = (&&) <$> isConnected s <*> isReadable s+-- | Create a connection wrapper around a socket.+newConnection :: N.Socket -> IO Connection+newConnection s = Connection s <$> newIORef B.empty --- | Serialize a response to a ByteString Builder.-szResponse :: Response -> Builder-szResponse res =- fromWord8 0x81- <> fromWord8 c- <> fromWord16be (fromIntegral keyl)- <> fromWord8 (fromIntegral extl)- <> fromWord8 0- <> fromWord16be 0- <> fromWord32be (fromIntegral $ extl + keyl + vall)- <> fromWord32be (resOpaque res)- <> fromWord64be (resCas res)- <> ext'- <> key'- <> val'- where- (c, k', v', ext', extl) = szOpResponse (resOp res)- (keyl, key') = case k' of- Just k -> (B.length k, fromByteString k)- Nothing -> (0, mempty)- (vall, val') = case v' of- Just v -> (B.length v, fromByteString v)- Nothing -> (0, mempty)+-- | Close the underlying socket.+closeConnection :: Connection -> IO ()+closeConnection (Connection s _) = N.close s -szOpResponse :: OpResponse -> (Word8, Maybe Key, Maybe Value, Builder, Int)-szOpResponse o = case o of- ResGet Loud v f -> (0x00, Nothing, Just v, fromWord32be f, 4)- ResGet Quiet v f -> (0x09, Nothing, Just v, fromWord32be f, 4)- ResGetK Loud k v f -> (0x0C, Just k, Just v, fromWord32be f, 4)- ResGetK Quiet k v f -> (0x0D, Just k, Just v, fromWord32be f, 4)- ResSet Loud -> (0x01, Nothing, Nothing, mempty, 0)- ResSet Quiet -> (0x11, Nothing, Nothing, mempty, 0)- ResAdd Loud -> (0x02, Nothing, Nothing, mempty, 0)- ResAdd Quiet -> (0x12, Nothing, Nothing, mempty, 0)- ResReplace Loud -> (0x03, Nothing, Nothing, mempty, 0)- ResReplace Quiet -> (0x13, Nothing, Nothing, mempty, 0)- ResDelete Loud -> (0x04, Nothing, Nothing, mempty, 0)- ResDelete Quiet -> (0x14, Nothing, Nothing, mempty, 0)- ResIncrement Loud f -> (0x05, Nothing, Nothing, fromWord64be f, 8)- ResIncrement Quiet f -> (0x15, Nothing, Nothing, fromWord64be f, 8)- ResDecrement Loud f -> (0x06, Nothing, Nothing, fromWord64be f, 8)- ResDecrement Quiet f -> (0x16, Nothing, Nothing, fromWord64be f, 8)- ResAppend Loud -> (0x0E, Nothing, Nothing, mempty, 0)- ResAppend Quiet -> (0x19, Nothing, Nothing, mempty, 0)- ResPrepend Loud -> (0x0F, Nothing, Nothing, mempty, 0)- ResPrepend Quiet -> (0x1A, Nothing, Nothing, mempty, 0)- ResTouch -> (0x1C, Nothing, Nothing, mempty, 0)- ResGAT Loud v f -> (0x1D, Nothing, Just v, fromWord32be f, 4)- ResGAT Quiet v f -> (0x1E, Nothing, Just v, fromWord32be f, 4)- ResGATK Loud k v f -> (0x23, Just k, Just v, fromWord32be f, 4)- ResGATK Quiet k v f -> (0x24, Just k, Just v, fromWord32be f, 4)- ResFlush Loud -> (0x08, Nothing, Nothing, mempty, 0)- ResFlush Quiet -> (0x18, Nothing, Nothing, mempty, 0)- ResNoop -> (0x0A, Nothing, Nothing, mempty, 0)- ResVersion v -> (0x0B, Nothing, Just v, mempty, 0)- ResStat k v -> (0x10, Just k, Just v, mempty, 0)- ResQuit Loud -> (0x07, Nothing, Nothing, mempty, 0)- ResQuit Quiet -> (0x17, Nothing, Nothing, mempty, 0)- ResSASLList v -> (0x20, Nothing, Just v, mempty, 0)- ResSASLStart -> (0x21, Nothing, Nothing, mempty, 0)- ResSASLStep -> (0x22, Nothing, Nothing, mempty, 0)+-- | Send raw bytes on the underlying socket.+sendRaw :: Connection -> ByteString -> IO ()+sendRaw (Connection s _) = N.sendAll s --- | Serialize a request to a ByteString Builder.-szRequest :: Request -> Builder-szRequest req =- fromWord8 0x80- <> fromWord8 c- <> fromWord16be (fromIntegral keyl)- <> fromWord8 (fromIntegral extl)- <> fromWord8 0- <> fromWord16be 0- <> fromWord32be (fromIntegral $ extl + keyl + vall)- <> fromWord32be (reqOpaque req)- <> fromWord64be (reqCas req)- <> ext'- <> key'- <> val'+-- | Receive and decode one Meta Text response.+recvResponse :: Connection -> IO Response+recvResponse s = do+ line <- recvLine s+ case C.words line of+ [] -> throwIO $ ProtocolError UnknownPkt { protocolError = "" }+ code:tokens -> case tokens of+ n:_ | code == "VA" -> do+ value <- recvExact s =<< parseLength n+ trailer <- recvLine s+ unless (trailer == "\r") $ throwIO $ ProtocolError BadLength+ { protocolError = "value is not followed by an empty line" }+ return $ Response code tokens value+ _ -> return $ Response code tokens B.empty where- (c, k', v', ext', extl) = szOpRequest (reqOp req)- (keyl, key') = case k' of- Just k -> (B.length k, fromByteString k)- Nothing -> (0, mempty)- (vall, val') = case v' of- Just v -> (B.length v, fromByteString v)- Nothing -> (0, mempty)+ parseLength x = case readMaybe (C.unpack x) of+ Just n | n >= 0 && n <= 128 * 1024 * 1024 -> return n+ _ -> throwIO $ ProtocolError BadLength { protocolError = "invalid value length" } --- Extract needed info from an OpRequest for serialization.--- FIXME: Make sure this is optimized well (no tuple, boxing, unboxing, inlined)-szOpRequest :: OpRequest -> (Word8, Maybe Key, Maybe Value, Builder, Int)-szOpRequest o = case o of- -- FIXME: make sure this isn't a thunk! (c)- ReqGet q k key -> let c | q == Loud && k == NoKey = 0x00- | q == Loud && k == IncludeKey = 0x0C- | k == NoKey = 0x09 -- Quiet- | otherwise = 0x0D -- Quiet && IncludeKey- in (c, Just key, Nothing, mempty, 0)+-- | Convert a response code and tokens to the corresponding status.+responseStatus :: Response -> IO Status+responseStatus (Response code tokens _) = case code of+ "VA" -> return NoError+ "HD" -> return NoError+ "EN" -> return ErrKeyNotFound+ "NF" -> return ErrKeyNotFound+ "EX" -> return ErrKeyExists+ "NS" -> return ErrItemNotStored+ "ST" -> return NoError+ "STORED" -> return NoError+ "VERSION" -> return NoError+ "STAT" -> return NoError+ "OK" -> return NoError+ "END" -> return NoError+ "MN" -> return NoError+ "SERVER_ERROR" -> serverFailure+ "CLIENT_ERROR" -> clientFailure+ "ERROR" -> throwIO $ ProtocolError UnknownOp { protocolError = "ERROR" }+ _ -> throwIO $ ProtocolError UnknownStatus { protocolError = C.unpack code }+ where+ message = C.unpack $ C.unwords tokens - ReqSet Loud key v e -> (0x01, Just key, Just v, szSESet e, 8)- ReqSet Quiet key v e -> (0x11, Just key, Just v, szSESet e, 8)- ReqAdd Loud key v e -> (0x02, Just key, Just v, szSESet e, 8)- ReqAdd Quiet key v e -> (0x12, Just key, Just v, szSESet e, 8)- ReqReplace Loud key v e -> (0x03, Just key, Just v, szSESet e, 8)- ReqReplace Quiet key v e -> (0x13, Just key, Just v, szSESet e, 8)- ReqDelete Loud key -> (0x04, Just key, Nothing, mempty, 0)- ReqDelete Quiet key -> (0x14, Just key, Nothing, mempty, 0)- ReqIncrement Loud key e -> (0x05, Just key, Nothing, szSEIncr e, 20)- ReqIncrement Quiet key e -> (0x15, Just key, Nothing, szSEIncr e, 20)- ReqDecrement Loud key e -> (0x06, Just key, Nothing, szSEIncr e, 20)- ReqDecrement Quiet key e -> (0x16, Just key, Nothing, szSEIncr e, 20)- ReqAppend Loud key v -> (0x0E, Just key, Just v, mempty, 0)- ReqAppend Quiet key v -> (0x19, Just key, Just v, mempty, 0)- ReqPrepend Loud key v -> (0x0F, Just key, Just v, mempty, 0)- ReqPrepend Quiet key v -> (0x1A, Just key, Just v, mempty, 0)- ReqTouch key e -> (0x1C, Just key, Nothing, szSETouch e, 4)- -- FIXME: beware allocation.- ReqGAT q k key e -> let c | q == Quiet && k == IncludeKey = 0x24- | q == Quiet && k == NoKey = 0x1E- | k == IncludeKey = 0x23- | otherwise = 0x1D- in (c, Just key, Nothing, szSETouch e, 4)- ReqFlush Loud (Just e) -> (0x08, Nothing, Nothing, szSETouch e, 4)- ReqFlush Quiet (Just e) -> (0x18, Nothing, Nothing, szSETouch e, 4)- ReqFlush Loud Nothing -> (0x08, Nothing, Nothing, mempty, 0)- ReqFlush Quiet Nothing -> (0x18, Nothing, Nothing, mempty, 0)- ReqNoop -> (0x0A, Nothing, Nothing, mempty, 0)- ReqVersion -> (0x0B, Nothing, Nothing, mempty, 0)- ReqStat key -> (0x10, key, Nothing, mempty, 0)- ReqQuit Loud -> (0x07, Nothing, Nothing, mempty, 0)- ReqQuit Quiet -> (0x17, Nothing, Nothing, mempty, 0)- ReqSASLList -> (0x20, Nothing, Nothing, mempty, 0)- ReqSASLStart key v -> (0x21, Just key, Just v, mempty, 0)- ReqSASLStep key v -> (0x22, Just key, Just v, mempty, 0)+ -- Meta Text reports these as free-form error text where the binary+ -- protocol had dedicated status codes; match them back so callers keep+ -- seeing 'OpError' rather than an opaque protocol failure.+ serverFailure+ | "object too large" `isInfixOf` message = throwStatus ErrValueTooLarge+ | "out of memory" `isInfixOf` message = throwStatus ErrOutOfMemory+ | otherwise = throwIO $ ProtocolError ServerError { protocolError = message } - -- XXX: Should kill in future, ugly- ReqRaw c k v (SERaw e n) -> (c, k, v, e, n)+ clientFailure+ | "non-numeric" `isInfixOf` message = throwStatus ErrValueNonNumeric+ | otherwise = throwIO $ ProtocolError BadCommand { protocolError = message } - where- szSESet (SESet f e) = fromWord32be f <> fromWord32be e- szSEIncr (SEIncr i d e) = fromWord64be i <> fromWord64be d <> fromWord32be e- szSETouch (SETouch e) = fromWord32be e+-- | Extract the CAS version from a response, or zero when absent or invalid.+responseCas :: Response -> Version+responseCas = token64 "c" . responseTokens --- | Deserialize a Header from a ByteString.-dzHeader :: PktType -> Get Header-{-# INLINE dzHeader #-}-dzHeader pkt = do- m <- getWord8- case pkt of- PktResponse -> when (m /= 0x81) $- throw $ ProtocolError UnknownPkt { protocolError = show m }- PktRequest -> when (m /= 0x80) $- throw $ ProtocolError UnknownPkt { protocolError = show m }- o <- getWord8- kl <- getWord16be- el <- getWord8- skip 1 -- unused data type field- st <- dzStatus- bl <- getWord32be- opq <- getWord32be- ver <- getWord64be- return Header {- op = o,- keyLen = kl,- extraLen = el,- status = st,- bodyLen = bl,- opaque = opq,- cas = ver- }+-- | Extract the flags from a response, or zero when absent or invalid.+responseFlags :: Response -> Flags+responseFlags = fromIntegral . token64 "f" . responseTokens --- | Deserialize a Response body.-dzResponse :: Header -> L.ByteString -> Response-dzResponse h = runGet $- case op h of- 0x00 -> dzGetResponse h $ ResGet Loud- 0x09 -> dzGetResponse h $ ResGet Quiet- 0x1D -> dzGetResponse h $ ResGAT Loud- 0x1E -> dzGetResponse h $ ResGAT Quiet- 0x0C -> dzGetKResponse h $ ResGetK Loud- 0x0D -> dzGetKResponse h $ ResGetK Quiet- 0x23 -> dzGetKResponse h $ ResGATK Loud- 0x24 -> dzGetKResponse h $ ResGATK Quiet- 0x05 -> dzNumericResponse h $ ResIncrement Loud- 0x15 -> dzNumericResponse h $ ResIncrement Quiet- 0x06 -> dzNumericResponse h $ ResDecrement Loud- 0x16 -> dzNumericResponse h $ ResDecrement Quiet- 0x01 -> dzGenericResponse h $ ResSet Loud- 0x11 -> dzGenericResponse h $ ResSet Quiet- 0x02 -> dzGenericResponse h $ ResAdd Loud- 0x12 -> dzGenericResponse h $ ResAdd Quiet- 0x03 -> dzGenericResponse h $ ResReplace Loud- 0x13 -> dzGenericResponse h $ ResReplace Quiet- 0x04 -> dzGenericResponse h $ ResDelete Loud- 0x14 -> dzGenericResponse h $ ResDelete Quiet- 0x0E -> dzGenericResponse h $ ResAppend Loud- 0x19 -> dzGenericResponse h $ ResAppend Quiet- 0x0F -> dzGenericResponse h $ ResPrepend Loud- 0x1A -> dzGenericResponse h $ ResPrepend Quiet- 0x1C -> dzGenericResponse h ResTouch- 0x07 -> dzGenericResponse h $ ResQuit Loud- 0x17 -> dzGenericResponse h $ ResQuit Quiet- 0x08 -> dzGenericResponse h $ ResFlush Loud- 0x18 -> dzGenericResponse h $ ResFlush Quiet- 0x0A -> dzGenericResponse h ResNoop- 0x10 -> dzKeyValueResponse h ResStat- 0x0B -> dzValueResponse h ResVersion- -- SASL- 0x20 -> dzValueResponse h ResSASLList- 0x21 -> dzGenericResponse h ResSASLStart- 0x22 -> dzGenericResponse h ResSASLStep+-- | Parse the numeric value in a response, or zero when it is absent or invalid.+responseNumber :: Response -> Word64+responseNumber = fromMaybe 0 . readMaybe . C.unpack . responseValue - _ -> throw $ ProtocolError UnknownOp { protocolError = show (op h) }+token64 :: ByteString -> [ByteString] -> Word64+token64 prefix xs = fromMaybe 0 $ do+ token <- find (B.isPrefixOf prefix) xs+ readMaybe (C.unpack (B.drop (B.length prefix) token)) --- | Deserialize the body of a Response that contains nothing.-dzGenericResponse :: Header -> OpResponse -> Get Response-dzGenericResponse h o = do- skip (fromIntegral $ bodyLen h)- chkLength 0 (extraLen h) "Extra"- chkLength 0 (keyLen h) "Key"- chkLength 0 (bodyLen h) "Body"- return Res {- resOp = o,- resStatus = status h,- resOpaque = opaque h,- resCas = cas h- }+-- | Longest response line accepted before assuming the stream is desynchronised.+maxLineLength :: Int+maxLineLength = 8 * 1024 --- | Deserialize the body of a Get Response (Extras [flags] & Value).-dzGetResponse :: Header -> (Value -> Flags -> OpResponse) -> Get Response-dzGetResponse h o = do- e <- if status h == NoError && el == 4- then getWord32be- else skip el >> return 0- v <- getByteString vl- chkLength 4 el "Extra"- chkLength 0 (keyLen h) "Key"- return Res {- resOp = o v e,- resStatus = status h,- resOpaque = opaque h,- resCas = cas h- }+-- | Receive one CRLF-terminated line, without the line-feed byte.+recvLine :: Connection -> IO ByteString+recvLine s = do+ buffer <- readBuffer s+ case B.elemIndex 10 buffer of+ Just i -> writeBuffer s (B.drop (i + 1) buffer) >> return (B.take i buffer)+ Nothing+ | B.length buffer > maxLineLength ->+ throwIO $ ProtocolError BadLength { protocolError = "response line too long" }+ | otherwise -> readChunk 0 s >> recvLine s++-- | Receive exactly the requested number of bytes.+recvExact :: Connection -> Int -> IO ByteString+recvExact s n+ | n < 0 = throwIO $ ProtocolError BadLength { protocolError = "negative length" }+ | otherwise = B.concat <$> go n [] where- el = fromIntegral $ extraLen h- vl = fromIntegral (bodyLen h) - el- --- | Deserialize the body of a GetK Response (Extras [flags] & Key & Value).-dzGetKResponse :: Header -> (Key -> Value -> Flags -> OpResponse) -> Get Response-dzGetKResponse h o = do- e <- if status h == NoError && el == 4- then getWord32be- else skip el >> return 0- k <- getByteString kl- v <- getByteString vl- chkLength 4 el "Extra"- -- FIXME: check strictness ($!)- return Res {- resOp = o k v e,- resStatus = status h,- resOpaque = opaque h,- resCas = cas h- }+ go 0 acc = return (reverse acc)+ go wanted acc = do+ buffer <- readBuffer s+ let takeN = min wanted (B.length buffer)+ if takeN > 0+ then writeBuffer s (B.drop takeN buffer) >> go (wanted - takeN) (B.take takeN buffer : acc)+ else readChunk wanted s >> go wanted acc++-- | Receive through the first occurrence of a delimiter, including it.+recvRawUntil :: ByteString -> Connection -> IO ByteString+recvRawUntil end s = go B.empty where- el = fromIntegral $ extraLen h- kl = fromIntegral $ keyLen h- vl = fromIntegral (bodyLen h) - el - kl+ go acc = do+ buffer <- readBuffer s+ let combined = acc <> buffer+ case B.breakSubstring end combined of+ (before, rest) | not (B.null rest) -> do+ writeBuffer s (B.drop (B.length end) rest)+ return (before <> end)+ _ -> writeBuffer s B.empty >> readChunk 0 s >> go combined --- | Deserialize the body of a Incr/Decr Response (Value [Word64]).-dzNumericResponse :: Header -> (Word64 -> OpResponse) -> Get Response-dzNumericResponse h o = do- v <- if status h == NoError && bodyLen h == 8- then getWord64be- else skip (fromIntegral $ bodyLen h) >> return 0- chkLength 0 (extraLen h) "Extra"- chkLength 0 (keyLen h) "Key"- chkLength 8 (bodyLen h) "Body"- return Res {- resOp = o v,- resStatus = status h,- resOpaque = opaque h,- resCas = cas h- }+readChunk :: Int -> Connection -> IO ()+readChunk n (Connection s ref) = do+ chunk <- N.recv s (max 4096 (min maxRecvSize n))+ if B.null chunk then throwIO eofError else modifyIORef' ref (<> chunk) --- | Deserialize the body of a general response that just has a value (no key--- or extras).-dzValueResponse :: Header -> (Value -> OpResponse) -> Get Response-dzValueResponse h o = do- v <- getByteString (fromIntegral $ bodyLen h)- chkLength 0 (extraLen h) "Extra"- chkLength 0 (keyLen h) "Key"- return Res {- resOp = o v,- resStatus = status h,- resOpaque = opaque h,- resCas = cas h- }+-- | Largest single receive. 'N.recv' allocates a buffer of the full requested+-- size on every call, so asking for the whole remainder of a large value+-- turns reading it into quadratic allocation.+maxRecvSize :: Int+maxRecvSize = 64 * 1024 --- | Deserialize the body of a general response that just has a key and value--- (no extras).-dzKeyValueResponse :: Header -> (Key -> Value -> OpResponse) -> Get Response-dzKeyValueResponse h o = do- k <- getByteString kl- v <- getByteString vl- chkLength 0 (extraLen h) "Extra"- return Res {- resOp = o k v,- resStatus = status h,- resOpaque = opaque h,- resCas = cas h- }+readBuffer :: Connection -> IO ByteString+readBuffer (Connection _ ref) = readIORef ref++writeBuffer :: Connection -> ByteString -> IO ()+writeBuffer (Connection _ ref) = writeIORef ref++eofError :: MemcacheError+eofError = ProtocolError UnexpectedEOF { protocolError = "" }++-- | A key encoded for the Meta Text wire protocol.+data WireKey = WireKey+ -- | The bytes sent as the key.+ { wireKey :: !ByteString+ -- | Whether the key uses the protocol's base64 binary-key mode.+ , wireB64 :: !Bool+ } deriving (Eq, Show)++-- | Encode a key, using base64 when it contains protocol-invalid bytes.+encodeKey :: Key -> Either ClientError WireKey+encodeKey key+ | B.null key = Left (KeyTooLong "empty key")+ | B.length key > 250 = Left (KeyTooLong "key exceeds 250 bytes")+ | B.length wire > 250 = Left (KeyTooLong "encoded key exceeds 250 bytes")+ | otherwise = Right (WireKey wire b64) where- kl = fromIntegral $ keyLen h- vl = fromIntegral (bodyLen h) - kl+ b64 = B.any (\c -> c <= 0x20 || c == 0x7f) key+ wire = if b64 then Base64.encode key else key --- | Deserialize a Response status code.-dzStatus :: Get Status-dzStatus = do- st <- getWord16be- return $ case st of- 0x00 -> NoError- 0x01 -> ErrKeyNotFound- 0x02 -> ErrKeyExists- 0x03 -> ErrValueTooLarge- 0x04 -> ErrInvalidArgs- 0x05 -> ErrItemNotStored- 0x06 -> ErrValueNonNumeric- 0x81 -> ErrUnknownCommand- 0x82 -> ErrOutOfMemory- 0x20 -> SaslAuthFail- 0x21 -> SaslAuthContinue- _ -> throw $ ProtocolError UnknownStatus { protocolError = show st }+keyArgs :: WireKey -> ByteString+keyArgs (WireKey wire b64) = wire <> if b64 then " b" else B.empty --- | Check the length of a header field is as expected, throwing a--- ProtocolError exception if it is not.-chkLength :: (Eq a, Show a) => a -> a -> String -> Get ()-{-# INLINE chkLength #-}-chkLength expected l msg = when (l /= expected) $- return $ throw $ ProtocolError BadLength { protocolError =- msg ++ " length expected " ++ show expected ++ " got " ++ show l- }+keyFlag :: WireKey -> ByteString+keyFlag (WireKey _ b64) = if b64 then " b" else B.empty +-- | Write a request that carries a value or a number into one buffer. The+-- value is copied once (each strict 'ByteString' append copies both sides, so+-- a value at the end of a chain of appends is copied once per piece before+-- it), and numbers are written in place rather than through 'show'. The size+-- is the length of the request's key and value. Values go in with+-- 'byteStringCopy', as 'byteString' would add a large one as a chunk of its+-- own for 'BL.toStrict' to copy again. Inlined so the pieces are written+-- directly instead of first being allocated as closures.+--+-- Requests with neither keep plain appends, which are cheaper for a few short+-- strings than running a 'Builder'.+render :: Int -> Builder -> ByteString+{-# INLINE render #-}+render size = BL.toStrict . toLazyByteStringWith strategy BL.empty+ where strategy = untrimmedStrategy (size + requestHeadroom) smallChunkSize++-- | Room for everything in a request besides its key and value: the command,+-- its meta flags and up to four numbers of at most 20 digits. Room left over+-- stays with the request rather than being trimmed off by another copy. A+-- request that needed more would still be encoded correctly, but copied once+-- more.+requestHeadroom :: Int+requestHeadroom = 128++-- | A piece of a request. Fixed pieces are 'ByteString' literals passed+-- through this rather than 'Builder' literals, which are encoded a character+-- at a time and measured noticeably slower.+bytes :: ByteString -> Builder+bytes = byteString++-- | CRLF used to terminate Meta Text requests.+crlf :: ByteString+crlf = "\r\n"++-- | Meta Text storage operation.+data StoreMode+ -- | Store whether or not the key exists.+ = Set+ -- | Store only when the key does not exist.+ | Add+ -- | Store only when the key exists.+ | Replace+ -- | Add bytes to the end of an existing value.+ | Append+ -- | Add bytes to the beginning of an existing value.+ | Prepend+ deriving (Eq, Show)++-- | Meta Text arithmetic operation.+data ArithMode+ -- | Increment a numeric value.+ = Incr+ -- | Decrement a numeric value.+ | Decr+ deriving (Eq, Show)++storeMode :: StoreMode -> ByteString+storeMode Set = "S"+storeMode Add = "E"+storeMode Replace = "R"+storeMode Append = "A"+storeMode Prepend = "P"++arithMode :: ArithMode -> ByteString+arithMode Incr = "I"+arithMode Decr = "D"++cmdMetaGet, cmdMetaSet, cmdMetaDelete, cmdMetaArith, cmdMetaNoOp :: ByteString+cmdMetaGet = "mg"+cmdMetaSet = "ms"+cmdMetaDelete = "md"+cmdMetaArith = "ma"+cmdMetaNoOp = "mn"++cmdFlush, cmdVersion, cmdQuit, cmdStats, cmdSet :: ByteString+cmdFlush = "flush_all"+cmdVersion = "version"+cmdQuit = "quit"+cmdStats = "stats"+cmdSet = "set"++cas :: Version -> Builder+cas n = if n == 0 then mempty else bytes " C" <> word64Dec n++-- | Encode a get request for one key.+getRequest :: WireKey -> ByteString+getRequest key = cmdMetaGet <> " " <> keyArgs key <> " v f c" <> crlf++-- | Encode a quiet multi-get request for one key.+getManyRequest :: WireKey -> ByteString+getManyRequest key = cmdMetaGet <> " " <> keyArgs key <> " v f c k q" <> crlf++-- | Encode a get-and-touch request.+gatRequest :: WireKey -> Expiration -> ByteString+gatRequest key expiry = render (B.length (wireKey key)) $+ bytes cmdMetaGet <> bytes " " <> bytes (keyArgs key) <> bytes " v f c T" <> word32Dec expiry+ <> bytes crlf++-- | Encode a touch request.+touchRequest :: WireKey -> Expiration -> ByteString+touchRequest key expiry = render (B.length (wireKey key)) $+ bytes cmdMetaGet <> bytes " " <> bytes (keyArgs key) <> bytes " T" <> word32Dec expiry+ <> bytes " c" <> bytes crlf++-- | Encode a storage request.+storeRequest :: StoreMode -> WireKey -> Value -> Flags -> Expiration -> Version -> ByteString+storeRequest mode key value flags expiry version = render (B.length (wireKey key) + B.length value) $+ bytes cmdMetaSet <> bytes " " <> bytes (wireKey key) <> bytes " " <> intDec (B.length value)+ <> bytes " F" <> word32Dec flags <> bytes " T" <> word32Dec expiry <> cas version+ <> bytes " M" <> bytes (storeMode mode) <> bytes (keyFlag key) <> bytes " c"+ <> bytes crlf <> byteStringCopy value <> bytes crlf++-- | Encode an increment or decrement request. Unless the expiration is+-- 'maxBound', the request asks the server to create a missing key with the+-- initial value and that expiration.+arithmeticRequest :: ArithMode -> WireKey -> Initial -> Delta -> Expiration -> Version -> ByteString+arithmeticRequest mode key initial delta expiry version = render (B.length (wireKey key)) $+ bytes cmdMetaArith <> bytes " " <> bytes (keyArgs key) <> bytes " D" <> word64Dec delta+ <> cas version <> bytes " M" <> bytes (arithMode mode) <> bytes " v c"+ <> (if expiry == maxBound+ then mempty+ else bytes " N" <> word32Dec expiry <> bytes " J" <> word64Dec initial)+ <> bytes crlf++-- | Encode an append or prepend request.+modifyRequest :: StoreMode -> WireKey -> Value -> Version -> ByteString+modifyRequest mode key value version = render (B.length (wireKey key) + B.length value) $+ bytes cmdMetaSet <> bytes " " <> bytes (wireKey key) <> bytes " " <> intDec (B.length value)+ <> bytes " M" <> bytes (storeMode mode) <> bytes (keyFlag key) <> cas version <> bytes " c"+ <> bytes crlf <> byteStringCopy value <> bytes crlf++-- | Encode a delete request.+deleteRequest :: WireKey -> Version -> ByteString+deleteRequest key 0 = cmdMetaDelete <> " " <> keyArgs key <> crlf+deleteRequest key version = render (B.length (wireKey key)) $+ bytes cmdMetaDelete <> bytes " " <> bytes (keyArgs key) <> cas version <> bytes crlf++-- | Encode a flush request, optionally delayed until the given expiration.+flushRequest :: Maybe Expiration -> ByteString+flushRequest Nothing = cmdFlush <> crlf+flushRequest (Just expiry) = render 0 $ bytes cmdFlush <> bytes " " <> word32Dec expiry <> bytes crlf++-- | Encode a version request.+versionRequest, quitRequest :: ByteString+versionRequest = cmdVersion <> crlf+quitRequest = cmdQuit <> crlf++-- | Encode a stats request, optionally scoped to a key.+statsRequest :: Maybe Key -> ByteString+statsRequest Nothing = cmdStats <> crlf+statsRequest (Just key) = cmdStats <> " " <> key <> crlf++-- | Encode the text auth-file request, authenticating using the value as its+-- space-separated credentials.+authRequest :: Value -> ByteString+authRequest value = render (B.length value) $+ bytes cmdSet <> bytes " auth 0 0 " <> intDec (B.length value)+ <> bytes crlf <> byteStringCopy value <> bytes crlf++-- | Encode a Meta Text no-op request.+noOpRequest :: ByteString+noOpRequest = cmdMetaNoOp <> crlf
Database/Memcache/Types.hs view
@@ -11,195 +11,52 @@ representation of the protocol. -} module Database.Memcache.Types (- -- * SASL Authentication+ -- * Authentication Authentication(..), Username, Password, -- * Fields & Values- Q(..), K(..), Key, Value, Extras, Initial, Delta, Expiration, Flags,+ Key, Value, Initial, Delta, Expiration, Flags, Version, Status(..),-- -- * Header- Header(..), mEMCACHE_HEADER_SIZE, PktType(..),-- -- * Requests- Request(..), OpRequest(..), SESet(..), SEIncr(..), SETouch(..),- SERaw(..), emptyReq,-- -- * Responses- Response(..), OpResponse(..), emptyRes+ ServerSpec(..), parseServerSpec ) where -import Blaze.ByteString.Builder (Builder)-import Data.ByteString (ByteString)-import Data.Word---- | SASL Authentication information for a server.-data Authentication- = Auth { username :: !Username, password :: !Password }- | NoAuth- deriving (Eq, Show)---- | Username for authentication.-type Username = ByteString---- | Password for authentication.-type Password = ByteString--{- MEMCACHED MESSAGE:-- header {- magic :: Word8- op :: Word8- keyLen :: Word16- extraLen :: Word8- datatype :: Word8- status / reserved :: Word16- bodyLen :: Word32 (total body length)- opaque :: Word32- cas :: Word64- }- extras :: ByteString- key :: ByteString- value :: ByteString- -}--mEMCACHE_HEADER_SIZE :: Int-mEMCACHE_HEADER_SIZE = 24---- | Memcached packet header (for both 'Request' and 'Response').-data Header = Header {- op :: Word8,- keyLen :: Word16,- extraLen :: Word8,- status :: Status,- bodyLen :: Word32,- opaque :: Word32,- cas :: Version- } deriving (Eq, Show)--data PktType = PktRequest | PktResponse- deriving (Eq, Show)+import Data.ByteString (ByteString)+import Data.Word+import Database.Memcache.Types.Authentication (Authentication(..), Username, Password)+import Database.Memcache.Types.ServerSpec (ServerSpec(..), parseServerSpec) -data Q = Loud | Quiet deriving (Eq, Show)-data K = NoKey | IncludeKey deriving (Eq, Show)+-- | Memcached key bytes. type Key = ByteString+-- | Memcached value bytes. type Value = ByteString-type Extras = ByteString+-- | Initial value used when creating an arithmetic counter. type Initial = Word64+-- | Arithmetic increment or decrement amount. type Delta = Word64+-- | Expiration in protocol time units. type Expiration = Word32+-- | Application-defined value flags. type Flags = Word32+-- | CAS version associated with a value. Operations that take a version skip+-- the CAS check when it is @0@. type Version = Word64 -data OpRequest- = ReqGet Q K Key- | ReqSet Q Key Value SESet- | ReqAdd Q Key Value SESet- | ReqReplace Q Key Value SESet- | ReqDelete Q Key- | ReqIncrement Q Key SEIncr- | ReqDecrement Q Key SEIncr- | ReqAppend Q Key Value- | ReqPrepend Q Key Value- | ReqTouch Key SETouch- | ReqGAT Q K Key SETouch- | ReqFlush Q (Maybe SETouch)- | ReqNoop- | ReqVersion- | ReqStat (Maybe Key)- | ReqQuit Q- | ReqSASLList- | ReqSASLStart Key Value -- key: auth method, value: auth data- | ReqSASLStep Key Value -- key: auth method, value: auth data (continued)-- -- | Raw request is custom requests, dangerous as no corresponding raw- -- response...- | ReqRaw Word8 (Maybe Key) (Maybe Value) SERaw- deriving (Eq, Show)--{-# WARNING ReqRaw "This is dangerous; no future compatability guaranteed" #-}--data SESet = SESet Flags Expiration deriving (Eq, Show)-data SEIncr = SEIncr Initial Delta Expiration deriving (Eq, Show)-data SETouch = SETouch Expiration deriving (Eq, Show)-data SERaw = SERaw Builder Int--instance Show SERaw where- show _ = "SERaw _"--instance Eq SERaw where- (==) _ _ = False--data Request = Req {- reqOp :: OpRequest,- reqOpaque :: Word32,- reqCas :: Version- } deriving (Eq, Show)---- | Noop request.-emptyReq :: Request-emptyReq = Req { reqOp = ReqNoop, reqOpaque = 0, reqCas = 0 }--data OpResponse- = ResGet Q Value Flags- | ResGetK Q Key Value Flags- | ResSet Q- | ResAdd Q- | ResReplace Q- | ResDelete Q- | ResIncrement Q Word64- | ResDecrement Q Word64- | ResAppend Q- | ResPrepend Q- | ResTouch- | ResGAT Q Value Flags- | ResGATK Q Key Value Flags- | ResFlush Q- | ResNoop- | ResVersion Value- | ResStat Key Value- | ResQuit Q- | ResSASLList Value- | ResSASLStart- | ResSASLStep- deriving (Eq, Show)---- | The status (success or error) of a Memcached operation returned in a--- 'Response'.+-- | Status reported by a Memcached operation. data Status- -- | Operation successful.- = NoError -- All- -- | Key not found.- | ErrKeyNotFound -- Get, GAT, Touch, Replace, Del, Inc, Dec, App, Pre, Set (key not there and version specified...)- -- | Key exists when not expected.- | ErrKeyExists -- Add, (version): Set, Rep, Del, Inc, Dec, App, Pre- -- | Value too large to store at server.- | ErrValueTooLarge -- Set, Add, Rep, Pre, App- -- | Invalid arguments for operation.- | ErrInvalidArgs -- All- -- | Key-Value pair not stored at server (internal error).- | ErrItemNotStored -- ?- -- | Value not numeric when increment or decrement requested.- | ErrValueNonNumeric -- Incr, Decr- -- | Server doesn't know requested command.- | ErrUnknownCommand -- All- -- | Server out of memory.- | ErrOutOfMemory -- All- -- | SASL authentication failed.- | SaslAuthFail -- SASL- -- | SASL authentication requires more steps.- | SaslAuthContinue -- SASL+ -- | The operation succeeded.+ = NoError+ -- | The key was not found.+ | ErrKeyNotFound+ -- | The key already exists.+ | ErrKeyExists+ -- | The value exceeds the server limit.+ | ErrValueTooLarge+ -- | The item was not stored.+ | ErrItemNotStored+ -- | The value is not numeric.+ | ErrValueNonNumeric+ -- | The server is out of memory.+ | ErrOutOfMemory+ -- | Text authentication failed.+ | TextAuthFail deriving (Eq, Show)---- | Memcached response packet.-data Response = Res {- resOp :: OpResponse,- resStatus :: Status,- resOpaque :: Word32,- resCas :: Version- } deriving (Eq, Show)---- | Noop response.-emptyRes :: Response-emptyRes = Res { resOp = ResNoop, resStatus = NoError, resOpaque = 0, resCas = 0 }-
+ Database/Memcache/Types/Authentication.hs view
@@ -0,0 +1,32 @@+{-|+Module : Database.Memcache.Types.Authentication+Description : Meta Text authentication types+Copyright : (c) David Terei, 2016+License : BSD+Maintainer : code@davidterei.com+Stability : stable+Portability : GHC+-}+module Database.Memcache.Types.Authentication (+ -- * Text authentication+ Authentication(..), Username, Password,+ ) where++import Data.ByteString (ByteString)+-- | Username and password information for text auth-file authentication.+data Authentication+ = Auth+ { -- | Username to send to the server.+ username :: !Username+ -- | Password to send to the server.+ , password :: !Password+ }+ -- | Do not authenticate the connection.+ | NoAuth+ deriving (Eq, Show)++-- | Username for text authentication.+type Username = ByteString++-- | Password for text authentication.+type Password = ByteString
+ Database/Memcache/Types/ServerSpec.hs view
@@ -0,0 +1,89 @@+{-# LANGUAGE OverloadedStrings #-}++{-|+Module : Database.Memcache.Types.ServerSpec+Description : Memcached server specifications+Copyright : (c) David Terei, 2016+License : BSD+Maintainer : code@davidterei.com+Stability : stable+Portability : GHC++Server connection settings, including optional Meta Text auth-file+credentials.+-}+module Database.Memcache.Types.ServerSpec+ ( ServerSpec (..),+ parseServerSpec,+ )+where++import Control.Monad (guard)+import Data.Default+import Data.Maybe (fromMaybe)+import qualified Data.Text as T+import Data.Text.Encoding (encodeUtf8)+import Database.Memcache.Types.Authentication (Authentication (..))+import Database.Memcache.Util (note)+import Network.Socket (HostName, ServiceName)+import Network.URI (URI (..), URIAuth (..), parseAbsoluteURI)+import Data.Bifunctor (second)++-- | A Memcached server connection configuration.+data ServerSpec = ServerSpec+ { -- | Hostname of server to connect to.+ ssHost :: HostName,+ -- | Port number server is running on.+ ssPort :: ServiceName,+ -- | Credentials to use for text auth-file authentication with this server.+ ssAuth :: Authentication+ }+ deriving (Eq, Show)++-- | Parse a @memcached://@ URI into a @ServerSpec@.+parseServerSpec :: String -> Either String ServerSpec+parseServerSpec s = do+ uri <- note ("Not a valid URI: " <> s) $ parseAbsoluteURI s+ note "Must begin memcached://" $ guard $ uriScheme uri == "memcached:"++ let mAuth = uriAuthority uri++ pure+ . maybe id setHost mAuth+ . maybe id setPort mAuth+ . maybe id setAuth (readAuthentication . uriUserInfo =<< mAuth)+ $ def++readAuthentication :: String -> Maybe Authentication+readAuthentication = go . T.pack+ where+ go a = do+ (u, p) <- second (T.drop 1) . T.breakOn ":" <$> T.stripSuffix "@" a++ guard $ not $ T.null u+ guard $ not $ T.null p++ pure+ Auth+ { username = encodeUtf8 u,+ password = encodeUtf8 p+ }++setHost :: URIAuth -> ServerSpec -> ServerSpec+setHost auth ss = case uriRegName auth of+ "" -> ss+ rn -> ss {ssHost = rn}++setPort :: URIAuth -> ServerSpec -> ServerSpec+setPort auth ss = fromMaybe ss $ do+ p <- case uriPort auth of+ "" -> Nothing+ (':' : p) -> Just p+ p -> Just p+ pure $ ss {ssPort = p}++setAuth :: Authentication -> ServerSpec -> ServerSpec+setAuth auth ss = ss {ssAuth = auth}++instance Default ServerSpec where+ def = ServerSpec "127.0.0.1" "11211" NoAuth
+ Database/Memcache/Util.hs view
@@ -0,0 +1,18 @@+{-|+Module : Database.Memcache.Util+Description : Internal helpers+Copyright : (c) David Terei, 2026+License : BSD+Maintainer : code@davidterei.com+Stability : internal+Portability : GHC++Small helpers shared by the library's modules. Not exposed.+-}+module Database.Memcache.Util (+ note+ ) where++-- | Turn a 'Nothing' into an error.+note :: e -> Maybe a -> Either e a+note e = maybe (Left e) Right
LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2012, David Terei.+Copyright (c) 2012-2026, David Terei. All rights reserved.
README.md view
@@ -1,26 +1,122 @@ # memcache: Haskell Memcached Client - [](https://hackage.haskell.org/package/memcache)-[](http://packdeps.haskellers.com/reverse/memcache)-[][tl;dr Legal: BSD3]-[](https://travis-ci.org/dterei/memcache-hs)-[](https://gitter.im/dterei/memcache-hs?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge)+[][tl;dr Legal: BSD3]+[](https://github.com/dterei/memcache-hs/actions/workflows/ci.yml) -[tl;dr Legal: BSD3]:- https://tldrlegal.com/license/bsd-3-clause-license-(revised)- "BSD3 License"+[tl;dr Legal: BSD3]: https://tldrlegal.com/license/bsd-3-clause-license-(revised) "BSD3 License" A client library for a memcached cluster. -It supports the binary memcached protocol and SASL authentication. No support-for the ASCII protocol is provided. It supports connecting to a single, or a-cluster of memcached servers. When connecting to a cluser, consistent hashing-is used for routing requests to the appropriate server.+It supports the Meta Text memcached protocol and username/password+authentication. Keys containing whitespace or control bytes are sent using its+base64 binary-key mode. The encoded key must still fit within 250 bytes. It+supports connecting to a single, or a cluster of memcached servers. When+connecting to a cluster, consistent hashing is used for routing requests to+the appropriate server. -Complete coverage of the memcached protocol is provided except for multi-get-and other pipelined operations.+**This library requires memcached 1.6.0 or newer**, the first release with the+meta commands. Against older servers every operation is rejected with `ERROR`. +Authentication uses a memcached auth file with `-Y authfile`, an experimental+ASCII-protocol auth mode added in memcached 1.5.15; the old binary protocol+`-S` SASL mode is not supported. Enabling `-Y` also disables binary and UDP+protocols on the server. `getMany` uses quiet misses, so a server connection+failure can be indistinguishable from a cache miss.++The client operations, including `getMany`, use direct Meta Text encoders and+the `Response` type; the former request and response wrapper types are no longer+provided.++## Usage++Add `memcache` to your package's `build-depends`. Everything most programs need+is exported from `Database.Memcache.Client`:++```haskell+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import qualified Database.Memcache.Client as M++main :: IO ()+main = do+ -- Connect to memcached on 127.0.0.1:11211 with the default options.+ mc <- M.newClient [M.def] M.def++ -- Store a value with flags 0 that never expires, then read it back along+ -- with its flags and CAS version.+ _ <- M.set mc "greeting" "hello" 0 0+ hit <- M.get mc "greeting"+ case hit of+ Nothing -> putStrLn "miss"+ Just (value, _flags, version) -> do+ print value+ -- Only overwrite the value if it hasn't changed since we read it.+ updated <- M.cas mc "greeting" "hello again" 0 0 version+ putStrLn (maybe "changed by someone else" (const "updated") updated)++ -- Create the counter as 1 if it is missing, otherwise add 1 to it.+ visits <- M.increment mc "visits" 1 1 0 0+ print (fst <$> visits)++ M.quit mc+```++Keys and values are strict `ByteString`s. Misses are returned as `Nothing`,+while unexpected failures are thrown as a `MemcacheError` or a network+`IOException`. Operations that take a CAS `Version` skip the check when it is+`0`. `increment` and `decrement` create a missing key with the given initial+value, unless the expiration is `maxBound`. The+[`Database.Memcache.Client`](https://hackage.haskell.org/package/memcache/docs/Database-Memcache-Client.html)+documentation covers every operation, along with the timeouts, retries and+failover configured through `Options`.++### Servers and authentication++`newClient` takes one `ServerSpec` per server, and routes each key to one of+them. Credentials are set per server:++```haskell+ mc <- M.newClient+ [ M.ServerSpec "cache1.internal" "11211" (M.Auth "user" "secret")+ , M.ServerSpec "cache2.internal" "11211" (M.Auth "user" "secret")+ ]+ M.def+```++`parseServerSpec`, from `Database.Memcache.Types`, reads a server from a+`memcached://[user:password@]host[:port]` URI instead. The host defaults to+`127.0.0.1` and the port to `11211`:++```haskell+import Database.Memcache.Types (parseServerSpec)++ spec <- either fail pure (parseServerSpec "memcached://user:secret@cache1.internal:11211")+ mc <- M.newClient [spec] M.def+```++The server must be started with `-Y <authfile>`, where the auth file holds one+`username:password` line per user. This library connects without TLS, so+credentials and all data are sent unencrypted; only rely on them within a+trusted network.++### Amazon ElastiCache++`Database.Memcache.ElastiCacheClient` creates a client from an ElastiCache+configuration endpoint (the address containing `.cfg`). It discovers the+cluster's nodes, then refreshes the node list in a background thread on the+given interval, in microseconds as for `threadDelay`:++```haskell+import qualified Database.Memcache.ElastiCacheClient as EC++ endpoint <- either fail pure $+ EC.parseConfigurationEndpoint "memcached://mycluster.abc123.cfg.use1.cache.amazonaws.com:11211"+ mc <- EC.newClient M.def (60 * 1000000) endpoint+```+ ## Licensing This library is BSD-licensed.@@ -30,31 +126,31 @@ This library also includes a few tools for manipulating and experimenting with memcached servers. -* `OpGen` -- A load generator for memcached. Doesn't collect timing statistics,+- `OpGen` -- A load generator for memcached. Doesn't collect timing statistics, other tools like [mutilate](https://github.com/leverich/mutilate) already do that very well. This tool is useful in conjunction with mutilate.-* `Loader` -- A tool to load random data of a certain size into a memcached+- `Loader` -- A tool to load random data of a certain size into a memcached server. Useful for priming a server for testing. ## Architecture Notes -We're relying on `Data.Pool` for thread safety right now, which is fine but is-a blocking API in that when we grab a socket (`withResource`) we are blocking-any other requests being sent over that connection until we get a response.-That is, we can't pipeline.+We're relying on `Data.Pool` for thread safety right now. Each server has its+own pool, and grabbing a connection from it (`withResource`) blocks other+requests using that connection until the operation completes. `getMany` is the+exception: it pipelines quiet Meta Text requests over one pooled connection. -Now, use of multiple connections through the pool abstraction is an easy way to-solve this and perhaps the right approach. But, could also implement own pool-abstraction that allowed pipelining. This wouldn't be a pool abstraction so-much as just round-robbining over multiple connections for performance.+By default each server's pool holds a single connection (`soNumResources = 1`),+so concurrent operations against the same server wait for each other. Raising+`soNumResources` in the `ServerOptions` set through `optsServerOptions` allows+concurrent operations without requiring a separate pool implementation. Either way, a pool is fine for now. ## Other clients -* [C: libmemcached](http://libmemcached.org/libMemcached.html)-* [Java: SpyMemcached](http://code.google.com/p/spymemcached/)-* [Ruby: Dalli](https://github.com/mperham/dalli)+- [C: libmemcached](http://libmemcached.org/libMemcached.html)+- [Java: SpyMemcached](http://code.google.com/p/spymemcached/)+- [Ruby: Dalli](https://github.com/mperham/dalli) ## Get involved! @@ -62,18 +158,63 @@ other improvements. Please report bugs via the-[github issue tracker](http://github.com/dterei/memcache-hs/issues).+[github issue tracker](https://github.com/dterei/memcache-hs/issues), and+security issues as described in the+[security policy](https://github.com/dterei/memcache-hs/security/policy). -Master [git repository](http://github.com/dterei/memcache-hs):+Master [git repository](https://github.com/dterei/memcache-hs): -* `git clone https://github.com/dterei/memcache-hs.git`+- `git clone https://github.com/dterei/memcache-hs.git` +## Development++The package builds with [Stack](https://docs.haskellstack.org/) or+[cabal](https://www.haskell.org/cabal/), on GHC 9.2 or newer:++```sh+stack build+stack test++cabal build+cabal test all+```++`stack.yaml` uses Stackage LTS 22.14 (GHC 9.6). The oldest and newest snapshots+CI tests against have their own files, for example+`stack --stack-yaml stack-lts-20.26.yaml test` for GHC 9.2.++There are two test suites:++- `spec` holds the unit and property tests, which run against an in-process+ mock server and need nothing installed.+- `integration` starts its own `memcached` servers on free local ports and+ tests against them, so it needs memcached 1.6 or newer on your `PATH`.+ Without it, the suite reports a single pending test instead of failing.+ memcached refuses to run as root without `-u`, so run the suite as an+ ordinary user.++CI lints with [HLint](https://github.com/ndmitchell/hlint) 3.10, so please+check that `hlint .` prints `No hints` before opening a pull request. For any+user-visible change, add a line to [CHANGELOG.md](CHANGELOG.md) under an+`Unreleased` heading at the top, adding the heading if it isn't there.++CI runs both build tools on every pull request. The `CI` workflow builds and+tests with Stack against each `stack*.yaml` snapshot, and runs HLint and+weeder. The `Cabal` workflow builds and tests with cabal on GHC 9.2 to 9.14,+builds the library against the oldest dependency versions its bounds allow,+and builds and tests the source distribution; it also runs every Monday, so a+new dependency release that breaks the build shows up without a push. Both+install memcached, so the integration suite runs in full.+ ## Authors This library is written and maintained by David Terei (<code@davidterei.com>). Contributions have been made by the following great people: -* Alfredo Di Napoli (<alfredo.dinapoli@gmail.com>)-* Amit Levy-+- Alfredo Di Napoli (<alfredo.dinapoli@gmail.com>)+- Amit Levy+- Chris Martin+- drlkf+- Joel McCracken+- Steven Leiva
− Setup.hs
@@ -1,7 +0,0 @@-module Main (main) where--import Distribution.Simple--main :: IO ()-main = defaultMain-
TODO.md view
@@ -1,33 +1,34 @@ ## ToDo +Behavior and API fixes, including routing and virtual nodes, are tracked in+[#46](https://github.com/dterei/memcache-hs/issues/46).+ Optional:-* Typeclass abstracting key and values-* MonadIO?-* Safe Haskell-* Differentiate between two expiration modes?-* Simpler interface?-* Configurable fail-over mode +- Typeclass abstracting key and values+- MonadIO?+- Safe Haskell+- Differentiate between two expiration modes?+- Simpler interface?+- Configurable fail-over mode+ Protocol:-* Multi-get-* Generic multi operation support +- Review quiet multi-get behavior when a server closes before `mn`.+ Performance:-* Is System.Timeout still slow? Switch to Warp/Snap timeout handler?-* Is NominalDiffTime slow? Switch to unix epoch time? +- System.Timeout forks a thread per server request on the non-threaded RTS and+ on Windows; consider a shared timer.+ Nice-to-have:-* Smarter connection handling to minimize system calls (buffered)-* Customizable server sharding -- mod & virtual servers-* Server weights-* Asynchronous support?-* Customizable -- timeout, max connection retries, hash algorithm-* Max value validation-* Optimizations -- http://code.google.com/p/spymemcached/wiki/Optimizations-* UDP-* ASCII-* Server error handling mode where we return misses and ignore sets +- Built-in modulo sharding (custom routing works through `optsGetServerForKey`)+- Server weights+- Asynchronous support?+- Max value validation before sending+- Server error handling mode where we return misses and ignore sets+ Maybe:-* Automatic encoding and decoding of Haskell data types (using flags?) +- Automatic encoding and decoding of Haskell data types (using flags?)
bench/Parser.hs view
@@ -1,59 +1,38 @@ {-# LANGUAGE OverloadedStrings #-} --- | Benchmark the parsing and serialization aspects of memcache.-module Main where--import Database.Memcache.Types-import Database.Memcache.Wire+-- | Micro-benchmarks for the pure parts of the Meta Text protocol layer: key+-- encoding, request encoding and reading fields out of responses.+module Main (main) where -import Criterion.Main-import Data.ByteString (ByteString)-import qualified Data.ByteString as B-import qualified Data.ByteString.Lazy as L-import qualified Data.ByteString.Lazy.Char8 as LC-import Data.Monoid+import Criterion.Main+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as C+import Database.Memcache.Socket main :: IO ()-main =- defaultMain [- bgroup "serialize" [- bench "get" $ whnf szRequest' getReqMsg,- bench "set" $ whnf szRequest' setReqMsg- ],- bgroup "deserialize" [- bench "get" $ whnf dzResponse' getRespBytes- ]+main = defaultMain+ [ bgroup "encodeKey"+ [ bench "plain" $ whnf encodedLength "user:1234:profile"+ , bench "base64" $ whnf encodedLength "a key with spaces"+ , bench "250 bytes" $ whnf encodedLength (C.replicate 250 'k') ]--getReqMsg :: Request-getReqMsg = Req {- reqOp = ReqGet Loud NoKey "key!",- reqOpaque = 123,- reqCas = 999- }--setReqMsg :: Request-setReqMsg = Req {- reqOp = ReqSet Loud "key!" "hello world" (SESet 10 0),- reqOpaque = 123,- reqCas = 999- }--getRespHeaderBytes :: L.ByteString-getRespHeaderBytes =- -- magic, op, extral, 0- L.pack $ [0x81, 0x00] ++ keyl' ++ [0x04, 0x00] ++ status' ++ bodyl' ++ opaque' ++ cas'- where- keyl' = [0x00, 0x00]- status' = [0x00, 0x00, 0x00, 0x00]- bodyl' = [0x00, 0x00, 0x00, 0x08]- opaque' = [0x00, 0x00, 0x00, 0x07]- cas' = [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x09]--getRespBytes :: L.ByteString-getRespBytes = getRespHeaderBytes <> extras' <> key' <> value'+ , bgroup "request"+ [ bench "get" $ whnf getRequest key+ , bench "set 100 B" $ whnf (storeRequest Set key (C.replicate 100 'v') 10 0) 0+ , bench "set 64 KiB" $ whnf (storeRequest Set key (C.replicate 65536 'v') 10 0) 0+ , bench "increment" $ whnf (arithmeticRequest Incr key 0 1 0) 0+ ]+ , bgroup "response"+ [ bench "cas" $ whnf responseCas valueResponse+ , bench "flags" $ whnf responseFlags valueResponse+ , bench "number" $ whnf responseNumber (Response "VA" ["5"] "12345")+ ]+ ] where- extras' = L.pack [0x00, 0x00, 0x00, 0x01] -- BE: so 1?- key' = L.pack []- value' = LC.pack "12345678"+ key = either (error . show) id (encodeKey "key!")+ valueResponse = Response "VA" ["3", "f7", "c123456789", "kkey"] "abc" +-- | Force the whole encoded key by measuring it, failing if the key is+-- rejected so a benchmark can't quietly time the error path instead.+encodedLength :: B.ByteString -> Int+encodedLength = either (error . show) (B.length . wireKey) . encodeKey
memcache.cabal view
@@ -1,134 +1,156 @@-name: memcache-version: 0.2.0.1-homepage: https://github.com/dterei/memcache-hs-bug-reports: https://github.com/dterei/memcache-hs/issues-synopsis: A memcached client library.-description: +cabal-version: 3.0+name: memcache+version: 0.4.2.0+synopsis: A memcached client library.+description: A client library for a Memcached cluster. Memcached is an in-memory key-value store typically used as a distributed and shared cache. Clients connect to a group of Memcached servers and perform out-of-band caching for things like SQL results, rendered pages, or third-party APIs.- .- It supports the binary Memcached protocol and SASL authentication. No support- for the ASCII protocol is provided. It supports connecting to a single, or a- cluster of Memcached servers. When connecting to a cluser, consistent hashing- is used for routing requests to the appropriate server. Timeouts, retrying- failed operations, and failover to a different server are all supported.- .- Complete coverage of the Memcached protocol is provided except for multi-get- and other pipelined operations.- .++ It uses the Memcached Meta Text protocol and requires memcached 1.6.0 or+ newer. Username/password authentication is supported through memcached's+ @-Y authfile@ mode; the binary protocol and SASL authentication are no longer+ supported. It supports connecting to a single, or a cluster of Memcached+ servers. When connecting to a cluster, consistent hashing is used for routing+ requests to the appropriate server. Timeouts, retrying failed operations, and+ failover to a different server are all supported, as is Amazon ElastiCache+ auto discovery.++ Multi-get is supported through the pipelined @getMany@. Keys containing+ whitespace or control bytes are sent using the protocol's base64 binary-key+ mode.+ Basic usage is:- .++ > {-# LANGUAGE OverloadedStrings #-}+ > > import qualified Database.Memcache.Client as M- > - > mc <- M.newClient [M.ServerSpec "localhost" 11211 M.NoAuth] M.def- > M.set mc "key" "value" 0 0- > v <- M.get mc "key"- .- You should only need to import 'Database.Memcache.Client', but for now other- modules are exposed.-license: BSD3-license-file: LICENSE-author: David Terei <code@davidterei.com>-maintainer: David Terei <code@davidterei.com>-copyright: 2016 David Terei.-category: Database-build-type: Simple-cabal-version: >= 1.10-extra-source-files:- README.md, CHANGELOG.md, TODO.md+ >+ > main :: IO ()+ > main = do+ > mc <- M.newClient [M.ServerSpec "localhost" "11211" M.NoAuth] M.def+ > _ <- M.set mc "key" "value" 0 0+ > v <- M.get mc "key"+ > print v -Source-repository this- type: git- location: https://github.com/dterei/memcache-hs.git- tag: 0.2.0.1+ You should only need to import 'Database.Memcache.Client', but other modules+ are exposed for lower-level use. Version 0.4 is a breaking change from 0.3;+ see the changelog for details. +category: Database+homepage: https://github.com/dterei/memcache-hs#readme+bug-reports: https://github.com/dterei/memcache-hs/issues+author: David Terei <code@davidterei.com>+maintainer: David Terei <code@davidterei.com>+copyright: 2012-2026 David Terei.+license: BSD-3-Clause+license-file: LICENSE+build-type: Simple+-- Keep in step with the GHC matrix in .github/workflows/cabal.yml.+tested-with:+ GHC ==9.2.*+ || ==9.4.*+ || ==9.6.*+ || ==9.8.*+ || ==9.10.*+ || ==9.12.*+ || ==9.14.*++extra-doc-files:+ CHANGELOG.md+ README.md+ TODO.md+ source-repository head type: git- location: https://github.com/dterei/memcache-hs.git+ location: https://github.com/dterei/memcache-hs +-- Settings shared by every component.+common shared+ default-language: GHC2021+ ghc-options: -Wall -Wextra -Wunused-packages+ build-depends:+ , base >=4.16 && <5+ , bytestring >=0.11.4.0+ library+ import: shared+ hs-source-dirs: . exposed-modules:+ Database.Memcache.Auth Database.Memcache.Client Database.Memcache.Cluster+ Database.Memcache.ElastiCacheClient Database.Memcache.Errors- Database.Memcache.SASL Database.Memcache.Server Database.Memcache.Socket Database.Memcache.Types- build-depends:- base < 5,- binary >= 0.6.2.0,- blaze-builder >= 0.3.1.0,- bytestring >= 0.9.2.1,- data-default-class >= 0.1.0,- hashable >= 1.2.0.3,- network >= 2.4,- resource-pool >= 0.2.1.0,- vector >= 0.7,- time >= 1.4- default-language: Haskell2010- other-extensions:- BangPatterns,- CPP,- DeriveDataTypeable,- FlexibleInstances,- RecordWildCards,- ScopedTypeVariables- ghc-options: -Wall -fwarn-tabs --- executable opgen--- hs-source-dirs: tools--- main-is: OpGen.hs--- ghc-options: -threaded--- build-depends:--- async,--- base < 5,--- bytestring,--- memcache,--- network------ executable load--- hs-source-dirs: tools--- main-is: Loader.hs--- ghc-options: -threaded--- build-depends:--- async,--- base < 5,--- bytestring,--- memcache,--- network+ other-modules:+ Database.Memcache.Types.Authentication+ Database.Memcache.Types.ServerSpec+ Database.Memcache.Util -test-suite full- type: exitcode-stdio-1.0- hs-source-dirs: test- main-is: Full.hs- build-depends:- base < 5,- binary >= 0.6.2.0,- blaze-builder >= 0.3.1.0,- bytestring >= 0.9.2.1,- memcache,- network >= 2.4- default-language: Haskell2010 other-extensions:- BangPatterns,- CPP,+ CPP OverloadedStrings- ghc-options: -fwarn-tabs+ RecordWildCards -benchmark parser- type: exitcode-stdio-1.0- hs-source-dirs: bench- main-is: Parser.hs+ ghc-options:+ -Wcompat -Widentities -Wincomplete-record-updates -Wredundant-constraints+ -fwarn-tabs+ build-depends:- base < 5,- bytestring >= 0.9.2.1,- criterion > 0.6.0.0,- memcache- default-language: Haskell2010- other-extensions:- OverloadedStrings- ghc-options: -fwarn-tabs+ , base64-bytestring >=1.2.1.0+ , data-default >=0.7.1.1+ , hashable >=1.4.2.0+ , network >=3.1.4.0+ , network-uri >=2.6.4.2+ , resource-pool >=0.2.3.2+ , text >=1.2.5.0+ , time >=1.11.1.1+ , unliftio >=0.2.25.0+ , vector >=0.12.3.1 +test-suite spec+ import: shared+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules: MockServer+ other-extensions: OverloadedStrings+ ghc-options: -threaded -rtsopts "-with-rtsopts=-N"+ build-depends:+ , base64-bytestring >=1.2.1.0+ , hspec+ , memcache+ , network+ , QuickCheck+ , unliftio++-- Runs against a real memcached; reports a pending test if none is on PATH.+test-suite integration+ import: shared+ type: exitcode-stdio-1.0+ hs-source-dirs: test-integration+ main-is: Integration.hs+ other-extensions: OverloadedStrings+ ghc-options: -threaded -rtsopts "-with-rtsopts=-N"+ build-depends:+ , directory+ , hspec+ , memcache+ , network+ , process+ , time++benchmark parser+ import: shared+ type: exitcode-stdio-1.0+ hs-source-dirs: bench+ main-is: Parser.hs+ other-extensions: OverloadedStrings+ build-depends:+ , criterion+ , memcache
+ test-integration/Integration.hs view
@@ -0,0 +1,304 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Control.Concurrent (threadDelay)+import Control.Exception (SomeException, bracket, try)+import Control.Monad (forM_, void)+import qualified Data.ByteString.Char8 as B+import Data.Either (isRight)+import Data.Maybe (isJust)+import Data.Time.Clock.POSIX (getPOSIXTime)+import qualified Database.Memcache.Client as M+import Database.Memcache.Types (ServerSpec (..))+import Network.Socket (Family (AF_INET), SocketOption (ReuseAddr),+ SocketType (Stream), addrAddress, addrFamily,+ addrSocketType, bind, close, connect,+ defaultHints, defaultProtocol, getAddrInfo,+ setSocketOption, socket, socketPort,+ tupleToHostAddress)+import qualified Network.Socket as N+import System.Directory (findExecutable, getTemporaryDirectory, removeFile)+import System.IO (hClose, hPutStr, openTempFileWithDefaultPermissions)+import System.Process (ProcessHandle, createProcess, proc,+ getProcessExitCode, terminateProcess,+ waitForProcess)+import Test.Hspec++main :: IO ()+main = do+ executable <- findExecutable "memcached"+ case executable of+ Nothing ->+ hspec $+ it "requires memcached" $+ pendingWith "memcached command not found"+ Just path ->+ withMemcached path [] $ \port ->+ withAuthfile $ \authfile ->+ withMemcached path ["-Y", authfile] $ \authPort ->+ withMemcacheds path 3 $ \clusterPorts ->+ hspec $ do+ describe "single server" $ integrationSpecs port+ describe "auth-file authentication" $ authSpecs authPort+ describe "three-server cluster" $ clusterSpecs clusterPorts++integrationSpecs :: Int -> Spec+integrationSpecs port = beforeAll (newClient port) $ do+ it "reports its version" $ \client -> do+ version <- M.version client+ version `shouldSatisfy` (not . B.null)++ it "sets and gets values with flags and CAS" $ \client -> do+ cas <- M.set client "integration-set" "value" 7 0+ M.get client "integration-set" `shouldReturn` Just ("value", 7, cas)++ it "returns misses" $ \client ->+ M.get client "integration-missing" `shouldReturn` Nothing++ it "adds and replaces values" $ \client -> do+ stamp <- round . (* 1000000) <$> getPOSIXTime :: IO Integer+ let key = B.pack ("integration-add-" <> show port <> "-" <> show stamp)+ _ <- M.set client key "one" 0 0++ M.add client key "two" 0 0 `shouldReturn` Nothing++ M.replace client "integration-no-replace" "two" 0 0 0 `shouldReturn` Nothing++ it "updates values with CAS" $ \client -> do+ old <- M.set client "integration-cas" "old" 0 0+ new <- M.cas client "integration-cas" "new" 0 0 old++ new `shouldSatisfy` isJust++ M.cas client "integration-cas" "stale" 0 0 old `shouldReturn` Nothing++ it "deletes values" $ \client -> do+ _ <- M.set client "integration-delete" "value" 0 0++ M.delete client "integration-delete" 0 `shouldReturn` True+ M.delete client "integration-delete" 0 `shouldReturn` False++ it "touches and gets with expiry" $ \client -> do+ _ <- M.set client "integration-touch" "value" 0 0+ touched <- M.touch client "integration-touch" 60+ touched `shouldSatisfy` isJust++ gated <- M.gat client "integration-touch" 60+ gated `shouldSatisfy` isJust++ M.touch client "integration-no-touch" 60 `shouldReturn` Nothing++ it "tells a plain key from another key's base64 form" $ \client -> do+ -- " " is sent, and echoed, as IA== with the b flag.+ _ <- M.set client "IA==" "plain" 0 0+ _ <- M.set client " " "space" 0 0+ values <- M.getMany client ["IA==", " "]+ map (\(key, (value, _, _)) -> (key, value)) values+ `shouldMatchList` [("IA==", "plain"), (" ", "space")]++ it "gets many keys, including encoded keys" $ \client -> do+ _ <- M.set client "integration-many" "one" 0 0+ _ <- M.set client "integration key" "two" 0 0++ values <-+ M.getMany+ client+ [ "integration-many"+ , "integration key"+ , "integration-no-many"+ ]+ lookup "integration-many" values `shouldSatisfy` isJust+ lookup "integration key" values `shouldSatisfy` isJust++ it "increments and decrements counters" $ \client -> do+ _ <- M.delete client "integration-counter" 0+ incremented <- M.increment client "integration-counter" 10 2 0 0+ incremented `shouldSatisfy` maybe False ((== 10) . fst)++ decremented <- M.decrement client "integration-counter" 0 3 0 0+ decremented `shouldSatisfy` isJust++ it "appends and prepends" $ \client -> do+ cas <- M.set client "integration-modify" "middle" 0 0+ appended <- M.append client "integration-modify" "-end" cas+ appended `shouldSatisfy` isJust++ prepended <- M.prepend client "integration-modify" "start-" 0+ prepended `shouldSatisfy` isJust++ result <- M.get client "integration-modify"+ fmap first result `shouldBe` Just "start-middle-end"++ it "modifies encoded keys" $ \client -> do+ _ <- M.set client "integration modify key" "middle" 0 0+ _ <- M.append client "integration modify key" "-end" 0+ result <- M.get client "integration modify key"+ fmap first result `shouldBe` Just "middle-end"++ it "flushes values" $ \client -> do+ _ <- M.set client "integration-flush" "value" 0 0+ M.flush client Nothing+ M.get client "integration-flush" `shouldReturn` Nothing++ it "reads stats" $ \client -> do+ stats <- M.stats client Nothing+ stats `shouldSatisfy` not . all (null . snd)++ it "roundtrips a large value" $ \client -> do+ let value = B.replicate 100000 'x'+ _ <- M.set client "integration-large" value 0 0+ result <- M.get client "integration-large"++ fmap first result `shouldBe` Just value++ it "roundtrips an encoded key" $ \client -> do+ _ <- M.set client "integration unsafe key" "value" 0 0+ result <- M.get client "integration unsafe key"++ fmap first result `shouldBe` Just "value"++ it "supports quit" M.quit++authSpecs :: Int -> Spec+authSpecs port = do+ it "serves a client with the right credentials" $ do+ client <- M.newClient [server (M.Auth "user" "secret")] M.def+ _ <- M.set client "integration-auth" "value" 0 0+ fmap first <$> M.get client "integration-auth" `shouldReturn` Just "value"++ it "rejects a wrong password" $ do+ client <- M.newClient [server (M.Auth "user" "wrong")] M.def+ M.get client "integration-auth" `shouldThrow` (== M.OpError M.TextAuthFail)++ it "rejects a client without credentials" $ do+ client <- M.newClient [server M.NoAuth] M.def+ M.get client "integration-auth" `shouldThrow` isBadCommand+ where+ server = ServerSpec "127.0.0.1" (show port)+ isBadCommand (M.ProtocolError (M.BadCommand _)) = True+ isBadCommand _ = False++-- | Tests against several servers. Key routing currently leaves at least one+-- server without keys (#46), so these don't assume keys reach every server.+clusterSpecs :: [Int] -> Spec+clusterSpecs ports = beforeAll newClusterClient $ do+ it "stores and reads back keys" $ \client -> do+ storeKeys client+ forM_ keys $ \key -> fmap first <$> M.get client key `shouldReturn` Just key++ it "gets many keys" $ \client -> do+ storeKeys client+ values <- M.getMany client keys+ map fst values `shouldMatchList` keys++ it "reads stats from every server" $ \client -> do+ stats <- M.stats client Nothing+ length stats `shouldBe` length ports++ it "flushes every server" $ \client -> do+ storeKeys client+ counts <- flushCounts client+ M.flush client Nothing+ flushCounts client `shouldReturn` map (+ 1) counts+ M.getMany client keys `shouldReturn` []+ where+ keys = [B.pack ("integration-cluster-" <> show n) | n <- [1 .. 300 :: Int]]+ storeKeys client = forM_ keys $ \key -> M.set client key key 0 0+ -- How many times each server has run flush_all, in server order.+ flushCounts client = do+ stats <- M.stats client Nothing+ return [maybe 0 (read @Int . B.unpack) (lookup "cmd_flush" s) | (_, s) <- stats]+ newClusterClient = do+ client <- M.newClient [ServerSpec "127.0.0.1" (show port) M.NoAuth | port <- ports] M.def+ M.flush client Nothing+ return client++newClient :: Int -> IO M.Client+newClient port = do+ client <- M.newClient [ServerSpec "127.0.0.1" (show port) M.NoAuth] M.def+ M.flush client Nothing+ return client++first :: (a, b, c) -> a+first (value, _, _) = value++-- | Run an action against a memcached started with the given extra+-- arguments on a free port.+withMemcached+ :: FilePath+ -> [String]+ -> (Int -> IO a)+ -> IO a+withMemcached executable args action = bracket acquire release (action . fst)+ where+ acquire = do+ port <- freePort+ (_, _, _, process) <-+ createProcess (proc executable (["-l", "127.0.0.1", "-p", show port, "-U", "0"] ++ args))+ waitForPort port process+ return (port, process)+ release (_, process) = do+ terminateProcess process+ void (waitForProcess process)++-- | Run an action against @n@ memcached servers.+withMemcacheds :: FilePath -> Int -> ([Int] -> IO a) -> IO a+withMemcacheds _ 0 action = action []+withMemcacheds executable n action =+ withMemcached executable [] $ \port ->+ withMemcacheds executable (n - 1) $ \ports -> action (port : ports)++-- | Run an action with an auth file allowing @user@ / @secret@.+withAuthfile :: (FilePath -> IO a) -> IO a+withAuthfile action = do+ tmp <- getTemporaryDirectory+ -- Default permissions, so a memcached that drops privileges (e.g. when+ -- started as root with -u) can still read it.+ bracket (openTempFileWithDefaultPermissions tmp "memcached-auth") (removeFile . fst) $ \(path, handle) -> do+ hPutStr handle "user:secret\n"+ hClose handle+ action path++freePort :: IO Int+freePort = bracket (socket AF_INET Stream defaultProtocol) close $ \sock -> do+ setSocketOption sock ReuseAddr 1+ bind sock (N.SockAddrInet 0 (tupleToHostAddress (127, 0, 0, 1)))+ fromIntegral <$> socketPort sock++waitForPort+ :: Int+ -> ProcessHandle+ -> IO ()+waitForPort port process = go 50+ where+ go :: Int -> IO ()+ go 0 = do+ -- Still running but never listening: stop it rather than waiting on it.+ terminateProcess process+ void (waitForProcess process)+ fail $ "memcached did not start listening on port " <> show port+ go attempts = do+ ready <- tryConnect port+ if ready+ then return ()+ else do+ exited <- getProcessExitCode process+ case exited of+ Just status -> fail $ "memcached exited before becoming ready: " <> show status+ Nothing -> threadDelay 100000 >> go (attempts - 1)++tryConnect+ :: Int+ -> IO Bool+tryConnect port = do+ result <- try $ do+ addr : _ <-+ getAddrInfo+ (Just defaultHints{addrSocketType = Stream})+ (Just "127.0.0.1")+ (Just (show port))+ sock <- socket (addrFamily addr) Stream defaultProtocol+ connect sock (addrAddress addr)+ close sock+ return (isRight @SomeException @() result)
− test/Full.hs
@@ -1,104 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}---- | Our testsuite, which we run expecting there to be a local Memcached server--- running on `localhost:11211`.-module Main where--import MockServer--import qualified Database.Memcache.Client as M-import Database.Memcache.Errors-import Database.Memcache.Socket-import Database.Memcache.Types--import Blaze.ByteString.Builder-import Control.Concurrent-import Control.Exception-import Control.Monad-import Data.Binary.Get-import qualified Data.ByteString.Char8 as BC-import qualified Network.Socket as N-import qualified Network.Socket.ByteString as N-import System.Exit-import System.IO--main :: IO ()-main = do- putStr "GET: "- getTest- putStrLn "PASSED"- putStr "DELETE: "- deleteTest- putStrLn "PASSED"- putStr "RETRY: "- retryTest- putStrLn "PASSED"- putStr "TIMEOUT1: "- timeoutTest- putStrLn "PASSED"- putStr "TIMEOUT2: "- timeoutRetryTest- putStrLn "PASSED"- exitSuccess--getTest :: IO ()-getTest = withMCServer False res $ do- c <- M.newClient [M.def] M.def- void $ M.set c (BC.pack "key") (BC.pack "world") 0 0- Just (v', _, _) <- M.get c "key"- when (v' /= "world") $ do- putStrLn $ "bad value returned! " ++ show v'- exitFailure - where- res = [ MR $ emptyRes { resOp = ResSet Loud }- , MR $ emptyRes { resOp = ResGet Loud "world" 0 }- ]--deleteTest :: IO ()-deleteTest = withMCServer False res $ do- c <- M.newClient [M.def] M.def- v1 <- M.set c "key" "world" 0 0- v2 <- M.set c "key" "world2" 0 0- when (v1 == v2) $ do- putStrLn $ "bad versions! " ++ show v1 ++ ", " ++ show v2- exitFailure- r <- M.delete c "key" 0- unless r $ do- putStrLn "delete failed!"- exitFailure- where- res = [ MR $ emptyRes { resOp = ResSet Loud, resCas = 1 }- , MR $ emptyRes { resOp = ResSet Loud, resCas = 2 }- , MR $ emptyRes { resOp = ResDelete Loud }- ]--retryTest :: IO ()-retryTest = withMCServer False res $ do- c <- M.newClient [M.def] M.def- void $ M.set c (BC.pack "key") (BC.pack "world") 0 0- where- res = [ CloseConnection- , MR $ emptyRes { resOp = ResSet Loud }- ]--timeoutTest :: IO ()-timeoutTest = withMCServer True res $ do- c <- M.newClient [M.def] M.def- void $ M.set c (BC.pack "key") (BC.pack "world") 0 0- r <- try $ M.get c "key" - case r of- Left (ClientError Timeout) -> return ()- Left _ -> putStrLn "unexpected exception!" >> exitFailure- Right _ -> putStrLn "no timeout occured!" >> exitFailure- where- res = [ MR $ emptyRes { resOp = ResSet Loud } ]--timeoutRetryTest :: IO ()-timeoutRetryTest = withMCServer False res $ do- c <- M.newClient [M.def] M.def- void $ M.set c (BC.pack "key") (BC.pack "world") 0 0- where- res = [ DelayMS 800 Noop- , MR $ emptyRes { resOp = ResSet Loud }- ]-
+ test/MockServer.hs view
@@ -0,0 +1,142 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Mock Memcached server - just enough for testing client.+--+-- Each mock listens on its own ephemeral port on 127.0.0.1, so tests can run+-- several at once and never clash with a real memcached on the default port.+module MockServer (+ MockResponse(..), MockServer(..), withMockServer, mockServerSpec+ ) where++import Control.Concurrent+import Control.Monad+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as C+import Data.IORef+import Database.Memcache.Types (Authentication (NoAuth), ServerSpec (..))+import qualified Network.Socket as N+import qualified Network.Socket.ByteString as N+import Text.Read (readMaybe)+import UnliftIO.Exception (SomeException, bracket, finally, handle,+ throwIO)++-- | Actions the mock server can take to a request.+data MockResponse+ -- | Reply with these bytes.+ = MR B.ByteString+ -- | Close the client connection instead of replying.+ | CloseConnection+ -- | Wait this many milliseconds, then take the given action.+ | DelayMS Int MockResponse+ -- | Read the request but send nothing back.+ | Noop++-- | A running mock server.+data MockServer = MockServer+ { -- | The port the mock is listening on.+ mockPort :: N.PortNumber+ -- | Every request received so far, oldest first, including any data+ -- block that followed the command line.+ , mockRequests :: IO [B.ByteString]+ }++-- | A server specification for connecting to the mock without authentication.+mockServerSpec :: MockServer -> ServerSpec+mockServerSpec server = ServerSpec "127.0.0.1" (show (mockPort server)) NoAuth++-- | Run an action with a mock Memcached server running in the background,+-- killing it once done. The server answers requests with the given responses+-- in order, one per request, across however many connections the client+-- makes; once they run out it closes the connection and stops listening.+withMockServer :: [MockResponse] -> (MockServer -> IO a) -> IO a+withMockServer responses action = do+ ready <- newEmptyMVar+ received <- newIORef []+ bracket+ (forkIO (serve ready received))+ killThread+ (const $ do+ port <- takeMVar ready+ action MockServer+ { mockPort = port+ , mockRequests = reverse <$> readIORef received+ })+ where+ serve ready received = bracket+ (N.socket N.AF_INET N.Stream N.defaultProtocol)+ N.close+ $ \sock -> do+ N.setSocketOption sock N.ReuseAddr 1+ N.bind sock (N.SockAddrInet 0 (N.tupleToHostAddress (127, 0, 0, 1)))+ N.listen sock 10+ N.socketPort sock >>= putMVar ready+ remaining <- newIORef responses+ -- Once the responses run out the listening socket closes, so later+ -- connection attempts are refused straight away.+ acceptLoop sock remaining received++ acceptLoop sock remaining received = do+ pending <- readIORef remaining+ unless (null pending) $ do+ (client, _) <- N.accept sock+ handle ignore (clientHandler client remaining received)+ `finally` N.close client+ acceptLoop sock remaining received++ ignore :: SomeException -> IO ()+ ignore = const (return ())++ -- Answer requests on one connection until the responses run out or the+ -- script closes it.+ clientHandler client remaining received = do+ pending <- readIORef remaining+ case pending of+ [] -> return ()+ (r:rest) -> do+ request <- recvReq client+ modifyIORef' received (request :)+ writeIORef remaining rest+ continue <- respond client r+ when continue $ clientHandler client remaining received++ respond client r = case r of+ Noop -> return True+ MR bytes -> N.sendAll client bytes >> return True+ DelayMS ms r' -> threadDelay (ms * 1000) >> respond client r'+ CloseConnection -> return False++-- | Read one request: its command line and, for storage commands, the data+-- block that follows it.+recvReq :: N.Socket -> IO B.ByteString+recvReq s = do+ line <- recvLine s []+ case payloadLen (C.words line) of+ Just n -> do+ payload <- recvAll s n+ trailer <- recvLine s []+ return (line <> "\n" <> payload <> trailer <> "\n")+ Nothing -> return (line <> "\n")+ where+ -- Only these carry a data block; guessing by token position instead sees+ -- a payload in commands like @stats cachedump 1 100@ and then hangs.+ payloadLen ("ms":_:size:_) = readMaybe (C.unpack size)+ payloadLen ("set":_:_:_:size:_) = readMaybe (C.unpack size)+ payloadLen _ = Nothing++recvAll :: N.Socket -> Int -> IO B.ByteString+recvAll _ 0 = return B.empty+recvAll socket n = do+ chunk <- N.recv socket n+ if B.null chunk+ then throwIO eof+ else (chunk <>) <$> recvAll socket (n - B.length chunk)+ where eof = userError "mock server EOF"++-- | Read up to and excluding the next @\\n@, one byte at a time so that+-- nothing past the line is consumed.+recvLine :: N.Socket -> [B.ByteString] -> IO B.ByteString+recvLine socket acc = do+ chunk <- N.recv socket 1+ if B.null chunk then throwIO (userError "mock server EOF")+ else if chunk == "\n" then return (B.concat (reverse acc))+ else recvLine socket (chunk:acc)
+ test/Spec.hs view
@@ -0,0 +1,720 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import MockServer+import qualified Database.Memcache.Auth as Auth+import qualified Database.Memcache.Client as Client+import qualified Database.Memcache.ElastiCacheClient as EC+import Database.Memcache.Errors+import qualified Database.Memcache.Errors as E+import Database.Memcache.Socket+import Database.Memcache.Types (ServerSpec (..), parseServerSpec)+import Test.Hspec+import Test.Hspec.QuickCheck (modifyMaxSuccess, prop)+import Test.QuickCheck+import qualified Data.ByteString as B+import qualified Data.ByteString.Base64 as Base64+import qualified Data.ByteString.Char8 as C+import qualified Network.Socket as N+import qualified Network.Socket.ByteString as N+import Control.Concurrent (forkIO, threadDelay)+import Control.Exception (bracket)+import Data.Either (isLeft)+import System.Timeout (timeout)++main :: IO ()+main = hspec $ do+ describe "meta command encoding" $ do+ it "leaves safe keys unchanged and encodes unsafe keys" $ do+ fmap wireKey (encodeKey "safe-key") `shouldBe` Right "safe-key"+ fmap wireKey (encodeKey "a b") `shouldBe` Right (Base64.encode "a b")++ it "terminates arithmetic commands" $+ arithmeticRequest Incr (wire "counter") 0 1 maxBound 0 `shouldSatisfy` C.isSuffixOf "\r\n"++ it "validates encoded key length" $+ encodeKey (C.replicate 251 'a') `shouldSatisfy` isLeft++ it "rejects empty keys" $+ encodeKey "" `shouldSatisfy` isLeft++ it "accepts a key of exactly 250 bytes unchanged" $+ encodeKey (C.replicate 250 'a') `shouldBe` Right (WireKey (C.replicate 250 'a') False)++ it "base64-encodes keys containing control bytes or DEL" $ do+ fmap wireB64 (encodeKey "tab\tkey") `shouldBe` Right True+ fmap wireB64 (encodeKey "line\nkey") `shouldBe` Right True+ fmap wireB64 (encodeKey "del\DELkey") `shouldBe` Right True++ it "sends bytes above 0x7f unencoded" $+ encodeKey "caf\xc3\xa9" `shouldBe` Right (WireKey "caf\xc3\xa9" False)++ it "rejects a key whose base64 form exceeds 250 bytes" $ do+ -- 186 bytes encode to 248 base64 bytes; 187 bytes encode to 252.+ fmap (B.length . wireKey) (encodeKey (C.replicate 186 ' ')) `shouldBe` Right 248+ encodeKey (C.replicate 187 ' ') `shouldSatisfy` isLeft++ prop "round-trips every key it accepts" $+ forAll keyGen $ \key -> case encodeKey key of+ Right (WireKey bytes True) ->+ Base64.decode bytes === Right key .&&. B.length bytes <= 250+ Right (WireKey bytes False) ->+ bytes === key .&&. not (B.any unsafeByte key)+ Left _ ->+ property (B.length key > 250 || (B.any unsafeByte key && base64Length key > 250))++ prop "only ever sends safe bytes as a key" $+ forAll keyGen $ \key -> case encodeKey key of+ Right (WireKey bytes _) -> not (B.any unsafeByte bytes)+ Left _ -> True++ prop "encodes each non-storage request as exactly one line" $+ forAll keyGen $ \key -> case encodeKey key of+ Left _ -> property True+ Right w -> conjoin+ [ oneLine (getRequest w)+ , oneLine (getManyRequest w)+ , oneLine (gatRequest w 30)+ , oneLine (touchRequest w 30)+ , oneLine (deleteRequest w 7)+ , oneLine (arithmeticRequest Incr w 1 2 30 7)+ ]++ it "rejects over-long keys without contacting the server" $+ withClient [MR "VA 3 f2 c7\r\nfoo\r\n"] $ \client -> do+ Client.get client (C.replicate 251 'a') `shouldThrow` isKeyTooLong+ Client.get client "key" `shouldReturn` Just ("foo", 2, 7)++ it "rejects unsafe stats keys before sending them" $ do+ client <- Client.newClient [Client.def] Client.def+ Client.stats client (Just "items\r\nset injected 0 0 1\r\nx")+ `shouldThrow` isKeyTooLong++ it "encodes all core request forms" $ do+ getRequest (wire "key") `shouldBe` "mg key v f c\r\n"+ getManyRequest (wire "key") `shouldBe` "mg key v f c k q\r\n"+ gatRequest (wire "key") 30 `shouldBe` "mg key v f c T30\r\n"+ touchRequest (wire "key") 30 `shouldBe` "mg key T30 c\r\n"+ deleteRequest (wire "key") 9 `shouldBe` "md key C9\r\n"+ flushRequest Nothing `shouldBe` "flush_all\r\n"+ flushRequest (Just 30) `shouldBe` "flush_all 30\r\n"+ versionRequest `shouldBe` "version\r\n"+ statsRequest Nothing `shouldBe` "stats\r\n"+ statsRequest (Just "key") `shouldBe` "stats key\r\n"+ quitRequest `shouldBe` "quit\r\n"+ noOpRequest `shouldBe` "mn\r\n"+ authRequest "user pass" `shouldBe` "set auth 0 0 9\r\nuser pass\r\n"++ it "flags encoded keys on every keyed request" $ do+ let encoded = Base64.encode "a b"+ getRequest (wire "a b") `shouldBe` "mg " <> encoded <> " b v f c\r\n"+ deleteRequest (wire "a b") 0 `shouldBe` "md " <> encoded <> " b\r\n"+ modifyRequest Append (wire "a b") "x" 0 `shouldBe` "ms " <> encoded <> " 1 MA b c\r\nx\r\n"++ it "places the binary-key flag after the data length" $+ storeRequest Set (wire "a b") "value" 3 20 0+ `shouldBe` "ms " <> Base64.encode "a b" <> " 5 F3 T20 MS b c\r\nvalue\r\n"++ it "encodes store modes and CAS" $ do+ storeRequest Add (wire "key") "v" 1 2 0 `shouldBe` "ms key 1 F1 T2 ME c\r\nv\r\n"+ storeRequest Replace (wire "key") "v" 1 2 5 `shouldBe` "ms key 1 F1 T2 C5 MR c\r\nv\r\n"+ modifyRequest Append (wire "key") "xyz" 5 `shouldBe` "ms key 3 MA C5 c\r\nxyz\r\n"+ modifyRequest Prepend (wire "key") "xyz" 0 `shouldBe` "ms key 3 MP c\r\nxyz\r\n"++ it "encodes arithmetic with auto-creation" $+ arithmeticRequest Decr (wire "n") 10 2 60 7 `shouldBe` "ma n D2 C7 MD v c N60 J10\r\n"++ it "omits arithmetic defaults at the maximum expiration" $+ arithmeticRequest Incr (wire "key") 0 1 maxBound 0+ `shouldBe` "ma key D1 MI v c\r\n"++ it "omits zero CAS values" $+ storeRequest Set (wire "key") "v" 0 0 0 `shouldBe` "ms key 1 F0 T0 MS c\r\nv\r\n"++ -- Most encoders write into a single buffer for speed; check them all+ -- against the plainest possible spelling of each request.+ prop "encodes every request as the concatenation of its pieces" $+ forAll wireKeyGen $ \key ->+ forAll requestValueGen $ \value ->+ forAll ((,,) <$> numberGen <*> numberGen <*> numberGen) $ \(flags, expiry, version) ->+ forAll ((,) <$> numberGen <*> numberGen) $ \(initial, delta) ->+ forAll ((,) <$> elements [Set, Add, Replace, Append, Prepend] <*> elements [Incr, Decr]) $ \(mode, arith) ->+ let keyArgs = wireKey key <> binaryFlag key+ binaryFlag k = if wireB64 k then " b" else ""+ casArg n = if n == 0 then "" else " C" <> decimal n+ decimal :: Show a => a -> B.ByteString+ decimal = C.pack . show+ modeLetter m = case m of+ Set -> "S"+ Add -> "E"+ Replace -> "R"+ Append -> "A"+ Prepend -> "P"+ arithLetter m = case m of+ Incr -> "I"+ Decr -> "D"+ in conjoin+ [ getRequest key === B.concat ["mg ", keyArgs, " v f c\r\n"]+ , getManyRequest key === B.concat ["mg ", keyArgs, " v f c k q\r\n"]+ , gatRequest key expiry === B.concat ["mg ", keyArgs, " v f c T", decimal expiry, "\r\n"]+ , touchRequest key expiry === B.concat ["mg ", keyArgs, " T", decimal expiry, " c\r\n"]+ , storeRequest mode key value flags expiry version === B.concat+ [ "ms ", wireKey key, " ", decimal (B.length value), " F", decimal flags+ , " T", decimal expiry, casArg version, " M", modeLetter mode, binaryFlag key+ , " c\r\n", value, "\r\n" ]+ , modifyRequest mode key value version === B.concat+ [ "ms ", wireKey key, " ", decimal (B.length value), " M", modeLetter mode+ , binaryFlag key, casArg version, " c\r\n", value, "\r\n" ]+ , arithmeticRequest arith key initial delta expiry version === B.concat+ [ "ma ", keyArgs, " D", decimal delta, casArg version, " M", arithLetter arith, " v c"+ , if expiry == maxBound then "" else " N" <> decimal expiry <> " J" <> decimal initial+ , "\r\n" ]+ , deleteRequest key version === B.concat ["md ", keyArgs, casArg version, "\r\n"]+ , flushRequest (Just expiry) === B.concat ["flush_all ", decimal expiry, "\r\n"]+ , flushRequest Nothing === "flush_all\r\n"+ , statsRequest Nothing === "stats\r\n"+ , statsRequest (Just (wireKey key)) === B.concat ["stats ", wireKey key, "\r\n"]+ , versionRequest === "version\r\n"+ , quitRequest === "quit\r\n"+ , noOpRequest === "mn\r\n"+ , authRequest value === B.concat ["set auth 0 0 ", decimal (B.length value), "\r\n", value, "\r\n"]+ ]++ describe "response parsing" $ do+ it "parses values and protocol errors" $ do+ recvResponseFrom "VA 3 f7 c9 O2\r\nabc\r\n" `shouldReturn` Response "VA" ["3", "f7", "c9", "O2"] "abc"+ (recvResponseFrom "SERVER_ERROR broken\r\n" >>= responseStatus) `shouldThrow` isServerError+ (recvResponseFrom "CLIENT_ERROR bad\r\n" >>= responseStatus) `shouldThrow` isBadCommand++ it "maps protocol statuses" $ do+ (recvResponseFrom "VA 0 f1 c2\r\n\r\n" >>= responseStatus) `shouldReturn` NoError+ (recvResponseFrom "EN\r\n" >>= responseStatus) `shouldReturn` ErrKeyNotFound+ (recvResponseFrom "NF\r\n" >>= responseStatus) `shouldReturn` ErrKeyNotFound+ (recvResponseFrom "HD\r\n" >>= responseStatus) `shouldReturn` NoError+ (recvResponseFrom "EX\r\n" >>= responseStatus) `shouldReturn` ErrKeyExists+ (recvResponseFrom "NS\r\n" >>= responseStatus) `shouldReturn` ErrItemNotStored+ (recvResponseFrom "MN\r\n" >>= responseStatus) `shouldReturn` NoError+ (recvResponseFrom "VERSION 1.6.0\r\n" >>= responseStatus) `shouldReturn` NoError+ (recvResponseFrom "STAT curr_items 2\r\n" >>= responseStatus) `shouldReturn` NoError++ it "maps server error text back to operation statuses" $ do+ (recvResponseFrom "SERVER_ERROR object too large for cache\r\n" >>= responseStatus)+ `shouldThrow` (== OpError ErrValueTooLarge)+ (recvResponseFrom "SERVER_ERROR out of memory storing object\r\n" >>= responseStatus)+ `shouldThrow` (== OpError ErrOutOfMemory)+ (recvResponseFrom "CLIENT_ERROR cannot increment or decrement non-numeric value\r\n" >>= responseStatus)+ `shouldThrow` (== OpError ErrValueNonNumeric)++ it "rejects unknown and empty response lines" $ do+ (recvResponseFrom "ERROR\r\n" >>= responseStatus) `shouldThrow` isUnknownOp+ (recvResponseFrom "XX\r\n" >>= responseStatus) `shouldThrow` isUnknownStatus+ recvResponseFrom "\r\n" `shouldThrow` isUnknownPkt++ it "rejects a value without its trailing line break" $+ recvResponseFrom "VA 3\r\nabcX\r\n" `shouldThrow` isBadLength++ it "reads flags, CAS and numbers from responses" $ do+ responseCas (Response "HD" ["c42"] "") `shouldBe` 42+ responseCas (Response "HD" [] "") `shouldBe` 0+ responseCas (Response "HD" ["cxyz"] "") `shouldBe` 0+ responseFlags (Response "VA" ["5", "f7", "c9"] "") `shouldBe` 7+ responseFlags (Response "VA" ["5"] "") `shouldBe` 0+ responseNumber (Response "VA" ["2"] "42") `shouldBe` 42+ responseNumber (Response "VA" ["3"] "abc") `shouldBe` 0++ describe "connection buffering" $ do+ it "keeps bytes after a parsed response" $+ bracket+ (N.socketPair N.AF_UNIX N.Stream 0)+ (\(left, right) -> N.close left >> N.close right) $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "VA 3 f1 c2\r\nabc\r\nMN\r\n"+ recvResponse connection `shouldReturn` Response "VA" ["3", "f1", "c2"] "abc"+ recvResponse connection `shouldReturn` Response "MN" [] ""++ it "handles a value split across packets" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "VA 5 f1 c2\r\nhe" >> N.sendAll right "llo\r\n"+ recvResponse connection `shouldReturn` Response "VA" ["5", "f1", "c2"] "hello"++ modifyMaxSuccess (const 50) $+ prop "parses responses however they are split across packets" $+ forAll ((,) <$> valueGen <*> listOf (choose (0, 1.0 :: Double))) $ \(value, cuts) ->+ ioProperty $ do+ let bytes = "VA " <> C.pack (show (B.length value)) <> " f1 c2\r\n"+ <> value <> "\r\nHD c3\r\n"+ pieces = splitAtFractions cuts bytes+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ mapM_ (\piece -> N.sendAll right piece >> threadDelay 200) pieces+ first <- recvResponse connection+ second <- recvResponse connection+ return $ first === Response "VA" [C.pack (show (B.length value)), "f1", "c2"] value+ .&&. second === Response "HD" ["c3"] ""++ it "reads through a delimiter split across packets" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "CONFIG x\r\nEN" >> threadDelay 1000 >> N.sendAll right "D\r\nrest"+ recvRawUntil "END\r\n" connection `shouldReturn` "CONFIG x\r\nEND\r\n"+ recvExact connection 4 `shouldReturn` "rest"++ it "reports EOF in a value" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "VA 5\r\nhe" >> N.close right+ recvResponse connection `shouldThrow` isUnexpectedEOF++ it "rejects unreasonably large value lengths" $+ recvResponseFrom "VA 134217729\r\n" `shouldThrow` isBadLength++ it "rejects an unterminated response line" $+ recvResponseFrom (C.replicate (16 * 1024) 'x') `shouldThrow` isBadLength++ describe "getMany" $ do+ it "matches echoed keys and skips misses" $+ withServerAndClient [MR "VA 5 f1 c11 kone\r\nfirst\r\n", Noop, MR "MN\r\n"] $ \server client -> do+ Client.getMany client ["one", "two"] `shouldReturn` [("one", ("first", 1, 11))]+ mockRequests server `shouldReturn`+ ["mg one v f c k q\r\n", "mg two v f c k q\r\n", "mn\r\n"]++ it "maps encoded keys back to the caller's key" $ do+ let encoded = Base64.encode "a b"+ withClient [MR ("VA 3 f0 c4 k" <> encoded <> " b\r\nabc\r\n"), MR "MN\r\n"] $ \client ->+ Client.getMany client ["a b"] `shouldReturn` [("a b", ("abc", 0, 4))]++ it "tells a plain key from another key's base64 form" $+ -- " " is sent, and echoed, as IA== with the b flag.+ withClient [MR "VA 1 f0 c1 kIA==\r\na\r\n", MR "VA 1 f0 c2 kIA== b\r\nb\r\n", MR "MN\r\n"] $ \client ->+ Client.getMany client ["IA==", " "] `shouldReturn` [("IA==", ("a", 0, 1)), (" ", ("b", 0, 2))]++ describe "client operations" $ do+ it "gets a hit" $+ withServerAndClient [MR "VA 3 f2 c7\r\nfoo\r\n"] $ \server client -> do+ Client.get client "key" `shouldReturn` Just ("foo", 2, 7)+ mockRequests server `shouldReturn` ["mg key v f c\r\n"]++ it "returns Nothing for a miss" $+ withClient [MR "EN\r\n"] $ \client ->+ Client.get client "key" `shouldReturn` Nothing++ it "rejects a reply meant for a different command" $+ withClient [MR "HD\r\n"] $ \client ->+ Client.get client "key" `shouldThrow` isUnexpectedResponse++ it "rejects an unknown reply" $+ withClient [MR "XX\r\n"] $ \client ->+ Client.get client "key" `shouldThrow` isUnknownStatus++ it "gets and touches a value" $+ withServerAndClient [MR "VA 3 f2 c7\r\nfoo\r\n", MR "EN\r\n"] $ \server client -> do+ Client.gat client "key" 60 `shouldReturn` Just ("foo", 2, 7)+ Client.gat client "key" 60 `shouldReturn` Nothing+ mockRequests server `shouldReturn` ["mg key v f c T60\r\n", "mg key v f c T60\r\n"]++ it "touches a value" $+ withClient [MR "HD c3\r\n", MR "EN\r\n"] $ \client -> do+ Client.touch client "key" 60 `shouldReturn` Just 3+ Client.touch client "key" 60 `shouldReturn` Nothing++ it "stores and returns the server CAS" $+ withServerAndClient [MR "HD c9\r\n"] $ \server client -> do+ Client.set client "key" "foo" 2 30 `shouldReturn` 9+ mockRequests server `shouldReturn` ["ms key 3 F2 T30 MS c\r\nfoo\r\n"]++ it "reports a set the server did not store" $+ withClient [MR "NS\r\n"] $ \client ->+ Client.set client "key" "foo" 0 0 `shouldThrow` (== OpError ErrItemNotStored)++ it "adds only when the key is new" $+ withClient [MR "HD c5\r\n", MR "NS\r\n"] $ \client -> do+ Client.add client "key" "v" 0 0 `shouldReturn` Just 5+ Client.add client "key" "v" 0 0 `shouldReturn` Nothing++ it "replaces only when the key exists and the CAS matches" $+ withClient [MR "HD c8\r\n", MR "NS\r\n", MR "NF\r\n", MR "EX\r\n"] $ \client -> do+ Client.replace client "key" "v" 0 0 0 `shouldReturn` Just 8+ Client.replace client "key" "v" 0 0 0 `shouldReturn` Nothing+ Client.replace client "key" "v" 0 0 4 `shouldReturn` Nothing+ Client.replace client "key" "v" 0 0 4 `shouldReturn` Nothing++ it "compares and swaps" $+ withServerAndClient [MR "HD c3\r\n", MR "EX\r\n", MR "NF\r\n"] $ \server client -> do+ Client.cas client "key" "v" 0 0 2 `shouldReturn` Just 3+ Client.cas client "key" "v" 0 0 2 `shouldReturn` Nothing+ Client.cas client "key" "v" 0 0 2 `shouldReturn` Nothing+ requests <- mockRequests server+ requests `shouldSatisfy` all (== "ms key 1 F0 T0 C2 MS c\r\nv\r\n")++ it "deletes a present key" $+ withClient [MR "HD\r\n"] $ \client ->+ Client.delete client "key" 0 `shouldReturn` True++ it "reports a delete of a missing key or with a stale CAS" $+ withClient [MR "NF\r\n", MR "EX\r\n"] $ \client -> do+ Client.delete client "key" 0 `shouldReturn` False+ Client.delete client "key" 5 `shouldReturn` False++ it "does not retry a command the server rejected" $+ withClient [MR "CLIENT_ERROR bad\r\n", MR "VA 3 f2 c7\r\nfoo\r\n"] $ \client ->+ Client.get client "key" `shouldThrow` isBadCommand++ it "reads version" $+ withClient [MR "VERSION 1.6.0\r\n"] $ \client ->+ Client.version client `shouldReturn` "1.6.0"++ it "collects stats until END, keeping values that contain spaces" $+ withClient [MR "STAT curr_items 2\r\nSTAT rusage_user 0.1 0.2\r\nEND\r\n"] $ \client -> do+ result <- Client.stats client Nothing+ result `shouldSatisfy` allStats++ it "allows stats arguments containing spaces" $+ withServerAndClient [MR "END\r\n"] $ \server client -> do+ result <- Client.stats client (Just "cachedump 1 100")+ map snd result `shouldBe` [[]]+ mockRequests server `shouldReturn` ["stats cachedump 1 100\r\n"]++ it "keeps non-STAT stats lines" $+ withClient [MR "ITEM key [3 b; 0 s]\r\nEND\r\n"] $ \client -> do+ result <- Client.stats client (Just "cachedump 1 1")+ map snd result `shouldBe` [[("ITEM", "key [3 b; 0 s]")]]++ it "returns Nothing when arithmetic cannot create the item" $+ withClient [MR "NS\r\n"] $ \client ->+ Client.increment client "counter" 0 1 30 0 `shouldReturn` Nothing++ it "returns the new counter value and CAS" $+ withServerAndClient [MR "VA 1 c2\r\n0\r\n"] $ \server client -> do+ Client.decrement client "counter" 10 1 60 0 `shouldReturn` Just (0, 2)+ mockRequests server `shouldReturn` ["ma counter D1 MD v c N60 J10\r\n"]++ it "reports arithmetic on a non-numeric value" $+ withClient [MR "CLIENT_ERROR cannot increment or decrement non-numeric value\r\n"] $ \client ->+ Client.increment client "key" 0 1 0 0 `shouldThrow` (== OpError ErrValueNonNumeric)++ it "returns Nothing appending to a missing key" $+ withClient [MR "NS\r\n"] $ \client ->+ Client.append client "key" "more" 0 `shouldReturn` Nothing++ it "prepends to an existing key" $+ withServerAndClient [MR "HD c4\r\n"] $ \server client -> do+ Client.prepend client "key" "pre" 0 `shouldReturn` Just 4+ mockRequests server `shouldReturn` ["ms key 3 MP c\r\npre\r\n"]++ it "flushes" $+ withServerAndClient [MR "OK\r\n"] $ \server client -> do+ Client.flush client (Just 10) `shouldReturn` ()+ mockRequests server `shouldReturn` ["flush_all 10\r\n"]++ it "reports an oversized value as a status, not a protocol failure" $+ withClient [MR "SERVER_ERROR object too large for cache\r\n"] $ \client ->+ Client.set client "key" "foo" 0 0 `shouldThrow` (== OpError ErrValueTooLarge)++ it "reports a server out of memory as a status" $+ withClient [MR "SERVER_ERROR out of memory storing object\r\n"] $ \client ->+ Client.set client "key" "foo" 0 0 `shouldThrow` (== OpError ErrOutOfMemory)++ it "returns no stats on a reset reply" $+ withClient [MR "RESET\r\n"] $ \client -> do+ result <- Client.stats client Nothing+ map snd result `shouldBe` [[]]++ it "sends quit" $+ withServerAndClient [Noop] $ \server client -> do+ Client.quit client+ requests <- waitForRequests server 1+ requests `shouldBe` ["quit\r\n"]++ describe "retries and failover" $ do+ let fast = Client.def+ { Client.optsServerTimeout = 300+ , Client.optsFailRetryDelay = 10+ , Client.optsDeadRetryDelay = 300+ }++ it "retries after the server drops the connection" $+ withServerAndClientOpts fast [CloseConnection, MR "VA 3 f2 c7\r\nfoo\r\n"] $ \server client -> do+ Client.get client "key" `shouldReturn` Just ("foo", 2, 7)+ length <$> mockRequests server `shouldReturn` 2++ it "retries after an operation times out" $+ withServerAndClientOpts fast [DelayMS 350 Noop, MR "VA 3 f2 c7\r\nfoo\r\n"] $ \_ client ->+ Client.get client "key" `shouldReturn` Just ("foo", 2, 7)++ it "marks a server dead once its retries run out, then revives it" $+ withServerAndClientOpts fast [CloseConnection, CloseConnection, MR "VA 3 f2 c7\r\nfoo\r\n"] $ \server client -> do+ -- The default of two retries means two attempts in total.+ Client.get client "key" `shouldThrow` isUnexpectedEOF+ length <$> mockRequests server `shouldReturn` 2+ Client.get client "key" `shouldThrow` (== ClientError NoServersReady)+ threadDelay 400000+ Client.get client "key" `shouldReturn` Just ("foo", 2, 7)++ describe "authenticate" $ do+ it "accepts credentials once the server confirms auth-file mode" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "CLIENT_ERROR unauthenticated\r\nSTORED\r\n"+ Auth.authenticate connection (Auth.Auth "user" "pass") `shouldReturn` ()+ recvExactly right (B.length expectedAuthRequests) `shouldReturn` expectedAuthRequests++ it "reports rejected credentials" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "CLIENT_ERROR unauthenticated\r\nCLIENT_ERROR authentication failure\r\n"+ Auth.authenticate connection (Auth.Auth "user" "pass") `shouldThrow` (== OpError TextAuthFail)++ it "refuses to send credentials when the server is not auth-gated" $+ bracket+ (N.socketPair N.AF_UNIX N.Stream 0)+ (\(left, right) -> N.close left >> N.close right) $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right "VERSION 1.6\r\n"+ Auth.authenticate connection (Auth.Auth "user" "pass") `shouldThrow` isInvalidAuthentication++ it "rejects credentials containing delimiters before contacting the server" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, _) -> do+ connection <- newConnection left+ Auth.authenticate connection (Auth.Auth "us er" "pass") `shouldThrow` isInvalidAuthentication++ it "does nothing without credentials" $+ bracket (N.socketPair N.AF_UNIX N.Stream 0) closePair $ \(left, _) -> do+ connection <- newConnection left+ Auth.authenticate connection Auth.NoAuth `shouldReturn` ()++ it "validates credential delimiters" $ do+ Auth.validateCredentials "user" "pass" `shouldBe` Right ()+ mapM_ (\c -> Auth.validateCredentials ("us" <> c <> "er") "pass" `shouldSatisfy` isLeft)+ [" ", "\t", "\r", "\n", ":"]+ mapM_ (\c -> Auth.validateCredentials "user" ("pa" <> c <> "ss") `shouldSatisfy` isLeft)+ [" ", "\t", "\r", "\n"]++ it "authenticates each new client connection" $+ withMockServer [MR "CLIENT_ERROR unauthenticated\r\n", MR "STORED\r\n", MR "VA 3 f2 c7\r\nfoo\r\n"] $ \server -> do+ let spec = (mockServerSpec server) { ssAuth = Auth.Auth "user" "pass" }+ client <- Client.newClient [spec] Client.def+ Client.get client "key" `shouldReturn` Just ("foo", 2, 7)+ mockRequests server `shouldReturn`+ ["version\r\n", "set auth 0 0 9\r\nuser pass\r\n", "mg key v f c\r\n"]++ describe "server URIs" $ do+ it "parses host and port" $+ parseServerSpec "memcached://cache.example:1234" `shouldBe`+ Right (ServerSpec "cache.example" "1234" Auth.NoAuth)++ it "defaults the port and host" $ do+ parseServerSpec "memcached://cache.example" `shouldBe`+ Right (ServerSpec "cache.example" "11211" Auth.NoAuth)+ parseServerSpec "memcached://" `shouldBe`+ Right (ServerSpec "127.0.0.1" "11211" Auth.NoAuth)++ it "parses credentials" $+ parseServerSpec "memcached://user:secret@cache.example:1234" `shouldBe`+ Right (ServerSpec "cache.example" "1234" (Auth.Auth "user" "secret"))++ it "ignores incomplete credentials" $ do+ parseServerSpec "memcached://user@cache.example" `shouldBe`+ Right (ServerSpec "cache.example" "11211" Auth.NoAuth)+ parseServerSpec "memcached://:secret@cache.example" `shouldBe`+ Right (ServerSpec "cache.example" "11211" Auth.NoAuth)++ it "rejects other schemes and invalid URIs" $ do+ parseServerSpec "http://cache.example" `shouldSatisfy` isLeft+ parseServerSpec "not a uri" `shouldSatisfy` isLeft++ describe "clusters" $ do+ it "fails over to another server once one is marked dead" $+ withMockServer [MR "VA 1 f0 c1\r\nA\r\n", CloseConnection, CloseConnection] $ \a ->+ withMockServer [MR "VA 1 f0 c1\r\nB\r\n", CloseConnection, CloseConnection] $ \b -> do+ client <- Client.newClient [mockServerSpec a, mockServerSpec b] fastFailover+ first <- Client.get client "key"+ -- The server that answered now drops two connections in a row, which+ -- uses up its retries and marks it dead...+ Client.get client "key" `shouldThrow` isUnexpectedEOF+ -- ...so the same key is now served by the other server.+ second <- Client.get client "key"+ map (fmap fst3) [first, second] `shouldSatisfy` (`elem` [[Just "A", Just "B"], [Just "B", Just "A"]])++ it "throws NoServersReady once every server is dead" $+ withMockServer [CloseConnection, CloseConnection] $ \a -> do+ client <- Client.newClient [mockServerSpec a] fastFailover+ Client.get client "key" `shouldThrow` isUnexpectedEOF+ Client.get client "key" `shouldThrow` (== ClientError NoServersReady)+ Client.flush client Nothing `shouldThrow` (== ClientError NoServersReady)+ Client.version client `shouldThrow` (== ClientError NoServersReady)++ it "flushes every server" $+ withMockServer [MR "OK\r\n"] $ \a ->+ withMockServer [MR "OK\r\n"] $ \b -> do+ client <- Client.newClient [mockServerSpec a, mockServerSpec b] Client.def+ Client.flush client Nothing+ mockRequests a `shouldReturn` ["flush_all\r\n"]+ mockRequests b `shouldReturn` ["flush_all\r\n"]++ it "collects stats from every server" $+ withMockServer [MR "STAT pid 1\r\nEND\r\n"] $ \a ->+ withMockServer [MR "STAT pid 2\r\nEND\r\n"] $ \b -> do+ client <- Client.newClient [mockServerSpec a, mockServerSpec b] Client.def+ results <- Client.stats client Nothing+ concatMap snd results `shouldMatchList` [("pid", "1"), ("pid", "2")]++ it "asks a single server for the version" $+ withMockServer [MR "VERSION 1.6.1\r\n"] $ \a ->+ withMockServer [MR "VERSION 1.6.1\r\n"] $ \b -> do+ client <- Client.newClient [mockServerSpec a, mockServerSpec b] Client.def+ Client.version client `shouldReturn` "1.6.1"+ requests <- (++) <$> mockRequests a <*> mockRequests b+ requests `shouldBe` ["version\r\n"]++ describe "ElastiCache discovery" $ do+ it "rejects an endpoint that is not a configuration endpoint" $+ isLeft (EC.parseConfigurationEndpoint "memcached://cache.example:11211") `shouldBe` True++ it "discovers nodes with the config command" $+ withMockServer [MR "VA 3 f0 c1\r\nfoo\r\n"] $ \node -> do+ let config = "CONFIG cluster 0 " <> nodeList node <> "END\r\n"+ withMockServer [MR "VERSION 1.6.12\r\n", MR config, Noop] $ \cfg -> do+ client <- EC.newClient Client.def hour (endpoint cfg)+ Client.get client "key" `shouldReturn` Just ("foo", 0, 1)+ waitForRequests cfg 3 `shouldReturn`+ ["version\r\n", "config get cluster\r\n", "quit\r\n"]++ it "discovers nodes from the cluster key on engines before 1.4.14" $+ withMockServer [MR "VA 3 f0 c1\r\nfoo\r\n"] $ \node -> do+ let value = "VALUE AmazonElastiCache:cluster 0 " <> nodeList node <> "END\r\n"+ withMockServer [MR "VERSION 1.4.5\r\n", MR value, Noop] $ \cfg -> do+ client <- EC.newClient Client.def hour (endpoint cfg)+ Client.get client "key" `shouldReturn` Just ("foo", 0, 1)+ waitForRequests cfg 3 `shouldReturn`+ ["version\r\n", "get AmazonElastiCache:cluster\r\n", "quit\r\n"]++ where+ recvResponseFrom bytes =+ bracket+ (N.socketPair N.AF_UNIX N.Stream 0)+ (\(left, right) -> N.close left >> N.close right) $ \(left, right) -> do+ connection <- newConnection left+ _ <- forkIO $ N.sendAll right bytes+ recvResponse connection+ wire key = either (error . show) id (encodeKey key)+ isKeyTooLong (ClientError (KeyTooLong _)) = True+ isKeyTooLong _ = False+ isInvalidAuthentication (ClientError (InvalidAuthentication _)) = True+ isInvalidAuthentication _ = False+ isServerError (ProtocolError (E.ServerError _)) = True+ isServerError _ = False+ isBadCommand (ProtocolError (E.BadCommand _)) = True+ isBadCommand _ = False+ isUnexpectedEOF (ProtocolError (UnexpectedEOF _)) = True+ isUnexpectedEOF _ = False+ isBadLength (ProtocolError (BadLength _)) = True+ isBadLength _ = False+ isUnknownOp (ProtocolError (UnknownOp _)) = True+ isUnknownOp _ = False+ isUnknownStatus (ProtocolError (UnknownStatus _)) = True+ isUnknownStatus _ = False+ isUnknownPkt (ProtocolError (UnknownPkt _)) = True+ isUnknownPkt _ = False+ isUnexpectedResponse (ProtocolError (UnexpectedResponse _)) = True+ isUnexpectedResponse _ = False+ closePair (left, right) = N.close left >> N.close right+ allStats [(_, [ ("curr_items", "2"), ("rusage_user", "0.1 0.2") ])] = True+ allStats _ = False+ expectedAuthRequests = "version\r\nset auth 0 0 9\r\nuser pass\r\n"+ fastFailover = Client.def { Client.optsFailRetryDelay = 10, Client.optsDeadRetryDelay = 60000 }+ fst3 (value, _, _) = value+ hour = 3600 * 1000000+ -- The endpoint check only looks for ".cfg" somewhere in the URI, so put it+ -- in the path and keep the host a plain loopback address.+ endpoint cfg = either error id $+ EC.parseConfigurationEndpoint ("memcached://127.0.0.1:" <> show (mockPort cfg) <> "/x.cfg")+ -- A config reply body listing one node, from its length line onwards.+ nodeList node =+ let body = "1\n127.0.0.1|127.0.0.1|" <> C.pack (show (mockPort node)) <> "\n"+ in C.pack (show (B.length body)) <> "\r\n" <> body <> "\r\n"++-- | Run an action with a client connected to a fresh mock server.+withClient :: [MockResponse] -> (Client.Client -> IO a) -> IO a+withClient responses action = withServerAndClient responses (const action)++withServerAndClient :: [MockResponse] -> (MockServer -> Client.Client -> IO a) -> IO a+withServerAndClient = withServerAndClientOpts Client.def++withServerAndClientOpts+ :: Client.Options -> [MockResponse] -> (MockServer -> Client.Client -> IO a) -> IO a+withServerAndClientOpts opts responses action = withMockServer responses $ \server -> do+ client <- Client.newClient [mockServerSpec server] opts+ action server client++-- | Wait (up to a second) for the mock to have received at least @n@ requests.+waitForRequests :: MockServer -> Int -> IO [B.ByteString]+waitForRequests server n = do+ result <- timeout 1000000 go+ maybe (mockRequests server) return result+ where+ go = do+ requests <- mockRequests server+ if length requests >= n then return requests else threadDelay 1000 >> go++-- | Receive exactly @n@ bytes from a socket.+recvExactly :: N.Socket -> Int -> IO B.ByteString+recvExactly _ 0 = return B.empty+recvExactly socket n = do+ chunk <- N.recv socket n+ if B.null chunk+ then return B.empty+ else (chunk <>) <$> recvExactly socket (n - B.length chunk)++-- | Keys of up to 260 arbitrary bytes, biased towards safe ones.+keyGen :: Gen B.ByteString+keyGen = B.pack <$> do+ len <- frequency [(4, choose (1, 64)), (1, choose (180, 260))]+ vectorOf len (frequency [(8, choose (0x21, 0x7e)), (1, arbitrary)])++valueGen :: Gen B.ByteString+valueGen = B.pack <$> do+ len <- choose (0, 3000)+ vectorOf len arbitrary++-- | Keys the encoder accepts, plain and base64.+wireKeyGen :: Gen WireKey+wireKeyGen = keyGen `suchThatMap` (either (const Nothing) Just . encodeKey)++-- | Request values, sometimes over the size above which+-- 'Data.ByteString.Builder.byteString' stops copying.+requestValueGen :: Gen B.ByteString+requestValueGen = frequency+ [ (4, valueGen)+ , (1, B.replicate <$> choose (8 * 1024, 20 * 1024) <*> arbitrary)+ ]++-- | Numbers of any size, with the edges of their range picked often.+numberGen :: (Arbitrary a, Bounded a, Num a) => Gen a+numberGen = frequency [(1, elements [0, 1, maxBound]), (4, arbitrary)]++unsafeByte :: (Ord a, Num a) => a -> Bool+unsafeByte c = c <= 0x20 || c == 0x7f++base64Length :: B.ByteString -> Int+base64Length key = 4 * ((B.length key + 2) `div` 3)++-- | A request line: ends in CRLF and contains no other line break.+oneLine :: B.ByteString -> Property+oneLine request =+ counterexample (show request) $+ C.count '\n' request === 1 .&&. C.count '\r' request === 1 .&&.+ property ("\r\n" `B.isSuffixOf` request)++-- | Split bytes at the given fractions of their length.+splitAtFractions :: [Double] -> B.ByteString -> [B.ByteString]+splitAtFractions cuts bytes = go 0 (map toIndex cuts) bytes+ where+ toIndex f = floor (f * fromIntegral (B.length bytes))+ go _ [] rest = [rest]+ go offset (i:is) rest+ | i <= offset = go offset is rest+ | otherwise =+ let (now, later) = B.splitAt (i - offset) rest+ in now : go i is later