Stay writable after partial socket writes.

webrtc/base/physicalsocketserver_unittest.cc

 /*
  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
@@ skipping 11 matching lines @@
 namespace rtc {
 
 class PhysicalSocketTest;
 
 class FakeSocketDispatcher : public SocketDispatcher {
  public:
   explicit FakeSocketDispatcher(PhysicalSocketServer* ss)
     : SocketDispatcher(ss) {
   }
 
+  FakeSocketDispatcher(SOCKET s, PhysicalSocketServer* ss)
+    : SocketDispatcher(s, ss) {
+  }
+
  protected:
   SOCKET DoAccept(SOCKET socket, sockaddr* addr, socklen_t* addrlen) override;
+  int DoSend(SOCKET socket, const char* buf, int len, int flags) override;
+  int DoSendTo(SOCKET socket, const char* buf, int len, int flags,
+               const struct sockaddr* dest_addr, socklen_t addrlen) override;
 };
 
 class FakePhysicalSocketServer : public PhysicalSocketServer {
  public:
   explicit FakePhysicalSocketServer(PhysicalSocketTest* test)
     : test_(test) {
   }
 
   AsyncSocket* CreateAsyncSocket(int type) override {
     SocketDispatcher* dispatcher = new FakeSocketDispatcher(this);
     if (dispatcher->Create(type)) {
       return dispatcher;
     } else {
       delete dispatcher;
       return nullptr;
     }
   }
 
   AsyncSocket* CreateAsyncSocket(int family, int type) override {
     SocketDispatcher* dispatcher = new FakeSocketDispatcher(this);
     if (dispatcher->Create(family, type)) {
       return dispatcher;
     } else {
       delete dispatcher;
       return nullptr;
     }
   }
 
+  AsyncSocket* WrapSocket(SOCKET s) override {
+    SocketDispatcher* dispatcher = new FakeSocketDispatcher(s, this);
+    if (dispatcher->Initialize()) {
+      return dispatcher;
+    } else {
+      delete dispatcher;
+      return nullptr;
+    }

[pthatcher1, 2016/01/30 00:03:10]

Can you flip this around?

if (!dispatcher->Initialize()) {
  delete dispatcher;
  return nullptr;
}
return dispatcher;

[joachim, 2016/01/31 23:29:07]

Done. This was implemented similar to the existing "CreateAsyncSocket" methods,
so I changed these as well.

+  }
+
   PhysicalSocketTest* GetTest() const { return test_; }
 
  private:
   PhysicalSocketTest* test_;
 };
 
 class PhysicalSocketTest : public SocketTest {
  public:
   // Set flag to simluate failures when calling "::accept" on a AsyncSocket.
   void SetFailAccept(bool fail) { fail_accept_ = fail; }
   bool FailAccept() const { return fail_accept_; }
 
+  // Maximum size to ::send to a socket. Set to 0 to disable limiting.

[pthatcher1, 2016/01/30 00:03:10]

Can you make "disabled limited" < 0, such as -1.  0 sounds to me like "it can't
send", which would be a valid option for a test.

[joachim, 2016/01/31 23:29:06]

Done.

+  void SetMaximumSendSize(int max_size) { max_send_size_ = max_size; }
+  int MaximumSendSize() const { return max_send_size_; }

[pthatcher1, 2016/01/30 00:03:10]

Please change MaximumSendSize => MaxSendSize in both places.

[joachim, 2016/01/31 23:29:07]

Done.

+
  protected:
   PhysicalSocketTest()
     : server_(new FakePhysicalSocketServer(this)),
       scope_(server_.get()),
-      fail_accept_(false) {
+      fail_accept_(false),
+      max_send_size_(0) {
   }
 
   void ConnectInternalAcceptError(const IPAddress& loopback);
+  void PartialWriteStayWritable(const IPAddress& loopback);

[pthatcher1, 2016/01/30 00:03:10]

Can you name this WritableAfterPartialWrite?

[joachim, 2016/01/31 23:29:06]

Done.

 
   rtc::scoped_ptr<FakePhysicalSocketServer> server_;
   SocketServerScope scope_;
   bool fail_accept_;
+  int max_send_size_;
 };
 
 SOCKET FakeSocketDispatcher::DoAccept(SOCKET socket,
                                       sockaddr* addr,
                                       socklen_t* addrlen) {
   FakePhysicalSocketServer* ss =
       static_cast<FakePhysicalSocketServer*>(socketserver());
   if (ss->GetTest()->FailAccept()) {
     return INVALID_SOCKET;
   }
 
   return SocketDispatcher::DoAccept(socket, addr, addrlen);
 }
 
+int FakeSocketDispatcher::DoSend(SOCKET socket, const char* buf, int len,
+    int flags) {
+  FakePhysicalSocketServer* ss =
+      static_cast<FakePhysicalSocketServer*>(socketserver());
+  if (ss->GetTest()->MaximumSendSize() > 0) {
+    len = std::min(len, ss->GetTest()->MaximumSendSize());
+  }
+
+  return SocketDispatcher::DoSend(socket, buf, len, flags);
+}
+
+int FakeSocketDispatcher::DoSendTo(SOCKET socket, const char* buf, int len,
+    int flags, const struct sockaddr* dest_addr, socklen_t addrlen) {
+  FakePhysicalSocketServer* ss =
+      static_cast<FakePhysicalSocketServer*>(socketserver());
+  if (ss->GetTest()->MaximumSendSize() > 0) {
+    len = std::min(len, ss->GetTest()->MaximumSendSize());
+  }
+
+  return SocketDispatcher::DoSendTo(socket, buf, len, flags, dest_addr,
+      addrlen);
+}
+
 TEST_F(PhysicalSocketTest, TestConnectIPv4) {
   SocketTest::TestConnectIPv4();
 }
 
 // Crashes on Linux. See webrtc:4923.
 #if defined(WEBRTC_LINUX)
 #define MAYBE_TestConnectIPv6 DISABLED_TestConnectIPv6
 #else
 #define MAYBE_TestConnectIPv6 TestConnectIPv6
 #endif
@@ skipping 92 matching lines @@
 // Crashes on Linux. See webrtc:4923.
 #if defined(WEBRTC_LINUX)
 #define MAYBE_TestConnectAcceptErrorIPv6 DISABLED_TestConnectAcceptErrorIPv6
 #else
 #define MAYBE_TestConnectAcceptErrorIPv6 TestConnectAcceptErrorIPv6
 #endif
 TEST_F(PhysicalSocketTest, MAYBE_TestConnectAcceptErrorIPv6) {
   ConnectInternalAcceptError(kIPv6Loopback);
 }
 
+void PhysicalSocketTest::PartialWriteStayWritable(const IPAddress& loopback) {
+  testing::StreamSink sink;
+  SocketAddress accept_addr;
+
+  // Simulate a really small maximum send size.
+  const int kMaximumSendSize = 128;

[pthatcher1, 2016/01/30 00:03:10]

kMaxSendSize

[joachim, 2016/01/31 23:29:06]

Done.

+  SetMaximumSendSize(kMaximumSendSize);
+
+  // Create test data.
+  const size_t kDataSize = 128 * 1024;
+  scoped_ptr<char[]> send_buffer(new char[kDataSize]);
+  scoped_ptr<char[]> recv_buffer(new char[kDataSize]);

[pthatcher1, 2016/01/30 00:03:10]

It seems like this would benefit from using rtc::Buffer.


rtc::Buffer recv_buffer(0, kDataSize);
rtc::Buffer send_buffer(0, kDataSize);
for (size_t i = 0; i < kDataSize; ++i) {
  send_buffer.AppendData(static_cast<char>(i % 256));
}


Then later:



while (recv_buffer.size() < send_buffer.size()) {
  int unsent_count = send_buffer.size() - sent_size;
  sender->Send(send_buffer.data() + sent_size, unsent_size);
  ...

  char[kDataSize] recved_data;
  int recved_size receiver->Recv(temp, kDataSize);
  ...
  if (recved_data > 0) {
    recv_buffer.AppendData(recved_data, recved_size);
  }

 ... 
EXPECT_EQ(send_buffer, receive_buffer)

[joachim, 2016/01/31 23:29:07]

Done. As I used "SocketTest::TcpInternal" as base for this test, I updated
"TcpInternal" instead with your suggested changes here and below and added
parameters to it where it needs to run differently from here.

+  size_t send_pos = 0, recv_pos = 0;

[pthatcher1, 2016/01/30 00:03:10]

You can replace all instances of recv_pos with recv_buffer.size().

Can you call send_pos "sent_size" instead?

[joachim, 2016/01/31 23:29:06]

Done.

+  for (size_t i = 0; i < kDataSize; ++i) {
+    send_buffer[i] = static_cast<char>(i % 256);
+    recv_buffer[i] = 0;
+  }
+
+  // Create client.
+  scoped_ptr<AsyncSocket> client(
+      server_->CreateAsyncSocket(loopback.family(), SOCK_STREAM));
+  sink.Monitor(client.get());
+
+  // Create server and listen.
+  scoped_ptr<AsyncSocket> server(
+      server_->CreateAsyncSocket(loopback.family(), SOCK_STREAM));
+  sink.Monitor(server.get());
+  EXPECT_EQ(0, server->Bind(SocketAddress(loopback, 0)));
+  EXPECT_EQ(0, server->Listen(5));
+
+  // Attempt connection.
+  EXPECT_EQ(0, client->Connect(server->GetLocalAddress()));
+
+  // Accept connection.
+  EXPECT_TRUE_WAIT((sink.Check(server.get(), testing::SSE_READ)), kTimeout);
+  scoped_ptr<AsyncSocket> accepted(server->Accept(&accept_addr));
+  ASSERT_TRUE(accepted);
+  sink.Monitor(accepted.get());
+
+  // Both sides are now connected.
+  EXPECT_EQ_WAIT(AsyncSocket::CS_CONNECTED, client->GetState(), kTimeout);
+  EXPECT_TRUE(sink.Check(client.get(), testing::SSE_OPEN));
+  EXPECT_EQ(client->GetRemoteAddress(), accepted->GetLocalAddress());
+  EXPECT_EQ(accepted->GetRemoteAddress(), client->GetLocalAddress());
+
+  // Send and receive a bunch of data.

[pthatcher1, 2016/01/30 00:03:10]

Can you move the definition of the send and recv buffers to down here just
before they are used?  I think that would make it more readable.

[joachim, 2016/01/31 23:29:06]

Done.

+  bool send_waiting_for_writability = false;

[pthatcher1, 2016/01/30 00:03:10]

Why not just "writable", with the reversed meaning (writable = true here)?  That
would seem more clear.  Below it would be while (writable && send_pos <
kDataSize) { ... }

[pthatcher1, 2016/01/30 00:03:10]

Can you call 

sender = accepted;
receiver = client;

That would make the code below more readable.

[joachim, 2016/01/31 23:29:06]

Done.

[joachim, 2016/01/31 23:29:06]

Done.

+  bool send_expect_success = true;

[pthatcher1, 2016/01/30 00:03:10]

Can you call this send_called = false?  That would make it more clear how you
use it below.

[joachim, 2016/01/31 23:29:07]

Done.

+  bool recv_waiting_for_readability = true;

[pthatcher1, 2016/01/30 00:03:10]

Similarly, "readable = false" here.

[joachim, 2016/01/31 23:29:06]

Done.

+  bool recv_expect_success = false;

[pthatcher1, 2016/01/30 00:03:10]

And this "recv_called = false"?

[joachim, 2016/01/31 23:29:06]

Done.

+  int data_in_flight = 0;
+  while (recv_pos < kDataSize) {
+    // Send as much as we can if we've been cleared to send.

[pthatcher1, 2016/01/30 00:03:09]

if we've been cleared to send => while we're cleared to send

?

[joachim, 2016/01/31 23:29:06]

Done.

+    while (!send_waiting_for_writability && send_pos < kDataSize) {
+      int tosend = static_cast<int>(kDataSize - send_pos);
+      int sent = accepted->Send(send_buffer.get() + send_pos, tosend);
+      if (send_expect_success) {
+        // The first Send() after connecting or getting writability should
+        // succeed and send some data.
+        EXPECT_GT(sent, 0);
+        send_expect_success = false;
+      }
+      if (sent >= 0) {
+        EXPECT_LE(sent, tosend);
+        EXPECT_LE(sent, kMaximumSendSize);
+        send_pos += sent;
+        data_in_flight += sent;

[pthatcher1, 2016/01/30 00:03:10]

I think you can replace all instances of data_in_flight with (sent_size -
recv_buffer.size())

[joachim, 2016/01/31 23:29:06]

Done.

+      }
+      if (sent < tosend) {
+        // Partial data or nothing sent.
+        ASSERT_TRUE(accepted->IsBlocking());
+        send_waiting_for_writability = true;
+      }
+    }
+
+    // Read all the sent data.
+    while (data_in_flight > 0) {

[pthatcher1, 2016/01/30 00:03:10]

Or in this case:

while (recv_buffer.size() < sent_size)

[joachim, 2016/01/31 23:29:06]

Done.

+      if (recv_waiting_for_readability) {
+        // Wait until data is available.
+        EXPECT_TRUE_WAIT(sink.Check(client.get(), testing::SSE_READ), kTimeout);
+        recv_waiting_for_readability = false;
+        recv_expect_success = true;
+      }
+
+      // Receive as much as we can get in a single recv call.
+      int rcvd = client->Recv(recv_buffer.get() + recv_pos,
+                              kDataSize - recv_pos);
+
+      if (recv_expect_success) {
+        // The first Recv() after getting readability should succeed and receive
+        // some data.
+        EXPECT_GT(rcvd, 0);
+        recv_expect_success = false;
+      }
+      if (rcvd >= 0) {
+        EXPECT_LE(rcvd, data_in_flight);
+        recv_pos += rcvd;
+        data_in_flight -= rcvd;
+      } else {
+        ASSERT_TRUE(client->IsBlocking());
+        recv_waiting_for_readability = true;
+      }
+    }
+
+    // Once all that we've sent has been rcvd, expect to be able to send again.
+    if (send_waiting_for_writability) {
+      EXPECT_TRUE_WAIT(sink.Check(accepted.get(), testing::SSE_WRITE),
+                       kTimeout);
+      send_waiting_for_writability = false;
+      send_expect_success = true;
+    }
+  }
+
+  // The received data matches the sent data.
+  EXPECT_EQ(kDataSize, send_pos);
+  EXPECT_EQ(kDataSize, recv_pos);
+  EXPECT_EQ(0, memcmp(recv_buffer.get(), send_buffer.get(), kDataSize));
+
+  // Close down.
+  accepted->Close();
+  EXPECT_EQ_WAIT(AsyncSocket::CS_CLOSED, client->GetState(), kTimeout);
+  EXPECT_TRUE(sink.Check(client.get(), testing::SSE_CLOSE));
+  client->Close();
+}
+
+TEST_F(PhysicalSocketTest, TestPartialWriteStayWritableIPv4) {
+  PartialWriteStayWritable(kIPv4Loopback);
+}
+
+// Crashes on Linux. See webrtc:4923.
+#if defined(WEBRTC_LINUX)
+#define MAYBE_TestPartialWriteStayWritableIPv6 \
+    DISABLED_TestPartialWriteStayWritableIPv6
+#else
+#define MAYBE_TestPartialWriteStayWritableIPv6 TestPartialWriteStayWritableIPv6
+#endif
+TEST_F(PhysicalSocketTest, MAYBE_TestPartialWriteStayWritableIPv6) {
+  PartialWriteStayWritable(kIPv6Loopback);
+}
+
 // Crashes on Linux. See webrtc:4923.
 #if defined(WEBRTC_LINUX)
 #define MAYBE_TestConnectFailIPv6 DISABLED_TestConnectFailIPv6
 #else
 #define MAYBE_TestConnectFailIPv6 TestConnectFailIPv6
 #endif
 TEST_F(PhysicalSocketTest, MAYBE_TestConnectFailIPv6) {
   SocketTest::TestConnectFailIPv6();
 }
 
@@ skipping 295 matching lines @@
   thread->Start(runnable.get());
   EXPECT_TRUE(ss_->Wait(1500, true));
   EXPECT_TRUE(ExpectSignal(SIGTERM));
   EXPECT_EQ(Thread::Current(), signaled_thread_);
   EXPECT_TRUE(ExpectNone());
 }
 
 #endif
 
 }  // namespace rtc