Fixed calculation of throughput and added output for job duration

nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/broadcast/BroadcastJob.java

@@ -166,19 +166,21 @@ public static void main(String[] args) {
  }
 
  try {
- final BufferedWriter writer = new BufferedWriter(new FileWriter(getFilename()));
+ final BufferedWriter throughputWriter = new BufferedWriter(new FileWriter(getThroughputFilename()));
+ final BufferedWriter durationWriter = new BufferedWriter(new FileWriter(getDurationFilename()));
 
  // Execute the individual job runs
  for (int i = 0; i < NUMBER_OF_RUNS; ++i) {
  try {
- runJob(i, writer);
+ runJob(i, throughputWriter, durationWriter);
  } catch (Exception e) {
  System.err.println("Error executing run " + i + ": " + StringUtils.stringifyException(e));
  break;
  }
  }
 
- writer.close();
+ throughputWriter.close();
+ durationWriter.close();
  } catch (IOException ioe) {
  System.err.println("An IO exception occurred " + StringUtils.stringifyException(ioe));
  }
@@ -194,11 +196,13 @@ public static void main(String[] args) {
  * 
  * @param run
  * the run of the job
- * @param outputWriter
+ * @param throughputWriter
  * writer object to write the throughput results for each run
+ * @param durationWriter
+ * writer object to write the duration results for each run
  */
- private static void runJob(final int run, final BufferedWriter outputWriter) throws JobGraphDefinitionException,
- IOException, JobExecutionException {
+ private static void runJob(final int run, final BufferedWriter throughputWriter, final BufferedWriter durationWriter)
+ throws JobGraphDefinitionException, IOException, JobExecutionException {
 
  // Construct job graph
  final JobGraph jobGraph = new JobGraph("Broadcast Job (Run " + run + ")");
@@ -236,22 +240,32 @@ private static void runJob(final int run, final BufferedWriter outputWriter) thr
  final JobClient jobClient = new JobClient(jobGraph, conf);
  final long jobDuration = jobClient.submitJobAndWait();
 
- final long numberOfBytesSent = BroadcastRecord.RECORD_SIZE * NUMBER_OF_RECORDS * NUMBER_OF_CONSUMERS;
+ final long numberOfBytesSent = (long) BroadcastRecord.RECORD_SIZE * (long) NUMBER_OF_RECORDS
+ * (long) NUMBER_OF_CONSUMERS;
  // Throughput in bits per second
- final long throughput = numberOfBytesSent / jobDuration * 1000 * 8;
+ final double throughput = (double) (numberOfBytesSent * 1000L * 8L) / (double) (jobDuration * 1024L * 1024L);
 
  // Write calculated throughput to file
- outputWriter.write(throughput + "\n");
+ throughputWriter.write(throughput + "\n");
+
+ // Write the job duration
+ durationWriter.write(jobDuration + "\n");
  }
 
  /**
  * Constructs the filename for the throughput result.
  * 
  * @return the filename for the throughput result
  */
- private static String getFilename() {
+ private static String getThroughputFilename() {
 
  return OUTPUT_PATH + File.separator + "throughput_" + INSTANCE_TYPE + "_" + TOPOLOGY_TREE + "_"
  + NUMBER_OF_CONSUMERS + "_" + NUMBER_OF_RECORDS + ".dat";
  }
+
+ private static String getDurationFilename() {
+
+ return OUTPUT_PATH + File.separator + "duration_" + INSTANCE_TYPE + "_" + TOPOLOGY_TREE + "_"
+ + NUMBER_OF_CONSUMERS + "_" + NUMBER_OF_RECORDS + ".dat";
+ }
 }