diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2016 Lars Petersen + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/MissileLauncher.hs b/MissileLauncher.hs new file mode 100644 --- /dev/null +++ b/MissileLauncher.hs @@ -0,0 +1,66 @@ +module MissileLauncher where + +import Control.Exception +import Control.Monad +import Control.Concurrent +import Control.Concurrent.Async +import Control.Concurrent.RPC +import Data.Word +import System.Random + +type Missile = Word +type LaunchSite = String + +main :: IO () +main = do + (launchMissile, withMissile) <- newRPC + runMissileProduction launchMissile + `race_` runLaunchSite withMissile "Redmond" + `race_` runLaunchSite withMissile "Cambridge" + +runLaunchSite :: WithRPC Missile LaunchSite -> LaunchSite -> IO () +runLaunchSite withMissile site = forever $ do + sleepRandom + catch + ( withMissile $ \missile-> do + r <- random100 + if r < 10 + then error $ "bad weather in " ++ site + else do + printThread $ site ++ ": LAUNCH THE MISSILE!" + return site + ) + ( \e-> do + let _ = e :: SomeException + printThread $ site ++ ": Couldn't launch. Waiting for next missile." + ) + +runMissileProduction :: RPC Missile LaunchSite -> IO () +runMissileProduction launchMissile = + produce `race_` produce `race_` produce `race_` produce + where + produce = forever $ do + sleepRandom + missile <- randomIO :: IO Missile + catch + ( do + printThread $ "Production: Ready to launch missile " ++ show missile + site <- launchMissile missile + printThread $ "Production: Missile " ++ show missile ++ " launched in " ++ site + ) + ( \e-> + printThread $ "Production: Missile " ++ show missile ++ + " failed to launch due to " ++ show (e :: SomeException) + ) + +printThread :: Show a => a -> IO () +printThread x = do + threadId <- myThreadId + random100 >>= \x-> threadDelay (x * 100) + putStrLn $ show threadId ++ ": " ++ show x + +random100 :: IO Int +random100 = (`mod` 100) <$> randomIO + +sleepRandom :: IO () +sleepRandom = random100 >>= \x-> threadDelay (x * 100000) diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,95 @@ +concurrent-rpc +============== + +[![Available on Hackage][badge-hackage]][hackage] +[![License MIT][badge-license]][license] +[![Build Status][badge-travis]][travis] + +### Summary + +This library is small wrapper around `Control.Concurrent.MVar.MVar`s that can +be used to implement request-response communication between different threads. + +### Example + +```haskell +module MissileLauncher where + +import Control.Exception +import Control.Monad +import Control.Concurrent +import Control.Concurrent.Async +import Control.Concurrent.RPC +import Data.Word +import System.Random + +type Missile = Word +type LaunchSite = String + +main :: IO () +main = do + (launchMissile, withMissile) <- newRPC + runMissileProduction launchMissile + `race_` runLaunchSite withMissile "Redmond" + `race_` runLaunchSite withMissile "Cambridge" + +runLaunchSite :: WithRPC Missile LaunchSite -> LaunchSite -> IO () +runLaunchSite withMissile site = forever $ do + sleepRandom + catch + ( withMissile $ \missile-> do + r <- random100 + if r < 10 + then error $ "bad weather in " ++ site + else do + printThread $ site ++ ": LAUNCH THE MISSILE!" + return site + ) + ( \e-> do + let _ = e :: SomeException + printThread $ site ++ ": Couldn't launch. Waiting for next missile." + ) + +runMissileProduction :: RPC Missile LaunchSite -> IO () +runMissileProduction launchMissile = + produce `race_` produce `race_` produce `race_` produce + where + produce = forever $ do + sleepRandom + missile <- randomIO :: IO Missile + catch + ( do + printThread $ "Production: Ready to launch missile " ++ show missile + site <- launchMissile missile + printThread $ "Production: Missile " ++ show missile ++ " launched in " ++ site + ) + ( \e-> + printThread $ "Production: Missile " ++ show missile ++ + " failed to launch due to " ++ show (e :: SomeException) + ) + +printThread :: Show a => a -> IO () +printThread x = do + threadId <- myThreadId + random100 >>= \x-> threadDelay (x * 100) + putStrLn $ show threadId ++ ": " ++ show x + +random100 :: IO Int +random100 = (`mod` 100) <$> randomIO + +sleepRandom :: IO () +sleepRandom = random100 >>= \x-> threadDelay (x * 100000) +``` + +### Dependencies + + - base >= 4.7 && < 5 + +[badge-travis]: https://img.shields.io/travis/lpeterse/haskell-concurrent-rpc.svg +[travis]: https://travis-ci.org/lpeterse/haskell-concurrent-rpc +[badge-hackage]: https://img.shields.io/hackage/v/concurrent-rpc.svg?dummy +[hackage]: https://hackage.haskell.org/package/concurrent-rpc +[badge-license]: https://img.shields.io/badge/license-MIT-green.svg?dummy +[license]: https://github.com/lpeterse/haskell-concurrent-rpc/blob/master/LICENSE +[issues]: https://github.com/lpeterse/haskell-concurrent-rpc/issues +[Github]: https://github.com/lpeterse/haskell-concurrent-rpc diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/concurrent-rpc.cabal b/concurrent-rpc.cabal new file mode 100644 --- /dev/null +++ b/concurrent-rpc.cabal @@ -0,0 +1,24 @@ +name: concurrent-rpc +version: 0.1.0.0 +synopsis: An abstraction for inter-thread RPC based on MVars +description: + This library is small wrapper around `Control.Concurrent.MVar.MVar`s that can + be used to implement request-response communication between different threads. +homepage: https://github.com/lpeterse/haskell-concurrent-rpc +license: MIT +license-file: LICENSE +author: Lars Petersen +maintainer: info@lars-petersen.net +category: Concurrency +build-type: Simple +cabal-version: >=1.10 + +extra-source-files: MissileLauncher.hs, + README.md + +library + ghc-options: -Wall + exposed-modules: Control.Concurrent.RPC + build-depends: base >=4.7 && <5 + hs-source-dirs: src + default-language: Haskell2010 diff --git a/src/Control/Concurrent/RPC.hs b/src/Control/Concurrent/RPC.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/RPC.hs @@ -0,0 +1,108 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.RPC +-- Copyright : (c) Lars Petersen 2016 +-- License : MIT +-- +-- Maintainer : info@lars-petersen.net +-- Stability : experimental +-- +-- This module offers the single method `newRPC` which creates a typed +-- communication channel. Give the one end of the channel to client threads +-- and the other end to worker threads. +-- +-- Also see https://gist.github.com/lpeterse/da7b574da5c7a0dc9794 for an example. +----------------------------------------------------------------------------- +module Control.Concurrent.RPC where + +import Control.Exception +import Control.Concurrent.MVar + +-- | The client interface for threads invoking /remote procedure calls/. +-- +-- * The operation blocks until another thread processed the request. +-- * If the other thread throws an exception during processing, the exception +-- is re-thrown in the thread waiting for the response. +-- +-- > type Rocket = String +-- > type Liftoff = Bool +-- > +-- > main = do +-- > (launchRocket, withRocket) <- newRPC :: IO (RPC Rocket Liftoff, WithRPC Rocket Liftoff) +-- > .. +-- > catch +-- > ( launchRocket "Apollo 11" >>= \liftoff-> if liftoff +-- > then print "Houston, we have a liftoff!" +-- > else print "Launch cancelled!" +-- > ) +-- > ( \e-> print $ "Houston, we have a problem: " ++ show (e :: IOError) ) +type RPC request response = request -> IO response + +-- | The interface for threads that serve and process /remote procedure calls/. +-- +-- * More than one thread may be used to process requests all using the same +-- interface object. All processing threads will block on `Control.Concurrent.take` +-- on the same `MVar` and only one will be served at a time. Fairness +-- properties of `Control.Concurrent.MVar` apply. +-- * Exceptions thrown by the handler operation will be re-thrown in both +-- the processing thread and the requesting thread. +-- +-- > type Rocket = String +-- > type Liftoff = Bool +-- > +-- > main = do +-- > (launchRocket, withRocket) <- newRPC :: IO (RPC Rocket Liftoff, WithRPC Rocket Liftoff) +-- > .. +-- > -- This is the rocket launch site thread. It forever waits for rockets and fires them into space one after the other. +-- > forkIO $ withFile "/dev/null" WriteMode $ \space-> +-- > forever $ catch +-- > ( withRocket $ \rocket-> do +-- > weather <- getWeatherCondition +-- > when (isGood weather) $ +-- > hPutStrLn space rocket -- The actual launch may throw an IOError! +-- > return weather +-- > ) +-- > ( \e-> print "A rocket exploded during launch phase: " ++ (e :: IOError) ) +type WithRPC request response = (request -> IO response) -> IO () + +-- | Creates a new request-response communication channel that may be used +-- by arbitrary many requesting and/or processing threads. +-- +-- > main :: IO () +-- > main = do +-- > (rpc, withRpc) <- newRPC +-- > forkIO $ forever $ withRpc $ \request-> +-- > response <- doSomethingWith request +-- > return response +-- > response <- rpc request +-- > .. +-- +-- * `newRPC` initially creates one empty `Control.Concurrent.MVar.MVar` +-- for queueing requests. +-- * Each call of `rpc` creates a temporary `Control.Concurrent.MVar` for the +-- reponse. +-- * If the handler given to `withRpc` throws an exception, the exception is +-- re-thrown in the `withRpc` thread as well as in the `rpc` thread that +-- issued the call. +-- * If an `rpc` thread that issued a call dies before the request +-- processing has started, the request gets discarded. If processing has +-- already started, the processing will finish and the response gets discarded. +newRPC :: IO (RPC request response, WithRPC request response) +newRPC = do + requestMVar <- newEmptyMVar + return (createRequest requestMVar, processRequest requestMVar) + where + createRequest requestMVar request = do + responseMVar <- newEmptyMVar + putMVar requestMVar (request, responseMVar) + response <- readMVar responseMVar + case response of + Left e -> throwIO (e :: SomeException) + Right result -> return result + processRequest requestMVar process = do + (request, responseMVar) <- takeMVar requestMVar + response <- try (process request) + putMVar responseMVar response + case response of + Left e -> throwIO (e :: SomeException) + Right _ -> return ()