daemons 0.1.1 → 0.1.2
raw patch · 24 files changed
+1188/−851 lines, 24 filesdep +daemonsdep ~pipesnew-component:exe:namePVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: daemons
Dependency ranges changed: pipes
API changes (from Hackage documentation)
- System.Daemon: startDaemon :: (Serialize a, Serialize b) => String -> DaemonOptions -> (a -> IO b) -> IO ()
- System.Daemon: startDaemonWithHandler :: String -> DaemonOptions -> Handler () -> IO ()
+ System.Daemon: ensureDaemonRunning :: (Serialize a, Serialize b) => String -> DaemonOptions -> (a -> IO b) -> IO ()
+ System.Daemon: ensureDaemonWithHandlerRunning :: String -> DaemonOptions -> Handler () -> IO ()
+ System.Daemon: printOnDaemonStarted :: DaemonOptions -> Bool
+ System.Daemon: runInForeground :: Port -> Handler () -> IO ()
+ System.Posix.Daemon: killAndWait :: FilePath -> IO ()
- System.Daemon: DaemonOptions :: Port -> PidFile -> DaemonOptions
+ System.Daemon: DaemonOptions :: Port -> PidFile -> Bool -> DaemonOptions
Files
- AddOne.hs +0/−15
- Control/Pipe/C3.hs +0/−40
- Control/Pipe/Serialize.hs +0/−48
- Control/Pipe/Socket.hs +0/−79
- Makefile +21/−10
- Memo.hs +0/−59
- NEWS.md +5/−0
- PMTQ.hs +0/−98
- README.md +29/−10
- System/Daemon.hs +0/−158
- System/Posix/Daemon.hs +0/−191
- Test/Daemon.hs +0/−122
- daemons.cabal +24/−21
- examples/AddOne.hs +15/−0
- examples/Memo.hs +60/−0
- examples/Memo.md +171/−0
- examples/Name.hs +83/−0
- examples/Queue.hs +98/−0
- src/Control/Pipe/C3.hs +40/−0
- src/Control/Pipe/Serialize.hs +48/−0
- src/Control/Pipe/Socket.hs +79/−0
- src/System/Daemon.hs +189/−0
- src/System/Posix/Daemon.hs +204/−0
- test/Daemon.hs +122/−0
− AddOne.hs
@@ -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)
− Control/Pipe/C3.hs
@@ -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)
− Control/Pipe/Serialize.hs
@@ -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)
− Control/Pipe/Socket.hs
@@ -1,79 +0,0 @@-{-# LANGUAGE ScopedTypeVariables #-}---- Thank you:--- <a href="https://github.com/pcapriotti/pipes-network">pipes-network</a>--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)
Makefile view
@@ -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
− Memo.hs
@@ -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")--}
NEWS.md view
@@ -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) -------------------
− PMTQ.hs
@@ -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
README.md view
@@ -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 ------------
− System/Daemon.hs
@@ -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)
− System/Posix/Daemon.hs
@@ -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: <http://www.enderunix.org/docs/eng/daemon.php>------ 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)
− Test/Daemon.hs
@@ -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
daemons.cabal view
@@ -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
+ examples/AddOne.hs view
@@ -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)
+ examples/Memo.hs view
@@ -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")+-}
+ examples/Memo.md view
@@ -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 <key> <value>` and `get <key>`+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 <key>`, 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 <key>` 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 <key> <value>` 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.
+ examples/Name.hs view
@@ -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))
+ examples/Queue.hs view
@@ -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
+ src/Control/Pipe/C3.hs view
@@ -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)
+ src/Control/Pipe/Serialize.hs view
@@ -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)
+ src/Control/Pipe/Socket.hs view
@@ -0,0 +1,79 @@+{-# LANGUAGE ScopedTypeVariables #-}++-- Thank you:+-- <a href="https://github.com/pcapriotti/pipes-network">pipes-network</a>++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)
+ src/System/Daemon.hs view
@@ -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)
+ src/System/Posix/Daemon.hs view
@@ -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: <http://www.enderunix.org/docs/eng/daemon.php>+--+-- 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)
+ test/Daemon.hs view
@@ -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