#include <ray/api.h>
#include <cmath>
#include <iostream>
#include <vector>
#include <chrono>
#include <memory>
#include <cstdlib>
// Remote function to calculate the partial sum of a segment
double partial_sum(std::tuple<long long, long long, double> seg) {
auto [start, end, dx] = seg;
double total = 0.0;
for (long long i = start; i < end; ++i) {
double x = i * dx;
total += std::sin(x) * dx;
}
return total;
}
// Declare as a remote function
RAY_REMOTE(partial_sum);
// Distributed integration calculation function
double distributed_integration(long long m, int num_tasks) {
const double pi = std::acos(-1.0); // Define π value
double dx = pi / m; // Width of rectangles
// Task division
long long segment_size = m / num_tasks;
std::vector<std::tuple<long long, long long, double>> segments;
for (long long i = 0; i < num_tasks; ++i) {
long long start = i * segment_size;
long long end = (i == num_tasks – 1) ? m : (i + 1) * segment_size;
segments.emplace_back(start, end, dx);
}
// Submit tasks in parallel
std::vector<ray::ObjectRef<double>> futures;
for (auto& seg : segments) {
ray::internal::CallOptions options;
options.resources[“CPU”] = 2;
auto task = ray::Task(partial_sum);
task.SetResources(options.resources);
futures.push_back(task.Remote(seg));
}
// Retrieve and merge results
auto partial_results = ray::Get(futures);
double final_result = 0.0;
for (auto res : partial_results) {
final_result += *res;
}
return final_result;
}
int main(int argc, char** argv) {
try {
ray::RayConfig config;
// Set Ray address
const char* ray_address = std::getenv(“RAY_ADDRESS”);
if (ray_address) {
config.address = ray_address;
std::cout << “Using Ray address from environment variable: ” << config.address << std::endl;
} else {
config.address = “111.111.111.111:1234”;
std::cout << “Using default Ray address: ” << config.address << std::endl;
}
// Set Redis password
config.redis_password_ = “5241590000000000”;
std::cout << “Redis password has been set” << std::endl;
// Initialize Ray
std::cout << “Attempting to initialize Ray cluster…” << std::endl;
ray::Init(config, argc, argv);
std::cout << “Ray cluster initialized successfully” << std::endl;
// Integration calculation
const long long m = 20000000000; // 100 million rectangles
const double theoretical = 2.0; // Theoretical integral value
std::cout << “Calculating the integral of sin(x) over [0, π] (using ” << m << ” rectangles)” << std::endl;
std::cout << “Theoretical value: ” << std::fixed << std::setprecision(10) << theoretical << std::endl;
// High parallelism test
auto start_high = std::chrono::high_resolution_clock::now();
double result_high = distributed_integration(m, 120);
auto duration_high = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::high_resolution_clock::now() – start_high);
std::cout << “Multiple tasks: Time taken ” << duration_high.count() / 1000.0 << ” seconds”
<< ” | Result: ” << std::fixed << std::setprecision(10) << result_high
<< ” | Error: ” << std::scientific << std::fabs(result_high – theoretical) << std::endl;
ray::Shutdown();
return 0;
}
catch (const ray::internal::RayTaskException& e)
{
std::cerr << “\nProgram exception 1: ” << e.msg_ << std::endl;
ray::Shutdown();
return 1;
}
catch (const ray::internal::RayWorkerException& e)
{
std::cerr << “\nProgram exception 2: ” << e.msg_ << std::endl;
ray::Shutdown();
return 1;
}
catch (const ray::internal::UnreconstructableException& e)
{
std::cerr << “\nProgram exception 3: ” << e.msg_ << std::endl;
ray::Shutdown();
return 1;
}
catch (…)
{
std::cerr << “\nAn unknown exception occurred” << std::endl;
ray::Shutdown();
return 1;
}
}
Note that
ray::internal::CallOptions options;
options.resources[“CPU”] = 2;
auto task = ray::Task(partial_sum);
task.SetResources(options.resources); This will affect the scheduling strategy. Additionally, there is no need to synchronize the C++ source code to all nodes in advance; it is sufficient for the source code to be present on any node.