attach_script.py

def load_python_helper_lib():
    import sys
    try:
        import ctypes
    except ImportError:
        ctypes = None

    # Note: we cannot use import platform because it may end up importing threading,
    # but that should be ok because at this point we can only be in CPython (other
    # implementations wouldn't get to this point in the attach process).
    # IS_CPYTHON = platform.python_implementation() == 'CPython'
    IS_CPYTHON = True

    import os
    IS_64BIT_PROCESS = sys.maxsize > (2 ** 32)
    IS_WINDOWS = sys.platform == 'win32'
    IS_LINUX = sys.platform in ('linux', 'linux2')
    IS_MAC = sys.platform == 'darwin'

    if not IS_CPYTHON or ctypes is None or sys.version_info[:2] > (3, 7):
        return None

    if IS_WINDOWS:
        if IS_64BIT_PROCESS:
            suffix = 'amd64'
        else:
            suffix = 'x86'

        filename = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'pydevd_attach_to_process', 'attach_%s.dll' % (suffix,))

    elif IS_LINUX:
        if IS_64BIT_PROCESS:
            suffix = 'amd64'
        else:
            suffix = 'x86'

        filename = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'pydevd_attach_to_process', 'attach_linux_%s.so' % (suffix,))

    elif IS_MAC:
        if IS_64BIT_PROCESS:
            suffix = 'x86_64.dylib'
        else:
            suffix = 'x86.dylib'

        filename = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'pydevd_attach_to_process', 'attach_%s' % (suffix,))

    else:
        return None

    if not os.path.exists(filename):
        return None

    try:
        # Load as pydll so that we don't release the gil.
        lib = ctypes.pydll.LoadLibrary(filename)
        return lib
    except:
        return None


def get_main_thread_instance(threading):
    if hasattr(threading, 'main_thread'):
        return threading.main_thread()
    else:
        # On Python 2 we don't really have an API to get the main thread,
        # so, we just get it from the 'shutdown' bound method.
        return threading._shutdown.im_self


def get_main_thread_id(unlikely_thread_id=None):
    '''
    :param unlikely_thread_id:
        Pass to mark some thread id as not likely the main thread.

    :return tuple(thread_id, critical_warning)
    '''
    import sys
    import os

    current_frames = sys._current_frames()
    possible_thread_ids = []
    for thread_ident, frame in current_frames.items():
        while frame.f_back is not None:
            frame = frame.f_back

        basename = os.path.basename(frame.f_code.co_filename)
        if basename.endswith(('.pyc', '.pyo')):
            basename = basename[:-1]

        if (frame.f_code.co_name, basename) in [
                ('_run_module_as_main', 'runpy.py'),
                ('run_module_as_main', 'runpy.py'),
                ('run_module', 'runpy.py'),
                ('run_path', 'runpy.py'),
            ]:
            # This is the case for python -m <module name> (this is an ideal match, so,
            # let's return it).
            return thread_ident, ''

        if frame.f_code.co_name == '<module>':
            if frame.f_globals.get('__name__') == '__main__':
                possible_thread_ids.insert(0, thread_ident)  # Add with higher priority
                continue

            # Usually the main thread will be started in the <module>, whereas others would
            # be started in another place (but when Python is embedded, this may not be
            # correct, so, just add to the available possibilities as we'll have to choose
            # one if there are multiple).
            possible_thread_ids.append(thread_ident)

    if len(possible_thread_ids) > 0:
        if len(possible_thread_ids) == 1:
            return possible_thread_ids[0], ''  # Ideal: only one match

        while unlikely_thread_id in possible_thread_ids:
            possible_thread_ids.remove(unlikely_thread_id)

        if len(possible_thread_ids) == 1:
            return possible_thread_ids[0], ''  # Ideal: only one match

        elif len(possible_thread_ids) > 1:
            # Bad: we can't really be certain of anything at this point.
            return possible_thread_ids[0], \
                'Multiple thread ids found (%s). Choosing main thread id randomly (%s).' % (
                    possible_thread_ids, possible_thread_ids[0])

    # If we got here we couldn't discover the main thread id.
    return None, 'Unable to discover main thread id.'


def fix_main_thread_id(on_warn=lambda msg:None, on_exception=lambda msg:None, on_critical=lambda msg:None):
    # This means that we weren't able to import threading in the main thread (which most
    # likely means that the main thread is paused or in some very long operation).
    # In this case we'll import threading here and hotfix what may be wrong in the threading
    # module (if we're on Windows where we create a thread to do the attach and on Linux
    # we are not certain on which thread we're executing this code).
    #
    # The code below is a workaround for https://bugs.python.org/issue37416
    import sys
    import threading

    try:
        with threading._active_limbo_lock:
            main_thread_instance = get_main_thread_instance(threading)

            if sys.platform == 'win32':
                # On windows this code would be called in a secondary thread, so,
                # the current thread is unlikely to be the main thread.
                if hasattr(threading, '_get_ident'):
                    unlikely_thread_id = threading._get_ident()  # py2
                else:
                    unlikely_thread_id = threading.get_ident()  # py3
            else:
                unlikely_thread_id = None

            main_thread_id, critical_warning = get_main_thread_id(unlikely_thread_id)

            if main_thread_id is not None:
                main_thread_id_attr = '_ident'
                if not hasattr(main_thread_instance, main_thread_id_attr):
                    main_thread_id_attr = '_Thread__ident'
                    assert hasattr(main_thread_instance, main_thread_id_attr)

                if main_thread_id != getattr(main_thread_instance, main_thread_id_attr):
                    # Note that we also have to reset the '_tstack_lock' for a regular lock.
                    # This is needed to avoid an error on shutdown because this lock is bound
                    # to the thread state and will be released when the secondary thread
                    # that initialized the lock is finished -- making an assert fail during
                    # process shutdown.
                    main_thread_instance._tstate_lock = threading._allocate_lock()
                    main_thread_instance._tstate_lock.acquire()

                    # Actually patch the thread ident as well as the threading._active dict
                    # (we should have the _active_limbo_lock to do that).
                    threading._active.pop(getattr(main_thread_instance, main_thread_id_attr), None)
                    setattr(main_thread_instance, main_thread_id_attr, main_thread_id)
                    threading._active[getattr(main_thread_instance, main_thread_id_attr)] = main_thread_instance

        # Note: only import from pydevd after the patching is done (we want to do the minimum
        # possible when doing that patching).
        on_warn('The threading module was not imported by user code in the main thread. The debugger will attempt to work around https://bugs.python.org/issue37416.')

        if critical_warning:
            on_critical('Issue found when debugger was trying to work around https://bugs.python.org/issue37416:\n%s' % (critical_warning,))
    except:
        on_exception('Error patching main thread id.')


def attach(port, host, protocol=''):
    try:
        import sys
        fix_main_thread = 'threading' not in sys.modules

        if fix_main_thread:

            def on_warn(msg):
                from _pydev_bundle import pydev_log
                pydev_log.warn(msg)

            def on_exception(msg):
                from _pydev_bundle import pydev_log
                pydev_log.exception(msg)

            def on_critical(msg):
                from _pydev_bundle import pydev_log
                pydev_log.critical(msg)

            fix_main_thread_id(on_warn=on_warn, on_exception=on_exception, on_critical=on_critical)

        else:
            from _pydev_bundle import pydev_log  # @Reimport
            pydev_log.debug('The threading module is already imported by user code.')

        if protocol:
            from _pydevd_bundle import pydevd_defaults
            pydevd_defaults.PydevdCustomization.DEFAULT_PROTOCOL = protocol

        import pydevd

        # I.e.: disconnect/reset if already connected.

        pydevd.SetupHolder.setup = None

        py_db = pydevd.get_global_debugger()
        if py_db is not None:
            py_db.dispose_and_kill_all_pydevd_threads(wait=False)

        # pydevd.DebugInfoHolder.DEBUG_RECORD_SOCKET_READS = True
        # pydevd.DebugInfoHolder.DEBUG_TRACE_BREAKPOINTS = 3
        # pydevd.DebugInfoHolder.DEBUG_TRACE_LEVEL = 3
        pydevd.settrace(
            port=port,
            host=host,
            stdoutToServer=True,
            stderrToServer=True,
            overwrite_prev_trace=True,
            suspend=False,
            trace_only_current_thread=False,
            patch_multiprocessing=False,
        )
    except:
        import traceback
        traceback.print_exc()