#!/usr/bin/python
# Copyright 2015 The Chromium OS Authors.

"""Reproduction for semaphore race bug

See https://bugs.python.org/issue24303"""

from __future__ import print_function

import ctypes
import ctypes.util
import os
import sys


# How many times should we run the test.  It usually reproduces fast.
MAX_ATTEMPTS = 3

# Max children at once.  We don't want to fork bomb the system :).
# But we need a bunch at the same time to help trigger the bug.
MAX_CHILDREN = 20


def exitstatus(status):
    """Decode the status from wait()"""
    if os.WIFSIGNALED(status):
        return 128 + os.WTERMSIG(status)
    elif os.WIFEXITED(status):
        return os.WEXITSTATUS(status)
    else:
        raise ValueError('unknown status: %s' % status)


def waitall():
    """reap all children and return an array of all their statuses"""
    statuses = []
    while True:
        try:
            _pid, status = os.wait()
            statuses.append(exitstatus(status))
        except OSError:
            break
    return statuses


def unshare_pidns():
    """Run unshare(CLONE_NEWPID)"""
    CLONE_NEWUSER = 0x10000000
    CLONE_NEWPID = 0x20000000
    libc = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True)
    # First unshare the user namespace so this test can run as non-root.
    libc.unshare(ctypes.c_int(CLONE_NEWUSER))
    # Then unshare the pid namespace.
    if libc.unshare(ctypes.c_int(CLONE_NEWPID)) != 0:
        e = ctypes.get_errno()
        raise OSError(e, os.strerror(e))


def forkone():
    """Fork a single child in the right way"""
    # This has to be in a sep process so that the multiprocessing module is not
    # able to communicate back and prevent the race.
    cmd = [
        sys.executable, '-c', 'import multiprocessing; multiprocessing.Event()',
    ]

    if os.fork() == 0:
        # The first child has to unshare the pid ns.
        unshare_pidns()
        if os.fork() == 0:
            # While the second actually runs the code.
            os.execvp(cmd[0], cmd)
        else:
            # Pass the child status back up.
            _pid, status = os.wait()
            os._exit(exitstatus(status))


def main():
    # Sanity check!
    forkone()
    statuses = waitall()
    if sum(statuses):
        print('test error!', file=sys.stderr)
        return False

    failed = False
    total = curr = 0
    while not failed and total < MAX_ATTEMPTS * MAX_CHILDREN:
        total += 1
        forkone()
        curr += 1
        if curr >= MAX_CHILDREN:
            curr = 0
            statuses = waitall()
            if sum(statuses):
                failed = True

    waitall()

    if failed:
        print('failed', file=sys.stderr)
    else:
        print('passed!')

    return failed


if __name__ == '__main__':
    sys.exit(main())