Python - Shrinking pool of threads dynamically / Stop a thread

Question

I am writing a small multi-threaded http file downloader and would like to be able to shrink the available threads as the code encounters errors

The errors would be specific to http errors returned where the web server is not allowing any more connections

eg. If I setup a pool of 5 threads, each thread is attempting to open it's own connection and download a chunk of the file. The server may only allow 2 connections and will I believe return 503 errors, I want to detect this and shut down a thread, eventually limiting the size of the pool to presumably only the 2 that the server will allow

Can I make a thread stop itself?

Is self.Thread_stop() sufficient?

Do I also need to join()?

Here's my worker class that does the downloading, grabs from the queue to process, once downloaded it dumps the result into resultQ to be saved to file by the main thread

It's in here where I would like to detect a http 503 and stop/kill/remove a thread from the available pools - and of course re-add the failed chunk back to the queue so the remaining threads will process it

class Downloader(threading.Thread):
    def __init__(self, queue, resultQ, file_name):
        threading.Thread.__init__(self)
        self.workQ = queue
        self.resultQ = resultQ
        self.file_name = file_name

    def run(self):
        while True:
            block_num, url, start, length = self.workQ.get()
            print 'Starting Queue #: %s' % block_num
            print start
            print length

            #Download the file
            self.download_file(url, start, length)

            #Tell queue that this task is done
            print 'Queue #: %s finished' % block_num
            self.workQ.task_done()


    def download_file(self, url, start, length):        

        request = urllib2.Request(url, None, headers)
        if length == 0:
            return None
        request.add_header('Range', 'bytes=%d-%d' % (start, start + length))

        while 1:
            try:
                data = urllib2.urlopen(request)
            except urllib2.URLError, u:
                print "Connection did not start with", u
            else:
                break

        chunk = ''
        block_size = 1024
        remaining_blocks = length

        while remaining_blocks > 0:

            if remaining_blocks >= block_size:
                fetch_size = block_size
            else:
                fetch_size = int(remaining_blocks)

            try:
                data_block = data.read(fetch_size)
                if len(data_block) == 0:
                    print "Connection: [TESTING]: 0 sized block" + \
                        " fetched."
                if len(data_block) != fetch_size:
                    print "Connection: len(data_block) != length" + \
                        ", but continuing anyway."
                    self.run()
                    return

            except socket.timeout, s:
                print "Connection timed out with", s
                self.run()
                return

            remaining_blocks -= fetch_size
            chunk += data_block

        resultQ.put([start, chunk])

Below is where I init the thread pool, further down I put items to the queue

# create a thread pool and give them a queue
for i in range(num_threads):
    t = Downloader(workQ, resultQ, file_name)
    t.setDaemon(True)
    t.start()

Answer 1

Can I make a thread stop itself?

Don't use self._Thread__stop(). It is enough to exit the thread's run() method (you can check a flag or read a sentinel value from a queue to know when to exit).

It's in here where I would like to detect a http 503 and stop/kill/remove a thread from the available pools - and of course re-add the failed chunk back to the queue so the remaining threads will process it

You can simplify the code by separating responsibilities:

• download_file() should not try to reconnect in the infinite loop. If there is an error; let's the code that calls download_file() resubmit it if necessary
• the control about the number of concurrent connections can be encapsulated in a Semaphore object. Number of threads may differ from number of concurrent connections in this case

import concurrent.futures # on Python 2.x: pip install futures 
from threading import BoundedSemaphore

def download_file(args):
    nconcurrent.acquire(timeout=args['timeout']) # block if too many connections
    # ...
    nconcurrent.release() #NOTE: don't release it on exception,
                          #      allow the caller to handle it

# you can put it into a dictionary: server -> semaphore instead of the global
nconcurrent = BoundedSemaphore(5) # start with at most 5 concurrent connections
with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
    future_to_args = dict((executor.submit(download_file, args), args)
                           for args in generate_initial_download_tasks())

    while future_to_args:
        for future in concurrent.futures.as_completed(dict(**future_to_args)):
            args = future_to_args.pop(future)
            try: 
                result = future.result()
            except Exception as e:
                print('%r generated an exception: %s' % (args, e))
                if getattr(e, 'code') != 503:
                   # don't decrease number of concurrent connections
                   nconcurrent.release() 
                # resubmit
                args['timeout'] *= 2                    
                future_to_args[executor.submit(download_file, args)] = args
            else: # successfully downloaded `args`
                print('f%r returned %r' % (args, result))

See ThreadPoolExecutor() example.

Answer 2

you should be using a threadpool to control the life of your threads:

• http://www.inductiveload.com/posts/easy-thread-pools-in-python-with-threadpool/

Then when a thread exists, you can send a message to the main thread (that is handling the threadpool) and then change the size of the threadpool, and postpone new requests or failed requests in a stack that you'll empty.

tedelanay is absolutely right about the daemon status you're giving to your threads. There is no need to set them as daemons.

Basically, you can simplify your code, you could do something as follows:

import threadpool

def process_tasks():
    pool = threadpool.ThreadPool(4)

    requests = threadpool.makeRequests(download_file, arguments)

    for req in requests:
        pool.putRequest(req) 

    #wait for them to finish (or you could go and do something else)
    pool.wait()

if __name__ == '__main__': 
    process_tasks()

where arguments is up to your strategy. Either you give your threads a queue as argument and then empty the queue. Or you can get process the queue in process_tasks, block while the pool is full, and open a new thread when a thread is done, but the queue is not empty. It all depends on your needs and the context of your downloader.

resources:

• http://chrisarndt.de/projects/threadpool/
• http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/203871
• http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/196618
• http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/302746
• http://lethain.com/using-threadpools-in-python/