Cluster-based Architecture

// worker_cluster.dart
import 'dart:math';

class Job {
  final String id;
  final int processingTime;

  Job({required this.id, required this.processingTime});
}

class WorkerCluster {
  final String name;
  final int id;
  bool isBusy = false;

  WorkerCluster({required this.name, required this.id});

  Future<void> processJob(Job job) async {
    isBusy = true;
    print('Cluster $name ($id) processing job ${job.id} for ${job.processingTime}ms');
    await Future.delayed(Duration(milliseconds: job.processingTime));
    print('Cluster $name ($id) finished job ${job.id}');
    isBusy = false;
  }
}

// dispatcher.dart
class Dispatcher {
  final List<WorkerCluster> clusters;

  Dispatcher({required this.clusters});

  Future<void> dispatchJob(Job job) async {
    final availableCluster = clusters.firstWhere((cluster) => !cluster.isBusy);
    await availableCluster.processJob(job);
  }
}

// main.dart
import 'worker_cluster.dart';
import 'dispatcher.dart';

void main() async {
  final clusters = [
    WorkerCluster(name: 'Alpha', id: 1),
    WorkerCluster(name: 'Beta', id: 2),
    WorkerCluster(name: 'Gamma', id: 3),
  ];

  final dispatcher = Dispatcher(clusters: clusters);

  final jobs = [
    Job(id: 'A', processingTime: 500),
    Job(id: 'B', processingTime: 1000),
    Job(id: 'C', processingTime: 750),
    Job(id: 'D', processingTime: 250),
  ];

  await Future.wait(jobs.map((job) => dispatcher.dispatchJob(job)));

  print('All jobs completed.');
}

import akka.actor.{Actor, ActorSystem, Props}
import scala.collection.mutable
import scala.util.Random

// Messages
case class Task(id: Int, payload: String)
case class Result(taskId: Int, outcome: String)
case class RegisterWorker
case class Work(task: Task)

// Worker Actor
class Worker extends Actor {
  override def receive: Receive = {
    case Work(task) =>
      println(s"Worker received task ${task.id}: ${task.payload}")
      val result = s"Processed: ${task.payload}"
      sender() ! Result(task.id, result) // Send result back to the master
    case _ => println("Worker received unknown message")
  }
}

// Master Actor
class Master extends Actor {
  val workers = new mutable.ListBuffer[Actor.Receive]()
  var nextTaskId = 1

  override def receive: Receive = {
    case RegisterWorker =>
      val worker = sender()
      workers += worker
      println(s"Worker registered: ${worker}")
    case task: Task =>
      println(s"Master received task ${task.id}")
      if (workers.isEmpty) {
        println("No workers available")
      } else {
        val worker = workers(Random.nextInt(workers.length))
        worker ! Work(task)
      }
    case result: Result =>
      println(s"Master received result for task ${result.taskId}: ${result.outcome}")
  }
}

// Main Application
object ClusterExample {
  def main(args: Array[String]): Unit = {
    val system = ActorSystem("ClusterSystem")

    val master = system.actorOf(Props[Master], "master")

    val worker1 = system.actorOf(Props[Worker], "worker1")
    val worker2 = system.actorOf(Props[Worker], "worker2")
    val worker3 = system.actorOf(Props[Worker], "worker3")

    worker1 ! RegisterWorker
    worker2 ! RegisterWorker
    worker3 ! RegisterWorker

    master ! Task(1, "Data to process 1")
    master ! Task(2, "Data to process 2")
    master ! Task(3, "Data to process 3")

    Thread.sleep(2000)
    system.terminate()
  }
}

<?php

/**
 * Interface for a worker task.
 */
interface Task
{
    public function execute(): string;
}

/**
 * A concrete task example.
 */
class ExampleTask implements Task
{
    private string $data;

    public function __construct(string $data)
    {
        $this->data = $data;
    }

    public function execute(): string
    {
        return "Processed: " . $this->data . " on server " . uniqid();
    }
}

/**
 *  Worker Server - processes tasks.
 */
class WorkerServer
{
    public function processTask(Task $task): string
    {
        return $task->execute();
    }
}

// Simulate a cluster
$workers = [new WorkerServer(), new WorkerServer(), new WorkerServer()];
$tasks = [new ExampleTask("Task 1"), new ExampleTask("Task 2"), new ExampleTask("Task 3")];

// Distribute tasks round-robin
$workerIndex = 0;
foreach ($tasks as $task) {
    $result = $workers[$workerIndex]->processTask($task);
    echo $result . PHP_EOL;
    $workerIndex = ($workerIndex + 1) % count($workers);
}

?>

# request_distributor.rb
require 'socket'

class RequestDistributor
  def initialize(clusters)
    @clusters = clusters
    @cluster_index = 0
  end

  def distribute_request(request)
    cluster = @clusters[@cluster_index]
    cluster.process_request(request)
    @cluster_index = (@cluster_index + 1) % @clusters.length
  end
end

# worker_cluster.rb
class WorkerCluster
  def initialize(id)
    @id = id
  end

  def process_request(request)
    puts "Cluster #{@id} processing request: #{request}"
    # Simulate processing time
    sleep(rand(0.1..0.5))
    puts "Cluster #{@id} finished processing request: #{request}"
  end
end

# main.rb
require_relative 'request_distributor'
require_relative 'worker_cluster'

cluster1 = WorkerCluster.new(1)
cluster2 = WorkerCluster.new(2)
cluster3 = WorkerCluster.new(3)

distributor = RequestDistributor.new([cluster1, cluster2, cluster3])

# Simulate incoming requests
10.times do |i|
  request = "Request #{i + 1}"
  distributor.distribute_request(request)
end

// Define a common protocol for Media Clusters
protocol MediaCluster {
    func load(filePath: String) -> Data?
    func process(data: Data) -> String
}

// Image Cluster
class ImageCluster: MediaCluster {
    func load(filePath: String) -> Data? {
        print("Loading image from: \(filePath)")
        return Data(contentsOfFile: filePath) // Simplified for example
    }

    func process(data: Data) -> String {
        print("Processing image data")
        return "Image processed successfully."
    }
}

// Video Cluster
class VideoCluster: MediaCluster {
    func load(filePath: String) -> Data? {
        print("Loading video from: \(filePath)")
        return Data(contentsOfFile: filePath) // Simplified for example
    }

    func process(data: Data) -> String {
        print("Processing video data")
        return "Video processed successfully."
    }
}

// Audio Cluster
class AudioCluster: MediaCluster {
    func load(filePath: String) -> Data? {
        print("Loading audio from: \(filePath)")
        return Data(contentsOfFile: filePath) // Simplified for example
    }

    func process(data: Data) -> String {
        print("Processing audio data")
        return "Audio processed successfully."
    }
}

// Media Manager - the Coordinator
class MediaManager {
    private var imageCluster: ImageCluster = ImageCluster()
    private var videoCluster: VideoCluster = VideoCluster()
    private var audioCluster: AudioCluster = AudioCluster()

    enum MediaType {
        case image, video, audio, unknown
    }

    func getMediaType(filePath: String) -> MediaType {
        let extensionValue = filePath.split(separator: ".").last?.lowercased()
        switch extensionValue {
        case "jpg", "jpeg", "png":
            return .image
        case "mp4", "mov":
            return .video
        case "mp3", "wav":
            return .audio
        default:
            return .unknown
        }
    }

    func processMedia(filePath: String) -> String {
        let mediaType = getMediaType(filePath: filePath)
        guard let data = loadMedia(filePath: filePath, type: mediaType) else {
            return "Failed to load media."
        }

        switch mediaType {
        case .image:
            return imageCluster.process(data: data)
        case .video:
            return videoCluster.process(data: data)
        case .audio:
            return audioCluster.process(data: data)
        case .unknown:
            return "Unsupported media type."
        }
    }

    private func loadMedia(filePath: String, type: MediaType) -> Data? {
        switch type {
        case .image: return imageCluster.load(filePath: filePath)
        case .video: return videoCluster.load(filePath: filePath)
        case .audio: return audioCluster.load(filePath: filePath)
        case .unknown: return nil
        }
    }
}

// Example Usage
let mediaManager = MediaManager()
print(mediaManager.processMedia(filePath: "video.mp4"))
print(mediaManager.processMedia(filePath: "image.jpg"))
print(mediaManager.processMedia(filePath: "audio.mp3"))
print(mediaManager.processMedia(filePath: "document.pdf"))

// Worker.kt
data class Task(val id: Int, val data: String)

class Worker(val id: Int) {
    fun executeTask(task: Task): String {
        println("Worker $id executing task $task")
        return "Result from ${task.id}"
    }
}

// WorkerCluster.kt
class WorkerCluster(val clusterId: Int, private val workers: List<Worker>) {
    fun executeTasks(tasks: List<Task>): List<String> {
        return tasks.parallelStream().map { task ->
            workers.random().executeTask(task) // Distribute tasks randomly
        }.toList()
    }
}

// ClusterManager.kt
class ClusterManager(private val clusters: List<WorkerCluster>) {
    fun executeTasksAcrossClusters(tasks: List<Task>): List<String> {
        val results = mutableListOf<String>()
        for (cluster in clusters) {
            results.addAll(cluster.executeTasks(tasks.subList(0, tasks.size / clusters.size)))
        }
        return results
    }
}

// Main.kt
fun main() {
    val worker1 = Worker(1)
    val worker2 = Worker(2)
    val worker3 = Worker(3)

    val cluster1 = WorkerCluster(1, listOf(worker1, worker2))
    val cluster2 = WorkerCluster(2, listOf(worker3))

    val clusterManager = ClusterManager(listOf(cluster1, cluster2))

    val tasks = List(10) { Task(it, "Data $it") }
    val results = clusterManager.executeTasksAcrossClusters(tasks)

    println("Results: $results")
}

use std::thread;
use std::sync::{Arc, Mutex};

struct Worker {
    data: Vec<i32>,
}

impl Worker {
    fn new(data: Vec<i32>) -> Self {
        Worker { data }
    }

    fn process(&self) -> i32 {
        self.data.iter().map(|&x| x * x).sum()
    }
}

struct Cluster {
    workers: Vec<Worker>,
    result: Arc<Mutex<i32>>,
}

impl Cluster {
    fn new(data: Vec<i32>, num_workers: usize) -> Self {
        let chunk_size = (data.len() + num_workers - 1) / num_workers; // Ensure even distribution
        let workers: Vec<Worker> = data
            .chunks(chunk_size)
            .map(|chunk| Worker::new(chunk.to_vec()))
            .collect();

        Cluster {
            workers,
            result: Arc::new(Mutex::new(0)),
        }
    }

    fn execute(&self) {
        let result = Arc::clone(&self.result);
        let mut handles = vec![];

        for worker in &self.workers {
            let worker_data = worker.data.clone();
            let result_clone = result.clone();
            let handle = thread::spawn(move || {
                let partial_sum = worker.process();
                let mut guard = result_clone.lock().unwrap();
                *guard += partial_sum;
            });
            handles.push(handle);
        }

        for handle in handles {
            handle.join().unwrap();
        }
    }

    fn get_result(&self) -> i32 {
        *self.result.lock().unwrap()
    }
}


fn main() {
    let data: Vec<i32> = (1..=100).collect();
    let num_workers = 4;

    let cluster = Cluster::new(data, num_workers);
    cluster.execute();

    let total_sum_of_squares = cluster.get_result();
    println!("Total sum of squares: {}", total_sum_of_squares);
}

package main

import (
	"fmt"
	"sync"
	"time"
)

type Worker struct {
	ID     int
	Master *Master
}

type Task struct {
	ID     int
	Input  int
	Result int
}

type Master struct {
	Workers   []*Worker
	TaskChan  chan Task
	Results   chan Task
	Register  chan *Worker
	WaitGroup sync.WaitGroup
}

func NewMaster() *Master {
	return &Master{
		TaskChan:  make(chan Task),
		Results:   make(chan Task),
		Register:  make(chan *Worker),
		WaitGroup: sync.WaitGroup{},
	}
}

func (m *Master) Run() {
	for worker := range m.Register {
		m.Workers = append(m.Workers, worker)
	}

	go m.dispatchTasks()

	m.WaitGroup.Wait() // Wait for all workers to finish
	close(m.Results)
}

func (m *Master) dispatchTasks() {
	workerIndex := 0
	for task := range m.TaskChan {
		worker := m.Workers[workerIndex]
		worker.processTask(task)
		workerIndex = (workerIndex + 1) % len(m.Workers)
	}
	m.WaitGroup.Done()
}


func (w *Worker) processTask(task Task) {
	time.Sleep(time.Millisecond * 50) // Simulate work
	task.Result = task.Input * 2
	w.Master.Results <- task
}

func main() {
	master := NewMaster()
	numWorkers := 3

	for i := 0; i < numWorkers; i++ {
		worker := &Worker{ID: i, Master: master}
		master.WaitGroup.Add(1)
		go func(w *Worker) {
			master.Register <- w
		}(worker)
	}

	go master.Run()

	// Send tasks
	for i := 0; i < 10; i++ {
		master.TaskChan <- Task{ID: i, Input: i}
	}
	close(master.TaskChan)


	// Collect results
	for result := range master.Results {
		fmt.Printf("Task %d: Input = %d, Result = %d\n", result.ID, result.Input, result.Result)
	}
}

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>

#define NUM_WORKERS 4
#define QUEUE_SIZE 10

typedef struct {
  int id;
  char* data;
} WorkItem;

WorkItem work_queue[QUEUE_SIZE];
int queue_head = 0;
int queue_tail = 0;
pthread_mutex_t queue_mutex;

void* worker_thread(void* arg) {
  int worker_id = *(int*)arg;
  WorkItem* item;

  while (1) {
    pthread_mutex_lock(&queue_mutex);
    if (queue_head != queue_tail) {
      item = &work_queue[queue_head++];
      pthread_mutex_unlock(&queue_mutex);

      printf("Worker %d processing item %d: %s\n", worker_id, item->id, item->data);
      free(item->data); // Free allocated data
      item->data = NULL;
    } else {
      pthread_mutex_unlock(&queue_mutex);
      // Queue is empty.  Sleep to avoid busy-waiting.  Better would be a condition variable.
      usleep(10000); // 10ms
    }
  }

  return NULL;
}


int main() {
  pthread_t workers[NUM_WORKERS];
  int worker_ids[NUM_WORKERS];

  pthread_mutex_init(&queue_mutex, NULL);

  // Create worker threads
  for (int i = 0; i < NUM_WORKERS; i++) {
    worker_ids[i] = i;
    if (pthread_create(&workers[i], NULL, worker_thread, &worker_ids[i]) != 0) {
      perror("Thread creation failed");
      return 1;
    }
  }

  // Add work to the queue
  for (int i = 0; i < 20; i++) {
    char* data = (char*)malloc(50);
    snprintf(data, 50, "Task %d data", i);

    pthread_mutex_lock(&queue_mutex);
    if ((queue_tail + 1) % QUEUE_SIZE == queue_head) {
      pthread_mutex_unlock(&queue_mutex);
      printf("Queue is full, dropping task %d\n", i);
      free(data);
      continue;
    }
    work_queue[queue_tail].id = i;
    work_queue[queue_tail].data = data;
    queue_tail = (queue_tail + 1) % QUEUE_SIZE;
    pthread_mutex_unlock(&queue_mutex);
    usleep(50000); // Add a small delay to simulate work creation rate
  }

  // Allow workers to finish (in a real system, you'd have a proper shutdown mechanism)
  sleep(5);


  pthread_mutex_destroy(&queue_mutex);
  return 0;
}

#include <iostream>
#include <vector>

// Abstract Interface for Order Cluster
class IOrderService {
public:
  virtual void createOrder(int productId, int quantity) = 0;
  virtual int getOrderId() = 0; //Returns the most recent order id
};

// Abstract Interface for Payment Cluster
class IPaymentService {
public:
  virtual bool processPayment(int orderId, double amount) = 0;
};

// Order Cluster Implementation
class OrderService : public IOrderService {
private:
  static int nextOrderId;
  std::vector<int> orderIds;
public:
  void createOrder(int productId, int quantity) override {
    std::cout << "Order: Creating order for product " << productId << ", quantity " << quantity << std::endl;
    orderIds.push_back(nextOrderId++);
  }

  int getOrderId() override {
    return orderIds.back();
  }
};

int OrderService::nextOrderId = 1;

// Payment Cluster Implementation
class PaymentService : public IPaymentService {
public:
  bool processPayment(int orderId, double amount) override {
    std::cout << "Payment: Processing payment of " << amount << " for order " << orderId << std::endl;
    return true; // Simulate successful payment
  }
};

// Client code (glue - minimized for demonstration)
int main() {
  IOrderService* orderSvc = new OrderService();
  IPaymentService* paymentSvc = new PaymentService();

  orderSvc->createOrder(123, 2);
  int orderId = orderSvc->getOrderId();
  
  if (paymentSvc->processPayment(orderId, 29.99)) {
    std::cout << "Transaction complete!" << std::endl;
  }
  else {
    std::cout << "Transaction failed." << std::endl;
  }

  delete orderSvc;
  delete paymentSvc;
  return 0;
}

// Interfaces defining cluster communication
public interface IOrderCluster
{
    bool PlaceOrder(Order order);
}

public interface IPaymentCluster
{
    bool ProcessPayment(Payment payment);
}

public interface IInventoryCluster
{
    bool ReserveInventory(InventoryReservation reservation);
}

// Cluster Implementations
public class OrderCluster : IOrderCluster
{
    public bool PlaceOrder(Order order)
    {
        // Order processing logic
        Console.WriteLine($"Order placed: {order.OrderId}");
        return true;
    }
}

public class PaymentCluster : IPaymentCluster
{
    public bool ProcessPayment(Payment payment)
    {
        // Payment processing logic
        Console.WriteLine($"Payment processed for amount: {payment.Amount}");
        return true;
    }
}

public class InventoryCluster : IInventoryCluster
{
    public bool ReserveInventory(InventoryReservation reservation)
    {
        // Inventory reservation logic
        Console.WriteLine($"Inventory reserved: {reservation.Quantity} of item {reservation.ItemId}");
        return true;
    }
}

// Data Models (Simplified)
public class Order { public int OrderId { get; set; } }
public class Payment { public decimal Amount { get; set; } }
public class InventoryReservation { public int ItemId { get; set; } public int Quantity { get; set; } }

// Orchestrator
public class Shop
{
    private readonly IOrderCluster _orderCluster;
    private readonly IPaymentCluster _paymentCluster;
    private readonly IInventoryCluster _inventoryCluster;

    public Shop(IOrderCluster orderCluster, IPaymentCluster paymentCluster, IInventoryCluster inventoryCluster)
    {
        _orderCluster = orderCluster;
        _paymentCluster = paymentCluster;
        _inventoryCluster = inventoryCluster;
    }

    public bool Purchase(Order order, Payment payment, InventoryReservation reservation)
    {
        if (_inventoryCluster.ReserveInventory(reservation) && _paymentCluster.ProcessPayment(payment))
        {
            return _orderCluster.PlaceOrder(order);
        }
        return false;
    }
}

// Example Usage
public class Program
{
    public static void Main(string[] args)
    {
        var orderCluster = new OrderCluster();
        var paymentCluster = new PaymentCluster();
        var inventoryCluster = new InventoryCluster();

        var shop = new Shop(orderCluster, paymentCluster, inventoryCluster);

        var order = new Order { OrderId = 123 };
        var payment = new Payment { Amount = 100.00m };
        var reservation = new InventoryReservation { ItemId = 456, Quantity = 2 };

        bool purchaseSuccessful = shop.Purchase(order, payment, reservation);

        Console.WriteLine($"Purchase successful: {purchaseSuccessful}");
    }
}

// cluster.ts
import * as cluster from 'cluster';
import * as http from 'http';
import { cpus } from 'os';

const numCPUs = cpus().length;
const port = 3000;

if (cluster.isMaster) {
  console.log(`Master process running on PID ${process.pid}`);

  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log(`Worker process exited with code ${code} and signal ${signal}`);
    cluster.fork(); // Restart the worker
  });
} else {
  const workerId = cluster.worker.id;
  console.log(`Worker process running on PID ${process.pid}, ID ${workerId}`);

  const server = http.createServer((req, res) => {
    res.writeHead(200, { 'Content-Type': 'text/plain' });
    res.end(`Hello from worker ${workerId}!`);
  });

  server.listen(port, () => {
    console.log(`Worker ${workerId} listening on port ${port}`);
  });

  process.on('uncaughtException', (err) => {
    console.error(`Worker ${workerId} uncaught exception: ${err}`);
    process.exit(1);
  });
}

// master.js
const { Worker, isMainThread, parentPort } = require('worker_threads');

if (isMainThread) {
  const numTasks = 5;
  const workers = [];
  const results = [];

  for (let i = 0; i < numTasks; i++) {
    const worker = new Worker('./worker.js', { workerId: i });
    workers.push(worker);

    worker.on('message', (result) => {
      results[worker.workerId] = result;
      if (results.every(x => x !== undefined)) {
        console.log('All results received:', results);
      }
    });
    worker.on('error', (err) => {
      console.error(`Worker ${worker.workerId} error:`, err);
    });
    worker.on('exit', (code) => {
      if (code !== 0) {
        console.error(`Worker ${worker.workerId} stopped with exit code ${code}`);
      }
    });

    worker.postMessage({ task: 'fibonacci', data: i + 1 }); // Send task to worker
  }
}

// worker.js
const { parentPort } = require('worker_threads');

function fibonacci(n) {
  if (n <= 1) return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

parentPort.on('message', (message) => {
  const { task, data } = message;
  if (task === 'fibonacci') {
    const result = fibonacci(data);
    parentPort.postMessage(result);
  }
});

import threading

class Worker(threading.Thread):
    """
    Represents a worker cluster.
    """
    def __init__(self, worker_id, task_queue, result_queue):
        threading.Thread.__init__(self)
        self.worker_id = worker_id
        self.task_queue = task_queue
        self.result_queue = result_queue

    def run(self):
        """
        Processes tasks from the task queue and puts results into the result queue.
        """
        while True:
            task = self.task_queue.get()
            if task is None:
                break  # Signal to terminate
            result = self.process_task(task)
            self.result_queue.put(result)
            self.task_queue.task_done()

    def process_task(self, task):
        """
        Simulates task processing.  Replace with actual logic.
        """
        return f"Worker {self.worker_id} processed: {task}"


class Master:
    """
    Distributes tasks to workers and collects results.
    """
    def __init__(self, num_workers, tasks):
        self.num_workers = num_workers
        self.tasks = tasks
        self.task_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.workers = []

    def start(self):
        """
        Starts the worker threads and distributes tasks.
        """
        for i in range(self.num_workers):
            worker = Worker(i, self.task_queue, self.result_queue)
            self.workers.append(worker)
            worker.start()

        for task in self.tasks:
            self.task_queue.put(task)

        # Signal workers to terminate
        for _ in range(self.num_workers):
            self.task_queue.put(None)

        self.task_queue.join()  # Wait for all tasks to be processed

    def collect_results(self):
        """
        Collects results from the result queue.
        """
        results = []
        while not self.result_queue.empty():
            results.append(self.result_queue.get())
        return results

import queue

if __name__ == "__main__":
    tasks = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    num_workers = 3

    master = Master(num_workers, tasks)
    master.start()
    results = master.collect_results()

    print("Results:")
    for result in results:
        print(result)

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

class LogProcessor implements Runnable {
    private final List<String> logs;
    private final int processorId;
    private int processedCount = 0;

    public LogProcessor(List<String> logs, int processorId) {
        this.logs = logs;
        this.processorId = processorId;
    }

    @Override
    public void run() {
        for (String log : logs) {
            processLog(log);
        }
    }

    private void processLog(String log) {
        // Simulate log processing
        try {
            Thread.sleep(new Random().nextInt(50)); // Simulate processing time
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        processedCount++;
        System.out.println("Processor " + processorId + " processed log: " + log);
    }

    public int getProcessedCount() {
        return processedCount;
    }
}

class ClusterManager {
    private final List<Thread> processors = new ArrayList<>();
    private final List<LogProcessor> processorInstances = new ArrayList<>();

    public ClusterManager(List<String> logs, int numProcessors) {
        int logsPerProcessor = logs.size() / numProcessors;
        for (int i = 0; i < numProcessors; i++) {
            List<String> processorLogs = logs.subList(i * logsPerProcessor,
                    (i == numProcessors - 1) ? logs.size() : (i + 1) * logsPerProcessor);

            LogProcessor processor = new LogProcessor(processorLogs, i + 1);
            processorInstances.add(processor);
            Thread thread = new Thread(processor);
            processors.add(thread);
            thread.start();
        }
    }

    public void waitForCompletion() throws InterruptedException {
        for (Thread thread : processors) {
            thread.join();
        }
    }
    
    public int getTotalProcessedLogs(){
        int total = 0;
        for(LogProcessor processor : processorInstances){
            total += processor.getProcessedCount();
        }
        return total;
    }
}

public class ClusterExample {
    public static void main(String[] args) throws InterruptedException {
        List<String> logs = new ArrayList<>();
        for (int i = 1; i <= 100; i++) {
            logs.add("Log entry " + i);
        }

        int numProcessors = 4;
        ClusterManager clusterManager = new ClusterManager(logs, numProcessors);

        clusterManager.waitForCompletion();

        System.out.println("Total logs processed: " + clusterManager.getTotalProcessedLogs());
    }
}