6 files changed, 282 insertions, 11 deletions

diff --git a/absl/strings/internal/cord_internal.cc b/absl/strings/internal/cord_internal.cc
index 1767e6fc..c9ceac90 100644
--- a/absl/strings/internal/cord_internal.cc
+++ b/absl/strings/internal/cord_internal.cc
@@ -19,6 +19,7 @@
 
 #include "absl/container/inlined_vector.h"
 #include "absl/strings/internal/cord_rep_btree.h"
+#include "absl/strings/internal/cord_rep_crc.h"
 #include "absl/strings/internal/cord_rep_flat.h"
 #include "absl/strings/internal/cord_rep_ring.h"
 
@@ -70,6 +71,9 @@ void CordRep::Destroy(CordRep* rep) {
         rep = child;
         continue;
       }
+    } else if (rep->tag == CRC) {
+      CordRepCrc::Destroy(rep->crc());
+      rep = nullptr;
     } else {
       CordRepFlat::Delete(rep);
       rep = nullptr;
diff --git a/absl/strings/internal/cord_internal.h b/absl/strings/internal/cord_internal.h
index bfe5564e..bf135ae2 100644
--- a/absl/strings/internal/cord_internal.h
+++ b/absl/strings/internal/cord_internal.h
@@ -192,6 +192,7 @@ struct CordRepConcat;
 struct CordRepExternal;
 struct CordRepFlat;
 struct CordRepSubstring;
+struct CordRepCrc;
 class CordRepRing;
 class CordRepBtree;
 
@@ -199,18 +200,19 @@ class CordRepBtree;
 enum CordRepKind {
   CONCAT = 0,
   SUBSTRING = 1,
-  BTREE = 2,
-  RING = 3,
-  EXTERNAL = 4,
+  CRC = 2,
+  BTREE = 3,
+  RING = 4,
+  EXTERNAL = 5,
 
   // We have different tags for different sized flat arrays,
-  // starting with FLAT, and limited to MAX_FLAT_TAG. The 225 value is based on
+  // starting with FLAT, and limited to MAX_FLAT_TAG. The 226 value is based on
   // the current 'size to tag' encoding of 8 / 32 bytes. If a new tag is needed
   // in the future, then 'FLAT' and 'MAX_FLAT_TAG' should be adjusted as well
   // as the Tag <---> Size logic so that FLAT stil represents the minimum flat
   // allocation size. (32 bytes as of now).
-  FLAT = 5,
-  MAX_FLAT_TAG = 225
+  FLAT = 6,
+  MAX_FLAT_TAG = 226
 };
 
 // There are various locations where we want to check if some rep is a 'plain'
@@ -251,6 +253,7 @@ struct CordRep {
   constexpr bool IsRing() const { return tag == RING; }
   constexpr bool IsConcat() const { return tag == CONCAT; }
   constexpr bool IsSubstring() const { return tag == SUBSTRING; }
+  constexpr bool IsCrc() const { return tag == CRC; }
   constexpr bool IsExternal() const { return tag == EXTERNAL; }
   constexpr bool IsFlat() const { return tag >= FLAT; }
   constexpr bool IsBtree() const { return tag == BTREE; }
@@ -261,6 +264,8 @@ struct CordRep {
   inline const CordRepConcat* concat() const;
   inline CordRepSubstring* substring();
   inline const CordRepSubstring* substring() const;
+  inline CordRepCrc* crc();
+  inline const CordRepCrc* crc() const;
   inline CordRepExternal* external();
   inline const CordRepExternal* external() const;
   inline CordRepFlat* flat();
diff --git a/absl/strings/internal/cord_rep_crc.cc b/absl/strings/internal/cord_rep_crc.cc
new file mode 100644
index 00000000..ee140354
--- /dev/null
+++ b/absl/strings/internal/cord_rep_crc.cc
@@ -0,0 +1,54 @@
+// Copyright 2021 The Abseil Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "absl/strings/internal/cord_rep_crc.h"
+
+#include <cassert>
+#include <cstdint>
+
+#include "absl/base/config.h"
+#include "absl/strings/internal/cord_internal.h"
+
+namespace absl {
+ABSL_NAMESPACE_BEGIN
+namespace cord_internal {
+
+CordRepCrc* CordRepCrc::New(CordRep* child, uint32_t crc) {
+  assert(child != nullptr);
+  if (child->IsCrc()) {
+    if (child->refcount.IsOne()) {
+      child->crc()->crc = crc;
+      return child->crc();
+    }
+    CordRep* old = child;
+    child = old->crc()->child;
+    CordRep::Ref(child);
+    CordRep::Unref(old);
+  }
+  auto* new_cordrep = new CordRepCrc;
+  new_cordrep->length = child->length;
+  new_cordrep->tag = cord_internal::CRC;
+  new_cordrep->child = child;
+  new_cordrep->crc = crc;
+  return new_cordrep;
+}
+
+void CordRepCrc::Destroy(CordRepCrc* node) {
+  CordRep::Unref(node->child);
+  delete node;
+}
+
+}  // namespace cord_internal
+ABSL_NAMESPACE_END
+}  // namespace absl
diff --git a/absl/strings/internal/cord_rep_crc.h b/absl/strings/internal/cord_rep_crc.h
new file mode 100644
index 00000000..3ead94a1
--- /dev/null
+++ b/absl/strings/internal/cord_rep_crc.h
@@ -0,0 +1,93 @@
+// Copyright 2021 The Abseil Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef ABSL_STRINGS_INTERNAL_CORD_REP_CRC_H_
+#define ABSL_STRINGS_INTERNAL_CORD_REP_CRC_H_
+
+#include <cassert>
+#include <cstdint>
+
+#include "absl/base/config.h"
+#include "absl/base/optimization.h"
+#include "absl/strings/internal/cord_internal.h"
+
+namespace absl {
+ABSL_NAMESPACE_BEGIN
+namespace cord_internal {
+
+// CordRepCrc is a CordRep node intended only to appear at the top level of a
+// cord tree.  It associates an "expected CRC" with the contained data, to allow
+// for easy passage of checksum data in Cord data flows.
+//
+// From Cord's perspective, the crc value has no semantics; any validation of
+// the contained checksum is the user's responsibility.
+struct CordRepCrc : public CordRep {
+  CordRep* child;
+  uint32_t crc;
+
+  // Consumes `child` and returns a CordRepCrc prefixed tree containing `child`.
+  // If the specified `child` is itself a CordRepCrc node, then this method
+  // either replaces the existing node, or directly updates the crc value in it
+  // depending on the node being shared or not, i.e.: refcount.IsOne().
+  // `child` must not be null. Never returns null.
+  static CordRepCrc* New(CordRep* child, uint32_t crc);
+
+  // Destroys (deletes) the provided node. `node` must not be null.
+  static void Destroy(CordRepCrc* node);
+};
+
+// Consumes `rep` and returns a CordRep* with any outer CordRepCrc wrapper
+// removed.  This is usually a no-op (returning `rep`), but this will remove and
+// unref an outer CordRepCrc node.
+inline CordRep* RemoveCrcNode(CordRep* rep) {
+  assert(rep != nullptr);
+  if (ABSL_PREDICT_FALSE(rep->IsCrc())) {
+    CordRep* child = rep->crc()->child;
+    if (rep->refcount.IsOne()) {
+      delete rep->crc();
+    } else {
+      CordRep::Ref(child);
+      CordRep::Unref(rep);
+    }
+    return child;
+  }
+  return rep;
+}
+
+// Returns `rep` if it is not a CordRepCrc node, or its child if it is.
+// Does not consume or create a reference on `rep` or the returned value.
+inline CordRep* SkipCrcNode(CordRep* rep) {
+  assert(rep != nullptr);
+  if (ABSL_PREDICT_FALSE(rep->IsCrc())) {
+    return rep->crc()->child;
+  } else {
+    return rep;
+  }
+}
+
+inline CordRepCrc* CordRep::crc() {
+  assert(IsCrc());
+  return static_cast<CordRepCrc*>(this);
+}
+
+inline const CordRepCrc* CordRep::crc() const {
+  assert(IsCrc());
+  return static_cast<const CordRepCrc*>(this);
+}
+
+}  // namespace cord_internal
+ABSL_NAMESPACE_END
+}  // namespace absl
+
+#endif  // ABSL_STRINGS_INTERNAL_CORD_REP_CRC_H_
diff --git a/absl/strings/internal/cord_rep_crc_test.cc b/absl/strings/internal/cord_rep_crc_test.cc
new file mode 100644
index 00000000..80f53485
--- /dev/null
+++ b/absl/strings/internal/cord_rep_crc_test.cc
@@ -0,0 +1,115 @@
+// Copyright 2021 The Abseil Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "absl/strings/internal/cord_rep_crc.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "absl/base/config.h"
+#include "absl/strings/internal/cord_internal.h"
+#include "absl/strings/internal/cord_rep_test_util.h"
+
+namespace absl {
+ABSL_NAMESPACE_BEGIN
+namespace cord_internal {
+namespace {
+
+using ::absl::cordrep_testing::MakeFlat;
+using ::testing::Eq;
+using ::testing::Ne;
+
+#if !defined(NDEBUG) && GTEST_HAS_DEATH_TEST
+
+TEST(CordRepCrc, NewWithNullPtr) {
+  EXPECT_DEATH(CordRepCrc::New(nullptr, 0), "");
+}
+
+TEST(CordRepCrc, RemoveCrcWithNullptr) {
+  EXPECT_DEATH(RemoveCrcNode(nullptr), "");
+}
+
+#endif  // !NDEBUG && GTEST_HAS_DEATH_TEST
+
+TEST(CordRepCrc, NewDestroy) {
+  CordRep* rep = cordrep_testing::MakeFlat("Hello world");
+  CordRepCrc* crc = CordRepCrc::New(rep, 12345);
+  EXPECT_TRUE(crc->refcount.IsOne());
+  EXPECT_THAT(crc->child, Eq(rep));
+  EXPECT_THAT(crc->crc, Eq(12345));
+  EXPECT_TRUE(rep->refcount.IsOne());
+  CordRepCrc::Destroy(crc);
+}
+
+TEST(CordRepCrc, NewExistingCrcNotShared) {
+  CordRep* rep = cordrep_testing::MakeFlat("Hello world");
+  CordRepCrc* crc = CordRepCrc::New(rep, 12345);
+  CordRepCrc* new_crc = CordRepCrc::New(crc, 54321);
+  EXPECT_THAT(new_crc, Eq(crc));
+  EXPECT_TRUE(new_crc->refcount.IsOne());
+  EXPECT_THAT(new_crc->child, Eq(rep));
+  EXPECT_THAT(new_crc->crc, Eq(54321));
+  EXPECT_TRUE(rep->refcount.IsOne());
+  CordRepCrc::Destroy(new_crc);
+}
+
+TEST(CordRepCrc, NewExistingCrcShared) {
+  CordRep* rep = cordrep_testing::MakeFlat("Hello world");
+  CordRepCrc* crc = CordRepCrc::New(rep, 12345);
+  CordRep::Ref(crc);
+  CordRepCrc* new_crc = CordRepCrc::New(crc, 54321);
+
+  EXPECT_THAT(new_crc, Ne(crc));
+  EXPECT_TRUE(new_crc->refcount.IsOne());
+  EXPECT_TRUE(crc->refcount.IsOne());
+  EXPECT_FALSE(rep->refcount.IsOne());
+  EXPECT_THAT(crc->child, Eq(rep));
+  EXPECT_THAT(new_crc->child, Eq(rep));
+  EXPECT_THAT(crc->crc, Eq(12345));
+  EXPECT_THAT(new_crc->crc, Eq(54321));
+
+  CordRep::Unref(crc);
+  CordRep::Unref(new_crc);
+}
+
+TEST(CordRepCrc, RemoveCrcNotCrc) {
+  CordRep* rep = cordrep_testing::MakeFlat("Hello world");
+  CordRep* nocrc = RemoveCrcNode(rep);
+  EXPECT_THAT(nocrc, Eq(rep));
+  CordRep::Unref(nocrc);
+}
+
+TEST(CordRepCrc, RemoveCrcNotShared) {
+  CordRep* rep = cordrep_testing::MakeFlat("Hello world");
+  CordRepCrc* crc = CordRepCrc::New(rep, 12345);
+  CordRep* nocrc = RemoveCrcNode(crc);
+  EXPECT_THAT(nocrc, Eq(rep));
+  EXPECT_TRUE(rep->refcount.IsOne());
+  CordRep::Unref(nocrc);
+}
+
+TEST(CordRepCrc, RemoveCrcShared) {
+  CordRep* rep = cordrep_testing::MakeFlat("Hello world");
+  CordRepCrc* crc = CordRepCrc::New(rep, 12345);
+  CordRep::Ref(crc);
+  CordRep* nocrc = RemoveCrcNode(crc);
+  EXPECT_THAT(nocrc, Eq(rep));
+  EXPECT_FALSE(rep->refcount.IsOne());
+  CordRep::Unref(nocrc);
+  CordRep::Unref(crc);
+}
+
+}  // namespace
+}  // namespace cord_internal
+ABSL_NAMESPACE_END
+}  // namespace absl
diff --git a/absl/strings/internal/cord_rep_flat.h b/absl/strings/internal/cord_rep_flat.h
index 4d0f9886..62cf5840 100644
--- a/absl/strings/internal/cord_rep_flat.h
+++ b/absl/strings/internal/cord_rep_flat.h
@@ -44,11 +44,11 @@ static constexpr size_t kMaxFlatLength = kMaxFlatSize - kFlatOverhead;
 static constexpr size_t kMinFlatLength = kMinFlatSize - kFlatOverhead;
 
 constexpr uint8_t AllocatedSizeToTagUnchecked(size_t size) {
-  return static_cast<uint8_t>((size <= 1024) ? size / 8 + 1
-                                             : 129 + size / 32 - 1024 / 32);
+  return static_cast<uint8_t>((size <= 1024) ? size / 8 + 2
+                                             : 130 + size / 32 - 1024 / 32);
 }
 
-static_assert(kMinFlatSize / 8 + 1 >= FLAT, "");
+static_assert(kMinFlatSize / 8 + 2 >= FLAT, "");
 static_assert(AllocatedSizeToTagUnchecked(kMaxFlatSize) <= MAX_FLAT_TAG, "");
 
 // Helper functions for rounded div, and rounding to exact sizes.
@@ -73,7 +73,7 @@ inline uint8_t AllocatedSizeToTag(size_t size) {
 
 // Converts the provided tag to the corresponding allocated size
 constexpr size_t TagToAllocatedSize(uint8_t tag) {
-  return (tag <= 129) ? ((tag - 1) * 8) : (1024 + (tag - 129) * 32);
+  return (tag <= 130) ? ((tag - 2) * 8) : (1024 + (tag - 130) * 32);
 }
 
 // Converts the provided tag to the corresponding available data length
@@ -82,7 +82,7 @@ constexpr size_t TagToLength(uint8_t tag) {
 }
 
 // Enforce that kMaxFlatSize maps to a well-known exact tag value.
-static_assert(TagToAllocatedSize(225) == kMaxFlatSize, "Bad tag logic");
+static_assert(TagToAllocatedSize(226) == kMaxFlatSize, "Bad tag logic");
 
 struct CordRepFlat : public CordRep {
   // Creates a new flat node.