0

我有一堆长时间运行的进程，我想将它们分成多个进程。那部分我可以做没有问题。我遇到的问题是有时这些进程会进入挂起状态。为了解决这个问题，我希望能够为进程正在处理的每个任务设置时间阈值。当超过该时间阈值时，我想重新启动或终止任务。

最初我的代码使用进程池非常简单，但是使用池我无法弄清楚如何检索池内的进程，更不用说如何重新启动/终止池中的进程。

我已使用队列和进程对象，如本示例所示（https://pymotw.com/2/multiprocessing/communication.html#passing-messages-to-processes进行了一些更改。

我试图弄清楚这一点在下面的代码中。在其当前状态下，进程实际上并未终止。除此之外，我无法弄清楚在当前任务终止后如何让进程转移到下一个任务。任何建议/帮助表示赞赏，也许我会以错误的方式解决这个问题。

谢谢

    import multiprocess
    import time

    class Consumer(multiprocess.Process):
        def __init__(self, task_queue, result_queue, startTimes, name=None):
            multiprocess.Process.__init__(self)
            if name:
                self.name = name
            print 'created process: {0}'.format(self.name)
            self.task_queue = task_queue
            self.result_queue = result_queue
            self.startTimes = startTimes

        def stopProcess(self):
            elapseTime = time.time() - self.startTimes[self.name]
            print 'killing process {0} {1}'.format(self.name, elapseTime)
            self.task_queue.cancel_join_thread()
            self.terminate()
            # now want to get the process to start procesing another job

        def run(self):
            '''
            The process subclass calls this on a separate process.
            '''    
            proc_name = self.name
            print proc_name
            while True:
                # pulling the next task off the queue and starting it
                # on the current process.
                task = self.task_queue.get()
                self.task_queue.cancel_join_thread()

                if task is None:
                    # Poison pill means shutdown
                    #print '%s: Exiting' % proc_name
                    self.task_queue.task_done()
                    break
                self.startTimes[proc_name] = time.time()
                answer = task()
                self.task_queue.task_done()
                self.result_queue.put(answer)
            return

    class Task(object):
        def __init__(self, a, b, startTimes):
            self.a = a
            self.b = b
            self.startTimes = startTimes
            self.taskName = 'taskName_{0}_{1}'.format(self.a, self.b)

        def __call__(self):
            import time
            import os

            print 'new job in process pid:', os.getpid(), self.taskName

            if self.a == 2:
                time.sleep(20000) # simulate a hung process
            else:
                time.sleep(3) # pretend to take some time to do the work
            return '%s * %s = %s' % (self.a, self.b, self.a * self.b)

        def __str__(self):
            return '%s * %s' % (self.a, self.b)

    if __name__ == '__main__':
        # Establish communication queues
        # tasks = this is the work queue and results is for results or completed work
        tasks = multiprocess.JoinableQueue()
        results = multiprocess.Queue()

        #parentPipe, childPipe = multiprocess.Pipe(duplex=True)
        mgr = multiprocess.Manager()
        startTimes = mgr.dict()

        # Start consumers
        numberOfProcesses = 4
        processObjs = []
        for processNumber in range(numberOfProcesses):
            processObj = Consumer(tasks, results, startTimes)
            processObjs.append(processObj)

        for process in processObjs:
            process.start()

        # Enqueue jobs
        num_jobs = 30
        for i in range(num_jobs):
            tasks.put(Task(i, i + 1, startTimes))

        # Add a poison pill for each process object
        for i in range(numberOfProcesses):
            tasks.put(None)

        # process monitor loop, 
        killProcesses = {}
        executing = True
        while executing:
            allDead = True
            for process in processObjs:
                name = process.name
                #status = consumer.status.getStatusString()
                status = process.is_alive()
                pid = process.ident
                elapsedTime = 0
                if name in startTimes:
                    elapsedTime = time.time() - startTimes[name]
                if elapsedTime > 10:
                    process.stopProcess()

                print "{0} - {1} - {2} - {3}".format(name, status, pid, elapsedTime)
                if  allDead and status:
                    allDead = False
            if allDead:
                executing = False
            time.sleep(3)

        # Wait for all of the tasks to finish
        #tasks.join()

        # Start printing results
        while num_jobs:
            result = results.get()
            print 'Result:', result
            num_jobs -= 1
4

2 回答 2

2

我通常建议不要子类化multiprocessing.Process，因为它会导致代码难以阅读。

我宁愿将您的逻辑封装在一个函数中并在一个单独的进程中运行它。这使代码更加简洁和直观。

不过，与其重新发明轮子，我建议您使用一些已经为您解决问题的库，例如Pebble或billiard。

例如，Pebble库允许轻松为独立运行或在Pool.

在具有超时的单独进程中运行您的函数：

from pebble import concurrent
from concurrent.futures import TimeoutError

@concurrent.process(timeout=10)
def function(foo, bar=0):
    return foo + bar

future = function(1, bar=2)

try:
    result = future.result()  # blocks until results are ready
except TimeoutError as error:
    print("Function took longer than %d seconds" % error.args[1])

相同的示例，但有一个进程池。

with ProcessPool(max_workers=5, max_tasks=10) as pool:
   future = pool.schedule(function, args=[1], timeout=10)

   try:
       result = future.result()  # blocks until results are ready
    except TimeoutError as error:
        print("Function took longer than %d seconds" % error.args[1])

在这两种情况下，超时过程都会为您自动终止。

于 2017-08-02T08:16:08.077 回答
2

一种更简单的解决方案是继续使用 a 而不是重新实现 aPool是设计一种使您正在运行的功能超时的机制。例如：

from time import sleep
import signal

class TimeoutError(Exception):
    pass    

def handler(signum, frame):
    raise TimeoutError()

def run_with_timeout(func, *args, timeout=10, **kwargs):
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(timeout)
    try:
        res = func(*args, **kwargs)
    except TimeoutError as exc:
        print("Timeout")
        res = exc
    finally:
        signal.alarm(0)
    return res


def test():
    sleep(4)
    print("ok")

if __name__ == "__main__":
    import multiprocessing as mp

    p = mp.Pool()
    print(p.apply_async(run_with_timeout, args=(test,),
                        kwds={"timeout":1}).get())

设置一个超时，当这个signal.alarm超时时，它运行处理程序，停止你的函数的执行。

编辑：如果您使用的是 Windows 系统，它似乎有点复杂，因为signal没有实现SIGALRM. 另一种解决方案是使用 C 级 python API。此代码已根据此SO 答案进行了改编，并针对 64 位系统进行了一些修改。我只在linux上测试过它，但它应该在windows上工作。

import threading
import ctypes
from time import sleep


class TimeoutError(Exception):
    pass


def run_with_timeout(func, *args, timeout=10, **kwargs):
    interupt_tid = int(threading.get_ident())

    def interupt_thread():
        # Call the low level C python api using ctypes. tid must be converted 
        # to c_long to be valid.
        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
            ctypes.c_long(interupt_tid), ctypes.py_object(TimeoutError))
        if res == 0:
            print(threading.enumerate())
            print(interupt_tid)
            raise ValueError("invalid thread id")
        elif res != 1:
            # "if it returns a number greater than one, you're in trouble,
            # and you should call it again with exc=NULL to revert the effect"
            ctypes.pythonapi.PyThreadState_SetAsyncExc(
                ctypes.c_long(interupt_tid), 0)
            raise SystemError("PyThreadState_SetAsyncExc failed")

    timer = threading.Timer(timeout, interupt_thread)
    try:
        timer.start()
        res = func(*args, **kwargs)
    except TimeoutError as exc:
        print("Timeout")
        res = exc
    else:
        timer.cancel()
    return res


def test():
    sleep(4)
    print("ok")


if __name__ == "__main__":
    import multiprocessing as mp

    p = mp.Pool()
    print(p.apply_async(run_with_timeout, args=(test,),
                        kwds={"timeout": 1}).get())
    print(p.apply_async(run_with_timeout, args=(test,),
                        kwds={"timeout": 5}).get())
于 2017-08-02T08:38:36.573 回答