# Diseño de una DHT
*Autores: Carlos Aznar Olmos, David Pisonero Fuentes y Pedro Rico Pinazo*

## 1. Diseño global de la práctica

Mediante el diseño que se va a implementar y que se explicará más detenidamente en los siguientes apartados, se pretende que cumpla con las características requeridas:

* *Coherencia:* Se pretende que las tablas de los servidores replicados deban tener los mismos valores, esto es, el usuario deberá obtener el mismo valor independientemente del servidor al que se refiera.
Con la implementación de nuestro diseño, la coherencia se consigue mediante la comunicación con el Ensemble de Zookeeper, ya que es el que gestiona las operaciones y las posesiones de las tablas con replicación.
Las operaciones put y remove nunca se transmitirán a las tablas locales directamente sino que estas operaciones se realizan a través de un znode. A su vez se establecen *watchers* sobre los hijos de estos znodes en cada tabla, las cuales están gestionadas por el servidor para que cuando se añada una nueva operación, ésta se ejecute sobre las tablas locales. El objetivo final es que la ejecución de todas las tablas se realicen en el mismo orden garantizando asi la coherencia.
Por otro lado, la operación get se realizará a través de TCP.

* *Tolerancia a fallos*: Se pretende que la aplicación tolere un número de fallos, al menos un fallo. Por tanto, en el caso de que un servidor se caiga, se deberá de crear uno nuevo para tolerar el número de fallos requeridos. Con la creación de este nuevo servidor, habrá que transferirle el mismo estado de las tablas asociadas.
Nuestra aplicación tolerará $r-1$ fallos si cada partición tiene una réplica en $r$ servidores. Por tanto, los servidores son los encargados de almacenar las diferentes particiones de la tabla hash distribuida. Además existen conexiones TCP/IP entre los servidores, lo que supone que si un servidor se cae, cuando se incorpora uno nuevo, éste recibirá las particiones correspondientes.

* *Disponibilidad*: A través de la disponibilidad, los servidores podrán invocar a cualquier otro servidor.
Esta disponibilidad se conseguirá a traves de las conexiones TCP/IP que existen entre los servidores. Esto garantiza que cuando se inscribe un nuevo servidor, ya sea porque hay otro servidor que se ha caído o porque se crea un servidor nuevo, este solicitará las tablas a los demás miembros.

* *Transparencia*: el software planteado, por diseño, garantiza que la interfaz con el usuario es única y permite realizar operaciones sobre la DHT de la misma forma que se harían si no se tratase de una aplicación distribuida

## 2. Arquitectura del despliegue

Para implementar las funcionalidades anteriormente descritas, hemos diseñado la siguiente arquitectura:


En ella podemos observar, en primer lugar, a los servidores que actúan como clientes de Zookeeper. Ellos van a ser los encargados de mantener las diferentes particiones o tablas que forman la DHT, tal y como se ve en el dibujo.

Lo siguiente en lo que reparamos es que estos servidores se conectan a un Ensemble de Zookeeper, que será el encargado de gestionar las operaciones que se llevan a cabo sobre las tablas, y coordinar a los diferentes servidores. Así, se establece una comunicación entre dichos servidores y el Ensemble, necesaria para poder guardar las operaciones en forma de znodes, configurar watchers, recibir eventos al activarse estos watchers, etc. Se profundizará en este funcionamiento en los apartados siguientes.

Finalmente, también podemos destacar que existen conexiones TCP/IP entre los diferentes servidores. Ello se debe a que, en caso de caída de algún servidor, el nuevo que se incorpore ha de recibir las particiones correspondientes en aras de mantener la tolerancia a fallos del sistema. Para ello, aunque el nuevo servidor utilice los znodes para averiguar cuáles son las direcciones IP del resto de servidores de los que va a recuperar las tablas que él mismo ha de guardar, el envío de dichas tablas no se hará por Zookeeper, pues es una herramienta de coordinación, no una base de datos, de modo que se empleará para ello una conexión IP corriente.

## 3. Diseño software de los servidores

Los servidores de nuestro sistema van a ser los encargados de almacenar las diferentes particiones de la tabla hash distribuida. Básicamente, la DHT se encuentra subdividida en diferentes tablas en función del rango de índices, y cada servidor va a poseer algunas de estas tablas, de modo que las diferentes partes de la DHT se encuentren replicadas para conseguir tolerancia a fallos.

Los datos son parejas clave-valor de tipo `<String, Integer>` almacenados en un `HashMap` de Java. Sobre estos datos, empleando los servidores, se van a poder llevar a cabo operaciones de `put` (para ingresar nuevos datos), `remove` (para eliminar datos) y `get` (para obtener un valor determinado). De este modo, cada vez que se solicite realizar una de estas operaciones proporcionando una clave, la función de hash permitirá obtener el índice, a partir del cual se sabrá a qué servidor hay que acudir, que será aquel que posea la tabla en cuestión, manteniendo la transparencia de cara al usuario.

Estos servidores se comunicarán con el Ensemble de Zookeeper para gestionar las operaciones y la asignación de las tablas con replicación, que será dinámica. Los servidores que poseen una de las réplicas de una partición pueden registrarse como miembros de dicha partición publicando su dirección IP y su puerto en Zookeeper de forma que los servidores nuevos pueden solicitarles las copias de las particiones mediante peticiones TCP/IP.

En cuanto a las operaciones, a priori, hemos establecido que las peticiones de `put` y `remove` se escriban en ZK, pero, en el caso de la operación `get`, se envíe una petición por TCP a algún miembro de la partición, que devolverá el valor correspondiente.

Por último, las principales clases que componen el diseño propuesto tienen las siguientes funciones:

- `DHTMain`: encargada de ofrecer una interfaz textual al usuario e inicializar las diferentes clases.
- `TableManager`: para manejar las tablas, y ejecutar sobre ellas las operaciones.
- `Listener`: para atender las peticiones TCP recibidas.
- `UserManager`: como de interfaz entre las peticiones del usuario y el resto de clases.
- `ZkManager` la parte encargada de la comunicación con Zookeeper.

En el siguiente apartado, se puede observar el diagrama de las clases referidas.

## 4. Diagrama de clases

El diagrma de clases desarrollado para el funcionamiento de la aplicación es el siguiente:


## 5. Zookeeper

### 5.1. Árbol de znodes


El árbol de znodes que se emplea en Zookeeper es el que se puede observar en la figura (los znodes efímeros se marcan en rojo). En primer lugar, existe un znode en la raíz representando cada una de las particiones del DHT. Bajo cada uno de dichos znodes cuelga, una vez la DHT está funcionando:

- Un znode `lock`, que determina que existe un servidor cuya secuencia de particiones replicadas comienza en dicha partición, como se verá más adelante.
- Un znode `mbs` que tiene un hijo con la dirección IP y puerto por cada servidor que contiene una copia local de la partición y por tanto puede atender peticiones.
- Un znode `ops` que tiene como hijos las operaciones que deben realizar todos los servidores sobre sus copias locales de forma secuencial. Cada vez que un servidor ejecuta una operación, añade un nuevo hijo al znode de la operación

Este planteamiento simplifica la lógica del programa en la medida en que no se necesita llevar a cabo un mapeo entre servidores y particiones. Dicho mapeo queda implícito en el propio árbol de znodes de Zookeeper.

### 5.2 Tolerancia a fallos

En primer lugar es importante indicar que en la implementación se considera que, en estado estacionario (sin fallos en los servidores), el número de servidores coincide con el número de particiones de la DHT.

Se puede garantizar la tolerancia a $r-1$ fallos si cada partición tiene una réplica en $r$ servidores diferentes. Para distribuir las réplicas entre los servidores cumpliendo dicho criterio, se siguen dos reglas:

- Cada servidor mantiene en local una secuencia **contigua** de $r$ particiones (considerando que la secuencia de particiones es circular)
- No pueden existir dos servidores cuya secuencia comience en la misma partición

Para cumplir con la segunda regla, se lleva a cabo una asignación dinámica. Todo servidor que quiera comenzar su secuencia en una determinada partición debe añadir un hijo efímero `lock` bajo el znode de dicha partición. Si se consigue crear el hijo con éxito, se garantiza que mientras el servidor sea correcto no podrá existir otro servidor que comience la secuencia en la misma partición.

Como ejemplo de todo lo anterior, si la DHT tiene por ejemplo 4 particiones y 3 réplicas y un servidor comienza su secuencia en la partición 2, este mantendría en local las particiones 2, 3 y 0.

Para conseguir llegar al escenario anterior, se detalla a continuación el algoritmo que se lleva a cabo cuando aparecen nuevos servidores.

Cuando un servidor nuevo aparece y quiere hacerse con una copia local de una partición, en primer lugar consulta los hijos del znode `mbs`. Si todavía no existen miembros en la partición se pueden presentar dos escenarios:

- El servidor ha establecido un `lock` sobre dicha partición: crea la partición local y se isncribe como miembro añadiendo un hijo en `mbs` con su dirección para que otros servidores puedan hacerle consultas sobre las tablas.
- No se trata de la primera partición de la secuencia de particiones replicadas por el servidor: establece un watcher sobre `mbs` esperando a que otro servidor cree la tabla.

Si un servidor que quiere adquirir una copia de la partición encuentra algún miembro bajo el znode `mbs`, establece una nueva operación en modo efímero bajo el znode `ops` de tipo `CHECKPOINT`. Los servidores que se encuentran dicha operación establecen un hijo bajo ella con su dirección para que el nodo nuevo pueda solicitarles la tabla por TCP. Una vez el nuevo servidor ha obtenido la tabla, borra el `CHECKPOINT`, marca como realizadas todas las operaciones anteriores (y las borra si corresponde) y continúa con normalidad.

### 5.3 Coherencia

Para garantizar la coherencia entre las diferentes tablas, las peticiones de operaciones `put` y `remove` sobre la DHT recibidas de los usuarios no son en ningún caso transmitidas a las tablas locales directamente. La respuesta es siempre escribir las operaciones a realizar en el znode `ops`. Al mismo tiempo, cada servidor establece watchers sobre los hijos de dicho znode en cada tabla que gestiona y cada vez que se añade una nueva operación en Zookeeper, se ejecuta sobre las tablas locales. De esa forma, la petición inicial del usuario se acaba traduciendo eventualmente en la ejecución de la operaciones sobre todas las tablas en el mismo orden, garantizando por tanto la coherencia entre tablas.

Las operaciones son representadas por znodes persistentes. Si fueran efímeros, podría darse una situación en la que una operación que ha sido ejecutada solo por algunos servidores desapareciese como consecuencia de la caída de algún servidor. Por ello, es necesario establecer un criterio para borrarlas de forma segura.

Cuando un servidor ejecuta una operación, añade un hijo efímero al znode que la representa. Si al añadirlo se alcanzan el número máximo de miembros, se borra la operación. De esta forma, se garantiza que solo se borra una operación cuando se ha ejecutado sobre todas las copias locales.

Hay que remarcar que para las operaciones `get` no se lleva a cabo ninguna gestión a través de Zookeeper, sino que se llevan a cabo a través de TCP. Simplemente, el hecho de que eventualmente las tablas tienden a un estado consistente entre sí garantiza que las operaciones `get` acaban devolviendo el valor esperado.