Re: Asynchronous BEEP draft status

On Wed, Nov 26, 2008 at 10:17 AM, Sam Roberts <[email protected]> wrote:
> a - This draft describes how to build into BEEP something you can
> already do with BEEP (several ways).
>
> I can write a trivial example using vortex that implements the
> motivating use-case for those unconvinced.

Sorry. Couldn't resist implementing this.

Draft states following use case:

"""
   Asynchronous applications require a protocol that is able to support
   a large number of concurrent outstanding requests.  The analogy of a
   channel as a thread does not scale to the large number of threads
   used in modern systems.  Modern applications regularly have large
   numbers of concurrent processing threads.  Thus, a better way of
   multiplexing large numbers of concurrent requests is required.

   This document describes an BEEP feature, an extension to BEEP, that
   enables the creation of an asynchronous channel.  An asynchronous
   channel is a channel where response ordering is not fixed to the
   order of the requests sent by the client peer.  An asynchronous
   channel is identical to other channels, using unmodified framing;
   only requests may be processed in parallel and responses may be sent
   in any order.
"""


Note that Vortex delivers each MSG on a seperate thread, exactly as
described above.

BEEP msgnos are ordered across a channel, of course, but application
msgs can be replied whenever you want. The profile designer specifies
the content and meaning of beep message payload.

Profile designers are not required to map BEEP internal msgno 1-to-1
to their profile's message identifiers. Often, this is convenient.
When it is not convenient, don't do it.


Server:

#include <vortex.h>

#define PROFILE "http://example.com/beep/delayed"
/*
MSG payload is msgid

server waits a random amount of time, and sends reply.

RPY payload is "msgid delay", where msgid is the msgid of the request being
responded to, and delay is how long the server delayed before sending the
reply.

Actual code might put msgid in a MIME header, and the payload would be the
request information. Or the request would be xml encoded, and include a msgid.

There are lots of variations of this, none of them requiring extending the BEEP
protocol.
*/

VortexQueue* msgno_queue;

void frame_received(VortexChannel* chan, VortexConnection* conn,
VortexFrame* frame, void*v)
{
  int msgno = vortex_frame_get_msgno (frame);
  int delay = (rand() % 30) + 1;
  char* msgid = vortex_frame_get_payload(frame);
  char str[128];

  vortex_queue_push(msgno_queue, INT_TO_PTR(msgno));

  printf("msgno %d msgid %s delay %d\n", msgno, msgid, delay);

  sleep(delay);

  msgno = PTR_TO_INT(vortex_queue_pop(msgno_queue));

  sprintf(str, "%s %d", msgid, delay);

  vortex_channel_send_rpy(chan, str, strlen(str), msgno);
}

const char USAGE[] = "usage: %s <port>";

int main(int argc, char ** argv)
{
  if(!argv[1]) {
    printf("%s\n", USAGE);
    return 1;
  }

  msgno_queue = vortex_queue_new();

  vortex_init();

  vortex_log_enable(1);

  vortex_profiles_register(PROFILE, NULL, NULL, NULL, NULL,
frame_received, NULL);

  vortex_listener_new("0.0.0.0", argv[1], NULL, NULL);

  vortex_listener_wait();

  vortex_exit();

  return 0;
}



Client:

#include <vortex.h>
#include <assert.h>

#define PROFILE "http://example.com/beep/delayed"

int msgid = 0;

void send_msg(VortexChannel* channel)
{
  int ok = vortex_channel_send_msgv(channel, 0, "%d", ++msgid);
  assert(ok);
}

void on_frame(VortexChannel* channel, VortexConnection* conn,
VortexFrame* frame, void* v)
{
  int msgno = vortex_frame_get_msgno(frame);
  const char* content = vortex_frame_get_payload(frame);

  printf("(msgno %d) %s\n", msgno, content);

  send_msg(channel);
}

int main (int argc, char ** argv)
{
  VortexConnection * connection = NULL;
  VortexChannel * channel = NULL;

  vortex_init ();

  connection = vortex_connection_new(argv[1], argv[2], NULL, NULL);

  if (!vortex_connection_is_ok(connection, false)) {
    fprintf(stderr, "Unable to connect remote server, error was: %s\n",
	vortex_connection_get_message(connection));
    return 1;
  }

  channel = vortex_channel_new(connection, 0,
      PROFILE,
      NULL, NULL, /* no close handling */
      on_frame, NULL,
      NULL, NULL /* no async channel creation */
      );

  if (channel == NULL) {
    fprintf(stderr, "Unable to create the channel..\n");
    return 1;
  }

  printf(".. send msg\n");

  send_msg(channel);
  send_msg(channel);
  send_msg(channel);
  send_msg(channel);
  send_msg(channel);

  {
    char c;
    read(0, &c, 1);
  }
  vortex_exit ();

  return 0 ;
}


Client output:

% ./async-client localhost 3333
.. send msg
(msgno 0) 3 28
(msgno 1) 1 14
(msgno 2) 2 17
(msgno 3) 5 24
(msgno 4) 4 26
(msgno 5) 10 2
(msgno 6) 11 3
(msgno 7) 9 10
(msgno 8) 8 13
(msgno 9) 12 8


Not BEEP msgno order is maintained, application msg order is asynchronous.

Cheers,
Sam