Path: csiph.com!usenet.pasdenom.info!weretis.net!feeder1.news.weretis.net!feeder.erje.net!newsfeed.xs4all.nl!newsfeed6.news.xs4all.nl!xs4all!news2.euro.net!newsgate.cistron.nl!newsgate.news.xs4all.nl!post.news.xs4all.nl!not-for-mail
MIME-Version: 1.0
In-Reply-To: <OFB845143C.E36D6F51-ON85257AA1.0066C7C0-85257AA1.0067DD35@us.ibm.com>
References: <OFB845143C.E36D6F51-ON85257AA1.0066C7C0-85257AA1.0067DD35@us.ibm.com>
From: Ian Kelly <ian.g.kelly@gmail.com>
Date: Wed, 24 Oct 2012 14:19:28 -0600
Subject: Re: A lock that prioritizes acquire()s?
To: Python <python-list@python.org>
Content-Type: text/plain; charset=ISO-8859-1
Precedence: list
Newsgroups: comp.lang.python
Message-ID: <mailman.2794.1351110001.27098.python-list@python.org>
Lines: 88
NNTP-Posting-Host: 2001:888:2000:d::a6
Xref: csiph.com comp.lang.python:32065

On Wed, Oct 24, 2012 at 12:54 PM, David M Chess <chess@us.ibm.com> wrote:
> Seeking any thoughts on other/better ways to do this, or whether the
> inefficiency will be too eyerolling if we get say one request per second
> with an average service time a bit under a second but maximum service time
> well over a second, and most of them are importance zero, but every (many)
> seconds there will be one or two with higher importance.

I used a PriorityQueue and Conditions to get rid of the ugly while True loop.


import threading
from Queue import PriorityQueue, Empty

class PriorityLock(object):

    def __init__(self):
        self._is_available = True
        self._mutex = threading.Lock()
        self._waiter_queue = PriorityQueue()

    def acquire(self, priority=0):
        self._mutex.acquire()
        # First, just check the lock.
        if self._is_available:
            self._is_available = False
            self._mutex.release()
            return True
        condition = threading.Condition()
        condition.acquire()
        self._waiter_queue.put((priority, condition))
        self._mutex.release()
        condition.wait()
        condition.release()
        return True

    def release(self):
        self._mutex.acquire()
        # Notify the next thread in line, if any.
        try:
            _, condition = self._waiter_queue.get_nowait()
        except Empty:
            self._is_available = True
        else:
            condition.acquire()
            condition.notify()
            condition.release()
        self._mutex.release()

def test():
    import random, time

    def thread(lock, priority):
        lock.acquire(priority)
        print("Thread %d running" % priority)
        time.sleep(1)
        lock.release()
    lock = PriorityLock()
    threads = [threading.Thread(target=thread, args=(lock, x)) for x
in range(10)]
    random.shuffle(threads)
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    test()


Output:

Thread 9 running
Thread 0 running
Thread 1 running
Thread 2 running
Thread 3 running
Thread 4 running
Thread 5 running
Thread 6 running
Thread 7 running
Thread 8 running

Note that with the PriorityQueue, lower priority values are retrieved
first.  Thread 9 ran first just by virtue of being first to the gate,
and after that you can see that everything went in order.

Cheers,
Ian