diff --git a/runtime/vm/clustered_snapshot.cc b/runtime/vm/clustered_snapshot.cc
index b13f0ba1f96..31ce0fc12ff 100644
--- a/runtime/vm/clustered_snapshot.cc
+++ b/runtime/vm/clustered_snapshot.cc
@@ -52,6 +52,69 @@ DEFINE_FLAG(charp,
             "Write a snapshot profile in V8 format to a file.");
 #endif  // defined(DART_PRECOMPILER)
 
+namespace {
+// StorageTrait for HashTable which allows to create hash tables backed by
+// zone memory. Used to compute cluster order for canonical clusters.
+struct GrowableArrayStorageTraits {
+  class Array {
+   public:
+    explicit Array(Zone* zone, intptr_t length)
+        : length_(length), array_(zone->Alloc<ObjectPtr>(length)) {}
+
+    intptr_t Length() const { return length_; }
+    void SetAt(intptr_t index, const Object& value) const {
+      array_[index] = value.ptr();
+    }
+    ObjectPtr At(intptr_t index) const { return array_[index]; }
+
+   private:
+    intptr_t length_ = 0;
+    ObjectPtr* array_ = nullptr;
+    DISALLOW_COPY_AND_ASSIGN(Array);
+  };
+
+  using ArrayPtr = Array*;
+  class ArrayHandle : public ZoneAllocated {
+   public:
+    explicit ArrayHandle(ArrayPtr ptr) : ptr_(ptr) {}
+    ArrayHandle() {}
+
+    void SetFrom(const ArrayHandle& other) { ptr_ = other.ptr_; }
+    void Clear() { ptr_ = nullptr; }
+    bool IsNull() const { return ptr_ == nullptr; }
+    ArrayPtr ptr() { return ptr_; }
+
+    intptr_t Length() const { return ptr_->Length(); }
+    void SetAt(intptr_t index, const Object& value) const {
+      ptr_->SetAt(index, value);
+    }
+    ObjectPtr At(intptr_t index) const { return ptr_->At(index); }
+
+   private:
+    ArrayPtr ptr_ = nullptr;
+    DISALLOW_COPY_AND_ASSIGN(ArrayHandle);
+  };
+
+  static ArrayHandle& PtrToHandle(ArrayPtr ptr) {
+    return *new ArrayHandle(ptr);
+  }
+
+  static void SetHandle(ArrayHandle& dst, const ArrayHandle& src) {  // NOLINT
+    dst.SetFrom(src);
+  }
+
+  static void ClearHandle(ArrayHandle& dst) {  // NOLINT
+    dst.Clear();
+  }
+
+  static ArrayPtr New(Zone* zone, intptr_t length, Heap::Space space) {
+    return new (zone) Array(zone, length);
+  }
+
+  static bool IsImmutable(const ArrayHandle& handle) { return false; }
+};
+}  // namespace
+
 #if defined(DART_PRECOMPILER) && !defined(TARGET_ARCH_IA32)
 
 static void RelocateCodeObjects(
@@ -406,10 +469,227 @@ class ClassDeserializationCluster : public DeserializationCluster {
   intptr_t predefined_stop_index_;
 };
 
-#if !defined(DART_PRECOMPILED_RUNTIME)
-class TypeArgumentsSerializationCluster : public SerializationCluster {
+// Super classes for writing out clusters which contain objects grouped into
+// a canonical set (e.g. String, Type, TypeArguments, etc).
+// To save space in the snapshot we avoid writing such canonical sets
+// explicitly as Array objects into the snapshot and instead utilize a different
+// encoding: objects in a cluster representing a canonical set are sorted
+// to appear in the same order they appear in the Array representing the set,
+// and we additionaly write out array of values describing gaps between objects.
+//
+// In some situations not all canonical objects of the some type need to
+// be added to the resulting canonical set because they are cached in some
+// special way (see Type::Canonicalize as an example, which caches declaration
+// types in a special way). In this case subclass can set
+// kAllCanonicalObjectsAreIncludedIntoSet to |false| and override
+// IsInCanonicalSet filter.
+#if !defined(DART_PRECOMPILED_RUNTIME) && !defined(DART_COMPRESSED_POINTERS)
+template <typename SetType,
+          typename HandleType,
+          typename PointerType,
+          bool kAllCanonicalObjectsAreIncludedIntoSet = true>
+class CanonicalSetSerializationCluster : public SerializationCluster {
+ protected:
+  CanonicalSetSerializationCluster(bool represents_canonical_set,
+                                   const char* name,
+                                   intptr_t target_instance_size = 0)
+      : SerializationCluster(name, target_instance_size),
+        represents_canonical_set_(represents_canonical_set) {}
+
+  virtual bool IsInCanonicalSet(Serializer* s, PointerType ptr) {
+    // Must override this function if kAllCanonicalObjectsAreIncludedIntoSet
+    // is set to |false|.
+    ASSERT(kAllCanonicalObjectsAreIncludedIntoSet);
+    return true;
+  }
+
+  void ReorderObjects(Serializer* s) {
+    if (!represents_canonical_set_) {
+      return;
+    }
+
+    // Sort objects before writing them out so that they appear in the same
+    // order as they would appear in a CanonicalStringSet.
+    using ZoneCanonicalSet =
+        HashTable<typename SetType::Traits, 0, 0, GrowableArrayStorageTraits>;
+
+    // Compute required capacity for the hashtable (to avoid overallocating).
+    intptr_t required_capacity = 0;
+    for (auto ptr : objects_) {
+      if (kAllCanonicalObjectsAreIncludedIntoSet || IsInCanonicalSet(s, ptr)) {
+        required_capacity++;
+      }
+    }
+
+    intptr_t num_occupied = 0;
+
+    // Build canonical set out of objects that should belong to it.
+    // Objects that don't belong to it are copied to the prefix of objects_.
+    ZoneCanonicalSet table(
+        s->zone(), HashTables::New<ZoneCanonicalSet>(required_capacity));
+    HandleType& element = HandleType::Handle(s->zone());
+    for (auto ptr : objects_) {
+      if (kAllCanonicalObjectsAreIncludedIntoSet || IsInCanonicalSet(s, ptr)) {
+        element ^= ptr;
+        intptr_t entry = -1;
+        const bool present = table.FindKeyOrDeletedOrUnused(element, &entry);
+        ASSERT(!present);
+        table.InsertKey(entry, element);
+      } else {
+        objects_[num_occupied++] = ptr;
+      }
+    }
+
+    const auto prefix_length = num_occupied;
+
+    // Compute objects_ order and gaps based on canonical set layout.
+    auto& arr = table.Release();
+    intptr_t last_occupied = ZoneCanonicalSet::kFirstKeyIndex - 1;
+    for (intptr_t i = ZoneCanonicalSet::kFirstKeyIndex, length = arr.Length();
+         i < length; i++) {
+      ObjectPtr v = arr.At(i);
+      ASSERT(v != ZoneCanonicalSet::DeletedMarker().ptr());
+      if (v != ZoneCanonicalSet::UnusedMarker().ptr()) {
+        const intptr_t unused_run_length = (i - 1) - last_occupied;
+        gaps_.Add(unused_run_length);
+        objects_[num_occupied++] = static_cast<PointerType>(v);
+        last_occupied = i;
+      }
+    }
+    ASSERT(num_occupied == objects_.length());
+    ASSERT(prefix_length == (objects_.length() - gaps_.length()));
+    table_length_ = arr.Length();
+  }
+
+  void WriteCanonicalSetLayout(Serializer* s) {
+    if (represents_canonical_set_) {
+      s->WriteUnsigned(table_length_);
+      if (kAllCanonicalObjectsAreIncludedIntoSet) {
+        ASSERT(objects_.length() == gaps_.length());
+      } else {
+        s->WriteUnsigned(objects_.length() - gaps_.length());
+      }
+      for (auto gap : gaps_) {
+        s->WriteUnsigned(gap);
+      }
+    }
+  }
+
+  GrowableArray<PointerType> objects_;
+
+ private:
+  const bool represents_canonical_set_;
+  GrowableArray<intptr_t> gaps_;
+  intptr_t table_length_ = 0;
+};
+#endif
+
+template <typename SetType, bool kAllCanonicalObjectsAreIncludedIntoSet = true>
+class CanonicalSetDeserializationCluster : public DeserializationCluster {
  public:
-  TypeArgumentsSerializationCluster() : SerializationCluster("TypeArguments") {}
+  CanonicalSetDeserializationCluster(bool is_root_unit, const char* name)
+      : DeserializationCluster(name),
+        is_root_unit_(is_root_unit),
+        table_(Array::Handle()) {}
+
+  void BuildCanonicalSetFromLayout(Deserializer* d, bool is_canonical) {
+    if (!is_root_unit_ || !is_canonical) {
+      return;
+    }
+
+    const auto table_length = d->ReadUnsigned();
+    first_element_ =
+        kAllCanonicalObjectsAreIncludedIntoSet ? 0 : d->ReadUnsigned();
+    const intptr_t count = stop_index_ - (start_index_ + first_element_);
+    auto table = StartDeserialization(d, table_length, count);
+    for (intptr_t i = start_index_ + first_element_; i < stop_index_; i++) {
+      table.FillGap(d->ReadUnsigned());
+      table.WriteElement(d, d->Ref(i));
+    }
+    table_ = table.Finish();
+  }
+
+ protected:
+  const bool is_root_unit_;
+  intptr_t first_element_;
+  Array& table_;
+
+  void VerifyCanonicalSet(Deserializer* d,
+                          const Array& refs,
+                          const Array& current_table) {
+#if defined(DEBUG)
+    // First check that we are not overwriting a table and loosing information.
+    if (!current_table.IsNull()) {
+      SetType current_set(d->zone(), current_table.ptr());
+      ASSERT(current_set.NumOccupied() == 0);
+      current_set.Release();
+    }
+
+    // Now check that manually created table behaves correctly as a canonical
+    // set.
+    SetType canonical_set(d->zone(), table_.ptr());
+    Object& key = Object::Handle();
+    for (intptr_t i = start_index_ + first_element_; i < stop_index_; i++) {
+      key = refs.At(i);
+      ASSERT(canonical_set.GetOrNull(key) != Object::null());
+    }
+    canonical_set.Release();
+#endif  // defined(DEBUG)
+  }
+
+ private:
+  struct DeserializationFinger {
+    ArrayPtr table;
+    intptr_t current_index;
+    ObjectPtr gap_element;
+
+    void FillGap(int length) {
+      for (intptr_t j = 0; j < length; j++) {
+        table->untag()->data()[current_index + j] = gap_element;
+      }
+      current_index += length;
+    }
+
+    void WriteElement(Deserializer* d, ObjectPtr object) {
+      table->untag()->data()[current_index++] = object;
+    }
+
+    ArrayPtr Finish() {
+      if (table != Array::null()) {
+        FillGap(Smi::Value(table->untag()->length_) - current_index);
+      }
+      auto result = table;
+      table = Array::null();
+      return result;
+    }
+  };
+
+  static DeserializationFinger StartDeserialization(Deserializer* d,
+                                                    intptr_t length,
+                                                    intptr_t count) {
+    const intptr_t instance_size = Array::InstanceSize(length);
+    ArrayPtr table = static_cast<ArrayPtr>(
+        AllocateUninitialized(d->heap()->old_space(), instance_size));
+    Deserializer::InitializeHeader(table, kArrayCid, instance_size);
+    table->untag()->type_arguments_ = TypeArguments::null();
+    table->untag()->length_ = Smi::New(length);
+    for (intptr_t i = 0; i < SetType::kFirstKeyIndex; i++) {
+      table->untag()->data()[i] = Smi::New(0);
+    }
+    table->untag()->data()[SetType::kOccupiedEntriesIndex] = Smi::New(count);
+    return {table, SetType::kFirstKeyIndex, SetType::UnusedMarker().ptr()};
+  }
+};
+
+#if !defined(DART_PRECOMPILED_RUNTIME)
+class TypeArgumentsSerializationCluster
+    : public CanonicalSetSerializationCluster<CanonicalTypeArgumentsSet,
+                                              TypeArguments,
+                                              TypeArgumentsPtr> {
+ public:
+  explicit TypeArgumentsSerializationCluster(bool represents_canonical_set)
+      : CanonicalSetSerializationCluster(represents_canonical_set,
+                                         "TypeArguments") {}
   ~TypeArgumentsSerializationCluster() {}
 
   void Trace(Serializer* s, ObjectPtr object) {
@@ -427,6 +707,7 @@ class TypeArgumentsSerializationCluster : public SerializationCluster {
     s->WriteCid(kTypeArgumentsCid);
     const intptr_t count = objects_.length();
     s->WriteUnsigned(count);
+    ReorderObjects(s);
     for (intptr_t i = 0; i < count; i++) {
       TypeArgumentsPtr type_args = objects_[i];
       s->AssignRef(type_args);
@@ -436,6 +717,7 @@ class TypeArgumentsSerializationCluster : public SerializationCluster {
       target_memory_size_ +=
           compiler::target::TypeArguments::InstanceSize(length);
     }
+    WriteCanonicalSetLayout(s);
   }
 
   void WriteFill(Serializer* s) {
@@ -455,16 +737,14 @@ class TypeArgumentsSerializationCluster : public SerializationCluster {
       }
     }
   }
-
- private:
-  GrowableArray<TypeArgumentsPtr> objects_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME
 
-class TypeArgumentsDeserializationCluster : public DeserializationCluster {
+class TypeArgumentsDeserializationCluster
+    : public CanonicalSetDeserializationCluster<CanonicalTypeArgumentsSet> {
  public:
-  TypeArgumentsDeserializationCluster()
-      : DeserializationCluster("TypeArguments") {}
+  explicit TypeArgumentsDeserializationCluster(bool is_root_unit)
+      : CanonicalSetDeserializationCluster(is_root_unit, "TypeArguments") {}
   ~TypeArgumentsDeserializationCluster() {}
 
   void ReadAlloc(Deserializer* d, bool stamp_canonical) {
@@ -477,6 +757,7 @@ class TypeArgumentsDeserializationCluster : public DeserializationCluster {
                                          TypeArguments::InstanceSize(length)));
     }
     stop_index_ = d->next_index();
+    BuildCanonicalSetFromLayout(d, stamp_canonical);
   }
 
   void ReadFill(Deserializer* d, bool stamp_canonical) {
@@ -498,7 +779,12 @@ class TypeArgumentsDeserializationCluster : public DeserializationCluster {
   }
 
   void PostLoad(Deserializer* d, const Array& refs, bool canonicalize) {
-    if (canonicalize) {
+    if (!table_.IsNull()) {
+      auto object_store = d->isolate_group()->object_store();
+      VerifyCanonicalSet(
+          d, refs, Array::Handle(object_store->canonical_type_arguments()));
+      object_store->set_canonical_type_arguments(table_);
+    } else if (canonicalize) {
       Thread* thread = Thread::Current();
       TypeArguments& type_arg = TypeArguments::Handle(d->zone());
       for (intptr_t i = start_index_; i < stop_index_; i++) {
@@ -2374,12 +2660,20 @@ class CompressedStackMapsDeserializationCluster
 
 #if !defined(DART_PRECOMPILED_RUNTIME) && !defined(DART_COMPRESSED_POINTERS)
 // PcDescriptor, CompressedStackMaps, OneByteString, TwoByteString
-class RODataSerializationCluster : public SerializationCluster {
+class RODataSerializationCluster
+    : public CanonicalSetSerializationCluster<CanonicalStringSet,
+                                              String,
+                                              ObjectPtr> {
  public:
-  RODataSerializationCluster(Zone* zone, const char* type, intptr_t cid)
-      : SerializationCluster(ImageWriter::TagObjectTypeAsReadOnly(zone, type)),
+  RODataSerializationCluster(Zone* zone,
+                             const char* type,
+                             intptr_t cid,
+                             bool is_canonical)
+      : CanonicalSetSerializationCluster(
+            is_canonical && IsStringClassId(cid),
+            ImageWriter::TagObjectTypeAsReadOnly(zone, type)),
+        zone_(zone),
         cid_(cid),
-        objects_(),
         type_(type) {}
   ~RODataSerializationCluster() {}
 
@@ -2399,15 +2693,18 @@ class RODataSerializationCluster : public SerializationCluster {
   }
 
   void WriteAlloc(Serializer* s) {
+    const bool is_string_cluster = IsStringClassId(cid_);
     s->WriteCid(cid_);
 
     intptr_t count = objects_.length();
     s->WriteUnsigned(count);
+    ReorderObjects(s);
+
     uint32_t running_offset = 0;
     for (intptr_t i = 0; i < count; i++) {
       ObjectPtr object = objects_[i];
       s->AssignRef(object);
-      if (cid_ == kOneByteStringCid || cid_ == kTwoByteStringCid) {
+      if (is_string_cluster) {
         s->TraceStartWritingObject(type_, object, String::RawCast(object));
       } else {
         s->TraceStartWritingObject(type_, object, nullptr);
@@ -2422,6 +2719,7 @@ class RODataSerializationCluster : public SerializationCluster {
       running_offset = offset;
       s->TraceEndWritingObject();
     }
+    WriteCanonicalSetLayout(s);
   }
 
   void WriteFill(Serializer* s) {
@@ -2429,17 +2727,18 @@ class RODataSerializationCluster : public SerializationCluster {
   }
 
  private:
+  Zone* zone_;
   const intptr_t cid_;
-  GrowableArray<ObjectPtr> objects_;
   const char* const type_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME && !DART_COMPRESSED_POINTERS
 
 #if !defined(DART_COMPRESSED_POINTERS)
-class RODataDeserializationCluster : public DeserializationCluster {
+class RODataDeserializationCluster
+    : public CanonicalSetDeserializationCluster<CanonicalStringSet> {
  public:
-  explicit RODataDeserializationCluster(intptr_t cid)
-      : DeserializationCluster("ROData"), cid_(cid) {}
+  explicit RODataDeserializationCluster(bool is_root_unit, intptr_t cid)
+      : CanonicalSetDeserializationCluster(is_root_unit, "ROData"), cid_(cid) {}
   ~RODataDeserializationCluster() {}
 
   void ReadAlloc(Deserializer* d, bool stamp_canonical) {
@@ -2448,9 +2747,11 @@ class RODataDeserializationCluster : public DeserializationCluster {
     uint32_t running_offset = 0;
     for (intptr_t i = 0; i < count; i++) {
       running_offset += d->ReadUnsigned() << kObjectAlignmentLog2;
-      d->AssignRef(d->GetObjectAt(running_offset));
+      ObjectPtr object = d->GetObjectAt(running_offset);
+      d->AssignRef(object);
     }
     stop_index_ = d->next_index();
+    BuildCanonicalSetFromLayout(d, cid_ == kStringCid);
   }
 
   void ReadFill(Deserializer* d, bool stamp_canonical) {
@@ -2458,7 +2759,14 @@ class RODataDeserializationCluster : public DeserializationCluster {
   }
 
   void PostLoad(Deserializer* d, const Array& refs, bool canonicalize) {
-    if (canonicalize) {
+    if (!table_.IsNull()) {
+      auto object_store = d->isolate_group()->object_store();
+      VerifyCanonicalSet(d, refs, Array::Handle(object_store->symbol_table()));
+      object_store->set_symbol_table(table_);
+      if (d->isolate_group() == Dart::vm_isolate_group()) {
+        Symbols::InitFromSnapshot(d->isolate_group());
+      }
+    } else if (canonicalize) {
       FATAL("Cannot recanonicalize RO objects.");
     }
   }
@@ -3482,10 +3790,18 @@ static constexpr intptr_t kNullabilityBitSize = 2;
 static constexpr intptr_t kNullabilityBitMask = (1 << kNullabilityBitSize) - 1;
 
 #if !defined(DART_PRECOMPILED_RUNTIME)
-class TypeSerializationCluster : public SerializationCluster {
+class TypeSerializationCluster
+    : public CanonicalSetSerializationCluster<
+          CanonicalTypeSet,
+          Type,
+          TypePtr,
+          /*kAllCanonicalObjectsAreIncludedIntoSet=*/false> {
  public:
-  TypeSerializationCluster()
-      : SerializationCluster("Type", compiler::target::Type::InstanceSize()) {}
+  explicit TypeSerializationCluster(bool represents_canonical_set)
+      : CanonicalSetSerializationCluster(
+            represents_canonical_set,
+            "Type",
+            compiler::target::Type::InstanceSize()) {}
   ~TypeSerializationCluster() {}
 
   void Trace(Serializer* s, ObjectPtr object) {
@@ -3509,10 +3825,12 @@ class TypeSerializationCluster : public SerializationCluster {
     s->WriteCid(kTypeCid);
     intptr_t count = objects_.length();
     s->WriteUnsigned(count);
+    ReorderObjects(s);
     for (intptr_t i = 0; i < count; i++) {
       TypePtr type = objects_[i];
       s->AssignRef(type);
     }
+    WriteCanonicalSetLayout(s);
   }
 
   void WriteFill(Serializer* s) {
@@ -3523,6 +3841,27 @@ class TypeSerializationCluster : public SerializationCluster {
   }
 
  private:
+  Type& type_ = Type::Handle();
+  Class& cls_ = Class::Handle();
+
+  // Type::Canonicalize does not actually put all canonical Type objects into
+  // canonical_types set. Some of the canonical declaration types (but not all
+  // of them) are simply cached in UntaggedClass::declaration_type_ and are not
+  // inserted into the canonical_types set.
+  // Keep in sync with Type::Canonicalize.
+  virtual bool IsInCanonicalSet(Serializer* s, TypePtr type) {
+    SmiPtr raw_type_class_id = Smi::RawCast(type->untag()->type_class_id_);
+    ClassPtr type_class =
+        s->isolate_group()->class_table()->At(Smi::Value(raw_type_class_id));
+    if (type_class->untag()->declaration_type_ != type) {
+      return true;
+    }
+
+    type_ = type;
+    cls_ = type_class;
+    return !type_.IsDeclarationTypeOf(cls_);
+  }
+
   void WriteType(Serializer* s, TypePtr type) {
     AutoTraceObject(type);
     WriteFromTo(type);
@@ -3538,14 +3877,16 @@ class TypeSerializationCluster : public SerializationCluster {
     ASSERT_EQUAL(type->untag()->nullability_, combined & kNullabilityBitMask);
     s->Write<uint8_t>(combined);
   }
-
-  GrowableArray<TypePtr> objects_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME
 
-class TypeDeserializationCluster : public DeserializationCluster {
+class TypeDeserializationCluster
+    : public CanonicalSetDeserializationCluster<
+          CanonicalTypeSet,
+          /*kAllCanonicalObjectsAreIncludedIntoSet=*/false> {
  public:
-  TypeDeserializationCluster() : DeserializationCluster("Type") {}
+  explicit TypeDeserializationCluster(bool is_root_unit)
+      : CanonicalSetDeserializationCluster(is_root_unit, "Type") {}
   ~TypeDeserializationCluster() {}
 
   void ReadAlloc(Deserializer* d, bool stamp_canonical) {
@@ -3553,9 +3894,11 @@ class TypeDeserializationCluster : public DeserializationCluster {
     PageSpace* old_space = d->heap()->old_space();
     const intptr_t count = d->ReadUnsigned();
     for (intptr_t i = 0; i < count; i++) {
-      d->AssignRef(AllocateUninitialized(old_space, Type::InstanceSize()));
+      ObjectPtr object = AllocateUninitialized(old_space, Type::InstanceSize());
+      d->AssignRef(object);
     }
     stop_index_ = d->next_index();
+    BuildCanonicalSetFromLayout(d, stamp_canonical);
   }
 
   void ReadFill(Deserializer* d, bool stamp_canonical) {
@@ -3571,7 +3914,12 @@ class TypeDeserializationCluster : public DeserializationCluster {
   }
 
   void PostLoad(Deserializer* d, const Array& refs, bool canonicalize) {
-    if (canonicalize) {
+    if (!table_.IsNull()) {
+      auto object_store = d->isolate_group()->object_store();
+      VerifyCanonicalSet(d, refs,
+                         Array::Handle(object_store->canonical_types()));
+      object_store->set_canonical_types(table_);
+    } else if (canonicalize) {
       Thread* thread = Thread::Current();
       AbstractType& type = AbstractType::Handle(d->zone());
       for (intptr_t i = start_index_; i < stop_index_; i++) {
@@ -3601,11 +3949,16 @@ class TypeDeserializationCluster : public DeserializationCluster {
 };
 
 #if !defined(DART_PRECOMPILED_RUNTIME)
-class FunctionTypeSerializationCluster : public SerializationCluster {
+class FunctionTypeSerializationCluster
+    : public CanonicalSetSerializationCluster<CanonicalFunctionTypeSet,
+                                              FunctionType,
+                                              FunctionTypePtr> {
  public:
-  FunctionTypeSerializationCluster()
-      : SerializationCluster("FunctionType",
-                             compiler::target::FunctionType::InstanceSize()) {}
+  explicit FunctionTypeSerializationCluster(bool represents_canonical_set)
+      : CanonicalSetSerializationCluster(
+            represents_canonical_set,
+            "FunctionType",
+            compiler::target::FunctionType::InstanceSize()) {}
   ~FunctionTypeSerializationCluster() {}
 
   void Trace(Serializer* s, ObjectPtr object) {
@@ -3618,10 +3971,13 @@ class FunctionTypeSerializationCluster : public SerializationCluster {
     s->WriteCid(kFunctionTypeCid);
     intptr_t count = objects_.length();
     s->WriteUnsigned(count);
+    ReorderObjects(s);
+
     for (intptr_t i = 0; i < count; i++) {
       FunctionTypePtr type = objects_[i];
       s->AssignRef(type);
     }
+    WriteCanonicalSetLayout(s);
   }
 
   void WriteFill(Serializer* s) {
@@ -3650,15 +4006,14 @@ class FunctionTypeSerializationCluster : public SerializationCluster {
     s->Write<uint8_t>(combined);
     s->Write<uint32_t>(type->untag()->packed_fields_);
   }
-
-  GrowableArray<FunctionTypePtr> objects_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME
 
-class FunctionTypeDeserializationCluster : public DeserializationCluster {
+class FunctionTypeDeserializationCluster
+    : public CanonicalSetDeserializationCluster<CanonicalFunctionTypeSet> {
  public:
-  FunctionTypeDeserializationCluster()
-      : DeserializationCluster("FunctionType") {}
+  explicit FunctionTypeDeserializationCluster(bool is_root_unit)
+      : CanonicalSetDeserializationCluster(is_root_unit, "FunctionType") {}
   ~FunctionTypeDeserializationCluster() {}
 
   void ReadAlloc(Deserializer* d, bool stamp_canonical) {
@@ -3666,10 +4021,12 @@ class FunctionTypeDeserializationCluster : public DeserializationCluster {
     PageSpace* old_space = d->heap()->old_space();
     const intptr_t count = d->ReadUnsigned();
     for (intptr_t i = 0; i < count; i++) {
-      d->AssignRef(
-          AllocateUninitialized(old_space, FunctionType::InstanceSize()));
+      ObjectPtr object =
+          AllocateUninitialized(old_space, FunctionType::InstanceSize());
+      d->AssignRef(object);
     }
     stop_index_ = d->next_index();
+    BuildCanonicalSetFromLayout(d, stamp_canonical);
   }
 
   void ReadFill(Deserializer* d, bool stamp_canonical) {
@@ -3687,7 +4044,12 @@ class FunctionTypeDeserializationCluster : public DeserializationCluster {
   }
 
   void PostLoad(Deserializer* d, const Array& refs, bool canonicalize) {
-    if (canonicalize) {
+    if (!table_.IsNull()) {
+      auto object_store = d->isolate_group()->object_store();
+      VerifyCanonicalSet(
+          d, refs, Array::Handle(object_store->canonical_function_types()));
+      object_store->set_canonical_function_types(table_);
+    } else if (canonicalize) {
       Thread* thread = Thread::Current();
       AbstractType& type = AbstractType::Handle(d->zone());
       for (intptr_t i = start_index_; i < stop_index_; i++) {
@@ -3810,11 +4172,17 @@ class TypeRefDeserializationCluster : public DeserializationCluster {
 };
 
 #if !defined(DART_PRECOMPILED_RUNTIME)
-class TypeParameterSerializationCluster : public SerializationCluster {
+class TypeParameterSerializationCluster
+    : public CanonicalSetSerializationCluster<CanonicalTypeParameterSet,
+                                              TypeParameter,
+                                              TypeParameterPtr> {
  public:
-  TypeParameterSerializationCluster()
-      : SerializationCluster("TypeParameter",
-                             compiler::target::TypeParameter::InstanceSize()) {}
+  explicit TypeParameterSerializationCluster(
+      bool cluster_represents_canonical_set)
+      : CanonicalSetSerializationCluster(
+            cluster_represents_canonical_set,
+            "TypeParameter",
+            compiler::target::TypeParameter::InstanceSize()) {}
   ~TypeParameterSerializationCluster() {}
 
   void Trace(Serializer* s, ObjectPtr object) {
@@ -3828,10 +4196,12 @@ class TypeParameterSerializationCluster : public SerializationCluster {
     s->WriteCid(kTypeParameterCid);
     intptr_t count = objects_.length();
     s->WriteUnsigned(count);
+    ReorderObjects(s);
     for (intptr_t i = 0; i < count; i++) {
       TypeParameterPtr type = objects_[i];
       s->AssignRef(type);
     }
+    WriteCanonicalSetLayout(s);
   }
 
   void WriteFill(Serializer* s) {
@@ -3859,15 +4229,14 @@ class TypeParameterSerializationCluster : public SerializationCluster {
     ASSERT_EQUAL(type->untag()->nullability_, combined & kNullabilityBitMask);
     s->Write<uint8_t>(combined);
   }
-
-  GrowableArray<TypeParameterPtr> objects_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME
 
-class TypeParameterDeserializationCluster : public DeserializationCluster {
+class TypeParameterDeserializationCluster
+    : public CanonicalSetDeserializationCluster<CanonicalTypeParameterSet> {
  public:
-  TypeParameterDeserializationCluster()
-      : DeserializationCluster("TypeParameter") {}
+  explicit TypeParameterDeserializationCluster(bool is_root_unit)
+      : CanonicalSetDeserializationCluster(is_root_unit, "TypeParameter") {}
   ~TypeParameterDeserializationCluster() {}
 
   void ReadAlloc(Deserializer* d, bool stamp_canonical) {
@@ -3879,6 +4248,7 @@ class TypeParameterDeserializationCluster : public DeserializationCluster {
           AllocateUninitialized(old_space, TypeParameter::InstanceSize()));
     }
     stop_index_ = d->next_index();
+    BuildCanonicalSetFromLayout(d, stamp_canonical);
   }
 
   void ReadFill(Deserializer* d, bool stamp_canonical) {
@@ -3898,7 +4268,12 @@ class TypeParameterDeserializationCluster : public DeserializationCluster {
   }
 
   void PostLoad(Deserializer* d, const Array& refs, bool canonicalize) {
-    if (canonicalize) {
+    if (!table_.IsNull()) {
+      auto object_store = d->isolate_group()->object_store();
+      VerifyCanonicalSet(
+          d, refs, Array::Handle(object_store->canonical_type_parameters()));
+      object_store->set_canonical_type_parameters(table_);
+    } else if (canonicalize) {
       Thread* thread = Thread::Current();
       TypeParameter& type_param = TypeParameter::Handle(d->zone());
       for (intptr_t i = start_index_; i < stop_index_; i++) {
@@ -5128,8 +5503,10 @@ class FakeSerializationCluster : public SerializationCluster {
 #if !defined(DART_PRECOMPILED_RUNTIME)
 class VMSerializationRoots : public SerializationRoots {
  public:
-  explicit VMSerializationRoots(const Array& symbols)
-      : symbols_(symbols), zone_(Thread::Current()->zone()) {}
+  explicit VMSerializationRoots(const Array& symbols, bool should_write_symbols)
+      : symbols_(symbols),
+        should_write_symbols_(should_write_symbols),
+        zone_(Thread::Current()->zone()) {}
 
   void AddBaseObjects(Serializer* s) {
     // These objects are always allocated by Object::InitOnce, so they are not
@@ -5199,7 +5576,13 @@ class VMSerializationRoots : public SerializationRoots {
   }
 
   void PushRoots(Serializer* s) {
-    s->Push(symbols_.ptr());
+    if (should_write_symbols_) {
+      s->Push(symbols_.ptr());
+    } else {
+      for (intptr_t i = 0; i < symbols_.Length(); i++) {
+        s->Push(symbols_.At(i));
+      }
+    }
     if (Snapshot::IncludesCode(s->kind())) {
       for (intptr_t i = 0; i < StubCode::NumEntries(); i++) {
         s->Push(StubCode::EntryAt(i).ptr());
@@ -5208,17 +5591,37 @@ class VMSerializationRoots : public SerializationRoots {
   }
 
   void WriteRoots(Serializer* s) {
-    s->WriteRootRef(symbols_.ptr(), "symbol-table");
+    s->WriteRootRef(should_write_symbols_ ? symbols_.ptr() : Object::null(),
+                    "symbol-table");
     if (Snapshot::IncludesCode(s->kind())) {
       for (intptr_t i = 0; i < StubCode::NumEntries(); i++) {
         s->WriteRootRef(StubCode::EntryAt(i).ptr(),
                         zone_->PrintToString("Stub:%s", StubCode::NameAt(i)));
       }
     }
+
+    if (!should_write_symbols_ && s->profile_writer() != nullptr) {
+      // If writing V8 snapshot profile create an artifical node representing
+      // VM isolate symbol table.
+      auto symbols_ref = s->AssignArtificialRef(symbols_.ptr());
+      const V8SnapshotProfileWriter::ObjectId symbols_snapshot_id(
+          V8SnapshotProfileWriter::kSnapshot, symbols_ref);
+      s->profile_writer()->AddRoot(symbols_snapshot_id, "vm_symbols");
+      s->profile_writer()->SetObjectTypeAndName(symbols_snapshot_id, "Symbols",
+                                                nullptr);
+      for (intptr_t i = 0; i < symbols_.Length(); i++) {
+        const V8SnapshotProfileWriter::ObjectId code_id(
+            V8SnapshotProfileWriter::kSnapshot, s->RefId(symbols_.At(i)));
+        s->profile_writer()->AttributeReferenceTo(
+            symbols_snapshot_id,
+            {code_id, V8SnapshotProfileWriter::Reference::kElement, i});
+      }
+    }
   }
 
  private:
   const Array& symbols_;
+  const bool should_write_symbols_;
   Zone* zone_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME
@@ -5282,7 +5685,9 @@ class VMDeserializationRoots : public DeserializationRoots {
 
   void ReadRoots(Deserializer* d) {
     symbol_table_ ^= d->ReadRef();
-    d->isolate_group()->object_store()->set_symbol_table(symbol_table_);
+    if (!symbol_table_.IsNull()) {
+      d->isolate_group()->object_store()->set_symbol_table(symbol_table_);
+    }
     if (Snapshot::IncludesCode(d->kind())) {
       for (intptr_t i = 0; i < StubCode::NumEntries(); i++) {
         Code* code = Code::ReadOnlyHandle();
@@ -5297,7 +5702,9 @@ class VMDeserializationRoots : public DeserializationRoots {
     // allocations (e.g., FinalizeVMIsolate) before allocating new pages.
     d->heap()->old_space()->AbandonBumpAllocation();
 
-    Symbols::InitFromSnapshot(d->isolate_group());
+    if (!symbol_table_.IsNull()) {
+      Symbols::InitFromSnapshot(d->isolate_group());
+    }
 
     Object::set_vm_isolate_snapshot_object_table(refs);
   }
@@ -5319,29 +5726,51 @@ static const char* kObjectStoreFieldNames[] = {
 class ProgramSerializationRoots : public SerializationRoots {
  public:
   ProgramSerializationRoots(ZoneGrowableArray<Object*>* base_objects,
-                            ObjectStore* object_store)
+                            ObjectStore* object_store,
+                            Snapshot::Kind snapshot_kind)
       : base_objects_(base_objects),
         object_store_(object_store),
-        dispatch_table_entries_(Array::Handle()) {
+        dispatch_table_entries_(Array::Handle()),
+        saved_symbol_table_(Array::Handle()),
+        saved_canonical_types_(Array::Handle()),
+        saved_canonical_function_types_(Array::Handle()),
+        saved_canonical_type_arguments_(Array::Handle()),
+        saved_canonical_type_parameters_(Array::Handle()) {
+    saved_symbol_table_ = object_store->symbol_table();
+    if (Snapshot::IncludesCode(snapshot_kind)) {
+      object_store->set_symbol_table(
+          Array::Handle(HashTables::New<CanonicalStringSet>(4)));
+    } else {
 #if defined(DART_PRECOMPILER)
-    if (FLAG_precompiled_mode) {
-      // Elements of constant tables are treated as weak so literals used only
-      // in deferred libraries do not end up in the main snapshot.
-      Array& table = Array::Handle();
-      table = object_store->symbol_table();
-      HashTables::Weaken(table);
-      table = object_store->canonical_types();
-      HashTables::Weaken(table);
-      table = object_store->canonical_function_types();
-      HashTables::Weaken(table);
-      table = object_store->canonical_type_parameters();
-      HashTables::Weaken(table);
-      table = object_store->canonical_type_arguments();
-      HashTables::Weaken(table);
-    }
+      if (FLAG_precompiled_mode) {
+        HashTables::Weaken(saved_symbol_table_);
+      }
 #endif
+    }
+    saved_canonical_types_ = object_store->canonical_types();
+    object_store->set_canonical_types(
+        Array::Handle(HashTables::New<CanonicalTypeSet>(4)));
+    saved_canonical_function_types_ = object_store->canonical_function_types();
+    object_store->set_canonical_function_types(
+        Array::Handle(HashTables::New<CanonicalFunctionTypeSet>(4)));
+    saved_canonical_type_arguments_ = object_store->canonical_type_arguments();
+    object_store->set_canonical_type_arguments(
+        Array::Handle(HashTables::New<CanonicalTypeArgumentsSet>(4)));
+    saved_canonical_type_parameters_ =
+        object_store->canonical_type_parameters();
+    object_store->set_canonical_type_parameters(
+        Array::Handle(HashTables::New<CanonicalTypeParameterSet>(4)));
+  }
+  ~ProgramSerializationRoots() {
+    object_store_->set_symbol_table(saved_symbol_table_);
+    object_store_->set_canonical_types(saved_canonical_types_);
+    object_store_->set_canonical_function_types(
+        saved_canonical_function_types_);
+    object_store_->set_canonical_type_arguments(
+        saved_canonical_type_arguments_);
+    object_store_->set_canonical_type_parameters(
+        saved_canonical_type_parameters_);
   }
-  ~ProgramSerializationRoots() {}
 
   void AddBaseObjects(Serializer* s) {
     if (base_objects_ == nullptr) {
@@ -5397,6 +5826,11 @@ class ProgramSerializationRoots : public SerializationRoots {
   ZoneGrowableArray<Object*>* base_objects_;
   ObjectStore* object_store_;
   Array& dispatch_table_entries_;
+  Array& saved_symbol_table_;
+  Array& saved_canonical_types_;
+  Array& saved_canonical_function_types_;
+  Array& saved_canonical_type_arguments_;
+  Array& saved_canonical_type_parameters_;
 };
 #endif  // !DART_PRECOMPILED_RUNTIME
 
@@ -5855,6 +6289,9 @@ const char* Serializer::ReadOnlyObjectType(intptr_t cid) {
       return "CodeSourceMap";
     case kCompressedStackMapsCid:
       return "CompressedStackMaps";
+    case kStringCid:
+      RELEASE_ASSERT(current_loading_unit_id_ <= LoadingUnit::kRootId);
+      return "CanonicalString";
     case kOneByteStringCid:
       return current_loading_unit_id_ <= LoadingUnit::kRootId
                  ? "OneByteStringCid"
@@ -5868,7 +6305,8 @@ const char* Serializer::ReadOnlyObjectType(intptr_t cid) {
   }
 }
 
-SerializationCluster* Serializer::NewClusterForClass(intptr_t cid) {
+SerializationCluster* Serializer::NewClusterForClass(intptr_t cid,
+                                                     bool is_canonical) {
 #if defined(DART_PRECOMPILED_RUNTIME)
   UNREACHABLE();
   return NULL;
@@ -5899,16 +6337,20 @@ SerializationCluster* Serializer::NewClusterForClass(intptr_t cid) {
   // compressed pointers.
   if (Snapshot::IncludesCode(kind_)) {
     if (auto const type = ReadOnlyObjectType(cid)) {
-      return new (Z) RODataSerializationCluster(Z, type, cid);
+      return new (Z) RODataSerializationCluster(Z, type, cid, is_canonical);
     }
   }
 #endif
 
+  const bool cluster_represents_canonical_set =
+      current_loading_unit_id_ <= LoadingUnit::kRootId && is_canonical;
+
   switch (cid) {
     case kClassCid:
       return new (Z) ClassSerializationCluster(num_cids_ + num_tlc_cids_);
     case kTypeArgumentsCid:
-      return new (Z) TypeArgumentsSerializationCluster();
+      return new (Z)
+          TypeArgumentsSerializationCluster(cluster_represents_canonical_set);
     case kPatchClassCid:
       return new (Z) PatchClassSerializationCluster();
     case kFunctionCid:
@@ -5960,13 +6402,15 @@ SerializationCluster* Serializer::NewClusterForClass(intptr_t cid) {
     case kLibraryPrefixCid:
       return new (Z) LibraryPrefixSerializationCluster();
     case kTypeCid:
-      return new (Z) TypeSerializationCluster();
+      return new (Z) TypeSerializationCluster(cluster_represents_canonical_set);
     case kFunctionTypeCid:
-      return new (Z) FunctionTypeSerializationCluster();
+      return new (Z)
+          FunctionTypeSerializationCluster(cluster_represents_canonical_set);
     case kTypeRefCid:
       return new (Z) TypeRefSerializationCluster();
     case kTypeParameterCid:
-      return new (Z) TypeParameterSerializationCluster();
+      return new (Z)
+          TypeParameterSerializationCluster(cluster_represents_canonical_set);
     case kClosureCid:
       return new (Z) ClosureSerializationCluster();
     case kMintCid:
@@ -6195,11 +6639,15 @@ void Serializer::Trace(ObjectPtr object) {
     cid = object->GetClassId();
     is_canonical = object->untag()->IsCanonical();
   }
+  if (Snapshot::IncludesCode(kind_) && is_canonical && IsStringClassId(cid) &&
+      current_loading_unit_id_ <= LoadingUnit::kRootId) {
+    cid = kStringCid;
+  }
 
   SerializationCluster** cluster_ref =
       is_canonical ? &canonical_clusters_by_cid_[cid] : &clusters_by_cid_[cid];
   if (*cluster_ref == nullptr) {
-    *cluster_ref = NewClusterForClass(cid);
+    *cluster_ref = NewClusterForClass(cid, is_canonical);
     if (*cluster_ref == nullptr) {
       UnexpectedObject(object, "No serialization cluster defined");
     }
@@ -6673,13 +7121,16 @@ DeserializationCluster* Deserializer::ReadCluster() {
       case kPcDescriptorsCid:
       case kCodeSourceMapCid:
       case kCompressedStackMapsCid:
-        return new (Z) RODataDeserializationCluster(cid);
+        return new (Z) RODataDeserializationCluster(!is_non_root_unit_, cid);
       case kOneByteStringCid:
       case kTwoByteStringCid:
         if (!is_non_root_unit_) {
-          return new (Z) RODataDeserializationCluster(cid);
+          return new (Z) RODataDeserializationCluster(!is_non_root_unit_, cid);
         }
         break;
+      case kStringCid:
+        RELEASE_ASSERT(!is_non_root_unit_);
+        return new (Z) RODataDeserializationCluster(!is_non_root_unit_, cid);
     }
   }
 #endif
@@ -6688,7 +7139,7 @@ DeserializationCluster* Deserializer::ReadCluster() {
     case kClassCid:
       return new (Z) ClassDeserializationCluster();
     case kTypeArgumentsCid:
-      return new (Z) TypeArgumentsDeserializationCluster();
+      return new (Z) TypeArgumentsDeserializationCluster(!is_non_root_unit_);
     case kPatchClassCid:
       return new (Z) PatchClassDeserializationCluster();
     case kFunctionCid:
@@ -6742,13 +7193,13 @@ DeserializationCluster* Deserializer::ReadCluster() {
     case kLibraryPrefixCid:
       return new (Z) LibraryPrefixDeserializationCluster();
     case kTypeCid:
-      return new (Z) TypeDeserializationCluster();
+      return new (Z) TypeDeserializationCluster(!is_non_root_unit_);
     case kFunctionTypeCid:
-      return new (Z) FunctionTypeDeserializationCluster();
+      return new (Z) FunctionTypeDeserializationCluster(!is_non_root_unit_);
     case kTypeRefCid:
       return new (Z) TypeRefDeserializationCluster();
     case kTypeParameterCid:
-      return new (Z) TypeParameterDeserializationCluster();
+      return new (Z) TypeParameterDeserializationCluster(!is_non_root_unit_);
     case kClosureCid:
       return new (Z) ClosureDeserializationCluster();
     case kMintCid:
@@ -7251,7 +7702,8 @@ ZoneGrowableArray<Object*>* FullSnapshotWriter::WriteVMSnapshot() {
   serializer.ReserveHeader();
   serializer.WriteVersionAndFeatures(true);
   VMSerializationRoots roots(
-      Array::Handle(Dart::vm_isolate_group()->object_store()->symbol_table()));
+      Array::Handle(Dart::vm_isolate_group()->object_store()->symbol_table()),
+      /*should_write_symbols=*/!Snapshot::IncludesCode(kind_));
   ZoneGrowableArray<Object*>* objects = serializer.Serialize(&roots);
   serializer.FillHeader(serializer.kind());
   clustered_vm_size_ = serializer.bytes_written();
@@ -7293,7 +7745,7 @@ void FullSnapshotWriter::WriteProgramSnapshot(
 
   serializer.ReserveHeader();
   serializer.WriteVersionAndFeatures(false);
-  ProgramSerializationRoots roots(objects, object_store);
+  ProgramSerializationRoots roots(objects, object_store, kind_);
   objects = serializer.Serialize(&roots);
   if (units != nullptr) {
     (*units)[LoadingUnit::kRootId]->set_objects(objects);
diff --git a/runtime/vm/clustered_snapshot.h b/runtime/vm/clustered_snapshot.h
index 3262967f10c..2966da62f76 100644
--- a/runtime/vm/clustered_snapshot.h
+++ b/runtime/vm/clustered_snapshot.h
@@ -226,7 +226,7 @@ class Serializer : public ThreadStackResource {
   ObjectPtr ParentOf(const Object& object);
 #endif
 
-  SerializationCluster* NewClusterForClass(intptr_t cid);
+  SerializationCluster* NewClusterForClass(intptr_t cid, bool is_canonical);
 
   void ReserveHeader() {
     // Make room for recording snapshot buffer size.
diff --git a/runtime/vm/hash_table.h b/runtime/vm/hash_table.h
index ef895561b43..d065d8f71f8 100644
--- a/runtime/vm/hash_table.h
+++ b/runtime/vm/hash_table.h
@@ -10,6 +10,37 @@
 
 namespace dart {
 
+// Storage traits control how memory is allocated for HashTable.
+// Default ArrayStorageTraits use an Array to store HashTable contents.
+struct ArrayStorageTraits {
+  using ArrayHandle = Array;
+  using ArrayPtr = ArrayPtr;
+
+  static ArrayHandle& PtrToHandle(ArrayPtr ptr) { return Array::Handle(ptr); }
+
+  static void SetHandle(ArrayHandle& dst, const ArrayHandle& src) {  // NOLINT
+    dst = src.ptr();
+  }
+
+  static void ClearHandle(ArrayHandle& handle) {  // NOLINT
+    handle = Array::null();
+  }
+
+  static ArrayPtr New(Zone* zone, intptr_t length, Heap::Space space) {
+    return Array::New(length, space);
+  }
+
+  static bool IsImmutable(const ArrayHandle& handle) {
+    return handle.ptr()->untag()->InVMIsolateHeap();
+  }
+};
+
+class HashTableBase : public ValueObject {
+ public:
+  static const Object& UnusedMarker() { return Object::transition_sentinel(); }
+  static const Object& DeletedMarker() { return Object::sentinel(); }
+};
+
 // OVERVIEW:
 //
 // Hash maps and hash sets all use RawArray as backing storage. At the lowest
@@ -71,29 +102,34 @@ namespace dart {
 //    uword Hash(const Key& key) for any number of desired lookup key types.
 //  kPayloadSize: number of components of the payload in each entry.
 //  kMetaDataSize: number of elements reserved (e.g., for iteration order data).
-template <typename KeyTraits, intptr_t kPayloadSize, intptr_t kMetaDataSize>
-class HashTable : public ValueObject {
+template <typename KeyTraits,
+          intptr_t kPayloadSize,
+          intptr_t kMetaDataSize,
+          typename StorageTraits = ArrayStorageTraits>
+class HashTable : public HashTableBase {
  public:
   typedef KeyTraits Traits;
+  typedef StorageTraits Storage;
+
   // Uses the passed in handles for all handle operations.
   // 'Release' must be called at the end to obtain the final table
   // after potential growth/shrinkage.
-  HashTable(Object* key, Smi* index, Array* data)
+  HashTable(Object* key, Smi* index, typename StorageTraits::ArrayHandle* data)
       : key_handle_(key),
         smi_handle_(index),
         data_(data),
         released_data_(NULL) {}
   // Uses 'zone' for handle allocation. 'Release' must be called at the end
   // to obtain the final table after potential growth/shrinkage.
-  HashTable(Zone* zone, ArrayPtr data)
+  HashTable(Zone* zone, typename StorageTraits::ArrayPtr data)
       : key_handle_(&Object::Handle(zone)),
         smi_handle_(&Smi::Handle(zone)),
-        data_(&Array::Handle(zone, data)),
+        data_(&StorageTraits::PtrToHandle(data)),
         released_data_(NULL) {}
 
   // Returns the final table. The handle is cleared when this HashTable is
   // destroyed.
-  Array& Release() {
+  typename StorageTraits::ArrayHandle& Release() {
     ASSERT(data_ != NULL);
     ASSERT(released_data_ == NULL);
     // Ensure that no methods are called after 'Release'.
@@ -106,7 +142,7 @@ class HashTable : public ValueObject {
     // In DEBUG mode, calling 'Release' is mandatory.
     ASSERT(data_ == NULL);
     if (released_data_ != NULL) {
-      *released_data_ = Array::null();
+      StorageTraits::ClearHandle(*released_data_);
     }
   }
 
@@ -245,9 +281,6 @@ class HashTable : public ValueObject {
     NOT_IN_PRECOMPILED(ASSERT(NumOccupied() < NumEntries()));
   }
 
-  const Object& UnusedMarker() const { return Object::transition_sentinel(); }
-  const Object& DeletedMarker() const { return *data_; }
-
   bool IsUnused(intptr_t entry) const {
     return InternalGetKey(entry) == UnusedMarker().ptr();
   }
@@ -314,7 +347,7 @@ class HashTable : public ValueObject {
   }
   void UpdateCollisions(intptr_t collisions) const {
     if (KeyTraits::ReportStats()) {
-      if (data_->ptr()->untag()->InVMIsolateHeap()) {
+      if (Storage::IsImmutable(*data_)) {
         return;
       }
       AdjustSmiValueAt(kNumProbesIndex, collisions + 1);
@@ -403,10 +436,17 @@ class HashTable : public ValueObject {
   Object* key_handle_;
   Smi* smi_handle_;
   // Exactly one of these is non-NULL, depending on whether Release was called.
-  Array* data_;
-  Array* released_data_;
+  typename StorageTraits::ArrayHandle* data_;
+  typename StorageTraits::ArrayHandle* released_data_;
 
   friend class HashTables;
+  template <typename Table, bool kAllCanonicalObjectsAreIncludedIntoSet>
+  friend class CanonicalSetDeserializationCluster;
+  template <typename Table,
+            typename HandleType,
+            typename PointerType,
+            bool kAllCanonicalObjectsAreIncludedIntoSet>
+  friend class CanonicalSetSerializationCluster;
 };
 
 // Table with unspecified iteration order. No payload overhead or metadata.
@@ -448,17 +488,20 @@ class HashTables : public AllStatic {
  public:
   // Allocates and initializes a table.
   template <typename Table>
-  static ArrayPtr New(intptr_t initial_capacity,
-                      Heap::Space space = Heap::kNew) {
+  static typename Table::Storage::ArrayPtr New(intptr_t initial_capacity,
+                                               Heap::Space space = Heap::kNew) {
+    auto zone = Thread::Current()->zone();
     Table table(
-        Thread::Current()->zone(),
-        Array::New(Table::ArrayLengthForNumOccupied(initial_capacity), space));
+        zone,
+        Table::Storage::New(
+            zone, Table::ArrayLengthForNumOccupied(initial_capacity), space));
     table.Initialize();
     return table.Release().ptr();
   }
 
   template <typename Table>
-  static ArrayPtr New(const Array& array) {
+  static typename Table::Storage::ArrayPtr New(
+      const typename Table::Storage::ArrayHandle& array) {
     Table table(Thread::Current()->zone(), array.ptr());
     table.Initialize();
     return table.Release().ptr();
@@ -513,7 +556,7 @@ class HashTables : public AllStatic {
     Table new_table(New<Table>(new_capacity,  // Is rounded up to power of 2.
                                table.data_->IsOld() ? Heap::kOld : Heap::kNew));
     Copy(table, new_table);
-    *table.data_ = new_table.Release().ptr();
+    Table::Storage::SetHandle(*table.data_, new_table.Release());
     NOT_IN_PRODUCT(table.UpdateGrowth(); table.PrintStats();)
   }
 
@@ -547,7 +590,8 @@ class HashTables : public AllStatic {
       for (intptr_t i = 0; i < table.Length(); i++) {
         element = table.At(i);
         if (!element.IsSmi()) {
-          element = WeakSerializationReference::New(element, table);
+          element = WeakSerializationReference::New(
+              element, HashTableBase::DeletedMarker());
           table.SetAt(i, element);
         }
       }
diff --git a/runtime/vm/object.cc b/runtime/vm/object.cc
index dc3e61d52d4..91c228e5bec 100644
--- a/runtime/vm/object.cc
+++ b/runtime/vm/object.cc
@@ -20339,6 +20339,7 @@ bool Type::IsDeclarationTypeOf(const Class& cls) const {
   return nullability() == Nullability::kNonNullable;
 }
 
+// Keep in sync with TypeSerializationCluster::IsInCanonicalSet.
 AbstractTypePtr Type::Canonicalize(Thread* thread, TrailPtr trail) const {
   Zone* zone = thread->zone();
   ASSERT(IsFinalized());
diff --git a/runtime/vm/object.h b/runtime/vm/object.h
index 01cfb33fe18..205665ca811 100644
--- a/runtime/vm/object.h
+++ b/runtime/vm/object.h
@@ -1740,6 +1740,7 @@ class Class : public Object {
   friend class Intrinsifier;
   friend class ProgramWalker;
   friend class Precompiler;
+  friend class ClassFinalizer;
 };
 
 // Classification of type genericity according to type parameter owners.
diff --git a/runtime/vm/raw_object.h b/runtime/vm/raw_object.h
index f53df118ca5..7a9ceb51364 100644
--- a/runtime/vm/raw_object.h
+++ b/runtime/vm/raw_object.h
@@ -908,6 +908,7 @@ class UntaggedClass : public UntaggedObject {
   friend class UntaggedTypeArguments;
   friend class SnapshotReader;
   friend class InstanceSerializationCluster;
+  friend class TypeSerializationCluster;
   friend class CidRewriteVisitor;
   friend class Api;
 };
@@ -2744,7 +2745,8 @@ class UntaggedArray : public UntaggedInstance {
   friend class ICData;            // For high performance access.
   friend class SubtypeTestCache;  // For high performance access.
   friend class ReversePc;
-
+  template <typename Table, bool kAllCanonicalObjectsAreIncludedIntoSet>
+  friend class CanonicalSetDeserializationCluster;
   friend class OldPage;
 };