[Java] Concurrency met meerdere losstaande taskqueues

[Gerco]

Ik heb een open source programma in Java geschreven met Swing en om de UI responsive te houden maak ik gebruik van een ThreadPoolExecuter met 1 thread om het dure werk op een achtergrondthread uit te voeren. Nu ik de functionaliteit van het programma heb uitgebreid kom ik tot de conclusie dat dit een beetje te beperkt is.

Wat ik nodig heb is een manier om meerdere taken onafhankelijk van elkaar, tegelijk te laten uitvoeren. Daarnaast zijn sommige taken wel van elkaar afhankelijk. Bovendien is iedere taak geassocieerd met een bepaalde JMS connectie en per JMS connectie mag er maar 1 taak tegelijk draaien.

Ik zou dit kunnen doen met meerdere Executors, voor elke JMS connectie 1. Dan heb je dus wel per connectie een thread draaien, terwijl er waarschijnlijk geen taken in de queue staan voor die connectie (meerdere dingen tegelijk doen is een uitzondering). Bovendien los ik daarmee het afhankelijkheidsprobleem niet op, een taak in de queue van connectie 1 kan best afhankelijk zijn van een taak voor connectie 2. Die zouden dus technisch gezien tegelijk kunnen worden uitgevoerd, maar mijn applicatielogica vind dat natuurlijk niet leuk.

Kent iemand een framework of misschien een startpunt om zo'n scheduler te maken?

[Macros]

Ik ken geen framework, maar is het niet mogelijk om de afhankelijkheden expliciet te maken tussen de tasks mbv queue's. Dus je hebt dan per JMS connectie een ThreadPool, maakt niet erg veel uit wat voor 1. En dan als een task in 1 threadpool afhankelijk is van een andere task, dan doet hij een pop van de queue van de andere task waar hij afhankelijk van is (en de data waarschijnlijk voor nodig heeft). Is die data er niet, dan blockt die call. Zodra die task klaar is en data in de queue heeft gezet gaat de afhankelijke thread verder met executie omdat zijn call unblockt.

Is dit een mogelijke oplossing voor je probleem?

[Gerco]

Klinkt wel als iets wat zou werken. Die taken hebben overigens meestal geen data van elkaar nodig, behalve een "ik ben klaar"-achtig vlaggetje. Dat is iets wat de scheduler zou kunnen regelen natuurlijk. Bij gebrek aan goede ideeen daarvoor kan ik het natuurlijk in de taken zelf coderen.

Nadeel daarvan is natuurlijk dat het de task queue van connectie A blokkeert terwijl deze op connectie B aan het wachten is en het best mogelijk is dat er in de tussentijd een andere taak op connectie A uitgevoerd had kunnen worden. Ik weet nog niet hoe vaak die situatie voor gaat komen, maar ik verwacht dat ik daar wel mee kan leven.

Voor de duidelijkheid, je bedoelt iets als dit?:

Java:

class Task1 implements Runnable {
  public BlockingQueue queue = new SynchronousQueue();

  public void run() {
    // Do stuff
    Object data = doStuff();

    queue.put(d);
  }
}

class Task2 implements Runnable {
  private Task1 task1;

  public void run() {
    // Wait for task 1 to finish
    Object data = task1.queue.take();

    // Do stuff
    doOtherStuff(data);
  }
}

[BCC]

Gerco schreef op dinsdag 11 augustus 2009 @ 00:07:

Nadeel daarvan is natuurlijk dat het de task queue van connectie A blokkeert terwijl deze op connectie B aan het wachten is en het best mogelijk is dat er in de tussentijd een andere taak op connectie A uitgevoerd had kunnen worden. Ik weet nog niet hoe vaak die situatie voor gaat komen, maar ik verwacht dat ik daar wel mee kan leven.

Hoera, het digning philosophers probleem: Wikipedia: Dining philosophers problem Daar zijn eindeloze verhalen en scheduling algoritmes voor geschreven

Maar volgens mij moet je in jouw eenvoudige geval gewoon een scheduler maken die de eerste van de rij probeert te doen. Kan dat niet, ga dan de rij af totdat je wel iets kan. Iets gedaan of einde van de rij? Begin opnieuw.

[ Voor 14% gewijzigd door BCC op 11-08-2009 00:13 ]

[Gerco]

BCC schreef op dinsdag 11 augustus 2009 @ 00:09:
Hoera, het digning philosophers probleem: Wikipedia: Dining philosophers problem Daar zijn eindeloze verhalen en scheduling algoritmes voor geschreven

Inderdaad, behalve dat dat er helemaal niets mee te maken heeft.

Mijn punt is dat connectie A (laten dat resource A noemen) gelocked is terwijl dat niet nodig is. Die resource is niet nodig voor het uitvoeren van de taak die op dit moment draait en is alleen gelocked omdat er geen mechanisme is om afhankelijkheden tussen taken fatsoenlijk aan te geven in java.util.concurrent.

Dit heeft helemaal niets met locking te maken en alles met scheduling.

Maar volgens mij moet je in jouw eenvoudige geval gewoon een scheduler maken die de eerste van de rij probeert te doen. Kan dat niet, ga dan de rij af totdat je wel iets kan. Iets gedaan of einde van de rij? Begin opnieuw.

Om dat te doen kan ik een aantal attributen aan een taak koppelen. Een lijst van taken waarvan deze afhankelijk is (die dus eerst uitgevoerd moeten worden) en een lijst van resources die nodig zijn voor die taak. Die laatste lijst bevat 1 item (een jms connectie) of geen items.

Dan hoeft de scheduler dus alleen te kijken naar de taken in de queue die een lege depends lijst hebben en waarvan de resources (connecties) momenteel niet in gebruik zijn. Dan kan die taak gestart worden en wanneer er een taak klaar is doen we hetzelfde gewoon nog een keer.

[ Voor 33% gewijzigd door Gerco op 11-08-2009 00:19 ]

[BCC]

Volgens mij spreek je nu jezelf tegen

[Gerco]

BCC schreef op dinsdag 11 augustus 2009 @ 00:19:
Volgens mij spreek je nu jezelf tegen

Wat ik bedoel is dat het helemaal niet nodig is om iets te locken wanneer die taak gewoon zou kunnen zeggen dat hij afhankelijk is van een andere taak. Dat gebeurt in mijn eerste codevoorbeeld puur en alleen omdat er geen andere manier is om die afhankelijkheid aan te geven.

Dining Philisophers draait om het uitdelen van meerdere resources aan processen die ze willen locken. Dat is hier niet de bedoeling omdat het puur om afhankelijkheden tussen taken gaat en deze geen resources willen locken. Elke resource heeft zeg maar zijn eigen task queue en dus is er nooit locking nodig.

1 taak heeft nooit 2 connecties nodig, soms niet eens 1 maar dat is een uitzondering.

[ Voor 5% gewijzigd door Gerco op 11-08-2009 00:48 ]

[BCC]

Dan heb je toch gewoon genoeg aan twee queues? Of een slimme queue zoals ik die daarstraks beschreef?

[Salandur]

waarom gebruik je JMS? Volgens mij maakt dit je probleem erg ingewikkeld. JMS is namelijk gewoon een Queue oplossing, die je ook met een BlockingQueue op kan lossen.

[Gerco]

BCC schreef op dinsdag 11 augustus 2009 @ 09:23:
Dan heb je toch gewoon genoeg aan twee queues? Of een slimme queue zoals ik die daarstraks beschreef?

Ik ga N "task queues" hebben, dus ik denk dat ik inderdaad voor die slimme queue van jou ga. Ik ga het eens uitwerken en kijken of ik daarmee nog problemen veroorzaak die ik nu nog niet voorzie.

Salandur schreef op dinsdag 11 augustus 2009 @ 09:29:
waarom gebruik je JMS? Volgens mij maakt dit je probleem erg ingewikkeld. JMS is namelijk gewoon een Queue oplossing, die je ook met een BlockingQueue op kan lossen.

Ik gebruik JMS omdat mijn programma een JMS management tool is. JMS heeft niets met het probleem te maken, die connecties die de tasks nodig hebben kunnen net zo goed resource A en B heten. Dat zou het "probleem" niet veranderen.

[ Voor 27% gewijzigd door Gerco op 11-08-2009 11:23 ]

[Woy]

Aangezien een Task maar aan 1 JMS instantie gelinkt kan zijn, en een JMS maar 1 Task tegelijk kan faciliteren, zul je inderdaad per JMS 1 TaskQueue moeten hebben.

Als je nu per Task een list met Tasks hebt waar hij dependent van is, kun je gewoon een canExecute method maken, en dan sorteren op canExecute ( en eventueel andere priority kenmerken )

CanExecute is dan gewoon een simpele method

code:

foreach( dependencies )
{
   if(!dependency.isFinished())
      return false;
}
return true;

Op die manier krijg je gewoon een eenvoudige PriorityQueue. Je zult dan hoogstens nog moeten zorgen dat je thread tijdelijk blockt op het moment dat er geen Tasks meer zijn die uitgevoerd kunnen worden, en weer resumed op het moment dat er weer een uitvoerbare Task is.

[ Voor 20% gewijzigd door Woy op 11-08-2009 12:37 ]

[.oisyn]

dependency

[ Voor 5% gewijzigd door .oisyn op 11-08-2009 12:08 ]

[Woy]

[Gerco]

Zo, het werkt. Ik zal nog even toelichten welke oplossingsrichting ik heb gekozen en waarom:

Wat ik nu gemaakt heb is een redelijk eenvoudig systeem. Alle taken erven van een baseclass Task (die een handige static factory method heeft om een Runnable te wrappen). Wanneer de Task uitgevoerd gaat worden, zal hij een method waitForDependencies() uitvoeren, deze blockt tot alle dependencies klaar zijn. Daarna voert de taak nog een method uit om zijn JMS connectie te locken die weer wacht als die nog in gebruik is.

Al deze taken worden aan een TaskExecutor gegeven die een CachedThreadPoolExecutor gebruikt om de taken op uit te voeren.

Gevolg van dit systeem is natuurlijk dat er voor elke taak een thread "bezig" is. Als de taak aan het executen is dan is die thread wat aan het doen. Als de taak aan het wachten is staat de thread in wait(). Dit is feitelijk threadverspilling, maar het maakte de oplossing wel een stuk eenvoudiger te maken

Ik heb deze oplossing gekozen omdat het behoorlijk non-triviaal is om een robuuste, thread-safe Executor te implementeren (zie ook de code van de verschillende Executors in de JRE). Ik besloot om dat pokkewerk over te laten aan de mensen die daar beter in zijn dan ik.

Er zijn nu soms meer threads actief dan ik strikt nodig heb, maar die worden na een paar seconden vanzelf weer opgeruimd en met de aantallen valt het wel mee. Aangezien de meeste taken vlug klaar zijn en er queues nooit erg lang zijn heb ik nog niet meer dan 6 actieve (uitvoerend of wachtend) threads gezien. Wel voelt de applicatie nu een stuk sneller omdat dingen die eerst serieel gingen terwijl daar geen reden voor was parallel gaan.

[ Voor 20% gewijzigd door Gerco op 13-08-2009 14:55 ]