packages feed

network-rpca (empty) → 0.0.1

raw patch · 16 files changed

+1778/−0 lines, 16 filesdep +arraydep +basedep +binarybuild-type:Customsetup-changed

Dependencies added: array, base, binary, binary-strict, bytestring, codec-libevent, containers, control-timeout, network, network-bytestring, stm

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) Adam Langley++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:++1. Redistributions of source code must retain the above copyright+   notice, this list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright+   notice, this list of conditions and the following disclaimer in the+   documentation and/or other materials provided with the distribution.++3. Neither the name of the author nor the names of his contributors+   may be used to endorse or promote products derived from this software+   without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE CONTRIBUTORS ``AS IS'' AND ANY EXPRESS+OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS+OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,+STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN+ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE+POSSIBILITY OF SUCH DAMAGE.
+ SEMANTICS.html view
@@ -0,0 +1,26 @@+<html>+  <head><title>RPCA</title></head>++  <body>++<h6>Semantics</h6>++<p>RPCA is an RPC system, but that's a pretty loose term covering everything from I2C messages to SOAP. So this is the definition of exactly what an RPCA endpoint should do.</p>++<p>RPCA RPCs are request, response pairs. Each request has, at most, one response and every response is generated by a single request. That means, at the moment, so unsolicited messages from a server and no streaming replies.</p>++<p>RPCs are carried over TCP connections and each RPC on a given connection is numbered by the client. Each RPC id must be unique over all RPCs inflight on that TCP connection. (Inflight means that a request has been send, but the client hasn't processed the reply yet.) A reply must come back over the same TCP connection as the request which prompted it. If a TCP connection fails, all RPCs inflight on that connection also fail.</p>++<p>An RPC request or reply is a pair of byte strings. The first is the header, which is specific to RPCA. The only part of the header which applications need be concerned with is the error code in the reply header. The second is the payload (either the arguments in the case of a request, or the result in the case of a reply). This may be in any form of the applications' choosing, but it expects that it'll be a libevent tagged data structure.</p>++<p>An RPC is targeted at a service, method pair. A server can export many services but each must have a unique name on that server. (A server is a TCP host + port number.) Each service can have many methods, the names of which need only be unique within that service.</p>++<p>A Channel is an abstract concept on the client side of a way of delivering RPCs, and getting the replies back from a given server, service pair. It's distinct from a connection in that a Channel can have many connections (usually only one at a time, though) and that a Channel targets a specific service on a server.</p>++<p>On a given server a service may be up, lame or down. There's no difference between a service which is down and a service which a server doesn't export. Services which are lame are still capable of serving requests, but are requesting that clients stop sending them because, for example, the server is about to shutdown. When a service becomes lame it sends special health messages along all inbound connections to the server, so that clients may be asynchronously notified. (Note that health messages aren't RPCs so this doesn't contradict the above assertion that there are no unsolicited RPC replies.)</p>++<p>If a Channel is targeted at a single server, service pair, then it's free to assume that the service is immediately up. If not, the server will set the error code in the RPC replies accordingly. If a Channel is load-balancing (i.e. is has multiple possible servers that a request could be routed to) it must wait to perform a health check before routing any requests to any server. A load-balancing Channel stops routing requests to any servers which report lameness.</p>++<p>Note that lameness is a per-service value so that some services on a server may be lame with others are up.</p>++</html>
+ Setup.lhs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ evrpca/evrpca.c view
@@ -0,0 +1,602 @@+#include <stdio.h>+#include <stdlib.h>+#include <assert.h>+#include <string.h>++#include <errno.h>+#include <unistd.h>+#include <fcntl.h>+#include <arpa/inet.h>+#include <sys/socket.h>+#include <netinet/in.h>++#include <event.h>+#include <evdns.h>++#include <glib.h>+#include <glib/gtree.h>++#include "../rpca.gen.h"++// -----------------------------------------------------------------------------+// -----------------------------------------------------------------------------+// BUFFERS:++struct buffer {+  uint8_t *data;+  unsigned length;+};++static void+buffer_free_data(struct buffer *const buf) {+  free(buf->data);+}+// -----------------------------------------------------------------------------+// -----------------------------------------------------------------------------+++// -----------------------------------------------------------------------------+// -----------------------------------------------------------------------------+// CONNECTIONS:++struct connection {+  int fd;+  unsigned state;+  struct bufferevent *be;+  struct sockaddr_in sin;+  struct event connect_event;++  void (*state_cb) (struct connection *, void *);+  void (*read_cb) (struct connection *, void *);+  void *arg;+};++// These are the values of the @state member of a struct connection+enum {+  kConnStateResolving,+  kConnStateConnecting,+  kConnStateUp,+  kConnStateDead+};++// -----------------------------------------------------------------------------+// This is called when something is fatally wrong with the connection+// -----------------------------------------------------------------------------+static void+connection_fatal(struct connection *const conn) {+  close(conn->fd);+  conn->state = kConnStateDead;+  conn->state_cb(conn, conn->arg);+}++// -----------------------------------------------------------------------------+// A callback which is called when there's data in the bufferevent to read+// -----------------------------------------------------------------------------+static void+connection_read(struct bufferevent *const be, void *arg) {+  struct connection *const conn = (struct connection *) arg;+  conn->read_cb(conn, conn->arg);+}++// -----------------------------------------------------------------------------+// A callback which is called when the bufferevent's outbound queue is empty.+// -----------------------------------------------------------------------------+static void+connection_write(struct bufferevent *const be, void *arg) {+  // * Stop trying to write+  struct connection *const conn = (struct connection *) arg;+  bufferevent_enable(conn->be, EV_READ);+}++// -----------------------------------------------------------------------------+// This callback is made when there's an error on the fd.+// -----------------------------------------------------------------------------+static void+connection_error(struct bufferevent *const be, short what, void *arg) {+  struct connection *const conn = (struct connection *) arg;+  connection_fatal(conn);+}++// -----------------------------------------------------------------------------+// Called when the connection's fd is ready+// -----------------------------------------------------------------------------+static void+connection_up(struct connection *const conn) {+  conn->be = bufferevent_new+    (conn->fd, connection_read, connection_write, connection_error, conn);++  conn->state = kConnStateUp;+  conn->state_cb(conn, conn->arg);++  bufferevent_enable(conn->be, EV_READ);+}++// -----------------------------------------------------------------------------+// A callback for when the connecting socket is writable.+// -----------------------------------------------------------------------------+static void+connection_connected(int fd, short event, void *arg) {+  struct connection *const conn = (struct connection *) arg;++  int error;+  socklen_t error_len = sizeof(error);+  getsockopt(conn->fd, SOL_SOCKET, SO_ERROR, &error, &error_len);+  if (error) {+    connection_fatal(conn);+  } else {+    connection_up(conn);+  }+}++// -----------------------------------------------------------------------------+// Called when conn->sin has already been filled in to start the connection.+// -----------------------------------------------------------------------------+static void+connection_connect(struct connection *const conn) {+  const int n = connect(conn->fd, (struct sockaddr *) &conn->sin,+                        sizeof(conn->sin));++  if (n == -1) {+    switch (errno) {+      case EINPROGRESS:+      case EWOULDBLOCK:+        event_set(&conn->connect_event, conn->fd, EV_WRITE,+                  connection_connected, conn);+        event_add(&conn->connect_event, NULL);+        return;+      default:+        connection_fatal(conn);+    }+  } else {+    connection_up(conn);+  }+}++// -----------------------------------------------------------------------------+// This is a callback which is made when a connection's hostname is resolved.+// -----------------------------------------------------------------------------+static void+connection_resolved(int result, char type, int count, int ttl,+                    void *addresses, void *arg) {+  struct connection *const conn = (struct connection *) arg;++  if (result != DNS_ERR_NONE || count < 1) {+    connection_fatal(conn);+    return;+  }++  memcpy(&conn->sin.sin_addr.s_addr, addresses, 4);+  connection_connect(conn);+}++static void+connection_free(struct connection *const conn) {+  close(conn->fd);+  if (conn->be) bufferevent_free(conn->be);+  free(conn);+}++// -----------------------------------------------------------------------------+// Create a new connection to the given host:port pair+// -----------------------------------------------------------------------------+static struct connection *+connection_new(const char *const host, int port,+               void (*state_cb) (struct connection *, void *),+               void (*read_cb) (struct connection *, void *),+               void *arg) {+  struct connection *const conn = malloc(sizeof(struct connection));+  memset(conn, 0, sizeof(struct connection));+  conn->fd = socket(AF_INET, SOCK_STREAM, 0);+  // set socket non-blocking+  const long flags = fcntl(conn->fd, F_GETFL);+  fcntl(conn->fd, F_SETFL, flags | O_NONBLOCK);++  conn->state_cb = state_cb;+  conn->read_cb = read_cb;+  conn->arg = arg;++  conn->sin.sin_family = PF_INET;+  conn->sin.sin_port = htons(port);+  // see if the given hostname is an ip address+  if (inet_aton(host, &conn->sin.sin_addr)) {+    // it's an IP address+    connection_connect(conn);+  } else {+    // it's not an IP address, we need to resolve it+    conn->state = kConnStateResolving;+    evdns_resolve_ipv4(host, 0, connection_resolved, conn);+  }++  return conn;+}++// -----------------------------------------------------------------------------+// Write some data to the connection+// -----------------------------------------------------------------------------+static void+connection_enqueue(struct connection *const conn,+                   const struct buffer *const buf) {+  bufferevent_write(conn->be, buf->data, buf->length);+}++// -----------------------------------------------------------------------------+// -----------------------------------------------------------------------------+++// -----------------------------------------------------------------------------+// -----------------------------------------------------------------------------+// CHANNELS++// -----------------------------------------------------------------------------+// This is the information which is kept about each outstanding RPC+// -----------------------------------------------------------------------------++typedef void (*rpc_callback) (struct rpcreply *, const uint8_t *body,+                              unsigned len, void *);++struct dispatch {+  rpc_callback cb;+  void *arg;+  struct event timeout_event;+  char timeout_event_enabled;+};++static void+dispatch_free(struct dispatch *const disp) {+  if (disp->timeout_event_enabled) {+    event_del(&disp->timeout_event);+  }++  free(disp);+}++static void+prelude_serialise(uint8_t *const out,+                  unsigned header_len, unsigned body_len, unsigned chksum) {+  uint32_t header_len_be = htonl(header_len);+  uint32_t body_len_be = htonl(body_len);+  uint32_t chksum_be = htonl(chksum);++  memcpy(out, &header_len_be, 4);+  memcpy(out + 4, &body_len_be, 4);+  memcpy(out + 8, &chksum_be, 4);+}++static const unsigned kChanStateUp = 1;+static const unsigned kChanStateDown = 2;++struct channel {+  struct connection *conn;+  char *host;+  int port;+  char *service;+  int state;+  uint32_t next_id;++  struct event reconnect_timeout;+  char reconnect_timeout_enabled;++  // This is a mapping from RPC id to a pair of pointers to buffers (which is+  // the header and payload of that RPC). These are requests which are getting+  // buffered here until we have a connection to send them on. In the case that+  // they time out, they are removed from this structure.+  GTree *outq;+  // This is mapping from RPC id to a dispatch structure+  GTree *dispatch;+};++static void+channel_free(struct channel *const chan) {+  free(chan->host);+  free(chan->service);+  g_tree_destroy(chan->outq);+  g_tree_destroy(chan->dispatch);+  if (chan->reconnect_timeout_enabled) evtimer_del(&chan->reconnect_timeout);+  // FIXME: delete connection (if nonnull)+}++// -----------------------------------------------------------------------------+// This is the key compare function for our dispatch and outq trees. The keys+// are pointers with uints packed into them.+// -----------------------------------------------------------------------------+static int+channel_uint_key_cmp(gconstpointer a, gconstpointer b, gpointer arg) {+  const guint ua = GPOINTER_TO_UINT(a);+  const guint ub = GPOINTER_TO_UINT(b);++  if (ua < ub) {+    return -1;+  } else if (ua > ub) {+    return 1;+  } else {+    return 0;+  }+}++// -----------------------------------------------------------------------------+// Our keys are uints packed into the pointer, thus we don't need to do+// anything to destroy them+// -----------------------------------------------------------------------------+static void+channel_uint_key_destroy(gpointer key) { }++// -----------------------------------------------------------------------------+// Free an outq tree values, which is an alloced array of two pointers to+// buffers.+// -----------------------------------------------------------------------------+static void+channel_outq_value_destroy(gpointer value) {+  struct buffer *buffers = (struct buffer *) value;+  buffer_free_data(&buffers[0]);+  buffer_free_data(&buffers[1]);+  buffer_free_data(&buffers[2]);+  free(buffers);+}++// -----------------------------------------------------------------------------+// This is called when we destroy a dispatch tree to delete all the values. In+// a dispatch tree, the values are allocated dispatch objects+// -----------------------------------------------------------------------------+static void+channel_dispatch_value_destroy(gpointer value) {+  struct dispatch *const disp = (struct dispatch *) value;+  dispatch_free(disp);+}++// -----------------------------------------------------------------------------+// When a connect is made, this function is called for each enqueued block of+// data. These blocks of data are rpcrequests (and their payloads) for RPCs+// which have been waiting for the connection to come up.+// -----------------------------------------------------------------------------+static gboolean+channel_block_to_conn(gpointer key, gpointer value, gpointer data) {+  struct channel *const chan = (struct channel *) data;++  struct buffer *const bufs = (struct buffer *) value;+  // * The arg is a pointer to an array of two pointers to buffers+  connection_enqueue(chan->conn, &bufs[0]);+  connection_enqueue(chan->conn, &bufs[1]);+  connection_enqueue(chan->conn, &bufs[2]);++  return FALSE;+}++static const unsigned kMaxSerialisedRPCHeaderSize = 512;+static const unsigned kMaxSerialisedPayloadSize = 1024 * 1024;++static void evbuffer_fake(struct evbuffer *const buf,+                          const uint8_t *header, unsigned length) {+  memset(buf, 0, sizeof(struct evbuffer));+  buf->buffer = (uint8_t *) header;+  buf->totallen = buf->off = length;+}++static void+channel_have_message(struct channel *const chan,+                     const uint8_t *const header, const unsigned header_len,+                     const uint8_t *const body, const unsigned body_len) {+  struct inboundreply reply;+  struct evbuffer buf;++  memset(&reply, 0, sizeof(reply));+  evbuffer_fake(&buf, header, header_len);+  if (inboundreply_unmarshal(&reply, &buf)) {+    fprintf(stderr, "Failed to decode RPC reply header");+    return;+  }++  if (reply.rpc_set) {+    const gpointer id_as_ptr = GUINT_TO_POINTER(reply.id_data);+    const gpointer value = g_tree_lookup(chan->dispatch, id_as_ptr);+    if (!value) {+      fprintf(stderr, "Got reply for unknown RPC id %d\n", reply.id_data);+      return;+    }++    struct dispatch *const disp = (struct dispatch *) value;+    disp->cb(reply.rpc_data, body, body_len, disp->arg);+    if (disp->timeout_event_enabled) {+      evtimer_del(&disp->timeout_event);+      disp->timeout_event_enabled = 0;+    }+    g_tree_remove(chan->dispatch, id_as_ptr);+  }+}++// -----------------------------------------------------------------------------+// This callback is made when there's data in the connection's input buffer+// -----------------------------------------------------------------------------+static void+channel_conn_read(struct connection *conn, void *arg) {+  struct channel *const chan = (struct channel *) arg;++  // We need to read the 12 byte header first+  fprintf(stderr, "channel_conn_read\n");+  for (;;) {+    if (EVBUFFER_LENGTH(conn->be->input) < 12) return;+    uint32_t t;+    memcpy(&t, EVBUFFER_DATA(conn->be->input), sizeof(t));+    const uint32_t header_len = ntohl(t);+    memcpy(&t, EVBUFFER_DATA(conn->be->input) + sizeof(t), sizeof(t));+    const uint32_t body_len = ntohl(t);++    if (header_len > kMaxSerialisedRPCHeaderSize ||+        body_len > kMaxSerialisedPayloadSize) {+      fprintf(stderr, "Got oversized message: %u %u\n", header_len, body_len);+      // FIXME: kill channel+      return;+    }++    // * If we don't have the whole message, give up+    const uint32_t total_len = 12 + header_len + body_len;+    if (EVBUFFER_LENGTH(conn->be->input) < (12 + header_len + body_len)) return;++    channel_have_message(chan,+                         EVBUFFER_DATA(conn->be->input) + 12, header_len,+                         EVBUFFER_DATA(conn->be->input) + 12 + header_len, body_len);+    evbuffer_drain(conn->be->input, total_len);+  }+}++static void channel_conn_state(struct connection *conn, void *arg);+static struct timeval kReconnectTimeout = {5, 0};++// -----------------------------------------------------------------------------+// This is a timeout callback which is called when we want to reconnect+// -----------------------------------------------------------------------------+static void+channel_reconnect(int fd, short events, void *arg) {+  struct channel *const chan = (struct channel *) arg;++  chan->reconnect_timeout_enabled = 0;++  connection_free(chan->conn);+  chan->conn = connection_new(chan->host, chan->port,+                              channel_conn_state,+                              channel_conn_read, chan);+}++// -----------------------------------------------------------------------------+// This callback is made when the state of the connection changes.+// -----------------------------------------------------------------------------+static void+channel_conn_state(struct connection *conn, void *arg) {+  struct channel *const chan = (struct channel *) arg;++  switch (conn->state) {+    case kConnStateUp:+      // the connection has just finished connecting+      // in this case we need to move all the pending data to the connection+      chan->state = kChanStateUp;+      g_tree_foreach(chan->outq, channel_block_to_conn, chan);+      g_tree_destroy(chan->outq);+      chan->outq = g_tree_new_full+        (channel_uint_key_cmp, NULL, channel_uint_key_destroy,+         channel_outq_value_destroy);+      break;+    case kConnStateDead:+      // the connection has just failed. Try to connect again+      chan->state = kChanStateDown;+      evtimer_set(&chan->reconnect_timeout, channel_reconnect, chan);+      evtimer_add(&chan->reconnect_timeout, &kReconnectTimeout);+      chan->reconnect_timeout_enabled = 1;+      break;+  }+}++static struct channel *+channel_new(const char *const host, int port, const char *const service) {+  struct channel *const chan =+    (struct channel *) malloc(sizeof(struct channel));+  memset(chan, 0, sizeof(struct channel));++  chan->host = strdup(host);+  chan->port = port;+  chan->service = strdup(service);+  chan->conn = connection_new(host, port, channel_conn_state,+                              channel_conn_read, chan);++  chan->dispatch = g_tree_new_full+    (channel_uint_key_cmp, NULL, channel_uint_key_destroy,+     channel_dispatch_value_destroy);++  chan->outq = g_tree_new_full+    (channel_uint_key_cmp, NULL, channel_uint_key_destroy,+     channel_outq_value_destroy);++  return chan;+}++static void+channel_enqueue(struct channel *const chan,+                const char *method, rpc_callback cb, void *arg,+                const uint8_t *body, const unsigned body_len,+                unsigned timeout) {+  const uint32_t id = chan->next_id++;+  struct outboundrequest obr;+  struct rpcrequest rpcr;++  memset(&rpcr, 0, sizeof(rpcr));+  rpcrequest_service_assign(&rpcr, chan->service);+  rpcrequest_method_assign(&rpcr, method);++  memset(&obr, 0, sizeof(obr));+  outboundrequest_id_assign(&obr, id);+  outboundrequest_rpc_assign(&obr, &rpcr);++  struct evbuffer *buf = evbuffer_new();+  outboundrequest_marshal(buf, &obr);+  const unsigned header_len = EVBUFFER_LENGTH(buf);++  uint8_t prelude[12];+  prelude_serialise(prelude, header_len, body_len, 0);++  if (chan->state != kChanStateUp) {+    // * Enqueue the RPC for when we have a connection+    uint8_t *const header = (uint8_t *) malloc(header_len);+    memcpy(header, EVBUFFER_DATA(buf), header_len);++    struct buffer *const buffers =+      (struct buffer *) malloc(sizeof(struct buffer ) * 3);+    uint8_t *const prelude_copy = (uint8_t *) malloc(12);+    memcpy(prelude_copy, prelude, 12);++    buffers[0].data = prelude_copy;+    buffers[0].length = 12;++    buffers[1].data = header;+    buffers[1].length = header_len;++    uint8_t *const body_copy = (uint8_t *) malloc(body_len);+    memcpy(body_copy, body, body_len);+    buffers[2].data = (uint8_t *) body;+    buffers[2].length = body_len;++    g_tree_insert(chan->outq, GUINT_TO_POINTER(id), buffers);+  } else {+    // * We have a connection - send the request+    struct buffer dbuf;+    dbuf.data = prelude;+    dbuf.length = 12;+    connection_enqueue(chan->conn, &dbuf);+    dbuf.data = EVBUFFER_DATA(buf);+    dbuf.length = header_len;+    connection_enqueue(chan->conn, &dbuf);+    dbuf.data = (uint8_t *) body;+    dbuf.length = body_len;+    connection_enqueue(chan->conn, &dbuf);+  }++  // * Record the RPC id with the given callback information+  struct dispatch *const disp = (struct dispatch *) malloc(sizeof(struct dispatch));+  memset(disp, 0, sizeof(struct dispatch));+  disp->cb = cb;+  disp->arg = arg;+  disp->timeout_event_enabled = 0;++  g_tree_insert(chan->dispatch, GUINT_TO_POINTER(id), disp);++  evbuffer_free(buf);+}++// -----------------------------------------------------------------------------+// -----------------------------------------------------------------------------++static void callback(struct rpcreply *reply, const uint8_t *body, unsigned len,+                     void *arg) {+  fprintf(stderr, "OMG\n");+}++int+main() {+  event_init();+  evdns_init();+  evtag_init();++  struct channel *const chan = channel_new("127.0.0.1", 4545, "test");+  channel_enqueue(chan, "halfsec", callback, NULL, NULL, 0, 0);++  event_loop(0);++  return 0;+}
+ examples/README view
@@ -0,0 +1,10 @@+This directory contains a couple of very short examples of a client and server.+To build it, you need to run codec-libevent-generate(*) on testrpc.rpc like:+  % codec-libevent-generate testrpc.rpc TestRPC > TestRPC.hs++Then run the programs with+  % cd ../src+  % ghci -i../examples ../examples/client.hs  (or service.hs)+++(* this is a binary which is installed by the codec-libevent package)
+ examples/client.hs view
@@ -0,0 +1,26 @@+module Main where++import Control.Concurrent+import Control.Concurrent.STM+import qualified Data.ByteString as BS+import Network.RPCA.Channel+import Network.RPCA.Structs+import Network.RPCA.Util++import System.Time++import qualified TestRPC++import Text.Printf (printf)++cb :: Either ErrorCode TestRPC.Addnumbersreply -> IO ()+cb reply =+  case reply of+       (Left errorcode) -> print errorcode+       (Right (TestRPC.Addnumbersreply { TestRPC.addnumbersreply_c = c })) -> return ()++main = do+  chan <- networkChannel "test" "127.0.0.1" 4545+  getClockTime >>= print+  sequence $ replicate 10000 ((rpc chan "add" (TestRPC.Addnumbersrequest 400 300) 1) :: IO (Either ErrorCode TestRPC.Addnumbersreply))+  getClockTime >>= print
+ examples/service.hs view
@@ -0,0 +1,45 @@+module Main where++import qualified Data.ByteString as BS+import qualified Data.Map as Map+import Control.Concurrent+import Control.Concurrent.STM+import Control.Timeout++import Network.RPCA.Structs+import Network.RPCA.ExportedService+import Network.RPCA.Util+import qualified Network.RPCA.Port as Port++import Codec.Libevent.Class as Tagged+import qualified TestRPC++-- These two callbacks demo the use of raw callback (where the payload doesn't+-- have to be a Libevent structure)+quickRPC :: Callback+quickRPC rpcreq payload cb = do+  print "Got quick request"+  atomically $ cb rpcreplyEmpty BS.empty++halfsecRPC :: Callback+halfsecRPC rpcreq payload cb = do+  print "Got 1/2 second request"+  addTimeout 0.5 $ atomically $ cb rpcreplyEmpty BS.empty+  return ()++-- This one uses rpcFunction to wrap Libevent parsing of the payload+addNumbers :: TestRPC.Addnumbersrequest -> ((Maybe TestRPC.Addnumbersreply) -> STM ()) -> IO ()+addNumbers (TestRPC.Addnumbersrequest { TestRPC.addnumbersrequest_a = a+                                      , TestRPC.addnumbersrequest_b = b }) cb = do+  atomically $ cb $ Just $ TestRPC.addnumbersreplyEmpty { TestRPC.addnumbersreply_c = a + b }++methods = Map.fromList+  [  ("quick", MethodInfo quickRPC () ())+   , ("halfsec", MethodInfo halfsecRPC () ())+   , ("add", MethodInfo (rpcFunction addNumbers TestRPC.addnumbersrequestEmpty TestRPC.addnumbersreplyEmpty) () ())+  ]++main = do+  port <- Port.new 4545+  serv <- newService port "test" Up methods+  threadDelay 100000000
+ examples/testrpc.rpc view
@@ -0,0 +1,8 @@+struct addnumbersrequest {+  int a = 1;+  int b = 2;+}++struct addnumbersreply {+  int c = 1;+}
+ network-rpca.cabal view
@@ -0,0 +1,19 @@+name:            network-rpca+version:         0.0.1+license:         BSD3+license-file:    LICENSE+author:          Adam Langley <agl@imperialviolet.org>+description:     A cross-platform RPC library+synopsis:        A cross-platform RPC library+category:        Networking+build-depends:   base, containers, array, bytestring>=0.9, codec-libevent>=0.1.2, network-bytestring, network>=2.1, control-timeout>=0.1.1, stm>=2.1, binary>=0.4, binary-strict+stability:       provisional+tested-with:     GHC == 6.8.2+exposed-modules:   Network.RPCA.Channel+                 , Network.RPCA.Connection+                 , Network.RPCA.ExportedService+		 , Network.RPCA.Port+		 , Network.RPCA.Structs+		 , Network.RPCA.Util+hs-source-dirs:  src+extra-source-files: examples/service.hs, examples/client.hs, examples/testrpc.rpc, evrpca/evrpca.c, SEMANTICS.html, rpca.rpc, examples/README
+ rpca.rpc view
@@ -0,0 +1,30 @@+struct outboundrequest {+  int id = 1;+  optional struct[rpcrequest] rpc = 2;+  optional struct[healthprobe] probe = 3;+}++struct rpcrequest {+  string service = 1;+  string method = 2;+  optional int checksum = 3;+}++struct healthprobe {+  string service = 1;+}++struct inboundreply {+  int id = 1;+  optional struct[rpcreply] rpc = 2;+  optional struct[healthreply] health = 3;+}++struct rpcreply {+  int reply_code = 1;+  optional int checksum = 2;+}++struct healthreply {+  int good = 1;+}
+ src/Network/RPCA/Channel.hs view
@@ -0,0 +1,290 @@+-- | A channel carries RPCs to a remote server+module Network.RPCA.Channel(+    ChannelStatus(..)+  , ErrorCode(..)+  , Channel+  , networkChannel+  , rpc+  , rpcAsync+  ) where++import Control.Concurrent.STM+import GHC.Conc+import Control.Timeout+import Control.Monad (when)++import Text.Printf (printf)++import Data.Word+import Data.Maybe (isJust, fromJust)++import Network.Socket hiding (send, sendTo, recv, recvFrom)++import qualified Data.ByteString as BS++import qualified Data.Map as Map+import qualified Data.Sequence as Seq++import Codec.Libevent.Class++import Network.RPCA.Structs+import Network.RPCA.Util+import qualified Network.RPCA.Connection as C++-- | These are the various states that a channel can be in+data ChannelStatus = Down  -- ^ channel is down. Requests will be enqueued+                   | Lame  -- ^ remote end has signaled that it's shutting down+                   | Connecting  -- ^ a connection is being attempted.+                                 --   Requests will be enqueued+                   | Up  -- ^ channel is ready. Requests will be sent to the transport++-- | This just factors out some common code from @rpc@ and @rpcAsync@+genericRPCCallback :: (TaggedStructure a)+                   => ((Either ErrorCode a) -> IO ())+                   -> Rpcreply+                   -> BS.ByteString+                   -> IO ()+genericRPCCallback cont reply payload =+ let+   replyCode = toEnum $ fromIntegral $ rpcreply_reply_code reply+ in+   if replyCode == ErrNone+      then case deserialise payload of+                Left _ -> cont $ Left ErrReplyPayloadParseFailed+                Right x -> cont $ Right x+      else cont $ Left replyCode++-- | Perform an asyncronous RPC call+rpcAsync :: (Channel c, TaggedStructure a, TaggedStructure b)+         => c  -- ^ the channel+         -> String  -- ^ the method name+         -> a  -- ^ the request arguments+         -> ((Either ErrorCode b) -> IO ())  -- ^ callback+         -> Float  -- ^ timeout+         -> IO ()+rpcAsync channel method request cb timeout = do+  nqueue channel (rpcrequestEmpty { rpcrequest_method = method }) (serialise request) (Just timeout) $ genericRPCCallback cb++-- | Perform a syncronous RPC+rpc :: (Channel c, TaggedStructure a, TaggedStructure b)+    => c  -- ^ the channel+    -> String  -- ^ the method name+    -> a  -- ^ request arguments+    -> Float -- ^ timeout+    -> IO (Either ErrorCode b)+rpc channel method request timeout = do+  result <- atomically newEmptyTMVar+  nqueue channel (rpcrequestEmpty {rpcrequest_method = method }) (serialise request) Nothing $ genericRPCCallback (atomically . putTMVar result)+  atomically (readTMVar result) >>= return++-- | This is a channel over which RPCs can be submitted+class Channel c where+  -- | Add an RPC to the outbound queue+  nqueue :: c  -- ^ the channel+         -> Rpcrequest  -- ^ the request - the id and service is filled in by the channel+         -> BS.ByteString  -- ^ payload+         -> Maybe Float  -- ^ timeout in seconds+         -> (Rpcreply -> BS.ByteString -> IO ())  -- ^ callback+         -> IO ()++  -- | Get the current status of the channel+  channelStatus :: c -> TVar ChannelStatus++-- | A simple network channel which tries to maintain a connection to a+--   hostname:port pair.+data NetworkChannel = NetworkChannel+  { ncid :: TVar Word32  -- ^ the next RPC id+  , ncservice :: String  -- ^ the target service name+  , nchost :: String  -- ^ the target host+  , ncport :: Int  -- ^ the target port number+  , ncoutq :: TVar (Seq.Seq (BS.ByteString, BS.ByteString))+    -- ^ requests waiting for a connection+  , ncdispatch :: TVar (Map.Map Word32 (Rpcreply -> BS.ByteString -> IO ()))+    -- ^ the dispatch table for incomming replies. Maps the RPC id to the+    --   handler function+  , nctimeouts :: TVar (Map.Map Word32 TimeoutTag)+    -- ^ maps RPC id to the timeout tag that we can use to cancel the timeout+    --   when the reply comes in+  , ncstatus :: TVar ChannelStatus  -- ^ the status of this channel+  , ncdead :: TVar Bool  -- ^ if set, the channel is to be shutdown+  , ncconn :: TVar C.Connection }++-- | This is the "reading" thread for the network connection. If it throws an+--   exception the connection is shutdown and we end up in closeConnection+readAction :: Socket -> NetworkChannel -> IO ()+readAction socket nc = do+  atomically $ writeTVar (ncstatus nc) Connecting+  hostaddr <- inet_addr $ nchost nc+  print "Connecting"+  connect socket (SockAddrInet (PortNum $ htons $ fromIntegral $ ncport nc) hostaddr)++  -- Set the status and drain the outbound queue+  (q, conn) <- atomically (do+    writeTVar (ncstatus nc) Up  -- TODO: maybe Lame?+    q <- readTVar $ ncoutq nc+    writeTVar (ncoutq nc) Seq.empty+    conn <- readTVar $ ncconn nc+    return (q, conn))++  atomically $ mapM (C.writePacket conn) $ seqToList q++  readReply nc socket++-- | This also runs in the read thread of the connection and loops forever,+--   reading replies from the network and processing them.+readReply :: NetworkChannel -> Socket -> IO ()+readReply nc socket = do+  (a, payload) <- C.readPacket socket+  let mibreply = inboundreplyDeserialiseBS a+  case mibreply of+    Left _ -> error "Protocol error"+    Right ibreply -> do+      -- Lookup the id in the dispatch table, removing it if found. Also,+      -- possibly cancel a timeout linked to the RPC+      mcb <- atomically (do+        dispatch <- readTVar $ ncdispatch nc+        timeouts <- readTVar $ nctimeouts nc++        let id = inboundreply_id ibreply+        let mtimeoutTag = Map.lookup id timeouts++        -- If we have a timeout, cancel it and remove it from our map+        when (isJust mtimeoutTag) (do+          cancelTimeout $ fromJust mtimeoutTag+          writeTVar (nctimeouts nc) $ Map.delete id timeouts)++        case Map.lookup id dispatch of+             Nothing -> return Nothing+             Just cb -> do+               writeTVar (ncdispatch nc) $ Map.delete id dispatch+               return $ Just cb)+      case mcb of+           Nothing -> printf "RPC reply id:%d\n" $ ((fromIntegral $ inboundreply_id ibreply) :: Int)+           Just _ -> return ()++      case mcb of+           Nothing -> printf "Unknown RPC reply" >> readReply nc socket+           Just cb -> do+             case inboundreply_rpc ibreply of+               Nothing -> error "No RPC reply in reply"+               Just rpcreply -> do+                 cb rpcreply payload+                 readReply nc socket++-- | This is called by the Connection when the connection fails. If this+--   channel has been shutdown then we have nothing else to do as the+--   Connection will close the socket and kill the threads. Otherwise, we+--   sleep, make a new socket and retry the connection+closeAction :: NetworkChannel -> IO ()+closeAction nc = do+  (dead, cbs) <- atomically (do+    writeTVar (ncstatus nc) Down+    cbs <- readTVar (ncdispatch nc) >>= return . Map.elems+    timeoutTags <- readTVar (nctimeouts nc) >>= return . Map.elems+    mapM_ cancelTimeout timeoutTags+    writeTVar (ncdispatch nc) Map.empty+    writeTVar (nctimeouts nc) Map.empty+    dead <- readTVar $ ncdead nc+    return (dead, cbs))++  -- fail all enqueued RPCs+  mapM_ (abortRPC ErrTransportFailed) cbs++  if dead+     then return ()+     else do+       threadDelay 1000000+       sock <- socket AF_INET Stream 0+       setSocketOption sock NoDelay 1+       conn <- atomically (do+         conn <- C.new sock $ closeAction nc+         writeTVar (ncconn nc) conn+         return conn)+       C.forkThreads conn $ readAction sock nc++-- | Create a new networkChannel+networkChannel :: String  -- ^ service name+               -> String  -- ^ hostname+               -> Int  -- ^ port number+               -> IO NetworkChannel+networkChannel service host port = do+  sock <- socket AF_INET Stream 0+  setSocketOption sock NoDelay 1+  c <- atomically (do ncid <- newTVar 0+                      ncdispatch <- newTVar Map.empty+                      nctimeouts <- newTVar Map.empty+                      ncstatus <- newTVar Down+                      ncoutq <- newTVar Seq.empty+                      ncdead <- newTVar False+                      ncconn <- newTVar undefined+                      let c = NetworkChannel ncid service host port ncoutq ncdispatch nctimeouts ncstatus ncdead ncconn+                      conn <- C.new sock $ closeAction c+                      writeTVar ncconn conn+                      return c)+  conn <- atomically $ readTVar $ ncconn c+  C.forkThreads conn $ readAction sock c+  return c++-- | Run the given callback with a constructued Rpcreply object which has the+--   correct error code for the given error.+abortRPC :: ErrorCode  -- ^ the error code to give+         -> (Rpcreply -> BS.ByteString -> IO ())  -- ^ the callback+         -> IO ()+abortRPC err cb = do+  -- make up the reply with the correct error code and run the callback+  let rpcreply = rpcreplyEmpty { rpcreply_reply_code = fromIntegral $ fromEnum err }+  cb rpcreply BS.empty++handleTimeout :: NetworkChannel  -- ^ the channel which contains the request+              -> Word32  -- ^ the id which has timed out+              -> IO ()+handleTimeout nc id = do+  -- remove our timeout tag from the map of timeouts and return the callback+  mcb <- atomically (do+    updateTVar (Map.delete id) $ nctimeouts nc+    updateTVar (Map.delete id) (ncdispatch nc) >>= return . Map.lookup id)+  case mcb of+       Nothing -> return ()+       Just cb -> abortRPC ErrTimeout cb++instance Channel NetworkChannel where+  channelStatus = ncstatus++  nqueue nc rpcreq payload mtimeout cb = do+    -- Get an ID number for this RPC+    id <- atomically $ updateTVar ((+) 1) $ ncid nc++    -- If we have a timeout, do the IO preparation now+    timeoutTag <-+      case mtimeout of+           Nothing -> return Nothing+           Just timeout -> addTimeoutAtomic timeout (handleTimeout nc id) >>= return . Just++    -- Update all the bookkeeping+    atomically (do+      let rpcreq' = rpcreq { rpcrequest_service = ncservice nc }+      updateTVar (Map.insert id cb) $ ncdispatch nc+      -- if we have a timeout, run the STM action to enable it and record the+      -- tag in case we want to cancel it.+      when (isJust timeoutTag) (do+        tag <- fromJust timeoutTag+        updateTVar (Map.insert id tag) $ nctimeouts nc+        return ())++      -- build the outgoing request+      let obreq = outboundrequestEmpty { outboundrequest_id = id+                                       , outboundrequest_rpc = Just rpcreq' }+          obreqbytes = outboundrequestSerialiseBS obreq++      -- depending on the channel status we either enqueue the request here or+      -- in the Connection. (If the Connection is still connecting etc, then+      -- enqueuing in the Connection will result in it getting dropped if the+      -- connection fails etc)+      status <- readTVar $ channelStatus nc+      case status of+        Up -> do+          conn <- readTVar $ ncconn nc+          C.writePacket conn (obreqbytes, payload)+        _ -> do+          updateTVar ((Seq.<|) (obreqbytes, payload)) $ ncoutq nc+          return ())
+ src/Network/RPCA/Connection.hs view
@@ -0,0 +1,166 @@+module Network.RPCA.Connection+  ( Connection+  , new+  , forkThreads+  , readPacket+  , writePacket+  , buildHeader+  , connoutq+  ) where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Control.Monad++import Data.Binary.Strict.Get+import Data.Binary.Put+import Foreign.C.Types++import Data.Maybe (fromJust)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL+import qualified Data.Map as Map+import qualified Data.Sequence as Seq++import Network.Socket hiding (send, sendTo, recv, recvFrom)+import Network.Socket.ByteString++import Network.RPCA.Structs++data Connection = Connection { connsocket :: Socket+                             , connoutq :: TVar (Seq.Seq BS.ByteString)+                             , connreaderthread :: TVar (Maybe ThreadId)+                             , connwriterthread :: TVar (Maybe ThreadId)+                             , conndeath :: IO ()+                             , conndead :: TVar Bool }++-- Threading: each connection has two threads: one pumping data out and one+-- reading it in. The failure (by throwing an exception) of either is enough+-- to close the socket and kill the connection. This could happen to both at+-- the same time, in which case they race to set conndead to True.+-- However, we also need to record their ThreadIds in the Connection record+-- so that they can kill each other. It's possible that they could try to+-- kill the connection right away - before the creating thread has recorded+-- the correct ThreadIds in the Connection. Thus, at startup, they both wait+-- for the controlling thread to set conndead to False before doing anything+-- else++new :: Socket  -- ^ the socket to make a connection from+    -> IO ()  -- ^ the action run when the connection fails+    -> STM Connection+new socket deathaction = do+  dead <- newTVar True+  outq <- newTVar Seq.empty+  p1 <- newTVar Nothing+  p2 <- newTVar Nothing++  let conn = Connection socket outq p1 p2 deathaction dead+  return conn++forkThreads :: Connection  -- ^ the connection to fork the threads for+            -> IO ()  -- ^ the action which reads from the socket+            -> IO ()+forkThreads conn readeraction = do+  reader <- forkIO $ waitForReadySignal conn $+                     connectionThreadWrapper conn connwriterthread $+                     readeraction+  writer <- forkIO $ waitForReadySignal conn $+                     connectionThreadWrapper conn connreaderthread $+                     seqToSocket (connoutq conn) (connsocket conn)+  -- update the thread ids in the Connection and set the ready flag+  atomically (writeTVar (connreaderthread conn) (Just reader) >>+              writeTVar (connwriterthread conn) (Just writer) >>+              writeTVar (conndead conn) False)+  return ()++-- | Wait for conndead to be set to False on the given connection, then run+--   the given action+waitForReadySignal :: Connection -> IO a -> IO a+waitForReadySignal conn action = do+  atomically (do dead <- readTVar (conndead conn)+                 if dead == True then retry else return ())+  action++-- | Wrap a connection thread so that, when the thread dies, it races to set+--   the dead flag. If it does so, it closes the socket and kills the other+--   thread+connectionThreadWrapper :: Connection -> (Connection -> TVar (Maybe ThreadId)) -> IO a -> IO a+connectionThreadWrapper conn otherthread action = do+  finally action+          (do isDead <- atomically (do dead <- readTVar (conndead conn)+                                       when (not dead) $ writeTVar (conndead conn) True+                                       return dead)+              when (not isDead) (do t <- atomically (readTVar $ otherthread conn)+                                    killThread $ fromJust t+                                    sClose (connsocket conn)+                                    conndeath conn))++-- | Atomically take elements from the end of the given sequence and write them+--   to the given socket. Throw an exception when the write fails+seqToSocket :: TVar (Seq.Seq BS.ByteString)  -- ^ data is removed from the end+            -> Socket  -- ^ the socket to write to+            -> IO ()+seqToSocket q sock = do+  -- Atomically remove an element from the end of the sequence+  bs <- atomically (do q' <- readTVar q+                       (bs, rest) <-+                         case Seq.viewr q' of+                              Seq.EmptyR -> retry+                              rest Seq.:> head -> return (head, rest)+                       writeTVar q rest+                       return bs)+  -- Write the data to the socket+  bytes <- send sock bs+  when (bytes /= BS.length bs) $ error "Short write"+  seqToSocket q sock++-- | Read a given number of bytes from a socket. Block until that many bytes+--   are availible. Throw an exception otherwise+reada :: Socket -> Int -> IO BS.ByteString+reada _ 0 = return BS.empty+reada sock bytes = do+  bs <- recv sock bytes+  if BS.length bs < bytes+     then if BS.null bs+             then error "Disconnected"+             else reada sock (bytes - BS.length bs) >>= return . BS.append bs+     else return bs++kMaxSerialisedRPCHeaderSize = 512+kMaxSerialisedPayloadSize = 1024 * 1024++-- | Read an rpc header and payload from the wire.+readPacket :: Socket -> IO (BS.ByteString, BS.ByteString)+readPacket sock = do+  -- Read the 12 byte header+  header <- reada sock 12+  -- Parse the header from the wire+  let deserialiseHeader = do+        alen <- getWord32be+        plen <- getWord32be+        chksum <- getWord32be+        return (alen, plen, chksum)+  case fst (runGet deserialiseHeader header) of+       Left _ -> error "Protocol error in header"+       Right (alen, plen, chksum) -> do+         when (alen > kMaxSerialisedRPCHeaderSize) $ error "RPC header too large"+         when (plen > kMaxSerialisedPayloadSize) $ error "Payload too large"+         -- Read the rpcheader and payload+         a <- reada sock $ fromIntegral alen+         payload <- reada sock $ fromIntegral plen+         return (a, payload)++writePacket :: Connection -> (BS.ByteString, BS.ByteString) -> STM ()+writePacket conn (a, payload) = do+  let header = buildHeader (a, payload)+  q <- readTVar $ connoutq conn+  writeTVar (connoutq conn) $ payload Seq.<| a Seq.<| header Seq.<| q++-- | Build a 12-byte header for the given ib/ob header and payload+buildHeader :: (BS.ByteString, BS.ByteString) -> BS.ByteString+buildHeader (a, b) =+  BS.concat $ BSL.toChunks $ runPut (do+    putWord32be $ fromIntegral $ BS.length a+    putWord32be $ fromIntegral $ BS.length b+    putWord32be 0)
+ src/Network/RPCA/ExportedService.hs view
@@ -0,0 +1,111 @@+-- | An ExportedService is a set of methods which can be involved over the+--   network. Each method has a name, which is unique within the service and+--   the name of the service is unique within the Port that it's bound to.+module Network.RPCA.ExportedService+  ( Callback+  , DispatchTable+  , MethodInfo(..)+  , State(..)+  , newService+  , rpcFunction+  ) where++import Control.Concurrent.STM++import qualified Data.ByteString as BS+import qualified Data.Map as Map++import Network.RPCA.Structs+import Network.RPCA.Util+import qualified Network.RPCA.Port as Port++import qualified Codec.Libevent.Class as Tagged++type DispatchTable = Map.Map String MethodInfo++-- | This is the type of a callback from the RPC system to the client+--   application. The first ByteString is the payload of the RPC+--   and the application should call the callback with a reply structure+--   that has the reply_code filled in+type Callback = Rpcrequest+              -> BS.ByteString+              -> (Rpcreply -> BS.ByteString -> STM ())+              -> IO ()++data MethodInfo = MethodInfo { methcallback :: Callback+                             , methargument :: ()  -- ^ for future reflection+                             , methreply :: () -- ^ for future reflection+                             }++data Service = Service { servname :: String+                       , servstate :: TVar State+                       , servport :: Port.Port+                       , servcbs :: DispatchTable }++data State = Up | Lame | Down deriving (Show, Eq)++-- | This wraps a callback function into the format expected by newService.+--   Until this point, the RPC system is payload format agnostic. This+--   wrapper introduces the assumption that the payload is a codec-libevent+--   structure.+rpcFunction :: (Tagged.TaggedStructure a, Tagged.TaggedStructure b)+            => (a -> ((Maybe b) -> STM ()) -> IO ())+            -- ^ the callback function. The first argument is the decoded+            --   arguments structure. The callback is passed a continuation+            --   which takes a possible reply structure.+            -> a  -- ^ dummy value needed for type inference, can be undefined+            -> b  -- ^ dummy value needed for type inference, can be undefined+            -> Callback+rpcFunction handler _ _ request payload cb = do+  let request = Tagged.deserialise payload+  case request of+       (Left _) -> atomically $ cb (rpcreplyEmpty { rpcreply_reply_code = fromIntegral $ fromEnum ErrPayloadParseFailed }) BS.empty+       (Right request) -> do+         let cb' Nothing = cb (rpcreplyEmpty { rpcreply_reply_code = 32 }) BS.empty+             cb' (Just reply) = cb rpcreplyEmpty $ Tagged.serialise reply+         handler request cb'++-- | Export a new service on a given Port+newService :: Port.Port  -- ^ the port to export the service on+           -> String  -- ^ the service name+           -> State  -- ^ initial state of the service+           -> DispatchTable  -- ^ information about the methods+           -> IO ()+newService port name initstate dispatcht = do+  st <- atomically $ newTVar initstate+  let serv = Service name st port dispatcht+  atomically $ Port.addService port name $ dispatch serv++dispatch :: Service -> Port.Callback+dispatch serv obreq payload cb = do+  state <- atomically $ readTVar $ servstate serv+  let good = case state of Up -> 1; otherwise -> 0+  let id = outboundrequest_id obreq++  case outboundrequest_probe obreq of+    Just _ -> atomically $ cb (inboundreplyEmpty { inboundreply_health = Just (healthreplyEmpty { healthreply_good = good}), inboundreply_id = id }) BS.empty+    Nothing -> return ()++  case outboundrequest_rpc obreq of+    Just rpcreq -> do+      let mmethod = Map.lookup (rpcrequest_method rpcreq) $ servcbs serv+      case mmethod of+        Nothing -> do+          -- return unknown method+          let replyCode = fromIntegral $ fromEnum ErrMethodUnknown+          atomically $ cb (inboundreplyEmpty { inboundreply_rpc = Just (rpcreplyEmpty { rpcreply_reply_code = replyCode }), inboundreply_id = id }) BS.empty+        Just method -> (methcallback method) rpcreq payload $ handleReply obreq cb+    Nothing -> return ()++-- | This is a callback from the service implementation which is called when an+--   RPC is finished and the reply is ready+handleReply :: Outboundrequest  -- ^ the request which resulted in this reply+            -> (Inboundreply -> BS.ByteString -> STM ())+               -- ^ the Port's callback action+            -> Rpcreply  -- ^ the method's reply+            -> BS.ByteString  -- ^ ... and payload+            -> STM ()+handleReply obreq cb rpcreply payload = do+  let ibreply = inboundreplyEmpty { inboundreply_id = outboundrequest_id obreq+                                  , inboundreply_rpc = Just rpcreply }+  cb ibreply payload
+ src/Network/RPCA/Port.hs view
@@ -0,0 +1,135 @@+module Network.RPCA.Port+  ( Port+  , Callback+  , new+  , addService+  ) where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Control.Monad++import Data.Binary.Strict.Get+import Data.Maybe (isJust, isNothing, fromJust)+import Data.Either (either)+import Data.Int+import Foreign.C.Types++import qualified Data.ByteString as BS+import qualified Data.Map as Map+import qualified Data.Sequence as Seq++import Network.Socket hiding (send, sendTo, recv, recvFrom)+import Network.Socket.ByteString++import Network.RPCA.Structs+import Network.RPCA.Util+import qualified Network.RPCA.Connection as C++type Callback = Outboundrequest -> BS.ByteString -> (Inboundreply -> BS.ByteString -> STM ()) -> IO ()++data Port = Port { portsock :: Socket+                 , portnextid :: TVar Int64+                 , portconnections :: TVar (Map.Map Int64 C.Connection)+                 , portservices :: TVar (Map.Map String Callback)+                 , portacceptorthread :: ThreadId }++new :: Int -> IO Port+new portno = do+  s <- socket AF_INET Stream 0+  setSocketOption s ReuseAddr 1+  let sockaddr = SockAddrInet (PortNum $ htons $ fromIntegral portno) 0+  bindSocket s sockaddr+  listen s 1+  emptyServices <- atomically $ newTVar Map.empty+  emptyConnections <- atomically $ newTVar Map.empty+  initId <- atomically $ newTVar 1+  myid <- myThreadId+  let port = Port s initId emptyConnections emptyServices myid+  threadid <- forkIO $ accepting port+  return $ port { portacceptorthread = threadid }++-- | Add a service to a port+addService :: Port  -- ^ the port to which the service is added+           -> String  -- ^ the name of the service+           -> Callback  -- ^ the handler callback for this service+           -> STM ()+addService port name cb = do+  updateTVar (Map.insert name cb) $ portservices port+  return ()++accepting :: Port -> IO ()+accepting port = do+  (newsock, addr) <- accept $ portsock port+  setSocketOption newsock NoDelay 1+  id <- atomically $ updateTVar ((+) 1) (portnextid port)+  print $ "Connection from " ++ show addr+  conn <- atomically (do+    conn <- C.new newsock (closeAction port id)+    conns <- readTVar $ portconnections port+    writeTVar (portconnections port) $ Map.insert id conn conns+    return conn)+  C.forkThreads conn (readAction port newsock id)+  accepting port++readAction :: Port -> Socket -> Int64 -> IO ()+readAction port sock id = do+  (abytes, payload) <- C.readPacket sock+  let a = either (const Nothing) Just $ outboundrequestDeserialiseBS abytes+      healthservice = a >>= outboundrequest_probe >>= return . healthprobe_service+      rpcservice = a >>= outboundrequest_rpc >>= return . rpcrequest_service+      servicename = healthservice `mplus` rpcservice+  when (isJust servicename) (do+    let Just serv = servicename+    mservice <- atomically $ readTVar (portservices port) >>= return . Map.lookup serv+    case mservice of+      Nothing -> do+        Just conn <- atomically $ readTVar (portconnections port) >>= return . Map.lookup id+        let replyCode = fromIntegral $ fromEnum ErrServiceUnknown+        nqReplyRPC conn (fromJust a) $ rpcreplyEmpty { rpcreply_reply_code = replyCode }+      Just callback -> callback (fromJust a) payload $ handleReply port id)+  readAction port sock id++-- | Used to send payloadless replies to RPCs which are handled within the Port+--   layer (probably because it's an error)+nqReplyRPC :: C.Connection  -- ^ the connection to send to+           -> Outboundrequest  -- ^ the request to which we are replying+           -> Rpcreply  -- ^ the header of the reply+           -> IO ()+nqReplyRPC conn obrq reply = do+  let ibreply = inboundreplyEmpty { inboundreply_id = outboundrequest_id obrq+                                  , inboundreply_rpc = Just reply }+      ibbytes = inboundreplySerialiseBS ibreply+      header = C.buildHeader (ibbytes, BS.empty)++  atomically (do+    updateTVar (\s -> ibbytes Seq.<| header Seq.<| s) $ C.connoutq conn)+  return ()++-- | A callback which is passed to the service. It takes the reply header+--   and the payload and sends it to the connection, if it still exists+handleReply :: Port  -- ^ the port which the request came in on+            -> Int64  -- ^ the id of the connection+            -> Inboundreply  -- ^ the reply header+            -> BS.ByteString   --- ^ ... and payload+            -> STM ()+handleReply port id ibreply payload = do+  let ibbytes = inboundreplySerialiseBS ibreply+      header = C.buildHeader (ibbytes, payload)+  conns <- readTVar $ portconnections port+  let mconn = Map.lookup id conns+  case mconn of+       Nothing -> return ()+       Just conn -> do+         q <- readTVar $ C.connoutq conn+         writeTVar (C.connoutq conn) $ payload Seq.<| ibbytes Seq.<| header Seq.<| q+         return ()++-- | This is called when a connection is closed+closeAction :: Port  -- ^ the port which owns the connection+            -> Int64  -- ^ the id of the connection+            -> IO ()+closeAction port id = atomically (do+  updateTVar (Map.delete id) $ portconnections port+  return ())
+ src/Network/RPCA/Structs.hs view
@@ -0,0 +1,239 @@+module Network.RPCA.Structs where++import Codec.Libevent+import Data.Word++import qualified Data.IntSet as IS++import Data.Binary.Put+import Data.Binary.Strict.Get++import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL++import Codec.Libevent.Class+data Outboundrequest = Outboundrequest { outboundrequest_id :: Word32, outboundrequest_rpc :: Maybe Rpcrequest, outboundrequest_probe :: Maybe Healthprobe } deriving (Show, Eq)++outboundrequestSerialise :: Outboundrequest -> Put+outboundrequestSerialise x = do+  putTaggedWord32 1 (outboundrequest_id x)+  case (outboundrequest_rpc x) of+       Nothing -> return ()+       (Just x) -> putTaggedVarBytes 2 $ rpcrequestSerialiseBS (x)+  case (outboundrequest_probe x) of+       Nothing -> return ()+       (Just x) -> putTaggedVarBytes 3 $ healthprobeSerialiseBS (x)++outboundrequestSerialiseBS = BS.concat . BSL.toChunks . runPut . outboundrequestSerialise++outboundrequestEmpty = Outboundrequest 0 Nothing Nothing++outboundrequestRequiredElementsSet = IS.fromList [1]++outboundrequestDeserialise :: Get Outboundrequest+outboundrequestDeserialise = f outboundrequestEmpty IS.empty where+  f o set = do+    emptyp <- isEmpty+    if emptyp+       then if not (IS.isSubsetOf outboundrequestRequiredElementsSet set)+               then fail "Outboundrequest did not contain all required elements"+               else return o+       else do tag <- getBase128+               case tag of+                 1 -> getWord8 >> getLengthPrefixed >>= (\v -> f (o { outboundrequest_id = v }) (IS.insert 1 set))+                 2 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> case (rpcrequestDeserialiseBS v) of { Left err -> fail ("Failed to deserialse Outboundrequest: " ++ err) ; Right result -> f (o {outboundrequest_rpc = (Just result) }) (IS.insert 2 set) })+                 3 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> case (healthprobeDeserialiseBS v) of { Left err -> fail ("Failed to deserialse Outboundrequest: " ++ err) ; Right result -> f (o {outboundrequest_probe = (Just result) }) (IS.insert 3 set) })+                 otherwise -> getLengthPrefixed >>= getByteString . fromIntegral >> f o set+++outboundrequestDeserialiseBS = fst . runGet outboundrequestDeserialise++instance TaggedStructure Outboundrequest where+  empty = outboundrequestEmpty+  serialise = outboundrequestSerialiseBS+  deserialise = outboundrequestDeserialiseBS++data Rpcrequest = Rpcrequest { rpcrequest_service :: String, rpcrequest_method :: String, rpcrequest_checksum :: Maybe Word32 } deriving (Show, Eq)++rpcrequestSerialise :: Rpcrequest -> Put+rpcrequestSerialise x = do+  putTaggedString 1 (rpcrequest_service x)+  putTaggedString 2 (rpcrequest_method x)+  case (rpcrequest_checksum x) of+       Nothing -> return ()+       (Just x) -> putTaggedWord32 3 (x)++rpcrequestSerialiseBS = BS.concat . BSL.toChunks . runPut . rpcrequestSerialise++rpcrequestEmpty = Rpcrequest "" "" Nothing++rpcrequestRequiredElementsSet = IS.fromList [1,2]++rpcrequestDeserialise :: Get Rpcrequest+rpcrequestDeserialise = f rpcrequestEmpty IS.empty where+  f o set = do+    emptyp <- isEmpty+    if emptyp+       then if not (IS.isSubsetOf rpcrequestRequiredElementsSet set)+               then fail "Rpcrequest did not contain all required elements"+               else return o+       else do tag <- getBase128+               case tag of+                 1 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> f (o { rpcrequest_service = (decodeString v) }) (IS.insert 1 set))+                 2 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> f (o { rpcrequest_method = (decodeString v) }) (IS.insert 2 set))+                 3 -> getWord8 >> getLengthPrefixed >>= (\v -> f (o { rpcrequest_checksum = (Just v) }) (IS.insert 3 set))+                 otherwise -> getLengthPrefixed >>= getByteString . fromIntegral >> f o set+++rpcrequestDeserialiseBS = fst . runGet rpcrequestDeserialise++instance TaggedStructure Rpcrequest where+  empty = rpcrequestEmpty+  serialise = rpcrequestSerialiseBS+  deserialise = rpcrequestDeserialiseBS++data Healthprobe = Healthprobe { healthprobe_service :: String } deriving (Show, Eq)++healthprobeSerialise :: Healthprobe -> Put+healthprobeSerialise x = do+  putTaggedString 1 (healthprobe_service x)++healthprobeSerialiseBS = BS.concat . BSL.toChunks . runPut . healthprobeSerialise++healthprobeEmpty = Healthprobe ""++healthprobeRequiredElementsSet = IS.fromList [1]++healthprobeDeserialise :: Get Healthprobe+healthprobeDeserialise = f healthprobeEmpty IS.empty where+  f o set = do+    emptyp <- isEmpty+    if emptyp+       then if not (IS.isSubsetOf healthprobeRequiredElementsSet set)+               then fail "Healthprobe did not contain all required elements"+               else return o+       else do tag <- getBase128+               case tag of+                 1 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> f (o { healthprobe_service = (decodeString v) }) (IS.insert 1 set))+                 otherwise -> getLengthPrefixed >>= getByteString . fromIntegral >> f o set+++healthprobeDeserialiseBS = fst . runGet healthprobeDeserialise++instance TaggedStructure Healthprobe where+  empty = healthprobeEmpty+  serialise = healthprobeSerialiseBS+  deserialise = healthprobeDeserialiseBS++data Inboundreply = Inboundreply { inboundreply_id :: Word32, inboundreply_rpc :: Maybe Rpcreply, inboundreply_health :: Maybe Healthreply } deriving (Show, Eq)++inboundreplySerialise :: Inboundreply -> Put+inboundreplySerialise x = do+  putTaggedWord32 1 (inboundreply_id x)+  case (inboundreply_rpc x) of+       Nothing -> return ()+       (Just x) -> putTaggedVarBytes 2 $ rpcreplySerialiseBS (x)+  case (inboundreply_health x) of+       Nothing -> return ()+       (Just x) -> putTaggedVarBytes 3 $ healthreplySerialiseBS (x)++inboundreplySerialiseBS = BS.concat . BSL.toChunks . runPut . inboundreplySerialise++inboundreplyEmpty = Inboundreply 0 Nothing Nothing++inboundreplyRequiredElementsSet = IS.fromList [1]++inboundreplyDeserialise :: Get Inboundreply+inboundreplyDeserialise = f inboundreplyEmpty IS.empty where+  f o set = do+    emptyp <- isEmpty+    if emptyp+       then if not (IS.isSubsetOf inboundreplyRequiredElementsSet set)+               then fail "Inboundreply did not contain all required elements"+               else return o+       else do tag <- getBase128+               case tag of+                 1 -> getWord8 >> getLengthPrefixed >>= (\v -> f (o { inboundreply_id = v }) (IS.insert 1 set))+                 2 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> case (rpcreplyDeserialiseBS v) of { Left err -> fail ("Failed to deserialse Inboundreply: " ++ err) ; Right result -> f (o {inboundreply_rpc = (Just result) }) (IS.insert 2 set) })+                 3 -> getLengthPrefixed >>= getByteString . fromIntegral >>= (\v -> case (healthreplyDeserialiseBS v) of { Left err -> fail ("Failed to deserialse Inboundreply: " ++ err) ; Right result -> f (o {inboundreply_health = (Just result) }) (IS.insert 3 set) })+                 otherwise -> getLengthPrefixed >>= getByteString . fromIntegral >> f o set+++inboundreplyDeserialiseBS = fst . runGet inboundreplyDeserialise++instance TaggedStructure Inboundreply where+  empty = inboundreplyEmpty+  serialise = inboundreplySerialiseBS+  deserialise = inboundreplyDeserialiseBS++data Rpcreply = Rpcreply { rpcreply_reply_code :: Word32, rpcreply_checksum :: Maybe Word32 } deriving (Show, Eq)++rpcreplySerialise :: Rpcreply -> Put+rpcreplySerialise x = do+  putTaggedWord32 1 (rpcreply_reply_code x)+  case (rpcreply_checksum x) of+       Nothing -> return ()+       (Just x) -> putTaggedWord32 2 (x)++rpcreplySerialiseBS = BS.concat . BSL.toChunks . runPut . rpcreplySerialise++rpcreplyEmpty = Rpcreply 0 Nothing++rpcreplyRequiredElementsSet = IS.fromList [1]++rpcreplyDeserialise :: Get Rpcreply+rpcreplyDeserialise = f rpcreplyEmpty IS.empty where+  f o set = do+    emptyp <- isEmpty+    if emptyp+       then if not (IS.isSubsetOf rpcreplyRequiredElementsSet set)+               then fail "Rpcreply did not contain all required elements"+               else return o+       else do tag <- getBase128+               case tag of+                 1 -> getWord8 >> getLengthPrefixed >>= (\v -> f (o { rpcreply_reply_code = v }) (IS.insert 1 set))+                 2 -> getWord8 >> getLengthPrefixed >>= (\v -> f (o { rpcreply_checksum = (Just v) }) (IS.insert 2 set))+                 otherwise -> getLengthPrefixed >>= getByteString . fromIntegral >> f o set+++rpcreplyDeserialiseBS = fst . runGet rpcreplyDeserialise++instance TaggedStructure Rpcreply where+  empty = rpcreplyEmpty+  serialise = rpcreplySerialiseBS+  deserialise = rpcreplyDeserialiseBS++data Healthreply = Healthreply { healthreply_good :: Word32 } deriving (Show, Eq)++healthreplySerialise :: Healthreply -> Put+healthreplySerialise x = do+  putTaggedWord32 1 (healthreply_good x)++healthreplySerialiseBS = BS.concat . BSL.toChunks . runPut . healthreplySerialise++healthreplyEmpty = Healthreply 0++healthreplyRequiredElementsSet = IS.fromList [1]++healthreplyDeserialise :: Get Healthreply+healthreplyDeserialise = f healthreplyEmpty IS.empty where+  f o set = do+    emptyp <- isEmpty+    if emptyp+       then if not (IS.isSubsetOf healthreplyRequiredElementsSet set)+               then fail "Healthreply did not contain all required elements"+               else return o+       else do tag <- getBase128+               case tag of+                 1 -> getWord8 >> getLengthPrefixed >>= (\v -> f (o { healthreply_good = v }) (IS.insert 1 set))+                 otherwise -> getLengthPrefixed >>= getByteString . fromIntegral >> f o set+++healthreplyDeserialiseBS = fst . runGet healthreplyDeserialise++instance TaggedStructure Healthreply where+  empty = healthreplyEmpty+  serialise = healthreplySerialiseBS+  deserialise = healthreplyDeserialiseBS++
+ src/Network/RPCA/Util.hs view
@@ -0,0 +1,38 @@+module Network.RPCA.Util where++import Control.Concurrent.STM+import Data.Word+import Data.Bits+import qualified Data.Sequence as Seq++-- | These are the errors which can occur. You'll find the code in the+--   reply_code member of the Rpcreply+data ErrorCode = ErrNone  -- ^ no error occured+               | ErrServiceUnknown  -- ^ the service was unknown at the remote end+               | ErrMethodUnknown  -- ^ the method was unknown at the remote end+               | ErrTransportFailed  -- ^ the connection died+               | ErrTimeout  -- ^ the request timed out+               | ErrPayloadParseFailed  -- ^ remote end failed to deserialise the payload+               | ErrReplyPayloadParseFailed  -- ^ local parsing of the reply failed+               deriving (Enum, Eq, Show)++-- | Atomically update a value by applying a function to it and return+--   the old value+updateTVar :: (a -> a)  -- ^ function to apply+           -> TVar a  -- ^ value to update+           -> STM a+updateTVar f var = do+  old <- readTVar var+  writeTVar var $ f old+  return old++-- | This assumes a little endian system because I can't find a nice way to probe+--   the endianness+htons :: Word16 -> Word16+htons x = ((x .&. 255) `shiftL` 8) .|. (x `shiftR` 8)++seqToList :: Seq.Seq a -> [a]+seqToList s =+  case Seq.viewl s of+    x Seq.:< xs -> x : seqToList xs+    Seq.EmptyL -> []