ZeroMQ - asynchronní klient

[Clyde]

Ahoj, snažím se porozumět NetMQ (potažmo zeromq), ale nějak mi to pořád nejde.
Mám server, který přijímá zprávy z N klientů a vrací jim odpovědi (ten mi snad funguje), ale potřebuji udělat klienta, který dokáže odesílat zprávy asynchronně. Příklad: v hlavním vlákně se připojím na server. Vytvořím Task, který během své existence čile komunikuje se serverem. Následně vytvořím další Tasky a počkám si na jejich dokončení. Na serveru používám pro propojení soketu s vnějškem a inproc soketů objekt Proxy. Ten ale blokuje hlavní vlákno. Dá se to nějak vyřešit asynchronně?

Ukázka kódu na klientovi:

Kód: [Vybrat]

using var frontend = new DealerSocket();
using var backend = new RouterSocket();

frontend.Connect("tcp://127.0.0.1:5555");
backend.Bind("inproc://workers2");

var queue = new List<Task>();

for (int i = 0; i < 1; ++i)
{
   int id = i;

   var task = Task.Run(() =>
   {
      Worker(id);
   });

   queue.Add(task);
}

//var proxy = new Proxy(frontend, backend);
//proxy.Start();
//??

Task.WaitAll(queue.ToArray());


[Reklama]

[IDontCare]

https://netmq.readthedocs.io/en/latest/async-await/

[Clyde]

https://netmq.readthedocs.io/en/latest/async-await/

Na tohle jsem už samozřejmě taky narazil, ale asi nejsem moc důvtipný, protože mi to stejně nepomohlo vyřešit můj problém. Pokud bych z každého Tasku přímo připojoval na server, pak by se mi předpokládám vytvářelo nové TCP spojení pro každý úkol, což by asi nebylo dobré. A pokud bych se připojoval na inproc soket, kter mám v příkladu uvedený, chybí mi most mezi inproc a soketem spojeným se serverem... to by mohla dělat Proxy, ale ta blokuje hlavní vlákno Je vůbec cesta jak to používat bez blokace hlavního vlákna? Chtěl bych použít NetMQ v gui aplikaci, ale tam samozřejmě nechci. aby mi zamrzlo hlavní vlákno při odesílání a přijímání zpráv.
Mohl byste být s Vaší radou trošku konkrétnější? Děkuji.

[alex6bbc]

Pokud bych z každého Tasku přímo připojoval na server, pak by se mi předpokládám vytvářelo nové TCP spojení pro každý úkol, což by asi nebylo dobré.

nejsem C#, takze jen obecneji. proc by to nebylo dobre vytvaret nove tcp spojeni?
v asynchronnim tasku si udelate spojeni, vyresite nezavisle na ostatnich zdrojich ukol a task i spojeni se muze ukoncit.

[Clyde]

Pokud bych z každého Tasku přímo připojoval na server, pak by se mi předpokládám vytvářelo nové TCP spojení pro každý úkol, což by asi nebylo dobré.

nejsem C#, takze jen obecneji. proc by to nebylo dobre vytvaret nove tcp spojeni?
v asynchronnim tasku si udelate spojeni, vyresite nezavisle na ostatnich zdrojich ukol a task i spojeni se muze ukoncit.

Nejsem odborník, takže mě klidně opravte. Zaprvé to píšou přímo v dokumentaci, že to není moc vhodné, takže bych tomu i věřil. A zadruhé si myslím, že vybudování TCP spojení je nákladné, kdybychom hovořili v kontextu velkého množství Tasků.

[Reklama]

[TakyPatrik]

V Unity jsem na tohle používal Coroutines - jinak by se hra zasekla při čekání na odpověď ze serveru.
https://en.wikipedia.org/wiki/Coroutine#Implementations_for_C%23

[LadislavBohm]

Nevim, jestli jsem problem uplne pochopil a taky nevim, jestli tu nemichate dve veci: 1) asynchronni zpracovani tasku na server 2) zpracovani vice tasku najednou (paralelne).

Nicmene i tak se to da resit ruzne:

Task.WaitAll je mimochodem blokujici, nevim jestli je to jen pro ukazku, ale tim si nepomuzete. Proste nastartujte Task.Run, jako to mate ve Vasi ukazce. Tim si regulujte, kolik paralelnich tasku chcete na serveru zpracovavat. V nich muzete v cyklu napriklad pres CancellationToken kontrolovat, zda se nemaji ukoncit (napriklad pri vypnuti serveru).
Uvnitr techto tasku muzete pouzivate async await jak je uvedeno v dokumentaci.

Kod jsem nijak netestoval, ale obecne s frontou by to mohlo pracovat treba takto:

Kód: [Vybrat]

var cts = new CancellationTokenSource();
      Task.Run(async () => {
        while (!cts.IsCancellationRequested) {
          try {
            var (message, more) = await client.ReceiveFrameStringAsync();
            //process message
          }
          finally {
            //nejaky delay pred kontrolou dalsi zpravy?
            Task.Delay(TimeSpan.FromMilliseconds(100));
          }
        }
});

Predpokladam, ze z jedne instance 'client' muze cist vice vlaken. Takze tech Task.Run muzete nastartovat vice a token idealne predava i do te metody ReceiveFrameStringAsync, pokud to podporuje.

[Clyde]

Nevim, jestli jsem problem uplne pochopil a taky nevim, jestli tu nemichate dve veci: 1) asynchronni zpracovani tasku na server 2) zpracovani vice tasku najednou (paralelne).

Nicmene i tak se to da resit ruzne:

Task.WaitAll je mimochodem blokujici, nevim jestli je to jen pro ukazku, ale tim si nepomuzete. Proste nastartujte Task.Run, jako to mate ve Vasi ukazce. Tim si regulujte, kolik paralelnich tasku chcete na serveru zpracovavat. V nich muzete v cyklu napriklad pres CancellationToken kontrolovat, zda se nemaji ukoncit (napriklad pri vypnuti serveru).
Uvnitr techto tasku muzete pouzivate async await jak je uvedeno v dokumentaci.

Kod jsem nijak netestoval, ale obecne s frontou by to mohlo pracovat treba takto:

Kód: [Vybrat]

var cts = new CancellationTokenSource();
      Task.Run(async () => {
        while (!cts.IsCancellationRequested) {
          try {
            var (message, more) = await client.ReceiveFrameStringAsync();
            //process message
          }
          finally {
            //nejaky delay pred kontrolou dalsi zpravy?
            Task.Delay(TimeSpan.FromMilliseconds(100));
          }
        }
});

Predpokladam, ze z jedne instance 'client' muze cist vice vlaken. Takze tech Task.Run muzete nastartovat vice a token idealne predava i do te metody ReceiveFrameStringAsync, pokud to podporuje.

Ahoj, možná jsem se vyjádřil špatně, ale neřeším stranu serveru (Task.WaitAll byl jen pro ukázku, vím že je blokující). Spíš mě zajímá klient. Chci přes jedem ZeroMQ soket multiplexovat komunikaci z několika vláken. Je to možné, lze pouít např zmíněný objekt proxy, ale ten je blokující a nemá metody jako IsRunning a podobně (metody pro kontrolu a ovládání běhu). Myslím si, že řešení bude objekt Poller, ale stále se mi ho nepodařilo rozběhat ke své spokojenosti. Řešením by samozřejmě mohlo bt spustit Proxy ve vedlejším vlákně, ale jak už jsem zmínil, pak se špatně hlídí jeho stav.