The while Loop and State-Driven Execution
The while Loop and State-Driven Execution
After reading this chapter, you will master state-driven iteration using the while loop, configure daemon loops with while True, implement robust polling algorithms with exponential backoff, handle stream termination sentinels, and prevent runaway CPU-consuming infinite loops.
Mental model
A while loop repeatedly executes a block of code as long as a boolean test condition evaluates to True. The condition is checked at the entry of every iteration:
The while Loop Execution Cycle:
┌──────────────────────┐
▼ │
[ Test Condition ] │
/ \ │
True False │
/ \ │
[ Loop Body ] [ Terminate Loop ]
│ │
└────────────────────────────┘
If the condition evaluates to False on the very first evaluation, the loop body never executes.
Minimal example
Save as while_loop_mechanics.py:
# while_loop_mechanics.py
import time
def poll_node_readiness(target_node: str, max_retries: int = 4) -> bool:
"""Poll a cluster node until it reports ready or retries are exhausted."""
# Simulated cluster states: 0=initializing, 1=starting, 2=healthy
mock_states = ["INITIALIZING", "STARTING", "READY"]
attempt = 0
print(f"Polling node '{target_node}'...")
while attempt < max_retries:
attempt += 1
# Retrieve current mock state
current_state = mock_states[attempt - 1] if attempt - 1 < len(mock_states) else "READY"
print(f" Attempt {attempt}/{max_retries}: State is '{current_state}'")
if current_state == "READY":
print(" Node is ready! Exiting polling loop.")
return True
# Polling delay
time.sleep(0.02)
print(" Failed: Maximum retries exceeded.")
return False
def main() -> None:
is_ready = poll_node_readiness("k8s-worker-01", max_retries=4)
print(f"Final readiness status: {is_ready}")
if __name__ == "__main__":
main()Run via uv run python while_loop_mechanics.py:
Polling node 'k8s-worker-01'...
Attempt 1/4: State is 'INITIALIZING'
Attempt 2/4: State is 'STARTING'
Attempt 3/4: State is 'READY'
Node is ready! Exiting polling loop.
Final readiness status: True
Worked examples
Case 1: Exponential Backoff with Jitter for Network Retries
In production distributed systems, polling an overloaded server at fixed intervals worsens congestion. An exponential backoff loop doubles the wait duration on each attempt:
# exponential_backoff.py
import time
def call_unstable_service(attempt: int) -> bool:
# Simulates recovery on 4th attempt
return attempt >= 4
def retry_with_backoff(max_attempts: int = 5, base_delay: float = 0.01) -> bool:
attempt = 1
delay = base_delay
while attempt <= max_attempts:
print(f"[Attempt {attempt}] Contacting payment gateway...")
success = call_unstable_service(attempt)
if success:
print(f"[Success] Gateway response received on attempt {attempt}!")
return True
print(f" Transient failure. Backing off for {delay*1000:.1f}ms...")
time.sleep(delay)
# Exponential backoff: double the delay duration
delay *= 2
attempt += 1
return False
if __name__ == "__main__":
retry_with_backoff(max_attempts=5, base_delay=0.01)Run:
uv run python exponential_backoff.pyOutput:
[Attempt 1] Contacting payment gateway...
Transient failure. Backing off for 10.0ms...
[Attempt 2] Contacting payment gateway...
Transient failure. Backing off for 20.0ms...
[Attempt 3] Contacting payment gateway...
Transient failure. Backing off for 40.0ms...
[Attempt 4] Contacting payment gateway...
[Success] Gateway response received on attempt 4!
Case 2: State-Machine Dispatcher Loop
while loops are the foundation of state-machine processing, where the loop continues running until a terminal state is reached:
# state_machine_loop.py
from enum import Enum
class State(Enum):
PENDING = "pending"
PROCESSING = "processing"
VERIFYING = "verifying"
COMPLETED = "completed"
FAILED = "failed"
def run_job_pipeline() -> None:
current_state = State.PENDING
step_count = 0
print("Initiating state machine loop:")
while current_state not in (State.COMPLETED, State.FAILED):
step_count += 1
print(f" Step {step_count}: Current state is '{current_state.value}'")
if current_state == State.PENDING:
current_state = State.PROCESSING
elif current_state == State.PROCESSING:
current_state = State.VERIFYING
elif current_state == State.VERIFYING:
current_state = State.COMPLETED
print(f"Terminal state reached: '{current_state.value}' after {step_count} steps.")
if __name__ == "__main__":
run_job_pipeline()Run:
uv run python state_machine_loop.pyOutput:
Initiating state machine loop:
Step 1: Current state is 'pending'
Step 2: Current state is 'processing'
Step 3: Current state is 'verifying'
Terminal state reached: 'completed' after 3 steps.
Pitfalls
Pitfall 1: The Accidental Runaway Infinite Loop
If the loop variable is not updated within the loop body, the condition remains permanently True, consuming 100% of a CPU core:
# FATAL BUG:
count = 5
while count > 0:
print(count)
# BUG: Forgot count -= 1! Runs forever!
# FIX:
while count > 0:
print(count)
count -= 1Pitfall 2: Busy-Waiting Without Sleeping
A while loop that checks a flag or condition without yielding (time.sleep() or await asyncio.sleep()) creates a busy-wait loop that starves other threads and overheats the CPU. Always insert a sleep or use an event primitive (threading.Event.wait()).
Exercises
- Write a
whileloop that implements Euclid’s algorithm to find the greatest common divisor (GCD) of two numbers \(A\) and \(B\). - Implement a countdown loop that starts at 10 and decrements down to 1, printing
"BLASTOFF!"after the loop finishes. - Write a sentinel-controlled
whileloop that pops elements from a queue list until an element withstatus == "DONE"is encountered. - Implement an exponential backoff loop that caps the maximum delay at
1.0second.
Further reading
- Python Language Reference: The
whilestatement. - Python Standard Library:
time.sleepandthreading.Event.