src/queueing_test.cc

// Copyright 2022 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <limits>
#include <string>

#include "ipcz/ipcz.h"
#include "test/multinode_test.h"
#include "testing/gtest/include/gtest/gtest.h"
#include "third_party/abseil-cpp/absl/synchronization/notification.h"

namespace ipcz {
namespace {

using QueueingTestNode = test::TestNode;
using QueueingTest = test::MultinodeTest<QueueingTestNode>;

MULTINODE_TEST_NODE(QueueingTestNode, RemoteQueueFeedbackClient) {
  IpczHandle b = ConnectToBroker();

  // Wait for the first parcel to arrive.
  EXPECT_EQ(IPCZ_RESULT_OK,
            WaitForConditions(b, {.flags = IPCZ_TRAP_ABOVE_MIN_LOCAL_PARCELS,
                                  .min_local_parcels = 0}));

  // Send and ack and wait for another parcel to arrive.
  absl::Notification new_parcel_arrived;
  EXPECT_EQ(IPCZ_RESULT_OK,
            Trap(b, {.flags = IPCZ_TRAP_NEW_LOCAL_PARCEL},
                 [&](const IpczTrapEvent&) { new_parcel_arrived.Notify(); }));
  EXPECT_EQ(IPCZ_RESULT_OK, Put(b, "ok"));
  new_parcel_arrived.WaitForNotification();

  std::string data;
  EXPECT_EQ(IPCZ_RESULT_OK, Get(b, &data));
  EXPECT_EQ("1234", data);

  std::string ack;
  EXPECT_EQ(IPCZ_RESULT_OK, WaitToGet(b, &ack));
  EXPECT_EQ("ok", ack);

  EXPECT_EQ(IPCZ_RESULT_OK, WaitForConditionFlags(b, IPCZ_TRAP_PEER_CLOSED));
  Close(b);
}

MULTINODE_TEST(QueueingTest, RemoteQueueFeedback) {
  // Exercises operations which rely on feedback from the remote peer regarding
  // its inbound parcel queue state.
  IpczHandle c = SpawnTestNode<RemoteQueueFeedbackClient>();

  // This trap can only be set while the remote portal appears to be non-empty.
  const IpczTrapConditions all_bytes_consumed = {
      .size = sizeof(all_bytes_consumed),
      .flags = IPCZ_TRAP_BELOW_MAX_REMOTE_BYTES,
      .max_remote_bytes = 1,
  };
  EXPECT_EQ(IPCZ_RESULT_FAILED_PRECONDITION,
            Trap(c, all_bytes_consumed, [&](const auto&) {}));

  // Send 4 bytes and wait for acknowledgement that the parcel was received.
  std::string ack;
  EXPECT_EQ(IPCZ_RESULT_OK, Put(c, "1234"));
  EXPECT_EQ(IPCZ_RESULT_OK, WaitToGet(c, &ack));
  EXPECT_EQ("ok", ack);

  // Now these operations should always fail due to the specified limits.
  EXPECT_EQ(IPCZ_RESULT_RESOURCE_EXHAUSTED,
            PutWithLimits(c, {.max_queued_parcels = 1}, "meh"));
  EXPECT_EQ(IPCZ_RESULT_RESOURCE_EXHAUSTED,
            PutWithLimits(c, {.max_queued_bytes = 4}, "?"));

  // Now we should be able to install traps for both queued parcels and bytes on
  // the remote side.
  absl::Notification consumed_parcels;
  const IpczTrapConditions all_parcels_consumed = {
      .size = sizeof(all_parcels_consumed),
      .flags = IPCZ_TRAP_BELOW_MAX_REMOTE_PARCELS,
      .max_remote_parcels = 1,
  };
  EXPECT_EQ(
      IPCZ_RESULT_OK,
      Trap(c, all_parcels_consumed, [&](const IpczTrapEvent& event) {
        EXPECT_TRUE(event.condition_flags & IPCZ_TRAP_BELOW_MAX_REMOTE_PARCELS);
        consumed_parcels.Notify();
      }));

  absl::Notification consumed_bytes;
  EXPECT_EQ(
      IPCZ_RESULT_OK,
      Trap(c, all_bytes_consumed, [&](const IpczTrapEvent& event) {
        EXPECT_TRUE(event.condition_flags & IPCZ_TRAP_BELOW_MAX_REMOTE_BYTES);
        consumed_bytes.Notify();
      }));

  // Ack back to the client so it will read its queue. Then we can wait for both
  // traps to notify.
  EXPECT_EQ(IPCZ_RESULT_OK, Put(c, "ok"));
  consumed_parcels.WaitForNotification();
  consumed_bytes.WaitForNotification();

  // And now this Put operation should succeed.
  EXPECT_EQ(IPCZ_RESULT_OK,
            PutWithLimits(c, {.max_queued_parcels = 1, .max_queued_bytes = 4},
                          "meh!"));

  Close(c);
}

MULTINODE_TEST_NODE(QueueingTestNode, TwoPhaseQueueingClient) {
  IpczHandle b = ConnectToBroker();
  WaitForDirectRemoteLink(b);
  EXPECT_EQ(IPCZ_RESULT_OK, Put(b, "go"));

  EXPECT_EQ(IPCZ_RESULT_OK,
            WaitForConditions(b, {.flags = IPCZ_TRAP_ABOVE_MIN_LOCAL_PARCELS,
                                  .min_local_parcels = 0}));
  size_t num_bytes;
  const void* data;
  EXPECT_EQ(IPCZ_RESULT_OK, ipcz().BeginGet(b, IPCZ_NO_FLAGS, nullptr, &data,
                                            &num_bytes, nullptr));

  // The producer should only have been able to put 3 out of its 4 bytes.
  EXPECT_EQ("ipc",
            std::string_view(reinterpret_cast<const char*>(data), num_bytes));
  EXPECT_EQ(IPCZ_RESULT_OK, ipcz().EndGet(b, num_bytes, 0, IPCZ_NO_FLAGS,
                                          nullptr, nullptr, nullptr));

  Close(b);
}

MULTINODE_TEST(QueueingTest, TwoPhaseQueueing) {
  IpczHandle c = SpawnTestNode<TwoPhaseQueueingClient>();
  WaitForDirectRemoteLink(c);

  std::string message;
  EXPECT_EQ(IPCZ_RESULT_OK, WaitToGet(c, &message));
  EXPECT_EQ("go", message);

  const IpczPutLimits limits = {
      .size = sizeof(limits),
      .max_queued_parcels = 1,
      .max_queued_bytes = 3,
  };

  size_t num_bytes = 4;
  void* data;
  const IpczBeginPutOptions options = {.size = sizeof(options),
                                       .limits = &limits};
  EXPECT_EQ(IPCZ_RESULT_OK, ipcz().BeginPut(c, IPCZ_BEGIN_PUT_ALLOW_PARTIAL,
                                            &options, &num_bytes, &data));

  // There should not be enough space for all 4 bytes.
  EXPECT_EQ(3u, num_bytes);
  memcpy(data, "ipc", 3);
  EXPECT_EQ(IPCZ_RESULT_OK,
            ipcz().EndPut(c, num_bytes, nullptr, 0, IPCZ_NO_FLAGS, nullptr));

  EXPECT_EQ(IPCZ_RESULT_OK, WaitForConditionFlags(c, IPCZ_TRAP_PEER_CLOSED));
  Close(c);
}

MULTINODE_TEST_NODE(QueueingTestNode, TwoPhaseFeedbackClient) {
  IpczHandle b = ConnectToBroker();
  WaitForDirectRemoteLink(b);

  EXPECT_EQ(IPCZ_RESULT_OK,
            WaitForConditions(b, {.flags = IPCZ_TRAP_ABOVE_MIN_LOCAL_PARCELS,
                                  .min_local_parcels = 0}));
  size_t num_bytes;
  const void* data;
  EXPECT_EQ(IPCZ_RESULT_OK, ipcz().BeginGet(b, IPCZ_NO_FLAGS, nullptr, &data,
                                            &num_bytes, nullptr));

  EXPECT_EQ("hello?",
            std::string_view(reinterpret_cast<const char*>(data), num_bytes));
  EXPECT_EQ(IPCZ_RESULT_OK, ipcz().EndGet(b, num_bytes, 0, IPCZ_NO_FLAGS,
                                          nullptr, nullptr, nullptr));
  Close(b);
}

// TODO(https://crbug.com/1361670): Fix flakiness and re-enable this test.
MULTINODE_TEST(QueueingTest, DISABLED_TwoPhaseFeedback) {
  IpczHandle c = SpawnTestNode<TwoPhaseFeedbackClient>();
  WaitForDirectRemoteLink(c);
  EXPECT_EQ(IPCZ_RESULT_OK, Put(c, "hello?"));
  EXPECT_EQ(IPCZ_RESULT_OK,
            WaitForConditions(c, {.flags = IPCZ_TRAP_BELOW_MAX_REMOTE_PARCELS,
                                  .max_remote_parcels = 1}));
  Close(c);
}

constexpr size_t kStressTestPortalCapacity = 256;
constexpr size_t kStressTestPayloadSize = 4 * 1024 * 1024;

MULTINODE_TEST_NODE(QueueingTestNode, RemoteQueueFeedbackStressTestClient) {
  IpczHandle b = ConnectToBroker();

  size_t bytes_received = 0;
  while (bytes_received < kStressTestPayloadSize) {
    // Consistency check: ensure that the portal never has more than
    // kStressTestPortalCapacity bytes available to retrieve. Otherwise limits
    // were not properly enforced by the sender.
    IpczPortalStatus status = {.size = sizeof(status)};
    EXPECT_EQ(IPCZ_RESULT_OK,
              ipcz().QueryPortalStatus(b, IPCZ_NO_FLAGS, nullptr, &status));
    EXPECT_LE(status.num_local_bytes, kStressTestPortalCapacity);

    const void* data;
    size_t num_bytes;
    const IpczResult begin_result =
        ipcz().BeginGet(b, IPCZ_NO_FLAGS, nullptr, &data, &num_bytes, nullptr);
    if (begin_result == IPCZ_RESULT_OK) {
      bytes_received += num_bytes;
      EXPECT_EQ(std::string_view(static_cast<const char*>(data), num_bytes),
                std::string(num_bytes, '!'));
      EXPECT_EQ(IPCZ_RESULT_OK, ipcz().EndGet(b, num_bytes, 0, IPCZ_NO_FLAGS,
                                              nullptr, nullptr, nullptr));
      continue;
    }

    ASSERT_EQ(IPCZ_RESULT_UNAVAILABLE, begin_result);
    WaitForConditions(
        b, {.flags = IPCZ_TRAP_ABOVE_MIN_LOCAL_BYTES, .min_local_bytes = 0});
  }

  Close(b);
}

MULTINODE_TEST(QueueingTest, RemoteQueueFeedbackStressTest) {
  IpczHandle c = SpawnTestNode<RemoteQueueFeedbackStressTestClient>();

  size_t bytes_remaining = kStressTestPayloadSize;
  while (bytes_remaining) {
    void* data;
    size_t capacity = bytes_remaining;
    const IpczPutLimits limits = {
        .size = sizeof(limits),
        .max_queued_parcels = std::numeric_limits<size_t>::max(),
        .max_queued_bytes = kStressTestPortalCapacity,
    };
    const IpczBeginPutOptions options = {
        .size = sizeof(options),
        .limits = &limits,
    };
    const IpczResult begin_result = ipcz().BeginPut(
        c, IPCZ_BEGIN_PUT_ALLOW_PARTIAL, &options, &capacity, &data);
    if (begin_result == IPCZ_RESULT_OK) {
      size_t num_bytes = std::min(bytes_remaining, capacity);
      bytes_remaining -= num_bytes;
      memset(data, '!', num_bytes);
      EXPECT_EQ(IPCZ_RESULT_OK, ipcz().EndPut(c, num_bytes, nullptr, 0,
                                              IPCZ_NO_FLAGS, nullptr));
      continue;
    }
    ASSERT_EQ(IPCZ_RESULT_RESOURCE_EXHAUSTED, begin_result);

    EXPECT_EQ(
        IPCZ_RESULT_OK,
        WaitForConditions(c, {.flags = IPCZ_TRAP_BELOW_MAX_REMOTE_BYTES,
                              .max_remote_bytes = kStressTestPortalCapacity}));
  }

  WaitForConditionFlags(c, IPCZ_TRAP_PEER_CLOSED);
  Close(c);
}

}  // namespace
}  // namespace ipcz