Extracted BenchStandBuilder

Author: faucct

benchmark/ansible/flamestream.yml

@@ -47,7 +47,7 @@
 - name: Run the benchmark
   hosts: bench
   roles:
-    - { role: flamestream-bench , bench_class: com.spbsu.flamestream.example.benchmark.LentaBenchStand }
+    - { role: flamestream-lenta-bench , bench_class: com.spbsu.flamestream.example.benchmark.LentaBenchStand }
 
 - name: Destroy flamestream
   hosts: workers

benchmark/ansible/group_vars/all.yml

@@ -12,7 +12,7 @@ tarball_dir: /opt/src
 
 results_dir: results/{{ lookup("pipe", "git rev-parse --short HEAD") }}/{{ groups['workers'] | length }}/{{ rate }}/{{ guarantees }}/{{ millis_between_commits }}
 
-zookeeper_version: 3.4.13
+zookeeper_version: 3.4.14
 zookeeper_client_port: 2181
 
 flamestream_worker_port: 5412

benchmark/flink-benchmark/src/main/java/com/spbsu/benchmark/flink/index/FlinkBench.java

@@ -6,7 +6,7 @@
 import com.spbsu.benchmark.flink.index.ops.OrderEnforcer;
 import com.spbsu.benchmark.flink.index.ops.TwoPCKryoSocketSink;
 import com.spbsu.benchmark.flink.index.ops.WikipediaPageToWordPositions;
-import com.spbsu.flamestream.example.benchmark.BenchStand;
+import com.spbsu.flamestream.example.benchmark.WikiBenchStand;
 import com.spbsu.flamestream.example.benchmark.GraphDeployer;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
@@ -24,7 +24,7 @@
 import java.nio.file.Paths;
 
 public class FlinkBench {
-  public static void main(String[] args) throws IOException, InterruptedException {
+  public static void main(String[] args) throws Exception {
     final Config benchConfig;
     final Config deployerConfig;
     if (args.length == 2) {
@@ -34,9 +34,8 @@ public static void main(String[] args) throws IOException, InterruptedException
       benchConfig = ConfigFactory.load("flink-bench.conf").getConfig("benchmark");
       deployerConfig = ConfigFactory.load("flink-deployer.conf").getConfig("deployer");
     }
-    final BenchStand.StandConfig standConfig = new BenchStand.StandConfig(benchConfig);
-
-    final GraphDeployer deployer = new GraphDeployer() {
+    WikiBenchStand wikiBenchStand = new WikiBenchStand(benchConfig);
+    wikiBenchStand.run(new GraphDeployer() {
       @Override
       public void deploy() {
         final int parallelism = deployerConfig.getInt("parallelism");
@@ -59,12 +58,12 @@ public void deploy() {
         final SinkFunction<Result> sinkFunction;
         if (guarantees.equals("EXACTLY_ONCE")) {
           sinkFunction = new TwoPCKryoSocketSink(
-                  standConfig.benchHost(),
-                  standConfig.rearPort(),
+                  wikiBenchStand.benchHost,
+                  wikiBenchStand.rearPort,
                   environment.getConfig()
           );
         } else {
-          sinkFunction = new KryoSocketSink(standConfig.benchHost(), standConfig.rearPort());
+          sinkFunction = new KryoSocketSink(wikiBenchStand.benchHost, wikiBenchStand.rearPort);
         }
 
         if (guarantees.equals("EXACTLY_ONCE") || guarantees.equals("AT_LEAST_ONCE")) {
@@ -81,7 +80,7 @@ public void deploy() {
 
 
         environment
-                .addSource(new KryoSocketSource(standConfig.benchHost(), standConfig.frontPort()))
+                .addSource(new KryoSocketSource(wikiBenchStand.benchHost, wikiBenchStand.frontPort))
                 .setParallelism(parallelism)
                 .shuffle()
                 .flatMap(new WikipediaPageToWordPositions())
@@ -105,11 +104,7 @@ public void deploy() {
       public void close() {
         // It will close itself on completion
       }
-    };
-
-    try (BenchStand benchStand = new BenchStand(standConfig, deployer)) {
-      benchStand.run();
-    }
+    });
     System.exit(0);
   }
 }