[Python] Why does my first test run timeout (but second run is fast) when running multiple...

I am working on an automated grading tool for student programming submissions. The process is:

1. Students submit their code (Python projects).
2. I clean and organise the submissions.
3. I set up a separate virtual environment for each submission.
4. When I press “Run Tests,” the system grades all submissions in parallel using ThreadPoolExecutor.

The problem is when I press “Run Tests” for the first time the program runs extremely slowly and eventually every submission hits a timeout resulting in having an empty report. However, when I run the same tests again immediately afterward, they complete very quickly without any issue.

What I tried:

• I created a warm-up function that pre-compiles Python files in each submission compileall before running tests. It did not solve the timeout; the first run still hangs.
• I replaced ThreadPoolExecutor with ProcessPoolExecutor but it made no noticeable difference (and was even slightly slower on the second run).
• Creating venvs does not interfere with running tests — each step (cleaning, venv setup, testing) is separated clearly.
• I suspect it may be related to ThreadPoolExecutor or how many submissions I am trying to grade in parallel (~200 submission) as I do not encounter this issue when running tests sequentially.

What can I do to run these tasks in parallel safely, without submissions hitting a timeout on first run?

• Should I limit the number of parallel jobs?
• Should I change the way subprocesses are created or warmed up?
• Is there a better way to handle parallelism across many venvs?

def grade_all_submissions(tasks: list, submissions_root: Path) -> None:
threads = int(os.cpu_count() * 1.5)

for task in tasks:
config = TASK_CONFIG.get(task)
if not config:
continue

submissions = [
submission for submission in submissions_root.iterdir()
if submission.is_dir() and submission.name.startswith("Portfolio")
]

with ThreadPoolExecutor(max_workers=threads) as executor:
future_to_submission = {
executor.submit(grade_single_submission, task, submission): submission
for submission in submissions
}

for future in as_completed(future_to_submission):
submission = future_to_submission[future]
try:
future.result()
except Exception as e:
print(f"Error in {submission.name} for {task}: {e}")

def run_python(self, args, cwd) -> str:
pythonPath = str(self.get_python_path())
command = [pythonPath] + args
result = subprocess.run(
command,
capture_output=True,
text=True,
cwd = str(cwd) if cwd else None,
timeout=59.0
)



grade_single_submission() uses run_python() to run -m unittest path/to/testscript

Continue reading...