value based thread lock

Updated

Good news: I was able to reproduce the release_lock problem you encountered using my original answer via a somewhat crude testbed I cobbled together, and fix the issue using a counting mechanism (as you suggested) — at least a far as I can tell with my testing apparatus.

Now two separate shared dictionaries are used, one to keep track of the "names" or values associated with each lock as before, and another to keep track of how many threads are using each one at a given time.

As before, lock names must be hashable values so they can be used as keys in dictionaries.

import threading

namespace_lock = threading.Lock()
namespace = {}
counters = {}

def aquire_lock(value):
    with namespace_lock:
        if value in namespace:
            counters[value] += 1
        else:
            namespace[value] = threading.Lock()
            counters[value] = 1

    namespace[value].acquire()

def release_lock(value):
    with namespace_lock:
        if counters[value] == 1:
            del counters[value]
            lock = namespace.pop(value)
        else:
            counters[value] -= 1
            lock = namespace[value]

    lock.release()

# sample usage    
def foo(bar):
    aquire_lock(bar)
    # Code block ALPHA (two threads with equivalent bar should not be in here)
    release_lock(bar)

Have one lock, acquired whenever a thread tries to enter or exit the critical section, and use separate condition variables for each value of bar. The following could probably be optimized to create less condition variables, but doing so for this post felt like premature optimization:

import collections
import contextlib
import threading

lock = threading.Lock()

wait_tracker = collections.defaultdict(lambda: (False, 0, threading.Condition(lock)))

@contextlib.contextmanager
def critical(bar):
    with lock:
        busy, waiters, condition = wait_tracker[bar]
        if busy:
            # Someone with the same bar value is in the critical section.

            # Record that we're waiting.
            waiters += 1
            wait_tracker[bar] = busy, waiters, condition

            # Wait for our turn.
            while wait_tracker[bar][0]:
                condition.wait()

            # Record that we're not waiting any more.
            busy, waiters, condition = wait_tracker[bar]
            waiters -= 1

        # Record that we're entering the critical section.
        busy = True
        wait_tracker[bar] = busy, waiters, condition
    try:
        # Critical section runs here.
        yield
    finally:
        with lock:
            # Record that we're out of the critical section.
            busy, waiters, condition = wait_tracker[bar]
            busy = False
            if waiters:
                # Someone was waiting for us. Tell them it's their turn now.
                wait_tracker[bar] = busy, waiters, condition
                condition.notify()
            else:
                # No one was waiting for us. Clean up a bit so the wait_tracker
                # doesn't grow forever.
                del wait_tracker[bar]

Then each thread that wants to enter the critical section does the following:

with critical(bar):
    # Critical section.

This code is untested, and parallelism is hard, especially locks-and-shared-memory parallelism. I make no guarantees that it will work.