[Alg] Concurrency en mutexes bij maken scheduler

[MrBucket]

Ik wil een applicatie maken die informatie van meerdere websites, webservices en/of rss-feeds ophaalt en combineert. Nu wil ik voorkomen dat ik op een enkele webserver teveel requests achter elkaar uitzet (ook wel "hammeren" genoemd), en om de requests min of meer gelijkmatig te verdelen over de webservers wil ik een scheduler maken die garandeert dat er niet te snel achter elkaar requests op dezelfde server worden afgeschoten.

Deze scheduler loopt in een aparte thread, en beheert een X aantal job (of "worker") threads die het van de ingeschedulde jobs moet voorzien, nl. het opvragen en parsen van een pagina/feed/whatever. Wanneer een job thread tijdelijk niets te doen heeft, zal deze thread blocken totdat het een nieuwe job krijgt van de scheduler. Dit voorkomt dat er continue nieuwe threads worden aangemaakt en opgeruimd.

Globaal werkt de scheduler als volgt: de scheduler wacht totdat er een job thread vrijkomt. Dan zal de scheduler kijken of er op dit moment al een job is die uitgevoerd mag worden. Zo niet, dan zal de scheduler wachten totdat een van de aanwezige jobs wel uitgevoerd mag worden, of totdat er een nieuwe job wordt toegevoegd die meteen uitgevoerd kan worden. Als de job aan de job thread is meegegeven, dan zal het weer wachten totdat er een job thread vrijkomt (als die er al niet is).

Goed en wel, maar het punt waar ik moeite mee had, is het thread-safe doorgeven van een job aan een job thread. Hieronder heb ik schematisch mijn oplossing weergegeven, waarbij ik per job thread 2 mutexen gebruik om zowel de scheduler te kunnen laten wachten op een vrije job thread, en een job thread kan laten wachten op een job.

Nu vraag ik me af: kan dit niet makkelijker?

code:

      Scheduler thread      :        Job Thread
                            :
    IdleMutex  HelpMutex    :   IdleMutex  HelpMutex    
       | |        |X|       :      |X|         |
 a)    | |        |X|       :      |X|         |
       | |        |X|       :      |X|         |
       | |/_______|X|______(1)_____|X|         |
       |X|\       |X|       :       |          |
       |X|        |X|       :       |          |
 b)    |X|        |X| Set Job on_\  |          |
       |X|        |X| Job Thread /  |          |
       |X|        |X|       :       |         | |
       |X|        |X|______(2)______|________\| |
       |X|         |        :       |        /|X|
 c)    |X|         |        :       |         |X|
       |X|         |        :      | |        |X|
       |X|_________|_______(3)____\| |        |X|
        |          |        :     /|X|        |X|
 d)     |          |        :      |X|        |X|
        |         | |       :      |X|        |X|
        |         | |/_____(4)_____|X|________|X|
        |         |X|\      :      |X|         | 
 e)     |         |X|       :      |X|         | 
        |         |X|       :      |X|         | 

Legenda:     |   = Mutex is niet in het bezit van de thread.
            |X|  = Mutex is in het bezit van de thread.
            | |  = Thread wacht totdat het de mutex krijgt
            -->  = Thread geeft mutex vrij

Onderstaand stappenplan gaat uit van een Scheduler Thread met 1 Job Thread; in het echt zal de Scheduler Thread bij stap a) de Idle Mutexen van meerdere Job Threads in de gaten houden totdat er een van vrij komt (d.m.v. WaitForAny()), en met deze Job Thread de rest van de stappen doorlopen.

a) Hier is de Job Thread nog bezig met een job. Het bezit de Idle Mutex, en zolang de job thread deze heeft zal de Scheduler Thread blocken.
1) De Job Thread geeft de Idle mutex vrij, om aan de Scheduler Thread aan te geven dat het klaar is met zijn job.
b) De Job Thread wacht op de Help mutex, zodat deze Job Thread geblockt blijft totdat de scheduler het een job toewijst. Wanneer de Scheduler een job heeft voor deze thread, zet hij dit als een Job object op een member variabele van de Job Thread.
2) De Scheduler thread geeft de Help Mutex vrij, zodat de Job Thread weer running wordt en weet dat het een job toegewezen heeft gekregen.
c) De Job Thread wacht totdat het de Idle Mutex terugkrijgt.
3) De Scheduler geeft de Idle mutex vrij zodat de Job Thread deze weer kan pakken.
d) De Job Thread heeft de Idle mutex gepakt zodat het als "bezet" gemarkeerd is; de Scheduler thread zal blocken wanneer het probeert de Idle mutex van deze Job Thread te pakken om een nieuwe job toe te wijzen.
4) Nu de Job Thread de Idle Mutex heeft, kan het de Help Mutex teruggeven aan de Scheduler thread en beginnen aan zijn Job.
e) Door het pakken van de Help Mutex weet de Scheduler thread zeker dat de Job Thread de Idle Mutex heeft, en kan weer wachten totdat de Idle Mutex vrijkomt.

Het over en weer geven van de mutexen is volgens mij nodig omdat je nooit zeker weet dat wanneer de ene thread een mutex vrijgeeft, dat de andere thread deze meteen pakt.

Maar denk ik te ingewikkeld?
_{Pfoe, dat is een flinke lap tekst geworden}

[whoami]

MrBucket schreef op vrijdag 15 augustus 2008 @ 19:37:
Wanneer een job thread tijdelijk niets te doen heeft, zal deze thread blocken totdat het een nieuwe job krijgt van de scheduler. Dit voorkomt dat er continue nieuwe threads worden aangemaakt en opgeruimd.

Heb je geen threadpool die dit voor jou kan afhandelen (voorkomen dat er nieuwe threads aangemaakt & opgeruimd worden) ?
Dan kan je dit al abstraheren, en hoef jij er geen aandacht meer aan te besteden.

[MrBucket]

whoami schreef op vrijdag 15 augustus 2008 @ 19:51:
[...]
Heb je geen threadpool die dit voor jou kan afhandelen (voorkomen dat er nieuwe threads aangemaakt & opgeruimd worden) ?
Dan kan je dit al abstraheren, en hoef jij er geen aandacht meer aan te besteden.

Klopt, dat kan. Ik zou evt. met een semafoor het aantal lopende threads kunnen limiteren, en pas een nieuwe job kunnen toekennen wanneer er een job thread klaar is en de semafoor weer ophoogt.

Maar ik wil het hardcore doen, zonder hulp van een threadpool. Je zou zeggen dat dat niet zoveel moeilijker moet zijn, maar dat over en weer gooien van 2 mutexen is de enig werkbare oplossing die ik kon verzinnen. Is er geen makkelijkere constructie (met mutexen, wait handles of semaforen, maakt me niet uit eigenlijk)?

Oh, en even ter verduidelijking: dit is een hobbyprojectje waarin ik een aantal dingen wil uitproberen. Dus vandaar dat ik niet voor de makkelijkste oplossing ga

[ Voor 9% gewijzigd door MrBucket op 15-08-2008 20:03 ]

[H!GHGuY]

Er al eens over nagedacht dat een lock niet per definitie thread-safe hoeft te zijn?
Klinkt idioot, maar kijk hier maar eens naar.
Ik zou dus opletten met het "doorgeven" van mutex-objecten tussen threads.

Ik denk dat je beter je algoritme eens onder de loep neemt. Bij wat jij doet denk ik aan het volgende:
Neem een priority-queue <ScheduleTime, Job> en steek er elke job in.
Neem een message queue waar je op kunt blocken voor een gegeven tijd.

Laat je scheduler thread blocken op de message queue voor de duur van (queue->front()->ScheduleTime - Time->Now).
Wanneer de je een timeout krijgt (double check je de tijd natuurlijk! en) neem je de Job uit de queue en geef je hem aan een worker thread. Daarna ga je terug wachten op een message met de timeout.
Wanneer je Job klaar is in de worker thread, zend hij een message met de volgende Job en een nieuwe ScheduleTimeout naar de scheduler thread. Die zal terugkeren van de message queue (dit keer zonder Timeout) en stopt die Job in de priority-queue. Daarna gaat hij terug wachten op de timeout.

Geen hassle met locks, deadlocks/livelocks, race condities en whatever else en toch een ultra-schaalbaar design.

[MrBucket]

H!GHGuY schreef op vrijdag 15 augustus 2008 @ 20:25:
Er al eens over nagedacht dat een lock niet per definitie thread-safe hoeft te zijn?
Klinkt idioot, maar kijk hier maar eens naar.
Ik zou dus opletten met het "doorgeven" van mutex-objecten tussen threads.

Voor zover ik kan overzien is deze constructie niet vatbaar voor race condities en/of deadlocks. Toch?

Ik denk dat je beter je algoritme eens onder de loep neemt. Bij wat jij doet denk ik aan het volgende:
Neem een priority-queue <ScheduleTime, Job> en steek er elke job in.
Neem een message queue waar je op kunt blocken voor een gegeven tijd.

Laat je scheduler thread blocken op de message queue voor de duur van (queue->front()->ScheduleTime - Time->Now).
Wanneer de je een timeout krijgt (double check je de tijd natuurlijk! en) neem je de Job uit de queue en geef je hem aan een worker thread. Daarna ga je terug wachten op een message met de timeout.

Dit is precies zoals ik het nu doe (ik houd meerdere gewone queues bij in plaats van een enkele priority queue, maar a la).

Wanneer je Job klaar is in de worker thread, zend hij een message met de volgende Job en een nieuwe ScheduleTimeout naar de scheduler thread. Die zal terugkeren van de message queue (dit keer zonder Timeout) en stopt die Job in de priority-queue. Daarna gaat hij terug wachten op de timeout.

En dit stuk kon ik niet helemaal volgen, zou je dit misschien wat kunnen verduidelijken?
Zoals ik het nu lees schiet een Job Thread zelf een nieuwe Job in bij de Scheduler, of...?

[H!GHGuY]

MrBucket schreef op zaterdag 16 augustus 2008 @ 12:18:
Voor zover ik kan overzien is deze constructie niet vatbaar voor race condities en/of deadlocks. Toch?

Denk er maar eens zo over:

C#:

class Mutex
{
  private WaitQueue Queue = new WaitQueue(); // implementatie elders
  private Event evt = new Event(); // implementatie elders
  //...
  public void Lock()
  {
     this.WaitQueue.Add(System.Threading.Thread.CurrentThread)
     evt.WaitOne();
  }
  public void Unlock()
  {
     Thread wakeupthread = this.Waitqueue.Next();
     evt.ReleaseOne();
     wakeupThread.NotifyOne();
  }
  //...
}

De implementatie zelf is fout, maar wat er achterliggend kan gebeuren is wel correct.

Thread A: Mutex.Lock();
Thread A: SendToThreadB(Mutex)
Thread B: ReceiveMutex().Lock()
Thread A: Mutex.Unlock();

Dan wordt thread A wakker gemaakt, hoewel die al wakker was. B wordt nooit wakker gemaakt en is dus voor altijd gelocked.

Dit is precies zoals ik het nu doe (ik houd meerdere gewone queues bij in plaats van een enkele priority queue, maar a la).

Wat doe je dan met die queues? 1 per 1 pollen? 1 thread/queue ? locking ?
Stuk voor stuk inefficient, tenzij je iets beters hebt natuurlijk.

En dit stuk kon ik niet helemaal volgen, zou je dit misschien wat kunnen verduidelijken?
Zoals ik het nu lees schiet een Job Thread zelf een nieuwe Job in bij de Scheduler, of...?

In pseudoCode:

code:

SchedulerThread:
Schedule()
{
Timeout = Prioqueue.Front.Time - Now

Message msg = MessageQueue.DeQueue(Timeout)
if (msg = nothing)
{
  Job = PrioQueue.Front.Job;
  PrioQueue.Pop_First();
  ThreadPool.ScheduleWork(Job)
}
else
{
  PrioQueue.Push(msg.Time, msg.Job);
}
}

Job:
Execute() { DoWork(); ScheduleNext() }
ScheduleNext() { Scheduler.SendMessage(NextTimeout, this); }