process_pool.js

/**
 * Process Pool Manager
 * Maintains a pool of reusable Python processes to avoid memory leaks
 * 
 * Problem: Spawning new Python process per request = OOM at 50+ concurrent users
 * Solution: Keep N persistent processes, queue requests to them
 * Benefit: 100x memory efficiency, stable under load
 */

const { spawn } = require('child_process');
const path = require('path');

class ProcessPool {
  constructor(poolSize = 4, scriptPath = 'M3U_Matrix_Pro.py') {
    this.poolSize = poolSize;
    this.scriptPath = scriptPath;
    this.processes = [];
    this.requestQueue = [];
    this.processingRequest = new Array(poolSize).fill(false);
    this.initialized = false;
  }

  /**
   * Initialize the process pool
   * Creates N persistent Python processes
   */
  async init() {
    if (this.initialized) return;

    console.log(`[ProcessPool] Initializing ${this.poolSize} Python processes...`);
    
    for (let i = 0; i < this.poolSize; i++) {
      const processIndex = i;
      const proc = this._createProcess(i);
      
      this.processes.push({
        id: i,
        process: proc,
        busy: false,
        requestCount: 0
      });

      // Wait for process to be ready
      await this._waitForProcessReady(proc, i);
    }

    this.initialized = true;
    console.log(`[ProcessPool] ✓ Pool initialized with ${this.poolSize} processes`);
  }

  /**
   * Create a single Python process
   * Each process runs in "server mode" listening for requests
   */
  _createProcess(index) {
    const proc = spawn('python3', [
      this.scriptPath,
      '--process-pool',
      String(index)
    ], {
      stdio: ['pipe', 'pipe', 'pipe'],
      detached: false
    });

    proc.on('error', (error) => {
      console.error(`[ProcessPool] Process ${index} error:`, error.message);
    });

    proc.on('exit', (code) => {
      console.warn(`[ProcessPool] Process ${index} exited with code ${code}`);
    });

    // Monitor stderr for errors
    proc.stderr.on('data', (data) => {
      const message = data.toString().trim();
      if (message && !message.includes('debug')) {
        console.warn(`[ProcessPool] Process ${index} stderr:`, message);
      }
    });

    return proc;
  }

  /**
   * Wait for process to signal it's ready
   */
  _waitForProcessReady(proc, index) {
    return new Promise((resolve) => {
      const timeout = setTimeout(() => {
        console.warn(`[ProcessPool] Process ${index} initialization timeout`);
        resolve(); // Assume ready after timeout
      }, 2000);

      proc.stdout.once('data', (data) => {
        clearTimeout(timeout);
        console.log(`[ProcessPool] Process ${index} ready`);
        resolve();
      });
    });
  }

  /**
   * Execute a command using the process pool
   * Returns promise that resolves with output
   */
  async execute(args) {
    if (!this.initialized) {
      await this.init();
    }

    return new Promise((resolve, reject) => {
      const request = { args, resolve, reject, timestamp: Date.now() };
      
      // Try to find available process
      const processIndex = this._findAvailableProcess();
      
      if (processIndex !== -1) {
        // Process available, execute immediately
        this._executeOnProcess(processIndex, request);
      } else {
        // Queue request, will execute when process available
        this.requestQueue.push(request);
      }
    });
  }

  /**
   * Find an available process
   */
  _findAvailableProcess() {
    for (let i = 0; i < this.processes.length; i++) {
      if (!this.processes[i].busy) {
        return i;
      }
    }
    return -1;
  }

  /**
   * Execute request on specific process
   */
  _executeOnProcess(processIndex, request) {
    const pool = this.processes[processIndex];
    pool.busy = true;
    pool.requestCount++;

    const args = request.args;
    const argString = JSON.stringify(args) + '\n';
    let output = '';
    let errorOutput = '';

    // Set up output handlers
    const onData = (chunk) => {
      output += chunk.toString();
    };

    const onError = (chunk) => {
      errorOutput += chunk.toString();
    };

    const onClose = () => {
      // Cleanup
      pool.process.stdout.removeListener('data', onData);
      pool.process.stderr.removeListener('data', onError);
      pool.busy = false;

      // Process result
      try {
        if (errorOutput.trim()) {
          request.reject(new Error(errorOutput));
        } else if (output.trim()) {
          request.resolve(output.trim());
        } else {
          request.reject(new Error('No output from process'));
        }
      } catch (e) {
        request.reject(e);
      }

      // Process next queued request
      if (this.requestQueue.length > 0) {
        const nextRequest = this.requestQueue.shift();
        this._executeOnProcess(processIndex, nextRequest);
      }

      // Log pool status every 100 requests
      if (pool.requestCount % 100 === 0) {
        console.log(`[ProcessPool] Process ${processIndex}: ${pool.requestCount} requests handled`);
      }
    };

    pool.process.stdout.once('data', onData);
    pool.process.stderr.once('data', onError);

    // Timeout after 30 seconds
    const timeout = setTimeout(() => {
      pool.process.stdout.removeListener('data', onData);
      pool.process.stderr.removeListener('data', onError);
      pool.busy = false;
      request.reject(new Error('Process execution timeout'));

      // Process next request
      if (this.requestQueue.length > 0) {
        const nextRequest = this.requestQueue.shift();
        this._executeOnProcess(processIndex, nextRequest);
      }
    }, 30000);

    pool.process.stdout.once('close', () => {
      clearTimeout(timeout);
      onClose();
    });

    // Send request to process (assuming Python script reads from stdin)
    pool.process.stdin.write(argString);
  }

  /**
   * Get pool statistics
   */
  getStats() {
    return {
      poolSize: this.poolSize,
      busyProcesses: this.processes.filter(p => p.busy).length,
      totalRequests: this.processes.reduce((sum, p) => sum + p.requestCount, 0),
      queuedRequests: this.requestQueue.length,
      avgRequests: Math.round(this.processes.reduce((sum, p) => sum + p.requestCount, 0) / this.poolSize)
    };
  }

  /**
   * Shutdown the pool gracefully
   */
  async shutdown() {
    console.log('[ProcessPool] Shutting down...');
    
    for (const pool of this.processes) {
      pool.process.kill();
    }

    this.processes = [];
    this.initialized = false;
    console.log('[ProcessPool] ✓ Shutdown complete');
  }
}

/**
 * Simple fallback: execute without pool (for testing)
 * Creates new process per request (slower, but works)
 */
async function executeProcessSimple(args) {
  return new Promise((resolve, reject) => {
    const python = spawn('python3', args);
    let output = '';
    let errorOutput = '';

    python.stdout.on('data', (data) => {
      output += data.toString();
    });

    python.stderr.on('data', (data) => {
      errorOutput += data.toString();
    });

    const timeout = setTimeout(() => {
      python.kill();
      reject(new Error('Process execution timeout'));
    }, 30000);

    python.on('close', (code) => {
      clearTimeout(timeout);
      if (code === 0) {
        resolve(output.trim());
      } else {
        reject(new Error(errorOutput || `Process exited with code ${code}`));
      }
    });

    python.on('error', reject);
  });
}

// Export both pool and simple executor
module.exports = {
  ProcessPool,
  executeProcessSimple
};