GenServer

[hyperthunk]

See issue #1 and https://gist.github.com/4025934

[rodlogic]

Continuing over here since it is purely genserver related...

How do I register a process name given I have a SendPort? Is there an easy way to extract the ProcessId from it?

I also could not find a timeout version of receiveChan.

[edsko]

How do I register a process name given I have a SendPort? Is there an easy way to extract the ProcessId from it?

Yes, sendPortProcessId . sendPortId

I also could not find a timeout version of receiveChan.

This is available in HEAD.

[rodlogic]

I am not 100% if this is an issue with the prototype code I have or an issue with spawnChannelLocal. Basically, spawnChannelLocal returns a SendPort a, but spawnChannelLocal is called from GenServer which is a step removed from the concrete types in Counter? Is this something we will have to defer to the Server callbacks too to make sure it is fully typed?

This is the compilation error in line 132 of GenServer.hs (https://github.com/rodlogic/distributed-process-platform/blob/master/src/Control/Distributed/Platform/GenServer.hs).

  Couldn't match type `rq' with `(ProcessId, rq)'
      `rq' is a rigid type variable bound by
           the type signature for
             serverStart :: (Serializable rq, Serializable rs) =>
                            Name -> Process (Server rq rs) -> Process (SendPort rq)
           at src/Control/Distributed/Platform/GenServer.hs:96:1
    Expected type: SendPort rq
      Actual type: SendPort (ProcessId, rq)
    In the first argument of `return', namely `sreq'
    In a stmt of a 'do' block: return sreq

@hyperthunk I am committing my experiments in the above fork, btw. Still putting my fingers on the different areas of distributed-process.

[rodlogic]

@edsko Is there a way to register the SendPort by name instead? If the server is using channels, the pid is a bit useless.

[edsko]

@edsko Is there a way to register the SendPort by name instead? If the server is using channels, the pid is a bit useless.

No. You could add your own registry for sendports if you wish.

[edsko]

Is this something we will have to defer to the Server callbacks too to make sure it is fully typed?

No, that's not the issue; the logic in your code doesn't quite add up. On line 123 you ask for a message from the channel taht containts both the request and a process ID, but yet you claim in the type of the function that the channel only carries requests.

[hyperthunk]

@rodlogic - I like these ideas and think we should start to collaboratively work on GenServer now. I have some feedback I'd like to make so once you've got this compiling, please send me a pull request so I can comment on individual lines of code in the diff. Once we're merged I'll start writing some HUnit tests for it and we can split up some more fine grained tasks to work on individually.

I quite like the setup you've got now, bar some stylistic questions, especially the use of temporary channels which is really nice. Most of the things I'd like to suggest are renaming, and a bit of splitting up 'where' into separate functions. Some things we can add quite quickly once we've merged this (apart from tests!) include:

Tagging internal messages

OTP's gen_server differentiates between messages sent via call/cast and 'info' messages by sending the former in a tagged tuple {'$gen', Msg} and we can do the same easily. That way the server 'loop' can use receiveWait to drain the message queue and match on these easily, with something like

type ReplyTo = ReceivePort
data CallRequest = CallReq Serialisable ReplyTo
data CastRequest = CastReq Serialisable

and instead of expect we can use

receiveWait [ match (\(CallReq m replyTo) -> handleCall server m replyTo)
    , match (\(CastReq m) -> handleCast server m)
    , match (\(Serializable m) -> handleInfo server m)
    ]

When the callbacks return a Timeout in the reply/noreply data, we can defer to receiveTimeout instead.

An explicit asynchronous call API

As well as providing a call implementation, we might consider beginCall and endCall as well, with beginCall returning the ReceivePort and endCall doing the blocking receive (with an optional timeout). That way clients can choose to defer getting the result until they want it, whilst call can be defined in terms of endCall . beginCall or some such with a bit of glue around it.

[rodlogic]

@hyperthunk Sounds good. Let me get it into a running state again and then I'll send a pull request. I have paid no attention to the code structure so far and was more focused on the actual call interaction with expect/channels/etc. I am/was expecting we'll iterate this and refactor the code as necessary.

Tagging internal messages

If we go the typed-channels route, which imo we should unless we have specific reason not to, we will have to use receiveChan instead of expect/receiveWait and to handle multiple types of messages we will have to "merge" the different ports with mergePort{RR,Biased}. Aside from the order in which the ports are selected, I am assuming that the underlying semantics is the same as receiveWait's with multiple matches. Is that so?

Now, what about info messages? They seem to be used for 'all the other kinds of messages', but what kind of messages are expected there in Erlang? The gen_server.erl has simple examples like:

handle_info(Info, State) Info is e.g. {'EXIT', P, R}, {nodedown, N}, ...

And from the documentation:

If the gen_server should be able to receive other messages than requests, the callback function handle_info(Info, State) must be implemented to handle them. Examples of other messages are exit messages, if the gen_server is linked to other processes (than the supervisor) and trapping exit signals.

handle_info({'EXIT', Pid, Reason}, State) ->
    ..code to handle exits here..
    {noreply, State1}.

Are there other specific examples of how info messages are used in Erlang?

An explicit asynchronous call API

Definitely. In the Java world is common to see APIs like:

public interface Counter {
   int count();
   Future<Integer> countAsync();
}

And the Future interface, which I think is what you are more or less referring to with the difference that it hide the internals from the client process.

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

I am sure Haskell has similar patterns. In it's simplest form, this Future value (the reply) will wrap the reply's receive port completely from the call. The only limitation here is that it won't be possible to send future values to other processes, but unless we have a clear and valuable use case it is not worth solving that.

Another point for us to consider and somewhat related to the async API for calls, is the handling of deferred reply's by the server process, which in Erlang I think happens when you return {noreply, State} from the handle_call callback. It would be nice to have a dual of Future that would represent the promise to send a typed reply back by any process and only once since there is a caller waiting for this reply. The server process's implementation could decide to send this promise to another process, which would then finally reply to the caller. For instance, instead of something like:

handleCall      :: req -> Process (CallResult reply)

it would look like:

handleCall      :: req -> Promise reply -> Process (CallResult)

where Promise would be a simple function/API such as:

serverReply :: Promise reply -> reply -> Process ()

The server implemenation would have the option of simply calling serverReply and returning CallOk or sending the Promise to another process and returning CallDeferred. The promise wrapper will guarantee that serverReply can be called only once.

[rodlogic]

Created a pull request: https://github.com/hyperthunk/distributed-process-platform/pull/5

[hyperthunk]

@rodlogic good, I think we're on the same track here. Some thoughts...

If we go the typed-channels route, which imo we should unless we have specific reason not to, we will have to use receiveChan instead of expect/receiveWait and to handle multiple types of messages we will have to "merge" the different ports with mergePort{RR,Biased}. Aside from the order in which the ports are selected, I am assuming that the underlying semantics is the same as receiveWait's with multiple matches. Is that so?

I do like the typed channels route, but it kind of kills off the handle_info concept, which is for handling 'unexpected' messages or (in other words) messages that aren't a primary part of this particular server's protocol, but are possible such as system messages (for dynamic debugging) and monitor signals and so on.

Also, we can't monitor other processes very easily in this way, because monitor signals come via expect/receiveWait rather than on channels. Basically the typical use of handle_info in OTP is to deal with processes that are interested in other processes, though perhaps not supervising them. Consider this example from RabbitMQ:

handle_info({'DOWN', _MRef, process, {rabbit, Node}, _Reason},
            State = #state{monitors = Monitors}) ->
    rabbit_log:info("rabbit on node ~p down~n", [Node]),
    {AllNodes, DiscNodes, RunningNodes} = read_cluster_status(),
    write_cluster_status({AllNodes, DiscNodes, del_node(Node, RunningNodes)}),
    ok = handle_dead_rabbit(Node),
    {noreply, State#state{monitors = pmon:erase({rabbit, Node}, Monitors)}};

This is a classic example of receiving a monitor notification and doing some hidden internal state change in response without replying.

Personally, I think we should actually have two kinds of gen-server: one using channels and one using bare messages. They can probably share a lot of infrastructure, and you'll pick the bare messages one only if you need to do things like monitor and/or link handling.

Killing Processes (also needed for supervision)

Unless I'm really missing something, there doesn't appear to be a corollary to erlang:exit(Pid, Term) in the API, which means that in order to kill a process you have to dream up some protocol to allow (for example) supervisors to instruct their children to terminate and brutally kill them if they don't respond in a timely fashion by exiting.

One way to do this is to pass some Exit reason structure to the process, another approach would be to spawn a 'live' process to which all the children link, and then kill it when you need to terminate them abruptly. Anyway, we need to be able to kill processes in various ways.

The point is partly that we should prioritize 'shutdown' messages. If the Channel based GenServer is going to support this, then need to check we understand the semantics of mergePort_ properly.

Futures

I like the things you've said about this: There is a similar set of concepts in the .NET world (BeginInvoke, EndInvoke).

Another point for us to consider and somewhat related to the async API for calls, is the handling of deferred reply's by the server process, which in Erlang I think happens when you return {noreply, State} from the handle_call callback. It would be nice to have a dual of Future that would represent the promise to send a typed reply back by any process and only once since there is a caller waiting for this reply. The server process's implementation could decide to send this promise to another process, which would then finally reply to the caller. [snip]

Yes the 'reply later' option is very useful. I do wonder whether this will work more cleanly for a Channels based server API wise though.

Another thing Promises/Futures enable is the ability to construct higher order channels, dealing with things like delegation and proxying, which is very very useful (and important!) in building fault tolerant systems. One of the key things about OTP's gen_server is that the implementation is no longer in control of their own mailbox. This can be vitally important. Consider this wrapping code, for example, which deals with the common situation where you need fine grained control over a process that might get into a blocking state very easily - in this case we're managing an external resource (like an OS process) that could take forever to respond, but we must be able to interact with the wrapper in a timely fashion if we need to shut it down: https://github.com/nebularis/systest/blob/resource2/src/systest_resource.erl

Insulating yourself and providing middle-man processes to make sure that you can avoid deadlocks is vital in a complex actor based system, and I suspect doing that purely with Channels might be hard, though I'm willing to be persuaded otherwise.

[hyperthunk]

Created a pull request: hyperthunk/distributed-process-platform#5

@rodlogic thanks that's perfect - I'll try and give it proper attention over the next day or so and hopefully we can get it merged over the weekend.

[rodlogic]

Now that we have some sort of a starting point, I was pondering a bit about the direction we are taking here and wondering if the GenServer design could be even simpler by leveraging more of Haskell's strengths.

Barred some of the limitations of the current design, which we will fix over many iterations, can we simplify even more what someone building a server has to implement? Can we make this 'design survace area' even smaller? I think it should be dead simple to create a server and the type system should keep me honest even if that means a bit more complexity under the covers.

What is a server, really? A server is nothing more than a Process with an id and one or more channels of interaction (in a more abstract sense). These channels are either a call (request/response) or a cast (one-way) and a client may choose to use a call channel synchronously or asynchronously, and a server may choose to 'reply later', delegate the reply to another server/process, or just reply right there. I am ignoring info messages here.

So what could be better in the "Counter" example from the point of view of who is designing and implementing it? Having to implement CounterRequest and CounterResponse smells like cruft to me and somewhat error prone. It would be nice if I could design Counter with mostly functions and have the rest inferred automatically somehow. For instance, the count service in the Counter server could be just:

count :: () -> Int

This is already specifying that there is a Request () and a Response Int. Could this Request and Response message be automatically generated for us? Isn't this was Closure in CloudHaskell is proposing/implementing?

In addition, it would be great if the following could also be automatically derived from the above definition (and implementation):

countAsync :: () -> Future Int

And something like the following for the handler type:

countHandler :: () -> Promise Int -> Maybe Int

[hyperthunk]

@rodlogic I like these ideas in general, though obviously we'll need to understand what the plumbing looks like before we can generate any of it. :)

As the input to any GenServer callback function must be an instance of Typeable then we can presumably use {g}cast to differentiate between the messages passed at runtime and possibly using funResultTy match the possibilities against the supplied callbacks, leaving the implementor to simply list the functions they want to expose as callbacks. With a bit of magic (a la Template Haskell or some such) then we can probably just magically grab everything the callback module is exporting.

And yes, the return type should tell us whether we're dealing with a future/promise or an immediate reply. Of course there are servers that never reply, dealing only with casts and casts in general would need to be handled a little differently, but some indication that we're not replying shouldn't be too hard to dream up.

Of course we always need to deal with timeouts as well, so even if we do avoid making the implementation return a data type indicating whether they're replying or not, we will need to somehow deal with timeouts, hibernation and so on. Ideally we should be able to do this with a type, though I'm struggling to see how we'd wrap the return type (whether it is Int, Future Int or Promise Int) without forcing the author to do something in their code.

I suppose, and I shudder to suggest this, that we can have timeout/hibernate throw an exception containing the required span, but that feels really clunky and I'm sure there's a better way.

One thing though:

What is a server, really? A server is nothing more than a Process with an id and one or more channels of interaction (in a more abstract sense). These channels are either a call (request/response) or a cast (one-way) and a client may choose to use a call channel synchronously or asynchronously, and a server may choose to 'reply later', delegate the reply to another server/process, or just reply right there. I am ignoring info messages here.

Yes, and I am holding to my point about info messages, which is that we probably need a non-channel based option for people who do care about info messages coming from monitors.

[hyperthunk]

Guys, I'm getting a bit stuck with this. I do not want to specify the exact input type for my gen server, as this means that it can only deal with one kind of input message. Honestly, what use is that for writing a supervisor, that needs to handle the following instructions:

• Monitor notifications from the children
• add_child
• delete_child
• restart_child
• stop_children

So how can we write a generic process that accepts all these different types of input messages and handles them uniformly by evaluating a callback function? The callback can be written in terms of generic inputs, for example

handleCall :: (Serializable m) => m -> ProcessId -> Process ()

We need to handle these 5 different input types now, so how does a record with 1 input type help us at all? As I've said in distributed-process issue 71 this seems to completely defeat the purpose of having expect and receiveWait defined in such a way that we can receive anything because we can't use the messages without fully specifying the types we're dealing with.

Now I completely understand why this is the case - how on earth is the type system supposed to guess what we mean if we don't specify the types? I don't know if opening up AbstractMessage as I've suggested in that (previously mentioned) issue is the right thing to do or not. It feels to me like we're dealing with chalk and cheese here - the idea of gen server which deals with a totally open domain of input types just doesn't fit Haskell's type system, which is exactly what I felt to begin with.

Having a gen server that accepts just one type of input is fine, if all you want to do is centralize the error handling, timeouts and so on. If that's what we want to do, then the gen server infrastructure might be overkill.

How I arrived at this....

What I tried (over the weekend) to get this working was several things. First of all, I tried to stop limiting the input domain of the handleX functions to a specific type. As long as we don't mind not being able to use record accessors we can match out the handleX functions we need...

-- we can *try* using existential types for this....
data Server s = forall m. (Serializable m) => Server {
    init             :: .....
    handleCall :: m -> state -> ProcessAction s
    state          :: s
  }

handleRequest :: (Serializable m) => Server s -> m -> ProcessAction s
handleRequest Server{ handleCall = hc } = hc

But is Serializable specific enough to use in receiveWait and can it be applied to data taken out of expect using an m <- expect expression? I tried numerous approaches here, included attempting to encode the input messages in various forms of other types.

So @edsko my question is, taking the following, which compiles (therefore it must work!) ;) are we on the right track and is there some type trickery I'm missing to use the record instead of the type class, because that completely bombed out for me, and what's the neat way of dealing with the Maybe s that receiveTimeout returns in our handleRequest implementation?

{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE FlexibleContexts          #-}
{-# LANGUAGE RankNTypes                #-}
{-# LANGUAGE DeriveDataTypeable        #-}
{-# LANGUAGE ScopedTypeVariables       #-}
{-# LANGUAGE TemplateHaskell           #-}
{-# LANGUAGE TypeFamilies              #-}

module Control.Distributed.Platform.GenProcess where

import Prelude hiding (catch, init)
import Control.Distributed.Process
import Control.Distributed.Process.Serializable
import Control.Monad (forever)
import Control.Concurrent
import Data.Typeable (Typeable)
import Data.Binary
import Data.DeriveTH

type Name = String

data TimeUnit = Hours | Minutes | Seconds | Millis 
data Timeout = Timeout TimeUnit Int | Infinity  

data ProcessAction = 
    ProcessContinue
  | ProcessTimeout Timeout
  | ProcessStop String

data GenMessage m where 
    Message :: (Typeable m) => GenMessage (m, ReplyTo)
        deriving (Typeable)

data ReplyTo = ReplyTo ProcessId | None 
    deriving (Typeable, Show)

data Gen = Call | Cast
    deriving (Typeable, Show)

$(derive makeBinary ''ReplyTo)
$(derive makeBinary ''Gen)

class (Typeable s) => GenProcess s where
    init       :: Process () -> a -> s
    state      :: Process (s) -> s
    handleCall :: (Serializable m) => s -> m -> ReplyTo -> (ProcessAction, s)

serverContinue :: (GenProcess s) => s -> Timeout -> Process (s)
serverContinue s t = do
  case t of
    (Infinity)    -> receiveWait handlers
    -- (Timeout u v) -> receiveTimeout (timeToMs u v) handlers
  where handlers = [ (match (\(Call, m, r) -> handleRequest s m r)) ]

handleRequest :: (GenProcess s, Serializable m) =>
    s -> m -> ReplyTo -> Process (s)
handleRequest s m r = do
    let (action, s2) = handleCall s m r
    case action of
        ProcessContinue    -> serverContinue s2 Infinity
        (ProcessTimeout t) -> serverContinue s2 t

timeToMs :: TimeUnit -> Int -> Int
timeToMs Millis  ms  = ms
timeToMs Seconds sec = sec * 1000
timeToMs Minutes min = (min * 60) * 1000

reply :: (Serializable m) => ReplyTo -> m -> Process ()
reply (ReplyTo pid) m = send pid m
reply _             _ = return () 

replyVia :: (Serializable m) => SendPort m -> m -> Process ()
replyVia p m = sendChan p m

[hyperthunk]

So @edsko my question is, taking the following, which compiles (therefore it must work!)

Urgh - I take that back, I just hadn't reconfigured recently enough.

src/Control/Distributed/Platform/GenProcess.hs:53:47:
    Ambiguous type variable `m0' in the constraints:
      (Typeable m0)
        arising from a use of `handleRequest'
        at src/Control/Distributed/Platform/GenProcess.hs:53:47-59
      (Binary m0)
        arising from a use of `handleRequest'
        at src/Control/Distributed/Platform/GenProcess.hs:53:47-59
    Probable fix: add a type signature that fixes these type variable(s)
    In the expression: handleRequest s m r
    In the first argument of `match', namely
      `(\ (Call, m, r) -> handleRequest s m r)'
    In the expression: (match (\ (Call, m, r) -> handleRequest s m r))

So how do I handle messages in the general case? I can be more specific by using an existential type, but it's still too ambiguous:

type GenMessage = forall m. (Typeable m, Serializable m) => m
-- snip

serverContinue :: (GenProcess s) => s -> Timeout -> Process (s)
serverContinue s t = do
  case t of
    (Infinity)    -> receiveWait handlers
    -- (Timeout u v) -> receiveTimeout (timeToMs u v) handlers
  where handlers = [ (match (\(Call, m, r) -> handleRequest s m r)) ]

handleRequest :: (GenProcess s) =>
    s -> GenMessage -> ReplyTo -> Process (s)
handleRequest s m r = do
    let (action, s2) = handleCall s m r
    case action of
        ProcessContinue    -> serverContinue s2 Infinity
        (ProcessTimeout t) -> serverContinue s2 t

will still yield the same error. So do we actually need an API on AbstractMessage in order to do this then? Or is there some Typeable magic I can do here?

Interestingly, I wonder if instead of trying to define handleCall just once for the gen server, we should take the gen server state's type and a list of possible handlers, each being a pure function mapping from some domain of inputs to a ProcessAction state instead. That would alleviate the problem of knowing a single handleCall input type and we should be able, in theory, to pass these to receiveWait verbatim and get back the ProcessAction s for further inspection.