Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

bpo-30696: Fix the REPL looping endlessly when no memory #4160

Merged
merged 8 commits into from
Nov 12, 2017
Prev Previous commit
Next Next commit
Create test_repl.py and move run_pty() to test.support
xdegaye committed Oct 29, 2017
commit 44c335e318ac824aaf268b06e6e02432a1078359
56 changes: 56 additions & 0 deletions Lib/test/support/__init__.py
Original file line number Diff line number Diff line change
@@ -32,6 +32,9 @@
import unittest
import urllib.error
import warnings
import selectors
from contextlib import ExitStack
from errno import EIO

try:
import multiprocessing.process
@@ -105,6 +108,7 @@
"run_with_locale", "swap_item",
"swap_attr", "Matcher", "set_memlimit", "SuppressCrashReport", "sortdict",
"run_with_tz", "PGO", "missing_compiler_executable", "fd_count",
"run_pty",
]

class Error(Exception):
@@ -2755,3 +2759,55 @@ def fd_count():
msvcrt.CrtSetReportMode(report_type, old_modes[report_type])

return count

def run_pty(script, input=b"dummy input\r", env=None):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I propose to add a docstring to document the side effect of import_module('pty'):

"""
Run a script in a temporary pseudoterminal created just for the script.

Raise SkipTest if the pty module cannot be imported.
"""

pty = import_module('pty')
output = bytearray()
[master, slave] = pty.openpty()
if script is None:
args = (sys.executable, '-q')
else:
args = (sys.executable, '-c', script)
proc = subprocess.Popen(args, stdin=slave, stdout=slave, stderr=slave, env=env)
os.close(slave)
with ExitStack() as cleanup:
cleanup.enter_context(proc)
def terminate(proc):
try:
proc.terminate()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use kill() instead, no need to be nice if something goes wrong.

Moreover, you should always call proc.wait() after, to prevent zombie process.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FYI exiting the with statement (after the enter_context call) should already guarantee one call to wait. See the bottom of https://docs.python.org/3/library/subprocess.html#subprocess.Popen: “Popen objects are supported as context managers . . . the process is waited for.”

except ProcessLookupError:
# Workaround for Open/Net BSD bug (Issue 16762)
pass
cleanup.callback(terminate, proc)
cleanup.callback(os.close, master)
# Avoid using DefaultSelector and PollSelector. Kqueue() does not
# work with pseudo-terminals on OS X < 10.9 (Issue 20365) and Open
# BSD (Issue 20667). Poll() does not work with OS X 10.6 or 10.4
# either (Issue 20472). Hopefully the file descriptor is low enough
# to use with select().
sel = cleanup.enter_context(selectors.SelectSelector())
sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE)
os.set_blocking(master, False)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had prefer to set master to non-blocking mode before registering it into the selector.

By the way, a selector must not accept a FD in blocking more :-)

while True:
for [_, events] in sel.select():
if events & selectors.EVENT_READ:
try:
chunk = os.read(master, 0x10000)
except OSError as err:
# Linux raises EIO when slave is closed (Issue 5380)
if err.errno != EIO:
raise
chunk = b""
if not chunk:
return output
output.extend(chunk)
if events & selectors.EVENT_WRITE:
try:
input = input[os.write(master, input):]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please call os.write() on a separated line:

written = os.write(...)
input = input[written:]

You might move the "input = ..." in an else block. It's up to you.

except OSError as err:
# Apparently EIO means the slave was closed
if err.errno != EIO:
raise
input = b"" # Stop writing
if not input:
sel.modify(master, selectors.EVENT_READ)
83 changes: 1 addition & 82 deletions Lib/test/test_readline.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,11 @@
"""
Very minimal unittests for parts of the readline module.
"""
from contextlib import ExitStack
from errno import EIO
import os
import selectors
import subprocess
import sys
import tempfile
import unittest
from textwrap import dedent
from test.support import (import_module, unlink, temp_dir, TESTFN, verbose,
SuppressCrashReport)
run_pty)
from test.support.script_helper import assert_python_ok

# Skip tests if there is no readline module
@@ -273,80 +267,5 @@ def test_history_size(self):
self.assertEqual(lines[-1].strip(), b"last input")


# Tests for the interactive interpreter.
class TestInteractiveInterpreter(unittest.TestCase):

def test_interactive_no_memory(self):
# Issue #30696: Fix the interactive interpreter looping endlessly when
# no memory. Check also that the fix does not break the interactive
# loop when an exception is raised.
user_input = """
import sys, _testcapi
1/0
_testcapi.set_nomemory(0)
sys.exit(0)
"""
user_input = dedent(user_input)
user_input = b'\r'.join(x.encode() for x in user_input.split('\n'))
user_input += b'\r'
with SuppressCrashReport():
output = run_pty(None, input=user_input)
self.assertIn(b"Fatal Python error: Cannot recover from MemoryErrors",
output)


def run_pty(script, input=b"dummy input\r", env=None):
pty = import_module('pty')
output = bytearray()
[master, slave] = pty.openpty()
if script is None:
args = (sys.executable, '-q')
else:
args = (sys.executable, '-c', script)
proc = subprocess.Popen(args, stdin=slave, stdout=slave, stderr=slave, env=env)
os.close(slave)
with ExitStack() as cleanup:
cleanup.enter_context(proc)
def terminate(proc):
try:
proc.terminate()
except ProcessLookupError:
# Workaround for Open/Net BSD bug (Issue 16762)
pass
cleanup.callback(terminate, proc)
cleanup.callback(os.close, master)
# Avoid using DefaultSelector and PollSelector. Kqueue() does not
# work with pseudo-terminals on OS X < 10.9 (Issue 20365) and Open
# BSD (Issue 20667). Poll() does not work with OS X 10.6 or 10.4
# either (Issue 20472). Hopefully the file descriptor is low enough
# to use with select().
sel = cleanup.enter_context(selectors.SelectSelector())
sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE)
os.set_blocking(master, False)
while True:
for [_, events] in sel.select():
if events & selectors.EVENT_READ:
try:
chunk = os.read(master, 0x10000)
except OSError as err:
# Linux raises EIO when slave is closed (Issue 5380)
if err.errno != EIO:
raise
chunk = b""
if not chunk:
return output
output.extend(chunk)
if events & selectors.EVENT_WRITE:
try:
input = input[os.write(master, input):]
except OSError as err:
# Apparently EIO means the slave was closed
if err.errno != EIO:
raise
input = b"" # Stop writing
if not input:
sel.modify(master, selectors.EVENT_READ)


if __name__ == "__main__":
unittest.main()
28 changes: 28 additions & 0 deletions Lib/test/test_repl.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
"""Test the interactive interpreter."""

import unittest
from textwrap import dedent
from test.support import run_pty, SuppressCrashReport

class TestInteractiveInterpreter(unittest.TestCase):

def test_no_memory(self):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please decorate the test with @support.cpython_only, or use support.import_module('_testcapi'), to skip the test on Python implementations without _testcapi.

# Issue #30696: Fix the interactive interpreter looping endlessly when
# no memory. Check also that the fix does not break the interactive
# loop when an exception is raised.
user_input = """
import sys, _testcapi
1/0
_testcapi.set_nomemory(0)
sys.exit(0)
"""
user_input = dedent(user_input)
user_input = b'\r'.join(x.encode() for x in user_input.split('\n'))
user_input += b'\r'
with SuppressCrashReport():
output = run_pty(None, input=user_input)
self.assertIn(b"Fatal Python error: Cannot recover from MemoryErrors",
output)

if __name__ == "__main__":
unittest.main()