network-stack-regressions.cc

/*
 * Copyright (c) 2026, Pedigree Developers
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted.
 */
 
#include "pedigree/kernel/Atomic.h"
#include "pedigree/kernel/Log.h"
#include "pedigree/kernel/process/Scheduler.h"
#include "pedigree/kernel/process/Semaphore.h"
#include "pedigree/kernel/process/Thread.h"
#include "pedigree/kernel/processor/Processor.h"
#include "pedigree/kernel/time/Time.h"
#include "pedigree/kernel/utilities/new"
 
#include "modules/system/lwip/include/lwip/err.h"
#include "modules/system/lwip/include/lwip/netif.h"
#include "modules/system/lwip/include/lwip/sys.h"
#include "modules/system/network-stack/NetworkStack.h"
#include "system/kernel/core/processor/DeviceHardIrqContext.h"
 
namespace {
class HostedNetworkDevice final : public Network {
 public:
  bool send(size_t, uintptr_t) override {
    return true;
  }
 
  const StationInfo& getStationInfo() override {
    return m_StationInfo;
  }
};
 
struct ReceivePublicationContext {
  static constexpr size_t MaxTrackedBuffers = 257;
 
  ReceivePublicationContext(Network* device, bool holdFirst)
      : device(device),
        holdFirst(holdFirst),
        workerEntered(0),
        allowDispatch(0),
        queued(0),
        queuedWithInterruptsDisabled(0),
        beforeDispatch(0),
        delivered(0),
        staleDiscards(0),
        cancellations(0),
        failures(0) {}
 
  Network* device;
  bool holdFirst;
  Semaphore workerEntered;
  Semaphore allowDispatch;
  Atomic<size_t> queued;
  Atomic<size_t> queuedWithInterruptsDisabled;
  Atomic<size_t> beforeDispatch;
  Atomic<size_t> delivered;
  Atomic<size_t> staleDiscards;
  Atomic<size_t> cancellations;
  Atomic<size_t> failures;
  uintptr_t buffers[MaxTrackedBuffers] = {};
  uint8_t terminalCounts[MaxTrackedBuffers] = {};
 
  bool everyBufferTerminatedOnce(size_t expected) const {
    if (expected > MaxTrackedBuffers) {
      return false;
    }
    for (size_t i = 0; i < expected; ++i) {
      if (!buffers[i] || terminalCounts[i] != 1) {
        return false;
      }
    }
    return true;
  }
};
 
ReceivePublicationContext* g_ReceivePublicationContext = nullptr;
 
void receivePublicationHook(NetworkStack::HostedReceiveEvent event, uintptr_t buffer,
                            Network* device, size_t) {
  ReceivePublicationContext* context = g_ReceivePublicationContext;
  if (!context || device != context->device) {
    return;
  }
 
  bool terminalEvent = false;
  switch (event) {
    case NetworkStack::HostedReceiveEvent::Queued: {
      const size_t index = (context->queued += 1) - 1;
      if (index >= ReceivePublicationContext::MaxTrackedBuffers) {
        context->failures += 1;
        break;
      }
      context->buffers[index] = buffer;
      if (!Processor::getInterrupts()) {
        context->queuedWithInterruptsDisabled += 1;
      }
      break;
    }
    case NetworkStack::HostedReceiveEvent::BeforeDispatch:
      if ((context->beforeDispatch += 1) == 1 && context->holdFirst) {
        context->workerEntered.release();
        if (!context->allowDispatch.acquireForCompletion()) {
          context->failures += 1;
        }
      }
      break;
    case NetworkStack::HostedReceiveEvent::Delivered:
      context->delivered += 1;
      terminalEvent = true;
      break;
    case NetworkStack::HostedReceiveEvent::DiscardedStale:
      context->staleDiscards += 1;
      terminalEvent = true;
      break;
    case NetworkStack::HostedReceiveEvent::Cancelled:
      context->cancellations += 1;
      terminalEvent = true;
      break;
  }
 
  if (!terminalEvent) {
    return;
  }
 
  bool found = false;
  for (size_t i = 0; i < context->queued; ++i) {
    if (context->buffers[i] != buffer) {
      continue;
    }
    found = true;
    if (++context->terminalCounts[i] != 1) {
      context->failures += 1;
    }
    break;
  }
  if (!found) {
    context->failures += 1;
  }
}
 
struct ReceiveAbaContext {
  explicit ReceiveAbaContext(Network* device)
      : device(device),
        workerEntered(0),
        allowDispatch(0),
        buffer(0),
        generation(0),
        queued(0),
        beforeDispatch(0),
        delivered(0),
        staleDiscards(0),
        cancellations(0),
        failures(0) {}
 
  Network* device;
  Semaphore workerEntered;
  Semaphore allowDispatch;
  Atomic<uintptr_t> buffer;
  Atomic<size_t> generation;
  Atomic<size_t> queued;
  Atomic<size_t> beforeDispatch;
  Atomic<size_t> delivered;
  Atomic<size_t> staleDiscards;
  Atomic<size_t> cancellations;
  Atomic<size_t> failures;
};
 
ReceiveAbaContext* g_ReceiveAbaContext = nullptr;
 
void receiveHook(NetworkStack::HostedReceiveEvent event, uintptr_t buffer, Network* device,
                 size_t generation) {
  ReceiveAbaContext* context = g_ReceiveAbaContext;
  if (!context || device != context->device) {
    return;
  }
 
  if (event == NetworkStack::HostedReceiveEvent::Queued) {
    context->queued += 1;
    if (!context->buffer.compareAndSwap(0, buffer)) {
      context->failures += 1;
    }
    context->generation = generation;
    return;
  }
 
  if (buffer != static_cast<uintptr_t>(context->buffer)) {
    return;
  }
 
  switch (event) {
    case NetworkStack::HostedReceiveEvent::BeforeDispatch:
      if ((context->beforeDispatch += 1) != 1) {
        context->failures += 1;
        return;
      }
      context->workerEntered.release();
      if (!context->allowDispatch.acquireForCompletion()) {
        context->failures += 1;
      }
      break;
    case NetworkStack::HostedReceiveEvent::Delivered:
      context->delivered += 1;
      break;
    case NetworkStack::HostedReceiveEvent::DiscardedStale:
      context->staleDiscards += 1;
      break;
    case NetworkStack::HostedReceiveEvent::Cancelled:
      context->cancellations += 1;
      break;
    case NetworkStack::HostedReceiveEvent::Queued:
      break;
  }
}
 
bool check(bool condition, const char* detail) {
  if (condition) {
    return true;
  }
 
  ERROR("HOSTED-NETWORK-TEST: FAIL receive-generation-aba: " << detail);
  return false;
}
 
bool check(bool condition, const char* test, const char* detail) {
  if (condition) {
    return true;
  }
 
  ERROR("HOSTED-NETWORK-TEST: FAIL " << test << ": " << detail);
  return false;
}
 
bool isRegistered(NetworkStack& stack, Network* device) {
  NetworkStack::DeviceLease lease;
  return stack.acquireDevice(device, lease);
}
 
bool waitUntilQueued(Thread* thread, size_t debugState) {
  const Time::Timestamp deadline = Time::getTicks() + (500 * Time::Multiplier::Millisecond);
  while (Time::getTicks() < deadline) {
    Thread::WaitDebugInfo info = {};
    uintptr_t debugAddress = 0;
    if (thread->getWaitDebugInfo(info) && info.queue && info.queued &&
        thread->getDebugState(debugAddress) == debugState) {
      return true;
    }
    Scheduler::instance().yield();
  }
  return false;
}
 
struct DeviceDeregisterContext {
  DeviceDeregisterContext(NetworkStack* stack, Network* device)
      : stack(stack), device(device), entered(0), returned(0) {}
 
  NetworkStack* stack;
  Network* device;
  Atomic<size_t> entered;
  Atomic<size_t> returned;
};
 
int deregisterLeasedDevice(void* parameter) {
  DeviceDeregisterContext* context = reinterpret_cast<DeviceDeregisterContext*>(parameter);
  context->entered += 1;
  context->stack->deRegisterDevice(context->device);
  context->returned += 1;
  return 0;
}
 
bool deviceLeaseDeregisterDrain() {
  static const char* Test = "device-lease-deregister-drain";
 
  NetworkStack& stack = NetworkStack::instance();
  alignas(HostedNetworkDevice) uint8_t deviceStorage[sizeof(HostedNetworkDevice)];
  HostedNetworkDevice* original = new (deviceStorage) HostedNetworkDevice();
  stack.registerDevice(original);
  const size_t originalGeneration = NetworkStack::getHostedRegistrationGeneration(original);
 
  NetworkStack::DeviceLease held;
  const bool acquired = stack.acquireDevice(original, held);
  struct netif* originalInterface = acquired ? held.interface() : nullptr;
  DeviceDeregisterContext context(&stack, original);
  Thread* remover = nullptr;
  bool drainPublished = false;
  bool lateRejected = false;
  bool usableWhileDraining = false;
 
  if (acquired) {
    remover = new Thread(Scheduler::instance().getKernelProcess(), deregisterLeasedDevice, &context,
                         nullptr, false, true);
    remover->setName("hosted network device deregister");
    drainPublished = waitUntilQueued(remover, Thread::CallbackDrain);
 
    NetworkStack::DeviceLease late;
    lateRejected = !stack.acquireDevice(original, late);
 
    String name;
    held.device()->getName(name);
    const StationInfo& info = held.device()->getStationInfo();
    usableWhileDraining = held.device() == original && held.interface() == originalInterface &&
                          originalInterface && originalInterface->state == original &&
                          info.nPackets == 0;
    held = NetworkStack::DeviceLease();
  }
 
  const bool removerJoined = remover && remover->join();
  const bool unregistered =
      !isRegistered(stack, original) && !NetworkStack::getHostedRegistrationGeneration(original);
  if (!remover) {
    stack.deRegisterDevice(original);
  }
  original->~HostedNetworkDevice();
 
  HostedNetworkDevice* replacement = new (deviceStorage) HostedNetworkDevice();
  stack.registerDevice(replacement);
  const size_t replacementGeneration = NetworkStack::getHostedRegistrationGeneration(replacement);
  NetworkStack::DeviceLease replacementLease;
  const bool replacementRegistered = stack.acquireDevice(replacement, replacementLease);
  replacementLease = NetworkStack::DeviceLease();
  stack.deRegisterDevice(replacement);
  replacement->~HostedNetworkDevice();
 
  const bool passed =
      check(acquired && originalGeneration && context.entered == 1 && drainPublished &&
                context.returned == 1 && lateRejected && usableWhileDraining && removerJoined &&
                unregistered && replacementRegistered && replacementGeneration &&
                replacementGeneration != originalGeneration,
            Test, "deregistration did not unpublish, drain, and retire one held registration");
  if (passed) {
    NOTICE("HOSTED-NETWORK-TEST: PASS " << Test);
  }
  return passed;
}
 
bool mailboxTryPostRejectsHardIrq() {
  static const char* Test = "mailbox-hard-irq-trypost";
 
  sys_mbox_t mailbox = nullptr;
  if (sys_mbox_new(&mailbox, 1) != ERR_OK) {
    return check(false, Test, "mailbox creation failed");
  }
 
  int payload = 0;
  err_t result = ERR_OK;
  size_t previousDepth = 0;
  bool restorationArmed = false;
  {
    DeviceHardIrqContext hardIrq(previousDepth, restorationArmed);
    result = sys_mbox_trypost(&mailbox, &payload);
  }
 
  void* fetched = nullptr;
  const u32_t fetchResult = sys_arch_mbox_tryfetch(&mailbox, &fetched);
  sys_mbox_free(&mailbox);
 
  const bool passed =
      check(result == ERR_WOULDBLOCK && fetchResult == SYS_MBOX_EMPTY && !fetched, Test,
            "the ISR call enqueued data or entered a blocking notification path");
  if (passed) {
    NOTICE("HOSTED-NETWORK-TEST: PASS " << Test);
  }
  return passed;
}
 
bool interruptReceivePublication() {
  static const char* Test = "receive-interrupt-publication";
 
  NetworkStack& stack = NetworkStack::instance();
  HostedNetworkDevice device;
  stack.registerDevice(&device);
 
  ReceivePublicationContext context(&device, true);
  g_ReceivePublicationContext = &context;
  NetworkStack::setHostedReceiveHook(receivePublicationHook);
 
  uint8_t packet[64] = {};
  const bool interrupts = Processor::getInterrupts();
  Processor::setInterrupts(false);
  stack.receive(sizeof(packet), reinterpret_cast<uintptr_t>(packet), &device, 0);
  const bool remainedDisabled = !Processor::getInterrupts();
  Processor::setInterrupts(interrupts);
 
  const bool workerHeld = context.workerEntered.acquire(1, 2);
  bool unregistered = false;
  if (workerHeld) {
    stack.deRegisterDevice(&device);
    unregistered =
        !isRegistered(stack, &device) && !NetworkStack::getHostedRegistrationGeneration(&device);
  }
  context.allowDispatch.release();
  const bool drained = stack.drain();
  NetworkStack::setHostedReceiveHook(nullptr);
  g_ReceivePublicationContext = nullptr;
  if (!unregistered) {
    stack.deRegisterDevice(&device);
  }
 
  const size_t terminalOwnershipEvents = static_cast<size_t>(context.delivered) +
                                         static_cast<size_t>(context.staleDiscards) +
                                         static_cast<size_t>(context.cancellations);
 
  bool passed = true;
  passed &= check(interrupts && remainedDisabled && context.queued == 1 &&
                      context.queuedWithInterruptsDisabled == 1,
                  Test, "receive did not publish while preserving IF=0");
  passed &= check(workerHeld && unregistered && drained && context.beforeDispatch == 1 &&
                      !context.delivered && context.staleDiscards == 1 && !context.cancellations &&
                      terminalOwnershipEvents == 1 && context.everyBufferTerminatedOnce(1) &&
                      !context.failures,
                  Test, "the interrupt-published pbuf did not transfer exactly once");
  passed &= check(NetworkStack::getHostedReceiveRequestCapacity() == 256, Test,
                  "the preallocated bank no longer preserves receive capacity");
 
  if (passed) {
    NOTICE("HOSTED-NETWORK-TEST: PASS " << Test);
  }
  return passed;
}
 
bool boundedReceiveBurst() {
  static const char* Test = "receive-bounded-burst";
 
  NetworkStack& stack = NetworkStack::instance();
  HostedNetworkDevice device;
  stack.registerDevice(&device);
 
  ReceivePublicationContext context(&device, true);
  g_ReceivePublicationContext = &context;
  NetworkStack::setHostedReceiveHook(receivePublicationHook);
 
  uint8_t packet[64] = {};
  stack.receive(sizeof(packet), reinterpret_cast<uintptr_t>(packet), &device, 0);
  const bool workerHeld = context.workerEntered.acquire(1, 2);
  const size_t capacity = NetworkStack::getHostedReceiveRequestCapacity();
  if (workerHeld) {
    // The active request owns one token. Fill every remaining token, then
    // submit one packet whose ownership must be cancelled synchronously.
    for (size_t i = 1; i <= capacity; ++i) {
      stack.receive(sizeof(packet), reinterpret_cast<uintptr_t>(packet), &device, 0);
    }
  }
  const bool rejectedWhileHeld = context.cancellations == 1;
 
  bool unregistered = false;
  if (workerHeld) {
    stack.deRegisterDevice(&device);
    unregistered =
        !isRegistered(stack, &device) && !NetworkStack::getHostedRegistrationGeneration(&device);
  }
 
  // Also releases a worker which arrived after the bounded wait above.
  context.allowDispatch.release();
  const bool drained = stack.drain();
 
  NetworkStack::setHostedReceiveHook(nullptr);
  g_ReceivePublicationContext = nullptr;
  if (!unregistered) {
    stack.deRegisterDevice(&device);
  }
 
  const size_t terminalOwnershipEvents = static_cast<size_t>(context.delivered) +
                                         static_cast<size_t>(context.staleDiscards) +
                                         static_cast<size_t>(context.cancellations);
 
  bool passed = true;
  passed &= check(workerHeld && rejectedWhileHeld && unregistered && !context.failures, Test,
                  "the full token bank did not reject one pbuf synchronously");
  passed &= check(drained && context.queued == capacity + 1 && context.beforeDispatch == capacity &&
                      !context.delivered && context.staleDiscards == capacity &&
                      context.cancellations == 1,
                  Test, "the accepted/rejected burst counts were inconsistent");
  passed &= check(
      terminalOwnershipEvents == capacity + 1 && context.everyBufferTerminatedOnce(capacity + 1),
      Test, "a burst pbuf did not have exactly one terminal owner");
 
  if (passed) {
    NOTICE("HOSTED-NETWORK-TEST: PASS " << Test);
  }
  return passed;
}
 
bool queuedReceiveGenerationAba() {
  NetworkStack& stack = NetworkStack::instance();
  alignas(HostedNetworkDevice) uint8_t deviceStorage[sizeof(HostedNetworkDevice)];
  HostedNetworkDevice* original = new (deviceStorage) HostedNetworkDevice();
 
  stack.registerDevice(original);
  const size_t originalGeneration = NetworkStack::getHostedRegistrationGeneration(original);
  bool passed = true;
  passed &= check(isRegistered(stack, original) && originalGeneration,
                  "the original device was not registered");
 
  ReceiveAbaContext context(original);
  g_ReceiveAbaContext = &context;
  NetworkStack::setHostedReceiveHook(receiveHook);
 
  uint8_t packet[64] = {};
  stack.receive(sizeof(packet), reinterpret_cast<uintptr_t>(packet), original, 0);
 
  const bool workerHeld = context.workerEntered.acquire(1, 2);
  HostedNetworkDevice* replacement = nullptr;
  size_t replacementGeneration = 0;
  bool unregistered = false;
  bool reusedAddress = false;
  bool replacementRegistered = false;
 
  if (workerHeld) {
    stack.deRegisterDevice(original);
    unregistered =
        !isRegistered(stack, original) && !NetworkStack::getHostedRegistrationGeneration(original);
    original->~HostedNetworkDevice();
 
    replacement = new (deviceStorage) HostedNetworkDevice();
    reusedAddress = replacement == original;
    stack.registerDevice(replacement);
    replacementGeneration = NetworkStack::getHostedRegistrationGeneration(replacement);
    replacementRegistered = isRegistered(stack, replacement) && replacementGeneration;
  }
 
  // Also makes a late-arriving worker safe after a timeout above.
  context.allowDispatch.release();
  const bool drained = stack.drain();
 
  NetworkStack::setHostedReceiveHook(nullptr);
  g_ReceiveAbaContext = nullptr;
 
  if (replacement) {
    stack.deRegisterDevice(replacement);
    replacement->~HostedNetworkDevice();
  } else {
    stack.deRegisterDevice(original);
    original->~HostedNetworkDevice();
  }
 
  const size_t terminalOwnershipEvents = static_cast<size_t>(context.delivered) +
                                         static_cast<size_t>(context.staleDiscards) +
                                         static_cast<size_t>(context.cancellations);
 
  passed &= check(workerHeld && drained && !context.failures,
                  "the queued receive could not be held and drained deterministically");
  passed &= check(context.queued == 1 && context.beforeDispatch == 1 &&
                      context.generation == originalGeneration,
                  "the queued request did not retain the original registration");
  passed &= check(unregistered && reusedAddress && replacementRegistered &&
                      replacementGeneration != originalGeneration,
                  "unregister/re-register did not create a new identity at one address");
  passed &= check(context.staleDiscards == 1 && !context.delivered && !context.cancellations,
                  "stale queued work reached the replacement or the wrong release path");
  passed &= check(terminalOwnershipEvents == 1,
                  "the receive buffer did not have exactly one terminal owner");
 
  if (passed) {
    NOTICE("HOSTED-NETWORK-TEST: PASS receive-generation-aba");
  }
  return passed;
}
}  // namespace
 
bool runHostedNetworkStackRegressions() {
  bool passed = mailboxTryPostRejectsHardIrq();
  passed &= deviceLeaseDeregisterDrain();
  passed &= interruptReceivePublication();
  passed &= boundedReceiveBurst();
  passed &= queuedReceiveGenerationAba();
  return passed;
}