{
    "file_diffs": [
        {
            "old_file_content": "#!/usr/bin/env python\n#\n# Copyright 2011 Facebook\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\n\"\"\"Utilities for working with multiple processes, including both forking\nthe server into multiple processes and managing subprocesses.\n\"\"\"\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\nimport errno\nimport os\nimport signal\nimport subprocess\nimport sys\nimport time\n\nfrom binascii import hexlify\n\nfrom tornado.concurrent import Future\nfrom tornado import ioloop\nfrom tornado.iostream import PipeIOStream\nfrom tornado.log import gen_log\nfrom tornado.platform.auto import set_close_exec\nfrom tornado import stack_context\nfrom tornado.util import errno_from_exception\n\ntry:\n    import multiprocessing\nexcept ImportError:\n    # Multiprocessing is not available on Google App Engine.\n    multiprocessing = None\n\ntry:\n    long  # py2\nexcept NameError:\n    long = int  # py3\n\n\ndef cpu_count():\n    \"\"\"Returns the number of processors on this machine.\"\"\"\n    if multiprocessing is None:\n        return 1\n    try:\n        return multiprocessing.cpu_count()\n    except NotImplementedError:\n        pass\n    try:\n        return os.sysconf(\"SC_NPROCESSORS_CONF\")\n    except ValueError:\n        pass\n    gen_log.error(\"Could not detect number of processors; assuming 1\")\n    return 1\n\n\ndef _reseed_random():\n    if 'random' not in sys.modules:\n        return\n    import random\n    # If os.urandom is available, this method does the same thing as\n    # random.seed (at least as of python 2.6).  If os.urandom is not\n    # available, we mix in the pid in addition to a timestamp.\n    try:\n        seed = long(hexlify(os.urandom(16)), 16)\n    except NotImplementedError:\n        seed = int(time.time() * 1000) ^ os.getpid()\n    random.seed(seed)\n\n\ndef _pipe_cloexec():\n    r, w = os.pipe()\n    set_close_exec(r)\n    set_close_exec(w)\n    return r, w\n\n\n_task_id = None\n\n\ndef fork_processes(num_processes, max_restarts=100):\n    \"\"\"Starts multiple worker processes.\n\n    If ``num_processes`` is None or <= 0, we detect the number of cores\n    available on this machine and fork that number of child\n    processes. If ``num_processes`` is given and > 0, we fork that\n    specific number of sub-processes.\n\n    Since we use processes and not threads, there is no shared memory\n    between any server code.\n\n    Note that multiple processes are not compatible with the autoreload\n    module (or the ``autoreload=True`` option to `tornado.web.Application`\n    which defaults to True when ``debug=True``).\n    When using multiple processes, no IOLoops can be created or\n    referenced until after the call to ``fork_processes``.\n\n    In each child process, ``fork_processes`` returns its *task id*, a\n    number between 0 and ``num_processes``.  Processes that exit\n    abnormally (due to a signal or non-zero exit status) are restarted\n    with the same id (up to ``max_restarts`` times).  In the parent\n    process, ``fork_processes`` returns None if all child processes\n    have exited normally, but will otherwise only exit by throwing an\n    exception.\n    \"\"\"\n    global _task_id\n    assert _task_id is None\n    if num_processes is None or num_processes <= 0:\n        num_processes = cpu_count()\n    if ioloop.IOLoop.initialized():\n        raise RuntimeError(\"Cannot run in multiple processes: IOLoop instance \"\n                           \"has already been initialized. You cannot call \"\n                           \"IOLoop.instance() before calling start_processes()\")\n    gen_log.info(\"Starting %d processes\", num_processes)\n    children = {}\n\n    def start_child(i):\n        pid = os.fork()\n        if pid == 0:\n            # child process\n            _reseed_random()\n            global _task_id\n            _task_id = i\n            return i\n        else:\n            children[pid] = i\n            return None\n    for i in range(num_processes):\n        id = start_child(i)\n        if id is not None:\n            return id\n    num_restarts = 0\n    while children:\n        try:\n            pid, status = os.wait()\n        except OSError as e:\n            if errno_from_exception(e) == errno.EINTR:\n                continue\n            raise\n        if pid not in children:\n            continue\n        id = children.pop(pid)\n        if os.WIFSIGNALED(status):\n            gen_log.warning(\"child %d (pid %d) killed by signal %d, restarting\",\n                            id, pid, os.WTERMSIG(status))\n        elif os.WEXITSTATUS(status) != 0:\n            gen_log.warning(\"child %d (pid %d) exited with status %d, restarting\",\n                            id, pid, os.WEXITSTATUS(status))\n        else:\n            gen_log.info(\"child %d (pid %d) exited normally\", id, pid)\n            continue\n        num_restarts += 1\n        if num_restarts > max_restarts:\n            raise RuntimeError(\"Too many child restarts, giving up\")\n        new_id = start_child(id)\n        if new_id is not None:\n            return new_id\n    # All child processes exited cleanly, so exit the master process\n    # instead of just returning to right after the call to\n    # fork_processes (which will probably just start up another IOLoop\n    # unless the caller checks the return value).\n    sys.exit(0)\n\n\ndef task_id():\n    \"\"\"Returns the current task id, if any.\n\n    Returns None if this process was not created by `fork_processes`.\n    \"\"\"\n    global _task_id\n    return _task_id\n\n\nclass Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)\n",
            "new_file_content": "#!/usr/bin/env python\n#\n# Copyright 2011 Facebook\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\n\"\"\"Utilities for working with multiple processes, including both forking\nthe server into multiple processes and managing subprocesses.\n\"\"\"\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\nimport errno\nimport os\nimport signal\nimport subprocess\nimport sys\nimport time\n\nfrom binascii import hexlify\n\nfrom tornado.concurrent import Future\nfrom tornado import ioloop\nfrom tornado.iostream import PipeIOStream\nfrom tornado.log import gen_log\nfrom tornado.platform.auto import set_close_exec\nfrom tornado import stack_context\nfrom tornado.util import errno_from_exception\n\ntry:\n    import multiprocessing\nexcept ImportError:\n    # Multiprocessing is not available on Google App Engine.\n    multiprocessing = None\n\ntry:\n    long  # py2\nexcept NameError:\n    long = int  # py3\n\n\n# Re-export this exception for convenience.\nCalledProcessError = subprocess.CalledProcessError\n\n\ndef cpu_count():\n    \"\"\"Returns the number of processors on this machine.\"\"\"\n    if multiprocessing is None:\n        return 1\n    try:\n        return multiprocessing.cpu_count()\n    except NotImplementedError:\n        pass\n    try:\n        return os.sysconf(\"SC_NPROCESSORS_CONF\")\n    except ValueError:\n        pass\n    gen_log.error(\"Could not detect number of processors; assuming 1\")\n    return 1\n\n\ndef _reseed_random():\n    if 'random' not in sys.modules:\n        return\n    import random\n    # If os.urandom is available, this method does the same thing as\n    # random.seed (at least as of python 2.6).  If os.urandom is not\n    # available, we mix in the pid in addition to a timestamp.\n    try:\n        seed = long(hexlify(os.urandom(16)), 16)\n    except NotImplementedError:\n        seed = int(time.time() * 1000) ^ os.getpid()\n    random.seed(seed)\n\n\ndef _pipe_cloexec():\n    r, w = os.pipe()\n    set_close_exec(r)\n    set_close_exec(w)\n    return r, w\n\n\n_task_id = None\n\n\ndef fork_processes(num_processes, max_restarts=100):\n    \"\"\"Starts multiple worker processes.\n\n    If ``num_processes`` is None or <= 0, we detect the number of cores\n    available on this machine and fork that number of child\n    processes. If ``num_processes`` is given and > 0, we fork that\n    specific number of sub-processes.\n\n    Since we use processes and not threads, there is no shared memory\n    between any server code.\n\n    Note that multiple processes are not compatible with the autoreload\n    module (or the ``autoreload=True`` option to `tornado.web.Application`\n    which defaults to True when ``debug=True``).\n    When using multiple processes, no IOLoops can be created or\n    referenced until after the call to ``fork_processes``.\n\n    In each child process, ``fork_processes`` returns its *task id*, a\n    number between 0 and ``num_processes``.  Processes that exit\n    abnormally (due to a signal or non-zero exit status) are restarted\n    with the same id (up to ``max_restarts`` times).  In the parent\n    process, ``fork_processes`` returns None if all child processes\n    have exited normally, but will otherwise only exit by throwing an\n    exception.\n    \"\"\"\n    global _task_id\n    assert _task_id is None\n    if num_processes is None or num_processes <= 0:\n        num_processes = cpu_count()\n    if ioloop.IOLoop.initialized():\n        raise RuntimeError(\"Cannot run in multiple processes: IOLoop instance \"\n                           \"has already been initialized. You cannot call \"\n                           \"IOLoop.instance() before calling start_processes()\")\n    gen_log.info(\"Starting %d processes\", num_processes)\n    children = {}\n\n    def start_child(i):\n        pid = os.fork()\n        if pid == 0:\n            # child process\n            _reseed_random()\n            global _task_id\n            _task_id = i\n            return i\n        else:\n            children[pid] = i\n            return None\n    for i in range(num_processes):\n        id = start_child(i)\n        if id is not None:\n            return id\n    num_restarts = 0\n    while children:\n        try:\n            pid, status = os.wait()\n        except OSError as e:\n            if errno_from_exception(e) == errno.EINTR:\n                continue\n            raise\n        if pid not in children:\n            continue\n        id = children.pop(pid)\n        if os.WIFSIGNALED(status):\n            gen_log.warning(\"child %d (pid %d) killed by signal %d, restarting\",\n                            id, pid, os.WTERMSIG(status))\n        elif os.WEXITSTATUS(status) != 0:\n            gen_log.warning(\"child %d (pid %d) exited with status %d, restarting\",\n                            id, pid, os.WEXITSTATUS(status))\n        else:\n            gen_log.info(\"child %d (pid %d) exited normally\", id, pid)\n            continue\n        num_restarts += 1\n        if num_restarts > max_restarts:\n            raise RuntimeError(\"Too many child restarts, giving up\")\n        new_id = start_child(id)\n        if new_id is not None:\n            return new_id\n    # All child processes exited cleanly, so exit the master process\n    # instead of just returning to right after the call to\n    # fork_processes (which will probably just start up another IOLoop\n    # unless the caller checks the return value).\n    sys.exit(0)\n\n\ndef task_id():\n    \"\"\"Returns the current task id, if any.\n\n    Returns None if this process was not created by `fork_processes`.\n    \"\"\"\n    global _task_id\n    return _task_id\n\n\nclass Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self, raise_error=True):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        By default, raises `subprocess.CalledProcessError` if the process\n        has a non-zero exit status. Use ``wait_for_exit(raise_error=False)``\n        to suppress this behavior and return the exit status without raising.\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n\n        def callback(ret):\n            if ret != 0 and raise_error:\n                # Unfortunately we don't have the original args any more.\n                future.set_exception(CalledProcessError(ret, None))\n            else:\n                future.set_result(ret)\n        self.set_exit_callback(callback)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)\n",
            "header": {
                "file": {
                    "path": "tornado/process.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "fb9254fc",
                "new_commit_hash": "f580e192",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/process.py"
            },
            "plus_file": {
                "path": "b/tornado/process.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 49,
                            "length": 6
                        },
                        "new_range": {
                            "start": 49,
                            "length": 10
                        },
                        "section": "except NameError:"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    long = int  # py3",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "# Re-export this exception for convenience.",
                                "type": "added"
                            },
                            {
                                "content": "CalledProcessError = subprocess.CalledProcessError",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "def cpu_count():",
                                "type": "context"
                            },
                            {
                                "content": "    \"\"\"Returns the number of processors on this machine.\"\"\"",
                                "type": "context"
                            },
                            {
                                "content": "    if multiprocessing is None:",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "tornado/process.py",
                            "type": "statement",
                            "name": "",
                            "content": "CalledProcessError = subprocess.CalledProcessError",
                            "ast_type_str": "Assign",
                            "start_lineno": 53,
                            "end_lineno": 53,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 259,
                            "length": 7
                        },
                        "new_range": {
                            "start": 263,
                            "length": 7
                        },
                        "section": "class Subprocess(object):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        Subprocess._waiting[self.pid] = self",
                                "type": "context"
                            },
                            {
                                "content": "        Subprocess._try_cleanup_process(self.pid)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def wait_for_exit(self):",
                                "type": "deleted"
                            },
                            {
                                "content": "    def wait_for_exit(self, raise_error=True):",
                                "type": "added"
                            },
                            {
                                "content": "        \"\"\"Returns a `.Future` which resolves when the process exits.",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "        Usage::",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/process.py",
                            "type": "method",
                            "name": "Subprocess.wait_for_exit",
                            "content": "class Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 262,
                            "end_lineno": 276,
                            "parent": {
                                "file_name": "tornado/process.py",
                                "type": "class",
                                "name": "Subprocess",
                                "content": "class Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 185,
                                "end_lineno": 335,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/process.py",
                            "type": "class",
                            "name": "Subprocess",
                            "content": "class Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 185,
                            "end_lineno": 335,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 269,
                            "length": 10
                        },
                        "new_range": {
                            "start": 273,
                            "length": 21
                        },
                        "section": "class Subprocess(object):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        This is a coroutine-friendly alternative to `set_exit_callback`",
                                "type": "context"
                            },
                            {
                                "content": "        (and a replacement for the blocking `subprocess.Popen.wait`).",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "        By default, raises `subprocess.CalledProcessError` if the process",
                                "type": "added"
                            },
                            {
                                "content": "        has a non-zero exit status. Use ``wait_for_exit(raise_error=False)``",
                                "type": "added"
                            },
                            {
                                "content": "        to suppress this behavior and return the exit status without raising.",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        .. versionadded:: 4.2",
                                "type": "context"
                            },
                            {
                                "content": "        \"\"\"",
                                "type": "context"
                            },
                            {
                                "content": "        future = Future()",
                                "type": "context"
                            },
                            {
                                "content": "        self.set_exit_callback(future.set_result)",
                                "type": "deleted"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        def callback(ret):",
                                "type": "added"
                            },
                            {
                                "content": "            if ret != 0 and raise_error:",
                                "type": "added"
                            },
                            {
                                "content": "                # Unfortunately we don't have the original args any more.",
                                "type": "added"
                            },
                            {
                                "content": "                future.set_exception(CalledProcessError(ret, None))",
                                "type": "added"
                            },
                            {
                                "content": "            else:",
                                "type": "added"
                            },
                            {
                                "content": "                future.set_result(ret)",
                                "type": "added"
                            },
                            {
                                "content": "        self.set_exit_callback(callback)",
                                "type": "added"
                            },
                            {
                                "content": "        return future",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    @classmethod",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/process.py",
                            "type": "method",
                            "name": "Subprocess.wait_for_exit",
                            "content": "class Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 262,
                            "end_lineno": 276,
                            "parent": {
                                "file_name": "tornado/process.py",
                                "type": "class",
                                "name": "Subprocess",
                                "content": "class Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 185,
                                "end_lineno": 335,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/process.py",
                            "type": "class",
                            "name": "Subprocess",
                            "content": "class Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 185,
                            "end_lineno": 335,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "#!/usr/bin/env python\n\n\nfrom __future__ import absolute_import, division, print_function, with_statement\nimport logging\nimport os\nimport signal\nimport subprocess\nimport sys\nfrom tornado.httpclient import HTTPClient, HTTPError\nfrom tornado.httpserver import HTTPServer\nfrom tornado.ioloop import IOLoop\nfrom tornado.log import gen_log\nfrom tornado.process import fork_processes, task_id, Subprocess\nfrom tornado.simple_httpclient import SimpleAsyncHTTPClient\nfrom tornado.testing import bind_unused_port, ExpectLog, AsyncTestCase, gen_test\nfrom tornado.test.util import unittest, skipIfNonUnix\nfrom tornado.web import RequestHandler, Application\n\n\ndef skip_if_twisted():\n    if IOLoop.configured_class().__name__.endswith(('TwistedIOLoop',\n                                                    'AsyncIOMainLoop')):\n        raise unittest.SkipTest(\"Process tests not compatible with \"\n                                \"TwistedIOLoop or AsyncIOMainLoop\")\n\n# Not using AsyncHTTPTestCase because we need control over the IOLoop.\n\n\n@skipIfNonUnix\nclass ProcessTest(unittest.TestCase):\n    def get_app(self):\n        class ProcessHandler(RequestHandler):\n            def get(self):\n                if self.get_argument(\"exit\", None):\n                    # must use os._exit instead of sys.exit so unittest's\n                    # exception handler doesn't catch it\n                    os._exit(int(self.get_argument(\"exit\")))\n                if self.get_argument(\"signal\", None):\n                    os.kill(os.getpid(),\n                            int(self.get_argument(\"signal\")))\n                self.write(str(os.getpid()))\n        return Application([(\"/\", ProcessHandler)])\n\n    def tearDown(self):\n        if task_id() is not None:\n            # We're in a child process, and probably got to this point\n            # via an uncaught exception.  If we return now, both\n            # processes will continue with the rest of the test suite.\n            # Exit now so the parent process will restart the child\n            # (since we don't have a clean way to signal failure to\n            # the parent that won't restart)\n            logging.error(\"aborting child process from tearDown\")\n            logging.shutdown()\n            os._exit(1)\n        # In the surviving process, clear the alarm we set earlier\n        signal.alarm(0)\n        super(ProcessTest, self).tearDown()\n\n    def test_multi_process(self):\n        # This test can't work on twisted because we use the global reactor\n        # and have no way to get it back into a sane state after the fork.\n        skip_if_twisted()\n        with ExpectLog(gen_log, \"(Starting .* processes|child .* exited|uncaught exception)\"):\n            self.assertFalse(IOLoop.initialized())\n            sock, port = bind_unused_port()\n\n            def get_url(path):\n                return \"http://127.0.0.1:%d%s\" % (port, path)\n            # ensure that none of these processes live too long\n            signal.alarm(5)  # master process\n            try:\n                id = fork_processes(3, max_restarts=3)\n                self.assertTrue(id is not None)\n                signal.alarm(5)  # child processes\n            except SystemExit as e:\n                # if we exit cleanly from fork_processes, all the child processes\n                # finished with status 0\n                self.assertEqual(e.code, 0)\n                self.assertTrue(task_id() is None)\n                sock.close()\n                return\n            try:\n                if id in (0, 1):\n                    self.assertEqual(id, task_id())\n                    server = HTTPServer(self.get_app())\n                    server.add_sockets([sock])\n                    IOLoop.current().start()\n                elif id == 2:\n                    self.assertEqual(id, task_id())\n                    sock.close()\n                    # Always use SimpleAsyncHTTPClient here; the curl\n                    # version appears to get confused sometimes if the\n                    # connection gets closed before it's had a chance to\n                    # switch from writing mode to reading mode.\n                    client = HTTPClient(SimpleAsyncHTTPClient)\n\n                    def fetch(url, fail_ok=False):\n                        try:\n                            return client.fetch(get_url(url))\n                        except HTTPError as e:\n                            if not (fail_ok and e.code == 599):\n                                raise\n\n                    # Make two processes exit abnormally\n                    fetch(\"/?exit=2\", fail_ok=True)\n                    fetch(\"/?exit=3\", fail_ok=True)\n\n                    # They've been restarted, so a new fetch will work\n                    int(fetch(\"/\").body)\n\n                    # Now the same with signals\n                    # Disabled because on the mac a process dying with a signal\n                    # can trigger an \"Application exited abnormally; send error\n                    # report to Apple?\" prompt.\n                    # fetch(\"/?signal=%d\" % signal.SIGTERM, fail_ok=True)\n                    # fetch(\"/?signal=%d\" % signal.SIGABRT, fail_ok=True)\n                    # int(fetch(\"/\").body)\n\n                    # Now kill them normally so they won't be restarted\n                    fetch(\"/?exit=0\", fail_ok=True)\n                    # One process left; watch it's pid change\n                    pid = int(fetch(\"/\").body)\n                    fetch(\"/?exit=4\", fail_ok=True)\n                    pid2 = int(fetch(\"/\").body)\n                    self.assertNotEqual(pid, pid2)\n\n                    # Kill the last one so we shut down cleanly\n                    fetch(\"/?exit=0\", fail_ok=True)\n\n                    os._exit(0)\n            except Exception:\n                logging.error(\"exception in child process %d\", id, exc_info=True)\n                raise\n\n\n@skipIfNonUnix\nclass SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n",
            "new_file_content": "#!/usr/bin/env python\n\n\nfrom __future__ import absolute_import, division, print_function, with_statement\nimport logging\nimport os\nimport signal\nimport subprocess\nimport sys\nfrom tornado.httpclient import HTTPClient, HTTPError\nfrom tornado.httpserver import HTTPServer\nfrom tornado.ioloop import IOLoop\nfrom tornado.log import gen_log\nfrom tornado.process import fork_processes, task_id, Subprocess\nfrom tornado.simple_httpclient import SimpleAsyncHTTPClient\nfrom tornado.testing import bind_unused_port, ExpectLog, AsyncTestCase, gen_test\nfrom tornado.test.util import unittest, skipIfNonUnix\nfrom tornado.web import RequestHandler, Application\n\n\ndef skip_if_twisted():\n    if IOLoop.configured_class().__name__.endswith(('TwistedIOLoop',\n                                                    'AsyncIOMainLoop')):\n        raise unittest.SkipTest(\"Process tests not compatible with \"\n                                \"TwistedIOLoop or AsyncIOMainLoop\")\n\n# Not using AsyncHTTPTestCase because we need control over the IOLoop.\n\n\n@skipIfNonUnix\nclass ProcessTest(unittest.TestCase):\n    def get_app(self):\n        class ProcessHandler(RequestHandler):\n            def get(self):\n                if self.get_argument(\"exit\", None):\n                    # must use os._exit instead of sys.exit so unittest's\n                    # exception handler doesn't catch it\n                    os._exit(int(self.get_argument(\"exit\")))\n                if self.get_argument(\"signal\", None):\n                    os.kill(os.getpid(),\n                            int(self.get_argument(\"signal\")))\n                self.write(str(os.getpid()))\n        return Application([(\"/\", ProcessHandler)])\n\n    def tearDown(self):\n        if task_id() is not None:\n            # We're in a child process, and probably got to this point\n            # via an uncaught exception.  If we return now, both\n            # processes will continue with the rest of the test suite.\n            # Exit now so the parent process will restart the child\n            # (since we don't have a clean way to signal failure to\n            # the parent that won't restart)\n            logging.error(\"aborting child process from tearDown\")\n            logging.shutdown()\n            os._exit(1)\n        # In the surviving process, clear the alarm we set earlier\n        signal.alarm(0)\n        super(ProcessTest, self).tearDown()\n\n    def test_multi_process(self):\n        # This test can't work on twisted because we use the global reactor\n        # and have no way to get it back into a sane state after the fork.\n        skip_if_twisted()\n        with ExpectLog(gen_log, \"(Starting .* processes|child .* exited|uncaught exception)\"):\n            self.assertFalse(IOLoop.initialized())\n            sock, port = bind_unused_port()\n\n            def get_url(path):\n                return \"http://127.0.0.1:%d%s\" % (port, path)\n            # ensure that none of these processes live too long\n            signal.alarm(5)  # master process\n            try:\n                id = fork_processes(3, max_restarts=3)\n                self.assertTrue(id is not None)\n                signal.alarm(5)  # child processes\n            except SystemExit as e:\n                # if we exit cleanly from fork_processes, all the child processes\n                # finished with status 0\n                self.assertEqual(e.code, 0)\n                self.assertTrue(task_id() is None)\n                sock.close()\n                return\n            try:\n                if id in (0, 1):\n                    self.assertEqual(id, task_id())\n                    server = HTTPServer(self.get_app())\n                    server.add_sockets([sock])\n                    IOLoop.current().start()\n                elif id == 2:\n                    self.assertEqual(id, task_id())\n                    sock.close()\n                    # Always use SimpleAsyncHTTPClient here; the curl\n                    # version appears to get confused sometimes if the\n                    # connection gets closed before it's had a chance to\n                    # switch from writing mode to reading mode.\n                    client = HTTPClient(SimpleAsyncHTTPClient)\n\n                    def fetch(url, fail_ok=False):\n                        try:\n                            return client.fetch(get_url(url))\n                        except HTTPError as e:\n                            if not (fail_ok and e.code == 599):\n                                raise\n\n                    # Make two processes exit abnormally\n                    fetch(\"/?exit=2\", fail_ok=True)\n                    fetch(\"/?exit=3\", fail_ok=True)\n\n                    # They've been restarted, so a new fetch will work\n                    int(fetch(\"/\").body)\n\n                    # Now the same with signals\n                    # Disabled because on the mac a process dying with a signal\n                    # can trigger an \"Application exited abnormally; send error\n                    # report to Apple?\" prompt.\n                    # fetch(\"/?signal=%d\" % signal.SIGTERM, fail_ok=True)\n                    # fetch(\"/?signal=%d\" % signal.SIGABRT, fail_ok=True)\n                    # int(fetch(\"/\").body)\n\n                    # Now kill them normally so they won't be restarted\n                    fetch(\"/?exit=0\", fail_ok=True)\n                    # One process left; watch it's pid change\n                    pid = int(fetch(\"/\").body)\n                    fetch(\"/?exit=4\", fail_ok=True)\n                    pid2 = int(fetch(\"/\").body)\n                    self.assertNotEqual(pid, pid2)\n\n                    # Kill the last one so we shut down cleanly\n                    fetch(\"/?exit=0\", fail_ok=True)\n\n                    os._exit(0)\n            except Exception:\n                logging.error(\"exception in child process %d\", id, exc_info=True)\n                raise\n\n\n@skipIfNonUnix\nclass SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)\n",
            "header": {
                "file": {
                    "path": "tornado/test/process_test.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "72474167",
                "new_commit_hash": "58cc410b",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/test/process_test.py"
            },
            "plus_file": {
                "path": "b/tornado/test/process_test.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 222,
                            "length": 3
                        },
                        "new_range": {
                            "start": 222,
                            "length": 22
                        },
                        "section": "class SubprocessTest(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        ret = self.wait()",
                                "type": "context"
                            },
                            {
                                "content": "        self.assertEqual(subproc.returncode, ret)",
                                "type": "context"
                            },
                            {
                                "content": "        self.assertEqual(ret, -signal.SIGTERM)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    @gen_test",
                                "type": "added"
                            },
                            {
                                "content": "    def test_wait_for_exit_raise(self):",
                                "type": "added"
                            },
                            {
                                "content": "        skip_if_twisted()",
                                "type": "added"
                            },
                            {
                                "content": "        Subprocess.initialize()",
                                "type": "added"
                            },
                            {
                                "content": "        self.addCleanup(Subprocess.uninitialize)",
                                "type": "added"
                            },
                            {
                                "content": "        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])",
                                "type": "added"
                            },
                            {
                                "content": "        with self.assertRaises(subprocess.CalledProcessError) as cm:",
                                "type": "added"
                            },
                            {
                                "content": "            yield subproc.wait_for_exit()",
                                "type": "added"
                            },
                            {
                                "content": "        self.assertEqual(cm.exception.returncode, 1)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    @gen_test",
                                "type": "added"
                            },
                            {
                                "content": "    def test_wait_for_exit_raise_disabled(self):",
                                "type": "added"
                            },
                            {
                                "content": "        skip_if_twisted()",
                                "type": "added"
                            },
                            {
                                "content": "        Subprocess.initialize()",
                                "type": "added"
                            },
                            {
                                "content": "        self.addCleanup(Subprocess.uninitialize)",
                                "type": "added"
                            },
                            {
                                "content": "        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])",
                                "type": "added"
                            },
                            {
                                "content": "        ret = yield subproc.wait_for_exit(raise_error=False)",
                                "type": "added"
                            },
                            {
                                "content": "        self.assertEqual(ret, 1)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/process_test.py",
                            "type": "class",
                            "name": "SubprocessTest",
                            "content": "class SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 138,
                            "end_lineno": 243,
                            "parent": null
                        }
                    ],
                    "added_entities": [
                        {
                            "file_name": "tornado/test/process_test.py",
                            "type": "method",
                            "name": "SubprocessTest.test_wait_for_exit_raise_disabled",
                            "content": "class SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 237,
                            "end_lineno": 243,
                            "parent": {
                                "file_name": "tornado/test/process_test.py",
                                "type": "class",
                                "name": "SubprocessTest",
                                "content": "class SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 138,
                                "end_lineno": 243,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/test/process_test.py",
                            "type": "method",
                            "name": "SubprocessTest.test_wait_for_exit_raise",
                            "content": "class SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 227,
                            "end_lineno": 234,
                            "parent": {
                                "file_name": "tornado/test/process_test.py",
                                "type": "class",
                                "name": "SubprocessTest",
                                "content": "class SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 138,
                                "end_lineno": 243,
                                "parent": null
                            }
                        }
                    ],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "2bd2f0db087c12738ac55aba75fe78fe31987dd0^",
    "new_commit_hash": "2bd2f0db087c12738ac55aba75fe78fe31987dd0",
    "commit_message": "Add raise_error behavior to Subprocess.wait_for_exit.\n\nCloses #1205.",
    "commit_date": "2015-04-18T14:14:08-04:00",
    "metadata": {}
}
{
    "repo_name": "tornado",
    "new_commit_hash": "2bd2f0db087c12738ac55aba75fe78fe31987dd0",
    "test_file_codes": [
        "#!/usr/bin/env python\n\n\nfrom __future__ import absolute_import, division, print_function, with_statement\nimport logging\nimport os\nimport signal\nimport subprocess\nimport sys\nfrom tornado.httpclient import HTTPClient, HTTPError\nfrom tornado.httpserver import HTTPServer\nfrom tornado.ioloop import IOLoop\nfrom tornado.log import gen_log\nfrom tornado.process import fork_processes, task_id, Subprocess\nfrom tornado.simple_httpclient import SimpleAsyncHTTPClient\nfrom tornado.testing import bind_unused_port, ExpectLog, AsyncTestCase, gen_test\nfrom tornado.test.util import unittest, skipIfNonUnix\nfrom tornado.web import RequestHandler, Application\n\n\ndef skip_if_twisted():\n    if IOLoop.configured_class().__name__.endswith(('TwistedIOLoop',\n                                                    'AsyncIOMainLoop')):\n        raise unittest.SkipTest(\"Process tests not compatible with \"\n                                \"TwistedIOLoop or AsyncIOMainLoop\")\n\n# Not using AsyncHTTPTestCase because we need control over the IOLoop.\n\n\n@skipIfNonUnix\nclass ProcessTest(unittest.TestCase):\n    def get_app(self):\n        class ProcessHandler(RequestHandler):\n            def get(self):\n                if self.get_argument(\"exit\", None):\n                    # must use os._exit instead of sys.exit so unittest's\n                    # exception handler doesn't catch it\n                    os._exit(int(self.get_argument(\"exit\")))\n                if self.get_argument(\"signal\", None):\n                    os.kill(os.getpid(),\n                            int(self.get_argument(\"signal\")))\n                self.write(str(os.getpid()))\n        return Application([(\"/\", ProcessHandler)])\n\n    def tearDown(self):\n        if task_id() is not None:\n            # We're in a child process, and probably got to this point\n            # via an uncaught exception.  If we return now, both\n            # processes will continue with the rest of the test suite.\n            # Exit now so the parent process will restart the child\n            # (since we don't have a clean way to signal failure to\n            # the parent that won't restart)\n            logging.error(\"aborting child process from tearDown\")\n            logging.shutdown()\n            os._exit(1)\n        # In the surviving process, clear the alarm we set earlier\n        signal.alarm(0)\n        super(ProcessTest, self).tearDown()\n\n    def test_multi_process(self):\n        # This test can't work on twisted because we use the global reactor\n        # and have no way to get it back into a sane state after the fork.\n        skip_if_twisted()\n        with ExpectLog(gen_log, \"(Starting .* processes|child .* exited|uncaught exception)\"):\n            self.assertFalse(IOLoop.initialized())\n            sock, port = bind_unused_port()\n\n            def get_url(path):\n                return \"http://127.0.0.1:%d%s\" % (port, path)\n            # ensure that none of these processes live too long\n            signal.alarm(5)  # master process\n            try:\n                id = fork_processes(3, max_restarts=3)\n                self.assertTrue(id is not None)\n                signal.alarm(5)  # child processes\n            except SystemExit as e:\n                # if we exit cleanly from fork_processes, all the child processes\n                # finished with status 0\n                self.assertEqual(e.code, 0)\n                self.assertTrue(task_id() is None)\n                sock.close()\n                return\n            try:\n                if id in (0, 1):\n                    self.assertEqual(id, task_id())\n                    server = HTTPServer(self.get_app())\n                    server.add_sockets([sock])\n                    IOLoop.current().start()\n                elif id == 2:\n                    self.assertEqual(id, task_id())\n                    sock.close()\n                    # Always use SimpleAsyncHTTPClient here; the curl\n                    # version appears to get confused sometimes if the\n                    # connection gets closed before it's had a chance to\n                    # switch from writing mode to reading mode.\n                    client = HTTPClient(SimpleAsyncHTTPClient)\n\n                    def fetch(url, fail_ok=False):\n                        try:\n                            return client.fetch(get_url(url))\n                        except HTTPError as e:\n                            if not (fail_ok and e.code == 599):\n                                raise\n\n                    # Make two processes exit abnormally\n                    fetch(\"/?exit=2\", fail_ok=True)\n                    fetch(\"/?exit=3\", fail_ok=True)\n\n                    # They've been restarted, so a new fetch will work\n                    int(fetch(\"/\").body)\n\n                    # Now the same with signals\n                    # Disabled because on the mac a process dying with a signal\n                    # can trigger an \"Application exited abnormally; send error\n                    # report to Apple?\" prompt.\n                    # fetch(\"/?signal=%d\" % signal.SIGTERM, fail_ok=True)\n                    # fetch(\"/?signal=%d\" % signal.SIGABRT, fail_ok=True)\n                    # int(fetch(\"/\").body)\n\n                    # Now kill them normally so they won't be restarted\n                    fetch(\"/?exit=0\", fail_ok=True)\n                    # One process left; watch it's pid change\n                    pid = int(fetch(\"/\").body)\n                    fetch(\"/?exit=4\", fail_ok=True)\n                    pid2 = int(fetch(\"/\").body)\n                    self.assertNotEqual(pid, pid2)\n\n                    # Kill the last one so we shut down cleanly\n                    fetch(\"/?exit=0\", fail_ok=True)\n\n                    os._exit(0)\n            except Exception:\n                logging.error(\"exception in child process %d\", id, exc_info=True)\n                raise\n\n\n@skipIfNonUnix\nclass SubprocessTest(AsyncTestCase):\n    def test_subprocess(self):\n        if IOLoop.configured_class().__name__.endswith('LayeredTwistedIOLoop'):\n            # This test fails non-deterministically with LayeredTwistedIOLoop.\n            # (the read_until('\\n') returns '\\n' instead of 'hello\\n')\n            # This probably indicates a problem with either TornadoReactor\n            # or TwistedIOLoop, but I haven't been able to track it down\n            # and for now this is just causing spurious travis-ci failures.\n            raise unittest.SkipTest(\"Subprocess tests not compatible with \"\n                                    \"LayeredTwistedIOLoop\")\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.write(b\"print('hello')\\n\")\n        subproc.stdout.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"hello\\n\")\n\n        subproc.stdout.read_until(b\">>> \", self.stop)\n        self.wait()\n        subproc.stdin.write(b\"raise SystemExit\\n\")\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\")\n\n    def test_close_stdin(self):\n        # Close the parent's stdin handle and see that the child recognizes it.\n        subproc = Subprocess([sys.executable, '-u', '-i'],\n                             stdin=Subprocess.STREAM,\n                             stdout=Subprocess.STREAM, stderr=subprocess.STDOUT,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stdout.read_until(b'>>> ', self.stop)\n        self.wait()\n        subproc.stdin.close()\n        subproc.stdout.read_until_close(self.stop)\n        data = self.wait()\n        self.assertEqual(data, b\"\\n\")\n\n    def test_stderr(self):\n        subproc = Subprocess([sys.executable, '-u', '-c',\n                              r\"import sys; sys.stderr.write('hello\\n')\"],\n                             stderr=Subprocess.STREAM,\n                             io_loop=self.io_loop)\n        self.addCleanup(lambda: os.kill(subproc.pid, signal.SIGTERM))\n        subproc.stderr.read_until(b'\\n', self.stop)\n        data = self.wait()\n        self.assertEqual(data, b'hello\\n')\n\n    def test_sigchild(self):\n        # Twisted's SIGCHLD handler and Subprocess's conflict with each other.\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        ret = self.wait()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    @gen_test\n    def test_sigchild_future(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'pass'])\n        ret = yield subproc.wait_for_exit()\n        self.assertEqual(ret, 0)\n        self.assertEqual(subproc.returncode, ret)\n\n    def test_sigchild_signal(self):\n        skip_if_twisted()\n        Subprocess.initialize(io_loop=self.io_loop)\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c',\n                              'import time; time.sleep(30)'],\n                             io_loop=self.io_loop)\n        subproc.set_exit_callback(self.stop)\n        os.kill(subproc.pid, signal.SIGTERM)\n        ret = self.wait()\n        self.assertEqual(subproc.returncode, ret)\n        self.assertEqual(ret, -signal.SIGTERM)\n\n    @gen_test\n    def test_wait_for_exit_raise(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        with self.assertRaises(subprocess.CalledProcessError) as cm:\n            yield subproc.wait_for_exit()\n        self.assertEqual(cm.exception.returncode, 1)\n\n    @gen_test\n    def test_wait_for_exit_raise_disabled(self):\n        skip_if_twisted()\n        Subprocess.initialize()\n        self.addCleanup(Subprocess.uninitialize)\n        subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])\n        ret = yield subproc.wait_for_exit(raise_error=False)\n        self.assertEqual(ret, 1)\n",
        "#!/usr/bin/env python\nimport unittest\nimport sys\nimport time\nimport traceback\nimport platform\nimport os\n\n\nclass PytestLikeResult(unittest.TestResult):\n    def __init__(self, *args, **kwargs):\n        super().__init__(*args, **kwargs)\n        self.start_time = None\n        # Store (test, outcome, info) for each test:\n        #   outcome ∈ { \"passed\", \"failed\", \"error\", \"skipped\" }\n        #   info can hold the exception or skip reason, etc.\n        self.results = []\n\n    def startTestRun(self):\n        \"\"\"Called once before all tests.\"\"\"\n        self.start_time = time.time()\n        # Print a header similar to pytest\n        print(\"=\" * 29 + \" test session starts \" + \"=\" * 29)\n        # You can customize the platform/versions/paths as you like:\n        print(\n            f\"platform {platform.system().lower()} -- Python {platform.python_version()}\"\n        )\n        print(f\"rootdir: {os.path.abspath(os.curdir)}\")\n\n    def stopTestRun(self):\n        \"\"\"Called once after all tests are done.\"\"\"\n        elapsed = time.time() - self.start_time\n        print()  # blank line after final dot\n\n        # Summarize\n        ntests = len(self.results)\n        print(f\"collected {ntests} items\")\n\n        # Show details of failures/errors\n        failures = [r for r in self.results if r[1] == \"failed\"]\n        errors = [r for r in self.results if r[1] == \"error\"]\n\n        if failures:\n            print(\"\\n\" + \"=\" * 35 + \" FAILURES \" + \"=\" * 35)\n            for test, outcome, err_info in failures:\n                test = test.split(\"::\")\n                if len(test) == 3:\n                    test = f\"{test[1]}.{test[2]}\"\n                else:\n                    test = f\"{test[1]}\"\n                print(f\"__________ {test} __________\")\n                self._print_traceback(err_info)\n        if errors:\n            print(\"\\n\" + \"=\" * 35 + \" ERRORS \" + \"=\" * 37)\n            for test, outcome, err_info in errors:\n                test = test.split(\"::\")\n                if len(test) == 3:\n                    test = f\"{test[1]}.{test[2]}\"\n                else:\n                    test = f\"{test[1]}\"\n                print(f\"__________ {test} __________\")\n                self._print_traceback(err_info)\n\n        # Optional short summary lines for passes/fails/skips\n        print(\"\\n==================== short test summary info ====================\")\n        for test, outcome, info in self.results:\n            if outcome in (\"passed\", \"failed\", \"error\", \"skipped\"):\n                print(f\"{outcome.upper()} {test}\")\n\n        # Final one-line summary: “1 failed, 1 errors, 10 passed in 2.04s”\n        passed_count = sum(1 for r in self.results if r[1] == \"passed\")\n        failed_count = len(failures)\n        error_count = len(errors)\n        skipped_count = sum(1 for r in self.results if r[1] == \"skipped\")\n\n        summary_bits = []\n        if failed_count:\n            summary_bits.append(f\"{failed_count} failed\")\n        if error_count:\n            summary_bits.append(f\"{error_count} error{'s' if error_count != 1 else ''}\")\n        if passed_count:\n            summary_bits.append(f\"{passed_count} passed\")\n        if skipped_count:\n            summary_bits.append(f\"{skipped_count} skipped\")\n\n        summary_str = \", \".join(summary_bits) if summary_bits else \"no tests run\"\n        print(\n            f\"=================== {summary_str} in {elapsed:.2f}s ===================\"\n        )\n\n    def startTest(self, test):\n        \"\"\"Called right before each test method.\"\"\"\n        super().startTest(test)\n        # Print a dot or some indicator\n        print(\".\", end=\"\", flush=True)\n\n    def addSuccess(self, test):\n        super().addSuccess(test)\n        self.results.append((self._test_id(test), \"passed\", None))\n\n    def addFailure(self, test, err):\n        super().addFailure(test, err)\n        self.results.append((self._test_id(test), \"failed\", err))\n\n    def addError(self, test, err):\n        super().addError(test, err)\n        self.results.append((self._test_id(test), \"error\", err))\n\n    def addSkip(self, test, reason):\n        super().addSkip(test, reason)\n        self.results.append((self._test_id(test), \"skipped\", reason))\n\n    def _print_traceback(self, err_info):\n        \"\"\"Helper to print the traceback like pytest does.\"\"\"\n        if isinstance(err_info, tuple) and len(err_info) == 3:\n            exc_type, exc_value, tb = err_info\n            traceback.print_exception(exc_type, exc_value, tb, file=sys.stdout)\n        else:\n            print(str(err_info))\n\n    def _test_id(self, test):\n        try:\n            return f\"{test.__class__.__module__}::{test.__class__.__name__}::{test._testMethodName}\"\n        except AttributeError:\n            # Fallback for _ErrorHolder objects\n            return f\"{test.__class__.__module__}::{test.__class__.__name__}\"\n\n\nclass PytestLikeRunner(unittest.TextTestRunner):\n    \"\"\"A custom TextTestRunner that uses the PytestLikeResult.\"\"\"\n\n    resultclass = PytestLikeResult\n\n    def run(self, test):\n        result = self._makeResult()\n        # Overriding run to ensure we call startTestRun/stopTestRun\n        result.startTestRun()\n        test(result)\n        result.stopTestRun()\n        return result\n\n    def tearDown(self):\n        # Reset signal handlers\n        from twisted.internet import reactor\n\n        reactor.removeAll()\n        reactor.stop()\n        super().tearDown()\n\n\ndef main():\n    loader = unittest.TestLoader()\n    suite = loader.discover(\".\", pattern=\"test_*.py\")  # discover tests in r2e_tests/\n    runner = PytestLikeRunner(verbosity=0)\n    result = runner.run(suite)\n    # Exit with code 0 if all tests passed/skipped, 1 if there were fails/errors\n    sys.exit(not result.wasSuccessful())\n\n\nif __name__ == \"__main__\":\n    main()\n"
    ],
    "test_file_names": [
        "test_1.py",
        "tornado_unittest_runner.py"
    ],
    "setup_res_code": 0,
    "setup_res_stdout": "",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 1 package in 786ms\nPrepared 1 package in 1.14s\nInstalled 1 package in 10ms\n + tornado==4.2.dev1 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0)\nResolved 29 packages in 16ms\nInstalled 29 packages in 125ms\n + attrs==24.3.0\n + automat==24.8.1\n + cachetools==5.5.0\n + cffi==1.17.1\n + chardet==5.2.0\n + colorama==0.4.6\n + constantly==23.10.4\n + distlib==0.3.9\n + exceptiongroup==1.2.2\n + filelock==3.16.1\n + hyperlink==21.0.0\n + idna==3.10\n + incremental==24.7.2\n + iniconfig==2.0.0\n + packaging==24.2\n + platformdirs==4.3.6\n + pluggy==1.5.0\n + pycares==4.5.0\n + pycparser==2.22\n + pycurl==7.45.4\n + pyproject-api==1.8.0\n + pytest==8.3.4\n + setuptools==75.6.0\n + tomli==2.2.1\n + tox==4.23.2\n + twisted==24.11.0\n + typing-extensions==4.12.2\n + virtualenv==20.28.0\n + zope-interface==7.2\n",
    "new_commit_res_code": 0,
    "new_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0\n.........\ncollected 9 items\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::ProcessTest::test_multi_process\nPASSED r2e_tests.test_1::SubprocessTest::test_close_stdin\nPASSED r2e_tests.test_1::SubprocessTest::test_sigchild\nPASSED r2e_tests.test_1::SubprocessTest::test_sigchild_future\nPASSED r2e_tests.test_1::SubprocessTest::test_sigchild_signal\nPASSED r2e_tests.test_1::SubprocessTest::test_stderr\nPASSED r2e_tests.test_1::SubprocessTest::test_subprocess\nPASSED r2e_tests.test_1::SubprocessTest::test_wait_for_exit_raise\nPASSED r2e_tests.test_1::SubprocessTest::test_wait_for_exit_raise_disabled\n=================== 9 passed in 0.29s ===================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0\n.........\ncollected 9 items\n\n=================================== FAILURES ===================================\n__________ SubprocessTest.test_wait_for_exit_raise __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 592, in run\n    self._callTestMethod(testMethod)\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 550, in _callTestMethod\n    method()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/testing.py\", line 120, in __call__\n    result = self.orig_method(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/testing.py\", line 499, in post_coroutine\n    return self.io_loop.run_sync(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/ioloop.py\", line 445, in run_sync\n    return future_cell[0].result()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/concurrent.py\", line 211, in result\n    raise_exc_info(self._exc_info)\n  File \"<string>\", line 3, in raise_exc_info\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/gen.py\", line 884, in run\n    yielded = self.gen.send(value)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/r2e_tests/test_1.py\", line 233, in test_wait_for_exit_raise\n    yield subproc.wait_for_exit()\nAssertionError: CalledProcessError not raised\n\n=================================== ERRORS =====================================\n__________ SubprocessTest.test_wait_for_exit_raise_disabled __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 592, in run\n    self._callTestMethod(testMethod)\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 550, in _callTestMethod\n    method()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/testing.py\", line 120, in __call__\n    result = self.orig_method(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/testing.py\", line 499, in post_coroutine\n    return self.io_loop.run_sync(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/ioloop.py\", line 445, in run_sync\n    return future_cell[0].result()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/concurrent.py\", line 211, in result\n    raise_exc_info(self._exc_info)\n  File \"<string>\", line 3, in raise_exc_info\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/tornado/gen.py\", line 230, in wrapper\n    yielded = next(result)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_2bd2f0db087c12738ac55aba75fe78fe31987dd0/r2e_tests/test_1.py\", line 242, in test_wait_for_exit_raise_disabled\n    ret = yield subproc.wait_for_exit(raise_error=False)\nTypeError: wait_for_exit() got an unexpected keyword argument 'raise_error'\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::ProcessTest::test_multi_process\nPASSED r2e_tests.test_1::SubprocessTest::test_close_stdin\nPASSED r2e_tests.test_1::SubprocessTest::test_sigchild\nPASSED r2e_tests.test_1::SubprocessTest::test_sigchild_future\nPASSED r2e_tests.test_1::SubprocessTest::test_sigchild_signal\nPASSED r2e_tests.test_1::SubprocessTest::test_stderr\nPASSED r2e_tests.test_1::SubprocessTest::test_subprocess\nFAILED r2e_tests.test_1::SubprocessTest::test_wait_for_exit_raise\nERROR r2e_tests.test_1::SubprocessTest::test_wait_for_exit_raise_disabled\n=================== 1 failed, 1 error, 7 passed in 0.20s ===================\n",
    "old_commit_res_stderr": ""
}
tornado/process.py tornado/test/process_test.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** `Subprocess.wait_for_exit()` Doesn’t Raise Exception on Non-Zero Exit and Lacks `raise_error` Parameter

**Description:**
When a subprocess exits with a non-zero status, the `wait_for_exit()` method does not raise a `CalledProcessError` as expected. Additionally, attempting to disable exception raising by passing the `raise_error=False` argument results in a `TypeError` since the method does not accept this parameter.

**Example Code:**
```python
subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])

# This should raise CalledProcessError but does not
yield subproc.wait_for_exit()

# This should return the exit status without raising an error
ret = yield subproc.wait_for_exit(raise_error=False)
```

**Expected Behavior:**
- Calling `wait_for_exit()` should raise a `CalledProcessError` when the subprocess exits with a non-zero status.
- The method should accept a `raise_error` parameter that allows developers to disable the exception raising and instead receive the exit status directly.

**Actual Behavior:**
- No exception is raised when the subprocess exits with a non-zero status, causing tests that expect an error to fail.
- Passing `raise_error=False` results in a `TypeError` because the `wait_for_exit()` method does not recognize the `raise_error` argument.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### tornado/process.py
#!/usr/bin/env python
#
# Copyright 2011 Facebook
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

"""Utilities for working with multiple processes, including both forking
the server into multiple processes and managing subprocesses.
"""

from __future__ import absolute_import, division, print_function, with_statement

import errno
import os
import signal
import subprocess
import sys
import time

from binascii import hexlify

from tornado.concurrent import Future
from tornado import ioloop
from tornado.iostream import PipeIOStream
from tornado.log import gen_log
from tornado.platform.auto import set_close_exec
from tornado import stack_context
from tornado.util import errno_from_exception

try:
    import multiprocessing
except ImportError:
    # Multiprocessing is not available on Google App Engine.
    multiprocessing = None

try:
    long  # py2
except NameError:
    long = int  # py3


def cpu_count():
    """Returns the number of processors on this machine."""
    if multiprocessing is None:
        return 1
    try:
        return multiprocessing.cpu_count()
    except NotImplementedError:
        pass
    try:
        return os.sysconf("SC_NPROCESSORS_CONF")
    except ValueError:
        pass
    gen_log.error("Could not detect number of processors; assuming 1")
    return 1


def _reseed_random():
    if 'random' not in sys.modules:
        return
    import random
    # If os.urandom is available, this method does the same thing as
    # random.seed (at least as of python 2.6).  If os.urandom is not
    # available, we mix in the pid in addition to a timestamp.
    try:
        seed = long(hexlify(os.urandom(16)), 16)
    except NotImplementedError:
        seed = int(time.time() * 1000) ^ os.getpid()
    random.seed(seed)


def _pipe_cloexec():
    r, w = os.pipe()
    set_close_exec(r)
    set_close_exec(w)
    return r, w


_task_id = None


def fork_processes(num_processes, max_restarts=100):
    """Starts multiple worker processes.

    If ``num_processes`` is None or <= 0, we detect the number of cores
    available on this machine and fork that number of child
    processes. If ``num_processes`` is given and > 0, we fork that
    specific number of sub-processes.

    Since we use processes and not threads, there is no shared memory
    between any server code.

    Note that multiple processes are not compatible with the autoreload
    module (or the ``autoreload=True`` option to `tornado.web.Application`
    which defaults to True when ``debug=True``).
    When using multiple processes, no IOLoops can be created or
    referenced until after the call to ``fork_processes``.

    In each child process, ``fork_processes`` returns its *task id*, a
    number between 0 and ``num_processes``.  Processes that exit
    abnormally (due to a signal or non-zero exit status) are restarted
    with the same id (up to ``max_restarts`` times).  In the parent
    process, ``fork_processes`` returns None if all child processes
    have exited normally, but will otherwise only exit by throwing an
    exception.
    """
    global _task_id
    assert _task_id is None
    if num_processes is None or num_processes <= 0:
        num_processes = cpu_count()
    if ioloop.IOLoop.initialized():
        raise RuntimeError("Cannot run in multiple processes: IOLoop instance "
                           "has already been initialized. You cannot call "
                           "IOLoop.instance() before calling start_processes()")
    gen_log.info("Starting %d processes", num_processes)
    children = {}

    def start_child(i):
        pid = os.fork()
        if pid == 0:
            # child process
            _reseed_random()
            global _task_id
            _task_id = i
            return i
        else:
            children[pid] = i
            return None
    for i in range(num_processes):
        id = start_child(i)
        if id is not None:
            return id
    num_restarts = 0
    while children:
        try:
            pid, status = os.wait()
        except OSError as e:
            if errno_from_exception(e) == errno.EINTR:
                continue
            raise
        if pid not in children:
            continue
        id = children.pop(pid)
        if os.WIFSIGNALED(status):
            gen_log.warning("child %d (pid %d) killed by signal %d, restarting",
                            id, pid, os.WTERMSIG(status))
        elif os.WEXITSTATUS(status) != 0:
            gen_log.warning("child %d (pid %d) exited with status %d, restarting",
                            id, pid, os.WEXITSTATUS(status))
        else:
            gen_log.info("child %d (pid %d) exited normally", id, pid)
            continue
        num_restarts += 1
        if num_restarts > max_restarts:
            raise RuntimeError("Too many child restarts, giving up")
        new_id = start_child(id)
        if new_id is not None:
            return new_id
    # All child processes exited cleanly, so exit the master process
    # instead of just returning to right after the call to
    # fork_processes (which will probably just start up another IOLoop
    # unless the caller checks the return value).
    sys.exit(0)


def task_id():
    """Returns the current task id, if any.

    Returns None if this process was not created by `fork_processes`.
    """
    global _task_id
    return _task_id


class Subprocess(object):
    """Wraps ``subprocess.Popen`` with IOStream support.

    The constructor is the same as ``subprocess.Popen`` with the following
    additions:

    * ``stdin``, ``stdout``, and ``stderr`` may have the value
      ``tornado.process.Subprocess.STREAM``, which will make the corresponding
      attribute of the resulting Subprocess a `.PipeIOStream`.
    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.

    .. versionchanged:: 4.1
       The ``io_loop`` argument is deprecated.
    """
    STREAM = object()

    _initialized = False
    _waiting = {}

    def __init__(self, *args, **kwargs):
        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()
        # All FDs we create should be closed on error; those in to_close
        # should be closed in the parent process on success.
        pipe_fds = []
        to_close = []
        if kwargs.get('stdin') is Subprocess.STREAM:
            in_r, in_w = _pipe_cloexec()
            kwargs['stdin'] = in_r
            pipe_fds.extend((in_r, in_w))
            to_close.append(in_r)
            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)
        if kwargs.get('stdout') is Subprocess.STREAM:
            out_r, out_w = _pipe_cloexec()
            kwargs['stdout'] = out_w
            pipe_fds.extend((out_r, out_w))
            to_close.append(out_w)
            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)
        if kwargs.get('stderr') is Subprocess.STREAM:
            err_r, err_w = _pipe_cloexec()
            kwargs['stderr'] = err_w
            pipe_fds.extend((err_r, err_w))
            to_close.append(err_w)
            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)
        try:
            self.proc = subprocess.Popen(*args, **kwargs)
        except:
            for fd in pipe_fds:
                os.close(fd)
            raise
        for fd in to_close:
            os.close(fd)
        for attr in ['stdin', 'stdout', 'stderr', 'pid']:
            if not hasattr(self, attr):  # don't clobber streams set above
                setattr(self, attr, getattr(self.proc, attr))
        self._exit_callback = None
        self.returncode = None

    def set_exit_callback(self, callback):
        """Runs ``callback`` when this process exits.

        The callback takes one argument, the return code of the process.

        This method uses a ``SIGCHLD`` handler, which is a global setting
        and may conflict if you have other libraries trying to handle the
        same signal.  If you are using more than one ``IOLoop`` it may
        be necessary to call `Subprocess.initialize` first to designate
        one ``IOLoop`` to run the signal handlers.

        In many cases a close callback on the stdout or stderr streams
        can be used as an alternative to an exit callback if the
        signal handler is causing a problem.
        """
        self._exit_callback = stack_context.wrap(callback)
        Subprocess.initialize(self.io_loop)
        Subprocess._waiting[self.pid] = self
        Subprocess._try_cleanup_process(self.pid)

    def wait_for_exit(self):
        """Returns a `.Future` which resolves when the process exits.

        Usage::

            ret = yield proc.wait_for_exit()

        This is a coroutine-friendly alternative to `set_exit_callback`
        (and a replacement for the blocking `subprocess.Popen.wait`).

        .. versionadded:: 4.2
        """
        future = Future()
        self.set_exit_callback(future.set_result)
        return future

    @classmethod
    def initialize(cls, io_loop=None):
        """Initializes the ``SIGCHLD`` handler.

        The signal handler is run on an `.IOLoop` to avoid locking issues.
        Note that the `.IOLoop` used for signal handling need not be the
        same one used by individual Subprocess objects (as long as the
        ``IOLoops`` are each running in separate threads).

        .. versionchanged:: 4.1
           The ``io_loop`` argument is deprecated.
        """
        if cls._initialized:
            return
        if io_loop is None:
            io_loop = ioloop.IOLoop.current()
        cls._old_sigchld = signal.signal(
            signal.SIGCHLD,
            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))
        cls._initialized = True

    @classmethod
    def uninitialize(cls):
        """Removes the ``SIGCHLD`` handler."""
        if not cls._initialized:
            return
        signal.signal(signal.SIGCHLD, cls._old_sigchld)
        cls._initialized = False

    @classmethod
    def _cleanup(cls):
        for pid in list(cls._waiting.keys()):  # make a copy
            cls._try_cleanup_process(pid)

    @classmethod
    def _try_cleanup_process(cls, pid):
        try:
            ret_pid, status = os.waitpid(pid, os.WNOHANG)
        except OSError as e:
            if errno_from_exception(e) == errno.ECHILD:
                return
        if ret_pid == 0:
            return
        assert ret_pid == pid
        subproc = cls._waiting.pop(pid)
        subproc.io_loop.add_callback_from_signal(
            subproc._set_returncode, status)

    def _set_returncode(self, status):
        if os.WIFSIGNALED(status):
            self.returncode = -os.WTERMSIG(status)
        else:
            assert os.WIFEXITED(status)
            self.returncode = os.WEXITSTATUS(status)
        if self._exit_callback:
            callback = self._exit_callback
            self._exit_callback = None
            callback(self.returncode)

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
[ISSUE]
**Title:** `Subprocess.wait_for_exit()` Doesn’t Raise Exception on Non-Zero Exit and Lacks `raise_error` Parameter

**Description:**
When a subprocess exits with a non-zero status, the `wait_for_exit()` method does not raise a `CalledProcessError` as expected. Additionally, attempting to disable exception raising by passing the `raise_error=False` argument results in a `TypeError` since the method does not accept this parameter.

**Example Code:**
```python
subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])

# This should raise CalledProcessError but does not
yield subproc.wait_for_exit()

# This should return the exit status without raising an error
ret = yield subproc.wait_for_exit(raise_error=False)
```

**Expected Behavior:**
- Calling `wait_for_exit()` should raise a `CalledProcessError` when the subprocess exits with a non-zero status.
- The method should accept a `raise_error` parameter that allows developers to disable the exception raising and instead receive the exit status directly.

**Actual Behavior:**
- No exception is raised when the subprocess exits with a non-zero status, causing tests that expect an error to fail.
- Passing `raise_error=False` results in a `TypeError` because the `wait_for_exit()` method does not recognize the `raise_error` argument.

[/ISSUE]
{
    "ProcessTest.test_multi_process": "PASSED",
    "SubprocessTest.test_close_stdin": "PASSED",
    "SubprocessTest.test_sigchild": "PASSED",
    "SubprocessTest.test_sigchild_future": "PASSED",
    "SubprocessTest.test_sigchild_signal": "PASSED",
    "SubprocessTest.test_stderr": "PASSED",
    "SubprocessTest.test_subprocess": "PASSED",
    "SubprocessTest.test_wait_for_exit_raise": "PASSED",
    "SubprocessTest.test_wait_for_exit_raise_disabled": "PASSED"
}
diff --git a/tornado/process.py b/tornado/process.py
index fb9254fc..f580e192 100644
--- a/tornado/process.py
+++ b/tornado/process.py
@@ -49,6 +49,10 @@ except NameError:
     long = int  # py3
 
 
+# Re-export this exception for convenience.
+CalledProcessError = subprocess.CalledProcessError
+
+
 def cpu_count():
     """Returns the number of processors on this machine."""
     if multiprocessing is None:
@@ -259,7 +263,7 @@ class Subprocess(object):
         Subprocess._waiting[self.pid] = self
         Subprocess._try_cleanup_process(self.pid)
 
-    def wait_for_exit(self):
+    def wait_for_exit(self, raise_error=True):
         """Returns a `.Future` which resolves when the process exits.
 
         Usage::
@@ -269,10 +273,21 @@ class Subprocess(object):
         This is a coroutine-friendly alternative to `set_exit_callback`
         (and a replacement for the blocking `subprocess.Popen.wait`).
 
+        By default, raises `subprocess.CalledProcessError` if the process
+        has a non-zero exit status. Use ``wait_for_exit(raise_error=False)``
+        to suppress this behavior and return the exit status without raising.
+
         .. versionadded:: 4.2
         """
         future = Future()
-        self.set_exit_callback(future.set_result)
+
+        def callback(ret):
+            if ret != 0 and raise_error:
+                # Unfortunately we don't have the original args any more.
+                future.set_exception(CalledProcessError(ret, None))
+            else:
+                future.set_result(ret)
+        self.set_exit_callback(callback)
         return future
 
     @classmethod

{"tornado/process.py": "#!/usr/bin/env python\n#\n# Copyright 2011 Facebook\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\n\"\"\"Utilities for working with multiple processes, including both forking\nthe server into multiple processes and managing subprocesses.\n\"\"\"\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\nimport errno\nimport os\nimport signal\nimport subprocess\nimport sys\nimport time\n\nfrom binascii import hexlify\n\nfrom tornado.concurrent import Future\nfrom tornado import ioloop\nfrom tornado.iostream import PipeIOStream\nfrom tornado.log import gen_log\nfrom tornado.platform.auto import set_close_exec\nfrom tornado import stack_context\nfrom tornado.util import errno_from_exception\n\ntry:\n    import multiprocessing\nexcept ImportError:\n    # Multiprocessing is not available on Google App Engine.\n    multiprocessing = None\n\ntry:\n    long  # py2\nexcept NameError:\n    long = int  # py3\n\n\ndef cpu_count():\n    \"\"\"Returns the number of processors on this machine.\"\"\"\n    if multiprocessing is None:\n        return 1\n    try:\n        return multiprocessing.cpu_count()\n    except NotImplementedError:\n        pass\n    try:\n        return os.sysconf(\"SC_NPROCESSORS_CONF\")\n    except ValueError:\n        pass\n    gen_log.error(\"Could not detect number of processors; assuming 1\")\n    return 1\n\n\ndef _reseed_random():\n    if 'random' not in sys.modules:\n        return\n    import random\n    # If os.urandom is available, this method does the same thing as\n    # random.seed (at least as of python 2.6).  If os.urandom is not\n    # available, we mix in the pid in addition to a timestamp.\n    try:\n        seed = long(hexlify(os.urandom(16)), 16)\n    except NotImplementedError:\n        seed = int(time.time() * 1000) ^ os.getpid()\n    random.seed(seed)\n\n\ndef _pipe_cloexec():\n    r, w = os.pipe()\n    set_close_exec(r)\n    set_close_exec(w)\n    return r, w\n\n\n_task_id = None\n\n\ndef fork_processes(num_processes, max_restarts=100):\n    \"\"\"Starts multiple worker processes.\n\n    If ``num_processes`` is None or <= 0, we detect the number of cores\n    available on this machine and fork that number of child\n    processes. If ``num_processes`` is given and > 0, we fork that\n    specific number of sub-processes.\n\n    Since we use processes and not threads, there is no shared memory\n    between any server code.\n\n    Note that multiple processes are not compatible with the autoreload\n    module (or the ``autoreload=True`` option to `tornado.web.Application`\n    which defaults to True when ``debug=True``).\n    When using multiple processes, no IOLoops can be created or\n    referenced until after the call to ``fork_processes``.\n\n    In each child process, ``fork_processes`` returns its *task id*, a\n    number between 0 and ``num_processes``.  Processes that exit\n    abnormally (due to a signal or non-zero exit status) are restarted\n    with the same id (up to ``max_restarts`` times).  In the parent\n    process, ``fork_processes`` returns None if all child processes\n    have exited normally, but will otherwise only exit by throwing an\n    exception.\n    \"\"\"\n    global _task_id\n    assert _task_id is None\n    if num_processes is None or num_processes <= 0:\n        num_processes = cpu_count()\n    if ioloop.IOLoop.initialized():\n        raise RuntimeError(\"Cannot run in multiple processes: IOLoop instance \"\n                           \"has already been initialized. You cannot call \"\n                           \"IOLoop.instance() before calling start_processes()\")\n    gen_log.info(\"Starting %d processes\", num_processes)\n    children = {}\n\n    def start_child(i):\n        pid = os.fork()\n        if pid == 0:\n            # child process\n            _reseed_random()\n            global _task_id\n            _task_id = i\n            return i\n        else:\n            children[pid] = i\n            return None\n    for i in range(num_processes):\n        id = start_child(i)\n        if id is not None:\n            return id\n    num_restarts = 0\n    while children:\n        try:\n            pid, status = os.wait()\n        except OSError as e:\n            if errno_from_exception(e) == errno.EINTR:\n                continue\n            raise\n        if pid not in children:\n            continue\n        id = children.pop(pid)\n        if os.WIFSIGNALED(status):\n            gen_log.warning(\"child %d (pid %d) killed by signal %d, restarting\",\n                            id, pid, os.WTERMSIG(status))\n        elif os.WEXITSTATUS(status) != 0:\n            gen_log.warning(\"child %d (pid %d) exited with status %d, restarting\",\n                            id, pid, os.WEXITSTATUS(status))\n        else:\n            gen_log.info(\"child %d (pid %d) exited normally\", id, pid)\n            continue\n        num_restarts += 1\n        if num_restarts > max_restarts:\n            raise RuntimeError(\"Too many child restarts, giving up\")\n        new_id = start_child(id)\n        if new_id is not None:\n            return new_id\n    # All child processes exited cleanly, so exit the master process\n    # instead of just returning to right after the call to\n    # fork_processes (which will probably just start up another IOLoop\n    # unless the caller checks the return value).\n    sys.exit(0)\n\n\ndef task_id():\n    \"\"\"Returns the current task id, if any.\n\n    Returns None if this process was not created by `fork_processes`.\n    \"\"\"\n    global _task_id\n    return _task_id\n\n\nclass Subprocess(object):\n    \"\"\"Wraps ``subprocess.Popen`` with IOStream support.\n\n    The constructor is the same as ``subprocess.Popen`` with the following\n    additions:\n\n    * ``stdin``, ``stdout``, and ``stderr`` may have the value\n      ``tornado.process.Subprocess.STREAM``, which will make the corresponding\n      attribute of the resulting Subprocess a `.PipeIOStream`.\n    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.\n\n    .. versionchanged:: 4.1\n       The ``io_loop`` argument is deprecated.\n    \"\"\"\n    STREAM = object()\n\n    _initialized = False\n    _waiting = {}\n\n    def __init__(self, *args, **kwargs):\n        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()\n        # All FDs we create should be closed on error; those in to_close\n        # should be closed in the parent process on success.\n        pipe_fds = []\n        to_close = []\n        if kwargs.get('stdin') is Subprocess.STREAM:\n            in_r, in_w = _pipe_cloexec()\n            kwargs['stdin'] = in_r\n            pipe_fds.extend((in_r, in_w))\n            to_close.append(in_r)\n            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)\n        if kwargs.get('stdout') is Subprocess.STREAM:\n            out_r, out_w = _pipe_cloexec()\n            kwargs['stdout'] = out_w\n            pipe_fds.extend((out_r, out_w))\n            to_close.append(out_w)\n            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)\n        if kwargs.get('stderr') is Subprocess.STREAM:\n            err_r, err_w = _pipe_cloexec()\n            kwargs['stderr'] = err_w\n            pipe_fds.extend((err_r, err_w))\n            to_close.append(err_w)\n            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)\n        try:\n            self.proc = subprocess.Popen(*args, **kwargs)\n        except:\n            for fd in pipe_fds:\n                os.close(fd)\n            raise\n        for fd in to_close:\n            os.close(fd)\n        for attr in ['stdin', 'stdout', 'stderr', 'pid']:\n            if not hasattr(self, attr):  # don't clobber streams set above\n                setattr(self, attr, getattr(self.proc, attr))\n        self._exit_callback = None\n        self.returncode = None\n\n    def set_exit_callback(self, callback):\n        \"\"\"Runs ``callback`` when this process exits.\n\n        The callback takes one argument, the return code of the process.\n\n        This method uses a ``SIGCHLD`` handler, which is a global setting\n        and may conflict if you have other libraries trying to handle the\n        same signal.  If you are using more than one ``IOLoop`` it may\n        be necessary to call `Subprocess.initialize` first to designate\n        one ``IOLoop`` to run the signal handlers.\n\n        In many cases a close callback on the stdout or stderr streams\n        can be used as an alternative to an exit callback if the\n        signal handler is causing a problem.\n        \"\"\"\n        self._exit_callback = stack_context.wrap(callback)\n        Subprocess.initialize(self.io_loop)\n        Subprocess._waiting[self.pid] = self\n        Subprocess._try_cleanup_process(self.pid)\n\n    def wait_for_exit(self):\n        \"\"\"Returns a `.Future` which resolves when the process exits.\n\n        Usage::\n\n            ret = yield proc.wait_for_exit()\n\n        This is a coroutine-friendly alternative to `set_exit_callback`\n        (and a replacement for the blocking `subprocess.Popen.wait`).\n\n        .. versionadded:: 4.2\n        \"\"\"\n        future = Future()\n        self.set_exit_callback(future.set_result)\n        return future\n\n    @classmethod\n    def initialize(cls, io_loop=None):\n        \"\"\"Initializes the ``SIGCHLD`` handler.\n\n        The signal handler is run on an `.IOLoop` to avoid locking issues.\n        Note that the `.IOLoop` used for signal handling need not be the\n        same one used by individual Subprocess objects (as long as the\n        ``IOLoops`` are each running in separate threads).\n\n        .. versionchanged:: 4.1\n           The ``io_loop`` argument is deprecated.\n        \"\"\"\n        if cls._initialized:\n            return\n        if io_loop is None:\n            io_loop = ioloop.IOLoop.current()\n        cls._old_sigchld = signal.signal(\n            signal.SIGCHLD,\n            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))\n        cls._initialized = True\n\n    @classmethod\n    def uninitialize(cls):\n        \"\"\"Removes the ``SIGCHLD`` handler.\"\"\"\n        if not cls._initialized:\n            return\n        signal.signal(signal.SIGCHLD, cls._old_sigchld)\n        cls._initialized = False\n\n    @classmethod\n    def _cleanup(cls):\n        for pid in list(cls._waiting.keys()):  # make a copy\n            cls._try_cleanup_process(pid)\n\n    @classmethod\n    def _try_cleanup_process(cls, pid):\n        try:\n            ret_pid, status = os.waitpid(pid, os.WNOHANG)\n        except OSError as e:\n            if errno_from_exception(e) == errno.ECHILD:\n                return\n        if ret_pid == 0:\n            return\n        assert ret_pid == pid\n        subproc = cls._waiting.pop(pid)\n        subproc.io_loop.add_callback_from_signal(\n            subproc._set_returncode, status)\n\n    def _set_returncode(self, status):\n        if os.WIFSIGNALED(status):\n            self.returncode = -os.WTERMSIG(status)\n        else:\n            assert os.WIFEXITED(status)\n            self.returncode = os.WEXITSTATUS(status)\n        if self._exit_callback:\n            callback = self._exit_callback\n            self._exit_callback = None\n            callback(self.returncode)\n"}
A user will ask you to solve a task. You should first draft your thinking process (inner monologue). Then, generate the solution.

Your response format must follow the template below:
<think>
Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate a correct solution.
</think>
<solution>
Final solution presented to the user.
</solution>
system
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** `Subprocess.wait_for_exit()` Doesn’t Raise Exception on Non-Zero Exit and Lacks `raise_error` Parameter

**Description:**
When a subprocess exits with a non-zero status, the `wait_for_exit()` method does not raise a `CalledProcessError` as expected. Additionally, attempting to disable exception raising by passing the `raise_error=False` argument results in a `TypeError` since the method does not accept this parameter.

**Example Code:**
```python
subproc = Subprocess([sys.executable, '-c', 'import sys; sys.exit(1)'])

# This should raise CalledProcessError but does not
yield subproc.wait_for_exit()

# This should return the exit status without raising an error
ret = yield subproc.wait_for_exit(raise_error=False)
```

**Expected Behavior:**
- Calling `wait_for_exit()` should raise a `CalledProcessError` when the subprocess exits with a non-zero status.
- The method should accept a `raise_error` parameter that allows developers to disable the exception raising and instead receive the exit status directly.

**Actual Behavior:**
- No exception is raised when the subprocess exits with a non-zero status, causing tests that expect an error to fail.
- Passing `raise_error=False` results in a `TypeError` because the `wait_for_exit()` method does not recognize the `raise_error` argument.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### tornado/process.py
#!/usr/bin/env python
#
# Copyright 2011 Facebook
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

"""Utilities for working with multiple processes, including both forking
the server into multiple processes and managing subprocesses.
"""

from __future__ import absolute_import, division, print_function, with_statement

import errno
import os
import signal
import subprocess
import sys
import time

from binascii import hexlify

from tornado.concurrent import Future
from tornado import ioloop
from tornado.iostream import PipeIOStream
from tornado.log import gen_log
from tornado.platform.auto import set_close_exec
from tornado import stack_context
from tornado.util import errno_from_exception

try:
    import multiprocessing
except ImportError:
    # Multiprocessing is not available on Google App Engine.
    multiprocessing = None

try:
    long  # py2
except NameError:
    long = int  # py3


def cpu_count():
    """Returns the number of processors on this machine."""
    if multiprocessing is None:
        return 1
    try:
        return multiprocessing.cpu_count()
    except NotImplementedError:
        pass
    try:
        return os.sysconf("SC_NPROCESSORS_CONF")
    except ValueError:
        pass
    gen_log.error("Could not detect number of processors; assuming 1")
    return 1


def _reseed_random():
    if 'random' not in sys.modules:
        return
    import random
    # If os.urandom is available, this method does the same thing as
    # random.seed (at least as of python 2.6).  If os.urandom is not
    # available, we mix in the pid in addition to a timestamp.
    try:
        seed = long(hexlify(os.urandom(16)), 16)
    except NotImplementedError:
        seed = int(time.time() * 1000) ^ os.getpid()
    random.seed(seed)


def _pipe_cloexec():
    r, w = os.pipe()
    set_close_exec(r)
    set_close_exec(w)
    return r, w


_task_id = None


def fork_processes(num_processes, max_restarts=100):
    """Starts multiple worker processes.

    If ``num_processes`` is None or <= 0, we detect the number of cores
    available on this machine and fork that number of child
    processes. If ``num_processes`` is given and > 0, we fork that
    specific number of sub-processes.

    Since we use processes and not threads, there is no shared memory
    between any server code.

    Note that multiple processes are not compatible with the autoreload
    module (or the ``autoreload=True`` option to `tornado.web.Application`
    which defaults to True when ``debug=True``).
    When using multiple processes, no IOLoops can be created or
    referenced until after the call to ``fork_processes``.

    In each child process, ``fork_processes`` returns its *task id*, a
    number between 0 and ``num_processes``.  Processes that exit
    abnormally (due to a signal or non-zero exit status) are restarted
    with the same id (up to ``max_restarts`` times).  In the parent
    process, ``fork_processes`` returns None if all child processes
    have exited normally, but will otherwise only exit by throwing an
    exception.
    """
    global _task_id
    assert _task_id is None
    if num_processes is None or num_processes <= 0:
        num_processes = cpu_count()
    if ioloop.IOLoop.initialized():
        raise RuntimeError("Cannot run in multiple processes: IOLoop instance "
                           "has already been initialized. You cannot call "
                           "IOLoop.instance() before calling start_processes()")
    gen_log.info("Starting %d processes", num_processes)
    children = {}

    def start_child(i):
        pid = os.fork()
        if pid == 0:
            # child process
            _reseed_random()
            global _task_id
            _task_id = i
            return i
        else:
            children[pid] = i
            return None
    for i in range(num_processes):
        id = start_child(i)
        if id is not None:
            return id
    num_restarts = 0
    while children:
        try:
            pid, status = os.wait()
        except OSError as e:
            if errno_from_exception(e) == errno.EINTR:
                continue
            raise
        if pid not in children:
            continue
        id = children.pop(pid)
        if os.WIFSIGNALED(status):
            gen_log.warning("child %d (pid %d) killed by signal %d, restarting",
                            id, pid, os.WTERMSIG(status))
        elif os.WEXITSTATUS(status) != 0:
            gen_log.warning("child %d (pid %d) exited with status %d, restarting",
                            id, pid, os.WEXITSTATUS(status))
        else:
            gen_log.info("child %d (pid %d) exited normally", id, pid)
            continue
        num_restarts += 1
        if num_restarts > max_restarts:
            raise RuntimeError("Too many child restarts, giving up")
        new_id = start_child(id)
        if new_id is not None:
            return new_id
    # All child processes exited cleanly, so exit the master process
    # instead of just returning to right after the call to
    # fork_processes (which will probably just start up another IOLoop
    # unless the caller checks the return value).
    sys.exit(0)


def task_id():
    """Returns the current task id, if any.

    Returns None if this process was not created by `fork_processes`.
    """
    global _task_id
    return _task_id


class Subprocess(object):
    """Wraps ``subprocess.Popen`` with IOStream support.

    The constructor is the same as ``subprocess.Popen`` with the following
    additions:

    * ``stdin``, ``stdout``, and ``stderr`` may have the value
      ``tornado.process.Subprocess.STREAM``, which will make the corresponding
      attribute of the resulting Subprocess a `.PipeIOStream`.
    * A new keyword argument ``io_loop`` may be used to pass in an IOLoop.

    .. versionchanged:: 4.1
       The ``io_loop`` argument is deprecated.
    """
    STREAM = object()

    _initialized = False
    _waiting = {}

    def __init__(self, *args, **kwargs):
        self.io_loop = kwargs.pop('io_loop', None) or ioloop.IOLoop.current()
        # All FDs we create should be closed on error; those in to_close
        # should be closed in the parent process on success.
        pipe_fds = []
        to_close = []
        if kwargs.get('stdin') is Subprocess.STREAM:
            in_r, in_w = _pipe_cloexec()
            kwargs['stdin'] = in_r
            pipe_fds.extend((in_r, in_w))
            to_close.append(in_r)
            self.stdin = PipeIOStream(in_w, io_loop=self.io_loop)
        if kwargs.get('stdout') is Subprocess.STREAM:
            out_r, out_w = _pipe_cloexec()
            kwargs['stdout'] = out_w
            pipe_fds.extend((out_r, out_w))
            to_close.append(out_w)
            self.stdout = PipeIOStream(out_r, io_loop=self.io_loop)
        if kwargs.get('stderr') is Subprocess.STREAM:
            err_r, err_w = _pipe_cloexec()
            kwargs['stderr'] = err_w
            pipe_fds.extend((err_r, err_w))
            to_close.append(err_w)
            self.stderr = PipeIOStream(err_r, io_loop=self.io_loop)
        try:
            self.proc = subprocess.Popen(*args, **kwargs)
        except:
            for fd in pipe_fds:
                os.close(fd)
            raise
        for fd in to_close:
            os.close(fd)
        for attr in ['stdin', 'stdout', 'stderr', 'pid']:
            if not hasattr(self, attr):  # don't clobber streams set above
                setattr(self, attr, getattr(self.proc, attr))
        self._exit_callback = None
        self.returncode = None

    def set_exit_callback(self, callback):
        """Runs ``callback`` when this process exits.

        The callback takes one argument, the return code of the process.

        This method uses a ``SIGCHLD`` handler, which is a global setting
        and may conflict if you have other libraries trying to handle the
        same signal.  If you are using more than one ``IOLoop`` it may
        be necessary to call `Subprocess.initialize` first to designate
        one ``IOLoop`` to run the signal handlers.

        In many cases a close callback on the stdout or stderr streams
        can be used as an alternative to an exit callback if the
        signal handler is causing a problem.
        """
        self._exit_callback = stack_context.wrap(callback)
        Subprocess.initialize(self.io_loop)
        Subprocess._waiting[self.pid] = self
        Subprocess._try_cleanup_process(self.pid)

    def wait_for_exit(self):
        """Returns a `.Future` which resolves when the process exits.

        Usage::

            ret = yield proc.wait_for_exit()

        This is a coroutine-friendly alternative to `set_exit_callback`
        (and a replacement for the blocking `subprocess.Popen.wait`).

        .. versionadded:: 4.2
        """
        future = Future()
        self.set_exit_callback(future.set_result)
        return future

    @classmethod
    def initialize(cls, io_loop=None):
        """Initializes the ``SIGCHLD`` handler.

        The signal handler is run on an `.IOLoop` to avoid locking issues.
        Note that the `.IOLoop` used for signal handling need not be the
        same one used by individual Subprocess objects (as long as the
        ``IOLoops`` are each running in separate threads).

        .. versionchanged:: 4.1
           The ``io_loop`` argument is deprecated.
        """
        if cls._initialized:
            return
        if io_loop is None:
            io_loop = ioloop.IOLoop.current()
        cls._old_sigchld = signal.signal(
            signal.SIGCHLD,
            lambda sig, frame: io_loop.add_callback_from_signal(cls._cleanup))
        cls._initialized = True

    @classmethod
    def uninitialize(cls):
        """Removes the ``SIGCHLD`` handler."""
        if not cls._initialized:
            return
        signal.signal(signal.SIGCHLD, cls._old_sigchld)
        cls._initialized = False

    @classmethod
    def _cleanup(cls):
        for pid in list(cls._waiting.keys()):  # make a copy
            cls._try_cleanup_process(pid)

    @classmethod
    def _try_cleanup_process(cls, pid):
        try:
            ret_pid, status = os.waitpid(pid, os.WNOHANG)
        except OSError as e:
            if errno_from_exception(e) == errno.ECHILD:
                return
        if ret_pid == 0:
            return
        assert ret_pid == pid
        subproc = cls._waiting.pop(pid)
        subproc.io_loop.add_callback_from_signal(
            subproc._set_returncode, status)

    def _set_returncode(self, status):
        if os.WIFSIGNALED(status):
            self.returncode = -os.WTERMSIG(status)
        else:
            assert os.WIFEXITED(status)
            self.returncode = os.WEXITSTATUS(status)
        if self._exit_callback:
            callback = self._exit_callback
            self._exit_callback = None
            callback(self.returncode)

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
user