/*

 * Copyright (C) Max Dymond, <cmeister2@gmail.com>, et al.

 *

 * SPDX-License-Identifier: curl

 */

/// @file

/// @brief Implementation of the bounded concurrent easy-handle runner.

#include "proto_fuzzer/multi_transfer_runner.h"

#define CURL_ALLOW_OLD_MULTI_SOCKET

#include <curl/curl.h>

#include <curl/multi.h>

#include <algorithm>

#include <array>

#include <cstddef>

#include <cstdio>

#include <memory>

#include <string>

#include "proto_fuzzer/curl_raii.h"

#include "proto_fuzzer/mock_server.h"

#include "proto_fuzzer/multi_socket_driver.h"

#include "proto_fuzzer/option_apply.h"

#include "proto_fuzzer/request_data.h"

#include "proto_fuzzer/scenario_limits.h"

#include "proto_fuzzer/transfer_session.h"

namespace proto_fuzzer {

namespace {

/// Track shared-multi membership separately from each transfer's resources.

struct TransferState {

  TransferSession session;

  bool attached = false;

};

constexpr int kMaxDriveIterations = 512;

constexpr int kMaxIdleIterations = 8;

constexpr curl_off_t kStateProbeReceiveRate = 1;

constexpr long kStateProbeTimeoutMs = 20;

struct FileCloser {

  void operator()(FILE* file) const {

    if (file != nullptr) {

      std::fclose(file);

    }

  }

};

using FilePtr = std::unique_ptr<FILE, FileCloser>;

/// Install the one intentionally slow transfer shape without exposing its

/// values through generic SetOption mutations. The anonymous stream is owned

/// by RunMultiTransferScenario and outlives every configured easy handle.

bool ApplyStateProbeOptions(CURL* easy, curl::fuzzer::proto::MultiStateProbeProfile profile, FILE* progress_sink) {

  if (profile != curl::fuzzer::proto::MULTI_STATE_PROBE_PROGRESS_RATE_LIMIT || progress_sink == nullptr) {

    return false;

  }

  return curl_easy_setopt(easy, CURLOPT_STDERR, progress_sink) == CURLE_OK &&

         curl_easy_setopt(easy, CURLOPT_TIMEOUT_MS, kStateProbeTimeoutMs) == CURLE_OK &&

         curl_easy_setopt(easy, CURLOPT_NOPROGRESS, 0L) == CURLE_OK &&

         curl_easy_setopt(easy, CURLOPT_MAX_RECV_SPEED_LARGE, kStateProbeReceiveRate) == CURLE_OK;

}

std::size_t RuntimeTransferCount(const curl::fuzzer::proto::MultiPlan& plan) {

  return std::max(scenario_limits::kMinMultiTransfers,

                  std::min(static_cast<std::size_t>(plan.transfer_count()), scenario_limits::kMaxMultiTransfers));

}

void RecordNotification(CURLM*, unsigned int notification, CURL*, void* userdata) {

  auto* stats = static_cast<MultiTransferRunStats*>(userdata);

  if (stats == nullptr) {

    return;

  }

  if (notification == CURLMNOTIFY_INFO_READ) {

    ++stats->info_read_notifications;

  } else if (notification == CURLMNOTIFY_EASY_DONE) {

    ++stats->easy_done_notifications;

  }

}

/// Consume every currently queued completion before handle removal can discard

/// it. Re-added handles may complete more than once, so count messages rather

/// than only distinct easy pointers.

bool DrainCompletionMessages(CURLM* multi, std::array<TransferState, scenario_limits::kMaxMultiTransfers>* transfers,

                             std::size_t transfer_count, MultiTransferRunStats* stats) {

  bool consumed = false;

  int messages_remaining = 0;

  CURLMsg* message = nullptr;

  while ((message = curl_multi_info_read(multi, &messages_remaining)) != nullptr) {

    consumed = true;

    if (message->msg != CURLMSG_DONE) {

      continue;

    }

    for (std::size_t index = 0; index < transfer_count; ++index) {

      TransferState& transfer = (*transfers)[index];

      if (transfer.session.easy() == message->easy_handle) {

        ++stats->completion_messages;

        break;

      }

    }

  }

  return consumed;

}

/// Apply one schema-safe lifecycle transition. Return true only when libcurl

/// accepted a state change; consuming an ineffective action is tracked

/// separately so later ordered actions remain reachable.

bool ApplyAction(const curl::fuzzer::proto::MultiAction& action, CURLM* multi,

                 std::array<TransferState, scenario_limits::kMaxMultiTransfers>* transfers,

                 std::size_t transfer_count) {

  if (transfer_count == 0) {

    return false;

  }

  TransferState& transfer = (*transfers)[static_cast<std::size_t>(action.transfer_selector()) % transfer_count];

  CURL* easy = transfer.session.easy();

  if (easy == nullptr) {

    return false;

  }

  switch (action.kind()) {

    case curl::fuzzer::proto::MULTI_ACTION_PAUSE_RECV:

      return curl_easy_pause(easy, CURLPAUSE_RECV) == CURLE_OK;

    case curl::fuzzer::proto::MULTI_ACTION_PAUSE_SEND:

      return curl_easy_pause(easy, CURLPAUSE_SEND) == CURLE_OK;

    case curl::fuzzer::proto::MULTI_ACTION_PAUSE_ALL:

      return curl_easy_pause(easy, CURLPAUSE_ALL) == CURLE_OK;

    case curl::fuzzer::proto::MULTI_ACTION_RESUME:

      return curl_easy_pause(easy, CURLPAUSE_CONT) == CURLE_OK;

    case curl::fuzzer::proto::MULTI_ACTION_REMOVE:

      if (transfer.attached && curl_multi_remove_handle(multi, easy) == CURLM_OK) {

        transfer.attached = false;

        return true;

      }

      return false;

    case curl::fuzzer::proto::MULTI_ACTION_READD:

      if (!transfer.attached && curl_multi_add_handle(multi, easy) == CURLM_OK) {

        transfer.attached = true;

        return true;

      }

      return false;

    case curl::fuzzer::proto::MULTI_ACTION_NONE:

    default:

      return false;

  }

}

/// Cover the public wait/poll/fdset queries from a valid shared-multi state

/// without introducing wall-clock delay.

void ProbeMultiWaitApis(CURLM* multi) {

  int numfds = 0;

  (void)curl_multi_poll(multi, nullptr, 0, 0, &numfds);

  (void)curl_multi_wait(multi, nullptr, 0, 0, &numfds);

  unsigned int waitfd_count = 0;

  (void)curl_multi_waitfds(multi, nullptr, 0, &waitfd_count);

  std::array<curl_waitfd, scenario_limits::kMaxMultiTransfers * 2> waitfds{};

  (void)curl_multi_waitfds(multi, waitfds.data(), static_cast<unsigned int>(waitfds.size()), &waitfd_count);

  fd_set readfds;

  fd_set writefds;

  fd_set exceptfds;

  FD_ZERO(&readfds);

  FD_ZERO(&writefds);

  FD_ZERO(&exceptfds);

  int maxfd = -1;

  (void)curl_multi_fdset(multi, &readfds, &writefds, &exceptfds, &maxfd);

  long timeout_ms = -1;

  (void)curl_multi_timeout(multi, &timeout_ms);

  CURL** handles = curl_multi_get_handles(multi);

  curl_free(handles);

  constexpr CURLMinfo_offt kOffsetInfo[] = {

      CURLMINFO_XFERS_CURRENT, CURLMINFO_XFERS_RUNNING, CURLMINFO_XFERS_PENDING,

      CURLMINFO_XFERS_DONE,    CURLMINFO_XFERS_ADDED,

  };

  for (const CURLMinfo_offt info : kOffsetInfo) {

    curl_off_t value = 0;

    (void)curl_multi_get_offt(multi, info, &value);

  }

}

/// Exercise the two deprecated public entrypoints as real symbols rather than

/// curl_multi_socket's compatibility macro. Their output is deliberately not

/// used to select the primary runner: either call may advance a transfer, and

/// the normal bounded loop consumes whatever state remains.

void ProbeLegacyMultiSocketApis(CURLM* multi) {

  int running_handles = 0;

#if defined(__GNUC__)

#pragma GCC diagnostic push

#pragma GCC diagnostic ignored "-Wdeprecated-declarations"

#endif

  (void)curl_multi_socket(multi, CURL_SOCKET_TIMEOUT, &running_handles);

  (void)curl_multi_socket_all(multi, &running_handles);

#if defined(__GNUC__)

#pragma GCC diagnostic pop

#endif

}

}  // namespace

/// Drive two to four easy handles through one CURLM and one bounded in-process

/// HTTP peer. Runtime bounds repeat the mutation policy's limits so direct

/// unit-test and standalone calls cannot make work proportional to unchecked

/// input.

/// @param scenario HTTP transfer shape and concurrent lifecycle plan.

/// @return Bounded setup, action, and completion counts.

MultiTransferRunStats RunMultiTransferScenario(const curl::fuzzer::proto::Scenario& scenario) {

  MultiTransferRunStats stats;

  const auto& plan = scenario.multi_plan();

  const std::size_t transfer_count = RuntimeTransferCount(plan);

  stats.configured_handles = transfer_count;

  FilePtr progress_sink;

  if (plan.state_probe_profile() == curl::fuzzer::proto::MULTI_STATE_PROBE_PROGRESS_RATE_LIMIT) {

    progress_sink.reset(std::tmpfile());

  }

  // The peer and socket callback state must outlive multi cleanup, which can

  // synchronously announce socket removals and close cached connections.

  MockServer mock;

  mock.SetScripts(scenario);

  mock.SetKeepConnectionsOpen(plan.keep_connections_open());

  MultiSocketDriver socket_driver;

  CurlMultiPtr multi(curl_multi_init());

  if (!multi) {

    return stats;

  }

  const long max_host_connections =

      static_cast<long>(std::min<std::size_t>(plan.max_host_connections(), transfer_count));

  const long max_total_connections =

      static_cast<long>(std::min<std::size_t>(plan.max_total_connections(), transfer_count));

  const long cache_size =

      static_cast<long>(std::min<std::size_t>(plan.connection_cache_size(), scenario_limits::kMaxMultiTransfers * 2));

  (void)curl_multi_setopt(multi.get(), CURLMOPT_MAX_HOST_CONNECTIONS, max_host_connections);

  (void)curl_multi_setopt(multi.get(), CURLMOPT_MAX_TOTAL_CONNECTIONS, max_total_connections);

  (void)curl_multi_setopt(multi.get(), CURLMOPT_MAXCONNECTS, cache_size);

  (void)curl_multi_setopt(multi.get(), CURLMOPT_PIPELINING,

                          plan.multiplex() ? static_cast<long>(CURLPIPE_MULTIPLEX) : 0L);

  (void)curl_multi_setopt(multi.get(), CURLMOPT_NOTIFYFUNCTION, &RecordNotification);

  (void)curl_multi_setopt(multi.get(), CURLMOPT_NOTIFYDATA, &stats);

  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_INFO_READ);

  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_EASY_DONE);

  (void)curl_multi_notify_disable(multi.get(), CURLMNOTIFY_INFO_READ);

  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_INFO_READ);

  (void)curl_multi_notify_disable(multi.get(), CURLMNOTIFY_EASY_DONE);

  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_EASY_DONE);

  const bool socket_mode = plan.drive_mode() == curl::fuzzer::proto::MULTI_DRIVE_SOCKET;

  const bool socket_driver_installed = socket_mode && socket_driver.Install(multi.get());

  std::array<TransferState, scenario_limits::kMaxMultiTransfers> transfers;

  const std::string url = "http://" + scenario.host_path();

  for (std::size_t index = 0; index < transfer_count; ++index) {

    TransferState& transfer = transfers[index];

    if (!transfer.session.Initialize()) {

      continue;

    }

    transfer.session.ApplyBaseline(scenario);

    (void)curl_easy_setopt(transfer.session.easy(), CURLOPT_URL, url.c_str());

    mock.Install(transfer.session.easy());

    (void)ApplyScenarioOptions(transfer.session.easy(), scenario);

    if (ApplyStateProbeOptions(transfer.session.easy(), plan.state_probe_profile(), progress_sink.get())) {

      ++stats.state_probe_handles;

    }

    mock.ConfigureRequestData(transfer.session.InstallRequestData(scenario));

    if (curl_multi_add_handle(multi.get(), transfer.session.easy()) == CURLM_OK) {

      transfer.attached = true;

      ++stats.added_handles;

    }

  }

  if (stats.added_handles == 0) {

    multi.reset();

    return stats;

  }

  ProbeLegacyMultiSocketApis(multi.get());

  const std::size_t action_count = std::min<std::size_t>(plan.actions_size(), scenario_limits::kMaxMultiActions);

  std::size_t next_action = 0;

  bool initial_action_changed = false;

  if (next_action < action_count) {

    initial_action_changed =

        ApplyAction(plan.actions(static_cast<int>(next_action++)), multi.get(), &transfers, transfer_count);

    ++stats.actions_consumed;

  }

  int running_handles = 0;

  CURLMcode result = socket_driver_installed ? socket_driver.Start(&running_handles)

                                             : curl_multi_perform(multi.get(), &running_handles);

  bool completion_progress = DrainCompletionMessages(multi.get(), &transfers, transfer_count, &stats);

  ProbeMultiWaitApis(multi.get());

  if (plan.wake_multi()) {

    if (socket_driver_installed) {

      socket_driver.ProbeControlApis();

    } else {

      (void)curl_multi_wakeup(multi.get());

    }

  }

  int idle_iterations = initial_action_changed || completion_progress ? 0 : 1;

  for (int iteration = 0; result == CURLM_OK && iteration < kMaxDriveIterations; ++iteration) {

    if (running_handles == 0 && next_action >= action_count) {

      break;

    }

    bool made_progress = mock.ServiceConnections();

    if (next_action < action_count) {

      made_progress =

          ApplyAction(plan.actions(static_cast<int>(next_action++)), multi.get(), &transfers, transfer_count) ||

          made_progress;

      ++stats.actions_consumed;

      // Consuming an ordered no-op is still harness progress: continue to the

      // next action instead of letting an idle transport hide its suffix.

      made_progress = true;

    }

    const int running_before = running_handles;

    if (socket_driver_installed) {

      const MultiSocketDriver::DriveResult drive = socket_driver.DriveReady(&running_handles);

      result = drive.code;

      made_progress = drive.made_progress || made_progress;

    } else {

      result = curl_multi_perform(multi.get(), &running_handles);

      made_progress = running_handles != running_before || made_progress;

    }

    made_progress = DrainCompletionMessages(multi.get(), &transfers, transfer_count, &stats) || made_progress;

    if (made_progress) {

      idle_iterations = 0;

    } else if (++idle_iterations >= kMaxIdleIterations) {

      break;

    }

  }

  (void)DrainCompletionMessages(multi.get(), &transfers, transfer_count, &stats);

  stats.opened_connections = mock.opened_connection_count();

  for (std::size_t index = 0; index < transfer_count; ++index) {

    TransferState& transfer = transfers[index];

    if (transfer.session.easy() == nullptr) {

      continue;

    }

    (void)curl_easy_pause(transfer.session.easy(), CURLPAUSE_CONT);

    if (transfer.attached) {

      (void)curl_multi_remove_handle(multi.get(), transfer.session.easy());

      transfer.attached = false;

    }

  }

  // Destroy the multi while both its socket callback storage and mock peer are

  // live, then detach request-data pointers before cleaning each easy handle.

  multi.reset();

  for (std::size_t index = 0; index < transfer_count; ++index) {

    transfers[index].session.Close();

  }

  return stats;

}

}  // namespace proto_fuzzer