Return to Snippet

Revision: 60942
at November 18, 2012 08:55 by ytech

Initial Code
import threading
import time

class ThM(object):
    """ThM (ThreadManager)
    Handles very simple thread operations:
        Creating single-shot threads ->
        Creating 'looping per interval' threads -> with loop set to True
        Stopping looping threads based on name -> ThM.stop_loop(...)
        Joining all threads into the calling thread ThM.joinall()
        Removing stopped threads from 'running_threads' - > ThM.free_dead()

    The class has been designed for very simple operations, mainly
    for programs that need "workers" that mindlessly loop over a function.

    NOTE: Locks,Events,Semaphores etc. have not been taken into consideration
    and may cause unexpected behaviour if used!
    running_threads = []

    def run(cls,targetfunc,thname,loop,interval,arglist=[]):
        """Statrs a new thread and appends it to the running_threads list
        along with the specified values.
        Loop and interval needs to be specified even if you dont
        want it to loop over. This is to avoid lot of keyword arguments
        and possible confusion.
        Example of starting a looping thread:

        To stop it, use:
        Note, a stopped thread cannot be started again!

        Example of a single-shot thread:

        th = threading.Thread(target=cls._thread_runner_,args=(targetfunc,thname,interval,arglist))

    def free_dead(cls):
        """Removes all threads that return FALSE on isAlive() from the running_threads list """
        for th in cls.running_threads[:]:
            if th[0].isAlive() == False:

    def stop_loop(cls,threadname):
        """Stops a looping function that was started with"""
        for i,thlis in enumerate(cls.running_threads):
            if thlis[1] == threadname:
                cls.running_threads[i][2] = False
    def joinall(cls):
        """Joins all the threads together into the calling thread."""
        for th in cls.running_threads[:]:
            while th[0].isAlive():
         #   print "Thread:",th[1],"joined","isalive:",th[0].isAlive() --- Debug stuff

    def get_all_params(cls):
        """Returns parameters from the running_threads list for external manipulation"""
        for thli in cls.running_threads:

    #This method is only intended for threads started with ThM !
    def _thread_runner_(cls,targetfunc,thname,interval,arglist):
        """Internal function handling the running and looping of the threads
        Note: threading.Event() has not been taken into consideration and neither the
        other thread managing objects (semaphores, locks, etc.)"""
        for thread in cls.running_threads[:]:
            if thname == thread[1]:
        while cls.running_threads[indx][2] == True:
            if interval != 0:

Initial URL

Initial Description
A python static class that handles simple thread management.

Initial Title
Python simple thread manager

Initial Tags

Initial Language