Skip to content

Commit 66dd175

Browse files
flangletflanglet
committed
Rewrite event logging code
1 parent 1eb4d8c commit 66dd175

2 files changed

Lines changed: 91 additions & 56 deletions

File tree

‎src/app/InfoPrinter.cpp‎

Lines changed: 86 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -30,17 +30,15 @@ InfoPrinter::InfoPrinter(int infoLevel, InfoPrinter::Type type, OutputStream& os
3030
, _level(infoLevel)
3131
, _headerInfo(0)
3232
{
33-
STORE_ATOMIC(_lastEmittedBlockId, max(firstBlockId - 1, 0));
33+
STORE_ATOMIC(_nextBlockId, firstBlockId);
3434

35-
// Select the ONLY ordered phase
3635
if (type == InfoPrinter::COMPRESSION) {
3736
_thresholds[0] = Event::COMPRESSION_START;
3837
_thresholds[1] = Event::BEFORE_TRANSFORM;
3938
_thresholds[2] = Event::AFTER_TRANSFORM;
4039
_thresholds[3] = Event::BEFORE_ENTROPY;
4140
_thresholds[4] = Event::AFTER_ENTROPY;
4241
_thresholds[5] = Event::COMPRESSION_END;
43-
_orderedPhase = Event::AFTER_ENTROPY;
4442
}
4543
else {
4644
_thresholds[0] = Event::DECOMPRESSION_START;
@@ -49,7 +47,6 @@ InfoPrinter::InfoPrinter(int infoLevel, InfoPrinter::Type type, OutputStream& os
4947
_thresholds[3] = Event::BEFORE_TRANSFORM;
5048
_thresholds[4] = Event::AFTER_TRANSFORM;
5149
_thresholds[5] = Event::DECOMPRESSION_END;
52-
_orderedPhase = Event::BEFORE_ENTROPY;
5350
}
5451
}
5552

@@ -62,8 +59,11 @@ void InfoPrinter::processEvent(const Event& evt)
6259
}
6360

6461
#ifdef CONCURRENCY_ENABLED
65-
if (evt.getType() == _orderedPhase) {
66-
processOrderedPhase(evt);
62+
Event::Type t = evt.getType();
63+
64+
if (t == Event::BEFORE_TRANSFORM || t == Event::AFTER_TRANSFORM ||
65+
t == Event::BEFORE_ENTROPY || t == Event::AFTER_ENTROPY) {
66+
processBlockEventOrdered(evt);
6767
return;
6868
}
6969
#endif
@@ -73,34 +73,79 @@ void InfoPrinter::processEvent(const Event& evt)
7373

7474

7575
#ifdef CONCURRENCY_ENABLED
76-
void InfoPrinter::processOrderedPhase(const Event& evt)
76+
void InfoPrinter::processBlockEventOrdered(const Event& evt)
7777
{
78+
const int blockId = evt.getId();
79+
const Event::Type type = evt.getType();
80+
bool blockComplete = false;
81+
82+
// Determine completion condition
83+
if (_type == InfoPrinter::COMPRESSION) {
84+
blockComplete = (type == Event::AFTER_ENTROPY);
85+
}
86+
else if (_type == InfoPrinter::DECOMPRESSION) {
87+
blockComplete = (type == Event::AFTER_TRANSFORM);
88+
}
89+
7890
{
79-
std::lock_guard<std::mutex> lock(_mutex);
80-
_orderedPending.insert(std::make_pair(evt.getId(), evt));
91+
#ifdef CONCURRENCY_ENABLED
92+
std::lock_guard<std::mutex> lock(_mutex1);
93+
#endif
94+
_pendingBlocks[blockId].push_back(evt);
95+
96+
// Do not attempt to release unless this block is complete
97+
if (blockComplete == false)
98+
return;
8199
}
82100

101+
// Try to release completed blocks in strict blockId order
83102
for (;;)
84103
{
85-
WallTimer timer;
86-
Event next(Event::BLOCK_INFO, 0, "", timer.getCurrentTime());
104+
std::vector<Event> events;
87105

88106
{
89-
std::lock_guard<std::mutex> lock(_mutex);
90-
int nextId = LOAD_ATOMIC(_lastEmittedBlockId) + 1;
91-
std::map<int, Event>::iterator it = _orderedPending.find(nextId);
107+
#ifdef CONCURRENCY_ENABLED
108+
std::lock_guard<std::mutex> lock(_mutex1);
109+
#endif
110+
int expectedId = LOAD_ATOMIC(_nextBlockId);
111+
std::map<int, std::vector<Event> >::iterator it = _pendingBlocks.find(expectedId);
92112

93-
if (it == _orderedPending.end())
113+
if (it == _pendingBlocks.end())
94114
return;
95115

96-
next = it->second;
97-
_orderedPending.erase(it);
98-
STORE_ATOMIC(_lastEmittedBlockId, nextId);
116+
// The block must be complete before release
117+
bool complete = false;
118+
const std::vector<Event>& evts = it->second;
119+
120+
for (size_t i = 0; i < evts.size(); i++) {
121+
if ((_type == InfoPrinter::COMPRESSION) && (evts[i].getType() == Event::AFTER_ENTROPY)) {
122+
complete = true;
123+
break;
124+
}
125+
126+
if ((_type == InfoPrinter::DECOMPRESSION) && (evts[i].getType() == Event::AFTER_TRANSFORM)) {
127+
complete = true;
128+
break;
129+
}
130+
}
131+
132+
if (complete == false)
133+
return;
134+
135+
// Release block
136+
events.swap(it->second);
137+
_pendingBlocks.erase(it);
138+
STORE_ATOMIC(_nextBlockId, expectedId + 1);
99139
}
100140

101-
// Compression: AFTER_ENTROPY emitted in-order
102-
// Decompression: BEFORE_TRANSFORM emitted in-order
103-
processEventOrdered(next);
141+
// Process all events for this block in arrival order
142+
#ifdef CONCURRENCY_ENABLED
143+
std::lock_guard<std::mutex> lock(_mutex2);
144+
#endif
145+
146+
for (size_t i = 0; i < events.size(); i++) {
147+
processEventOrdered(events[i]);
148+
}
104149
}
105150
}
106151
#endif
@@ -110,21 +155,19 @@ void InfoPrinter::processEventOrdered(const Event& evt)
110155
{
111156
const int blockId = evt.getId();
112157
const Event::Type type = evt.getType();
158+
string msg;
113159

114-
if (type == _thresholds[1])
115-
{
160+
if (type == _thresholds[1]) {
116161
BlockInfo* bi = new BlockInfo();
117162
bi->_timeStamp1 = evt.getTime();
118163
bi->_stage0Size = evt.getSize();
119164
_blocks[blockId] = bi;
120165

121-
if (_level >= 5)
122-
_os << evt.toString() << std::endl;
123-
124-
_os.flush();
166+
if (_level >= 5) {
167+
msg = evt.toString();
168+
}
125169
}
126-
else if (type == _thresholds[2])
127-
{
170+
else if (type == _thresholds[2]) {
128171
std::map<int, BlockInfo*>::iterator it = _blocks.find(blockId);
129172

130173
if (it == _blocks.end())
@@ -137,13 +180,10 @@ void InfoPrinter::processEventOrdered(const Event& evt)
137180
double elapsed = WallTimer::calculateDifference(bi._timeStamp1, bi._timeStamp2);
138181
std::stringstream ss;
139182
ss << evt.toString() << " [" << int64(elapsed) << " ms]";
140-
_os << ss.str() << std::endl;
183+
msg = ss.str();
141184
}
142-
143-
_os.flush();
144185
}
145-
else if (type == _thresholds[3])
146-
{
186+
else if (type == _thresholds[3]) {
147187
std::map<int, BlockInfo*>::iterator it = _blocks.find(blockId);
148188

149189
if (it == _blocks.end())
@@ -154,12 +194,9 @@ void InfoPrinter::processEventOrdered(const Event& evt)
154194
bi._stage1Size = evt.getSize();
155195

156196
if (_level >= 5)
157-
_os << evt.toString() << std::endl;
158-
159-
_os.flush();
197+
msg = evt.toString();
160198
}
161-
else if (type == _thresholds[4])
162-
{
199+
else if (type == _thresholds[4]) {
163200
std::map<int, BlockInfo*>::iterator it = _blocks.find(blockId);
164201

165202
if (it == _blocks.end())
@@ -194,28 +231,25 @@ void InfoPrinter::processEventOrdered(const Event& evt)
194231
ss << std::uppercase << std::hex << " [" << evt.getHash() << "]";
195232
}
196233

197-
_os << ss.str() << std::endl;
234+
msg = ss.str();
198235
}
199236

200237
delete it->second;
201238
_blocks.erase(it);
202-
203-
_os.flush();
204239
}
205240
else if ((evt.getType() == Event::AFTER_HEADER_DECODING) && (_level >= 3)) {
206241
Event::HeaderInfo* info = evt.getInfo();
207242

208243
if (info == nullptr)
209244
return;
210245

211-
stringstream ss;
212-
213246
if (_level >= 5) {
214247
// JSON output
215-
ss << evt.toString();
248+
msg = evt.toString();
216249
}
217250
else {
218251
// Raw text output
252+
stringstream ss;
219253
ss << "Bitstream version: " << info->bsVersion << endl;
220254
string strCk = "NONE";
221255

@@ -233,13 +267,16 @@ void InfoPrinter::processEventOrdered(const Event& evt)
233267

234268
if (info->originalSize >= 0)
235269
ss << "Original size: " << info->originalSize << " byte(s)" << endl;
236-
}
237270

238-
_os << ss.str() << endl;
239-
_os.flush();
271+
msg = ss.str();
272+
}
240273
}
241274
else if (_level >= 5) {
242-
_os << evt.toString() << endl;
275+
msg = evt.toString();
276+
}
277+
278+
if (msg.size() > 0) {
279+
_os << msg << endl;
243280
_os.flush();
244281
}
245282
}

‎src/app/InfoPrinter.hpp‎

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ namespace kanzi
5757
private:
5858
#ifdef CONCURRENCY_ENABLED
5959
// Ordered-phase handling
60-
void processOrderedPhase(const Event& evt);
60+
void processBlockEventOrdered(const Event& evt);
6161
#endif
6262

6363
// Actual event processing + printing
@@ -74,15 +74,13 @@ namespace kanzi
7474
// Per-block state
7575
std::map<int, BlockInfo*> _blocks;
7676

77-
// Ordered-phase state
78-
Event::Type _orderedPhase;
79-
8077
Event::Type _thresholds[6];
8178
#ifdef CONCURRENCY_ENABLED
82-
std::mutex _mutex;
79+
std::mutex _mutex1;
80+
std::mutex _mutex2;
8381
#endif
84-
std::map<int, Event> _orderedPending;
85-
atomic_int_t _lastEmittedBlockId;
82+
std::map<int, std::vector<Event> > _pendingBlocks;
83+
atomic_int_t _nextBlockId;
8684
};
8785

8886
}

0 commit comments

Comments
 (0)