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"

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

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);
}

// 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::CreateAsyncPipe(&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::CreateAsyncPipe(&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));
}

TEST(AsyncLoop, AsyncHandleStartAccept) {
  std::string err_msg;
  auto server_handle = IpcServiceHandle::BindTo("test_service", &err_msg);
  if (!server_handle)
    fprintf(stderr, "BindTo() error: %s\n", err_msg.c_str());
  ASSERT_TRUE(server_handle);

  AsyncLoop& loop = GetNewLoop();

  AsyncResult result;
  auto async_handle =
      AsyncHandle::Create(std::move(server_handle), loop, result.callback());
  async_handle.StartAccept();

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

  // Synchronous connect.
  IpcHandle client = IpcServiceHandle::ConnectTo("test_service", &err_msg);
  if (!client)
    fprintf(stderr, "ConnectTo() error: %s\n", err_msg.c_str());
  ASSERT_TRUE(client);

  status = loop.RunOnce(50);
  EXPECT_EQ(AsyncLoop::ExitSuccess, status);
  EXPECT_TRUE(result.completed);
  EXPECT_EQ(0, result.error);

  IpcHandle accept_handle = async_handle.TakeAcceptedHandle();
  EXPECT_TRUE(accept_handle);
  EXPECT_NE(accept_handle.native_handle(), client.native_handle());
}

TEST(AsyncLoop, AsyncHandleStartConnect) {
  std::string err_msg;
  auto server_handle = IpcServiceHandle::BindTo("test_service", &err_msg);
  if (!server_handle)
    fprintf(stderr, "BindTo() error: %s\n", err_msg.c_str());
  ASSERT_TRUE(server_handle);

  bool did_connect = false;
  IpcHandle client_handle =
      IpcServiceHandle::AsyncConnectTo("test_service", &did_connect, &err_msg);
  if (!client_handle)
    fprintf(stderr, "AsyncClientTo() error: %s\n", err_msg.c_str());
  ASSERT_TRUE(client_handle);

  (void)did_connect;  // ignored intentionally.

  AsyncLoop& loop = GetNewLoop();

  AsyncResult result;
  auto async_handle =
      AsyncHandle::Create(std::move(client_handle), loop, result.callback());
  async_handle.StartConnect();

  auto status = loop.RunOnce(50);
  EXPECT_EQ(AsyncLoop::ExitSuccess, status);
  EXPECT_TRUE(result.completed);
  EXPECT_EQ(0, result.error);
  EXPECT_EQ(0, result.size);
}

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

  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::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::ExitSuccess, status);
}

TEST(AsyncLoop, TimerTest) {
  AsyncLoop& loop = GetNewLoop();
  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);

  EXPECT_EQ(AsyncLoop::ExitSuccess, loop.RunOnce(200));
  EXPECT_EQ(1, counter);
}

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);
}