diff --git a/AddOne.hs b/AddOne.hs deleted file mode 100644 --- a/AddOne.hs +++ /dev/null @@ -1,15 +0,0 @@ -module Main where - -import Data.Default ( def ) -import System.Environment ( getArgs ) -import System.Daemon - -addOne :: Int -> IO Int -addOne n = return (n + 1) - -main :: IO () -main = do - startDaemon "addOne" def addOne - [n] <- getArgs - res <- runClient "localhost" 5000 ((read n) :: Int) - print (res :: Maybe Int) diff --git a/Control/Pipe/C3.hs b/Control/Pipe/C3.hs deleted file mode 100644 --- a/Control/Pipe/C3.hs +++ /dev/null @@ -1,40 +0,0 @@ -module Control.Pipe.C3 ( - -- * Pipes - commandSender, commandReceiver - ) where - -import Control.Monad ( forever ) -import Control.Monad.Trans.Class ( lift ) -import Control.Pipe ( runPipe, await, yield, (<+<) ) -import Control.Pipe.Serialize ( serializer, deserializer ) -import Control.Pipe.Socket ( Handler ) -import Data.Serialize ( Serialize ) - --- | Send a single command over the outgoing pipe and wait for a --- response. If the incoming pipe is closed before a response --- arrives, returns @Nothing@. -commandSender :: (Serialize a, Serialize b) => a -> Handler (Maybe b) -commandSender command reader writer = do - runPipe (writer <+< serializer <+< sendCommand) - runPipe (receiveResponse - <+< (deserializer >> return Nothing) - <+< (reader >> return Nothing)) - where - sendCommand = do - yield command - - receiveResponse = do - res <- await - return (Just res) - --- | Wait for commands on the incoming pipe, handle them, and send the --- reponses over the outgoing pipe. -commandReceiver :: (Serialize a, Serialize b) => (a -> IO b) -> Handler () -commandReceiver executeCommand reader writer = do - runPipe (writer <+< serializer - <+< commandExecuter - <+< deserializer <+< reader) - where - commandExecuter = forever $ do - comm <- await - yield =<< lift (executeCommand comm) diff --git a/Control/Pipe/Serialize.hs b/Control/Pipe/Serialize.hs deleted file mode 100644 --- a/Control/Pipe/Serialize.hs +++ /dev/null @@ -1,48 +0,0 @@ --- | This module provides the 'deserializer' and 'serializer' pipes to --- convert 'B.ByteString's off of pipes into typed values. --- --- In order to use it, the types of the values need to have --- 'Serialize' instances. These can be derived automatically using --- "Ghc.Generics": --- --- > {-# LANGUAGE DeriveGeneric #-} --- > --- > data Foo = Bar String | Baz Int --- > deriving ( Generic ) --- > --- > instance Serialize Foo --- --- Note that in the above example: we use the @DeriveGeneric@ --- extension, derive a @Generic@ instance for our data-type, and write --- an /empty/ @Serialize@ instance. --- -module Control.Pipe.Serialize ( - -- * Pipes - serializer, deserializer - ) where - -import Data.ByteString.Char8 ( ByteString ) -import Data.Serialize ( Serialize, get, encode - , Result(..), runGetPartial ) -import Control.Pipe ( Pipe, await, yield ) -import Control.Monad ( forever ) - --- | De-serialize data from strict 'ByteString's. Uses @cereal@'s --- incremental 'Data.Serialize.Get' parser. -deserializer :: (Serialize a, Monad m) => Pipe ByteString a m () -deserializer = loop Nothing Nothing - where - loop mk mbin = do - bin <- maybe await return mbin - case (maybe (runGetPartial get) id mk) bin of - Fail reason -> fail reason - Partial k -> loop (Just k) Nothing - Done c bin' -> do - yield c - loop Nothing (Just bin') - --- | Serialize data into strict 'ByteString's. -serializer :: (Serialize a, Monad m) => Pipe a ByteString m () -serializer = forever $ do - x <- await - yield (encode x) diff --git a/Control/Pipe/Socket.hs b/Control/Pipe/Socket.hs deleted file mode 100644 --- a/Control/Pipe/Socket.hs +++ /dev/null @@ -1,79 +0,0 @@ -{-# LANGUAGE ScopedTypeVariables #-} - --- Thank you: --- pipes-network - -module Control.Pipe.Socket ( - -- * Socket pipes - socketReader, socketWriter, - - -- * Socket server/client - Handler, runSocketServer, runSocketClient - ) where - -import Control.Concurrent ( forkIO ) -import qualified Control.Exception as CE -import Control.Monad ( forever, unless ) -import Control.Monad.IO.Class ( MonadIO(..) ) -import Control.Monad.Trans.Class ( lift ) -import Control.Pipe ( Consumer, Producer, await, yield ) -import Data.ByteString.Char8 ( ByteString ) -import qualified Data.ByteString.Char8 as B -import Network.Socket ( Socket ) -import qualified Network.Socket as NS -import Network.Socket.ByteString ( sendAll, recv ) - --- | Stream data from the socket. -socketReader :: (MonadIO m) => Socket -> Producer ByteString m () -socketReader socket = do - bin <- lift . liftIO $ recv socket 4096 - unless (B.null bin) $ do - yield bin - socketReader socket - --- | Stream data to the socket. -socketWriter :: (MonadIO m) => Socket -> Consumer ByteString m () -socketWriter socket = forever $ do - bin <- await - lift . liftIO $ sendAll socket bin - --- | A simple handler: takes an incoming stream of 'ByteString's, an --- stream of 'ByteString's, and ties them together somehow. --- Conceptually, the simplest handler would be @identity@: --- --- > import Control.Monad --- > import Control.Pipe --- > import Data.ByteString.Char8 --- > --- > handler reader writer = do --- > let identity = forever $ do --- > x <- await --- > yield x --- > runPipe (writer <+< identity <+< reader) --- --- See the @pipes@ tutorial for more examples of writing pipes. --- --- Since 'ByteString's are fairly boring by themseleves, have a look --- at "Control.Pipe.Serialize" which lets you deserialize/serialize --- pipes of 'ByteString's easily. -type Handler r = Producer ByteString IO () - -> Consumer ByteString IO () - -> IO r - --- | Listen for connections on the given socket, and run 'Handler' on --- each received connection. The socket should previously have been --- bound to a port or to a file. Each handler is run in its own --- thread. Even in case of an error, the handlers' sockets are --- closed. -runSocketServer :: (MonadIO m) => Socket -> Handler () -> m () -runSocketServer lsocket handler = liftIO $ forever $ do - (socket, _addr) <- NS.accept lsocket - _ <- forkIO $ CE.finally - (handler (socketReader socket) (socketWriter socket)) - (NS.sClose socket) - return () - --- | Run 'Handler' on the given socket. -runSocketClient :: (MonadIO m) => Socket -> Handler r -> m r -runSocketClient socket handler = liftIO $ do - handler (socketReader socket) (socketWriter socket) diff --git a/Makefile b/Makefile --- a/Makefile +++ b/Makefile @@ -1,28 +1,39 @@ -.PHONY: all build dist install test clean doc p +CABAL := $(shell cabal-dev --version > /dev/null && echo cabal-dev || echo cabal) all: build test +.PHONY: all build dist install clean doc p test ghci + build: dist/setup-config - grep -E "$ " Memo.md | sed 's/$ //' > Memo.hs - cabal build + grep -E "$ " examples/Memo.md | sed 's/$ //' > examples/Memo.hs + $(CABAL) build -dist: test +dist: build cabal sdist install: build cabal install test: build - cabal test + $(CABAL) test clean: - cabal clean + $(CABAL) clean + rm -rf cabal-dev/ dist/setup-config: daemons.cabal - cabal configure --enable-tests +# If you don't have all the necessary packages installed on the first +# run, run `cabal-dev install`. + $(CABAL) configure --enable-tests || $(CABAL) install --enable-tests doc: build - cabal haddock + $(CABAL) haddock -p: clean - permamake.sh $(shell find */ -name '*.hs') *.cabal Makefile *.md +p: + permamake.sh $(shell find src/ -name '*.hs') \ + $(shell find test/ -name '*.hs') \ + *.cabal \ + Makefile + +ghci: build + cabal-dev ghci diff --git a/Memo.hs b/Memo.hs deleted file mode 100644 --- a/Memo.hs +++ /dev/null @@ -1,59 +0,0 @@ -{-# LANGUAGE DeriveGeneric, OverloadedStrings #-} - -module Main where - -import Control.Concurrent.MVar ( MVar, newMVar, modifyMVar ) -import Data.ByteString.Char8 ( ByteString ) -import Data.Default ( def ) -import Data.Serialize ( Serialize ) -import Data.String ( fromString ) -import qualified Data.Map as M -import GHC.Generics -import System.Environment ( getArgs ) -import System.Daemon - -data Command = Put ByteString ByteString - | Get ByteString - deriving ( Generic, Show ) - -instance Serialize Command - -data Response = Failed String - | Value ByteString - deriving ( Generic, Show ) - -instance Serialize Response - -type Book = M.Map ByteString ByteString - -handleCommand :: MVar Book -> Command -> IO Response -handleCommand bookVar comm = modifyMVar bookVar $ \book -> return $ - case comm of - Get key -> ( book - , maybe (Failed "not found") Value (M.lookup key book) ) - Put key value -> ( M.insert key value book - , Value "ok" ) - -main :: IO () -main = do - bookVar <- newMVar M.empty - let options = def { daemonPort = 7856 } - startDaemon "memo" options (handleCommand bookVar) - args <- getArgs - let args' = map fromString args - res <- case args' of - ["get", key] -> runClient "localhost" 7856 (Get key) - ["put", key, value] -> runClient "localhost" 7856 (Put key value) - _ -> error "invalid command" - print (res :: Maybe Response) - -{- -% dist/build/memo/memo get apples -Just (Failed "not found") - -% dist/build/memo/memo put apples 23 -Just (Value "ok") - -% dist/build/memo/memo get apples -Just (Value "23") --} diff --git a/NEWS.md b/NEWS.md --- a/NEWS.md +++ b/NEWS.md @@ -1,6 +1,11 @@ Changes ======= +0.1.2 (05 Apr 2013) +------------------- + + - add `System.Posix.Daemon.killAndWait`; thank you, @isturdy + 0.1.1 (18 Aug 2012) ------------------- diff --git a/PMTQ.hs b/PMTQ.hs deleted file mode 100644 --- a/PMTQ.hs +++ /dev/null @@ -1,98 +0,0 @@ -{-# LANGUAGE DeriveGeneric, OverloadedStrings #-} - -module Main where - -import Control.Concurrent.Chan ( Chan, newChan, readChan, writeChan ) -import Control.Concurrent.MVar ( MVar, newMVar, modifyMVar ) -import Control.Monad ( forever ) -import Control.Monad.Trans.Class ( lift ) -import Control.Pipe ( runPipe, (<+<), await, yield ) -import Control.Pipe.Serialize ( serializer, deserializer ) -import Control.Pipe.Socket ( Handler ) -import Data.ByteString.Char8 ( ByteString ) -import qualified Data.ByteString.Char8 as B -import Data.Char ( toLower ) -import Data.Default ( def ) -import Data.Serialize ( Serialize ) -import Data.String ( fromString ) -import qualified Data.Map as M -import GHC.Generics -import Network.Socket ( withSocketsDo ) -import System.Environment ( getArgs ) -import System.Daemon -import System.IO ( hPutStrLn, stderr ) - -data Command = Push ByteString ByteString - | Pop ByteString - | Consume ByteString - deriving ( Generic, Show ) - -instance Serialize Command - -data Response = Value ByteString - deriving ( Generic, Show ) - -instance Serialize Response - -type Registry = M.Map ByteString (Chan ByteString) - -handleCommands :: MVar Registry -> Handler () -handleCommands registryVar reader writer = do - runPipe (writer <+< serializer - <+< commandExecuter - <+< deserializer <+< reader) - where - commandExecuter = forever $ do - comm <- await - case comm of - Pop topic -> do - ch <- lift $ getCreateChan topic - transferToPipeFromChan ch - Consume topic -> do - ch <- lift $ getCreateChan topic - forever $ transferToPipeFromChan ch - Push topic val -> do - ch <- lift $ getCreateChan topic - lift $ writeChan ch val - yield (Value "ok") - - -- Transfer a value from the given channel to the pipe. - transferToPipeFromChan ch = do - val <- lift $ readChan ch - yield (Value val) - - -- Get the channel for the given topic, and create it if it does - -- not already exist. - getCreateChan topic = modifyMVar registryVar $ \registry -> do - case M.lookup topic registry of - Nothing -> do - ch <- newChan - return (M.insert topic ch registry, ch) - Just ch -> do - return (registry, ch) - -main :: IO () -main = withSocketsDo $ do - registryVar <- newMVar M.empty - let options = def { daemonPort = 7857 } - startDaemonWithHandler "pmtq" options (handleCommands registryVar) - args <- getArgs - let args' = map (fromString . map toLower) args - case args' of - ["pop", key] -> do - res <- runClient "localhost" 7857 (Pop key) - printResult res - ["push", key, value] -> do - res <- runClient "localhost" 7857 (Push key value) - printResult res - ["consume", key] -> do - runClientWithHandler "localhost" 7857 $ \reader writer -> do - runPipe (writer <+< serializer <+< yield (Consume key)) - runPipe ((forever $ await >>= \res -> lift (printResult (Just res))) - <+< deserializer <+< reader) - _ -> do - error "invalid command" - where - printResult :: Maybe Response -> IO () - printResult Nothing = hPutStrLn stderr "no response" - printResult (Just (Value val)) = B.putStrLn val diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ main :: IO () main = do - startDaemon "addOne" def addOne + ensureDaemonRunning "addOne" def addOne [n] <- getArgs res <- runClient "localhost" 5000 ((read n) :: Int) print (res :: Maybe Int) @@ -28,21 +28,40 @@ Running it, we see: % addone 22 + Daemon started on port 5000 Just 23 % addone 41 Just 42 -The two important functions above are `startDaemon`, which checks if a -daemon named `addOne` is already running, and starts it if not, and -`runClient` which connects to the daemon running on `localhost:5000`, -passes it a number, and waits for the response. +The two important functions above are `ensureDaemonRunning`, which +checks if a daemon named `addOne` is already running, and starts it if +not, and `runClient` which connects to the daemon running on +`localhost:5000`, passes it a number, and waits for the response. -For a less trivial example, see the in-memory key-value store, -[Memo](https://github.com/scvalex/daemons/blob/master/Memo.md). +What would I use this for? +-------------------------- -For an example that uses the streaming interface of `daemons`, see -[PMTQ](https://github.com/scvalex/daemons/blob/master/PMTQ.hs) (Poor -Man's Task Queue). + - You can use the `runDetached` from `System.Posix.Daemon` to turn + your program into a daemon for Unix-like systems. You'd want to do + this for practically every program that's meant to run as a server. + + - You can use the functions from `Control.Pipe.C3`, `Socket`, and + `Serialize` to communicate with running Haskell program. At the + simplest, you could query the program for its status, or instruct + it to shutdown cleanly. A more complex use would be adding a full + REPL into a running Haskell process (think `erl -remsh`). + + - You can use the helpers from `System.Daemon` to trivially do the + above. Check out the following tutorials and examples for details. + +Tutorials and examples +---------------------- + + - [Memo](https://github.com/scvalex/daemons/blob/master/examples/Memo.md) - + in which we write an in-memory key-value store, + + - [Queue](https://github.com/scvalex/daemons/blob/master/examples/Queue.hs) + - a task queue using the streaming interface of `daemons`. Installation ------------ diff --git a/System/Daemon.hs b/System/Daemon.hs deleted file mode 100644 --- a/System/Daemon.hs +++ /dev/null @@ -1,158 +0,0 @@ -module System.Daemon ( - -- * Daemons - startDaemon, startDaemonWithHandler, - - -- * Clients, - runClient, runClientWithHandler, - - -- * Types - DaemonOptions(..), PidFile(..), HostName, Port, - - -- * Helpers - bindPort, getSocket - ) where - -import Control.Concurrent ( threadDelay ) -import qualified Control.Exception as CE -import Control.Monad ( when ) -import Control.Pipe.C3 ( commandSender, commandReceiver ) -import Control.Pipe.Socket ( Handler, runSocketServer, runSocketClient ) -import Data.Default ( Default(..) ) -import Data.Serialize ( Serialize ) -import Data.String ( IsString(..) ) -import Network.Socket ( Socket, SockAddr(..), Family(..), SocketType(..) - , SocketOption(..), setSocketOption - , socket, sClose, connect, bindSocket, listen - , AddrInfo(..), getAddrInfo, addrAddress, defaultHints - , defaultProtocol, iNADDR_ANY, maxListenQueue ) -import System.Directory ( getHomeDirectory ) -import System.FilePath ( (), (<.>) ) -import System.Posix.Daemon ( runDetached, isRunning ) - -type Port = Int -type HostName = String - --- | The configuration options of a daemon. See 'startDaemon' for a --- description of each. -data DaemonOptions = DaemonOptions - { daemonPort :: Port - , daemonPidFile :: PidFile - } deriving ( Show ) - -instance Default DaemonOptions where - def = DaemonOptions { daemonPort = 5000 - , daemonPidFile = InHome - } - --- | The location of the daemon's pidfile. -data PidFile = InHome - | PidFile FilePath - deriving ( Show ) - -instance IsString PidFile where - fromString = PidFile - --- | Simple wrapper around 'startDaemonWithHandler' which uses a --- simple function to respond to commands and doesn't deal with pipes. --- --- The @handler@ is just a function that takes a command and returns a --- response. -startDaemon :: (Serialize a, Serialize b) - => String -- ^ name - -> DaemonOptions -- ^ options - -> (a -> IO b) -- ^ handler - -> IO () -startDaemon name options executeCommand = do - startDaemonWithHandler name options (commandReceiver executeCommand) - --- | Start a daemon running on the given port, using the given handler --- to respond to events. If the daemon is already running, just --- return. --- --- The pidfile @PidFile options@ will be created and locked. This --- function checks the pidfile to see if the daemon is already --- running. --- --- The daemon will listen for incoming connections on all interfaces --- on @daemonPort options@. --- --- The @handler@ is a function that takes the reader and writer --- 'ByteString' pipes and does something with them. See --- 'commandReceiver' for an example handler. -startDaemonWithHandler :: String -- ^ name - -> DaemonOptions -- ^ options - -> Handler () -- ^ handler - -> IO () -startDaemonWithHandler name options handler = do - home <- getHomeDirectory - let pidfile = case daemonPidFile options of - InHome -> home ("." ++ name) <.> "pid" - PidFile path -> path - running <- isRunning pidfile - when (not running) $ do - runDetached (Just pidfile) def $ do - CE.bracket - (bindPort (daemonPort options)) - sClose - (\lsocket -> - runSocketServer lsocket handler) - threadDelay 1000000 - --- | Send a command to the daemon running at the given network address --- and wait for a response. --- --- This is a simple wrapper around 'runClientWithHandler' that sends a --- single command and waits for a single response. --- --- If the connection is closed before receiving a response, return --- 'Nothing'. -runClient :: (Serialize a, Serialize b) - => HostName -- ^ hostname - -> Port -- ^ port - -> a -- ^ command - -> IO (Maybe b) -runClient hostname port comm = - runClientWithHandler hostname port (commandSender comm) - --- | Connect to the given network address and run the handler on the --- reader and wrier pipes for the socket. --- --- The @handler@ is a function that takes the reader and writer --- 'ByteString' pipes and does something with them. For an example --- handler, see 'commandSender', which sends a command and waits for a --- response. -runClientWithHandler :: HostName -- ^ hostname - -> Port -- ^ port - -> Handler a -- ^ command - -> IO a -runClientWithHandler hostname port handler = do - CE.bracket - (getSocket hostname port) - sClose - (\s -> runSocketClient s handler) - --- | Create a socket and bind it to the given port. -bindPort :: Port -> IO Socket -bindPort port = do - CE.bracketOnError - (socket AF_INET Stream defaultProtocol) - sClose - (\s -> do - setSocketOption s ReuseAddr 1 - bindSocket s (SockAddrInet (fromIntegral port) - iNADDR_ANY) - listen s maxListenQueue - return s) - --- | Create a socket connected to the given network address. -getSocket :: HostName -> Port -> IO Socket -getSocket hostname port = do - addrInfos <- getAddrInfo (Just (defaultHints { addrFamily = AF_INET })) - (Just hostname) - (Just $ show port) - CE.bracketOnError - (socket AF_INET Stream defaultProtocol) - sClose - (\s -> do - connect s (addrAddress $ head addrInfos) - return s) diff --git a/System/Posix/Daemon.hs b/System/Posix/Daemon.hs deleted file mode 100644 --- a/System/Posix/Daemon.hs +++ /dev/null @@ -1,191 +0,0 @@ --- | This module provides a simple interface to creating, checking the --- status of, and stopping background jobs. --- --- Use 'runDetached' to start a background job. For instance, here is --- a daemon that peridically hits a webserver: --- --- > import Control.Concurrent --- > import Control.Monad --- > import Data.Default --- > import Data.Maybe --- > import Network.BSD --- > import Network.HTTP --- > import Network.URI --- > import System.Posix.Daemon --- > --- > main :: IO () --- > main = runDetached (Just "diydns.pid") def $ forever $ do --- > hostname <- getHostName --- > _ <- simpleHTTP --- > (Request { rqURI = fromJust (parseURI "http://foo.com/dns") --- > , rqMethod = GET --- > , rqHeaders = [] --- > , rqBody = hostname }) --- > threadDelay (600 * 1000 * 1000) --- --- To check if the above job is running, use 'isRunning' with the same --- pidfile: --- --- > isRunning "diydns.pid" --- --- Finally, to stop the above job (maybe because we're rolling a new --- version of it), use 'kill': --- --- > kill "diydns.pid" --- --- As a side note, the code above is a script that the author uses as --- a sort of homebrew dynamic DNS: the remote address is a CGI script --- that records the IP addresses of all incoming requests in separate --- files named after the contents of the requests; the addresses are --- then viewable with any browser. -module System.Posix.Daemon ( - -- * Starting - runDetached, Redirection(..), - - -- * Status - isRunning, - - -- * Stopping - kill, brutalKill - ) where - -import Prelude hiding ( FilePath ) - -import Control.Monad ( when ) -import Data.Default ( Default(..) ) -import System.Directory ( doesFileExist ) -import System.FilePath ( FilePath ) -import System.IO ( SeekMode(..), hFlush, stdout ) -import System.Posix.Files ( stdFileMode ) -import System.Posix.IO ( openFd, OpenMode(..), defaultFileFlags, closeFd - , dupTo, stdInput, stdOutput, stdError, getLock - , createFile - , LockRequest (..), setLock, fdWrite, fdRead ) -import System.Posix.Process ( getProcessID, forkProcess, createSession ) -import System.Posix.Signals ( Signal, signalProcess, sigQUIT, sigKILL ) - --- | Where should the output (and input) of a daemon be redirected to? --- (we can't just leave it to the current terminal, because it may be --- closed, and that would kill the daemon). --- --- When in doubt, just use 'def', the default value. --- --- 'DevNull' causes the output to be redirected to @\/dev\/null@. This --- is safe and is what you want in most cases. --- --- If you don't want to lose the output (maybe because you're using it --- for logging), use 'ToFile', instead. -data Redirection = DevNull - | ToFile FilePath - deriving ( Show ) - -instance Default Redirection where - def = DevNull - --- | Run the given action detached from the current terminal; this --- creates an entirely new process. This function returns --- immediately. Uses the double-fork technique to create a well --- behaved daemon. If @pidfile@ is given, check/write it; if we --- cannot obtain a lock on the file, another process is already using --- it, so fail. The @redirection@ parameter controls what to do with --- the standard channels (@stdin@, @stderr@, and @stdout@). --- --- See: --- --- Note: All unnecessary fds should be close before calling this. --- Otherwise, you get an fd leak. -runDetached :: Maybe FilePath -- ^ pidfile - -> Redirection -- ^ redirection - -> IO () -- ^ program - -> IO () -runDetached maybePidFile redirection program = do - -- check if the pidfile exists; fail if it does - checkPidFile - -- fork first child - ignore $ forkProcess $ do - -- create a new session and make this process its leader; see - -- setsid(2) - ignore $ createSession - -- fork second child - ignore $ forkProcess $ do - -- create the pidfile - writePidFile - -- remap standard fds - remapFds - -- run the daemon - program - where - ignore act = act >> return () - - -- Remap the standard channels based on the @redirection@ - -- parameter. - remapFds = do - devnull <- openFd "/dev/null" ReadOnly Nothing defaultFileFlags - ignore (dupTo devnull stdInput) - closeFd devnull - - let file = case redirection of - DevNull -> "/dev/null" - ToFile filepath -> filepath - fd <- openFd file ReadWrite (Just stdFileMode) defaultFileFlags - hFlush stdout - mapM_ (dupTo fd) [stdOutput, stdError] - closeFd fd - - -- Convert the 'FilePath' @pidfile@ to a regular 'String' and run - -- the action with it. - withPidFile act = - case maybePidFile of - Nothing -> return () - Just pidFile -> act pidFile - - -- Check if the pidfile exists; fail if it does, and create it, otherwise - checkPidFile = withPidFile $ \pidFile -> do - running <- isRunning pidFile - when running $ fail "already running" - - writePidFile = withPidFile $ \pidFile -> do - fd <- createFile pidFile stdFileMode - setLock fd (WriteLock, AbsoluteSeek, 0, 0) - pid <- getProcessID - ignore $ fdWrite fd (show pid) - -- note that we do not close the fd; doing so would release - -- the lock - --- | Return 'True' if the given file is locked by a process. In our --- case, returns 'True' when the daemon that created the file is still --- alive. -isRunning :: FilePath -> IO Bool -isRunning pidFile = do - dfe <- doesFileExist pidFile - if dfe - then do - fd <- openFd pidFile ReadWrite Nothing defaultFileFlags - -- is there an *incompatible* lock on the pidfile? - ml <- getLock fd (WriteLock, AbsoluteSeek, 0, 0) - (pid, _) <- fdRead fd 100 - closeFd fd - case ml of - Nothing -> do - pid' <- getProcessID - return (read pid == pid') - Just _ -> do - return True - else do - return False - --- | Send 'sigQUIT' to the process recorded in the pidfile. This --- gives the process a chance to close cleanly. -kill :: FilePath -> IO () -kill = signalProcessByFilePath sigQUIT - --- | Send 'sigKILL' to the process recorded in the pidfile. This --- immediately kills the process. -brutalKill :: FilePath -> IO () -brutalKill = signalProcessByFilePath sigKILL - --- | Send a signal to a process whose pid is recorded in a file. -signalProcessByFilePath :: Signal -> FilePath -> IO () -signalProcessByFilePath signal pidFile = do - pid <- readFile pidFile - signalProcess signal (read pid) diff --git a/Test/Daemon.hs b/Test/Daemon.hs deleted file mode 100644 --- a/Test/Daemon.hs +++ /dev/null @@ -1,122 +0,0 @@ -{-# LANGUAGE OverloadedStrings, ScopedTypeVariables #-} - -module Main where - -import Control.Concurrent -import Control.Exception -import Control.Monad -import Data.Default -import Data.Monoid -import System.Directory -import System.Posix.Daemon -import System.Posix.Process - -import Test.Framework -import Test.Framework.Providers.HUnit -import Test.HUnit - -main :: IO () -main = defaultMainWithOpts - [ testCase "firstRun" testFirst - , testCase "withPid" testWithPid - , testCase "isRunning" testIsRunning - , testCase "exclusion" testExclusion - , testCase "release" testRelease - , testCase "redirection" testRedirection - ] mempty - -ensureRemoved :: [FilePath] -> IO () -ensureRemoved filepaths = forM_ filepaths $ \filepath -> do - exists <- doesFileExist filepath - when exists $ do - removeFile filepath - --- Wait the given number of ms. -sleep :: Int -> IO () -sleep n = threadDelay (n * 1000) - -testFirst :: Assertion -testFirst = flip finally (ensureRemoved ["tmp"]) $ do - let txtExp = "42" - runDetached Nothing def $ do - writeFile "tmp" txtExp - sleep 500 - txt <- readFile "tmp" - txt @?= txtExp - -testWithPid :: Assertion -testWithPid = flip finally (ensureRemoved ["pid", "tmp"]) $ do - let txtExp = "42" - runDetached (Just "pid") def $ do - pid <- getProcessID - pid' <- readFile "pid" - if show pid == pid' - then writeFile "tmp" txtExp - else writeFile "tmp" "wrong pid recorded" - sleep 500 - txt <- readFile "tmp" - txt @?= txtExp - pid <- readFile "pid" - null pid @?= False - -testIsRunning :: Assertion -testIsRunning = flip finally (ensureRemoved ["pid", "tmp"]) $ do - runDetached (Just "pid") def $ do - running <- isRunning "pid" - writeFile "tmp" (show running) - sleep 10000 - sleep 500 - --- FIXME There's some weird behaviour when the process that has locked --- the file (or its ancestors, or its descendents) use 'isRunning'. --- --- The semantics of 'fnctl' are "try to aquire the requested lock; if --- there is an incompatible lock in place, return it". Of course, --- this means that the process that acquired the lock sees it as --- unlocked. --- --- We mitigated the obvious part of the problem (same process) by --- checking the pid in the pidfile. Now, we're left with the case --- where an ancestor of the process thinks it can set the lock. - - -- running <- isRunning "pid" - -- running @?= True - txt <- readFile "tmp" - txt @?= "True" - -testExclusion :: Assertion -testExclusion = flip finally (ensureRemoved ["pid", "tmp"]) $ do - let txtExp = "ok" - runDetached (Just "pid") def $ do - sleep 1000 - sleep 500 - handle (\(_ :: SomeException) -> writeFile "tmp" txtExp) - (runDetached (Just "pid") def $ do - writeFile "tmp" "failed") - sleep 500 - txt <- readFile "tmp" - txt @?= txtExp - -testRelease :: Assertion -testRelease = flip finally (ensureRemoved ["pid", "tmp"]) $ do - let txtExp = "ok" - runDetached (Just "pid") def $ do - writeFile "tmp" txtExp - sleep 500 - txt <- readFile "tmp" - txt @?= txtExp - let txtExp' = "ok" - runDetached (Just "pid") def $ do - writeFile "tmp" txtExp' - sleep 500 - txt' <- readFile "tmp" - txt' @?= txtExp' - -testRedirection :: Assertion -testRedirection = flip finally (ensureRemoved ["tmp"]) $ do - let txtExp = "ok" - runDetached Nothing (ToFile "tmp") $ do - putStr "ok" - sleep 500 - txt <- readFile "tmp" - txt @?= txtExp diff --git a/daemons.cabal b/daemons.cabal --- a/daemons.cabal +++ b/daemons.cabal @@ -1,5 +1,5 @@ Name: daemons -Version: 0.1.1 +Version: 0.1.2 Cabal-Version: >= 1.8 License: GPL-3 License-File: LICENSE @@ -29,15 +29,17 @@ Extra-Source-Files: Makefile -Data-Files: README.md, NEWS.md, LICENSE +Data-Files: README.md, NEWS.md, LICENSE, examples/Memo.md Source-repository head Type: git Location: git://github.com/scvalex/daemons.git Library + Hs-Source-Dirs: src Build-depends: base >= 4 && < 5, bytestring, cereal, - data-default, directory, filepath, network, pipes, + data-default, directory, filepath, ghc-prim, network, + pipes >= 3.1, transformers, unix Ghc-options: -Wall Exposed-modules: Control.Pipe.C3, @@ -48,33 +50,34 @@ Other-modules: Executable memo - Build-depends: base >= 4 && < 5, bytestring, cereal, - containers, data-default, directory, filepath, - ghc-prim, pipes, transformers, network, unix - Main-Is: Memo.hs + Build-depends: base >= 4 && < 5, bytestring, cereal, containers, + daemons, data-default, ghc-prim + Main-Is: examples/Memo.hs Ghc-options: -Wall Executable addone - Build-depends: base >= 4 && < 5, bytestring, cereal, - containers, data-default, directory, filepath, - ghc-prim, pipes, transformers, network, unix - Main-Is: AddOne.hs + Build-depends: base >= 4 && < 5, daemons, data-default, ghc-prim + Main-Is: examples/AddOne.hs Ghc-options: -Wall Executable queue - Build-depends: base >= 4 && < 5, bytestring, cereal, - containers, data-default, directory, filepath, - ghc-prim, pipes, transformers, network, unix - Main-Is: PMTQ.hs + Build-depends: base >= 4 && < 5, bytestring, cereal, containers, + daemons, data-default, ghc-prim, network, + pipes >= 3.1, transformers + Main-Is: examples/Queue.hs Ghc-options: -Wall +Executable name + Build-depends: base >= 4 && < 5, bytestring, cereal, containers, + daemons, data-default, ghc-prim + Main-Is: examples/Name.hs + Ghc-options: -Wall + Test-suite daemon - Hs-Source-Dirs: Test, . + Hs-Source-Dirs: test Main-Is: Daemon.hs Type: exitcode-stdio-1.0 - - Build-Depends: base >= 4 && < 5, data-default, directory, - filepath, unix + Build-Depends: base >= 4 && < 5, daemons, data-default, directory, + ghc-prim, HUnit, test-framework, test-framework-hunit, + unix Ghc-Options: -Wall - - Build-Depends: test-framework, test-framework-hunit, HUnit diff --git a/examples/AddOne.hs b/examples/AddOne.hs new file mode 100644 --- /dev/null +++ b/examples/AddOne.hs @@ -0,0 +1,15 @@ +module Main where + +import Data.Default ( def ) +import System.Environment ( getArgs ) +import System.Daemon + +addOne :: Int -> IO Int +addOne n = return (n + 1) + +main :: IO () +main = do + ensureDaemonRunning "addOne" def addOne + [n] <- getArgs + res <- runClient "localhost" 5000 ((read n) :: Int) + print (res :: Maybe Int) diff --git a/examples/Memo.hs b/examples/Memo.hs new file mode 100644 --- /dev/null +++ b/examples/Memo.hs @@ -0,0 +1,60 @@ +{-# LANGUAGE DeriveGeneric, OverloadedStrings #-} + +module Main where + +import Control.Concurrent.MVar ( MVar, newMVar, modifyMVar ) +import Data.ByteString.Char8 ( ByteString ) +import Data.Default ( def ) +import Data.Serialize ( Serialize ) +import Data.String ( fromString ) +import qualified Data.Map as M +import GHC.Generics +import System.Environment ( getArgs ) +import System.Daemon + +data Command = Put ByteString ByteString + | Get ByteString + deriving ( Generic, Show ) + +instance Serialize Command + +data Response = Failed String + | Value ByteString + deriving ( Generic, Show ) + +instance Serialize Response + +type Book = M.Map ByteString ByteString + +handleCommand :: MVar Book -> Command -> IO Response +handleCommand bookVar comm = modifyMVar bookVar $ \book -> return $ + case comm of + Get key -> ( book + , maybe (Failed "not found") Value (M.lookup key book) ) + Put key value -> ( M.insert key value book + , Value "ok" ) + +main :: IO () +main = do + bookVar <- newMVar M.empty + let options = def { daemonPort = 7856 } + ensureDaemonRunning "memo" options (handleCommand bookVar) + args <- getArgs + let args' = map fromString args + res <- case args' of + ["get", key] -> runClient "localhost" 7856 (Get key) + ["put", key, value] -> runClient "localhost" 7856 (Put key value) + _ -> error "invalid command" + print (res :: Maybe Response) + +{- +% dist/build/memo/memo get apples +Daemon started on port 7856 +Just (Failed "not found") + +% dist/build/memo/memo put apples 23 +Just (Value "ok") + +% dist/build/memo/memo get apples +Just (Value "23") +-} diff --git a/examples/Memo.md b/examples/Memo.md new file mode 100644 --- /dev/null +++ b/examples/Memo.md @@ -0,0 +1,171 @@ +Memo +==== + +> A simple in-memory key-value store + +Welcome to the first `daemons` tutorial, in which we walk through +writing an in-memory key-value store with an RPC-like interface. The +code for this tutorial is +[Memo.hs](https://github.com/scvalex/daemons/blob/master/examples/Memo.hs). + +Concretely, we want a program such that: + + - `memo put x 42` associates the value `42` with the key `x`, and + + - `memo get x` returns the value `42`. + +First of all, the extensions and imports: + + {-# LANGUAGE DeriveGeneric, OverloadedStrings #-} + + +We need `DeriveGenerics` for +[cereal](http://hackage.haskell.org/package/cereal) to generate +serializers and deserializers automatically, and we enable +`OverloadedStrings` because it makes working with `ByteString`s much +nicer. + + module Main where + + import Control.Concurrent.MVar ( MVar, newMVar, modifyMVar ) + import Data.ByteString.Char8 ( ByteString ) + import Data.Default ( def ) + import Data.Serialize ( Serialize ) + import Data.String ( fromString ) + import qualified Data.Map as M + import GHC.Generics + import System.Environment ( getArgs ) + import System.Daemon + + +Our key-value store will be a `Map ByteString ByteString` and we'll +store it in an `MVar` to synchronize concurrent accesses. Instead of +handcrafting a binary protocol for our daemon, we take the easy road +and generate it automatically with `Data.Serialize` and +[GHC.Generics](http://www.haskell.org/ghc/docs/7.4.2/html/users_guide/generic-programming.html). + +We import `System.Daemon` which is the high-level interface to the +`daemons` library. The daemons' configuration is an instance of +[Data.Default](http://hackage.haskell.org/package/data-default), so +we'll be able to use the defaults. + + data Command = Put ByteString ByteString + | Get ByteString + deriving ( Generic, Show ) + + instance Serialize Command + + +We define a datatype for the `put ` and `get ` +commands. We let GHC derive the `Generics` instance, which gives us a +pure Haskell representation of the type; this is used by the +`Serialize` instance to generate all the necessary binary +serialization and deserialization code. + + data Response = Failed String + | Value ByteString + deriving ( Generic, Show ) + + instance Serialize Response + + +Similarly, we define a datatype for the possible responses. These can +either be values requested by `get `, or failure messages. + + type Book = M.Map ByteString ByteString + + handleCommand :: MVar Book -> Command -> IO Response + handleCommand bookVar comm = modifyMVar bookVar $ \book -> return $ + +Our "book" is just a map of `ByteString`s; our command handler takes +this map and a command, and returns a response. + +Whenever the daemon receives a command, it spawns a new thread and +runs the command handler. We want to share the book between these +concurrent calls to the handler, so we stick it in an +[MVar](http://www.haskell.org/ghc/docs/7.4.2/html/libraries/base/Control-Concurrent-MVar.html). + +An `MVar` is basically a thread-safe box which holds at most *one* +item. We use `modifyMVar` which takes the book out of the `MVar`, +runs our function with it, and puts the returned book back in the +`MVar`. + + case comm of + Get key -> ( book + , maybe (Failed "not found") Value (M.lookup key book) ) + +A `get ` command does not change the book, so we just return it. +We look up the key and return its value or a failure message. + + Put key value -> ( M.insert key value book + , Value "ok" ) + + +A `put ` command inserts the key-value pair into the +book, and returns a confirmation message. + + main :: IO () + main = do + bookVar <- newMVar M.empty + let options = def { daemonPort = 7856 } + ensureDaemonRunning "memo" options (handleCommand bookVar) + +Before doing anything else, we need to ensure that the daemon is +running: we create an empty book, customize the daemon's default +options, and finally start it. Note that `ensureDaemonRunning` checks +if the daemon is running and starts it otherwise; so, the daemon will +be started the first time the program is run, and all later runs will +use the initial daemon. + + args <- getArgs + let args' = map fromString args + +Now it's time to handle the user input. First, we convert all the +arguments to `ByteString`s for ease of use. + + res <- case args' of + ["get", key] -> runClient "localhost" 7856 (Get key) + ["put", key, value] -> runClient "localhost" 7856 (Put key value) + _ -> error "invalid command" + +Next, we parse the arguments into a command and send it to the daemon. +We call `runClient` with the port we gave earlier to +`ensureDaemonRunning` and with the parsed command. + + print (res :: Maybe Response) + + +Finally, we print the returned response. Note that `runClient` is +polymorphic in its return so we *need* to specify the type of the +response. + +Now let's see it in action: + + {- + % dist/build/memo/memo get apples + Daemon started on port 7856 + Just (Failed "not found") + + % dist/build/memo/memo put apples 23 + Just (Value "ok") + + % dist/build/memo/memo get apples + Just (Value "23") + -} + +To recap, we: + + - wrote data-types for commands and responses and gave them + `Serialize` instances, + + - wrote a handler that takes a command and returns a response, + + - ensured that our daemon is running with `ensureDaemonRunning`, and + + - sent commands and received responses with `runClient`. + +This tutorial illustrates the basic concepts behind `daemons`, but +hides a powerful feature: the interface is *streaming*. See the +[Queue](https://github.com/scvalex/daemons/blob/master/examples/Queue.hs) +(Poor Man's Task Queue) tutorial for an example use of the streaming +interface. diff --git a/examples/Name.hs b/examples/Name.hs new file mode 100644 --- /dev/null +++ b/examples/Name.hs @@ -0,0 +1,83 @@ +{-# LANGUAGE DeriveGeneric, OverloadedStrings, MultiParamTypeClasses, FunctionalDependencies #-} + +module Main where + +import Control.Concurrent.MVar ( MVar, newMVar, modifyMVar ) +import Data.ByteString.Char8 ( ByteString, unpack ) +import Data.Default ( def ) +import Data.Serialize ( Serialize ) +import Data.String ( fromString ) +import qualified Data.Map as M +import GHC.Generics +import System.Environment ( getArgs ) +import System.Daemon + +data CommandV0 = Register ByteString Port + | WhereIs ByteString + deriving ( Generic, Show ) + +instance Serialize CommandV0 + +data Command = CommandV0 CommandV0 + deriving ( Generic, Show ) + +instance Serialize Command + +data ResponseV0 = Ok + | NotFound ByteString + | AtPort ByteString Port + deriving ( Generic, Show ) + +instance Serialize ResponseV0 + +data Response = ResponseV0 ResponseV0 + deriving ( Generic, Show ) + +instance Serialize Response + +type Registry = M.Map ByteString Port + +class VersionOf a b | a -> b where + toLatest :: a -> b + fromLatest :: b -> a + +instance VersionOf Command CommandV0 where + toLatest (CommandV0 v0) = v0 + fromLatest v0 = CommandV0 v0 + +instance VersionOf Response ResponseV0 where + toLatest (ResponseV0 v0) = v0 + fromLatest v0 = ResponseV0 v0 + +namePort :: Port +namePort = 4370 + +handleCommand :: MVar Registry -> CommandV0 -> IO ResponseV0 +handleCommand registryVar cmd = modifyMVar registryVar $ \registry -> return $ + case cmd of + WhereIs name -> ( registry + , maybe (NotFound name) (AtPort name) (M.lookup name registry) ) + Register name port -> ( M.insert name port registry + , Ok ) + +wrapVersion :: (CommandV0 -> IO ResponseV0) -> Command -> IO Response +wrapVersion f cmd = do + rsp <- f (toLatest cmd) + return (fromLatest rsp) + +main :: IO () +main = do + registryVar <- newMVar M.empty + let options = def { daemonPort = namePort } + ensureDaemonRunning "name" options (wrapVersion (handleCommand registryVar)) + args <- getArgs + let args' = map fromString args + res <- case args' of + ["where-is", key] -> + runClient "localhost" namePort (CommandV0 (WhereIs key)) + ["register", name, port] -> + let portNum = read (unpack port) in + runClient "localhost" namePort (CommandV0 (Register name portNum)) + _ -> + error "invalid command" + print (fmap toLatest (res :: Maybe Response)) diff --git a/examples/Queue.hs b/examples/Queue.hs new file mode 100644 --- /dev/null +++ b/examples/Queue.hs @@ -0,0 +1,98 @@ +{-# LANGUAGE DeriveGeneric, OverloadedStrings #-} + +module Main where + +import Control.Concurrent.Chan ( Chan, newChan, readChan, writeChan ) +import Control.Concurrent.MVar ( MVar, newMVar, modifyMVar ) +import Control.Monad ( forever ) +import Control.Monad.Trans.Class ( lift ) +import Control.Pipe ( runPipe, (<+<), await, yield ) +import Control.Pipe.Serialize ( serializer, deserializer ) +import Control.Pipe.Socket ( Handler ) +import Data.ByteString.Char8 ( ByteString ) +import qualified Data.ByteString.Char8 as B +import Data.Char ( toLower ) +import Data.Default ( def ) +import Data.Serialize ( Serialize ) +import Data.String ( fromString ) +import qualified Data.Map as M +import GHC.Generics +import Network.Socket ( withSocketsDo ) +import System.Environment ( getArgs ) +import System.Daemon +import System.IO ( hPutStrLn, stderr ) + +data Command = Push ByteString ByteString + | Pop ByteString + | Consume ByteString + deriving ( Generic, Show ) + +instance Serialize Command + +data Response = Value ByteString + deriving ( Generic, Show ) + +instance Serialize Response + +type Registry = M.Map ByteString (Chan ByteString) + +handleCommands :: MVar Registry -> Handler () +handleCommands registryVar reader writer = do + runPipe (writer <+< serializer + <+< commandExecuter + <+< deserializer <+< reader) + where + commandExecuter = forever $ do + comm <- await + case comm of + Pop topic -> do + ch <- lift $ getCreateChan topic + transferToPipeFromChan ch + Consume topic -> do + ch <- lift $ getCreateChan topic + forever $ transferToPipeFromChan ch + Push topic val -> do + ch <- lift $ getCreateChan topic + lift $ writeChan ch val + yield (Value "ok") + + -- Transfer a value from the given channel to the pipe. + transferToPipeFromChan ch = do + val <- lift $ readChan ch + yield (Value val) + + -- Get the channel for the given topic, and create it if it does + -- not already exist. + getCreateChan topic = modifyMVar registryVar $ \registry -> do + case M.lookup topic registry of + Nothing -> do + ch <- newChan + return (M.insert topic ch registry, ch) + Just ch -> do + return (registry, ch) + +main :: IO () +main = withSocketsDo $ do + registryVar <- newMVar M.empty + let options = def { daemonPort = 7857 } + ensureDaemonWithHandlerRunning "queue" options (handleCommands registryVar) + args <- getArgs + let args' = map (fromString . map toLower) args + case args' of + ["pop", key] -> do + res <- runClient "localhost" 7857 (Pop key) + printResult res + ["push", key, value] -> do + res <- runClient "localhost" 7857 (Push key value) + printResult res + ["consume", key] -> do + runClientWithHandler "localhost" 7857 $ \reader writer -> do + runPipe (writer <+< serializer <+< yield (Consume key)) + runPipe ((forever $ await >>= \res -> lift (printResult (Just res))) + <+< deserializer <+< reader) + _ -> do + error "invalid command" + where + printResult :: Maybe Response -> IO () + printResult Nothing = hPutStrLn stderr "no response" + printResult (Just (Value val)) = B.putStrLn val diff --git a/src/Control/Pipe/C3.hs b/src/Control/Pipe/C3.hs new file mode 100644 --- /dev/null +++ b/src/Control/Pipe/C3.hs @@ -0,0 +1,40 @@ +module Control.Pipe.C3 ( + -- * Pipes + commandSender, commandReceiver + ) where + +import Control.Monad ( forever ) +import Control.Monad.Trans.Class ( lift ) +import Control.Pipe ( runPipe, await, yield, (<+<) ) +import Control.Pipe.Serialize ( serializer, deserializer ) +import Control.Pipe.Socket ( Handler ) +import Data.Serialize ( Serialize ) + +-- | Send a single command over the outgoing pipe and wait for a +-- response. If the incoming pipe is closed before a response +-- arrives, returns @Nothing@. +commandSender :: (Serialize a, Serialize b) => a -> Handler (Maybe b) +commandSender command reader writer = do + runPipe (writer <+< serializer <+< sendCommand) + runPipe (receiveResponse + <+< (deserializer >> return Nothing) + <+< (reader >> return Nothing)) + where + sendCommand = do + yield command + + receiveResponse = do + res <- await + return (Just res) + +-- | Wait for commands on the incoming pipe, handle them, and send the +-- reponses over the outgoing pipe. +commandReceiver :: (Serialize a, Serialize b) => (a -> IO b) -> Handler () +commandReceiver executeCommand reader writer = do + runPipe (writer <+< serializer + <+< commandExecuter + <+< deserializer <+< reader) + where + commandExecuter = forever $ do + comm <- await + yield =<< lift (executeCommand comm) diff --git a/src/Control/Pipe/Serialize.hs b/src/Control/Pipe/Serialize.hs new file mode 100644 --- /dev/null +++ b/src/Control/Pipe/Serialize.hs @@ -0,0 +1,48 @@ +-- | This module provides the 'deserializer' and 'serializer' pipes to +-- convert 'B.ByteString's off of pipes into typed values. +-- +-- In order to use it, the types of the values need to have +-- 'Serialize' instances. These can be derived automatically using +-- "Ghc.Generics": +-- +-- > {-# LANGUAGE DeriveGeneric #-} +-- > +-- > data Foo = Bar String | Baz Int +-- > deriving ( Generic ) +-- > +-- > instance Serialize Foo +-- +-- Note that in the above example: we use the @DeriveGeneric@ +-- extension, derive a @Generic@ instance for our data-type, and write +-- an /empty/ @Serialize@ instance. +-- +module Control.Pipe.Serialize ( + -- * Pipes + serializer, deserializer + ) where + +import Data.ByteString.Char8 ( ByteString ) +import Data.Serialize ( Serialize, get, encode + , Result(..), runGetPartial ) +import Control.Pipe ( Pipe, await, yield ) +import Control.Monad ( forever ) + +-- | De-serialize data from strict 'ByteString's. Uses @cereal@'s +-- incremental 'Data.Serialize.Get' parser. +deserializer :: (Serialize a, Monad m) => Pipe ByteString a m () +deserializer = loop Nothing Nothing + where + loop mk mbin = do + bin <- maybe await return mbin + case (maybe (runGetPartial get) id mk) bin of + Fail reason -> fail reason + Partial k -> loop (Just k) Nothing + Done c bin' -> do + yield c + loop Nothing (Just bin') + +-- | Serialize data into strict 'ByteString's. +serializer :: (Serialize a, Monad m) => Pipe a ByteString m () +serializer = forever $ do + x <- await + yield (encode x) diff --git a/src/Control/Pipe/Socket.hs b/src/Control/Pipe/Socket.hs new file mode 100644 --- /dev/null +++ b/src/Control/Pipe/Socket.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +-- Thank you: +-- pipes-network + +module Control.Pipe.Socket ( + -- * Socket pipes + socketReader, socketWriter, + + -- * Socket server/client + Handler, runSocketServer, runSocketClient + ) where + +import Control.Concurrent ( forkIO ) +import qualified Control.Exception as CE +import Control.Monad ( forever, unless ) +import Control.Monad.IO.Class ( MonadIO(..) ) +import Control.Monad.Trans.Class ( lift ) +import Control.Pipe ( Consumer, Producer, await, yield ) +import Data.ByteString.Char8 ( ByteString ) +import qualified Data.ByteString.Char8 as B +import Network.Socket ( Socket ) +import qualified Network.Socket as NS +import Network.Socket.ByteString ( sendAll, recv ) + +-- | Stream data from the socket. +socketReader :: (MonadIO m) => Socket -> Producer ByteString m () +socketReader socket = do + bin <- lift . liftIO $ recv socket 4096 + unless (B.null bin) $ do + yield bin + socketReader socket + +-- | Stream data to the socket. +socketWriter :: (MonadIO m) => Socket -> Consumer ByteString m () +socketWriter socket = forever $ do + bin <- await + lift . liftIO $ sendAll socket bin + +-- | A simple handler: takes an incoming stream of 'ByteString's, an +-- stream of 'ByteString's, and ties them together somehow. +-- Conceptually, the simplest handler would be @identity@: +-- +-- > import Control.Monad +-- > import Control.Pipe +-- > import Data.ByteString.Char8 +-- > +-- > handler reader writer = do +-- > let identity = forever $ do +-- > x <- await +-- > yield x +-- > runPipe (writer <+< identity <+< reader) +-- +-- See the @pipes@ tutorial for more examples of writing pipes. +-- +-- Since 'ByteString's are fairly boring by themseleves, have a look +-- at "Control.Pipe.Serialize" which lets you deserialize/serialize +-- pipes of 'ByteString's easily. +type Handler r = Producer ByteString IO () + -> Consumer ByteString IO () + -> IO r + +-- | Listen for connections on the given socket, and run 'Handler' on +-- each received connection. The socket should previously have been +-- bound to a port or to a file. Each handler is run in its own +-- thread. Even in case of an error, the handlers' sockets are +-- closed. +runSocketServer :: (MonadIO m) => Socket -> Handler () -> m () +runSocketServer lsocket handler = liftIO $ forever $ do + (socket, _addr) <- NS.accept lsocket + _ <- forkIO $ CE.finally + (handler (socketReader socket) (socketWriter socket)) + (NS.sClose socket) + return () + +-- | Run 'Handler' on the given socket. +runSocketClient :: (MonadIO m) => Socket -> Handler r -> m r +runSocketClient socket handler = liftIO $ do + handler (socketReader socket) (socketWriter socket) diff --git a/src/System/Daemon.hs b/src/System/Daemon.hs new file mode 100644 --- /dev/null +++ b/src/System/Daemon.hs @@ -0,0 +1,189 @@ +-- | An RPC-like interface for daemons is provided by +-- 'ensureDaemonRunning' and 'runClient'. +-- +-- A more versatile interface that lets you supply your own `Handler` +-- is provided by `ensureDaemonWithHandlerRunning` and +-- `runClientWithHandler`. These are useful if, for instance, you +-- need streaming requests or replies, or if you need to change your +-- event handler at runtime. +-- +-- The event handling loop is provided by `runInForeground`. You may +-- want to use this for debugging purposes or if you want to handle +-- daemonization manually. +module System.Daemon ( + -- * Daemons + ensureDaemonRunning, ensureDaemonWithHandlerRunning, + + -- * Clients, + runClient, runClientWithHandler, + + -- * Types + DaemonOptions(..), PidFile(..), HostName, Port, + + -- * Helpers + runInForeground, bindPort, getSocket + ) where + +import Control.Concurrent ( threadDelay ) +import qualified Control.Exception as CE +import Control.Monad ( when ) +import Control.Pipe.C3 ( commandSender, commandReceiver ) +import Control.Pipe.Socket ( Handler, runSocketServer, runSocketClient ) +import Data.Default ( Default(..) ) +import Data.Serialize ( Serialize ) +import Data.String ( IsString(..) ) +import Network.Socket ( Socket, SockAddr(..), Family(..), SocketType(..) + , SocketOption(..), setSocketOption + , socket, sClose, connect, bindSocket, listen + , AddrInfo(..), getAddrInfo, addrAddress, defaultHints + , defaultProtocol, iNADDR_ANY, maxListenQueue ) +import System.Directory ( getHomeDirectory ) +import System.FilePath ( (), (<.>) ) +import System.Posix.Daemon ( runDetached, isRunning ) +import Text.Printf ( printf ) + +type Port = Int +type HostName = String + +-- | The configuration options of a daemon. See 'ensureDaemonRunning' +-- for a description of each. +data DaemonOptions = DaemonOptions + { daemonPort :: Port + , daemonPidFile :: PidFile + , printOnDaemonStarted :: Bool + } deriving ( Show ) + +instance Default DaemonOptions where + def = DaemonOptions { daemonPort = 5000 + , daemonPidFile = InHome + , printOnDaemonStarted = True + } + +-- | The location of the daemon's pidfile. +data PidFile = InHome + | PidFile FilePath + deriving ( Show ) + +instance IsString PidFile where + fromString = PidFile + +-- | Simple wrapper around 'ensureDaemonWithHandlerRunning' which uses +-- a simple function to respond to commands and doesn't deal with +-- pipes. +-- +-- The @handler@ is just a function that takes a command and returns a +-- response. +ensureDaemonRunning :: (Serialize a, Serialize b) + => String -- ^ name + -> DaemonOptions -- ^ options + -> (a -> IO b) -- ^ handler + -> IO () +ensureDaemonRunning name options executeCommand = do + ensureDaemonWithHandlerRunning name options (commandReceiver executeCommand) + +-- FIXME Add set-up and tear-down action. The reason the threaded +-- runtime wouldn't work was because we were creating the mvar in a +-- different thread! + +-- | Start a daemon running on the given port, using the given handler +-- to respond to events. If the daemon is already running, don't do +-- anything. Returns immediately. +-- +-- The pidfile @PidFile options@ will be created and locked. This +-- function checks the pidfile to see if the daemon is already +-- running. +-- +-- The daemon will listen for incoming connections on all interfaces +-- on @daemonPort options@. +-- +-- The @handler@ is a function that takes the reader and writer +-- 'ByteString' pipes and does something with them. See +-- 'commandReceiver' for an example handler. +ensureDaemonWithHandlerRunning :: String -- ^ name + -> DaemonOptions -- ^ options + -> Handler () -- ^ handler + -> IO () +ensureDaemonWithHandlerRunning name options handler = do + home <- getHomeDirectory + let pidfile = case daemonPidFile options of + InHome -> home ("." ++ name) <.> "pid" + PidFile path -> path + running <- isRunning pidfile + when (not running) $ do + runDetached (Just pidfile) def + (runInForeground (daemonPort options) handler) + when (printOnDaemonStarted options) + (printf "Daemon started on port %d\n" (daemonPort options)) + threadDelay (1 * 1000 * 1000) -- 1s delay + +-- | Start the given handler in the foreground. It will listen and +-- respond to events on the given port. +-- +-- This is the function that 'ensureDaemonWithHandlerRunning' runs on +-- the daemon thread. +runInForeground :: Port -> Handler () -> IO () +runInForeground port handler = do + CE.bracket + (bindPort port) + sClose + (\lsocket -> + runSocketServer lsocket handler) + +-- | Send a command to the daemon running at the given network address +-- and wait for a response. +-- +-- This is a simple wrapper around 'runClientWithHandler' that sends a +-- single command and waits for a single response. +-- +-- If the connection is closed before receiving a response, return +-- 'Nothing'. +runClient :: (Serialize a, Serialize b) + => HostName -- ^ hostname + -> Port -- ^ port + -> a -- ^ command + -> IO (Maybe b) +runClient hostname port comm = + runClientWithHandler hostname port (commandSender comm) + +-- | Connect to the given network address and run the handler on the +-- reader and wrier pipes for the socket. +-- +-- The @handler@ is a function that takes the reader and writer +-- 'ByteString' pipes and does something with them. For an example +-- handler, see 'commandSender', which sends a command and waits for a +-- response. +runClientWithHandler :: HostName -- ^ hostname + -> Port -- ^ port + -> Handler a -- ^ command + -> IO a +runClientWithHandler hostname port handler = do + CE.bracket + (getSocket hostname port) + sClose + (\s -> runSocketClient s handler) + +-- | Create a socket and bind it to the given port. +bindPort :: Port -> IO Socket +bindPort port = do + CE.bracketOnError + (socket AF_INET Stream defaultProtocol) + sClose + (\s -> do + -- FIXME See the examples at the end of Network.Socket.ByteString + setSocketOption s ReuseAddr 1 + bindSocket s (SockAddrInet (fromIntegral port) iNADDR_ANY) + listen s maxListenQueue + return s) + +-- | Create a socket connected to the given network address. +getSocket :: HostName -> Port -> IO Socket +getSocket hostname port = do + addrInfos <- getAddrInfo (Just (defaultHints { addrFamily = AF_INET })) + (Just hostname) + (Just $ show port) + CE.bracketOnError + (socket AF_INET Stream defaultProtocol) + sClose + (\s -> do + connect s (addrAddress $ head addrInfos) + return s) diff --git a/src/System/Posix/Daemon.hs b/src/System/Posix/Daemon.hs new file mode 100644 --- /dev/null +++ b/src/System/Posix/Daemon.hs @@ -0,0 +1,204 @@ +-- | This module provides a simple interface to creating, checking the +-- status of, and stopping background jobs. +-- +-- Use 'runDetached' to start a background job. For instance, here is +-- a daemon that peridically hits a webserver: +-- +-- > import Control.Concurrent +-- > import Control.Monad +-- > import Data.Default +-- > import Data.Maybe +-- > import Network.BSD +-- > import Network.HTTP +-- > import Network.URI +-- > import System.Posix.Daemon +-- > +-- > main :: IO () +-- > main = runDetached (Just "diydns.pid") def $ forever $ do +-- > hostname <- getHostName +-- > _ <- simpleHTTP +-- > (Request { rqURI = fromJust (parseURI "http://foo.com/dns") +-- > , rqMethod = GET +-- > , rqHeaders = [] +-- > , rqBody = hostname }) +-- > threadDelay (600 * 1000 * 1000) +-- +-- To check if the above job is running, use 'isRunning' with the same +-- pidfile: +-- +-- > isRunning "diydns.pid" +-- +-- Finally, to stop the above job (maybe because we're rolling a new +-- version of it), use 'kill': +-- +-- > kill "diydns.pid" +-- +-- To stop a job and wait for it to close (release its pidfile), +-- such as when restarting it), use killAndWait: +-- +-- > killAndWait "diydns.pid" >> doSomething +-- +-- As a side note, the code above is a script that the author uses as +-- a sort of homebrew dynamic DNS: the remote address is a CGI script +-- that records the IP addresses of all incoming requests in separate +-- files named after the contents of the requests; the addresses are +-- then viewable with any browser. +module System.Posix.Daemon ( + -- * Starting + runDetached, Redirection(..), + + -- * Status + isRunning, + + -- * Stopping + kill, killAndWait, brutalKill + ) where + +import Prelude hiding ( FilePath ) + +import Control.Monad ( when ) +import Data.Default ( Default(..) ) +import System.Directory ( doesFileExist ) +import System.FilePath ( FilePath ) +import System.IO ( SeekMode(..), hFlush, stdout ) +import System.Posix.Files ( stdFileMode ) +import System.Posix.IO ( openFd, OpenMode(..), defaultFileFlags, closeFd + , dupTo, stdInput, stdOutput, stdError, getLock + , createFile, fdWrite, fdRead + , LockRequest (..), setLock, waitToSetLock ) +import System.Posix.Process ( getProcessID, forkProcess, createSession ) +import System.Posix.Signals ( Signal, signalProcess, sigQUIT, sigKILL ) + +-- | Where should the output (and input) of a daemon be redirected to? +-- (we can't just leave it to the current terminal, because it may be +-- closed, and that would kill the daemon). +-- +-- When in doubt, just use 'def', the default value. +-- +-- 'DevNull' causes the output to be redirected to @\/dev\/null@. This +-- is safe and is what you want in most cases. +-- +-- If you don't want to lose the output (maybe because you're using it +-- for logging), use 'ToFile', instead. +data Redirection = DevNull + | ToFile FilePath + deriving ( Show ) + +instance Default Redirection where + def = DevNull + +-- | Run the given action detached from the current terminal; this +-- creates an entirely new process. This function returns +-- immediately. Uses the double-fork technique to create a well +-- behaved daemon. If @pidfile@ is given, check/write it; if we +-- cannot obtain a lock on the file, another process is already using +-- it, so fail. The @redirection@ parameter controls what to do with +-- the standard channels (@stdin@, @stderr@, and @stdout@). +-- +-- See: +-- +-- Note: All unnecessary fds should be close before calling this. +-- Otherwise, you get an fd leak. +runDetached :: Maybe FilePath -- ^ pidfile + -> Redirection -- ^ redirection + -> IO () -- ^ program + -> IO () +runDetached maybePidFile redirection program = do + -- check if the pidfile exists; fail if it does + checkPidFile + -- fork first child + ignore $ forkProcess $ do + -- create a new session and make this process its leader; see + -- setsid(2) + ignore $ createSession + -- fork second child + ignore $ forkProcess $ do + -- create the pidfile + writePidFile + -- remap standard fds + remapFds + -- run the daemon + program + where + ignore act = act >> return () + + -- Remap the standard channels based on the @redirection@ + -- parameter. + remapFds = do + devnull <- openFd "/dev/null" ReadOnly Nothing defaultFileFlags + ignore (dupTo devnull stdInput) + closeFd devnull + + let file = case redirection of + DevNull -> "/dev/null" + ToFile filepath -> filepath + fd <- openFd file ReadWrite (Just stdFileMode) defaultFileFlags + hFlush stdout + mapM_ (dupTo fd) [stdOutput, stdError] + closeFd fd + + -- Convert the 'FilePath' @pidfile@ to a regular 'String' and run + -- the action with it. + withPidFile act = + case maybePidFile of + Nothing -> return () + Just pidFile -> act pidFile + + -- Check if the pidfile exists; fail if it does, and create it, otherwise + checkPidFile = withPidFile $ \pidFile -> do + running <- isRunning pidFile + when running $ fail "already running" + + writePidFile = withPidFile $ \pidFile -> do + fd <- createFile pidFile stdFileMode + setLock fd (WriteLock, AbsoluteSeek, 0, 0) + pid <- getProcessID + ignore $ fdWrite fd (show pid) + -- note that we do not close the fd; doing so would release + -- the lock + +-- | Return 'True' if the given file is locked by a process. In our +-- case, returns 'True' when the daemon that created the file is still +-- alive. +isRunning :: FilePath -> IO Bool +isRunning pidFile = do + dfe <- doesFileExist pidFile + if dfe + then do + fd <- openFd pidFile ReadWrite Nothing defaultFileFlags + -- is there an *incompatible* lock on the pidfile? + ml <- getLock fd (WriteLock, AbsoluteSeek, 0, 0) + (pid, _) <- fdRead fd 100 + closeFd fd + case ml of + Nothing -> do + pid' <- getProcessID + return (read pid == pid') + Just _ -> do + return True + else do + return False + +-- | Send 'sigQUIT' to the process recorded in the pidfile. This +-- gives the process a chance to close cleanly. +kill :: FilePath -> IO () +kill = signalProcessByFilePath sigQUIT + +-- | Kill a process and wait for it to release its pidfile +killAndWait :: FilePath -> IO () +killAndWait pidFile = do + signalProcessByFilePath sigQUIT pidFile + fd <- openFd pidFile ReadWrite Nothing defaultFileFlags + waitToSetLock fd (WriteLock, AbsoluteSeek, 0, 0) + closeFd fd + +-- | Send 'sigKILL' to the process recorded in the pidfile. This +-- immediately kills the process. +brutalKill :: FilePath -> IO () +brutalKill = signalProcessByFilePath sigKILL + +-- | Send a signal to a process whose pid is recorded in a file. +signalProcessByFilePath :: Signal -> FilePath -> IO () +signalProcessByFilePath signal pidFile = do + pid <- readFile pidFile + signalProcess signal (read pid) diff --git a/test/Daemon.hs b/test/Daemon.hs new file mode 100644 --- /dev/null +++ b/test/Daemon.hs @@ -0,0 +1,122 @@ +{-# LANGUAGE OverloadedStrings, ScopedTypeVariables #-} + +module Main where + +import Control.Concurrent +import Control.Exception +import Control.Monad +import Data.Default +import Data.Monoid +import System.Directory +import System.Posix.Daemon +import System.Posix.Process + +import Test.Framework +import Test.Framework.Providers.HUnit +import Test.HUnit + +main :: IO () +main = defaultMainWithOpts + [ testCase "firstRun" testFirst + , testCase "withPid" testWithPid + , testCase "isRunning" testIsRunning + , testCase "exclusion" testExclusion + , testCase "release" testRelease + , testCase "redirection" testRedirection + ] mempty + +ensureRemoved :: [FilePath] -> IO () +ensureRemoved filepaths = forM_ filepaths $ \filepath -> do + exists <- doesFileExist filepath + when exists $ do + removeFile filepath + +-- Wait the given number of ms. +sleep :: Int -> IO () +sleep n = threadDelay (n * 1000) + +testFirst :: Assertion +testFirst = flip finally (ensureRemoved ["tmp"]) $ do + let txtExp = "42" + runDetached Nothing def $ do + writeFile "tmp" txtExp + sleep 500 + txt <- readFile "tmp" + txt @?= txtExp + +testWithPid :: Assertion +testWithPid = flip finally (ensureRemoved ["pid", "tmp"]) $ do + let txtExp = "42" + runDetached (Just "pid") def $ do + pid <- getProcessID + pid' <- readFile "pid" + if show pid == pid' + then writeFile "tmp" txtExp + else writeFile "tmp" "wrong pid recorded" + sleep 500 + txt <- readFile "tmp" + txt @?= txtExp + pid <- readFile "pid" + null pid @?= False + +testIsRunning :: Assertion +testIsRunning = flip finally (ensureRemoved ["pid", "tmp"]) $ do + runDetached (Just "pid") def $ do + running <- isRunning "pid" + writeFile "tmp" (show running) + sleep 10000 + sleep 500 + +-- FIXME There's some weird behaviour when the process that has locked +-- the file (or its ancestors, or its descendents) use 'isRunning'. +-- +-- The semantics of 'fnctl' are "try to aquire the requested lock; if +-- there is an incompatible lock in place, return it". Of course, +-- this means that the process that acquired the lock sees it as +-- unlocked. +-- +-- We mitigated the obvious part of the problem (same process) by +-- checking the pid in the pidfile. Now, we're left with the case +-- where an ancestor of the process thinks it can set the lock. + + -- running <- isRunning "pid" + -- running @?= True + txt <- readFile "tmp" + txt @?= "True" + +testExclusion :: Assertion +testExclusion = flip finally (ensureRemoved ["pid", "tmp"]) $ do + let txtExp = "ok" + runDetached (Just "pid") def $ do + sleep 1000 + sleep 500 + handle (\(_ :: SomeException) -> writeFile "tmp" txtExp) + (runDetached (Just "pid") def $ do + writeFile "tmp" "failed") + sleep 500 + txt <- readFile "tmp" + txt @?= txtExp + +testRelease :: Assertion +testRelease = flip finally (ensureRemoved ["pid", "tmp"]) $ do + let txtExp = "ok" + runDetached (Just "pid") def $ do + writeFile "tmp" txtExp + sleep 500 + txt <- readFile "tmp" + txt @?= txtExp + let txtExp' = "ok" + runDetached (Just "pid") def $ do + writeFile "tmp" txtExp' + sleep 500 + txt' <- readFile "tmp" + txt' @?= txtExp' + +testRedirection :: Assertion +testRedirection = flip finally (ensureRemoved ["tmp"]) $ do + let txtExp = "ok" + runDetached Nothing (ToFile "tmp") $ do + putStr "ok" + sleep 500 + txt <- readFile "tmp" + txt @?= txtExp