Skip to content

Commit

Permalink
Make sure that the debugger is cleared after fork in non-python proce…
Browse files Browse the repository at this point in the history
…ss. Fixes microsoft#1005
  • Loading branch information
fabioz committed Aug 18, 2022
1 parent 71d42ed commit 99440cd
Show file tree
Hide file tree
Showing 3 changed files with 106 additions and 1 deletion.
4 changes: 3 additions & 1 deletion src/debugpy/_vendored/pydevd/_pydev_bundle/pydev_monkey.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import sys
from _pydev_bundle._pydev_saved_modules import threading
from _pydevd_bundle.pydevd_constants import get_global_debugger, IS_WINDOWS, IS_JYTHON, get_current_thread_id, \
sorted_dict_repr
sorted_dict_repr, set_global_debugger
from _pydev_bundle import pydev_log
from contextlib import contextmanager
from _pydevd_bundle import pydevd_constants
Expand Down Expand Up @@ -904,6 +904,8 @@ def new_fork():
if is_new_python_process:
PydevdCustomization.DEFAULT_PROTOCOL = protocol
_on_forked_process(setup_tracing=apply_arg_patch and not is_subprocess_fork)
else:
set_global_debugger(None)
else:
if is_new_python_process:
send_process_created_message()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
from gevent import monkey
monkey.patch_all()

import subprocess
import sys
import os

if __name__ == "__main__":
if '-foo' in sys.argv:
print('foo called')
else:
if os.environ.get('CALL_PYTHON_SUB') == '1':
assert 'foo called' in subprocess.check_output([sys.executable, __file__, '-foo']).decode('utf-8')
else:
subprocess.check_output("tput -T xterm-256color bold".split())
print('TEST SUCEEDED')
87 changes: 87 additions & 0 deletions src/debugpy/_vendored/pydevd/tests_python/test_debugger_json.py
Original file line number Diff line number Diff line change
Expand Up @@ -3990,6 +3990,93 @@ def get_environ(writer):
writer.finished_ok = True


@pytest.mark.skipif(not TEST_GEVENT, reason='Gevent not installed.')
def test_gevent_subprocess_not_python(case_setup):

def get_environ(writer):
env = os.environ.copy()
env['GEVENT_SUPPORT'] = 'True'
env['CALL_PYTHON_SUB'] = '0'
return env

with case_setup.test_file('_debugger_case_gevent_subprocess.py', get_environ=get_environ) as writer:
json_facade = JsonFacade(writer)

break1_line = writer.get_line_index_with_content("print('TEST SUCEEDED')")
json_facade.write_set_breakpoints(break1_line)
json_facade.write_make_initial_run()
json_facade.wait_for_thread_stopped(line=break1_line)

json_facade.write_continue(wait_for_response=False)
writer.finished_ok = True


@pytest.mark.skipif(not TEST_GEVENT, reason='Gevent not installed.')
def test_gevent_subprocess_python(case_setup_multiprocessing):
import threading
from tests_python.debugger_unittest import AbstractWriterThread

def get_environ(writer):
env = os.environ.copy()
env['GEVENT_SUPPORT'] = 'True'
env['CALL_PYTHON_SUB'] = '1'
return env

with case_setup_multiprocessing.test_file(
'_debugger_case_gevent_subprocess.py',
get_environ=get_environ,
) as writer:
json_facade = JsonFacade(writer)
json_facade.write_launch()

break1_line = writer.get_line_index_with_content("print('foo called')")
json_facade.write_set_breakpoints([break1_line])

server_socket = writer.server_socket
secondary_finished_ok = [False]

class SecondaryProcessWriterThread(AbstractWriterThread):

TEST_FILE = writer.get_main_filename()
_sequence = -1

class SecondaryProcessThreadCommunication(threading.Thread):

def run(self):
from tests_python.debugger_unittest import ReaderThread
server_socket.listen(1)
self.server_socket = server_socket
new_sock, addr = server_socket.accept()

reader_thread = ReaderThread(new_sock)
reader_thread.name = ' *** Multiprocess Reader Thread'
reader_thread.start()

writer2 = SecondaryProcessWriterThread()
writer2.reader_thread = reader_thread
writer2.sock = new_sock
json_facade2 = JsonFacade(writer2)

json_facade2.write_set_breakpoints([break1_line, ])
json_facade2.write_make_initial_run()

json_facade2.wait_for_thread_stopped()
json_facade2.write_continue()
secondary_finished_ok[0] = True

secondary_process_thread_communication = SecondaryProcessThreadCommunication()
secondary_process_thread_communication.start()
time.sleep(.1)

json_facade.write_make_initial_run()
secondary_process_thread_communication.join(10)
if secondary_process_thread_communication.is_alive():
raise AssertionError('The SecondaryProcessThreadCommunication did not finish')

assert secondary_finished_ok[0]
writer.finished_ok = True


@pytest.mark.skipif(
not TEST_GEVENT or IS_WINDOWS,
reason='Gevent not installed / Sometimes the debugger crashes on Windows as the compiled extensions conflict with gevent.'
Expand Down

0 comments on commit 99440cd

Please sign in to comment.