Multi-threaded python (mp.Pool) server with task queue
<p>So, I am writing a free python task server for Autodesk Maya that holds a queue of x number of 'workers'. At any time the server can accept a 'task' and toss that task on the queue that the workers churn through.</p>

<p>From the queue each worker gets a 'taskDict' which is a dictionary sent to the server that says where the Maya file is, and what code to run when we open a headless Maya application (mayapy.exe/standalone)</p>

<p>I have rewritten this many times, first using my own queue system, but then i decided to use python's. Next using a pool, using Queue.Queue, using mp.Manager.Queue and pool, etc.  I have a hard time finding any examples of a simple multithreaded server that receives information and kicks off a thread, but uses a queue for when it gets too many requests.</p>

<p>I just fundamentally do not understand how to place information in a queue, and have an mp.pool churn through the queue, kicking off apply_async processes that use that data and telling the queue when it's complete.</p>

<p>Here's the current state of the code:</p>

<pre><code>import tempfile
import os
import subprocess
import threading
import multiprocessing as mp
import socket
import sys

from PySide import QtGui, QtCore

import serverUtils

selfDirectory = os.path.dirname(__file__)
uiFile = selfDirectory + '/server.ui'
if os.path.isfile(uiFile):
    form_class, base_class = serverUtils.loadUiType(uiFile)
else:
    print('Cannot find UI file: ' + uiFile)


def show():
    global mayaTaskServerWindow
    try:
        mayaTaskServerWindow.close()
    except:
        pass

        mayaTaskServerWindow = mayaTaskServer()
        mayaTaskServerWindow.show()
    return mayaTaskServerWindow

class MayaTaskServer(base_class, form_class):
    refreshSignal = QtCore.Signal()

    def __init__(self):
        super(MayaTaskServer, self).__init__()

        self.setupUi(self)

        self.mainJobServer = None
        self.mpPool = None
        self.manager = None
        self.q = None

        self.workerDict = {}

        self.refreshSignal.connect(self.refreshTree)
        self.startLocalCoresBTN.clicked.connect(self.startLocalCoresFn)
        self.killLocalCoresBTN.clicked.connect(self.killLocalCoresFn)
        self.jobTree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
        self.jobTree.customContextMenuRequested.connect(self.openMenu)

        self.startJobServer(6006)
        self.startQueue()

        # set the default temp folder
        filepath = os.path.realpath(__file__)
        self.localTempFolderEDT.setText(filepath.replace(filepath.split('\\')[-1], ''))

    ## JOB SYSTEM
    ####################################################################

    class MayaWorker(object):
        def __init__(self, host, port, cpuID):
            self.host = host
            self.port = port
            self.location = None
            self.cpuID = cpuID

            self.location = self.host

            self.busy = False
            self.task = None
            self.taskHistory = {}

        def runTask(self, task):
            print 'starting task - ', self.task['description']
            self.busy = True
            serverUtils.spawnMaya(task)
            win.refreshSignal.emit()

        def taskComplete(self, arg):
            self.busy = False
            self.task = None
            self.mayaFile = None
            win.refreshSignal.emit()

    def bootUpLocalWorkers(self, numProcs):
        self.mpPool = mp.Pool(processes=numProcs)
        for i in range(0, numProcs):
            mw = self.MayaWorker('localhost', 6006, i)
            win.mpPool.apply_async(mw, args=(win.q))
            win.workerDict['CPU_' + str(i)] = mw

    ## USER INTERFACE
    ####################################################################

    #UI code here you don't care about

    ## JOB LISTENER / SERVER / QUEUE
    ####################################################################
    class JobServer(threading.Thread):
        def __init__(self, port):
            threading.Thread.__init__(self)
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.bind(('localhost', port))
            self.server_socket.listen(5)

            self.port = port
            self.running = True

            self.mpPool = None

        def addToQueue(self, task):
            #add to queue
            win.q.put(task, timeout=1000)

            #update UI
            wid1 = QtGui.QTreeWidgetItem()
            wid1.setText(0, str(task))
            win.queTree.addTopLevelItem(wid1)

        def run(self, debug=1):
            print 'Starting Task Server @' + socket.gethostbyname(socket.gethostname()) + ':' + str(self.port)
            while self.running:
                client_socket, address = self.server_socket.accept()
                ip = str(address[0])
                data = client_socket.recv(512)
                if 'runTask' in data:
                    taskDict = eval(data.split(' &gt;&gt; ')[-1])
                    print 'SERVER&gt;&gt; Received task:', str(taskDict)
                    self.addToQueue(taskDict)

    class TaskQueueServer(threading.Thread):
        def __init__(self):
            q = self.q_in
            while True:
                if self.q_in:
                    worker = win.findLocalWorker()
                    if worker:
                        taskDict = self.q_in[0]
                        worker.task = taskDict
                        worker.startTask()
                        self.q_in.pop[0]


    def startJobServer(self, port):
        self.mainJobServer = self.JobServer(port)
        self.mainJobServer.start()

    def startQueue(self):
        self.manager = mp.Manager()
        self.q = self.manager.Queue()


if __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)
    win = MayaTaskServer()
    win.show()
    sys.exit(app.exec_())
</code></pre>

<p>So here's how I did it. A very simple, pragmatic solution.</p>

<p>I have a method called 'finaLocalWorker', you can see the worker class can get marked as 'busy'. If a worker isn't busy, an incoming task is sent to it.</p>

<p>If all workers are busy, then an incoming task gets added to a simple list called 'self.q'.</p>

<p>When a worker finishes a task, mpPool.apply_async has a callback that fires the taskComplete method.  This method says 'if self.q, take the [0] item of the list and pop (remove) it. Else mark myself as not busy'.</p>

<p>This allows for overflowing incoming requests like a batch of 500 animations to be queued up in the task list, but also the server still is able to receive no tasks for some time and immediately work on any task that comes in.</p>

<p>I will put the final code up on github.</p>

python multithreading server queue pool