|
1 | | -{-# LANGUAGE DeriveDataTypeable #-} |
2 | 1 | {-# LANGUAGE ExistentialQuantification #-} |
3 | 2 | {-# LANGUAGE ScopedTypeVariables #-} |
4 | | -{-# LANGUAGE TemplateHaskell #-} |
5 | | -{-# LANGUAGE RecordWildCards #-} |
6 | 3 |
|
7 | 4 | ----------------------------------------------------------------------------- |
8 | 5 | -- | |
|
229 | 226 | -- Using a prioritised process is as simple as calling 'pserve' instead of |
230 | 227 | -- 'serve', and passing an initialised 'PrioritisedProcessDefinition'. |
231 | 228 | -- |
| 229 | +-- Note that prioritised process definitions cannot utilise control channels, |
| 230 | +-- nor can the @handleExternal@ family of expressions be used with them. This |
| 231 | +-- constraint is currenly not enforced by the compiler, and calling @pserve@ |
| 232 | +-- with a @ProcessDefinition@ containing any of these items will fail with |
| 233 | +-- either @ExitOther "IllegalControlChannel"@ or @ExitOther "IllegalSTMAction"@ |
| 234 | +-- at runtime. |
| 235 | +-- |
232 | 236 | -- [Control Channels] |
233 | 237 | -- |
234 | 238 | -- For advanced users and those requiring very low latency, a prioritised |
|
301 | 305 | -- > sendControlMessage cp $ Request str sp |
302 | 306 | -- > receiveChan rp |
303 | 307 | -- |
| 308 | +-- [External (STM) Input Channels] |
| 309 | +-- |
| 310 | +-- Both client and server APIs provide a mechanism for interacting with a running |
| 311 | +-- server process via STM. This is primarily intended for code that runs outside |
| 312 | +-- of Cloud Haskell's /Process/ monad, but can also be used as a channel for |
| 313 | +-- sending and/or receiving non-serializable data to or from a managed process. |
| 314 | +-- Obviously if you attempt to do this across a remote boundary, things will go |
| 315 | +-- spectacularly wrong. The APIs provided do not attempt to restrain this, or |
| 316 | +-- to impose any particular scheme on the programmer, therefore you're on your |
| 317 | +-- own when it comes to writing the /STM/ code for reading and writing data |
| 318 | +-- between client and server. |
| 319 | +-- |
| 320 | +-- For code running inside the /Process/ monad and passing Serializable thunks, |
| 321 | +-- there is no real advantage to this approach, and indeed there are several |
| 322 | +-- serious disadvantages - none of Cloud Haskell's ordering guarantees will hold |
| 323 | +-- when passing data to and from server processes in this fashion, nor are there |
| 324 | +-- any guarantees the runtime system can make with regards interleaving between |
| 325 | +-- messages passed across Cloud Haskell's communication fabric vs. data shared |
| 326 | +-- via STM. This is true even when client(s) and server(s) reside on the same |
| 327 | +-- local node. |
| 328 | +-- |
| 329 | +-- |
| 330 | +-- A server wishing to receive data via STM can do so using the @handleExternal@ |
| 331 | +-- API. By way of example, here is a simple echo server implemented using STM: |
| 332 | +-- |
| 333 | +-- > demoExternal = do |
| 334 | +-- > inChan <- liftIO newTQueueIO |
| 335 | +-- > replyQ <- liftIO newTQueueIO |
| 336 | +-- > let procDef = statelessProcess { |
| 337 | +-- > apiHandlers = [ |
| 338 | +-- > handleExternal |
| 339 | +-- > (readTQueue inChan) |
| 340 | +-- > (\s (m :: String) -> do |
| 341 | +-- > liftIO $ atomically $ writeTQueue replyQ m |
| 342 | +-- > continue s) |
| 343 | +-- > ] |
| 344 | +-- > } |
| 345 | +-- > let txt = "hello 2-way stm foo" |
| 346 | +-- > pid <- spawnLocal $ serve () (statelessInit Infinity) procDef |
| 347 | +-- > echoTxt <- liftIO $ do |
| 348 | +-- > -- firstly we write something that the server can receive |
| 349 | +-- > atomically $ writeTQueue inChan txt |
| 350 | +-- > -- then sit and wait for it to write something back to us |
| 351 | +-- > atomically $ readTQueue replyQ |
| 352 | +-- > |
| 353 | +-- > say (show $ echoTxt == txt) |
| 354 | +-- |
| 355 | +-- For request/reply channels such as this, a convenience based on the call API |
| 356 | +-- is also provided, which allows the server author to write an ordinary call |
| 357 | +-- handler, and the client author to utilise an API that monitors the server and |
| 358 | +-- does the usual stuff you'd expect an RPC style client to do. Here is another |
| 359 | +-- example of this in use, demonstrating the @callSTM@ and @handleCallExternal@ |
| 360 | +-- APIs in practise. |
| 361 | +-- |
| 362 | +-- > data StmServer = StmServer { serverPid :: ProcessId |
| 363 | +-- > , writerChan :: TQueue String |
| 364 | +-- > , readerChan :: TQueue String |
| 365 | +-- > } |
| 366 | +-- > |
| 367 | +-- > instance Resolvable StmServer where |
| 368 | +-- > resolve = return . Just . serverPid |
| 369 | +-- > |
| 370 | +-- > echoStm :: StmServer -> String -> Process (Either ExitReason String) |
| 371 | +-- > echoStm StmServer{..} = callSTM serverPid |
| 372 | +-- > (writeTQueue writerChan) |
| 373 | +-- > (readTQueue readerChan) |
| 374 | +-- > |
| 375 | +-- > launchEchoServer :: CallHandler () String String -> Process StmServer |
| 376 | +-- > launchEchoServer handler = do |
| 377 | +-- > (inQ, replyQ) <- liftIO $ do |
| 378 | +-- > cIn <- newTQueueIO |
| 379 | +-- > cOut <- newTQueueIO |
| 380 | +-- > return (cIn, cOut) |
| 381 | +-- > |
| 382 | +-- > let procDef = statelessProcess { |
| 383 | +-- > apiHandlers = [ |
| 384 | +-- > handleCallExternal |
| 385 | +-- > (readTQueue inQ) |
| 386 | +-- > (writeTQueue replyQ) |
| 387 | +-- > handler |
| 388 | +-- > ] |
| 389 | +-- > } |
| 390 | +-- > |
| 391 | +-- > pid <- spawnLocal $ serve () (statelessInit Infinity) procDef |
| 392 | +-- > return $ StmServer pid inQ replyQ |
| 393 | +-- > |
| 394 | +-- > testExternalCall :: TestResult Bool -> Process () |
| 395 | +-- > testExternalCall result = do |
| 396 | +-- > let txt = "hello stm-call foo" |
| 397 | +-- > srv <- launchEchoServer (\st (msg :: String) -> reply msg st) |
| 398 | +-- > echoStm srv txt >>= stash result . (== Right txt) |
| 399 | +-- |
304 | 400 | -- [Performance Considerations] |
305 | 401 | -- |
306 | 402 | -- The various server loops are fairly optimised, but there /is/ a definite |
@@ -386,6 +482,10 @@ module Control.Distributed.Process.ManagedProcess |
386 | 482 | , channelControlPort |
387 | 483 | , handleControlChan |
388 | 484 | , handleControlChan_ |
| 485 | + -- * Arbitrary STM actions |
| 486 | + , handleExternal |
| 487 | + , handleExternal_ |
| 488 | + , handleCallExternal |
389 | 489 | -- * Prioritised mailboxes |
390 | 490 | , module Control.Distributed.Process.ManagedProcess.Server.Priority |
391 | 491 | -- * Constructing handler results |
@@ -522,4 +622,3 @@ statelessProcess = defaultProcess :: ProcessDefinition () |
522 | 622 | -- state (i.e., unit) and the given 'Delay'. |
523 | 623 | statelessInit :: Delay -> InitHandler () () |
524 | 624 | statelessInit d () = return $ InitOk () d |
525 | | - |
|
0 commit comments