diff --git a/scripts/base/frameworks/cluster/setup-connections.bro b/scripts/base/frameworks/cluster/setup-connections.bro index b5a0d25e1f..20646525be 100644 --- a/scripts/base/frameworks/cluster/setup-connections.bro +++ b/scripts/base/frameworks/cluster/setup-connections.bro @@ -44,7 +44,7 @@ event bro_init() &priority=9 { if ( n$node_type == WORKER && n$proxy == node ) Communication::nodes[i] = - [$host=n$ip, $connect=F, $class=i, $sync=T, $auth=T, $events=worker2proxy_events]; + [$host=n$ip, $connect=F, $class=i, $sync=F, $auth=T, $events=worker2proxy_events]; # accepts connections from the previous one. # (This is not ideal for setups with many proxies) diff --git a/src/Stats.cc b/src/Stats.cc index a2e7496c5f..c3035231e9 100644 --- a/src/Stats.cc +++ b/src/Stats.cc @@ -210,11 +210,16 @@ void ProfileLogger::Log() i != thread_stats.end(); ++i ) { threading::MsgThread::Stats s = i->second; - file->Write(fmt("%0.6f %-25s in=%" PRIu64 " out=%" PRIu64 " pending=%" PRIu64 "/%" PRIu64 "\n", + file->Write(fmt("%0.6f %-25s in=%" PRIu64 " out=%" PRIu64 " pending=%" PRIu64 "/%" PRIu64 + " (#queue r/w: in=%" PRIu64 "/%" PRIu64 " out=%" PRIu64 "/%" PRIu64 ")" + "\n", network_time, i->first.c_str(), s.sent_in, s.sent_out, - s.pending_in, s.pending_out)); + s.pending_in, s.pending_out, + s.queue_in_stats.num_reads, s.queue_in_stats.num_writes, + s.queue_out_stats.num_reads, s.queue_out_stats.num_writes + )); } // Script-level state. diff --git a/src/logging/WriterFrontend.h b/src/logging/WriterFrontend.h index 3e05d17c9e..4d22bd9b1f 100644 --- a/src/logging/WriterFrontend.h +++ b/src/logging/WriterFrontend.h @@ -212,7 +212,7 @@ protected: const threading::Field* const* fields; // The log fields. // Buffer for bulk writes. - static const int WRITER_BUFFER_SIZE = 50; + static const int WRITER_BUFFER_SIZE = 1000; int write_buffer_pos; // Position of next write in buffer. threading::Value*** write_buffer; // Buffer of size WRITER_BUFFER_SIZE. }; diff --git a/src/threading/MsgThread.cc b/src/threading/MsgThread.cc index 145e16c57b..ddcd3df1dd 100644 --- a/src/threading/MsgThread.cc +++ b/src/threading/MsgThread.cc @@ -283,5 +283,7 @@ void MsgThread::GetStats(Stats* stats) stats->sent_out = cnt_sent_out; stats->pending_in = queue_in.Size(); stats->pending_out = queue_out.Size(); + queue_in.GetStats(&stats->queue_in_stats); + queue_out.GetStats(&stats->queue_out_stats); } diff --git a/src/threading/MsgThread.h b/src/threading/MsgThread.h index 28c7690dfa..5ac1c0f780 100644 --- a/src/threading/MsgThread.h +++ b/src/threading/MsgThread.h @@ -154,6 +154,10 @@ public: uint64_t sent_out; //! Number of messages sent from the child thread to the main thread uint64_t pending_in; //! Number of messages sent to the child but not yet processed. uint64_t pending_out; //! Number of messages sent from the child but not yet processed by the main thread. + + /// Statistics from our queues. + Queue::Stats queue_in_stats; + Queue::Stats queue_out_stats; }; /** diff --git a/src/threading/Queue.h b/src/threading/Queue.h index a25f897d23..985ba31714 100644 --- a/src/threading/Queue.h +++ b/src/threading/Queue.h @@ -58,6 +58,22 @@ public: */ uint64_t Size(); + /** + * Statistics about inter-thread communication. + */ + struct Stats + { + uint64_t num_reads; //! Number of messages read from the queue. + uint64_t num_writes; //! Number of messages written to the queue. + }; + + /** + * Returns statistics about the queue's usage. + * + * @param stats A pointer to a structure that will be filled with + * current numbers. */ + void GetStats(Stats* stats); + private: static const int NUM_QUEUES = 8; @@ -67,6 +83,10 @@ private: int read_ptr; // Where the next operation will read from int write_ptr; // Where the next operation will write to + + // Statistics. + uint64_t num_reads; + uint64_t num_writes; }; inline static void safe_lock(pthread_mutex_t* mutex) @@ -86,6 +106,7 @@ inline Queue::Queue() { read_ptr = 0; write_ptr = 0; + num_reads = num_writes = 0; for( int i = 0; i < NUM_QUEUES; ++i ) { @@ -121,6 +142,7 @@ inline T Queue::Get() messages[read_ptr].pop(); read_ptr = (read_ptr + 1) % NUM_QUEUES; + ++num_reads; safe_unlock(&mutex[old_read_ptr]); @@ -142,6 +164,7 @@ inline void Queue::Put(T data) pthread_cond_signal(&has_data[write_ptr]); write_ptr = (write_ptr + 1) % NUM_QUEUES; + ++num_writes; safe_unlock(&mutex[old_write_ptr]); } @@ -177,7 +200,23 @@ inline uint64_t Queue::Size() return size; } +template +inline void Queue::GetStats(Stats* stats) + { + // To be safe, we look all queues. That's probably unneccessary, but + // doesn't really hurt. + for ( int i = 0; i < NUM_QUEUES; i++ ) + safe_lock(&mutex[i]); + + stats->num_reads = num_reads; + stats->num_writes = num_writes; + + for ( int i = 0; i < NUM_QUEUES; i++ ) + safe_unlock(&mutex[i]); + } + } + #endif