我刚刚注意到 Linux 上的进程终止(来自multiprocessing
库)方法的问题。我有使用multiprocessing
库的应用程序,但是......当我terminate
在 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_ip)
user_test_process = Process(target=start_test, args=(timestamp,current_test_suite,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"