c++ - Usando ZeroMQ junto con Boost:: ASIO

sockets boost-asio

Question

Tengo una aplicación C ++ que está usando ZeroMQ para algunos mensajes. Pero también debe proporcionar una conexión SGCI para un servicio web basado en AJAX / Comet.

Para esto necesito un socket TCP normal. Podría hacerlo con tomas normales de Posix, pero para mantenerme portátil en varias plataformas y hacer mi vida más fácil (espero ...) estaba pensando en utilizar Boost :: ASIO.

Pero ahora tengo el choque de ZMQ que quiere usar su propio zmq_poll() y ASIO es io_service.run() ...

¿Hay alguna manera de hacer que ASIO trabaje junto con 0MQ zmq_poll() ?

¿O hay otra forma recomendada de lograr dicha configuración?

Nota: Podría resolverlo usando múltiples hilos, pero es solo un pequeño núcleo / CPU que ejecutará ese programa con una cantidad muy baja de tráfico SCGI, por lo que el multihilo sería una pérdida de recursos ...

Answer 1

Al final, descubrí que hay dos posibles soluciones:

• Sam Miller es donde usamos el ciclo de eventos de ASIO
• El bucle de eventos de ZeroMQ al obtener los descriptores de archivos ASIO a través de los métodos .native() del acceptor y el socket e insertarlos en la matriz de zmq_pollitem_t

He aceptado la respuesta de Sam Miller, ya que para mí es la mejor solución en el caso SCGI en el que se crean y finalizan conexiones constantemente nuevas. Manejar el así cada cambio de matriz zmq_pollitem_t es una gran molestia que se puede evitar usando el ciclo de eventos ASIO.

Answer 2

Después de leer la documentación here y here , específicamente este párrafo

ZMQ_FD: recupera el descriptor de archivo asociado con el socket La opción ZMQ_FD recuperará el descriptor de archivo asociado con el socket especificado. El descriptor de archivo devuelto se puede utilizar para integrar el socket en un bucle de evento existente; la biblioteca ØMQ señalará cualquier evento pendiente en el socket de una manera desencadenada haciendo que el descriptor de archivo esté listo para la lectura.

Creo que puedes usar null_buffers para cada zmq_pollitem_t y diferir el bucle de eventos a un io_service , pasando zmq_poll() alto completamente a zmq_poll() . Sin embargo, parece que hay algunas advertencias en la documentación mencionada, notablemente

La capacidad de lectura del descriptor de archivo devuelto no indica necesariamente que los mensajes estén disponibles para ser leídos o pueden escribirse en el socket subyacente; las aplicaciones deben recuperar el estado real del evento con una recuperación posterior de la opción ZMQ_EVENTS.

Entonces, cuando se dispare el controlador de uno de sus sockets zmq, tendrá que trabajar un poco más antes de manejar el evento, creo. Pseudo-código no completado está debajo

const int fd = getZmqDescriptorSomehow();
boost::asio::posix::stream_descriptor socket( _io_service, fd );
socket->async_read_some(
    boost::asio::null_buffers(),
    [=](const boost::system::error_code& error)
    {
       if (!error) {
           // handle data ready to be read
       }
     }
);

tenga en cuenta que no tiene que usar una lambda aquí, boost::bind a una función miembro sería suficiente.

Answer 3

La solución es sondear tu io_service también en lugar de run ().

Consulte esta solución para obtener información sobre poll() .

Usar encuesta en lugar de ejecutar le permitirá sondear las conexiones de zmq sin ningún problema de bloqueo.

Answer 4

Obtener el socket a ZeroMQ es la parte más pequeña de la batalla. ZeroMQ se basa en un protocol que se superpone a través de TCP, por lo que tendrá que volver a implementar ZeroMQ dentro de un Boost.Asio io_service personalizado si realiza esta ruta. Me encontré con el mismo problema al crear un servicio ENet asíncrono usando Boost.Asio simplemente tratando de captar tráfico de un cliente ENet usando un servicio Boost.Asio UDP. ENet es un protocolo tipo TCP en capas sobre UDP, por lo que todo lo que logré en ese momento fue capturar paquetes en un estado prácticamente inútil.

Boost.Asio está basado en plantillas, y las plantillas de uso de io_service incorporadas básicamente envuelven la biblioteca de sockets del sistema para crear el servicio TCP y UDP. Mi solución final fue crear un io_servicio personalizado que envolviera la biblioteca ENet en lugar de la biblioteca de sockets de sistemas, permitiéndole usar las funciones de transporte de ENet en lugar de tener que volver a implementarlas usando el transporte UDP incorporado.

Lo mismo se puede hacer para ZeroMQ, pero ZeroMQ ya es una biblioteca de red de muy alto rendimiento que ya proporciona E / S asincrónica. Creo que puede crear una solución viable al recibir mensajes usando la API existente de ZeroMQ y pasar los mensajes a un grupo de hilos io_service. De esta forma, los mensajes / tareas se manejarán de forma asíncrona utilizando el patrón de reactores de Boost.Asio sin tener que volver a escribir nada. ZeroMQ proporcionará la E / S asíncrona, Boost.Asio proporcionará los controladores / trabajadores de la tarea asincrónica.

El io_service existente también se puede acoplar a un socket TCP existente, lo que permite que el threadpool maneje tanto TCP (HTTP en su caso) como ZeroMQ. Es completamente posible en tal configuración para los manejadores de tareas ZeroMQ acceder a los objetos de sesión de los servicios TCP, lo que le permite enviar los resultados del mensaje / tarea ZeroMQ a un cliente TCP.

Lo siguiente es solo para ilustrar el concepto.

// Create a pool of threads to run all of the io_services.
std::vector<boost::shared_ptr<boost::thread> > threads;
for(std::size_t i = 0; i < thread_pool_size_; ++i) {
    boost::shared_ptr<boost::thread> thread(new boost::thread(boost::bind(&boost::asio::io_service::run, &io_service_)));
    threads.push_back(thread);
}

while (1) {
    char buffer [10];
    zmq_recv (responder_, buffer, 10, 0);
    io_service_.post(boost::bind(&server::handle_zeromq_message, buffer, this));
}