Colas en Python con RabbitMQ y Pika (parte 3)

Antes de este, debes leer los dos artículos anteriores de esta serie. En el primero, desarrollamos un simple flujo de mensajes y en el segundo, estructuramos dichos mensajes como trabajos.

Ahora vamos a afrontar otra de las funcionalidades que requiere cualquier cola de trabajos: debe haber un mecanismo para que el proceso consumidor pueda confirmar que un trabajo se ha completado, y en caso contrario, sea reasignado a otro proceso.

Se ruega acuse de recibo

La confirmación, acuse de recibo o acknowledgment, es un concepto usado en prácticamente todos los protocolos (incluso los no digitales) para aquellas operaciones en las que queremos asegurar que se completan correctamente.

En el programa que hemos hecho en esta serie de artículos, hasta ahora le estábamos indicando a la cola de RabbitMQ que no hiciera acuse de recibo, mediante la opción no_ack=True en la llamada a basic_consume(). Esto hacía que la cola no esperase ninguna confirmación y eliminase los trabajos tan pronto como eran despachados al consumidor. Esto es muy eficiente, pero evidentemente corremos el peligro de que algún trabajo quede sin hacer y sea silenciosamente olvidado.

Para hacer el acuse de recibo, eliminamos la opción no_ack (por defecto es False) y añadimos en la función callback una llamada a ch.basic_ack(). Esta llamada avisa a la cola que el trabajo se ha completado y ya puede ser retirado. Para identificar el trabajo al que nos referimos, se incluye el argumento delivery_tag, que a su vez hemos recibido a través del argumento method.

En resumen, el código cambia de esta manera:

# Definir la función callback
def procesar(ch, method, properties, body):
    datos = body.decode('utf-8')
    print("Se ha recibido un trabajo: %s" % datos)
    t = Trabajo.importar(datos)
    ejecutar(t)
    ch.basic_ack(delivery_tag=method.delivery_tag)

# Enganchar el callback
ch.basic_consume(procesar, queue='trabajos')

Si ejecutamos este consumidor, en principio no notaremos ningún cambio, a no ser que miremos en la cola de mensajes y nos demos cuenta que el mensaje se elimina después de pasar el tiempo de espera, no antes.

Desde la línea de comandos de rabbitmqctl, hay una forma para ver no solo los mensajes en espera, sino cuántos han sido despachados y están esperando confirmación:

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

En nuestro ejemplo, los posibles valores arrojados por esta consulta pueden ser:

a) No hay trabajos.

Listing queues ...
trabajos 0 0

b) Se ha enviado un trabajo con productor.py.

Listing queues ...
trabajos 1 0

c) Se ha lanzado consumidor.py y está procesando el trabajo.

Listing queues ...
trabajos 0 1

d) El trabajo se ha terminado de procesar, y el acuse de recibo se ha completado.

Listing queues ...
trabajos 0 0

¿No es maravilloso? ¡Seguimos para bingo!

Competencia entre consumidores

El ejemplo anterior es muy trivial, debido a que sólo contamos con un consumidor. La cosa se pone más interesante cuando hay varios consumidores procesando los mensajes, lo que inevitablemente genera una "competencia" entre ellos. Veremos que RabbitMQ puede lidiar con esa competencia de forma brillante.

Para simular una competencia, vamos a modificar nuestros inocentes programas para que fuercen la máquina. Vamos a hacer lo siguiente:

• Haremos que productor.py genere trabajos de manera constante.
• Haremos que consumidor.py, además de ejecutar trabajos en espera simulando que trabaja, aleatoriamente falle lanzando una excepción. De esta manera comprobaremos si dicho trabajo se redirige a otro consumidor.
• Lanzaremos varios productores y varios consumidores a la vez, en distintas terminales.

Nuestros programas se quedan de esta manera:

productor.py

import pika
from trabajo import Trabajo
from random import randint
import time

# Establecer conexión
con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = con.channel()

# Declarar la cola
ch.queue_declare(queue='trabajos')

# Bucle infinito
while True:
    # Espera aleatoria antes de lanzar el siguiente trabajo
    time.sleep(randint(1,4))

    # Generar un trabajo
    t = Trabajo('esperar', randint(1,10))

    # Publicar el mensaje
    ch.basic_publish(exchange='', routing_key='trabajos', body=t.exportar().encode('utf-8'))
    print("Trabajo enviado: %s" % t.exportar())

consumidor.py

import pika
from trabajo import Trabajo
from proceso import ejecutar

# Establecer conexión
con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = con.channel()

# Declarar la cola
ch.queue_declare(queue='trabajos')


# Definir la función callback
def procesar(ch, method, properties, body):
    datos = body.decode('utf-8')
    print("Se ha recibido un trabajo: %s" % datos)
    t = Trabajo.importar(datos)
    ejecutar(t)
    ch.basic_ack(delivery_tag=method.delivery_tag)

# Enganchar el callback
ch.basic_consume(procesar, queue='trabajos')

# Poner en espera
print('Esperando trabajos...')
ch.start_consuming()

proceso.py

import time
import random

def ejecutar(trabajo):
    if trabajo.operacion == 'esperar':
        print('Esperando %d segundos...' % trabajo.entrada)
        time.sleep(trabajo.entrada)

        # Simular un fallo el 20% de las veces
        if random.random() < 0.2:
            raise Exception

        print('Hecho!')
    else:
        raise NotImplementedError('Operación "%s" no soportada.' % trabajo.operacion)

Si, con esta configuración, lanzamos, por ejemplo, dos productores y tres consumidores, veremos cómo los trabajos van fluyendo y cuando, eventualmente, uno de los consumidores aborta, los otros siguen procesando con normalidad, sin que se pierda ningún trabajo.

Dos productores y tres consumidores trabajando a la vez. Podemos observar que, a pesar de que los consumidores 1 y 2 han abortado, el segundo sigue en la brecha procesando trabajos.

TCP/IP (9): Algoritmos de rutado

Richard Bellman

Lester R. Ford

Rutado por vector-distancia (Bellman-Ford)

El algoritmo de vector-distancia, también conocido como Ford Fulkerson o Bellman-Ford, define una forma sencilla de mantener las pasarelas actualizadas.

Cuando una pasarela arranca, inicia su tabla de rutas para que contenga una entrada por cada red conectada directamente, indicando para cada una de ellas la distancia hacia ella, por lo general medida en saltos (que será cero pues están conectadas directamente).

Periódicamente, la pasarela J envía una copia de su tabla a cualquier otro que pueda alcanzar de manera directa (por ejemplo, K). Éste utilizará esta información para actualizar su propia tabla, en tres casos:

* Si J tiene una ruta más corta para un determinado destino.

* Si J lista un destino que K no tiene en su tabla.

* Si ha cambiado la distancia hacia algún destino hacia donde K ruta actualmente a través de J.Como es lógico, K tendrá que añadir una unidad a cada distancia que le reporte J.

En este tipo de diseño, todos las pasarelas han de participar en el intercambio de información para que las rutas sean consistentes.
El nombre vector-distancia proviene de que la información intercambiada es una lista de pares de destinos (o vectores) y distancias hacia ellos.

El protocolo GGP (Gateway-to-Gateway Protocol) se utilizó originalmente para el intercambio de rutas en el sistema central de pasarelas núcleo del INOC, y hace uso del algoritmo de vector-distancia.

Se transporta en datagramas IP igual que los protocolos UDP o TCP, con un encabezado de formato fijo que indica el formato del resto del mensaje.

El contenido consiste en un conjunto de pares de direcciones de red IP (netid) y distancias medidas en saltos (hops). Cero saltos indica que la red en cuestión está accesible directamente; es decir, el número representa la cantidad de pasarelas que un datagrama se encontrará en su camino hacia dicha red.

El punto débil de este planteamiento es que no siempre un número de saltos menor ha de producir menor retardo, pues las redes son heterogéneas, y tienen diferentes velocidades. Muchas pasarelas suman artificialmente varios saltos para trayectorias que cruzan redes lentas.

Para mantener los mensajes lo más cortos posible, se agrupan las entradas por distancias, enviándose para cada grupo un valor de distancia (8 bits), el número de redes que están a esa distancia (8 bits), seguidos por la lista de los netid de todas ellas.

En el mensaje puede también el emisor indicarle al receptor, mediante un campo de la cabecera, que necesita a su vez una actualización de él.

Este tipo de mensaje GGP se llama UPDATE (actualización). Otros tipos son los acuses de recibo positivos y negativos, las solicitudes de eco y sus respuestas. Un campo de 8 bits al principio de la cabecera identifica el tipo de mensaje.

Algoritmo de enlace-estado o SPF

El mítico Edsger Dijkstra inventó el SPF.

La principal desventaja de los algoritmos de vector-distancia es que no se extienden con la suficiente rapidez ante los cambios en la red. Además, requiere del intercambio de mensajes largos, cuyo tamaño es proporcional al número de redes. Esto, unido a la necesaria participación de todas las pasarelas, hace que la cantidad de información a intercambiar sea enorme.

La alternativa más aceptada es el algoritmo de enlace-estado, también conocido como SPF (Shortest Path First, primero el camino más corto), el cual hace que cada pasarela tenga una idea de la topología completa en forma de un grafo de nodos y arcos.

Cada pasarela prueba periódicamente el estado de sus vecinos en el grafo, o sea, aquellos a los que está conectado directamente, difundiendo esta información hacia otras pasarelas.

Para localizar los vecinos activos (up) o inactivos (down), se usa la regla k-out-of-n (k-de-n), que significa que el enlace se considera activo si un porcentaje significativo de solicitudes de eco tiene réplica.

Periódicamente, cada pasarela difunde el estado en que está cada uno de sus enlaces. El software del protocolo se encarga de hacer llegar a cada pasarela participante una copia sin modificación alguna de dicho mensaje. No hay en él información de rutado, sólo indica la disponibilidad de líneas y, en todo caso, la métrica hacia las pasarelas directamente conectadas a ella.

Cuando un mensaje de este tipo llega a una pasarela, éste actualiza su grafo y recomputa las rutas usando en conocido algoritmo del camino más corto de Dijkstra.

Como todos los participantes usan idéntica información de base para recomputar las rutas de manera local, la convergencia entre ellas está asegurada y los problemas de depuración son fáciles de resolver. Cada pasarela mantiene una base de datos completa, la misma para todas.

Simulación animada del SPF (fuente: Wikipedia)

La aplicación del algoritmo en cada pasarela lleva a la computación de un árbol de caminos óptimos distinto en cada una, tomándose a sí misma como raíz del árbol.

Entre los años 1989 y 1994 el IETF desarrolló un protocolo muy usado, basado en este algoritmo, que se denomina OSPF.

Colas en Python con RabbitMQ y Pika (parte 2)

En el anterior artículo de esta serie, aprendimos a crear un flujo de mensajes entre un proceso que denominamos productor y otro que llamamos consumidor. Pero eso dista mucho de lo que podríamos llamar una cola de trabajos. Para eso, necesitamos añadir ciertas funcionalidades.

Una de esas funcionalidades, que es imprescindible para nuestra cola de trabajo, es que los mensajes (que denominaremos trabajos a partir de ahora) deben tener un formato estructurado y un significado concreto, reconocido por todos los procesos implicados.

Dando formato a los mensajes

Vamos a convertir los simples mensajes que manejábamos hasta ahora en auténticas órdenes de trabajo. Para ello, hace falta estructurar la información. Tendremos que definir el objeto Trabajo y luego insertarlo en la cola como mensaje.

Dado que RabbitMQ acepta mensajes binarios, somos libres de elegir el formato que más nos plazca. En cualquier caso, ese formato deberá saber representar ciertos objetos de los que maneja el programa. En programación, la acción de convertir objetos de programa en bloques de datos almacenables o transmisibles se denomina serialización o marshalling. Algunos buenos candidatos para esto en Python son:

• XML: Es un estándar muy utilizado para representar objetos, independientemente del lenguaje de programación.
• JSON: Pronunciado "yeison", es similar a XML pero mucho más simple y legible para los humanos. Es menos estándar, pero está implementado en casi todos los lenguajes.
• YAML: Tiene una filosofía parecida a JSON, pero más flexible. Hay quien considera a JSON un subconjunto de YAML.
• Pickle: Es un mecanismo desarrollado específicamente para Python. Es complejo y muy potente, pero no es legible por humanos (son datos binarios), lo que dificulta bastante la depuración del código.

En la elección del formato, hay que tener en cuenta principalmente dos factores: la complejidad de los datos que vamos a serializar y la portabilidad de éstos a otros lenguajes. Este segundo punto no es trivial. Si realizamos un sistema de mensajes que esté atado a un lenguaje determinado, perdemos la libertad de maniobra que nos da una plataforma como RabbitMQ.

Otros factores que podemos tener en cuenta son la legibilidad (a efectos de depuración) y la eficiencia (que el coste de tiempo de proceso en la serialización y deserialización sea bajo).

En este caso, me voy a decantar por JSON, que es un formato muy utilizado en Python, que también tiene soporte en casi todos los lenguajes y además es muy legible y bastante eficiente en cuanto a rendimiento.

Su uso básico en Python es muy simple. Para serializar un objeto en formato cadena (legible) usamos el método json.dumps(), y para deserializar una cadena de vuelta al objeto original, json.loads(). El módulo json viene "de fábrica" en cualquier instalación de Python, y además maneja automáticamente todos los tipos nativos del lenguaje, como listas y diccionarios. Su uso es tan común en Python, que los conversores estándar str() y repr() son sospechosamente similares a él.

Define trabajo

Vamos a definir el objeto genérico Trabajo como una operación con unos datos (opcionales) de entrada. Una primera versión de trabajo.py puede ser algo así:

class Trabajo:
    def __init__(self, operacion, entrada=None):
        self.operacion = operacion
        self.entrada = entrada

En esta definición, hemos abusado sin piedad del duck typing de Python para dejar abierta la elección de qué consideramos una operación y una entrada.

La simple existencia de un objeto no lo hace serializable. Si ahora definimos una instancia de Trabajo, no podremos pasarla a JSON y nos lanzará un TypeError. Hay muchas maneras en Python de hacerlo serializable, por ejemplo usando un JSONEncoder. Aquí vamos a ir a lo más simple, que es definir métodos ad hoc para exportar e importar los objetos. La versión serializada será el resultado de exportar todas las variables en forma de diccionario:

import json


class Trabajo:
    def __init__(self, operacion, entrada=None):
        self.operacion = operacion
        self.entrada = entrada

    def exportar(self):
        return json.dumps(self.__dict__)

    @classmethod
    def importar(cls, datos):
        dic = json.loads(datos)
        return cls(dic['operacion'], dic['entrada'])

Con esta implementación, las operaciones posibles son:

Crear un trabajo:
t = Trabajo('hola', 'mundo')

Serializarlo:
cadena = t.exportar()

Deserializarlo:

t2 = Trabajo.importar(cadena)

Esta implementación tan simple sólo va a funcionar si los tipos que usamos para la operación y la entrada son a su vez serializables. Para nuestro ejemplo, nos vale perfectamente.

Con todo esto, vamos a modificar los procesos productor y consumidor del anterior artículo para que manejen el objeto que acabamos de definir.

Para simular un trabajo, definiremos la operación "esperar" con un número aleatorio de segundos.

productor.py

import pika
from trabajo import Trabajo
from random import randint

# Establecer conexión
con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = con.channel()

# Declarar la cola
ch.queue_declare(queue='trabajos')

# Generar un trabajo
t = Trabajo('esperar', randint(1,10))

# Publicar el mensaje
ch.basic_publish(exchange='', routing_key='trabajos', body=t.exportar().encode('utf-8'))
print("Trabajo enviado.")

# Cerrar conexión
con.close()

consumidor.py

import pika
from trabajo import Trabajo
from proceso import ejecutar

# Establecer conexión
con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = con.channel()

# Declarar la cola
ch.queue_declare(queue='trabajos')


# Definir la función callback
def procesar(ch, method, properties, body):
    datos = body.decode('utf-8')
    print("Se ha recibido un trabajo: %s" % datos)
    t = Trabajo.importar(datos)
    ejecutar(t)

# Enganchar el callback
ch.basic_consume(procesar, queue='trabajos', no_ack=True)

# Poner en espera
print('Esperando trabajos...')
ch.start_consuming()

Un par de detalles a destacar:

• Dado que el body de los mensajes ha de ser binario, hacemos la correspondiente conversión desde/hacia UTF-8.
• Para mantener el código debidamente modular, hemos separado la gestión de los trabajos por un lado y la propia ejecución de los mismos. Para ello hemos creado un módulo proceso.py con una función ejecutar().

proceso.py

import time


def ejecutar(trabajo):
    if trabajo.operacion == 'esperar':
        print('Esperando %d segundos...' % trabajo.entrada)
        time.sleep(trabajo.entrada)
        print('Hecho!')
    else:
        raise NotImplementedError('Operación "%s" no soportada.' % trabajo.operacion)

Con esto, tenemos un precioso consumidor de trabajos:

$ python consumidor.py
Esperando trabajos...
Se ha recibido un trabajo: {"entrada": 5, "operacion": "esperar"}
Esperando 5 segundos...
Hecho!

Colas en Python con RabbitMQ y Pika (parte 1)

La cola (queue) es un objeto clásico en programación (y, por desgracia, en la vida real también), que consiste en un conjunto de elementos ordenados que deben consumirse siguiendo el concepto FIFO (First In, First Out: primero en entrar, primero en salir).

Un ejemplo típico de esta estructura son las colas de trabajos. Un determinado dispositivo o recurso sólo puede atender un trabajo a la vez y organiza sus tareas en una cola que se va sirviendo por orden. Esto es lo que todos tenemos en nuestro ordenador para los trabajos de impresión.

En general, cuando tenemos uno o varios procesos que deben atender una gran carga de trabajo (no interactiva), no es buena idea que se sirvan las peticiones en paralelo, sino que es más factible organizar dichas tareas en una cola, de manera que los procesos no se colapsen.

Un ejemplo de eso podría ser un sistema de entrega de mensajes como Twitter. Cada vez que escribes un tuit, dicho mensaje entra en una o varias colas de trabajo desde las cuales unos procesos especializados van a a ir haciéndolo llegar a todos tus seguidores. El sistema es muy rápido, pero asíncrono al fin y al cabo.

Desde el punto de vista de la Ingeniería del Software, se trata de una solución del clásico problema del productor-consumidor.

Una cola de paso de mensajes

En este artículo vamos a dar un primer paso, haciendo un flujo de "mensajes" (es decir, fragmentos de información) basado en colas. Esto aún no se puede considerar una cola de trabajos, dado que habría que añadir mucha más funcionalidad, que veremos en futuros artículos. Lo llamaremos "cola de mensajes" para no confundir términos.

Generar un flujo de mensajes entre procesos en Python, es relativamente sencillo usando dos herramientas de software libre basadas en el protocolo AMPQ: el servicio de mensajería RabbitMQ y el cliente Pika.

RabbitMQ es un gestor de mensajes (message broker) desarrollado en el exótico lenguaje Erlang (que curiosamente lleva el nombre del inventor de la Teoría de Colas).

Su único trabajo consiste en aceptar y despachar "mensajes" (donde un mensaje es cualquier fragmento de información). En este sentido, RabbitMQ hace las veces de buzón, de oficina de correos y de cartero, todo en uno. La diferencia con los sistemas de correo electrónico es que no trata con personas, sino con procesos. Usaremos este servicio de bajo nivel como base para nuestro sistema de cola de trabajos.

Para el intercambio de mensajes, RabbitMQ implementa un protocolo estándar llamado AMQP (Advanced Message Queue Protocol). Dado que es un estándar, podemos interactuar con él usando cualquier lenguaje que tenga una librería cliente para dicho protocolo. En este artículo vamos a usar la librería Pika en Python, pero hay otras similares para montones de lenguajes, incluidos Java, PHP, Javascript, C# y un largo etcétera.

NOTA: En este artículo no voy a entrar en la instalación. Cuento con que tienes RabbitMQ instalado y corriendo en tu máquina local. Desde la página de RabbitMQ se puede descargar el software para Linux, Windows y Mac OS X. También se puede obtener mediante Docker. Por su parte, Pika está disponible en el repositorio estándar de Python. Los programas de ejemplo en este artículo están probados en Python 3.5.2 sobre Ubuntu Desktop 16.

Comprobar que el servicio está arriba

Ttanto el emisor como el receptor deberán conectar con el servicio RabbitMQ, que deberá estar corriendo en la máquina local. Para ver que está corriendo, podemos usar el siguiente comando:

$ sudo rabbitmqctl status

Si el servicio está corriendo, recibiremos una larga serie de información de depuración. Si no es así, obtendremos un error de conexión.

La instalación por defecto en Ubuntu nos levanta el servicio automáticamente, así que en condiciones normales debería estar arriba. En caso contrario, se puede usar el mismo comando rabbitmqctl para arrancar el servicio, pero es más recomendable delegar este trabajo en la gestión de servicios del sistema operativo. Así pues, por ejemplo en Ubuntu haríamos:

$ sudo service rabbitmq-server start

Una vez que estamos seguros que tenemos el servicio de mensajería arriba, podemos pasar a nuestro primer programita.

Prueba de concepto: envío de mensajes

Vamos a hacer la prueba de concepto mínima de envío y recepción de mensajes.

La idea es crear un proceso que espera la llegada de un mensaje, y otro que hace el envío. Así comprobaremos que la emisión y recepción funcionan correctamente.

Ambos procesos, el receptor (consumidor) y el emisor (productor) van a tener que hacer la conexión con el servicio de mensajería RabbitMQ de la máquina local. Para ello, instanciamos la clase pika.BlockingConnection. De los tipos de conexión que soporta Pika, el "blocking" es el más sencillo de utilizar, dado que nos resulta muy familiar y similar al paradigma clásico de apertura y cierre de un canal de comunicación, igual que si de una conexión a base de datos se tratara.

A esta clase (denominada adaptador), le pasamos una instancia de pika.ConnectionParameters, a la que definimos un único parámetro diciendo que el destino es la propia máquina local.

con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

Una vez instanciado, abrimos el canal llamando al método channel().

ch = con.channel()

Siempre que sea posible, nos cuidaremos de liberar recursos cerrando la conexión cuando ya no haga falta:

con.close()

Una vez tenemos acceso al servicio de mensajería, podemos empezar a usar una cola para enviar mensajes o recibirlos. Pero hay que ser precavidos, puesto que nada nos garantiza que la cola ya exista. Así que lo primero será declarar la cola, por si acaso no existiera:

ch.queue_declare(queue='prueba')

En caso de que la cola ya exista, esto no tendrá efecto alguno.

Para enviar un mensaje a la cola, hay también varios métodos. El más simple es basic_publish(), que recibe como mínimo dos parámetros:

• exchange: El intercambiador es una abstracción que nos permite configurar la entrega de los mensajes como una caja negra, de manera que en principio no tendríamos que decir a qué cola o colas van los mensajes ni cómo se hace la entrega. Esto es muy potente, pero complejo. Para una simple prueba podemos obviar todo esto dejando vacío este parámetro. En ese caso, añadiremos el parámetro adicional routing_key diciendo directamente el nombre de la cola de destino.
• body: El cuerpo del mensaje. Recordemos que puede ser cualquier trozo de información, ya sea binaria o de caracteres.

La orden queda de esta manera:

ch.basic_publish(exchange='', routing_key='prueba', body='Hola Mundo')

Así pues, ya podemos poner todo junto para nuestro envío de mensajes. Lo vamos a poner en un módulo llamado productor.py:

import pika

# Establecer conexión
con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = con.channel()

# Declarar la cola
ch.queue_declare(queue='prueba')

# Publicar el mensaje
ch.basic_publish(exchange='', routing_key='prueba', body='Hola Mundo')
print("Mensaje enviado.")

# Cerrar conexión
con.close()

Si lanzamos este programa, terminará con éxito a pesar de que no hay ningún proceso esperando los mensajes:

$ python productor.py
Mensaje enviado.

$

Esto es así porque la entrega de mensajes es asíncrona. El mensaje quedará en la cola hasta que alguien lo consuma. En cualquier momento, podemos ver las colas que hay y cuántos mensajes tienen esperando, mediante este comando:

$ sudo rabbitmqctl list_queues
Listing queues ...
prueba 1
$

NOTA: El procedimiento basic_publish() no tiene mecanismos para comprobar si el mensaje realmente ha llegado a la cola. En nuestro ejemplo, el programa productor sólo sabe que se ha intentado el envío. El protocolo permite usar acuses de recibo, pero de momento obviaremos esa parte.

Prueba de concepto: recepción de mensajes

La recepción no es tan sencilla. Al ser un sistema asíncrono, necesitamos de alguna forma "suscribirnos" a la cola para recibir una llamada cada vez que haya un mensaje. Esto se hace mediante una función tipo callback. Por ejemplo, definimos lo siguiente:

def recepcion(ch, method, properties, body):
    print("Se ha recibido el siguiente mensaje: %s" % body)

Es obligatorio declarar estos parámetros, aunque por el momento vamos a ignorarlos a excepción del body.

Para enganchar esta función a la cola, usamos el método basic_consume().

ch.basic_consume(recepcion, queue='prueba', no_ack=True)

El primer parámetro es la referencia a la función que hemos definido. Recuerda que se trata de un callback, es decir, no usamos recepción() con paréntesis porque en este caso queremos usar la referencia (o puntero), y no el resultado de la función.

Los otros dos parámetros que hemos usado es la cola a la que nos queremos enganchar (queue) y un booleano (no_ack), que indica a la cola que borre el mensaje en cuanto sea servido, sin esperar confirmación del proceso consumidor. Esto simplifica la programación, y también hace el proceso más eficiente, a costa de un cierto riesgo de pérdida de datos.

Por último, nos ponemos en espera de mensajes;

ch.start_consuming()

Este comando dejará bloqueado el proceso eternamente hasta que se aborte con CTRL+C. Mientras tanto, la función recepcion() recibirá un callout cada vez que tenga un mensaje en cola.

El código completo de consumidor.py es el siguiente:

import pika

# Establecer conexión
con = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = con.channel()

# Declarar la cola
ch.queue_declare(queue='prueba')

# Definir la función callback
def recepcion(ch, method, properties, body):
    print("Se ha recibido el siguiente mensaje: %s" % body)

# Enganchar el callback
ch.basic_consume(recepcion, queue='prueba', no_ack=True)

# Poner en espera
print('Esperando mensajes...')
ch.start_consuming()

Lanzamos el consumidor:

$ python consumidor.py
Esperando mensajes...
Se ha recibido el siguiente mensaje: b'Hola Mundo'

El consumidor ha recibido inmediatamente el mensaje que estaba en la cola, y se ha quedado esperando más. Si en otro terminal lanzamos el productor seguirá recibiéndolos.

Obsérvese que nuestro "Hola Mundo" va precedido de una b, indicando que es de tipo byte. Es una confirmación de lo que dijimos antes: que lo que la cola maneja son datos binarios y podremos poner ahí cualquier cosa que se nos ocurra.

Si miramos en la cola, veremos que ya no hay mensajes en espera:

$ sudo rabbitmqctl list_queues
Listing queues ...
prueba 0
$

Hemos conseguido crear un flujo de mensajes entre dos procesos independientes en la misma máquina. Esto se podría hacer en distintas máquinas, simplemente cambiando el "localhost" por la dirección de la máquina donde reside la cola. Como ves, el potencial que nos ofrece esta herramienta y su capacidad de abstracción es muy grande, con la ventaja de que hay un protocolo subyacente del que no tenemos que preocuparnos.

En futuros artículos, iremos añadiendo más funcionalidad hasta convertir este simple programa en un completo gestor de colas de trabajos.

TCP/IP (8): Manejo de la tabla de rutas

Comandos para el manejo de la tabla de rutas

Los comandos para mantener la tabla de rutas se desarrollaron inicialmente en Unix. En otros sistemas, los comandos son similares. 

Vamos a ver ejemplos de comandos válidos para Linux. Casi todos ellos requieren permisos de administrador, así que probablemente necesites usar "sudo" para ejecutarlos. 

route add 128.6.2.0 128.6.4.1 1

Este comando añade una entrada en la tabla para que las comunicaciones con la red 128.6.2 se dirijan a la pasarela 128.6.4.1, con métrica 1. La primera dirección del par se puede cambiar por la palabra default, para crear una entrada por defecto.

ifconfig ie0 128.6.4.4 netmask 255.255.255.0

Esto añade una entrada que especifica que la dirección propia 128.6.4.4 corresponde a la interfaz de red ie0, con la máscara 255.255.255.0. Este comando se incluye en el fichero de arranque, y si la máquina tiene varias interfaces se añade una orden por cada uno.

netstat -n -r

Esto presenta por pantalla el estado actual de la tabla de rutas, con diferentes estadísticas, etc. Básicamente, netstat indica para cada nodo o red de destino, el gateway a usar, la interfaz de red (si hay varios) y una serie de banderas (flags) que indican diferentes aspectos de rutado. Los más importantes son:

* G: Indica si la ruta usa una pasarela para llegar al nodo de destino. Cuando esta bandera no está presente, el destino está conectado directamente, y la dirección de pasarela indicada no está siendo usada en realidad.

* U: Indica si la ruta está activa (up) en este momento. Cuando se han registrado varios fallos de conexión consecutivos, esta bandera se elimina, para que se deje de usar esta ruta.

* H: Indica que esta ruta se refiere a un nodo (host), y no a una red. Normalmente, las entradas de la tabla de rutas no son de este tipo, aunque puede darse el caso que esto sea necesario.

* D: Indica que esta ruta ha sido establecida dinámicamente por medio de una redirección ICMP.Una entrada en la tabla de rutas indicará la ruta por defecto, con la palabra default en la columna de direcciones de destino.

Manejo de errores en el rutado

Cuando un nodo que no sea una pasarela, por error en el rutado, recibe un datagrama no destinado a él, no debe intentar reenviarlo a su correcto destinatario. En vez de ello, el datagrama debe ser descartado. Las razones para establecer esta norma son varias:

* La acción correctiva evitaría que el error en el rutado se hiciera patente, de manera que éste persistiría.

* El rutado añadido genera mayor tráfico de red de manera innecesaria, además del gasto adicional en tiempo de CPU de la máquina que reenvía.

* Cualquier error simple puede reproducirse en cadena hasta convertir la red en un caos.

* Los nodos no participan de los mecanismos de control de rutado por medio de los cuales las pasarelas se informan unas a otras y mantienen actualizadas sus tablas. Las propias pasarelas no podrían detectar el rutado adicional producido por los nodos, y las consecuencias serían imprevisibles.

Separación estricta del direccionamiento IP

Durante su viaje desde el nodo origen al destino final, los datagramas no se ven alterados, a excepción del tiempo de vida y el checksum, que se recalculan. Esto implica que, aunque un datagrama atraviese muchas pasarelas, las direcciones de origen y destino en la cabecera del mismo siempre reflejan las del origen y destino finales.

Cuando una pasarela se dispone a reenviar un datagrama al salto siguiente, la dirección IP de dicho salto no es guardada en ningún lugar del datagrama, sino que simplemente es puesta en manos de la interfaz de red, el cual se encargará de resolver la dirección MAC correspondiente, empaquetar el datagrama en una trama física y transmitirlo. Luego, tanto la dirección IP como la física son desechadas.

Este comportamiento resulta paradójico; se puede pensar que si el rutado manejase directamente la dirección física, resultaría mucho más eficiente. Si en vez de un sólo datagrama, el nodo origen ha emitido una secuencia larga de ellos hacia el mismo destino, como suele ocurrir, resulta especialmente costoso resolver una y otra vez la misma dirección MAC. Las razones para que IP trabaje así son varias:

* Se ha de respetar escrupulosamente la frontera de direcciones entre el software IP y la red física. Si no se hiciera así, todo el sentido del protocolo de la red de redes y la independencia de la red física se vendrían abajo.

* La tabla de rutado es una interfaz muy transparente entre el software IP y el software de alto nivel con el que los administradores monitorizan y controlan las rutas. Habitualmente se suelen examinar las tablas de rutado, resultando mucho más fácil este trabajo usando direcciones IP.

* Al respetar la frontera se facilita el desarrollo y actualización de los restantes protocolos TCP/IP, que de esta manera sólo tienen que entender de direcciones de la red de redes.

Por tanto, IP establece una estricta frontera de direcciones entre el esquema de direccionamiento IP y otros esquemas, como pueden ser las direcciones MAC que se usan en Ethernet y otros protocolos similares. La traducción entre un tipo de dirección y otra la realizan otros protocolos que veremos en próximos artículos.

Protocolos para gestión del rutado

Para que las pasarelas mantengan actualizadas sus tablas de rutas, se impone la presencia de unos protocolos que les permitan mantenerse en contacto e intercambiar información de manera automática. 

Los protocolos clásicos de gestión del rutado son:

RIP (Routing Information Protocol): Es un protocolo diseñado para redes de pequeño y mediano tamaño, donde las velocidades de las diferentes líneas no difieren demasiado. Tiene una serie de limitaciones:

• No se puede usar en redes donde los paquetes puedan atravesar más de 15 pasarelas.
• No puede repartir el tráfico en conexiones en paralelo (dos o mas líneas con el mismo inicio y fin).
• No se adapta a los cambios en la carga de la red.
• No maneja bien las rutas alternativas cuando hay demasiada diferencia de velocidades entre ellas.
• No es estable en las redes dónde hay contínuos cambios de líneas o de pasarelas.

La principal ventaja que tiene este protocolo es ser el único que es totalmente estándar. Así pues, es fundamental cuando se usan pasarelas de diferentes fabricantes.

EGP (External Gateway Protocol): Es un protocolo especialmente diseñado para redes dispersas unidas por medio una línea central. Permite intercambiar información sobre accesibilidad con la línea central.

Por otra parte, se han desarrollado también protocolos específicos para monitorización:

SGMP: Permite recoger información y hacer ajustes en los parámetros de los pasarelas y otros elementos de la red. El correspondiente conjunto de programas interfaz pueden ejecutarse desde cualquier nodo de la red. Está aceptado como un estándar por la mayoría de los fabricantes; existe un limitado conjunto de datos que se espera que proporcione, así como unos mecanismos comunes para la información añadida propia.

SNMP: Es la generación siguiente al SGMP. Prácticamente efectúa la misma misión, pudiendo monitorizar un conjunto de parámetros más completo, llamado MIB (Management Information Base). Esta base estándar de parámetros es el resultado de las aportaciones más o menos desorganizadas de los diferentes fabricantes a su antecesor, una vez pasadas por los comités de estandarización.

CMIP: Es el protocolo oficialmente propuesto por el ISO dentro de su especificación CMIS de monitorización.

TCP/IP (7): Introducción al rutado IP

Cómo encontrar la mejor ruta, la eterna pregunta.

Nota lingüística: La palabra correcta en castellano para tomar un camino es encaminar, pero en estos artículos seguiremos la jerga informática rutado o enrutamiento, por ser lo habitual en los textos técnicos.

La separación de tareas entre los distintos protocolos es básica para cualquier sistema de red. Se trata de una versión moderna del clásico divide y vencerás. En el caso de TCP/IP, este reparto de tareas ha sido una de las claves de su éxito, que ha hecho que tres décadas después sigan siendo la base de nuestra vida digital. 

Ya vimos en los anteriores artículos que TCP se encarga de negociar la conexión entre los extremos finales y de asegurarse que el flujo de datos llega a su destino intacto y en el orden correcto.

Hoy vamos a ver la otra cara de la moneda. En este caso, el protocolo IP (Internet Protocol) se encarga de encontrar la ruta (saltos intermedios) entre los dos extremos e ir entregando los paquetes uno a uno y paso a paso en dicha ruta.

IP sólo entiende de paquetes (datagramas en la literatura original). No tiene el concepto de conexión entre extremos. Se encarga de entregar cada paquete individual a su destinatario. La ruta no es mantenida ni recordada, sino que cada paquete se entrega por separado, y por consiguiente suele pasar que no sigan la misma ruta y que no lleguen ordenados. 

Haciendo una analogía, TCP sería el vendedor que negocia la venta con el cliente y se asegura de que recibe la mercancía, mientras que IP sería el mensajero que lleva el paquete a la dirección de destino y decide la ruta sin importarle los detalles de la venta o cuántos paquetes se han acordado.

Este planteamiento es muy diferente de otros protocolos orientados a la conexión, como por ejemplo la red de telefonía. En esos instantes que pasan desde que marcamos un número de teléfono hasta que comienza la llamada, se produce la negociación de una ruta fija que va a mantenerse durante toda la llamada. Eso es muy distinto a lo que ocurre cuando pedimos una descarga por Internet.

Buscando caminos

La tarea de encontrar el camino para llevar un datagrama a su destino se llama rutado, enrutamiento o encaminamiento (routing).

IP supone que cada equipo está unido a una determinada red local. Se da por sentado que puede enviar datagramas a máquinas de su misma red de forma natural. A esto le podemos llamar rutado interno o entrega directa, porque está completamente contenido en los mecanismos de la red física. Sin embargo, cuando es necesario enviar un datagrama a una máquina de otra red diferente, entran en acción las pasarelas (gateways), en lo que podemos llamar rutado IP o entrega indirecta. El rutado en la red de redes (o rutado IP) es análogo al rutado en la red local, porque busca un camino por el que enviar los datos; sin embargo se diferencia en que debe atravesar varias redes físicas heterogéneas.

Para saber cuándo debe usar la entrega directa y cuándo hacer uso de una pasarela, el nodo emisor debe comparar su dirección con la del destinatario. Si la parte denominada netid (identificador de red) coincide, es que están en la misma red y, por lo tanto, se debe usar la entrega directa. 

Esta operación es muy eficiente, porque se basa, como veremos, en la aplicación de una operación AND a los bits de la dirección IP. Esto se realiza con una única instrucción de código máquina.

Cada dirección IP (en su versión 4, que es la más extendida) tiene 32 bits, que suelen escribirse usando cuatro números de 8 bits cada uno. Una parte de esos bits identifican la red (netid) y el resto identifican al nodo (dispositivo individual). 


La máscara de subred es una serie de bits tal que, aplicada la operación AND, generan el netid a partir de la dirección IP. En la figura se muestra un ejemplo de aplicación de la máscara.

En próximos artículos nos extenderemos en las clases de direcciones que existen y también veremos cómo cambian el direccionamiento en la versión 6 del protocolo.

Pasarelas

Una pasarela (gateway) es un sistema que conecta una red con una o más redes diferentes. En teoría, pueden ser ordenadores que se ocupen de otras cosas, aparte de este trabajo. Básicamente puede serlo cualquier máquina que esté conectada a dos o más interfaces de red. 

En la práctica, al equipo que hace la función de pasarela se le suele llamar router, aunque este suele incluir otras funciones y dicha denominación se presta a confusión. En este texto usaremos el término pasarela, que es el concepto funcional clásico que utilizaron los creadores de IP.

IP está pensado para que cada interfaz tenga su propia dirección. P.e., tenemos una máquina que está conectada a las redes 128.6.4 y 128.6.3; esto significa que puede ser capaz de recibir un datagrama de una de ellas y reenviarlo a la otra. Esto es posible porque posee dos direcciones; una para cada red. En cada caso, la máquina que le envía el datagrama lo hará con la dirección que le corresponda en su red.

Si bien cualquier sistema de uso general puede hacer de pasarela, para lograr mayor eficiencia, se usan sistemas dedicados, con un hardware especializado. Los algoritmos de rutado se programan en esas máquinas directamente en el firmware y en chips especializados de una manera extremadamente eficiente.

Rutado controlado por tabla

Las dos preguntas fundamentales en el rutado IP son: ¿Cómo sabe un nodo qué pasarela usar para llegar a un destino determinado? y ¿cómo sabe una pasarela a dónde enviar un datagrama?

En una red de redes TCP/IP, los nodos finales también intervienen en el rutado. Concretamente, cuando un nodo genera un datagrama, decide en primera instancia a qué pasarela va a dirigirlo. Cada pasarela puede proporcionarle un mejor camino hacia determinados destinos.

Cada sistema tiene una tabla de direcciones de redes, cada una asociada al gateway mejor situado para enviar algo a dicha dirección. La tabla contiene básicamente pares de direcciones (R, G), donde G es la dirección del siguiente gateway en el camino hacia la red R. El sistema usará dicho gateway en cada envío que tenga que hacer a esa red.

En el ejemplo de la figura, una máquina de la red 128.6.3 tendrá como mejor pasarela al 128.6.3.12, para cualquier comunicación fuera de la red local. Por otra parte, cualquier sistema de la red 128.6.4, usará el 128.6.4.1 como primer paso. Luego, el datagrama será reenviado a través del mencionado 128.6.3.12.

Este mecanismo de rutado se llama rutado con salto al siguiente porque cada pasarela sólo especifica un paso en el camino hacia el nodo destino; no tiene que saber el camino completo. Es necesario, además, que el gateway al que apunta cada paso siguiente pueda ser alcanzado a través de una sola red física.

La tabla de acceso suele también contener información que servirá a los algoritmos de rutado para medir lo "lejana" que está la red de destino; esto se llama métrica (metric). En el caso más básico, la métrica corresponde simplemente en contar el número de pasarelas que habrá de atravesar el mensaje.

Las tablas de rutas son el corazón del sistema de rutado; de hecho, los diferentes algoritmos no son más que formas más o menos sofisticadas de configurar y mantener dichas tablas.

Las entradas de la tabla de rutas pueden tener diferentes características. Se pueden diferenciar:

* Entradas referentes a redes y subredes, o en casos especiales las que se refieren a un sólo nodo.

* Entradas que tienen como destinatario a una pasarela, o a un nodo concreto.

* Entradas que se han añadido directamente a la tabla, o aquellas que se han puesto automáticamente a petición de una pasarela, por medio de un ICMP redirect (lo veremos en otro artículo). Estas entradas se llaman dinámicas.

* Entradas activas (up), o que se han desactivado (down) por problemas con las conexiones.

Optimización de las tablas de rutas

Si un sistema tuviese una tabla de rutas con entradas para los miles de redes que hay en Internet, ésta se haría demasiado grande. Además, es inútil tener una tabla así cuando muchas redes sólo acceden al "mundo exterior" por una o dos pasarelas.

La solución es tener rutas por defecto de las que hacer uso cuando no se posea ninguna referencia en la tabla para una dirección dada.

Cuando hay varias pasarelas en una red, éstas deben de tener alguna manera de avisar a un nodo de que, para una determinada dirección, no es la mejor opción. Esto se hace mediante el protocolo ICMP (Internet Control Message Protocol).

El procedimiento es el siguiente:

1. Un nodo quiere enviar un datagrama a una determinada dirección.
2. Como no posee entrada en su tabla para esa dirección, lo envía a su pasarela por defecto.
3. Este reconoce que no es el mejor camino, y reenvía el datagrama a otra pasarela de la misma red.
4. Una vez hecho esto, envía un mensaje vía ICMP al nodo origen diciendo: "No soy el mejor gateway para acceder a la red 18, usa mejor el 128.6.4.1". Este mensaje se llama "ICMP redirect".
5. El software de red añadirá entonces una entrada en la tabla para que los próximos mensajes hacia la red 18 sean enviados directamente a la pasarela 128.6.4.1.

Así pues, la mayoría de los sistemas empiezan a trabajar con pasarelas por defecto, para luego ir dejando que sean las propias pasarelas los que les vayan diciendo las mejores rutas.

Ejemplo de tabla de rutas

Como ejemplo de una tabla de rutas muy simple, nada mejor que la de nuestro propio PC. Para verla, sólo tienes que abrir una línea de comandos y ejecutar "route list" (Windows) o "sudo route -n" (Linux). En el resultado encontraremos, entre otras cosas algo así:

  

En esta tabla hay dos interfaces de red:

• 192.168.1.171 es la interfaz real (en este caso el adaptador WiFi).
• 127.0.0.1 es el "loopback" o "localhost", un bucle por el cual cualquier aplicación puede comunicarse con el propio PC.

Todos los enlaces "en vínculo" son los que son accesibles directamente, mientras que la puerta de enlace 192.168.1.1 es la pasarela del router inalámbrico al que estoy conectado, y que me da salida a cualquier dirección IP (0.0.0.0). Este es el "gateway por defecto", que en esta configuración tan simple es el único que tengo.

Unas direcciones IP especiales son las 192.168.1.255 y 255.255.255.255, que son el "broadcast" o "difusión", que pueden usar mi equipo cuando quiera que cualquiera dentro de la red local le escuche. Esto se usa en muchas aplicaciones requieren "descubrir" lo que hay en la red, como las de compartir archivos.

Por últmo, otra dirección especial es 224.0.0.0, la cual se usa para "multicast", que también tendrá su artículo próximamente.