Usually, without interrupting it, you cannot interactively debug, call functions in, or change the values of variables in a running python program. This article’s technique removes the limitation.

There are plenty of python examples showing how an executing script can interrupt its execution to present a console, where you can interact, and then resume execution. Here, the console is your choice of the Python interactive shell prompt, with its Read-Eval-Print Loop (aka REPL), or alternatively, the python PDB debugger.

I wanted to do that interacting, but without theinterrupting and the resuming parts. While the long running execution continues to run, and without interrupting the program’s flow, I wanted to be able to access objects and invoke other functions concurrently. In the Snow White films, there is a song “Whistle While You Work.” (Watch the 1937 version or 2025 version). I wanted to “REPL While It Works.”

Audience: The audience for this technique is writers of programs that they use for themselves. Normally, a program’s data and functions, their names, semantics, and usage, are not known by others.

Achieving this objective is not hard. The main features of the solution are:

  1. A convenient repository for pseudo-global objects which can be created, set, and used both by the threads and by you while in the REPL or PDB. Avoid having to use theglobalkeyword.
  2. The application’s real work happens in a started thread or threads.
  3. The main thread, after any initiation and starting the threads, drops into REPL, or PDB, or both, one after the other.

More details:

1. The repository

Create a function to hold the pseudoglobal objects as attributes. Add, change, and use them from anywhere. Avoid having to use the global keyword.

def g(): # This function is never called. 
    # It just holds attributes which can be added, set and used from anywhere.
    pass
# example attributes
g.d = defaultdict(list)
g.finishedthreads=0
g.startedat=time.time()

  1. The application functions, are (a) first defined, and then (b) some of them are started as threads. These do the application’s work. See the details in the full program’s code below.
  1. Drop into the interactive REPL ro PDB console, or one after the other.
#This goes at the bottom of the script
while True:	
    # Drop into the PDB console	
    print(f'n Type os._exit()  to terminate the process abruptly, without flushing file buffers, running finally blocks, or executing atexit cleanups.')	  
    print(f'n Type sys.exit(0) to terminate the main and daemon threads. It first raises a SystemExit exception in each thread. ')	
    print(" Dropping into the PDB debugger console.")	
    print("	 Type c or continue to leave PDB and continue script execution.")	

    breakpoint()

    # Drop into the REPL console	
    print(" Dropping into the REPL console")	
    print("	 Type <ctrl>D to leave the REPL and continue script execution.")	

    code.interact(local=dict(globals(), **locals()))

Full program’s code. This is the full program, including the features above.

#!/usr/bin/python3
# run multithreaded tasks, with a repl while they run

import code
import time
from collections import defaultdict
import threading 
import os
import sys

# 1 Create the repository

def g(): # This function is never called. It just holds attributes which set and used from anywhere.
	pass
# example attributes
g.threads=[]
g.glock=threading.Lock() #Global lock available to be used by all threads.
g.d = defaultdict(list)
g.finishedthreads=0
g.startedat=time.time()

# 2A Define functions

def aworker(i=1,art=0,ep=1,op=1000):
	wpytid = threading.get_ident()
	wostid=  threading.get_native_id()
	wname = threading.current_thread().name
	g.d[i]={"i":i, "name":wname, "wpytid":wpytid, "wostid": wostid}
	print("Thread started: ",g.d[i])
	print(f"Parms = {i=}, {art=}, {ep=}, {op=}") 
	time.sleep(5)
	
	# Resource Lock aquire and use
	#   defined elsewhere: g.glock=threading.Lock()
	g.glock.acquire()
	try: # This try - finally avoids deadlocks if what is inside throws an exception
		print("Acquired ", g.glock)
		whatever=0
	finally:
		g.glock.release()  

	time.sleep(50)
	g.finishedthreads+=1
    g.d[i]["Terminated"]=True
	print(f'Thread terminating {wname}')



def analyze():
		pass

def periodicresultssofar():
	# started as a thread
	while True:
		time.sleep(600)
		analyze()
		if g.finishedthreads==g.qthreads: return

# 2b Launch threads	

qthreads=12
g.qthreads=qthreads
for t in range(1,qthreads+1):
	w = threading.Thread(name=f'Worker-{t}',target=aworker, args=([t]), kwargs={"ep":1,"op": 3}, daemon=True)

 
	g.threads.append(w)
	w.start()
	print(f"Thread Name: {w.name}")
	print(f"Python Thread ID (.ident): {w.ident}")
	print(f"OS Native Thread ID (.native_id): {w.native_id}")

# periodic results so far

p=threading.Thread(name=f'PeriodicResultsSoFar',target=periodicresultssofar, daemon=True)
p.start()
g.threads.append(p)
print(f"Thread Name: {p.name}")
print(f"Python Thread ID (.ident): {p.ident}")
print(f"OS Native Thread ID (.native_id): {p.native_id}")

# 3 Repeatedly drop into the PDB and then the REPL console.

while True:
	# Drop into the PDB console
	print(" Dropping into the PDB console.")
	print(f'n Type os._exit()  to terminate the process abruptly, without flushing file buffers, running finally blocks, or executing atexit cleanups.')
	print(f'n Type sys.exit(0) to terminate the main thread and daemon threads. It first raises a SystemExit exception. ')
	print("	Type c or continue to exit the debugger and let the script continue.")
	print("	Type analyze()  to analyze what has been processed so far")
	breakpoint() 

	# Drop into the REPL console
	print(" Dropping into the REPL console")
	print("	Type <ctrl>D to exit the REPL and let the script continue.")
	code.interact(local=dict(globals(), **locals()))

How did I use this? I created a multithreaded program which accesses my gmail account. The threads divided up the work of collecting the sender and size of each of the account’s 100K emails. With gmail’s IMAP response time of about 3 seconds per request, it took many hours. This technique let me run my analyze function whenever I wanted even while the data was still being collected. This let me satisfy my curiosity to find the top senders, so far, in terms of total bytes in all the emails for that sender.