From ae931b4dba60f9114398c43c5c2fa782daeb147c Mon Sep 17 00:00:00 2001 From: virusdefender <1670873886@qq.com> Date: Fri, 3 Jul 2015 16:01:53 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=86=E5=88=A4=E9=A2=98=E6=8E=A7=E5=88=B6?= =?UTF-8?q?=E5=99=A8=E7=94=B1=E5=A4=9A=E7=BA=BF=E7=A8=8B=E4=BF=AE=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=E5=A4=9A=E8=BF=9B=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 我一定是sb了,使用Python的多线程跑cpu密集型的应用。 由于Python的GIL的存在,导致在cpu上每次只能有一个线程在运行。如果一个线程运行,而且cpu时间是3秒的话,那实际运行时间将大约3秒。如果两个线程同时在进行,那总运行时间几乎要翻倍。 而换用多进程之后,单个进行运行实际时间只是稍微大于cpu时间,两个进程同时运行的时候,总的时间也是cpu时间稍微增加。 同时Python2在多进程之间运行类方法的时候存在bug,使用了 http://stackoverflow.com/questions/1816958/cant-pickle-type-instancemethod-when-using-pythons-multiprocessing-pool-ma/7309686#7309686 的方法进行patch。然后不同进程之间共享的时候,要防止循环依赖,参考 http://stackoverflow.com/questions/25382455/python-notimplementederror-pool-objects-cannot-be-passed-between-processes --- judge/client.py | 90 ++++++++++++++++++++++++++++++++++--------------- 1 file changed, 63 insertions(+), 27 deletions(-) diff --git a/judge/client.py b/judge/client.py index 8302f759..c9ff7338 100644 --- a/judge/client.py +++ b/judge/client.py @@ -2,11 +2,32 @@ import json import time import commands -from Queue import Queue -from thread_pool import ThreadPool +from multiprocessing import Pool from settings import max_running_number, lrun_gid, lrun_uid, use_tmpfs from consts import Language, Result +from copy_reg import pickle +from types import MethodType + + +# 下面两个函数告诉Python怎么pickle类实例中的方法,否则Python2会报错,是Python2的已知bug +def _pickle_method(method): + func_name = method.im_func.__name__ + obj = method.im_self + cls = method.im_class + return _unpickle_method, (func_name, obj, cls) + + +def _unpickle_method(func_name, obj, cls): + for cls in cls.mro(): + try: + func = cls.__dict__[func_name] + except KeyError: + pass + else: + break + return func.__get__(obj, cls) + class JudgeClientException(Exception): pass @@ -15,24 +36,25 @@ class JudgeClientException(Exception): class JudgeClient(object): def __init__(self, language, exec_file_path, max_cpu_time, max_real_time, max_memory, test_case_dir): - # 语言 c cpp 或者 java + """ + :param language: 语言,见consts.py + :param exec_file_path: 可执行文件路径 + :param max_cpu_time: 最大cpu时间,单位ms + :param max_real_time: 最大执行时间,单位ms + :param max_memory: 最大内存,单位MB + :param test_case_dir: 测试用户文件夹路径 + :return:返回结果list + """ self.language = language - # 可执行文件路径,比如 /root/1/a.out /root/1/Main.class self.exec_file_path = exec_file_path - # 最大的cpu时间 单位ms self.max_cpu_time = max_cpu_time - # 最大实际运行时间 单位ms self.max_real_time = max_real_time - # 最大cpu占用,注意不要小于500000 单位byte self.max_memory = max_memory - # 测试用例文件路径,比如/root/testcase/1/ self.test_case_dir = test_case_dir - # 判题结果队列 - self.result_queue = Queue() - # 线程池 - self.thread_pool = ThreadPool(size=max_running_number, - result_queue=self.result_queue) - self.thread_pool.start() + # 进程池 + self.pool = Pool(processes=max_running_number) + # 结果数组 + self.results = [] # 测试用例配置项 self.test_case_info = self.load_test_case_info() @@ -67,9 +89,9 @@ class JudgeClient(object): # todo 系统调用白名单 chroot等参数 # fixme 时间的单位问题 command = "lrun" + \ - " --max-cpu-time " + str(self.max_cpu_time) + \ - " --max-real-time " + str(self.max_real_time) + \ - " --max-memory " + str(self.max_memory) + \ + " --max-cpu-time " + str(self.max_cpu_time / 1000.0) + \ + " --max-real-time " + str(self.max_real_time / 1000.0) + \ + " --max-memory " + str(self.max_memory * 1000 * 1000) + \ " --network false" + \ " --uid " + str(lrun_uid) + \ " --gid " + str(lrun_gid) @@ -98,6 +120,8 @@ class JudgeClient(object): "CPU_TIME": "cpu_time", "REALTIME": "real_time", "REAL_TIME": "real_time", + "TERMSIG": "term_sig", + "SIGNALED": "siginaled", "EXITCODE": "exit_code", "EXCEED": "exceed"} for line in lines: @@ -112,6 +136,10 @@ class JudgeClient(object): result[translate[name]] = float(value) * 1000 elif name == "EXITCODE": result[translate[name]] = int(value) + elif name == "TERMSIG": + result[translate[name]] = int(value) + elif name == "SIGNALED": + result[translate[name]] = int(value) elif name == "EXCEED": if value == "none": result[translate[name]] = None @@ -130,7 +158,7 @@ class JudgeClient(object): run_result["test_case_id"] = test_case_id # 如果返回值非0,代表非正常结束 - if run_result["exit_code"]: + if run_result["exit_code"] or run_result["term_sig"] or run_result["siginaled"]: run_result["result"] = Result.RUNTIME_ERROR return run_result @@ -148,22 +176,30 @@ class JudgeClient(object): run_result["result"] = Result.ACCEPTED return run_result + def collect_result(self, result): + self.results.append(result) + def run(self): # 添加到任务队列 for i in range(self.test_case_info["test_case_number"]): - self.thread_pool.append_job(self.judge_one, i + 1) + self.pool.apply_async(self.judge_one, args=(i + 1, ), + callback=self.collect_result) + self.pool.close() + self.pool.join() + print self.results - self.thread_pool.join() - self.thread_pool.stop() - - for i in range(self.test_case_info["test_case_number"]): - print self.result_queue.get(block=False) + def __getstate__(self): + # 不同的pool之间进行pickle的时候要排除自己,否则报错 + self_dict = self.__dict__.copy() + del self_dict['pool'] + return self_dict +pickle(MethodType, _pickle_method, _unpickle_method) client = JudgeClient(language=Language.C, exec_file_path="/var/judge/a.out", - max_cpu_time=1000, - max_real_time=2000, - max_memory=600000, + max_cpu_time=1000000, + max_real_time=200000, + max_memory=1, test_case_dir="/var/test_case/1/") client.run()