Merge branch 'main' into v1.0.0

Author: etwest

CMakeLists.txt

@@ -151,6 +151,17 @@ if (BUILD_EXE)
   add_executable(to_binary_format
     tools/to_binary_format.cpp)
   target_link_libraries(to_binary_format PRIVATE GraphZeppelinCommon)
+
+  # executable for processing a binary graph stream
+  add_executable(process_stream
+    tools/process_stream.cpp)
+  target_link_libraries(process_stream PRIVATE GraphZeppelin)
+
+  # tool for validating that a binary stream appears correct  
+  add_executable(validate_binary_stream
+    tools/validate_binary_stream.cpp
+  )
+  target_link_libraries(validate_binary_stream PRIVATE GraphZeppelin)
 endif()
 
 if (BUILD_BENCH)

include/binary_graph_stream.h

@@ -11,6 +11,12 @@ class BadStreamException : public std::exception {
   }
 };
 
+class StreamFailedException : public std::exception {
+  virtual const char* what() const throw() {
+    return "ERROR: read_data() encountered failed stream. Is stream file corrupted?";
+  }
+};
+
 // A class for reading from a binary graph stream
 class BinaryGraphStream {
 public:
@@ -56,6 +62,10 @@ class BinaryGraphStream {
     // set buf back to the beginning of the buffer read in data
     buf = start_buf;
     bin_file.read(buf, buf_size);
+  
+    if (bin_file.fail() && !bin_file.eof()) {
+      throw StreamFailedException();
+    }  
   }
   const uint32_t edge_size = sizeof(uint8_t) + 2 * sizeof(uint32_t); // size of binary encoded edge
   std::ifstream bin_file; // file to read from
@@ -72,7 +82,7 @@ class BinaryGraphStream_MT {
 public:
   BinaryGraphStream_MT(std::string file_name, uint32_t _b) {
     stream_fd = open(file_name.c_str(), O_RDONLY, S_IRUSR);
-    if (stream_fd == -1) {
+    if (!stream_fd) {
       throw BadStreamException();
     }
 
@@ -167,7 +177,9 @@ class BinaryGraphStream_MT {
       data_to_read = end_of_file - read_off; // EOF truncates the read
 
     while (data_read < data_to_read) {
-      data_read += pread(stream_fd, buf, data_to_read, read_off + data_read); // perform the read
+      int ret = pread(stream_fd, buf, data_to_read, read_off + data_read); // perform the read
+      if (ret == -1) throw StreamFailedException();
+      data_read += ret;
     }
     return data_read;
   }

src/graph_configuration.cpp

@@ -53,6 +53,6 @@ std::ostream& operator<< (std::ostream &out, const GraphConfiguration &conf) {
     out << " Size of groups        = " << conf._group_size << std::endl;
     out << " On disk data location = " << conf._disk_dir << std::endl;
     out << " Backup sketch to RAM  = " << (conf._backup_in_mem? "ON" : "OFF") << std::endl;
-    out << conf._gutter_conf << std::endl;
+    out << conf._gutter_conf;
     return out;
   }

tools/experiment/BCHMK_Graph.cpp

@@ -0,0 +1,41 @@
+#include <graph.h>
+#include <binary_graph_stream.h>
+
+int main(int argc, char **argv) {
+  if (argc != 3) {
+    std::cout << "ERROR: Incorrect number of arguments!" << std::endl;
+    std::cout << "Arguments: stream_file, graph_workers" << std::endl;
+  }
+
+  std::string stream_file = argv[1];
+  int num_threads = std::atoi(argv[2]);
+  if (num_threads < 1) {
+    std::cout << "ERROR: Invalid number of graph workers! Must be > 0." << std::endl;
+  }
+
+  BinaryGraphStream stream(stream_file, 1024*32);
+  node_id_t num_nodes = stream.nodes();
+  size_t num_updates  = stream.edges();
+  std::cout << "Processing stream: " << stream_file << std::endl;
+  std::cout << "nodes       = " << num_nodes << std::endl;
+  std::cout << "num_updates = " << num_updates << std::endl;
+  std::cout << std::endl;
+
+  auto config = GraphConfiguration().gutter_sys(STANDALONE).num_groups(num_threads);
+  config.gutter_conf().gutter_factor(-4);
+  Graph g{num_nodes, config};
+
+  auto ins_start = std::chrono::steady_clock::now();
+  for (size_t e = 0; e < num_updates; e++)
+    g.update(stream.get_edge());
+
+  auto cc_start = std::chrono::steady_clock::now();
+  auto CC_num = g.connected_components().size();
+  std::chrono::duration<double> insert_time = g.flush_end - ins_start;
+  std::chrono::duration<double> cc_time = std::chrono::steady_clock::now() - cc_start;
+  double num_seconds = insert_time.count();
+  std::cout << "Total insertion time was: " << num_seconds << std::endl;
+  std::cout << "Insertion rate was:       " << stream.edges() / num_seconds << std::endl;
+  std::cout << "CC query latency:         " << cc_time.count() << std::endl;
+  std::cout << "Connected Components:     " << CC_num << std::endl;
+}

tools/experiment/BCHMK_post_proc.cpp

@@ -1,26 +1,43 @@
 #include <iostream>
 #include <fstream>
 #include <vector>
+#include <errno.h>
+#include <string.h>
 #include <graph_zeppelin_common.h>
 
 int main(int argc, char **argv) {
-  if (argc != 2 && argc != 3) {
+  if (argc < 3 || argc > 5) {
     std::cout << "Incorrect number of arguments. "
-                 "Expected at either one or two but got " << argc-1 << std::endl;
-    std::cout << "Arguments are: text_stream [update_type]" << std::endl;
-    std::cout << "text_stream is the file to parse into binary format" << std::endl;
-    std::cout << "update_type is a flag. If present then stream indicates insertions vs deletions" << std::endl;
+                 "Expected [2-4] but got " << argc-1 << std::endl;
+    std::cout << "Arguments are: ascii_stream out_file_name [--update_type] [--verbose]" << std::endl;
+    std::cout << "ascii_stream:  The file to parse into binary format" << std::endl;
+    std::cout << "out_file_name: Where the binary stream will be written" << std::endl;
+    std::cout << "--update_type: If present then ascii stream indicates insertions vs deletions" << std::endl;
+    std::cout << "--silent:      If present then no warnings are printed when stream corrections are made" << std::endl;
+    exit(EXIT_FAILURE);
   }
 
   std::ifstream txt_file(argv[1]);
-  std::ofstream out_file("binary_stream.data", std::ios_base::binary);
+  if (!txt_file) {
+    std::cerr << "ERROR: could not open input file!" << std::endl;
+    exit(EXIT_FAILURE);
+  }
+  std::ofstream out_file(argv[2], std::ios_base::binary | std::ios_base::out);
+  if (!out_file) {
+    std::cerr << "ERROR: could not open output file! " << argv[2] << ": " << strerror(errno) << std::endl;
+    exit(EXIT_FAILURE);
+  }
 
   bool update_type = false;
-  if (argc == 3) {
-    if (std::string(argv[2]) == "update_type")
+  bool silent = false;
+  for (int i = 3; i < argc; i++) {
+    if (std::string(argv[i]) == "--update_type")
       update_type = true;
+    else if (std::string(argv[i]) == "--silent") {
+      silent = true;
+    }
     else {
-      std::cerr << "Did not recognize second argument! Expected 'update_type'";
+      std::cerr << "Did not recognize argument: " << argv[i] << " Expected '--update_type' or '--silent'";
       return EXIT_FAILURE;
     }
   }
@@ -29,6 +46,16 @@ int main(int argc, char **argv) {
   edge_id_t num_edges;
 
   txt_file >> num_nodes >> num_edges;
+  
+  std::cout << "Parsed ascii stream header. . ." << std::endl;
+  std::cout << "Number of nodes:   " << num_nodes << std::endl;
+  std::cout << "Number of updates: " << num_edges << std::endl; 
+  if (update_type)
+    std::cout << "Assuming that update format is: upd_type src dst" << std::endl;
+  else
+    std::cout << "Assuming that update format is: src dst" << std::endl;
+  
+
   out_file.write((char *) &num_nodes, sizeof(num_nodes));
   out_file.write((char *) &num_edges, sizeof(num_edges));
 
@@ -48,14 +75,14 @@ int main(int argc, char **argv) {
       txt_file >> src >> dst;
 
     if (src > dst) {
-      if (u != adj_mat[dst][src - dst]) {
+      if (!silent && u != adj_mat[dst][src - dst]) {
         std::cout << "WARNING: update " << u << " " << src << " " << dst;
         std::cout << " is double insert or delete before insert. Correcting." << std::endl;
       }
       u = adj_mat[dst][src - dst];
       adj_mat[dst][src - dst] = !adj_mat[dst][src - dst];
     } else {
-      if (u != adj_mat[src][dst - src]) {
+      if (!silent && u != adj_mat[src][dst - src]) {
         std::cout << "WARNING: update " << u << " " << src << " " << dst;
         std::cout << " is double insert or delete before insert. Correcting." << std::endl;
       }
@@ -68,3 +95,4 @@ int main(int argc, char **argv) {
     out_file.write((char *) &dst, sizeof(dst));
   }
 }
+