trace: add coroutine and yield to log_parser

trace/gem5.cc

@@ -53,7 +53,7 @@ void gem5_parser::process_msg(uint64_t ts, char *comp_name,
       return;
 
     if (const std::string *s = syms.lookup(addr)) {
-      cur_event = std::make_shared<EHostCall>(ts, *s);
+      yield(std::make_shared<EHostCall>(ts, *s));
     }
   } else if (comp_name_len == 18 &&
              !memcmp(comp_name, "system.pc.ethernet", 18)) {
@@ -70,40 +70,40 @@ void gem5_parser::process_msg(uint64_t ts, char *comp_name,
     uint64_t size = 0;
     if (p.consume_str("received ")) {
       if (p.consume_str("MSI-X intr vec ") && p.consume_dec(id)) {
-        cur_event = std::make_shared<EHostMsiX>(ts, id);
+        yield(std::make_shared<EHostMsiX>(ts, id));
       } else if (p.consume_str("DMA read id ") && p.consume_dec(id) &&
                  p.consume_str(" addr ") && p.consume_hex(addr) &&
                  p.consume_str(" size ") && p.consume_dec(size)) {
         // cosim: received DMA read id 94113551511792 addr 23697ad60
         //          size 20
-        cur_event = std::make_shared<EHostDmaR>(ts, id, addr, size);
+        yield(std::make_shared<EHostDmaR>(ts, id, addr, size));
       } else if (p.consume_str("DMA write id ") && p.consume_dec(id) &&
                  p.consume_str(" addr ") && p.consume_hex(addr) &&
                  p.consume_str(" size ") && p.consume_dec(size)) {
         // cosim: received DMA write id 94113551528032 addr 236972000
         //          size 4
-        cur_event = std::make_shared<EHostDmaW>(ts, id, addr, size);
+        yield(std::make_shared<EHostDmaW>(ts, id, addr, size));
       } else if (p.consume_str("read completion id ") && p.consume_dec(id)) {
         // cosim: received read completion id 94583743418112
-        cur_event = std::make_shared<EHostMmioC>(ts, id);
+        yield(std::make_shared<EHostMmioC>(ts, id));
       } else if (p.consume_str("write completion id ") && p.consume_dec(id)) {
         // cosim: received write completion id 94583743418736
-        cur_event = std::make_shared<EHostMmioC>(ts, id);
+        yield(std::make_shared<EHostMmioC>(ts, id));
       }
     } else if (p.consume_str("sending ")) {
       if (p.consume_str("read addr ") && p.consume_hex(addr) &&
           p.consume_str(" size ") && p.consume_dec(size) &&
           p.consume_str(" id ") && p.consume_dec(id)) {
         // cosim: sending read addr c012a500 size 4 id 94583743418112
-        cur_event = std::make_shared<EHostMmioR>(ts, id, addr, size);
+        yield(std::make_shared<EHostMmioR>(ts, id, addr, size));
       } else if (p.consume_str("write addr ") && p.consume_hex(addr) &&
                  p.consume_str(" size ") && p.consume_dec(size) &&
                  p.consume_str(" id ") && p.consume_dec(id)) {
         // cosim: sending write addr c0108000 size 4 id 94584005188256
-        cur_event = std::make_shared<EHostMmioW>(ts, id, addr, size);
+        yield(std::make_shared<EHostMmioW>(ts, id, addr, size));
       }
     } else if (p.consume_str("completed DMA id ") && p.consume_dec(id)) {
-      cur_event = std::make_shared<EHostDmaC>(ts, id);
+      yield(std::make_shared<EHostDmaC>(ts, id));
     }
   }
 

trace/log_parser.cc

@@ -97,13 +97,12 @@ size_t log_parser::try_line() {
 }
 
 bool log_parser::next_event() {
-  cur_event.reset();
-
   if (buf_len == 0 && !next_block()) {
     std::cerr << "escape 0" << std::endl;
     return false;
   }
 
+  got_event = false;
   do {
     size_t newpos = try_line();
     if (!newpos) {
@@ -119,7 +118,19 @@ bool log_parser::next_event() {
       }
     }
     buf_pos = newpos;
-  } while (!cur_event);
+  } while (!got_event);
 
   return true;
 }
+
+void log_parser::read_coro(coro_t::push_type &sink_) {
+  sink = &sink_;
+  while (next_event());
+}
+
+void log_parser::yield(std::shared_ptr<event> ev)
+{
+  got_event = true;
+  ev->source = this;
+  (*sink)(ev);
+}
\ No newline at end of file

trace/nicbm.cc

@@ -47,35 +47,35 @@ void nicbm_parser::process_line(char *line, size_t line_len) {
   if (p.consume_str("read(off=0x")) {
     if (p.consume_hex(addr) && p.consume_str(", len=") && p.consume_dec(len) &&
         p.consume_str(", val=0x") && p.consume_hex(val)) {
-      cur_event = std::make_shared<e_nic_mmio_r>(ts, addr, len, val);
+      yield(std::make_shared<e_nic_mmio_r>(ts, addr, len, val));
     }
   } else if (p.consume_str("write(off=0x")) {
     if (p.consume_hex(addr) && p.consume_str(", len=") && p.consume_dec(len) &&
         p.consume_str(", val=0x") && p.consume_hex(val)) {
-      cur_event = std::make_shared<e_nic_mmio_w>(ts, addr, len, val);
+      yield(std::make_shared<e_nic_mmio_w>(ts, addr, len, val));
     }
   } else if (p.consume_str("issuing dma op 0x")) {
     if (p.consume_hex(id) && p.consume_str(" addr ") && p.consume_hex(addr) &&
         p.consume_str(" len ") && p.consume_hex(len)) {
-      cur_event = std::make_shared<e_nic_dma_i>(ts, id, addr, len);
+      yield(std::make_shared<e_nic_dma_i>(ts, id, addr, len));
     }
   } else if (p.consume_str("completed dma read op 0x") ||
              p.consume_str("completed dma write op 0x")) {
     if (p.consume_hex(id) && p.consume_str(" addr ") && p.consume_hex(addr) &&
         p.consume_str(" len ") && p.consume_hex(len)) {
-      cur_event = std::make_shared<e_nic_dma_c>(ts, id);
+      yield(std::make_shared<e_nic_dma_c>(ts, id));
     }
   } else if (p.consume_str("issue MSI-X interrupt vec ")) {
     if (p.consume_dec(id)) {
-      cur_event = std::make_shared<e_nic_msix>(ts, id);
+      yield(std::make_shared<e_nic_msix>(ts, id));
     }
   } else if (p.consume_str("eth tx: len ")) {
     if (p.consume_dec(len)) {
-      cur_event = std::make_shared<e_nic_tx>(ts, len);
+      yield(std::make_shared<e_nic_tx>(ts, len));
     }
   } else if (p.consume_str("eth rx: port 0 len ")) {
     if (p.consume_dec(len)) {
-      cur_event = std::make_shared<e_nic_rx>(ts, len);
+      yield(std::make_shared<e_nic_rx>(ts, len));
     }
 #if 1
   }

trace/process.cc

@@ -25,7 +25,6 @@
 #include "trace/process.h"
 
 #include <boost/bind.hpp>
-#include <boost/coroutine2/all.hpp>
 #include <boost/foreach.hpp>
 #include <iostream>
 #include <memory>
@@ -41,15 +40,6 @@ struct event_pair_cmp {
   }
 };
 
-typedef boost::coroutines2::asymmetric_coroutine<std::shared_ptr<event>> coro_t;
-
-void ReadEvents(coro_t::push_type &sink, log_parser &lp) {
-  while (lp.next_event() && lp.cur_event) {
-    lp.cur_event->source = &lp;
-    sink(lp.cur_event);
-  }
-}
-
 /** merge multiple event streams into one ordered by timestamp */
 void MergeEvents(coro_t::push_type &sink,
                  std::set<coro_t::pull_type *> &all_parsers) {
@@ -158,7 +148,7 @@ int main(int argc, char *argv[]) {
   std::set<coro_t::pull_type *> sources;
   for (auto p : all_parsers) {
     sources.insert(new coro_t::pull_type(
-      boost::bind(ReadEvents, _1, boost::ref(*p))));
+      boost::bind(&log_parser::read_coro, boost::ref(*p), _1)));
   }
 
   coro_t::pull_type merged(boost::bind(MergeEvents, _1, boost::ref(sources)));

trace/process.h

@@ -24,6 +24,7 @@
 
 #pragma once
 
+#include <boost/coroutine2/all.hpp>
 #include <boost/iostreams/filtering_streambuf.hpp>
 #include <map>
 #include <memory>
@@ -32,6 +33,8 @@
 
 #include "trace/events.h"
 
+typedef boost::coroutines2::asymmetric_coroutine<std::shared_ptr<event>> coro_t;
+
 class sym_map {
  protected:
   bool filter_en;
@@ -68,20 +71,25 @@ class log_parser {
   size_t buf_len;
   size_t buf_pos;
 
+  coro_t::push_type *sink;
+  bool got_event;
+
   bool next_block();
   size_t try_line();
   virtual void process_line(char *line, size_t len) = 0;
 
+  bool next_event();
+  void yield(std::shared_ptr<event> ev);
+
  public:
   const char *label;
-  std::shared_ptr<event> cur_event;
 
   log_parser();
   virtual ~log_parser();
   void open(const char *path);
   void open_gz(const char *path);
 
-  bool next_event();
+  void read_coro(coro_t::push_type &sink_);
 };
 
 class gem5_parser : public log_parser {