mojo/system/message_pipe_unittest.cc

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

#include "mojo/system/message_pipe.h"

#include "base/memory/ref_counted.h"
#include "base/threading/platform_thread.h"  // For |Sleep()|.
#include "base/time/time.h"
#include "mojo/system/waiter.h"
#include "mojo/system/waiter_test_utils.h"
#include "testing/gtest/include/gtest/gtest.h"

namespace mojo {
namespace system {
namespace {

// Tests:
//  - only default flags
//  - reading messages from a port
//    - when there are no/one/two messages available for that port
//    - with buffer size 0 (and null buffer) -- should get size
//    - with too-small buffer -- should get size
//    - also verify that buffers aren't modified when/where they shouldn't be
//  - writing messages to a port
//    - in the obvious scenarios (as above)
//    - to a port that's been closed
//  - writing a message to a port, closing the other (would be the source) port,
//    and reading it
TEST(MessagePipeTest, Basic) {
  scoped_refptr<MessagePipe> mp(new MessagePipe());

  int32_t buffer[2];
  const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer));
  uint32_t buffer_size;

  // Nothing to read yet on port 0.
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(kBufferSize, buffer_size);
  EXPECT_EQ(123, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Ditto for port 1.
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));

  // Write from port 1 (to port 0).
  buffer[0] = 789012345;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(1,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Read from port 0.
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);
  EXPECT_EQ(789012345, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Read again from port 0 -- it should be empty.
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));

  // Write two messages from port 0 (to port 1).
  buffer[0] = 123456789;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(0,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));
  buffer[0] = 234567890;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(0,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Read from port 1 with buffer size 0 (should get the size of next message).
  // Also test that giving a null buffer is okay when the buffer size is 0.
  buffer_size = 0;
  EXPECT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
            mp->ReadMessage(1,
                            NULL, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);

  // Read from port 1 with buffer size 1 (too small; should get the size of next
  // message).
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = 1;
  EXPECT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);
  EXPECT_EQ(123, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Read from port 1.
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);
  EXPECT_EQ(123456789, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Read again from port 1.
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);
  EXPECT_EQ(234567890, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Read again from port 1 -- it should be empty.
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));

  // Write from port 0 (to port 1).
  buffer[0] = 345678901;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(0,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Close port 0.
  mp->Close(0);

  // Try to write from port 1 (to port 0).
  buffer[0] = 456789012;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
            mp->WriteMessage(1,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Read from port 1; should still get message (even though port 0 was closed).
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);
  EXPECT_EQ(345678901, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Read again from port 1 -- it should be empty (and port 0 is closed).
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));

  mp->Close(1);
}

TEST(MessagePipeTest, CloseWithQueuedIncomingMessages) {
  scoped_refptr<MessagePipe> mp(new MessagePipe());

  int32_t buffer[1];
  const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer));
  uint32_t buffer_size;

  // Write some messages from port 1 (to port 0).
  for (int32_t i = 0; i < 5; i++) {
    buffer[0] = i;
    EXPECT_EQ(MOJO_RESULT_OK,
              mp->WriteMessage(1,
                               buffer, kBufferSize,
                               NULL,
                               MOJO_WRITE_MESSAGE_FLAG_NONE));
  }

  // Port 0 shouldn't be empty.
  buffer_size = 0;
  EXPECT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
            mp->ReadMessage(0,
                            NULL, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(kBufferSize, buffer_size);

  // Close port 0 first, which should have outstanding (incoming) messages.
  mp->Close(0);
  mp->Close(1);
}

TEST(MessagePipeTest, DiscardMode) {
  scoped_refptr<MessagePipe> mp(new MessagePipe());

  int32_t buffer[2];
  const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer));
  uint32_t buffer_size;

  // Write from port 1 (to port 0).
  buffer[0] = 789012345;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(1,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Read/discard from port 0 (no buffer); get size.
  buffer_size = 0;
  EXPECT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
            mp->ReadMessage(0,
                            NULL, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);

  // Read again from port 0 -- it should be empty.
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));

  // Write from port 1 (to port 0).
  buffer[0] = 890123456;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(1,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Read from port 0 (buffer big enough).
  buffer[0] = 123;
  buffer[1] = 456;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);
  EXPECT_EQ(890123456, buffer[0]);
  EXPECT_EQ(456, buffer[1]);

  // Read again from port 0 -- it should be empty.
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));

  // Write from port 1 (to port 0).
  buffer[0] = 901234567;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(1,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Read/discard from port 0 (buffer too small); get size.
  buffer_size = 1;
  EXPECT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));
  EXPECT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size);

  // Read again from port 0 -- it should be empty.
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));

  // Write from port 1 (to port 0).
  buffer[0] = 123456789;
  buffer[1] = 0;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(1,
                             buffer, static_cast<uint32_t>(sizeof(buffer[0])),
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Discard from port 0.
  buffer_size = 1;
  EXPECT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
            mp->ReadMessage(0,
                            NULL, NULL,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));

  // Read again from port 0 -- it should be empty.
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT,
            mp->ReadMessage(0,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_MAY_DISCARD));

  mp->Close(0);
  mp->Close(1);
}

TEST(MessagePipeTest, BasicWaiting) {
  scoped_refptr<MessagePipe> mp(new MessagePipe());
  Waiter waiter;

  int32_t buffer[1];
  const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer));
  uint32_t buffer_size;

  // Always writable (until the other port is closed).
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
            mp->AddWaiter(0, &waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0));
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
            mp->AddWaiter(0,
                          &waiter,
                          MOJO_HANDLE_SIGNAL_READABLE |
                              MOJO_HANDLE_SIGNAL_WRITABLE,
                          0));

  // Not yet readable.
  waiter.Init();
  ASSERT_EQ(MOJO_RESULT_OK,
            mp->AddWaiter(0, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 1));
  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, NULL));
  mp->RemoveWaiter(0, &waiter);

  // Write from port 0 (to port 1), to make port 1 readable.
  buffer[0] = 123456789;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->WriteMessage(0,
                             buffer, kBufferSize,
                             NULL,
                             MOJO_WRITE_MESSAGE_FLAG_NONE));

  // Port 1 should already be readable now.
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
            mp->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 2));
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
            mp->AddWaiter(1,
                          &waiter,
                          MOJO_HANDLE_SIGNAL_READABLE |
                              MOJO_HANDLE_SIGNAL_WRITABLE,
                          0));
  // ... and still writable.
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
            mp->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 3));

  // Close port 0.
  mp->Close(0);

  // Now port 1 should not be writable.
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
            mp->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 4));

  // But it should still be readable.
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
            mp->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 5));

  // Read from port 1.
  buffer[0] = 0;
  buffer_size = kBufferSize;
  EXPECT_EQ(MOJO_RESULT_OK,
            mp->ReadMessage(1,
                            buffer, &buffer_size,
                            0, NULL,
                            MOJO_READ_MESSAGE_FLAG_NONE));
  EXPECT_EQ(123456789, buffer[0]);

  // Now port 1 should no longer be readable.
  waiter.Init();
  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
            mp->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 6));

  mp->Close(1);
}

TEST(MessagePipeTest, ThreadedWaiting) {
  int32_t buffer[1];
  const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer));

  MojoResult result;
  uint32_t context;

  // Write to wake up waiter waiting for read.
  {
    scoped_refptr<MessagePipe> mp(new MessagePipe());
    test::SimpleWaiterThread thread(&result, &context);

    thread.waiter()->Init();
    ASSERT_EQ(MOJO_RESULT_OK,
              mp->AddWaiter(1, thread.waiter(), MOJO_HANDLE_SIGNAL_READABLE,
                            1));
    thread.Start();

    buffer[0] = 123456789;
    // Write from port 0 (to port 1), which should wake up the waiter.
    EXPECT_EQ(MOJO_RESULT_OK,
              mp->WriteMessage(0,
                               buffer, kBufferSize,
                               NULL,
                               MOJO_WRITE_MESSAGE_FLAG_NONE));

    mp->RemoveWaiter(1, thread.waiter());

    mp->Close(0);
    mp->Close(1);
  }  // Joins |thread|.
  // The waiter should have woken up successfully.
  EXPECT_EQ(MOJO_RESULT_OK, result);
  EXPECT_EQ(1u, context);

  // Close to cancel waiter.
  {
    scoped_refptr<MessagePipe> mp(new MessagePipe());
    test::SimpleWaiterThread thread(&result, &context);

    thread.waiter()->Init();
    ASSERT_EQ(MOJO_RESULT_OK,
              mp->AddWaiter(1, thread.waiter(), MOJO_HANDLE_SIGNAL_READABLE,
                            2));
    thread.Start();

    // Close port 1 first -- this should result in the waiter being cancelled.
    mp->CancelAllWaiters(1);
    mp->Close(1);

    // Port 1 is closed, so |Dispatcher::RemoveWaiter()| wouldn't call into the
    // |MessagePipe| to remove any waiter.

    mp->Close(0);
  }  // Joins |thread|.
  EXPECT_EQ(MOJO_RESULT_CANCELLED, result);
  EXPECT_EQ(2u, context);

  // Close to make waiter un-wake-up-able.
  {
    scoped_refptr<MessagePipe> mp(new MessagePipe());
    test::SimpleWaiterThread thread(&result, &context);

    thread.waiter()->Init();
    ASSERT_EQ(MOJO_RESULT_OK,
              mp->AddWaiter(1, thread.waiter(), MOJO_HANDLE_SIGNAL_READABLE,
                            3));
    thread.Start();

    // Close port 0 first -- this should wake the waiter up, since port 1 will
    // never be readable.
    mp->CancelAllWaiters(0);
    mp->Close(0);

    mp->RemoveWaiter(1, thread.waiter());

    mp->CancelAllWaiters(1);
    mp->Close(1);
  }  // Joins |thread|.
  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result);
  EXPECT_EQ(3u, context);
}

}  // namespace
}  // namespace system
}  // namespace mojo