Dining Philosopher Problem II

Inhaltsverzeichnis

Im letzten Beitrag "Dining Philosophers Problem I" begann Andre Adrian seine Analyse der klassischen Problemstellung. Heute verwendet er Atomics, Mutexe und Locks.

By Benjamin D. Esham / Wikimedia Commons, CC BY-SA 3.0, https://commons.wikimedia.org/w/index.php?curid=56559

Ich möchte kurz daran erinnern, wo Andres Analyse beim letzten Mal endete.

// dp_5.cpp
#include <iostream>
#include <thread>
#include <chrono>
#include <atomic>

int myrand(int min, int max) {
  return rand()%(max-min)+min;
}

void lock(std::atomic<int>& m) {
  while (m)
    ; // busy waiting
  m=1;
}

void unlock(std::atomic<int>& m) {
  m=0;
}

void phil(int ph, std::atomic<int>& ma, std::atomic<int>& mb) {
  while(true) {
    int duration=myrand(1000, 2000);
    std::cout<<ph<<" thinks "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    lock(ma);
    std::cout<<"\t\t"<<ph<<" got ma\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    lock(mb);
    std::cout<<"\t\t"<<ph<<" got mb\n";

    duration=myrand(1000, 2000);
    std::cout<<"\t\t\t\t"<<ph<<" eats "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    unlock(mb);
    unlock(ma);
  }
}

int main() {
  std::cout<<"dp_5\n";
  srand(time(nullptr));

  std::atomic<int> m1{0}, m2{0}, m3{0}, m4{0};

  std::thread t1([&] {phil(1, m1, m2);});
  std::thread t2([&] {phil(2, m2, m3);});
  std::thread t3([&] {phil(3, m3, m4);});
  std::thread t4([&] {phil(4, m1, m4);});

  t1.join();
  t2.join();
  t3.join();
  t4.join();
}

Das Programm sieht richtig aus. Es gibt immer noch eine winzige Chance für Fehlverhalten. Die beiden Operationen "ist eine Ressource verfügbar" und "Ressource als in Gebrauch markieren" in der Funktion lock() sind atomar, aber sie sind immer noch zwei Operationen. Zwischen diesen beiden Operationen kann der Scheduler einen Threadwechsel platzieren. Und dieser Threadwechsel zu diesem ungünstigsten Zeitpunkt kann sehr schwer zu findende Fehler im Programm erzeugen.

Richtiges Busy Waiting mit Resourcehierarchie

Glücklicherweise verfügen alle aktuellen Computer über eine atomare Operation "Ressource testen und bei positivem Test Ressource als verwendet markieren". In der Programmiersprache C++ stellt uns der Typ atomic_flag diese spezielle Operation "test and set" zur Verfügung. Die Datei dp_6.cpp ist die erste richtige Lösung für das Problem der Restaurantphilosophen:

// dp_6.cpp
#include <iostream>
#include <thread>
#include <chrono>
#include <atomic>

int myrand(int min, int max) {
  return rand()%(max-min)+min;
}

void lock(std::atomic_flag& m) {
  while (m.test_and_set())
    ; // busy waiting
}

void unlock(std::atomic_flag& m) {
  m.clear();
}

void phil(int ph, std::atomic_flag& ma, std::atomic_flag& mb) {
  while(true) {
    int duration=myrand(1000, 2000);
    std::cout<<ph<<" thinks "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    lock(ma);
    std::cout<<"\t\t"<<ph<<" got ma\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    lock(mb);
    std::cout<<"\t\t"<<ph<<" got mb\n";

    duration=myrand(1000, 2000);
    std::cout<<"\t\t\t\t"<<ph<<" eats "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    unlock(mb);
    unlock(ma);
  }
}

int main() {
  std::cout<<"dp_6\n";
  srand(time(nullptr));

  std::atomic_flag m1, m2, m3, m4;
  unlock(m1);
  unlock(m2);
  unlock(m3);
  unlock(m4);

  std::thread t1([&] {phil(1, m1, m2);});
  std::thread t2([&] {phil(2, m2, m3);});
  std::thread t3([&] {phil(3, m3, m4);});
  std::thread t4([&] {phil(4, m1, m4);});

  t1.join();
  t2.join();
  t3.join();
  t4.join();
}

Die Ausgabe der Programmversion 6 ist der letzten Ausgabe ähnlich. Das Problem der speisenden Philosophen ist gutmütig. Eine Ressource wird nur von zwei Threads gemeinsam genutzt. Der atomic_flag-Spinlock wird nur dann wirklich benötigt, wenn mehrere Threads dieselbe Ressource erhalten wollen.

Optimiertes Busy Waiting mit Resourcehierarchie

Der Nachteil von Spinlock ist das beschäftigte Warten (busy waiting). Die while-Schleife in lock() ist eine Verschwendung von CPU-Ressourcen. Eine Lösung für dieses Problem besteht darin, eine sleep_for()-Funktion in den Rumpf dieser while-Schleife einzufügen. Die Funktion sleep_for() führt das Warten im Scheduler durch. Dieses Warten ist viel besser als das Warten in der Anwendung. Wie immer gibt es einen Preis. Das sleep_for() verlangsamt den Programmfortschritt, weil nur noch alle 8 Millisekunden der Zustand der Ressource geprüft wird. Datei dp_7.cpp ist die zweite richtige Lösung:

// dp_7.cpp
void lock(std::atomic_flag& m) {
  while (m.test_and_set())
    std::this_thread::sleep_for(std::chrono::milliseconds(8));
}

Hinweis: Ein std::this_thread::yield() anstelle von sleep_for() reduziert die CPU-Last auf dem Computer des Autors nicht. Die Auswirkung von yield() ist implementierungsabhängig.

std::mutex mit Resourcehierarchie

Um geschäftiges Warten vollständig zu vermeiden, brauchen wir mehr Hilfe vom Scheduler. Wenn der Scheduler einen "Ressource belegen"-Aufruf erhält und diese Ressource ist schon belegt, schiebt der Scheduler den Thread in die Scheduler wait-Queue. Nachdem der Scheduler einen "Ressource ist verfügbar"-Aufruf erhalten hat, wird der wartende Thread von der wait-Queue wieder in die ready-Queue verschoben.

Der Informationsaustausch zwischen Thread und Scheduler ist teuer. Aus diesem Grund bietet C++ sowohl Spinlock als auch Mutex. Spinlock wartet im Thread und Mutex wartet im Scheduler.

Die Datei dp_8.cpp zeigt die Mutex-Lösung. Bitte beachte das #include <mutex> :

// dp_8.cpp
#include <iostream>
#include <thread>
#include <chrono>
#include <mutex>

int myrand(int min, int max) {
  return rand()%(max-min)+min;
}

void phil(int ph, std::mutex& ma, std::mutex& mb) {
  while(true) {
    int duration=myrand(1000, 2000);
    std::cout<<ph<<" thinks "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    ma.lock();
    std::cout<<"\t\t"<<ph<<" got ma\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    mb.lock();
    std::cout<<"\t\t"<<ph<<" got mb\n";

    duration=myrand(1000, 2000);
    std::cout<<"\t\t\t\t"<<ph<<" eats "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));
    mb.unlock(); // (9)
    ma.unlock();
  }
}

int main() {
  std::cout<<"dp_8\n";
  srand(time(nullptr));

  std::mutex m1, m2, m3, m4;

  std::thread t1([&] {phil(1, m1, m2);});
  std::thread t2([&] {phil(2, m2, m3);});
  std::thread t3([&] {phil(3, m3, m4);});
  std::thread t4([&] {phil(4, m1, m4);});

  t1.join();
  t2.join();
  t3.join();
  t4.join();
}

Programmversion 8 ist korrekt und verbraucht sehr wenig CPU-Ressourcen. C++ bietet einen Wrapper für Mutex, um Programmierern das Leben zu erleichtern.

std::lock_guard mit Resourcehierarchie

Wenn wir das lock_guard-Template verwenden, müssen wir ihm nur den Mutex übergeben.

Die Anweisung lock wird im automatisch im Konstruktor des Locks und die Anweisung unlock automatisch in seinem Destruktor am Ende des Blocks aufgerufen, also an der nächsten schließenden Klammer }. Durch lock_guard wird unlock auch dann ausgeführt, wenn der Block durch eine Exception verlassen wird.

Die bequeme Version ist dp_9.cpp:

// dp_9.cpp
void phil(int ph, std::mutex& ma, std::mutex& mb) {
  while(true) {
    int duration=myrand(1000, 2000);
    std::cout<<ph<<" thinks "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    std::lock_guard<std::mutex> ga(ma);
    std::cout<<"\t\t"<<ph<<" got ma\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    std::lock_guard<std::mutex> gb(mb);
    std::cout<<"\t\t"<<ph<<" got mb\n";

    duration=myrand(1000, 2000);
    std::cout<<"\t\t\t\t"<<ph<<" eats "<<duration<<"ms\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));
  }
}

Wir werden immer besser. Die Programmversionen 8 und 9 sind korrekt und haben geringe CPU-Auslastung. Aber achten Sie auf die Programmausgabe:

Die Programmausgabe ist leicht verstümmelt. Vielleicht haben Sie diese Ausgabeverzerrung schon bei anderen Programmversionen gesehen. An den Spinlock-Programmversionen 6 und 7 oder den Mutex-Programmversionen 8 und 9 ist nichts auszusetzen.

std::lock_guard mit synchronisierter Ausgabe und Resourcenhierarchie

Die Konsolenausgabe selbst ist eine Ressource. Dies ist der Grund für die verstümmelte Ausgabe in Multithread-Programmen. Die Lösung besteht darin, jede Konsolenausgabe mit einem lock_guard zu versehen. Siehe dp_10.cpp:

// dp_10.cpp

std::mutex mo;

void phil(int ph, std::mutex& ma, std::mutex& mb) {
  while(true) {
    int duration=myrand(1000, 2000);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<ph<<" thinks "<<duration<<"ms\n";
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    std::lock_guard<std::mutex> ga(ma);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<"\t\t"<<ph<<" got ma\n";
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    std::lock_guard<std::mutex> gb(mb);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<"\t\t"<<ph<<" got mb\n";
    }

    duration=myrand(1000, 2000);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<"\t\t\t\t"<<ph<<" eats "<<duration<<"ms\n";
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));
  }
}

Der globale Mutex mo schützt die Konsolenausgabe-Ressource. Jede cout-Anweisung befindet sich in ihrem Block und das lock_guard()-Template stellt sicher, dass die Konsolenausgabe nicht mehr verstümmelt wird.

std::lock_guard mit synchronisierter Ausgabe und Resourcenhierarchie und einem Zähler

Als kleinen Bonus habe ich dp_11.cpp hinzugefügt. Diese Programmversion zählt die Anzahl der Philosophen-Threads, die gleichzeitig essen. Da wir vier Gabeln haben, sollte es Zeiten geben, in denen zwei Philosophen-Thread gleichzeitig im Zustand essen sind. Bitte beachten Sie, dass Sie erneut #include <atomic> benötigen. Siehe dp_11.cpp:

// dp_11.cpp

std::mutex mo;
std::atomic<int> cnt = 0;

void phil(int ph, std::mutex& ma, std::mutex& mb) {
  while(true) {
    int duration=myrand(1000, 2000);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<ph<<" thinks "<<duration<<"ms\n";
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));

    std::lock_guard<std::mutex> ga(ma);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<"\t\t"<<ph<<" got ma\n";
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    std::lock_guard<std::mutex> gb(mb);
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<"\t\t"<<ph<<" got mb\n";
    }

    duration=myrand(1000, 2000);
    ++cnt;
    {
      std::lock_guard<std::mutex> g(mo);
      std::cout<<"\t\t\t\t"<<ph<<" eats "<<duration<<"ms "<<cnt<<"\n";
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(duration));
    --cnt;
  }
}

Die Ausgabe der Programmversion 11 ist:

Der Zusatz ist die Nummer 1 oder 2 am Ende der "eats"-Protokollierung.

Wie geht's weiter?

In seiner nächsten Folge des Dining Philosoher Problems verwendet Andre std::unique_lock (C++11), std::scoped_lock (C++17) und std::semaphore (C++20). ()