issue3319

Complexity

Total Complexity

5

Size/Duplication

Total Lines	58
Duplicated Lines	46.55 %

Importance

Changes

0

1 Function

Rating	Name	Duplication	Size	Complexity
A	exit_after()	27	27	3

####################################################################################
#
# Migration of the code issue3290.py to use multiprocessing with 'fork' start method
#
####################################################################################

import multiprocessing
import time

import psutil

# multiprocessing.set_start_method("fork")
ctx_mp_fork = multiprocessing.get_context('fork')


def exit_after(seconds, default=None):

This code seems to be duplicated in your project.

    """Exit the function if it takes more than 'second' seconds to complete.
    In this case, return the value of 'default' (default: None)."""

    def handler(q, func, args, kwargs):
        q.put(func(*args, **kwargs))

    def decorator(func):
        def wraps(*args, **kwargs):
            q = ctx_mp_fork.Queue()
            p = ctx_mp_fork.Process(target=handler, args=(q, func, args, kwargs))
            p.start()
            p.join(timeout=seconds)
            if not p.is_alive():
                return q.get()

            p.terminate()
            p.join(timeout=0.1)
            if p.is_alive():
                # Kill in case processes doesn't terminate
                # Happens with cases like broken NFS connections
                p.kill()
            return default

        return wraps

    return decorator


class Issue3290:
    @exit_after(1, default=None)
    def blocking_io_call(self, fs):
        try:
            return psutil.disk_usage(fs)

The variable psutil does not seem to be defined.

        except OSError:
            return None


issue = Issue3290()
while True:
    print(f"{time.time()} {issue.blocking_io_call('/home/nicolargo/tmp/hang')}")
    time.sleep(1)