python multiprocessing库使用记录

python multiprocessing库使用记录

需求是想并行调用形式化分析工具proverif,同时发起对多个query的分析(378个)。实验室有40核心80线程的服务器(双cpu,至强gold 5218R*2)。

观察到单个命令在分析时内存占用不大,且只使用单核心执行,因此考虑同时调用多个命令同时执行分析,加快结果输出。

最底层的逻辑是调用多个命令行语句,和在命令行直接执行proverif语句类似。在python中也就是使用 os.system()函数实现命令调用。然而由于存在如下问题,需要考虑使用多进程multiprocessing库。

  • 如果使用多线程threading库,由于GIL的存在,是否会因为一个进程未执行结束而无法发起新的进程?
  • query数量很大的原因来自于多场景分析,同时对于同一场景下的query也希望可以并行推进,同时分析。
  • query数量大+场景多,得到很多结果,每条分析语句都有各自不同的位置,需要生成大量的命令。
  • 每条query执行完成后会给出分析结果。虽然分析结果会以html文件的形式输出到指定结果文件夹,但是不能对分析结果做统一的分析,仍旧需要逐个阅读。希望能在输出后即时统计,原有输出不变,还能给出分析结果表。
  • 尽管proverif在分析上速度已经很好了,但是仍然有62条query在30000秒(8.3h)后未给出结果。希望能够统计每一条query的运行时间并记录,并能够提供当前仍在执行的query数量。
    • 进一步的,设置最高分析时长上限(如48h),若超出上限则终止分析。
    • 对于一些可达性查询(reachability,实现方法是:在实体执行最后,在公开信道上发送执行完成标记,检查攻击者是否检验实体代码是否正确,以及攻击者是否能够阻止合法实体正常执行程序(如何做?)),会出现构建攻击路径很慢的情况。但是实际上已经给出了goal reachable的结果。对于这种其实无需浪费更多时间,可以把reachability的query添加 set reconstructTrace = false .以提前结束。
    • 对于数量监控,需要多进程读写共享变量;对于运行时间记录,需要多进程读写同一个文件。

mutliprocessing库使用

主要使用multiprocessing.Pool()来创建进程池,当前python进程会创建新的python进程用于执行函数。(win下是子进程,linux下是fork)

由于存在操作系统上的差异,请使用if __name__ == '__main__':来编写主函数,否则可能出现问题。主函数内容如下。

query_num = multiprocessing.Value('i', 0)

def long_time_task(c, ):
    start = time.time()
    os.system(c)
    end = time.time()
    # task_name=...
    with query_num.get_lock():
        query_num.value -= 1
        print('Task %s runs %0.2f seconds. ' % (task_name, (end - start)) + str(query_num.value) + ' left.')
    return 'Task %s runs %0.2f seconds.' % (task_name, (end - start))


def call_back(s):
    with open('/home/dell/proverif/DDS/time.txt', "a+") as file:
        file.writelines(s + '\n')
        

if __name__ == '__main__':
    query_list = extract(path_query, 'query', '.')
    query_file_path_list = query_file(query_list)
    whole_cS = compromise_Scenarios(path_compromise, path_process_whole, work_path)
    MAC_cS = compromise_Scenarios(path_compromise, path_process_MAC, work_path)
    cmd = []
    cmd += (pv_cmd(query_file_path_list, whole_cS, path_result))
    cmd += (pv_cmd(query_file_path_list, MAC_cS, path_result))
    p = Pool(len(cmd))
    query_num.value = len(cmd)
    # for i in cmd:
    #     p.apply_async(long_time_task, args=(i,), callback=call_back)

    results = [p.apply_async(long_time_task, (i,), callback=call_back) for i in cmd]

    print('Waiting for all subprocesses done...')
    output = [result.get(timeout=24*60*60) for result in results]
    # p.close()
    # p.join()
    print('All subprocesses done.')
    

主函数前7行为文本处理,其内容不细表。

第8行p = Pool(len(cmd))创建了进程池,其长度为cmd的个数,也就是我们要同时发起这么多个进程。接下来注释掉的循环是常规的多进程发起办法,即使用apply_async函数执行我们要的函数。args是long_time_task的参数,由于需要为Iterable且只有一个参数,因此以元组形式传入。

call_back参数为回调函数,这里很像go语言下的defer,会在函数执行后再执行。回调函数接受long_time_task的返回值作为参数,我们使用这个机制实现多进程写文件。long_time_task在返回后会受到进程池p的调度,依次执行写文件操作,因此避免了同时写引起冲突。

对于剩余的query数量,使用全局变量query_num = multiprocessing.Value('i', 0),这样的变量具有锁,可以供多进程读写。每个query在完成后会将数量减一,输出时间和剩余数量。使用with query_num.get_lock():获得锁,避免读写冲突,并在使用完成后自动释放。

这已经满足了基本需求。还有一个定时终止的功能有待实现。接下来再介绍我不断修改的思路。

多进程定时终止

单进程定时终止

process = multiprocessing.Process(target=long_time_task)

# 启动进程
process.start()

# 设置运行时长上限(48小时)
timeout = 48 * 60 * 60  # 以秒为单位

# 创建定时器,在指定时间后终止进程
timer = multiprocessing.Timer(timeout, process.terminate)
timer.start()

# 等待进程结束
process.join()

使用定时器的办法,在一定时间后调用我们创建进程的process.terminate()方法结束进程。但我们需要多进程并行。

多进程定时终止

pool = multiprocessing.Pool()

# 准备要执行函数的参数列表
inputs = [1, 2, 3, 4, 5]

# 执行函数,并设置最大运行时长为30秒
result = pool.map_async(long_time_task, inputs)

# 获取结果,最多等待30秒
output = result.get(timeout=48 * 60 * 60)

map_async方法可以将函数应用于可迭代的参数列表,并返回一个AsyncResult对象,可以使用该对象的get方法获取结果。map_async方法将任务提交给进程池后会立即返回,并不会等待所有任务执行完成。如果在get方法获取结果时,其中某些任务仍在执行,将会等待直到超时。get方法拥有timeout参数,超时后会raise TimeoutError,报错终止python程序的运行。因此如果想输出已完成的结果,有两个思路:

  1. try-except捕获TimeoutError,并针对处理。
  2. 对每个结果都使用get方法并设置超时时间。

列表推导式

results = [p.apply_async(long_time_task, (i,), callback=call_back) for i in cmd]

print('Waiting for all subprocesses done...')
output = [result.get(timeout=24*60*60) for result in results]
# p.close()
# p.join()
print('All subprocesses done.')

使用apply_async方法来执行函数,该方法会也会返回一个AsyncResult对象。我们将这些对象放入results数组,接着使用数组中每个元素的结果组成output数组并定义超时时间。这样就可以执行call_back函数了。output内容其实不是很重要,主要是为了使用AsyncResult对象的get方法来设置定时器。

不过这样还是需要try-except捕获TimeoutError,以处理超时未完成的query。这样做比map_async好在哪里?我在使用的时候map_async似乎不能成功调用回调函数,还有待试验。此外,该方法并不能在设定时间时准时停下,例如我设置时间5s,则会在约12秒时才停止。

还有一个问题是,在pycharm里运行脚本时,会有部分进程无法结束。暂不清楚其原因,也不确定命令行下执行脚本是否存在同样的问题。

与Go相比

显著的感觉到python在处理多进程、多线程、并发等问题上有一定的弱点。虽然能够通过一系列操作实现,但是做起来比较吃力,也不算太优雅。现在的脚本已经可以并行分析了。然而在任务管理器中,除了看到了378个proverif进程,还看到了378个sh和378个python🤣

这378个python其实是没必要的,如果使用goroutine,发起多个协程执行shell命令即可。这样在开销和效率上都会更好。不过其实总的内存占用并不太高,所以这个点不算非常大的问题。但是如果使用更复杂的分析工具,mutliprocessing多进程调用就太笨拙了。

此外,在共享变量的访问上也不那么容易。例如query_num这个共享变量,多进程的访问就不是很方便,如果使用goroutine,则可以使用channel,创建一个monitor goroutine来接受各个goroutine的返回值,并做计数处理;文件读写也可以在channel+监视器内完成,无需考虑文件读写争用,或者也可以用mutex来互斥地读写文件。

在计时器的操作上,目前只了解到python的解决办法是抛出超时异常。这种方法会使得没执行完毕的脚本无法正常返回,不能给出信息。需要根据执行结果做一些特定的异常处理。也可能这是会有部分进程无法正常结束的原因。go语言有Timer类可以执行计时器操作,有望更优雅的解决问题。

后续计划

后续会考虑使用go语言编写一个专用于proverif的多进程并发分析调用工具,使用更优雅、效率更高的方法,实现:

  • 多进程并发调用
  • query分析时长记录、分析结果导出
  • 总分析时长上限设定,超时后正常退出,并标记超时query。

热门相关:有个人爱你很久   民国之文豪崛起   回眸医笑,冷王的神秘嫡妃   回眸医笑,冷王的神秘嫡妃   修真界败类