minor cleanup

Nathan Marz · Nathan Marz · commit fc05f1e097fc · 2012-03-21T04:59:09.000-07:00
diff --git a/src/dev/resources/tester_spout.py b/src/dev/resources/tester_spout.py
@@ -21,6 +21,7 @@ def ack(self, id):
         del self.pending[id]
 
     def fail(self, id):
+        log("emitting " + self.pending[id] + " on fail")
         emit([self.pending[id]], id=id)
 
 TesterSpout().run()
diff --git a/src/jvm/backtype/storm/task/ShellBolt.java b/src/jvm/backtype/storm/task/ShellBolt.java
@@ -53,6 +53,9 @@ public class ShellBolt implements IBolt {
     private ShellProcess _process;
     private volatile boolean _running = true;
     private LinkedBlockingQueue _pendingWrites = new LinkedBlockingQueue();
+    
+    private Thread _readerThread;
+    private Thread _writerThread;
 
     public ShellBolt(ShellComponent component) {
         this(component.get_execution_command(), component.get_script());
@@ -76,7 +79,7 @@ public void prepare(Map stormConf, TopologyContext context,
         }
 
         // reader
-        new Thread(new Runnable() {
+        _readerThread = new Thread(new Runnable() {
             public void run() {
                 while (_running) {
                     try {
@@ -99,14 +102,14 @@ public void run() {
                     } catch (IOException e) {
                         throw new RuntimeException(e);
                     } catch (InterruptedException e) {
-                        // ignore
                     }
                 }
             }
-        }).start();
+        });
+        
+        _readerThread.start();
 
-        // writer
-        new Thread(new Runnable() {
+        _writerThread = new Thread(new Runnable() {
             public void run() {
                 while (_running) {
                     try {
@@ -117,11 +120,12 @@ public void run() {
                     } catch (IOException e) {
                         throw new RuntimeException(e);
                     } catch (InterruptedException e) {
-                        // ignore
                     }
                 }
             }
-        }).start();
+        });
+        
+        _writerThread.start();
     }
 
     public void execute(Tuple input) {
@@ -142,9 +146,9 @@ public void execute(Tuple input) {
     }
 
     public void cleanup() {
+        _running = false;
         _process.destroy();
         _inputs.clear();
-        _running = false;
     }
 
     private void handleAck(Map action) {
diff --git a/src/jvm/backtype/storm/utils/ShellProcess.java b/src/jvm/backtype/storm/utils/ShellProcess.java
@@ -29,12 +29,6 @@ public String launch(Map conf, TopologyContext context) throws IOException {
         processIn = new DataOutputStream(_subprocess.getOutputStream());
         processOut = new BufferedReader(new InputStreamReader(_subprocess.getInputStream()));
 
-        // this doesn't seem to work when the jvm dies due to at least
-        // some kinds of errors... does it help at all?
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-            public void run() { _subprocess.destroy(); }
-        });
-
         writeString(context.getPIDDir());
         String pid = readString();
         writeObject(conf);

Original file line number	Diff line number	Diff line change
`@@ -53,6 +53,9 @@ public class ShellBolt implements IBolt {`
`53`	`53`	`private ShellProcess _process;`
`54`	`54`	`private volatile boolean _running = true;`
`55`	`55`	`private LinkedBlockingQueue _pendingWrites = new LinkedBlockingQueue();`
	`56`	`+`
	`57`	`+ private Thread _readerThread;`
	`58`	`+ private Thread _writerThread;`
`56`	`59`
`57`	`60`	`public ShellBolt(ShellComponent component) {`
`58`	`61`	`this(component.get_execution_command(), component.get_script());`
`@@ -76,7 +79,7 @@ public void prepare(Map stormConf, TopologyContext context,`
`76`	`79`	`}`
`77`	`80`
`78`	`81`	`// reader`
`79`		`- new Thread(new Runnable() {`
	`82`	`+ _readerThread = new Thread(new Runnable() {`
`80`	`83`	`public void run() {`
`81`	`84`	`while (_running) {`
`82`	`85`	`try {`
`@@ -99,14 +102,14 @@ public void run() {`
`99`	`102`	`} catch (IOException e) {`
`100`	`103`	`throw new RuntimeException(e);`
`101`	`104`	`} catch (InterruptedException e) {`
`102`		`- // ignore`
`103`	`105`	`}`
`104`	`106`	`}`
`105`	`107`	`}`
`106`		`- }).start();`
	`108`	`+ });`
	`109`	`+`
	`110`	`+ _readerThread.start();`
`107`	`111`
`108`		`- // writer`
`109`		`- new Thread(new Runnable() {`
	`112`	`+ _writerThread = new Thread(new Runnable() {`
`110`	`113`	`public void run() {`
`111`	`114`	`while (_running) {`
`112`	`115`	`try {`
`@@ -117,11 +120,12 @@ public void run() {`
`117`	`120`	`} catch (IOException e) {`
`118`	`121`	`throw new RuntimeException(e);`
`119`	`122`	`} catch (InterruptedException e) {`
`120`		`- // ignore`
`121`	`123`	`}`
`122`	`124`	`}`
`123`	`125`	`}`
`124`		`- }).start();`
	`126`	`+ });`
	`127`	`+`
	`128`	`+ _writerThread.start();`
`125`	`129`	`}`
`126`	`130`
`127`	`131`	`public void execute(Tuple input) {`
`@@ -142,9 +146,9 @@ public void execute(Tuple input) {`
`142`	`146`	`}`
`143`	`147`
`144`	`148`	`public void cleanup() {`
	`149`	`+ _running = false;`
`145`	`150`	`_process.destroy();`
`146`	`151`	`_inputs.clear();`
`147`		`- _running = false;`
`148`	`152`	`}`
`149`	`153`
`150`	`154`	`private void handleAck(Map action) {`