Move FifoBuffer to its own file and build target

Used only by test code and by pseudo_tcp.

Bug: webrtc:6424
Change-Id: I28903e74f7b69cbdd8c368f4444c8a233eb76868
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/128868
Reviewed-by: Karl Wiberg <[email protected]>
Reviewed-by: Mirko Bonadei <[email protected]>
Commit-Queue: Niels Moller <[email protected]>
Cr-Commit-Position: refs/heads/master@{#27341}

Author: Niels Möller

Diff for: p2p/BUILD.gn

@@ -92,8 +92,11 @@ rtc_static_library("rtc_p2p") {
     "../logging:rtc_event_log_api",
     "../rtc_base",
     "../rtc_base:checks",
+
+    # Needed by pseudo_tcp, which should move to a separate target.
     "../rtc_base:safe_minmax",
     "../rtc_base:weak_ptr",
+    "../rtc_base/memory:fifo_buffer",
     "../rtc_base/network:sent_packet",
     "../rtc_base/system:rtc_export",
     "../rtc_base/third_party/base64",

Diff for: p2p/base/pseudo_tcp.h

@@ -15,7 +15,7 @@
 #include <stdint.h>
 #include <list>
 
-#include "rtc_base/stream.h"
+#include "rtc_base/memory/fifo_buffer.h"
 #include "rtc_base/system/rtc_export.h"
 
 namespace cricket {

Diff for: rtc_base/BUILD.gn

@@ -1106,6 +1106,7 @@ rtc_source_set("rtc_base_tests_utils") {
     ":checks",
     ":rtc_base",
     "../api/units:time_delta",
+    "memory:fifo_buffer",
     "third_party/sigslot",
     "//third_party/abseil-cpp/absl/algorithm:container",
     "//third_party/abseil-cpp/absl/memory",
@@ -1383,7 +1384,6 @@ if (rtc_include_tests) {
       "rtc_certificate_unittest.cc",
       "signal_thread_unittest.cc",
       "sigslot_tester_unittest.cc",
-      "stream_unittest.cc",
       "test_client_unittest.cc",
       "thread_unittest.cc",
       "unique_id_generator_unittest.cc",
@@ -1416,6 +1416,7 @@ if (rtc_include_tests) {
       "../test:field_trial",
       "../test:fileutils",
       "../test:test_support",
+      "memory:fifo_buffer",
       "synchronization:synchronization_unittests",
       "third_party/sigslot",
       "//third_party/abseil-cpp/absl/algorithm:container",

Diff for: rtc_base/memory/BUILD.gn

@@ -30,15 +30,36 @@ rtc_source_set("aligned_malloc") {
   deps = []
 }
 
+rtc_source_set("fifo_buffer") {
+  visibility = [
+    "../../p2p:rtc_p2p",
+    "..:rtc_base_tests_utils",
+    "..:rtc_base_unittests",
+    ":unittests",
+  ]
+  sources = [
+    "fifo_buffer.cc",
+    "fifo_buffer.h",
+  ]
+  deps = [
+    "..:rtc_base",
+  ]
+  if (is_nacl) {
+    deps += [ "//native_client_sdk/src/libraries/nacl_io" ]
+  }
+}
+
 rtc_source_set("unittests") {
   testonly = true
   sources = [
     "aligned_array_unittest.cc",
     "aligned_malloc_unittest.cc",
+    "fifo_buffer_unittest.cc",
   ]
   deps = [
     ":aligned_array",
     ":aligned_malloc",
+    ":fifo_buffer",
     "../../test:test_support",
   ]
 }

Diff for: rtc_base/memory/fifo_buffer.cc

@@ -0,0 +1,247 @@
+/*
+ *  Copyright 2019 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.
+ */
+
+#include "rtc_base/memory/fifo_buffer.h"
+
+#include <algorithm>
+
+#include "rtc_base/thread.h"
+
+namespace rtc {
+
+FifoBuffer::FifoBuffer(size_t size)
+    : state_(SS_OPEN),
+      buffer_(new char[size]),
+      buffer_length_(size),
+      data_length_(0),
+      read_position_(0),
+      owner_(Thread::Current()) {
+  // all events are done on the owner_ thread
+}
+
+FifoBuffer::FifoBuffer(size_t size, Thread* owner)
+    : state_(SS_OPEN),
+      buffer_(new char[size]),
+      buffer_length_(size),
+      data_length_(0),
+      read_position_(0),
+      owner_(owner) {
+  // all events are done on the owner_ thread
+}
+
+FifoBuffer::~FifoBuffer() {}
+
+bool FifoBuffer::GetBuffered(size_t* size) const {
+  CritScope cs(&crit_);
+  *size = data_length_;
+  return true;
+}
+
+bool FifoBuffer::SetCapacity(size_t size) {
+  CritScope cs(&crit_);
+  if (data_length_ > size) {
+    return false;
+  }
+
+  if (size != buffer_length_) {
+    char* buffer = new char[size];
+    const size_t copy = data_length_;
+    const size_t tail_copy = std::min(copy, buffer_length_ - read_position_);
+    memcpy(buffer, &buffer_[read_position_], tail_copy);
+    memcpy(buffer + tail_copy, &buffer_[0], copy - tail_copy);
+    buffer_.reset(buffer);
+    read_position_ = 0;
+    buffer_length_ = size;
+  }
+  return true;
+}
+
+StreamResult FifoBuffer::ReadOffset(void* buffer,
+                                    size_t bytes,
+                                    size_t offset,
+                                    size_t* bytes_read) {
+  CritScope cs(&crit_);
+  return ReadOffsetLocked(buffer, bytes, offset, bytes_read);
+}
+
+StreamResult FifoBuffer::WriteOffset(const void* buffer,
+                                     size_t bytes,
+                                     size_t offset,
+                                     size_t* bytes_written) {
+  CritScope cs(&crit_);
+  return WriteOffsetLocked(buffer, bytes, offset, bytes_written);
+}
+
+StreamState FifoBuffer::GetState() const {
+  CritScope cs(&crit_);
+  return state_;
+}
+
+StreamResult FifoBuffer::Read(void* buffer,
+                              size_t bytes,
+                              size_t* bytes_read,
+                              int* error) {
+  CritScope cs(&crit_);
+  const bool was_writable = data_length_ < buffer_length_;
+  size_t copy = 0;
+  StreamResult result = ReadOffsetLocked(buffer, bytes, 0, &copy);
+
+  if (result == SR_SUCCESS) {
+    // If read was successful then adjust the read position and number of
+    // bytes buffered.
+    read_position_ = (read_position_ + copy) % buffer_length_;
+    data_length_ -= copy;
+    if (bytes_read) {
+      *bytes_read = copy;
+    }
+
+    // if we were full before, and now we're not, post an event
+    if (!was_writable && copy > 0) {
+      PostEvent(owner_, SE_WRITE, 0);
+    }
+  }
+  return result;
+}
+
+StreamResult FifoBuffer::Write(const void* buffer,
+                               size_t bytes,
+                               size_t* bytes_written,
+                               int* error) {
+  CritScope cs(&crit_);
+
+  const bool was_readable = (data_length_ > 0);
+  size_t copy = 0;
+  StreamResult result = WriteOffsetLocked(buffer, bytes, 0, &copy);
+
+  if (result == SR_SUCCESS) {
+    // If write was successful then adjust the number of readable bytes.
+    data_length_ += copy;
+    if (bytes_written) {
+      *bytes_written = copy;
+    }
+
+    // if we didn't have any data to read before, and now we do, post an event
+    if (!was_readable && copy > 0) {
+      PostEvent(owner_, SE_READ, 0);
+    }
+  }
+  return result;
+}
+
+void FifoBuffer::Close() {
+  CritScope cs(&crit_);
+  state_ = SS_CLOSED;
+}
+
+const void* FifoBuffer::GetReadData(size_t* size) {
+  CritScope cs(&crit_);
+  *size = (read_position_ + data_length_ <= buffer_length_)
+              ? data_length_
+              : buffer_length_ - read_position_;
+  return &buffer_[read_position_];
+}
+
+void FifoBuffer::ConsumeReadData(size_t size) {
+  CritScope cs(&crit_);
+  RTC_DCHECK(size <= data_length_);
+  const bool was_writable = data_length_ < buffer_length_;
+  read_position_ = (read_position_ + size) % buffer_length_;
+  data_length_ -= size;
+  if (!was_writable && size > 0) {
+    PostEvent(owner_, SE_WRITE, 0);
+  }
+}
+
+void* FifoBuffer::GetWriteBuffer(size_t* size) {
+  CritScope cs(&crit_);
+  if (state_ == SS_CLOSED) {
+    return nullptr;
+  }
+
+  // if empty, reset the write position to the beginning, so we can get
+  // the biggest possible block
+  if (data_length_ == 0) {
+    read_position_ = 0;
+  }
+
+  const size_t write_position =
+      (read_position_ + data_length_) % buffer_length_;
+  *size = (write_position > read_position_ || data_length_ == 0)
+              ? buffer_length_ - write_position
+              : read_position_ - write_position;
+  return &buffer_[write_position];
+}
+
+void FifoBuffer::ConsumeWriteBuffer(size_t size) {
+  CritScope cs(&crit_);
+  RTC_DCHECK(size <= buffer_length_ - data_length_);
+  const bool was_readable = (data_length_ > 0);
+  data_length_ += size;
+  if (!was_readable && size > 0) {
+    PostEvent(owner_, SE_READ, 0);
+  }
+}
+
+bool FifoBuffer::GetWriteRemaining(size_t* size) const {
+  CritScope cs(&crit_);
+  *size = buffer_length_ - data_length_;
+  return true;
+}
+
+StreamResult FifoBuffer::ReadOffsetLocked(void* buffer,
+                                          size_t bytes,
+                                          size_t offset,
+                                          size_t* bytes_read) {
+  if (offset >= data_length_) {
+    return (state_ != SS_CLOSED) ? SR_BLOCK : SR_EOS;
+  }
+
+  const size_t available = data_length_ - offset;
+  const size_t read_position = (read_position_ + offset) % buffer_length_;
+  const size_t copy = std::min(bytes, available);
+  const size_t tail_copy = std::min(copy, buffer_length_ - read_position);
+  char* const p = static_cast<char*>(buffer);
+  memcpy(p, &buffer_[read_position], tail_copy);
+  memcpy(p + tail_copy, &buffer_[0], copy - tail_copy);
+
+  if (bytes_read) {
+    *bytes_read = copy;
+  }
+  return SR_SUCCESS;
+}
+
+StreamResult FifoBuffer::WriteOffsetLocked(const void* buffer,
+                                           size_t bytes,
+                                           size_t offset,
+                                           size_t* bytes_written) {
+  if (state_ == SS_CLOSED) {
+    return SR_EOS;
+  }
+
+  if (data_length_ + offset >= buffer_length_) {
+    return SR_BLOCK;
+  }
+
+  const size_t available = buffer_length_ - data_length_ - offset;
+  const size_t write_position =
+      (read_position_ + data_length_ + offset) % buffer_length_;
+  const size_t copy = std::min(bytes, available);
+  const size_t tail_copy = std::min(copy, buffer_length_ - write_position);
+  const char* const p = static_cast<const char*>(buffer);
+  memcpy(&buffer_[write_position], p, tail_copy);
+  memcpy(&buffer_[0], p + tail_copy, copy - tail_copy);
+
+  if (bytes_written) {
+    *bytes_written = copy;
+  }
+  return SR_SUCCESS;
+}
+
+}  // namespace rtc