Parallel search benchmark (WIP)

applications/CMakeLists.txt

@@ -8,3 +8,4 @@ add_subdirectory(join)
 add_subdirectory(isopath)
 add_subdirectory(graphlab)
 add_subdirectory(util)
+add_subdirectory(hpcc)

applications/hpcc/CMakeLists.txt

@@ -0,0 +1,4 @@
+#set(SOURCES main.cpp common.hpp)
+
+	add_grappa_application(hpcc_random_access.exe hpcc_random_access.cpp)
+

applications/hpcc/hpcc_random_access.cpp

@@ -0,0 +1,118 @@
+// to run, do something like
+// make -j demo-hpcc_random_access
+// bin/grappa_run --ppn 8 --nnode 12 -- demo-hpcc_random_access.exe 
+
+#include <Grappa.hpp>
+#include <iomanip>
+
+using namespace Grappa;
+
+// define command-line flags (third-party 'gflags' library)
+DEFINE_int64( scale, 20, "table size = 2 ^ scale * sizeof(uint64_t) * number of PEs" );
+DEFINE_int64( iters, 4, "Number of iterations" );
+
+// define custom statistics which are logged by the runtime
+// (here we're not using these features, just printing them ourselves)
+GRAPPA_DEFINE_METRIC( SimpleMetric<double>, gups_runtime, 0.0 );
+GRAPPA_DEFINE_METRIC( SimpleMetric<double>, gups_throughput, 0.0 );
+
+#define POLY 0x0000000000000007ULL
+#define PERIOD 1317624576693539401LL
+
+typedef double time_type;
+std::string print_time(time_type t)
+{
+	std::ostringstream out;
+	out << std::setiosflags(std::ios::fixed) << std::setprecision(2) << t;
+	return out.str();
+}
+
+
+uint64_t N;
+uint64_t HPCC_starts(int64_t n);
+
+Grappa::GlobalCompletionEvent gce;
+
+double run_random_access() {
+	LOG(INFO) << "HPCC RandomAccess" << std::endl;
+	N = (1LL << FLAGS_scale) * cores();
+
+	// create target array that we'll be updating
+	auto hpcc_table = global_alloc<int64_t>(N);
+	Grappa::memset( hpcc_table, 0, N);
+
+	double tstart = walltime();
+
+	on_all_cores([hpcc_table] {
+		N = (1LL << FLAGS_scale) * cores();
+		uint64_t key = HPCC_starts(FLAGS_iters * mycore() * N / cores());
+		for(int k = 0; k < FLAGS_iters; k++)
+			for(uint64_t i = 0; i < N / cores(); i++) {
+				key = key << 1 ^ ((int64_t) key < 0 ? POLY : 0);
+				auto addr = hpcc_table + (key & N - 1);
+				//auto core = key >> (int)log2((double)(N / cores())) & cores() - 1;
+				delegate::call<async, &gce>(addr.core(), [addr, key] {
+					//uint64_t offset = key & (N / cores() - 1);
+					*(addr.pointer()) ^= key;
+				});
+			}
+	});
+	gce.wait();
+
+	double tend = walltime();
+
+	LOG(INFO) << "\tTime elapsed " << (double)(tend - tstart) << " sec" << std::endl;
+
+	global_free(hpcc_table);
+
+	return (double)(tend - tstart);
+}
+
+uint64_t HPCC_starts(int64_t n) {
+	int i, j;
+	uint64_t m2[64];
+	uint64_t temp, ran;
+	while (n < 0) n += PERIOD;
+	while (n > PERIOD) n -= PERIOD;
+	if (n == 0) return 0x1;
+	temp = 0x1;
+	for (i = 0; i < 64; i++) {
+		m2[i] = temp;
+		temp = temp << 1 ^ ((int64_t) temp < 0 ? POLY : 0);
+		temp = temp << 1 ^ ((int64_t) temp < 0 ? POLY : 0);
+	}
+	for (i = 62; i >= 0; i--)
+		if (n >> i & 1)
+			break;
+
+	ran = 0x2;
+	while (i > 0) {
+		temp = 0;
+		for (j = 0; j < 64; j++)
+			if (ran >> j & 1)
+				temp ^= m2[j];
+		ran = temp;
+		i -= 1;
+		if (n >> i & 1)
+			ran = ran << 1 ^ ((int64_t) ran < 0 ? POLY : 0);
+	}
+	return ran;
+}
+
+
+int main(int argc, char * argv[]) {
+  init( &argc, &argv );
+  run([]{
+
+		LOG(INFO) << "\tGlobal table size   = 2^" << FLAGS_scale << " * " << cores() << " = " << (1LL << FLAGS_scale) * cores() << " words\n";
+		LOG(INFO) << "\tNumber of processes = " << cores() << std::endl;
+		LOG(INFO) << "\tNumber of updates = " << FLAGS_iters * (1LL << FLAGS_scale) * cores() << std::endl;
+
+		gups_runtime = run_random_access();
+    gups_throughput = 1e-9 * FLAGS_iters * N / gups_runtime;
+
+    LOG(INFO) << "[Final] CPU time used " << print_time(gups_runtime.value()) << " seconds, " << print_time(gups_throughput.value()) << " GUPS\n";
+
+  });
+  finalize();
+}

applications/nativegraph/CMakeLists.txt

@@ -1,3 +1,4 @@
 add_subdirectory(bfs)
 add_subdirectory(cc)
 add_subdirectory(sssp)
+add_subdirectory(parallel_search)

applications/nativegraph/parallel_search/CMakeLists.txt

@@ -0,0 +1,3 @@
+set(SOURCES main.cpp common.hpp)
+
+add_grappa_application(parallel_search.exe parallel_search.cpp ${SOURCES})

applications/nativegraph/parallel_search/common.hpp

@@ -0,0 +1,39 @@
+#include <Grappa.hpp>
+#include <GlobalVector.hpp>
+#include <graph/Graph.hpp>
+#include "../verifier.hpp"
+
+using namespace Grappa;
+
+// additional data to attach to each vertex in the graph
+struct BFSData {
+  int64_t parent;
+  int64_t level;
+  bool seen;
+
+  void init() {
+    parent = -1;
+    level = 0;
+    seen = false;
+  }
+};
+
+using G = Graph<BFSData,Empty>;
+
+extern int64_t nedge_traversed;
+
+void bfs(GlobalAddress<G> g, int nbfs, TupleGraph tg);
+
+template< typename V, typename E >
+inline int64_t choose_root(GlobalAddress<Graph<V,E>> g) {
+  int64_t root;
+  do {
+    root = random() % g->nv;
+  } while (delegate::call(g->vs+root,[](typename G::Vertex& v){ return v.nadj; }) == 0);
+  return root;
+}
+
+inline int64_t verify(TupleGraph tg, GlobalAddress<G> g, int64_t root) {
+  return VerificatorBase<G>::verify(tg, g, root);
+}
+

applications/nativegraph/parallel_search/igor_bfs.rb

@@ -0,0 +1,51 @@
+#!/usr/bin/env ruby
+require 'igor'
+
+# inherit parser, sbatch_flags
+require_relative '../../util/igor_common.rb'
+
+Igor do
+  include Isolatable
+
+  database '~/osdi.sqlite', :bfs
+
+  # isolate everything needed for the executable so we can sbcast them for local execution
+  isolate(['bfs_bags.exe'])
+
+  GFLAGS.merge!({
+    path: '/pic/projects/grappa/twitter/bintsv4/twitter-all.bintsv4',
+    format: 'bintsv4',
+    nbfs: 3,
+    beamer_alpha: 20.0,
+    beamer_beta: 20.0,
+  })
+  GFLAGS.delete :flat_combining
+
+  params.merge!(GFLAGS)
+
+  @c = ->{ %Q[ %{tdir}/grappa_srun --no-freeze-on-error
+    -- %{tdir}/bfs_bags.exe --metrics --max_degree_source --vmodule bfs_bags=1
+    #{GFLAGS.expand}
+  ].gsub(/\s+/,' ') }
+  command @c[]
+
+  sbatch_flags << "--time=30:00"
+
+  params {
+    grappa_version 'beamer'
+    nnode       16
+    ppn         32
+    loop_threshold 1024
+    num_starting_workers 512
+    # aggregator_autoflush_ticks 3e6.to_i
+    # periodic_poll_ticks 2e5.to_i
+    shared_pool_chunk_size 2**15
+    global_heap_fraction 0.25
+  }
+
+  expect :total_time
+  @cols << :total_time_mean
+  @order = :total_time_mean
+
+  interact # enter interactive mode
+end

applications/nativegraph/parallel_search/main.cpp

@@ -0,0 +1,95 @@
+////////////////////////////////////////////////////////////////////////
+// Copyright (c) 2010-2015, University of Washington and Battelle
+// Memorial Institute.  All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions
+// are met:
+//     * Redistributions of source code must retain the above
+//       copyright notice, this list of conditions and the following
+//       disclaimer.
+//     * Redistributions in binary form must reproduce the above
+//       copyright notice, this list of conditions and the following
+//       disclaimer in the documentation and/or other materials
+//       provided with the distribution.
+//     * Neither the name of the University of Washington, Battelle
+//       Memorial Institute, or the names of their contributors may be
+//       used to endorse or promote products derived from this
+//       software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+// FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+// UNIVERSITY OF WASHINGTON OR BATTELLE MEMORIAL INSTITUTE BE LIABLE
+// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
+// OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+// BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
+// DAMAGE.
+////////////////////////////////////////////////////////////////////////
+
+////////////////////////////////////////////////////////////////////////
+/// Simon Kahan's 3-phase Connected Components (for Grappa Graph)
+////////////////////////////////////////////////////////////////////////
+
+#include <Grappa.hpp>
+#include "common.hpp"
+
+DEFINE_bool( metrics, false, "Dump metrics");
+
+DEFINE_int32(scale, 10, "Log2 number of vertices.");
+DEFINE_int32(edgefactor, 16, "Average number of edges per vertex.");
+DEFINE_int32(nbfs, 1, "Number of BFS traversals to do.");
+
+DEFINE_string(path, "", "Path to graph source file.");
+DEFINE_string(format, "bintsv4", "Format of graph source file.");
+
+GRAPPA_DEFINE_METRIC(SimpleMetric<double>, init_time, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<double>, tuple_time, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<double>, construction_time, 0);
+
+GRAPPA_DEFINE_METRIC(SummarizingMetric<double>, bfs_mteps, 0);
+GRAPPA_DEFINE_METRIC(SummarizingMetric<double>, total_time, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<int64_t>, bfs_nedge, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<double>, verify_time, 0);
+
+int64_t nedge_traversed;
+
+int main(int argc, char* argv[]) {
+  init(&argc, &argv);
+  run([]{
+
+    TupleGraph tg;
+
+    GRAPPA_TIME_REGION(tuple_time) {
+      if (FLAGS_path.empty()) {
+        int64_t NE = (1L << FLAGS_scale) * FLAGS_edgefactor;
+        tg = TupleGraph::Kronecker(FLAGS_scale, NE, 111, 222);
+      } else {
+        LOG(INFO) << "loading " << FLAGS_path;
+        tg = TupleGraph::Load(FLAGS_path, FLAGS_format);
+      }
+    }
+    LOG(INFO) << tuple_time;
+    LOG(INFO) << "constructing graph";
+
+    double t = walltime();
+
+    // construct the compact graph representation (roughly CSR)
+    auto g = G::Undirected( tg );
+
+    construction_time = (walltime()-t);
+    LOG(INFO) << construction_time;
+
+    bfs(g, FLAGS_nbfs, tg);
+
+    LOG(INFO) << "\n" << bfs_nedge << "\n" << total_time << "\n" << bfs_mteps;
+    if (FLAGS_metrics) Metrics::merge_and_print();
+    Metrics::merge_and_dump_to_file();
+  });
+  finalize();
+}