protobuf-c/protobuf-c-rpc/protobuf-c-rpc.c

/* KNOWN DEFECTS:
    - server does not obey max_pending_requests_per_connection
    - no ipv6 support
 */
#include <string.h>
#include <assert.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <netdb.h>
#include <stdarg.h>
#include <fcntl.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "protobuf-c-rpc.h"
#include "protobuf-c-data-buffer.h"
#include "gsklistmacros.h"

#define protobuf_c_assert(x) assert(x)

#undef TRUE
#define TRUE 1
#undef FALSE
#define FALSE 0

/* enabled for efficiency, can be useful to disable for debugging */
#define RECYCLE_REQUESTS                1

#define UINT_TO_POINTER(ui)      ((void*)(uintptr_t)(ui))
#define POINTER_TO_UINT(ptr)     ((unsigned)(uintptr_t)(ptr))

#define MAX_FAILED_MSG_LENGTH   512

typedef enum
{
  PROTOBUF_C_CLIENT_STATE_INIT,
  PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP,
  PROTOBUF_C_CLIENT_STATE_CONNECTING,
  PROTOBUF_C_CLIENT_STATE_CONNECTED,
  PROTOBUF_C_CLIENT_STATE_FAILED_WAITING,
  PROTOBUF_C_CLIENT_STATE_FAILED,               /* if no autoreconnect */
  PROTOBUF_C_CLIENT_STATE_DESTROYED
} ProtobufC_RPC_ClientState;

typedef struct _Closure Closure;
struct _Closure
{
  /* these will be NULL for unallocated request ids */
  const ProtobufCMessageDescriptor *response_type;
  ProtobufCClosure closure;

  /* this is the next request id, or 0 for none */
  void *closure_data;
};

static void
error_handler (ProtobufC_RPC_Error_Code code,
               const char              *message,
               void                    *error_func_data)
{
  fprintf (stderr, "*** error: %s: %s\n",
           (char*) error_func_data, message);
}

struct _ProtobufC_RPC_Client
{
  ProtobufCService base_service;
  ProtobufCDataBuffer incoming;
  ProtobufCDataBuffer outgoing;
  ProtobufCAllocator *allocator;
  ProtobufCDispatch *dispatch;
  ProtobufC_RPC_AddressType address_type;
  char *name;
  ProtobufC_FD fd;
  protobuf_c_boolean autoreconnect;
  unsigned autoreconnect_millis;
  ProtobufC_NameLookup_Func resolver;
  ProtobufC_RPC_Error_Func error_handler;
  void *error_handler_data;
  ProtobufC_RPC_ClientState state;
  union {
    struct {
      ProtobufCDispatchIdle *idle;
    } init;
    struct {
      protobuf_c_boolean pending;
      protobuf_c_boolean destroyed_while_pending;
      uint16_t port;
    } name_lookup;
    struct {
      unsigned closures_alloced;
      unsigned first_free_request_id;
      /* indexed by (request_id-1) */
      Closure *closures;
    } connected;
    struct {
      ProtobufCDispatchTimer *timer;
      char *error_message;
    } failed_waiting;
    struct {
      char *error_message;
    } failed;
  } info;
};

static void begin_name_lookup (ProtobufC_RPC_Client *client);
static void destroy_client_rpc (ProtobufCService *service);


static void
set_fd_nonblocking(int fd)
{
  int flags = fcntl (fd, F_GETFL);
  protobuf_c_assert (flags >= 0);
  fcntl (fd, F_SETFL, flags | O_NONBLOCK);
}

static void
handle_autoreconnect_timeout (ProtobufCDispatch *dispatch,
                          void              *func_data)
{
  ProtobufC_RPC_Client *client = func_data;
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_FAILED_WAITING);
  client->allocator->free (client->allocator,
                           client->info.failed_waiting.error_message);
  begin_name_lookup (client);
}

static void
client_failed (ProtobufC_RPC_Client *client,
               const char           *format_str,
               ...)
{
  va_list args;
  char buf[MAX_FAILED_MSG_LENGTH];
  size_t msg_len;
  char *msg;
  size_t n_closures = 0;
  Closure *closures = NULL;
  switch (client->state)
    {
    case PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP:
      protobuf_c_assert (!client->info.name_lookup.pending);
      break;
    case PROTOBUF_C_CLIENT_STATE_CONNECTING:
      /* nothing to do */
      break;
    case PROTOBUF_C_CLIENT_STATE_CONNECTED:
      n_closures = client->info.connected.closures_alloced;
      closures = client->info.connected.closures;
      break;

      /* should not get here */
    case PROTOBUF_C_CLIENT_STATE_INIT:
    case PROTOBUF_C_CLIENT_STATE_FAILED_WAITING:
    case PROTOBUF_C_CLIENT_STATE_FAILED:
    case PROTOBUF_C_CLIENT_STATE_DESTROYED:
      protobuf_c_assert (FALSE);
      break;
    }
  if (client->fd >= 0)
    {
      protobuf_c_dispatch_close_fd (client->dispatch, client->fd);
      client->fd = -1;
    }
  protobuf_c_data_buffer_reset (&client->incoming);
  protobuf_c_data_buffer_reset (&client->outgoing);

  /* Compute the message */
  va_start (args, format_str);
  vsnprintf (buf, sizeof (buf), format_str, args);
  va_end (args);
  buf[sizeof(buf)-1] = 0;
  msg_len = strlen (buf);
  msg = client->allocator->alloc (client->allocator, msg_len + 1);
  strcpy (msg, buf);

  /* go to one of the failed states */
  if (client->autoreconnect)
    {
      client->state = PROTOBUF_C_CLIENT_STATE_FAILED_WAITING;
      client->info.failed_waiting.timer
        = protobuf_c_dispatch_add_timer_millis (client->dispatch,
                                                client->autoreconnect_millis,
                                                handle_autoreconnect_timeout,
                                                client);
      client->info.failed_waiting.error_message = msg;
    }
  else
    {
      client->state = PROTOBUF_C_CLIENT_STATE_FAILED;
      client->info.failed.error_message = msg;
    }

  /* we defer calling the closures to avoid
     any re-entrancy issues (e.g. people further RPC should
     not see a socket in the "connected" state-- at least,
     it shouldn't be accessing the array of closures that we are considering */
  if (closures != NULL)
    {
      unsigned i;

      for (i = 0; i < n_closures; i++)
        if (closures[i].response_type != NULL)
          closures[i].closure (NULL, closures[i].closure_data);
      client->allocator->free (client->allocator, closures);
    }
}

static inline protobuf_c_boolean
errno_is_ignorable (int e)
{
#ifdef EWOULDBLOCK              /* for windows */
  if (e == EWOULDBLOCK)
    return 1;
#endif
  return e == EINTR || e == EAGAIN;
}

static void
set_state_connected (ProtobufC_RPC_Client *client)
{
  client->state = PROTOBUF_C_CLIENT_STATE_CONNECTED;

  client->info.connected.closures_alloced = 1;
  client->info.connected.first_free_request_id = 1;
  client->info.connected.closures = client->allocator->alloc (client->allocator, sizeof (Closure));
  client->info.connected.closures[0].closure = NULL;
  client->info.connected.closures[0].response_type = NULL;
  client->info.connected.closures[0].closure_data = UINT_TO_POINTER (0);
}

static void
handle_client_fd_connect_events (int         fd,
                                 unsigned    events,
                                 void       *callback_data)
{
  ProtobufC_RPC_Client *client = callback_data;
  socklen_t size_int = sizeof (int);
  int fd_errno = EINVAL;
  if (getsockopt (fd, SOL_SOCKET, SO_ERROR, &fd_errno, &size_int) < 0)
    {
      /* Note: this behavior is vaguely hypothetically broken,
       *       in terms of ignoring getsockopt's error;
       *       however, this shouldn't happen, and EINVAL is ok if it does.
       *       Furthermore some broken OS's return an error code when
       *       fetching SO_ERROR!
       */
    }

  if (fd_errno == 0)
    {
      /* goto state CONNECTED */
      protobuf_c_dispatch_watch_fd (client->dispatch,
                                    client->fd,
                                    0, NULL, NULL);
      set_state_connected (client);
    }
  else if (errno_is_ignorable (fd_errno))
    {
      /* remain in CONNECTING state */
      return;
    }
  else
    {
      /* Call error handler */
      client_failed (client,
                     "failed connecting to server: %s",
                     strerror (fd_errno));
    }
}

static void
begin_connecting (ProtobufC_RPC_Client *client,
                  struct sockaddr      *address,
                  size_t                addr_len)
{
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP);

  client->state = PROTOBUF_C_CLIENT_STATE_CONNECTING;
  client->fd = socket (address->sa_family, SOCK_STREAM, 0);
  if (client->fd < 0)
    {
      client_failed (client, "error creating socket: %s", strerror (errno));
      return;
    }
  set_fd_nonblocking (client->fd);
  if (connect (client->fd, address, addr_len) < 0)
    {
      if (errno == EINPROGRESS)
        {
          /* register interest in fd */
          protobuf_c_dispatch_watch_fd (client->dispatch,
                                        client->fd,
                                        PROTOBUF_C_EVENT_READABLE|PROTOBUF_C_EVENT_WRITABLE,
                                        handle_client_fd_connect_events,
                                        client);
          return;
        }
      close (client->fd);
      client->fd = -1;
      client_failed (client, "error connecting to remote host: %s", strerror (errno));
      return;
    }

  set_state_connected (client);
}
static void
handle_name_lookup_success (const uint8_t *address,
                            void          *callback_data)
{
  ProtobufC_RPC_Client *client = callback_data;
  struct sockaddr_in addr;
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP);
  protobuf_c_assert (client->info.name_lookup.pending);
  client->info.name_lookup.pending = 0;
  if (client->info.name_lookup.destroyed_while_pending)
    {
      destroy_client_rpc (&client->base_service);
      return;
    }
  memset (&addr, 0, sizeof (addr));
  addr.sin_family = AF_INET;
  memcpy (&addr.sin_addr, address, 4);
  addr.sin_port = htons (client->info.name_lookup.port);
  begin_connecting (client, (struct sockaddr *) &addr, sizeof (addr));
}

static void
handle_name_lookup_failure (const char    *error_message,
                            void          *callback_data)
{
  ProtobufC_RPC_Client *client = callback_data;
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP);
  protobuf_c_assert (client->info.name_lookup.pending);
  client->info.name_lookup.pending = 0;
  if (client->info.name_lookup.destroyed_while_pending)
    {
      destroy_client_rpc (&client->base_service);
      return;
    }
  client_failed (client, "name lookup failed (for name from %s): %s", client->name, error_message);
}

static void
begin_name_lookup (ProtobufC_RPC_Client *client)
{
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_INIT
                 ||  client->state == PROTOBUF_C_CLIENT_STATE_FAILED_WAITING
                 ||  client->state == PROTOBUF_C_CLIENT_STATE_FAILED);
  client->state = PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP;
  client->info.name_lookup.pending = 0;
  switch (client->address_type)
    {
    case PROTOBUF_C_RPC_ADDRESS_LOCAL:
      {
        struct sockaddr_un addr;
        addr.sun_family = AF_UNIX;
        strncpy (addr.sun_path, client->name, sizeof (addr.sun_path));
        begin_connecting (client, (struct sockaddr *) &addr,
                          sizeof (addr));
        return;
      }

    case PROTOBUF_C_RPC_ADDRESS_TCP:
      {
        /* parse hostname:port from client->name */
        const char *colon = strchr (client->name, ':');
        char *host;
        unsigned port;
        if (colon == NULL)
          {
            client_failed (client,
                           "name '%s' does not have a : in it (supposed to be HOST:PORT)",
                           client->name);
            return;
          }
        host = client->allocator->alloc (client->allocator, colon + 1 - client->name);
        memcpy (host, client->name, colon - client->name);
        host[colon - client->name] = 0;
        port = atoi (colon + 1);

        client->info.name_lookup.pending = 1;
        client->info.name_lookup.destroyed_while_pending = 0;
        client->info.name_lookup.port = port;
        client->resolver (client->dispatch,
                          host,
                          handle_name_lookup_success,
                          handle_name_lookup_failure,
                          client);

        /* cleanup */
        client->allocator->free (client->allocator, host);
        return;
      }
    default:
      assert (0);
    }
}

static void
handle_init_idle (ProtobufCDispatch *dispatch,
                  void              *data)
{
  ProtobufC_RPC_Client *client = data;
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_INIT);
  begin_name_lookup (client);
}

static void
grow_closure_array (ProtobufC_RPC_Client *client)
{
  /* resize array */
  unsigned old_size = client->info.connected.closures_alloced;
  unsigned new_size = old_size * 2;
  unsigned i;
  Closure *new_closures = client->allocator->alloc (client->allocator, sizeof (Closure) * new_size);
  memcpy (new_closures,
          client->info.connected.closures,
          sizeof (Closure) * old_size);

  /* build new free list */
  for (i = old_size; i < new_size - 1; i++)
    {
      new_closures[i].response_type = NULL;
      new_closures[i].closure = NULL;
      new_closures[i].closure_data = UINT_TO_POINTER (i+2);
    }
  new_closures[i].closure_data = UINT_TO_POINTER (client->info.connected.first_free_request_id);
  new_closures[i].response_type = NULL;
  new_closures[i].closure = NULL;
  client->info.connected.first_free_request_id = old_size + 1;

  client->allocator->free (client->allocator, client->info.connected.closures);
  client->info.connected.closures = new_closures;
  client->info.connected.closures_alloced = new_size;
}
static uint32_t 
uint32_to_le (uint32_t le)
{
#if !defined(WORDS_BIGENDIAN)
  return le;
#else
  return (le << 24) | (le >> 24)
       | ((le >> 8) & 0xff00)
       | ((le << 8) & 0xff0000);
#endif
}
#define uint32_from_le uint32_to_le             /* make the code more readable, i guess */

static void
enqueue_request (ProtobufC_RPC_Client *client,
                 unsigned          method_index,
                 const ProtobufCMessage *input,
                 ProtobufCClosure  closure,
                 void             *closure_data)
{
  uint32_t request_id;
  struct {
    uint32_t method_index;
    uint32_t packed_size;
    uint32_t request_id;
  } header;
  size_t packed_size;
  uint8_t *packed_data;
  Closure *cl;
  const ProtobufCServiceDescriptor *desc = client->base_service.descriptor;
  const ProtobufCMethodDescriptor *method = desc->methods + method_index;

  protobuf_c_assert (method_index < desc->n_methods);

  /* Allocate request_id */
  //protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_CONNECTED);
  if (client->info.connected.first_free_request_id == 0)
    grow_closure_array (client);
  request_id = client->info.connected.first_free_request_id;
  cl = client->info.connected.closures + (request_id - 1);
  client->info.connected.first_free_request_id = POINTER_TO_UINT (cl->closure_data);

  /* Pack message */
  packed_size = protobuf_c_message_get_packed_size (input);
  if (packed_size < client->allocator->max_alloca)
    packed_data = alloca (packed_size);
  else
    packed_data = client->allocator->alloc (client->allocator, packed_size);
  protobuf_c_message_pack (input, packed_data);

  /* Append to buffer */
  protobuf_c_assert (sizeof (header) == 12);
  header.method_index = uint32_to_le (method_index);
  header.packed_size = uint32_to_le (packed_size);
  header.request_id = request_id;
  protobuf_c_data_buffer_append (&client->outgoing, &header, 12);
  protobuf_c_data_buffer_append (&client->outgoing, packed_data, packed_size);

  /* Clean up if not using alloca() */
  if (packed_size >= client->allocator->max_alloca)
    client->allocator->free (client->allocator, packed_data);

  /* Add closure to request-tree */
  cl->response_type = method->output;
  cl->closure = closure;
  cl->closure_data = closure_data;
}

static void
handle_client_fd_events (int                fd,
                         unsigned           events,
                         void              *func_data)
{
  ProtobufC_RPC_Client *client = func_data;
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_CONNECTED);
  if (events & PROTOBUF_C_EVENT_WRITABLE)
    {
      int write_rv = protobuf_c_data_buffer_writev (&client->outgoing,
                                                    client->fd);
      if (write_rv < 0 && !errno_is_ignorable (errno))
        {
          client_failed (client,
                         "writing to file-descriptor: %s",
                         strerror (errno));
          return;
        }

      if (client->outgoing.size == 0)
        protobuf_c_dispatch_watch_fd (client->dispatch, client->fd,
                                      PROTOBUF_C_EVENT_READABLE,
                                      handle_client_fd_events, client);
    }
  if (events & PROTOBUF_C_EVENT_READABLE)
    {
      /* do read */
      int read_rv = protobuf_c_data_buffer_read_in_fd (&client->incoming,
                                                       client->fd);
      if (read_rv < 0)
        {
          if (!errno_is_ignorable (errno))
            {
              client_failed (client,
                             "reading from file-descriptor: %s",
                             strerror (errno));
            }
        }
      else if (read_rv == 0)
        {
          /* handle eof */
          client_failed (client,
                         "got end-of-file from server [%u bytes incoming, %u bytes outgoing]",
                         client->incoming.size, client->outgoing.size);
        }
      else
        {
          /* try processing buffer */
          while (client->incoming.size >= 16)
            {
              uint32_t header[4];
              unsigned status_code, method_index, message_length, request_id;
              Closure *closure;
              ProtobufCMessage *msg;
              protobuf_c_data_buffer_peek (&client->incoming, header, sizeof (header));
              status_code = uint32_from_le (header[0]);
              method_index = uint32_from_le (header[1]);
              message_length = uint32_from_le (header[2]);
              request_id = header[3];           /* already native-endian */

              if (16 + message_length > client->incoming.size)
                break;

              /* lookup request by id */
              if (request_id > client->info.connected.closures_alloced
               || request_id == 0
               || client->info.connected.closures[request_id-1].response_type == NULL)
                {
                  client_failed (client, "bad request-id in response from server");
                  return;
                }
              closure = client->info.connected.closures + (request_id - 1);

              /* Discard the RPC header */
              protobuf_c_data_buffer_discard (&client->incoming, 16);

              if (status_code == PROTOBUF_C_STATUS_CODE_SUCCESS)
              {
                 /* read message and unpack */
                 uint8_t *packed_data =
                    client->allocator->alloc (client->allocator, message_length);
                 protobuf_c_data_buffer_read (&client->incoming, packed_data, message_length);

                 /* TODO: use fast temporary allocator */
                 msg = protobuf_c_message_unpack (closure->response_type,
                       client->allocator,
                       message_length,
                       packed_data);

                 client->allocator->free (client->allocator, packed_data);
                 if (msg == NULL)
                 {
                    fprintf(stderr, "unable to unpack msg of length %u", message_length);
                    client_failed (client, "failed to unpack message");
                    return;
                 }
              }
              else
              {
                 /* Server did not send a response message */
                 protobuf_c_assert (message_length == 0);
                 msg = NULL;
              }

              /* invoke closure */
              closure->closure (msg, closure->closure_data);
              closure->response_type = NULL;
              closure->closure = NULL;
              closure->closure_data = UINT_TO_POINTER (client->info.connected.first_free_request_id);
              client->info.connected.first_free_request_id = request_id;

              /* clean up */
              if (msg)
                 protobuf_c_message_free_unpacked (msg, client->allocator);
            }
        }
    }
}

static void
update_connected_client_watch (ProtobufC_RPC_Client *client)
{
  unsigned events = PROTOBUF_C_EVENT_READABLE;
  protobuf_c_assert (client->state == PROTOBUF_C_CLIENT_STATE_CONNECTED);
  protobuf_c_assert (client->fd >= 0);
  if (client->outgoing.size > 0)
    events |= PROTOBUF_C_EVENT_WRITABLE;
  protobuf_c_dispatch_watch_fd (client->dispatch,
                                client->fd,
                                events,
                                handle_client_fd_events, client);
}

static void
invoke_client_rpc (ProtobufCService *service,
                   unsigned          method_index,
                   const ProtobufCMessage *input,
                   ProtobufCClosure  closure,
                   void             *closure_data)
{
  ProtobufC_RPC_Client *client = (ProtobufC_RPC_Client *) service;
  protobuf_c_assert (service->invoke == invoke_client_rpc);
  switch (client->state)
    {
    case PROTOBUF_C_CLIENT_STATE_INIT:
    case PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP:
    case PROTOBUF_C_CLIENT_STATE_CONNECTING:
      enqueue_request (client, method_index, input, closure, closure_data);
      break;
      
    case PROTOBUF_C_CLIENT_STATE_CONNECTED:
      {
        int had_outgoing = (client->outgoing.size > 0);
        enqueue_request (client, method_index, input, closure, closure_data);
        if (!had_outgoing)
          update_connected_client_watch (client);
      }
      break;

    case PROTOBUF_C_CLIENT_STATE_FAILED_WAITING:
    case PROTOBUF_C_CLIENT_STATE_FAILED:
    case PROTOBUF_C_CLIENT_STATE_DESTROYED:
      closure (NULL, closure_data);
      break;
    }
}

static void
destroy_client_rpc (ProtobufCService *service)
{
  ProtobufC_RPC_Client *client = (ProtobufC_RPC_Client *) service;
  ProtobufC_RPC_ClientState state = client->state;
  unsigned i;
  unsigned n_closures = 0;
  Closure *closures = NULL;
  switch (state)
    {
    case PROTOBUF_C_CLIENT_STATE_INIT:
      protobuf_c_dispatch_remove_idle (client->info.init.idle);
      break;
    case PROTOBUF_C_CLIENT_STATE_NAME_LOOKUP:
      if (client->info.name_lookup.pending)
        {
          client->info.name_lookup.destroyed_while_pending = 1;
          return;
        }
      break;
    case PROTOBUF_C_CLIENT_STATE_CONNECTING:
      break;
    case PROTOBUF_C_CLIENT_STATE_CONNECTED:
      n_closures = client->info.connected.closures_alloced;
      closures = client->info.connected.closures;
      break;
    case PROTOBUF_C_CLIENT_STATE_FAILED_WAITING:
      protobuf_c_dispatch_remove_timer (client->info.failed_waiting.timer);
      client->allocator->free (client->allocator, client->info.failed_waiting.timer);
      client->allocator->free (client->allocator, client->info.failed_waiting.error_message);
      break;
    case PROTOBUF_C_CLIENT_STATE_FAILED:
      client->allocator->free (client->allocator, client->info.failed.error_message);
      break;
    case PROTOBUF_C_CLIENT_STATE_DESTROYED:
      protobuf_c_assert (0);
      break;
    }
  if (client->fd >= 0)
    {
      protobuf_c_dispatch_close_fd (client->dispatch, client->fd);
      client->fd = -1;
    }
  protobuf_c_data_buffer_clear (&client->incoming);
  protobuf_c_data_buffer_clear (&client->outgoing);
  client->state = PROTOBUF_C_CLIENT_STATE_DESTROYED;
  client->allocator->free (client->allocator, client->name);

  /* free closures only once we are in the destroyed state */
  for (i = 0; i < n_closures; i++)
    if (closures[i].response_type != NULL)
      closures[i].closure (NULL, closures[i].closure_data);
  if (closures)
    client->allocator->free (client->allocator, closures);

  client->allocator->free (client->allocator, client);
}

static void
trivial_sync_libc_resolver (ProtobufCDispatch *dispatch,
                            const char        *name,
                            ProtobufC_NameLookup_Found found_func,
                            ProtobufC_NameLookup_Failed failed_func,
                            void *callback_data)
{
  struct hostent *ent;
  ent = gethostbyname (name);
  if (ent == NULL)
    failed_func (hstrerror (h_errno), callback_data);
  else
    found_func ((const uint8_t *) ent->h_addr_list[0], callback_data);
}

ProtobufCService *protobuf_c_rpc_client_new (ProtobufC_RPC_AddressType type,
                                             const char               *name,
                                             const ProtobufCServiceDescriptor *descriptor,
                                             ProtobufCDispatch       *orig_dispatch)
{
  ProtobufCDispatch *dispatch = orig_dispatch ? orig_dispatch : protobuf_c_dispatch_default ();
  ProtobufCAllocator *allocator = protobuf_c_dispatch_peek_allocator (dispatch);
  ProtobufC_RPC_Client *rv = allocator->alloc (allocator, sizeof (ProtobufC_RPC_Client));
  rv->base_service.descriptor = descriptor;
  rv->base_service.invoke = invoke_client_rpc;
  rv->base_service.destroy = destroy_client_rpc;
  protobuf_c_data_buffer_init (&rv->incoming, allocator);
  protobuf_c_data_buffer_init (&rv->outgoing, allocator);
  rv->allocator = allocator;
  rv->dispatch = dispatch;
  rv->address_type = type;
  rv->name = strcpy (allocator->alloc (allocator, strlen (name) + 1), name);
  rv->state = PROTOBUF_C_CLIENT_STATE_INIT;
  rv->fd = -1;
  rv->autoreconnect = 1;
  rv->autoreconnect_millis = 2*1000;
  rv->resolver = trivial_sync_libc_resolver;
  rv->error_handler = error_handler;
  rv->error_handler_data = "protobuf-c rpc client";
  rv->info.init.idle = protobuf_c_dispatch_add_idle (dispatch, handle_init_idle, rv);
  return &rv->base_service;
}

protobuf_c_boolean
protobuf_c_rpc_client_is_connected (ProtobufC_RPC_Client *client)
{
  return client->state == PROTOBUF_C_CLIENT_STATE_CONNECTED;
}

void
protobuf_c_rpc_client_set_autoreconnect_period (ProtobufC_RPC_Client *client,
                                            unsigned millis)
{
  client->autoreconnect = 1;
  client->autoreconnect_millis = millis;
}


void
protobuf_c_rpc_client_set_error_handler (ProtobufC_RPC_Client *client,
                                         ProtobufC_RPC_Error_Func func,
                                         void                    *func_data)
{
  client->error_handler = func;
  client->error_handler_data = func_data;
}

void
protobuf_c_rpc_client_disable_autoreconnect (ProtobufC_RPC_Client *client)
{
  client->autoreconnect = 0;
}

/* === Server === */
typedef struct _ServerRequest ServerRequest;
typedef struct _ServerConnection ServerConnection;
struct _ServerRequest
{
  uint32_t request_id;                  /* in little-endian */
  uint32_t method_index;                /* in native-endian */
  ServerConnection *conn;
  ProtobufC_RPC_Server *server;
  union {
    /* if conn != NULL, then the request is alive: */
    struct { ServerRequest *prev, *next; } alive;

    /* if conn == NULL, then the request is defunct: */
    struct { ProtobufCAllocator *allocator; } defunct;

    /* well, if it is in the recycled list, then it's recycled :/ */
    struct { ServerRequest *next; } recycled;
  } info;
};
struct _ServerConnection
{
  int fd;
  ProtobufCDataBuffer incoming, outgoing;

  ProtobufC_RPC_Server *server;
  ServerConnection *prev, *next;

  unsigned n_pending_requests;
  ServerRequest *first_pending_request, *last_pending_request;
};


/* When we get a response in the wrong thread,
   we proxy it over a system pipe.  Actually, we allocate one
   of these structures and pass the pointer over the pipe.  */
typedef struct _ProxyResponse ProxyResponse;
struct _ProxyResponse
{
  ServerRequest *request;
  unsigned len;
  /* data follows the structure */
};

struct _ProtobufC_RPC_Server
{
  ProtobufCDispatch *dispatch;
  ProtobufCAllocator *allocator;
  ProtobufCService *underlying;
  ProtobufC_RPC_AddressType address_type;
  char *bind_name;
  ServerConnection *first_connection, *last_connection;
  ProtobufC_FD listening_fd;

  ServerRequest *recycled_requests;

  /* multithreading support */
  ProtobufC_RPC_IsRpcThreadFunc is_rpc_thread_func;
  void * is_rpc_thread_data;
  int proxy_pipe[2];
  unsigned proxy_extra_data_len;
  uint8_t proxy_extra_data[sizeof (void*)];

  ProtobufC_RPC_Error_Func error_handler;
  void *error_handler_data;

  /* configuration */
  unsigned max_pending_requests_per_connection;
};

#define GET_PENDING_REQUEST_LIST(conn) \
  ServerRequest *, conn->first_pending_request, conn->last_pending_request, info.alive.prev, info.alive.next
#define GET_CONNECTION_LIST(server) \
  ServerConnection *, server->first_connection, server->last_connection, prev, next

static void
server_connection_close (ServerConnection *conn)
{
  ProtobufCAllocator *allocator = conn->server->allocator;

  /* general cleanup */
  protobuf_c_dispatch_close_fd (conn->server->dispatch, conn->fd);
  conn->fd = -1;
  protobuf_c_data_buffer_clear (&conn->incoming);
  protobuf_c_data_buffer_clear (&conn->outgoing);

  /* remove this connection from the server's list */
  GSK_LIST_REMOVE (GET_CONNECTION_LIST (conn->server), conn);

  /* disassocate all the requests from the connection */
  while (conn->first_pending_request != NULL)
    {
      ServerRequest *req = conn->first_pending_request;
      conn->first_pending_request = req->info.alive.next;
      req->conn = NULL;
      req->info.defunct.allocator = allocator;
    }

  /* free the connection itself */
  allocator->free (allocator, conn);
}

static void
server_failed_literal (ProtobufC_RPC_Server *server,
                       ProtobufC_RPC_Error_Code code,
                       const char *msg)
{
  if (server->error_handler != NULL)
    server->error_handler (code, msg, server->error_handler_data);
}

#if 0
static void
server_failed (ProtobufC_RPC_Server *server,
               ProtobufC_RPC_Error_Code code,
               const char           *format,
               ...)
{
  va_list args;
  char buf[MAX_FAILED_MSG_LENGTH];
  va_start (args, format);
  vsnprintf (buf, sizeof (buf), format, args);
  buf[sizeof(buf)-1] = 0;
  va_end (args);

  server_failed_literal (server, code, buf);
}
#endif

static protobuf_c_boolean
address_to_name (const struct sockaddr *addr,
                 unsigned               addr_len,
                 char                  *name_out,
                 unsigned               name_out_buf_length)
{
  if (addr->sa_family == PF_INET)
    {
      /* convert to dotted address + port */
      const struct sockaddr_in *addr_in = (const struct sockaddr_in *) addr;
      const uint8_t *addr = (const uint8_t *) &(addr_in->sin_addr);
      uint16_t port = htons (addr_in->sin_port);
      snprintf (name_out, name_out_buf_length,
                "%u.%u.%u.%u:%u",
                addr[0], addr[1], addr[2], addr[3], port);
      return TRUE;
    }
  return FALSE;
}

static void
server_connection_failed (ServerConnection *conn,
                          ProtobufC_RPC_Error_Code code,
                          const char       *format,
                          ...)
{
  char remote_addr_name[64];
  char msg[MAX_FAILED_MSG_LENGTH];
  char *msg_end = msg + sizeof (msg);
  char *msg_at;
  struct sockaddr addr;
  socklen_t addr_len = sizeof (addr);
  va_list args;

  /* if we can, find the remote name of this connection */
  if (getpeername (conn->fd, &addr, &addr_len) == 0
   && address_to_name (&addr, addr_len, remote_addr_name, sizeof (remote_addr_name)))
    snprintf (msg, sizeof (msg), "connection to %s from %s: ",
              conn->server->bind_name, remote_addr_name);
  else
    snprintf (msg, sizeof (msg), "connection to %s: ",
              conn->server->bind_name);
  msg[sizeof(msg)-1] = 0;
  msg_at = strchr (msg, 0);

  /* do vsnprintf() */
  va_start (args, format);
  vsnprintf(msg_at, msg_end - msg_at, format, args);
  va_end (args);
  msg[sizeof(msg)-1] = 0;

  /* invoke server error hook */
  server_failed_literal (conn->server, code, msg);

  server_connection_close (conn);
}

static ServerRequest *
create_server_request (ServerConnection *conn,
                       uint32_t          request_id,
                       uint32_t          method_index)
{
  ServerRequest *rv;
  if (conn->server->recycled_requests != NULL)
    {
      rv = conn->server->recycled_requests;
      conn->server->recycled_requests = rv->info.recycled.next;
    }
  else
    {
      ProtobufCAllocator *allocator = conn->server->allocator;
      rv = allocator->alloc (allocator, sizeof (ServerRequest));
    }
  rv->server = conn->server;
  rv->conn = conn;
  rv->request_id = request_id;
  rv->method_index = method_index;
  conn->n_pending_requests++;
  GSK_LIST_APPEND (GET_PENDING_REQUEST_LIST (conn), rv);
  return rv;
}

static void
free_server_request (ProtobufC_RPC_Server *server,
                     ServerRequest        *request)
{
#if RECYCLE_REQUESTS
  /* recycle request */
  request->info.recycled.next = server->recycled_requests;
  server->recycled_requests = request;
#else
  /* free the request immediately */
  server->allocator->free (server->allocator, request);
#endif
}


static void handle_server_connection_events (int fd,
                                             unsigned events,
                                             void *data);
static void
server_connection_response_closure (const ProtobufCMessage *message,
                                    void                   *closure_data)
{
  ServerRequest *request = closure_data;
  ProtobufC_RPC_Server *server = request->server;
  protobuf_c_boolean must_proxy = 0;
  ProtobufCAllocator *allocator = server->allocator;
  if (server->is_rpc_thread_func != NULL)
    {
      must_proxy = !server->is_rpc_thread_func (server,
                                                server->dispatch,
                                                server->is_rpc_thread_data);
    }


  uint8_t buffer_slab[512];
  ProtobufCBufferSimple buffer_simple = PROTOBUF_C_BUFFER_SIMPLE_INIT (buffer_slab);
  if (!protobuf_c_message_check (message))
    {
      /* send failed status */
      uint32_t header[4];
      header[0] = uint32_to_le (PROTOBUF_C_STATUS_CODE_SERVICE_FAILED);
      header[1] = uint32_to_le (request->method_index);
      header[2] = 0;            /* no message */
      header[3] = request->request_id;
      protobuf_c_buffer_simple_append (&buffer_simple.base,
                                       16, (uint8_t *) header);
      server_failed_literal (server, PROTOBUF_C_ERROR_CODE_BAD_REQUEST,
            "response message was malformed, sending failed status to client");
    }
  else
    {
      /* send success response */
      uint32_t header[4];
      header[0] = uint32_to_le (PROTOBUF_C_STATUS_CODE_SUCCESS);
      header[1] = uint32_to_le (request->method_index);
      header[3] = request->request_id;
      protobuf_c_buffer_simple_append (&buffer_simple.base,
                                       16, (uint8_t *) header);
      protobuf_c_message_pack_to_buffer (message, &buffer_simple.base);
      ((uint32_t *)buffer_simple.data)[2] = uint32_to_le (buffer_simple.len - 16);
    }

  if (must_proxy)
    {
      ProxyResponse *pr = allocator->alloc (allocator, sizeof (ProxyResponse) + buffer_simple.len);
      int rv;
      pr->request = request;
      pr->len = buffer_simple.len;
      memcpy (pr + 1, buffer_simple.data, buffer_simple.len);

      /* write pointer to proxy pipe */
retry_write:
      rv = write (server->proxy_pipe[1], &pr, sizeof (void*));
      if (rv < 0)
        {
          if (errno == EINTR || errno == EAGAIN)
            goto retry_write;
          server_failed_literal (server, PROTOBUF_C_ERROR_CODE_PROXY_PROBLEM,
                                 "error writing to proxy-pipe");
          allocator->free (allocator, pr);
        }
      else if (rv < sizeof (void *))
        {
          server_failed_literal (server, PROTOBUF_C_ERROR_CODE_PROXY_PROBLEM,
                                 "partial write to proxy-pipe");
          allocator->free (allocator, pr);
        }
    }
  else if (request->conn == NULL)
    {
      /* defunct request */
      allocator->free (allocator, request);
      free_server_request (server, request);
    }
  else
    {
      ServerConnection *conn = request->conn;
      protobuf_c_boolean must_set_output_watch = (conn->outgoing.size == 0);
      protobuf_c_data_buffer_append (&conn->outgoing, buffer_simple.data, buffer_simple.len);
      if (must_set_output_watch)
        protobuf_c_dispatch_watch_fd (conn->server->dispatch,
                                      conn->fd,
                                      PROTOBUF_C_EVENT_READABLE|PROTOBUF_C_EVENT_WRITABLE,
                                      handle_server_connection_events,
                                      conn);
      GSK_LIST_REMOVE (GET_PENDING_REQUEST_LIST (conn), request);
      conn->n_pending_requests--;

      free_server_request (server, request);
    }
  PROTOBUF_C_BUFFER_SIMPLE_CLEAR (&buffer_simple);
}

static void
handle_server_connection_events (int fd,
                                 unsigned events,
                                 void *data)
{
  ServerConnection *conn = data;
  ProtobufCService *service = conn->server->underlying;
  ProtobufCAllocator *allocator = conn->server->allocator;
  if (events & PROTOBUF_C_EVENT_READABLE)
    {
      int read_rv = protobuf_c_data_buffer_read_in_fd (&conn->incoming, fd);
      if (read_rv < 0)
        {
          if (!errno_is_ignorable (errno))
            {
              server_connection_failed (conn,
                                        PROTOBUF_C_ERROR_CODE_CLIENT_TERMINATED,
                                        "reading from file-descriptor: %s",
                                        strerror (errno));
              return;
            }
        }
      else if (read_rv == 0)
        {
          if (conn->first_pending_request != NULL)
            server_connection_failed (conn,
                                      PROTOBUF_C_ERROR_CODE_CLIENT_TERMINATED,
                                      "closed while calls pending");
          else
            server_connection_close (conn);
          return;
        }
      else
        while (conn->incoming.size >= 12)
          {
            uint32_t header[3];
            uint32_t method_index, message_length, request_id;
            uint8_t *packed_data;
            ProtobufCMessage *message;
            ServerRequest *server_request;
            protobuf_c_data_buffer_peek (&conn->incoming, header, 12);
            method_index = uint32_from_le (header[0]);
            message_length = uint32_from_le (header[1]);
            request_id = header[2];             /* store in whatever endianness it comes in */

            if (conn->incoming.size < 12 + message_length)
              break;

            if (method_index >= conn->server->underlying->descriptor->n_methods)
              {
                server_connection_failed (conn,
                                          PROTOBUF_C_ERROR_CODE_BAD_REQUEST,
                                          "bad method_index %u", method_index);
                return;
              }

            /* Read message */
            protobuf_c_data_buffer_discard (&conn->incoming, 12);
            packed_data = allocator->alloc (allocator, message_length);
            protobuf_c_data_buffer_read (&conn->incoming, packed_data, message_length);

            /* Unpack message */
            message = protobuf_c_message_unpack (service->descriptor->methods[method_index].input,
                                                 allocator, message_length, packed_data);
            allocator->free (allocator, packed_data);
            if (message == NULL)
              {
                server_connection_failed (conn,
                                          PROTOBUF_C_ERROR_CODE_BAD_REQUEST,
                                          "error unpacking message");
                return;
              }

            /* Invoke service (note that it may call back immediately) */
            server_request = create_server_request (conn, request_id, method_index);
            service->invoke (service, method_index, message,
                             server_connection_response_closure, server_request);
            protobuf_c_message_free_unpacked (message, allocator);
          }
    }
  if ((events & PROTOBUF_C_EVENT_WRITABLE) != 0
    && conn->outgoing.size > 0)
    {
      int write_rv = protobuf_c_data_buffer_writev (&conn->outgoing, fd);
      if (write_rv < 0)
        {
          if (!errno_is_ignorable (errno))
            {
              server_connection_failed (conn,
                                        PROTOBUF_C_ERROR_CODE_CLIENT_TERMINATED,
                                        "writing to file-descriptor: %s",
                                        strerror (errno));
              return;
            }
        }
      if (conn->outgoing.size == 0)
        protobuf_c_dispatch_watch_fd (conn->server->dispatch, conn->fd, PROTOBUF_C_EVENT_READABLE,
                                      handle_server_connection_events, conn);
    }
}

static void
handle_server_listener_readable (int fd,
                                 unsigned events,
                                 void *data)
{
  ProtobufC_RPC_Server *server = data;
  struct sockaddr addr;
  socklen_t addr_len = sizeof (addr);
  int new_fd = accept (fd, &addr, &addr_len);
  ServerConnection *conn;
  ProtobufCAllocator *allocator = server->allocator;
  if (new_fd < 0)
    {
      if (errno_is_ignorable (errno))
        return;
      fprintf (stderr, "error accept()ing file descriptor: %s\n",
               strerror (errno));
      return;
    }
  conn = allocator->alloc (allocator, sizeof (ServerConnection));
  conn->fd = new_fd;
  protobuf_c_data_buffer_init (&conn->incoming, server->allocator);
  protobuf_c_data_buffer_init (&conn->outgoing, server->allocator);
  conn->n_pending_requests = 0;
  conn->first_pending_request = conn->last_pending_request = NULL;
  conn->server = server;
  GSK_LIST_APPEND (GET_CONNECTION_LIST (server), conn);
  protobuf_c_dispatch_watch_fd (server->dispatch, conn->fd, PROTOBUF_C_EVENT_READABLE,
                                handle_server_connection_events, conn);
}

static ProtobufC_RPC_Server *
server_new_from_fd (ProtobufC_FD              listening_fd,
                    ProtobufC_RPC_AddressType address_type,
                    const char               *bind_name,
                    ProtobufCService         *service,
                    ProtobufCDispatch       *orig_dispatch)
{
  ProtobufCDispatch *dispatch = orig_dispatch ? orig_dispatch : protobuf_c_dispatch_default ();
  ProtobufCAllocator *allocator = protobuf_c_dispatch_peek_allocator (dispatch);
  ProtobufC_RPC_Server *server = allocator->alloc (allocator, sizeof (ProtobufC_RPC_Server));
  server->dispatch = dispatch;
  server->allocator = allocator;
  server->underlying = service;
  server->first_connection = server->last_connection = NULL;
  server->max_pending_requests_per_connection = 32;
  server->address_type = address_type;
  server->bind_name = allocator->alloc (allocator, strlen (bind_name) + 1);
  server->error_handler = error_handler;
  server->error_handler_data = "protobuf-c rpc server";
  server->listening_fd = listening_fd;
  server->recycled_requests = NULL;
  server->is_rpc_thread_func = NULL;
  server->is_rpc_thread_data = NULL;
  server->proxy_pipe[0] = server->proxy_pipe[1] = -1;
  server->proxy_extra_data_len = 0;
  strcpy (server->bind_name, bind_name);
  set_fd_nonblocking (listening_fd);
  protobuf_c_dispatch_watch_fd (dispatch, listening_fd, PROTOBUF_C_EVENT_READABLE, 
                                handle_server_listener_readable, server);
  return server;
}

/* this function is for handling the common problem 
   that we bind over-and-over again to the same
   unix path.

   ideally, you'd think the OS's SO_REUSEADDR flag would
   cause this to happen, but it doesn't,
   at least on my linux 2.6 box.

   in fact, we really need a way to test without
   actually connecting to the remote server,
   which might annoy it.

   XXX: we should survey what others do here... like x-windows...
 */
/* NOTE: stolen from gsk, obviously */
static void
_gsk_socket_address_local_maybe_delete_stale_socket (const char *path,
                                                     struct sockaddr *addr,
                                                     unsigned addr_len)
{
  int fd;
  struct stat statbuf;
  if (stat (path, &statbuf) < 0)
    return;
  if (!S_ISSOCK (statbuf.st_mode))
    {
      fprintf (stderr, "%s existed but was not a socket\n", path);
      return;
    }

  fd = socket (PF_UNIX, SOCK_STREAM, 0);
  if (fd < 0)
    return;
  set_fd_nonblocking (fd);
  if (connect (fd, addr, addr_len) < 0)
    {
      if (errno == EINPROGRESS)
        {
          close (fd);
          return;
        }
    }
  else
    {
      close (fd);
      return;
    }

  /* ok, we should delete the stale socket */
  close (fd);
  if (unlink (path) < 0)
    fprintf (stderr, "unable to delete %s: %s\n",
             path, strerror(errno));
}
ProtobufC_RPC_Server *
protobuf_c_rpc_server_new       (ProtobufC_RPC_AddressType type,
                                 const char               *name,
                                 ProtobufCService         *service,
                                 ProtobufCDispatch       *dispatch)
{
  int fd = -1;
  int protocol_family;
  struct sockaddr *address;
  socklen_t address_len;
  struct sockaddr_un addr_un;
  struct sockaddr_in addr_in;
  switch (type)
    {
    case PROTOBUF_C_RPC_ADDRESS_LOCAL:
      protocol_family = PF_UNIX;
      memset (&addr_un, 0, sizeof (addr_un));
      addr_un.sun_family = AF_UNIX;
      strncpy (addr_un.sun_path, name, sizeof (addr_un.sun_path));
      address_len = sizeof (addr_un);
      address = (struct sockaddr *) (&addr_un);
      _gsk_socket_address_local_maybe_delete_stale_socket (name,
                                                           address,
                                                           address_len);
      break;
    case PROTOBUF_C_RPC_ADDRESS_TCP:
      protocol_family = PF_INET;
      memset (&addr_in, 0, sizeof (addr_in));
      addr_in.sin_family = AF_INET;
      {
        unsigned port = atoi (name);
        addr_in.sin_port = htons (port);
      }
      address_len = sizeof (addr_in);
      address = (struct sockaddr *) (&addr_in);
      break;
    default:
    protobuf_c_assert (0);
    }

  fd = socket (protocol_family, SOCK_STREAM, 0);
  if (fd < 0)
    {
      fprintf (stderr, "protobuf_c_rpc_server_new: socket() failed: %s\n",
               strerror (errno));
      return NULL;
    }
  if (bind (fd, address, address_len) < 0)
    {
      fprintf (stderr, "protobuf_c_rpc_server_new: error binding to port: %s\n",
               strerror (errno));
      return NULL;
    }
  if (listen (fd, 255) < 0)
    {
      fprintf (stderr, "protobuf_c_rpc_server_new: listen() failed: %s\n",
               strerror (errno));
      return NULL;
    }
  return server_new_from_fd (fd, type, name, service, dispatch);
}

ProtobufCService *
protobuf_c_rpc_server_destroy (ProtobufC_RPC_Server *server,
                               protobuf_c_boolean    destroy_underlying)
{
  ProtobufCService *rv = destroy_underlying ? NULL : server->underlying;
  while (server->first_connection != NULL)
    server_connection_close (server->first_connection);

  if (server->address_type == PROTOBUF_C_RPC_ADDRESS_LOCAL)
    unlink (server->bind_name);
  server->allocator->free (server->allocator, server->bind_name);

  while (server->recycled_requests != NULL)
    {
      ServerRequest *req = server->recycled_requests;
      server->recycled_requests = req->info.recycled.next;
      server->allocator->free (server->allocator, req);
    }

  protobuf_c_dispatch_close_fd (server->dispatch, server->listening_fd);

  if (destroy_underlying)
    protobuf_c_service_destroy (server->underlying);

  server->allocator->free (server->allocator, server);

  return rv;
}

/* Number of proxied requests to try to grab in a single read */
#define PROXY_BUF_SIZE   256
static void
handle_proxy_pipe_readable (ProtobufC_FD   fd,
                            unsigned       events,
                            void          *callback_data)
{
  int nread;
  ProtobufC_RPC_Server *server = callback_data;
  ProtobufCAllocator *allocator = server->allocator;
  union {
    char buf[sizeof(void*) * PROXY_BUF_SIZE];
    ProxyResponse *responses[PROXY_BUF_SIZE];
  } u;
  unsigned amt, i;
  memcpy (u.buf, server->proxy_extra_data, server->proxy_extra_data_len);
  nread = read (fd, u.buf + server->proxy_extra_data_len, sizeof (u.buf) - server->proxy_extra_data_len);
  if (nread <= 0)
    return;             /* TODO: handle 0 and non-retryable errors separately */
  amt = server->proxy_extra_data_len + nread;
  for (i = 0; i < amt / sizeof(void*); i++)
    {
      ProxyResponse *pr = u.responses[i];
      ServerRequest *request = pr->request;
      if (request->conn == NULL)
        {
          /* defunct request */
          allocator->free (allocator, request);
        }
      else
        {
          ServerConnection *conn = request->conn;
          protobuf_c_boolean must_set_output_watch = (conn->outgoing.size == 0);
          protobuf_c_data_buffer_append (&conn->outgoing, (uint8_t*)(pr+1), pr->len);
          if (must_set_output_watch)
            protobuf_c_dispatch_watch_fd (conn->server->dispatch,
                                          conn->fd,
                                          PROTOBUF_C_EVENT_READABLE|PROTOBUF_C_EVENT_WRITABLE,
                                          handle_server_connection_events,
                                          conn);
          GSK_LIST_REMOVE (GET_PENDING_REQUEST_LIST (conn), request);
          conn->n_pending_requests--;

          free_server_request (conn->server, request);
        }
      allocator->free (allocator, pr);
    }
  memcpy (server->proxy_extra_data, u.buf + i * sizeof(void*), amt - i * sizeof(void*));
  server->proxy_extra_data_len = amt - i * sizeof(void*);
}

void
protobuf_c_rpc_server_configure_threading (ProtobufC_RPC_Server *server,
                                           ProtobufC_RPC_IsRpcThreadFunc func,
                                           void          *is_rpc_data)
{
  server->is_rpc_thread_func = func;
  server->is_rpc_thread_data = is_rpc_data;
retry_pipe:
  if (pipe (server->proxy_pipe) < 0)
    {
      if (errno == EINTR || errno == EAGAIN)
        goto retry_pipe;
      server_failed_literal (server, PROTOBUF_C_ERROR_CODE_PROXY_PROBLEM,
                             "error creating pipe for thread-proxying");
      return;
    }

  /* make the read side non-blocking, since we will use it from the main-loop;
     leave the write side blocking, since it will be used from foreign threads */
  set_fd_nonblocking (server->proxy_pipe[0]);
  protobuf_c_dispatch_watch_fd (server->dispatch, server->proxy_pipe[0],
                                PROTOBUF_C_EVENT_READABLE,
                                handle_proxy_pipe_readable, server);
}

void
protobuf_c_rpc_server_set_error_handler (ProtobufC_RPC_Server *server,
                                         ProtobufC_RPC_Error_Func func,
                                         void                 *error_func_data)
{
  server->error_handler = func;
  server->error_handler_data = error_func_data;
}