python 21 lines · 6 steps

Round-robin task distribution in Python

A cyclic iterator hands tasks to workers in turn, with a weighted variant that skews the rotation by capacity.

Explained by highlit
Intermediate round-robin iterators load-balancing defaultdict
1from itertools import cycle
2from collections import defaultdict
3 
4 
5def distribute(tasks, workers):
6 if not workers:
7 raise ValueError("at least one worker is required")
8 
9 assignments = defaultdict(list)
10 worker_pool = cycle(workers)
11 
12 for task in tasks:
13 worker = next(worker_pool)
14 assignments[worker].append(task)
15 
16 return dict(assignments)
17 
18 
19def distribute_weighted(tasks, workers):
20 expanded = [w for w in workers for _ in range(w.capacity)]
21 return distribute(tasks, expanded)
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1itertools.cycle turns any finite sequence into an endless round-robin without tracking an index yourself.
  2. 2defaultdict(list) removes the need to check for and initialize missing keys before appending.
  3. 3Weighting a round-robin can be as simple as repeating each item in the pool proportionally to its weight.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
rust 
use chrono::{Duration, NaiveDate};
 
#[derive(Debug)]
pub struct DateRange {

Parsing and iterating date ranges in Rust

error-handling iterators parsing
Intermediate 7 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps

Share this explainer

Here's the card — post it anywhere.

Round-robin task distribution in Python — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code