Skip to content

Commit

Permalink
partial revert to XMutex implementation (note: avert chunity hang-on-…
Browse files Browse the repository at this point in the history 
…exit)
  • Loading branch information
@gewang
gewang committed May 16, 2024
1 parent 3e9b8f3 commit d7dffba
Show file tree
Hide file tree
Showing 5 changed files with 206 additions and 112 deletions.
31 changes: 23 additions & 8 deletions src/core/chuck_errmsg.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,16 @@
#include "util_platforms.h"
#include "util_string.h"

#ifndef __DISABLE_THREADS__
#include "util_thread.h"
#endif

#include <stdlib.h>
#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <sstream>
#include <iostream>
#include <mutex> // c++11
using namespace std;


Expand Down Expand Up @@ -72,7 +75,9 @@ static std::string g_error2str = "";
// log globals
t_CKINT g_loglevel = CK_LOG_CORE;
t_CKINT g_logstack = 0;
std::mutex g_logmutex;
#ifndef __DISABLE_THREADS__
XMutex g_logmutex;
#endif

// more local globals
std::stringstream g_stdout_stream;
Expand Down Expand Up @@ -877,9 +882,6 @@ void EM_print2magenta( const char * message, ... )
//-----------------------------------------------------------------------------
void EM_log( t_CKINT level, const char * message, ... )
{
// 1.5.2.5 (ge) added
std::lock_guard<std::mutex> lock(g_logmutex);

va_list ap;

if( level > CK_LOG_ALL ) level = CK_LOG_ALL;
Expand All @@ -888,6 +890,10 @@ void EM_log( t_CKINT level, const char * message, ... )
// check level
if( level > g_loglevel ) return;

#ifndef __DISABLE_THREADS__
g_logmutex.acquire();
#endif

TC::off();
CK_FPRINTF_STDERR( "[%s:%s:%s]: ",
TC::blue("chuck", true).c_str(),
Expand All @@ -905,6 +911,10 @@ void EM_log( t_CKINT level, const char * message, ... )

CK_FPRINTF_STDERR( "\n" );
CK_FFLUSH_STDERR();

#ifndef __DISABLE_THREADS__
g_logmutex.release();
#endif
}


Expand All @@ -916,9 +926,6 @@ void EM_log( t_CKINT level, const char * message, ... )
//-----------------------------------------------------------------------------
void EM_log_opts( t_CKINT level, enum em_LogOpts options, const char * message, ... )
{
// 1.5.2.5 (ge) added
std::lock_guard<std::mutex> lock(g_logmutex);

va_list ap;

if( level > CK_LOG_ALL ) level = CK_LOG_ALL;
Expand All @@ -932,6 +939,10 @@ void EM_log_opts( t_CKINT level, enum em_LogOpts options, const char * message,
// whether to print newline at end
t_CKBOOL nl = !(options & EM_LOG_NO_NEWLINE);

#ifndef __DISABLE_THREADS__
g_logmutex.acquire();
#endif

// check option
if( prefix )
{
Expand All @@ -956,6 +967,10 @@ void EM_log_opts( t_CKINT level, enum em_LogOpts options, const char * message,

// flush
CK_FFLUSH_STDERR();

#ifndef __DISABLE_THREADS__
g_logmutex.release();
#endif
}


Expand Down
136 changes: 91 additions & 45 deletions src/core/chuck_oo.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3103,19 +3103,19 @@ t_CKUINT Chuck_Event::our_waiting_on = 0;
//-----------------------------------------------------------------------------
void Chuck_Event::signal_local()
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

// if queue not empty
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
if( !m_queue.empty() )
{
// get the shred on top of the queue
Chuck_VM_Shred * shred = m_queue.front();
// pop the top
m_queue.pop();
// release lock
m_queue_lock.unlock();

// release it!
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
// REFACTOR-2017: BUG-FIX
// release the extra ref we added when we started waiting for this event
CK_SAFE_RELEASE( shred->event );
Expand All @@ -3129,6 +3129,12 @@ void Chuck_Event::signal_local()
t_CKTIME *& sp = (t_CKTIME *&)shred->reg->sp;
push_( sp, shreduler->now_system );
}
else
{
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
}
}


Expand All @@ -3140,13 +3146,13 @@ void Chuck_Event::signal_local()
//-----------------------------------------------------------------------------
t_CKBOOL Chuck_Event::remove( Chuck_VM_Shred * shred )
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

// queue of shred pointers
queue<Chuck_VM_Shred *> temp;
t_CKBOOL removed = FALSE;

// lock
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// while something in queue
while( !m_queue.empty() )
{
Expand Down Expand Up @@ -3175,6 +3181,10 @@ t_CKBOOL Chuck_Event::remove( Chuck_VM_Shred * shred )

// copy temp back to queue
m_queue = temp;
// release lock
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif

return removed;
}
Expand Down Expand Up @@ -3256,12 +3266,13 @@ void Chuck_Event::global_listen( t_CKINT id, void (* cb)(t_CKINT),
//-----------------------------------------------------------------------------
t_CKBOOL Chuck_Event::remove_listen( void (* cb)(void) )
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
// temp queue
std::queue<Chuck_Global_Event_Listener> temp;
t_CKBOOL removed = FALSE;

// lock
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// while something in queue
while( !m_global_queue.empty() )
{
Expand All @@ -3283,6 +3294,10 @@ t_CKBOOL Chuck_Event::remove_listen( void (* cb)(void) )

// copy temp back to queue
m_global_queue = temp;
// release lock
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif

return removed;
}
Expand All @@ -3296,12 +3311,13 @@ t_CKBOOL Chuck_Event::remove_listen( void (* cb)(void) )
//-----------------------------------------------------------------------------
t_CKBOOL Chuck_Event::remove_listen( std::string name, void (* cb)(const char *) )
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

std::queue<Chuck_Global_Event_Listener> temp;
t_CKBOOL removed = FALSE;

// lock
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// while something in queue
while( !m_global_queue.empty() )
{
Expand All @@ -3323,6 +3339,10 @@ t_CKBOOL Chuck_Event::remove_listen( std::string name, void (* cb)(const char *)

// copy temp back to queue
m_global_queue = temp;
// release lock
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif

return removed;
}
Expand All @@ -3336,12 +3356,13 @@ t_CKBOOL Chuck_Event::remove_listen( std::string name, void (* cb)(const char *)
//-----------------------------------------------------------------------------
t_CKBOOL Chuck_Event::remove_listen( t_CKINT id, void (* cb)(t_CKINT) )
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
// temp queue of listeners
std::queue<Chuck_Global_Event_Listener> temp;
t_CKBOOL removed = FALSE;

// lock
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// while something in queue
while( !m_global_queue.empty() )
{
Expand All @@ -3363,6 +3384,10 @@ t_CKBOOL Chuck_Event::remove_listen( t_CKINT id, void (* cb)(t_CKINT) )

// copy temp back to queue
m_global_queue = temp;
// release lock
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif

return removed;
}
Expand All @@ -3376,10 +3401,10 @@ t_CKBOOL Chuck_Event::remove_listen( t_CKINT id, void (* cb)(t_CKINT) )
//-----------------------------------------------------------------------------
void Chuck_Event::signal_global()
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif

// if global queue not empty
if( !m_global_queue.empty() )
{
// get the listener on top of the queue
Expand Down Expand Up @@ -3414,6 +3439,10 @@ void Chuck_Event::signal_global()
m_global_queue.push( listener );
}
}

#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
}


Expand All @@ -3425,12 +3454,11 @@ void Chuck_Event::signal_global()
//-----------------------------------------------------------------------------
void Chuck_Event::broadcast_global()
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
// queue of event listeners
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
std::queue< Chuck_Global_Event_Listener > call_again;

// check globals queue
while( !m_global_queue.empty() )
{
// get the listener on top of the queue
Expand Down Expand Up @@ -3468,6 +3496,10 @@ void Chuck_Event::broadcast_global()

// for those that should be called again, store them again
m_global_queue = call_again;

#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
}


Expand All @@ -3481,22 +3513,27 @@ void Chuck_Event::broadcast_global()
//-----------------------------------------------------------------------------
void Chuck_Event::queue_broadcast( CBufferSimple * event_buffer )
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
// TODO: handle multiple VM

// if not empty
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
if( !m_queue.empty() )
{
// get shred (only to get the VM ref)
Chuck_VM_Shred * shred = m_queue.front();

// release lock
m_queue_lock.unlock();

#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
// queue the event on the vm (added 1.3.0.0: event_buffer)
shred->vm_ref->queue_event( this, 1, event_buffer );
}
else
{
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
}
}


Expand All @@ -3509,21 +3546,28 @@ void Chuck_Event::queue_broadcast( CBufferSimple * event_buffer )
//-----------------------------------------------------------------------------
void Chuck_Event::broadcast_local()
{
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

// lock queue
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// while not empty
while( !m_queue.empty() )
{
// release first
m_queue_lock.unlock();

#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
// signal the next shred
this->signal_local();

// lock again
m_queue_lock.lock();
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
}
// release
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
}


Expand All @@ -3548,12 +3592,14 @@ void Chuck_Event::wait( Chuck_VM_Shred * shred, Chuck_VM * vm )
// suspend
shred->is_running = FALSE;

// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
// add to waiting list
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
m_queue.push( shred );
// release
m_queue_lock.unlock();
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif

// add event to shred
assert( shred->event == NULL );
Expand Down
Loading

0 comments on commit d7dffba

Please sign in to comment.