src/ipc_service_test.cc

// Copyright 2024 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 "ipc_service.h"

#include <stdio.h>
#include <string.h>

#include "ipc_utils.h"
#include "test.h"

#ifndef _WIN32
#include <pthread.h>
#endif

#ifdef _WIN32
#include <windows.h>
#else
#include <fcntl.h>
#include <unistd.h>
#endif

namespace {

// Alternatives to ASSERT_{STREQ,STRNE}("", str) that
// works correctly if the str argument contains an initial \0.
#define ASSERT_STRING_EMPTY(str) ASSERT_EQ(std::string(), str)
#define ASSERT_STRING_NOT_EMPTY(str) ASSERT_NE(std::string(), str)

#define IPC_ERROR_CONTEXT(prefix) \
  prefix " did_timeout=" << did_timeout << ", error: " << error_message

ScopedHandle::NativeType CreateTestNativeHandle() {
#ifdef _WIN32
  return CreateFileW(L"NUL", GENERIC_READ | GENERIC_WRITE, 0, nullptr,
                     OPEN_EXISTING, 0, NULL);
#else
  return open("/dev/null", O_RDWR);
#endif
}

ScopedHandle CreateTestHandle() {
  return { CreateTestNativeHandle() };
}

// Create a pipe for testing. On success return true and
// set |*read| and |*write| to be the read and write ends
// of the pipe. Note that the pipe is unidirectional due
// to portability with Windows.
bool CreateTestPipe(ScopedHandle* read, ScopedHandle* write) {
  std::string error;
  if (!ScopedHandle::CreatePipe(read, write, &error)) {
    fprintf(stderr, "ERROR: when creating pipe: %s\n", error.c_str());
    return false;
  }
  return true;
}

// A background thread that will try to connect to a given IpcService
// instance for a given timeout, then report the result. Usage is:
//
//  - Create instance, passing service name and timeout in milliseconds.
//  - Call Start() to start the thread in the background.
//  - Call Join() to wait for its completion.
//  - Access |handle|, |did_timeout| and |error_message| members directly
//    to get the result of the operation.
//
// This is necessary because on Windows, a ConnectTo() will always timeout
// or fail if there was no prior AcceptPeer() from another thread. This
// does not happen on Posix though.
struct TestIpcServiceConnectingThread {
  TestIpcServiceConnectingThread(StringPiece service_name, int64_t timeout_ms)
      : service_name_(service_name.AsString()), timeout_ms_(timeout_ms) {}

  void Start() {
#ifdef _WIN32
    thread_ = CreateThread(nullptr, 65536, MainRoutine, this, 0, nullptr);
    ASSERT_NE(thread_, INVALID_HANDLE_VALUE) << GetLastErrorString();
#else   // !_WIN32
    int ret = pthread_create(&thread_, nullptr, MainRoutine, this);
    ASSERT_EQ(ret, 0) << strerror(ret);
#endif  // !_WIN32
  }

  void Join() {
#ifdef _WIN32
    (void)WaitForSingleObject(thread_, INFINITE);
#else   // !_WIN32
    void* ignored = nullptr;
    pthread_join(thread_, &ignored);
#endif  // !_WIN32
  }

 private:
  const std::string service_name_;
  const int64_t timeout_ms_;
#ifdef _WIN32
  HANDLE thread_ = INVALID_HANDLE_VALUE;

  static DWORD WINAPI MainRoutine(void* arg) {
    reinterpret_cast<TestIpcServiceConnectingThread*>(arg)->RunInBackground();
    return 0;
  }
#else   // !_WIN32
  pthread_t thread_;

  static void* MainRoutine(void* arg) {
    reinterpret_cast<TestIpcServiceConnectingThread*>(arg)->RunInBackground();
    return nullptr;
  }
#endif  // !_WIN32

  void RunInBackground() {
    handle = IpcService::ConnectTo(service_name_, timeout_ms_, &did_timeout,
                                   &error_message);
  }

 public:
  ScopedHandle handle;
  bool did_timeout = false;
  std::string error_message;
};

}  // namespace

TEST(IpcService, ServerHandleIsNotInheritable) {
  std::string service_name = "test_service";
  std::string error_message;

  ASSERT_FALSE(IpcService::HasServer(service_name));

  // Start server
  error_message.clear();
  IpcService service = IpcService::StartServer(service_name, &error_message);
  ASSERT_TRUE(service) << "StartServer(): " << error_message;
  ASSERT_FALSE(service.GetServerHandleForTest().IsInheritable());
  ASSERT_STRING_EMPTY(error_message);
  ASSERT_TRUE(service.IsValid());
}

TEST(IpcService, StartConnectAndAccept) {
  std::string service_name = "test_service";
  std::string error_message;

  // Tring to connect to a service without a server should fail.
  bool did_timeout = false;
  ScopedHandle no_client =
      IpcService::ConnectTo(service_name, -1, &did_timeout, &error_message);
  ASSERT_FALSE(no_client) << IPC_ERROR_CONTEXT("ConnectTo");
  ASSERT_FALSE(did_timeout);
  ASSERT_STRING_NOT_EMPTY(error_message);
  error_message.clear();

  ASSERT_FALSE(IpcService::HasServer(service_name));

  // Start server
  error_message.clear();
  IpcService service = IpcService::StartServer(service_name, &error_message);
  ASSERT_TRUE(service) << "StartServer(): " << error_message;
  ASSERT_STRING_EMPTY(error_message);
  ASSERT_TRUE(service.IsValid());

  ASSERT_TRUE(IpcService::HasServer(service_name));

  // Trying to create a secondary server for the same service should fail.
  IpcService no_service = IpcService::StartServer(service_name, &error_message);
  ASSERT_FALSE(no_service);
  ASSERT_STRING_NOT_EMPTY(error_message);
  error_message.clear();

  ASSERT_TRUE(IpcService::HasServer(service_name));

  // Start a connecting thread in the background with a large timeout.
  TestIpcServiceConnectingThread client(service_name, 1000);
  client.Start();

  // Accept new client connection on the server side, this should succeed
  // immediately, except if the background thread has not been scheduled
  // yet, so use a minimal timeout of 1 millisecond to ensure this happens.
  did_timeout = false;
  ScopedHandle peer = service.AcceptPeer(1, &did_timeout, &error_message);
  ASSERT_TRUE(peer) << IPC_ERROR_CONTEXT("AcceptPeer");
  ASSERT_FALSE(did_timeout);
  ASSERT_STRING_EMPTY(error_message);

  // Connect to the server, and receive a client handle.
  // Note that it will not be possible to write to it until the server
  // has called AcceptPeer() on its service instance.
  client.Join();
  ASSERT_TRUE(client.handle) << "client did_timeout=" << client.did_timeout
                             << ", error: " << client.error_message;
  ASSERT_FALSE(client.did_timeout);
  ASSERT_STRING_EMPTY(client.error_message);

  // Send data from the client to the peer.
  const char kInput[] = "sending data";
  const size_t kInputSize = sizeof(kInput);
  ASSERT_TRUE(client.handle.WriteFull(kInput, kInputSize, &error_message));
  ASSERT_STRING_EMPTY(error_message);

  char output[kInputSize] = {};
  ASSERT_TRUE(peer.ReadFull(output, sizeof(output), &error_message));
  ASSERT_STRING_EMPTY(error_message);

  ASSERT_TRUE(!memcmp(output, kInput, kInputSize));

  // Close all handles and verify that IsBound() returns false.
  // Note that on Unix, only closing the server handle is necessary for this,
  // but the way Win32 named pipes work require all client handles closed too.
  peer.Close();
  client.handle.Close();

  service.Stop();
  ASSERT_FALSE(IpcService::HasServer(service_name));
}

TEST(IpcService, StartAndAcceptWithTimeout) {
  std::string error_message;
  StringPiece service_name = "test_service";

  // Create server
  IpcService server = IpcService::StartServer(service_name, &error_message);
  ASSERT_TRUE(server) << error_message;
  ASSERT_STRING_EMPTY(error_message);

  // Accepting should time out since no client is trying to connect.
  bool did_timeout = false;
  ScopedHandle connection = server.AcceptPeer(10, &did_timeout, &error_message);
  ASSERT_FALSE(connection) << IPC_ERROR_CONTEXT("AcceptPeer");
  ASSERT_TRUE(did_timeout);
  ASSERT_STRING_NOT_EMPTY(error_message);
  error_message.clear();
  did_timeout = false;

  // Start a connecting thread in the background with a large timeout.
  TestIpcServiceConnectingThread client(service_name, 1000);
  client.Start();

  // Accepting should now work.
  ScopedHandle peer = server.AcceptPeer(1, &did_timeout, &error_message);
  ASSERT_TRUE(peer) << IPC_ERROR_CONTEXT("AcceptPeer");
  ASSERT_FALSE(did_timeout);
  ASSERT_STRING_EMPTY(error_message);

  // Connect to the server, and receive a peer handle.
  client.Join();
  ASSERT_TRUE(client.handle) << "client did_timeout=" << client.did_timeout
                             << ", error: " << client.error_message;
  ASSERT_FALSE(client.did_timeout);
  ASSERT_STRING_EMPTY(client.error_message);
}

TEST(IpcService, StartAndSendLoop) {
  std::string service_name = "test_service";
  std::string error_message;

  // Start server.
  IpcService server = IpcService::StartServer(service_name, &error_message);
  ASSERT_TRUE(server) << "StartServer: " << error_message;
  ASSERT_STRING_EMPTY(error_message);
  ASSERT_TRUE(server.IsValid());

  ASSERT_TRUE(IpcService::HasServer(service_name));

  for (size_t count = 0; count < 10; ++count) {
    std::string loop_info = std::string("#") + std::to_string(count) + ": ";

    // Start a first client in a background thread.
    TestIpcServiceConnectingThread client1(service_name, 1000);
    client1.Start();

    // Accept it as first peer. Give 10ms timeout to allow thread
    // to be scheduled.
    bool did_timeout = false;
    ScopedHandle peer1 = server.AcceptPeer(10, &did_timeout, &error_message);
    ASSERT_TRUE(peer1) << loop_info << IPC_ERROR_CONTEXT("AcceptPeer #1");
    ASSERT_FALSE(did_timeout);
    ASSERT_STRING_EMPTY(error_message);

    client1.Join();
    ASSERT_TRUE(client1.handle) << loop_info;

    // Send data through the first pipe from peer to client, then close the
    // peer handle, but do not close the client handle. This should maintain
    // the pipe buffer in kernel memory.
    const char kMessage[] = "One, Two, Three, Four";
    ASSERT_TRUE(peer1.WriteFull(kMessage, sizeof(kMessage), &error_message))
        << loop_info << error_message;
    peer1.Close();

    // Create second client in background thread.
    TestIpcServiceConnectingThread client2(service_name, 1000);
    client2.Start();

    // Accept it as second peer.
    ScopedHandle peer2 = server.AcceptPeer(10, &did_timeout, &error_message);
    ASSERT_TRUE(peer2) << loop_info << IPC_ERROR_CONTEXT("AcceptPeer #2");
    ASSERT_FALSE(did_timeout);
    ASSERT_STRING_EMPTY(error_message);

    client2.Join();
    ASSERT_TRUE(client2.handle);

    // Receive data from the first pipe.
    char buffer1[sizeof(kMessage)] = {};
    ASSERT_TRUE(
        client1.handle.ReadFull(buffer1, sizeof(kMessage), &error_message))
        << loop_info << error_message;
    ASSERT_STREQ(kMessage, buffer1) << loop_info;

    const char kMessage2[] = "Uno, Dos, Tres, Quatro";
    ASSERT_TRUE(peer2.WriteFull(kMessage2, sizeof(kMessage2), &error_message))
        << loop_info << error_message;
    peer2.Close();

    char buffer2[sizeof(kMessage2)] = {};
    ASSERT_TRUE(
        client2.handle.ReadFull(buffer2, sizeof(kMessage2), &error_message))
        << loop_info << error_message;
    ASSERT_STREQ(kMessage2, buffer2) << loop_info;
  }
}

TEST(IpcService, SendAndReceiveNativeHandle) {
  ScopedHandle pipe1_read;
  ScopedHandle pipe1_write;
  ASSERT_TRUE(CreateTestPipe(&pipe1_read, &pipe1_write));

  std::string error_message;
  StringPiece service_name = "test_service";
  IpcService server = IpcService::StartServer(service_name, &error_message);
  ASSERT_TRUE(server) << error_message;
  ASSERT_STRING_EMPTY(error_message);

  // Start a connecting thread in the background with a large timeout.
  TestIpcServiceConnectingThread client(service_name, 1000);
  client.Start();

  bool did_timeout = false;
  ScopedHandle peer = server.AcceptPeer(-1, &did_timeout, &error_message);
  ASSERT_TRUE(peer) << error_message;
  ASSERT_FALSE(did_timeout);
  ASSERT_STRING_EMPTY(error_message);

  client.Join();
  ASSERT_TRUE(client.handle) << "client did_timeout=" << client.did_timeout
                             << ", error: " << client.error_message;
  ASSERT_FALSE(client.did_timeout);
  ASSERT_STRING_EMPTY(client.error_message);

  // Send pipe1_write.get() from client to peer.
  ASSERT_TRUE(
      RemoteSendNativeHandle(client.handle, pipe1_write.get(), &error_message))
      << error_message;
  ASSERT_STRING_EMPTY(error_message);

  ScopedHandle received;
  ASSERT_TRUE(RemoteReceiveNativeHandle(peer, &received, &error_message))
      << error_message;
  ASSERT_STRING_EMPTY(error_message);

  // Write to |received|, this should send data to the first pipe.
  const char kInputData[] = "Bonjour monde!";
  const size_t kInputDataSize = sizeof(kInputData);

  ASSERT_TRUE(received.WriteFull(kInputData, kInputDataSize, &error_message));
  ASSERT_STRING_EMPTY(error_message);

  char data[kInputDataSize] = {};
  ssize_t count = pipe1_read.Read(data, sizeof(data), &error_message);
  ASSERT_EQ(static_cast<ssize_t>(sizeof(data)), count);
  ASSERT_STRING_EMPTY(error_message);

  // Verify content.
  ASSERT_TRUE(!memcmp(data, kInputData, kInputDataSize));
}