Adding intermediate results

Mobilyzer/src/com/mobilyzer/MeasurementScheduler.java

@@ -185,6 +185,9 @@ public void onCreate() {
     filter.addAction(UpdateIntent.MEASUREMENT_PROGRESS_UPDATE_ACTION);
     filter.addAction(UpdateIntent.GCM_MEASUREMENT_ACTION);
     filter.addAction(UpdateIntent.PLT_MEASUREMENT_ACTION);
+    //added by Clarence
+    filter.addAction(UpdateIntent.MEASUREMENT_INTERMEDIATE_PROGRESS_UPDATE_ACTION);
+
 
     broadcastReceiver = new BroadcastReceiver() {
 
@@ -323,7 +326,20 @@ public void onReceive(Context context, Intent intent) {
           } else if (intent.getStringExtra(UpdateIntent.TASK_STATUS_PAYLOAD).equals(
               Config.TASK_RESUMED)) {
             tasksStatus.put(taskid, TaskStatus.RUNNING);
-          }
+          }  //added by Clarence
+        } else if (intent.getAction().equals(UpdateIntent.MEASUREMENT_INTERMEDIATE_PROGRESS_UPDATE_ACTION)){
+        	String IM_taskKey = intent.getStringExtra(UpdateIntent.CLIENTKEY_PAYLOAD);
+            String IM_taskid = intent.getStringExtra(UpdateIntent.TASKID_PAYLOAD);
+            int IM_priority =
+                intent.getIntExtra(UpdateIntent.TASK_PRIORITY_PAYLOAD,
+                    MeasurementTask.INVALID_PRIORITY);
+            Parcelable[] IM_Results = intent.getParcelableArrayExtra(UpdateIntent.INTERMEDIATE_RESULT_PAYLOAD);
+            if (IM_Results != null && IM_Results.length!=0) {
+                sendResultToClient(IM_Results, IM_priority, IM_taskKey, IM_taskid);
+                Logger.i("Sending Intermediate results to client...");
+            	System.out.println("receive the intermediate results");
+            }
+
         } else if (intent.getAction().equals(UpdateIntent.CHECKIN_ACTION)
             || intent.getAction().equals(UpdateIntent.CHECKIN_RETRY_ACTION)) {
           Logger.d("Checkin intent received");

Mobilyzer/src/com/mobilyzer/MeasurementTask.java

@@ -29,6 +29,8 @@ public abstract class MeasurementTask
       Parcelable {
   protected MeasurementDesc measurementDesc;
   protected String taskId;
+
+  private MeasurementScheduler scheduler; // added by Clarence 
 
 
   public static final int USER_PRIORITY = Integer.MIN_VALUE;
@@ -114,6 +116,15 @@ public String getKey() {
   public void setKey(String key) {
     this.measurementDesc.key = key;
   }
+
+  // added by Clarence, pass scheduler to every specific task
+  public void setScheduler(MeasurementScheduler scheduler) {
+	    this.scheduler = scheduler;
+  }
+
+  public MeasurementScheduler getScheduler() {
+	  return this.scheduler;
+  }
 
 
   public MeasurementDesc getDescription() {

Mobilyzer/src/com/mobilyzer/ServerMeasurementTask.java

@@ -38,6 +38,7 @@ public class ServerMeasurementTask implements Callable<MeasurementResult[]> {
 	public ServerMeasurementTask(MeasurementTask task,
 			MeasurementScheduler scheduler, ResourceCapManager manager) {
 		realTask = task;
+		realTask.setScheduler(null);  //added by Clarence
 		this.scheduler = scheduler;
 		this.contextCollector = new ContextCollector();
 		this.rManager = manager;

Mobilyzer/src/com/mobilyzer/UpdateIntent.java

@@ -44,6 +44,8 @@ public class UpdateIntent extends Intent {
   public static final String VIDEO_TASK_PAYLOAD_REBUFFER_TIME = "VIDEO_TASK_PAYLOAD_REBUFFER_TIME";
   public static final String VIDEO_TASK_PAYLOAD_BBA_SWITCH_TIME = "VIDEO_TASK_PAYLOAD_BBA_SWITCH_TIME";
   public static final String VIDEO_TASK_PAYLOAD_BYTE_USED = "VIDEO_TASK_PAYLOAD_BYTE_USED";
+  //added by Clarence
+  public static final String INTERMEDIATE_RESULT_PAYLOAD = "INTERMEDIATE_RESULT_PAYLOAD";
 
 
   // Different types of actions that this intent can represent:
@@ -81,6 +83,11 @@ public class UpdateIntent extends Intent {
       PACKAGE_PREFIX + ".DATA_USAGE_ACTION";
   public static final String AUTH_ACCOUNT_ACTION =
       PACKAGE_PREFIX + ".AUTH_ACCOUNT_ACTION";
+
+
+  // added by Clarence
+  public static final String MEASUREMENT_INTERMEDIATE_PROGRESS_UPDATE_ACTION =
+	      APP_PREFIX + ".MEASUREMENT_INTERMEDIATE_PROGRESS_UPDATE_ACTION";
 
   /**
    * Creates an intent of the specified action with an optional message

Mobilyzer/src/com/mobilyzer/UserMeasurementTask.java

@@ -18,6 +18,7 @@
 import java.util.HashMap;
 import java.util.concurrent.Callable;
 
+import android.content.Context;
 import android.content.Intent;
 
 import com.mobilyzer.MeasurementResult.TaskProgress;
@@ -33,6 +34,7 @@ public class UserMeasurementTask implements Callable<MeasurementResult[]> {
   public UserMeasurementTask(MeasurementTask task,
                              MeasurementScheduler scheduler) {
     realTask = task;
+    realTask.setScheduler(scheduler);  //added by Clarence
     this.scheduler = scheduler;  
     this.contextCollector= new ContextCollector();
   }
@@ -48,7 +50,10 @@ private void broadcastMeasurementStart() {
     intent.putExtra(UpdateIntent.TASK_STATUS_PAYLOAD, Config.TASK_STARTED);
     intent.putExtra(UpdateIntent.TASKID_PAYLOAD, realTask.getTaskId());
     intent.putExtra(UpdateIntent.CLIENTKEY_PAYLOAD, realTask.getKey());
+//    Context context = scheduler.getApplicationContext();
+//    context.sendBroadcast(intent);
     scheduler.sendBroadcast(intent);
+
   }
 
   /**
@@ -96,9 +101,12 @@ public MeasurementResult[] call() throws MeasurementError {
       ArrayList<HashMap<String, String>> contextResults =
           contextCollector.stopCollector();
       for (MeasurementResult r: results){
+    	String testIntermediate1 = r.toString();
         r.addContextResults(contextResults);
         r.getDeviceProperty().dnResolvability=contextCollector.dnsConnectivity;
         r.getDeviceProperty().ipConnectivity=contextCollector.ipConnectivity;
+        String testIntermediate2 = r.toString();
+
       } 
     } catch (MeasurementError e) {
       Logger.e("User measurement " + realTask.getDescriptor() + " has failed");

Mobilyzer/src/com/mobilyzer/measurements/ParallelTask.java

@@ -165,6 +165,9 @@ public String getDescriptor() {
 
   @Override
   public MeasurementResult[] call() throws MeasurementError {
+	 for (MeasurementTask task: this.tasks){
+		 task.setScheduler(this.getScheduler());
+	 }           //added by Clarence
     long timeout=duration;
     executor=Executors.newFixedThreadPool(this.tasks.size());
 

Mobilyzer/src/com/mobilyzer/measurements/PingTask.java

@@ -15,6 +15,7 @@
 
 package com.mobilyzer.measurements;
 
+import android.content.Intent;
 import android.os.Parcel;
 import android.os.Parcelable;
 import android.util.Log;
@@ -23,7 +24,9 @@
 import com.mobilyzer.Config;
 import com.mobilyzer.MeasurementDesc;
 import com.mobilyzer.MeasurementResult;
+import com.mobilyzer.MeasurementScheduler;
 import com.mobilyzer.MeasurementTask;
+import com.mobilyzer.UpdateIntent;
 import com.mobilyzer.MeasurementResult.TaskProgress;
 import com.mobilyzer.exceptions.MeasurementError;
 import com.mobilyzer.util.Logger;
@@ -71,6 +74,32 @@ public class PingTask extends MeasurementTask{
   //Track data consumption for this task to avoid exceeding user's limit  
   private long dataConsumed;
 
+
+  private MeasurementScheduler scheduler = null;  // added by Clarence
+
+  private void broadcastIntermediateMeasurement(MeasurementResult[] results, MeasurementScheduler scheduler) {
+	  this.scheduler = scheduler;
+      Intent intent = new Intent();
+      intent.setAction(UpdateIntent.MEASUREMENT_INTERMEDIATE_PROGRESS_UPDATE_ACTION);
+      //TODO fixed one value priority for all users task?
+      intent.putExtra(UpdateIntent.TASK_PRIORITY_PAYLOAD,
+        MeasurementTask.USER_PRIORITY);
+      intent.putExtra(UpdateIntent.TASKID_PAYLOAD, this.getTaskId());
+      intent.putExtra(UpdateIntent.CLIENTKEY_PAYLOAD, this.getKey());
+
+      if (results != null){
+
+        //intent.putExtra(UpdateIntent.TASK_STATUS_PAYLOAD, Config.TASK_FINISHED);
+        intent.putExtra(UpdateIntent.INTERMEDIATE_RESULT_PAYLOAD, results);
+
+      this.scheduler.sendBroadcast(intent);
+      }else{
+    	 intent.putExtra(UpdateIntent.INTERMEDIATE_RESULT_PAYLOAD, "No intermediate results are broadcasted");
+
+     }
+
+    }
+
   /**
    * Encode ping specific parameters, along with common parameters inherited from MeasurmentDesc
    * @author wenjiezeng@google.com (Steve Zeng)
@@ -356,6 +385,11 @@ private MeasurementResult executePingCmdTask(int ipByteLen)
     PingDesc pingTask = (PingDesc) this.measurementDesc;
     String errorMsg = "";
     MeasurementResult measurementResult = null;
+
+    MeasurementResult IM_measurementResult = null;  //added by Clarence
+    MeasurementResult[] IM_result = null;
+    IM_result = new MeasurementResult[1];
+
     // TODO(Wenjie): Add a exhaustive list of ping locations for different
     //               Android phones
     pingTask.pingExe = Util.pingExecutableBasedOnIPType(ipByteLen);
@@ -384,11 +418,13 @@ private MeasurementResult executePingCmdTask(int ipByteLen)
       ArrayList<Integer> receivedIcmpSeq = new ArrayList<Integer>();
       double packetLoss = Double.MIN_VALUE;
       int packetsSent = Config.PING_COUNT_PER_MEASUREMENT;
+
       // Process each line of the ping output and store the rrt in array rrts.
       while ((line = br.readLine()) != null) {
         // Ping prints a number of 'param=value' pairs, among which we only need
         // the 'time=rrt_val' pair
         String[] extractedValues = Util.extractInfoFromPingOutput(line);
+
         if (extractedValues != null) {
           int curIcmpSeq = Integer.parseInt(extractedValues[0]);
           double rrtVal = Double.parseDouble(extractedValues[1]);
@@ -409,6 +445,18 @@ private MeasurementResult executePingCmdTask(int ipByteLen)
           int packetsReceived = packetLossInfo[1];
           packetLoss = 1 - ((double) packetsReceived / (double) packetsSent);
         }
+        this.scheduler = this.getScheduler();
+        if ( this.scheduler != null ){
+        	if (rrts.size() >= 2  && (rrts.size() < Config.PING_COUNT_PER_MEASUREMENT) && (extractedValues != null) ){
+        		IM_measurementResult = constructResult(rrts,packetLoss,
+            			rrts.size(),PING_METHOD_CMD);
+        		IM_result[0] = IM_measurementResult;
+        		broadcastIntermediateMeasurement(IM_result,this.scheduler);
+
+
+        	}
+        }
+        //IM_packetsSent++;
 
         Logger.i(line);
       }
@@ -452,6 +500,10 @@ private MeasurementResult executeJavaPingTask() throws MeasurementError {
     ArrayList<Double> rrts = new ArrayList<Double>();
     String errorMsg = "";
     MeasurementResult result = null;
+    double packetLoss = Double.MIN_VALUE;           //added by Clarence
+    MeasurementResult IM_measurementResult = null;  //added by Clarence
+    MeasurementResult[] IM_result = null;
+    IM_result = new MeasurementResult[1];
 
     try {       
       int timeOut = (int) (3000 * (double) pingTask.pingTimeoutSec /
@@ -466,11 +518,28 @@ private MeasurementResult executeJavaPingTask() throws MeasurementError {
         if (status) {
           totalPingDelay += rrtVal;
           rrts.add((double) rrtVal);
+          packetLoss = 1 -
+                ((double) rrts.size() / (i+1));
+          dataConsumed += pingTask.packetSizeByte * (i+1) * 2;
+          this.scheduler = this.getScheduler();
+          if ( this.scheduler != null){
+        	  IM_measurementResult = constructResult(rrts,packetLoss,
+            			(i+1),PING_METHOD_JAVA);
+              IM_result[0] = IM_measurementResult;
+              broadcastIntermediateMeasurement(IM_result,this.scheduler);
+
+          }
+
+
         }
       }
       Logger.i("java ping succeeds");
-      double packetLoss = 1 -
-          ((double) rrts.size() / (double) Config.PING_COUNT_PER_MEASUREMENT);
+//      double packetLoss = 1 -
+//          ((double) rrts.size() / (double) Config.PING_COUNT_PER_MEASUREMENT);  //deleted by Clarence
+      if (packetLoss == Double.MIN_VALUE) {
+          packetLoss = 1 - 
+              ((double) rrts.size() / (double) Config.PING_COUNT_PER_MEASUREMENT);
+        }
 
       dataConsumed += pingTask.packetSizeByte * Config.PING_COUNT_PER_MEASUREMENT * 2;
 
@@ -505,6 +574,10 @@ private MeasurementResult executeHttpPingTask() throws MeasurementError {
     PingDesc pingTask = (PingDesc) this.measurementDesc;
     String errorMsg = "";
     MeasurementResult result = null;
+    double packetLoss = Double.MIN_VALUE;           //added by Clarence
+    MeasurementResult IM_measurementResult = null;  //added by Clarence
+    MeasurementResult[] IM_result = null;
+    IM_result = new MeasurementResult[1];
 
     try {
       long totalPingDelay = 0;
@@ -525,12 +598,27 @@ private MeasurementResult executeHttpPingTask() throws MeasurementError {
         pingEndTime = System.currentTimeMillis();
         httpClient.disconnect();
         rrts.add((double) (pingEndTime - pingStartTime));
+        packetLoss = 1 -
+                ((double) rrts.size() / (i+1));
+        dataConsumed += pingTask.packetSizeByte * (i+1) * 2;
+        this.scheduler = this.getScheduler();
+        if ( this.scheduler != null){
+        	  IM_measurementResult = constructResult(rrts,packetLoss,
+            			(i+1),PING_METHOD_HTTP);
+              IM_result[0] = IM_measurementResult;
+              broadcastIntermediateMeasurement(IM_result,this.scheduler);
+
+         }
 
       }
       Logger.i("HTTP get ping succeeds");
       Logger.i("RTT is " + rrts.toString());
-      double packetLoss = 1
-          - ((double) rrts.size() / (double) Config.PING_COUNT_PER_MEASUREMENT);
+//      double packetLoss = 1
+//          - ((double) rrts.size() / (double) Config.PING_COUNT_PER_MEASUREMENT);
+      if (packetLoss == Double.MIN_VALUE) {
+          packetLoss = 1 - 
+              ((double) rrts.size() / (double) Config.PING_COUNT_PER_MEASUREMENT);
+        }
 
       dataConsumed += pingTask.packetSizeByte * Config.PING_COUNT_PER_MEASUREMENT * 2;
 

Mobilyzer/src/com/mobilyzer/measurements/SequentialTask.java

@@ -169,6 +169,7 @@ public MeasurementResult[] call() throws MeasurementError {
     try {
       //      futures=executor.invokeAll(this.tasks,timeout,TimeUnit.MILLISECONDS);
       for(MeasurementTask mt: tasks){
+    	mt.setScheduler(this.getScheduler());
         if(stopFlag){
           throw new MeasurementError("Cancelled");
         }