Синхронизация многопоточного вывода STD cout

Question

В последнее время я работал с многопоточным кодированием, через некоторое время я понял, что, если бы я использовал std :: cout в разных boost :: threads, вывод получился бы без логического порядка, программы, которую я 'Тестирование m выглядит примерно так:

#include <boost/thread/thread.hpp>
#include <iostream>

int my01( void )
{
    std::cout << "my01" << std::endl;
    return 0;
}
/* my02, my03 and my04 are the same with different outputs*/
[...]
int main( void )
{
    boost::thread t1(&my01);
    boost::thread t2(&my02);
    boost::thread t3(&my03);
    boost::thread t4(&my04);

    while(!t1.joinable() || !t2.joinable() || !t3.joinable() || !t4.joinable());

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

    std::cout << "The end!" << std::endl;
    getchar();
    return 0;
}

И вывод обычно такой (меняется):

my02my01
my04
my03
BLANK LINE
Конец!

Имея в виду эту проблему, я думал о создании единого потока для управления всеми выходными данными, чтобы они были в таком порядке, как:

my01
my02
my03
my04
Конец!

Какой оптимальный способнаписать такую ​​ветку или управлять этими выходами?
Пожалуйста, прочитайте ответы и на этот вопрос: Cout синхронизирован / потокобезопасен?

Ps: я использую Visual C ++ 2010Express и мой процессор имеет 8 различных ядер.

Спасибо за ваше время!

Answer 1

Прежде всего, вы можете рассмотреть возможность избежать всего явного управления потоками и вместо этого использовать std::async для запуска ваших задач в произвольном количестве отдельных потоков.

Во-вторых, вместо ввода-вывода в самих потоках, вы хотите создать результаты и сделать вывод сам по себе. Это означает, что функция потока просто создает некоторые данные и оставляет их вызывающей стороне, чтобы фактически записать это:

std::string process(int value) {
     std::ostringstream buffer;
     buffer << "my" << std::setfill('0') << std::setw(2) << value;
     return buffer.str();
}

Затем нам нужно запустить четыре копии этого асинхронно:

std::vector<std::future<std::string> > results;

for (int i=0; i<4; i++)
    results.push_back(std::async(std::launch::async, process, i));

Затем мы получаем результаты и распечатываем их по порядку:

for (auto &r : results)
    std::cout << r.get() << "\n";

Собрав их вместе, мы могли бы получить код, подобный этому:

#include <string>
#include <iostream>
#include <thread>
#include <future>
#include <sstream>
#include <vector>
#include <iomanip>

std::string process(int value) {
     std::ostringstream buffer;
     buffer << "my" << std::setfill('0') << std::setw(2) << value;
     return buffer.str();
}

int main() { 
    std::vector<std::future<std::string>> rets;

    for (int i=0; i<4; i++)
        rets.push_back(std::async(std::launch::async, process, i));

    for (auto & t : rets) {
        t.wait();
        std::cout << t.get() << "\n";
    }
}

Я должен добавить один незначительный момент: я основываю это на стандартных C ++ 11 future s. Я считаю, что основная идея должна также работать с Boost future s (на котором основан стандарт), но я не проверял это. Я ожидаю, что для работы с фьючерсами Boost потребуются некоторые незначительные изменения (например, в именах).

Answer 2

Я решил это путем кодирования тонкой оболочки, которая блокирует мьютекс при начале записи в поток и освобождает его вместе со сбросом потока после завершения оператора записи.

Использование: заменить std :: cout на safe_cout.

Имейте в виду, что он не поддерживает необычные функции std :: cout, такие как std :: endl.

См. Код ниже или возьмите его отсюда: https://github.com/dkorolev/felicity/blob/master/safe_ostream.h

#include <cassert>
#include <iostream>
#include <mutex>
#include <memory>

struct safe_ostream {
  struct guarded_impl {
    guarded_impl() = delete;
    guarded_impl(const guarded_impl&) = delete;
    void operator=(const guarded_impl&) = delete;
    guarded_impl(std::ostream& ostream, std::mutex& mutex) : ostream_(ostream), guard_(mutex) {
    }
    ~guarded_impl() {
      ostream_.flush();
    }
    template<typename T> void write(const T& x) {
      ostream_ << x;
    }
    std::ostream& ostream_;
    std::lock_guard<std::mutex> guard_;
  };
  struct impl {
    impl() = delete;
    void operator=(const impl&) = delete;
    impl(std::ostream& ostream, std::mutex& mutex) : unique_impl_(new guarded_impl(ostream, mutex)) {
    }
    impl(const impl& rhs) {
      assert(rhs.unique_impl_.get());
      unique_impl_.swap(rhs.unique_impl_);
    }
    template<typename T> impl& operator<<(const T& x) {
      guarded_impl* p = unique_impl_.get();
      assert(p);
      p->write(x);
      return *this;
    }
    mutable std::unique_ptr<guarded_impl> unique_impl_;
  };
  explicit safe_ostream(std::ostream& ostream) : ostream_(ostream) {
  }
  template<typename T> impl operator<<(const T& x) {
    return impl(ostream_, mutex_) << x;
  }
  std::ostream& ostream_;
  std::mutex mutex_;
};
safe_ostream safe_cout(std::cout);
safe_ostream safe_cerr(std::cerr);

Answer 3

Вам нужно либо наложить порядок на потоки так, чтобы порядок вывода был таким, как вы хотите (возможно, путем передачи экземпляров потоков или событий в соответствующие потоки, чтобы они могли выполняться только в вашем порядке), илиВы можете присвоить всем выходам порядковый номер потока, поставить все выходные в один поток «print» и, там, сохранить список любых неупорядоченных строк, чтобы распечатка была такой, как вы хотите.

В случае «реального» приложения (т. Е. Не тривиального тестового приложения, которое неправильно использует потоки), где потоки выполняют большую работу параллельно над последовательными буферами, порядок которых должен быть сохранен, заставляя потоки ждатьдруг друга обычно не является разумным вариантом.Обычно используют порядковые номера и впоследствии собирают буферный поток.

Answer 4

Дайте каждому потоку std::ostringstream для записи вывода. В конце программы выведите выходные данные каждого потока по порядку.

Как еще вы это сделаете, учитывая, что поток 4 может закончиться задолго до потока 1?

Answer 5

Использовать блокировку.Если вы можете использовать повышение, сделайте, например,

int my01(boost::mutex *coutGuard)
{
  {
     // lock cout until the closing brace
     boost::mutex::scoped_lock lock(*coutGuard);

     std::cout << "my01" << std::endl;
  }

  return 0;
}

int main( void )
{
   boost::mutex coutGuard;

   boost::thread t1(boost::bind(&my01, &coutGuard));
   ...
}

Вместо scoped_lock, можно использовать lock_guard.