PDF de programación - Programación concurrente - Colas de mensajes (3)

LinuxFocus article number 316
http://linuxfocus.org

Programación concurrente − Colas de
mensajes (3)

by Leonardo Giordani
<leo.giordani(at)libero.it>

About the author:

Acabo de recibir mi título de
la Facultad de Ingeniería de
Telecomunicaciones de la
Politécnica de Milan.
Interesado en la
programación
(principalmente en
Ensamblador y C/C++).
Desde 1999 trabajo casi
exclusivamente con
Linux/Unix.

Abstract:

Este es el último artículo de la serie sobre programación concurrente: en él
implementaremos la segunda y última capa de nuestro protocolo, creando
funciones que simularán el comportamiento del usuario basándose en la
primera capa desarrollada en el pasado artículo.

Podría ser una buenda idea leerse primero alguno de los artículos anteriores
de la serie:

•
•
•
•

Programación concurrente − Colas de mensajes (2)
Programación concurrente − Principios e introducción a los procesos
Programación concurrente − Comunicación entre procesos
Programación concurrente − Colas de mensajes (1)

_________________ _________________ _________________

Implementación del protocolo − Capa 2 − General

El programa ipcdemo ha sido desarrollado con el fin de ser una implementación simple de un conmutador
entre usuarios que intentan enviarse mensajes entre ellos. Para añadirle diversión a la simulación añadí el
concepto de "servicio": un mensaje de servicio (señalización) es un mensaje cuyo propósito principal no
consiste en transportar información entre usuarios, sino información de control. Los mensajes de servicio
serán enviados al conmutados por los usuarios para hacerle saber que están vivos, cómo alcanzarlos (enviando
un identificador de cola IPC) y que están cerrándose. Se han definido dos servicios más: Finalización y
temporización; el primero lo utiliza el conmutador para decirle al usuario que debe finalizar, el segundo
intenta medir el tiempo de respuesta del usuario. Hablaremos más sobre este tema luego, en la sección de
usuario y en la del conmutador.

1/7

La capa 2 contiene funciones de alto nivel para enviar y recibir mensajes, para pedir y responder a servicios, y
algún material de inicialización: esas funciones se construyen utilizando las de la capa 1 y, por tanto, son
realmente fáciles de entender. Nótese que en layer2.h declaré algunos alias para representar tipos de mensajes
(mensajes de usuario y mensajes de servicio) y diferentes servicios (entre ellos dos servicios definidos por el
usuario para experimentos).

El ipcdemo es sólamente un código de demostración: no está optimizado, y salta a la vista que he empleado
muchas variables globales, pero esto es sóolo para permitir que el lector se concentre en la parte de IPC y no
en los detalles del código. De cualquier forma, si encuentras algo realmente extraño, simplemente escríbeme y
lo comentaremos.

Implementación del proceso de usuario

El usuario es simplemente un proceso hijo del conmutador (o, mejor, del proceso padre, que actúa como un
conmutador). Esto significa que el usuario tiene todas las variables inicializadas justamente como el
conmutador: por ejemplo, conoce el identificador de la cola del conmutador, porque se guarda en una variable
local por parte del propio conmutador antes de la operación fork.

Cuando el usuario comienza su vida, lo primero que debería hacer es crear una cola y permitir que el
conmutador conozca cómo llegar a ella; para hacer esto el usuario envía dos mensajes de servicio,
SERV_BIRTH y SERV_QID.

/* Initialize queue */
qid = init_queue(i);

/* Let the switch know we are alive */
child_send_birth(i, sw);

/* Let the switch know how to reach us */
child_send_qid(i, qid, sw);

Entonces, entra en el bucle principal: en él, el usuario envía un mensaje, compruba si hay mensajes entrantes
de otros usuarios y comprueba si el conmutador solicitó un servicio.

La decisión de enviar mensajes se toma en base a una probabilidad: la función myrand() devuelve un número
aleatorio normalizado según el argumento pasado, en este caso 100, y se envía el mensaje sólo si este número
es menor que la probabilidad especificada; ya que el usuario se duerme durante 1 segundo entre dos
ejecuciones del bucle, más o menos enviará tantos mensajes como la probababilidad de envío cada 100
segundos, asumiendo que 100 extracciones son suficientes para transformar la probabilidad en realidad, las
cuales son insuficientes... Simplemente tendremos que poner atención en no utilizar probabilidades demasiado
bajas o nuestra simulación se ejecutará durante mucho tiempo.

if(myrand(100) < send_prob){
dest = 0;

/* Do not send messages to the switch, to you, */
/* and to the same receiver of the previous message */
while((dest == 0) || (dest == i) || (dest == olddest)){
dest = myrand(childs + 1);
}
olddest = dest;

printf("%d −− U %d −− Message to user %d\n", (int) time(NULL), i, dest);
child_send_msg(i, dest, 0, sw);

2/7

}

Los mensajes de otros usuarios son, de hecho, mensajes que los otros usuarios envían al conmutador y que el
conmutador nos envía a nosotros, y están marcados con el tipo TYPE_CONN (como CONNECTION).

/* Check the incoming box for simple messages */
if(child_get_msg(TYPE_CONN, &in)){
msg_sender = get_sender(&in);
msg_data = get_data(&in);
printf("%d −− U %d −− Message from user %d: %d\n",
(int) time(NULL), i, msg_sender, msg_data);
}

Si el conmutador solicitó un servicio utilizaremos un mensaje marcado con el tipo TYPE_SERV, y tenemos
que responder; en caso de finalización del servicio le enviaremos al conmutador un mensaje de asentimiento,
de forma que nos pueda marcar como inalcanzables y pare de enviarnos mensajes; entonces tenemos que leer
todos los mensajes restantes (para guardar las formas, podríamos saltarnos este paso), eliminar la cola y decir
adiós a la simulación. La petición del servicio de temporización que enviamos al conmutador es un mensaje
que contiene la fecha actual: el conmutador la resta del instante de tiempo en que el mensaje fue enviado para
registrar cuánto tiempo ha estado el mensaje en las colas. Como vemos, también estamos haciendo QoS
(Calidad de Servicio), de forma que la simulación ya es posiblemente mejor que el actual sistema telefónico...

/* Check if the switch requested a service */
if(child_get_msg(TYPE_SERV, &in)){
msg_service = get_service(&in);

switch(msg_service){
case SERV_TERM:
/* Sorry, we have to terminate */
/* Send an acknowledgement to the switch */
child_send_death(i, getpid(), sw);

/* Read the last messages we have in the queue */
while(child_get_msg(TYPE_CONN, &in)){
msg_sender = get_sender(&in);
msg_data = get_data(&in);
printf("%d −− U %d −− Message from user %d: %d\n",
(int) time(NULL), i, msg_sender, msg_data);
}

/* Remove the queue */
close_queue(qid);
printf("%d −− U %d −− Termination\n", (int) time(NULL), i);
exit(0);
break;
case SERV_TIME:
/* We have to time our work */
child_send_time(i, sw);
printf("%d −− U %d −− Timing\n", (int) time(NULL), i);
break;
}
}

Implementación del proceso conmutador

El proceso padre se divide en dos partes, antes y después de la creación de los hijos. Durante la primera parte
se tiene que inicializar el array para salvar los identificadores de cola de sus hijos y se tiene que crear su

3/7

propia cola; seguramente esta no es la forma correcta de implementar algo de este tipo, pero introducir listas
dinámicas en este contexto quedaría fuera del alcance de este artículo y, despues de todo, no sería útil; de
todas formas, si se planea desarrollar algo que acepte cualquier número de conexiones hay que acordarse de
utilizar estructuras dinámicas y reserva de memoria. Los identificadores de las colas son inicializados al
principio con el valor del identificador de la cola del conmutador, simbolizando que el usuario no está vivo
todavía: cuando un usuario termina se vuelve a establecer el identificador de la cola a su valor original.

En la segunda parte, el proceso padre actúa como un conmutador, ejecutando un bucle tal y como hace el
usuario, hasta que todos los usuarios han finalizado. El conmutador comprueba si hay mensajes entrantes
procedentes de usuarios y los enruta hacia sus destinos.

/* Check if some user has connected */
if(switch_get_msg(TYPE_CONN, &in)){

msg_receiver = get_receiver(&in);
msg_sender = get_sender(&in);
msg_data = get_data(&in);

/* If the destination is alive */
if(queues[msg_receiver] != sw){

/* Send a messge to the destination (follow−up the received message) */
switch_send_msg(msg_sender, msg_data, queues[msg_receiver]);

printf("%d −− S −− Sender: %d −− Destination: %d\n",
(int) time(NULL), msg_sender, msg_receiver);
}
else{
/* The destination is not alive */
printf("%d −− S −− Unreachable destination (Sender: %d − Destination: %d)\n",
(int) time(NULL), msg_sender, msg_receiver);
}

Pero si un usuario envió un mensaje a través del conmutador, puede ser el objeto de una petición de servicio
de acuerdo a una probabilidad (funcionando igual que antes); en el primer caso forzamos al usuario a
terminar, en el segundo comenzamos una operación de temporización: registramos el tiempo actual y
marcamos al usuario de forma que no intentemos temporizar a un usuario que ya esté realizando esta
operación. Si no recibimos un mensaje es posible que todos los usuarios hayan terminado: en este caso
esperamos que los procesos hijos finalicen realmente (el último usuario podría estar comprobando los
mensajes restantes en su cola), eliminamos nuestra cola y salimos.

/* Randomly request a service to the sender of the last message */
if((myrand(100) < death_prob) && (queues[msg_sender] != sw)){
switch(myrand(2))
{
case 0:
/* The user must terminate */
printf("%d −− S −− User %d chosen for termination\n",
(int) time(NULL), msg_sender);
switch_send_term(i, queues[msg_sender]);
break;
c