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