Jassif Team Institution

const {
  Worker,
  isMainThread,
  parentPort,
  workerData
} = require('worker_threads');

// main.js
const { Worker } = require('worker_threads');

// Function to create a new worker
function runWorker(workerData) {
  return new Promise((resolve, reject) => {
    // Create a new worker
    const worker = new Worker('./worker.js', { workerData });
    
    // Listen for messages from the worker
    worker.on('message', resolve);
    
    // Listen for errors
    worker.on('error', reject);
    
    // Listen for worker exit
    worker.on('exit', (code) => {
      if (code !== 0) {
        reject(new Error(`Worker stopped with exit code ${code}`));
      }
    });
  });
}

// Run the worker
async function run() {
  try {
    // Send data to the worker and get the result
    const result = await runWorker('Hello from main thread!');
    console.log('Worker result:', result);
  } catch (err) {
    console.error('Worker error:', err);
  }
}

run().catch(err => console.error(err));

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

// Receive message from the main thread
console.log('Worker received:', workerData);

// Simulate CPU-intensive task
function performCPUIntensiveTask() {
  // Simple example: Sum up to a large number
  let result = 0;
  for (let i = 0; i < 1_000_000; i++) {
    result += i;
  }
  return result;
}

// Perform the task
const result = performCPUIntensiveTask();

// Send the result back to the main thread
parentPort.postMessage({
  receivedData: workerData,
  calculatedSum: result
});

// main.js
const { Worker } = require('worker_threads');

// Create a worker
const worker = new Worker('./message_worker.js');

// Send messages to the worker
worker.postMessage('Hello worker!');
worker.postMessage({ type: 'task', data: [1, 2, 3, 4, 5] });

// Receive messages from the worker
worker.on('message', (message) => {
  console.log('Main thread received:', message);
});

// Handle worker completion
worker.on('exit', (code) => {
  console.log(`Worker exited with code ${code}`);
});

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

// Receive messages from the main thread
parentPort.on('message', (message) => {
  console.log('Worker received:', message);
  
  // Process different message types
  if (typeof message === 'object' && message.type === 'task') {
    const result = processTask(message.data);
    parentPort.postMessage({ type: 'result', data: result });
  } else {
    // Echo the message back
    parentPort.postMessage(`Worker echoing: ${message}`);
  }
});

// Example task processor
function processTask(data) {
  if (Array.isArray(data)) {
    return data.map(x => x * 2);
  }
  return null;
}

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

// Recursive Fibonacci function (deliberately inefficient to simulate CPU load)
function fibonacci(n) {
  if (n <= 1) return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

if (isMainThread) {
  // This code runs in the main thread
  
  // Function to run a worker
  function runFibonacciWorker(n) {
    return new Promise((resolve, reject) => {
      const worker = new Worker(__filename, { workerData: n });
      worker.on('message', resolve);
      worker.on('error', reject);
      worker.on('exit', (code) => {
        if (code !== 0) {
          reject(new Error(`Worker stopped with exit code ${code}`));
        }
      });
    });
  }
  
  // Measure execution time with and without workers
  async function run() {
    const numbers = [40, 41, 42, 43];
    
    // Using a single thread (blocking)
    console.time('Single thread');
    for (const n of numbers) {
      console.log(`Fibonacci(${n}) = ${fibonacci(n)}`);
    }
    console.timeEnd('Single thread');
    
    // Using worker threads (parallel)
    console.time('Worker threads');
    const results = await Promise.all(
      numbers.map(n => runFibonacciWorker(n))
    );
    for (let i = 0; i < numbers.length; i++) {
      console.log(`Fibonacci(${numbers[i]}) = ${results[i]}`);
    }
    console.timeEnd('Worker threads');
  }
  
  run().catch(err => console.error(err));
} else {
  // This code runs in worker threads
  
  // Calculate Fibonacci number
  const result = fibonacci(workerData);
  
  // Send the result back to the main thread
  parentPort.postMessage(result);
}

// transfer_main.js
const { Worker } = require('worker_threads');

// Create a large buffer
const buffer = new ArrayBuffer(100 * 1024 * 1024); // 100MB
const view = new Uint8Array(buffer);

// Fill with data
for (let i = 0; i < view.length; i++) {
  view[i] = i % 256;
}

console.log('Buffer created in main thread');
console.log('Buffer byteLength before transfer:', buffer.byteLength);

// Create a worker and transfer the buffer
const worker = new Worker('./transfer_worker.js');
worker.on('message', (message) => {
  console.log('Message from worker:', message);
  
  // After transfer, the buffer is no longer usable in main thread
  console.log('Buffer byteLength after transfer:', buffer.byteLength);
});

// Transfer ownership of the buffer to the worker
worker.postMessage({ buffer }, [buffer]);

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

parentPort.on('message', ({ buffer }) => {
  const view = new Uint8Array(buffer);
  
  // Calculate sum to verify data
  let sum = 0;
  for (let i = 0; i < view.length; i++) {
    sum += view[i];
  }
  
  console.log('Buffer received in worker');
  console.log('Buffer byteLength in worker:', buffer.byteLength);
  console.log('Sum of all values:', sum);
  
  // Send confirmation back
  parentPort.postMessage('Buffer processed successfully');
});

// shared_main.js
const { Worker } = require('worker_threads');

// Create a shared buffer
const sharedBuffer = new SharedArrayBuffer(4 * 10); // 10 Int32 values
const sharedArray = new Int32Array(sharedBuffer);

// Initialize the shared array
for (let i = 0; i < sharedArray.length; i++) {
  sharedArray[i] = i;
}

console.log('Initial shared array in main thread:', [...sharedArray]);

// Create a worker that will update the shared memory
const worker = new Worker('./shared_worker.js', {
  workerData: { sharedBuffer }
});

worker.on('message', (message) => {
  console.log('Message from worker:', message);
  console.log('Updated shared array in main thread:', [...sharedArray]);
  
  // The changes made in the worker are visible here
  // because we're accessing the same memory
});

// shared_worker.js
const { parentPort, workerData } = require('worker_threads');
const { sharedBuffer } = workerData;

// Create a new view on the shared buffer
const sharedArray = new Int32Array(sharedBuffer);

console.log('Initial shared array in worker:', [...sharedArray]);

// Modify the shared memory
for (let i = 0; i < sharedArray.length; i++) {
  // Double each value
  sharedArray[i] = sharedArray[i] * 2;
}

console.log('Updated shared array in worker:', [...sharedArray]);

// Notify the main thread
parentPort.postMessage('Shared memory updated');

// atomics_main.js
const { Worker } = require('worker_threads');

// Create a shared buffer with control flags and data
const sharedBuffer = new SharedArrayBuffer(4 * 10);
const sharedArray = new Int32Array(sharedBuffer);

// Initialize values
sharedArray[0] = 0; // Control flag: 0 = main thread's turn, 1 = worker's turn
sharedArray[1] = 0; // Data value to increment

// Create workers
const workerCount = 4;
const workerIterations = 10;
const workers = [];

console.log(`Creating ${workerCount} workers with ${workerIterations} iterations each`);

for (let i = 0; i < workerCount; i++) {
  const worker = new Worker('./atomics_worker.js', {
    workerData: { sharedBuffer, id: i, iterations: workerIterations }
  });
  
  workers.push(worker);
  
  worker.on('exit', () => {
    console.log(`Worker ${i} exited`);
    
    // If all workers have exited, show final value
    if (workers.every(w => w.threadId === -1)) {
      console.log(`Final value: ${sharedArray[1]}`);
      console.log(`Expected value: ${workerCount * workerIterations}`);
    }
  });
}

// Signal to the first worker to start
Atomics.store(sharedArray, 0, 1);
Atomics.notify(sharedArray, 0);

// atomics_worker.js
const { parentPort, workerData } = require('worker_threads');
const { sharedBuffer, id, iterations } = workerData;

// Create a typed array from the shared memory
const sharedArray = new Int32Array(sharedBuffer);

for (let i = 0; i < iterations; i++) {
  // Wait for this worker's turn
  while (Atomics.load(sharedArray, 0) !== id + 1) {
    // Wait for notification
    Atomics.wait(sharedArray, 0, Atomics.load(sharedArray, 0));
  }
  
  // Increment the shared counter
  const currentValue = Atomics.add(sharedArray, 1, 1);
  console.log(`Worker ${id} incremented counter to ${currentValue + 1}`);
  
  // Signal to the next worker
  const nextWorkerId = (id + 1) % (iterations === 0 ? 1 : iterations);
  Atomics.store(sharedArray, 0, nextWorkerId + 1);
  Atomics.notify(sharedArray, 0);
}

// Exit the worker
parentPort.close();

// worker_pool.js
const { Worker } = require('worker_threads');
const os = require('os');
const path = require('path');

class WorkerPool {
  constructor(workerScript, numWorkers = os.cpus().length) {
    this.workerScript = workerScript;
    this.numWorkers = numWorkers;
    this.workers = [];
    this.freeWorkers = [];
    this.tasks = [];
    
    // Initialize workers
    this._initialize();
  }
  
  _initialize() {
    // Create all workers
    for (let i = 0; i < this.numWorkers; i++) {
      this._createWorker();
    }
  }
  
  _createWorker() {
    const worker = new Worker(this.workerScript);
    
    worker.on('message', (result) => {
      // Get the current task
      const { resolve } = this.tasks.shift();
      
      // Resolve the task with the result
      resolve(result);
      
      // Add this worker back to the free workers pool
      this.freeWorkers.push(worker);
      
      // Process the next task if any
      this._processQueue();
    });
    
    worker.on('error', (err) => {
      // If a worker errors, terminate it and create a new one
      console.error(`Worker error: ${err}`);
      this._removeWorker(worker);
      this._createWorker();
      
      // Process the next task
      if (this.tasks.length > 0) {
        const { reject } = this.tasks.shift();
        reject(err);
        this._processQueue();
      }
    });
    
    worker.on('exit', (code) => {
      if (code !== 0) {
        console.error(`Worker exited with code ${code}`);
        this._removeWorker(worker);
        this._createWorker();
      }
    });
    
    // Add to free workers
    this.workers.push(worker);
    this.freeWorkers.push(worker);
  }
  
  _removeWorker(worker) {
    // Remove from the workers arrays
    this.workers = this.workers.filter(w => w !== worker);
    this.freeWorkers = this.freeWorkers.filter(w => w !== worker);
  }
  
  _processQueue() {
    // If there are tasks and free workers, process the next task
    if (this.tasks.length > 0 && this.freeWorkers.length > 0) {
      const { taskData } = this.tasks[0];
      const worker = this.freeWorkers.pop();
      worker.postMessage(taskData);
    }
  }
  
  // Run a task on a worker
  runTask(taskData) {
    return new Promise((resolve, reject) => {
      const task = { taskData, resolve, reject };
      this.tasks.push(task);
      this._processQueue();
    });
  }
  
  // Close all workers when done
  close() {
    for (const worker of this.workers) {
      worker.terminate();
    }
  }
}

module.exports = WorkerPool;

// pool_usage.js
const WorkerPool = require('./worker_pool');
const path = require('path');

// Create a worker pool with the worker script
const pool = new WorkerPool(path.resolve(__dirname, 'pool_worker.js'));

// Function to run tasks on the pool
async function runTasks() {
  const tasks = [
    { type: 'fibonacci', data: 40 },
    { type: 'factorial', data: 15 },
    { type: 'prime', data: 10000000 },
    { type: 'fibonacci', data: 41 },
    { type: 'factorial', data: 16 },
    { type: 'prime', data: 20000000 },
    { type: 'fibonacci', data: 42 },
    { type: 'factorial', data: 17 },
  ];
  
  console.time('All tasks');
  
  try {
    // Run all tasks in parallel
    const results = await Promise.all(
      tasks.map(task => {
        console.time(`Task: ${task.type}(${task.data})`);
        return pool.runTask(task)
          .then(result => {
            console.timeEnd(`Task: ${task.type}(${task.data})`);
            return result;
          });
      })
    );
    
    // Log results
    for (let i = 0; i < tasks.length; i++) {
      console.log(`${tasks[i].type}(${tasks[i].data}) = ${results[i].result}`);
    }
  } catch (err) {
    console.error('Error running tasks:', err);
  } finally {
    console.timeEnd('All tasks');
    pool.close();
  }
}

runTasks().catch(console.error);

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

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

// Factorial function
function factorial(n) {
  if (n <= 1) return 1;
  return n * factorial(n - 1);
}

// Prime count function
function countPrimes(max) {
  const sieve = new Uint8Array(max);
  let count = 0;
  
  for (let i = 2; i < max; i++) {
    if (!sieve[i]) {
      count++;
      for (let j = i * 2; j < max; j += i) {
        sieve[j] = 1;
      }
    }
  }
  
  return count;
}

// Handle messages from the main thread
parentPort.on('message', (task) => {
  const { type, data } = task;
  let result;
  
  // Perform different calculations based on task type
  switch (type) {
    case 'fibonacci':
      result = fibonacci(data);
      break;
    case 'factorial':
      result = factorial(data);
      break;
    case 'prime':
      result = countPrimes(data);
      break;
    default:
      throw new Error(`Unknown task type: ${type}`);
  }
  
  // Send the result back
  parentPort.postMessage({ result });
});

// image_main.js
const { Worker } = require('worker_threads');
const path = require('path');
const fs = require('fs');

// Function to process an image in a worker
function processImageInWorker(imagePath, options) {
  return new Promise((resolve, reject) => {
    const worker = new Worker('./image_worker.js', {
      workerData: {
        imagePath,
        options
      }
    });
    
    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) {
        reject(new Error(`Worker stopped with exit code ${code}`));
      }
    });
  });
}

// Main function to process multiple images in parallel
async function processImages() {
  const images = [
    { path: 'image1.jpg', options: { grayscale: true } },
    { path: 'image2.jpg', options: { blur: 5 } },
    { path: 'image3.jpg', options: { sharpen: 10 } },
    { path: 'image4.jpg', options: { resize: { width: 800, height: 600 } } }
  ];
  
  console.time('Image processing');
  
  try {
    // Process all images in parallel
    const results = await Promise.all(
      images.map(img => processImageInWorker(img.path, img.options))
    );
    
    console.log('All images processed successfully');
    console.log('Results:', results);
  } catch (err) {
    console.error('Error processing images:', err);
  }
  
  console.timeEnd('Image processing');
}

// Note: This is a conceptual example.
// In a real application, you would use an image processing library like sharp or jimp
// and provide actual image files.
// processImages().catch(console.error);
console.log('Image processing example (not actually running)');

// image_worker.js
const { parentPort, workerData } = require('worker_threads');
const { imagePath, options } = workerData;

// In a real application, you would import an image processing library here
// const sharp = require('sharp');

// Simulate image processing
function processImage(imagePath, options) {
  console.log(`Processing image: ${imagePath} with options:`, options);
  
  // Simulate processing time based on options
  let processingTime = 500; // Base time in ms
  
  if (options.grayscale) processingTime += 200;
  if (options.blur) processingTime += options.blur * 50;
  if (options.sharpen) processingTime += options.sharpen * 30;
  if (options.resize) processingTime += 300;
  
  // Simulate the actual processing
  return new Promise(resolve => {
    setTimeout(() => {
      // Return simulated result
      resolve({
        imagePath,
        outputPath: `processed_${imagePath}`,
        processing: options,
        dimensions: options.resize || { width: 1024, height: 768 },
        size: Math.floor(Math.random() * 1000000) + 500000 // Random file size
      });
    }, processingTime);
  });
}

// Process the image and send the result back
processImage(imagePath, options)
  .then(result => {
    parentPort.postMessage(result);
  })
  .catch(err => {
    throw err;
  });