Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Is it possible to prevent python from termination until block ends?

Basically, I'm curious if it's possible to execute a block of python code "atomically" without being interrupted by a signal.

For instance, I want to perform operations in a loop, let's say:

for i in range(100):
    do_stuff(1)
    do_stuff(2)
    do_stuff(3)

But I want to finish all of three do_stuff(1), do_stuff(2), do_stuff(3) if do_stuff(1) managed to start. Script should ignore CTRL+C, finish these three instructions and then terminate if SIGINT happened. All of 100 iterations does not have to be executed.

I believe it could be done with a custom signal handler

import signal

def handler(signum, frame):
    # wait for the loop iteration finish and exit
signal.signal(signal.SIGINT, handler)

threads and synchronization but I have no idea how to implement it.

  1. Is it possible?
  2. If it is, can it be done nicely? With some kind of a context manager, for example?
for i in range(100):
    with atomic_execution():
        do_stuff(1)
        do_stuff(2)
        do_stuff(3)

Edit: in the meantime I created this:

import threading
import sys
import signal


class atomic_execution:
    started = 0
    execution_in_progress = threading.Lock()

    def __enter__(self):
        atomic_execution.execution_in_progress.acquire()

    def __exit__(self, type, value, traceback):
        atomic_execution.execution_in_progress.release()


def handler(signum, frame):
  atomic_execution.execution_in_progress.acquire()
  sys.exit(0)

signal.signal(signal.SIGINT, handler)

while True:
  with atomic_execution():
    print(1)
    print(2)
    print(3)

I am not sure if it's good, though.

like image 461
asikorski Avatar asked May 21 '26 11:05

asikorski


1 Answers

This is the basic idea:

import signal
import time

stop = False

def sighandler(*unused):
    global stop
    print('signal caught')
    stop = True

def main():
    for i in range(10):
        print('a')
        time.sleep(0.5)
        print('b')
        time.sleep(0.5)
        print('c')
        time.sleep(0.5)
        print()
        if stop:
            print('STOP')
            break

if __name__ == '__main__':
    signal.signal(signal.SIGINT, sighandler)
    main()

I think it is not difficult to make an context manager for this purpose:

on enter:

  • save the current signal handler
  • install own handler setting a flag like in the example above

on exit:

  • restore the original signal handler
  • exit if the flag was set

But I do not like the idea, because you want to install the handler once before the loop and test the flag many times at each iteration.

like image 192
VPfB Avatar answered May 23 '26 03:05

VPfB



Donate For Us

If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!