makeParallel 🚀

The easiest way to speed up your Python code using all your CPU cores.

Got a slow, CPU-heavy task in Python? makeParallel lets you run it on a separate core with a single line of code, so you can get results up to 4x, 8x, or even 16x faster without blocking your main program.

It's powered by Rust to safely bypass Python's Global Interpreter Lock (GIL), giving you true parallelism without the complexity of multiprocessing.

🤔 What's the "GIL"?

Python has a rule called the Global Interpreter Lock (GIL) that only lets one thread run at a time, even on a multi-core CPU. For tasks that just wait for networks (I/O-bound), this is fine. But for heavy calculations (CPU-bound), it means Python can't use all the power your computer has. makeParallel fixes this.

✨ Why You'll Love `makeParallel`

So Simple: Just add the @parallel decorator to any function. That's it!
True Speed-Up: Uses Rust threads to run your code on all available CPU cores.
Doesn't Block: Your main application stays responsive while the work happens in the background.
Smart Callbacks: Monitor progress, handle completion, catch errors - all with simple callbacks.
Task Dependencies: Build complex pipelines where tasks automatically wait for their dependencies.
Auto Progress Tracking: Report progress from within tasks without managing task IDs.
No multiprocessing Headaches: Avoids the complexity, memory overhead, and data-sharing issues of multiprocessing.
Production Ready: Built-in error handling, timeouts, cancellation, and graceful shutdown.
Works with Your Code: Decorate any function, even class methods.

📦 Installation

Installing is as simple as:

pip install makeparallel

Or, to build it from the source:

# Clone the repository
git clone https://github.com/amiyamandal-dev/makeParallel.git
cd makeParallel
# Build and install locally
pip install .

🚀 Quick Start

Let's say you have a function that does a lot of math and slows down your program.

Before: Your code waits...

importtimedefcpu_intensive_task(n):
# A slow calculationreturnsum(i*iforiinrange(n))
start=time.time()
result=cpu_intensive_task(20_000_000) # This blocks everything!print(f"Got result: {result} in {time.time() -start:.2f}s")

After: Instant and non-blocking!

importtimefrommakeparallelimportparallel@parallel# Just add this decorator!defcpu_intensive_task(n):
# The same slow calculationreturnsum(i*iforiinrange(n))
start=time.time()
# The function returns instantly with a "handle"handle=cpu_intensive_task(20_000_000)
print("The task is running in the background, my app is still responsive!")
# You can do other work here...# Now, get the result (this will wait until it's ready)result=handle.get()
print(f"Got result: {result} in {time.time() -start:.2f}s")

In the example above, handle.get() blocks until the result is ready. You can also check if it's done without waiting:

ifhandle.is_ready():
print("It's done!")
else:
print("Still working...")

🤔 When Should I Use This?

makeParallel is for CPU-bound tasks. These are operations that require a lot of computation, like:

heavy data processing, or scientific computing.
Image or video processing.
Complex simulations.

For I/O-bound tasks (like waiting for a web request or reading a file), Python's built-in threading or asyncio are usually a better fit.

📚 Complete Feature Guide

makeParallel comes with many powerful decorators and utilities.

🔥 Parallel Execution Decorators

`@parallel` - Full-featured parallel execution with callbacks and advanced control

frommakeparallelimportparallel, report_progress@paralleldefcpu_intensive_task(n):
foriinrange(0, n, n//10):
# Report progress automatically (no task_id needed!)report_progress(i/n)
# Do work...returnsum(i*iforiinrange(n))
# Returns immediately with an AsyncHandlehandle=cpu_intensive_task(20_000_000, timeout=5.0)
# Set up callbacks (execute automatically when task completes)handle.on_progress(lambdap: print(f"Progress: {p*100:.0f}%"))
handle.on_complete(lambdaresult: print(f"Success! Result: {result}"))
handle.on_error(lambdaerror: print(f"Error occurred: {error}"))
# Check statusifhandle.is_ready():
result=handle.get() # Callbacks fire here# Try to get result without blockingresult=handle.try_get() # Returns None if not ready# Cancel a running taskhandle.cancel()
ifhandle.is_cancelled():
print("Task was cancelled")
# Get task infoprint(f"Task ID: {handle.get_task_id()}")
print(f"Elapsed: {handle.elapsed_time()}s")
print(f"Progress: {handle.get_progress()}")
# Add metadatahandle.set_metadata("user_id", "user-123")
metadata=handle.get_all_metadata()

`@parallel_fast` - Optimized with lock-free channels (crossbeam)

frommakeparallelimportparallel_fast@parallel_fastdeffast_task(x):
returnx**2handle=fast_task(10)
result=handle.get() # Faster channel communication

`@parallel_pool` - Uses Rayon thread pool (best for many small tasks)

frommakeparallelimportparallel_pool@parallel_pooldefsmall_task(x):
returnx*2# Efficiently handles many concurrent taskshandles= [small_task(i) foriinrange(1000)]
results= [h.get() forhinhandles]

`@parallel_priority` - Priority-based execution

frommakeparallelimportparallel_priority, start_priority_worker, stop_priority_worker# Start the priority worker before using priority tasksstart_priority_worker()
@parallel_prioritydeftask(data):
returnprocess(data)
# High priority tasks execute first (higher number = higher priority)low=task(data1, priority=1)
high=task(data2, priority=10) # Executes first# Get resultslow_result=low.get()
high_result=high.get()
# Stop the worker when donestop_priority_worker()

`@parallel_with_deps` - Task dependencies and pipelines

frommakeparallelimportparallel_with_deps@parallel_with_depsdefstep1():
return"data from step 1"@parallel_with_depsdefstep2(deps):
# deps is a tuple of all dependency resultsdata=deps[0] # Result from step1returnf"processed {data}"@parallel_with_depsdefstep3(deps):
result=deps[0] # Result from step2returnf"final: {result}"# Build dependency chainh1=step1()
h2=step2(depends_on=[h1]) # Automatically waits for h1h3=step3(depends_on=[h2]) # Automatically waits for h2# Execute entire pipelinefinal=h3.get() # Returns: "final: processed data from step 1"

🎯 Callbacks and Event Handling

makeParallel provides a powerful callback system for monitoring task execution:

frommakeparallelimportparallel, report_progress@paralleldefdownload_file(url):
# Simulate download with progressforiinrange(100):
download_chunk(url, i)
# Report progress (task_id is automatic!)report_progress(i/100.0)
returnf"Downloaded {url}"handle=download_file("https://example.com/large_file.zip")
# Set up callbackshandle.on_progress(lambdap: print(f"Downloaded: {p*100:.1f}%"))
handle.on_complete(lambdaresult: notify_user(result))
handle.on_error(lambdaerror: log_error(error))
# Callbacks fire automatically when you get the resultresult=handle.get()

Callback Types:

on_progress(callback) - Called when report_progress() is called inside task
on_complete(callback) - Called when task succeeds (receives result)
on_error(callback) - Called when task fails (receives error string)

Key Features:

✅ Automatic task_id tracking (no need to pass task_id!)
✅ Thread-safe callback execution
✅ Error isolation (callback failures don't crash tasks)
✅ Progress validation (NaN/Infinity rejected)

🗺️ Batch Processing

`parallel_map` - Process lists in parallel

frommakeparallelimportparallel_mapdefprocess_data(item):
returnitem*2my_large_list=list(range(10000))
results=parallel_map(process_data, my_large_list)

`gather` - Collect results from multiple handles

frommakeparallelimportparallel, gather@paralleldeftask(x):
returnx**2handles= [task(i) foriinrange(10)]
# Wait for all and collect resultsresults=gather(handles, on_error="raise") # or "skip" or "none"

`ParallelContext` - Context manager for parallel tasks

frommakeparallelimportParallelContext, parallel@paralleldeftask(x):
returnx*2withParallelContext(timeout=10.0) asctx:
handle1=ctx.submit(task, (5,))
handle2=ctx.submit(task, (10,))
# All tasks complete when exiting context

💾 Caching Decorators

`@memoize` - Cache function results

frommakeparallelimportmemoize@memoizedeffibonacci(n):
ifn<=1:
returnnreturnfibonacci(n-1) +fibonacci(n-2)
fibonacci(35) # Slow first timefibonacci(35) # Instant second time

`@memoize_fast` - Lock-free concurrent cache (DashMap)

frommakeparallelimportmemoize_fast@memoize_fastdefexpensive_computation(x, y):
returnx**y# Safe for concurrent access from multiple threads

🔁 Retry Logic

`@retry` - Simple retry with fixed delays

frommakeparallelimportretry@retry(max_retries=3)defflaky_api_call():
# Will retry up to 3 times on failurereturnmake_request()

`@retry_backoff` - Retry with exponential backoff

frommakeparallelimportretry_backoff@retry_backoff(max_attempts=5,backoff="exponential", # or "linear"initial_delay=1.0,max_delay=60.0)defunreliable_task():
returndo_something()

📊 Performance Monitoring

`@profiled` - Automatic performance tracking

frommakeparallelimportprofiled, get_metrics, get_all_metrics@profileddeftracked_function(x):
returnx**2foriinrange(100):
tracked_function(i)
# Get metrics for specific functionmetrics=get_metrics("tracked_function")
print(f"Total tasks: {metrics.total_tasks}")
print(f"Completed: {metrics.completed_tasks}")
print(f"Failed: {metrics.failed_tasks}")
print(f"Avg time: {metrics.average_execution_time_ms}ms")
# Get all metricsall_metrics=get_all_metrics()

`@timer` - Simple execution timing

frommakeparallelimporttimer@timerdefmy_function():
# Do work...passmy_function() # Prints execution time

`@CallCounter` - Count function invocations

frommakeparallelimportCallCounter@CallCounterdefcounted_function():
return"result"counted_function()
counted_function()
print(counted_function.call_count) # 2counted_function.reset()

⚙️ Advanced Configuration

Thread Pool Configuration

frommakeparallelimportconfigure_thread_pool, get_thread_pool_info# Configure global thread poolconfigure_thread_pool(num_threads=8, stack_size=2*1024*1024)
# Get current infoinfo=get_thread_pool_info()
print(info["current_num_threads"])

Backpressure and Resource Management

frommakeparallelimportset_max_concurrent_tasks, configure_memory_limit# Limit concurrent tasks to prevent overloadset_max_concurrent_tasks(100)
# Set memory limit (percentage)configure_memory_limit(max_memory_percent=80.0)

Progress Reporting and Callbacks

frommakeparallelimportparallel, report_progress@paralleldeflong_task():
foriinrange(100):
# Report progress from within task (task_id is automatic!)report_progress(i/100.0)
# Do work...return"done"handle=long_task()
# Set up callbackshandle.on_progress(lambdap: print(f"Progress: {p*100:.1f}%"))
handle.on_complete(lambdaresult: print(f"Finished: {result}"))
handle.on_error(lambdaerror: print(f"Error: {error}"))
# Get result (callbacks fire automatically)result=handle.get()

Task Dependencies

frommakeparallelimportparallel_with_deps@parallel_with_depsdeffetch_data():
return {"users": 100, "orders": 500}
@parallel_with_depsdefprocess_data(deps):
# deps[0] contains result from fetch_datadata=deps[0]
returnf"Processed {data['users']} users"@parallel_with_depsdefsave_results(deps):
# deps[0] contains result from process_dataprocessed=deps[0]
returnf"Saved: {processed}"# Build a dependency pipelineh1=fetch_data()
h2=process_data(depends_on=[h1]) # Waits for h1h3=save_results(depends_on=[h2]) # Waits for h2# Execute the entire pipelinefinal_result=h3.get() # Returns: "Saved: Processed 100 users"

Graceful Shutdown

frommakeparallelimportshutdown, get_active_task_count, reset_shutdown# Get active task countprint(f"Active tasks: {get_active_task_count()}")
# Graceful shutdown with timeoutsuccess=shutdown(timeout_secs=30.0, cancel_pending=True)
# Reset after shutdown (for testing)reset_shutdown()

🎯 Choosing the Right Decorator

Decorator	Best For	Performance	Features
`@parallel`	Most use cases	Good	Full control: timeout, cancel, metadata, progress
`@parallel_fast`	High-throughput tasks	Better	Lock-free channels (crossbeam)
`@parallel_pool`	Many small tasks	Best	Rayon thread pool, efficient resource usage
`@parallel_priority`	Priority-based scheduling	Good	Priority queue execution
`parallel_map`	Batch processing lists	Best	Automatic parallelization across items

Quick Decision Guide:

Single long task with monitoring? → @parallel
Thousands of small tasks? → @parallel_pool
Processing a large list? → parallel_map
Need priority scheduling? → @parallel_priority
Maximum throughput? → @parallel_fast

🏗️ How It Works

Here's a simple breakdown of what happens when you call a @parallel function:

Python Side: Your main program calls the function but doesn't run it directly. Instead, it sends the function and its arguments to the Rust backend.
Rust Backend:
- It immediately returns the AsyncHandle object to your Python code so it doesn't have to wait.
- It releases Python's Global Interpreter Lock (GIL).
- It spawns a new Rust OS thread (a real parallel thread).
- Inside the new thread, it re-acquires the GIL to safely execute your Python function.
Result: The result is sent back to the AsyncHandle, which your main program can access with .get().

This GIL-release-and-reacquire step is the key to unlocking true parallelism for CPU-bound Python code.

🔧 Best Practices

✅ Do's

Use for CPU-bound tasks: Heavy computation, data processing, mathematical operations
Combine with caching: Use @memoize or @memoize_fast to avoid redundant calculations
Monitor with profiling: Use @profiled to track performance and identify bottlenecks
Set resource limits: Use set_max_concurrent_tasks() to prevent overload
Handle errors gracefully: Use gather() with appropriate on_error strategy
Use timeouts for long tasks: Add timeout parameter to prevent hanging

❌ Don'ts

Don't use for I/O-bound tasks: Use asyncio or threading instead
Don't pass large objects: Minimize data transfer between threads
Don't ignore error handling: Always check results or use try/except
Don't spawn unlimited tasks: Use set_max_concurrent_tasks() for backpressure
Don't forget cleanup: Use shutdown() for graceful termination

💡 Performance Tips

# ✅ Good: Process large batches efficientlyresults=parallel_map(heavy_computation, large_list)
# ❌ Bad: Creating too many individual parallel taskshandles= [parallel_task(x) forxinrange(10000)] # Overhead!# ✅ Good: Use memoization for repeated calls@memoize_fast@parallel_pooldefcached_parallel_task(x):
returnexpensive_operation(x)
# ✅ Good: Configure thread pool for your workloadconfigure_thread_pool(num_threads=8) # Match your CPU cores

🆚 Comparison with Alternatives

Feature	makeParallel	multiprocessing	threading	asyncio
True Parallelism	✅ Yes	✅ Yes	❌ No (GIL)	❌ No (GIL)
CPU-bound Tasks	✅ Excellent	✅ Good	❌ Poor	❌ Poor
I/O-bound Tasks	⚠️ Okay	⚠️ Okay	✅ Good	✅ Excellent
Memory Overhead	✅ Low	❌ High	✅ Low	✅ Low
Easy to Use	✅ Very Easy	⚠️ Complex	✅ Easy	⚠️ Moderate
Data Sharing	✅ Simple	❌ Complex	✅ Simple	✅ Simple
Performance	✅✅ Fast (Rust)	✅ Good	⚠️ Limited	✅ Good
Cancellation	✅ Built-in	⚠️ Manual	⚠️ Manual	✅ Built-in
Progress Tracking	✅ Built-in	❌ Manual	❌ Manual	⚠️ Manual

📖 Real-World Examples

Example 1: Image Processing Pipeline

frommakeparallelimportparallel_map, profiledfromPILimportImageimportos@profileddefprocess_image(image_path):
img=Image.open(image_path)
# Resize, apply filters, etc.img_resized=img.resize((800, 600))
# Save processed imageoutput_path=f"processed_{os.path.basename(image_path)}"img_resized.save(output_path)
returnoutput_path# Process 1000 images in parallelimage_files= [f"image_{i}.jpg"foriinrange(1000)]
processed=parallel_map(process_image, image_files)
print(f"Processed {len(processed)} images")

Example 2: Web Scraping with Retry Logic

frommakeparallelimportparallel_pool, retry_backoffimportrequests@retry_backoff(max_attempts=3, backoff="exponential")@parallel_pooldeffetch_url(url):
response=requests.get(url, timeout=10)
returnresponse.texturls= ["https://example.com/page1", "https://example.com/page2", ...]
handles= [fetch_url(url) forurlinurls]
results= [h.get() forhinhandles]

Example 3: Data Analysis with Progress Tracking and Callbacks

frommakeparallelimportparallel, report_progressimportpandasaspd@paralleldefanalyze_dataset(file_path):
df=pd.read_csv(file_path)
total_rows=len(df)
results= []
fori, rowindf.iterrows():
# Report progress (task_id is automatic!)report_progress(i/total_rows)
# Perform analysisresult=complex_analysis(row)
results.append(result)
returnresultshandle=analyze_dataset("large_dataset.csv")
# Set up callbacks for monitoringhandle.on_progress(lambdap: print(f"Analyzed: {p*100:.1f}%"))
handle.on_complete(lambdaresults: print(f"Analysis complete! {len(results)} rows"))
handle.on_error(lambdae: print(f"Analysis failed: {e}"))
# Get results (callbacks fire automatically)final_results=handle.get()

Example 4: ETL Pipeline with Task Dependencies

frommakeparallelimportparallel_with_deps@parallel_with_depsdefextract_data(source):
# Fetch data from database/APIprint(f"Extracting from {source}...")
returnfetch_raw_data(source)
@parallel_with_depsdeftransform_data(deps):
# deps[0] contains result from extract_dataraw_data=deps[0]
print("Transforming data...")
returnclean_and_transform(raw_data)
@parallel_with_depsdefvalidate_data(deps):
# deps[0] contains result from transform_datatransformed=deps[0]
print("Validating data...")
returnrun_validation_checks(transformed)
@parallel_with_depsdefload_data(deps):
# deps[0] contains result from validate_datavalidated=deps[0]
print("Loading to warehouse...")
returninsert_into_warehouse(validated)
# Build ETL pipeline with dependenciesh1=extract_data("production_db")
h2=transform_data(depends_on=[h1]) # Waits for extracth3=validate_data(depends_on=[h2]) # Waits for transformh4=load_data(depends_on=[h3]) # Waits for validate# Execute entire pipelineresult=h4.get() # Blocks until all dependencies completeprint(f"Pipeline complete: {result}")

Example 5: Machine Learning Model Training

frommakeparallelimportparallel, gather, configure_thread_poolfromsklearn.model_selectionimporttrain_test_splitfromsklearn.ensembleimportRandomForestClassifier# Configure thread pool for ML workloadconfigure_thread_pool(num_threads=4)
@paralleldeftrain_model(params):
model=RandomForestClassifier(**params)
model.fit(X_train, y_train)
score=model.score(X_test, y_test)
return {"params": params, "score": score}
# Train multiple models with different hyperparametersparam_grid= [
{"n_estimators": 100, "max_depth": 10},
{"n_estimators": 200, "max_depth": 15},
{"n_estimators": 300, "max_depth": 20},
]
handles= [train_model(params) forparamsinparam_grid]
results=gather(handles)
# Find best modelbest=max(results, key=lambdax: x["score"])
print(f"Best params: {best['params']}, Score: {best['score']}")

🐛 Troubleshooting

My tasks are running slowly

Check if your task is CPU-bound (use @profiled to measure)
For I/O-bound tasks, use asyncio instead
Ensure you're not creating too many tasks (use parallel_map for batch processing)
Configure thread pool: configure_thread_pool(num_threads=<cpu_cores>)

Tasks are hanging

Add timeouts: @parallel def task(): ... then task(timeout=10.0)
Use cancel() to stop stuck tasks
Check for deadlocks in your Python code

Memory usage is too high

Limit concurrent tasks: set_max_concurrent_tasks(100)
Set memory limit: configure_memory_limit(max_memory_percent=80.0)
Use @parallel_pool instead of spawning individual threads
Process data in smaller batches

Errors are being swallowed

Always check handle.get() in a try/except block
Use gather() with on_error="raise" to see all errors
Enable profiling to see failed task counts: @profiled
Use on_error callbacks to capture errors: handle.on_error(lambda e: print(e))

Callbacks not firing

Make sure you call handle.get() or handle.wait() to trigger callbacks
Callbacks execute during result retrieval
Check callback syntax: handle.on_progress(lambda p: print(p))

Dependencies hanging

Check for circular dependencies (task A depends on B, B depends on A)
Verify all dependencies complete successfully
Use timeouts: task(depends_on=[h1], timeout=60.0)
Enable logging to see dependency errors: RUST_LOG=makeparallel=debug

🤝 Contributing

Contributions are welcome! If you want to help improve makeParallel, please feel free to open an issue or submit a pull request.

Development Setup

# Clone the repository
git clone https://github.com/amiyamandal-dev/makeParallel.git
cd makeParallel
# Create virtual environment
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate# Install development dependencies
pip install maturin
# Build and install in development mode
maturin develop
# Run tests
python tests/test_all.py

Running Tests

# Run all tests
python tests/test_all.py
# The test suite includes:# - 37 core tests covering all decorators and features# - 3 callback tests (on_progress, on_complete, on_error)# - 5 progress tracking tests# - Performance benchmarks# - Edge case validation# - Error handling verification# Run specific test suites
python test_simple_callbacks.py # Callback functionality
python test_progress_fix.py # Progress tracking

Code Quality

# Format Rust code
cargo fmt
# Lint Rust code
cargo clippy
# Format Python code (if you have ruff)
ruff format .# Check Python code
ruff check .

📝 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

Built with PyO3 for Python-Rust interop
Uses Rayon for efficient thread pool management
Uses Crossbeam for lock-free channels
Uses DashMap for concurrent caching

📬 Contact & Support

Issues: GitHub Issues
Discussions: GitHub Discussions
PyPI: pypi.org/project/makeparallel
Author: Amiya Mandal - GitHub

Made with ❤️ and Rust 🦀

Name		Name	Last commit message	Last commit date
.github		.github
benchmarks		benchmarks
docs		docs
examples		examples
src		src
tests		tests
.gitignore		.gitignore
.python-version		.python-version
Cargo.toml		Cargo.toml
LICENSE		LICENSE
PROJECT_STRUCTURE.md		PROJECT_STRUCTURE.md
README.md		README.md
README_UPDATES.md		README_UPDATES.md
pyproject.toml		pyproject.toml

Latest commit

History

Folders and files

Repository files navigation