Ray Distributed Framework

import ray
import time
import os

# Initialize Ray (connects to an existing cluster or starts a local one)
# For a cluster, you might use ray.init(address="auto") or specify an address.
# For local testing, ray.init() is sufficient.
ray.init(address=os.environ.get('RAY_ADDRESS', None), ignore_reinit_error=True)

@ray.remote
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci.remote(n - 1) + fibonacci.remote(n - 2)

@ray.remote
def slow_square(x):
    time.sleep(1) # Simulate a slow computation
    return x * x

if __name__ == '__main__':
    print("--- Ray Tasks Example ---")
    # Run tasks in parallel
    futures = [slow_square.remote(i) for i in range(5)]
    results = ray.get(futures)
    print(f"Results from slow_square: {results}")

    # Example of recursive Ray tasks (note: fibonacci is a common but inefficient example for Ray's overhead)
    # For larger N, this can quickly create too many tasks.
    # result_fib = ray.get(fibonacci.remote(10))
    # print(f"Fibonacci(10) using Ray: {result_fib}")

    print("Ray initialized successfully, dashboard at:", ray.get_dashboard_url())

    ray.shutdown()
    print("Ray shutdown.")