Create cleanup mechanism

[cgarciae]

• implement __del__

[mkarmona]

@cgarciae this project works great but it does not exit gracefully. I would like to get this somehow integrated with pypeln. Happy to help with it in the case it would be needed.

[cgarciae]

Thanks @mkarmona !

What kind of test have you ran? I know that if you try something like this:

iterable = iter(stage)
next(iterable)
# then do nothing

that doesn't fully consume the stage's iterator will leave the background processes/threads hanging.

I've thought about this and the proper way to do it would be

1. Have to_iterable return a custom iterable object.
2. Have the class of this object implement the __del__ method.
3. Inside __del__ call the .done() method of all the InputQueues

About SIGINT terminator, I think a way to handle this would be to just use a something like

try:
    # to_iterable code
finally:
    for queue in stage_input_queue.values():
        queue.done()

in to_iterable, the code insidefinally should be called regardless of the exception type.

If you wish to try fix this I can help you along the way!

[mkarmona]

@cgarciae thanks! Basically I am doing this in the main section of my pipeline

    logger.debug('create an iterable of handles from filenames %s', str(filenames))
    in_handles = itertools.imap(from_source_for_reading, filenames)

    logger.debug('create a iterable of lines from all file handles')
    chained_handles = itertools.chain.from_iterable(itertools.ifilter(lambda e: e is not None, in_handles))

    evs = more_itertools.take(first_n, chained_handles) \
        if first_n else chained_handles

    logger.debug('load LUTs')
    lookup_data = make_lookup_data(es_client, redis_client)

    logger.info('declare pipeline to run')
    write_evidences_on_start_f = functools.partial(write_evidences_on_start, enable_output_to_es, output_folder)
    validate_evidence_on_start_f = functools.partial(process_evidence_on_start, lookup_data)

    # here the pipeline definition
    pl_stage = pr.map(process_evidence, evs, workers=num_workers, maxsize=10000,
                      on_start=validate_evidence_on_start_f, on_done=process_evidence_on_done)
    pl_stage = pr.map(write_evidences, pl_stage, workers=num_writers, maxsize=10000, on_start=write_evidences_on_start_f)

    logger.info('run evidence processing pipeline')
    results = reduce_tuple_with_sum(pr.to_iterable(pl_stage))
    logger.info('done evidence processing pipeline')
    return results

So yes, I know is lazy evaluated and I work with iterators as inputs and outputs. But if you Ctrl+C from terminal it gets into a stage where you cannot exit without kill -9 the main process. Also, if one of the processes die caused by an exception from inside it will go into the same state again you have again to kill the whole process. That said, does it make sense from the pypeln core functions as map filter and so on just trying to catch general exceptions and exiting gracefully from those?
I will try your suggestions, whether you consider my comments or not. I will come back, eventually if I get some result.

[cgarciae]

@mkarmona I see.

But my comment was actually about modifying Pypeline's to_iterable function which is called under the hood every time you want to iterate over a stage as to clean after itself, so yes, I see this as a feature we should implement in Pypeline itself and not leave it to the user.

I'll try to take a shot at it during the weekend, but if you have time in the previous comment I left some ideas as how you can implement it if you want to create a PR.

[mkarmona]

@cgarciae let me see this weekend if I find the proper time slot to push this forward

[cgarciae]

@mkarmona I just realized that during the implementation I create all Threads and Processes with daemon = True, therefore they exit if the main process exits via Ctrl+C. I just verified this behavior with the following code:

from pypeln import process as pr
import time

def do_print(x):
    time.sleep(1)
    print(x)

stage = pr.map(do_print, range(1000), workers = 5)

pr.run(stage)

In htop I see that all background process exit and I don't see any hanging prints. What kind of behavior are you experiencing?

[mkarmona]

@cgarciae that was a really simple example, in my pipeline one step fills the queue as the second one is trying to cope with the messages at a slower pace so when it fills the maxsize the performance is reduce, which is ok as it was expected, but if I Ctrl+C I get this out on the console

^CTraceback (most recent call last):
  File "/usr/lib/python2.7/runpy.py", line 174, in _run_module_as_main
    "__main__", fname, loader, pkg_name)
  File "/usr/lib/python2.7/runpy.py", line 72, in _run_code
    exec code in run_globals
  File "/home/mkarmona/src/github/opent/data_pipeline_refactor/mrtarget/CommandLine.py", line 378, in <module>
    sys.exit(main())
  File "/home/mkarmona/src/github/opent/data_pipeline_refactor/mrtarget/CommandLine.py", line 316, in main
    num_writers=args.num_writers)
  File "mrtarget/modules/Evidences.py", line 357, in process_evidences_pipeline
    results = reduce_tuple_with_sum(pr.to_iterable(pl_stage))
  File "mrtarget/common/EvidencesHelpers.py", line 142, in reduce_tuple_with_sum
    return functools.reduce(lambda x, y: (x[0] + y[0], x[1] + y[1]), iterable, (0, 0))
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 841, in _to_iterable
    for x in input_queue:
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 236, in __iter__
    if self.pipeline_namespace.error:
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 1023, in __getattr__
    return callmethod('__getattribute__', (key,))
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 759, in _callmethod
    kind, result = conn.recv()
KeyboardInterrupt

and it keeps stuck there ad infinitum

[cgarciae]

@mkarmona I see.

I think this can be solved by fixing the pypeln.process._run_task function and adding a try/except. This can easily be fixed, I'll ping you when I have the new code on develop so you can test it.

[cgarciae]

@mkarmona this was actually a quick fix. Can you upgrade via

pip install -U git+https://github.com/cgarciae/pypeln@develop

and try the new code?

[mkarmona]

@cgarciae cool! how quick thanks. Here my attempt to test it with real code

^CTraceback (most recent call last):
  File "/usr/lib/python2.7/runpy.py", line 174, in _run_module_as_main
    "__main__", fname, loader, pkg_name)
  File "/usr/lib/python2.7/runpy.py", line 72, in _run_code
    exec code in run_globals
  File "/home/mkarmona/src/github/opent/data_pipeline_refactor/mrtarget/CommandLine.py", line 378, in <module>
    sys.exit(main())
  File "/home/mkarmona/src/github/opent/data_pipeline_refactor/mrtarget/CommandLine.py", line 316, in main
    num_writers=args.num_writers)
  File "mrtarget/modules/Evidences.py", line 357, in process_evidences_pipeline
    results = reduce_tuple_with_sum(pr.to_iterable(pl_stage))
  File "mrtarget/common/EvidencesHelpers.py", line 142, in reduce_tuple_with_sum
    return functools.reduce(lambda x, y: (x[0] + y[0], x[1] + y[1]), iterable, (0, 0))
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 848, in _to_iterable
    for x in input_queue:
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 233, in __iter__
    while not self.is_done():
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 258, in is_done
    return self.namespace.remaining == 0 and self.queue.empty()
KeyboardInterrupt
Process Process-2:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 267, in _bootstrap
Process Process-3:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 267, in _bootstrap
    self.run()
    self.run()
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
    self._target(*self._args, **self._kwargs)
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 341, in _map
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 341, in _map
    _run_task(f_task, params)
    _run_task(f_task, params)
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 326, in _run_task
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 326, in _run_task
    params.pipeline_namespace.error = True
    params.pipeline_namespace.error = True
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 1028, in __setattr__
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 1028, in __setattr__
    return callmethod('__setattr__', (key, value))
    return callmethod('__setattr__', (key, value))
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 758, in _callmethod
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 758, in _callmethod
    conn.send((self._id, methodname, args, kwds))
    conn.send((self._id, methodname, args, kwds))
IOError: [Errno 32] Broken pipe
IOError: [Errno 32] Broken pipe
2018-11-15 22:31:57,485 - mrtarget.common.EvidencesHelpers_31400 - DEBUG - closing files ./evidences-valid_259360b92bf743078e01b83b12fe4f89.json.gz ./evidences-invalid_e634a404df8447838ccc2ccc5ad05f4f.json.gz
2018-11-15 22:31:57,485 - mrtarget.common.EvidencesHelpers_31396 - DEBUG - closing files ./evidences-valid_bc4d6ac7f2364d9f959f823d8203c680.json.gz ./evidences-invalid_e753a055787a4910a77514be2665750d.json.gz

[cgarciae]

@mkarmona thanks for the feedback!

1. Can you create some minimal code that reproduces this behavior? I've tested my simple print code in both python 2 and 3 but I don't get this problem.
2. Looking at the traceback, the error is happening in code that is trying to communicate the exception to the main process but fails for some reason (probably the main process exited), I've added an extra try/except for this, you can test it as before on develop

[mkarmona]

@cgarciae I tried again and I now got a broken pipe. I found this on StackOverflow which may be worth to have a look.

^CTraceback (most recent call last):
  File "/usr/lib/python2.7/runpy.py", line 174, in _run_module_as_main
    "__main__", fname, loader, pkg_name)
  File "/usr/lib/python2.7/runpy.py", line 72, in _run_code
    exec code in run_globals
  File "/home/mkarmona/src/github/opent/data_pipeline_refactor/mrtarget/modules/Evidences.py", line 373, in <module>
    enable_output_to_es=False, output_folder='.')
  File "/home/mkarmona/src/github/opent/data_pipeline_refactor/mrtarget/modules/Evidences.py", line 357, in process_evidences_pipeline
    results = reduce_tuple_with_sum(pr.to_iterable(pl_stage))
  File "mrtarget/common/EvidencesHelpers.py", line 136, in reduce_tuple_with_sum
    return functools.reduce(lambda x, y: (x[0] + y[0], x[1] + y[1]), iterable, (0, 0))
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 851, in _to_iterable
    for x in input_queue:
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 233, in __iter__
    while not self.is_done():
  File "/home/mkarmona/.virtualenvs/mrtarget/local/lib/python2.7/site-packages/pypeln/process.py", line 258, in is_done
    return self.namespace.remaining == 0 and self.queue.empty()
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 1023, in __getattr__
[Errno 32] Broken pipe
    return callmethod('__getattribute__', (key,))
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 759, in _callmethod
    kind, result = conn.recv()
KeyboardInterrupt
[Errno 32] Broken pipe

I will try to simplify my main function as a working example, I am afraid it won't help anyway though as it works if I put a limit in the number of lines to be processed or I leave it to finish.