Avoid usage of six.moves in local functions

Currently it appears that using six.moves in threaded code isn't
working as expected (something there in six does not appear to
be thread safe) so until this is fixed avoid using those moves
in functions in the examples and in the utility code (and instead
import the moved function at the top of the module in code to
avoid any threaded usage problems).

Upstream bug filed at:

https://bitbucket.org/gutworth/six/issue/98/

Fixes bug 1377514

Change-Id: I3fc1819df8fb42d0c3d394bbc7d047b09152af68

taskflow/examples/jobboard_produce_consume_colors.py

@@ -30,7 +30,7 @@ top_dir = os.path.abspath(os.path.join(os.path.dirname(__file__),
                                        os.pardir))
 sys.path.insert(0, top_dir)
 
-import six
+from six.moves import range as compat_range
 from zake import fake_client
 
 from taskflow import exceptions as excp
@@ -137,7 +137,7 @@ def producer(ident, client):
     name = "P-%s" % (ident)
     safe_print(name, "started")
     with backends.backend(name, SHARED_CONF.copy(), client=client) as board:
-        for i in six.moves.xrange(0, PRODUCER_UNITS):
+        for i in compat_range(0, PRODUCER_UNITS):
             job_name = "%s-%s" % (name, i)
             details = {
                 'color': random.choice(['red', 'blue']),
@@ -151,13 +151,13 @@ def producer(ident, client):
 def main():
     with contextlib.closing(fake_client.FakeClient()) as c:
         created = []
-        for i in range(0, PRODUCERS):
+        for i in compat_range(0, PRODUCERS):
             p = threading.Thread(target=producer, args=(i + 1, c))
             p.daemon = True
             created.append(p)
             p.start()
         consumed = collections.deque()
-        for i in range(0, WORKERS):
+        for i in compat_range(0, WORKERS):
             w = threading.Thread(target=worker, args=(i + 1, c, consumed))
             w.daemon = True
             created.append(w)

taskflow/examples/wbe_mandelbrot.py

@@ -20,13 +20,13 @@ import os
 import sys
 import threading
 
-import six
-
 top_dir = os.path.abspath(os.path.join(os.path.dirname(__file__),
                                        os.pardir,
                                        os.pardir))
 sys.path.insert(0, top_dir)
 
+from six.moves import range as compat_range
+
 from taskflow import engines
 from taskflow.engines.worker_based import worker
 from taskflow.patterns import unordered_flow as uf
@@ -84,7 +84,7 @@ class MandelCalculator(task.Task):
         def mandelbrot(x, y, max_iters):
             c = complex(x, y)
             z = 0.0j
-            for i in six.moves.xrange(max_iters):
+            for i in compat_range(max_iters):
                 z = z * z + c
                 if (z.real * z.real + z.imag * z.imag) >= 4:
                     return i
@@ -95,10 +95,10 @@ class MandelCalculator(task.Task):
         pixel_size_x = (max_x - min_x) / width
         pixel_size_y = (max_y - min_y) / height
         block = []
-        for y in six.moves.xrange(chunk[0], chunk[1]):
+        for y in compat_range(chunk[0], chunk[1]):
             row = []
             imag = min_y + y * pixel_size_y
-            for x in six.moves.xrange(0, width):
+            for x in compat_range(0, width):
                 real = min_x + x * pixel_size_x
                 row.append(mandelbrot(real, imag, max_iters))
             block.append(row)
@@ -133,7 +133,7 @@ def calculate(engine_conf):
     # Compose our workflow.
     height, width = IMAGE_SIZE
     chunk_size = int(math.ceil(height / float(CHUNK_COUNT)))
-    for i in six.moves.xrange(0, CHUNK_COUNT):
+    for i in compat_range(0, CHUNK_COUNT):
         chunk_name = 'chunk_%s' % i
         task_name = "calculation_%s" % i
         # Break the calculation up into chunk size pieces.
@@ -225,7 +225,7 @@ def create_fractal():
     try:
         # Create a set of workers to simulate actual remote workers.
         print('Running %s workers.' % (WORKERS))
-        for i in range(0, WORKERS):
+        for i in compat_range(0, WORKERS):
             worker_conf['topic'] = 'calculator_%s' % (i + 1)
             worker_topics.append(worker_conf['topic'])
             w = worker.Worker(**worker_conf)

taskflow/utils/kazoo_utils.py

@@ -17,6 +17,7 @@
 from kazoo import client
 from kazoo import exceptions as k_exc
 import six
+from six.moves import zip as compat_zip
 
 from taskflow import exceptions as exc
 from taskflow.utils import reflection
@@ -100,7 +101,7 @@ def checked_commit(txn):
         return []
     results = txn.commit()
     failures = []
-    for op, result in six.moves.zip(txn.operations, results):
+    for op, result in compat_zip(txn.operations, results):
         if isinstance(result, k_exc.KazooException):
             failures.append((op, result))
     if len(results) < len(txn.operations):

taskflow/utils/misc.py

@@ -34,6 +34,8 @@ import traceback
 from oslo.serialization import jsonutils
 from oslo.utils import netutils
 import six
+from six.moves import map as compat_map
+from six.moves import range as compat_range
 from six.moves.urllib import parse as urlparse
 
 from taskflow import exceptions as exc
@@ -286,7 +288,7 @@ def item_from(container, index, name=None):
 
 def get_duplicate_keys(iterable, key=None):
     if key is not None:
-        iterable = six.moves.map(key, iterable)
+        iterable = compat_map(key, iterable)
     keys = set()
     duplicates = set()
     for item in iterable:
@@ -373,7 +375,7 @@ class ExponentialBackoff(object):
     def __iter__(self):
         if self.count <= 0:
             raise StopIteration()
-        for i in six.moves.range(0, self.count):
+        for i in compat_range(0, self.count):
             yield min(self.exponent ** i, self.max_backoff)
 
     def __str__(self):

tox.ini

@@ -54,7 +54,7 @@ builtins = _
 exclude = .venv,.tox,dist,doc,./taskflow/openstack/common,*egg,.git,build,tools
 
 [hacking]
-import_exceptions = six.moves.mock
+import_exceptions = six.moves
     taskflow.test.mock
     unittest.mock