Docs

Checkpointing & Recovery

CROWDio supports transparent checkpoint/resume for long-running tasks. This page covers the decorator API, checkpoint types, storage strategy, AST-based instrumentation, and recovery mechanics.

---

Why Checkpointing?

In Mobile Crowd Computing environments, devices are intermittently connected, battery-constrained, and prone to unexpected interruptions. Checkpointing is the primary resilience mechanism — replacing naive full restart strategies with deterministic, low-overhead resumption.

The checkpointing subsystem achieves four core objectives:

Objective	Description
Deterministic recovery	Always resume from a known-good consistent state
Bounded storage growth	Compaction prevents unbounded delta chain growth
Minimal runtime overhead	Delta-only persistence on hot paths
Transparent failure handling	No developer-managed retry logic required

---

Checkpoint Types

CROWDio defines three structured checkpoint types forming a tiered persistence model:

Type	Description	When Created
BASE	Full snapshot of task execution state	First checkpoint per task; after compaction threshold
DELTA	Incremental changes since the last checkpoint	Every subsequent checkpoint
COMPACTED	Merged BASE + deltas into a new full snapshot	After every 50 deltas (configurable)

BASE Checkpoints

A complete, authoritative state snapshot. Guarantees:

• Deterministic restart capability — independent of prior execution environments.
• Safe task migration — task can move to a different worker without loss.

DELTA Checkpoints

Only the modified state variables since the previous checkpoint are persisted:

Only {variable: new_value} pairs where value changed

Benefits:

• Reduced network transfer overhead
• Minimal serialization cost
• Small storage footprint

Particularly efficient for iterative workloads (e.g., Monte Carlo) where only a subset of variables changes per iteration.

COMPACTED Checkpoints

After a threshold (default: 50 deltas), the system reconstructs a new BASE by merging the original BASE + all accumulated deltas, then discards older entries.

This ensures O(k) recovery complexity (where k = compaction threshold) rather than O(n) for an unbounded delta chain.

---

Checkpoint Storage Strategy

Checkpoint size varies significantly by task complexity. CROWDio applies a hybrid persistence strategy:

Checkpoint Size	Storage	Format
Small / frequent	SQLite (embedded)	JSONB — atomic, fast indexed retrieval
Large / infrequent	Filesystem	Binary file with lightweight DB metadata reference

Small checkpoints in SQLite: Atomic transactional guarantees, fast indexed retrieval, no filesystem overhead.

Large checkpoints on filesystem: Prevents database bloat, avoids query performance degradation, maintains scalability for tensor-sized states (e.g., ML model intermediate states).

---

Decorator API

Apply the @crowdio.task decorator from developer_sdk/decorators.py:

from developer_sdk import crowdio


@crowdio.task(
    checkpoint=True,
    checkpoint_interval=5.0,           # seconds between checkpoint polls
    checkpoint_state=["i", "partial", "progress_percent"],  # variables to capture
    retry_on_failure=True,
    max_retries=3,
)
def long_task(n):
    partial = 0
    progress_percent = 0.0

    for i in range(n):
        partial += i
        progress_percent = ((i + 1) / n) * 100

    return partial

Decorator Parameters

Parameter	Type	Default	Description
checkpoint	bool	False	Enable checkpointing
checkpoint_interval	float	5.0	Seconds between checkpoint captures
checkpoint_state	list[str]	[]	Variable names to include in state snapshot
retry_on_failure	bool	False	Automatically retry failed tasks
max_retries	int	0	Maximum retry attempts

---

Checkpoint Payload Format

State is serialized on the worker side as:

Python dict → JSON string → GZIP bytes → hex-encoded string

Transported in the delta_data_hex field of the task_checkpoint WebSocket message.

---

Recovery Flow

Failure recovery is deterministic, scheduler-driven, and transparent to both developer and client.

sequenceDiagram
    participant W as Worker
    participant F as Foreman
    participant DB as SQLite / Storage
    participant W2 as New Worker

    W-->>F: worker heartbeat stops (timeout)
    F->>F: mark worker OFFLINE
    F->>DB: lock latest consistent checkpoint
    F->>DB: reconstruct BASE + ordered DELTAs
    F->>F: validate checkpoint integrity + chain completeness
    alt Inconsistency detected
        F->>DB: roll back to previous valid BASE
    end
    F->>F: apply AST code instrumentation
    F->>W2: resume_task (instrumented code + checkpoint payload)
    W2->>W2: restore state + resume loop from checkpoint index
    W2-->>F: task_result (recoveryStatus = resumed)

Failure Detection

Each worker emits a periodic heartbeat. If no heartbeat is received within the timeout window, the worker is classified as Offline.

To reduce false positives from temporary mobile network instability:

• Configurable timeout thresholds — tune per environment
• Grace period retries — wait before declaring failure
• Context-aware scheduling — don’t re-assign if reconnection is likely

---

AST-Based Code Instrumentation (Android Workers)

Android workers run Python via Chaquopy, which has significant restrictions versus a standard desktop Python:

Capability	Desktop Python	Android/Chaquopy
sys.settrace() frame introspection	✅ Supported	❌ Not supported
Full execution frame metadata	✅ Available	❌ Restricted
Low-level debugging hooks	✅ Available	❌ Unavailable

Because runtime-based checkpoint capture cannot be applied on Android, CROWDio uses Python AST (Abstract Syntax Tree) transformation to inject checkpoint logic at compile time.

Transformation Steps

# 1. Parse function source to AST
tree = ast.parse(source_code)

# 2. Traverse nodes using custom visitors
# 3. Modify FunctionDef, Assign, For, While nodes
# 4. Generate modified source
modified_source = ast.unparse(transformed_tree)

The transformation is deterministic and repeatable — instrumented code is always syntactically valid and semantically equivalent to the original.

---

[1] Variable State Injection (Resume Transformation)

Checkpoint variables are reassigned from recovered values before execution continues:

# Original:
count = 0
total = 0.0

# Instrumented (on resume):
count = <checkpoint_value>   # e.g. 499
total = <checkpoint_value>   # e.g. 124250.5

[2] Loop Progress Adjustment

To avoid recomputing completed iterations:

# Original:
for i in range(n):
    ...

# Instrumented (on resume, if 45% complete):
for i in range(resume_index, n):   # resume_index = int(0.45 * n)
    ...

Guarantees: no duplicate execution, no skipped iterations, deterministic continuation.

[3] Checkpoint Injection into Loop Bodies

An internal _ckpt_update() call is injected after each state mutation:

# Original:
for i in range(n):
    total += compute(data[i])
    progress_percent = ((i + 1) / n) * 100

# Instrumented:
for i in range(n):
    total += compute(data[i])
    progress_percent = ((i + 1) / n) * 100
    _ckpt_update({
        "total": total,
        "progress_percent": progress_percent
    })

Only variables declared in checkpoint_state are captured — internal variables are excluded, keeping state dict size minimal.

[4] Mobile Checkpoint Wrapper

A lightweight _MobileCheckpointState class is prepended to the instrumented code:

Responsibility	Detail
Maintain latest state snapshot	Thread-safe in-memory dict
_ckpt_update(state_dict)	Called from inside loop body
_ckpt_get_state()	Called by external checkpoint handler to serialize and transmit

This decouples state capture (inside user function) from state serialization (handled by worker runtime), ensuring modularity and minimal overhead.

---

Running the Checkpoint Demo

python tests/example_checkpoint_client.py localhost

---

Checkpoint REST Endpoints

Endpoint	Method	Description
/api/checkpoints/job/{job_id}	GET	All checkpoint records for a job
/api/checkpoints/recovery-events	GET	All recovery events across all jobs

curl http://localhost:8000/api/checkpoints/job/<job_id>
curl http://localhost:8000/api/checkpoints/recovery-events

---

Checkpoint Frequency Results

From the Monte Carlo evaluation — 5-second intervals provide the optimal performance/recovery tradeoff:

Interval	Execution Time (5M trials)
5 s	15.267 s
2 s	15.431 s
1 s	15.204 s
0.5 s	15.610 s
0.1 s	16.770 s
0.05 s	17.880 s

See full checkpointing evaluation results for throughput, fairness, and scalability data.

---

Inspecting Checkpoints Locally

python tests/view_checkpoints.py