concurrent-rpc (empty) → 0.1.0.0

raw patch · 6 files changed

+315/−0 lines, 6 filesdep +basesetup-changed

Dependencies added: base

Files

• LICENSE +20/−0
• MissileLauncher.hs +66/−0
• README.md +95/−0
• Setup.hs +2/−0
• concurrent-rpc.cabal +24/−0
• src/Control/Concurrent/RPC.hs +108/−0

@@ -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.

@@ -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)

@@ -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

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -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

@@ -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 ()