Line data Source code
1 : /*
2 : * DUNE DAQ modification notice:
3 : * This file has been modified from the original ATLAS ers source for the DUNE DAQ project.
4 : * Fork baseline commit: 8267df82a4f6fe6bf02c4014923eba19eddc4614 (2020-04-14).
5 : * Renamed since fork: yes (from src/StreamManager.cxx to src/StreamManager.cpp).
6 : *
7 : * Original copyright:
8 : * Copyright (C) 2001-2020 CERN for the benefit of the ATLAS collaboration.
9 : * Licensed under the Apache License, Version 2.0.
10 : */
11 :
12 : /*
13 : * StreamManager.cxx
14 : * ERS
15 : *
16 : * Created by Matthias Wiesmann on 21.01.05.
17 : * Modified by Serguei Kolos on 12.09.06.
18 : * Copyright 2005 CERN. All rights reserved.
19 : *
20 : */
21 :
22 : #include <assert.h>
23 : #include <iostream>
24 :
25 : #include <ers/Configuration.hpp>
26 : #include <ers/InputStream.hpp>
27 : #include <ers/Issue.hpp>
28 : #include <ers/OutputStream.hpp>
29 : #include <ers/Severity.hpp>
30 : #include <ers/StreamFactory.hpp>
31 : #include <ers/StreamManager.hpp>
32 : #include <ers/ers.hpp>
33 : #include <ers/internal/NullStream.hpp>
34 : #include <ers/internal/PluginManager.hpp>
35 : #include <ers/internal/SingletonCreator.hpp>
36 : #include <ers/internal/Util.hpp>
37 : #include <ers/internal/macro.hpp>
38 :
39 78 : ERS_DECLARE_ISSUE(ers,
40 : BadConfiguration,
41 : "The stream configuration string \"" << config << "\" has syntax errors.",
42 : ((std::string)config))
43 :
44 : namespace {
45 : /** This variable contains the default keys for building the default streams.
46 : * The default is to use the default stream, in verbose mode for errors and fatals.
47 : */
48 : const char SEPARATOR = ',';
49 :
50 : const char* const DefaultOutputStreams[] = {
51 : "lstdout", // Debug
52 : "lstdout", // Log
53 : "throttle,lstdout", // Information
54 : "throttle,lstderr", // Warning
55 : "throttle,lstderr", // Error
56 : "lstderr" // Fatal
57 : };
58 :
59 : const char*
60 59 : get_stream_description(ers::severity severity)
61 : {
62 59 : assert(ers::Debug <= severity && severity <= ers::Fatal);
63 :
64 59 : std::string env_name("DUNEDAQ_ERS_");
65 59 : env_name += ers::to_string(severity);
66 59 : const char* env = ::getenv(env_name.c_str());
67 118 : return env ? env : DefaultOutputStreams[severity];
68 59 : }
69 :
70 : void
71 59 : parse_stream_definition(const std::string& text, std::vector<std::string>& result)
72 : {
73 59 : std::string::size_type start_p = 0, end_p = 0;
74 59 : short brackets_open = 0;
75 679 : while (end_p < text.length()) {
76 620 : switch (text[end_p]) {
77 0 : case '(':
78 0 : ++brackets_open;
79 0 : break;
80 0 : case ')':
81 0 : --brackets_open;
82 0 : break;
83 23 : case SEPARATOR:
84 23 : if (!brackets_open) {
85 23 : result.push_back(text.substr(start_p, end_p - start_p));
86 23 : start_p = end_p + 1;
87 : }
88 : break;
89 : default:
90 : break;
91 : }
92 620 : end_p++;
93 : }
94 59 : if (brackets_open) {
95 0 : throw ers::BadConfiguration(ERS_HERE, text);
96 : }
97 59 : if (start_p != end_p) {
98 59 : result.push_back(text.substr(start_p, end_p - start_p));
99 : }
100 59 : }
101 : }
102 :
103 : namespace ers {
104 : // Performs lazy srteam initialization. Stream instances are created
105 : // at the first attempt of writing to the stream
106 : class StreamInitializer : public ers::OutputStream
107 : {
108 : public:
109 258 : StreamInitializer(StreamManager& manager)
110 516 : : m_manager(manager)
111 258 : , m_in_progress(false)
112 : {
113 258 : ;
114 258 : }
115 :
116 59 : void write(const Issue& issue)
117 : {
118 59 : ers::severity s = issue.severity();
119 59 : std::scoped_lock lock(m_mutex);
120 :
121 59 : if (!m_in_progress) {
122 59 : m_in_progress = true;
123 : } else {
124 : // The issue is coming from the stream constructor
125 : // We can't use ERS streams, so print it to std
126 0 : if (s < ers::Warning)
127 0 : std::cout << issue << std::endl;
128 : else
129 0 : std::cerr << issue << std::endl;
130 0 : return;
131 : }
132 :
133 59 : if (m_manager.m_out_streams[s].get() == this) {
134 59 : m_manager.m_out_streams[s] = std::shared_ptr<OutputStream>(m_manager.setup_stream(s));
135 : }
136 59 : m_manager.report_issue(s, issue);
137 59 : m_in_progress = false;
138 59 : }
139 :
140 : private:
141 : std::recursive_mutex m_mutex;
142 : StreamManager& m_manager;
143 : bool m_in_progress;
144 : };
145 :
146 : }
147 :
148 : /** This method returns the singleton instance.
149 : * It should be used for every operation on the factory.
150 : * \return a reference to the singleton instance
151 : */
152 : ers::StreamManager&
153 1083 : ers::StreamManager::instance()
154 : {
155 : /**Singleton instance
156 : */
157 1083 : static ers::StreamManager* instance = ers::SingletonCreator<ers::StreamManager>::create();
158 :
159 1083 : return *instance;
160 : } // instance
161 :
162 : /** Private constructor - can not be called by user code, use the \c instance() method instead
163 : * \see instance()
164 : */
165 43 : ers::StreamManager::StreamManager()
166 : {
167 301 : for (short ss = ers::Debug; ss <= ers::Fatal; ++ss) {
168 258 : m_init_streams[ss] = std::make_shared<StreamInitializer>(*this);
169 258 : m_out_streams[ss] = m_init_streams[ss];
170 : }
171 43 : }
172 :
173 : /** Destructor - basic cleanup
174 : */
175 0 : ers::StreamManager::~StreamManager()
176 : {
177 0 : ;
178 0 : }
179 :
180 : void
181 0 : ers::StreamManager::add_output_stream(ers::severity severity, ers::OutputStream* new_stream)
182 : {
183 0 : std::shared_ptr<OutputStream> head = m_out_streams[severity];
184 0 : if (head && !head->isNull()) {
185 0 : OutputStream* parent = head.get();
186 0 : for (OutputStream* stream = parent; !stream->isNull(); parent = stream, stream = &parent->chained())
187 : ;
188 :
189 0 : parent->chained(new_stream);
190 : } else {
191 0 : m_out_streams[severity] = std::shared_ptr<OutputStream>(new_stream);
192 : }
193 0 : }
194 :
195 : void
196 0 : ers::StreamManager::add_receiver(const std::string& stream, const std::string& filter, ers::IssueReceiver* receiver)
197 : {
198 0 : InputStream* in = ers::StreamFactory::instance().create_in_stream(stream, filter);
199 0 : in->set_receiver(receiver);
200 :
201 0 : std::scoped_lock lock(m_mutex);
202 0 : m_in_streams.push_back(std::shared_ptr<InputStream>(in));
203 0 : }
204 :
205 : void
206 0 : ers::StreamManager::add_receiver(const std::string& stream,
207 : const std::initializer_list<std::string>& params,
208 : ers::IssueReceiver* receiver)
209 : {
210 0 : InputStream* in = ers::StreamFactory::instance().create_in_stream(stream, params);
211 0 : in->set_receiver(receiver);
212 :
213 0 : std::scoped_lock lock(m_mutex);
214 0 : m_in_streams.push_back(std::shared_ptr<InputStream>(in));
215 0 : }
216 :
217 : void
218 0 : ers::StreamManager::remove_receiver(ers::IssueReceiver* receiver)
219 : {
220 0 : std::scoped_lock lock(m_mutex);
221 0 : for (std::list<std::shared_ptr<InputStream>>::iterator it = m_in_streams.begin(); it != m_in_streams.end();) {
222 0 : if ((*it)->m_receiver == receiver)
223 0 : m_in_streams.erase(it++);
224 : else
225 0 : ++it;
226 : }
227 0 : }
228 :
229 : ers::OutputStream*
230 59 : ers::StreamManager::setup_stream(ers::severity severity)
231 : {
232 59 : std::string config = get_stream_description(severity);
233 59 : std::vector<std::string> streams;
234 59 : try {
235 59 : parse_stream_definition(config, streams);
236 0 : } catch (ers::BadConfiguration& ex) {
237 0 : ERS_INTERNAL_ERROR("Configuration for the \"" << severity
238 : << "\" stream is invalid. "
239 0 : "Default configuration will be used.");
240 0 : }
241 :
242 59 : ers::OutputStream* main = setup_stream(streams);
243 :
244 59 : if (!main) {
245 0 : std::vector<std::string> default_streams;
246 0 : try {
247 0 : parse_stream_definition(DefaultOutputStreams[severity], default_streams);
248 0 : main = setup_stream(default_streams);
249 0 : } catch (ers::BadConfiguration& ex) {
250 0 : ERS_INTERNAL_ERROR("Can not configure the \"" << severity << "\" stream because of the following issue {" << ex
251 0 : << "}");
252 0 : }
253 0 : }
254 59 : return (main ? main : new ers::NullStream());
255 59 : }
256 :
257 : ers::OutputStream*
258 59 : ers::StreamManager::setup_stream(const std::vector<std::string>& streams)
259 : {
260 59 : size_t cnt = 0;
261 59 : ers::OutputStream* main = 0;
262 59 : for (; cnt < streams.size(); ++cnt) {
263 59 : main = ers::StreamFactory::instance().create_out_stream(streams[cnt]);
264 59 : if (main)
265 : break;
266 : }
267 :
268 59 : if (!main) {
269 : return 0;
270 : }
271 :
272 59 : ers::OutputStream* head = main;
273 82 : for (++cnt; cnt < streams.size(); ++cnt) {
274 23 : ers::OutputStream* chained = ers::StreamFactory::instance().create_out_stream(streams[cnt]);
275 :
276 23 : if (chained) {
277 23 : head->chained(chained);
278 23 : head = chained;
279 : }
280 : }
281 :
282 : return main;
283 : }
284 :
285 : /** Sends an Issue to an appropriate stream
286 : * \param type
287 : * \param issue
288 : */
289 : void
290 1141 : ers::StreamManager::report_issue(ers::severity type, const Issue& issue)
291 : {
292 1141 : ers::severity old_severity = issue.set_severity(type);
293 1141 : m_out_streams[type]->write(issue);
294 1141 : issue.set_severity(old_severity);
295 1141 : } // error
296 :
297 : /** Sends an Issue to the error stream
298 : * \param issue
299 : */
300 : void
301 0 : ers::StreamManager::error(const Issue& issue)
302 : {
303 0 : report_issue(ers::Error, issue);
304 0 : } // error
305 :
306 : /** Sends an issue to the debug stream
307 : * \param issue the Issue to send
308 : * \param level the debug level.
309 : */
310 : void
311 1 : ers::StreamManager::debug(const Issue& issue, int level)
312 : {
313 1 : if (Configuration::instance().debug_level() >= level) {
314 0 : ers::severity old_severity = issue.set_severity(ers::Severity(ers::Debug, level));
315 0 : m_out_streams[ers::Debug]->write(issue);
316 0 : issue.set_severity(old_severity);
317 : }
318 1 : }
319 :
320 : /** Sends an Issue to the fatal error stream
321 : * \param issue
322 : */
323 : void
324 0 : ers::StreamManager::fatal(const Issue& issue)
325 : {
326 0 : report_issue(ers::Fatal, issue);
327 0 : }
328 :
329 : /** Sends an Issue to the warning stream
330 : * \param issue the issue to send
331 : */
332 : void
333 0 : ers::StreamManager::warning(const Issue& issue)
334 : {
335 0 : report_issue(ers::Warning, issue);
336 0 : }
337 :
338 : /** Sends an issue to the info stream
339 : * \param issue the Issue to send
340 : */
341 : void
342 19 : ers::StreamManager::information(const Issue& issue)
343 : {
344 19 : report_issue(ers::Information, issue);
345 19 : }
346 :
347 : /** Sends an issue to the log stream
348 : * \param issue the Issue to send
349 : */
350 : void
351 1022 : ers::StreamManager::log(const Issue& issue)
352 : {
353 1022 : report_issue(ers::Log, issue);
354 1022 : }
355 :
356 : std::ostream&
357 0 : ers::operator<<(std::ostream& out, const ers::StreamManager&)
358 : {
359 0 : for (short ss = ers::Debug; ss <= ers::Fatal; ++ss) {
360 0 : out << (ers::severity)ss << "\t\"" << get_stream_description((ers::severity)ss) << "\"" << std::endl;
361 : }
362 0 : return out;
363 : }
|
