added shared executor data between tasks, make task data have a map-like interface

Nathan Marz · Nathan Marz · commit 933669968a43 · 2012-06-02T14:00:25.000-07:00
diff --git a/src/clj/backtype/storm/daemon/executor.clj b/src/clj/backtype/storm/daemon/executor.clj
@@ -160,6 +160,7 @@
      :receive-queue ((:executor-receive-queue-map worker) executor-id)
      :storm-id (:storm-id worker)
      :conf (:conf worker)
+     :shared-executor-data (HashMap.)
      :storm-active-atom (:storm-active-atom worker)
      :batch-transfer-queue batch-transfer->worker
      :transfer-fn (fn [task tuple]
diff --git a/src/clj/backtype/storm/daemon/task.clj b/src/clj/backtype/storm/daemon/task.clj
@@ -10,7 +10,7 @@
 
 (bootstrap)
 
-(defn mk-topology-context-builder [worker topology]
+(defn mk-topology-context-builder [worker executor-data topology]
   (let [conf (:conf worker)]
     #(TopologyContext.
       topology
@@ -27,17 +27,20 @@
       (:task-ids worker)
       (:default-shared-resources worker)
       (:user-shared-resources worker)
+      (:shared-executor-data executor-data)
       )))
 
-(defn system-topology-context [worker tid]
+(defn system-topology-context [worker executor-data tid]
   ((mk-topology-context-builder
     worker
+    executor-data
     (:system-topology worker))
    tid))
 
-(defn user-topology-context [worker tid]
+(defn user-topology-context [worker executor-data tid]
   ((mk-topology-context-builder
     worker
+    executor-data
     (:topology worker))
    tid))
 
@@ -140,8 +143,8 @@
   (recursive-map
     :executor-data executor-data
     :task-id task-id
-    :system-context (system-topology-context (:worker executor-data) task-id)
-    :user-context (user-topology-context (:worker executor-data) task-id)
+    :system-context (system-topology-context (:worker executor-data) executor-data task-id)
+    :user-context (user-topology-context (:worker executor-data) executor-data task-id)
     :tasks-fn (mk-tasks-fn <>)
     :object (get-task-object (.getRawTopology ^TopologyContext (:system-context <>)) (:component-id executor-data))
     ))
diff --git a/src/clj/backtype/storm/testing.clj b/src/clj/backtype/storm/testing.clj
@@ -8,6 +8,7 @@
   (:require [backtype.storm [process-simulator :as psim]])
   (:import [org.apache.commons.io FileUtils])
   (:import [java.io File])
+  (:import [java.util HashMap])
   (:import [java.util.concurrent.atomic AtomicInteger])
   (:import [java.util.concurrent ConcurrentHashMap])
   (:import [backtype.storm.utils Time Utils RegisteredGlobalState])
@@ -560,6 +561,7 @@
                   nil
                   [(int 1)]
                   {}
-                  {})]
+                  {}
+                  (HashMap.))]
     (TupleImpl. context values 1 stream)
     ))
diff --git a/src/jvm/backtype/storm/task/TopologyContext.java b/src/jvm/backtype/storm/task/TopologyContext.java
@@ -28,18 +28,20 @@ public class TopologyContext extends WorkerTopologyContext {
     private Integer _taskId;
     private Map<String, Object> _taskData = new HashMap<String, Object>();
     private List<ITaskHook> _hooks = new ArrayList<ITaskHook>();
+    private Map<String, Object> _executorData;
 
     
     public TopologyContext(StormTopology topology, Map stormConf,
             Map<Integer, String> taskToComponent, Map<String, List<Integer>> componentToSortedTasks,
             Map<String, Map<String, Fields>> componentToStreamToFields,
             String stormId, String codeDir, String pidDir, Integer taskId,
             Integer workerPort, List<Integer> workerTasks, Map<String, Object> defaultResources,
-            Map<String, Object> userResources) {
+            Map<String, Object> userResources, Map<String, Object> executorData) {
         super(topology, stormConf, taskToComponent, componentToSortedTasks,
                 componentToStreamToFields, stormId, codeDir, pidDir,
                 workerPort, workerTasks, defaultResources, userResources);
         _taskId = taskId;
+        _executorData = executorData;
     }
 
     /**
@@ -171,6 +173,14 @@ public void setTaskData(String name, Object data) {
     public Object getTaskData(String name) {
         return _taskData.get(name);
     }
+
+    public void setExecutorData(String name, Object data) {
+        _executorData.put(name, data);
+    }
+    
+    public Object getExecutorData(String name) {
+        return _executorData.get(name);
+    }    
     
     public void addTaskHook(ITaskHook hook) {
         hook.prepare(_stormConf, this);
