问题:
下面的3个while循环包含已经注释掉的代码。我搜索了("TAG1", "TAG2"和"TAG3")以便更容易地识别它们。我只想让while循环在条件为真时等待，同时尽可能地减少CPU资源的使用。我首先尝试使用Boost条件变量，但存在竞态条件。将线程休眠"x"微秒是低效的，因为没有办法精确计时唤醒。最后，boost::this_thread::yield()似乎没有做任何事情。可能是因为我在双核系统上只有2个活动线程。具体来说，我如何使下面标记的三个区域更有效地运行，同时尽可能地少引入不必要的阻塞。
背景
目标:
我有一个记录大量数据的应用程序。在进行性能分析后，我发现很多时间都花费在记录操作上(将文本或二进制记录到本地硬盘上的文件)。我的目标是通过用调用线程化的缓冲流记录器替换非线程化的直接写入调用来减少logData调用的延迟。
已探索的选项:
- 升级2005年代慢速硬盘到SSD...可能。成本不是禁止性的...但涉及很多工作...需要升级超过200台计算机...
- Boost ASIO...我不需要所有的proactor/网络开销，寻找更简单、更轻量级的东西。
设计:
- 生产者和消费者线程模式，应用程序将数据写入缓冲区，后台线程稍后将其写入磁盘。因此，最终目标是使由应用层调用的writeMessage函数尽可能快地返回，同时数据以FIFO顺序正确/完全记录到日志文件中。
- 仅一个应用程序线程，仅一个写入线程。
- 基于环形缓冲区。做出这个决定的原因是使用尽可能少的锁，并且理想情况下...如果我错了，请纠正我...我不认为我需要任何锁。
- 缓冲区是静态分配的字符数组，但出于性能原因，如果需要/希望，可以将其移动到堆中。
- 缓冲区具有指向应写入文件的下一个字符的起始指针。缓冲区具有指向应写入文件的最后一个字符后的数组索引的结束指针。结束指针永远不会超过开始指针。如果消息大于缓冲区，则写入器等待缓冲区被清空，并直接将新消息写入文件而不将超大消息放入缓冲区（一旦缓冲区被清空，工作线程将不会写入任何内容，因此没有争用）。
- 写入器（工作线程）仅更新环形缓冲区的起始指针。
- 主要的（应用程序线程）仅更新环形缓冲区的结束指针，并且仅在有可用空间时才将新数据插入缓冲区...否则它会等待缓冲区中的空间变得可用或按上述方式直接写入。
- 工作线程不断检查是否有要写入的数据（表示为缓冲区起始指针！=缓冲区结束指针的情况）。如果没有要写入的数据，则工作线程最好进入睡眠状态，并在应用程序线程插入某些内容（并更改缓冲区的结束指针，使其不再指向与开始指针相同的索引）后唤醒。我下面的内容涉及while循环不断检查该条件。这是一种非常糟糕/低效的等待缓冲区的方式。
结果：
- 在我的2009年双核笔记本电脑上使用SSD，我发现线程/缓冲基准测试的总写入时间与直接写入大约为1：6（0.609秒与0.095秒），但高度可变。通常，缓冲写入基准测试实际上比直接写入慢。我认为可变性是由于等待缓冲区释放空间的不良实现、等待缓冲区为空以及工作线程等待工作可用性的问题。我已经测量了一些while循环消耗超过10000个周期，并且怀疑这些周期实际上正在竞争其他线程（工作或应用程序）需要完成等待计算所需的硬件资源。
- 输出似乎检查通过。在启用TEST模式和缓冲区大小为10的压力测试中，我对数百MB的输出进行了差异化比较，并发现它与输入相等。
与当前版本的Boost（1.55）兼容
标题
#ifndef BufferedLogStream_h
#define BufferedLogStream_h
#include <stdio.h>
#include <iostream>
#include <iostream>
#include <cstdlib>
#include "boost\chrono\chrono.hpp"
#include "boost\thread\thread.hpp"
#include "boost\thread\locks.hpp"
#include "boost\thread\mutex.hpp"
#include "boost\thread\condition_variable.hpp"
#include <time.h>
using namespace std;
#define BENCHMARK_STR_SIZE 128
#define NUM_BENCHMARK_WRITES 524288
#define TEST 0
#define BENCHMARK 1
#define WORKER_LOOP_WAIT_MICROSEC 20
#define MAIN_LOOP_WAIT_MICROSEC 10
#if(TEST)
#define BUFFER_SIZE 10
#else
#define BUFFER_SIZE 33554432 //4 MB
#endif
class BufferedLogStream {
public:
BufferedLogStream();
void openFile(char* filename);
void flush();
void close();
inline void writeMessage(const char* message, unsigned int length);
void writeMessage(string message);
bool operator() () { return start != end; }
private:
void threadedWriter();
inline bool hasSomethingToWrite();
inline unsigned int getFreeSpaceInBuffer();
void appendStringToBuffer(const char* message, unsigned int length);
FILE* fp;
char* start;
char* end;
char* endofringbuffer;
char ringbuffer[BUFFER_SIZE];
bool workerthreadkeepalive;
boost::mutex mtx;
boost::condition_variable waitforempty;
boost::mutex workmtx;
boost::condition_variable waitforwork;
#if(TEST)
struct testbuffer {
int length;
char message[BUFFER_SIZE * 2];
};
public:
void test();
private:
void getNextRandomTest(testbuffer &tb);
FILE* datatowrite;
#endif
#if(BENCHMARK)
public:
void runBenchmark();
private:
void initBenchmarkString();
void runDirectWriteBaseline();
void runBufferedWriteBenchmark();
char benchmarkstr[BENCHMARK_STR_SIZE];
#endif
};
#if(TEST)
int main() {
BufferedLogStream* bl = new BufferedLogStream();
bl->openFile("replicated.txt");
bl->test();
bl->close();
cout << "Done" << endl;
cin.get();
return 0;
}
#endif
#if(BENCHMARK)
int main() {
BufferedLogStream* bl = new BufferedLogStream();
bl->runBenchmark();
cout << "Done" << endl;
cin.get();
return 0;
}
#endif //for benchmark
#endif
实施
#include "BufferedLogStream.h"
BufferedLogStream::BufferedLogStream() {
fp = NULL;
start = ringbuffer;
end = ringbuffer;
endofringbuffer = ringbuffer + BUFFER_SIZE;
workerthreadkeepalive = true;
}
void BufferedLogStream::openFile(char* filename) {
if(fp) close();
workerthreadkeepalive = true;
boost::thread t2(&BufferedLogStream::threadedWriter, this);
fp = fopen(filename, "w+b");
}
void BufferedLogStream::flush() {
fflush(fp);
}
void BufferedLogStream::close() {
workerthreadkeepalive = false;
if(!fp) return;
while(hasSomethingToWrite()) {
boost::unique_lock<boost::mutex> u(mtx);
waitforempty.wait_for(u, boost::chrono::microseconds(MAIN_LOOP_WAIT_MICROSEC));
}
flush();
fclose(fp);
fp = NULL;
}
void BufferedLogStream::threadedWriter() {
while(true) {
if(start != end) {
char* currentend = end;
if(start < currentend) {
fwrite(start, 1, currentend - start, fp);
}
else if(start > currentend) {
if(start != endofringbuffer) fwrite(start, 1, endofringbuffer - start, fp);
fwrite(ringbuffer, 1, currentend - ringbuffer, fp);
}
start = currentend;
waitforempty.notify_one();
}
else { //start == end...no work to do
if(!workerthreadkeepalive) return;
boost::unique_lock<boost::mutex> u(workmtx);
waitforwork.wait_for(u, boost::chrono::microseconds(WORKER_LOOP_WAIT_MICROSEC));
}
}
}
bool BufferedLogStream::hasSomethingToWrite() {
return start != end;
}
void BufferedLogStream::writeMessage(string message) {
writeMessage(message.c_str(), message.length());
}
unsigned int BufferedLogStream::getFreeSpaceInBuffer() {
if(end > start) return (start - ringbuffer) + (endofringbuffer - end) - 1;
if(end == start) return BUFFER_SIZE-1;
return start - end - 1; //case where start > end
}
void BufferedLogStream::appendStringToBuffer(const char* message, unsigned int length) {
if(end + length <= endofringbuffer) { //most common case for appropriately-sized buffer
memcpy(end, message, length);
end += length;
}
else {
int lengthtoendofbuffer = endofringbuffer - end;
if(lengthtoendofbuffer > 0) memcpy(end, message, lengthtoendofbuffer);
int remainderlength = length - lengthtoendofbuffer;
memcpy(ringbuffer, message + lengthtoendofbuffer, remainderlength);
end = ringbuffer + remainderlength;
}
}
void BufferedLogStream::writeMessage(const char* message, unsigned int length) {
if(length > BUFFER_SIZE - 1) { //if string is too large for buffer, wait for buffer to empty and bypass buffer, write directly to file
while(hasSomethingToWrite()); {
boost::unique_lock<boost::mutex> u(mtx);
waitforempty.wait_for(u, boost::chrono::microseconds(MAIN_LOOP_WAIT_MICROSEC));
}
fwrite(message, 1, length, fp);
}
else {
//wait until there is enough free space to insert new string
while(getFreeSpaceInBuffer() < length) {
boost::unique_lock<boost::mutex> u(mtx);
waitforempty.wait_for(u, boost::chrono::microseconds(MAIN_LOOP_WAIT_MICROSEC));
}
appendStringToBuffer(message, length);
}
waitforwork.notify_one();
}
#if(TEST)
void BufferedLogStream::getNextRandomTest(testbuffer &tb) {
tb.length = 1 + (rand() % (int)(BUFFER_SIZE * 1.05));
for(int i = 0; i < tb.length; i++) {
tb.message[i] = rand() % 26 + 65;
}
tb.message[tb.length] = '\n';
tb.length++;
tb.message[tb.length] = '\0';
}
void BufferedLogStream::test() {
cout << "Buffer size is: " << BUFFER_SIZE << endl;
testbuffer tb;
datatowrite = fopen("orig.txt", "w+b");
for(unsigned int i = 0; i < 7000000; i++) {
if(i % 1000000 == 0) cout << i << endl;
getNextRandomTest(tb);
writeMessage(tb.message, tb.length);
fwrite(tb.message, 1, tb.length, datatowrite);
}
fflush(datatowrite);
fclose(datatowrite);
}
#endif
#if(BENCHMARK)
void BufferedLogStream::initBenchmarkString() {
for(unsigned int i = 0; i < BENCHMARK_STR_SIZE - 1; i++) {
benchmarkstr[i] = rand() % 26 + 65;
}
benchmarkstr[BENCHMARK_STR_SIZE - 1] = '\n';
}
void BufferedLogStream::runDirectWriteBaseline() {
clock_t starttime = clock();
fp = fopen("BenchMarkBaseline.txt", "w+b");
for(unsigned int i = 0; i < NUM_BENCHMARK_WRITES; i++) {
fwrite(benchmarkstr, 1, BENCHMARK_STR_SIZE, fp);
}
fflush(fp);
fclose(fp);
clock_t elapsedtime = clock() - starttime;
cout << "Direct write baseline took " << ((double) elapsedtime) / CLOCKS_PER_SEC << " seconds." << endl;
}
void BufferedLogStream::runBufferedWriteBenchmark() {
clock_t starttime = clock();
openFile("BufferedBenchmark.txt");
cout << "Opend file" << endl;
for(unsigned int i = 0; i < NUM_BENCHMARK_WRITES; i++) {
writeMessage(benchmarkstr, BENCHMARK_STR_SIZE);
}
cout << "Wrote" << endl;
close();
cout << "Close" << endl;
clock_t elapsedtime = clock() - starttime;
cout << "Buffered write took " << ((double) elapsedtime) / CLOCKS_PER_SEC << " seconds." << endl;
}
void BufferedLogStream::runBenchmark() {
cout << "Buffer size is: " << BUFFER_SIZE << endl;
initBenchmarkString();
runDirectWriteBaseline();
runBufferedWriteBenchmark();
}
#endif
更新：2013年11月25日
我更新了下面的代码，使用了boost::condition_variables，具体来说是使用了wait_for()方法，正如Evgeny Panasyuk所建议的那样。这避免了不必要地反复检查相同的条件。目前，我看到缓冲版本的运行时间约为未缓冲 / 直接写入版本的1/6。这不是理想情况，因为两种情况都受硬盘限制（在我的情况下是2010年的SSD）。我计划在硬盘不会成为瓶颈的环境中使用下面的代码，大多数（如果不是全部）时间，缓冲区应该有可用空间来容纳writeMessage请求。这就带来了我的下一个问题。我应该把缓冲区设置多大？我不介意分配32 MB或64 MB以确保它永远不会填满。代码将在可以承受的系统上运行。直觉上，我觉得静态分配32 MB字符数组是个坏主意。是吗？无论如何，我预计当我为我的预期应用程序运行下面的代码时，logData()调用的延迟将大大降低，从而在总处理时间上产生显着的减少。
如果有人发现下面的代码有更好的方法（更快、更强大、更精简等），请告诉我。我感激反馈。Lazin，你的方法如何比我下面发布的方法更快或更有效？我有点喜欢只有一个缓冲区并使其足够大，以便它几乎永远不会填满的想法。然后我就不必担心从不同的缓冲区读取了。Evgeny Panasyuk，我喜欢在可能的情况下使用现有代码，特别是如果它是现有的boost库。然而，我也看不出spcs_queue比我下面的方法更有效。我宁愿处理一个大缓冲区而不是许多较小的缓冲区，并且必须担心在输入上将我的输入流分裂，并在输出上将其拼接回来。您的方法将允许我将格式化从主线程转移到工作线程。这是一种聪明的方法。但我还不确定它是否会节省很多时间，并且要实现全部效益，我必须修改我不拥有的代码。
boost::lockfree::spsc_queue- 这是一个无等待的单生产者单消费者队列。它可以配置为具有编译时容量（内部环形缓冲区的大小）。 - Evgeny Panasyuk