Metadata-Version: 2.4
Name: pyeasythreads
Version: 0.1.0
Summary: A simple wrapper around ThreadPoolExecutor with retries, backoff, and rate limiting
Author-email: Phant0m1zed <myworkdesk2007@gmail.com>
License: Apache-2.0
Project-URL: Homepage, https://github.com/Phant0m1zed/easythreads
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Dynamic: license-file

# easythreads

**Safe, Simple, Efficient parallelism for Python => All in one function call.**

## Before

```python
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_all(fn, items, workers=10, max_retries=3, rate_limit=5):
    values = [None] * len(items)
    errors = {}
    remaining = list(enumerate(items))
    timestamps = []
    lock = threading.Lock()

    def rate_limited(item):
        while True:
            with lock:
                now = time.monotonic()
                timestamps[:] = [t for t in timestamps if now - t < 1]
                if len(timestamps) < rate_limit:
                    timestamps.append(now)
                    break
            time.sleep(0.05)
        return fn(item)

    for attempt in range(max_retries + 1):
        with ThreadPoolExecutor(max_workers=workers) as executor:
            futures = {executor.submit(rate_limited, item): i for i, item in remaining}
            remaining = []
            for future in as_completed(futures):
                index = futures[future]
                try:
                    values[index] = future.result()
                except Exception as e:
                    remaining.append((index, items[index]))
                    errors[index] = e
        if not remaining or attempt == max_retries:
            break
        time.sleep(2 ** attempt)

    return values, errors

results, errors = fetch_all(fetch_url, urls)
```

Roughly 40 lines, and this version doesn't even validate its own inputs, guard against the `bool`-is-`int` trap, or cap thread count safely at scale. A production version runs closer to 60-80 lines.

## After

```python
from easythreads import map_parallel

results = map_parallel(fetch_url, urls, workers=10, retries=3, rate_limit=5)

results.values   # successful outputs, in original input order
results.errors   # {index: exception} for items that failed permanently
```

Four lines. Same results. Ordered output, isolated per-item failures, retries with backoff, a global rate limit reducing the chance to get any of it subtly wrong.

---

## What it is

`easythreads` is a small, dependency-free library that wraps `concurrent.futures.ThreadPoolExecutor` with the four things every real parallel workload eventually needs:

1. **Ordered results:** `.values` always matches your input order, even though threads finish in whatever order they finish in.
2. **Per-item error isolation:** one failing item never crashes the rest of the batch.
3. **Retries with backoff** failed items are automatically retried (linear or exponential delay), without re-running items that already succeeded.
4. **Global rate limiting:** cap total throughput across the *entire* pool, not per-worker, using a sliding-window algorithm, so you can safely respect an API's rate limit no matter how many workers you configure.

All of it in about 120 lines of readable, auditable source — lightweight, no dependencies beyond the Python standard library.

---

## Installation

```bash
pip install pyeasythreads
```

---

## Quick example

```python
from easythreads import map_parallel
import requests

def fetch(url):
    return requests.get(url, timeout=5).status_code

urls = [
    "https://example.com",
    "https://httpbin.org/status/500",   # will fail, then retry
    "https://github.com",
]

result = map_parallel(
    fetch, urls,
    workers=3,
    retries=2,
    backoff=("exponential", 0.5),
    rate_limit=5,          # never exceed 5 calls/sec, combined across all workers
)

print(result.values)   # [200, 500, 200]  (or None for the one that never recovered)
print(result.errors)   # {1: ConnectionError(...)} if it never succeeded
```

---

## What you get

| Feature | Description |
|---|---|
| **Ordered results** | `result.values` always matches your input order, regardless of which thread finishes first. |
| **Per-item error isolation** | One item failing never crashes the batch. Every other item still runs to completion. |
| **Retries with backoff** | Failed items are automatically retried, with linear or exponential delay between rounds. Only *failing* items are retried — successes are never redundantly re-run. |
| **Global rate limiting** | Cap total throughput (calls/sec) across the *entire* worker pool — not per-worker — using a sliding-window algorithm, so you never blow past an API's rate limit no matter how many workers you configure. |
| **Sane defaults, hard validation** | Bad input (wrong types, negative numbers, `True`/`False` silently masquerading as `1`/`0`) is rejected immediately with a clear error — not a confusing crash three layers deep. |

---

## API

```python
map_parallel(fn, items, workers=None, retries=None, backoff=("linear", 1), rate_limit=None)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `fn` | callable | — | Function to call on each item. |
| `items` | list / tuple | — | Inputs to process. Must be non-empty. |
| `workers` | int | `min(32, len(items))` | Max concurrent threads. Capped at 32 by default to avoid OS thread exhaustion on large inputs — pass an explicit value to override. |
| `retries` | int | `2` | Retry attempts after a failure, per item. |
| `backoff` | `(mode, delay)` | `("linear", 1)` | `mode` is `"linear"` or `"exponential"`; `delay` is the base wait in seconds between retry rounds. Only relevant when `retries > 0`. |
| `rate_limit` | int | `None` (unlimited) | Max combined calls/sec across all workers. |

**Returns** a `Data` object:
- `.values` — list of results, same length and order as `items`. Permanently-failed items are `None`.
- `.errors` — `{index: Exception}` for items that never succeeded, even after all retries.

---

## Benchmarks

All numbers below are from real runs, an I/O-bound simulated task (`time.sleep(0.01)` per call, standing in for a network request), measured with `time.perf_counter()`.

### easythreads vs. sequential loop, vs. raw `ThreadPoolExecutor`

| n | sequential | raw pool | easythreads | speedup vs sequential | overhead vs raw pool |
|---:|---:|---:|---:|---:|---:|
| 100 | 1.014s | 0.073s | 0.064s | **15.8x** | ~0% (noise) |
| 500 | 5.096s | 0.269s | 0.284s | **17.9x** | +5.5% |
| 1000 | 10.260s | 0.524s | 0.525s | **19.5x** | +0.2% |

The retry/error/ordering machinery costs almost nothing. Overhead shrinks toward zero as batch size grows, because it's a small fixed cost per item, not per-second.

### Effect of `workers` (n=500)

| workers | time | notes |
|---:|---:|---|
| 1 | 5.138s | effectively sequential |
| 5 | 1.127s | 4.5x |
| 20 | 0.268s | 19x |
| 50 | 0.132s | 39x |
| 100 | 0.098s | ~optimal for this workload |
| 500 | 0.105s | past the point of diminishing returns — more threads !=(not equal) more speed |

### Effect of `rate_limit` (n=100, workers=50)

| rate_limit | time | theoretical floor |
|---:|---:|---|
| unlimited | 0.033s | — |
| 50 | 1.017s | 2.0s |
| 20 | 4.015s | 5.0s |
| 10 | 9.016s | 10.0s |
| 5 | 19.018s | 20.0s |

This isn't overhead — It's the rate limiter doing exactly its job. The enforcement tracks the mathematical minimum closely, meaning very little time is wasted beyond what the constraint itself requires.

### Scaling safety: the `workers` default fix

An earlier version defaulted `workers` to `len(items)` => one OS thread per item, unbounded:

| n | old default (`workers=len(items)`) | current default (`workers=min(32, n)`) |
|---:|---|---|
| 30,000 | 9.586s | 9.909s |
| 50,000 | 19.813s | — |
| 100,000 | **crashed after 144s** — `RuntimeError: can't start new thread` | **33.048s, no crash** |

At scale, an unbounded thread-per-item default is a real footgun, not just a theoretical concern. `easythreads` caps concurrency by default specifically to avoid this.

---

## Complexity

- **Time:** `O(n)` calls to `fn` in the common case; `O(n · r)` in the worst case where every item exhausts all `r` retries (`r` is a small constant, so this remains effectively linear). Wall-clock time approaches `O(n / workers)` for I/O-bound work.
- **Space:** `O(n)` — dominated by the `values` list and per-round bookkeeping. The rate-limiter's timestamp list is bounded by `O(rate_limit)`, independent of `n`.

---

## Tested

83 pytest tests, 0 failures, covering input validation, ordering under variable completion times, per-item error isolation, retry/backoff timing, and rate-limit enforcement (including that extra workers can't be used to bypass a configured `rate_limit`).

```bash
pytest tests/ -v --tb=short
```

---

## Roadmap

- [ ] Token-bucket rate limiting option (currently sliding-window log Algorithm)
- [ ] `asyncio`-based variant for coroutine workloads
- [ ] Optional progress callback
- [ ] Per-item timeout support

---

## License

Apache License 2.0 — see [LICENSE](LICENSE).
