multi_socket_driver.cc

Go to the documentation of this file.

/*
 * Copyright (C) Max Dymond, <cmeister2@gmail.com>, et al.
 *
 * SPDX-License-Identifier: curl
 */

 
#include "proto_fuzzer/multi_socket_driver.h"
 
#include <poll.h>
 
#include <array>
#include <cstddef>
#include <cstdint>
 
namespace proto_fuzzer {
 
namespace {

short PollEventsForInterest(int interest) {
  short events = 0;
  if ((interest & CURL_POLL_IN) != 0) {
    events |= POLLIN;
  }
  if ((interest & CURL_POLL_OUT) != 0) {
    events |= POLLOUT;
  }
  return events;
}

int CurlEventsForPollResult(short events) {
  int result = 0;
  if ((events & (POLLIN | POLLHUP)) != 0) {
    // A stream hangup remains readable until curl consumes EOF.
    result |= CURL_CSELECT_IN;
  }
  if ((events & POLLOUT) != 0) {
    result |= CURL_CSELECT_OUT;
  }
  if ((events & (POLLERR | POLLHUP | POLLNVAL)) != 0) {
    result |= CURL_CSELECT_ERR;
  }
  return result;
}
 
}  // namespace

MultiSocketDriver::MultiSocketDriver()
    : multi_(nullptr), timeout_ms_(-1), timer_pending_(false), watch_exhausted_(false), generation_(0) {}

MultiSocketDriver::~MultiSocketDriver() = default;

bool MultiSocketDriver::Install(CURLM* multi) {
  multi_ = multi;
  if (multi_ == nullptr) {
    return false;
  }
  return curl_multi_setopt(multi_, CURLMOPT_SOCKETFUNCTION, &MultiSocketDriver::SocketCallback) == CURLM_OK &&
         curl_multi_setopt(multi_, CURLMOPT_SOCKETDATA, this) == CURLM_OK &&
         curl_multi_setopt(multi_, CURLMOPT_TIMERFUNCTION, &MultiSocketDriver::TimerCallback) == CURLM_OK &&
         curl_multi_setopt(multi_, CURLMOPT_TIMERDATA, this) == CURLM_OK;
}

CURLMcode MultiSocketDriver::Start(int* running_handles) {
  if (multi_ == nullptr) {
    return CURLM_BAD_HANDLE;
  }
  timer_pending_ = false;
  return curl_multi_socket_action(multi_, CURL_SOCKET_TIMEOUT, 0, running_handles);
}

MultiSocketDriver::DriveResult MultiSocketDriver::DriveReady(int* running_handles) {
  DriveResult result;
  if (multi_ == nullptr) {
    result.code = CURLM_BAD_HANDLE;
    return result;
  }
 
  std::array<struct pollfd, kMaxWatches> poll_fds{};
  std::size_t poll_count = 0;
  for (const Watch& watch : watches_) {
    if (!watch.active) {
      continue;
    }
    poll_fds[poll_count].fd = watch.socket;
    poll_fds[poll_count].events = PollEventsForInterest(watch.interest);
    ++poll_count;
  }
 
  const std::uint64_t generation_before = generation_;
  const int running_before = running_handles == nullptr ? 0 : *running_handles;
  bool action_dispatched = false;
  if (poll_count != 0 && ::poll(poll_fds.data(), static_cast<nfds_t>(poll_count), 0) > 0) {
    for (std::size_t index = 0; index < poll_count; ++index) {
      if (running_handles != nullptr && *running_handles == 0) {
        break;
      }
      // An earlier action can synchronously remove or repurpose any later fd
      // in the snapshot. Revalidate it against the callback-owned table so a
      // stale readiness notification never reaches curl under a new meaning.
      Watch* current = FindWatch(poll_fds[index].fd);
      if (current == nullptr) {
        continue;
      }
      const int events = CurlEventsForPollResult(poll_fds[index].revents);
      if (events == 0) {
        continue;
      }
      result.code = curl_multi_socket_action(multi_, poll_fds[index].fd, events, running_handles);
      action_dispatched = true;
      // Process at most one snapshot entry. Its callbacks can repurpose an fd
      // with the same numeric value and interest, which no post-hoc lookup can
      // distinguish; the outer loop rebuilds readiness from fresh watches.
      break;
    }
  }
 
  if (result.code == CURLM_OK && timer_pending_ && timeout_ms_ == 0) {
    // Clear first: the action may synchronously install another zero timer,
    // which belongs to the next outer loop turn rather than recursive work.
    timer_pending_ = false;
    result.code = curl_multi_socket_action(multi_, CURL_SOCKET_TIMEOUT, 0, running_handles);
    action_dispatched = true;
  }
 
  const int running_after = running_handles == nullptr ? 0 : *running_handles;
  // A successful action can consume buffered protocol bytes without changing
  // either callbacks or handle count, so dispatch itself is observable
  // progress. The outer fixed operation cap still bounds permanently-ready
  // sockets.
  result.made_progress = action_dispatched || generation_ != generation_before || running_after != running_before;
  return result;
}

void MultiSocketDriver::ProbeControlApis() {
  if (multi_ == nullptr) {
    return;
  }
  long timeout_ms = -1;
  (void)curl_multi_timeout(multi_, &timeout_ms);
  (void)curl_multi_wakeup(multi_);
}

int MultiSocketDriver::SocketCallback(CURL* /*easy*/, curl_socket_t socket, int what, void* user_data,
                                      void* socket_data) {
  return static_cast<MultiSocketDriver*>(user_data)->UpdateSocket(socket, what, socket_data);
}

int MultiSocketDriver::TimerCallback(CURLM* /*multi*/, long timeout_ms, void* user_data) {
  return static_cast<MultiSocketDriver*>(user_data)->UpdateTimer(timeout_ms);
}

int MultiSocketDriver::UpdateSocket(curl_socket_t socket, int what, void* socket_data) {
  Watch* watch = FindWatch(socket);
  if (what == CURL_POLL_REMOVE) {
    if (watch != nullptr) {
      (void)curl_multi_assign(multi_, socket, nullptr);
      watch->active = false;
      watch->socket = CURL_SOCKET_BAD;
      watch->interest = CURL_POLL_NONE;
      ++generation_;
    }
    return 0;
  }
 
  if (watch == nullptr) {
    watch = FindFreeWatch();
    if (watch == nullptr) {
      watch_exhausted_ = true;
      return -1;
    }
    watch->socket = socket;
    watch->active = true;
    (void)curl_multi_assign(multi_, socket, watch);
    ++generation_;
  } else if (socket_data != nullptr && socket_data != watch) {
    // Reassert the stable association if an unusual transition supplied a
    // different application pointer. Never dereference foreign socket_data.
    (void)curl_multi_assign(multi_, socket, watch);
  }
 
  if (watch->interest != what) {
    watch->interest = what;
    ++generation_;
  }
  return 0;
}

int MultiSocketDriver::UpdateTimer(long timeout_ms) {
  if (!timer_pending_ || timeout_ms_ != timeout_ms) {
    ++generation_;
  }
  timeout_ms_ = timeout_ms;
  timer_pending_ = timeout_ms >= 0;
  return 0;
}

MultiSocketDriver::Watch* MultiSocketDriver::FindWatch(curl_socket_t socket) {
  for (Watch& watch : watches_) {
    if (watch.active && watch.socket == socket) {
      return &watch;
    }
  }
  return nullptr;
}

MultiSocketDriver::Watch* MultiSocketDriver::FindFreeWatch() {
  for (Watch& watch : watches_) {
    if (!watch.active) {
      return &watch;
    }
  }
  return nullptr;
}
 
}  // namespace proto_fuzzer