[snee] r1368 committed - fixes for join simulation within noise model. ...

[sn...@googlecode.com]

Revision: 1368
Author: alan.bar...@gmail.com
Date: Tue Mar 5 10:56:33 2013
Log: fixes for join simulation within noise model.

NOTE: node 5 in query 60 exhibitis the noise stagnation effect. May be
worth checking other topolgoies for this behaviour
http://code.google.com/p/snee/source/detail?r=1368

Modified:

/branches/alan_2012_07_19_unreliableChannels/snee-core/src/main/java/uk/ac/manchester/cs/snee/compiler/costmodels/cardinalitymodel/CardinalityEstimatedCostModel.java

/branches/alan_2012_07_19_unreliableChannels/snee-core/src/main/java/uk/ac/manchester/cs/snee/compiler/costmodels/cardinalitymodel/CollectionOfPackets.java

/branches/alan_2012_07_19_unreliableChannels/snee-wsn-manager/src/main/java/uk/ac/manchester/cs/snee/manager/planner/costbenifitmodel/model/channel/ChannelModelSite.java

=======================================
---
/branches/alan_2012_07_19_unreliableChannels/snee-core/src/main/java/uk/ac/manchester/cs/snee/compiler/costmodels/cardinalitymodel/CardinalityEstimatedCostModel.java
Mon Mar 4 09:34:19 2013
+++
/branches/alan_2012_07_19_unreliableChannels/snee-core/src/main/java/uk/ac/manchester/cs/snee/compiler/costmodels/cardinalitymodel/CardinalityEstimatedCostModel.java
Tue Mar 5 10:56:33 2013
@@ -388,7 +388,7 @@
{
if(operator.isNodeDead())
return new CardinalityDataStructureChannel(new ArrayList<Window>());
- String extent =
operator.getSensornetOperator().getAttributes().get(1).toString();
+ String extent = operator.getExtent();
ArrayList<Window> inputWindows = inputs.getWindowsOfExtent(operator,
extent);
ArrayList<Window> outputWindows = new ArrayList<Window>();

@@ -491,7 +491,7 @@

ArrayList<Window> output = new ArrayList<Window>();

- for(int index =0; index <= beta; index++)
+ for(int index =1; index <= beta; index++)
{
Window extent1Window = inputs.getWindow(index, windowsOfExtent1);
Window extent2Window = inputs.getWindow(index, windowsOfExtent2);
@@ -536,8 +536,7 @@
}
else if(operator.getSensornetOperator() instanceof
SensornetProjectOperator)
{
- InstanceOperator op =
(InstanceOperator)(operator.getInstanceInput(0));
- return model(op, inputs, beta);
+ return project(operator, inputs, beta);
}
else if(operator.getSensornetOperator() instanceof
SensornetRStreamOperator)
{
@@ -562,4 +561,22 @@
return new CardinalityDataStructureChannel(new ArrayList<Window>());
}
}
+
+ private CardinalityDataStructureChannel project(InstanceOperator
operator,
+ CollectionOfPackets
inputs, long beta)
+ {
+ if(operator.isNodeDead())
+ return new CardinalityDataStructureChannel(new ArrayList<Window>());
+ String extent = operator.getExtent();
+ ArrayList<Window> inputWindows = inputs.getWindowsOfExtent(operator,
extent);
+ ArrayList<Window> outputWindows = new ArrayList<Window>();
+
+ for(int index = 0; index < inputWindows.size(); index++)
+ {
+ int windowCard = inputWindows.get(index).getTuples();
+ outputWindows.add(new
Window(windowCard,inputWindows.get(index).getWindowID()));
+ }
+ CardinalityDataStructureChannel output = new
CardinalityDataStructureChannel(outputWindows);
+ return output;
+ }
}
=======================================
---
/branches/alan_2012_07_19_unreliableChannels/snee-core/src/main/java/uk/ac/manchester/cs/snee/compiler/costmodels/cardinalitymodel/CollectionOfPackets.java
Mon Mar 4 09:34:19 2013
+++
/branches/alan_2012_07_19_unreliableChannels/snee-core/src/main/java/uk/ac/manchester/cs/snee/compiler/costmodels/cardinalitymodel/CollectionOfPackets.java
Tue Mar 5 10:56:33 2013
@@ -23,18 +23,18 @@
{
}

- public void updateCollection(InstanceOperator extent, final
ArrayList<Window> cWindows)
+ public void updateCollection(InstanceOperator operator, final
ArrayList<Window> cWindows)
{
- ArrayList<Window> extentWindows = this.windows.get(extent);
+ ArrayList<Window> extentWindows = this.windows.get(operator);
Iterator<Window> newWindows = cWindows.iterator();
while(newWindows.hasNext())
{
Window newWindow = newWindows.next();
Window oldWindow = getWindow(newWindow.getWindowID(), extentWindows);
Window addedWindow = new Window(oldWindow.getTuples() +
newWindow.getTuples(), newWindow.getWindowID());
- if(this.windows.keySet().contains(extent))
- this.windows.remove(extent, oldWindow);
- this.windows.add(extent, addedWindow);
+ if(this.windows.keySet().contains(operator))
+ this.windows.remove(operator, oldWindow);
+ this.windows.add(operator, addedWindow);
}
}

@@ -98,11 +98,6 @@
}
return new Window(0,index);
}
-
- public void removeExtent(InstanceOperator extentName)
- {
- this.windows.remove(extentName);
- }

public static int determineNoTuplesFromWindows(ArrayList<Window>
outputWindows)
{
@@ -141,6 +136,8 @@
int tuplesCounted = 0;
ArrayList<Window> doneWindows = new ArrayList<Window>();
ArrayList<Window> windows = this.windows.get(extent);
+ if(windows.size() == 0)
+ System.out.println();
Iterator<Window> windowIterator = windows.iterator();
boolean done = false;
while(windowIterator.hasNext() && !done)
=======================================
---
/branches/alan_2012_07_19_unreliableChannels/snee-wsn-manager/src/main/java/uk/ac/manchester/cs/snee/manager/planner/costbenifitmodel/model/channel/ChannelModelSite.java
Mon Mar 4 13:20:13 2013
+++
/branches/alan_2012_07_19_unreliableChannels/snee-wsn-manager/src/main/java/uk/ac/manchester/cs/snee/manager/planner/costbenifitmodel/model/channel/ChannelModelSite.java
Tue Mar 5 10:56:33 2013
@@ -415,7 +415,7 @@
ArrayList<Window> windows =

packetToTupleConversion(transmittablePacketsCount, exOp, exOp.getPrevious(),

currentUsedRecievedPacketCount);
- tuples.updateCollection(exOp, windows);
+ tuples.updateCollection(preExOp, windows);
int outputPackets = this.tupleToPacketConversion(tuples,
preExOp, exOp, cPacketCount);
addToPacketCounts(outputPackets, cPacketCount, isleaf,
packetIds);
currentPacketCount.remove(exOp.getSite().getID());
@@ -426,7 +426,7 @@
int noOfTuples =
CollectionOfPackets.determineNoTuplesFromWindows(outputWindows);
exOp.setTupleValueForExtent(noOfTuples);
this.tupleToPacketConversion(tuples, preExOp, exOp,
cPacketCount);
- this.transmitableWindows.removeExtent(exOp);
+ tuples.updateCollection(exOp, outputWindows);
this.transmitableWindows.updateCollection(exOp,
tuples.getWindowsOfExtent(exOp, exOp.getExtent()));

}
@@ -452,6 +452,7 @@
packetToTupleConversion(transmittablePacketsCount, exOp,
exOp.getPrevious(),
currentUsedRecievedPacketCount);
tuples.updateCollection(exOp, windows);
+ tuples.updateCollection(exOp.getPrevious(), windows);
this.tupleToPacketConversion(tuples, exOp.getPrevious(), exOp,
cPacketCount);

currentUsedRecievedPacketCount.remove(preExOp.getSite().getID());
currentUsedRecievedPacketCount.put(
@@ -498,7 +499,6 @@
extent = preOp.getExtent();
}
CardinalityDataStructureChannel outputs = cardModel.model(op,
tuples, beta);
- tuples.removeExtent(op);
tuples.updateCollection(op, outputs.getWindows());
this.tupleToPacketConversion(tuples, preOp, op, 0);
}
@@ -517,7 +517,6 @@
}
CardinalityDataStructureChannel outputs = cardModel.model(op,
tuples, beta);
ArrayList<Window> outputWindows = outputs.getWindows();
- tuples.removeExtent(op);
tuples.updateCollection(op, outputWindows);
previousOpOutput = this.tupleToPacketConversion(tuples, op,
op, 0);
for(int index = 0; index < previousOpOutput; index++)
@@ -560,7 +559,6 @@
if(!doneExtents.contains(currentExtent))
{
doneExtents.add(currentExtent);
- tuples.removeExtent(cOp);
}
}
return doneExtents;
@@ -999,7 +997,13 @@
CollectionOfPackets completeRecievedWindows = new
CollectionOfPackets();
if(noPackets == 0)
{
- return completeRecievedWindows.getWindowsOfExtent(op,
op.getExtent());
+ ArrayList<Window> windows = new ArrayList<Window>();
+ for(int index =1; index <=beta; index++)
+ {
+ Window newWindow = new Window(0, index);
+ windows.add(newWindow);
+ }
+ return windows;
}
else
{
@@ -1057,18 +1061,15 @@
}
else
{
- String extent;
if(op instanceof InstanceExchangePart)
{
InstanceExchangePart exOp = (InstanceExchangePart) op;
tupleSize =

exOp.getSourceFrag().getRootOperator().getSensornetOperator().getPhysicalTupleSize();
- extent = exOp.getExtent();
}
else
{
tupleSize = op.getSensornetOperator().getPhysicalTupleSize();
- extent = op.getExtent();
}

int maxMessagePayloadSize = costs.getMaxMessagePayloadSize();
@@ -1109,11 +1110,22 @@
update.add(receivedWindows.next());
completeRecievedWindows.updateCollection(op, update);
}
+ countedPacket++;
}
tupleCount += numTuplesPerMessage;
- countedPacket++;
}
- return completeRecievedWindows.getWindowsOfExtent(op);
+ ArrayList<Window> output =
completeRecievedWindows.getWindowsOfExtent(op);
+ if(output.size() == 0)
+ {
+ ArrayList<Window> windows = new ArrayList<Window>();
+ for(int index =1; index <=beta; index++)
+ {
+ Window newWindow = new Window(0, index);
+ windows.add(newWindow);
+ }
+ return windows;
+ }
+ return output;
}
}
}
@@ -1168,6 +1180,8 @@
counter++;
packetRecieved = false;
}
+ if(packetsRecieved == 0)
+ System.out.println();
return packetsRecieved;
}