forked from linyacool/WebServer
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathAsyncLogging.cpp
More file actions
executable file
·126 lines (113 loc) · 3.57 KB
/
Copy pathAsyncLogging.cpp
File metadata and controls
executable file
·126 lines (113 loc) · 3.57 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
// @Author Lin Ya
// @Email [email protected]
#include "AsyncLogging.h"
#include "LogFile.h"
#include <stdio.h>
#include <assert.h>
#include <unistd.h>
#include <functional>
AsyncLogging::AsyncLogging(std::string logFileName_,int flushInterval)
: flushInterval_(flushInterval),
running_(false),
basename_(logFileName_),
thread_(std::bind(&AsyncLogging::threadFunc, this), "Logging"),
mutex_(),
cond_(mutex_),
currentBuffer_(new Buffer),
nextBuffer_(new Buffer),
buffers_(),
latch_(1)
{
assert(logFileName_.size() > 1);
currentBuffer_->bzero();
nextBuffer_->bzero();
buffers_.reserve(16);
}
void AsyncLogging::append(const char* logline, int len)
{
MutexLockGuard lock(mutex_);
if (currentBuffer_->avail() > len)
currentBuffer_->append(logline, len);
else
{
buffers_.push_back(currentBuffer_);
currentBuffer_.reset();
if (nextBuffer_)
currentBuffer_ = std::move(nextBuffer_);
else
currentBuffer_.reset(new Buffer);
currentBuffer_->append(logline, len);
cond_.notify();
}
}
void AsyncLogging::threadFunc()
{
assert(running_ == true);
latch_.countDown();
LogFile output(basename_);
BufferPtr newBuffer1(new Buffer);
BufferPtr newBuffer2(new Buffer);
newBuffer1->bzero();
newBuffer2->bzero();
BufferVector buffersToWrite;
buffersToWrite.reserve(16);
while (running_)
{
assert(newBuffer1 && newBuffer1->length() == 0);
assert(newBuffer2 && newBuffer2->length() == 0);
assert(buffersToWrite.empty());
{
MutexLockGuard lock(mutex_);
if (buffers_.empty()) // unusual usage!
{
cond_.waitForSeconds(flushInterval_);
}
buffers_.push_back(currentBuffer_);
currentBuffer_.reset();
currentBuffer_ = std::move(newBuffer1);
buffersToWrite.swap(buffers_);
if (!nextBuffer_)
{
nextBuffer_ = std::move(newBuffer2);
}
}
assert(!buffersToWrite.empty());
if (buffersToWrite.size() > 25)
{
//char buf[256];
// snprintf(buf, sizeof buf, "Dropped log messages at %s, %zd larger buffers\n",
// Timestamp::now().toFormattedString().c_str(),
// buffersToWrite.size()-2);
//fputs(buf, stderr);
//output.append(buf, static_cast<int>(strlen(buf)));
buffersToWrite.erase(buffersToWrite.begin()+2, buffersToWrite.end());
}
for (size_t i = 0; i < buffersToWrite.size(); ++i)
{
// FIXME: use unbuffered stdio FILE ? or use ::writev ?
output.append(buffersToWrite[i]->data(), buffersToWrite[i]->length());
}
if (buffersToWrite.size() > 2)
{
// drop non-bzero-ed buffers, avoid trashing
buffersToWrite.resize(2);
}
if (!newBuffer1)
{
assert(!buffersToWrite.empty());
newBuffer1 = buffersToWrite.back();
buffersToWrite.pop_back();
newBuffer1->reset();
}
if (!newBuffer2)
{
assert(!buffersToWrite.empty());
newBuffer2 = buffersToWrite.back();
buffersToWrite.pop_back();
newBuffer2->reset();
}
buffersToWrite.clear();
output.flush();
}
output.flush();
}