Producer - Consumer: Hoe te stoppen?

[EddoH]

Hallo,

ik ben vandaag met een stukje code bezig geweest wat sterk lijkt op het zogenoemde consumer-producer pattern, ik denk de meeste wel bekend.

Wikipedia:

code:

semaphore fillCount = 0; // items produced
semaphore emptyCount = BUFFER_SIZE; // remaining space
 
procedure producer() {
    while (true) {
        item = produceItem();
        down(emptyCount);
            putItemIntoBuffer(item);
        up(fillCount);
    }
}
 
procedure consumer() {
    while (true) {
        down(fillCount);
            item = removeItemFromBuffer();
        up(emptyCount);
        consumeItem(item);
    }
}

Nu is mijn probleem simpel, maar een mooie elegante oplossing heb ik nog niet gevonden, ik eindig steeds met 'vieze' hacks.

Doel: het stoppen van de consumer, op het moment dat deze niet meer nodig is.

Probleem: in de consumer is down(fillCount), uiteraard, blocking. Hoe kom ik ooit uit deze state als de producer al gestopt is? Merk op dat de producer in theorie 'oneindig' tijd nodig heeft om een een item voor de Buffer te produceren. Oplossingen met timeouts kunnen dus per definitie niet.

De enige 'mooie' oplossing die ik nu kan bedenken is een notification object in de Buffer te stoppen. De consumer leest deze als zijnde het ' stop'-signal en breakt uit zijn loop.
Handmatig de semaphore (in mijn geval is het geen semaphore, maar een blocking queue, dus gaat het ook moeilijker dan in het voorbeeld) 'uppen' als ik wil stoppen, en daarna niets uit de buffer lezen klinkt mij ook nogal vies in de oren.

Misschien zie ik iets heel eenvoudigs over het hoofd, maar op dit moment kom ik er gewoon even niet op...

Suggesties?

[D-Raven]

Welke ontwikkeltaal gebruik je?

[EddoH]

In dit geval Java, maar het maakt voor het concept niet uit denk ik, vandaar dat ik het er expres niet had bijgezet.

[D-Raven]

Het kan van invloed zijn op de oplossing die word aangedragen .Net heeft de TPL welke dit soort dingen kinderspel maken.

Maargoed.

Wat ik doe in dit soort gevallen is de consumer thread gewoon laten stoppen indien er geen items meer in queue zitten. De producer heeft dan een extra controle na het genereren van een item, welke kijkt of de producer bezig is, en hem start indien nodig.

Dat laten stoppen zou je natuurlijk op meerdere manieren kunnen doen afhankelijk van je behoefte. Je zou met een timeout kunnen werken, zoiets als er zijn al 5 sec geen items meer aangereikt, einde van consumer thread, of gewoon kijken naar queue.count == 0 dan stoppen.

[EddoH]

Bedankt, maar timeouts zijn geen mogelijke oplossing (zoals in de topicstart vermeld staat )
In theorie kan het produceren ' oneindig' duren, een valide timeout is dus neit te geven.

Het is niet gegarandeerd dat er dus altijd een item in de queue aanwezig is. De consumer moet blijven lopen, ook als er nog niks te consumen valt, maar ik wil hem ook op een 'willekeurig' punt in de tijd laten stoppen.

[Hydra]

Ik heb meestal een specaal 'token' message object. Als de consumer deze token tegenkomt is dat een teken dat 'ie op kan houden.

Psuedo:

Java:

        while(true)
        {
            Object o = queue.nextObject();
            if(o == specialTokenObject)
                break;
            else
                doWhateverWeNeedToDo(o);
        }

[Zoijar]

Weet niet of het het beste is, maar ik deed meestal iets van

C++:

while (empty()) {
                m_cond_empty.wait(lock);
                if (m_aborted) {
                    throw std::runtime_error("Operation aborted on request");
                }
            }
[...]

void abort() {
            boost::unique_lock<boost::mutex> lock(m_mutex);
            m_aborted = true;
            m_cond_empty.notify_all();
            m_cond_full.notify_all();
        }

[D-Raven]

Waarom _moet_ de consumer blijven lopen? Waarom mag hij niet stoppen als er geen werk is?

Maargoed, wat je ook kan doen is een eigen blocking collection implementeren. Waarin je dus via een semaphore blokkeert indien er geen items in zitten, deze semaphore kun je dan exposen in de public interface van je collectie, zo kun je via je collectie een pulse geven aan deze semaphore om zo je consumer uit de blocking te laten komen, zodat deze kan quitten indien je dit wilt.


Ik ben zelf geen fan van special token objecten overigens Ik vind dat dit een implementatie detail is welke los staat van je producer/consumer implementatie.

[ Voor 14% gewijzigd door D-Raven op 14-11-2012 14:54 ]

[Hydra]

D-Raven schreef op woensdag 14 november 2012 @ 14:53:
Waarom _moet_ de consumer blijven lopen? Waarom mag hij niet stoppen als er geen werk is?

Euh, het is doodnormaal dat workers wachten op werk?

Maargoed, wat je ook kan doen is een eigen blocking collection implementeren. Waarin je dus via een semaphore blokkeert indien er geen items in zitten, deze semaphore kun je dan exposen in de public interface van je collectie, zo kun je via je collectie een pulse geven aan deze semaphore om zo je consumer uit de blocking te laten komen, zodat deze kan quitten indien je dit wilt.

Waarom een eigen blocking collection implementeren als er gewoon blocking collections in java aanwezig zijn?

D-Raven schreef op woensdag 14 november 2012 @ 14:53:
Ik ben zelf geen fan van special token objecten overigens Ik vind dat dit een implementatie detail is welke los staat van je producer/consumer implementatie.

Als je er geen fan van bent geef dan tenminste tegenargumenten?

Daarnaast: dit is gewoon een manier hoe messagequeue's vaak werken. Dit heeft niks met implementatiedetails te maken.

[ Voor 23% gewijzigd door Hydra op 14-11-2012 14:57 ]

[D-Raven]

Hydra schreef op woensdag 14 november 2012 @ 14:54:
[...]


Euh, het is doodnormaal dat workers wachten op werk?

Ja duh, maar hij wil dat zijn consumer juist stopt met een druk op de knop. Daarnaast mag van mij een consumer thread best stoppen als hij al langere tijd geen werk meer heeft gezien.

[...]


Waarom een eigen blocking collection implementeren als er gewoon blocking collections in java aanwezig zijn?

Ja goed die zijn er vast. Maar kun je daarop ook een method aanroepen die waiting threads pulsed die op dat moment blokkeren op said collectie?

[EddoH]

@Hydra: thanks. die oplossing gebruik ik nu ook. Werkt op zich goed, alleen ergens voelt het ' vies'. Maar dat ligt misschien aan mij Enige probleem is dat als de consumer nog bezig is met een vorig object uit de queue (wat soms best lang kan duren), de abort-token wordt dan pas laat afgehandeld.

@Zoijar: wat als de notify wordt geissued voordat de wait wordt uitgevoerd? Dan wordt de notify gemist.

@Deathraven: er is werk in aankomst, maar nog niet op dat moment. Het opzetten van de Consumer kost tijd en bevind zich in een bepaalde state. Die state wil ik niet verliezen.
De semaphore exposen kan idd, maar ik gebruik al een BlockQueue. Redelijk wat werk en error-prone, maar moet werkbaar te maken zijn.

[ Voor 3% gewijzigd door EddoH op 14-11-2012 14:58 ]

[Hydra]

EddoH schreef op woensdag 14 november 2012 @ 14:57:
@Hydra: thanks. die oplossing gebruik ik nu ook. Werkt op zich goed, alleen ergens voelt het ' vies'. Maar dat ligt misschien aan mij Enige probleem is dat als de consumer nog bezig is met een vorig object uit de queue (wat soms best lang kan duren), de abort-token wordt dan pas laat afgehandeld.

Het lijkt mij sowieso wenselijk dat een worker z'n taak netjes afhandelt. Anders kun je bij wijze van net zo goed een thread afschieten.

Dergelijke token objecten zijn niet vies hoor. Het is doodnormaal in message queue's en als je bijvoorbeeld naar een LZW implementatie kijkt zie je dat dergelijke eindtokens daar ook gebruikt worden. Het is een erg simpele manier om een workerthread te laten weten dat z'n werk gedaan is zonder dat je extra complexiteit introduceert. Het werkt allemaal via de queue waar 'ie toch al op synchroniseert.

[Zoijar]

EddoH schreef op woensdag 14 november 2012 @ 14:57:
@Zoijar: wat als de notify wordt geissued voordat de wait wordt uitgevoerd? Dan wordt de notify gemist.

Condition-var onthoudt wake-ups (net als een semaphore)

[Hydra]

D-Raven schreef op woensdag 14 november 2012 @ 14:57:
Ja duh, maar hij wil dat zijn consumer juist stopt met een druk op de knop. Daarnaast mag van mij een consumer thread best stoppen als hij al langere tijd geen werk meer heeft gezien.

Fijn dat dat van jou mag, maar dat is volledig afhankelijk van de implementatie. Waarom zou je het onnodig complex maken met workers die automatisch stoppen/starten?

Ja goed die zijn er vast. Maar kun je daarop ook een method aanroepen die waiting threads pulsed die op dat moment blokkeren op said collectie?

Je kunt hoe dan ook een thread die op iets blockt interrupten.

[EddoH]

Zoijar schreef op woensdag 14 november 2012 @ 15:00:
[...]

Condition-var onthoudt wake-ups (net als een semaphore)

Ah, op die fiets. Dan is het idd een oplossing (mits C++ )

[D-Raven]

EddoH schreef op woensdag 14 november 2012 @ 14:57:
@Hydra: thanks. die oplossing gebruik ik nu ook. Werkt op zich goed, alleen ergens voelt het ' vies'. Maar dat ligt misschien aan mij Enige probleem is dat als de consumer nog bezig is met een vorig object uit de queue (wat soms best lang kan duren), de abort-token wordt dan pas laat afgehandeld.

@Zoijar: wat als de notify wordt geissued voordat de wait wordt uitgevoerd? Dan wordt de notify gemist.

@Deathraven: er is werk in aankomst, maar nog niet op dat moment. Het opzetten van de Consumer kost tijd en bevind zich in een bepaalde state. Die state wil ik niet verliezen.
De semaphore exposen kan idd, maar ik gebruik al een BlockQueue. Redelijk wat werk en error-prone, maar moet werkbaar te maken zijn.

Toegegeven, het is niet de meest ideale oplossing en als java een blocking collection heeft die deze functionaliteit aanbied, zeker gebruiken.

Maar de state van de consumer, kan je deze niet ergens tijdelijk bijhouden? zodat je altijd weer rap/goedkoop een nieuwe consumer thread kan starten? IMO zou dat de voorkeur hebben boven die custom collectie oplossing.

In .Net zou ik cancellationtokens gebruiken icm de TPL.

[D-Raven]

Hydra schreef op woensdag 14 november 2012 @ 15:00:
[...]
Fijn dat dat van jou mag, maar dat is volledig afhankelijk van de implementatie.

Klopt, maargoed dat is redelijk obvious.

Waarom zou je het onnodig complex maken met workers die automatisch stoppen/starten?

In hoeverre is dat complexer dan de andere oplossingen?

[...]
Je kunt hoe dan ook een thread die op iets blockt interrupten.

Nog niet eens bij stil gestaan, zit met mn hoofd veel te snel bij semaforen en mutexen.

[Zoijar]

Een consumer thread die geen werk heeft wordt toch als het goed is overigens ook gewoon niet gescheduled, dus waarom zou je threads gaan killen en opnieuw aanmaken? Maar voor program exit ed. is het wel netjes om je consumer een nette kans te geven om op te ruimen.

[EddoH]

Hydra schreef op woensdag 14 november 2012 @ 14:59:
[...]


Het lijkt mij sowieso wenselijk dat een worker z'n taak netjes afhandelt. Anders kun je bij wijze van net zo goed een thread afschieten.

Dergelijke token objecten zijn niet vies hoor. Het is doodnormaal in message queue's en als je bijvoorbeeld naar een LZW implementatie kijkt zie je dat dergelijke eindtokens daar ook gebruikt worden. Het is een erg simpele manier om een workerthread te laten weten dat z'n werk gedaan is zonder dat je extra complexiteit introduceert. Het werkt allemaal via de queue waar 'ie toch al op synchroniseert.

Je hebt ergens gelijk dat de consumer eerst zn taak netjes af zou mowten handelen. In dit geval is het echter geen vereiste, maar soit. Deze oplossing is idd tot nu toe het meest clean.

[EddoH]

Zoijar schreef op woensdag 14 november 2012 @ 15:07:
Een consumer thread die geen werk heeft wordt toch als het goed is overigens ook gewoon niet gescheduled, dus waarom zou je threads gaan killen en opnieuw aanmaken? Maar voor program exit ed. is het wel netjes om je consumer een nette kans te geven om op te ruimen.

in dit geval is het component die tijdens zn werk wel wat cpu/memory mag gebruiken, maar daarna moet hij zoveel mogelijk opruimen, om een ander component de ruimte te geven.

[Paul]

Je moet bij het sturen van dat soort tokens echter wel rekening houden met het aantal consumers, als een consumer het exit-token pop()t dan krijgen de overige (if any) threads geen signaal meer.

[Hydra]

Paul schreef op woensdag 14 november 2012 @ 15:51:
Je moet bij het sturen van dat soort tokens echter wel rekening houden met het aantal consumers, als een consumer het exit-token pop()t dan krijgen de overige (if any) threads geen signaal meer.

Idd. Je kunt of meerdere tokens op de queue zetten (als je aantal consumer constant is) of een consumer de token laten peek-en zonder deze van de queue te halen.

[EddoH]

In dit geval is er maar 1 consumer, dus geen probleem. Maar goed om in gedachte te houden.

[TheNameless]

Met een timeout werken op je pop() functie vind ik zelf altijd ook wel een aardige oplossing.
In samenwerking met een simpele boolean die aangeeft of de consumer nog steeds moet consumen vlak voor de pop() werkt het heel behoorlijk.

Gelukkig heeft .NET hiervoor een CancellationToken

[EddoH]

Maar hoe groot wordt de timeout als je niet weet hoe lang je op je data moet wachten?
Daarnaast, als je je timeout gebruikt om de de boolean te 'pollen' kun je net zo good peek/sleepen, waardoor je thread nog steeds gescheduled wordt, wat je uiteraard niet wilt als er niets te doen is.

[.oisyn]

Kun je niet gewoon wachten op meerdere waitable objects tegelijk, zoals WaitForMultipleObjects() in de win32 API? Dat gebruik ik altijd. Ik heb dan altijd gewoon een stop event waar ik óók op wacht.

[EddoH]

Hmm, in Java heb ik een soortgelijk iets niet kunnen vinden.
Op SO geven ze als alternatief suggestie: "You could easily set up a BlockingQueue that you could have producers drop event objects into and consumers blocking on removing the events."

En dat is precies wat ik nu doe

[TheNameless]

EddoH schreef op donderdag 15 november 2012 @ 09:53:
Maar hoe groot wordt de timeout als je niet weet hoe lang je op je data moet wachten?
Daarnaast, als je je timeout gebruikt om de de boolean te 'pollen' kun je net zo good peek/sleepen, waardoor je thread nog steeds gescheduled wordt, wat je uiteraard niet wilt als er niets te doen is.

Zoiets bedoel ik:

C#:

void RunWorker() {
  while(running){
    Job j;
    if(queue.TryPop(j, TimeSpan.FromSeconds(1))({
      DoWork(j);
    }
  }
}

//hier je worker stoppen
running = false;

Als je er dan voor zorgt dat je TryPop() async wacht (via semaphore of wat dan ook), dan kan het maximaal 1 seconden duren voordat je worker reageerd en wordt je thread niet geblocked door sleeps. Lijkt me acceptabel toch?

@edit: cancellen tijdens het runnen van de job kan op deze manier echter niet.

[ Voor 4% gewijzigd door TheNameless op 15-11-2012 13:59 ]

[EddoH]

Tja, op deze manier poll je om de seconde. Dan vind ik een wait-signal mechanisme mooier.

[pedorus]

Je kunt natuurlijk ook exceptions en try/catch gebruiken. Ik ben bijvoorbeeld wel eens een ExitThreadException tegen gekomen.

[kenneth]

Exceptions voor flow control lijkt me geen goede gewoonte.

[Tarilo]

kenneth schreef op donderdag 15 november 2012 @ 15:32:
Exceptions voor flow control lijkt me geen goede gewoonte.

Ik ga er toch vanuit dat pedorus hier enig sarcasme heeft gebruikt.

Anders:
* Tarilo zoekt zijn ClueBat

[Kwistnix]

Als de producer en consumer(s) niet van elkaar weten en ontkoppeld zijn via een queue dan kan je inderdaad het beste met poison pill(s) werken om de consumer(s) te stoppen. Als dat niet het geval is dan is de makkelijkste aanpak om je taken Runnable/Callable te maken en ze via (ThreadPool)ExecuterService te starten. Dan kan de producer gewoon een gracefull shutdown van de ExecutorService initiëren als hij klaar is met produceren, welke dan netjes wacht tot alle nog gequeuede taken afgrond zijn.

[ Voor 0% gewijzigd door Kwistnix op 15-11-2012 16:41 . Reden: typos ]

[Sleepkever]

Tarilo schreef op donderdag 15 november 2012 @ 15:50:
[...]

Ik ga er toch vanuit dat pedorus hier enig sarcasme heeft gebruikt.

Anders:
* Tarilo zoekt zijn ClueBat

Nope, Exceptions zijn exceptions. Als in, excepties op het standaardgedrag. In dit geval is het stoppen van een consumer standaard gedrag, het gebeurt immers vaak en is gedefineert. Orace/Sun zelft zegt dat je volgens best practice in Java _nooit_ exceptions voor flowcontrol moet gebruiken. En persoonlijk kan ik me daar wel in vinden. Maar dit is eigenlijk meer een persoonlijke keus dan wat anders. Het is namelijk niet fout volgens de compiler om dat te doen.

[Zoijar]

Sleepkever schreef op donderdag 15 november 2012 @ 16:29:
Orace/Sun zelft zegt dat je volgens best practice in Java _nooit_ exceptions voor flowcontrol moet gebruiken.

Exceptions zijn flow control, daarom vind ik die stelling altijd een beetje vreemd. Hoewel ik wel snap wat er bedoeld wordt uiteraard. Exceptions zijn voor flow control buiten het normale programma om, een soort van side-channel voor fouten en... uitzonderlijke situaties. Je kan best verdedigen dat een abort van een consumer zo'n situatie is, zeker als de user dit bijvoorbeeld kan doen. Niet heel anders dan een sigterm.

[TheNameless]

EddoH schreef op donderdag 15 november 2012 @ 14:20:
Tja, op deze manier poll je om de seconde. Dan vind ik een wait-signal mechanisme mooier.

Je polled niet om de seconde, je wacht een seconde lang op een signal
Maar misschien verschillen onze kijk op de definitie "pollen"

[Hydra]

Gaan we nu serieus die discussie weer voeren? Als men stelt dat je geen exceptions voor flow control moet gebruiken hebben ze het over de normale programmaflow. Nu kun je wel zeggen dat dat volgens jouw definitie ook "flow control" is, maar dan zorg je alleen maar voor een hoop onduidelijkheid.

TheNameless schreef op donderdag 15 november 2012 @ 17:00:
Je polled niet om de seconde, je wacht een seconde lang op een signal
Maar misschien verschillen onze kijk op de definitie "pollen"

Je kijkt elke seconde of er 'iets' is, dat is gewoon pollen.

[ Voor 35% gewijzigd door Hydra op 15-11-2012 17:02 ]

[.oisyn]

TheNameless schreef op donderdag 15 november 2012 @ 17:00:
[...]

Je polled niet om de seconde, je wacht een seconde lang op een signal

Elke seconde wordt er toch gekeken of de 'running' variabele op true is gezet, of niet dan?

[Zoijar]

Hydra schreef op donderdag 15 november 2012 @ 17:01:
Gaan we nu serieus die discussie weer voeren? Als men stelt dat je geen exceptions voor flow control moet gebruiken hebben ze het over de normale programmaflow. Nu kun je wel zeggen dat dat volgens jouw definitie ook "flow control" is, maar dan zorg je alleen maar voor een hoop onduidelijkheid.

Dan moeten ze dat zeggen en niet "_nooit_ exceptions voor flowcontrol moet gebruiken." Ik heb nou eenmaal iets tegen die "nooit" regeltjes ivm programmeren; als het voor jou (en andere) goed werkt en duidelijk is, prima. Dat een consumer voortijdig afbreekt zonder zijn werk af te maken, niet heel normaal Maar goed, laat maar.

[.oisyn]

Kun je in Java een thread trouwens niet gewoon interrupten? Dan gaat ie uit waiting state en wordt er een ThreadInterruptedException gegooid op die thread (oei, moord, brand, exceptions, it's every man for himself en meer van dat soort kolder)

[ Voor 17% gewijzigd door .oisyn op 15-11-2012 17:49 ]

[pedorus]

In java moet je inderdaad toch al besluiten wat je met InterruptedException doet. (Mijn voorkeur gaat meestal uit naar loggen en stug doorgaan, maar ik werk veelal met threads die nooit horen te stoppen.)

[Herko_ter_Horst]

Mij lijkt bij ontkoppelde consumers een speciaal stop-bericht cleaner dan het alternatief via een lock/interrupt (dan heb je ook weer twee threads nodig per consumer).

[ Voor 67% gewijzigd door Herko_ter_Horst op 15-11-2012 18:18 ]

[Hydra]

Zoijar schreef op donderdag 15 november 2012 @ 17:44:
Dan moeten ze dat zeggen en niet "_nooit_ exceptions voor flowcontrol moet gebruiken." Ik heb nou eenmaal iets tegen die "nooit" regeltjes ivm programmeren; als het voor jou (en andere) goed werkt en duidelijk is, prima. Dat een consumer voortijdig afbreekt zonder zijn werk af te maken, niet heel normaal Maar goed, laat maar.

Misschien dan gewoon ff kijken naar wat er bedoeld wordt met een stukje tekst in plaats van alleen maar op het woordje "nooit" in een anafylactische shock te schieten?

Herko_ter_Horst schreef op donderdag 15 november 2012 @ 18:11:
Maar ik neem aan dat de producer de consumers niet kent, dus wie gaat die thread interrupt dan doen?

Ze hoeven de thread niet te kennen, ze kunnen interrupten op de lock.

[ Voor 52% gewijzigd door Hydra op 15-11-2012 18:16 ]

[Herko_ter_Horst]

Hydra schreef op donderdag 15 november 2012 @ 18:15:

[...]

Ze hoeven de thread niet te kennen, ze kunnen interrupten op de lock.

Dat had ik inmiddels ook bedacht (heb mijn bericht aangepast), maar dan heb je weer een tweede thread nodig die op die lock staat te wachten.

[Zoijar]

Hydra schreef op donderdag 15 november 2012 @ 18:15:
Misschien dan gewoon ff kijken naar wat er bedoeld wordt met een stukje tekst in plaats van alleen maar op het woordje "nooit" in een anafylactische shock te schieten?

Als je denkt dat dit: Zoijar in "Producer - Consumer: Hoe te stoppen?" al een anafylactische shock is... Ik zei het toch gewoon netjes


(verder, het gaat om een thread afschieten, doe gewoon iets, vang een exception, check een var, stuur een bericht... just get it over with already )

[ Voor 14% gewijzigd door Zoijar op 15-11-2012 18:23 ]

[Soultaker]

EddoH schreef op woensdag 14 november 2012 @ 14:37:
De enige 'mooie' oplossing die ik nu kan bedenken is een notification object in de Buffer te stoppen. De consumer leest deze als zijnde het ' stop'-signal en breakt uit zijn loop.

Je hebt een mooie oplossing waarbij je geen nadelen noemt — waarom gebruik je die dan niet gewoon?

(Ik mik zelf meestal gewoon n keer een null pointer in de queue om EOF te signaleren, waarbij n het aantal consumer threads is. In sommige situaties heb je daarna nog een manier nodig om te wachten tot alle consumers geëxit zijn.)

[.oisyn]

Hydra schreef op donderdag 15 november 2012 @ 18:15:
[...]


Misschien dan gewoon ff kijken naar wat er bedoeld wordt met een stukje tekst in plaats van alleen maar op het woordje "nooit" in een anafylactische shock te schieten?

Sorry hoor Hydra, maar je mag de hand volledig in eigen boezem slaan in dit geval. Als je de originele reactie van Zoijar namelijk nog eens doorleest dan wordt de soep helemaal niet zo heet gegeten als ie wordt opgediend:

Exceptions zijn flow control, daarom vind ik die stelling altijd een beetje vreemd. Hoewel ik wel snap wat er bedoeld wordt uiteraard. Exceptions zijn voor flow control buiten het normale programma om, een soort van side-channel voor fouten en... uitzonderlijke situaties. Je kan best verdedigen dat een abort van een consumer zo'n situatie is, zeker als de user dit bijvoorbeeld kan doen. Niet heel anders dan een sigterm.

Hij zegt dus letterlijk dat ie snapt wat er bedoeld wordt, maar dat deze situatie best als uitzonderlijk kan worden beschouwd. Waar jij die anafylactische shock in terugziet is mij een raadsel.

.edit: en ik moet even verder lezen voor ik reageer, Zoijar zei al zoiets

Herko_ter_Horst schreef op donderdag 15 november 2012 @ 18:11:
Mij lijkt bij ontkoppelde consumers een speciaal stop-bericht cleaner dan het alternatief via een lock/interrupt (dan heb je ook weer twee threads nodig per consumer).

Waarom heb je nou weer 2 threads nodig per consumer

Hydra schreef op donderdag 15 november 2012 @ 18:15:
Ze hoeven de thread niet te kennen, ze kunnen interrupten op de lock.

Nou heb ik al in geen jaren meer iets met Java gedaan, maar de documentatie leert dat er op een Semaphore iig geen interrupt mechanisme zit. Wel kun je gewoon de wachtende threads opvragen overigens, dus dan kun je ze alsnog individueel gaan interrupten. Het is wel jammer dat je op die manier geen threads vangt die nog niet aan het wachten zijn maar dat wel gaan doen in de toekomst, dus dat riekt dan weer naar race conditions.

[ Voor 38% gewijzigd door .oisyn op 15-11-2012 23:56 ]

[Herko_ter_Horst]

.oisyn schreef op donderdag 15 november 2012 @ 23:42:
[...]

Waarom heb je nou weer 2 threads nodig per consumer

Eén thread die de message queue leegleest en één die wacht op het lock en dan de lees-thread interrupt, lijkt mij toch?

(overigens hoeft dit niet persé een Semaphore object te zijn, je kunt in Java op elk object wait()en)

[Zoijar]

Volgens mij is dit de functie om aan te roepen: http://docs.oracle.com/ja...ject.html#notifyAll%28%29 In context: http://docs.oracle.com/ja...rent/locks/Condition.html

[EddoH]

.oisyn schreef op donderdag 15 november 2012 @ 23:42:
Nou heb ik al in geen jaren meer iets met Java gedaan, maar de documentatie leert dat er op een Semaphore iig geen interrupt mechanisme zit. Wel kun je gewoon de wachtende threads opvragen overigens, dus dan kun je ze alsnog individueel gaan interrupten. Het is wel jammer dat je op die manier geen threads vangt die nog niet aan het wachten zijn maar dat wel gaan doen in de toekomst, dus dat riekt dan weer naar race conditions.

Yep, been there done that, uiteindelijk eindig je met hele lappen coden en synchronized blocks om race conditions tegen te gaan.

[EddoH]

Soultaker schreef op donderdag 15 november 2012 @ 18:50:
[...]

Je hebt een mooie oplossing waarbij je geen nadelen noemt — waarom gebruik je die dan niet gewoon?

(Ik mik zelf meestal gewoon n keer een null pointer in de queue om EOF te signaleren, waarbij n het aantal consumer threads is. In sommige situaties heb je daarna nog een manier nodig om te wachten tot alle consumers geëxit zijn.)

Ik heb deze oplossing nu ook geimplementeerd, maar ik wist toen ik het maakte niet of dit een gangbare oplossing was, of dat er misschien nog mooiere constructies te bedenken waren, vandaar dit topic.

Het werkt prima nu en dit is ook mijn uiteindelijk oplossing

Wel leuk om te zien hoe iedereen het 'probleem' kent en er op zijn eigen manier mee omgaat.

[EddoH]

Zoijar schreef op vrijdag 16 november 2012 @ 10:15:
Volgens mij is dit de functie om aan te roepen: http://docs.oracle.com/ja...ject.html#notifyAll%28%29 In context: http://docs.oracle.com/ja...rent/locks/Condition.html

Dan zit je weer met situaties waarin een thread nog niet aan het waiten is, maar dat wel gaat doen, en dan de notify mist => race condition.

[D-Raven]

EddoH schreef op vrijdag 16 november 2012 @ 10:20:
[...]

Wel leuk om te zien hoe iedereen het 'probleem' kent en er op zijn eigen manier mee omgaat.

Vele wegen lijden tot rome.

[.oisyn]

Herko_ter_Horst schreef op vrijdag 16 november 2012 @ 08:59:
[...]

Eén thread die de message queue leegleest en één die wacht op het lock en dan de lees-thread interrupt, lijkt mij toch?

(overigens hoeft dit niet persé een Semaphore object te zijn, je kunt in Java op elk object wait()en)

Het ging om deze opmerking:

Hydra schreef op donderdag 15 november 2012 @ 18:15:
Ze hoeven de thread niet te kennen, ze kunnen interrupten op de lock.

Als een waitable object kan interrupten dan is die 2e thread dus niet nodig, want dan interrupt je de message queue zelf (en daardoor alle threads die daarop aan het wachten zijn of dat gaan doen). Maar ik trek nu de validiteit van die opmerking in twijfel. Of ik begrijp 'm verkeerd, dat kan natuurlijk ook.

[ Voor 6% gewijzigd door .oisyn op 16-11-2012 10:36 ]

[Zoijar]

EddoH schreef op vrijdag 16 november 2012 @ 10:23:
Dan zit je weer met situaties waarin een thread nog niet aan het waiten is, maar dat wel gaat doen, en dan de notify mist => race condition.

Ik had misschien beter volledige code kunnen plaatsen

Dit bedoel ik:

C++:

void put(const T& item) {
    boost::unique_lock<boost::mutex> lock(m_mutex);
    while(!m_aborted && full()) {
        m_cond_full.wait(lock);
    }

    if (m_aborted) {
        throw std::runtime_error("Operation aborted on request");
    }

    m_items.push_back(item);
    m_cond_empty.notify_one();
}

T get() {
    boost::unique_lock<boost::mutex> lock(m_mutex);
    while (!m_aborted && empty()) {
        m_cond_empty.wait(lock);
    }

    if (m_aborted) {
        throw std::runtime_error("Operation aborted on request");
    }

    T result = m_items.front();
    m_items.pop_front();
    m_cond_full.notify_one();

    return result;
}

void abort() {
    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_aborted = true;
    m_cond_empty.notify_all();
    m_cond_full.notify_all();
}

boost::condition_variable m_cond_empty, m_cond_full;
boost::mutex m_mutex;
std::list<T> m_items;
bool m_aborted;

Dit is C++ maar vertaalt een-op-een naar Java. Niet dat dit nou "De beste" manier is, maar ik zie hier geen fouten in (?)(nadeel is dat je niet kan stoppen tijdens het verwerken van data; dan zal je toch echt de thread moeten killen)

Stel we kijkn naar get(). Op het moment dat ik abort() aanroep, acquire ik het lock m_mutex in regel 33. De thread die get draait kan dan maar op twee plaatsen wachten: regel 16 (normale lock), of regel 18 (condition wait). In het eerste geval draait die thread verder nadat abort() z'n lock vrijgeeft omdat die gewoon nog active is, en in het tweede geval draait die thread later verder door de notify_all wakeup. In beide gevallen als hij daar verder gaat en m_abort=true valt hij door naar regel 22.

Feitelijk verbeedt je je definite van de conditie van "!empty" naar "aborted || !empty" en notify je nog steeds een wijziging in die conditie.

Wel nog een vraagje aan jullie: regel 12 en 27 doet een notify_one(). Nu wordt er altijd gezegd dat je notify_all moet gebruiken om lost wakeup te voorkomen; i.e., producer maakt snel twee items aan voor consumers erbij kunnen, en dan wordt er maar 1 consumer thread wakker gemaakt. Echter, dat is hier niet het geval toch? Als je twee keer achter elkaar notify_one aanroept, heb je dan de garantie dat er altijd twee verschillende threads wakker worden gemaakt? Dat lijkt mij wel, toch? Maarja, om hier dus zeker van te zijn kan je 12 en 27 naar notify_all veranderen... maar volgens mij hoeft dat niet (?).

[ Voor 4% gewijzigd door Zoijar op 16-11-2012 14:32 ]

[Soultaker]

Dat hangt natuurlijk van de precieze semantics van je threading library af. Ik zou denken dat je gelijk hebt en dat je gewoon notify_one() kunt gebruiken.

Als ik in de man-page van pthread_cond_signal() kijk (wat vergelijkbaar is met notify_one()) dan wordt het zo geformuleerd:

The pthread_cond_signal() function shall unblock at least one of the threads that are blocked on the specified condition variable cond (if any threads are blocked on cond).

Ik interpreteer dat ook zo dat het niet mogelijk is dat twee calls dezelfde thread unblocken (zonder dat die tussentijds uitgevoerd wordt), want eenmaal unblocked is een thread niet meer blocked (ook al is 'ie nog niet aan executie toegekomen).

[.oisyn]

Maar unblockt 2x een notify_one() ook daadwerkelijk 2 threads als geen van de threads staat te wachten als je die notifies doet?

[ Voor 19% gewijzigd door .oisyn op 16-11-2012 17:31 ]

[Soultaker]

Daar gaat het niet om

Als er geen threads blocked zijn op een condition variable, worden er ook geen unblocked (dat staat ook tussen haakjes in die quote uit de man-page). Dat zou je als een “lost wake-up” kunnen beschouwen, maar het maakt in Zoijars code niet uit, want hij lockt netjes de mutex in alledrie de functies, dus threads die niet blocked zijn op de condition zijn helemaal nog niet aan executie begonnen, en zullen dus gewoon exiten zodra ze m_aborted uitlezen.

Behalve dat dit dus geen probleem is, is het ook niet iets wat notify_all() zou oplossen, want die gedraagt zich op exact dezelfde manier als er geen blocked threads zijn.

edit:
Snel editen hè? Nu snap ik helemaal niet meer wat je bezwaar is...

[ Voor 6% gewijzigd door Soultaker op 16-11-2012 17:38 ]

[.oisyn]

Geen bezwaar, maar een vraag . Ik heb Zoijar's niet goed bekeken of dat überhaupt een issue zou zijn.

[Zoijar]

.oisyn schreef op vrijdag 16 november 2012 @ 17:30:
Daar gaat het niet om, het gaat erom of 2x een notify_one() ook daadwerkelijk 2 threads unblockt als er geen van de threads staat te wachten als je die notifies doet.

Dat is volgens mij niet erg, als ze toch al draaien en niet staan te wachten gaan ze nooit de wacht in als je queue niet leeg is, omdat je de conditie voor de wacht altijd nog een keer nagaat.

Het ging mij hierom: stel je hebt 3 consumers en produceert heel snel 10 items, dat gaat zo snel dat er geen consumer tussen komt om het lock te krijgen. Dan staan er 10 items in je queue, staan er 3 consumers te wachten, en is er 10 keer de sequence "lock, notify_one, unlock" aangeroepen. Zijn dan gegarandeerd alle 3 de consumers wakker, of kan het dat er 10 keer een wake-up naar consumer 1 is gestuurd, die vanaf dan al het werk (10 items) doet onderhand dat 2 en 3 nog steeds slapen.

Maar als ik de documentatie zie kan dat niet. Als er 100 threads slapen, en je roept in een loop waar je het lock nog vast hebt 100x notify_one aan (geen een thread kan dan onderhand draaien) dan zijn daarna gegarandeerd al je 100 threads wakker. Notify_one zet volgens mij gewoon een 'running' vlag op een thread waar die vlag op 'sleep' staat, zodat die scheduled wordt, en vervolgens gaat de thread zelf later proberen het lock te krijgen.

Zoiets dus:

C++:

notify_one() {
  for (auto thread : m_waitingThreads) {
    lock(thread);
    if (thread->sleeping()) {
      thread->wakeup();
      break;
    }
  }
}

Waarbij het verschil met notify_one en notify_all dat 'break' statement is.

Maar het moet niet zo zijn dat het is:

C++:

notify_one() {
   lock(m_waitingThreads[0]);
   m_waitingThreads[0]->wakeup();
}

Want dan gaat het niet helemaal goed. Tenzij je dan nog toevoegt, m_waitingThreads.pop_front(); en er alleen gegarandeert sleeping threads in die lijst staan.

(sorry voor warrige uitleg; snel typen, honger... )

[Soultaker]

Als notify_one() op de tweede manier werkt is er toch geen enkele situatie denkbaar waarin je die functie zou kunnen/willen gebruiken? Immers met meer dan één consumer thread werkt het niet goed, en met één enkele consumer thread doet notify_all() hetzelfde; in beide gevallen gebruik je dan dus de laatste. Dat lijkt me al reden genoeg om aan te nemen dat 't niet zo werkt (los van de documentatie).

[ Voor 27% gewijzigd door Soultaker op 16-11-2012 18:07 ]

[pedorus]

Wat wel kan gebeuren natuurlijk is dat er een interrruptiemechanisme heeft plaatsgevonden op de ontwaakte thread. Je ziet bij java in de implementatie van een BlockingQueue<> dan ook zoiets:

Java:

        lock.lockInterruptibly();
        try {
            try {
                while (count == 0)
                    notEmpty.await();
            } catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            E x = extract();
            return x;
        } finally {
            lock.unlock();
        }

Waar hydra vandaal haalt dat je de threads op de queue/lock zou kunnen interrrupten snap ik trouwens niet?

---

Verder heb je nog het probleem hoe je weet of alles klaar is. Wat ik nu veelal doe is een protected AtomicInteger gebruiken die ik voor elke toevoeging aan de queue eentje ophoog (inProcessing). Bij het einde van de behandeling verlaag ik deze eentje, en signaleer ik zonodig een Semaphore(1) (processingDone):

Java:

if (inProcessing.decrementAndGet() <= 0) {
    processingDone.release();
}

Bij de producer gebruik ik aan het einde zonodig de semaphore:

Java:

if (inProcessing.get() != 0) {
    processingDone.drainPermits();
    if (inProcessing.get() != 0) {
        processingDone.acquire();
    }
}

Dit gaat uit van 1 producer. anders zou je na die acquire bijvoorbeeld gelijk weer een release moeten doen (buiten de binnenste if). Of misschien dat er een ander mechanisme is in java dat dit beter kan, in .NET is een ManualResetEventSlim waarschijnlijk handiger.

Als iemand hiervoor een betere oplossing heeft, graag