src/async_loop_test.cc

// Copyright 2023 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "async_loop.h"

// This file also includes tests for the AsyncLoopTimers class.
#include "async_loop_timers.h"
#include "ipc_handle.h"
#include "test.h"

namespace {

AsyncLoop& GetNewLoop() {
  AsyncLoop::ResetForTesting();
  return AsyncLoop::Get();
}

}  // namespace

TEST(AsyncLoop, NowMs) {
  AsyncLoop& loop = GetNewLoop();
  // Ensure the result of NowMs() is always positive, otherwise many
  // things will not work correctly since a negative expiration date is
  // interpreted as infinite.
  EXPECT_GE(loop.NowMs(), 0LL);
}

TEST(AsyncLoop, ScopedTestClock) {
  AsyncLoop& loop = GetNewLoop();
  // Ensure the result of NowMs() is always positive, otherwise many
  // things will not work correctly since a negative expiration date is
  // interpreted as infinite.
  int64_t start_time_ms = loop.NowMs();
  EXPECT_GE(start_time_ms, 0LL);

  {
    AsyncLoop::ScopedTestClock test_clock(loop);
    EXPECT_EQ(0LL, loop.NowMs());
    test_clock.AdvanceTimeMillis(120);
    EXPECT_EQ(120LL, loop.NowMs());
    test_clock.AdvanceTimeMillis(120);
    EXPECT_EQ(240LL, loop.NowMs());
    test_clock.AdvanceTimeMillis(10000);
    EXPECT_EQ(10240LL, loop.NowMs());
  }

  int64_t end_time_ms = loop.NowMs();
  ASSERT_GE(end_time_ms, start_time_ms);
  ASSERT_LT(end_time_ms - start_time_ms, 10240LL);
}

// Helper type to store async operation results.
struct AsyncResult {
  bool completed = false;
  AsyncError error = 0;
  size_t size = 0;

  AsyncHandle::Callback callback() {
    return [this](AsyncError error, size_t size) {
      this->completed = true;
      this->error = error;
      this->size = size;
    };
  }
};

TEST(AsyncLoop, AsyncHandleStartRead) {
  std::string err_msg;
  IpcHandle read_handle, write_handle;
  bool ret = IpcHandle::CreatePipe(&read_handle, &write_handle, &err_msg);
  if (!ret)
    fprintf(stderr, "CreatePipe() error: %s\n", err_msg.c_str());
  ASSERT_TRUE(ret);

  AsyncLoop& loop = GetNewLoop();
  char read_buffer[10] = {};

  AsyncResult result;
  auto async_handle =
      AsyncHandle::Create(std::move(read_handle), loop, result.callback());
  async_handle.StartRead(read_buffer, sizeof(read_buffer));

  // Nothing should be accepted in the next 50ms
  auto status = loop.RunOnce(50);
  EXPECT_EQ(AsyncLoop::ExitTimeout, status);
  EXPECT_FALSE(result.completed);

  // Write to the pipe.
  char buf[1] = { 'x' };
  EXPECT_EQ(1, write_handle.Write(buf, sizeof(buf), &err_msg));

  status = loop.RunOnce(50);
  EXPECT_EQ(AsyncLoop::ExitSuccess, status);
  EXPECT_TRUE(result.completed);
  EXPECT_EQ(0u, result.error);
  EXPECT_EQ(1u, result.size);
  EXPECT_EQ('x', read_buffer[0]);
}

TEST(AsyncLoop, AsyncHandleStartWrite) {
  std::string err_msg;
  IpcHandle read_handle, write_handle;
  bool ret = IpcHandle::CreatePipe(&read_handle, &write_handle, &err_msg);
  if (!ret)
    fprintf(stderr, "CreatePipe() error: %s\n", err_msg.c_str());
  ASSERT_TRUE(ret);

  AsyncLoop& loop = GetNewLoop();
  const char buffer[] = "foo bar";

  AsyncResult result;
  auto async_handle =
      AsyncHandle::Create(std::move(write_handle), loop, result.callback());
  async_handle.StartWrite(buffer, sizeof(buffer));

  // Writing should work directly.
  auto status = loop.RunOnce(50);
  EXPECT_EQ(AsyncLoop::ExitSuccess, status);

  EXPECT_TRUE(result.completed);
  EXPECT_EQ(0u, result.error);
  EXPECT_EQ(sizeof(buffer), result.size);

  // Read from pipe synchronously.
  char read_buffer[sizeof(buffer)];
  ssize_t actual_bytes =
      read_handle.Read(read_buffer, sizeof(buffer), &err_msg);
  EXPECT_EQ(sizeof(buffer), static_cast<size_t>(actual_bytes));
}

#ifdef _WIN32
TEST(AsyncLoop, AsyncHandleStartConnectNamedPipe) {
  AsyncLoop& loop = GetNewLoop();

  std::wstring pipe_path =
      IpcHandle::CreateUniqueNamedPipePath(L"AsyncLoopTest");
  IpcHandle pipe(IpcHandle::CreateAsyncNamedPipeInstance(pipe_path, true, 1));
  ASSERT_TRUE(pipe) << "CreateNamedPipeW(): " << GetLastErrorString();

  struct TestPeer {
    AsyncError error = 0;
    bool completed = false;
  };

  TestPeer peer;

  AsyncHandle handle = AsyncHandle::Create(std::move(pipe), loop,
                                           [&peer](AsyncError error, size_t) {
                                             peer.error = error;
                                             peer.completed = true;
                                           });
  ASSERT_TRUE(handle);

  for (size_t trial = 1; trial <= 3; ++trial) {
    peer = TestPeer();

    std::string context = "Try #" + std::to_string(trial);

    if (trial > 1)
      DisconnectNamedPipe(handle.native_handle());

    handle.StartConnectNamedPipe();
    EXPECT_FALSE(peer.completed) << context;
    EXPECT_TRUE(handle.IsRunning()) << context;

    // Nothing should be accepted in the next 50ms
    auto status = loop.RunOnce(50);
    EXPECT_EQ(AsyncLoop::ExitTimeout, status) << context;
    EXPECT_FALSE(peer.completed) << context;

    // Synchronous connect.
    IpcHandle client(IpcHandle::ClientConnectToNamedPipe(pipe_path));
    ASSERT_TRUE(client) << context
                        << " CreateFileW(): " << GetLastErrorString();

    status = loop.RunOnce(50);
    EXPECT_EQ(AsyncLoop::ExitSuccess, status) << context;
    EXPECT_TRUE(peer.completed) << context;
    EXPECT_EQ(0, peer.error) << context;
  }
}
#endif  // _WIN32

TEST(AsyncLoop, RunUntil) {
  AsyncLoop& loop = GetNewLoop();
  AsyncLoop::ScopedTestClock test_clock(loop);

  auto always_false = []() { return false; };

  auto status = loop.RunUntil(always_false, -1);
  EXPECT_EQ(AsyncLoop::ExitIdle, status);

  status = loop.RunUntil(always_false, 10);
  EXPECT_EQ(AsyncLoop::ExitTimeout, status);

  bool flag = false;
  auto flag_is_set = [&flag]() { return flag; };
  AsyncTimer timer(loop, [&flag]() { flag = true; });

  timer.SetDurationMs(100LL);
  status = loop.RunUntil(flag_is_set, -1);
  EXPECT_EQ(AsyncLoop::ExitTimeout, status);

  test_clock.AdvanceTimeMillis(200);
  status = loop.RunUntil(flag_is_set, -1);
  EXPECT_EQ(AsyncLoop::ExitSuccess, status);

  flag = false;
  timer.SetDurationMs(1000LL);
  status = loop.RunUntil(flag_is_set, 10LL);
  EXPECT_EQ(AsyncLoop::ExitTimeout, status);

  status = loop.RunUntil(flag_is_set, -1);
  EXPECT_EQ(AsyncLoop::ExitTimeout, status);

  test_clock.AdvanceTimeMillis(1000);
  status = loop.RunUntil(flag_is_set, -1);
  EXPECT_EQ(AsyncLoop::ExitSuccess, status);
}

TEST(AsyncLoop, TimerTest) {
  AsyncLoop& loop = GetNewLoop();
  AsyncLoop::ScopedTestClock test_clock(loop);

  int counter = 0;
  AsyncTimer timer_1(loop, [&counter]() { counter += 1; });
  timer_1.SetDurationMs(100LL);

  EXPECT_EQ(AsyncLoop::ExitTimeout, loop.RunOnce(0));
  EXPECT_EQ(0, counter);

  test_clock.AdvanceTimeMillis(99);
  EXPECT_EQ(AsyncLoop::ExitTimeout, loop.RunOnce(0));
  EXPECT_EQ(0, counter);

  test_clock.AdvanceTimeMillis(1);
  EXPECT_EQ(AsyncLoop::ExitSuccess, loop.RunOnce(0));

  timer_1.SetDurationMs(100LL);
  EXPECT_EQ(AsyncLoop::ExitTimeout, loop.RunOnce(0));
  EXPECT_EQ(1, counter);

  test_clock.AdvanceTimeMillis(100);
  EXPECT_EQ(AsyncLoop::ExitSuccess, loop.RunOnce(300));
  EXPECT_EQ(2, counter);

  EXPECT_EQ(200LL, loop.NowMs());
}

TEST(AsyncLoopTimers, Test) {
  AsyncLoopTimers timers;

  EXPECT_EQ(-1LL, timers.ComputeNextExpiration());

  int counter = 0;

  // Create local loop, required by AsyncTimer::State construtor
  // but not used by the test though because its RunOnce() method is never
  // called.
  auto loop = AsyncLoop::CreateLocal();

  auto timer_1 = std::unique_ptr<AsyncTimer::State>(
      new AsyncTimer::State(*loop, [&counter]() { counter += 1; }));

  auto timer_2 = std::unique_ptr<AsyncTimer::State>(
      new AsyncTimer::State(*loop, [&counter]() { counter += 100; }));

  timers.AttachTimer(timer_1.get());
  timers.AttachTimer(timer_2.get());

  EXPECT_EQ(-1LL, timers.ComputeNextExpiration());

  timer_1->SetExpirationMs(100LL);

  EXPECT_EQ(100LL, timers.ComputeNextExpiration());
  EXPECT_EQ(0, counter);

  timer_2->SetExpirationMs(200LL);
  EXPECT_EQ(100LL, timers.ComputeNextExpiration());
  EXPECT_EQ(0, counter);

  EXPECT_FALSE(timers.ProcessExpiration(99LL));
  EXPECT_EQ(100LL, timers.ComputeNextExpiration());
  EXPECT_EQ(0, counter);

  EXPECT_TRUE(timers.ProcessExpiration(100LL));
  EXPECT_EQ(1, counter);
  EXPECT_EQ(200LL, timers.ComputeNextExpiration());

  EXPECT_TRUE(timers.ProcessExpiration(200LL));
  EXPECT_EQ(101, counter);
  EXPECT_EQ(-1LL, timers.ComputeNextExpiration());

  timer_1->SetExpirationMs(200LL);
  timers.DetachTimer(timer_1.get());
  timers.DetachTimer(timer_2.get());

  EXPECT_EQ(-1LL, timers.ComputeNextExpiration());

  EXPECT_FALSE(timers.ProcessExpiration(300LL));
  EXPECT_EQ(101, counter);
}