使用 f2py 在 Python GUI 中更新长时间运行的 Fortran 子例程

Question

我有一个 Python GUI (wxPython)，它使用 f2py 封装了一个 fortran“后端”。有时，fortran 进程可能会运行很长时间，我们希望在 GUI 中放置一个进度条，以通过 Fortran 例程更新进度。有什么方法可以在不涉及文件 I/O 的情况下获取 Fortran 例程的状态/进度？

Answer 1

如果您对代码的哪些部分花费最多时间有基本的了解，则可以添加进度指示器。考虑以下示例代码：

program main
implicit none
integer, parameter :: N = 100000
integer :: i
real, allocatable :: a(:), b(:)

! -- Initialization
write(*,*) 'FORFTP: Beginning routine'

allocate(a(N), b(N))
a = 0.
b = 0.

write(*,*) 'FORFTP: Completed initialization'

do i=1,N
   call RANDOM_NUMBER(a(i))
   b(i) = exp(a(i))     ! Some expensive calculation

   if (mod(i,N/100)==0) then     ! -- Assumes N is evenly divisible by 100
      write(*,*) 'FORFTP: completed ', i*100/N, ' percent of the calculation'
   endif
enddo

write(*,*) 'FORFTP: completed routine'

end program main

然后，用户将在初始化后和“昂贵的计算”的每个百分比完成后获得更新。

我不知道f2py 是如何工作的，但我认为python 有某种方法可以读取fortran 在运行时输出的内容并将其显示在它的gui 中。在此示例中，任何标记为 FORFTP 的内容都将在 gui 中输出，而我使用的是标准输出。

不过，此示例说明了调查进度的问题。与计算相比，很难理解分配需要多少时间。所以很难说初始化是总执行时间的 15%，例如。

但是，即使他们没有确切的进度表，更新正在发生的事情仍然很有用。

编辑 该例程提供以下输出：

 >  pgfortran main.f90
 >  ./a.out 
 FORFTP: Beginning routine
 FORFTP: Completed initialization
 FORFTP: completed             1  percent of the calculation
 FORFTP: completed             2  percent of the calculation
  ...
 FORFTP: completed            99  percent of the calculation
 FORFTP: completed           100  percent of the calculation
 FORFTP: completed routine

Answer 2

很有可能被否决，如果您大致知道每项任务需要多长时间，最简单的选择是根据任务开始后经过的时间确定进度，并根据任务的预期持续时间来衡量.
为了保持相关性，您始终可以存储每次任务的运行持续时间，并将其或平均值用作您的基本时间线。
有时，我们可以把事情复杂化;)

Answer 3

我找到了这样做的方法——它是否是一个好方法，我不能说，但我已经尝试过了，它确实有效。我写了elsewhere，我也会在这里发布。但是，此方法确实涉及到 Fortran 源代码，因此如果您只使用已编译的文件，它可能对您不起作用。

首先，在 Fortran 代码中，在模块中定义名为 progress 和 max_prog 的整数：

module fblock
  use iso_fortran_env
  integer(kind=int32), save :: progress = 0
  integer(kind=int32), save :: max_prog = 1
  
  contains
    subroutine long_runner(...)

这里使用了保存标志，因此变量不会超出其范围undefined（即在调用/完成子例程之前或之后访问它们，这可能会在接下来的步骤中发生）。如果您使用的是 Fortran 2008 或更高版本，则不需要保存标志，因为始终保存模块变量。

然后，在长时间运行的子程序中，添加一行告诉 f2py 解锁 python 的Global Interpreter Lock (GIL)：

module fblock
  use iso_fortran_env
  integer(kind=int32), save :: progress = 0
  integer(kind=int32), save :: max_prog = 1

  contains
    subroutine long_runner(...)
      !f2py threadsafe

解锁 GIL 可防止 Python 在此 Fortran 块运行时完全无响应，从而允许 Python 中的单独线程在其执行期间运行（这是此后的下一步；我对线程安全知之甚少很多，但是这一步对于使整个工作正常进行有点需要）。最后，只需在代码中的进度变量中添加一个：

module fblock
  use iso_fortran_env
  integer(kind=int32), save :: progress = 0
  integer(kind=int32), save :: max_prog = 1

  contains
    subroutine long_runner(input_data, output_data)
      !f2py threadsafe

      ! other code ...

      max_prog = giant_number

      ! possibly more code...

      do i = 1, giant_number
        progress = progress + 1
        ! yet more code...

您必须根据代码是否在巨大的 do 循环中运行来调整它以适应代码的运行方式，但您只是增加了一个数字。请注意，如果您使用 openmp 进行并行工作，请仅在第一个线程/处理器上汇总进度：

subroutine long_runner(input_data, output_data)
  use omp

  ! code...

  max_prog = giant_number / omp_get_num_procs()

  !$OMP PARALLEL
  proc_num = omp_get_thread_num()
  ! ...

  !$OMP DO
  do i = 1, giant_number
    ! ...
    if (proc_num == 0) then
      progress = progress + 1
    end if
    ! ...

现在，一旦您使用 f2py 将其编译到 Python 模块中，就该进行第二步并处理 Python 端了。例如，你的 Fortran 模块 'fblock' 和子例程 'long_runner' 已被编译到文件 'pyblk.pyd' 中。导入 pyblk 以及 threading 和 time 模块：

import pyblk  # your f2py compiled fortran block
import threading
import time
 
global bg_runn
bg_runn = True
 
def background():
    "background query/monitoring thread"
    time.sleep(0.1)  # wait a bit for foreground code to start
    while bg_runn:
        a = pyblk.fblock.progress
        b = pyblk.fblock.max_prog
        if a >= b: break
 
        print(a, 'of', b, '({:.3%})'.format(a / b))
        time.sleep(2)
 
# start the background thread
thrd = threading.Thread(target=background)
thrd.start()
 
print(time.ctime())
# then call the compiled fortran
output_data = pyblk.fblock.long_runner(init_data)
 
bg_runn = False  # stop the background thread once done
thrd.join()  # wait for it to stop (wait out the sleep cycle)
print(time.ctime())

在 Python 端的整个过程中，只有 progress 和 max_prog 被读取（读取不会修改）。 Fortran 块上的其他所有内容都在子程序内部，并且无论如何都没有设置任何干扰这些变量 - progess 和 max_prog 是唯一要在子程序之外查看的变量。 Python 的输出可能如下所示：

...
1026869 of 4793490 (21.422%)
1056318 of 4793490 (22.037%)
1086679 of 4793490 (22.670%)
1116830 of 4793490 (23.299%)
...

这给运行时间增加的时间可以忽略不计（如果有的话；我在测试时没有注意到时间差异）。

---

现在，要将其与带有精美进度条的 GUI 联系起来，它变得更加复杂，因为 GUI 和 Fortran 块都必须在前台运行。您不能只在后台线程中运行 Fortran（至少，我不能——Python 对我来说完全崩溃了）。因此，您必须使用 Python 的 multiprocessing 模块启动一个可以运行 Fortran 的完全独立的进程。

+===========+---------------+===========+
|  Primary  | -- Spawns --> | Secondary |
|  Process  |               |  Process  |
+===========+               +===========+
|foreground:|               |foreground:|
|    GUI    |               |  fortran  |
+- - - - - -+               +- - - - - -+
|background:|               |background:|
|           |  <-- Queue -- |   query   |
+-----------+               +-----------+                   

因此设置是将 GUI 作为主要进程，它会启动一个辅助进程，您的 Fortran 代码可以在其中使用自己的后台查询线程运行。设置了multiprocessing.Queue（将信息传递回GUI）并将其提供给multiprocessing.Process（辅助），然后启动查询线程并运行。辅助节点上的这个查询线程将其发现放入队列中，而不是像上面那样打印出来。回到主要流程，信息被拉出队列并用于设置进度条。我不熟悉 wxPython，但这里有一个使用另一个 GUI 库 PySide2 的整个复杂情况的示例：

from multiprocessing import shared_memory
import multiprocessing as mp
import numpy as np
import threading
import sys, os
import time
 
import pyblk  # your f2py compiled fortran block
 
# to use PyQt5, replace 'PySide2' with 'PyQt5'
from PySide2.QtWidgets import (QApplication, QWidget, QVBoxLayout,
                               QProgressBar, QPushButton)
 
 
global bg_runn
bg_runn = True
 
# this query is run in the background on the secondary process 
def bg_query(prog_que):
    "background query thread for compiled fortran block"
    global bg_runn
    current, total = 0, 1
 
    # wait a bit for fortran code to initialize the queried variables
    time.sleep(0.25)
     
    while bg_runn:
        # query the progress
        current = pyblk.fblock.progress
        total   = pyblk.fblock.max_prog
 
        if current >= total: break
        prog_que.put((current, total))
        time.sleep(0.1)  # this can be more or less depending on need
         
    prog_que.put((current, total))
    prog_que.put('DONE')  # inform other end that this is complete
    return
 
# this fortran block is run on the secondary process
def run_fortran(prog_que, init_data):
    "call to run compiled fortran block"
    global bg_runn
     
    # setup/start background query thread
    thrd = threading.Thread(target=bg_query, args=(prog_que, ))
    thrd.start()
     
    # call the compiled fortran code
    results = pyblk.fblock.long_runner(init_data)
     
    bg_runn = False  # inform query to stop
    thrd.join()  # wait for it to stop (wait out the sleep cycle)
     
    # now, do something with the results or
    # copy the results out from this process
    ##shm = shared_memory.SharedMemory('results')  # connect to shared mem
    ##b = np.ndarray(results.shape, dtype=results.dtype, buffer=shm.buf)
    ##b[:] = img_arr[:]  # copy results (memory is now allocated)
    ##shm.close()  # disconnect from shared mem
    return
 
 
# this GUI is run on the primary process
class ProgTest(QWidget):
    "progess test of compiled fortran code through python"
    def __init__(self, parent=None):
        super().__init__()
        # setup/layout of widget
        self.pbar = QProgressBar()
        self.pbar.setTextVisible(False)
 
        self.start_button = QPushButton('Start')
        self.start_button.clicked.connect(self.run_the_thing)
         
        ly = QVBoxLayout()
        ly.addWidget(self.start_button)
        ly.addWidget(self.pbar)
        self.setLayout(ly)
         
    def run_the_thing(self):
        "called on clicking the start button"
        self.setEnabled(False)  # prevent interaction during run
        app.processEvents()
 
        t0 = time.time()
        print('start:', time.ctime(t0))
         
        prog_que = mp.Queue()  # progress queue
 
        # if wanting the results on the primary process:
        # create shared memory to later copy result array into
        # (array size is needed; no memory is used/allocated at this time)
        ##shm = shared_memory.SharedMemory('results', create=True,
        ##                                 size=np.int32(1).nbytes * amount)
 
        init_data = None  # your initial information, if any
        # if it's large and on disk, read it in on the secondary process
 
        # setup/start the secondary process with the compiled fortran code
        run = mp.Process(target=run_fortran, args=(prog_que, init_data))
        run.start()
 
        # listen in on the query through the Queue
        while True:
            res = prog_que.get()
            if res == 'DONE': break
            current, total = res  # unpack from queue
 
            if total != self.pbar.maximum(): self.pbar.setMaximum(total)
 
            self.pbar.setValue(current)
            self.setWindowTitle('{:.3%}'.format(current / total))
            app.processEvents()
        # this while loop can be done on a separate background thread
        # but isn't done for this example
             
        run.join()  # wait for the secondary process to complete
 
        # extract the results from secondary process with SharedMemory
        # (shape and dtype need to be known)
        ##results = np.ndarray(shape, dtype=np.int32, buffer=shm.buf)
 
        t1 = time.time()
        print('end:', time.ctime(t1))
        print('{:.3f} seconds'.format(t1 - t0))
 
        self.pbar.setValue(total)
        self.setWindowTitle('Done!')
        self.setEnabled(True)
        return
 
if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = ProgTest()
    window.show()
    sys.exit(app.exec_())

但是，这种在辅助进程上运行它的方法会产生一个新问题 - 您的结果是在辅助进程上！如何处理这完全取决于您的结果。如果您可以在辅助服务器上处理它们（例如，在计算它们后保存数据），最好在那里进行。但是，如果您需要让它们回到主要进程进行交互，则必须将它们复制出来。

通常，使用 f2py 涉及numpy，因此您的结果可能是某种 numpy 数组。以下是我尝试的一些方法（使用 1600000000 字节数组）将其从辅助进程获取到主进程：

• 使用 multiprocessing.Array 创建和复制结果以将它们从辅助变为主 - 这会使整个运行时间的内存翻倍，并增加约 30 到 45 秒的运行时间。
• 将结果填充到 multiprocessing.Queue 以将它们从辅助变为主 - 填充期间内存增加一倍以上，并增加了约 5 到 10 秒的运行时间 - 不一定能很好地使用队列。李>
• 使用shared_memory.SharedMemory 复制结果以将它们从次要复制到主要-复制期间内存加倍并增加约1 秒的运行时间。这是上面 GUI 示例中注释掉的方法。当被问到这个问题时，我意识到这个选项不存在。