c# - tutorial - zmq socket types

0MQ: ¿Cómo usar ZeroMQ de una manera segura para subprocesos?

Question

Leí la guía ZeroMq y me topé con lo siguiente:

NO DEBE compartir los zócalos ØMQ entre las roscas. Los zócalos ØMQ no son a prueba de roscas. Técnicamente es posible hacer esto, pero exige semáforos, bloqueos o mutexes. Esto hará que su aplicación sea lenta y frágil. El único lugar en el que está remotamente sano para compartir sockets entre hilos es en enlaces de idiomas que necesitan hacer magia como la recolección de basura en sockets.

y más tarde:

Recuerde: No use ni cierre los enchufes, excepto en el hilo que los creó.

También entendí que el Context ZeroMQ es seguro para subprocesos.

Si una clase se registra para un evento de otra clase, en .Net, este evento podría invocarse desde un hilo diferente al hilo en el que se creó el oyente.

Creo que solo hay dos opciones para poder enviar algo a través de ZeroMQ-Sockets desde un manejador de eventos:

• Sincronice el thread de invocación de evento con el thread en el que se creó el ZeroMQ- Socket
• Cree un nuevo ZeroMQ- Socket / obtenga el ZeroMQ- Socket existente para el subproceso dentro del manejador de eventos utilizando el contexto seguro ZeroMQ de threadsafe

Parece que la Guía 0MQ para desalentar a la primera y no creo que la creación de un ZeroMq-Socket nuevo para cada subproceso sea eficaz / la forma de hacerlo.

Mi pregunta :
¿Cuál es el patrón correcto (la forma en que debe ser) para publicar mensajes a través de 0MQ desde un manejador de eventos?

Además, los autores de la guía tuvieron en mente el ZeroMQ-Binding para .Net cuando escribieron:

El único lugar en el que está remotamente sano para compartir sockets entre hilos es en enlaces de idiomas que necesitan hacer magia como la recolección de basura en sockets. ?

Aquí hay un código de muestra para enfatizar mi problema / pregunta:

public class ExampleClass
{
    public event EventHandler<ByteEventArgs> SomethinIsCalledFromAnotherThread;
}

public class ByteEventArgs : EventArgs
{
    public byte[] BytesToSend;
}


public class Dispatcher
{
    ZMQ.Context ctx;

    public Dispatcher(ZMQ.Context mqcontext, ExampleClass exampleClassInstance)
    {
        this.ctx = mqcontext;
        exampleClassInstance.SomethinIsCalledFromAnotherThread += new EventHandler<ByteEventArgs>(exampleClass_SomethinIsCalledFromAnotherThread);
    }

    void exampleClass_SomethinIsCalledFromAnotherThread(object sender, ByteEventArgs e)
    {
        // this method might be called by a different thread. So I have to get a new socket etc?
        using (var socket = ctx.Socket(ZMQ.SocketType.PUSH))
        {
            // init socket etc..... and finally: 
            socket.Send(e.BytesToSend);
        }
        // isn''t that too much overhead?
    }
}

Answer 1

En .net framework v4 y superiores, puede utilizar la recopilación concurrente para resolver este problema. Es decir, patrón productor-consumidor. Varios subprocesos (controladores) pueden enviar datos a una cola segura de subprocesos y solo un único subproceso consume datos de la cola y los envía utilizando el socket.

Aquí está la idea:

sendQueue = new BlockingCollection<MyStuff>(new ConcurrentQueue<MyStuff>());
// concurrent queue can accept from multiple threads/handlers safely
MyHandler += (MyStuff stuffToSend) => sendQueue.Add(stuffToSend);

// start single-threaded data send loop
Task.Factory.StartNew(() => {
    using(var socket = context.Socket()) {
        MyStuff stuffToSend;
        // this enumerable will be blocking until CompleteAdding is called
        foreach(var stuff in sendQueue.GetConsumingEnumerable())
            socket.Send(stuff.Serialize());
    }
});

// break out of the send loop when done
OnMyAppExit += sendQueue.CompleteAdding;

Answer 2

No olvides echar un vistazo al transporte inproc. Podría ser útil usar inproc: // sockets para la comunicación entre hilos y tener un hilo que abra sockets para hablar con otros procesos / servidores.

Aún necesita al menos un socket por subproceso, pero los en proceso no implican en absoluto la capa de red IP.

Answer 3

Puedes crear muchos sockets 0MQ, ciertamente tantos como tengas hilos. Si creas un socket en un hilo y lo usas en otro, debes ejecutar una barrera de memoria completa entre las dos operaciones. Cualquier otra cosa resultará en extraños fallos aleatorios en libzmq, ya que los objetos de socket no son seguros para subprocesos.

Hay algunos patrones convencionales, aunque no sé cómo se relacionan específicamente con .NET:

1. Crea zócalos en los hilos que los utilizan, punto. Comparta contextos entre hilos que están estrechamente vinculados en un proceso, y cree contenidos separados en hilos que no están estrechamente vinculados. En la API de C de alto nivel (czmq), se denominan subprocesos adjuntos y separados.
2. Cree un socket en un subproceso principal y pase en el momento de la creación de subprocesos a un subproceso adjunto. La llamada de creación de hilo ejecutará una barrera de memoria completa. A partir de entonces, use el zócalo solo en el hilo secundario. "uso" significa recv, send, setsockopt, getsockopt y close.
3. Cree un socket en un hilo y utilícelo en otro, ejecutando su propia barrera de memoria completa entre cada uso. Esto es extremadamente delicado y si no sabe qué es una "barrera de memoria completa", no debe hacer esto.