Add P2PDatagramSocket and P2PStreamSocket interfaces.

remoting/base/buffered_socket_writer.cc

 // Copyright 2015 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 "remoting/base/buffered_socket_writer.h"
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/thread_task_runner_handle.h"
+#include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
+#include "net/socket/socket.h"
 
 namespace remoting {
 
-struct BufferedSocketWriterBase::PendingPacket {
-  PendingPacket(scoped_refptr<net::IOBufferWithSize> data,
+namespace {
+
+int WriteNetSocket(net::Socket* socket,
+                   const scoped_refptr<net::IOBuffer>& buf,
+                   int buf_len,
+                   const net::CompletionCallback& callback) {
+  return socket->Write(buf.get(), buf_len, callback);
+}
+
+}  // namespace
+
+struct BufferedSocketWriter::PendingPacket {
+  PendingPacket(scoped_refptr<net::DrainableIOBuffer> data,
                 const base::Closure& done_task)
       : data(data),
         done_task(done_task) {
   }
 
-  scoped_refptr<net::IOBufferWithSize> data;
+  scoped_refptr<net::DrainableIOBuffer> data;
   base::Closure done_task;
 };
 
-BufferedSocketWriterBase::BufferedSocketWriterBase()
-    : socket_(nullptr),
-      write_pending_(false),
-      closed_(false),
-      destroyed_flag_(nullptr) {
+// static
+scoped_ptr<BufferedSocketWriter> BufferedSocketWriter::CreateForSocket(
+    net::Socket* socket, const WriteFailedCallback& write_failed_callback) {
+  scoped_ptr<BufferedSocketWriter> result(new BufferedSocketWriter());
+  result->Init(base::Bind(&WriteNetSocket, socket), write_failed_callback);
+  return result.Pass();
 }
 
-void BufferedSocketWriterBase::Init(net::Socket* socket,
-                                    const WriteFailedCallback& callback) {
-  DCHECK(CalledOnValidThread());
-  DCHECK(socket);
-  socket_ = socket;
-  write_failed_callback_ = callback;
+BufferedSocketWriter::BufferedSocketWriter() : weak_factory_(this) {
 }
 
-bool BufferedSocketWriterBase::Write(
-    scoped_refptr<net::IOBufferWithSize> data, const base::Closure& done_task) {
-  DCHECK(CalledOnValidThread());
-  DCHECK(socket_);
+BufferedSocketWriter::~BufferedSocketWriter() {
+  STLDeleteElements(&queue_);
+}
+
+void BufferedSocketWriter::Init(
+    const WriteCallback& write_callback,
+    const WriteFailedCallback& write_failed_callback) {
+  write_callback_ = write_callback;
+  write_failed_callback_ = write_failed_callback;
+}
+
+bool BufferedSocketWriter::Write(
+    const scoped_refptr<net::IOBufferWithSize>& data,
+    const base::Closure& done_task) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(data.get());
 
-  // Don't write after Close().
-  if (closed_)
+  // Don't write after error.
+  if (is_closed())
     return false;
 
-  queue_.push_back(new PendingPacket(data, done_task));
+  queue_.push_back(new PendingPacket(
+      new net::DrainableIOBuffer(data.get(), data->size()), done_task));
 
   DoWrite();
 
-  // DoWrite() may trigger OnWriteError() to be called.
-  return !closed_;
+  return !is_closed();
 }
 
-void BufferedSocketWriterBase::DoWrite() {
-  DCHECK(CalledOnValidThread());
-  DCHECK(socket_);
+bool BufferedSocketWriter::is_closed() {
+  return write_callback_.is_null();
+}
+
+void BufferedSocketWriter::DoWrite() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!write_callback_.is_null());
 
   // Don't try to write if there is another write pending.
   if (write_pending_)
     return;
 
-  // Don't write after Close().
-  if (closed_)
+  // Don't write after error.
+  if (is_closed())
     return;
 
   while (true) {
-    net::IOBuffer* current_packet;
-    int current_packet_size;
-    GetNextPacket(&current_packet, &current_packet_size);
+    if (queue_.empty())
+      break;
 
-    // Return if the queue is empty.
-    if (!current_packet)
-      return;
-
-    int result = socket_->Write(
-        current_packet, current_packet_size,
-        base::Bind(&BufferedSocketWriterBase::OnWritten,
-                   base::Unretained(this)));
+    int result = write_callback_.Run(
+        queue_.front()->data.get(), queue_.front()->data->size(),
+        base::Bind(&BufferedSocketWriter::OnWritten,
+                   weak_factory_.GetWeakPtr()));
     bool write_again = false;
     HandleWriteResult(result, &write_again);
     if (!write_again)
       return;
   }
 }
 
-void BufferedSocketWriterBase::HandleWriteResult(int result,
-                                                 bool* write_again) {
+void BufferedSocketWriter::HandleWriteResult(int result, bool* write_again) {
   *write_again = false;
   if (result < 0) {
     if (result == net::ERR_IO_PENDING) {
       write_pending_ = true;
     } else {
-      HandleError(result);
+      write_callback_.Reset();
       if (!write_failed_callback_.is_null())
         write_failed_callback_.Run(result);
     }
     return;
   }
 
-  base::Closure done_task = AdvanceBufferPosition(result);
-  if (!done_task.is_null()) {
-    bool destroyed = false;
-    destroyed_flag_ = &destroyed;
-    done_task.Run();
-    if (destroyed) {
-      // Stop doing anything if we've been destroyed by the callback.
+  DCHECK(!queue_.empty());
+
+  queue_.front()->data->DidConsume(result);
+
+  if (queue_.front()->data->BytesRemaining() == 0) {
+    base::Closure done_task = queue_.front()->done_task;
+    delete queue_.front();
+    queue_.pop_front();
+
+    if (!done_task.is_null()) {
+      // |done_task| is allowed to delete the writer, so here we post a task to
+      // continue writing after calling |done_task| and return with
+      // |write_again| set to false.
+      base::ThreadTaskRunnerHandle::Get()->PostTask(
+          FROM_HERE, base::Bind(&BufferedSocketWriter::DoWrite,
+                                weak_factory_.GetWeakPtr()));

[Wez, 2015/07/14 21:49:20]

Now that you're using WeakPtr you don't even need to PostTask - you can just
take a copy of the WeakPtr and test it immediately after running the
|done_task|, to decide whether to call DoWrite() before you return?

If you're going to call DoWrite() here as I suggested, then you don't want a
write loop in DoWrite() itself - just let it call HandleWriteResult and let that
call back round to DoWrite().

[Sergey Ulanov, 2015/07/14 22:53:20]

We don't want to call DoWrite() directly here because it can potentially
overflow the stack. I.e. DoWrite() calls HandleWriteResult() and
HandleWriteResult() will be calling DoWrite()

[Wez, 2015/07/16 21:55:36]

If it's the last thing that HandleWriteResult() calls then won't the compiler
optimize that to use tail recursion, though?

[Sergey Ulanov, 2015/07/16 22:11:03]

AFAIK some C++ compilers implement tail call optimization, but it's not required
by the standard so we cannot rely on it here. E.g. it's not enabled in debug
builds. It also may be broken by other changes, e.g. making this method virtual.

[Sergey Ulanov, 2015/07/16 22:25:57]

Also there are cases when tail call optimization is just not possible in C++,
e.g.:

int foo() {
  A a;
  return bar();
}

in this case ~A() must be called after bar().

+      done_task.Run();
       return;
     }
-    destroyed_flag_ = nullptr;
   }
 
   *write_again = true;

[Wez, 2015/07/14 21:49:20]

Why is |write_again| an out parameter rather than a return value? Or, actually
with DoWrite() called in HandleWriteResult you can get ridof this entirely, I
think?

[Sergey Ulanov, 2015/07/14 22:53:20]

Just for readability, to make the purpose of the result clear.
Done, but this comes at cost of posting a task every time in
HandleWriteResult(), even when it can be avoided.

[Wez, 2015/07/16 21:55:36]

You don't need the PostTask, though - you can just take a WeakPtr before you
call |done_task| and then check it afterward, and if it's still valid, then call
on to DoWrite() directly, as the last thing you do, so that the compiler can
optimize it to a tail recursion?

The alternative would be to have HandleWriteResult() return a bool that is false
if the caller should immediately pop its stack - much like the previous
|write_again|, except that you'd have HandleWriteResult return |weak_ptr_|
coerced to bool in the case where there |might| be something more to write. So
basically like the old code but without the strange out-value and the
pointer-to-bool hack.

[Sergey Ulanov, 2015/07/16 22:11:03]

So if I understand correctly you are suggesting that HandleWriteResult() should
return a bool and a weak_ptr. Is that right interpretation?

 }
 
-void BufferedSocketWriterBase::OnWritten(int result) {
-  DCHECK(CalledOnValidThread());
+void BufferedSocketWriter::OnWritten(int result) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(write_pending_);
   write_pending_ = false;
 
   bool write_again;
   HandleWriteResult(result, &write_again);
   if (write_again)
     DoWrite();

[Wez, 2015/07/14 21:49:20]

As noted above, if you have HandleWriteResult() call DoWrite via WeakPtr then
you don't need this.

[Sergey Ulanov, 2015/07/14 22:53:21]

Done.

 }
 
-void BufferedSocketWriterBase::HandleError(int result) {
-  DCHECK(CalledOnValidThread());
-
-  closed_ = true;
-
-  STLDeleteElements(&queue_);
-
-  // Notify subclass that an error is received.
-  OnError(result);
-}
-
-void BufferedSocketWriterBase::Close() {
-  DCHECK(CalledOnValidThread());
-  closed_ = true;
-}
-
-BufferedSocketWriterBase::~BufferedSocketWriterBase() {
-  if (destroyed_flag_)
-    *destroyed_flag_ = true;
-
-  STLDeleteElements(&queue_);
-}
-
-base::Closure BufferedSocketWriterBase::PopQueue() {
-  base::Closure result = queue_.front()->done_task;
-  delete queue_.front();
-  queue_.pop_front();
-  return result;
-}
-
-BufferedSocketWriter::BufferedSocketWriter() {
-}
-
-void BufferedSocketWriter::GetNextPacket(
-    net::IOBuffer** buffer, int* size) {
-  if (!current_buf_.get()) {
-    if (queue_.empty()) {
-      *buffer = nullptr;
-      return;  // Nothing to write.
-    }
-    current_buf_ = new net::DrainableIOBuffer(queue_.front()->data.get(),
-                                              queue_.front()->data->size());
-  }
-
-  *buffer = current_buf_.get();
-  *size = current_buf_->BytesRemaining();
-}
-
-base::Closure BufferedSocketWriter::AdvanceBufferPosition(int written) {
-  current_buf_->DidConsume(written);
-
-  if (current_buf_->BytesRemaining() == 0) {
-    current_buf_ = nullptr;
-    return PopQueue();
-  }
-  return base::Closure();
-}
-
-void BufferedSocketWriter::OnError(int result) {
-  current_buf_ = nullptr;
-}
-
-BufferedSocketWriter::~BufferedSocketWriter() {
-}
-
-BufferedDatagramWriter::BufferedDatagramWriter() {
-}
-
-void BufferedDatagramWriter::GetNextPacket(
-    net::IOBuffer** buffer, int* size) {
-  if (queue_.empty()) {
-    *buffer = nullptr;
-    return;  // Nothing to write.
-  }
-  *buffer = queue_.front()->data.get();
-  *size = queue_.front()->data->size();
-}
-
-base::Closure BufferedDatagramWriter::AdvanceBufferPosition(int written) {
-  DCHECK_EQ(written, queue_.front()->data->size());
-  return PopQueue();
-}
-
-void BufferedDatagramWriter::OnError(int result) {
-  // Nothing to do here.
-}
-
-BufferedDatagramWriter::~BufferedDatagramWriter() {
-}
-
 }  // namespace remoting