packages feed

Neks (empty) → 0.3.0.0

raw patch · 10 files changed

+365/−0 lines, 10 filesdep +basedep +bytestringdep +cerealsetup-changed

Dependencies added: base, bytestring, cereal, containers, directory, hashable, messagepack, network, stm, vector

Files

+ LICENSE view
@@ -0,0 +1,19 @@+Copyright (c) 2014 William Yager++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ Neks.cabal view
@@ -0,0 +1,54 @@+-- Initial Neks.cabal generated by cabal init.  For further +-- documentation, see http://haskell.org/cabal/users-guide/++name:                Neks+version:             0.3.0.0+synopsis:            Simple networked key/value store+description:         Neks is a fast, simple in-memory key/value server. +license:             MIT+license-file:        LICENSE+author:              William Yager+maintainer:          will.yager@gmail.com+-- copyright:           +category:            Database+build-type:          Simple+extra-source-files:  Network/Neks/*.hs+cabal-version:       >=1.10      ++source-repository head+  type:     git+  location: https://github.com/wyager/Neks.git++executable Client+  main-is: Network/Neks/Client.hs           +  -- other-modules:       +  -- other-extensions:    +  build-depends:       base >=4.6 && <4.8,+                       messagepack >= 0.2.1,+                       bytestring >= 0.10.4.0,+                       cereal >= 0.4.1.0,+                       containers >= 0.5.5.1,+                       stm >= 2.4.2,+                       network >= 2.4.2.3,+                       vector >= 0.10.9.1,+                       hashable >= 0.1.0.0,+                       directory >= 1.2.1.0+  default-language:    Haskell2010+  ghc-options:         -O2++executable Server+  main-is: Network/Neks/Server.hs           +  -- other-modules:       +  -- other-extensions:    +  build-depends:       base >=4.6 && <4.8,+                       messagepack >= 0.2.1,+                       bytestring >= 0.10.4.0,+                       cereal >= 0.4.1.0,+                       containers >= 0.5.5.1,+                       stm >= 2.4.2,+                       network >= 2.4.2.3,+                       vector >= 0.10.9.1,+                       hashable >= 0.1.0.0,+                       directory >= 1.2.1.0+  default-language:    Haskell2010+  ghc-options:         -O2 -threaded -with-rtsopts=-N2
+ Network/Neks/Actions.hs view
@@ -0,0 +1,8 @@+module Network.Neks.Actions (+        Request(Set, Get, Delete, Atomic), Reply(Found, NotFound)+) where++import Data.ByteString (ByteString)++data Request = Set ByteString ByteString | Get ByteString | Delete ByteString | Atomic [Request] deriving (Show, Eq)+data Reply = Found ByteString | NotFound deriving (Show, Eq)
+ Network/Neks/Client.hs view
@@ -0,0 +1,70 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Network.Neks.Message (formatRequests, parseResponses)+import Network.Neks.NetPack (netWrite, netRead)+import Network.Neks.Actions (Request(Set, Get, Delete, Atomic), Reply(Found, NotFound))++import qualified Network as Net+import System.IO (Handle)+import System.Environment (getArgs)+import Data.ByteString.Char8 (pack)++import Control.Monad (when)+import Control.Concurrent (forkIO)+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)++main = Net.withSocketsDo $ do+        args <- getArgs+        case args of+                (host:port:command) -> do +                        let portID = Net.PortNumber . fromInteger . read $ port+                        server <- Net.connectTo host portID+                        case command of+                                ["--set", k, v] -> set k v server+                                ["--get", k]    -> get k server+                                ["--test"]      -> test host portID+                                _               -> putStrLn instructions+                _ -> putStrLn instructions++request :: Handle -> [Request] -> IO (Either String [Reply])+request server requests = do+        netWrite server $ formatRequests requests+        responseData <- netRead server+        return (responseData >>= parseResponses)++set :: String -> String -> Handle -> IO ()+set k v server = do+        response <- request server [Set (pack k) (pack v)]+        case response of+                Left error -> putStrLn ("Error setting value: " ++ error)+                Right [] -> putStrLn "Set successful"++get :: String -> Handle -> IO ()+get k server = do+        response <- request server [Get (pack k)]+        case response of+                Left error -> putStrLn ("Error getting value: " ++ error)+                Right response -> putStrLn ("Response received: " ++ show response)++test :: String -> Net.PortID -> IO ()+test host port = do+        putStrLn "Spawning 50 threads x 200 requests x 100 transactions"+        locks <- sequence [newEmptyMVar | _ <- [1..50]] -- 50 threads+        threads <- sequence [forkIO (testWith lock host port) | lock <- locks]+        sequence_ [takeMVar lock | lock <- locks] -- Wait for threads to finish+        putStrLn "Test complete"+        where testWith lock host port = do+                server <- Net.connectTo host port+                sequence . replicate 200 $ do+                        let requests = [Set k v | (k, v) <- zip testKeys testValues] ++ [Get k | k <- testKeys]+                        responses <- request server requests+                        when (responses /= Right [Found v | v <- testValues]) (error "Bad response")+                putMVar lock ()++testKeys = ["66991fb944", "afe0c0261a", "a4242d5dda", "d10db90845", "4384ecbfe", "a839702a82", "1ed8680b95", "0d2189d279", "f4b0795239", "a24d4e7e87", "28e24e1d51", "9bb0dfbfbd", "9776bad265", "89f79a8c71", "d50de7c1cd", "167a350f93", "36f41a6205", "f5bbd3bc20", "69a3d20bef", "33644bede7", "8744571558", "cd4ab79d3a", "8c26e6936c", "88c1d42e4e", "f31d532d05", "a9ad46aea2", "e9b0aeee64", "dffc6a25af", "90952b9dd", "04a136756e", "31ca38445e", "21c27b172", "5c09e01c46", "9b23b5ef27", "a9fd5ea170", "aa1718e735", "1ce6781a57", "a927b0584e", "e7aea00872", "52223f7078", "e620de282a", "1a4c71def8", "75bd1abc65", "af93442708", "2257127db4", "68ec4b4f7", "9b5473f839", "d453871c0f", "9657631a3d", "95503a22b9"]+testValues = ["5e7a195e90", "accdfc69c4", "43be950623", "afed0a6890", "0d23711bcf", "3b3d9b4043", "139ba09036", "a54b56630d", "61a729c150", "34891805ca", "d3dc68c9d3", "e1b4943d72", "8731015486", "f8f626c071", "4262ca1f24", "3c55632f50", "d32b8b30ca", "3311af7221", "29144d27ea", "0e0f97257e", "d6a2e1086", "aae1906c17", "d57f58433f", "9232138b5e", "fd1711214f", "84a66c50ac", "9b65ffc322", "d2d447396e", "6fc6c53265", "5183bca85", "884a5cc1cf", "7914d452ae", "6e2a351fd8", "7fb80954be", "3c3f1bf0cd", "112e60a719", "4917c12e1c", "9aaf5cc6d1", "7ccd97a418", "48c91da08c", "349524f781", "7d248047c", "9bfec0c3a4", "c0de587385", "216dd64a29", "eac5049f63", "133a259613", "843e1f1ee3", "e9c11331c0", "48e720933e"]++instructions = "Usage: Client <host> <port> <args>\n" +++                           "<args> are \"--test\", \"--get <key>\", or \"--set <key> <value>\""
+ Network/Neks/DataStore.hs view
@@ -0,0 +1,39 @@+module Network.Neks.DataStore (+        DataStore, createStore, dump, load, insert, get, delete+) where++import qualified Data.Map.Strict as Map (Map, empty, insert, lookup, delete)+import Control.Concurrent.STM (STM)+import Control.Concurrent.STM.TMVar (TMVar, newTMVar, takeTMVar, putTMVar, readTMVar)+import Data.Hashable (Hashable, hash)+import Data.Vector as Vector (Vector, fromList, toList, mapM, (!))+import Data.Bits ((.&.))++newtype DataStore k v = DataStore {mapsOf :: Vector (TMVar (Map.Map k v))}++createStore :: STM (DataStore k v)+createStore = load [Map.empty | _ <- [0..4096]]++insert :: (Hashable k, Ord k) => k -> v -> DataStore k v -> STM ()+insert k v (DataStore maps) = do+        let atomicMap = maps ! (hash k .&. 0xFFF)+        map <- takeTMVar atomicMap+        putTMVar atomicMap $! Map.insert k v map++get :: (Hashable k, Ord k) => k -> DataStore k v -> STM (Maybe v)+get k (DataStore maps) = do+        let atomicMap = maps ! (hash k .&. 0xFFF)+        map <- readTMVar atomicMap+        return (Map.lookup k map)++delete :: (Hashable k, Ord k) => k -> DataStore k v -> STM ()+delete k (DataStore maps) = do+        let atomicMap = maps ! (hash k .&. 0xFFF)+        map <- takeTMVar atomicMap+        putTMVar atomicMap $! Map.delete k map++dump :: DataStore k v -> STM [Map.Map k v]+dump = Prelude.mapM readTMVar . toList . mapsOf++load :: [Map.Map k v] -> STM (DataStore k v)+load = fmap DataStore . Vector.mapM newTMVar . fromList
+ Network/Neks/Disk.hs view
@@ -0,0 +1,28 @@+module Network.Neks.Disk (saveTo, loadFrom) where++import Control.Concurrent.STM (atomically)+import Network.Neks.DataStore (DataStore, dump, load)+import Network.Neks.NetPack (netRead, netWrite) -- also works on files+import System.IO (Handle, withFile, IOMode(ReadMode, WriteMode))+import System.Directory (renameFile, doesFileExist)+import Data.Serialize (Serialize, encode, decode)+import Control.Applicative ((<$>))++saveTo :: (Serialize a, Ord a, Serialize b) => String -> DataStore a b -> IO ()+saveTo path store = do+        withFile (path ++ "~") WriteMode (`saveToHandle` store)+        renameFile (path ++ "~") path+saveToHandle handle store = do+        maps <- atomically (dump store)+        sequence_ [netWrite handle (encode map) | map <- maps]++loadFrom :: (Serialize a, Ord a, Serialize b) => String -> IO (Maybe (DataStore a b))+loadFrom path = doesFileExist path >>= \exists -> if exists+        then Just <$> withFile path ReadMode loadFromHandle+        else return Nothing+loadFromHandle handle = do+        maps <- sequence . replicate 4096 $ do+                mapData <- netRead handle+                let Right map = mapData >>= decode+                return map+        atomically (load maps)
+ Network/Neks/Message.hs view
@@ -0,0 +1,42 @@+module Network.Neks.Message (+        parseRequests, formatRequests,+        parseResponses, formatResponses+) where++import Data.ByteString (ByteString)+import Network.Neks.Actions (Request(Set, Get, Delete, Atomic), Reply(Found, NotFound))+import qualified Data.Serialize as Serialize+import Data.MessagePack (Object(ObjectArray, ObjectInt, ObjectBinary))+import Control.Applicative ((<$>))++formatRequests :: [Request] -> ByteString+formatRequests = Serialize.encode . ObjectArray . map format+        where format request = ObjectArray $ case request of+                Get k           -> [ObjectInt 0, ObjectBinary k]+                Set k v         -> [ObjectInt 1, ObjectBinary k, ObjectBinary v]+                Delete k        -> [ObjectInt 2, ObjectBinary k]+                Atomic requests -> [ObjectInt 3, ObjectArray (map format requests)]++formatResponses :: [Reply] -> ByteString+formatResponses = Serialize.encode . ObjectArray . map format+        where format response = ObjectArray $ case response of+                Found v  -> [ObjectInt (-1), ObjectBinary v]+                NotFound -> [ObjectInt (-2)]++decode bs = case Serialize.decode bs of+        Right (ObjectArray messages) -> Right messages+        error -> Left "Response decode failure"++parseRequests :: ByteString -> Either String [Request]+parseRequests bs = decode bs >>= mapM parse where+        parse (ObjectArray [ObjectInt 0, ObjectBinary k]) = Right (Get k)+        parse (ObjectArray [ObjectInt 1, ObjectBinary k, ObjectBinary v]) = Right (Set k v)+        parse (ObjectArray [ObjectInt 2, ObjectBinary k]) = Right (Delete k)+        parse (ObjectArray [ObjectInt 3, ObjectArray requests]) = Atomic <$> mapM parse requests+        parse _ = Left "Invalid request structure"++parseResponses :: ByteString -> Either String [Reply]+parseResponses bs = decode bs >>= mapM parse where+        parse (ObjectArray [ObjectInt (-1), ObjectBinary v]) = Right (Found v)+        parse (ObjectArray [ObjectInt (-2)]) = Right NotFound+        parse _ = Left "Incorrect response type"
+ Network/Neks/NetPack.hs view
@@ -0,0 +1,24 @@+module Network.Neks.NetPack (netWrite, netRead) where++import Data.ByteString as BS (ByteString, hGet, length, hPut, append)+import Data.Serialize (encode, decode)+import System.IO (Handle)+import Data.Word (Word64)++netWrite :: Handle -> ByteString -> IO ()+netWrite hdl msg = do+        let len = fromIntegral (BS.length msg) :: Word64+        BS.hPut hdl (encode len `append` msg)++netRead :: Handle -> IO (Either String ByteString)+netRead hdl = do+        lengthBytes <- hGet hdl 8+        case decode lengthBytes of+                Left err -> return (Left "Network stream ended while reading length")+                Right len  | len < (10^6 :: Word64) -> readData (fromIntegral len)+                           | otherwise -> return (Left "Message too long")+        where readData length = do+                dataBytes <- hGet hdl length+                if BS.length dataBytes /= length+                        then return (Left "Connection dropped during message read")+                        else return (Right dataBytes)
+ Network/Neks/Server.hs view
@@ -0,0 +1,79 @@+module Main where++import qualified Network as Net+import System.IO (Handle, hClose)+import Data.ByteString (ByteString)+import Control.Monad (forever, unless)+import Control.Concurrent.STM (STM, atomically)+import Control.Concurrent (ThreadId, forkIO, threadDelay)+import Control.Exception (SomeException, catch, finally) +import Network.Neks.Disk (saveTo, loadFrom)+import Network.Neks.NetPack (netRead, netWrite)+import Network.Neks.Message (parseRequests, formatResponses)+import Network.Neks.DataStore (DataStore, createStore, insert, get, delete)+import Network.Neks.Actions (Request(Set, Get, Delete, Atomic), Reply(Found, NotFound))++type Store = DataStore ByteString ByteString++-- With disk persistence:+main = do+        loaded <- loadFrom "store.kvs"+        globalStore <- case loaded of+                Just store -> return store+                Nothing -> atomically createStore+        let periodically action = forever (threadDelay (30*10^6) >> action)+        forkIO $ periodically (saveTo "store.kvs" globalStore)+        putStrLn "Serving on port 9999 \nSaving DB to store.kvs"+        serve globalStore (Net.PortNumber 9999)++-- Without disk persistence: +-- main = do+--     globalStore <- atomically createStore+--     serve globalStore++serve :: Store -> Net.PortID -> IO ()+serve store port = Net.withSocketsDo $ do+        sock <- Net.listenOn port+        forever (wait sock store)++wait :: Net.Socket -> Store -> IO ThreadId+wait sock store = do+        (client, _, _) <- Net.accept sock+        forkIO (run client)+        where+        run client = (handle client store `finally` hClose client) `catch` exceptionHandler+        exceptionHandler :: SomeException -> IO ()+        exceptionHandler exception = return () -- Optional exception reporting++handle :: Handle -> Store -> IO ()+handle client store = do+        result <- processRequests client store+        case result of+                Right success -> handle client store+                Left failure -> return () -- Optional soft error reporting++processRequests :: Handle -> Store -> IO (Either String ())+processRequests client store = do+        requestData <- netRead client+        case requestData >>= parseRequests of+                Left err -> return (Left err)+                Right requests -> do+                        results <- mapM (atomically . processWith store) requests+                        netWrite client . formatResponses . concat $ results+                        return (Right ())++processWith :: Store -> Request -> STM [Reply]+processWith store (Set k v) = do+        insert k v store+        return []+processWith store (Get k) = do+        result <- get k store+        return $ case result of+                Nothing -> [NotFound]+                Just v -> [Found v]+processWith store (Delete k) = do+        delete k store+        return []+processWith store (Atomic requests) = do+        results <- mapM (processWith store) requests+        return (concat results)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain