1

Hello and thanks for your help :)

The program I was writing basically disassembles executables. I just wanted to see if I could make it faster by using pathos. The problem is that it does not run reliably. I'll explain in a second what I mean by reliable.

The program is run like this:

from ControlFlow import Disassembler, DisassemblerWorker
from ControlFlow import FlowGraph
import networkx as nx
import time

file_path = "/vagrant/SimpleTestBinaries/example3-x64"
start = time.time()
flow = Disassembler(file_path)
graph = FlowGraph(flow)
end = time.time()

print("Finished in: ", end - start, " seconds")

Normally it would reply with:

Finished in: 0.8992343274389473 seconds

But sometimes it just seems like it is stuck. After all, as you can see above it should take less than a second. So I proceed to kill it and it gives me a bunch of errors, which maybe hints where it's getting stuck.

Process ForkPoolWorker-11:
Process ForkPoolWorker-13:
Process ForkPoolWorker-10:
Traceback (most recent call last):
Traceback (most recent call last):
Process ForkPoolWorker-12:
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 258, in _bootstrap
    self.run()
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 258, in _bootstrap
    self.run()
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/pool.py", line 108, in worker
    task = get()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/queues.py", line 337, in get
    with self._rlock:
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/pool.py", line 108, in worker
    task = get()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/synchronize.py", line 101, in __enter__
    return self._semlock.__enter__()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/pool.py", line 108, in worker
    task = get()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/pool.py", line 108, in worker
    task = get()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/queues.py", line 337, in get
    with self._rlock:
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/synchronize.py", line 101, in __enter__
    return self._semlock.__enter__()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/queues.py", line 337, in get
    with self._rlock:
KeyboardInterrupt
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/queues.py", line 338, in get
    res = self._reader.recv_bytes()
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/synchronize.py", line 101, in __enter__
    return self._semlock.__enter__()
KeyboardInterrupt
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/connection.py", line 219, in recv_bytes
    buf = self._recv_bytes(maxlength)
KeyboardInterrupt
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/connection.py", line 410, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/lib/python3.6/dist-packages/multiprocess/connection.py", line 382, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt
---------------------------------------------------------------------------
KeyboardInterrupt                         Traceback (most recent call last)
 in 
      6 file_path = "/vagrant/SimpleTestBinaries/example3-x64"
      7 start = time.time()
----> 8 flow = Disassembler(file_path)
      9 graph = FlowGraph(flow)
     10 end = time.time()

/vagrant/BinaryResearch/ControlFlow.py in __init__(self, path)
     34         self.regs_write_map = {}
     35         self.section_map = {}
---> 36         self._analyze_flow()
     37 
     38     def disassembler_setup(self, architecture, details=True):

/vagrant/BinaryResearch/ControlFlow.py in _analyze_flow(self)
     77             jumps = p.amap(worker.get_jump_map, imagebase)
     78             returns = p.amap(worker.get_return_map, imagebase)
---> 79             p.close(), p.join()
     80 
     81             call_results, jump_results, return_results = calls.get()[0], jumps.get()[0], returns.get()[0]

/usr/local/lib/python3.6/dist-packages/pathos/multiprocessing.py in join(self)
    206         _pool = __STATE.get(self._id, None)
    207         if _pool and self.__nodes == _pool.__nodes:
--> 208             _pool.join()
    209         return
    210     # interface

/usr/local/lib/python3.6/dist-packages/multiprocess/pool.py in join(self)
    544         util.debug('joining pool')
    545         assert self._state in (CLOSE, TERMINATE)
--> 546         self._worker_handler.join()
    547         self._task_handler.join()
    548         self._result_handler.join()

/usr/lib/python3.6/threading.py in join(self, timeout)
   1054 
   1055         if timeout is None:
-> 1056             self._wait_for_tstate_lock()
   1057         else:
   1058             # the behavior of a negative timeout isn't documented, but

/usr/lib/python3.6/threading.py in _wait_for_tstate_lock(self, block, timeout)
   1070         if lock is None:  # already determined that the C code is done
   1071             assert self._is_stopped
-> 1072         elif lock.acquire(block, timeout):
   1073             lock.release()
   1074             self._stop()

KeyboardInterrupt:

So, I went to check the portion of code that it is referencing. I don't know if this means that it is getting stuck somewhere between the p.close() and p.join(). This is the snippet within ControlFlow.py that it points to.

from pathos.multiprocessing import ProcessPool

# More code ...

p = ProcessPool()

for section in available_sections:
    worker = DisassemblerWorker(self.architecture, section.content, section.virtual_address)

    p.clear()
    calls = p.amap(worker.get_call_map, imagebase)
    jumps = p.amap(worker.get_jump_map, imagebase)
    returns = p.amap(worker.get_return_map, imagebase)
    p.close(), p.join()

    call_results, jump_results, return_results = calls.get()[0], jumps.get()[0], returns.get()[0]

    # More code ...

So I really don't know what it is causing it to be unreliable. I know this sounds crazy but once the program has it's first "success" it seems to run fine afterwards. Also, I should say that I am running this in a Jupyter Notebook. I've read that multiprocessing is incompatible with notebooks so I'm using multiprocess instead.

Any ideas as to what is going on?

Thanks once again!

Xavier Merino
  • 679
  • 1
  • 5
  • 19

1 Answers1

0

I have had the same issue with the underlying python multiprocessing library (Pool): works great many times, and sometimes seems to get stuck.

The documentation points at the requirement to have the method you call with pool outside the call multiprocessing it. https://docs.python.org/2/library/multiprocessing.html

In practice, if seems you need to define your method outside the if main. I don't know why the behaviour appears inconsistent, but following the guideline made it work in some cases for me.

I found yet other cases that didn't work, and the ultimate solution seems to be to write the methods you parallelise in a separate file. I found this last bit in the question below, and that solved it for me: Jupyter notebook never finishes processing using multiprocessing (Python 3)

Below is an example of having your function f in a separate file, and outside the main.

**file 1: my_methods.py**
  def f(x):
  return x.count()



**file 2: main.py or your jupyter notebook, in the same directory here**

import multiprocessing as mp
from my_methods import f

def parallelize(df, func, n_cores=4):
    df_split = np.array_split(df, n_cores)
    pool = mp.Pool(n_cores)
    df = pd.concat(pool.map(func, df_split))
    pool.close()
    pool.join()
    return df   

if __name__ == '__main__':
  output = parallelize(
    df=chosen_input_dataframe,
    func = f,
    n_cores=4,
    )
   
   print(output)
Dharman
  • 30,962
  • 25
  • 85
  • 135
Romain
  • 1
  • 1