Search code examples
python-3.xlinuxmultiprocessing

Error automatically calling decorated function in Python


There is Python code that contains a singleton_process decorator and a decodable hello function. The decorator is designed to protect against re-running the script in another terminal/process.

from multiprocessing import shared_memory
import time
import os
import sys


def singleton_process(*args, **kwargs):
    def wrapper(func):
        flag = True
        try:
            shared_name = kwargs.get('shared_name')
            shm = shared_memory.SharedMemory(
                name=shared_name,
                create=True,
                size=1
            )
            print("The first copy of the script has been launched.")
            func()
        except FileExistsError:
            flag = False
            print("Another instance of the script is already running.")
            sys.exit(1)
        except KeyboardInterrupt:
            shm.close()
            shm.unlink()
        finally:
            if flag:
                shm.close()
                try:
                    shm.unlink()
                except FileNotFoundError:
                    pass

    return wrapper 


@singleton_process(shared_name='my_lock')
def hello():
    c = 0
    while True:
        c += 1
        if c == 5:
            break
        pid = os.getpid()
        print(f'PID={pid}. Script it work...')
        time.sleep(1)
    return True


if __name__ == '__main__':
    #print('output=', type(hello))
    #hello()
    pass

The code automatically calls the function being decoded, even when it is not explicitly asked to do so. How can I fix the script and make it run the hello() function only in the if __name__ == “__main__” statement?


Solution

  • The way you define the decorator runs hello in the decorating process.

    You need an inner function :

    from multiprocessing import shared_memory
    import time
    import os
    import sys
    
    
    def singleton_process(*args, **kwargs):
        def wrapper(func):
            def inner():
                flag = True
                try:
                    shared_name = kwargs.get('shared_name')
                    shm = shared_memory.SharedMemory(
                        name=shared_name,
                        create=True,
                        size=1
                    )
                    print("The first copy of the script has been launched.")
                    func()
                except FileExistsError:
                    flag = False
                    print("Another instance of the script is already running.")
                    sys.exit(1)
                except KeyboardInterrupt:
                    shm.close()
                    shm.unlink()
                finally:
                    if flag:
                        shm.close()
                        try:
                            shm.unlink()
                        except FileNotFoundError:
                            pass
            return inner
        return wrapper
    
    
    @singleton_process(shared_name='my_lock')
    def hello():
        c = 0
        while True:
            c += 1
            if c == 5:
                break
            pid = os.getpid()
            print(f'PID={pid}. Script it work...')
            time.sleep(1)
        return True
    
    
    if __name__ == '__main__':
        hello()