Mirror/zeek
1
0
Fork 0
mirror of https://github.com/zeek/zeek.git synced 2025-10-06 16:48:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

make input framework source (hopefully) adhere to the usual indentation

Browse source 
style. No functional changes.
This commit is contained in:
Bernhard Amann 2012-05-25 10:33:22 -07:00
parent 82a6f3832a
commit 2034c10e97
6 changed files with 1102 additions and 897 deletions
Download patch file Download diff file Expand all files Collapse all files

1224
src/input/Manager.cc
View file

File diff suppressed because it is too large Load diff

148
src/input/ReaderBackend.cc
View file

@ -15,10 +15,11 @@ public:
: threading::OutputMessage<ReaderFrontend>("Put", reader),
val(val) {}
virtual bool Process() {
virtual bool Process()
{
input_mgr->Put(Object(), val);
return true;
}
}
private:
Value* *val;
@ -30,9 +31,10 @@ public:
: threading::OutputMessage<ReaderFrontend>("Delete", reader),
val(val) {}
virtual bool Process() {
virtual bool Process()
{
return input_mgr->Delete(Object(), val);
}
}
private:
Value* *val;
@ -43,10 +45,11 @@ public:
ClearMessage(ReaderFrontend* reader)
: threading::OutputMessage<ReaderFrontend>("Clear", reader) {}
virtual bool Process() {
virtual bool Process()
{
input_mgr->Clear(Object());
return true;
}
}
private:
};
@ -57,14 +60,15 @@ public:
: threading::OutputMessage<ReaderFrontend>("SendEvent", reader),
name(name), num_vals(num_vals), val(val) {}
virtual bool Process() {
virtual bool Process()
{
bool success = input_mgr->SendEvent(name, num_vals, val);
if ( !success )
reporter->Error("SendEvent for event %s failed", name.c_str());
return true; // we do not want to die if sendEvent fails because the event did not return.
}
}
private:
const string name;
@ -78,10 +82,11 @@ public:
: threading::OutputMessage<ReaderFrontend>("SendEntry", reader),
val(val) { }
virtual bool Process() {
virtual bool Process()
{
input_mgr->SendEntry(Object(), val);
return true;
}
}
private:
Value* *val;
@ -92,10 +97,11 @@ public:
EndCurrentSendMessage(ReaderFrontend* reader)
: threading::OutputMessage<ReaderFrontend>("EndCurrentSend", reader) {}
virtual bool Process() {
virtual bool Process()
{
input_mgr->EndCurrentSend(Object());
return true;
}
}
private:
};
@ -105,9 +111,10 @@ public:
ReaderClosedMessage(ReaderFrontend* reader)
: threading::OutputMessage<ReaderFrontend>("ReaderClosed", reader) {}
virtual bool Process() {
virtual bool Process()
{
return input_mgr->RemoveStreamContinuation(Object());
}
}
private:
};
@ -119,12 +126,16 @@ public:
DisableMessage(ReaderFrontend* writer)
: threading::OutputMessage<ReaderFrontend>("Disable", writer) {}
virtual bool Process() { Object()->SetDisable(); return true; }
virtual bool Process()
{
Object()->SetDisable();
return true;
}
};
ReaderBackend::ReaderBackend(ReaderFrontend* arg_frontend) : MsgThread()
{
{
buf = 0;
buf_len = 1024;
disabled = true; // disabled will be set correcty in init.
@ -132,45 +143,45 @@ ReaderBackend::ReaderBackend(ReaderFrontend* arg_frontend) : MsgThread()
frontend = arg_frontend;
SetName(frontend->Name());
}
}
ReaderBackend::~ReaderBackend()
{
}
{
}
void ReaderBackend::Put(Value* *val)
{
{
SendOut(new PutMessage(frontend, val));
}
}
void ReaderBackend::Delete(Value* *val)
{
{
SendOut(new DeleteMessage(frontend, val));
}
}
void ReaderBackend::Clear()
{
{
SendOut(new ClearMessage(frontend));
}
}
void ReaderBackend::SendEvent(const string& name, const int num_vals, Value* *vals)
{
{
SendOut(new SendEventMessage(frontend, name, num_vals, vals));
}
}
void ReaderBackend::EndCurrentSend()
{
{
SendOut(new EndCurrentSendMessage(frontend));
}
}
void ReaderBackend::SendEntry(Value* *vals)
{
{
SendOut(new SendEntryMessage(frontend, vals));
}
}
bool ReaderBackend::Init(string arg_source, int mode, const int arg_num_fields, const threading::Field* const* arg_fields)
{
bool ReaderBackend::Init(string arg_source, int mode, const int arg_num_fields,
const threading::Field* const* arg_fields)
{
source = arg_source;
SetName("InputReader/"+source);
@ -180,89 +191,90 @@ bool ReaderBackend::Init(string arg_source, int mode, const int arg_num_fields,
// disable if DoInit returns error.
int success = DoInit(arg_source, mode, arg_num_fields, arg_fields);
if ( !success ) {
if ( !success )
{
Error("Init failed");
DisableFrontend();
}
}
disabled = !success;
return success;
}
}
void ReaderBackend::Close()
{
{
DoClose();
disabled = true;
DisableFrontend();
SendOut(new ReaderClosedMessage(frontend));
if ( fields != 0 ) {
for ( unsigned int i = 0; i < num_fields; i++ ) {
if ( fields != 0 )
{
for ( unsigned int i = 0; i < num_fields; i++ )
delete(fields[i]);
}
delete[] (fields);
fields = 0;
}
}
}
bool ReaderBackend::Update()
{
{
if ( disabled )
return false;
bool success = DoUpdate();
if ( !success ) {
if ( !success )
DisableFrontend();
}
return success;
}
}
void ReaderBackend::DisableFrontend()
{
disabled = true; // we also set disabled here, because there still may be other messages queued and we will dutifully ignore these from now
{
disabled = true;
// we also set disabled here, because there still may be other messages queued and we will dutifully ignore these from now
SendOut(new DisableMessage(frontend));
}
}
bool ReaderBackend::DoHeartbeat(double network_time, double current_time)
{
{
MsgThread::DoHeartbeat(network_time, current_time);
return true;
}
TransportProto ReaderBackend::StringToProto(const string &proto) {
if ( proto == "unknown" ) {
return TRANSPORT_UNKNOWN;
} else if ( proto == "tcp" ) {
return TRANSPORT_TCP;
} else if ( proto == "udp" ) {
return TRANSPORT_UDP;
} else if ( proto == "icmp" ) {
return TRANSPORT_ICMP;
}
TransportProto ReaderBackend::StringToProto(const string &proto)
{
if ( proto == "unknown" )
return TRANSPORT_UNKNOWN;
else if ( proto == "tcp" )
return TRANSPORT_TCP;
else if ( proto == "udp" )
return TRANSPORT_UDP;
else if ( proto == "icmp" )
return TRANSPORT_ICMP;
Error(Fmt("Tried to parse invalid/unknown protocol: %s", proto.c_str()));
return TRANSPORT_UNKNOWN;
}
}
// more or less verbose copy from IPAddr.cc -- which uses reporter
Value::addr_t ReaderBackend::StringToAddr(const string &s) {
Value::addr_t ReaderBackend::StringToAddr(const string &s)
{
Value::addr_t val;
if ( s.find(':') == std::string::npos ) // IPv4.
{
val.family = IPv4;
if ( inet_aton(s.c_str(), &(val.in.in4)) <= 0 ) {
if ( inet_aton(s.c_str(), &(val.in.in4)) <= 0 )
{
Error(Fmt("Bad addres: %s", s.c_str()));
memset(&val.in.in4.s_addr, 0, sizeof(val.in.in4.s_addr));
}
}
}
@ -277,6 +289,6 @@ Value::addr_t ReaderBackend::StringToAddr(const string &s) {
}
return val;
}
}
}

40
src/input/ReaderFrontend.cc
View file

@ -46,19 +46,23 @@ public:
};
ReaderFrontend::ReaderFrontend(bro_int_t type) {
ReaderFrontend::ReaderFrontend(bro_int_t type)
{
disabled = initialized = false;
ty_name = "<not set>";
backend = input_mgr->CreateBackend(this, type);
assert(backend);
backend->Start();
}
}
ReaderFrontend::~ReaderFrontend() {
}
ReaderFrontend::~ReaderFrontend()
{
}
void ReaderFrontend::Init(string arg_source, int mode, const int num_fields, const threading::Field* const* fields) {
void ReaderFrontend::Init(string arg_source, int mode, const int num_fields,
const threading::Field* const* fields)
{
if ( disabled )
return;
@ -69,39 +73,43 @@ void ReaderFrontend::Init(string arg_source, int mode, const int num_fields, con
initialized = true;
backend->SendIn(new InitMessage(backend, arg_source, mode, num_fields, fields));
}
}
void ReaderFrontend::Update() {
void ReaderFrontend::Update()
{
if ( disabled )
return;
if ( !initialized ) {
if ( !initialized )
{
reporter->Error("Tried to call update on uninitialized reader");
return;
}
}
backend->SendIn(new UpdateMessage(backend));
}
}
void ReaderFrontend::Close() {
void ReaderFrontend::Close()
{
if ( disabled )
return;
if ( !initialized ) {
if ( !initialized )
{
reporter->Error("Tried to call finish on uninitialized reader");
return;
}
}
backend->SendIn(new CloseMessage(backend));
}
}
string ReaderFrontend::Name() const
{
{
if ( source.size() )
return ty_name;
return ty_name + "/" + source;
}
}
}

314
src/input/readers/Ascii.cc
View file

@ -23,69 +23,73 @@ using threading::Field;
FieldMapping::FieldMapping(const string& arg_name, const TypeTag& arg_type, int arg_position)
: name(arg_name), type(arg_type)
{
{
position = arg_position;
secondary_position = -1;
present = true;
}
}
FieldMapping::FieldMapping(const string& arg_name, const TypeTag& arg_type, const TypeTag& arg_subtype, int arg_position)
FieldMapping::FieldMapping(const string& arg_name, const TypeTag& arg_type,
const TypeTag& arg_subtype, int arg_position)
: name(arg_name), type(arg_type), subtype(arg_subtype)
{
{
position = arg_position;
secondary_position = -1;
present = true;
}
}
FieldMapping::FieldMapping(const FieldMapping& arg)
: name(arg.name), type(arg.type), subtype(arg.subtype), present(arg.present)
{
{
position = arg.position;
secondary_position = arg.secondary_position;
}
}
FieldMapping FieldMapping::subType() {
FieldMapping FieldMapping::subType()
{
return FieldMapping(name, subtype, position);
}
}
Ascii::Ascii(ReaderFrontend *frontend) : ReaderBackend(frontend)
{
{
file = 0;
//keyMap = new map<string, string>();
separator.assign( (const char*) BifConst::InputAscii::separator->Bytes(), BifConst::InputAscii::separator->Len());
if ( separator.size() != 1 ) {
separator.assign( (const char*) BifConst::InputAscii::separator->Bytes(),
BifConst::InputAscii::separator->Len());
if ( separator.size() != 1 )
Error("separator length has to be 1. Separator will be truncated.");
}
set_separator.assign( (const char*) BifConst::InputAscii::set_separator->Bytes(), BifConst::InputAscii::set_separator->Len());
if ( set_separator.size() != 1 ) {
set_separator.assign( (const char*) BifConst::InputAscii::set_separator->Bytes(),
BifConst::InputAscii::set_separator->Len());
if ( set_separator.size() != 1 )
Error("set_separator length has to be 1. Separator will be truncated.");
}
empty_field.assign( (const char*) BifConst::InputAscii::empty_field->Bytes(), BifConst::InputAscii::empty_field->Len());
unset_field.assign( (const char*) BifConst::InputAscii::unset_field->Bytes(), BifConst::InputAscii::unset_field->Len());
empty_field.assign( (const char*) BifConst::InputAscii::empty_field->Bytes(),
BifConst::InputAscii::empty_field->Len());
unset_field.assign( (const char*) BifConst::InputAscii::unset_field->Bytes(),
BifConst::InputAscii::unset_field->Len());
}
Ascii::~Ascii()
{
{
DoClose();
}
}
void Ascii::DoClose()
{
if ( file != 0 ) {
{
if ( file != 0 )
{
file->close();
delete(file);
file = 0;
}
}
}
bool Ascii::DoInit(string path, int arg_mode, int arg_num_fields, const Field* const* arg_fields)
{
{
fname = path;
mode = arg_mode;
mtime = 0;
@ -93,124 +97,135 @@ bool Ascii::DoInit(string path, int arg_mode, int arg_num_fields, const Field* c
num_fields = arg_num_fields;
fields = arg_fields;
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) ) {
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) )
{
Error(Fmt("Unsupported read mode %d for source %s", mode, path.c_str()));
return false;
}
}
file = new ifstream(path.c_str());
if ( !file->is_open() ) {
if ( !file->is_open() )
{
Error(Fmt("Init: cannot open %s", fname.c_str()));
delete(file);
file = 0;
return false;
}
}
if ( ReadHeader(false) == false ) {
if ( ReadHeader(false) == false )
{
Error(Fmt("Init: cannot open %s; headers are incorrect", fname.c_str()));
file->close();
delete(file);
file = 0;
return false;
}
}
DoUpdate();
return true;
}
}
bool Ascii::ReadHeader(bool useCached) {
bool Ascii::ReadHeader(bool useCached)
{
// try to read the header line...
string line;
map<string, uint32_t> ifields;
if ( !useCached ) {
if ( !GetLine(line) ) {
if ( !useCached )
{
if ( !GetLine(line) )
{
Error("could not read first line");
return false;
}
}
headerline = line;
} else {
}
else
line = headerline;
}
// construct list of field names.
istringstream splitstream(line);
int pos=0;
while ( splitstream ) {
while ( splitstream )
{
string s;
if ( !getline(splitstream, s, separator[0]))
break;
ifields[s] = pos;
pos++;
}
}
//printf("Updating fields from description %s\n", line.c_str());
columnMap.clear();
for ( unsigned int i = 0; i < num_fields; i++ ) {
for ( unsigned int i = 0; i < num_fields; i++ )
{
const Field* field = fields[i];
map<string, uint32_t>::iterator fit = ifields.find(field->name);
if ( fit == ifields.end() ) {
if ( field->optional ) {
if ( fit == ifields.end() )
{
if ( field->optional )
{
// we do not really need this field. mark it as not present and always send an undef back.
FieldMapping f(field->name, field->type, field->subtype, -1);
f.present = false;
columnMap.push_back(f);
continue;
}
}
Error(Fmt("Did not find requested field %s in input data file %s.", field->name.c_str(), fname.c_str()));
return false;
}
}
FieldMapping f(field->name, field->type, field->subtype, ifields[field->name]);
if ( field->secondary_name != "" ) {
if ( field->secondary_name != "" )
{
map<string, uint32_t>::iterator fit2 = ifields.find(field->secondary_name);
if ( fit2 == ifields.end() ) {
if ( fit2 == ifields.end() )
{
Error(Fmt("Could not find requested port type field %s in input data file.", field->secondary_name.c_str()));
return false;
}
}
f.secondary_position = ifields[field->secondary_name];
}
}
columnMap.push_back(f);
}
}
// well, that seems to have worked...
return true;
}
}
bool Ascii::GetLine(string& str) {
while ( getline(*file, str) ) {
if ( str[0] != '#' ) {
bool Ascii::GetLine(string& str)
{
while ( getline(*file, str) )
{
if ( str[0] != '#' )
return true;
}
if ( str.compare(0,8, "#fields\t") == 0 ) {
if ( str.compare(0,8, "#fields\t") == 0 )
{
str = str.substr(8);
return true;
}
}
}
return false;
}
Value* Ascii::EntryToVal(string s, FieldMapping field) {
if ( s.compare(unset_field) == 0 ) { // field is not set...
return new Value(field.type, false);
}
Value* Ascii::EntryToVal(string s, FieldMapping field)
{
if ( s.compare(unset_field) == 0 ) // field is not set...
return new Value(field.type, false);
Value* val = new Value(field.type, true);
switch ( field.type ) {
@ -220,14 +235,15 @@ Value* Ascii::EntryToVal(string s, FieldMapping field) {
break;
case TYPE_BOOL:
if ( s == "T" ) {
if ( s == "T" )
val->val.int_val = 1;
} else if ( s == "F" ) {
else if ( s == "F" )
val->val.int_val = 0;
} else {
else
{
Error(Fmt("Field: %s Invalid value for boolean: %s", field.name.c_str(), s.c_str()));
return false;
}
}
break;
case TYPE_INT:
@ -250,7 +266,8 @@ Value* Ascii::EntryToVal(string s, FieldMapping field) {
val->val.port_val.proto = TRANSPORT_UNKNOWN;
break;
case TYPE_SUBNET: {
case TYPE_SUBNET:
{
size_t pos = s.find("/");
if ( pos == s.npos ) {
Error(Fmt("Invalid value for subnet: %s", s.c_str()));
@ -261,8 +278,8 @@ Value* Ascii::EntryToVal(string s, FieldMapping field) {
val->val.subnet_val.prefix = StringToAddr(addr);
val->val.subnet_val.length = width;
}
break;
}
case TYPE_ADDR:
val->val.addr_val = StringToAddr(s);
@ -287,47 +304,56 @@ Value* Ascii::EntryToVal(string s, FieldMapping field) {
Value** lvals = new Value* [length];
if ( field.type == TYPE_TABLE ) {
if ( field.type == TYPE_TABLE )
{
val->val.set_val.vals = lvals;
val->val.set_val.size = length;
} else if ( field.type == TYPE_VECTOR ) {
}
else if ( field.type == TYPE_VECTOR )
{
val->val.vector_val.vals = lvals;
val->val.vector_val.size = length;
} else {
}
else
{
assert(false);
}
}
if ( length == 0 )
break; //empty
istringstream splitstream(s);
while ( splitstream ) {
while ( splitstream )
{
string element;
if ( !getline(splitstream, element, set_separator[0]) )
break;
if ( pos >= length ) {
Error(Fmt("Internal error while parsing set. pos %d >= length %d. Element: %s", pos, length, element.c_str()));
if ( pos >= length )
{
Error(Fmt("Internal error while parsing set. pos %d >= length %d."
" Element: %s", pos, length, element.c_str()));
break;
}
}
Value* newval = EntryToVal(element, field.subType());
if ( newval == 0 ) {
if ( newval == 0 )
{
Error("Error while reading set");
return 0;
}
}
lvals[pos] = newval;
pos++;
}
}
if ( pos != length ) {
if ( pos != length )
{
Error("Internal error while parsing set: did not find all elements");
return 0;
}
}
break;
}
@ -340,24 +366,23 @@ Value* Ascii::EntryToVal(string s, FieldMapping field) {
}
return val;
}
}
// read the entire file and send appropriate thingies back to InputMgr
bool Ascii::DoUpdate() {
bool Ascii::DoUpdate()
{
switch ( mode ) {
case REREAD:
// check if the file has changed
struct stat sb;
if ( stat(fname.c_str(), &sb) == -1 ) {
if ( stat(fname.c_str(), &sb) == -1 )
{
Error(Fmt("Could not get stat for %s", fname.c_str()));
return false;
}
}
if ( sb.st_mtime <= mtime ) {
// no change
if ( sb.st_mtime <= mtime ) // no change
return true;
}
mtime = sb.st_mtime;
// file changed. reread.
@ -366,57 +391,56 @@ bool Ascii::DoUpdate() {
case MANUAL:
case STREAM:
// dirty, fix me. (well, apparently after trying seeking, etc - this is not that bad)
if ( file && file->is_open() ) {
if ( mode == STREAM ) {
// dirty, fix me. (well, apparently after trying seeking, etc
// - this is not that bad)
if ( file && file->is_open() )
{
if ( mode == STREAM )
{
file->clear(); // remove end of file evil bits
if ( !ReadHeader(true) ) // in case filters changed
{
return false; // header reading failed
}
break;
}
}
file->close();
}
}
file = new ifstream(fname.c_str());
if ( !file->is_open() ) {
if ( !file->is_open() )
{
Error(Fmt("cannot open %s", fname.c_str()));
return false;
}
}
if ( ReadHeader(false) == false ) {
if ( ReadHeader(false) == false )
{
return false;
}
}
break;
default:
assert(false);
}
//
// file->seekg(0, ios::beg); // do not forget clear.
}
string line;
while ( GetLine(line ) ) {
while ( GetLine(line ) )
{
// split on tabs
istringstream splitstream(line);
map<int, string> stringfields;
int pos = 0;
while ( splitstream ) {
while ( splitstream )
{
string s;
if ( !getline(splitstream, s, separator[0]) )
break;
stringfields[pos] = s;
pos++;
}
}
pos--; // for easy comparisons of max element.
@ -426,69 +450,60 @@ bool Ascii::DoUpdate() {
int fpos = 0;
for ( vector<FieldMapping>::iterator fit = columnMap.begin();
fit != columnMap.end();
fit++ ){
fit++ )
{
if ( ! fit->present ) {
if ( ! fit->present )
{
// add non-present field
fields[fpos] = new Value((*fit).type, false);
fpos++;
continue;
}
}
assert(fit->position >= 0 );
if ( (*fit).position > pos || (*fit).secondary_position > pos ) {
if ( (*fit).position > pos || (*fit).secondary_position > pos )
{
Error(Fmt("Not enough fields in line %s. Found %d fields, want positions %d and %d", line.c_str(), pos, (*fit).position, (*fit).secondary_position));
return false;
}
}
Value* val = EntryToVal(stringfields[(*fit).position], *fit);
if ( val == 0 ) {
if ( val == 0 )
{
Error("Could not convert String value to Val");
return false;
}
}
if ( (*fit).secondary_position != -1 ) {
if ( (*fit).secondary_position != -1 )
{
// we have a port definition :)
assert(val->type == TYPE_PORT );
// Error(Fmt("Got type %d != PORT with secondary position!", val->type));
val->val.port_val.proto = StringToProto(stringfields[(*fit).secondary_position]);
}
}
fields[fpos] = val;
fpos++;
}
}
//printf("fpos: %d, second.num_fields: %d\n", fpos, (*it).second.num_fields);
assert ( (unsigned int) fpos == num_fields );
if ( mode == STREAM ) {
if ( mode == STREAM )
Put(fields);
} else {
else
SendEntry(fields);
}
/* Do not do this, ownership changes to other thread
* for ( unsigned int i = 0; i < (*it).second.num_fields; i++ ) {
delete fields[i];
}
delete [] fields;
*/
}
//file->clear(); // remove end of file evil bits
//file->seekg(0, ios::beg); // and seek to start.
if ( mode != STREAM ) {
if ( mode != STREAM )
EndCurrentSend();
}
return true;
}
}
bool Ascii::DoHeartbeat(double network_time, double current_time)
{
@ -500,12 +515,13 @@ bool Ascii::DoHeartbeat(double network_time, double current_time)
break;
case REREAD:
case STREAM:
Update(); // call update and not DoUpdate, because update actually checks disabled.
Update(); // call update and not DoUpdate, because update
// checks disabled.
break;
default:
assert(false);
}
return true;
}
}

118
src/input/readers/Benchmark.cc
View file

@ -22,7 +22,7 @@ using threading::Field;
Benchmark::Benchmark(ReaderFrontend *frontend) : ReaderBackend(frontend)
{
{
multiplication_factor = double(BifConst::InputBenchmark::factor);
autospread = double(BifConst::InputBenchmark::autospread);
spread = int(BifConst::InputBenchmark::spread);
@ -32,19 +32,19 @@ Benchmark::Benchmark(ReaderFrontend *frontend) : ReaderBackend(frontend)
timedspread = double(BifConst::InputBenchmark::timedspread);
heart_beat_interval = double(BifConst::Threading::heart_beat_interval);
}
}
Benchmark::~Benchmark()
{
{
DoClose();
}
}
void Benchmark::DoClose()
{
}
{
}
bool Benchmark::DoInit(string path, int arg_mode, int arg_num_fields, const Field* const* arg_fields)
{
{
mode = arg_mode;
num_fields = arg_num_fields;
@ -54,18 +54,20 @@ bool Benchmark::DoInit(string path, int arg_mode, int arg_num_fields, const Fiel
if ( autospread != 0.0 )
autospread_time = (int) ( (double) 1000000 / (autospread * (double) num_lines) );
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) ) {
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) )
{
Error(Fmt("Unsupported read mode %d for source %s", mode, path.c_str()));
return false;
}
}
heartbeatstarttime = CurrTime();
DoUpdate();
return true;
}
}
string Benchmark::RandomString(const int len) {
string Benchmark::RandomString(const int len)
{
string s(len, ' ');
static const char values[] =
@ -73,65 +75,65 @@ string Benchmark::RandomString(const int len) {
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz";
for (int i = 0; i < len; ++i) {
for (int i = 0; i < len; ++i)
s[i] = values[rand() / (RAND_MAX / sizeof(values))];
}
return s;
}
}
double Benchmark::CurrTime() {
double Benchmark::CurrTime()
{
struct timeval tv;
assert ( gettimeofday(&tv, 0) >= 0 );
return double(tv.tv_sec) + double(tv.tv_usec) / 1e6;
}
}
// read the entire file and send appropriate thingies back to InputMgr
bool Benchmark::DoUpdate() {
bool Benchmark::DoUpdate()
{
int linestosend = num_lines * heart_beat_interval;
for ( int i = 0; i < linestosend; i++ ) {
for ( int i = 0; i < linestosend; i++ )
{
Value** field = new Value*[num_fields];
for (unsigned int j = 0; j < num_fields; j++ ) {
for (unsigned int j = 0; j < num_fields; j++ )
field[j] = EntryToVal(fields[j]->type, fields[j]->subtype);
}
if ( mode == STREAM ) {
if ( mode == STREAM )
// do not do tracking, spread out elements over the second that we have...
Put(field);
} else {
else
SendEntry(field);
}
if ( stopspreadat == 0 || num_lines < stopspreadat ) {
if ( stopspreadat == 0 || num_lines < stopspreadat )
{
if ( spread != 0 )
usleep(spread);
if ( autospread_time != 0 )
usleep( autospread_time );
}
}
if ( timedspread != 0.0 ) {
if ( timedspread != 0.0 )
{
double diff;
do {
do
diff = CurrTime() - heartbeatstarttime;
//printf("%d %f\n", i, diff);
//} while ( diff < i/threading::Manager::HEART_BEAT_INTERVAL*(num_lines + (num_lines * timedspread) ) );
} while ( diff/heart_beat_interval < i/(linestosend + (linestosend * timedspread) ) );
//} while ( diff < 0.8);
}
while ( diff/heart_beat_interval < i/(linestosend
+ (linestosend * timedspread) ) );
}
}
if ( mode != STREAM ) {
if ( mode != STREAM )
EndCurrentSend();
}
return true;
}
threading::Value* Benchmark::EntryToVal(TypeTag type, TypeTag subtype) {
threading::Value* Benchmark::EntryToVal(TypeTag type, TypeTag subtype)
{
Value* val = new Value(type, true);
// basically construct something random from the fields that we want.
@ -170,7 +172,8 @@ threading::Value* Benchmark::EntryToVal(TypeTag type, TypeTag subtype) {
val->val.port_val.proto = TRANSPORT_UNKNOWN;
break;
case TYPE_SUBNET: {
case TYPE_SUBNET:
{
val->val.subnet_val.prefix = StringToAddr("192.168.17.1");
val->val.subnet_val.length = 16;
}
@ -192,28 +195,32 @@ threading::Value* Benchmark::EntryToVal(TypeTag type, TypeTag subtype) {
Value** lvals = new Value* [length];
if ( type == TYPE_TABLE ) {
if ( type == TYPE_TABLE )
{
val->val.set_val.vals = lvals;
val->val.set_val.size = length;
} else if ( type == TYPE_VECTOR ) {
}
else if ( type == TYPE_VECTOR )
{
val->val.vector_val.vals = lvals;
val->val.vector_val.size = length;
} else {
}
else
assert(false);
}
if ( length == 0 )
break; //empty
for ( unsigned int pos = 0; pos < length; pos++ ) {
for ( unsigned int pos = 0; pos < length; pos++ )
{
Value* newval = EntryToVal(subtype, TYPE_ENUM);
if ( newval == 0 ) {
if ( newval == 0 )
{
Error("Error while reading set");
return 0;
}
}
lvals[pos] = newval;
}
}
break;
}
@ -226,20 +233,11 @@ threading::Value* Benchmark::EntryToVal(TypeTag type, TypeTag subtype) {
return val;
}
}
bool Benchmark::DoHeartbeat(double network_time, double current_time)
{
/*
* This does not work the way I envisioned it, because the queueing is the problem.
printf("%f\n", CurrTime() - current_time);
if ( CurrTime() - current_time > 0.25 ) {
// event has hung for a time. refuse.
SendEvent("EndBenchmark", 0, 0);
return true;
} */
ReaderBackend::DoHeartbeat(network_time, current_time);
num_lines = (int) ( (double) num_lines*multiplication_factor);
num_lines += add;
@ -251,7 +249,8 @@ bool Benchmark::DoHeartbeat(double network_time, double current_time)
break;
case REREAD:
case STREAM:
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.
Value** v = new Value*[2];
v[0] = new Value(TYPE_COUNT, true);
@ -260,12 +259,11 @@ bool Benchmark::DoHeartbeat(double network_time, double current_time)
v[1]->val.double_val = CurrTime();
SendEvent("lines_changed", 2, v);
}
}
if ( autospread != 0.0 ) {
autospread_time = (int) ( (double) 1000000 / (autospread * (double) num_lines) );
if ( autospread != 0.0 )
// because executing this in every loop is apparently too expensive.
}
autospread_time = (int) ( (double) 1000000 / (autospread * (double) num_lines) );
Update(); // call update and not DoUpdate, because update actually checks disabled.

155
src/input/readers/Raw.cc
View file

@ -24,79 +24,86 @@ using threading::Value;
using threading::Field;
Raw::Raw(ReaderFrontend *frontend) : ReaderBackend(frontend)
{
{
file = 0;
in = 0;
//keyMap = new map<string, string>();
separator.assign( (const char*) BifConst::InputRaw::record_separator->Bytes(), BifConst::InputRaw::record_separator->Len());
if ( separator.size() != 1 ) {
if ( separator.size() != 1 )
Error("separator length has to be 1. Separator will be truncated.");
}
}
Raw::~Raw()
{
{
DoClose();
}
}
void Raw::DoClose()
{
if ( file != 0 ) {
{
if ( file != 0 )
{
Close();
}
}
}
bool Raw::Open()
{
if ( execute ) {
{
if ( execute )
{
file = popen(fname.c_str(), "r");
if ( file == NULL ) {
if ( file == NULL )
{
Error(Fmt("Could not execute command %s", fname.c_str()));
return false;
}
}
} else {
else
{
file = fopen(fname.c_str(), "r");
if ( file == NULL ) {
if ( file == NULL )
{
Error(Fmt("Init: cannot open %s", fname.c_str()));
return false;
}
}
}
in = new boost::fdistream(fileno(file));
if ( execute && mode == STREAM ) {
if ( execute && mode == STREAM )
{
fcntl(fileno(file), F_SETFL, O_NONBLOCK);
}
}
return true;
}
}
bool Raw::Close()
{
if ( file == NULL ) {
{
if ( file == NULL )
{
InternalError(Fmt("Trying to close closed file for stream %s", fname.c_str()));
return false;
}
}
if ( execute ) {
if ( execute )
{
delete(in);
pclose(file);
} else {
}
else
{
delete(in);
fclose(file);
}
}
in = NULL;
file = NULL;
return true;
}
}
bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* const* arg_fields)
{
{
fname = path;
mode = arg_mode;
mtime = 0;
@ -107,24 +114,30 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
num_fields = arg_num_fields;
fields = arg_fields;
if ( path.length() == 0 ) {
if ( path.length() == 0 )
{
Error("No source path provided");
return false;
}
}
if ( arg_num_fields != 1 ) {
Error("Filter for raw reader contains more than one field. Filters for the raw reader may only contain exactly one string field. Filter ignored.");
if ( arg_num_fields != 1 )
{
Error("Filter for raw reader contains more than one field. "
"Filters for the raw reader may only contain exactly one string field. "
"Filter ignored.");
return false;
}
}
if ( fields[0]->type != TYPE_STRING ) {
if ( fields[0]->type != TYPE_STRING )
{
Error("Filter for raw reader contains a field that is not of type string.");
return false;
}
}
// do Initialization
char last = path[path.length()-1];
if ( last == '|' ) {
if ( last == '|' )
{
execute = true;
fname = path.substr(0, fname.length() - 1);
@ -137,19 +150,17 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
} else {
execute = false;
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) ) {
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) )
{
Error(Fmt("Unsupported read mode %d for source %s", mode, fname.c_str()));
return false;
}
}
result = Open();
}
}
if ( result == false ) {
if ( result == false )
return result;
}
#ifdef DEBUG
Debug(DBG_INPUT, "Raw reader created, will perform first update");
@ -162,62 +173,68 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
Debug(DBG_INPUT, "First update went through");
#endif
return true;
}
bool Raw::GetLine(string& str) {
while ( getline(*in, str, separator[0]) ) {
return true;
}
bool Raw::GetLine(string& str)
{
while ( getline(*in, str, separator[0]) )
return true;
return false;
}
}
// read the entire file and send appropriate thingies back to InputMgr
bool Raw::DoUpdate() {
if ( firstrun ) {
bool Raw::DoUpdate()
{
if ( firstrun )
firstrun = false;
} else {
else
{
switch ( mode ) {
case REREAD:
{
// check if the file has changed
struct stat sb;
if ( stat(fname.c_str(), &sb) == -1 ) {
if ( stat(fname.c_str(), &sb) == -1 )
{
Error(Fmt("Could not get stat for %s", fname.c_str()));
return false;
}
}
if ( sb.st_mtime <= mtime ) {
if ( sb.st_mtime <= mtime )
// no change
return true;
}
mtime = sb.st_mtime;
// file changed. reread.
// fallthrough
}
case MANUAL:
case STREAM:
if ( mode == STREAM && file != NULL && in != NULL ) {
if ( mode == STREAM && file != NULL && in != NULL )
{
//fpurge(file);
in->clear(); // remove end of file evil bits
break;
}
}
Close();
if ( !Open() ) {
if ( !Open() )
return false;
}
break;
default:
assert(false);
}
}
}
string line;
while ( GetLine(line) ) {
while ( GetLine(line) )
{
assert (num_fields == 1);
Value** fields = new Value*[1];
@ -228,14 +245,14 @@ bool Raw::DoUpdate() {
fields[0] = val;
Put(fields);
}
}
return true;
}
}
bool Raw::DoHeartbeat(double network_time, double current_time)
{
{
ReaderBackend::DoHeartbeat(network_time, current_time);
switch ( mode ) {
@ -244,12 +261,12 @@ bool Raw::DoHeartbeat(double network_time, double current_time)
break;
case REREAD:
case STREAM:
Update(); // call update and not DoUpdate, because update actually checks disabled.
Update(); // call update and not DoUpdate, because update
// checks disabled.
break;
default:
assert(false);
}
return true;
}
}