simple_demo.cpp

// Simplified Distributed Computing Demo#include <iostream>

// Works with older compilers (C++11) - No threading needed#include <vector>

// Simulates distributed computing concepts#include <queue>

#include <ctime>

#include <iostream>#include <cstdlib>

#include <vector>#include <algorithm>

#include <ctime>#include <numeric>

#include <cstdlib>#include <cmath>

#include <algorithm>

#include <numeric>// Simplified Distributed Computing Framework - Single-threaded version

#include <cmath>// Simulates distributed computing without threads (works with older compilers)



struct Task {// Task structure

    int id;struct Task {

    std::vector<double> data;    int id;

};    std::vector<double> data;

    double estimated_time;

struct Result {};

    int task_id;

    int worker_id;// Result structure

    double mean;struct Result {

    double variance;    int task_id;

    double std_dev;    int worker_id;

    double min_val;    std::vector<double> data;

    double max_val;    double execution_time;

    double execution_time;    bool success;

};};



class Worker {// Global task queue and results

public:std::queue<Task> task_queue;

    int id;std::vector<Result> results;

    int tasks_completed;std::mutex result_mutex;

    double total_time;

    // Worker status tracking

    Worker(int worker_id) : id(worker_id), tasks_completed(0), total_time(0.0) {}struct WorkerStatus {

        int id;

    Result executeTask(const Task& task) {    bool active;

        clock_t start = clock();    int tasks_completed;

            int tasks_failed;

        Result result;    double total_time;

        result.task_id = task.id;};

        result.worker_id = id;

        std::vector<WorkerStatus> worker_stats;

        // Calculate statistics

        double sum = 0.0;// Worker function - simulates a worker node

        for (size_t i = 0; i < task.data.size(); i++) {void worker_node(int worker_id, int& tasks_assigned) {

            sum += task.data[i];    {

        }        std::lock_guard<std::mutex> lock(cout_mutex);

        result.mean = sum / task.data.size();        std::cout << "Worker " << worker_id << " started and ready!\n";

            }

        // Variance    

        double sq_sum = 0.0;    while (true) {

        for (size_t i = 0; i < task.data.size(); i++) {        Task task;

            sq_sum += task.data[i] * task.data[i];        bool has_task = false;

        }        

        result.variance = sq_sum / task.data.size() - result.mean * result.mean;        // Try to get a task from queue

        result.std_dev = sqrt(result.variance);        {

                    std::lock_guard<std::mutex> lock(queue_mutex);

        // Min and Max            if (!task_queue.empty()) {

        result.min_val = *std::min_element(task.data.begin(), task.data.end());                task = task_queue.front();

        result.max_val = *std::max_element(task.data.begin(), task.data.end());                task_queue.pop();

                        has_task = true;

        clock_t end = clock();                tasks_assigned++;

        result.execution_time = double(end - start) / CLOCKS_PER_SEC;            } else if (tasks_assigned >= 20) {

                        // All tasks distributed and queue empty

        tasks_completed++;                break;

        total_time += result.execution_time;            }

                }

        return result;        

    }        if (has_task) {

};            {

                std::lock_guard<std::mutex> lock(cout_mutex);

class LoadBalancer {                std::cout << "Worker " << worker_id << " executing Task " << task.id << "...\n";

    int num_workers;            }

    int current_worker;            

                auto start = std::chrono::high_resolution_clock::now();

public:            

    LoadBalancer(int workers) : num_workers(workers), current_worker(0) {}            // Simulate computation: calculate statistics

                Result result;

    int assignTask() {            result.task_id = task.id;

        int assigned = current_worker;            result.worker_id = worker_id;

        current_worker = (current_worker + 1) % num_workers;            result.success = true;

        return assigned;            

    }            try {

};                // Calculate mean

                double sum = std::accumulate(task.data.begin(), task.data.end(), 0.0);

int main() {                double mean = sum / task.data.size();

    std::cout << "========================================\n";                result.data.push_back(mean);

    std::cout << "  Distributed Computing Framework Demo  \n";                

    std::cout << "  Aug 2022 - Nov 2022 Project          \n";                // Calculate variance

    std::cout << "========================================\n\n";                double sq_sum = std::inner_product(task.data.begin(), task.data.end(), 

                                                       task.data.begin(), 0.0);

    const int NUM_WORKERS = 4;                double variance = sq_sum / task.data.size() - mean * mean;

    const int NUM_TASKS = 20;                result.data.push_back(variance);

                    

    std::cout << "System Configuration:\n";                // Calculate std dev

    std::cout << "  Number of Workers: " << NUM_WORKERS << "\n";                result.data.push_back(std::sqrt(variance));

    std::cout << "  Number of Tasks: " << NUM_TASKS << "\n";                

    std::cout << "  Load Balancing: Round-Robin\n";                // Min and max

    std::cout << "  Data per Task: 1000 elements\n\n";                result.data.push_back(*std::min_element(task.data.begin(), task.data.end()));

                    result.data.push_back(*std::max_element(task.data.begin(), task.data.end()));

    // Initialize workers                

    std::vector<Worker> workers;                // Simulate some computation time

    for (int i = 0; i < NUM_WORKERS; i++) {                std::this_thread::sleep_for(std::chrono::milliseconds(100 + (rand() % 200)));

        workers.push_back(Worker(i));                

    }            } catch (...) {

                    result.success = false;

    LoadBalancer balancer(NUM_WORKERS);            }

                

    // Create tasks            auto end = std::chrono::high_resolution_clock::now();

    std::cout << "Creating tasks with random data...\n";            std::chrono::duration<double> elapsed = end - start;

    srand(static_cast<unsigned>(time(0)));            result.execution_time = elapsed.count();

                

    std::vector<Task> tasks;            // Store result

    for (int i = 0; i < NUM_TASKS; i++) {            {

        Task task;                std::lock_guard<std::mutex> lock(result_mutex);

        task.id = i;                results.push_back(result);

        for (int j = 0; j < 1000; j++) {                worker_stats[worker_id].tasks_completed++;

            task.data.push_back(static_cast<double>(rand()) / RAND_MAX * 100.0);                worker_stats[worker_id].total_time += elapsed.count();

        }            }

        tasks.push_back(task);            

    }            {

                    std::lock_guard<std::mutex> lock(cout_mutex);

    std::cout << "Distributing tasks to workers...\n\n";                std::cout << "Worker " << worker_id << " completed Task " << task.id 

                              << " in " << elapsed.count() << "s\n";

    clock_t overall_start = clock();            }

    std::vector<Result> results;        } else {

                // No task available, wait a bit

    // Distribute and execute tasks            std::this_thread::sleep_for(std::chrono::milliseconds(50));

    for (int i = 0; i < NUM_TASKS; i++) {        }

        int worker_id = balancer.assignTask();    }

        std::cout << "Task " << i << " assigned to Worker " << worker_id << "... ";    

            {

        Result result = workers[worker_id].executeTask(tasks[i]);        std::lock_guard<std::mutex> lock(cout_mutex);

        results.push_back(result);        std::cout << "Worker " << worker_id << " finished!\n";

            }

        std::cout << "completed in " << result.execution_time << "s\n";}

    }

    int main() {

    clock_t overall_end = clock();    std::cout << "========================================\n";

    double total_time = double(overall_end - overall_start) / CLOCKS_PER_SEC;    std::cout << "  Distributed Computing Framework Demo  \n";

        std::cout << "  (Simplified - No MPI Required)       \n";

    // Display statistics    std::cout << "========================================\n\n";

    std::cout << "\n========================================\n";    

    std::cout << "         Execution Complete!            \n";    const int NUM_WORKERS = 4;

    std::cout << "========================================\n\n";    const int NUM_TASKS = 20;

        

    std::cout << "Overall Statistics:\n";    std::cout << "Configuration:\n";

    std::cout << "  Total Execution Time: " << total_time << " seconds\n";    std::cout << "  Workers: " << NUM_WORKERS << "\n";

    std::cout << "  Tasks Completed: " << results.size() << "/" << NUM_TASKS << "\n";    std::cout << "  Tasks: " << NUM_TASKS << "\n\n";

    std::cout << "  Success Rate: 100%\n";    

    std::cout << "  Avg Task Time: " << (total_time / NUM_TASKS) << "s\n\n";    // Initialize worker stats

        worker_stats.resize(NUM_WORKERS);

    // Worker performance    for (int i = 0; i < NUM_WORKERS; i++) {

    std::cout << "Worker Performance:\n";        worker_stats[i].id = i;

    std::cout << "  Worker | Tasks | Avg Time | Total Time\n";        worker_stats[i].active = true;

    std::cout << "  -------|-------|----------|------------\n";        worker_stats[i].tasks_completed = 0;

            worker_stats[i].tasks_failed = 0;

    for (int i = 0; i < NUM_WORKERS; i++) {        worker_stats[i].total_time = 0.0;

        double avg_time = (workers[i].tasks_completed > 0)     }

                         ? workers[i].total_time / workers[i].tasks_completed     

                         : 0.0;    // Create tasks with random data

            std::cout << "Creating " << NUM_TASKS << " tasks...\n";

        printf("  %6d | %5d | %8.4fs | %10.4fs\n",     std::random_device rd;

               i, workers[i].tasks_completed, avg_time, workers[i].total_time);    std::mt19937 gen(rd());

    }    std::uniform_real_distribution<> dis(0.0, 100.0);

        

    // Load balance analysis    for (int i = 0; i < NUM_TASKS; i++) {

    std::vector<int> task_counts;        Task task;

    for (int i = 0; i < NUM_WORKERS; i++) {        task.id = i;

        task_counts.push_back(workers[i].tasks_completed);        task.estimated_time = 0.2;

    }        

            // Generate random data

    int max_tasks = *std::max_element(task_counts.begin(), task_counts.end());        for (int j = 0; j < 1000; j++) {

    int min_tasks = *std::min_element(task_counts.begin(), task_counts.end());            task.data.push_back(dis(gen));

    double avg_tasks = std::accumulate(task_counts.begin(), task_counts.end(), 0.0) / NUM_WORKERS;        }

            

    std::cout << "\nLoad Balance Analysis:\n";        task_queue.push(task);

    std::cout << "  Max tasks per worker: " << max_tasks << "\n";    }

    std::cout << "  Min tasks per worker: " << min_tasks << "\n";    

    std::cout << "  Avg tasks per worker: " << avg_tasks << "\n";    std::cout << "Starting workers...\n\n";

    std::cout << "  Load imbalance: " << ((max_tasks - min_tasks) * 100.0 / avg_tasks) << "%\n";    

        auto overall_start = std::chrono::high_resolution_clock::now();

    // Sample computation results    

    std::cout << "\nSample Computation Result (Task 0):\n";    // Launch worker threads

    const Result& sample = results[0];    std::vector<std::thread> workers;

    std::cout << "  Mean: " << sample.mean << "\n";    int tasks_assigned = 0;

    std::cout << "  Std Dev: " << sample.std_dev << "\n";    

    std::cout << "  Min: " << sample.min_val << "\n";    for (int i = 0; i < NUM_WORKERS; i++) {

    std::cout << "  Max: " << sample.max_val << "\n";        workers.emplace_back(worker_node, i, std::ref(tasks_assigned));

        }

    std::cout << "\n========================================\n";    

    std::cout << "Key Features Demonstrated:\n";    // Wait for all workers to complete

    std::cout << "  [x] Distributed task execution\n";    for (auto& worker : workers) {

    std::cout << "  [x] Round-robin load balancing\n";        worker.join();

    std::cout << "  [x] Data partitioning\n";    }

    std::cout << "  [x] Performance monitoring\n";    

    std::cout << "  [x] Statistical computation\n";    auto overall_end = std::chrono::high_resolution_clock::now();

    std::cout << "========================================\n";    std::chrono::duration<double> total_elapsed = overall_end - overall_start;

        

    return 0;    // Display results

}    std::cout << "\n========================================\n";

    std::cout << "         Execution Complete!            \n";
    std::cout << "========================================\n\n";
    
    std::cout << "Overall Statistics:\n";
    std::cout << "  Total Execution Time: " << total_elapsed.count() << " seconds\n";
    std::cout << "  Tasks Completed: " << results.size() << "/" << NUM_TASKS << "\n";
    std::cout << "  Success Rate: " << (100.0 * results.size() / NUM_TASKS) << "%\n\n";
    
    // Worker statistics
    std::cout << "Worker Performance:\n";
    std::cout << "  Worker | Tasks | Avg Time | Total Time\n";
    std::cout << "  -------|-------|----------|------------\n";
    
    for (int i = 0; i < NUM_WORKERS; i++) {
        double avg_time = (worker_stats[i].tasks_completed > 0) 
                         ? worker_stats[i].total_time / worker_stats[i].tasks_completed 
                         : 0.0;
        
        printf("  %6d | %5d | %8.3fs | %10.3fs\n", 
               i, worker_stats[i].tasks_completed, avg_time, worker_stats[i].total_time);
    }
    
    // Load balance analysis
    std::vector<int> task_counts;
    for (const auto& ws : worker_stats) {
        task_counts.push_back(ws.tasks_completed);
    }
    
    int max_tasks = *std::max_element(task_counts.begin(), task_counts.end());
    int min_tasks = *std::min_element(task_counts.begin(), task_counts.end());
    double avg_tasks = std::accumulate(task_counts.begin(), task_counts.end(), 0.0) / NUM_WORKERS;
    
    std::cout << "\nLoad Balance Analysis:\n";
    std::cout << "  Max tasks per worker: " << max_tasks << "\n";
    std::cout << "  Min tasks per worker: " << min_tasks << "\n";
    std::cout << "  Avg tasks per worker: " << avg_tasks << "\n";
    std::cout << "  Load imbalance: " << ((max_tasks - min_tasks) * 100.0 / avg_tasks) << "%\n";
    
    // Sample results
    if (!results.empty()) {
        std::cout << "\nSample Result (Task 0):\n";
        const auto& sample = results[0];
        std::cout << "  Mean: " << sample.data[0] << "\n";
        std::cout << "  Variance: " << sample.data[1] << "\n";
        std::cout << "  Std Dev: " << sample.data[2] << "\n";
        std::cout << "  Min: " << sample.data[3] << "\n";
        std::cout << "  Max: " << sample.data[4] << "\n";
    }
    
    std::cout << "\n========================================\n";
    std::cout << "Key Features Demonstrated:\n";
    std::cout << "  ✓ Parallel computation across workers\n";
    std::cout << "  ✓ Task queue distribution\n";
    std::cout << "  ✓ Load balancing (Round-robin)\n";
    std::cout << "  ✓ Performance monitoring\n";
    std::cout << "  ✓ Thread-safe operations\n";
    std::cout << "========================================\n";
    
    return 0;
}