Better Python 2 / 3 compatibility

This fixes the breaking changes for Python 2 that were made in the
last attempt and improves Python 3 compatibility a bit more.

Python 3 supports the new `file` argument to the `print` function
that allows you to specify where to print to.  In the code, this
was being used to print to stderr.  Unfortunately Python 2 does not
yet support this, so we are now using `sys.stderr.write` instead.

Python 2's `Queue` module was renamed to `queue` in Python 3.  We
should be importing them both to ensure that the code runs on both
versions.  By default, we are assuming that Python 3 will be used
and are using it as the first import before falling back to the old
name.

`threading._sleep` was a private method and was removed in Python 3.
It has been replaced by `time.sleep` for now, as that is both public
and consistent across versions.

When determining chunk sizes and the number of chunks, there was a
possibility of floating values being passed in or calculated.  To
ensure that the generated range does not throw an exception, we are
manually casting these to integers.

The iterator was previously changed so `next` became `__next__`,
which was the Python 3 method for the `next` function.  This adds
Python 2 compatibility alongside of these changes.

Author: kevin-brown

async/__init__.py

@@ -30,7 +30,7 @@ def thread_interrupt_handler(signum, frame):
         signal.signal(signal.SIGINT, thread_interrupt_handler)
     except ValueError:
         # happens if we don't try it from the main thread
-        print("Failed to setup thread-interrupt handler. This is usually not critical", file=sys.stderr)
+        sys.stderr.write("Failed to setup thread-interrupt handler. This is usually not critical")
     # END exception handling
 
 

async/channel.py

@@ -3,9 +3,15 @@
 # This module is part of async and is released under
 # the New BSD License: http://www.opensource.org/licenses/bsd-license.php
 """Contains a queue based channel implementation"""
-from queue import (
-    Empty, 
-    Full
+try:
+    from queue import (
+        Empty,
+        Full
+    )
+except ImportError:
+    from Queue import (
+        Empty,
+        Full
     )
 
 from .util import (
@@ -161,6 +167,10 @@ def __next__(self):
             return items[0]
         raise StopIteration
 
+    def next(self):
+        """Support the Python 2 iterator syntax"""
+        return self.__next__()
+
     #} END iterator protocol
 
 
@@ -332,7 +342,7 @@ class IteratorReader(Reader):
 
     def __init__(self, iterator):
         self._empty = False
-        if not hasattr(iterator, 'next'):
+        if not hasattr(iterator, 'next') and not (hasattr(iterator, "__next__")):
             raise ValueError("Iterator %r needs a next() function" % iterator)
         self._iter = iterator
         self._lock = self.lock_type()

async/pool.py

@@ -14,9 +14,15 @@
         DummyLock
     )
 
-from queue import (
-    Queue, 
-    Empty
+try:
+    from queue import (
+        Queue,
+        Empty
+        )
+except ImportError:
+    from Queue import (
+        Queue,
+        Empty
     )
 
 from .graph import Graph
@@ -277,6 +283,10 @@ def _prepare_channel_read(self, task, count):
                 remainder = actual_count - (numchunks * chunksize)
             # END handle chunking
 
+            # These both need to be integers, not floats
+            chunksize = int(chunksize)
+            numchunks = int(numchunks)
+
             # the following loops are kind of unrolled - code duplication
             # should make things execute faster. Putting the if statements
             # into the loop would be less code, but ... slower

async/test/test_channel.py

@@ -3,8 +3,11 @@
 # This module is part of async and is released under
 # the New BSD License: http://www.opensource.org/licenses/bsd-license.php
 """Channel testing"""
-from .lib import *
-from async.channel import *
+from .lib import TestBase
+from async.channel import (
+    CallbackChannelWriter, CallbackChannelReader, ChannelWriter, ChannelReader,
+    IteratorReader, mkchannel, ReadOnly
+)
 
 import time
 

async/test/test_performance.py

@@ -38,7 +38,7 @@ def test_base(self):
                     if read_mode == 1:
                         mode_info = "read(1) * %i" % ni
                     # END mode info
-                    fmt = "Threadcount=%%i: Produced %%i items using %s in %%i transformations in %%f s (%%f items / s)" % mode_info
+                    fmt = "Threadcount=%%i: Produced %%i items using %s in %%i transformations in %%f s (%%f items / s)\n" % mode_info
                     reader = rcs[-1]
                     st = time.time()
                     if read_mode == 1:
@@ -49,7 +49,7 @@ def test_base(self):
                         assert len(reader.read(0)) == ni
                     # END handle read mode
                     elapsed = time.time() - st
-                    print(fmt % (num_threads, ni, num_transformers, elapsed, ni / elapsed), file=sys.stderr)
+                    sys.stderr.write(fmt % (num_threads, ni, num_transformers, elapsed, ni / elapsed))
                 # END for each read-mode
             # END for each amount of processors
         # END for each thread count

async/test/test_pool.py

@@ -23,7 +23,7 @@ class TestThreadPool(TestBase):
 
     def _assert_single_task(self, p, async=False):
         """Performs testing in a synchronized environment"""
-        print("Threadpool: Starting single task (async = %i) with %i threads" % (async, p.size()), file=sys.stderr)
+        sys.stderr.write("Threadpool: Starting single task (async = %i) with %i threads\n" % (async, p.size()))
         null_tasks = p.num_tasks()      # in case we had some before
 
         # add a simple task
@@ -71,10 +71,10 @@ def _assert_single_task(self, p, async=False):
                 assert i == items[0]
         # END for each item
         elapsed = time.time() - st
-        print("Threadpool: processed %i individual items, with %i threads, one at a time, in %f s ( %f items / s )" % (ni, p.size(), elapsed, ni / elapsed), file=sys.stderr)
-        
-        # it couldn't yet notice that the input is depleted as we pulled exaclty 
-        # ni items - the next one would remove it. Instead, we delete our channel 
+        sys.stderr.write("Threadpool: processed %i individual items, with %i threads, one at a time, in %f s ( %f items / s )\n" % (ni, p.size(), elapsed, ni / elapsed))
+
+        # it couldn't yet notice that the input is depleted as we pulled exaclty
+        # ni items - the next one would remove it. Instead, we delete our channel
         # which triggers orphan handling
         assert not task.is_done()
         assert p.num_tasks() == 1 + null_tasks
@@ -162,8 +162,8 @@ def _assert_single_task(self, p, async=False):
         # END pull individual items
         # too many processing counts ;)
         elapsed = time.time() - st
-        print("Threadpool: processed %i individual items in chunks of %i, with %i threads, one at a time, in %f s ( %f items / s )" % (ni, ni/4, p.size(), elapsed, ni / elapsed), file=sys.stderr)
-        
+        sys.stderr.write("Threadpool: processed %i individual items in chunks of %i, with %i threads, one at a time, in %f s ( %f items / s)\n" % (ni, ni/4, p.size(), elapsed, ni / elapsed))
+
         task._assert(ni, ni)
         assert p.num_tasks() == 1 + null_tasks
         assert p.remove_task(task) is p     # del manually this time
@@ -207,17 +207,17 @@ def _assert_single_task(self, p, async=False):
         assert len(rc.read()) == nri
         assert task.is_done()
         assert isinstance(task.error(), AssertionError)
-        
-        print("done with everything", file=sys.stderr)
-        
-        
-        
+
+        sys.stderr.write("done with everything\n")
+
+
+
     def _assert_async_dependent_tasks(self, pool):
         # includes failure in center task, 'recursive' orphan cleanup
         # This will also verify that the channel-close mechanism works
         # t1 -> t2 -> t3
-    
-        print("Threadpool: starting async dependency test in %i threads" % pool.size(), file=sys.stderr)
+
+        sys.stderr.write("Threadpool: starting async dependency test in %i threads\n" % pool.size())
         null_tasks = pool.num_tasks()
         ni = 1000
         count = 3
@@ -243,9 +243,9 @@ def _assert_async_dependent_tasks(self, pool):
         time.sleep(0.15)
         assert sys.getrefcount(ts[-1]) == 2 # ts + call
         assert sys.getrefcount(ts[0]) == 2  # ts + call
-        print("Dependent Tasks: evaluated %i items of %i dependent in %f s ( %i items / s )" % (ni, aic, elapsed, ni / elapsed), file=sys.stderr)
-        
-        
+        sys.stderr.write("Dependent Tasks: evaluated %i items of %i dependent in %f s ( %i items / s )\n" % (ni, aic, elapsed, ni / elapsed))
+
+
         # read(1)
         #########
         ts, rcs = make_task()
@@ -257,9 +257,9 @@ def _assert_async_dependent_tasks(self, pool):
         elapsed_single = time.time() - st
         # another read yields nothing, its empty
         assert len(rcs[-1].read()) == 0
-        print("Dependent Tasks: evaluated %i items with read(1) of %i dependent in %f s ( %i items / s )" % (ni, aic, elapsed_single, ni / elapsed_single), file=sys.stderr)
-        
-        
+        sys.stderr.write("Dependent Tasks: evaluated %i items with read(1) of %i dependent in %f s ( %i items / s )\n" % (ni, aic, elapsed_single, ni / elapsed_single))
+
+
         # read with min-count size
         ###########################
         # must be faster, as it will read ni / 4 chunks
@@ -277,8 +277,8 @@ def _assert_async_dependent_tasks(self, pool):
         elapsed_minsize = time.time() - st
         # its empty
         assert len(rcs[-1].read()) == 0
-        print("Dependent Tasks: evaluated %i items with read(1), min_size=%i, of %i dependent in %f s ( %i items / s )" % (ni, nri, aic, elapsed_minsize, ni / elapsed_minsize), file=sys.stderr)
-        
+        sys.stderr.write("Dependent Tasks: evaluated %i items with read(1), min_size=%i, of %i dependent in %f s ( %i items / s )\n" % (ni, nri, aic, elapsed_minsize, ni / elapsed_minsize))
+
         # it should have been a bit faster at least, and most of the time it is
         # Sometimes, its not, mainly because:
         # * The test tasks lock a lot, hence they slow down the system
@@ -319,10 +319,10 @@ def _assert_async_dependent_tasks(self, pool):
         items = p2rcs[-1].read()
         elapsed = time.time() - st
         assert len(items) == ni
-        
-        print("Dependent Tasks: evaluated 2 connected pools and %i items with read(0), of %i dependent tasks in %f s ( %i items / s )" % (ni, aic + aic-1, elapsed, ni / elapsed), file=sys.stderr)
-        
-        
+
+        sys.stderr.write("Dependent Tasks: evaluated 2 connected pools and %i items with read(0), of %i dependent tasks in %f s ( %i items / s )\n" % (ni, aic + aic-1, elapsed, ni / elapsed))
+
+
         # loose the handles of the second pool to allow others to go as well
         del(p2rcs); del(p2ts)
         assert p2.num_tasks() == 0
@@ -344,9 +344,9 @@ def _assert_async_dependent_tasks(self, pool):
         # END for each item to get
         elapsed = time.time() - st
         del(reader)     # decrement refcount
-        
-        print("Dependent Tasks: evaluated 2 connected pools and %i items with read(1), of %i dependent tasks in %f s ( %i items / s )" % (ni, aic + aic-1, elapsed, ni / elapsed), file=sys.stderr)
-        
+
+        sys.stderr.write("Dependent Tasks: evaluated 2 connected pools and %i items with read(1), of %i dependent tasks in %f s ( %i items / s )\n" % (ni, aic + aic-1, elapsed, ni / elapsed))
+
         # another read is empty
         assert len(rcs[-1].read()) == 0
 
@@ -481,6 +481,5 @@ def test_base(self):
         # DEPENDENT TASK ASYNC MODE
         ###########################
         self._assert_async_dependent_tasks(p)
-        
-        print("Done with everything", file=sys.stderr)
-        
+
+        sys.stderr.write("Done with everything\n")

async/test/test_thread.py

@@ -6,7 +6,11 @@
 """ Test thead classes and functions"""
 from .lib import *
 from async.thread import *
-from queue import Queue
+try:
+    from queue import Queue
+except ImportError:
+    from Queue import Queue
+
 import time
 
 class TestWorker(WorkerThread):

async/thread.py

@@ -7,7 +7,10 @@
 __docformat__ = "restructuredtext"
 import threading
 import inspect
-import queue
+try:
+    import queue
+except ImportError:
+    import Queue as queue
 
 import sys
 

async/util.py

@@ -7,17 +7,18 @@
 from threading import (
     Lock,
     _allocate_lock,
-    _sleep,
     _time,
-    )
+)
 
-from queue import (
-        Empty,
-        )
+try:
+    from queue import Empty
+except ImportError:
+    from Queue import Empty
 
 from collections import deque
 import sys
 import os
+import time
 
 #{ Routines
 
@@ -130,7 +131,7 @@ def wait(self, timeout=None):
                     # this makes 4 threads working as good as two, but of course
                     # it causes more frequent micro-sleeping
                     #delay = min(delay * 2, remaining, .05)
-                    _sleep(delay)
+                    time.sleep(delay)
                 # END endless loop
                 if not gotit:
                     try: