Merge branch 'master' into 0.8.2

Nathan Marz · Nathan Marz · commit 138a2f90ebba · 2012-12-10T15:45:51.000-08:00
diff --git a/src/clj/backtype/storm/messaging/zmq.clj b/src/clj/backtype/storm/messaging/zmq.clj
@@ -26,6 +26,8 @@
 (defprotocol ZMQContextQuery
   (zmq-context [this]))
 
+(def NOBLOCK-SNDMORE (bit-or ZMQ/SNDMORE ZMQ/NOBLOCK))
+
 (deftype ZMQConnection [socket ^ByteBuffer bb]
   Connection
   (recv-with-flags [this flags]
@@ -37,8 +39,8 @@
   (send [this task message]
     (.clear bb)
     (.putShort bb (short task))
-    (mq/send socket (.array bb) ZMQ/SNDMORE)
-    (mq/send socket message)) ;; TODO: temporarily remove the noblock flag
+    (mq/send socket (.array bb) NOBLOCK-SNDMORE)
+    (mq/send socket message ZMQ/NOBLOCK)) ;; TODO: how to do backpressure if doing noblock?... need to only unblock if the target disappears
   (close [this]
     (.close socket)
     ))
diff --git a/src/jvm/backtype/storm/task/IMetricsContext.java b/src/jvm/backtype/storm/task/IMetricsContext.java
@@ -0,0 +1,6 @@
+package backtype.storm.task;
+
+
+public interface IMetricsContext {
+    
+}
diff --git a/src/jvm/backtype/storm/task/TopologyContext.java b/src/jvm/backtype/storm/task/TopologyContext.java
@@ -24,7 +24,7 @@
  * <p>The TopologyContext is also used to declare ISubscribedState objects to
  * synchronize state with StateSpouts this object is subscribed to.</p>
  */
-public class TopologyContext extends WorkerTopologyContext {
+public class TopologyContext extends WorkerTopologyContext implements IMetricsContext {
     private Integer _taskId;
     private Map<String, Object> _taskData = new HashMap<String, Object>();
     private List<ITaskHook> _hooks = new ArrayList<ITaskHook>();
diff --git a/src/jvm/storm/trident/operation/TridentOperationContext.java b/src/jvm/storm/trident/operation/TridentOperationContext.java
@@ -1,11 +1,12 @@
 package storm.trident.operation;
 
+import backtype.storm.task.IMetricsContext;
 import backtype.storm.task.TopologyContext;
 import backtype.storm.tuple.Fields;
 import storm.trident.tuple.TridentTuple;
 import storm.trident.tuple.TridentTupleView.ProjectionFactory;
 
-public class TridentOperationContext {
+public class TridentOperationContext implements IMetricsContext{
     TridentTuple.Factory _factory;
     TopologyContext _topoContext;
     
diff --git a/src/jvm/storm/trident/planner/SubtopologyBolt.java b/src/jvm/storm/trident/planner/SubtopologyBolt.java
@@ -49,7 +49,7 @@ public void prepare(Map conf, TopologyContext context, BatchOutputCollector batc
         int thisComponentNumTasks = context.getComponentTasks(context.getThisComponentId()).size();
         for(Node n: _nodes) {
             if(n.stateInfo!=null) {
-                State s = n.stateInfo.spec.stateFactory.makeState(conf, context.getThisTaskIndex(), thisComponentNumTasks);
+                State s = n.stateInfo.spec.stateFactory.makeState(conf, context, context.getThisTaskIndex(), thisComponentNumTasks);
                 context.setTaskData(n.stateInfo.id, s);
             }
         }
diff --git a/src/jvm/storm/trident/state/StateFactory.java b/src/jvm/storm/trident/state/StateFactory.java
@@ -1,8 +1,9 @@
 package storm.trident.state;
 
+import backtype.storm.task.IMetricsContext;
 import java.io.Serializable;
 import java.util.Map;
 
 public interface StateFactory extends Serializable {
-    State makeState(Map conf, int partitionIndex, int numPartitions);
+    State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions);
 }
diff --git a/src/jvm/storm/trident/testing/LRUMemoryMapState.java b/src/jvm/storm/trident/testing/LRUMemoryMapState.java
@@ -1,5 +1,6 @@
 package storm.trident.testing;
 
+import backtype.storm.task.IMetricsContext;
 import storm.trident.state.ITupleCollection;
 import backtype.storm.tuple.Values;
 import java.util.*;
@@ -70,7 +71,7 @@ public Factory(int maxSize) {
         }
 
         @Override
-        public State makeState(Map conf, int partitionIndex, int numPartitions) {
+        public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
             return new LRUMemoryMapState(_maxSize, _id);
         }
     }
diff --git a/src/jvm/storm/trident/testing/MemoryMapState.java b/src/jvm/storm/trident/testing/MemoryMapState.java
@@ -1,5 +1,6 @@
 package storm.trident.testing;
 
+import backtype.storm.task.IMetricsContext;
 import storm.trident.state.ITupleCollection;
 import backtype.storm.tuple.Values;
 import java.util.*;
@@ -67,7 +68,7 @@ public Factory() {
         }
 
         @Override
-        public State makeState(Map conf, int partitionIndex, int numPartitions) {
+        public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
             return new MemoryMapState(_id);
         }
     }
diff --git a/src/jvm/storm/trident/tuple/ComboList.java b/src/jvm/storm/trident/tuple/ComboList.java
@@ -8,8 +8,10 @@
 public class ComboList extends AbstractList<Object> {    
     public static class Factory implements Serializable {
         Pointer[] index;
+        int numLists;
         
         public Factory(int... sizes) {
+            numLists = sizes.length;
             int total = 0;
             for(int size: sizes) {
                 total+=size;
@@ -27,6 +29,9 @@ public Factory(int... sizes) {
         }
         
         public ComboList create(List[] delegates) {
+            if(delegates.length!=numLists) {
+                throw new RuntimeException("Expected " + numLists + " lists, but instead got " + delegates.length + " lists");
+            }
             return new ComboList(delegates, index);
         }
     }

-Original file line number
+Diff line change
@@ @@ -0,0 +1,6 @@ @@
 +package backtype.storm.task;
++
++
 +public interface IMetricsContext {
++
 +}
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ public void prepare(Map conf, TopologyContext context, BatchOutputCollector batc`
`49`	`49`	`int thisComponentNumTasks = context.getComponentTasks(context.getThisComponentId()).size();`
`50`	`50`	`for(Node n: _nodes) {`
`51`	`51`	`if(n.stateInfo!=null) {`
`52`		`- State s = n.stateInfo.spec.stateFactory.makeState(conf, context.getThisTaskIndex(), thisComponentNumTasks);`
	`52`	`+ State s = n.stateInfo.spec.stateFactory.makeState(conf, context, context.getThisTaskIndex(), thisComponentNumTasks);`
`53`	`53`	`context.setTaskData(n.stateInfo.id, s);`
`54`	`54`	`}`
`55`	`55`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,8 +1,9 @@`
`1`	`1`	`package storm.trident.state;`
`2`	`2`
	`3`	`+import backtype.storm.task.IMetricsContext;`
`3`	`4`	`import java.io.Serializable;`
`4`	`5`	`import java.util.Map;`
`5`	`6`
`6`	`7`	`public interface StateFactory extends Serializable {`
`7`		`- State makeState(Map conf, int partitionIndex, int numPartitions);`
	`8`	`+ State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions);`
`8`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`package storm.trident.testing;`
`2`	`2`
	`3`	`+import backtype.storm.task.IMetricsContext;`
`3`	`4`	`import storm.trident.state.ITupleCollection;`
`4`	`5`	`import backtype.storm.tuple.Values;`
`5`	`6`	`import java.util.*;`
`@@ -70,7 +71,7 @@ public Factory(int maxSize) {`
`70`	`71`	`}`
`71`	`72`
`72`	`73`	`@Override`
`73`		`- public State makeState(Map conf, int partitionIndex, int numPartitions) {`
	`74`	`+ public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {`
`74`	`75`	`return new LRUMemoryMapState(_maxSize, _id);`
`75`	`76`	`}`
`76`	`77`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,8 +8,10 @@`
`8`	`8`	`public class ComboList extends AbstractList<Object> {`
`9`	`9`	`public static class Factory implements Serializable {`
`10`	`10`	`Pointer[] index;`
	`11`	`+ int numLists;`
`11`	`12`
`12`	`13`	`public Factory(int... sizes) {`
	`14`	`+ numLists = sizes.length;`
`13`	`15`	`int total = 0;`
`14`	`16`	`for(int size: sizes) {`
`15`	`17`	`total+=size;`
`@@ -27,6 +29,9 @@ public Factory(int... sizes) {`
`27`	`29`	`}`
`28`	`30`
`29`	`31`	`public ComboList create(List[] delegates) {`
	`32`	`+ if(delegates.length!=numLists) {`
	`33`	`+ throw new RuntimeException("Expected " + numLists + " lists, but instead got " + delegates.length + " lists");`
	`34`	`+ }`
`30`	`35`	`return new ComboList(delegates, index);`
`31`	`36`	`}`
`32`	`37`	`}`