Skip to content

Commit

Permalink
update event mutex
Browse files Browse the repository at this point in the history
  • Loading branch information
@gewang
gewang committed May 14, 2024
1 parent b8bcf5f commit 146a8e9
Show file tree
Hide file tree
Showing 4 changed files with 65 additions and 129 deletions.
31 changes: 8 additions & 23 deletions src/core/chuck_errmsg.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,16 +37,13 @@
#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 @@ -75,9 +72,7 @@ static std::string g_error2str = "";
// log globals
t_CKINT g_loglevel = CK_LOG_CORE;
t_CKINT g_logstack = 0;
#ifndef __DISABLE_THREADS__
XMutex g_logmutex;
#endif
std::mutex g_logmutex;

// more local globals
std::stringstream g_stdout_stream;
Expand Down Expand Up @@ -882,6 +877,9 @@ 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 @@ -890,10 +888,6 @@ 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 @@ -911,10 +905,6 @@ 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 @@ -926,6 +916,9 @@ 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 @@ -939,10 +932,6 @@ 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 @@ -967,10 +956,6 @@ 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: 45 additions & 91 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()
{
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

// if queue not empty
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 it!
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
// release lock
m_queue_lock.unlock();

// 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,12 +3129,6 @@ 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 @@ -3146,13 +3140,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 @@ -3181,10 +3175,6 @@ 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 @@ -3266,13 +3256,12 @@ 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 @@ -3294,10 +3283,6 @@ 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 @@ -3311,13 +3296,12 @@ 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 @@ -3339,10 +3323,6 @@ 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 @@ -3356,13 +3336,12 @@ 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 @@ -3384,10 +3363,6 @@ 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 @@ -3401,10 +3376,10 @@ t_CKBOOL Chuck_Event::remove_listen( t_CKINT id, void (* cb)(t_CKINT) )
//-----------------------------------------------------------------------------
void Chuck_Event::signal_global()
{
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

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

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


Expand All @@ -3454,11 +3425,12 @@ void Chuck_Event::signal_global()
//-----------------------------------------------------------------------------
void Chuck_Event::broadcast_global()
{
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);
// queue of event listeners
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 @@ -3496,10 +3468,6 @@ 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 @@ -3513,27 +3481,22 @@ 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
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif

// if not empty
if( !m_queue.empty() )
{
// get shred (only to get the VM ref)
Chuck_VM_Shred * shred = m_queue.front();

// release lock
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
m_queue_lock.unlock();

// 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 @@ -3546,28 +3509,21 @@ void Chuck_Event::queue_broadcast( CBufferSimple * event_buffer )
//-----------------------------------------------------------------------------
void Chuck_Event::broadcast_local()
{
// lock queue
#ifndef __DISABLE_THREADS__
m_queue_lock.acquire();
#endif
// 1.5.2.5 (ge) updated to std::mutex
std::lock_guard<std::mutex> lock(m_queue_lock);

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

// signal the next shred
this->signal_local();

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


Expand All @@ -3592,14 +3548,12 @@ 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 );
#ifndef __DISABLE_THREADS__
m_queue_lock.release();
#endif
// release
m_queue_lock.unlock();

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

0 comments on commit 146a8e9

Please sign in to comment.