multithreading - 保存 fprintf 调用的变量参数列表

Question

我正在编写一个重型多线程 [>170 个线程] c++11 程序。每个线程都将信息记录到所有线程使用的一个文件中。出于性能原因，我想创建一个 log 线程，它通过 fprintf() 将信息写入全局文件。我不知道如何组织工作线程写入信息的结构，然后可以由日志线程读取该信息。

为什么我不在每个工作线程中调用sprintf()，然后只向log线程提供输出缓冲区？对于日志文件中的格式化输出，我在 fprintf() 函数中使用了 locale，这与线程的其余部分不同。因此，我必须永久切换和锁定/保护 xprintf() 调用，以便区分 locale 输出。 在log线程中，我有一个locale设置用于整个输出，而worker线程有其locale版本.

log线程的另一个原因是我必须对输出进行“分组”，否则来自每个worker线程的信息将不会位于一个 block 中:

错误:

Information A Thread #1
Information A Thread #2
Information B Thread #1
Information B Thread #2

正确:

Information A Thread #1
Information B Thread #1
Information A Thread #2
Information B Thread #2

为了实现这种分组，我必须保护每个工作线程中的输出，这会减慢线程执行时间。

如何将 va_list 保存到一个结构中，以便 log 线程可以读取它并传递回 fprintf() ？

Answer 1

我不明白如何使用旧版 C vprintf 和 va_list 轻松完成此操作。当您想要在线程之间传递东西时，迟早您将需要以某种方式使用堆。

下面是使用 Boost.Format 的解决方案用于格式化和 Boost.Variant用于参数传递。如果您按顺序连接以下代码块，则该示例完整且有效。如果使用 GCC 编译，则需要传递 -pthread 链接器标志。当然，您还需要两个仅包含头文件的 Boost 库。这是我们将使用的 header 。

#include <condition_variable>
#include <iostream>
#include <list>
#include <locale>
#include <mutex>
#include <random>
#include <string>
#include <thread>
#include <utility>
#include <vector>

#include <boost/format.hpp>
#include <boost/variant.hpp>

首先，我们需要某种机制来异步执行一些任务，在本例中，打印我们的日志消息。由于这个概念是通用的，因此我为此使用“抽象”基类 Spooler。其代码基于 Herb Sutter 在 CppCon 2014 上的演讲“无锁编程(或杂耍 Razor Blade )”(part 1、part 2)。我不会详细介绍这段代码，因为它主要是与您的问题没有直接关系的脚手架，并且我假设您已经具备了这项功能。我的 Spooler 使用受 std::mutex 保护的 std::list 作为任务队列。考虑使用无锁数据结构可能是值得的。

class Spooler
{
private:

  bool done_ {};
  std::list<std::function<void(void)>> queue_ {};
  std::mutex mutex_ {};
  std::condition_variable condvar_ {};
  std::thread worker_ {};

public:

  Spooler() : worker_ {[this](){ work(); }}
  {
  }

  ~Spooler()
  {
    auto poison = [this](){ done_ = true; };
    this->submit(std::move(poison));
    if (this->worker_.joinable())
      this->worker_.join();
  }

protected:

  void
  submit(std::function<void(void)> task)
  {
    // This is basically a push_back but avoids potentially blocking
    // calls while in the critical section.
    decltype(this->queue_) tmp {std::move(task)};
    {
      std::unique_lock<std::mutex> lck {this->mutex_};
      this->queue_.splice(this->queue_.cend(), tmp);
    }
    this->condvar_.notify_all();
  }

private:

  void
  work()
  {
    do
      {
        std::unique_lock<std::mutex> lck {this->mutex_};
        while (this->queue_.empty())
          this->condvar_.wait(lck);
        const auto task = std::move(this->queue_.front());
        this->queue_.pop_front();
        lck.unlock();
        task();
      }
    while (!this->done_);
  }
};

从 Spooler 中，我们现在派生出一个 Logger，它(私下)从 Spooler 继承其异步功能，并添加特定于日志记录的功能。它只有一个名为 log 的函数成员，它采用格式字符串和零个或多个参数作为参数，将其格式化为 boost 的 std::vector: :变体。

不幸的是，这限制了我们可以支持的固定数量的类型，但这不应该是一个大问题，因为 C printf 也不支持任意类型。在本示例中，我仅使用 int 和 double，但您可以使用 std::string 扩展列表，void * 指针或者你有什么。

log 函数构造一个 lambda 表达式，该表达式创建一个 boost::format 对象，为其提供所有参数，然后将其写入 std::log 或您想要格式化消息的任何位置。

boost::format 的构造函数具有接受格式字符串和区域设置的重载。您可能对此感兴趣，因为您在评论中提到设置自定义区域设置。通常的构造函数只接受一个参数，即格式字符串。

注意所有格式化和输出是如何在假脱机程序的线程上完成的。

class Logger : Spooler
{
 public:

  void
  log(const std::string& fmt,
      const std::vector<boost::variant<int, double>>& args)
  {
    auto task = [fmt, args](){
      boost::format msg {fmt, std::locale {"C"}};  // your locale here
      for (const auto& arg : args)
        msg % arg;  // feed the next argument
      std::clog << msg << std::endl;  // print the formatted message
    };
    this->submit(std::move(task));
  }
};

这就是所需要的一切。我们现在可以像本示例中那样使用 Logger。重要的是，在破坏 Logger 之前，所有工作线程都必须 join() ，否则它将无法处理所有消息。

int
main()
{
  Logger logger {};
  std::vector<std::thread> threads {};
  std::random_device rnddev {};
  for (int i = 0; i < 4; ++i)
    {
      const auto seed = rnddev();
      auto task = [&logger, i, seed](){
        std::default_random_engine rndeng {seed};
        std::uniform_real_distribution<double> rnddist {0.0, 0.5};
        for (double p = 0.0; p < 1.0; p += rnddist(rndeng))
          logger.log("thread #%d is %6.2f %% done", {i, 100.0 * p});
        logger.log("thread #%d has completed its work", {i});
      };
      threads.emplace_back(std::move(task));
    }
  for (auto& thread : threads)
    thread.join();
}

可能的输出:

thread #1 is   0.00 % done
thread #0 is   0.00 % done
thread #0 is  26.84 % done
thread #0 is  76.15 % done
thread #3 is   0.00 % done
thread #0 has completed its work
thread #3 is  34.70 % done
thread #3 is  78.92 % done
thread #3 is  91.89 % done
thread #3 has completed its work
thread #1 is  26.98 % done
thread #1 is  73.84 % done
thread #1 has completed its work
thread #2 is   0.00 % done
thread #2 is  10.17 % done
thread #2 is  29.85 % done
thread #2 is  79.03 % done
thread #2 has completed its work