roughly 10-fold speedup in navcat, plus fix ordering bug
parent
e20a0b40f8
commit
9776a0882d
33
navcat.cc
33
navcat.cc
|
@ -98,14 +98,14 @@ static size_t writen2(int fd, const void *buf, size_t count)
|
|||
|
||||
void sendProtobuf(string_view dir, time_t startTime, time_t stopTime=0)
|
||||
{
|
||||
pair<uint64_t, uint64_t> start = {startTime, 0};
|
||||
pair<int64_t, int64_t> start(startTime, 0);
|
||||
|
||||
// so we have a ton of files, and internally these are not ordered
|
||||
map<string,uint32_t> fpos;
|
||||
vector<NavMonMessage> nmms;
|
||||
vector<pair<timespec,string> > rnmms;
|
||||
for(;;) {
|
||||
auto srcs = getSources(dir);
|
||||
nmms.clear();
|
||||
rnmms.clear();
|
||||
for(const auto& src: srcs) {
|
||||
string fname = getPath(dir, start.first, src);
|
||||
FILE* fp = fopen(fname.c_str(), "r");
|
||||
|
@ -118,36 +118,35 @@ void sendProtobuf(string_view dir, time_t startTime, time_t stopTime=0)
|
|||
continue;
|
||||
}
|
||||
cerr <<"Seeked to position "<<fpos[fname]<<" of "<<fname<<endl;
|
||||
NavMonMessage nmm;
|
||||
|
||||
uint32_t looked=0;
|
||||
while(getNMM(fp, nmm, offset)) {
|
||||
string msg;
|
||||
struct timespec ts;
|
||||
while(getRawNMM(fp, ts, msg, offset)) {
|
||||
// don't drop data that is only 5 seconds too old
|
||||
if(make_pair(nmm.localutcseconds() + 5, nmm.localutcnanoseconds()) >= start) {
|
||||
nmms.push_back(nmm);
|
||||
if(make_pair(ts.tv_sec + 5, ts.tv_nsec) >= start) {
|
||||
rnmms.push_back({ts, msg});
|
||||
}
|
||||
++looked;
|
||||
}
|
||||
cerr<<"Harvested "<<nmms.size()<<" events out of "<<looked<<endl;
|
||||
cerr<<"Harvested "<<rnmms.size()<<" events out of "<<looked<<endl;
|
||||
fpos[fname]=offset;
|
||||
fclose(fp);
|
||||
}
|
||||
|
||||
sort(nmms.begin(), nmms.end(), [](const auto& a, const auto& b)
|
||||
sort(rnmms.begin(), rnmms.end(), [](const auto& a, const auto& b)
|
||||
{
|
||||
return make_pair(a.localutcseconds(), b.localutcnanoseconds()) <
|
||||
make_pair(b.localutcseconds(), b.localutcnanoseconds());
|
||||
return std::tie(a.first.tv_sec, a.first.tv_nsec)
|
||||
< std::tie(b.first.tv_sec, b.first.tv_nsec);
|
||||
});
|
||||
|
||||
for(const auto& nmm: nmms) {
|
||||
if(nmm.localutcseconds() > stopTime)
|
||||
for(const auto& nmm: rnmms) {
|
||||
if(nmm.first.tv_sec > stopTime)
|
||||
break;
|
||||
std::string out;
|
||||
nmm.SerializeToString(&out);
|
||||
std::string buf="bert";
|
||||
uint16_t len = htons(out.size());
|
||||
uint16_t len = htons(nmm.second.size());
|
||||
buf.append((char*)(&len), 2);
|
||||
buf+=out;
|
||||
buf += nmm.second;
|
||||
//fwrite(buf.c_str(), 1, buf.size(), stdout);
|
||||
writen2(1, buf.c_str(), buf.size());
|
||||
}
|
||||
|
|
24
storage.cc
24
storage.cc
|
@ -78,3 +78,27 @@ bool getNMM(FILE* fp, NavMonMessage& nmm, uint32_t& offset)
|
|||
offset += 4 + 2 + len;
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
bool getRawNMM(FILE* fp, timespec& t, string& raw, uint32_t& offset)
|
||||
{
|
||||
char bert[4];
|
||||
if(fread(bert, 1, 4, fp) != 4 || bert[0]!='b' || bert[1]!='e' || bert[2] !='r' || bert[3]!='t') {
|
||||
return false;
|
||||
}
|
||||
|
||||
uint16_t len;
|
||||
if(fread(&len, 1, 2, fp) != 2)
|
||||
return false;
|
||||
len = htons(len);
|
||||
char buffer[len];
|
||||
if(fread(buffer, 1, len, fp) != len)
|
||||
return false;
|
||||
NavMonMessage nmm;
|
||||
raw.assign(buffer, len);
|
||||
nmm.ParseFromString(raw);
|
||||
t.tv_sec = nmm.localutcseconds();
|
||||
t.tv_nsec = nmm.localutcnanoseconds();
|
||||
offset += 4 + 2 + len;
|
||||
return true;
|
||||
}
|
||||
|
|
|
@ -8,3 +8,4 @@ std::vector<std::string> getPathComponents(std::string_view root, time_t s, uint
|
|||
std::string getPath(std::string_view root, time_t s, uint64_t sourceid, bool create=false);
|
||||
bool getNMM(int fd, NavMonMessage& nmm, uint32_t& offset);
|
||||
bool getNMM(FILE* fp, NavMonMessage& nmm, uint32_t& offset);
|
||||
bool getRawNMM(FILE* fp, timespec& t, std::string& raw, uint32_t& offset);
|
||||
|
|
Loading…
Reference in New Issue