Channel.idr

-- ------------------------------------------------------------- [ Channel.idr ]
-- Module    : Channel.idr
-- Copyright : (c) Jan de Muijnck-Hughes
-- License   : see LICENSE
--
-- A effectful API for `System.Concurrency.Sessions`.
--
-- The code was partially adapted from existing code developed by
-- Edwin Brady (see `FILE_IO` effect) and Simon Fowler---see `Process`
-- effect in `IdrisNet2`.
--
-- --------------------------------------------------------------------- [ EOH ]
module Effect.Channel

import Effects

import public System.Concurrency.Channels

%access public export

-- ------------------------------------------------------------ [ The Resource ]

||| States of an active session.
data SSTATE =  Unconnected | Spawned | Active

||| Capture the state of a session and use it as the underlying
||| resource as used by the effect.
|||
||| @st The state the session is in.
data OpenSesh  : (st : SSTATE) -> Type where

   ||| The initial and end state of the session.
   NoSesh      : OpenSesh Unconnected

   ||| Spawned a thread to communicate with and have access to its `PID`.
   ||| @pid The `PID` of the process we are talking to.
   SpawnedSesh : (pid : PID) -> OpenSesh Spawned

   ||| Established a connection to a process.
   |||
   ||| We need to record the PID and resulting session if we have
   ||| spawned and have connected to a process, and just the session
   ||| if we are listening for someone to connect,
   |||
   ||| @pid `Just PID` if we connected to a process, `Nothing` if we
   |||       are listening.
   ||| @sesh The resulting session.
   ActiveSesh  : (pid : Maybe PID) -> (sesh : Channel) -> OpenSesh Active

Default (OpenSesh Unconnected) where
  default = NoSesh

-- ----------------------------------------------------------- [ Type Synonyms ]

||| The initial and end state of the session.
public export
INIT : Type
INIT = OpenSesh Unconnected

||| The state for we have spawned and are waiting to connect.
WAIT : Type
WAIT = OpenSesh Spawned

||| An active session.
ACTIVE : Type
ACTIVE = OpenSesh Active

-- ------------------------------------------------------- [ State Transitions ]

||| If connection was successful then move to active, otherwise go
||| back to wait.
connectedOK : Bool -> Type
connectedOK False = OpenSesh Spawned
connectedOK True  = OpenSesh Active

||| If attempt to listen was successful then move to active, otherwise
||| go back to unconnected.
listenedOK : Bool -> Type
listenedOK False = OpenSesh Unconnected
listenedOK True  = OpenSesh Active

||| If message was sent then stay active, otherwise there was a
||| failure and move to unconnected.
sentOK : Bool -> Type
sentOK True  = OpenSesh Active
sentOK False = OpenSesh Unconnected

||| If message was received then stay active, otherwise there was a
||| failure and move to unconnected.
receivedOK : Maybe a -> Type
receivedOK (Just v) = OpenSesh Active
receivedOK Nothing  = OpenSesh Unconnected


-- ------------------------------------------------------------------ [ Effect ]

mutual -- Need mutual for definition of `ProcessE`.

  ||| Description of an effectful Process that requires use of at
  ||| least one session.
  |||
  ||| @st      The state the session will be in.
  ||| @rTy     The processes return type.
  ||| @inEffs  The state of the effects at the start.
  ||| @outEffs The state of the effects at the end.
  ProcessE : (st      : Type)
          -> (rTy     : Type)
          -> (inEffs  : List EFFECT)
          -> (outEffs : List EFFECT)
          -> Type
  ProcessE st ty inEffs outEffs = Eff ty
                                      (CHANNEL st :: inEffs)
                                      (\_ => (CHANNEL st :: outEffs))

  ||| The effectful description for `Session`.
  data ChannelE : Effect where

    ||| Spawn an effectful process to communicate with, transitioning
    ||| from `INIT` to `WAIT`.
    Spawn : Env IO esi                       -- make m and use rewrite
         -> ProcessE INIT () esi eso
         -> sig ChannelE (PID) (INIT) (WAIT) -- make more fault tolerant

    ||| Connect to a spawned process, transitioning from `INIT` to
    ||| result of `connectedOK`.
    Connect : sig ChannelE (Bool) (WAIT) (\res => connectedOK res)

    ||| Listen for a connection, traisitioning from `INIT` to result
    ||| of `listenedOK`.
    Listen : (tout : Int)
          -> sig ChannelE (Bool) (INIT) (\res => listenedOK  res)

    ||| Send a message of type `a`, transitioning from `ACTIVE` to
    ||| result of `sentOK`.
    Send : a -> sig ChannelE (Bool) (ACTIVE) (\res => sentOK res)

    ||| Receive a message of type `a`, trasitioning from `ACTIVE` to
    ||| result of `receivedOK`.
    Recv : (a : Type)
        -> sig ChannelE (Maybe a) (ACTIVE) (\res => receivedOK res)

    ||| End a session, trasitioning back to `INIT`.
    End : sig ChannelE () (ty) (INIT)


  ||| An effectful API for `System.Concurrency.Sessions`.
  |||
  ||| @st The state the session is in.
  CHANNEL : (st : Type) -> EFFECT
  CHANNEL st = MkEff st ChannelE


-- ---------------------------------------------------------------------- [ IO ]
||| Dealing with the `IO` context.
Handler ChannelE IO where

  handle NoSesh (Spawn env proc) k = do
      pid <- spawn $ runInit (NoSesh::env) (proc)
      k (pid) (SpawnedSesh pid)

  handle (SpawnedSesh pid) Connect k = do
      msesh <- connect pid

      case msesh of
          Nothing      => k False (SpawnedSesh pid)
          r@(Just res) => k True  (ActiveSesh (Just pid) res)

  handle (NoSesh) (Listen tout) k = do
      msesh <- listen tout

      case msesh of
          Nothing      => k False NoSesh
          r@(Just res) => k True  (ActiveSesh Nothing res)

  handle st@(ActiveSesh pid sesh) (Send m) k = do
      sent <- unsafeSend sesh m
      if sent
        then k True st
        else k False NoSesh

  handle st@(ActiveSesh pid sesh) (Recv ty) k = do
      res <- unsafeRecv ty sesh
      case res of
        Nothing    => k Nothing    NoSesh
        (Just val) => k (Just val) st

  handle st End k = k () NoSesh

-- ----------------------------------------------------------------- [ The API ]

||| A `Process` is a function that can communicate using the session
||| effect.
|||
||| The definition already specifies the session required for
||| communicating.
|||
||| @st  The state the session is in.
||| @rTy The function's return type.
||| @es  The list of effects required.
Process : (st  : Type)
       -> (rTy : Type)
       -> (es  : List EFFECT)
       -> Type
Process st ty inEffs = ProcessE st ty inEffs inEffs


||| Attempt to spawn an effectful process that communicates using a
||| `Sessions'.
|||
||| Returns the `PID` of the spawned process.
|||
||| @env  The environment for the process' effects.
||| @proc The process to be spawned.
spawn : (env  : Env IO es)
     -> (proc : Process INIT () es)
     -> Eff (PID)
            [CHANNEL INIT]
            [CHANNEL WAIT]
spawn env proc = call $ Spawn env proc


||| Attempt to establish a connection to the spawned process.
|||
||| The function can only be used once a session has been spawned.
|||
||| Returns `Bool` describing success.
connect : Eff (Bool)
              [CHANNEL WAIT]
              (\res => [CHANNEL (connectedOK res)])
connect = call $ Connect


||| Listen for a fixed amount of time for a connection from an
||| external process.
|||
||| The function is used by the spawned process to accept incomming
||| requests.
|||
||| Returns a `Bool` describing success.
|||
||| @timeout the period in which to listen for a connection.
listen : (timeout : Int)
      -> Eff (Bool)
             [CHANNEL INIT]
             (\res => [CHANNEL (listenedOK res)])
listen t = call $ Listen t

||| Send a message of type `ty`.
|||
||| Returns `Bool` describing success.
|||
||| @msg The message being sent.
send : (msg : ty)
    -> Eff Bool
           [CHANNEL ACTIVE]
           (\res => [CHANNEL (sentOK res)])
send msg = call $ Send msg

||| Expect to receive a message of the specifyed type.
|||
||| This is unsafe as receive *expects* the incomming message to be of
||| the specified type.
recv : (ty : Type)
    -> Eff (Maybe ty)
           [CHANNEL ACTIVE]
           (\res => [CHANNEL (receivedOK res)])
recv ty = call $ Recv ty

||| Stop the session.
stop : Eff () [CHANNEL ty] [CHANNEL INIT]
stop = call End

-- --------------------------------------------------------------------- [ EOF ]

PingPong.idr

-- ------------------------------------------------------------ [ PingPong.idr ]
-- Module    : PingPong.idr
-- Copyright : (c) Jan de Muijnck-Hughes
-- License   : see LICENSE
--
-- Simple ping pong using a session.
--
-- --------------------------------------------------------------------- [ EOH ]
module Session.Example.PingPong

import Effects
import Effect.StdIO
import Effect.Channel

ping : Process INIT () [STDIO]
ping = do
  True <- listen 10
          | False => stop
  True <- send "ping"
          | False => stop

  msg <- recv String
  case msg of
    Nothing => pure ()
    (Just v) => do
        putStrLn "Ping received"
        printLn v
        stop

pong : Process INIT () [STDIO]
pong = do
  pid <- spawn [()] ping
  True <- connect | False => stop
  msg <- recv String
  case msg of
    Nothing => stop
    (Just v) => do
        putStrLn "pong received"
        printLn v
        True <- send "pong" | False => pure ()
        stop

play : IO ()
play = runInit [default,default] pong


namespace Main
  main : IO ()
  main = play