Prevent Queue.qsize() from raising NotImplementedError on Mac OS X

multiprocessing.Queue.qsize() may raise NotImplementedError on Unix
platforms like Mac OS X where sem_getvalue() is not implemented. In
order to address this problem, subclass Queue and use an internal
synchronized shared counter whose value is updated when the put() and
get() methods are called. As a side benefit, this makes it possible
to have a reliable version of both qsize() and empty(). Thanks to
Jean-Baptiste Marquette for reporting this problem (#11).

This merges branch 'fix-Queue.qsize-NotImplementedError'

Misc/ACKS

@@ -25,3 +25,4 @@ Javier Blasco, our most enthusiastic user, for such thoughtful suggestions
 Pablo Ramírez, the second person to ever use LEMON — that we know of
 Sofía León, for designing for us such a marvelous LEMON icon
 José Enrique Ruiz, for suggesting that we set extglob
+Jean-Baptiste Marquette, for reporting a bug under Mac OS X

diffphot.py

@@ -637,7 +637,7 @@ def best(self, n, fraction = 0.1, pct = 0.01, max_iters = None, minimum = None):
 # The Queue is global -- this works, but note that we could have
 # passed its reference to the function managed by pool.map_async.
 # See http://stackoverflow.com/a/3217427/184363
-queue = multiprocessing.Queue()
+queue = methods.Queue()
 
 def parallel_light_curves(args):
     """ Method argument of map_async to compute light curves in parallel.

methods.py

@@ -22,6 +22,8 @@
 import functools
 import logging
 import math
+import multiprocessing
+import multiprocessing.queues
 import numpy
 import os
 import os.path
@@ -553,3 +555,66 @@ def log_error(function, path, excinfo):
             else:
                 msg = "Temporary file '%s' removed"
                 logging.debug(msg % path)
+
+class SharedCounter(object):
+    """ A synchronized shared counter.
+
+    The locking done by multiprocessing.Value ensures that only a single
+    process or thread may read or write the in-memory ctypes object. However,
+    in order to do n += 1, Python performs a read followed by a write, so a
+    second process may read the old value before the new one is written by the
+    first process. The solution is to use a multiprocessing.Lock to guarantee
+    the atomicity of the modifications to Value.
+
+    This class comes almost entirely from Eli Bendersky's blog:
+    http://eli.thegreenplace.net/2012/01/04/shared-counter-with-pythons-multiprocessing/
+
+    """
+
+    def __init__(self, n = 0):
+        self.count = multiprocessing.Value('i', n)
+
+    def increment(self, n = 1):
+        """ Increment the counter by n (default = 1) """
+        with self.count.get_lock():
+            self.count.value += n
+
+    @property
+    def value(self):
+        """ Return the value of the counter """
+        return self.count.value
+
+
+class Queue(multiprocessing.queues.Queue):
+    """ A portable implementation of multiprocessing.Queue.
+
+    Because of multithreading / multiprocessing semantics, Queue.qsize() may
+    raise the NotImplementedError exception on Unix platforms like Mac OS X
+    where sem_getvalue() is not implemented. This subclass addresses this
+    problem by using a synchronized shared counter (initialized to zero) and
+    increasing / decreasing its value every time the put() and get() methods
+    are called, respectively. This not only prevents NotImplementedError from
+    being raised, but also allows us to implement a reliable version of both
+    qsize() and empty().
+
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(Queue, self).__init__(*args, **kwargs)
+        self.size = SharedCounter(0)
+
+    def put(self, *args, **kwargs):
+        self.size.increment(1)
+        super(Queue, self).put(*args, **kwargs)
+
+    def get(self, *args, **kwargs):
+        self.size.increment(-1)
+        return super(Queue, self).get(*args, **kwargs)
+
+    def qsize(self):
+        """ Reliable implementation of multiprocessing.Queue.qsize() """
+        return self.size.value
+
+    def empty(self):
+        """ Reliable implementation of multiprocessing.Queue.empty() """
+        return not self.qsize()

periods.py

@@ -264,7 +264,7 @@ def best_period(self, initial_step, exhaustive_step):
 # The Queue is global -- this works, but note that we could have
 # passed its reference to the function managed by pool.map_async.
 # See http://stackoverflow.com/a/3217427/184363
-queue = multiprocessing.Queue()
+queue = methods.Queue()
 
 def parallel_periods(args):
     """ Method argument of map_async to compute periods in parallel.

photometry.py

@@ -79,7 +79,7 @@
 # The Queue is global -- this works, but note that we could have
 # passed its reference to the function managed by pool.map_async.
 # See http://stackoverflow.com/a/3217427/184363
-queue = multiprocessing.Queue()
+queue = methods.Queue()
 
 class InputFITSFiles(collections.defaultdict):
     """ Map each photometric filter to a list of FITS files.
@@ -197,7 +197,7 @@ def parallel_photometry(args):
     the FITS image listed in options.coordinates, using the aperture, annulus
     and dannulus defined by the PhotometricParameters object. The result is
     another three-element tuple, which is put into the module-level 'queue'
-    multiprocessing.Queue object. This tuple contains (1) a database.Image
+    object, a process shared queue. This tuple contains (1) a database.Image
     object, (2) a database.PhotometricParameters object and (3) a qphot.QPhot
     object -- therefore mapping each FITS file and the parameters used for
     photometry to the measurements returned by qphot.

seeing.py

@@ -425,7 +425,7 @@ def elongation(self, per = 50, mode = 'median'):
 # This Queue is global -- this works, but note that we could have
 # passed its reference to the function managed by pool.map_async.
 # See http://stackoverflow.com/a/3217427/184363
-queue = multiprocessing.Queue()
+queue = methods.Queue()
 
 parser = customparser.get_parser(description)
 parser.usage = "%prog [OPTION]... INPUT_IMGS... OUTPUT_DIR"