Synchronizace vlaken v C#

[anonym]

Mám takový problém. Dal bych krk na to, že v Javě Semafor a Mutex nerozlišuje, jaké vlákno si zámek vzalo. C#, ta bestie, mi dělá tohle:

Mutex m = new Mutex();
m.WaitOne();
m.WaitOne(); <<< tady by se melo vlakno zastavit ale nezastavi se, protože si ten vyčůraný mutex pamatuje, že je to stejné vlákno

Jenže, já potřebuju Mutex a Semafor klasický, něco co má metodu Acquire() a nestará se, jaké vlákno to chce. A pokud nemůže získat zámek, bude prostě čekat. Jak na to pls?

[Reklama]

[Filip Jirsák]

V Javě i v jakémkoli jiném alespoň trochu příčetném frameworku je vlastníkem zámku pro synchronizaci vláken vždy vlákno, takže když se pokusí získat zámek, který už vlastní, dostane ho. Žádné jiné řešení totiž nedává smysl – kdyby se to chovalo tak, jak byste chtěl, na tom třetím řádku dojde k deadlocku a to vlákno už se nikdy neprobudí, protože by čekalo na uvolnění zámku, k jehož uvolnění by ale bylo potřeba, aby to vlákno běželo.

Takže evidentně potřebujete něco jiného. Zkuste raději popsat problém, který řešíte.

[Neviditelný]

To, co popisujete je chování rekurzivních mutexů, které jsou v microsoftích frameworkách běžné. Kromě toho, že rozumně navržený kód by rekurzivní synchronizační primitiva neměl potřebovat mi není jasné, proč chcete z jednoho vlákna zamykat mutex dvakrát a při tom druhém zamknutí čekat. Jakým způsobem zařídíte, že vám to vlákno nevydeadlockuje?

[Filip Jirsák]

Předpokládám,že nechce synchronizovat vlákna, ale něco jiného, takže to odemkne z jiného vlákna. Ale co doopravdy chce, to musí napsat anonymní tazatel, já jenom hádám.

[anonym]

Ok našel jsem, jmenuje se to AutoResetEvent.

Potřeboval jsem udělat toto:

IEnumerable<T> ThreadedPipe.IterateInThread(IEnumerable<T> input)

Tzn. input, ktery bych normalne prochazel primo pres foreach, budu prochazet pres tuto metodu, a ta metoda spustí načítání inputu v samostatném threadu. Můžu si takto paralelizovat proudové zpracování. Neříká se tomu Pipe and Filters? Tak tohle co jsem udělal je Pipe, Filtr na který to napojím líně načítá XMLko z disku.

Pokud má někdo lepší řešení, tak sem s nim, já se u toho pěkně zapotil a stejně mi to trochu hapruje.

[Reklama]

[anonym]

Ještě tu připojuju zdroják, kdyby někdo měl chuť se podívat na mé řešení 

Kód: [Vybrat]

public class ThreadedPipe<T>
    {
        Queue<T> buffer = new Queue<T>();
        int bufferSize = 100;

        AutoResetEvent taskFinished = new AutoResetEvent(false);
        AutoResetEvent taskEnqueued = new AutoResetEvent(false);
        AutoResetEvent[] taskEnqueuedOrFinished = new AutoResetEvent[2];

        AutoResetEvent readyEnqueue = new AutoResetEvent(false);


        public ThreadedPipe(int bufferSize)
        {
            taskEnqueuedOrFinished[0] = taskFinished;
            taskEnqueuedOrFinished[1] = taskEnqueued;

            this.bufferSize = bufferSize;
        }

        private void Enqueue(T item)
        {
            lock(buffer)
            {
                buffer.Enqueue(item);
            }
        }
        private T Dequeue()
        {
            lock (buffer)
            {
                return buffer.Dequeue();
            }
        }

        private int Count()
        {
            lock (buffer)
            {
                return buffer.Count;
            }
        }
        public IEnumerable<T> IterateInThread(IEnumerable<T> input)
        {
            buffer.Clear();

            Task t = new Task(() =>
            {
                foreach (var entry in input)
                {
                    if (Count() >= bufferSize)
                    {
                        readyEnqueue.WaitOne();
                    }

                    Enqueue(entry);
                    taskEnqueued.Set();
                    Trace.WriteLine("Enqueue. Count:" + Count());
                }
                taskFinished.Set();
            });
            t.Start();

            while (true)
            {
                if(Count()>0)
                {
                    yield return Dequeue();
                    readyEnqueue.Set();
                }else
                {
                    int num = WaitHandle.WaitAny(taskEnqueuedOrFinished);
                   
                    if(num == 0) // Task finished
                    {
                        break;
                    }
                }
            }
        }

    }


[Neviditelný]

Chápu správně, že jde o problém producent-konzument kdy producent čte nějaká data a láduje je do bufferu, odkud si je konzument sebere? Pokud je to tak, nešlo by to vyřešit semaforem?

[anonym]

To uplne neni producent/consumer



Spíše je to Pipe and Filters

[anonym]

Hm ono to je asi jedno... jo je to producent/konzumer. Ale semafor mi bude nanic.

[Neviditelný]

Hm ono to je asi jedno... jo je to producent/konzumer. Ale semafor mi bude nanic.

Proč myslíte? Mám dojem, že třeba příklad na použití semaforů v Qt řeší hodně podobný, ne-li identický problém a nemusí zamykat celý buffer.

[anonym]

Haha už jsem to našel!

https://docs.microsoft.com/en-us/dotnet/standard/collections/thread-safe/blockingcollection-overview

Jde se refaktorovat.

[kvr kvr kvr]

Pominu-li race condition v každém druhém bloku, což by asi bylo lépe vidět, kdy by třída měla dokončené finální konzumenty, tak... v Java by na to byl ideální Executor, s limitovaným počtem threadů, které můžou paralelně procesovat. Předpokládám, že C# má něco podobného - namátkou ThreadPool....? BlockingQueue je struktura na nižší úrovni, která se dá pro Executor využít.

[jpu]

pokial ti ide o pattern producent/konzument, tak tam je najlepsie pouzit DataFlow. Tu je aj navod od MS:
https://docs.microsoft.com/en-us/dotnet/standard/parallel-programming/how-to-implement-a-producer-consumer-dataflow-pattern