Revert "Implement full duplex mode for TCPClientSocketLibevent."

Revert "Disable the flaky fullduplex tests."

Review URL: http://codereview.chromium.org/113749

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@16685 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 156 insertions, 275 deletions

diff --git a/net/base/listen_socket.cc b/net/base/listen_socket.cc
index d31c66d..a9d16ef 100644
--- a/net/base/listen_socket.cc
+++ b/net/base/listen_socket.cc
@@ -23,7 +23,7 @@
 #if defined(OS_WIN)
 #define socklen_t int
 #elif defined(OS_POSIX)
-const int INVALID_SOCKET = -1;  // Used same name as in Windows to avoid #ifdef
+const int INVALID_SOCKET = -1; // Used same name as in Windows to avoid #ifdef
 const int SOCKET_ERROR = -1;
 #endif
 
@@ -31,9 +31,7 @@ const int kReadBufSize = 200;
 
 ListenSocket::ListenSocket(SOCKET s, ListenSocketDelegate *del)
     : socket_(s),
-      socket_delegate_(del),
-      reads_paused_(false),
-      has_pending_reads_(false) {
+      socket_delegate_(del) {
 #if defined(OS_WIN)
   socket_event_ = WSACreateEvent();
   // TODO(ibrar): error handling in case of socket_event_ == WSA_INVALID_EVENT
@@ -85,7 +83,7 @@ ListenSocket* ListenSocket::Listen(std::string ip, int port,
 }
 
 void ListenSocket::Listen() {
-  int backlog = 10;  // TODO(erikkay): maybe don't allow any backlog?
+  int backlog = 10; // TODO(erikkay): maybe don't allow any backlog?
   listen(socket_, backlog);
   // TODO(erikkay): error handling
 #if defined(OS_POSIX)
@@ -120,7 +118,7 @@ void ListenSocket::Accept() {
 }
 
 void ListenSocket::Read() {
-  char buf[kReadBufSize + 1];  // +1 for null termination
+  char buf[kReadBufSize + 1]; // +1 for null termination
   int len;
   do {
     len = HANDLE_EINTR(recv(socket_, buf, kReadBufSize, 0));
@@ -145,7 +143,7 @@ void ListenSocket::Read() {
     } else {
       // TODO(ibrar): maybe change DidRead to take a length instead
       DCHECK(len > 0 && len <= kReadBufSize);
-      buf[len] = 0;  // already create a buffer with +1 length
+      buf[len] = 0; // already create a buffer with +1 length
       socket_delegate_->DidRead(this, buf);
     }
   } while (len == kReadBufSize);
@@ -200,7 +198,7 @@ void ListenSocket::SendInternal(const char* bytes, int len) {
 #elif defined(OS_POSIX)
     if (errno == EWOULDBLOCK || errno == EAGAIN) {
 #endif
-    // TODO(ibrar): there should be logic here to handle this because
+    // TODO (ibrar): there should be logic here to handle this because
     // it is not an error
     }
   } else if (sent != len) {
@@ -219,21 +217,7 @@ void ListenSocket::Send(const std::string& str, bool append_linefeed) {
   Send(str.data(), static_cast<int>(str.length()), append_linefeed);
 }
 
-void ListenSocket::PauseReads() {
-  DCHECK(!reads_paused_);
-  reads_paused_ = true;
-}
-
-void ListenSocket::ResumeReads() {
-  DCHECK(reads_paused_);
-  reads_paused_ = false;
-  if (has_pending_reads_) {
-    has_pending_reads_ = false;
-    Read();
-  }
-}
-
-// TODO(ibrar): We can add these functions into OS dependent files
+// TODO (ibrar): We can add these functions into OS dependent files
 #if defined(OS_WIN)
 // MessageLoop watcher callback
 void ListenSocket::OnObjectSignaled(HANDLE object) {
@@ -255,11 +239,7 @@ void ListenSocket::OnObjectSignaled(HANDLE object) {
     Accept();
   }
   if (ev.lNetworkEvents & FD_READ) {
-    if (reads_paused_) {
-      has_pending_reads_ = true;
-    } else {
-      Read();
-    }
+    Read();
   }
   if (ev.lNetworkEvents & FD_CLOSE) {
     Close();
@@ -271,11 +251,7 @@ void ListenSocket::OnFileCanReadWithoutBlocking(int fd) {
     Accept();
   }
   if (wait_state_ == WAITING_READ) {
-    if (reads_paused_) {
-      has_pending_reads_ = true;
-    } else {
-      Read();
-    }
+    Read();
   }
   if (wait_state_ == WAITING_CLOSE) {
     // Close() is called by Read() in the Linux case.
diff --git a/net/base/listen_socket.h b/net/base/listen_socket.h
index cadc962..c6650ac 100644
--- a/net/base/listen_socket.h
+++ b/net/base/listen_socket.h
@@ -45,8 +45,6 @@ class ListenSocket : public base::RefCountedThreadSafe<ListenSocket>,
   // should be split up similarly.
   class ListenSocketDelegate {
    public:
-    virtual ~ListenSocketDelegate() {}
-
     // server is the original listening Socket, connection is the new
     // Socket that was created.  Ownership of connection is transferred
     // to the delegate with this call.
@@ -61,16 +59,10 @@ class ListenSocket : public base::RefCountedThreadSafe<ListenSocket>,
                               ListenSocketDelegate* del);
   virtual ~ListenSocket();
 
-  // Send data to the socket.
+  // send data to the socket
   void Send(const char* bytes, int len, bool append_linefeed = false);
   void Send(const std::string& str, bool append_linefeed = false);
 
-  // NOTE: This is for unit test use only!
-  // Pause/Resume calling Read().  Note that ResumeReads() will also call
-  // Read() if there is anything to read.
-  void PauseReads();
-  void ResumeReads();
-
  protected:
   ListenSocket(SOCKET s, ListenSocketDelegate* del);
   static SOCKET Listen(std::string ip, int port);
@@ -114,10 +106,7 @@ class ListenSocket : public base::RefCountedThreadSafe<ListenSocket>,
   ListenSocketDelegate *socket_delegate_;
 
  private:
-  bool reads_paused_;
-  bool has_pending_reads_;
-
-  DISALLOW_COPY_AND_ASSIGN(ListenSocket);
+  DISALLOW_EVIL_CONSTRUCTORS(ListenSocket);
 };
 
 #endif  // NET_BASE_LISTEN_SOCKET_H_
diff --git a/net/base/tcp_client_socket_libevent.cc b/net/base/tcp_client_socket_libevent.cc
index 6ad3d2d..3e23f1c 100644
--- a/net/base/tcp_client_socket_libevent.cc
+++ b/net/base/tcp_client_socket_libevent.cc
@@ -19,13 +19,11 @@
 
 namespace net {
 
-namespace {
-
 const int kInvalidSocket = -1;
 
 // Return 0 on success, -1 on failure.
 // Too small a function to bother putting in a library?
-int SetNonBlocking(int fd) {
+static int SetNonBlocking(int fd) {
     int flags = fcntl(fd, F_GETFL, 0);
     if (-1 == flags)
       return flags;
@@ -33,7 +31,7 @@ int SetNonBlocking(int fd) {
 }
 
 // Convert values from <errno.h> to values from "net/base/net_errors.h"
-int MapPosixError(int err) {
+static int MapPosixError(int err) {
   // There are numerous posix error codes, but these are the ones we thus far
   // find interesting.
   switch (err) {
@@ -66,8 +64,6 @@ int MapPosixError(int err) {
   }
 }
 
-}  // namespace
-
 //-----------------------------------------------------------------------------
 
 TCPClientSocketLibevent::TCPClientSocketLibevent(const AddressList& addresses)
@@ -75,8 +71,6 @@ TCPClientSocketLibevent::TCPClientSocketLibevent(const AddressList& addresses)
       addresses_(addresses),
       current_ai_(addresses_.head()),
       waiting_connect_(false),
-      read_watcher_(this),
-      write_watcher_(this),
       read_callback_(NULL),
       write_callback_(NULL) {
 }
@@ -117,12 +111,12 @@ int TCPClientSocketLibevent::Connect(CompletionCallback* callback) {
     return MapPosixError(errno);
   }
 
-  // Initialize write_socket_watcher_ and link it to our MessagePump.
+  // Initialize socket_watcher_ and link it to our MessagePump.
   // POLLOUT is set if the connection is established.
   // POLLIN is set if the connection fails.
   if (!MessageLoopForIO::current()->WatchFileDescriptor(
-          socket_, true, MessageLoopForIO::WATCH_WRITE, &write_socket_watcher_,
-          &write_watcher_)) {
+          socket_, true, MessageLoopForIO::WATCH_WRITE, &socket_watcher_,
+          this)) {
     DLOG(INFO) << "WatchFileDescriptor failed: " << errno;
     close(socket_);
     socket_ = kInvalidSocket;
@@ -130,7 +124,7 @@ int TCPClientSocketLibevent::Connect(CompletionCallback* callback) {
   }
 
   waiting_connect_ = true;
-  write_callback_ = callback;
+  read_callback_ = callback;
   return ERR_IO_PENDING;
 }
 
@@ -140,10 +134,7 @@ void TCPClientSocketLibevent::Disconnect() {
 
   TRACE_EVENT_INSTANT("socket.disconnect", this, "");
 
-  bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
-  DCHECK(ok);
-  ok = write_socket_watcher_.StopWatchingFileDescriptor();
-  DCHECK(ok);
+  socket_watcher_.StopWatchingFileDescriptor();
   close(socket_);
   socket_ = kInvalidSocket;
   waiting_connect_ = false;
@@ -206,7 +197,7 @@ int TCPClientSocketLibevent::Read(IOBuffer* buf,
 
   if (!MessageLoopForIO::current()->WatchFileDescriptor(
           socket_, true, MessageLoopForIO::WATCH_READ,
-          &read_socket_watcher_, &read_watcher_)) {
+          &socket_watcher_, this)) {
     DLOG(INFO) << "WatchFileDescriptor failed on read, errno " << errno;
     return MapPosixError(errno);
   }
@@ -238,7 +229,7 @@ int TCPClientSocketLibevent::Write(IOBuffer* buf,
 
   if (!MessageLoopForIO::current()->WatchFileDescriptor(
           socket_, true, MessageLoopForIO::WATCH_WRITE,
-          &write_socket_watcher_, &write_watcher_)) {
+          &socket_watcher_, this)) {
     DLOG(INFO) << "WatchFileDescriptor failed on write, errno " << errno;
     return MapPosixError(errno);
   }
@@ -309,13 +300,12 @@ void TCPClientSocketLibevent::DidCompleteConnect() {
     result = Connect(read_callback_);
   } else {
     result = MapPosixError(error_code);
-    bool ok = write_socket_watcher_.StopWatchingFileDescriptor();
-    DCHECK(ok);
+    socket_watcher_.StopWatchingFileDescriptor();
     waiting_connect_ = false;
   }
 
   if (result != ERR_IO_PENDING) {
-    DoWriteCallback(result);
+    DoReadCallback(result);
   }
 }
 
@@ -336,8 +326,7 @@ void TCPClientSocketLibevent::DidCompleteRead() {
   if (result != ERR_IO_PENDING) {
     read_buf_ = NULL;
     read_buf_len_ = 0;
-    bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
-    DCHECK(ok);
+    socket_watcher_.StopWatchingFileDescriptor();
     DoReadCallback(result);
   }
 }
@@ -359,11 +348,27 @@ void TCPClientSocketLibevent::DidCompleteWrite() {
   if (result != ERR_IO_PENDING) {
     write_buf_ = NULL;
     write_buf_len_ = 0;
-    write_socket_watcher_.StopWatchingFileDescriptor();
+    socket_watcher_.StopWatchingFileDescriptor();
     DoWriteCallback(result);
   }
 }
 
+void TCPClientSocketLibevent::OnFileCanReadWithoutBlocking(int fd) {
+  // When a socket connects it signals both Read and Write, we handle
+  // DidCompleteConnect() in the write handler.
+  if (!waiting_connect_ && read_callback_) {
+    DidCompleteRead();
+  }
+}
+
+void TCPClientSocketLibevent::OnFileCanWriteWithoutBlocking(int fd) {
+  if (waiting_connect_) {
+    DidCompleteConnect();
+  } else if (write_callback_) {
+    DidCompleteWrite();
+  }
+}
+
 int TCPClientSocketLibevent::GetPeerName(struct sockaddr *name,
                                          socklen_t *namelen) {
   return ::getpeername(socket_, name, namelen);
diff --git a/net/base/tcp_client_socket_libevent.h b/net/base/tcp_client_socket_libevent.h
index 041ca7c..4b3dd45 100644
--- a/net/base/tcp_client_socket_libevent.h
+++ b/net/base/tcp_client_socket_libevent.h
@@ -8,8 +8,6 @@
 #include <sys/socket.h>  // for struct sockaddr
 
 #include "base/message_loop.h"
-#include "base/ref_counted.h"
-#include "base/scoped_ptr.h"
 #include "net/base/address_list.h"
 #include "net/base/client_socket.h"
 #include "net/base/completion_callback.h"
@@ -19,14 +17,15 @@ struct event;  // From libevent
 namespace net {
 
 // A client socket that uses TCP as the transport layer.
-class TCPClientSocketLibevent : public ClientSocket {
+class TCPClientSocketLibevent : public ClientSocket,
+                                public MessageLoopForIO::Watcher {
  public:
   // The IP address(es) and port number to connect to.  The TCP socket will try
   // each IP address in the list until it succeeds in establishing a
   // connection.
   explicit TCPClientSocketLibevent(const AddressList& addresses);
 
-  virtual ~TCPClientSocketLibevent();
+  ~TCPClientSocketLibevent();
 
   // ClientSocket methods:
   virtual int Connect(CompletionCallback* callback);
@@ -45,46 +44,9 @@ class TCPClientSocketLibevent : public ClientSocket {
   virtual int GetPeerName(struct sockaddr *name, socklen_t *namelen);
 
  private:
-  class ReadWatcher : public MessageLoopForIO::Watcher {
-   public:
-    explicit ReadWatcher(TCPClientSocketLibevent* socket) : socket_(socket) {}
-
-    // MessageLoopForIO::Watcher methods
-
-    virtual void OnFileCanReadWithoutBlocking(int /* fd */) {
-      if (socket_->read_callback_)
-        socket_->DidCompleteRead();
-    }
-
-    virtual void OnFileCanWriteWithoutBlocking(int /* fd */) {}
-
-   private:
-    TCPClientSocketLibevent* const socket_;
-
-    DISALLOW_COPY_AND_ASSIGN(ReadWatcher);
-  };
-
-  class WriteWatcher : public MessageLoopForIO::Watcher {
-   public:
-    explicit WriteWatcher(TCPClientSocketLibevent* socket) : socket_(socket) {}
-
-    // MessageLoopForIO::Watcher methods
-
-    virtual void OnFileCanReadWithoutBlocking(int /* fd */) {}
-
-    virtual void OnFileCanWriteWithoutBlocking(int /* fd */) {
-      if (socket_->waiting_connect_) {
-        socket_->DidCompleteConnect();
-      } else if (socket_->write_callback_) {
-        socket_->DidCompleteWrite();
-      }
-    }
-
-   private:
-    TCPClientSocketLibevent* const socket_;
-
-    DISALLOW_COPY_AND_ASSIGN(WriteWatcher);
-  };
+  // Called by MessagePumpLibevent when the socket is ready to do I/O
+  void OnFileCanReadWithoutBlocking(int fd);
+  void OnFileCanWriteWithoutBlocking(int fd);
 
   void DoReadCallback(int rv);
   void DoWriteCallback(int rv);
@@ -105,13 +67,8 @@ class TCPClientSocketLibevent : public ClientSocket {
   // Whether we're currently waiting for connect() to complete
   bool waiting_connect_;
 
-  // The socket's libevent wrappers
-  MessageLoopForIO::FileDescriptorWatcher read_socket_watcher_;
-  MessageLoopForIO::FileDescriptorWatcher write_socket_watcher_;
-
-  // The corresponding watchers for reads and writes.
-  ReadWatcher read_watcher_;
-  WriteWatcher write_watcher_;
+  // The socket's libevent wrapper
+  MessageLoopForIO::FileDescriptorWatcher socket_watcher_;
 
   // The buffer used by OnSocketReady to retry Read requests
   scoped_refptr<IOBuffer> read_buf_;
diff --git a/net/base/tcp_client_socket_unittest.cc b/net/base/tcp_client_socket_unittest.cc
index abd1c0d..3fbb02c 100644
--- a/net/base/tcp_client_socket_unittest.cc
+++ b/net/base/tcp_client_socket_unittest.cc
@@ -2,24 +2,16 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "net/base/tcp_client_socket.h"
-
-#include "base/basictypes.h"
 #include "net/base/address_list.h"
 #include "net/base/host_resolver.h"
 #include "net/base/listen_socket.h"
 #include "net/base/net_errors.h"
+#include "net/base/tcp_client_socket.h"
 #include "net/base/test_completion_callback.h"
 #include "net/base/winsock_init.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "testing/platform_test.h"
 
-namespace net {
-
-namespace {
-
-const char kServerReply[] = "HTTP/1.1 404 Not Found";
-
 class TCPClientSocketTest
     : public PlatformTest, public ListenSocket::ListenSocketDelegate {
  public:
@@ -32,31 +24,18 @@ class TCPClientSocketTest
   }
   virtual void DidRead(ListenSocket*, const std::string& s) {
     // TODO(dkegel): this might not be long enough to tickle some bugs.
-    connected_sock_->Send(kServerReply,
-                          arraysize(kServerReply) - 1,
-                          false /* don't append line feed */);
+    connected_sock_->Send(std::string("HTTP/1.1 404 Not Found"), true);
+    // Close socket by destroying it, else read test below will hang.
+    connected_sock_ = NULL;
   }
   virtual void DidClose(ListenSocket* sock) {}
 
   // Testcase hooks
   virtual void SetUp();
 
-  void CloseServerSocket() {
-    // delete the connected_sock_, which will close it.
-    connected_sock_ = NULL;
-  }
-
-  void PauseServerReads() {
-    connected_sock_->PauseReads();
-  }
-
-  void ResumeServerReads() {
-    connected_sock_->ResumeReads();
-  }
-
  protected:
   int listen_port_;
-  scoped_ptr<TCPClientSocket> sock_;
+  scoped_ptr<net::TCPClientSocket> sock_;
 
  private:
   scoped_refptr<ListenSocket> listen_sock_;
@@ -73,7 +52,7 @@ void TCPClientSocketTest::SetUp() {
   const int kMinPort = 10100;
   const int kMaxPort = 10200;
 #if defined(OS_WIN)
-  EnsureWinsockInit();
+  net::EnsureWinsockInit();
 #endif
   for (port = kMinPort; port < kMaxPort; port++) {
     sock = ListenSocket::Listen("127.0.0.1", port, this);
@@ -84,11 +63,11 @@ void TCPClientSocketTest::SetUp() {
   listen_sock_ = sock;
   listen_port_ = port;
 
-  AddressList addr;
-  HostResolver resolver;
+  net::AddressList addr;
+  net::HostResolver resolver;
   int rv = resolver.Resolve("localhost", listen_port_, &addr, NULL);
-  CHECK(rv == OK);
-  sock_.reset(new TCPClientSocket(addr));
+  CHECK(rv == net::OK);
+  sock_.reset(new net::TCPClientSocket(addr));
 }
 
 TEST_F(TCPClientSocketTest, Connect) {
@@ -96,11 +75,11 @@ TEST_F(TCPClientSocketTest, Connect) {
   EXPECT_FALSE(sock_->IsConnected());
 
   int rv = sock_->Connect(&callback);
-  if (rv != OK) {
-    ASSERT_EQ(rv, ERR_IO_PENDING);
+  if (rv != net::OK) {
+    ASSERT_EQ(rv, net::ERR_IO_PENDING);
 
     rv = callback.WaitForResult();
-    EXPECT_EQ(rv, OK);
+    EXPECT_EQ(rv, net::OK);
   }
 
   EXPECT_TRUE(sock_->IsConnected());
@@ -116,222 +95,197 @@ TEST_F(TCPClientSocketTest, Connect) {
 TEST_F(TCPClientSocketTest, Read) {
   TestCompletionCallback callback;
   int rv = sock_->Connect(&callback);
-  if (rv != OK) {
-    ASSERT_EQ(rv, ERR_IO_PENDING);
+  if (rv != net::OK) {
+    ASSERT_EQ(rv, net::ERR_IO_PENDING);
 
     rv = callback.WaitForResult();
-    EXPECT_EQ(rv, OK);
+    EXPECT_EQ(rv, net::OK);
   }
 
   const char request_text[] = "GET / HTTP/1.0\r\n\r\n";
-  scoped_refptr<IOBuffer> request_buffer =
-      new IOBuffer(arraysize(request_text) - 1);
+  scoped_refptr<net::IOBuffer> request_buffer =
+      new net::IOBuffer(arraysize(request_text) - 1);
   memcpy(request_buffer->data(), request_text, arraysize(request_text) - 1);
 
   rv = sock_->Write(request_buffer, arraysize(request_text) - 1, &callback);
-  EXPECT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+  EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-  if (rv == ERR_IO_PENDING) {
+  if (rv == net::ERR_IO_PENDING) {
     rv = callback.WaitForResult();
     EXPECT_EQ(rv, static_cast<int>(arraysize(request_text) - 1));
   }
 
-  scoped_refptr<IOBuffer> buf = new IOBuffer(4096);
-  uint32 bytes_read = 0;
-  while (bytes_read < arraysize(kServerReply) - 1) {
+  scoped_refptr<net::IOBuffer> buf = new net::IOBuffer(4096);
+  for (;;) {
     rv = sock_->Read(buf, 4096, &callback);
-    EXPECT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+    EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-    if (rv == ERR_IO_PENDING)
+    if (rv == net::ERR_IO_PENDING)
       rv = callback.WaitForResult();
 
-    ASSERT_GE(rv, 0);
-    bytes_read += rv;
+    EXPECT_GE(rv, 0);
+    if (rv <= 0)
+      break;
   }
-
-  // All data has been read now.  Read once more to force an ERR_IO_PENDING, and
-  // then close the server socket, and note the close.
-
-  rv = sock_->Read(buf, 4096, &callback);
-  ASSERT_EQ(ERR_IO_PENDING, rv);
-  CloseServerSocket();
-  EXPECT_EQ(0, callback.WaitForResult());
 }
 
 TEST_F(TCPClientSocketTest, Read_SmallChunks) {
   TestCompletionCallback callback;
   int rv = sock_->Connect(&callback);
-  if (rv != OK) {
-    ASSERT_EQ(rv, ERR_IO_PENDING);
+  if (rv != net::OK) {
+    ASSERT_EQ(rv, net::ERR_IO_PENDING);
 
     rv = callback.WaitForResult();
-    EXPECT_EQ(rv, OK);
+    EXPECT_EQ(rv, net::OK);
   }
 
   const char request_text[] = "GET / HTTP/1.0\r\n\r\n";
-  scoped_refptr<IOBuffer> request_buffer =
-      new IOBuffer(arraysize(request_text) - 1);
+  scoped_refptr<net::IOBuffer> request_buffer =
+      new net::IOBuffer(arraysize(request_text) - 1);
   memcpy(request_buffer->data(), request_text, arraysize(request_text) - 1);
 
   rv = sock_->Write(request_buffer, arraysize(request_text) - 1, &callback);
-  EXPECT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+  EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-  if (rv == ERR_IO_PENDING) {
+  if (rv == net::ERR_IO_PENDING) {
     rv = callback.WaitForResult();
     EXPECT_EQ(rv, static_cast<int>(arraysize(request_text) - 1));
   }
 
-  scoped_refptr<IOBuffer> buf = new IOBuffer(1);
-  uint32 bytes_read = 0;
-  while (bytes_read < arraysize(kServerReply) - 1) {
+  scoped_refptr<net::IOBuffer> buf = new net::IOBuffer(1);
+  for (;;) {
     rv = sock_->Read(buf, 1, &callback);
-    EXPECT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+    EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-    if (rv == ERR_IO_PENDING)
+    if (rv == net::ERR_IO_PENDING)
       rv = callback.WaitForResult();
 
-    ASSERT_EQ(1, rv);
-    bytes_read += rv;
+    EXPECT_GE(rv, 0);
+    if (rv <= 0)
+      break;
   }
-
-  // All data has been read now.  Read once more to force an ERR_IO_PENDING, and
-  // then close the server socket, and note the close.
-
-  rv = sock_->Read(buf, 1, &callback);
-  ASSERT_EQ(ERR_IO_PENDING, rv);
-  CloseServerSocket();
-  EXPECT_EQ(0, callback.WaitForResult());
 }
 
 TEST_F(TCPClientSocketTest, Read_Interrupted) {
   TestCompletionCallback callback;
   int rv = sock_->Connect(&callback);
-  if (rv != OK) {
-    ASSERT_EQ(ERR_IO_PENDING, rv);
+  if (rv != net::OK) {
+    ASSERT_EQ(rv, net::ERR_IO_PENDING);
 
     rv = callback.WaitForResult();
-    EXPECT_EQ(rv, OK);
+    EXPECT_EQ(rv, net::OK);
   }
 
   const char request_text[] = "GET / HTTP/1.0\r\n\r\n";
-  scoped_refptr<IOBuffer> request_buffer =
-      new IOBuffer(arraysize(request_text) - 1);
+  scoped_refptr<net::IOBuffer> request_buffer =
+      new net::IOBuffer(arraysize(request_text) - 1);
   memcpy(request_buffer->data(), request_text, arraysize(request_text) - 1);
 
   rv = sock_->Write(request_buffer, arraysize(request_text) - 1, &callback);
-  EXPECT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+  EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-  if (rv == ERR_IO_PENDING) {
+  if (rv == net::ERR_IO_PENDING) {
     rv = callback.WaitForResult();
     EXPECT_EQ(rv, static_cast<int>(arraysize(request_text) - 1));
   }
 
   // Do a partial read and then exit.  This test should not crash!
-  scoped_refptr<IOBuffer> buf = new IOBuffer(16);
-  rv = sock_->Read(buf, 16, &callback);
-  EXPECT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+  scoped_refptr<net::IOBuffer> buf = new net::IOBuffer(512);
+  rv = sock_->Read(buf, 512, &callback);
+  EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-  if (rv == ERR_IO_PENDING)
+  if (rv == net::ERR_IO_PENDING)
     rv = callback.WaitForResult();
 
-  EXPECT_NE(0, rv);
+  EXPECT_NE(rv, 0);
 }
 
-TEST_F(TCPClientSocketTest, DISABLED_FullDuplex_ReadFirst) {
+TEST_F(TCPClientSocketTest, FullDuplex_ReadFirst) {
   TestCompletionCallback callback;
   int rv = sock_->Connect(&callback);
-  if (rv != OK) {
-    ASSERT_EQ(rv, ERR_IO_PENDING);
+  if (rv != net::OK) {
+    ASSERT_EQ(rv, net::ERR_IO_PENDING);
 
     rv = callback.WaitForResult();
-    EXPECT_EQ(rv, OK);
+    EXPECT_EQ(rv, net::OK);
   }
 
-  // Read first.  There's no data, so it should return ERR_IO_PENDING.
   const int kBufLen = 4096;
-  scoped_refptr<IOBuffer> buf = new IOBuffer(kBufLen);
+  scoped_refptr<net::IOBuffer> buf = new net::IOBuffer(kBufLen);
   rv = sock_->Read(buf, kBufLen, &callback);
-  EXPECT_EQ(ERR_IO_PENDING, rv);
+  EXPECT_EQ(net::ERR_IO_PENDING, rv);
 
-  PauseServerReads();
-  const int kWriteBufLen = 64 * 1024;
-  scoped_refptr<IOBuffer> request_buffer = new IOBuffer(kWriteBufLen);
-  char* request_data = request_buffer->data();
-  memset(request_data, 'A', kWriteBufLen);
+  const char request_text[] = "GET / HTTP/1.0\r\n\r\n";
+  scoped_refptr<net::IOBuffer> request_buffer =
+      new net::IOBuffer(arraysize(request_text) - 1);
+  memcpy(request_buffer->data(), request_text, arraysize(request_text) - 1);
   TestCompletionCallback write_callback;
 
-  while (true) {
-    rv = sock_->Write(request_buffer, kWriteBufLen, &write_callback);
-    ASSERT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
+  rv = sock_->Write(request_buffer, arraysize(request_text) - 1,
+                    &write_callback);
+  EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-    if (rv == ERR_IO_PENDING) {
-      ResumeServerReads();
-      rv = write_callback.WaitForResult();
-      break;
-    }
+  if (rv == net::ERR_IO_PENDING) {
+    rv = write_callback.WaitForResult();
+    EXPECT_EQ(rv, static_cast<int>(arraysize(request_text) - 1));
   }
 
-  // At this point, both read and write have returned ERR_IO_PENDING, and the
-  // write callback has executed.  We wait for the read callback to run now to
-  // make sure that the socket can handle full duplex communications.
-
   rv = callback.WaitForResult();
   EXPECT_GE(rv, 0);
+  while (rv > 0) {
+    rv = sock_->Read(buf, kBufLen, &callback);
+    EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
+
+    if (rv == net::ERR_IO_PENDING)
+      rv = callback.WaitForResult();
+
+    EXPECT_GE(rv, 0);
+    if (rv <= 0)
+      break;
+  }
 }
 
-TEST_F(TCPClientSocketTest, DISABLED_FullDuplex_WriteFirst) {
+TEST_F(TCPClientSocketTest, FullDuplex_WriteFirst) {
   TestCompletionCallback callback;
   int rv = sock_->Connect(&callback);
-  if (rv != OK) {
-    ASSERT_EQ(ERR_IO_PENDING, rv);
+  if (rv != net::OK) {
+    ASSERT_EQ(rv, net::ERR_IO_PENDING);
 
     rv = callback.WaitForResult();
-    EXPECT_EQ(OK, rv);
+    EXPECT_EQ(rv, net::OK);
   }
 
-  PauseServerReads();
-  const int kWriteBufLen = 64 * 1024;
-  scoped_refptr<IOBuffer> request_buffer = new IOBuffer(kWriteBufLen);
-  char* request_data = request_buffer->data();
-  memset(request_data, 'A', kWriteBufLen);
+  const char request_text[] = "GET / HTTP/1.0\r\n\r\n";
+  scoped_refptr<net::IOBuffer> request_buffer =
+      new net::IOBuffer(arraysize(request_text) - 1);
+  memcpy(request_buffer->data(), request_text, arraysize(request_text) - 1);
   TestCompletionCallback write_callback;
 
-  while (true) {
-    rv = sock_->Write(request_buffer, kWriteBufLen, &write_callback);
-    ASSERT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
-
-    if (rv == ERR_IO_PENDING)
-      break;
-  }
-
-  // Now we have the Write() blocked on ERR_IO_PENDING.  It's time to force the
-  // Read() to block on ERR_IO_PENDING too.
+  rv = sock_->Write(request_buffer, arraysize(request_text) - 1,
+                    &write_callback);
+  EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
   const int kBufLen = 4096;
-  scoped_refptr<IOBuffer> buf = new IOBuffer(kBufLen);
-  while (true) {
-    rv = sock_->Read(buf, kBufLen, &callback);
-    ASSERT_TRUE(rv >= 0 || rv == ERR_IO_PENDING);
-    if (rv == ERR_IO_PENDING)
-      break;
-  }
-
-  // At this point, both read and write have returned ERR_IO_PENDING.  Now we
-  // run the write and read callbacks to make sure they can handle full duplex
-  // communications.
+  scoped_refptr<net::IOBuffer> buf = new net::IOBuffer(kBufLen);
+  int read_rv = sock_->Read(buf, kBufLen, &callback);
+  EXPECT_TRUE(read_rv >= 0 || read_rv == net::ERR_IO_PENDING);
 
-  ResumeServerReads();
-  rv = write_callback.WaitForResult();
-  EXPECT_GE(rv, 0);
-
-  // It's possible the read is blocked because it's already read all the data.
-  // Close the server socket, so there will at least be a 0-byte read.
-  CloseServerSocket();
+  if (rv == net::ERR_IO_PENDING) {
+    rv = write_callback.WaitForResult();
+    EXPECT_EQ(static_cast<int>(arraysize(request_text) - 1), rv);
+  }
 
   rv = callback.WaitForResult();
   EXPECT_GE(rv, 0);
-}
+  while (rv > 0) {
+    rv = sock_->Read(buf, kBufLen, &callback);
+    EXPECT_TRUE(rv >= 0 || rv == net::ERR_IO_PENDING);
 
-}  // namespace
+    if (rv == net::ERR_IO_PENDING)
+      rv = callback.WaitForResult();
 
-}  // namespace net
+    EXPECT_GE(rv, 0);
+    if (rv <= 0)
+      break;
+  }
+}