threadgroup-manager 0.1.2

A Python library for hierarchical thread group management with event-based coordination.

Thread Management Library

This library addresses the below requirements:

• Logically group Thread objects, with thread pooling functionality, to limit the number of concurrent threads.
• Maintaining a parent-child hierarchy of threads, addressing the limitation of Python's built-in library wherein all threads are equal.
• Allowing threads to communicate with their siblings or children through an abstracted event based system.

Threads are grouped by ThreadGroup objects. A ThreadGroup object is created in the parent thread, and tasks are added as per the use case. We can specify the maximum number of concurrent threads this group should hold, at instantiation time.

Example usage:

from threadmanager import ThreadGroup

thread_group = ThreadGroup("Verifying Users", max_concurrent_threads=10)  
# Thread group to verify users in a data structure parallely

for user in users:
    thread_group.add_task(verify_user, args=(user))
    # If number of users are greater than 10, 
    # ThreadGroup will automatically limit the number of active threads

thread_group.join(600)  #  Waits for threads to terminate with a timeout of 600 seconds.

What if we wanted just one verified user? There is no default way for a thread to be aware that this condition is met. Neither is there a mechanism to kill an active thread, as that would be unsafe.

Every time a ThreadGroup object is created there is an associated Event object that is maintained. The parent thread has this context through the ThreadGroup object, and all the threads spawned by the thread group (i.e. the child threads) are aware of this object too. This is facilitated by maintaining an n-ary tree to track the hierarchy.

If two thread group objects are created inside the main thread, the tree's state will be

         MainThread(ThreadEventTreeNode)
                    /    \
                   /      \
                  /        \
                 /          \
      ThreadGroup1         ThreadGroup2

Each node has an Event object associated with it. And each thread group would have spawned several worker threads which are logically encapsulated inside the ThreadGroup object.

         MainThread(ThreadEventTreeNode)
                    /    \
                   /      \
                  /        \
                 /          \
      ThreadGroup1         ThreadGroup2
       ~ Thread1            ~ Thread4
       ~ Thread2            ~ Thread5
       ~ Thread3            ~ Thread6

Each thread group has 3 threads. There are 2 use cases for thread communication:

• A parent thread signalling to its children.
• A child thread signalling to its siblings.

Let's consider the first case and the earlier example of User verification.

from threadmanager import event_aware_sleep

valid_users = []

def verify_user(user):
    retries = 3
    while retries > 0:
        try:
            verify_user_internal()  # internal call made to verify user
            valid_users.append(user)
        except UserVerificationException:
            logger.info("Failed to verify user, sleeping for 5 seconds and retrying")

            # At this stage, if the event is set before the next 5 seconds, we will exit the verification loop.
            if not event_aware_sleep(5):  
                break

Note that checks for the state of the event can be added in key places as required, to ensure that we are not unnecessarily performing operations, and we are safely terminating threads.

When event-aware methods are used by child threads, it automatically fetches the event associated with the thread group of the currently executing child thread. If no event is found, i.e. this is the main thread with no parent group, then event_aware_sleep will perform a regular sleep.

thread_group = ThreadGroup("Verifying Users")
for user in users:
    thread_group.add_task(verify_user, args=(user))

threadmanager cannot auto-fetch the event associated with the current (parent) thread, as it could have created several different ThreadGroups. It needs context of which ThreadGroup to perform operations on, so its usage would be through the ThreadGroup object.

thread_group.wait_for_condition(lambda: len(valid_users) > 0, timeout=600)

The parent thread will wait for the length of the list valid_users to be greater than zero, and then inform the child threads to terminate, since we need just one valid user.

thread_group.join(600)

The ThreadEvents Controller maintains the hierarchy of all the thread groups created.

def worker_method():
    # We are creating multiple thread groups inside this method, each executing respective tasks in parallel
    thread_group_2_1 = ThreadGroup()
    thread_group_2_2 = ThreadGroup()
    thread_group_2_3 = ThreadGroup()

thread_group_2 = ThreadGroup()
thread_group_2.add_task(worker_method)

         MainThread(ThreadEventTreeNode)
                    /    \
                   /      \
                  /        \
                 /          \
      ThreadGroup1         ThreadGroup2
                            /    |    \
                           /     |     \
                          /      |      \
                         /       |       \
           ThreadGroup2.1        |        ThreadGroup2.3
                                 |
                           ThreadGroup2.2

Let's say ThreadGroup2 realizes it needs to abort all operations, We can simply call

thread_group_2.set_event_for_child_tasks()

and all events for the thread group and its child thread groups will be set, effectively signalling all threads spawned as part of this set of operations.

The hierarchy is maintained and computed, completely abstracted. It doesn't require any additional consideration when creating a ThreadGroup object.

It is also possible for the n-ary tree to become a forest, in case we spawn daemon threads from a thread group, and they still need to communicate amongst each other. Information about this disjoint tree will also be maintained automatically.