El patrón Reactor es una arquitectura de diseño para el manejo concurrente de operaciones de entrada/salida (I/O), especialmente útil en programación de redes. Se basa fundamentalmente en la distribución de eventos.

Componentes Principales del Patrón Reactor

Este patrón se compone principalmente de dos elementos:

• Reactor: Responsable de monitorear y distribuir eventos. Los eventos típicos incluyen solicitudes de conexión y operaciones de lectura/escritura. Utiliza mecanismos como epoll_wait para supervisar múltiples descriptores de archivo de manera eficiente.
• Pool de Recursos de Procesamiento: Conjunto de hilos o procesos encargados de ejecutar las tareas de manejo de eventos (lectura y escritura de datos, lógica de negocio).

Existen diversas implementaciones del patrón Reactor, clasificadas según el número de instancias de Reactor y el número de hilos/procesos en el pool de procesamiento:

1. Single Reactor, Single Thread/Process
2. Single Reactor, Multiple Threads/Processes
3. Multiple Reactors, Single Thread/Process
4. Multiple Reactors, Multiple Threads/Processes

1.1 Single Reactor, Single Thread/Process

Ventajas: Simplicidad en la implementación al evitar la complejidad de la comunicación entre procesos/hilos y la sincronización de datos. Desventajas: No aprovecha completamente los procesadores multinúcleo y puede experimentar latencias si las operaciones de manejo de eventos son prolongadas. Casos de uso: Escenarios no intensivos en cómputo o donde las operaciones se completan rápidamente.

2. Single Reactor, Multiple Threads/Processes

Ventajas: Supera la limitación de no aprovechar los CPU multinúcleo mediante el uso de múltiples hilos o procesos. Desventajas: Un único Reactor maneja toda la supervisión y respuesta de eventos, lo que puede convertirse en un cuello de botella ante picos de concurrencia elevados. Sin embargo, en ciertos contextos como el proyecto "Lower", esta situación puede no ser crítica.

En el proyecto "Lower", se adoptó un enfoque de Single Reactor, Single Process.

Roles: Reactor, Acceptor y Handler

• Reactor: Supervisa los eventos y los dirige.
• Acceptor: Gestiona la aceptación de nuevas conexiones.
• Handler: Se encarga del procesamiento de datos una vez establecida la conexión.

Detallando su funcionamiento en "Lower":

• El Reactor, mediante Epoll_wait(), detecta eventos y los canaliza hacia un objeto TcpChannel (actuando como Acceptor) para nuevas conexiones, o hacia un objeto TestItemChannel (actuando como Handler) para otros eventos.

• Para eventos de conexión, el TcpChannel (Acceptor) utiliza accept() para establecer la conexión y crea un TestItemChannel (Handler) para gestionar las interacciones posteriores. Se emplea un bucle while(accept()) para aceptar todas las conexiones entrantes disponibles de manera no bloqueante. ```cpp

  void EpollEx::acceptTcpChannel() { while ((connfd = accept(m_listenfd, (sockaddr *)&clientSocketAddress, &socketAddressLength)) > 0) { // ... lógica para crear y añadir el canal ... auto channelPtr = ChannelFactory::createTcpChannel(connfd, chipName, shared_from_this()); addChannelToMap(channelPtr); } }

• Para eventos de lectura/escritura, el TestItemChannel (Handler) lee los datos entrantes y los transfiere a un proceso TestItem para su procesamiento. Una vez que el proceso TestItem completa el procesamiento, devuelve el resultado al TestItemChannel, que a su vez escribe la respuesta al cliente. ```cpp

  void EpollEx::processReadWrite(const struct epoll_event &event) { // ... verificación y obtención del puntero al canal ... if (isOk && (event.events & EPOLLIN)) // Datos recibidos { isOk = (channelPtr->readData() > 0); } // Manejo de escritura para TcpChannel y TestItemChannel if (isOk && (event.events & EPOLLOUT)) // Envío de datos { isOk = (channelPtr->writeData() > 0); } // ... resto de la lógica ... }

• La comunicación entre procesos (padre-hijo) en el enfoque Single Reactor, Multiple Processes puede ser compleja, utilizando mecanismos como colas de mensajes o señales. Una alternativa más sencilla es el enfoque Single Reactor, Multiple Threads, debido a la menor complejidad en la comunicación entre hilos.

Resolución de Bloqueo en read()

Para evitar que read() bloqueante impida el manejo de otros eventos, se emplea una combinación de sockets no bloqueantes y multiplexación de I/O (como epoll). Esto permite al epoll_wait monitorear múltiples descriptores de archivo y notificar sobre aquellos listos para operaciones sin bloquear la ejecución principal.

bool EpollEx::run()
{
   while (m_run > 0)
   {
       // Reactor: Espera eventos
       std::int32_t eventsSize = epoll_wait(m_epfd, eventsPtr.get(), EPOLL_EVENTS_MAX_LENGTH - 1, 100); // Timeout en milisegundos
       // Pool de Procesamiento: Maneja los eventos detectados
       processEvents(eventsPtr.get(), eventsSize);
       heartBeat(); // Mantenimiento periódico
   }
   return true;
}
   

Funciones de Utilidad (Ejemplos)

Cálculo de CRC16 MODBUS

Función para calcular el checksum CRC16 utilizando el algoritmo MODBUS.

unsigned short CRC16_MODBUS(unsigned char *Data, unsigned int DataLen)
{
   unsigned short CRCin = 0xffff;
   unsigned short CRCret = 0;
   for (int i = 0; i < DataLen; i++)
   {
       CRCin ^= *Data;
       for (int j = 0; j < 8; j++)
       {
           if (CRCin & 0x01)
           {
               CRCin = CRCin >> 1;
               CRCin = CRCin ^ 0xa001; // Generado a partir de 0x8005
           }
           else
           {
               CRCin = CRCin >> 1;
           }
       }
       Data++;
   }
   // Reordenamiento de bytes para el resultado final
   CRCret = ((CRCin & 0x00FF) << 8) | ((CRCin & 0xFF00) >> 8);
   return CRCret;
}
   

Aálisis de Petición

Función para validar y parsear datos de una petición, incluyendo verificación CRC.

int ParseReq(char *buf, unsigned char &ReqId)
{
   unsigned short CrcResu = 0;
   unsigned short ResCrc = 0;
   // Verificación de cabecera (ej. 0xF1, 0x01)
   if ((unsigned char)buf[0] == 0xF1 && buf[1] == 0x01)
   {
       // Log de éxito
       TestItemPreprocess::instance().logInfo("ParseReq succ!");
       // Cálculo del CRC sobre los primeros 11 bytes
       CrcResu = CRC16_MODBUS((unsigned char *)buf, 11);
       // Extracción del CRC recibido (bytes 11 y 12)
       ResCrc = (buf[11] << 8) | buf[12];
       // Comparación de CRC
       if (CrcResu == ResCrc)
       {
           ReqId = buf[2]; // Extracción del ID de la petición
           return 0; // Éxito
       }
   }
   return -1; // Fallo
}