SimGrid: Lesson 11: Explicitely waiting for messages

Introduction

The messaging primitive we have seen so far are somehow limited on the receiver side. You have to attach a callback to a message type, and then go into an infinite loop. Sometimes, you want to block your execution until a message of a given type arrives. This often occures when you want to deal with synchronization problems.

As an example, we will study a simple centralized algorithm for mutual exclusion. In short, when the client wants to enter the critical section (CS), it sends a specific message to the server, and waits for the server to answer back with another message. While the client is "locked" waiting for the message, nothing else should occure for it (no callback or timer should be served).

The simplest interface to explicit message waiting allows you to specify the message type you are willing to accept (using gras_msg_wait()). But you can also specify a list of accepted message types (using gras_msg_wait_or()), or even provide your own message filters to decide precisly the kind of message you are waiting for (for example depending on message content -- this is done using gras_msg_wait_ext()).

Any message received not matching your expectation will be queued for further use. Ie, they will be stored in memory and are candidates for the next gras_msg_handle() or gras_msg_wait().

Example: mutual exclusion with centralized coordinator

GRAS modelization of the algorithm

This section naturally provides an example of how to use gras_msg_wait(), but it can also be seen as an example of the guidelines provided in HOWTO design a GRAS application.

So, here are the caracteristics of our example:

There is two types of processes:

several clients which want to enter a critical section
a server, which grants clients to enter the CS when no other process is already in it.

There is three kind of messages in the system:

request (from clients to server) to ask for the permission to enter the CS
release (from clients to server) to express that they exited the CS
grant (from server to clients) to allow the given process to enter the CS

The server has 2 callbacks attached:

When it gets a request, it checks whether their is already a process in the CS. If yes, it adds the requester into a FIFO list. If not, it sends a grant message to the asking client.
When it gets a release, it checks whether there is some waiting processes in the waiting queue. If yes, it dequeues the first one and send a grant message to it. If no, it notes that the CS is free.

The server has two private data (for the callbacks to work):

a boolean indicating whether there is a process in the CS already
a waiting queue (xbt_dynar_t is quite natural to code this).

The client interface is composed of two functions:

lock(), which tries to get the grant from the server to enter the CS. This is where explicit waiting comes into the game. This function sends a request to the server, and blocks until the server issues a grant back.
unlock(), which informs the server that we exited the CS (by sending a release message)

The code step-by-step

a) Messages declaration

First of all, we should have a function declaring all used messages. As said before, this should be in a separate function so that it can be shared between all process kinds and avoid code dupplication which may result in definition discrepency.

Here, there is no payload attached to the messages.

void message_declaration(void)
{
  gras_msgtype_declare("request", NULL);
  gras_msgtype_declare("grant", NULL);
  gras_msgtype_declare("release", NULL);
}

b) Defining private data and callbacks of the server

Then, we define the callbacks that should be invoqued on the server side when some messages are received, as previously. For this, we also have to declare a structure for the private data of the server.

typedef struct {
  int process_in_CS;
  xbt_dynar_t waiting_queue;
} server_data_t;

int server_request_cb(gras_msg_cb_ctx_t ctx, void *payload)
{
  server_data_t *globals = (server_data_t *) gras_userdata_get();
  gras_socket_t s = gras_msg_cb_ctx_from(ctx);

  if (globals->process_in_CS) {
    xbt_dynar_push(globals->waiting_queue, &s);
    XBT_INFO("put %s:%d in waiting queue", gras_socket_peer_name(s),
          gras_socket_peer_port(s));
  } else {
    globals->process_in_CS = 1;
    XBT_INFO("grant %s:%d since nobody wanted it", gras_socket_peer_name(s),
          gras_socket_peer_port(s));
    gras_msg_send(s, "grant", NULL);
  }
  return 0;
}                               /* end_of_request_callback */

int server_release_cb(gras_msg_cb_ctx_t ctx, void *payload)
{
  server_data_t *globals = (server_data_t *) gras_userdata_get();

  if (!xbt_dynar_is_empty(globals->waiting_queue)) {
    gras_socket_t s;
    xbt_dynar_pop(globals->waiting_queue, &s);

    XBT_INFO("grant %s:%d since token released", gras_socket_peer_name(s),
          gras_socket_peer_port(s));
    gras_msg_send(s, "grant", NULL);
  } else {
    globals->process_in_CS = 0;
  }

  return 0;
}                               /* end_of_release_callback */

c) Client-side API

Now, we define the functions that the client must call to use the service. Note that this is where the explicit wait feature is used.

void lock(gras_socket_t toserver)
{
  gras_msg_send(toserver, "request", NULL);
  gras_msg_wait(-1, "grant", NULL, NULL);
  XBT_INFO("Granted by server");
}                               /* end_of_lock */

void unlock(gras_socket_t toserver)
{
  XBT_INFO("Release the token");
  gras_msg_send(toserver, "release", NULL);
}                               /* end_of_unlock */

d) Server-side initialization

The core of our distributed service is implemented (protocol, actions on server side, and accessing function on client side). We should now initialize the server and let it wait for incoming messages.

Defining when to stop the server can become tricky. The simplest solution is to never let the server stop. It simply runs forever. But the simulator will raise an error at the end, so I won't do so here to keep the output clean. Another solution would be to deal with client membership properly: clients registers, use the service and quit afterward. When no client use the service, the server stops. This would be a bit difficult to implement (actually, there is an AMOK module to do so simply: Peer Management). Here, we will just hardcode that the clients ask 5 times for the token, and that there is two clients. This clearly simplify the problem.

  globals = gras_userdata_new(server_data_t);
  globals->process_in_CS = 0;
  globals->waiting_queue =
      xbt_dynar_new(sizeof(gras_socket_t),
                    NULL /* not closing sockets */ );

  message_declaration();
  gras_cb_register("request", &server_request_cb);
  gras_cb_register("release", &server_release_cb);

  for (i = 0; i < 20; i++)      /* 5 requests of each process, 2 processes, 2 messages per request */
    gras_msg_handle(-1);

e) Client-side use

And now, the client is really simple to write:

  message_declaration();
  toserver = gras_socket_client(argv[1], atoi(argv[2]));

  for (i = 0; i < 5; i++) {
    gras_os_sleep(0.1);
    lock(toserver);
    gras_os_sleep(0.1);
    unlock(toserver);
  }

Recapping everything together

The program now reads:

/* Copyright (c) 2007, 2010. The SimGrid Team.
 * All rights reserved.                                                     */

/* This program is free software; you can redistribute it and/or modify it
  * under the terms of the license (GNU LGPL) which comes with this package. */

#include <stdlib.h>
#include <gras.h>

XBT_LOG_NEW_DEFAULT_CATEGORY(test, "My little example");

void message_declaration(void)
{
  gras_msgtype_declare("request", NULL);
  gras_msgtype_declare("grant", NULL);
  gras_msgtype_declare("release", NULL);
}

typedef struct {
  int process_in_CS;
  xbt_dynar_t waiting_queue;
} server_data_t;

int server_request_cb(gras_msg_cb_ctx_t ctx, void *payload)
{
  server_data_t *globals = (server_data_t *) gras_userdata_get();
  gras_socket_t s = gras_msg_cb_ctx_from(ctx);

  if (globals->process_in_CS) {
    xbt_dynar_push(globals->waiting_queue, &s);
    XBT_INFO("put %s:%d in waiting queue", gras_socket_peer_name(s),
          gras_socket_peer_port(s));
  } else {
    globals->process_in_CS = 1;
    XBT_INFO("grant %s:%d since nobody wanted it", gras_socket_peer_name(s),
          gras_socket_peer_port(s));
    gras_msg_send(s, "grant", NULL);
  }
  return 0;
}                               /* end_of_request_callback */

int server_release_cb(gras_msg_cb_ctx_t ctx, void *payload)
{
  server_data_t *globals = (server_data_t *) gras_userdata_get();

  if (!xbt_dynar_is_empty(globals->waiting_queue)) {
    gras_socket_t s;
    xbt_dynar_pop(globals->waiting_queue, &s);

    XBT_INFO("grant %s:%d since token released", gras_socket_peer_name(s),
          gras_socket_peer_port(s));
    gras_msg_send(s, "grant", NULL);
  } else {
    globals->process_in_CS = 0;
  }

  return 0;
}                               /* end_of_release_callback */

int server(int argc, char *argv[])
{
  gras_socket_t mysock;         /* socket on which I listen */
  server_data_t *globals;
  int i;

  gras_init(&argc, argv);
  mysock = gras_socket_server(atoi(argv[1]));

  globals = gras_userdata_new(server_data_t);
  globals->process_in_CS = 0;
  globals->waiting_queue =
      xbt_dynar_new(sizeof(gras_socket_t),
                    NULL /* not closing sockets */ );

  message_declaration();
  gras_cb_register("request", &server_request_cb);
  gras_cb_register("release", &server_release_cb);

  for (i = 0; i < 20; i++)      /* 5 requests of each process, 2 processes, 2 messages per request */
    gras_msg_handle(-1);

  gras_exit();
  return 0;
}                               /* end_of_server */

void lock(gras_socket_t toserver)
{
  gras_msg_send(toserver, "request", NULL);
  gras_msg_wait(-1, "grant", NULL, NULL);
  XBT_INFO("Granted by server");
}                               /* end_of_lock */

void unlock(gras_socket_t toserver)
{
  XBT_INFO("Release the token");
  gras_msg_send(toserver, "release", NULL);
}                               /* end_of_unlock */

int client(int argc, char *argv[])
{
  int i;
  gras_socket_t mysock;         /* socket on which I listen */
  gras_socket_t toserver;       /* socket used to write to the server */

  gras_init(&argc, argv);

  mysock = gras_socket_server_range(1024, 10000, 0, 0);

  XBT_VERB("Client ready; listening on %d", gras_socket_my_port(mysock));

  gras_os_sleep(1.5);           /* sleep 1 second and half */
  message_declaration();
  toserver = gras_socket_client(argv[1], atoi(argv[2]));

  for (i = 0; i < 5; i++) {
    gras_os_sleep(0.1);
    lock(toserver);
    gras_os_sleep(0.1);
    unlock(toserver);
  }
  gras_exit();
  return 0;
}

Which produces the expected output:

$ ./test_server & ./test_client 127.0.0.1 12345 & ./test_client 127.0.0.1 12345 
[arthur:server:(28159) 0.000023] [test/INFO] grant 127.0.0.1:1024 since nobody wanted it
[arthur:client:(28160) 0.000013] [test/INFO] Granted by server
[arthur:server:(28159) 0.014750] [test/INFO] put 127.0.0.1:1025 in waiting queue
[arthur:client:(28160) 0.100254] [test/INFO] Release the token
[arthur:server:(28159) 0.100950] [test/INFO] grant 127.0.0.1:1025 since token released
[arthur:client:(28162) 0.000014] [test/INFO] Granted by server
[arthur:server:(28159) 0.201146] [test/INFO] put 127.0.0.1:1024 in waiting queue
[arthur:client:(28162) 0.100169] [test/INFO] Release the token
[arthur:server:(28159) 0.201593] [test/INFO] grant 127.0.0.1:1024 since token released
[arthur:client:(28160) 0.201472] [test/INFO] Granted by server
[arthur:client:(28160) 0.301560] [test/INFO] Release the token
[arthur:server:(28159) 0.302114] [test/INFO] put 127.0.0.1:1025 in waiting queue
[arthur:server:(28159) 0.302172] [test/INFO] grant 127.0.0.1:1025 since token released
[arthur:client:(28162) 0.201166] [test/INFO] Granted by server
[arthur:server:(28159) 0.402342] [test/INFO] put 127.0.0.1:1024 in waiting queue
[arthur:client:(28162) 0.301265] [test/INFO] Release the token
[arthur:server:(28159) 0.402687] [test/INFO] grant 127.0.0.1:1024 since token released
[arthur:client:(28160) 0.402605] [test/INFO] Granted by server
[arthur:server:(28159) 0.503013] [test/INFO] put 127.0.0.1:1025 in waiting queue
[arthur:client:(28160) 0.502739] [test/INFO] Release the token
[arthur:server:(28159) 0.503260] [test/INFO] grant 127.0.0.1:1025 since token released
[arthur:client:(28162) 0.402236] [test/INFO] Granted by server
[arthur:server:(28159) 0.603517] [test/INFO] put 127.0.0.1:1024 in waiting queue
[arthur:client:(28162) 0.502304] [test/INFO] Release the token
[arthur:server:(28159) 0.603699] [test/INFO] grant 127.0.0.1:1024 since token released
[arthur:client:(28160) 0.603564] [test/INFO] Granted by server
[arthur:client:(28160) 0.703651] [test/INFO] Release the token
[arthur:server:(28159) 0.704098] [test/INFO] put 127.0.0.1:1025 in waiting queue
[arthur:server:(28159) 0.704146] [test/INFO] grant 127.0.0.1:1025 since token released
[arthur:client:(28162) 0.603133] [test/INFO] Granted by server
[arthur:server:(28159) 0.804385] [test/INFO] put 127.0.0.1:1024 in waiting queue
[arthur:client:(28162) 0.703226] [test/INFO] Release the token
[arthur:server:(28159) 0.804647] [test/INFO] grant 127.0.0.1:1024 since token released
[arthur:client:(28160) 0.804530] [test/INFO] Granted by server
[arthur:client:(28160) 0.904608] [test/INFO] Release the token
[arthur:server:(28159) 0.905011] [test/INFO] put 127.0.0.1:1025 in waiting queue
[arthur:client:(28160) 0.904749] [gras/INFO] Exiting GRAS
[arthur:server:(28159) 0.905159] [test/INFO] grant 127.0.0.1:1025 since token released
[arthur:client:(28162) 0.804870] [test/INFO] Granted by server
[arthur:client:(28162) 0.905010] [test/INFO] Release the token
[arthur:client:(28162) 0.905181] [gras/INFO] Exiting GRAS
[arthur:server:(28159) 1.007620] [gras/INFO] Exiting GRAS
$
$
$ ./test_simulator platform-3nodes.xml test.xml
[Jacquelin:server:(1) 0.000000] [test/INFO] grant Boivin:1024 since nobody wanted it
[Jacquelin:server:(1) 0.000537] [test/INFO] put Geoff:1024 in waiting queue
[Boivin:client:(2) 0.000537] [test/INFO] Granted by server
[Boivin:client:(2) 0.100537] [test/INFO] Release the token
[Jacquelin:server:(1) 0.101074] [test/INFO] grant Geoff:1024 since token released
[Geoff:client:(3) 0.101264] [test/INFO] Granted by server
[Geoff:client:(3) 0.201264] [test/INFO] Release the token
[Jacquelin:server:(1) 0.201611] [test/INFO] put Boivin:1024 in waiting queue
[Jacquelin:server:(1) 0.201801] [test/INFO] grant Boivin:1024 since token released
[Boivin:client:(2) 0.202338] [test/INFO] Granted by server
[Jacquelin:server:(1) 0.301991] [test/INFO] put Geoff:1024 in waiting queue
[Boivin:client:(2) 0.302338] [test/INFO] Release the token
[Jacquelin:server:(1) 0.302875] [test/INFO] grant Geoff:1024 since token released
[Geoff:client:(3) 0.303065] [test/INFO] Granted by server
[Geoff:client:(3) 0.403065] [test/INFO] Release the token
[Jacquelin:server:(1) 0.403412] [test/INFO] put Boivin:1024 in waiting queue
[Jacquelin:server:(1) 0.403602] [test/INFO] grant Boivin:1024 since token released
[Boivin:client:(2) 0.404139] [test/INFO] Granted by server
[Jacquelin:server:(1) 0.503792] [test/INFO] put Geoff:1024 in waiting queue
[Boivin:client:(2) 0.504139] [test/INFO] Release the token
[Jacquelin:server:(1) 0.504675] [test/INFO] grant Geoff:1024 since token released
[Geoff:client:(3) 0.504865] [test/INFO] Granted by server
[Geoff:client:(3) 0.604865] [test/INFO] Release the token
[Jacquelin:server:(1) 0.605212] [test/INFO] put Boivin:1024 in waiting queue
[Jacquelin:server:(1) 0.605402] [test/INFO] grant Boivin:1024 since token released
[Boivin:client:(2) 0.605939] [test/INFO] Granted by server
[Jacquelin:server:(1) 0.705592] [test/INFO] put Geoff:1024 in waiting queue
[Boivin:client:(2) 0.705939] [test/INFO] Release the token
[Jacquelin:server:(1) 0.706476] [test/INFO] grant Geoff:1024 since token released
[Geoff:client:(3) 0.706666] [test/INFO] Granted by server
[Geoff:client:(3) 0.806666] [test/INFO] Release the token
[Jacquelin:server:(1) 0.807013] [test/INFO] put Boivin:1024 in waiting queue
[Jacquelin:server:(1) 0.807203] [test/INFO] grant Boivin:1024 since token released
[Boivin:client:(2) 0.807740] [test/INFO] Granted by server
[Jacquelin:server:(1) 0.907393] [test/INFO] put Geoff:1024 in waiting queue
[Boivin:client:(2) 0.907740] [test/INFO] Release the token
[Boivin:client:(2) 0.908277] [gras/INFO] Exiting GRAS
[Jacquelin:server:(1) 0.908277] [test/INFO] grant Geoff:1024 since token released
[Geoff:client:(3) 0.908467] [test/INFO] Granted by server
[Geoff:client:(3) 1.008467] [test/INFO] Release the token
[Geoff:client:(3) 1.008657] [gras/INFO] Exiting GRAS
[Jacquelin:server:(1) 1.008657] [gras/INFO] Exiting GRAS
$

Table of Contents