FragmentedPart (#1818)

Summary:

This sets us up for the next diff by creating `FragmentedPart` 

Currently our pickle is still not truly zero copy because the Pickler calls `Buffer::write()` which is copying bytes from `PyBytes` to `BytesMut` via `extend_from_slice()`. 

This is especially problematic for large messages (100KB+) as we are spending a lot of CPU cycles handling page faults. For a 1MB message pickling can take as long as 600us

To avoid copies, we can just make `Buffer` backed by a `Vec<PyBytes>` with each call to `Buffer::write()` pushing the PyBytes to the Vec. As a result of this, the PyBytes are physically fragmented despite being logically contiguous. 

To make this work, we will have a NewType with called `FragmentedPart` with a `::Fragmented` variant wrapping `Vec<Part>` and a `::Contiguous` variant wrapping `Part`. Similar to `Part`, `FragmentedPart` also just collects during serialization. When we receive the frame on the other end of the wire, we reconstruct it contiguously in the `FragmentedPart::Contiguous` variant so that we can easily consume it to create a single contiguous `bytes::Bytes`

Differential Revision: D86696390

Author: thomasywang

serde_multipart/src/de/bincode.rs

@@ -15,21 +15,32 @@ use bincode::ErrorKind;
 use bincode::Options;
 use serde::de::IntoDeserializer;
 
+use crate::FragmentedPart;
 use crate::part::Part;
 
 /// Multipart deserializer for bincode. This passes through to the underlying bincode
-/// deserializer, but dequeues serialized parts when they are needed by [`Part::deserialize`].
+/// deserializer, but dequeues serialized parts when they are needed by
+/// [`Part::deserialize`] or [`FragmentedPart::deserialize`].
 pub struct Deserializer<R, O: Options> {
     de: bincode::Deserializer<R, O>,
     parts: VecDeque<Part>,
+    fragmented_parts: VecDeque<FragmentedPart>,
 }
 
 impl<R, O> Deserializer<R, O>
 where
     O: Options,
 {
-    pub(crate) fn new(de: bincode::Deserializer<R, O>, parts: VecDeque<Part>) -> Self {
-        Self { de, parts }
+    pub(crate) fn new(
+        de: bincode::Deserializer<R, O>,
+        parts: VecDeque<Part>,
+        fragmented_parts: VecDeque<FragmentedPart>,
+    ) -> Self {
+        Self {
+            de,
+            parts,
+            fragmented_parts,
+        }
     }
 
     pub(crate) fn deserialize_part(&mut self) -> Result<Part, Error> {
@@ -38,8 +49,14 @@ where
         })
     }
 
+    pub(crate) fn deserialize_fragmented_part(&mut self) -> Result<FragmentedPart, Error> {
+        self.fragmented_parts.pop_front().ok_or_else(|| {
+            ErrorKind::Custom("fragmented part underrun while decoding".to_string()).into()
+        })
+    }
+
     pub(crate) fn end(self) -> Result<(), Error> {
-        if self.parts.is_empty() {
+        if self.parts.is_empty() && self.fragmented_parts.is_empty() {
             Ok(())
         } else {
             Err(ErrorKind::Custom("multipart overrun while decoding".to_string()).into())