perfsonar development work

Author: uros
Date: 2006-10-24 12:33:13 -0400 (Tue, 24 Oct 2006)
New Revision: 1714

Modified:
   
trunk/perfsonar/schema/example-instances/sonar/TCMPCaptureRequest-measurement.xml
   
trunk/perfsonar/src/org/perfsonar/commons/auxiliary/components/simplescheduler/SchedulerTask.java
   
trunk/perfsonar/src/org/perfsonar/service/measurementPoint/tracefileCaptureType/TCMPTypeMeasurementExecutor.java
Log:
Added scheduled start of TCMP measurement using simplescheduler component.

Added another constructor to SchedulerTask used to set up startup of 
scheduled action.

Updated request example accordingly.


Modified: 
trunk/perfsonar/schema/example-instances/sonar/TCMPCaptureRequest-measurement.xml
===================================================================
--- 
trunk/perfsonar/schema/example-instances/sonar/TCMPCaptureRequest-measurement.xml
   2006-10-24 14:01:02 UTC (rev 1713)
+++ 
trunk/perfsonar/schema/example-instances/sonar/TCMPCaptureRequest-measurement.xml
   2006-10-24 16:33:13 UTC (rev 1714)
@@ -44,6 +44,9 @@
 -->    
 
 <!-- endPoint example -->
+<!--
+      <nmwgt:endPoint type="hostname" value="sonar3.ljubljana.arnes.si" 
port="eth0" />
+-->
       <nmwgt:endPoint type="hostname" value="localhost" port="/dev/dag0" />
 
 <!-- endPointPair example
@@ -52,6 +55,10 @@
   -->
 <!--
       <nmwgt:endPointPair>
+        <nmwgt:src type="hostname" value="sonar3.ljubljana.arnes.si" 
port="eth0"/>
+        <nmwgt:dst type="hostname" value="sonar3.ljubljana.arnes.si" 
port="/dev/dag0"/>
+      </nmwgt:endPointPair> 
+      <nmwgt:endPointPair>
         <nmwgt:src type="hostname" value="localhost" port="eth0"/>
         <nmwgt:dst type="hostname" value="localhost" port="/dev/dag0"/>
       </nmwgt:endPointPair> 
@@ -94,15 +101,13 @@
         <nmwg:parameter name="characteristics">1</nmwg:parameter>
         <nmwg:parameter name="granularity">1</nmwg:parameter>
       -->
-      <!-- start_time specifies when to start capture
-           (currently ignored, waiting for scheduler)
-      -->
+      <!-- start_time specifies when to start capture -->
       <nmwg:parameter name="start_time">
-        <nmtm:start type="unix" value="1107492095"/>
+        <nmtm:start type="unix" value="1161707362"/>
       </nmwg:parameter>
       <!-- end_time specified when to end capture -->
       <nmwg:parameter name="end_time">
-        <nmtm:end type="unix" value="1150893383"/>
+        <nmtm:end type="unix" value="1161707362"/>
       </nmwg:parameter>            
     </passivetracefile:parameters>    
     
@@ -115,9 +120,9 @@
       <passivetracefile:datum filename="firstEndPoint.raw" />
       <!-- Specifies where to store PCAP file captured at the second 
endPoint -->
       <!-- Second datum is optiona, only required when endPointPair is 
specified in metadata subject -->
+      <!-- <passivetracefile:datum filename="secondEndPoint.raw" /> -->
+      <!--
       <passivetracefile:datum filename="secondEndPoint.raw" />
-      <!--
-        <passivetracefile:datum filename="secondEndPoint.raw" />
       -->
     </nmwg:commonTime>
   </nmwg:data>

Modified: 
trunk/perfsonar/src/org/perfsonar/commons/auxiliary/components/simplescheduler/SchedulerTask.java
===================================================================
--- 
trunk/perfsonar/src/org/perfsonar/commons/auxiliary/components/simplescheduler/SchedulerTask.java
   2006-10-24 14:01:02 UTC (rev 1713)
+++ 
trunk/perfsonar/src/org/perfsonar/commons/auxiliary/components/simplescheduler/SchedulerTask.java
   2006-10-24 16:33:13 UTC (rev 1714)
@@ -27,6 +27,23 @@
         
     }
 
+
+    /**
+     * This constructor is used, when some action should be run in the future
+     *
+     * @param interval interval for running action
+     * @param action action to run
+     * @param startTimestamp when should we start running action
+     */
+    public SchedulerTask(
+        long interval, SchedulerAction action, long startTimestamp) {
+
+        this(interval, action);
+
+        timestamp = startTimestamp - interval;
+
+    }
+
     
     // ------------------------------------------------ public methods
     

Modified: 
trunk/perfsonar/src/org/perfsonar/service/measurementPoint/tracefileCaptureType/TCMPTypeMeasurementExecutor.java
===================================================================
--- 
trunk/perfsonar/src/org/perfsonar/service/measurementPoint/tracefileCaptureType/TCMPTypeMeasurementExecutor.java
    2006-10-24 14:01:02 UTC (rev 1713)
+++ 
trunk/perfsonar/src/org/perfsonar/service/measurementPoint/tracefileCaptureType/TCMPTypeMeasurementExecutor.java
    2006-10-24 16:33:13 UTC (rev 1714)
@@ -27,6 +27,9 @@
 import org.perfsonar.commons.auxiliary.ComponentNames;
 import 
org.perfsonar.commons.auxiliary.components.configuration.ConfigurationComponent;
 import org.perfsonar.commons.auxiliary.components.logger.LoggerComponent;
+import org.perfsonar.commons.auxiliary.components.simplescheduler.Scheduler;
+import 
org.perfsonar.commons.auxiliary.components.simplescheduler.SchedulerAction;
+import 
org.perfsonar.commons.auxiliary.components.simplescheduler.SchedulerTask;
 
 import org.perfsonar.service.commons.measurementExecutor.MeasurementExecutor;
 
@@ -64,7 +67,7 @@
  * @author urosj
  */
 
-public class TCMPTypeMeasurementExecutor implements MeasurementExecutor, 
Runnable {
+public class TCMPTypeMeasurementExecutor implements MeasurementExecutor, 
SchedulerAction {
 
 
     //  ---------------------------------- class fields
@@ -88,8 +91,7 @@
 
     private static HashSet<String> localHostsIPs = null;
 
-    private TCMPTypeMeasurementExecutor parent = null;
-    private int threadCount = 0;
+    private boolean runnedAlready = false;
 
 
     // ---------------------------------- constructors
@@ -106,14 +108,12 @@
      * Private constructor used by makeMeasurement to run this
      * object in a new thread.
      *
-     * @param parent Parent TCMPTypeMeasurementExecutor object
      * @param request Filled by makeMeasurement
      * @param capIf Interface to capture from (@see CaptureInterface)
      */
-    private TCMPTypeMeasurementExecutor(TCMPTypeMeasurementExecutor parent,
+    private TCMPTypeMeasurementExecutor(
         TCMPTypeMeasurementRequest request, CaptureInterface capIf) {
 
-        this.parent = parent;
         this.request = request;
         this.capIf = capIf;
 
@@ -179,32 +179,43 @@
         // Throw either SystemException on RequestException if validation 
fails
         TCMPTypeMeasurementRequest req = validateTCMPRequest(request);
 
-        // Create new Thread of myself for first interface or both
-        // depending on second endPoint host
-        this.threadCount = 0;
+        // Create one or more Measurement executors
+        // (first or both interfaces)
+        // and pass them to the scheduler.
+        Scheduler scheduler = (Scheduler)
+            
AuxiliaryComponentManager.getInstance().getComponent("scheduler");
+
+        if (scheduler == null)
+
+            throw new SystemException(
+                "error.common.no_scheduler",
+                "No scheduler component in TCMPTypeMeasurementExecutor");
+
         if (req.getSecondEndPointHost() == null) {
-            new Thread(new TCMPTypeMeasurementExecutor(this, req,
-                        CaptureInterface.FIRST)).start();
-            // Wait for thread to start
-            while (this.threadCount < 1)
-                try { Thread.sleep(50); } catch (Exception e) { }
-            // Notify thread to start
-            synchronized (this) {
-                notifyAll();
-            }
+
+            scheduler.addSchedulerTask(
+                new SchedulerTask(
+                    0,
+                    new TCMPTypeMeasurementExecutor(req, 
CaptureInterface.FIRST),
+                    req.getStartTimeParam().getTime()));
+            logger.info(type + ".makeMeasurement: Scheduled start of 1 
measurement at "
+                + req.getStartTimeParam());
+            
         } else {
-            // Make synchronous start
-            new Thread(new TCMPTypeMeasurementExecutor(this, req,
-                        CaptureInterface.FIRST)).start();
-            new Thread(new TCMPTypeMeasurementExecutor(this, req,
-                        CaptureInterface.SECOND)).start();
-            // Wait for threads to start
-            while (this.threadCount < 2)
-                try { Thread.sleep(50); } catch (Exception e) { }
-            // Notify all threads to start
-            synchronized (this) {
-                notifyAll();
-            }
+
+            scheduler.addSchedulerTask(
+                new SchedulerTask(
+                    0,
+                    new TCMPTypeMeasurementExecutor(req, 
CaptureInterface.FIRST),
+                    req.getStartTimeParam().getTime()));
+            scheduler.addSchedulerTask(
+                new SchedulerTask(
+                    0,
+                    new TCMPTypeMeasurementExecutor(req, 
CaptureInterface.SECOND),
+                    req.getStartTimeParam().getTime()));
+            logger.info(type + ".makeMeasurement: Scheduled start of 2 
measurements at "
+                + req.getStartTimeParam());
+
         }
 
         // Construct a generic OK response, with URL to resulted data,
@@ -258,7 +269,9 @@
             new org.ggf.ns.nmwg.base.v2_0.Datum();
         firstData.setDatum(dDatum);
         mdEventType.setEventType("success.mp.tcmp");
-        dDatum.setDatum("Successfully initiated " + this.threadCount
+        dDatum.setDatum(
+            "Initiated "
+            + (req.getSecondEndPointHost() == null ? 1 : 2)
             + " capture(s).");
         
         logger.info(type + ": Returning a response");
@@ -268,30 +281,38 @@
     }
     
     /**
-     * Implementation of Runnable interface.
+     * Implementation of SchedulerAction interface.
      * This methods starts executing in a new thread.
      */
-    public void run() {
+    public void runAction() {
 
-        // Wait for all threads to startup
-        try {
-            synchronized (parent) {
-                parent.threadCount++;
-                parent.wait();
-            }
-        } catch (Exception e) { parent.threadCount++; }
+        if (runnedAlready) return;
 
-        // If request is null, then wrong constructor was called
-        // from outside?
-        // If so, return immediately!
-        if (this.request == null) return;
+        runnedAlready = true;
 
+
         try {
             
-            this.logger = 
(LoggerComponent)AuxiliaryComponentManager.getInstance().getComponent(ComponentNames.LOGGER);
+            this.logger = (LoggerComponent)
+                
AuxiliaryComponentManager.getInstance().getComponent(ComponentNames.LOGGER);
             
-            this.configuration = 
(ConfigurationComponent)AuxiliaryComponentManager.getInstance().getComponent(ComponentNames.CONFIG);
+            this.configuration = (ConfigurationComponent)
+                
AuxiliaryComponentManager.getInstance().getComponent(ComponentNames.CONFIG);
             
+            Scheduler scheduler = (Scheduler)
+                
AuxiliaryComponentManager.getInstance().getComponent("scheduler");
+
+            if (scheduler == null)
+
+                throw new SystemException(
+                    "error.common.no_scheduler",
+                    "No scheduler component in TCMPTypeMeasurementExecutor");
+            scheduler.remSchedulerTask(this);
+
+            // If request is null, then wrong constructor was called
+            // from outside?
+            if (this.request == null) return;
+
             if (this.configuration == null) {
                 
                 throw new SystemException(
@@ -310,7 +331,7 @@
     
             }
 
-            this.logger.info(type + ".run: Created new thread (tid="
+            this.logger.info(type + ".runAction: Created new thread (tid="
                 + Thread.currentThread().getId() + ")");
 
             // Create needed info ((interface, host:interface) and datum)
@@ -390,14 +411,14 @@
             // Create flow
             monReq.mapiCreateFlow(hostIf);
 
-            this.logger.info(type + ".run: Created flow to " + hostIf
+            this.logger.info(type + ".runAction: Created flow to " + hostIf
                 + "(tid=" + Thread.currentThread().getId() + ")");
 
             // Apply BPF_FILTER function to the flow if requested
             if (this.request.getFlowHFilterParam() != null) {
 
                 monReq.mapiApplyFunction("BPF_FILTER", 
this.request.getFlowHFilterParam());
-                this.logger.info(type + ".run: Applied BPF_FILTER function 
('"
+                this.logger.info(type + ".runAction: Applied BPF_FILTER 
function ('"
                     + this.request.getFlowHFilterParam() + "') to the flow 
on " + hostIf
                     + " (tid=" + Thread.currentThread().getId() + ")");
 
@@ -409,7 +430,7 @@
 
                 monReq.mapiApplyFunction("SAMPLE", 
this.request.getFlowSThreParam(),
                     this.request.getFlowSTypeParam());
-                this.logger.info(type + ".run: Applied SAMPLE function ("
+                this.logger.info(type + ".runAction: Applied SAMPLE function 
("
                     + this.request.getFlowSThreParam() + ", "
                     + this.request.getFlowSTypeParam() + ") to the flow on " 
+ hostIf
                     + " (tid=" + Thread.currentThread().getId() + ")");
@@ -420,7 +441,7 @@
             if (this.request.getFlowPFilterParam() != null) {
 
                 monReq.mapiApplyFunction("STR_SEARCH", 
this.request.getFlowPFilterParam(), 0, 0);
-                this.logger.info(type + ".run: Applied STR_SEARCH function 
('"
+                this.logger.info(type + ".runAction: Applied STR_SEARCH 
function ('"
                     + this.request.getFlowPFilterParam() + "', 0, 0) to the 
flow on " + hostIf
                     + " (tid=" + Thread.currentThread().getId() + ")");
 
@@ -430,14 +451,14 @@
             // TODO: another configuration option specifying maximum limit 
of packets
             //       that can be captured to one pcap file (disk space 
safeguard)
             monReq.mapiApplyFunction("TO_FILE", 
MonitorRequest.CAP_FORMAT_TYPE_MFF_PCAP, datumFile, 0L);
-            this.logger.info(type + ".run: Applied TO_FILE function ("
+            this.logger.info(type + ".runAction: Applied TO_FILE function ("
                 + MonitorRequest.CAP_FORMAT_TYPE_MFF_PCAP
                 + ", '" + datumFile + "', 0) to the flow on " + hostIf
                 + " (tid=" + Thread.currentThread().getId() + ")");
 
             // Connect to the flow
             monReq.mapiConnect();
-            this.logger.info(type + ".run: Connected to the flow on "
+            this.logger.info(type + ".runAction: Connected to the flow on "
                 + hostIf + ", waiting until " + 
this.request.getEndTimeParam()
                 + " (tid=" + Thread.currentThread().getId() + ")");
 
@@ -446,41 +467,41 @@
             // Close flow
             monReq.mapiCloseFlow();
 
-            this.logger.info(type + ".run: Closed flow to " + hostIf
+            this.logger.info(type + ".runAction: Closed flow to " + hostIf
                 + " (tid=" + Thread.currentThread().getId() + ")");
 
             
             // If remote flow, copy remotely created PCAP file to local
             // directory and remove it remotely
             if (isRemoteFlow) {
-                this.logger.info(type + ".run: Transporting remote PCAP file 
'"
+                this.logger.info(type + ".runAction: Transporting remote 
PCAP file '"
                     + datumFile + "' from host " + host + " to localhost"
                     + " (tid=" + Thread.currentThread().getId() + ")");
                 
                 ft.copy(fileName, remoteLoc, localLoc);
                 
-                this.logger.info(type + ".run: Removing remote PCAP file '"
+                this.logger.info(type + ".runAction: Removing remote PCAP 
file '"
                     + datumFile + "' on host " + host
                     + " (tid=" + Thread.currentThread().getId() + ")");
                 
                 ft.delete(fileName, remoteLoc);
             }
 
-            this.logger.info(type + ".run: Thread exiting (tid="
+            this.logger.info(type + ".runAction: Thread exiting (tid="
                 + Thread.currentThread().getId() + ")");
         } catch (SystemException sysEx) {
 
-            this.logger.fatal(type + ".run(): Received SystemException (tid="
+            this.logger.fatal(type + ".runAction: Received SystemException 
(tid="
                 + Thread.currentThread().getId() + "): " + sysEx);
 
         } catch (MAPIException mapiEx) {
 
-            this.logger.fatal(type + ".run(): Received MAPIException (tid="
+            this.logger.fatal(type + ".runAction: Received MAPIException 
(tid="
                 + Thread.currentThread().getId() + "): " + mapiEx);
 
         } catch (Exception e) {
 
-            this.logger.fatal(type + ".run(): Received Exception (tid="
+            this.logger.fatal(type + ".runAction: Received Exception (tid="
                 +  Thread.currentThread().getId() + "): " + e);
 
         }