Move threading classes to zeek namespaces

This commit is contained in:
Tim Wojtulewicz 2020-08-01 10:48:46 -07:00
parent f310795d79
commit 1262109e5a
42 changed files with 299 additions and 210 deletions

View file

@ -211,9 +211,9 @@ void ProfileLogger::Log()
current_timers[i])); current_timers[i]));
} }
file->Write(fmt("%0.6f Threads: current=%d\n", network_time, thread_mgr->NumThreads())); file->Write(fmt("%0.6f Threads: current=%d\n", network_time, zeek::thread_mgr->NumThreads()));
const threading::Manager::msg_stats_list& thread_stats = thread_mgr->GetMsgThreadStats(); const threading::Manager::msg_stats_list& thread_stats = zeek::thread_mgr->GetMsgThreadStats();
for ( threading::Manager::msg_stats_list::const_iterator i = thread_stats.begin(); for ( threading::Manager::msg_stats_list::const_iterator i = thread_stats.begin();
i != thread_stats.end(); ++i ) i != thread_stats.end(); ++i )
{ {

View file

@ -489,7 +489,8 @@ TableValPtr Val::GetRecordFields()
} }
// This is a static method in this file to avoid including rapidjson's headers in Val.h because they're huge. // This is a static method in this file to avoid including rapidjson's headers in Val.h because they're huge.
static void BuildJSON(threading::formatter::JSON::NullDoubleWriter& writer, Val* val, bool only_loggable=false, RE_Matcher* re=nullptr, const string& key="") static void BuildJSON(zeek::threading::formatter::JSON::NullDoubleWriter& writer, Val* val,
bool only_loggable=false, RE_Matcher* re=nullptr, const string& key="")
{ {
if ( !key.empty() ) if ( !key.empty() )
writer.Key(key); writer.Key(key);
@ -585,7 +586,7 @@ static void BuildJSON(threading::formatter::JSON::NullDoubleWriter& writer, Val*
else else
{ {
rapidjson::StringBuffer buffer; rapidjson::StringBuffer buffer;
threading::formatter::JSON::NullDoubleWriter key_writer(buffer); zeek::threading::formatter::JSON::NullDoubleWriter key_writer(buffer);
BuildJSON(key_writer, entry_key, only_loggable, re); BuildJSON(key_writer, entry_key, only_loggable, re);
string key_str = buffer.GetString(); string key_str = buffer.GetString();
@ -689,7 +690,7 @@ static void BuildJSON(threading::formatter::JSON::NullDoubleWriter& writer, Val*
StringValPtr Val::ToJSON(bool only_loggable, RE_Matcher* re) StringValPtr Val::ToJSON(bool only_loggable, RE_Matcher* re)
{ {
rapidjson::StringBuffer buffer; rapidjson::StringBuffer buffer;
threading::formatter::JSON::NullDoubleWriter writer(buffer); zeek::threading::formatter::JSON::NullDoubleWriter writer(buffer);
BuildJSON(writer, this, only_loggable, re, ""); BuildJSON(writer, this, only_loggable, re, "");

View file

@ -1264,7 +1264,7 @@ bool bro_broker::RecordIterator::DoUnserialize(const broker::data& data)
return true; return true;
} }
broker::data bro_broker::threading_field_to_data(const threading::Field* f) broker::data bro_broker::threading_field_to_data(const zeek::threading::Field* f)
{ {
auto name = f->name; auto name = f->name;
auto type = static_cast<uint64_t>(f->type); auto type = static_cast<uint64_t>(f->type);
@ -1279,7 +1279,7 @@ broker::data bro_broker::threading_field_to_data(const threading::Field* f)
return broker::vector({name, secondary, type, subtype, optional}); return broker::vector({name, secondary, type, subtype, optional});
} }
threading::Field* bro_broker::data_to_threading_field(broker::data d) zeek::threading::Field* bro_broker::data_to_threading_field(broker::data d)
{ {
if ( ! caf::holds_alternative<broker::vector>(d) ) if ( ! caf::holds_alternative<broker::vector>(d) )
return nullptr; return nullptr;
@ -1297,9 +1297,9 @@ threading::Field* bro_broker::data_to_threading_field(broker::data d)
if ( secondary != broker::nil && ! caf::holds_alternative<std::string>(secondary) ) if ( secondary != broker::nil && ! caf::holds_alternative<std::string>(secondary) )
return nullptr; return nullptr;
return new threading::Field(name->c_str(), return new zeek::threading::Field(name->c_str(),
secondary != broker::nil ? caf::get<std::string>(secondary).c_str() : nullptr, secondary != broker::nil ? caf::get<std::string>(secondary).c_str() : nullptr,
static_cast<zeek::TypeTag>(*type), static_cast<zeek::TypeTag>(*type),
static_cast<zeek::TypeTag>(*subtype), static_cast<zeek::TypeTag>(*subtype),
*optional); *optional);
} }

View file

@ -6,10 +6,13 @@
#include "Expr.h" #include "Expr.h"
ZEEK_FORWARD_DECLARE_NAMESPACED(ODesc, zeek); ZEEK_FORWARD_DECLARE_NAMESPACED(ODesc, zeek);
namespace zeek::threading {
struct Value;
struct Field;
}
namespace threading { namespace threading {
struct Value; using Value [[deprecated("Remove in v4.1. Use zeek::threading::Value.")]] = zeek::threading::Value;
struct Field; using Field [[deprecated("Remove in v4.1. Use zeek::threading::Field.")]] = zeek::threading::Field;
} }
namespace bro_broker { namespace bro_broker {
@ -65,34 +68,34 @@ broker::expected<broker::data> val_to_data(const zeek::Val* v);
zeek::ValPtr data_to_val(broker::data d, zeek::Type* type); zeek::ValPtr data_to_val(broker::data d, zeek::Type* type);
/** /**
* Convert a Bro threading::Value to a Broker data value. * Convert a zeek::threading::Value to a Broker data value.
* @param v a Bro threading::Value. * @param v a zeek::threading::Value.
* @return a Broker data value if the Bro threading::Value could be converted to one. * @return a Broker data value if the zeek::threading::Value could be converted to one.
*/ */
broker::expected<broker::data> threading_val_to_data(const threading::Value* v); broker::expected<broker::data> threading_val_to_data(const zeek::threading::Value* v);
/** /**
* Convert a Bro threading::Field to a Broker data value. * Convert a zeek::threading::Field to a Broker data value.
* @param f a Bro threading::Field. * @param f a zeek::threading::Field.
* @return a Broker data value if the Bro threading::Field could be converted to one. * @return a Broker data value if the zeek::threading::Field could be converted to one.
*/ */
broker::data threading_field_to_data(const threading::Field* f); broker::data threading_field_to_data(const zeek::threading::Field* f);
/** /**
* Convert a Broker data value to a Bro threading::Value. * Convert a Broker data value to a zeek::threading::Value.
* @param d a Broker data value. * @param d a Broker data value.
* @return a pointer to a new Bro threading::Value or a nullptr if the conversion was not * @return a pointer to a new zeek::threading::Value or a nullptr if the conversion was not
* possible. * possible.
*/ */
threading::Value* data_to_threading_val(broker::data d); zeek::threading::Value* data_to_threading_val(broker::data d);
/** /**
* Convert a Broker data value to a Bro threading::Value. * Convert a Broker data value to a zeek::threading::Value.
* @param d a Broker data value. * @param d a Broker data value.
* @return a pointer to a new Bro threading::Value or a nullptr if the conversion was not * @return a pointer to a new zeek::threading::Value or a nullptr if the conversion was not
* possible. * possible.
*/ */
threading::Field* data_to_threading_field(broker::data d); zeek::threading::Field* data_to_threading_field(broker::data d);
/** /**
* A Bro value which wraps a Broker data value. * A Bro value which wraps a Broker data value.

View file

@ -494,7 +494,7 @@ bool Manager::PublishIdentifier(std::string topic, std::string id)
bool Manager::PublishLogCreate(zeek::EnumVal* stream, zeek::EnumVal* writer, bool Manager::PublishLogCreate(zeek::EnumVal* stream, zeek::EnumVal* writer,
const zeek::logging::WriterBackend::WriterInfo& info, const zeek::logging::WriterBackend::WriterInfo& info,
int num_fields, const threading::Field* const * fields, int num_fields, const zeek::threading::Field* const * fields,
const broker::endpoint_info& peer) const broker::endpoint_info& peer)
{ {
if ( bstate->endpoint.is_shutdown() ) if ( bstate->endpoint.is_shutdown() )
@ -550,7 +550,7 @@ bool Manager::PublishLogCreate(zeek::EnumVal* stream, zeek::EnumVal* writer,
} }
bool Manager::PublishLogWrite(zeek::EnumVal* stream, zeek::EnumVal* writer, string path, bool Manager::PublishLogWrite(zeek::EnumVal* stream, zeek::EnumVal* writer, string path,
int num_fields, const threading::Value* const * vals) int num_fields, const zeek::threading::Value* const * vals)
{ {
if ( bstate->endpoint.is_shutdown() ) if ( bstate->endpoint.is_shutdown() )
return true; return true;
@ -1260,7 +1260,7 @@ bool bro_broker::Manager::ProcessLogCreate(broker::zeek::LogCreate lc)
} }
auto num_fields = fields_data->size(); auto num_fields = fields_data->size();
auto fields = new threading::Field* [num_fields]; auto fields = new zeek::threading::Field* [num_fields];
for ( size_t i = 0; i < num_fields; ++i ) for ( size_t i = 0; i < num_fields; ++i )
{ {
@ -1344,11 +1344,11 @@ bool bro_broker::Manager::ProcessLogWrite(broker::zeek::LogWrite lw)
return false; return false;
} }
auto vals = new threading::Value* [num_fields]; auto vals = new zeek::threading::Value* [num_fields];
for ( int i = 0; i < num_fields; ++i ) for ( int i = 0; i < num_fields; ++i )
{ {
vals[i] = new threading::Value; vals[i] = new zeek::threading::Value;
if ( ! vals[i]->Read(&fmt) ) if ( ! vals[i]->Read(&fmt) )
{ {

View file

@ -189,7 +189,7 @@ public:
bool PublishLogCreate(zeek::EnumVal* stream, zeek::EnumVal* writer, bool PublishLogCreate(zeek::EnumVal* stream, zeek::EnumVal* writer,
const zeek::logging::WriterBackend::WriterInfo& info, const zeek::logging::WriterBackend::WriterInfo& info,
int num_fields, int num_fields,
const threading::Field* const * fields, const zeek::threading::Field* const * fields,
const broker::endpoint_info& peer = NoPeer); const broker::endpoint_info& peer = NoPeer);
/** /**
@ -205,7 +205,7 @@ public:
*/ */
bool PublishLogWrite(zeek::EnumVal* stream, zeek::EnumVal* writer, bool PublishLogWrite(zeek::EnumVal* stream, zeek::EnumVal* writer,
std::string path, int num_vals, std::string path, int num_vals,
const threading::Value* const * vals); const zeek::threading::Value* const * vals);
/** /**
* Automatically send an event to any interested peers whenever it is * Automatically send an event to any interested peers whenever it is

View file

@ -23,8 +23,8 @@
#include "../threading/SerialTypes.h" #include "../threading/SerialTypes.h"
using namespace std; using namespace std;
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::input { namespace zeek::input {

View file

@ -4,8 +4,8 @@
#include "ReaderFrontend.h" #include "ReaderFrontend.h"
#include "Manager.h" #include "Manager.h"
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::input { namespace zeek::input {

View file

@ -12,10 +12,9 @@
#include "threading/SerialTypes.h" #include "threading/SerialTypes.h"
using namespace threading;
using namespace std; using namespace std;
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::input::reader::detail { namespace zeek::input::reader::detail {
@ -114,8 +113,8 @@ bool Ascii::DoInit(const ReaderInfo& info, int num_fields, const Field* const* f
if ( set_separator.size() != 1 ) if ( set_separator.size() != 1 )
Error("set_separator length has to be 1. Separator will be truncated."); Error("set_separator length has to be 1. Separator will be truncated.");
formatter::Ascii::SeparatorInfo sep_info(separator, set_separator, unset_field, empty_field); zeek::threading::formatter::Ascii::SeparatorInfo sep_info(separator, set_separator, unset_field, empty_field);
formatter = unique_ptr<threading::formatter::Formatter>(new formatter::Ascii(this, sep_info)); formatter = unique_ptr<zeek::threading::Formatter>(new zeek::threading::formatter::Ascii(this, sep_info));
return DoUpdate(); return DoUpdate();
} }
@ -368,7 +367,7 @@ bool Ascii::DoUpdate()
if ( ! fit->present ) if ( ! fit->present )
{ {
// add non-present field // add non-present field
fields[fpos] = new Value((*fit).type, false); fields[fpos] = new Value((*fit).type, false);
fpos++; fpos++;
continue; continue;
} }

View file

@ -82,7 +82,7 @@ private:
bool fail_on_file_problem; bool fail_on_file_problem;
std::string path_prefix; std::string path_prefix;
std::unique_ptr<threading::formatter::Formatter> formatter; std::unique_ptr<zeek::threading::Formatter> formatter;
}; };
} // namespace zeek::input::reader::detail } // namespace zeek::input::reader::detail

View file

@ -29,7 +29,7 @@ Benchmark::Benchmark(zeek::input::ReaderFrontend *frontend) : zeek::input::Reade
heartbeatstarttime = 0; heartbeatstarttime = 0;
heartbeat_interval = double(zeek::BifConst::Threading::heartbeat_interval); heartbeat_interval = double(zeek::BifConst::Threading::heartbeat_interval);
ascii = new threading::formatter::Ascii(this, threading::formatter::Ascii::SeparatorInfo()); ascii = new zeek::threading::formatter::Ascii(this, zeek::threading::formatter::Ascii::SeparatorInfo());
} }
Benchmark::~Benchmark() Benchmark::~Benchmark()
@ -43,7 +43,7 @@ void Benchmark::DoClose()
{ {
} }
bool Benchmark::DoInit(const ReaderInfo& info, int num_fields, const Field* const* fields) bool Benchmark::DoInit(const ReaderInfo& info, int num_fields, const zeek::threading::Field* const* fields)
{ {
num_lines = atoi(info.source); num_lines = atoi(info.source);
@ -88,7 +88,7 @@ bool Benchmark::DoUpdate()
int linestosend = num_lines * heartbeat_interval; int linestosend = num_lines * heartbeat_interval;
for ( int i = 0; i < linestosend; i++ ) for ( int i = 0; i < linestosend; i++ )
{ {
Value** field = new Value*[NumFields()]; zeek::threading::Value** field = new zeek::threading::Value*[NumFields()];
for (int j = 0; j < NumFields(); j++ ) for (int j = 0; j < NumFields(); j++ )
field[j] = EntryToVal(Fields()[j]->type, Fields()[j]->subtype); field[j] = EntryToVal(Fields()[j]->type, Fields()[j]->subtype);
@ -124,9 +124,9 @@ bool Benchmark::DoUpdate()
return true; return true;
} }
threading::Value* Benchmark::EntryToVal(zeek::TypeTag type, zeek::TypeTag subtype) zeek::threading::Value* Benchmark::EntryToVal(zeek::TypeTag type, zeek::TypeTag subtype)
{ {
Value* val = new Value(type, subtype, true); auto* val = new zeek::threading::Value(type, subtype, true);
// basically construct something random from the fields that we want. // basically construct something random from the fields that we want.
@ -189,7 +189,7 @@ threading::Value* Benchmark::EntryToVal(zeek::TypeTag type, zeek::TypeTag subtyp
// how many entries do we have... // how many entries do we have...
unsigned int length = random() / (RAND_MAX / 15); unsigned int length = random() / (RAND_MAX / 15);
Value** lvals = new Value* [length]; zeek::threading::Value** lvals = new zeek::threading::Value* [length];
if ( type == zeek::TYPE_TABLE ) if ( type == zeek::TYPE_TABLE )
{ {
@ -209,7 +209,7 @@ threading::Value* Benchmark::EntryToVal(zeek::TypeTag type, zeek::TypeTag subtyp
for ( unsigned int pos = 0; pos < length; pos++ ) for ( unsigned int pos = 0; pos < length; pos++ )
{ {
Value* newval = EntryToVal(subtype, zeek::TYPE_ENUM); zeek::threading::Value* newval = EntryToVal(subtype, zeek::TYPE_ENUM);
if ( newval == nullptr ) if ( newval == nullptr )
{ {
Error("Error while reading set"); Error("Error while reading set");
@ -250,10 +250,10 @@ bool Benchmark::DoHeartbeat(double network_time, double current_time)
if ( multiplication_factor != 1 || add != 0 ) if ( multiplication_factor != 1 || add != 0 )
{ {
// we have to document at what time we changed the factor to what value. // we have to document at what time we changed the factor to what value.
Value** v = new Value*[2]; zeek::threading::Value** v = new zeek::threading::Value*[2];
v[0] = new Value(zeek::TYPE_COUNT, true); v[0] = new zeek::threading::Value(zeek::TYPE_COUNT, true);
v[0]->val.uint_val = num_lines; v[0]->val.uint_val = num_lines;
v[1] = new Value(zeek::TYPE_TIME, true); v[1] = new zeek::threading::Value(zeek::TYPE_TIME, true);
v[1]->val.double_val = CurrTime(); v[1]->val.double_val = CurrTime();
SendEvent("lines_changed", 2, v); SendEvent("lines_changed", 2, v);

View file

@ -39,7 +39,7 @@ private:
double timedspread; double timedspread;
double heartbeat_interval; double heartbeat_interval;
threading::formatter::Ascii* ascii; zeek::threading::formatter::Ascii* ascii;
}; };
} // namespace zeek::input::reader } // namespace zeek::input::reader

View file

@ -8,8 +8,8 @@
#include "threading/SerialTypes.h" #include "threading/SerialTypes.h"
using namespace std; using namespace std;
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::input::reader::detail { namespace zeek::input::reader::detail {

View file

@ -15,9 +15,8 @@
#include "input/Manager.h" #include "input/Manager.h"
#include "threading/SerialTypes.h" #include "threading/SerialTypes.h"
using namespace threading; using zeek::threading::Value;
using threading::Value; using zeek::threading::Field;
using threading::Field;
namespace zeek::input::reader::detail { namespace zeek::input::reader::detail {
@ -72,8 +71,9 @@ bool Config::DoInit(const ReaderInfo& info, int num_fields, const Field* const*
empty_field.assign( (const char*) zeek::BifConst::InputConfig::empty_field->Bytes(), empty_field.assign( (const char*) zeek::BifConst::InputConfig::empty_field->Bytes(),
zeek::BifConst::InputConfig::empty_field->Len()); zeek::BifConst::InputConfig::empty_field->Len());
formatter::Ascii::SeparatorInfo sep_info("\t", set_separator, "", empty_field); zeek::threading::formatter::Ascii::SeparatorInfo sep_info("\t", set_separator, "", empty_field);
formatter = std::unique_ptr<threading::formatter::Formatter>(new formatter::Ascii(this, sep_info)); formatter = std::unique_ptr<zeek::threading::Formatter>(
new zeek::threading::formatter::Ascii(this, sep_info));
return DoUpdate(); return DoUpdate();
} }

View file

@ -49,7 +49,7 @@ private:
std::string set_separator; std::string set_separator;
std::string empty_field; std::string empty_field;
std::unique_ptr<threading::formatter::Formatter> formatter; std::unique_ptr<zeek::threading::Formatter> formatter;
std::unordered_map<std::string, std::tuple<zeek::TypeTag, zeek::TypeTag>> option_types; std::unordered_map<std::string, std::tuple<zeek::TypeTag, zeek::TypeTag>> option_types;
std::unordered_map<std::string, std::string> option_values; std::unordered_map<std::string, std::string> option_values;
}; };

View file

@ -20,8 +20,8 @@ extern "C" {
#include "setsignal.h" #include "setsignal.h"
} }
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::input::reader::detail { namespace zeek::input::reader::detail {

View file

@ -15,8 +15,8 @@
#include "threading/SerialTypes.h" #include "threading/SerialTypes.h"
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::input::reader::detail { namespace zeek::input::reader::detail {
@ -39,8 +39,8 @@ SQLite::SQLite(zeek::input::ReaderFrontend *frontend)
zeek::BifConst::InputSQLite::empty_field->Len() zeek::BifConst::InputSQLite::empty_field->Len()
); );
io = new threading::formatter::Ascii(this, threading::formatter::Ascii::SeparatorInfo(std::string(), io = new zeek::threading::formatter::Ascii(this, zeek::threading::formatter::Ascii::SeparatorInfo(
set_separator, unset_field, empty_field)); std::string(), set_separator, unset_field, empty_field));
} }
SQLite::~SQLite() SQLite::~SQLite()

View file

@ -21,7 +21,7 @@ public:
static zeek::input::ReaderBackend* Instantiate(zeek::input::ReaderFrontend* frontend) { return new SQLite(frontend); } static zeek::input::ReaderBackend* Instantiate(zeek::input::ReaderFrontend* frontend) { return new SQLite(frontend); }
protected: protected:
bool DoInit(const ReaderInfo& info, int arg_num_fields, const threading::Field* const* arg_fields) override; bool DoInit(const ReaderInfo& info, int arg_num_fields, const zeek::threading::Field* const* arg_fields) override;
void DoClose() override; void DoClose() override;
bool DoUpdate() override; bool DoUpdate() override;
bool DoHeartbeat(double network_time, double current_time) override { return true; } bool DoHeartbeat(double network_time, double current_time) override { return true; }
@ -29,7 +29,7 @@ protected:
private: private:
bool checkError(int code); bool checkError(int code);
threading::Value* EntryToVal(sqlite3_stmt *st, const threading::Field *field, int pos, int subpos); zeek::threading::Value* EntryToVal(sqlite3_stmt *st, const zeek::threading::Field *field, int pos, int subpos);
const threading::Field* const * fields; // raw mapping const threading::Field* const * fields; // raw mapping
unsigned int num_fields; unsigned int num_fields;
@ -38,7 +38,7 @@ private:
std::string query; std::string query;
sqlite3 *db; sqlite3 *db;
sqlite3_stmt *st; sqlite3_stmt *st;
threading::formatter::Ascii* io; zeek::threading::formatter::Ascii* io;
std::string set_separator; std::string set_separator;
std::string unset_field; std::string unset_field;

View file

@ -11,8 +11,8 @@
// Messages sent from backend to frontend (i.e., "OutputMessages"). // Messages sent from backend to frontend (i.e., "OutputMessages").
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::logging { namespace zeek::logging {

View file

@ -7,18 +7,18 @@
#include "WriterFrontend.h" #include "WriterFrontend.h"
#include "WriterBackend.h" #include "WriterBackend.h"
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::logging { namespace zeek::logging {
// Messages sent from frontend to backend (i.e., "InputMessages"). // Messages sent from frontend to backend (i.e., "InputMessages").
class InitMessage final : public threading::InputMessage<WriterBackend> class InitMessage final : public zeek::threading::InputMessage<WriterBackend>
{ {
public: public:
InitMessage(WriterBackend* backend, const int num_fields, const Field* const* fields) InitMessage(WriterBackend* backend, const int num_fields, const Field* const* fields)
: threading::InputMessage<WriterBackend>("Init", backend), : zeek::threading::InputMessage<WriterBackend>("Init", backend),
num_fields(num_fields), fields(fields) num_fields(num_fields), fields(fields)
{} {}
@ -29,12 +29,12 @@ private:
const Field * const* fields; const Field * const* fields;
}; };
class RotateMessage final : public threading::InputMessage<WriterBackend> class RotateMessage final : public zeek::threading::InputMessage<WriterBackend>
{ {
public: public:
RotateMessage(WriterBackend* backend, WriterFrontend* frontend, const char* rotated_path, const double open, RotateMessage(WriterBackend* backend, WriterFrontend* frontend, const char* rotated_path, const double open,
const double close, const bool terminating) const double close, const bool terminating)
: threading::InputMessage<WriterBackend>("Rotate", backend), : zeek::threading::InputMessage<WriterBackend>("Rotate", backend),
frontend(frontend), frontend(frontend),
rotated_path(copy_string(rotated_path)), open(open), rotated_path(copy_string(rotated_path)), open(open),
close(close), terminating(terminating) { } close(close), terminating(terminating) { }
@ -51,11 +51,11 @@ private:
const bool terminating; const bool terminating;
}; };
class WriteMessage final : public threading::InputMessage<WriterBackend> class WriteMessage final : public zeek::threading::InputMessage<WriterBackend>
{ {
public: public:
WriteMessage(WriterBackend* backend, int num_fields, int num_writes, Value*** vals) WriteMessage(WriterBackend* backend, int num_fields, int num_writes, Value*** vals)
: threading::InputMessage<WriterBackend>("Write", backend), : zeek::threading::InputMessage<WriterBackend>("Write", backend),
num_fields(num_fields), num_writes(num_writes), vals(vals) {} num_fields(num_fields), num_writes(num_writes), vals(vals) {}
bool Process() override { return Object()->Write(num_fields, num_writes, vals); } bool Process() override { return Object()->Write(num_fields, num_writes, vals); }
@ -66,11 +66,11 @@ private:
Value ***vals; Value ***vals;
}; };
class SetBufMessage final : public threading::InputMessage<WriterBackend> class SetBufMessage final : public zeek::threading::InputMessage<WriterBackend>
{ {
public: public:
SetBufMessage(WriterBackend* backend, const bool enabled) SetBufMessage(WriterBackend* backend, const bool enabled)
: threading::InputMessage<WriterBackend>("SetBuf", backend), : zeek::threading::InputMessage<WriterBackend>("SetBuf", backend),
enabled(enabled) { } enabled(enabled) { }
bool Process() override { return Object()->SetBuf(enabled); } bool Process() override { return Object()->SetBuf(enabled); }
@ -79,11 +79,11 @@ private:
const bool enabled; const bool enabled;
}; };
class FlushMessage final : public threading::InputMessage<WriterBackend> class FlushMessage final : public zeek::threading::InputMessage<WriterBackend>
{ {
public: public:
FlushMessage(WriterBackend* backend, double network_time) FlushMessage(WriterBackend* backend, double network_time)
: threading::InputMessage<WriterBackend>("Flush", backend), : zeek::threading::InputMessage<WriterBackend>("Flush", backend),
network_time(network_time) {} network_time(network_time) {}
bool Process() override { return Object()->Flush(network_time); } bool Process() override { return Object()->Flush(network_time); }

View file

@ -23,9 +23,8 @@
#include "ascii.bif.h" #include "ascii.bif.h"
using namespace std; using namespace std;
using namespace threading; using zeek::threading::Value;
using threading::Value; using zeek::threading::Field;
using threading::Field;
static constexpr auto shadow_file_prefix = ".shadow."; static constexpr auto shadow_file_prefix = ".shadow.";
@ -359,22 +358,22 @@ bool Ascii::InitFormatter()
if ( use_json ) if ( use_json )
{ {
formatter::JSON::TimeFormat tf = formatter::JSON::TS_EPOCH; zeek::threading::formatter::JSON::TimeFormat tf = zeek::threading::formatter::JSON::TS_EPOCH;
// Write out JSON formatted logs. // Write out JSON formatted logs.
if ( strcmp(json_timestamps.c_str(), "JSON::TS_EPOCH") == 0 ) if ( strcmp(json_timestamps.c_str(), "JSON::TS_EPOCH") == 0 )
tf = formatter::JSON::TS_EPOCH; tf = zeek::threading::formatter::JSON::TS_EPOCH;
else if ( strcmp(json_timestamps.c_str(), "JSON::TS_MILLIS") == 0 ) else if ( strcmp(json_timestamps.c_str(), "JSON::TS_MILLIS") == 0 )
tf = formatter::JSON::TS_MILLIS; tf = zeek::threading::formatter::JSON::TS_MILLIS;
else if ( strcmp(json_timestamps.c_str(), "JSON::TS_ISO8601") == 0 ) else if ( strcmp(json_timestamps.c_str(), "JSON::TS_ISO8601") == 0 )
tf = formatter::JSON::TS_ISO8601; tf = zeek::threading::formatter::JSON::TS_ISO8601;
else else
{ {
Error(Fmt("Invalid JSON timestamp format: %s", json_timestamps.c_str())); Error(Fmt("Invalid JSON timestamp format: %s", json_timestamps.c_str()));
return false; return false;
} }
formatter = new formatter::JSON(this, tf); formatter = new zeek::threading::formatter::JSON(this, tf);
// Using JSON implicitly turns off the header meta fields. // Using JSON implicitly turns off the header meta fields.
include_meta = false; include_meta = false;
} }
@ -387,8 +386,8 @@ bool Ascii::InitFormatter()
// Use the default "Bro logs" format. // Use the default "Bro logs" format.
desc.EnableEscaping(); desc.EnableEscaping();
desc.AddEscapeSequence(separator); desc.AddEscapeSequence(separator);
formatter::Ascii::SeparatorInfo sep_info(separator, set_separator, unset_field, empty_field); zeek::threading::formatter::Ascii::SeparatorInfo sep_info(separator, set_separator, unset_field, empty_field);
formatter = new formatter::Ascii(this, sep_info); formatter = new zeek::threading::formatter::Ascii(this, sep_info);
} }
return true; return true;
@ -424,7 +423,7 @@ void Ascii::CloseFile(double t)
gzfile = nullptr; gzfile = nullptr;
} }
bool Ascii::DoInit(const WriterInfo& info, int num_fields, const Field* const * fields) bool Ascii::DoInit(const WriterInfo& info, int num_fields, const zeek::threading::Field* const * fields)
{ {
assert(! fd); assert(! fd);
@ -596,8 +595,8 @@ bool Ascii::DoFinish(double network_time)
return true; return true;
} }
bool Ascii::DoWrite(int num_fields, const Field* const * fields, bool Ascii::DoWrite(int num_fields, const zeek::threading::Field* const * fields,
Value** vals) zeek::threading::Value** vals)
{ {
if ( ! fd ) if ( ! fd )
DoInit(Info(), NumFields(), Fields()); DoInit(Info(), NumFields(), Fields());

View file

@ -26,9 +26,9 @@ public:
protected: protected:
bool DoInit(const WriterInfo& info, int num_fields, bool DoInit(const WriterInfo& info, int num_fields,
const threading::Field* const* fields) override; const zeek::threading::Field* const* fields) override;
bool DoWrite(int num_fields, const threading::Field* const* fields, bool DoWrite(int num_fields, const zeek::threading::Field* const* fields,
threading::Value** vals) override; zeek::threading::Value** vals) override;
bool DoSetBuf(bool enabled) override; bool DoSetBuf(bool enabled) override;
bool DoRotate(const char* rotated_path, double open, bool DoRotate(const char* rotated_path, double open,
double close, bool terminating) override; double close, bool terminating) override;
@ -75,7 +75,7 @@ private:
bool enable_utf_8; bool enable_utf_8;
std::string json_timestamps; std::string json_timestamps;
threading::formatter::Formatter* formatter; zeek::threading::Formatter* formatter;
bool init_options; bool init_options;
}; };

View file

@ -12,8 +12,8 @@
#include "sqlite.bif.h" #include "sqlite.bif.h"
using namespace std; using namespace std;
using threading::Value; using zeek::threading::Value;
using threading::Field; using zeek::threading::Field;
namespace zeek::logging::writer::detail { namespace zeek::logging::writer::detail {
@ -36,8 +36,8 @@ SQLite::SQLite(zeek::logging::WriterFrontend* frontend)
zeek::BifConst::LogSQLite::empty_field->Len() zeek::BifConst::LogSQLite::empty_field->Len()
); );
threading::formatter::Ascii::SeparatorInfo sep_info(string(), set_separator, unset_field, empty_field); zeek::threading::formatter::Ascii::SeparatorInfo sep_info(string(), set_separator, unset_field, empty_field);
io = new threading::formatter::Ascii(this, sep_info); io = new zeek::threading::formatter::Ascii(this, sep_info);
} }
SQLite::~SQLite() SQLite::~SQLite()

View file

@ -49,7 +49,7 @@ private:
std::string unset_field; std::string unset_field;
std::string empty_field; std::string empty_field;
threading::formatter::Ascii* io; zeek::threading::formatter::Ascii* io;
}; };
} // namespace zeek::logging::writer::detail } // namespace zeek::logging::writer::detail

View file

@ -21,6 +21,13 @@ ZEEK_FORWARD_DECLARE_NAMESPACED(Event, zeek);
ZEEK_FORWARD_DECLARE_NAMESPACED(Func, zeek); ZEEK_FORWARD_DECLARE_NAMESPACED(Func, zeek);
ZEEK_FORWARD_DECLARE_NAMESPACED(Frame, zeek::detail); ZEEK_FORWARD_DECLARE_NAMESPACED(Frame, zeek::detail);
namespace zeek::threading {
struct Field;
}
namespace threading {
using Field [[deprecated("Remove in v4.1. Use zeek::threading::Field.")]] = zeek::threading::Field;
}
namespace zeek { namespace zeek {
template <class T> class IntrusivePtr; template <class T> class IntrusivePtr;
using ValPtr = zeek::IntrusivePtr<Val>; using ValPtr = zeek::IntrusivePtr<Val>;
@ -29,10 +36,6 @@ class Obj;
using BroObj [[deprecated("Remove in v4.1. Use zeek::Obj instead.")]] = zeek::Obj; using BroObj [[deprecated("Remove in v4.1. Use zeek::Obj instead.")]] = zeek::Obj;
namespace threading {
struct Field;
}
namespace zeek::plugin { namespace zeek::plugin {
class Manager; class Manager;

View file

@ -339,7 +339,7 @@ function get_thread_stats%(%): ThreadStats
auto r = zeek::make_intrusive<zeek::RecordVal>(ThreadStats); auto r = zeek::make_intrusive<zeek::RecordVal>(ThreadStats);
int n = 0; int n = 0;
r->Assign(n++, zeek::val_mgr->Count(thread_mgr->NumThreads())); r->Assign(n++, zeek::val_mgr->Count(zeek::thread_mgr->NumThreads()));
return r; return r;
%} %}

View file

@ -7,7 +7,7 @@
#include "Manager.h" #include "Manager.h"
#include "util.h" #include "util.h"
using namespace threading; namespace zeek::threading {
static const int STD_FMT_BUF_LEN = 2048; static const int STD_FMT_BUF_LEN = 2048;
@ -193,3 +193,5 @@ void* BasicThread::launcher(void *arg)
return nullptr; return nullptr;
} }
} // namespace zeek::threading

View file

@ -1,14 +1,16 @@
#pragma once #pragma once
#include "zeek-config.h"
#include <stdint.h> #include <stdint.h>
#include <iosfwd> #include <iosfwd>
#include <thread> #include <thread>
namespace threading { ZEEK_FORWARD_DECLARE_NAMESPACED(Manager, zeek, threading);
class Manager; namespace zeek::threading {
/** /**
* Base class for all threads. * Base class for all threads.
@ -211,4 +213,8 @@ private:
static uint64_t thread_counter; static uint64_t thread_counter;
}; };
} // namespace zeek::threading
namespace threading {
using BasicThread [[deprecated("Remove in v4.1. Use zeek::threading::BasicThread.")]] = zeek::threading::BasicThread;
} }

View file

@ -8,10 +8,10 @@
#include "MsgThread.h" #include "MsgThread.h"
#include "bro_inet_ntop.h" #include "bro_inet_ntop.h"
using namespace threading; using zeek::threading::Value;
using namespace formatter; using zeek::threading::Field;
using threading::Value;
using threading::Field; namespace zeek::threading {
Formatter::Formatter(threading::MsgThread* t) Formatter::Formatter(threading::MsgThread* t)
{ {
@ -125,3 +125,5 @@ std::string Formatter::Render(TransportProto proto)
else else
return "unknown"; return "unknown";
} }
} // namespace zeek::threading::formatter

View file

@ -7,11 +7,9 @@
#include "Type.h" #include "Type.h"
#include "SerialTypes.h" #include "SerialTypes.h"
namespace threading { ZEEK_FORWARD_DECLARE_NAMESPACED(MsgThread, zeek, threading);
class MsgThread; namespace zeek::threading {
namespace formatter {
/** /**
* A thread-safe class for converting values into some textual format. This * A thread-safe class for converting values into some textual format. This
@ -164,4 +162,8 @@ private:
threading::MsgThread* thread; threading::MsgThread* thread;
}; };
}} } // zeek::threading
namespace threading::formatter {
using Formatter [[deprecated("Remove in v4.1. Use zeek::threading::Formatter.")]] = zeek::threading::Formatter;
}

View file

@ -8,7 +8,8 @@
#include "Event.h" #include "Event.h"
#include "IPAddr.h" #include "IPAddr.h"
using namespace threading; namespace zeek::threading {
namespace detail {
void HeartbeatTimer::Dispatch(double t, bool is_expire) void HeartbeatTimer::Dispatch(double t, bool is_expire)
{ {
@ -19,6 +20,8 @@ void HeartbeatTimer::Dispatch(double t, bool is_expire)
thread_mgr->StartHeartbeatTimer(); thread_mgr->StartHeartbeatTimer();
} }
} // namespace detail
Manager::Manager() Manager::Manager()
{ {
DBG_LOG(zeek::DBG_THREADING, "Creating thread manager ..."); DBG_LOG(zeek::DBG_THREADING, "Creating thread manager ...");
@ -127,7 +130,7 @@ void Manager::SendHeartbeats()
void Manager::StartHeartbeatTimer() void Manager::StartHeartbeatTimer()
{ {
heartbeat_timer_running = true; heartbeat_timer_running = true;
zeek::detail::timer_mgr->Add(new HeartbeatTimer(network_time + zeek::BifConst::Threading::heartbeat_interval)); zeek::detail::timer_mgr->Add(new detail::HeartbeatTimer(network_time + zeek::BifConst::Threading::heartbeat_interval));
} }
// Raise everything in here as warnings so it is passed to scriptland without // Raise everything in here as warnings so it is passed to scriptland without
@ -269,3 +272,5 @@ const threading::Manager::msg_stats_list& threading::Manager::GetMsgThreadStats(
return stats; return stats;
} }
} // namespace zeek::threading

View file

@ -7,7 +7,9 @@
#include <list> #include <list>
#include <utility> #include <utility>
namespace zeek {
namespace threading { namespace threading {
namespace detail {
class HeartbeatTimer final : public zeek::detail::Timer { class HeartbeatTimer final : public zeek::detail::Timer {
public: public:
@ -21,6 +23,8 @@ protected:
void Init(); void Init();
}; };
} // namespace detail
/** /**
* The thread manager coordinates all child threads. Once a BasicThread is * The thread manager coordinates all child threads. Once a BasicThread is
* instantitated, it gets addedd to the manager, which will delete it later * instantitated, it gets addedd to the manager, which will delete it later
@ -103,7 +107,7 @@ public:
protected: protected:
friend class BasicThread; friend class BasicThread;
friend class MsgThread; friend class MsgThread;
friend class HeartbeatTimer; friend class detail::HeartbeatTimer;
/** /**
* Registers a new basic thread with the manager. This is * Registers a new basic thread with the manager. This is
@ -151,10 +155,21 @@ private:
bool heartbeat_timer_running = false; bool heartbeat_timer_running = false;
}; };
} } // namespace threading
/** /**
* A singleton instance of the thread manager. All methods must only be * A singleton instance of the thread manager. All methods must only be
* called from Bro's main thread. * called from Zeek's main thread.
*/ */
extern threading::Manager* thread_mgr; extern threading::Manager* thread_mgr;
} // namespace zeek
extern zeek::threading::Manager*& thread_mgr [[deprecated("Remove in v4.1. Use zeek::thread_mgr.")]];
namespace threading::detail {
using HeartbeatTimer [[deprecated("Remove in v4.1. Use zeek::threading::detail::HeartbeatTimer.")]] = zeek::threading::detail::HeartbeatTimer;
}
namespace threading {
using Manager [[deprecated("Remove in v4.1. Use zeek::threading::Manager.")]] = zeek::threading::Manager;
}

View file

@ -8,9 +8,11 @@
#include "Manager.h" #include "Manager.h"
#include "iosource/Manager.h" #include "iosource/Manager.h"
using namespace threading; // Set by Zeek's main signal handler.
extern int signal_val;
namespace threading { namespace zeek::threading {
namespace detail {
////// Messages. ////// Messages.
@ -124,8 +126,6 @@ private:
}; };
#endif #endif
}
// An event that the child wants to pass into the main event queue // An event that the child wants to pass into the main event queue
class SendEventMessage final : public OutputMessage<MsgThread> { class SendEventMessage final : public OutputMessage<MsgThread> {
public: public:
@ -151,13 +151,6 @@ private:
Value* *val; Value* *val;
}; };
////// Methods.
Message::~Message()
{
delete [] name;
}
bool ReporterMessage::Process() bool ReporterMessage::Process()
{ {
switch ( type ) { switch ( type ) {
@ -197,6 +190,15 @@ bool ReporterMessage::Process()
return true; return true;
} }
} // namespace detail
////// Methods.
Message::~Message()
{
delete [] name;
}
MsgThread::MsgThread() : BasicThread(), queue_in(this, nullptr), queue_out(nullptr, this) MsgThread::MsgThread() : BasicThread(), queue_in(this, nullptr), queue_out(nullptr, this)
{ {
cnt_sent_in = cnt_sent_out = 0; cnt_sent_in = cnt_sent_out = 0;
@ -219,9 +221,6 @@ MsgThread::~MsgThread()
zeek::iosource_mgr->UnregisterFd(flare.FD(), this); zeek::iosource_mgr->UnregisterFd(flare.FD(), this);
} }
// Set by Bro's main signal handler.
extern int signal_val;
void MsgThread::OnSignalStop() void MsgThread::OnSignalStop()
{ {
if ( main_finished || Killed() || child_sent_finish ) if ( main_finished || Killed() || child_sent_finish )
@ -229,7 +228,7 @@ void MsgThread::OnSignalStop()
child_sent_finish = true; child_sent_finish = true;
// Signal thread to terminate. // Signal thread to terminate.
SendIn(new FinishMessage(this, network_time), true); SendIn(new detail::FinishMessage(this, network_time), true);
} }
void MsgThread::OnWaitForStop() void MsgThread::OnWaitForStop()
@ -303,43 +302,43 @@ void MsgThread::Heartbeat()
if ( child_sent_finish ) if ( child_sent_finish )
return; return;
SendIn(new HeartbeatMessage(this, network_time, current_time())); SendIn(new detail::HeartbeatMessage(this, network_time, current_time()));
} }
void MsgThread::Finished() void MsgThread::Finished()
{ {
child_finished = true; child_finished = true;
SendOut(new FinishedMessage(this)); SendOut(new detail::FinishedMessage(this));
} }
void MsgThread::Info(const char* msg) void MsgThread::Info(const char* msg)
{ {
SendOut(new ReporterMessage(ReporterMessage::INFO, this, msg)); SendOut(new detail::ReporterMessage(detail::ReporterMessage::INFO, this, msg));
} }
void MsgThread::Warning(const char* msg) void MsgThread::Warning(const char* msg)
{ {
SendOut(new ReporterMessage(ReporterMessage::WARNING, this, msg)); SendOut(new detail::ReporterMessage(detail::ReporterMessage::WARNING, this, msg));
} }
void MsgThread::Error(const char* msg) void MsgThread::Error(const char* msg)
{ {
SendOut(new ReporterMessage(ReporterMessage::ERROR, this, msg)); SendOut(new detail::ReporterMessage(detail::ReporterMessage::ERROR, this, msg));
} }
void MsgThread::FatalError(const char* msg) void MsgThread::FatalError(const char* msg)
{ {
SendOut(new ReporterMessage(ReporterMessage::FATAL_ERROR, this, msg)); SendOut(new detail::ReporterMessage(detail::ReporterMessage::FATAL_ERROR, this, msg));
} }
void MsgThread::FatalErrorWithCore(const char* msg) void MsgThread::FatalErrorWithCore(const char* msg)
{ {
SendOut(new ReporterMessage(ReporterMessage::FATAL_ERROR_WITH_CORE, this, msg)); SendOut(new detail::ReporterMessage(detail::ReporterMessage::FATAL_ERROR_WITH_CORE, this, msg));
} }
void MsgThread::InternalWarning(const char* msg) void MsgThread::InternalWarning(const char* msg)
{ {
SendOut(new ReporterMessage(ReporterMessage::INTERNAL_WARNING, this, msg)); SendOut(new detail::ReporterMessage(detail::ReporterMessage::INTERNAL_WARNING, this, msg));
} }
void MsgThread::InternalError(const char* msg) void MsgThread::InternalError(const char* msg)
@ -353,7 +352,7 @@ void MsgThread::InternalError(const char* msg)
void MsgThread::Debug(zeek::DebugStream stream, const char* msg) void MsgThread::Debug(zeek::DebugStream stream, const char* msg)
{ {
SendOut(new DebugMessage(stream, this, msg)); SendOut(new detail::DebugMessage(stream, this, msg));
} }
#endif #endif
@ -390,7 +389,7 @@ void MsgThread::SendOut(BasicOutputMessage* msg, bool force)
void MsgThread::SendEvent(const char* name, const int num_vals, Value* *vals) void MsgThread::SendEvent(const char* name, const int num_vals, Value* *vals)
{ {
SendOut(new SendEventMessage(this, name, num_vals, vals)); SendOut(new detail::SendEventMessage(this, name, num_vals, vals));
} }
BasicOutputMessage* MsgThread::RetrieveOut() BasicOutputMessage* MsgThread::RetrieveOut()
@ -440,7 +439,7 @@ void MsgThread::Run()
// after all other outgoing messages (in particular // after all other outgoing messages (in particular
// error messages have been processed by then main // error messages have been processed by then main
// thread). // thread).
SendOut(new KillMeMessage(this)); SendOut(new detail::KillMeMessage(this));
failed = true; failed = true;
} }
} }
@ -483,3 +482,5 @@ void MsgThread::Process()
delete msg; delete msg;
} }
} }
} // namespace zeek::threading

View file

@ -8,13 +8,29 @@
#include "iosource/IOSource.h" #include "iosource/IOSource.h"
#include "Flare.h" #include "Flare.h"
namespace zeek::threading {
struct Value;
struct Field;
}
namespace threading { namespace threading {
using Value [[deprecated("Remove in v4.1. Use zeek::threading::Value.")]] = zeek::threading::Value;
using Field [[deprecated("Remove in v4.1. Use zeek::threading::Field.")]] = zeek::threading::Field;
}
namespace zeek::threading {
class BasicInputMessage; class BasicInputMessage;
class BasicOutputMessage; class BasicOutputMessage;
namespace detail {
// These classes are marked as friends later so they need to be forward declared.
class HeartbeatMessage; class HeartbeatMessage;
struct Value; class FinishMessage;
struct Field; class FinishedMessage;
class KillMeMessage;
}
/** /**
* A specialized thread that provides bi-directional message passing between * A specialized thread that provides bi-directional message passing between
@ -205,10 +221,10 @@ public:
protected: protected:
friend class Manager; friend class Manager;
friend class HeartbeatMessage; friend class detail::HeartbeatMessage;
friend class FinishMessage; friend class detail::FinishMessage;
friend class FinishedMessage; friend class detail::FinishedMessage;
friend class KillMeMessage; friend class detail::KillMeMessage;
/** /**
* Pops a message sent by the child from the child-to-main queue. * Pops a message sent by the child from the child-to-main queue.
@ -464,4 +480,13 @@ private:
O* object; O* object;
}; };
} } // namespace zeek::threading
namespace threading {
using MsgThread [[deprecated("Remove in v4.1. Use zeek::threading::MsgThread.")]] = zeek::threading::MsgThread;
using Message [[deprecated("Remove in v4.1. Use zeek::threading::Message.")]] = zeek::threading::Message;
using BasicInputMessage [[deprecated("Remove in v4.1. Use zeek::threading::BasicInputMessage.")]] = zeek::threading::BasicInputMessage;
using BasicOutputMessage [[deprecated("Remove in v4.1. Use zeek::threading::BasicOutputMessage.")]] = zeek::threading::BasicOutputMessage;
template<typename O> using InputMessage [[deprecated("Remove in v4.1. Use zeek::threading::InputMessage.")]] = zeek::threading::InputMessage<O>;
template<typename O> using OutputMessage [[deprecated("Remove in v4.1. Use zeek::threading::OutputMessage.")]] = zeek::threading::OutputMessage<O>;
} // namespace threading

View file

@ -12,7 +12,7 @@
#undef Queue // Defined elsewhere unfortunately. #undef Queue // Defined elsewhere unfortunately.
namespace threading { namespace zeek::threading {
/** /**
* A thread-safe single-reader single-writer queue. * A thread-safe single-reader single-writer queue.
@ -261,4 +261,8 @@ inline void Queue<T>::WakeUp()
} }
} }
} // namespace zeek::threading
namespace threading {
template<typename T> using Queue [[deprecated("Remove in v4.1. Use zeek::threading::Queue.")]] = zeek::threading::Queue<T>;
} }

View file

@ -14,7 +14,7 @@
#include "Scope.h" #include "Scope.h"
#include "IPAddr.h" #include "IPAddr.h"
using namespace threading; namespace zeek::threading {
bool Field::Read(zeek::detail::SerializationFormat* fmt) bool Field::Read(zeek::detail::SerializationFormat* fmt)
{ {
@ -636,3 +636,5 @@ zeek::Val* Value::ValueToVal(const std::string& source, const Value* val, bool&
assert(false); assert(false);
return nullptr; return nullptr;
} }
} // namespace zeek::threading

View file

@ -11,7 +11,7 @@
ZEEK_FORWARD_DECLARE_NAMESPACED(SerializationFormat, zeek::detail); ZEEK_FORWARD_DECLARE_NAMESPACED(SerializationFormat, zeek::detail);
namespace threading { namespace zeek::threading {
/** /**
* Definition of a log file, i.e., one column of a log stream. * Definition of a log file, i.e., one column of a log stream.
@ -214,4 +214,9 @@ private:
Value(const Value& other) = delete; Value(const Value& other) = delete;
}; };
} // namespace zeek::threading
namespace threading {
using Field [[deprecated("Remove in v4.1. Use zeek::threading::Field.")]] = zeek::threading::Field;
using Value [[deprecated("Remove in v4.1. Use zeek::threading::Value.")]] = zeek::threading::Value;
} }

View file

@ -10,7 +10,8 @@
#include <errno.h> #include <errno.h>
using namespace std; using namespace std;
using namespace threading::formatter;
namespace zeek::threading::formatter {
// If the value we'd write out would match exactly the a reserved string, we // If the value we'd write out would match exactly the a reserved string, we
// escape the first character so that the output won't be ambigious. If this // escape the first character so that the output won't be ambigious. If this
@ -47,7 +48,7 @@ Ascii::SeparatorInfo::SeparatorInfo(const string& arg_separator,
empty_field = arg_empty_field; empty_field = arg_empty_field;
} }
Ascii::Ascii(threading::MsgThread* t, const SeparatorInfo& info) : Formatter(t) Ascii::Ascii(zeek::threading::MsgThread* t, const SeparatorInfo& info) : zeek::threading::Formatter(t)
{ {
separators = info; separators = info;
} }
@ -56,8 +57,8 @@ Ascii::~Ascii()
{ {
} }
bool Ascii::Describe(zeek::ODesc* desc, int num_fields, const threading::Field* const * fields, bool Ascii::Describe(zeek::ODesc* desc, int num_fields, const zeek::threading::Field* const * fields,
threading::Value** vals) const zeek::threading::Value** vals) const
{ {
for ( int i = 0; i < num_fields; i++ ) for ( int i = 0; i < num_fields; i++ )
{ {
@ -71,7 +72,7 @@ bool Ascii::Describe(zeek::ODesc* desc, int num_fields, const threading::Field*
return true; return true;
} }
bool Ascii::Describe(zeek::ODesc* desc, threading::Value* val, const string& name) const bool Ascii::Describe(zeek::ODesc* desc, zeek::threading::Value* val, const string& name) const
{ {
if ( ! val->present ) if ( ! val->present )
{ {
@ -207,12 +208,12 @@ bool Ascii::Describe(zeek::ODesc* desc, threading::Value* val, const string& nam
} }
threading::Value* Ascii::ParseValue(const string& s, const string& name, zeek::TypeTag type, zeek::TypeTag subtype) const zeek::threading::Value* Ascii::ParseValue(const string& s, const string& name, zeek::TypeTag type, zeek::TypeTag subtype) const
{ {
if ( ! separators.unset_field.empty() && s.compare(separators.unset_field) == 0 ) // field is not set... if ( ! separators.unset_field.empty() && s.compare(separators.unset_field) == 0 ) // field is not set...
return new threading::Value(type, false); return new zeek::threading::Value(type, false);
threading::Value* val = new threading::Value(type, subtype, true); zeek::threading::Value* val = new zeek::threading::Value(type, subtype, true);
const char* start = s.c_str(); const char* start = s.c_str();
char* end = nullptr; char* end = nullptr;
errno = 0; errno = 0;
@ -373,7 +374,7 @@ threading::Value* Ascii::ParseValue(const string& s, const string& name, zeek::T
if ( separators.empty_field.empty() && s.empty() ) if ( separators.empty_field.empty() && s.empty() )
length = 0; length = 0;
threading::Value** lvals = new threading::Value* [length]; zeek::threading::Value** lvals = new zeek::threading::Value* [length];
if ( type == zeek::TYPE_TABLE ) if ( type == zeek::TYPE_TABLE )
{ {
@ -409,7 +410,7 @@ threading::Value* Ascii::ParseValue(const string& s, const string& name, zeek::T
break; break;
} }
threading::Value* newval = ParseValue(element, name, subtype); zeek::threading::Value* newval = ParseValue(element, name, subtype);
if ( newval == nullptr ) if ( newval == nullptr )
{ {
GetThread()->Warning("Error while reading set or vector"); GetThread()->Warning("Error while reading set or vector");
@ -474,7 +475,7 @@ parse_error:
bool Ascii::CheckNumberError(const char* start, const char* end) const bool Ascii::CheckNumberError(const char* start, const char* end) const
{ {
threading::MsgThread* thread = GetThread(); zeek::threading::MsgThread* thread = GetThread();
if ( end == start && *end != '\0' ) { if ( end == start && *end != '\0' ) {
thread->Warning(thread->Fmt("String '%s' contained no parseable number", start)); thread->Warning(thread->Fmt("String '%s' contained no parseable number", start));
@ -504,3 +505,5 @@ bool Ascii::CheckNumberError(const char* start, const char* end) const
return false; return false;
} }
} // namespace zeek::threading::formatter

View file

@ -4,9 +4,9 @@
#include "../Formatter.h" #include "../Formatter.h"
namespace threading { namespace formatter { namespace zeek::threading::formatter {
class Ascii final : public Formatter { class Ascii final : public zeek::threading::Formatter {
public: public:
/** /**
* A struct to pass the necessary configuration values to the * A struct to pass the necessary configuration values to the
@ -44,14 +44,14 @@ public:
* @param info SeparatorInfo structure defining the necessary * @param info SeparatorInfo structure defining the necessary
* separators. * separators.
*/ */
Ascii(threading::MsgThread* t, const SeparatorInfo& info); Ascii(zeek::threading::MsgThread* t, const SeparatorInfo& info);
virtual ~Ascii(); virtual ~Ascii();
virtual bool Describe(zeek::ODesc* desc, threading::Value* val, const std::string& name = "") const; virtual bool Describe(zeek::ODesc* desc, zeek::threading::Value* val, const std::string& name = "") const;
virtual bool Describe(zeek::ODesc* desc, int num_fields, const threading::Field* const * fields, virtual bool Describe(zeek::ODesc* desc, int num_fields, const zeek::threading::Field* const * fields,
threading::Value** vals) const; zeek::threading::Value** vals) const;
virtual threading::Value* ParseValue(const std::string& s, const std::string& name, virtual zeek::threading::Value* ParseValue(const std::string& s, const std::string& name,
zeek::TypeTag type, zeek::TypeTag subtype = zeek::TYPE_ERROR) const; zeek::TypeTag type, zeek::TypeTag subtype = zeek::TYPE_ERROR) const;
private: private:
bool CheckNumberError(const char* start, const char* end) const; bool CheckNumberError(const char* start, const char* end) const;
@ -59,4 +59,8 @@ private:
SeparatorInfo separators; SeparatorInfo separators;
}; };
}} } // namespace zeek::threading::formatter
namespace threading::formatter {
using Ascii [[deprecated("Remove in v4.1. Use zeek::threading::formatter::Ascii.")]] = zeek::threading::formatter::Ascii;
}

View file

@ -16,7 +16,7 @@
#include <math.h> #include <math.h>
#include <stdint.h> #include <stdint.h>
using namespace threading::formatter; namespace zeek::threading::formatter {
bool JSON::NullDoubleWriter::Double(double d) bool JSON::NullDoubleWriter::Double(double d)
{ {
@ -26,7 +26,7 @@ bool JSON::NullDoubleWriter::Double(double d)
return rapidjson::Writer<rapidjson::StringBuffer>::Double(d); return rapidjson::Writer<rapidjson::StringBuffer>::Double(d);
} }
JSON::JSON(MsgThread* t, TimeFormat tf) : Formatter(t), surrounding_braces(true) JSON::JSON(zeek::threading::MsgThread* t, TimeFormat tf) : zeek::threading::Formatter(t), surrounding_braces(true)
{ {
timestamps = tf; timestamps = tf;
} }
@ -35,8 +35,8 @@ JSON::~JSON()
{ {
} }
bool JSON::Describe(zeek::ODesc* desc, int num_fields, const Field* const * fields, bool JSON::Describe(zeek::ODesc* desc, int num_fields, const zeek::threading::Field* const * fields,
Value** vals) const zeek::threading::Value** vals) const
{ {
rapidjson::StringBuffer buffer; rapidjson::StringBuffer buffer;
NullDoubleWriter writer(buffer); NullDoubleWriter writer(buffer);
@ -55,7 +55,7 @@ bool JSON::Describe(zeek::ODesc* desc, int num_fields, const Field* const * fiel
return true; return true;
} }
bool JSON::Describe(zeek::ODesc* desc, Value* val, const std::string& name) const bool JSON::Describe(zeek::ODesc* desc, zeek::threading::Value* val, const std::string& name) const
{ {
if ( desc->IsBinary() ) if ( desc->IsBinary() )
{ {
@ -78,13 +78,14 @@ bool JSON::Describe(zeek::ODesc* desc, Value* val, const std::string& name) cons
return true; return true;
} }
threading::Value* JSON::ParseValue(const std::string& s, const std::string& name, zeek::TypeTag type, zeek::TypeTag subtype) const zeek::threading::Value* JSON::ParseValue(const std::string& s, const std::string& name,
zeek::TypeTag type, zeek::TypeTag subtype) const
{ {
GetThread()->Error("JSON formatter does not support parsing yet."); GetThread()->Error("JSON formatter does not support parsing yet.");
return nullptr; return nullptr;
} }
void JSON::BuildJSON(NullDoubleWriter& writer, Value* val, const std::string& name) const void JSON::BuildJSON(NullDoubleWriter& writer, zeek::threading::Value* val, const std::string& name) const
{ {
if ( ! val->present ) if ( ! val->present )
{ {
@ -204,3 +205,5 @@ void JSON::BuildJSON(NullDoubleWriter& writer, Value* val, const std::string& na
break; break;
} }
} }
} // namespace zeek::threading::formatter

View file

@ -8,13 +8,13 @@
#include "../Formatter.h" #include "../Formatter.h"
namespace threading { namespace formatter { namespace zeek::threading::formatter {
/** /**
* A thread-safe class for converting values into a JSON representation * A thread-safe class for converting values into a JSON representation
* and vice versa. * and vice versa.
*/ */
class JSON : public Formatter { class JSON : public zeek::threading::Formatter {
public: public:
enum TimeFormat { enum TimeFormat {
TS_EPOCH, // Doubles that represents seconds from the UNIX epoch. TS_EPOCH, // Doubles that represents seconds from the UNIX epoch.
@ -22,14 +22,14 @@ public:
TS_MILLIS // Milliseconds from the UNIX epoch. Some consumers need this (e.g., elasticsearch). TS_MILLIS // Milliseconds from the UNIX epoch. Some consumers need this (e.g., elasticsearch).
}; };
JSON(threading::MsgThread* t, TimeFormat tf); JSON(zeek::threading::MsgThread* t, TimeFormat tf);
~JSON() override; ~JSON() override;
bool Describe(zeek::ODesc* desc, threading::Value* val, const std::string& name = "") const override; bool Describe(zeek::ODesc* desc, zeek::threading::Value* val, const std::string& name = "") const override;
bool Describe(zeek::ODesc* desc, int num_fields, const threading::Field* const * fields, bool Describe(zeek::ODesc* desc, int num_fields, const zeek::threading::Field* const * fields,
threading::Value** vals) const override; zeek::threading::Value** vals) const override;
threading::Value* ParseValue(const std::string& s, const std::string& name, zeek::TypeTag type, zeek::threading::Value* ParseValue(const std::string& s, const std::string& name, zeek::TypeTag type,
zeek::TypeTag subtype = zeek::TYPE_ERROR) const override; zeek::TypeTag subtype = zeek::TYPE_ERROR) const override;
class NullDoubleWriter : public rapidjson::Writer<rapidjson::StringBuffer> { class NullDoubleWriter : public rapidjson::Writer<rapidjson::StringBuffer> {
public: public:
@ -38,10 +38,14 @@ public:
}; };
private: private:
void BuildJSON(NullDoubleWriter& writer, Value* val, const std::string& name = "") const; void BuildJSON(NullDoubleWriter& writer, zeek::threading::Value* val, const std::string& name = "") const;
TimeFormat timestamps; TimeFormat timestamps;
bool surrounding_braces; bool surrounding_braces;
}; };
}} } // namespace zeek::threading::formatter
namespace threading::formatter {
using JSON [[deprecated("Remove in v4.1. Use zeek::threading::formatter::JSON.")]] = zeek::threading::formatter::JSON;
}

View file

@ -104,7 +104,8 @@ zeek::detail::TimerMgr*& timer_mgr = zeek::detail::timer_mgr;
zeek::logging::Manager* zeek::log_mgr = nullptr; zeek::logging::Manager* zeek::log_mgr = nullptr;
zeek::logging::Manager*& log_mgr = zeek::log_mgr; zeek::logging::Manager*& log_mgr = zeek::log_mgr;
threading::Manager* thread_mgr = nullptr; zeek::threading::Manager* zeek::thread_mgr = nullptr;
zeek::threading::Manager*& thread_mgr = zeek::thread_mgr;
zeek::input::Manager* zeek::input_mgr = nullptr; zeek::input::Manager* zeek::input_mgr = nullptr;
zeek::input::Manager*& input_mgr = zeek::input_mgr; zeek::input::Manager*& input_mgr = zeek::input_mgr;
zeek::file_analysis::Manager* zeek::file_mgr = nullptr; zeek::file_analysis::Manager* zeek::file_mgr = nullptr;
@ -309,7 +310,7 @@ void terminate_bro()
notifier::registry.Terminate(); notifier::registry.Terminate();
zeek::log_mgr->Terminate(); zeek::log_mgr->Terminate();
zeek::input_mgr->Terminate(); zeek::input_mgr->Terminate();
thread_mgr->Terminate(); zeek::thread_mgr->Terminate();
broker_mgr->Terminate(); broker_mgr->Terminate();
zeek::detail::dns_mgr->Terminate(); zeek::detail::dns_mgr->Terminate();
@ -485,7 +486,7 @@ zeek::detail::SetupResult zeek::detail::setup(int argc, char** argv,
zeek::val_mgr = new ValManager(); zeek::val_mgr = new ValManager();
reporter = new Reporter(options.abort_on_scripting_errors); reporter = new Reporter(options.abort_on_scripting_errors);
thread_mgr = new threading::Manager(); zeek::thread_mgr = new threading::Manager();
zeek::plugin_mgr = new zeek::plugin::Manager(); zeek::plugin_mgr = new zeek::plugin::Manager();
#ifdef DEBUG #ifdef DEBUG