Zsynchronizowane strumienie wyjściowe w C++20

Zsynchronizowane strumienie wyjściowe w C++20

Jedną z wielu rzeczy zawartych w C++20 jest obsługa synchronizacji strumieni wyjściowych dla operacji, które mogą mieć warunki wyścigu. Aby zrozumieć problem, zacznijmy od następującego przykładu:

int main()
{
   std::vector<std::jthread> threads;
   for (int i = 1; i <= 10; ++i)
   {
      threads.push_back(
         std::jthread([](const int id)
            {
               std::cout << "I am thread [" << id << "]" << '\n';
            }, i));
   }
}

Uruchamiamy kilka wątków i to, co robią, to drukowanie tekstu na konsoli, a następnie kończenie. Więc można by się spodziewać wyniku takiego:

I am thread [4]
I am thread [3]
I am thread [8]
I am thread [5]
I am thread [9]
I am thread [6]
I am thread [10]
I am thread [7]
I am thread [2]
I am thread [1]

Nie można oczekiwać, że wątki będą wykonywane w kolejności, w której zostały uruchomione, ale intencją jest uzyskanie danych wyjściowych takich jak powyższy. Okazuje się jednak, że otrzymujesz raczej zaszyfrowany tekst, taki jak ten:

I am thread [I am thread [4I am thread [2I am thread [7]I am thread [9]
I am thread [3]

I am thread [5]
I am thread [10]I am thread [8]
I am thread [6]
]
]
1]

Poniższy przykład nie przedstawia tego problemu. Rzućmy okiem:

int main()
{
   std::vector<std::jthread> threads;

   auto worker = [](std::string text) { std::cout << text; };
   auto names = { "Alpha", "Beta", "Gamma", "Delta", "Epsilon" };

   using namespace std::string_literals;
   for (auto const& name : names)
      threads.push_back(std::jthread(worker, "Hello, "s + name + "!\n"));
}

Bez względu na to, ile razy uruchomisz ten kod, zawsze wyświetla wynik w następującej formie:

Hello, Alpha!
Hello, Delta!
Hello, Gamma!
Hello, Beta!
Hello, Epsilon!

W obu tych przykładach użyłem std::cout do drukowania do konsoli wyjściowej. Oczywiście są wyścigi danych, które występują w pierwszym przykładzie, ale nie w drugim. Jednak std::cout gwarantuje bezpieczeństwo wątków (chyba że sync_with_stdio(false) został nazwany). Użycie operator<< jest w porządku, jak widać w drugim przykładzie. Ale wiele wywołań do tego operator<< nie są atomowe i można je przerwać i wznowić po wznowieniu wykonywania wątku. Więc jeśli weźmiemy linię std::cout << "I am thread [" << id << "]" << '\n'; są cztery wywołania do operator<< . Tak więc wykonanie może zostać zatrzymane między dowolnym z tych, a inny wątek zapisze dane wyjściowe. Tak więc dane wyjściowe mogą mieć dowolną z tych postaci:

  • I am thread [1]\nI am thread [2]\n
  • I am thread[I am thread[2]\n1]\n
  • I am thread[1I am thread]\n[2]\n
  • itd. itp.

Oznacza to, że możesz rozwiązać ten problem, pisząc w strumieniu wyjściowym i po uzyskaniu całego tekstu, który powinien zostać zapisany w konsoli, używając std::cout obiekt. Pokazuje to następujący przykład:

int main()
{
   std::vector<std::jthread> threads;
   for (int i = 1; i <= 10; ++i)
   {
      threads.push_back(
         std::jthread([](const int id)
            {
               std::stringstream s;
               s << "I am thread [" << id << "]" << '\n';
               std::cout << s.str();
            }, i));
   }
}

W C++20 jest prostsze rozwiązanie:std::basic_osyncstream (dostępne w nowym <syncstream> header), który umożliwia wielu wątkom zapisywanie w tym samym strumieniu wyjściowym w sposób zsynchronizowany. Zmiany w pierwszym przykładzie, który miał wyścigi danych, są minimalne, ale mogą mieć dwie formy:

  • używając nazwanej zmiennej
int main()
{
   std::vector<std::jthread> threads;
   for (int i = 1; i <= 10; ++i)
   {
      threads.push_back(
         std::jthread([](const int id)
            {
               std::osyncstream scout{ std::cout };
               scout << "I am thread [" << id << "]" << '\n';
            }, i));
   }
}
  • używając tymczasowego obiektu
int main()
{
   std::vector<std::jthread> threads;
   for (int i = 1; i <= 10; ++i)
   {
      threads.push_back(
         std::jthread([](const int id)
            {
               std::osyncstream { std::cout } << "I am thread [" << id << "]" << '\n';
            }, i));
   }
}

Uwaga :Istnieją dwie specjalizacje std::basic_osyncstream dla typowych typów znaków, std::osyncstream dla char (które widzieliśmy w poprzednim fragmencie) i std::wosyncstream dla wchar_t .

Dopóki wszystkie zapisy do tego samego bufora docelowego (takiego jak standardowe wyjście w tym przykładzie) są zapisywane przez instancje std::basic_osyncstream klasy, gwarantuje się, że te operacje zapisu są wolne od wyścigów danych. Działa to tak, że std::basic_osyncstream otacza strumień wyjściowy, ale zawiera również wewnętrzny bufor (typu std::basic_syncbuf ), który gromadzi dane wyjściowe, gdzie pojawia się jako ciągła sekwencja znaków. Po zniszczeniu lub podczas jawnego wywołania emit() metoda, zawartość wewnętrznego bufora synchronizacji jest przesyłana do opakowanego strumienia. Zobaczmy kilka przykładów, aby zrozumieć, jak to działa.

int main()
{
   std::ostringstream str{ };
   {
      std::osyncstream syncstr{ str };
      syncstr << "Hello, World!";

      std::cout << "[1]:" << str.str() << '\n';
   }

   std::cout << "[2]:" << str.str() << '\n';
}

W tym przykładzie str to std::ostringstream . syncstr to std::osyncstream który owija ten strumień ciągów. Piszemy do zsynchronizowanego strumienia. W punkcie [1] , wywołując str() metoda ostringstream zwróci pusty ciąg, ponieważ strumień synchronizacji nie wyemitował zawartości swojego wewnętrznego bufora do opakowanego strumienia. Dzieje się tak po syncstr obiekt jest niszczony, gdy wychodzi poza zakres. Dlatego w punkcie [2] , str będzie zawierać tekst pisany. Wynik jest zatem następujący:

[1]:
[2]:Hello, World!

Możemy również jawnie wywołać emit() przenieść zawartość bufora wewnętrznego do opakowanego strumienia wyjściowego. Poniższy przykład to ilustruje:

int main()
{
   std::ostringstream str{ };
   {
      std::osyncstream syncstr{ str };
      syncstr << "Hello, World!";

      std::cout << "[1]:" << str.str() << '\n';

      syncstr.emit();

      std::cout << "[2]:" << str.str() << '\n';

      syncstr << "Hello, all!";

      std::cout << "[3]:" << str.str() << '\n';
   }

   std::cout << "[4]:" << str.str() << '\n';
}

To, co się tutaj dzieje, to:

  • w punkcie [1] , nic nie zostało wyemitowane, więc zawartość ostringstream jest pusty.
  • w punkcie [2] strumień ciągów będzie zawierał „Hello, World!” tekst od połączenia z emit() wcześniej wystąpiło
  • w punkcie [3] strumień ciągów zawiera tylko „Hello, World!” mimo że wcześniej w strumieniu wyjściowym synchronizacji zapisano więcej tekstu
  • w punkcie [4] strumień ciągów zawiera „Hello, World!Hello, all!” ponieważ strumień wyjściowy synchronizacji wyemitował resztę swojego wewnętrznego bufora po wyjściu poza zakres.

Dane wyjściowe są następujące:

[1]:
[2]:Hello, World!
[3]:Hello, World!
[4]:Hello, World!Hello, all!

Możesz także uzyskać wskaźnik do opakowanego strumienia std::basic_osyncstream z wezwaniem do get_wrapped() . Można to wykorzystać do sekwencjonowania zawartości do tego samego strumienia z wielu wystąpień std::basic_osyncstream . Oto przykład:

int main()
{
   std::ostringstream str{ };
   {
      std::osyncstream syncstr{ str };
      syncstr << "Hello, World!";

      std::cout << "[1]:" << str.str() << '\n';

      {
         std::osyncstream syncstr2{ syncstr.get_wrapped() };
         syncstr2 << "Hello, all!";

         std::cout << "[2]:" << str.str() << '\n';
      }

      std::cout << "[3]:" << str.str() << '\n';
   }

   std::cout << "[4]:" << str.str() << '\n';
}

W tym fragmencie mamy dwa std::osyncstream obiekty o różnych zakresach, oba owijające ten sam strumień ciągu. Co się dzieje, to:

  • w punkcie [1] , str jest pusty, ponieważ syncstr nie wyemitował swojej zawartości
  • w punkcie [2] , str jest nadal pusty, ponieważ żaden syncstr ani syncstr2 wyemitowali swoją treść
  • w punkcie [3] , str zawiera tekst „Witam wszystkich!” ponieważ syncstr2 wyszedł poza zakres i dlatego wyemitował swoją wewnętrzną zawartość
  • w punkcie [4] , str zawiera tekst „Witaj wszystkim! Witaj świecie!” ponieważ syncstr również wyszedł poza zakres i dlatego wyemitował swoją wewnętrzną zawartość

Dane wyjściowe tego przykładu są następujące:

[1]:
[2]:
[3]:Hello, all!
[4]:Hello, all!Hello, World!

std::osyncstream jest standardową alternatywą dla C++20 do jawnego używania mechanizmów synchronizacji (takich jak std::mutex ) do zapisywania treści do strumieni wyjściowych bez wyścigu danych.