Neks 0.3.0.0 → 0.4.0.0
raw patch · 5 files changed
+186/−156 lines, 5 filesdep ~basenew-component:exe:NeksClientnew-component:exe:NeksServer
Dependency ranges changed: base
Files
- Neks.cabal +25/−7
- Network/Neks/Client.hs +0/−70
- Network/Neks/NeksClient.hs +71/−0
- Network/Neks/NeksServer.hs +90/−0
- Network/Neks/Server.hs +0/−79
Neks.cabal view
@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: Neks-version: 0.3.0.0+version: 0.4.0.0 synopsis: Simple networked key/value store description: Neks is a fast, simple in-memory key/value server. license: MIT@@ -19,11 +19,11 @@ type: git location: https://github.com/wyager/Neks.git -executable Client- main-is: Network/Neks/Client.hs +executable NeksClient+ main-is: Network/Neks/NeksClient.hs -- other-modules: -- other-extensions: - build-depends: base >=4.6 && <4.8,+ build-depends: base >=4.6 && <= 4.8, messagepack >= 0.2.1, bytestring >= 0.10.4.0, cereal >= 0.4.1.0,@@ -36,11 +36,11 @@ default-language: Haskell2010 ghc-options: -O2 -executable Server- main-is: Network/Neks/Server.hs +executable NeksServer+ main-is: Network/Neks/NeksServer.hs -- other-modules: -- other-extensions: - build-depends: base >=4.6 && <4.8,+ build-depends: base >=4.6 && <= 4.8, messagepack >= 0.2.1, bytestring >= 0.10.4.0, cereal >= 0.4.1.0,@@ -52,3 +52,21 @@ directory >= 1.2.1.0 default-language: Haskell2010 ghc-options: -O2 -threaded -with-rtsopts=-N2++library+ exposed-modules: Network.Neks.Actions+ Network.Neks.DataStore+ Network.Neks.Disk+ Network.Neks.Message+ Network.Neks.NetPack+ 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
− Network/Neks/Client.hs
@@ -1,70 +0,0 @@-{-# 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/NeksClient.hs view
@@ -0,0 +1,71 @@+{-# 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+ ["--help"] -> putStrLn instructions -- To be explicit+ _ -> 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/NeksServer.hs view
@@ -0,0 +1,90 @@+module Main where++import qualified Network as Net+import System.IO (Handle, hClose)+import System.Environment (getArgs)+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++main = do+ args <- getArgs+ case args of+ [] -> serveWithPersistence+ ["--no-persistence"] -> serveWithoutPersistence+ ["--help"] -> putStrLn instructions -- To be explicit+ _ -> putStrLn instructions++serveWithPersistence = 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)++serveWithoutPersistence = do+ globalStore <- atomically createStore+ putStrLn "Serving on port 9999\nNot persisting to disk"+ serve globalStore (Net.PortNumber 9999)++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)++instructions = "Usage: Server <opt-args>\n" +++ "<opt-args> can be empty or can be \"--no-persistence\" to disable storing keys and values on disk"
− Network/Neks/Server.hs
@@ -1,79 +0,0 @@-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)