def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
## Modified parameters by Don Jayamanne
# Accept current Process id to pass back to debugger
python类start_new()的实例源码
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
## Modified parameters by Don Jayamanne
# Accept current Process id to pass back to debugger
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
# If threading has already been imported (i.e. we're attaching), we must hot-patch threading._start_new_thread
# so that new threads started using it will be intercepted by our code.
#
# On the other hand, if threading has not been imported, we must not import it ourselves, because it will then
# treat the current thread as the main thread, which is incorrect when attaching because this code is executing
# on an ephemeral debugger attach thread that will go away shortly. We don't need to hot-patch it in that case
# anyway, because it will pick up the new thread.start_new_thread that we have set above when it's imported.
global _threading
if _threading is None and 'threading' in sys.modules:
import threading
_threading = threading
_threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def test():
w = WindowOutput(queueing=flags.WQ_IDLE)
w.write("First bit of text\n")
import _thread
for i in range(5):
w.write("Hello from the main thread\n")
_thread.start_new(thread_test, (w,))
for i in range(2):
w.write("Hello from the main thread\n")
win32api.Sleep(50)
return w
def startPycheckerRun(self):
self.result=None
old=win32api.SetCursor(win32api.LoadCursor(0, win32con.IDC_APPSTARTING))
win32ui.GetApp().AddIdleHandler(self.idleHandler)
import _thread
_thread.start_new(self.threadPycheckerRun,())
##win32api.SetCursor(old)
def test():
w = WindowOutput(queueing=flags.WQ_IDLE)
w.write("First bit of text\n")
import _thread
for i in range(5):
w.write("Hello from the main thread\n")
_thread.start_new(thread_test, (w,))
for i in range(2):
w.write("Hello from the main thread\n")
win32api.Sleep(50)
return w
def __init__(self, *args):
winout.WindowOutput.__init__(*(self,)+args)
self.hStopThread = win32event.CreateEvent(None, 0, 0, None)
_thread.start_new(CollectorThread, (self.hStopThread, self))
def BeginThreadsSimpleMarshal(numThreads, cookie):
"""Creates multiple threads using simple (but slower) marshalling.
Single interpreter object, but a new stream is created per thread.
Returns the handles the threads will set when complete.
"""
ret = []
for i in range(numThreads):
hEvent = win32event.CreateEvent(None, 0, 0, None)
_thread.start_new(TestInterpInThread, (hEvent, cookie))
ret.append(hEvent)
return ret
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def intercept_threads(for_attach = False):
thread.start_new_thread = thread_creator
thread.start_new = thread_creator
global threading
if threading is None:
# we need to patch threading._start_new_thread so that
# we pick up new threads in the attach case when threading
# is already imported.
import threading
threading._start_new_thread = thread_creator
global _INTERCEPTING_FOR_ATTACH
_INTERCEPTING_FOR_ATTACH = for_attach
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
def detach_process():
global DETACHED
DETACHED = True
if not _INTERCEPTING_FOR_ATTACH:
if isinstance(sys.stdout, _DebuggerOutput):
sys.stdout = sys.stdout.old_out
if isinstance(sys.stderr, _DebuggerOutput):
sys.stderr = sys.stderr.old_out
if not _INTERCEPTING_FOR_ATTACH:
thread.start_new_thread = _start_new_thread
thread.start_new = _start_new_thread
