我刚刚注意到Linux上进程终止(来自多处理库)方法的问题.我有应用程序使用多处理库,但…当我在Windows上调用终止函数时,一切都很好,另一方面,Linux失败了这个解决方案.作为过程杀戮的替代,我被迫使用

os.system('kill -9 {}'.format(pid))

我知道这不是太聪明,但它确实有效.所以我只是想知道为什么这段代码在Windows上运行,但在Linux上运行失败.

例：

from multiprocessing import Process
import os
process=Process(target=foo,args=('bar',))
pid=process.pid
process.terminate() # works on Windows only
...
os.sytem('kill -9 {}'.format(pid)) # my replacements on Linux

我的配置：python 3.2.0 build 88445; Linux的2.6.32-Debian的6.0.4

这是我的代码中的示例.我希望这就足够了.

def start_test(timestamp,current_test_suite,user_ip):
    global_test_table[timestamp] = current_test_suite
    setattr(global_test_table[timestamp],"user_ip",user_ip)
    test_cases = global_test_table[timestamp].test_cases_table
    test_cases = test_cases*int(global_test_table[timestamp].count + 1)
    global_test_table[timestamp].test_cases_table = test_cases
    print(test_cases)
    print(global_test_table[timestamp].test_cases_table)
    case_num = len(test_cases)
    Report.basecounter = Report.casecounter = case_num
    setattr(global_test_table[timestamp],"case_num",case_num)
    setattr(global_test_table[timestamp],"user_current_test",0)
    try:
        dbobj=MysqLdb.connect(*dbconnector)
        dbcursor=dbobj.cursor()
        dbcursor.execute(sqlquery_insert_progress.format(progress_timestamp = str(timestamp),user_current_test = global_test_table[timestamp].user_current_tes$
    except :...
    for i in range(case_num):
        user_row = global_test_table[timestamp]
        current_test_from_tests_table = user_row.test_cases_table[i]
        unittest.TextTestRunner(verbosity=2).run(suite(CommonGUI.get_address(CommonGUI,current_test_from_tests_table[1],current_test_from_tests_table[2],user$
        global_test_table[timestamp].user_current_test = i + 1
        try:
            dbobj=MysqLdb.connect(*dbconnector)
            dbcursor=dbobj.cursor()
            dbcursor.execute(sqlquery_update_progress.format(progress_timestamp = str(timestamp),user_current_test = global_test_table[timestamp].user_current$
        except :...
@cherrypy.expose()
def start_test_page(self,**test_suite):
    timestamp = str(time.time())
    user_ip = cherrypy.request.remote.ip
    if on_server():
        sys.stdout=sys.stderr=open("/var/log/cherrypy/test_gui/{file}.log".format(file=timestamp),"a")
    current_test_suite = self.parse_result(**test_suite)
    #global_test_table[timestamp] = current_test_suite
    #setattr(global_test_table[timestamp],user_ip)
    user_test_process = Process(target=start_test,args=(timestamp,user_ip))
    users_process_table[timestamp] = user_test_process
    user_test_process.start()
    return '''{"testsuite_id" : "''' + str(timestamp) + '''"}'''
@cherrypy.expose()
def stop_test(self,timestamp):
    if timestamp in users_process_table:
        if on_server():
            user_process_pid = users_process_table[timestamp].pid
            os.system("kill -9 " + str(user_process_pid))
        else:
            users_process_table[timestamp].terminate()
        del users_process_table[timestamp]
    else:
        return "No process exists"