mirror of
https://github.com/zeek/zeek.git
synced 2025-10-10 02:28:21 +00:00
Further threading and API restructuring for logging and input
frameworks. There were a number of cases that weren't thread-safe. In particular, we don't use std::string anymore for anything that's passed between threads (but instead plain old const char*, with manual memmory managmenet). This is still a check-point commit, I'll do more testing.
This commit is contained in:
Robin Sommer
parent
490859cfef
commit
87e10b5f97
31 changed files with 692 additions and 381 deletions
|
|
@ -12,18 +12,23 @@
|
|||
|
||||
using namespace threading;
|
||||
|
||||
static const int STD_FMT_BUF_LEN = 2048;
|
||||
|
||||
uint64_t BasicThread::thread_counter = 0;
|
||||
|
||||
BasicThread::BasicThread()
|
||||
{
|
||||
started = false;
|
||||
terminating = false;
|
||||
killed = false;
|
||||
pthread = 0;
|
||||
|
||||
buf_len = 2048;
|
||||
buf_len = STD_FMT_BUF_LEN;
|
||||
buf = (char*) malloc(buf_len);
|
||||
|
||||
name = Fmt("thread-%d", ++thread_counter);
|
||||
strerr_buffer = 0;
|
||||
|
||||
name = copy_string(fmt("thread-%" PRIu64, ++thread_counter));
|
||||
|
||||
thread_mgr->AddThread(this);
|
||||
}
|
||||
|
|
@ -32,31 +37,41 @@ BasicThread::~BasicThread()
|
|||
{
|
||||
if ( buf )
|
||||
free(buf);
|
||||
|
||||
delete [] name;
|
||||
delete [] strerr_buffer;
|
||||
}
|
||||
|
||||
void BasicThread::SetName(const string& arg_name)
|
||||
void BasicThread::SetName(const char* name)
|
||||
{
|
||||
// Slight race condition here with reader threads, but shouldn't matter.
|
||||
name = arg_name;
|
||||
delete [] name;
|
||||
name = copy_string(name);
|
||||
}
|
||||
|
||||
void BasicThread::SetOSName(const string& name)
|
||||
void BasicThread::SetOSName(const char* name)
|
||||
{
|
||||
#ifdef HAVE_LINUX
|
||||
prctl(PR_SET_NAME, name.c_str(), 0, 0, 0);
|
||||
prctl(PR_SET_NAME, name, 0, 0, 0);
|
||||
#endif
|
||||
|
||||
#ifdef __APPLE__
|
||||
pthread_setname_np(name.c_str());
|
||||
pthread_setname_np(name);
|
||||
#endif
|
||||
|
||||
#ifdef FREEBSD
|
||||
pthread_set_name_np(pthread_self(), name, name.c_str());
|
||||
pthread_set_name_np(pthread_self(), name, name);
|
||||
#endif
|
||||
}
|
||||
|
||||
const char* BasicThread::Fmt(const char* format, ...)
|
||||
{
|
||||
if ( buf_len > 10 * STD_FMT_BUF_LEN )
|
||||
{
|
||||
// Shrink back to normal.
|
||||
buf = (char*) safe_realloc(buf, STD_FMT_BUF_LEN);
|
||||
buf_len = STD_FMT_BUF_LEN;
|
||||
}
|
||||
|
||||
va_list al;
|
||||
va_start(al, format);
|
||||
int n = safe_vsnprintf(buf, buf_len, format, al);
|
||||
|
|
@ -64,15 +79,13 @@ const char* BasicThread::Fmt(const char* format, ...)
|
|||
|
||||
if ( (unsigned int) n >= buf_len )
|
||||
{ // Not enough room, grow the buffer.
|
||||
int tmp_len = n + 32;
|
||||
char* tmp = (char*) malloc(tmp_len);
|
||||
buf_len = n + 32;
|
||||
buf = (char*) safe_realloc(buf, buf_len);
|
||||
|
||||
// Is it portable to restart?
|
||||
va_start(al, format);
|
||||
n = safe_vsnprintf(tmp, tmp_len, format, al);
|
||||
n = safe_vsnprintf(buf, buf_len, format, al);
|
||||
va_end(al);
|
||||
|
||||
free(tmp);
|
||||
}
|
||||
|
||||
return buf;
|
||||
|
|
@ -94,14 +107,14 @@ void BasicThread::Start()
|
|||
|
||||
int err = pthread_create(&pthread, 0, BasicThread::launcher, this);
|
||||
if ( err != 0 )
|
||||
reporter->FatalError("Cannot create thread %s:%s", name.c_str(), Strerror(err));
|
||||
reporter->FatalError("Cannot create thread %s: %s", name, Strerror(err));
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Started thread %s", name.c_str());
|
||||
DBG_LOG(DBG_THREADING, "Started thread %s", name);
|
||||
|
||||
OnStart();
|
||||
}
|
||||
|
||||
void BasicThread::Stop()
|
||||
void BasicThread::PrepareStop()
|
||||
{
|
||||
if ( ! started )
|
||||
return;
|
||||
|
|
@ -109,11 +122,28 @@ void BasicThread::Stop()
|
|||
if ( terminating )
|
||||
return;
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Signaling thread %s to terminate ...", name.c_str());
|
||||
DBG_LOG(DBG_THREADING, "Preparing thread %s to terminate ...", name);
|
||||
|
||||
terminating = true;
|
||||
OnPrepareStop();
|
||||
}
|
||||
|
||||
void BasicThread::Stop()
|
||||
{
|
||||
// XX fprintf(stderr, "stop1 %s %d %d\n", name, started, terminating);
|
||||
|
||||
if ( ! started )
|
||||
return;
|
||||
|
||||
if ( terminating )
|
||||
return;
|
||||
|
||||
// XX fprintf(stderr, "stop2 %s\n", name);
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Signaling thread %s to terminate ...", name);
|
||||
|
||||
OnStop();
|
||||
|
||||
terminating = true;
|
||||
}
|
||||
|
||||
void BasicThread::Join()
|
||||
|
|
@ -123,25 +153,33 @@ void BasicThread::Join()
|
|||
|
||||
assert(terminating);
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Joining thread %s ...", name.c_str());
|
||||
DBG_LOG(DBG_THREADING, "Joining thread %s ...", name);
|
||||
|
||||
if ( pthread && pthread_join(pthread, 0) != 0 )
|
||||
reporter->FatalError("Failure joining thread %s", name.c_str());
|
||||
reporter->FatalError("Failure joining thread %s", name);
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Done with thread %s", name.c_str());
|
||||
DBG_LOG(DBG_THREADING, "Joined with thread %s", name);
|
||||
|
||||
pthread = 0;
|
||||
}
|
||||
|
||||
void BasicThread::Kill()
|
||||
{
|
||||
// We don't *really* kill the thread here because that leads to race
|
||||
// conditions. Instead we set a flag that parts of the the code need
|
||||
// to check and get out of any loops they might be in.
|
||||
terminating = true;
|
||||
killed = true;
|
||||
OnKill();
|
||||
}
|
||||
|
||||
if ( ! (started && pthread) )
|
||||
return;
|
||||
void BasicThread::Done()
|
||||
{
|
||||
// XX fprintf(stderr, "DONE from thread %s\n", name);
|
||||
DBG_LOG(DBG_THREADING, "Thread %s has finished", name);
|
||||
|
||||
pthread = 0;
|
||||
pthread_kill(pthread, SIGTERM);
|
||||
terminating = true;
|
||||
killed = true;
|
||||
}
|
||||
|
||||
void* BasicThread::launcher(void *arg)
|
||||
|
|
@ -161,11 +199,12 @@ void* BasicThread::launcher(void *arg)
|
|||
sigdelset(&mask_set, SIGSEGV);
|
||||
sigdelset(&mask_set, SIGBUS);
|
||||
int res = pthread_sigmask(SIG_BLOCK, &mask_set, 0);
|
||||
assert(res == 0); //
|
||||
assert(res == 0);
|
||||
|
||||
// Run thread's main function.
|
||||
thread->Run();
|
||||
|
||||
thread->Done();
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@
|
|||
#include <pthread.h>
|
||||
#include <semaphore.h>
|
||||
|
||||
#include "Queue.h"
|
||||
#include "util.h"
|
||||
|
||||
using namespace std;
|
||||
|
|
@ -42,22 +41,25 @@ public:
|
|||
*
|
||||
* This method is safe to call from any thread.
|
||||
*/
|
||||
const string& Name() const { return name; }
|
||||
const char* Name() const { return name; }
|
||||
|
||||
/**
|
||||
* Sets a descriptive name for the thread. This should be a string
|
||||
* that's useful in output presented to the user and uniquely
|
||||
* identifies the thread.
|
||||
*
|
||||
* This method must be called only from the thread itself.
|
||||
* This method must be called only from main thread at initialization
|
||||
* time.
|
||||
*/
|
||||
void SetName(const string& name);
|
||||
void SetName(const char* name);
|
||||
|
||||
/**
|
||||
* Set the name shown by the OS as the thread's description. Not
|
||||
* supported on all OSs.
|
||||
*
|
||||
* Must be called only from the child thread.
|
||||
*/
|
||||
void SetOSName(const string& name);
|
||||
void SetOSName(const char* name);
|
||||
|
||||
/**
|
||||
* Starts the thread. Calling this methods will spawn a new OS thread
|
||||
|
|
@ -68,6 +70,18 @@ public:
|
|||
*/
|
||||
void Start();
|
||||
|
||||
/**
|
||||
* Signals the thread to prepare for stopping. This must be called
|
||||
* before Stop() and allows the thread to trigger shutting down
|
||||
* without yet blocking for doing so.
|
||||
*
|
||||
* Calling this method has no effect if Start() hasn't been executed
|
||||
* yet.
|
||||
*
|
||||
* Only Bro's main thread must call this method.
|
||||
*/
|
||||
void PrepareStop();
|
||||
|
||||
/**
|
||||
* Signals the thread to stop. The method lets Terminating() now
|
||||
* return true. It does however not force the thread to terminate.
|
||||
|
|
@ -88,6 +102,13 @@ public:
|
|||
*/
|
||||
bool Terminating() const { return terminating; }
|
||||
|
||||
/**
|
||||
* Returns true if Kill() has been called.
|
||||
*
|
||||
* This method is safe to call from any thread.
|
||||
*/
|
||||
bool Killed() const { return killed; }
|
||||
|
||||
/**
|
||||
* A version of fmt() that the thread can safely use.
|
||||
*
|
||||
|
|
@ -124,12 +145,24 @@ protected:
|
|||
virtual void OnStart() {}
|
||||
|
||||
/**
|
||||
* Executed with Stop(). This is a hook into stopping the thread. It
|
||||
* will be called from Bro's main thread after the thread has been
|
||||
* signaled to stop.
|
||||
* Executed with PrepareStop() (and before OnStop()). This is a hook
|
||||
* into preparing the thread for stopping. It will be called from
|
||||
* Bro's main thread before the thread has been signaled to stop.
|
||||
*/
|
||||
virtual void OnPrepareStop() {}
|
||||
|
||||
/**
|
||||
* Executed with Stop() (and after OnPrepareStop()). This is a hook
|
||||
* into stopping the thread. It will be called from Bro's main thread
|
||||
* after the thread has been signaled to stop.
|
||||
*/
|
||||
virtual void OnStop() {}
|
||||
|
||||
/**
|
||||
* Executed with Kill(). This is a hook into killing the thread.
|
||||
*/
|
||||
virtual void OnKill() {}
|
||||
|
||||
/**
|
||||
* Destructor. This will be called by the manager.
|
||||
*
|
||||
|
|
@ -153,14 +186,18 @@ protected:
|
|||
*/
|
||||
void Kill();
|
||||
|
||||
/** Called by child thread's launcher when it's done processing. */
|
||||
void Done();
|
||||
|
||||
private:
|
||||
// pthread entry function.
|
||||
static void* launcher(void *arg);
|
||||
|
||||
string name;
|
||||
const char* name;
|
||||
pthread_t pthread;
|
||||
bool started; // Set to to true once running.
|
||||
bool terminating; // Set to to true to signal termination.
|
||||
bool killed; // Set to true once forcefully killed.
|
||||
|
||||
// Used as a semaphore to tell the pthread thread when it may
|
||||
// terminate.
|
||||
|
|
|
|||
|
|
@ -30,6 +30,10 @@ void Manager::Terminate()
|
|||
do Process(); while ( did_process );
|
||||
|
||||
// Signal all to stop.
|
||||
|
||||
for ( all_thread_list::iterator i = all_threads.begin(); i != all_threads.end(); i++ )
|
||||
(*i)->PrepareStop();
|
||||
|
||||
for ( all_thread_list::iterator i = all_threads.begin(); i != all_threads.end(); i++ )
|
||||
(*i)->Stop();
|
||||
|
||||
|
|
@ -50,14 +54,14 @@ void Manager::Terminate()
|
|||
|
||||
void Manager::AddThread(BasicThread* thread)
|
||||
{
|
||||
DBG_LOG(DBG_THREADING, "Adding thread %s ...", thread->Name().c_str());
|
||||
DBG_LOG(DBG_THREADING, "Adding thread %s ...", thread->Name());
|
||||
all_threads.push_back(thread);
|
||||
idle = false;
|
||||
}
|
||||
|
||||
void Manager::AddMsgThread(MsgThread* thread)
|
||||
{
|
||||
DBG_LOG(DBG_THREADING, "%s is a MsgThread ...", thread->Name().c_str());
|
||||
DBG_LOG(DBG_THREADING, "%s is a MsgThread ...", thread->Name());
|
||||
msg_threads.push_back(thread);
|
||||
}
|
||||
|
||||
|
|
@ -114,6 +118,12 @@ void Manager::Process()
|
|||
{
|
||||
Message* msg = t->RetrieveOut();
|
||||
|
||||
if ( ! msg )
|
||||
{
|
||||
assert(t->Killed());
|
||||
break;
|
||||
}
|
||||
|
||||
if ( msg->Process() )
|
||||
{
|
||||
if ( network_time )
|
||||
|
|
@ -122,10 +132,9 @@ void Manager::Process()
|
|||
|
||||
else
|
||||
{
|
||||
string s = msg->Name() + " failed, terminating thread";
|
||||
reporter->Error("%s", s.c_str());
|
||||
reporter->Error("%s failed, terminating thread", msg->Name());
|
||||
t->Stop();
|
||||
}
|
||||
}
|
||||
|
||||
delete msg;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,16 +29,6 @@ private:
|
|||
double network_time;
|
||||
};
|
||||
|
||||
// A dummy message that's only purpose is unblock the current read operation
|
||||
// so that the child's Run() methods can check the termination status.
|
||||
class UnblockMessage : public InputMessage<MsgThread>
|
||||
{
|
||||
public:
|
||||
UnblockMessage(MsgThread* thread) : InputMessage<MsgThread>("Unblock", thread) { }
|
||||
|
||||
virtual bool Process() { return true; }
|
||||
};
|
||||
|
||||
/// Sends a heartbeat to the child thread.
|
||||
class HeartbeatMessage : public InputMessage<MsgThread>
|
||||
{
|
||||
|
|
@ -66,14 +56,16 @@ public:
|
|||
INTERNAL_WARNING, INTERNAL_ERROR
|
||||
};
|
||||
|
||||
ReporterMessage(Type arg_type, MsgThread* thread, const string& arg_msg)
|
||||
ReporterMessage(Type arg_type, MsgThread* thread, const char* arg_msg)
|
||||
: OutputMessage<MsgThread>("ReporterMessage", thread)
|
||||
{ type = arg_type; msg = arg_msg; }
|
||||
{ type = arg_type; msg = copy_string(arg_msg); }
|
||||
|
||||
~ReporterMessage() { delete [] msg; }
|
||||
|
||||
virtual bool Process();
|
||||
|
||||
private:
|
||||
string msg;
|
||||
const char* msg;
|
||||
Type type;
|
||||
};
|
||||
|
||||
|
|
@ -82,18 +74,19 @@ private:
|
|||
class DebugMessage : public OutputMessage<MsgThread>
|
||||
{
|
||||
public:
|
||||
DebugMessage(DebugStream arg_stream, MsgThread* thread, const string& arg_msg)
|
||||
DebugMessage(DebugStream arg_stream, MsgThread* thread, const char* arg_msg)
|
||||
: OutputMessage<MsgThread>("DebugMessage", thread)
|
||||
{ stream = arg_stream; msg = arg_msg; }
|
||||
{ stream = arg_stream; msg = copy_string(arg_msg); }
|
||||
|
||||
virtual ~DebugMessage() { delete [] msg; }
|
||||
|
||||
virtual bool Process()
|
||||
{
|
||||
string s = Object()->Name() + ": " + msg;
|
||||
debug_logger.Log(stream, "%s", s.c_str());
|
||||
debug_logger.Log(stream, "%s: %s", Object()->Name(), msg);
|
||||
return true;
|
||||
}
|
||||
private:
|
||||
string msg;
|
||||
const char* msg;
|
||||
DebugStream stream;
|
||||
};
|
||||
#endif
|
||||
|
|
@ -104,41 +97,39 @@ private:
|
|||
|
||||
Message::~Message()
|
||||
{
|
||||
delete [] name;
|
||||
}
|
||||
|
||||
bool ReporterMessage::Process()
|
||||
{
|
||||
string s = Object()->Name() + ": " + msg;
|
||||
const char* cmsg = s.c_str();
|
||||
|
||||
switch ( type ) {
|
||||
|
||||
case INFO:
|
||||
reporter->Info("%s", cmsg);
|
||||
reporter->Info("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
case WARNING:
|
||||
reporter->Warning("%s", cmsg);
|
||||
reporter->Warning("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
case ERROR:
|
||||
reporter->Error("%s", cmsg);
|
||||
reporter->Error("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
case FATAL_ERROR:
|
||||
reporter->FatalError("%s", cmsg);
|
||||
reporter->FatalError("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
case FATAL_ERROR_WITH_CORE:
|
||||
reporter->FatalErrorWithCore("%s", cmsg);
|
||||
reporter->FatalErrorWithCore("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
case INTERNAL_WARNING:
|
||||
reporter->InternalWarning("%s", cmsg);
|
||||
reporter->InternalWarning("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
case INTERNAL_ERROR :
|
||||
reporter->InternalError("%s", cmsg);
|
||||
reporter->InternalError("%s: %s", Object()->Name(), msg);
|
||||
break;
|
||||
|
||||
default:
|
||||
|
|
@ -148,62 +139,78 @@ bool ReporterMessage::Process()
|
|||
return true;
|
||||
}
|
||||
|
||||
MsgThread::MsgThread() : BasicThread()
|
||||
MsgThread::MsgThread() : BasicThread(), queue_in(this, 0), queue_out(0, this)
|
||||
{
|
||||
cnt_sent_in = cnt_sent_out = 0;
|
||||
finished = false;
|
||||
stopped = false;
|
||||
thread_mgr->AddMsgThread(this);
|
||||
}
|
||||
|
||||
// Set by Bro's main signal handler.
|
||||
extern int signal_val;
|
||||
|
||||
void MsgThread::OnStop()
|
||||
void MsgThread::OnPrepareStop()
|
||||
{
|
||||
if ( stopped )
|
||||
if ( finished || Killed() )
|
||||
return;
|
||||
|
||||
// XX fprintf(stderr, "Sending FINISH to thread %s ...\n", Name());
|
||||
|
||||
// Signal thread to terminate and wait until it has acknowledged.
|
||||
SendIn(new FinishMessage(this, network_time), true);
|
||||
}
|
||||
|
||||
void MsgThread::OnStop()
|
||||
{
|
||||
int signal_count = 0;
|
||||
int old_signal_val = signal_val;
|
||||
signal_val = 0;
|
||||
|
||||
int cnt = 0;
|
||||
bool aborted = 0;
|
||||
uint64_t last_size = 0;
|
||||
uint64_t cur_size = 0;
|
||||
|
||||
while ( ! finished )
|
||||
// XX fprintf(stderr, "WAITING for thread %s to stop ...\n", Name());
|
||||
|
||||
while ( ! (finished || Killed() ) )
|
||||
{
|
||||
// Terminate if we get another kill signal.
|
||||
if ( signal_val == SIGTERM || signal_val == SIGINT )
|
||||
{
|
||||
// Abort all threads here so that we won't hang next
|
||||
// on another one.
|
||||
fprintf(stderr, "received signal while waiting for thread %s, aborting all ...\n", Name().c_str());
|
||||
thread_mgr->KillThreads();
|
||||
aborted = true;
|
||||
break;
|
||||
++signal_count;
|
||||
|
||||
if ( signal_count == 1 )
|
||||
{
|
||||
// Abort all threads here so that we won't hang next
|
||||
// on another one.
|
||||
fprintf(stderr, "received signal while waiting for thread %s, aborting all ...\n", Name());
|
||||
thread_mgr->KillThreads();
|
||||
}
|
||||
else
|
||||
{
|
||||
// More than one signal. Abort processing
|
||||
// right away. on another one.
|
||||
fprintf(stderr, "received another signal while waiting for thread %s, aborting processing\n", Name());
|
||||
exit(1);
|
||||
}
|
||||
|
||||
signal_val = 0;
|
||||
}
|
||||
|
||||
if ( ++cnt % 10000 == 0 ) // Insurance against broken threads ...
|
||||
{
|
||||
fprintf(stderr, "killing thread %s ...\n", Name().c_str());
|
||||
Kill();
|
||||
aborted = true;
|
||||
break;
|
||||
}
|
||||
queue_in.WakeUp();
|
||||
|
||||
usleep(1000);
|
||||
}
|
||||
|
||||
Finished();
|
||||
|
||||
signal_val = old_signal_val;
|
||||
}
|
||||
|
||||
// One more message to make sure the current queue read operation unblocks.
|
||||
if ( ! aborted )
|
||||
SendIn(new UnblockMessage(this), true);
|
||||
void MsgThread::OnKill()
|
||||
{
|
||||
// Send a message to unblock the reader if its currently waiting for
|
||||
// input. This is just an optimization to make it terminate more
|
||||
// quickly, even without the message it will eventually time out.
|
||||
queue_in.WakeUp();
|
||||
}
|
||||
|
||||
void MsgThread::Heartbeat()
|
||||
|
|
@ -213,9 +220,7 @@ void MsgThread::Heartbeat()
|
|||
|
||||
void MsgThread::HeartbeatInChild()
|
||||
{
|
||||
string n = Name();
|
||||
|
||||
n = Fmt("bro: %s (%" PRIu64 "/%" PRIu64 ")", n.c_str(),
|
||||
string n = Fmt("bro: %s (%" PRIu64 "/%" PRIu64 ")", Name(),
|
||||
cnt_sent_in - queue_in.Size(),
|
||||
cnt_sent_out - queue_out.Size());
|
||||
|
||||
|
|
@ -283,7 +288,7 @@ void MsgThread::SendIn(BasicInputMessage* msg, bool force)
|
|||
return;
|
||||
}
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Sending '%s' to %s ...", msg->Name().c_str(), Name().c_str());
|
||||
DBG_LOG(DBG_THREADING, "Sending '%s' to %s ...", msg->Name(), Name());
|
||||
|
||||
queue_in.Put(msg);
|
||||
++cnt_sent_in;
|
||||
|
|
@ -306,9 +311,10 @@ void MsgThread::SendOut(BasicOutputMessage* msg, bool force)
|
|||
BasicOutputMessage* MsgThread::RetrieveOut()
|
||||
{
|
||||
BasicOutputMessage* msg = queue_out.Get();
|
||||
assert(msg);
|
||||
if ( ! msg )
|
||||
return 0;
|
||||
|
||||
DBG_LOG(DBG_THREADING, "Retrieved '%s' from %s", msg->Name().c_str(), Name().c_str());
|
||||
DBG_LOG(DBG_THREADING, "Retrieved '%s' from %s", msg->Name(), Name());
|
||||
|
||||
return msg;
|
||||
}
|
||||
|
|
@ -316,10 +322,12 @@ BasicOutputMessage* MsgThread::RetrieveOut()
|
|||
BasicInputMessage* MsgThread::RetrieveIn()
|
||||
{
|
||||
BasicInputMessage* msg = queue_in.Get();
|
||||
assert(msg);
|
||||
|
||||
if ( ! msg )
|
||||
return 0;
|
||||
|
||||
#ifdef DEBUG
|
||||
string s = Fmt("Retrieved '%s' in %s", msg->Name().c_str(), Name().c_str());
|
||||
string s = Fmt("Retrieved '%s' in %s", msg->Name(), Name());
|
||||
Debug(DBG_THREADING, s.c_str());
|
||||
#endif
|
||||
|
||||
|
|
@ -328,15 +336,18 @@ BasicInputMessage* MsgThread::RetrieveIn()
|
|||
|
||||
void MsgThread::Run()
|
||||
{
|
||||
while ( ! finished )
|
||||
while ( ! (finished || Killed() ) )
|
||||
{
|
||||
BasicInputMessage* msg = RetrieveIn();
|
||||
|
||||
if ( ! msg )
|
||||
continue;
|
||||
|
||||
bool result = msg->Process();
|
||||
|
||||
if ( ! result )
|
||||
{
|
||||
string s = msg->Name() + " failed, terminating thread (MsgThread)";
|
||||
string s = Fmt("%s failed, terminating thread (MsgThread)", Name());
|
||||
Error(s.c_str());
|
||||
break;
|
||||
}
|
||||
|
|
@ -344,7 +355,7 @@ void MsgThread::Run()
|
|||
delete msg;
|
||||
}
|
||||
|
||||
Finished();
|
||||
Finished();
|
||||
}
|
||||
|
||||
void MsgThread::GetStats(Stats* stats)
|
||||
|
|
|
|||
|
|
@ -228,6 +228,8 @@ protected:
|
|||
*/
|
||||
virtual void Run();
|
||||
virtual void OnStop();
|
||||
virtual void OnPrepareStop();
|
||||
virtual void OnKill();
|
||||
|
||||
private:
|
||||
/**
|
||||
|
|
@ -293,7 +295,6 @@ private:
|
|||
uint64_t cnt_sent_out; // Counts message sent by child.
|
||||
|
||||
bool finished; // Set to true by Finished message.
|
||||
bool stopped; // Set to true by OnStop().
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -312,7 +313,7 @@ public:
|
|||
* what's passed into the constructor and used mainly for debugging
|
||||
* purposes.
|
||||
*/
|
||||
const string& Name() const { return name; }
|
||||
const char* Name() const { return name; }
|
||||
|
||||
/**
|
||||
* Callback that must be overriden for processing a message.
|
||||
|
|
@ -326,10 +327,11 @@ protected:
|
|||
* @param arg_name A descriptive name for the type of message. Used
|
||||
* mainly for debugging purposes.
|
||||
*/
|
||||
Message(const string& arg_name) { name = arg_name; }
|
||||
Message(const char* arg_name)
|
||||
{ name = copy_string(arg_name); }
|
||||
|
||||
private:
|
||||
string name;
|
||||
const char* name;
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -344,7 +346,7 @@ protected:
|
|||
* @param name A descriptive name for the type of message. Used
|
||||
* mainly for debugging purposes.
|
||||
*/
|
||||
BasicInputMessage(const string& name) : Message(name) {}
|
||||
BasicInputMessage(const char* name) : Message(name) {}
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -359,7 +361,7 @@ protected:
|
|||
* @param name A descriptive name for the type of message. Used
|
||||
* mainly for debugging purposes.
|
||||
*/
|
||||
BasicOutputMessage(const string& name) : Message(name) {}
|
||||
BasicOutputMessage(const char* name) : Message(name) {}
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -384,7 +386,7 @@ protected:
|
|||
*
|
||||
* @param arg_object: An object to store with the message.
|
||||
*/
|
||||
InputMessage(const string& name, O* arg_object) : BasicInputMessage(name)
|
||||
InputMessage(const char* name, O* arg_object) : BasicInputMessage(name)
|
||||
{ object = arg_object; }
|
||||
|
||||
private:
|
||||
|
|
@ -413,7 +415,7 @@ protected:
|
|||
*
|
||||
* @param arg_object An object to store with the message.
|
||||
*/
|
||||
OutputMessage(const string& name, O* arg_object) : BasicOutputMessage(name)
|
||||
OutputMessage(const char* name, O* arg_object) : BasicOutputMessage(name)
|
||||
{ object = arg_object; }
|
||||
|
||||
private:
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
|
||||
#ifndef THREADING_QUEUE_H
|
||||
#define THREADING_QUEUE_H
|
||||
|
||||
|
|
@ -6,11 +5,28 @@
|
|||
#include <queue>
|
||||
#include <deque>
|
||||
#include <stdint.h>
|
||||
#include <sys/time.h>
|
||||
|
||||
#include "Reporter.h"
|
||||
#include "BasicThread.h"
|
||||
|
||||
#undef Queue // Defined elsewhere unfortunately.
|
||||
|
||||
#if 1
|
||||
// We don't have pthread spinlocks on DARWIN.
|
||||
# define PTHREAD_MUTEX_T pthread_mutex_t
|
||||
# define PTHREAD_MUTEX_LOCK(x) pthread_mutex_lock(x)
|
||||
# define PTHREAD_MUTEX_UNLOCK(x) pthread_mutex_unlock(x)
|
||||
# define PTHREAD_MUTEX_INIT(x) pthread_mutex_init(x, 0)
|
||||
# define PTHREAD_MUTEX_DESTROY(x) pthread_mutex_destroy(x)
|
||||
#else
|
||||
# define PTHREAD_MUTEX_T pthrea_spinlock_T
|
||||
# define PTHREAD_MUTEX_LOCK(x) pthrea_spin_lock(x)
|
||||
# define PTHREAD_MUTEX_UNLOCK(x) pthrea_spin_unlock(x)
|
||||
# define PTHREAD_MUTEX_INIT(x) pthrea_spin_init(x, PTHREAD_PROCESS_PRIVATE)
|
||||
# define PTHREAD_MUTEX_DESTROY(x) pthrea_spin_destroy(x)
|
||||
#endif
|
||||
|
||||
namespace threading {
|
||||
|
||||
/**
|
||||
|
|
@ -30,8 +46,12 @@ class Queue
|
|||
public:
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
* reader, writer: The corresponding threads. This is for checking
|
||||
* whether they have terminated so that we can abort I/O opeations.
|
||||
* Can be left null for the main thread.
|
||||
*/
|
||||
Queue();
|
||||
Queue(BasicThread* arg_reader, BasicThread* arg_writer);
|
||||
|
||||
/**
|
||||
* Destructor.
|
||||
|
|
@ -39,7 +59,9 @@ public:
|
|||
~Queue();
|
||||
|
||||
/**
|
||||
* Retrieves one elment.
|
||||
* Retrieves one elment. This may block for a little while of no
|
||||
* input is available and eventually return with a null element if
|
||||
* nothing shows up.
|
||||
*/
|
||||
T Get();
|
||||
|
||||
|
|
@ -60,6 +82,11 @@ public:
|
|||
*/
|
||||
bool MaybeReady() { return ( ( read_ptr - write_ptr) != 0 ); }
|
||||
|
||||
/** Wake up the reader if it's currently blocked for input. This is
|
||||
primarily to give it a chance to check termination quickly.
|
||||
**/
|
||||
void WakeUp();
|
||||
|
||||
/**
|
||||
* Returns the number of queued items not yet retrieved.
|
||||
*/
|
||||
|
|
@ -82,45 +109,50 @@ public:
|
|||
void GetStats(Stats* stats);
|
||||
|
||||
private:
|
||||
static const int NUM_QUEUES = 8;
|
||||
static const int NUM_QUEUES = 15;
|
||||
|
||||
pthread_mutex_t mutex[NUM_QUEUES]; // Mutex protected shared accesses.
|
||||
PTHREAD_MUTEX_T mutex[NUM_QUEUES]; // Mutex protected shared accesses.
|
||||
pthread_cond_t has_data[NUM_QUEUES]; // Signals when data becomes available
|
||||
std::queue<T> messages[NUM_QUEUES]; // Actually holds the queued messages
|
||||
|
||||
int read_ptr; // Where the next operation will read from
|
||||
int write_ptr; // Where the next operation will write to
|
||||
|
||||
BasicThread* reader;
|
||||
BasicThread* writer;
|
||||
|
||||
// Statistics.
|
||||
uint64_t num_reads;
|
||||
uint64_t num_writes;
|
||||
};
|
||||
|
||||
inline static void safe_lock(pthread_mutex_t* mutex)
|
||||
inline static void safe_lock(PTHREAD_MUTEX_T* mutex)
|
||||
{
|
||||
if ( pthread_mutex_lock(mutex) != 0 )
|
||||
if ( PTHREAD_MUTEX_LOCK(mutex) != 0 )
|
||||
reporter->FatalErrorWithCore("cannot lock mutex");
|
||||
}
|
||||
|
||||
inline static void safe_unlock(pthread_mutex_t* mutex)
|
||||
inline static void safe_unlock(PTHREAD_MUTEX_T* mutex)
|
||||
{
|
||||
if ( pthread_mutex_unlock(mutex) != 0 )
|
||||
if ( PTHREAD_MUTEX_UNLOCK(mutex) != 0 )
|
||||
reporter->FatalErrorWithCore("cannot unlock mutex");
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
inline Queue<T>::Queue()
|
||||
inline Queue<T>::Queue(BasicThread* arg_reader, BasicThread* arg_writer)
|
||||
{
|
||||
read_ptr = 0;
|
||||
write_ptr = 0;
|
||||
num_reads = num_writes = 0;
|
||||
reader = arg_reader;
|
||||
writer = arg_writer;
|
||||
|
||||
for( int i = 0; i < NUM_QUEUES; ++i )
|
||||
{
|
||||
if ( pthread_cond_init(&has_data[i], NULL) != 0 )
|
||||
if ( pthread_cond_init(&has_data[i], 0) != 0 )
|
||||
reporter->FatalError("cannot init queue condition variable");
|
||||
|
||||
if ( pthread_mutex_init(&mutex[i], NULL) != 0 )
|
||||
if ( PTHREAD_MUTEX_INIT(&mutex[i]) != 0 )
|
||||
reporter->FatalError("cannot init queue mutex");
|
||||
}
|
||||
}
|
||||
|
|
@ -131,19 +163,30 @@ inline Queue<T>::~Queue()
|
|||
for( int i = 0; i < NUM_QUEUES; ++i )
|
||||
{
|
||||
pthread_cond_destroy(&has_data[i]);
|
||||
pthread_mutex_destroy(&mutex[i]);
|
||||
PTHREAD_MUTEX_DESTROY(&mutex[i]);
|
||||
}
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
inline T Queue<T>::Get()
|
||||
{
|
||||
if ( (reader && reader->Killed()) || (writer && writer->Killed()) )
|
||||
return 0;
|
||||
|
||||
safe_lock(&mutex[read_ptr]);
|
||||
|
||||
int old_read_ptr = read_ptr;
|
||||
|
||||
if ( messages[read_ptr].empty() )
|
||||
pthread_cond_wait(&has_data[read_ptr], &mutex[read_ptr]);
|
||||
{
|
||||
struct timespec ts;
|
||||
ts.tv_sec = time(0) + 5;
|
||||
ts.tv_nsec = 0;
|
||||
|
||||
pthread_cond_timedwait(&has_data[read_ptr], &mutex[read_ptr], &ts);
|
||||
safe_unlock(&mutex[read_ptr]);
|
||||
return 0;
|
||||
}
|
||||
|
||||
T data = messages[read_ptr].front();
|
||||
messages[read_ptr].pop();
|
||||
|
|
@ -222,6 +265,17 @@ inline void Queue<T>::GetStats(Stats* stats)
|
|||
safe_unlock(&mutex[i]);
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
inline void Queue<T>::WakeUp()
|
||||
{
|
||||
for ( int i = 0; i < NUM_QUEUES; i++ )
|
||||
{
|
||||
safe_lock(&mutex[i]);
|
||||
pthread_cond_signal(&has_data[i]);
|
||||
safe_unlock(&mutex[i]);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -11,23 +11,54 @@ bool Field::Read(SerializationFormat* fmt)
|
|||
{
|
||||
int t;
|
||||
int st;
|
||||
string tmp_name;
|
||||
bool have_2nd;
|
||||
|
||||
bool success = (fmt->Read(&name, "name")
|
||||
&& fmt->Read(&secondary_name, "secondary_name")
|
||||
if ( ! fmt->Read(&have_2nd, "have_2nd") )
|
||||
return false;
|
||||
|
||||
if ( have_2nd )
|
||||
{
|
||||
string tmp_secondary_name;
|
||||
if ( ! fmt->Read(&tmp_secondary_name, "secondary_name") )
|
||||
return false;
|
||||
|
||||
secondary_name = copy_string(tmp_secondary_name.c_str());
|
||||
}
|
||||
else
|
||||
secondary_name = 0;
|
||||
|
||||
bool success = (fmt->Read(&tmp_name, "name")
|
||||
&& fmt->Read(&t, "type")
|
||||
&& fmt->Read(&st, "subtype")
|
||||
&& fmt->Read(&optional, "optional"));
|
||||
|
||||
if ( ! success )
|
||||
return false;
|
||||
|
||||
name = copy_string(tmp_name.c_str());
|
||||
|
||||
type = (TypeTag) t;
|
||||
subtype = (TypeTag) st;
|
||||
|
||||
return success;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Field::Write(SerializationFormat* fmt) const
|
||||
{
|
||||
assert(name);
|
||||
|
||||
if ( secondary_name )
|
||||
{
|
||||
if ( ! (fmt->Write(true, "have_2nd")
|
||||
&& fmt->Write(secondary_name, "secondary_name")) )
|
||||
return false;
|
||||
}
|
||||
else
|
||||
if ( ! fmt->Write(false, "have_2nd") )
|
||||
return false;
|
||||
|
||||
return (fmt->Write(name, "name")
|
||||
&& fmt->Write(secondary_name, "secondary_name")
|
||||
&& fmt->Write((int)type, "type")
|
||||
&& fmt->Write((int)subtype, "subtype"),
|
||||
fmt->Write(optional, "optional"));
|
||||
|
|
@ -51,7 +82,7 @@ Value::~Value()
|
|||
{
|
||||
if ( (type == TYPE_ENUM || type == TYPE_STRING || type == TYPE_FILE || type == TYPE_FUNC)
|
||||
&& present )
|
||||
delete val.string_val;
|
||||
delete [] val.string_val.data;
|
||||
|
||||
if ( type == TYPE_TABLE && present )
|
||||
{
|
||||
|
|
@ -224,10 +255,7 @@ bool Value::Read(SerializationFormat* fmt)
|
|||
case TYPE_STRING:
|
||||
case TYPE_FILE:
|
||||
case TYPE_FUNC:
|
||||
{
|
||||
val.string_val = new string;
|
||||
return fmt->Read(val.string_val, "string");
|
||||
}
|
||||
return fmt->Read(&val.string_val.data, &val.string_val.length, "string");
|
||||
|
||||
case TYPE_TABLE:
|
||||
{
|
||||
|
|
@ -339,7 +367,7 @@ bool Value::Write(SerializationFormat* fmt) const
|
|||
case TYPE_STRING:
|
||||
case TYPE_FILE:
|
||||
case TYPE_FUNC:
|
||||
return fmt->Write(*val.string_val, "string");
|
||||
return fmt->Write(val.string_val.data, val.string_val.length, "string");
|
||||
|
||||
case TYPE_TABLE:
|
||||
{
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@
|
|||
using namespace std;
|
||||
|
||||
class SerializationFormat;
|
||||
class RemoteSerializer;
|
||||
|
||||
namespace threading {
|
||||
|
||||
|
|
@ -19,10 +20,10 @@ namespace threading {
|
|||
* Definition of a log file, i.e., one column of a log stream.
|
||||
*/
|
||||
struct Field {
|
||||
string name; //! Name of the field.
|
||||
const char* name; //! Name of the field.
|
||||
//! Needed by input framework. Port fields have two names (one for the
|
||||
//! port, one for the type), and this specifies the secondary name.
|
||||
string secondary_name;
|
||||
const char* secondary_name;
|
||||
TypeTag type; //! Type of the field.
|
||||
TypeTag subtype; //! Inner type for sets.
|
||||
bool optional; //! True if field is optional.
|
||||
|
|
@ -30,13 +31,24 @@ struct Field {
|
|||
/**
|
||||
* Constructor.
|
||||
*/
|
||||
Field() { subtype = TYPE_VOID; optional = false; }
|
||||
Field(const char* name, const char* secondary_name, TypeTag type, TypeTag subtype, bool optional)
|
||||
: name(name ? copy_string(name) : 0),
|
||||
secondary_name(secondary_name ? copy_string(secondary_name) : 0),
|
||||
type(type), subtype(subtype), optional(optional) { }
|
||||
|
||||
/**
|
||||
* Copy constructor.
|
||||
*/
|
||||
Field(const Field& other)
|
||||
: name(other.name), type(other.type), subtype(other.subtype), optional(other.optional) { }
|
||||
: name(other.name ? copy_string(other.name) : 0),
|
||||
secondary_name(other.secondary_name ? copy_string(other.secondary_name) : 0),
|
||||
type(other.type), subtype(other.subtype), optional(other.optional) { }
|
||||
|
||||
~Field()
|
||||
{
|
||||
delete [] name;
|
||||
delete [] secondary_name;
|
||||
}
|
||||
|
||||
/**
|
||||
* Unserializes a field.
|
||||
|
|
@ -63,6 +75,12 @@ struct Field {
|
|||
* thread-safe.
|
||||
*/
|
||||
string TypeName() const;
|
||||
|
||||
private:
|
||||
friend class ::RemoteSerializer;
|
||||
|
||||
// Force usage of constructor above.
|
||||
Field() {};
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -102,7 +120,11 @@ struct Value {
|
|||
vec_t vector_val;
|
||||
addr_t addr_val;
|
||||
subnet_t subnet_val;
|
||||
string* string_val;
|
||||
|
||||
struct {
|
||||
char* data;
|
||||
int length;
|
||||
} string_val;
|
||||
} val;
|
||||
|
||||
/**
|
||||
|
|
@ -147,7 +169,7 @@ struct Value {
|
|||
static bool IsCompatibleType(BroType* t, bool atomic_only=false);
|
||||
|
||||
private:
|
||||
friend class ::IPAddr;
|
||||
friend class ::IPAddr;
|
||||
Value(const Value& other) { } // Disabled.
|
||||
};
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue