Tracing Linux Epoll Starvation and Thread Pool Congestion in High-Throughput Node.js Services via eBPF — code snippets

snippet-1.js

const fastify = require('fastify')({ logger: false });
const crypto = require('crypto');
const fs = require('fs').promises;
const path = require('path');

// Simulate a database verification using scrypt
const verifyPassword = (password, salt) => {
  return new Promise((resolve, reject) => {
    // This executes on the libuv thread pool
    crypto.scrypt(password, salt, 64, (err, derivedKey) => {
      if (err) return reject(err);
      resolve(derivedKey.toString('hex'));
    });
  });
};

fastify.post('/api/auth/login', async (request, reply) => {
  const { password, salt } = request.body;
  if (!password || !salt) {
    return reply.status(400).send({ error: 'Missing credentials' });
  }
  
  // High-latency CPU bound hashing task pushed to thread pool
  const hash = await verifyPassword(password, salt);
  return { status: 'authenticated', hash };
});

fastify.get('/health', async (request, reply) => {
  const filePath = path.join(__dirname, 'status.txt');
  try {
    // File I/O also relies on the libuv thread pool
    const data = await fs.readFile(filePath, 'utf8');
    return { status: 'ok', detail: data.trim() };
  } catch (err) {
    return { status: 'ok', detail: 'no status file' };
  }
});

const start = async () => {
  try {
    await fastify.listen({ port: 3000, host: '0.0.0.0' });
    console.log(`Server listening on ${fastify.server.address().port}`);
  } catch (err) {
    process.exit(1);
  }
};
start();

snippet-2.txt

#include <uapi/linux/ptrace.h>
#include <linux/sched.h>

struct val_t {
    u64 ts;
    u32 pid;
};

// Map to store system call start times
BPF_HASH(start_tb, u64, struct val_t);

// Histogram to record epoll_wait latency distributions
BPF_HISTOGRAM(dist);

// Hook into epoll_wait entry
TRACEPOINT_PROBE(syscalls, sys_enter_epoll_wait) {
    u64 id = bpf_get_current_pid_tgid();
    u32 tgid = id >> 32; // Process ID
    u32 pid = id;        // Thread ID

    // Target a specific Node.js process if configured (passed via Python loader)
    #ifdef TARGET_PID
    if (tgid != TARGET_PID) {
        return 0;
    }
    #endif

    struct val_t val = {};
    val.ts = bpf_ktime_get_ns();
    val.pid = tgid;

    start_tb.update(&id, &val);
    return 0;
}

// Hook into epoll_wait exit
TRACEPOINT_PROBE(syscalls, sys_exit_epoll_wait) {
    u64 id = bpf_get_current_pid_tgid();
    struct val_t *valp = start_tb.lookup(&id);

    if (valp == 0) {
        // Missed the entry event
        return 0;
    }

    u64 delta = bpf_ktime_get_ns() - valp->ts;
    start_tb.delete(&id);

    // Convert nanoseconds to microseconds
    u64 latency_us = delta / 1000;

    // Increment histogram bin
    dist.increment(bpf_log2l(latency_us));
    return 0;
}

snippet-3.py

from bcc import BPF
import time
import sys

if len(sys.argv) < 2:
    print("Usage: python3 trace_epoll.py <PID_OF_NODE_PROCESS>")
    sys.exit(1)

target_pid = int(sys.argv[1])

# Load and compile the BPF program, passing the Target PID macro
bpf_text = open('epoll_trace.c', 'r').read()
bpf_text = f"#define TARGET_PID {target_pid}\n" + bpf_text

b = BPF(text=bpf_text)
print(f"Tracing epoll_wait latencies for Node.js Process {target_pid}... Press Ctrl+C to stop.")

try:
    while True:
        time.sleep(5)
        print("\n--- epoll_wait Latency Distribution (Microseconds) ---")
        b["dist"].print_log2_hist("usecs")
        b["dist"].clear()
except KeyboardInterrupt:
    print("Detaching probes and exiting...")

snippet-4.txt

#include <uapi/linux/ptrace.h>
#include <linux/sched.h>

// Hash map to store wakeup timestamps for threads
BPF_HASH(wakeup_timestamp, u32, u64);

// Histogram to record scheduling delays (runqueue latencies)
BPF_HISTOGRAM(sched_latency);

// Hook scheduler wakeup: when a thread is marked runnable
TRACEPOINT_PROBE(sched, sched_wakeup) {
    u32 pid = args->pid; // Thread ID of the target being awakened
    u64 ts = bpf_ktime_get_ns();

    // Check if the thread group name belongs to 'node' to avoid tracking system noise
    char comm[TASK_COMM_LEN];
    bpf_get_current_comm(&comm, sizeof(comm));
    
    // We only record wakeup timestamps for our target thread groups
    wakeup_timestamp.update(&pid, &ts);
    return 0;
}

// Hook scheduler switch: when a thread starts executing on a CPU
TRACEPOINT_PROBE(sched, sched_switch) {
    u32 next_pid = args->next_pid; // Thread ID about to run
    u64 *tsp = wakeup_timestamp.lookup(&next_pid);

    if (tsp != 0) {
        u64 delta = bpf_ktime_get_ns() - *tsp;
        wakeup_timestamp.delete(&next_pid);

        // Convert latency to microseconds
        u64 latency_us = delta / 1000;

        // Populate the histogram
        sched_latency.increment(bpf_log2l(latency_us));
    }
    return 0;
}

snippet-5.py

from bcc import BPF
import time
import os

# Load the scheduler runqueue latency tracer
b = BPF(src_file="sched_trace.c")
print("Tracing kernel runqueue scheduling delay for Node.js threads... Press Ctrl+C to stop.")

try:
    while True:
        time.sleep(5)
        print("\n--- Thread Runqueue Scheduling Delay (Microseconds) ---")
        # Print the distribution log histogram
        b["sched_latency"].print_log2_hist("usecs")
        b["sched_latency"].clear()
except KeyboardInterrupt:
    print("Detaching scheduler probes...")

snippet-6.js

const fastify = require('fastify')({ logger: false });
const dns = require('dns').promises;
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const path = require('path');

// Ensure UV_THREADPOOL_SIZE environment variable is adjusted programmatically if not set
if (!process.env.UV_THREADPOOL_SIZE) {
  process.env.UV_THREADPOOL_SIZE = require('os').cpus().length.toString();
}

// 1. Decoupled DNS Resolution Configuration
// Replace the default lookup resolver with non-blocking dns.resolve4
const customLookup = async (hostname, options, cb) => {
  try {
    const addresses = await dns.resolve4(hostname);
    if (!addresses || addresses.length === 0) {
      return cb(new Error(`ENOTFOUND ${hostname}`));
    }
    // Return first resolved IP address
    cb(null, addresses[0], 4);
  } catch (err) {
    cb(err);
  }
};

// 2. Offloading CPU hashing to V8 Worker Threads
const runHashWorker = (password, salt) => {
  return new Promise((resolve, reject) => {
    const worker = new Worker(__filename, {
      workerData: { password, salt }
    });
    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) reject(new Error(`Worker stopped with exit code ${code}`));
    });
  });
};

if (isMainThread) {
  // Main Thread server execution
  fastify.post('/api/auth/login', async (request, reply) => {
    const { password, salt } = request.body;
    if (!password || !salt) {
      return reply.status(400).send({ error: 'Missing parameters' });
    }
    
    // Executes in a separate V8 Isolate, bypassing libuv threadpool entirely
    const hash = await runHashWorker(password, salt);
    return { status: 'authenticated', hash };
  });

  fastify.get('/api/external-fetch', async (request, reply) => {
    // Axios or native fetch can consume the optimized lookup to prevent thread exhaustion
    const response = await fetch('https://api.github.com/users/octocat', {
      // Custom lookup configuration avoids getaddrinfo threadpool allocation
      lookup: customLookup
    });
    const data = await response.json();
    return { data };
  });

  fastify.listen({ port: 3000 }, () => {
    console.log(`Mitigated Node.js server running on port 3000 (UV_THREADPOOL_SIZE=${process.env.UV_THREADPOOL_SIZE})`);
  });
} else {
  // Worker Thread execution context
  const crypto = require('crypto');
  const { password, salt } = workerData;

  // CPU bound crypt task executes on dedicated V8 worker CPU core
  crypto.scrypt(password, salt, 64, (err, derivedKey) => {
    if (err) {
      process.exit(1);
    }
    parentPort.postMessage(derivedKey.toString('hex'));
  });
}