diff --git a/test/functional/test_framework/util.py b/test/functional/test_framework/util.py
--- a/test/functional/test_framework/util.py
+++ b/test/functional/test_framework/util.py
@@ -9,10 +9,12 @@
 import logging
 import os
 import re
+import socket
 import time
 import unittest
 from base64 import b64encode
 from decimal import ROUND_DOWN, Decimal
+from functools import cache
 from io import BytesIO
 from subprocess import CalledProcessError
 from typing import Callable, Optional
@@ -329,15 +331,59 @@
     return coverage.AuthServiceProxyWrapper(proxy, coverage_logfile)
 
 
-def p2p_port(n):
-    assert n <= MAX_NODES
-    return PORT_MIN + n + \
-        (MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES)
+def is_port_available(port: int) -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        return sock.connect_ex(('127.0.0.1', port)) != 0
 
 
-def rpc_port(n):
-    return PORT_MIN + PORT_RANGE + n + \
-        (MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES)
+MAX_PORT_RETRY = 5
+LAST_P2P_PORT_USED = None
+
+
+@cache
+def p2p_port(n: int) -> int:
+    global LAST_P2P_PORT_USED
+    if LAST_P2P_PORT_USED is None:
+        # Initialize this here because PortSeed.n will be defined when this
+        # function is called.
+        assert PortSeed.n is not None
+        LAST_P2P_PORT_USED = PORT_MIN + (
+            MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES)
+
+    ntries = 0
+    try_port = LAST_P2P_PORT_USED + 1
+    while not is_port_available(try_port):
+        if ntries >= MAX_PORT_RETRY:
+            raise RuntimeError(
+                f"Could not find available port after {MAX_PORT_RETRY} attempts.")
+        ntries += 1
+        try_port += 1
+    LAST_P2P_PORT_USED = try_port
+    return try_port
+
+
+LAST_RPC_PORT_USED = None
+
+
+@cache
+def rpc_port(n: int) -> int:
+    global LAST_RPC_PORT_USED
+    if LAST_RPC_PORT_USED is None:
+        assert PortSeed.n is not None
+        LAST_RPC_PORT_USED = PORT_MIN + PORT_RANGE + (
+            MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES)
+
+    ntries = 0
+    try_port = LAST_RPC_PORT_USED + 1
+    while not is_port_available(try_port):
+        if ntries >= MAX_PORT_RETRY:
+            raise RuntimeError(
+                f"Could not find available port after {MAX_PORT_RETRY} attempts.")
+        ntries += 1
+        try_port += 1
+
+    LAST_RPC_PORT_USED = try_port
+    return try_port
 
 
 def rpc_url(datadir, chain, host, port):