NamedPipeClientStream BeginRead, BeginWrite - deadlock? – .NET – Fórum – Programujte.com

Ahoj,

chtěl jsem použít asynchronní komunikaci. Narazil jsem však na problém. Pokud zavolám BeginRead před BeginWrite, první volání BeginWrite sice data odešle, ale nedojde k zavolání callback funkce. Na příchozí data nereaguje. Pokud zavolám nejdříve BeginWrite a v jejím callback BeginRead, komunikace funguje - cyklus zahájil vysílání -> dokončil vysílání -> zahájil příjem -> dokončil příjem by fungoval, ale mojí aplikaci nevyhovuje. Jak dosáhnout stavu kde asynchronní čtení by pracovalo nezávisle na asynchronním zápisu?

hu

#1 hlucheucho
zkousel sis rozjet 2 vlakna v jenom synchrone cist a ve druhem synchrone zapisovat ? me to prijde jednoduzsi na implementaci.

Asynchronní mělo snadnou implementaci - 2 callback funkce, slibovalo možnost opakovaného volání BeginWrite a tím automatické tvoření fronty, s jednoduchým počitadlem se dala sledovat délka fronty a podle toho přizpůsobit rychlost vysílání.

U více vláken budu muset řešit jejich vzájemnou spolupráci - jak si budou předávat data, synchronizaci. Dělal jsem to takhle v C++, byla to otročina.

Ještě mne napadla možnost oddělené streamy - jeden in, druhý out a věřit, že budou pracovat nezávisle na sobě. Ale i tak se vytrácí "kouzlo jednoduchosti"

hu

Není možné, že se ten callback zavolá jen pokud ta operace probíhá asynchronně a pokud se provede synchronně, tak se tedy nezavolá? Koukni na ten buffer/result.

Nevím o co se pokoušíš, ale mám pocit, že na to jdeš špatně.

Na jednom streamu nelze provádět současně synchronní a asynchronní operace. Pokus o synchronní vysílání a asynchronní příjem skončil očekávaným výsledkem - nefungovalo to. Jestliže volám BeginRead nebo BeginWrite, provádím asynchronní operaci. Samotná operace je údajně prováděna v samostatném vlákně. Pro každé volání Begin... musí být volání End..., v příkladech to dělají uvnitř callbacku - nic jsem nevymýšlel, jen jsem to opsal. Bohužel jsem nenašel popis pravidel nebo omezení pro současné použití asynchronního čtení a zápisu na jednom streamu. Popsané chování naznačuje, že se oba směry vzájemně vylučují - během probíhajícího vysílání nelze zahájit příjem a naopak.

hu

Nakonec jsem narazil na toto  http://www.codeproject.com/…am-BeginRead.

hu

Myslel jsem to tak, že pokud tu operaci lze provést okamžitě bez blokování (typicky čtení, pokud v interním bufferu jsou nějaká data), tak neprobíhá na pozadí. Tak asynchronní I/O ve Windows funguje. Podle zdrojáků PipeStreamu to ale vypadá, že by se měl callback vyvolat vždy. Byl to jen nápad.

Pokud jsem článek z codeproject pochopil správně, používají jeden semafor společně pro obě metody. Zatím zkouším synchronní čtení v samostatném vlákně ...

hu

Synchronní se chová stejně nepochopitelně - "přijímací" vlákno zastaví na volání metody Read a spolu s tím přestane hlavní vlákno vysílat. Už mi došly nápady :(

hu

#9 hlucheucho
a nastavil su tu pipu na asynchroni zpracovani v konstruktoru?

tady je kod co funguje pro pralalelni cteni i zapis (je to jen ukazka takze neni moc hezky)

        static void Main(string[] args)
        {
            int serverMessage = 1;
            int clientMessage = 1;
            int sleepTime = 1;
            int bulkWrite = 200;
            int stopLimit = 5000;
            object serverSyncRoot = new object();
            object clientSyncRoot = new object();
            Task.Factory.StartNew(() =>
            {
                var server = new NamedPipeServerStream("moje", PipeDirection.InOut, 1, PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
                Console.WriteLine("Server-connecting");
                server.WaitForConnection();
                Console.WriteLine("Server-connected");
                Task.Factory.StartNew(() =>
                {
                    using (var reader = new StreamReader(server))
                    {
                        while (true)
                        {
                            var rawMessage = reader.ReadLine();
                            int recievedMessage = Convert.ToInt32(rawMessage);
                            if (recievedMessage%bulkWrite == 0)
                                Console.WriteLine("Server-read: " + recievedMessage);
                        }
                        Console.WriteLine("Exiting server reader");
                    }
                });

                Task.Factory.StartNew(() =>
                {
                    var writer = new StreamWriter(server);
                    writer.AutoFlush = true;
                    while (true)
                    {
                        Interlocked.Increment(ref serverMessage);
                        lock(serverSyncRoot)
                        writer.WriteLine(serverMessage);
                        
                        if (serverMessage%bulkWrite == 0)
                            Console.WriteLine("Server-writer 1: " + serverMessage);
                        Thread.Sleep(sleepTime);
                        if (serverMessage > stopLimit)
                            break;
                    }
                    Console.WriteLine("Exiting server writer 1");

                });

                Task.Factory.StartNew(() =>
                {
                    var writer = new StreamWriter(server);
                    writer.AutoFlush = true;
                    while (true)
                    {
                        Interlocked.Increment(ref serverMessage);
                        lock (serverSyncRoot)
                       writer.WriteLine(serverMessage);
                        
                        if (serverMessage%bulkWrite == 0)
                            Console.WriteLine("Server-writer 2: " + serverMessage);
                        Thread.Sleep(sleepTime);
                        if (serverMessage > stopLimit)
                            break;
                    }
                    Console.WriteLine("Exiting server writer 2");

                });
                Thread.Sleep(Timeout.Infinite);

            });

            Task.Factory.StartNew(() =>
            {
                Console.WriteLine("Client-connecting");
                var client = new NamedPipeClientStream("localhost", "moje", PipeDirection.InOut, PipeOptions.Asynchronous);
                client.Connect();
                Console.WriteLine("Client-connected");
                Task.Factory.StartNew(() =>
                {
                    using (var reader = new StreamReader(client))
                    {
                        while (true)
                        {
                            var rawMessage = reader.ReadLine();
                            int recievedMessage = Convert.ToInt32(rawMessage);
                            if (recievedMessage%bulkWrite == 0)
                                Console.WriteLine("Client-read: " + recievedMessage);
                        }
                        Console.WriteLine("Exiting client reader");
                    }
                });

                Task.Factory.StartNew(() =>
                {
                    var writer = new StreamWriter(client);
                    writer.AutoFlush = true;
                    while (true)
                    {
                        Interlocked.Increment(ref clientMessage);
                        lock (clientSyncRoot)
                            writer.WriteLine(clientMessage);
                        if (clientMessage%bulkWrite == 0)
                            Console.WriteLine("Client-writer 1: " + clientMessage);
                        Thread.Sleep(sleepTime);
                        if (clientMessage > stopLimit)
                            break;
                    }
                    Console.WriteLine("Exiting client writer 1");

                });

                Task.Factory.StartNew(() =>
                {
                    var writer = new StreamWriter(client);
                    writer.AutoFlush = true;
                    while (true)
                    {
                        Interlocked.Increment(ref clientMessage);
                        lock (clientSyncRoot)
                            writer.WriteLine(clientMessage);
                        if (clientMessage%bulkWrite == 0)
                            Console.WriteLine("Client-writer 2: " + clientMessage);
                        Thread.Sleep(sleepTime);
                        if (clientMessage > stopLimit)
                            break;
                    }
                    Console.WriteLine("Exiting server writer 2");

                });

                Thread.Sleep(Timeout.Infinite);
            });

            Console.WriteLine("Everything is started");
            Console.ReadKey();
        }

   hustýýýý.  Akorát jsem zvědav za jak dlouho to pochopím a za jak dlouho se mi to podaří upravit na klienta. "Idiots Guide" k tomu není?

hu

By mne zajímalo jak to je s obousměrností roury. Jsem zkusil udělat jednoduchého klienta v C++ a chová se stejně: 

HANDLE roura;
char z;
DWORD zapsal, cetl;
byte buf[128];

int main()
{
	roura = CreateFile(L"\\\\.\\pipe\\z001", GENERIC_READ | GENERIC_WRITE, 0, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL);
	if (roura == NULL) {
		cout << "nejde otevrit roura";
		cin >> z;
		return 0;
	}
	cout << "roura otevrena" << endl;
	Sleep(3000);
	WriteFile(roura, "text", 4, &zapsal, NULL);

	ReadFile(roura, buf, 13, &cetl, NULL);
	buf[cetl] = 0;
	cout << (char*)buf << endl;

	cin >> z;
	CloseHandle(roura);
	return 0;
}

Pokud server odesle data před voláním WriteFile z ukázkového kódu, ReadFile tyto data přečte. Pokud server odesle data až po volání WriteFile, ReadFile se nevrátí, ke čtení dat nedojde. Už tomu vůbec nerozumím   

hu