[Jython-users] concurrent.py

[Ype K. <yk...@xs...>]

This is especially for those interested in using multithreading and multiprocessors 
in Jython:

After the profiler showed me that quite a bit of time is spent non waiting in
threading.py (even in no more used debugging code), I looked around for a more
'native' solution and, to my surprise, found something very appropriate for jython.

I downloaded the zip file and wrote a python interface (less than 100 code lines)
below.
Thanks to Doug Lea and Mr. Laufer for making this so easy.

The package is called util.concurrent and it uses method names that are almost a
perfect match for the standard python names. I refrained from adapting the names
completely to the python standard for performance reasons. (Having small differences
between `platforms' is quite common in python.)

I think concurrent.py is an appropriate name for what follows:

""" Jython interface to the util.concurrent package of Doug Lea.

See:
http://g.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html

This is based on the java 1.1 precompiled version downloaded from:
http://www.cs.luc.edu/~laufer/courses/337/handouts/concurrent11.zip

This zip file should be on the class path when invoking jython
for using this module.

From the documentation of util.concurrent:

"
    This package provides standardized, efficient versions of utility classes
    commonly encountered in concurrent Java programming. The ideas behind and
    applications of several of these classes are discussed in the second edition of
    Concurrent Programming in Java. [...]

    The packege mainly consists of implementations of a few interfaces:

        Sync -- locks, conditions
        Channel -- queues, buffers
        Barrier -- multi-party synchronization
        SynchronizedVariable -- atomic ints, refs etc
        java.util.Collection -- collections
        Executor -- replacements for direct use of Thread

    Plus some utilities and frameworks that build upon these.
"
(One of these frameworks is the Fork/Join Task framework for
efficient use of threads on multiprocessor systems.
There is no python interface defined for this.)

This python module defines alternate and faster versions of:

Queue.Queue
threading.RLock
threading.Condition
threading.Semaphore

Also defined, but not faster:

threading.Lock
threading.Event

(The Event defined here is the standard version, it
may be faster because it is using a faster Condition).


Non standard behaviour:

Not available:              Use instead:

acquire() with a time out   attempt(msecs)
wait() with time out        timedwait(msecs)
Semaphore()                 Semaphore(1).

size() of unbounded Queue   nothing.


A java.lang.Interrupt can be generated when threads have been
interrupted elsewhere.
The standard behaviour in jython is that a java.lang.Interrupt
(eg. during time.sleep()) causes a message to standard output:
"Interrupted thread".
"""

import EDU.oswego.cs.dl.util.concurrent as _conc

# To access the full util.concurrent module:
# from _conc import *

class Queue:
   """Uses (Bounded)LinkedQueue from util.concurrent."""
   def __init__(self, maxSize = 0):
       if maxSize == 0:
           self.q = _conc.LinkedQueue() # no qsize available forLinkedQueue, why?
       else:
           self.q = _conc.BoundedLinkedQueue(maxSize)
           self.qsize = self.q.size
           self.full = lambda self=self: self.qsize() >= self.q.capacity()

       self.empty = self.q.isEmpty
       self.get = self.q.take
       self.put = self.q.put
       # Add when needed:
       #self.get_timedwait = self.q.poll # msecs
       #self.put_timedwait = self.q.offer # msecs

   def put_nowait(self, item): return self.q.offer(item, 0)
   def get_nowait(self): return self.q.poll(0)


Lock = _conc.Mutex
RLock = _conc.ReentrantLock

class Condition:
    def __init__(self, lock=None):
        if lock is None:
            lock = RLock()
        self.acquire = lock.acquire
        self.release = lock.release

        cv = _conc.CondVar(lock)
        self.notify = cv.signal
        self.notifyAll = cv.broadcast
        self.wait = cv.await
        self.timedwait = cv.timedwait

Semaphore = _conc.Semaphore # no more default 1 arg.

class Event:
    # Based on threading.py from the jython 21a3 distribution.
    # Uses the Condition and Lock of this module. Added a timedwait() method.

    # Might alternatively be based on _conc.WaitableBoolean, but then there is no timeout available.
    # (WaitableBoolean needs java 1.2 collections.)

    # After Tim Peters' event class (without is_posted())

    def __init__(self):
        self.__cond = Condition(Lock())
        self.__flag = 0

    def isSet(self):
        return self.__flag

    def set(self):
        self.__cond.acquire()
        self.__flag = 1
        self.__cond.notifyAll()
        self.__cond.release()

    def clear(self): # _cond.Latch can't do this.
        self.__cond.acquire()
        self.__flag = 0
        self.__cond.release()

    def wait(self): # time argument only in timedwait() method
        self.__cond.acquire()
        if not self.__flag:
            self.__cond.wait()
        self.__cond.release()

    def timedwait(self, msecs):
        self.__cond.acquire()
        if not self.__flag:
            self.__cond.timedwait(msecs)
        self.__cond.release()