001: /*
002: * <copyright>
003: *
004: * Copyright 1997-2004 BBNT Solutions, LLC
005: * under sponsorship of the Defense Advanced Research Projects
006: * Agency (DARPA).
007: *
008: * You can redistribute this software and/or modify it under the
009: * terms of the Cougaar Open Source License as published on the
010: * Cougaar Open Source Website (www.cougaar.org).
011: *
012: * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
013: * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
014: * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
015: * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
016: * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
017: * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
018: * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
019: * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
020: * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
021: * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
022: * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
023: *
024: * </copyright>
025: */
026:
027: package org.cougaar.core.qos.metrics;
028:
029: import java.util.Observable;
030: import java.util.Observer;
031:
032: import org.cougaar.core.component.ServiceBroker;
033: import org.cougaar.core.service.ThreadService;
034: import org.cougaar.core.thread.Schedulable;
035:
036: /**
037: * This test Plugin publishes data into the MetricsUpdateService at
038: * high rates and subscribes to formulas using that data. The
039: * arguments are <code>key</code> (the publish key) and
040: * <code>path</code> (the subscription path).
041: */
042: public class MetricsBlasterPlugin extends
043: org.cougaar.core.plugin.ParameterizedPlugin implements Observer {
044:
045: private MetricsUpdateService update;
046: private MetricsService svc;
047: private ThreadService tsvc;
048:
049: private String key, path;
050:
051: private long callbackCount = 0;
052: private long blastCount = 0;
053: private long lastCallbackDelay = 0;
054: private long lastPrintTime = 0;
055:
056: private int restCount = 0;
057: private Schedulable blastTask;
058: private Schedulable restTask;
059:
060: private void dumpCounters(long now) {
061: if (1000 < (now - lastPrintTime)) {
062: System.out.println("blast count=" + blastCount
063: + " callback count=" + callbackCount
064: + " Last delay=" + lastCallbackDelay);
065: lastPrintTime = now;
066: }
067: }
068:
069: public void execute() {
070: }
071:
072: public void setupSubscriptions() {
073: }
074:
075: public void load() {
076: super .load();
077:
078: ServiceBroker sb = getServiceBroker();
079: update = (MetricsUpdateService) sb.getService(this ,
080: MetricsUpdateService.class, null);
081: svc = (MetricsService) sb.getService(this ,
082: MetricsService.class, null);
083: tsvc = (ThreadService) sb.getService(this , ThreadService.class,
084: null);
085:
086: path = getParameter("path");
087: if (path != null) {
088: svc.subscribeToValue(path, this );
089: System.out.println("Subscribed to " + path);
090:
091: key = getParameter("key");
092: if (key != null) {
093: System.out.println("Blasting to " + key);
094: Blast blast = new Blast();
095: blastTask = tsvc.getThread(this , blast, "Blaster");
096: blastTask.schedule(10000);
097: }
098:
099: }
100: }
101:
102: public void update(Observable o, Object arg) {
103: callbackCount++;
104: long now = System.currentTimeMillis();
105: long value = ((Metric) arg).longValue();
106: lastCallbackDelay = now - value;
107: }
108:
109: private class Blast implements Runnable {
110: public void run() {
111: System.out.println("Starting Blaster");
112: long startTime = System.currentTimeMillis();
113: long startBlast = blastCount;
114: long startCallback = callbackCount;
115:
116: long now = startTime;
117: // Blast for 5 seconds and then stop
118: while (5000 > (now - startTime)) {
119: now = System.currentTimeMillis();
120: Metric m = new MetricImpl(new Long(now), 0.3, "",
121: "MetricsTestAspect");
122: update.updateValue(key, m);
123: blastCount++;
124: dumpCounters(now);
125: }
126: float deltaBlasts = (blastCount - startBlast);
127: float deltaCallback = (callbackCount - startCallback);
128: long deltaTime = now - startTime;
129: float blastRate = deltaBlasts / deltaTime;
130: float callbackRate = deltaCallback / deltaTime;
131: float callbackPercent = deltaBlasts > 0 ? (100 * deltaCallback)
132: / deltaBlasts
133: : 0.0f;
134: System.out.println("Stopped Blaster:" + "blasts/millisec ="
135: + blastRate + "callback/millisec =" + callbackRate
136: + " Callback % =" + callbackPercent);
137:
138: restCount = 0;
139: Rest rest = new Rest();
140: restTask = tsvc.getThread(this , rest, "Rest");
141: restTask.schedule(0, 1000);
142: }
143: }
144:
145: private class Rest implements Runnable {
146: public void run() {
147: dumpCounters(System.currentTimeMillis());
148: if (restCount++ == 10) {
149: restTask.cancelTimer();
150: blastTask.start();
151: }
152: }
153: }
154:
155: }
|