Merge pull request #720 from acmcelwee/amc-threaded-walker

ThreadedWalker constructor accepts a Semaphore, not an int

Author: ejholmes

stacker/actions/base.py

@@ -7,7 +7,7 @@
 import logging
 import threading
 
-from ..dag import walk, ThreadedWalker
+from ..dag import walk, ThreadedWalker, UnlimitedSemaphore
 from ..plan import Step, build_plan, build_graph
 
 import botocore.exceptions
@@ -53,7 +53,12 @@ def build_walker(concurrency):
     """
     if concurrency == 1:
         return walk
-    return ThreadedWalker(concurrency).walk
+
+    semaphore = UnlimitedSemaphore()
+    if concurrency > 1:
+        semaphore = threading.Semaphore(concurrency)
+
+    return ThreadedWalker(semaphore).walk
 
 
 def plan(description, stack_action, context,

stacker/dag/__init__.py

@@ -414,14 +414,12 @@ class ThreadedWalker(object):
     allows, using threads.
 
     Args:
-        semaphore (threading.Semaphore, optional): a semaphore object which
+        semaphore (threading.Semaphore): a semaphore object which
             can be used to control how many steps are executed in parallel.
-            By default, there is not limit to the amount of parallelism,
-            other than what the graph topology allows.
     """
 
-    def __init__(self, semaphore=None):
-        self.semaphore = semaphore or UnlimitedSemaphore()
+    def __init__(self, semaphore):
+        self.semaphore = semaphore
 
     def walk(self, dag, walk_func):
         """ Walks each node of the graph, in parallel if it can.

stacker/tests/test_dag.py

@@ -5,7 +5,12 @@
 
 from nose import with_setup
 from nose.tools import nottest, raises
-from stacker.dag import DAG, DAGValidationError, ThreadedWalker
+from stacker.dag import (
+    DAG,
+    DAGValidationError,
+    ThreadedWalker,
+    UnlimitedSemaphore
+)
 import threading
 
 dag = None
@@ -220,7 +225,7 @@ def test_transitive_deep_reduction():
 @with_setup(blank_setup)
 def test_threaded_walker():
     dag = DAG()
-    walker = ThreadedWalker()
+    walker = ThreadedWalker(UnlimitedSemaphore())
 
     # b and c should be executed at the same time.
     dag.from_dict({'a': ['b', 'c'],