// Solution 3 // (note: compiles with gcc 3.2, VC6) #include "CriticalSection.h" #include "message_handler_log.h" #include <string> #include <queue> #include <vector> #include <map> #include <assert.h> #include <algorithm> // the object to be started - on a given thread struct win32_thread_obj { virtual ~win32_thread_obj() {} virtual void operator()() = 0; }; struct win32_thread_manager { typedef win32_thread_obj thread_obj_base; static void sleep( int nMillisecs) { Sleep( nMillisecs); } static void create_thread( win32_thread_obj & obj) { DWORD dwThreadID; CreateThread( 0, 0, win32_thread_manager::ThreadProc, &obj, 0, &dwThreadID); } private: static DWORD WINAPI ThreadProc( LPVOID lpData) { win32_thread_obj * pThread = ( win32_thread_obj *)lpData; ( *pThread)(); return 0; } }; // forward declaration template< class char_type, class traits_type = std::char_traits< char_type> > class basic_internal_thread_safe_log; // allows thread-safe writing for multiple logs template< class char_type, class traits_type = std::char_traits< char_type> > class basic_thread_safe_log_writer_sharethread { typedef basic_thread_safe_log_writer_sharethread< char_type, traits_type> this_class; typedef std::basic_ostream< char_type, traits_type> ostream_type; typedef std::basic_string< char_type, traits_type> string_type; friend class basic_internal_thread_safe_log< char_type, traits_type>; // copying not allowed basic_thread_safe_log_writer_sharethread( const this_class &); this_class & operator=( const this_class &); // forward declaration struct thread_info; friend struct thread_info; // thread-related definitions typedef win32_thread_manager thread_manager; typedef typename thread_manager::thread_obj_base thread_obj_base; // so that from our thread we know the object we're manipulating struct thread_info : public thread_obj_base { thread_info() : m_bHasFinished( false), m_pThis( NULL) {} /* virtual */ void operator()() { while ( true) { // ... we might be writing multiple messages at once! std::vector< std::string *> astrMsgs; ostream_type * pLog = NULL; { CAutoLockUnlock locker( m_pThis->m_cs); if ( m_pThis->m_nSumOfPriorities <= 0) { // we don't have any logs yet... thread_manager::sleep( 1); continue; } // find a log that has messages to be written to it for ( int idx = 0; idx < m_pThis->m_nSumOfPriorities; ++idx) { LogWrites & writes = *( m_pThis->m_aWritesTo[ m_pThis->m_idxWrite]); if ( writes.m_astr.size() > 0) // we found a log that we should write to break; ++m_pThis->m_idxWrite; m_pThis->m_idxWrite %= m_pThis->m_nSumOfPriorities; } // did we find a log with messages that should be written to it? LogWrites & writes = *( m_pThis->m_aWritesTo[ m_pThis->m_idxWrite]); if ( writes.m_astr.size() > 0) { // we get the string(s) to write to this log pLog = writes.m_pDestLog; // optimization - if too many messages, write // multiple messages at once int nMessages = 1; if ( writes.m_astr.size() > 100) nMessages = writes.m_astr.size() / 10; while ( nMessages > 0) { astrMsgs.push_back( writes.m_astr.front()); writes.m_astr.pop(); --nMessages; } } // ... only when there are no more messages, // will we ask if we should be destructed else if ( m_pThis->m_bShouldBeDestructed) { // signal to the other thread we've finished m_bHasFinished = true; return; } } // write the string(s) if ( astrMsgs.size() > 0) { std::vector< std::string *>::iterator first = astrMsgs.begin(), last = astrMsgs.end(); while ( first != last) { std::string *pstr = *first; *pLog << *pstr; delete pstr; ++first; } } else // nothing to write - wait thread_manager::sleep( 1); } } this_class * m_pThis; volatile bool m_bHasFinished; }; public: basic_thread_safe_log_writer_sharethread() : m_bShouldBeDestructed( false) { m_info.m_pThis = this; thread_manager::create_thread( m_info); m_nSumOfPriorities = 0; m_idxWrite = 0; } ~basic_thread_safe_log_writer_sharethread() { // signal to the other thread we're about to be // destructed { CAutoLockUnlock locker( m_cs); m_bShouldBeDestructed = true; } // wait while the other thread writes all messages while ( true) { CAutoLockUnlock locker( m_cs); if ( m_info.m_bHasFinished) // the other thread has finished break; } } private: // note: only basic_internal_thread_safe_log can // call these functions // adds a message to be written to a given log void add_message( const string_type & str, ostream_type & log) { CAutoLockUnlock locker( m_cs); ostream_type * pLog = &log; m_collLogWrites[ pLog].m_astr.push( new string_type( str)); } // adds a log we can write to, with a given priority void add_log( ostream_type & log, int nPriority) { // priority should be at least one assert( nPriority > 0); CAutoLockUnlock locker( m_cs); ostream_type * pLog = &log; m_collLogWrites[ pLog].m_nLogPriority = nPriority; m_collLogWrites[ pLog].m_pDestLog = pLog; m_nSumOfPriorities += nPriority; m_idxWrite = 0; m_aWritesTo.resize( m_nSumOfPriorities); std::fill( m_aWritesTo.begin(), m_aWritesTo.end(), ( LogWrites *)0); LogWritesCollection::iterator first = m_collLogWrites.begin(), last = m_collLogWrites.end(); while ( first != last) { LogWrites & writes = first->second; for( int idx = 0; idx < writes.m_nLogPriority; ++idx) { int idxWrite = (double)( idx * m_nSumOfPriorities) / writes.m_nLogPriority; // ... find an empty spot while ( m_aWritesTo[ idxWrite] != 0) { ++idxWrite; idxWrite = idxWrite % m_nSumOfPriorities; } m_aWritesTo[ idxWrite] = &writes; } ++first; } } CCriticalSection & cs() const { return m_cs; }private: // the critical section used for thread-safe locking mutable CCriticalSection m_cs; // needed to create the other thread thread_info m_info; volatile bool m_bShouldBeDestructed; typedef std::queue< string_type* > StringsQueue; struct LogWrites { LogWrites() : m_nLogPriority( 0), m_pDestLog( NULL) {} // the priority of this log int m_nLogPriority; // the strings to write to this log StringsQueue m_astr; // the log we should write to ostream_type * m_pDestLog; }; // at each step, from which log should we write to? std::vector< LogWrites* > m_aWritesTo; // for each log, what should we write to it? typedef std::map< ostream_type*, LogWrites> LogWritesCollection; LogWritesCollection m_collLogWrites; // the sum of all log' priorities int m_nSumOfPriorities; // the index of the current write // ( always less than m_nSumOfPriorities) int m_idxWrite; }; typedef basic_thread_safe_log_writer_sharethread< char> thread_safe_log_writer_sharethread; typedef basic_thread_safe_log_writer_sharethread< wchar_t> wthread_safe_log_writer_sharethread; // forward declaration template< class char_type, class traits_type = std::char_traits< char_type> > class basic_thread_safe_log; template< class char_type, class traits_type> class basic_internal_thread_safe_log { typedef basic_internal_thread_safe_log< char_type, traits_type> this_class; typedef typename std::basic_ostream< char_type, traits_type> ostream_type; friend class basic_thread_safe_log< char_type, traits_type>; typedef class basic_thread_safe_log_writer_sharethread< char_type, traits_type> multiple_log_writer; // non-copyiable basic_internal_thread_safe_log( const this_class &); this_class & operator=( this_class &); public: basic_internal_thread_safe_log( ostream_type & underlyingLog, multiple_log_writer & writer, int nPriority) : m_underlyingLog( underlyingLog), m_writer( writer) { writer.add_log( m_underlyingLog, nPriority); } ~basic_internal_thread_safe_log() {} void write_message( const std::basic_string< char_type, traits_type> & str) { m_writer.add_message( str, m_underlyingLog); } void copy_state_to( ostream_type & dest) const { CAutoLockUnlock locker( m_writer.cs()); dest.copyfmt( m_underlyingLog); dest.setstate( m_underlyingLog.rdstate()); } void copy_state_from( const ostream_type & src) { CAutoLockUnlock locker( m_writer.cs()); m_underlyingLog.copyfmt( src); m_underlyingLog.setstate( m_underlyingLog.rdstate()); } private: ostream_type & m_underlyingLog; // IMPORTANT: keep it by reference! multiple_log_writer & m_writer; }; typedef basic_internal_thread_safe_log< char> internal_thread_safe_log; typedef basic_internal_thread_safe_log< wchar_t> winternal_thread_safe_log; template< class char_type, class traits_type> class basic_thread_safe_log // *** protected, not public !!! : protected basic_message_handler_log< char_type, traits_type> { typedef std::basic_ostream< char_type, traits_type> ostream_type; typedef basic_internal_thread_safe_log< char_type, traits_type> internal_type; public: basic_thread_safe_log( internal_type & tsLog) : m_tsLog( tsLog) { // get underlying stream state tsLog.copy_state_to( ts() ); } basic_thread_safe_log( const basic_thread_safe_log< char_type, traits_type> & from) : m_tsLog( from.m_tsLog), // ... on some platforms, a std::ostream base copy-constructor // might be defined as private... basic_message_handler_log< char_type, traits_type>() { // get underlying stream state m_tsLog.copy_state_to( ts() ); } ~basic_thread_safe_log() { // copy state to underlying stream m_tsLog.copy_state_from( ts() ); } // get base class - to which we can write std::basic_ostream< char_type, traits_type> & ts() { return *this; } protected: virtual void on_new_message( const string_type & str) { m_tsLog.write_message( str); } private: internal_type & m_tsLog; }; typedef basic_thread_safe_log< char> thread_safe_log; typedef basic_thread_safe_log< wchar_t> wthread_safe_log; ////////////////////////////////////////////////////////// // Test #include <iostream> #include <fstream> #include <iomanip> const int THREADS_COUNT = 200; const int WRITES_PER_THREAD = 500; // the writer thread_safe_log_writer_sharethread & get_ts_writer() { static thread_safe_log_writer_sharethread writer; return writer; } template< int i> struct int_to_type { int_to_type() {} }; // return out<idx>.txt // (example: for 3, return 'out3.txt') template< int idxLog> std::string get_out_name( int_to_type< idxLog> * = NULL /* workaround for VC6 bug */) { std::ostringstream out; out << "out" << idxLog << ".txt"; return out.str(); } // we have 10 logs // log <idx> has priority (<idx>+1)^2 * 10 // (example: log 6 has priority 490) template< int idxLog> thread_safe_log templ_get_log( int_to_type< idxLog> * = NULL /* workaround for VC6 bug */) { static std::ofstream out( get_out_name< idxLog>().c_str() ); static internal_thread_safe_log log( out, get_ts_writer(), 10 * ( idxLog + 1) * ( idxLog + 1)); return thread_safe_log( log); } // based on the index, return a different log thread_safe_log get_log( int idxLog) { switch( idxLog) { case 0: return templ_get_log< 0>(); case 1: return templ_get_log< 1>(); case 2: return templ_get_log< 2>(); case 3: return templ_get_log< 3>(); case 4: return templ_get_log< 4>(); case 5: return templ_get_log< 5>(); case 6: return templ_get_log< 6>(); case 7: return templ_get_log< 7>(); case 8: return templ_get_log< 8>(); case 9: return templ_get_log< 9>(); default: assert( false); return templ_get_log< 0>(); } } LONG nRemainingThreads = THREADS_COUNT; DWORD WINAPI WriteToLog( LPVOID lpData) { int *pnThreadID = ( int *)lpData; int idxLog = *pnThreadID % 10; // wait for all threads to be created, so that // we write at about the same time (stress it ;-)) Sleep( 500); for ( int idx = 0; idx < WRITES_PER_THREAD; idx++) { get_log( idxLog).ts() << "writing double: " << 5.23 << std::endl; get_log( idxLog).ts() << "message " << idx << " from thread " << *pnThreadID << std::endl; // ... get other threads a chance to write Sleep( 1); if ( ( idx == 10) && ( *pnThreadID == 10)) { // from now on, '5.23' will be written as '5,23' // (german locale) std::locale loc = std::locale( "german"); get_log( idxLog).ts().imbue( loc); } } InterlockedDecrement( &nRemainingThreads); delete pnThreadID; return 0; } int main(int argc, char* argv[]) { // make sure the statics are initialized get_ts_writer(); get_log( 0); get_log( 1); get_log( 2); get_log( 3); get_log( 4); get_log( 5); get_log( 6); get_log( 7); get_log( 8); get_log( 9); for ( int idx = 0; idx < THREADS_COUNT; ++idx) { DWORD dwThreadID; CreateThread( 0, 0, WriteToLog, new int( idx), 0, &dwThreadID); } // wait for all threads to end while ( true) { InterlockedIncrement( &nRemainingThreads); if ( InterlockedDecrement( &nRemainingThreads) == 0) break; Sleep( 100); } return 0; } | 