packages feed

starling (empty) → 0.0.1

raw patch · 7 files changed

+844/−0 lines, 7 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring

Files

+ CHANGES view
@@ -0,0 +1,3 @@+0.0.1++ - Initial release
+ LICENSE view
@@ -0,0 +1,9 @@+Copyright (c) 2009, Antoine Latter+All rights reserved.++Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:++Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.+Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.+Neither the name of Antoine Latter nor the names of any contributors may be used to endorse or promote products derived from this software without specific prior written permission.+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Network/Starling.hs view
@@ -0,0 +1,125 @@+++{-|++Module: Network.Starling+Copyright: Antoine Latter 2009+Maintainer: Antoine Latter <aslatter@gmail.com>++A haskell implementation of the memcahed+protocol.++This implements the new binary protocol, so+it only works with memcached version 1.3 and newer.++Example of usage, using the network package to obain+a handle, and the OverloadedStrings language extension:++> h <- connectTo "filename" $ UnixSocket "filename"+> hSetBuffering h NoBuffering+> con <- open h++> set con "hello" "world"+> get con "hello"++In the above example we connect to a unix socket in the file \"filename\",+set the key \"hello\" to the value \"world\" and then retrieve the value.++Operations are thread safe - multiple threads of execution may make+concurrent requests on the memcahced connection.++Operations are blocking, but do not block other concurrent threads+from placing requests on the connection.++-}+module Network.Starling+    ( open+    , close+    , Connection+    , Key+    , Value+    , Result+    , ResponseStatus(..)+    , set+    , get+    , delete+    -- , add+    -- , replace+    -- , increment+    -- , decrement+    , flush+    , stats+    -- , oneStat -- doesn't seem to work for me+    , version+    ) where++import Network.Starling.Connection+import Network.Starling.Core hiding+    ( get+    , set+    , delete+    , add+    , replace+    , increment+    , decrement+    , flush+    , stat+    , version+    , quit+    )+import qualified Network.Starling.Core as Core++import Data.ByteString.Lazy (ByteString)+import qualified Data.ByteString.Lazy as BS++type Result a = IO (Either (ResponseStatus, ByteString) a)++-- | Set a value in the cache+set :: Connection -> Key -> Value -> Result ()+set con key value = simpleRequest con (Core.set key value) (const ())++-- | Retrive a value from the cache+get :: Connection -> Key -> Result ByteString+get con key = simpleRequest con (Core.get key) rsBody++-- | Delete an entry in the cache+delete :: Connection -> Key -> Result ()+delete con key = simpleRequest con (Core.delete key) (const ())++-- | Delete all entries in the cache+flush :: Connection -> Result ()+flush con = simpleRequest con Core.flush (const ())++simpleRequest :: Connection -> Request -> (Response -> a) -> Result a+simpleRequest con req f+    = do+  response <- synchRequest con req+  if rsStatus response == NoError+   then return . Right . f $ response+   else return . errorResult $ response++errorResult response = Left (rsStatus response, rsBody response)++-- | Returns a list of stats about the server in key,value pairs+stats :: Connection -> Result [(ByteString,ByteString)]+stats con+    = do+  resps <- synchRequestMulti con $ Core.stat Nothing+  if null resps then error "fatal error in Network.Starling.stats"+   else do+     let resp = head resps+     if rsStatus resp == NoError+      then return . Right . unpackStats $ resps+      else return $ errorResult resp+ where unpackStats = filter (\(x,y) -> not (BS.null x && BS.null y)) .+                     map (\response -> (rsKey response, rsBody response))++-- | Returns a single stat. Example: 'stat con "pid"' will return+-- the +oneStat :: Connection -> Key -> Result ByteString+oneStat con key = simpleRequest con (Core.stat $ Just key) rsBody++-- | Returns the version of the server+version :: Connection -> Result ByteString+version con = simpleRequest con Core.version rsBody+
+ Network/Starling/Connection.hs view
@@ -0,0 +1,216 @@+{-# LANGUAGE RecordWildCards  #-}++{-|++All of the craziness for thread-safety+and asynchronous operations lives here.++The idea is that if someone comes up+with a better way of managing connection+state they can build what they want on top+of the Core module.++Operations are not entirely asynch -+they block until a response is returned.++But we don't hold the connection lock while+we're blocking, so other threads can still+put requests on the connection.++This should work well where each thread needs one+request to do it's job.++If you have a good idea for what an asynchronous+API should look like let me know. It shouldn't be too+hard to add on to what's already here.++-}+module Network.Starling.Connection+    ( Connection+    , open+    , close+    , synchRequest+    , synchRequestMulti+    , ignorantRequest+    ) where++import Network.Starling.Core++import Control.Concurrent (forkIO)+import Control.Concurrent.Chan+import Control.Concurrent.MVar+import Control.Exception (handle)+import Data.IORef++import System.IO++import qualified Data.Binary.Builder as B+import Data.ByteString.Lazy (ByteString)+import qualified Data.ByteString.Lazy as BS+++-- | For thread safety of operations, we perform+-- all requests on a wrapper around a handle.+data Connection = Conn +    { con_lock :: MVar ()+    , con_h :: Handle+    , con_q :: Chan QItem+    , con_opaque :: IORef Opaque+    }++-- | The connection keeps a queue of callbacks.+-- These are entries on that queue+data QItem+    = QDone+    | QHandleResp Opaque (Response -> IO ())+    | QHandleMulti Opaque ([Response] -> IO ())++-- | Create a connection.+-- Please don't use the handle after opening a+-- connection with it.+open :: Handle -> IO Connection+open h+    = do+  lock <- newMVar ()+  queue <- newChan+  opaque <- newIORef 0+  forkIO $ flip handle (readLoop h queue) $ \StarlingReadError ->+      return ()+  return $ Conn lock h queue opaque+           +-- | Process the callback queue+readLoop :: Handle -> Chan QItem -> IO ()+readLoop h q+    = do+  response <- getResponse h+  tryNextQItem h q response $ readLoop h q++tryNextQItem :: Handle -> Chan QItem -> Response -> IO () -> IO ()+tryNextQItem h q response k+    = do+  qItem <- readChan q+ +  case compareOpaque response qItem of+    KeepQ -> unGetChan q qItem >> k+    KeepR -> tryNextQItem h q response k+    Match -> processResponse h q response qItem k+    Done  -> return ()++-- | Until we have an implementation of RFC 1982, we+-- never return 'KeepR'+compareOpaque :: Response -> QItem -> CompareResult+compareOpaque response qItem+    = let qIdentm = qOpaque qItem+          rIdent = rsOpaque response+      in case qIdentm of+           Nothing -> Done+           Just qIdent ->+            case compare qIdent rIdent of+              EQ -> Match+              LT -> KeepQ+              GT -> KeepQ++qOpaque QDone = Nothing+qOpaque (QHandleResp ident _) = Just ident+qOpaque (QHandleMulti ident _) = Just ident++data CompareResult+    = KeepQ+    | KeepR+    | Match+    | Done++processResponse :: Handle -> Chan QItem+                -> Response -> QItem -> IO () -> IO ()+processResponse h q response qItem k+    = case qItem of+        QHandleResp _ f -> f response >> k+        QHandleMulti ident f -> do+                   (resps, left) <- takeResponses ident h+                   f (response : resps)+                   tryNextQItem h q left k++-- | Take many responses off of the queue as long as they+-- match the passed in senquence. The second returned+-- value is the response which did not match.+takeResponses :: Opaque -> Handle -> IO ([Response], Response)+takeResponses ident h+    = go []+ where go xs = do+         response <- getResponse h+         if rsOpaque response == ident+          then go (response:xs)+          else return (reverse xs, response)++withConLock :: Connection -> IO a -> IO a+withConLock Conn{..} k+    = withMVar_ con_lock k++withMVar_ :: MVar a -> IO b -> IO b+withMVar_ mvar f = withMVar mvar $ const f+++-- | This function ignores anything coming back from+-- the server.+-- Non-blocking.+ignorantRequest :: Connection -> Request -> IO ()+ignorantRequest conn req+    = withConLock conn $+      putRequest conn req >>= \_ -> return ()+      -- don't enqueue response handler++-- | Place a synchronous request which only returns+-- one reply+synchRequest :: Connection -> Request -> IO Response+synchRequest conn req+    = do+  result <- newEmptyMVar+  withConLock conn $ do+    opaque <- putRequest conn req+    enqueue conn $ QHandleResp opaque $ putMVar result+  readMVar result++-- | Place a synchronous request which may return+-- multiple response ('Stat', pretty much)+synchRequestMulti :: Connection -> Request -> IO [Response]+synchRequestMulti conn req+    = do+  result <- newEmptyMVar+  withConLock conn $ do+    opaque <- putRequest conn req+    enqueue conn $ QHandleMulti opaque $ putMVar result+    _ <- putRequest conn noop+    return ()+  readMVar result++-- | Shut down the connection.+-- Non-blocking.+close :: Connection -> IO ()+close conn+    =+  withConLock conn $ do+    enqueue conn QDone+    _ <- putRequest conn quit+    return ()++-- | Put a request onto the handle+putRequest :: Connection -> Request -> IO Opaque+putRequest conn@Conn{..} req+    = do+  opaque <- nextOpaque conn+  let chunk =  B.toLazyByteString $ serialize $ addOpaque opaque req+  BS.hPut con_h chunk+  return opaque++-- | Grab the next sequence number+nextOpaque :: Connection -> IO Opaque+nextOpaque Conn{..} = do+  current <- readIORef con_opaque+  let next = current + 1+  writeIORef con_opaque next+  return $ next `seq` current++-- | Add an item on the callback queue+enqueue :: Connection -> QItem -> IO ()+enqueue Conn{..} item+    = writeChan con_q item
+ Network/Starling/Core.hs view
@@ -0,0 +1,458 @@+{-# LANGUAGE+  RecordWildCards+  , DeriveDataTypeable+ #-}++{-|++Primitives for the memcached protocol.++-}+module Network.Starling.Core+    ( Request+    , requestOp+    , Key+    , Value+    , set+    , add+    , replace+    , get+    , increment+    , decrement+    , append+    , prepend+    , delete+    , quit+    , flush+    , noop+    , version+    , stat+    , addOpaque+    , addCAS+    , Response(..)+    , getResponse+    , StarlingReadError(..)+    , StarlingReadError+    , Serialize(..)+    , Deserialize(..)+    , Opaque+    , OpCode(..)+    , DataType(..)+    , CAS+    , nullCAS+    , ResponseStatus(..)+    ) where++import System.IO++import Control.Applicative ((<$>))+import Control.Exception+import Data.Maybe (fromMaybe)+import Data.Typeable+import Data.Word+import Data.Monoid (mconcat)++import Data.ByteString.Lazy (ByteString)+import qualified Data.ByteString.Lazy as BS++import Data.ByteString.Lazy.Char8()++import qualified Data.Binary.Builder as B+import qualified Data.Binary.Get as B++type Opaque = Word32+type Key = ByteString+type Value = ByteString+type ErrorInfo = ByteString++-- | Set a value in the cache.+set :: Key -> Value -> Request+set key value+    = let extras = setExtras 0 0+      in request Set extras key value++-- | Add a value to cache. Fails if+-- already present.+add :: Key -> Value -> Request+add key value+    = let extras = setExtras 0 0+      in request Add extras key value++-- | Replaces a value in cahce. Fails if+-- not present.+replace :: Key -> Value -> Request+replace key value+    = let extras = setExtras 0 0+      in request Replace extras key value++setExtras :: Word32 -> Word32 -> ByteString+setExtras flags expiry+    = B.toLazyByteString $ mconcat+      [ B.putWord32be flags+      , B.putWord32be expiry+      ]++-- | Get a value from cache+get :: Key -> Request+get key = request Get BS.empty key BS.empty++increment :: Key+          -> Word64 -- ^ amount+          -> Word64 -- ^ initial value+          -> Request+increment key amount init+    = let extras = incExtras amount init 0+      in request Increment extras key BS.empty++decrement :: Key+          -> Word64 -- ^ amount+          -> Word64 -- ^ initial value+          -> Request+decrement key amount init+    = let extras = incExtras amount init 0+      in request Decrement extras key BS.empty++incExtras :: Word64 -> Word64 -> Word32 -> ByteString+incExtras amount init expiry+    = B.toLazyByteString $ mconcat+      [ B.putWord64be amount+      , B.putWord64be init+      , B.putWord32be expiry+      ]++-- | Delete a cache entry+delete :: Key -> Request+delete key+    = request Delete BS.empty key BS.empty++-- | Quit+quit :: Request+quit = request Quit BS.empty BS.empty BS.empty++-- | Flush the cache+flush :: Request+flush+    = let extras = B.toLazyByteString $ B.putWord32be 0 -- expiry+      in request Flush extras BS.empty BS.empty++-- | Keepalive. Flushes responses for quiet operations.+noop :: Request+noop = request NoOp BS.empty BS.empty BS.empty++-- | Returns the server version+version :: Request+version = request Version BS.empty BS.empty BS.empty++-- | Appends the value to the value in the cache+append :: Key -> Value -> Request+append key value+    = request Append BS.empty key value++-- | Prepends the value to the value in the cache+prepend :: Key -> Value -> Request+prepend key value+    = request Prepend BS.empty key value++-- | Fetch statistics about the cahce. Returns a sequence+-- of responses.+stat :: Maybe Key -> Request+stat mkey = request Stat BS.empty (fromMaybe BS.empty mkey) BS.empty++-- | Add an opaque marker to a request.+-- This is returned unchanged in the corresponding+-- response.+addOpaque :: Opaque -> Request -> Request+addOpaque n req = req { rqOpaque = n }++-- | Add a version tag to a request. When+-- added to a set/replace request, the request+-- will fail if the data has been modified since+-- the CAS was retrieved for the item.+addCAS :: CAS -> Request -> Request+addCAS n req = req { rqCas = n }++class Serialize a where+    serialize :: a -> B.Builder++class Deserialize a where+    deserialize :: B.Get a++data Request+    = Req+      { rqMagic    :: RqMagic+      , rqOp       :: OpCode+      , rqDataType :: DataType+      , rqOpaque   :: Opaque+      , rqCas      :: CAS+      , rqExtras   :: ByteString+      , rqKey      :: ByteString+      , rqBody     :: ByteString+      }+ deriving (Eq, Ord, Read, Show)++instance Serialize Request where+    serialize Req{..} =+        let keyLen = BS.length rqKey+            extraLen = BS.length rqExtras+            bodyLen = BS.length rqBody+        in mconcat+            [ serialize            rqMagic+            , serialize            rqOp+            , B.putWord16be        (fromIntegral keyLen)+            , B.singleton           (fromIntegral extraLen)+            , serialize            rqDataType+            , B.putWord16be        0 -- reserved+            , B.putWord32be        (fromIntegral $ keyLen + extraLen + bodyLen)+            , B.putWord32be        rqOpaque+            , serialize            rqCas+            , B.fromLazyByteString rqExtras+            , B.fromLazyByteString rqKey+            , B.fromLazyByteString rqBody+            ]++-- | A starter request with fields set to reasonable+-- defaults. The opcode field is left undefined.+baseRequest :: Request+baseRequest+    = Req { rqMagic = Request+          , rqKey = BS.empty+          , rqExtras = BS.empty+          , rqDataType = RawData+          , rqBody = BS.empty+          , rqOpaque = 0+          , rqCas = nullCAS+          }++-- | Returns the operation the request will perform+requestOp :: Request -> OpCode+requestOp = rqOp++request :: OpCode+        -> BS.ByteString  -- ^ Extras+        -> BS.ByteString  -- ^ Key+        -> BS.ByteString  -- ^ Body+        -> Request+request opCode extras key body+    = let extraLen = fromIntegral (BS.length extras)+          keyLen = fromIntegral (BS.length key)+      in baseRequest+          { rqOp     = opCode+          , rqExtras = extras+          , rqKey    = key+          , rqBody   = body+          }++newtype CAS = CAS Word64+ deriving (Eq, Ord, Read, Show)++instance Serialize CAS where+    serialize (CAS n)+        = B.putWord64be n++instance Deserialize CAS where+    deserialize = CAS <$> B.getWord64be++nullCAS :: CAS+nullCAS = CAS 0++data Response+    = Res+      { rsMagic  :: RsMagic+      , rsOp     :: OpCode+      , rsDataType :: DataType+      , rsStatus :: ResponseStatus+      , rsOpaque :: Opaque+      , rsCas :: CAS+      , rsExtras :: ByteString+      , rsKey :: ByteString+      , rsBody :: ByteString+      }+ deriving (Eq, Ord, Read, Show)++instance Deserialize Response where+    deserialize = do+      rsMagic <- deserialize+      rsOp <- deserialize+      rsKeyLen <- B.getWord16be+      rsExtraLen <- B.getWord8+      rsDataType <- deserialize+      rsStatus <- deserialize+      rsTotalLen <- B.getWord32be+      let totalLen = fromIntegral rsTotalLen+          keyLen   = fromIntegral rsKeyLen+          extraLen = fromIntegral rsExtraLen+      rsOpaque <- B.getWord32be+      rsCas <- deserialize+      rsExtras <- B.getLazyByteString extraLen+      rsKey <- B.getLazyByteString keyLen+      rsBody <- B.getLazyByteString (totalLen - extraLen - keyLen)+      return Res{..}++newtype ResponseHeader+    = ResHead+      { rsHeadTotalLen :: Word32 }++instance Deserialize ResponseHeader where+    deserialize = do+      _ <- B.getBytes 8+      rsHeadTotalLen <- B.getWord32be+      _ <- B.getBytes 12+      return ResHead{..}++-- | Pulls a reponse to an operation+-- off of a handle.+-- May throw a 'StarlingReadError'+getResponse :: Handle -> IO Response+getResponse h = do+  chunk <- BS.hGet h 24+  if BS.length chunk /= 24 then throw StarlingReadError else do+  let resHeader = B.runGet deserialize chunk+      bodyLen = rsHeadTotalLen resHeader+  rest <- BS.hGet h $ fromIntegral bodyLen+  return . B.runGet deserialize $ chunk `BS.append` rest++data StarlingReadError = StarlingReadError+ deriving (Show, Typeable)++instance Exception StarlingReadError++data RqMagic = Request+ deriving (Eq, Ord, Read, Show)+instance Serialize RqMagic where+    serialize Request = B.singleton 0x80++data RsMagic = Response+ deriving (Eq, Ord, Read, Show)+instance Deserialize RsMagic where+    deserialize = do+      magic <- B.getWord8+      case magic of+        0x81 -> return Response++data DataType = RawData+ deriving (Eq, Ord, Read, Show)+instance Serialize DataType where+    serialize RawData = B.singleton 0x00+instance Deserialize DataType where+    deserialize = do+      dtype <- B.getWord8+      case dtype of+        0x00 -> return RawData++data ResponseStatus+    = NoError+    | KeyNotFound+    | KeyExists+    | ValueTooLarge+    | InvalidArguments+    | ItemNotStored+    | IncrDecrOnNonNumeric+    | UnknownCommand+    | OutOfMemory+ deriving (Eq, Ord, Read, Show)++instance Deserialize ResponseStatus where+    deserialize = do+      status <- B.getWord16be+      return $ case status of+        0x0000 -> NoError+        0x0001 -> KeyNotFound+        0x0002 -> KeyExists+        0x0003 -> ValueTooLarge+        0x0004 -> InvalidArguments+        0x0005 -> ItemNotStored+        0x0006 -> IncrDecrOnNonNumeric+        0x0081 -> UnknownCommand+        0x0082 -> OutOfMemory++data OpCode+    = Get+    | Set+    | Add+    | Replace+    | Delete+    | Increment+    | Decrement+    | Quit+    | Flush+    | GetQ+    | NoOp+    | Version+    | GetK+    | GetKQ+    | Append+    | Prepend+    | Stat+    | SetQ+    | AddQ+    | ReplaceQ+    | DeleteQ+    | IncrementQ+    | DecrementQ+    | QuitQ+    | FlushQ+    | AppendQ+    | PrependQ+ deriving (Eq, Ord, Read, Show)++instance Serialize OpCode where+    serialize Get = B.singleton 0x00+    serialize Set = B.singleton 0x01+    serialize Add = B.singleton 0x02+    serialize Replace = B.singleton 0x03+    serialize Delete = B.singleton 0x04+    serialize Increment = B.singleton 0x05+    serialize Decrement = B.singleton 0x06+    serialize Quit = B.singleton 0x07+    serialize Flush = B.singleton 0x08+    serialize GetQ = B.singleton 0x09+    serialize NoOp = B.singleton 0x0a+    serialize Version = B.singleton 0x0b+    serialize GetK = B.singleton 0x0c+    serialize GetKQ = B.singleton 0x0d+    serialize Append = B.singleton 0x0e+    serialize Prepend = B.singleton 0x0f+    serialize Stat = B.singleton 0x10+    serialize SetQ = B.singleton 0x11+    serialize AddQ = B.singleton 0x12+    serialize ReplaceQ = B.singleton 0x13+    serialize DeleteQ = B.singleton 0x14+    serialize IncrementQ = B.singleton 0x15+    serialize DecrementQ = B.singleton 0x16+    serialize QuitQ = B.singleton 0x17+    serialize FlushQ = B.singleton 0x18+    serialize AppendQ = B.singleton 0x19+    serialize PrependQ = B.singleton 0x1a++instance Deserialize OpCode where+    deserialize = do+      command <- B.getWord8+      return $ case command of+        0x00 -> Get+        0x01 -> Set+        0x02 -> Add+        0x03 -> Replace+        0x04 -> Delete+        0x05 -> Increment+        0x06 -> Decrement+        0x07 -> Quit+        0x08 -> Flush+        0x09 -> GetQ+        0x0a -> NoOp+        0x0b -> Version+        0x0c -> GetK+        0x0d -> GetKQ+        0x0e -> Append+        0x0f -> Prepend+        0x10 -> Stat+        0x11 -> SetQ+        0x12 -> AddQ+        0x13 -> ReplaceQ+        0x14 -> DeleteQ+        0x15 -> IncrementQ+        0x16 -> DecrementQ+        0x17 -> QuitQ+        0x18 -> FlushQ+        0x19 -> AppendQ+        0x20 -> PrependQ+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ starling.cabal view
@@ -0,0 +1,31 @@+name:            starling+version:         0.0.1+stability:       Alpha++synopsis:        A memcached client+description:     A haskell memcached client. See http:\/\/memcached.org+                 for more information.+                 .+                 This implements the new binary protocol,+		 so it only works with memcached version+                 1.3 and newer.+homepage:        http://community.haskell.org/~aslatter/code/starling++category:        Network++license:         BSD3+license-file:    LICENSE+author:          Antoine Latter <aslatter@gmail.com>+maintainer:      Antoine Latter <aslatter@gmail.com>++cabal-version:   >= 1.6++build-depends:   base==4.*, binary, bytestring+build-type:      Simple++extra-source-files: CHANGES++exposed-modules:+        Network.Starling+        Network.Starling.Core+        Network.Starling.Connection