From 98b66e4b9342787c09e95e28b27ee05b3d74c81c Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Sat, 7 Jan 2023 10:03:18 +0100
Subject: [PATCH 1/9] feat: Arena allocator

---
 stsync-proxy/src/utils/arena.rs | 235 ++++++++++++++++++++++++++++++++
 stsync-proxy/src/utils/mod.rs   |   1 +
 2 files changed, 236 insertions(+)
 create mode 100644 stsync-proxy/src/utils/arena.rs

diff --git a/stsync-proxy/src/utils/arena.rs b/stsync-proxy/src/utils/arena.rs
new file mode 100644
index 0000000..fa98e47
--- /dev/null
+++ b/stsync-proxy/src/utils/arena.rs
@@ -0,0 +1,235 @@
+//! A fast arena allocator for network buffers.
+
+use std::alloc::Layout;
+use std::alloc::{AllocError, Allocator};
+use std::cell::UnsafeCell;
+use std::ptr::NonNull;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+use parking_lot::RwLock;
+
+const PAGE: usize = 65536;
+
+#[derive(Clone, Debug)]
+pub struct Arena {
+    inner: Arc<ArenaInner>,
+}
+
+#[derive(Debug)]
+struct ArenaInner {
+    page_size: usize,
+    chunks: RwLock<Vec<(Chunk, *const u8)>>,
+}
+
+impl Arena {
+    #[inline]
+    pub fn new(page_size: usize) -> Self {
+        Self {
+            inner: Arc::new(ArenaInner {
+                page_size,
+                chunks: RwLock::new(Vec::new()),
+            }),
+        }
+    }
+
+    unsafe fn alloc(&self, mut layout: Layout) -> NonNull<[u8]> {
+        layout = layout.pad_to_align();
+
+        if layout.size() > self.inner.page_size {
+            panic!("exceeded page size");
+            // The requested layout is bigger than the maximum chunk size. We can never
+            // allocate this object with this arena. Forward to global allocator.
+            // return unsafe { NonNull::new_unchecked(std::alloc::alloc(layout)) };
+            // Equivalent to ptr.cast() but is unsized.
+        }
+
+        let chunks = self.inner.chunks.read();
+        for (chunk, _) in &*chunks {
+            if let Ok(ptr) = unsafe { chunk.allocate_padded(layout) } {
+                return ptr;
+            }
+        }
+
+        let (chunk, chunk_ptr) = self.make_chunk();
+
+        drop(chunks);
+        let mut chunks = self.inner.chunks.write();
+        let ptr = unsafe { chunk.allocate_padded(layout).unwrap() };
+
+        chunks.push((chunk, chunk_ptr));
+        ptr
+    }
+
+    unsafe fn dealloc(&self, ptr: NonNull<u8>, layout: Layout) {
+        let mut chunks = self.inner.chunks.write();
+        for (i, (chunk, base)) in chunks.iter().enumerate() {
+            // Pointer within base + page_size; the pointer is a part of this chunk.
+            if ptr.as_ptr() as usize <= *base as usize + self.inner.page_size {
+                // SAFETY: `ptr` was previously allocated in this `chunk`.
+                unsafe {
+                    chunk.deallocate(layout, ptr);
+                }
+
+                chunks.remove(i);
+                break;
+            }
+        }
+    }
+
+    #[inline]
+    fn make_chunk(&self) -> (Chunk, *const u8) {
+        let layout = unsafe {
+            Layout::from_size_align_unchecked(
+                std::mem::size_of::<u8>() * self.inner.page_size,
+                std::mem::align_of::<usize>(),
+            )
+        };
+
+        let ptr = unsafe { std::alloc::alloc(layout) };
+        assert!(!ptr.is_null());
+
+        let ptr = unsafe { NonNull::new_unchecked(ptr) };
+
+        (
+            Chunk {
+                arena: self.clone(),
+                ptr,
+                layout,
+                head: AtomicUsize::new(0),
+                entries: AtomicUsize::new(0),
+            },
+            ptr.as_ptr(),
+        )
+    }
+}
+
+impl Default for Arena {
+    #[inline]
+    fn default() -> Self {
+        Self::new(PAGE)
+    }
+}
+
+unsafe impl Allocator for Arena {
+    fn allocate(&self, layout: Layout) -> Result<NonNull<[u8]>, std::alloc::AllocError> {
+        unsafe { Ok(self.alloc(layout)) }
+    }
+
+    unsafe fn deallocate(&self, ptr: NonNull<u8>, layout: Layout) {
+        unsafe { self.dealloc(ptr, layout) };
+    }
+}
+
+/// A fixed size memory page.
+#[derive(Debug)]
+struct Chunk {
+    arena: Arena,
+    ptr: NonNull<u8>,
+    layout: Layout,
+    head: AtomicUsize,
+    /// The number of active references to the buffer of this chunk.
+    entries: AtomicUsize,
+}
+
+impl Chunk {
+    /// # Safety
+    ///
+    /// The `Chunk` must have enough [`contiguous_capacity`] to allocate using `layout`.
+    ///
+    /// [`contiguous_capacity`]: Self::contiguous_capacity
+    unsafe fn allocate_padded(&self, layout: Layout) -> Result<NonNull<[u8]>, AllocError> {
+        // Acquire the head offset.
+        let mut head = self.head.load(Ordering::SeqCst);
+        if head > head + layout.size() {
+            return Err(AllocError);
+        }
+
+        while let Err(current) = self.head.compare_exchange_weak(
+            head,
+            head + layout.size(),
+            Ordering::SeqCst,
+            Ordering::SeqCst,
+        ) {
+            head = current;
+
+            // Another thread won the race and we no longer have enough space to
+            // allocate using this `layout`.
+            if head > head + layout.size() {
+                return Err(AllocError);
+            }
+        }
+
+        self.entries.fetch_add(1, Ordering::SeqCst);
+
+        let ptr = unsafe { NonNull::new_unchecked(self.ptr.as_ptr().add(head)) };
+        let len = layout.size();
+
+        // SAFETY: The caller guarantees that `end < buffer.len()`.
+        // let fatptr = [self.buffer.as_ptr() as usize + head, layout.size()];
+        // Ok(unsafe { std::mem::transmute::<[usize; 2], NonNull<[u8]>>(fatptr) })
+        // Ok(unsafe { self.buffer.get_unchecked(head..end).into() })
+        Ok(NonNull::from_raw_parts(ptr.cast(), len))
+    }
+
+    /// Deallocate given memory block owned by this `Chunk`.
+    ///
+    /// # Safety
+    ///
+    /// The given `ptr` must be owned by this `Chunk`, i.e. have been previously allocated using
+    /// `allocate` of the same `Chunk`. The `layout` must be same layout used to allocate.
+    unsafe fn deallocate(&self, layout: Layout, ptr: NonNull<u8>) {
+        self.entries.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
+impl Drop for Chunk {
+    fn drop(&mut self) {
+        unsafe {
+            std::alloc::dealloc(self.ptr.as_ptr(), self.layout);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::alloc::{Allocator, Layout};
+    use std::slice;
+
+    use super::{Arena, PAGE};
+
+    #[test]
+    fn test_arena() {
+        let arena = Arena::default();
+        let ptr = arena.allocate(Layout::new::<u8>()).unwrap();
+
+        unsafe { arena.deallocate(ptr.to_raw_parts().0.cast(), Layout::new::<u8>()) };
+
+        let mut ptr = arena.allocate(Layout::array::<u8>(PAGE).unwrap()).unwrap();
+        let slice = unsafe { ptr.as_mut() };
+        for b in slice {
+            *b = 25;
+        }
+
+        unsafe {
+            arena.deallocate(
+                ptr.to_raw_parts().0.cast(),
+                Layout::array::<u8>(PAGE).unwrap(),
+            );
+        }
+    }
+
+    #[test]
+    fn test_arena_vec() {
+        let arena = Arena::default();
+
+        let mut vec = Vec::new_in(arena);
+        vec.push("Hello World");
+        vec.push("str1");
+        vec.push("str2");
+        vec.push("str3");
+        vec.truncate(0);
+        vec.shrink_to_fit();
+        drop(vec);
+    }
+}
diff --git a/stsync-proxy/src/utils/mod.rs b/stsync-proxy/src/utils/mod.rs
index 86e0ab3..5d63a72 100644
--- a/stsync-proxy/src/utils/mod.rs
+++ b/stsync-proxy/src/utils/mod.rs
@@ -1,3 +1,4 @@
+pub mod arena;
 pub mod serial;
 
 use std::cmp::Ordering;

From 03a659a421387db38544976db305ffa77e338c49 Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Thu, 12 Jan 2023 20:35:26 +0100
Subject: [PATCH 2/9] Arena allocator take 2

---
 stsync-proxy/src/utils/arena.rs | 667 ++++++++++++++++++++++++--------
 1 file changed, 515 insertions(+), 152 deletions(-)

diff --git a/stsync-proxy/src/utils/arena.rs b/stsync-proxy/src/utils/arena.rs
index fa98e47..4101c61 100644
--- a/stsync-proxy/src/utils/arena.rs
+++ b/stsync-proxy/src/utils/arena.rs
@@ -1,235 +1,598 @@
 //! A fast arena allocator for network buffers.
 
 use std::alloc::Layout;
-use std::alloc::{AllocError, Allocator};
-use std::cell::UnsafeCell;
+use std::ops::{Deref, DerefMut};
 use std::ptr::NonNull;
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::slice;
+use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
 
-use parking_lot::RwLock;
+use bytes::{Bytes, Vtable};
+use parking_lot::Mutex;
 
 const PAGE: usize = 65536;
 
-#[derive(Clone, Debug)]
-pub struct Arena {
-    inner: Arc<ArenaInner>,
-}
-
 #[derive(Debug)]
-struct ArenaInner {
-    page_size: usize,
-    chunks: RwLock<Vec<(Chunk, *const u8)>>,
+pub struct Arena {
+    chunk_size: usize,
+    // FIXME: Maybe this can be a lock-free intrusive doubly linked list.
+    chunks: Mutex<Vec<Chunk>>,
 }
 
 impl Arena {
-    #[inline]
-    pub fn new(page_size: usize) -> Self {
+    pub fn new(chunk_size: usize) -> Self {
         Self {
-            inner: Arc::new(ArenaInner {
-                page_size,
-                chunks: RwLock::new(Vec::new()),
-            }),
+            chunk_size,
+            chunks: Mutex::new(Vec::new()),
         }
     }
 
-    unsafe fn alloc(&self, mut layout: Layout) -> NonNull<[u8]> {
-        layout = layout.pad_to_align();
-
-        if layout.size() > self.inner.page_size {
-            panic!("exceeded page size");
-            // The requested layout is bigger than the maximum chunk size. We can never
-            // allocate this object with this arena. Forward to global allocator.
-            // return unsafe { NonNull::new_unchecked(std::alloc::alloc(layout)) };
-            // Equivalent to ptr.cast() but is unsized.
+    pub fn get(&self, size: usize) -> BytesMut {
+        if size > self.chunk_size {
+            panic!("Cannot allocate larger than chunk size");
         }
 
-        let chunks = self.inner.chunks.read();
-        for (chunk, _) in &*chunks {
-            if let Ok(ptr) = unsafe { chunk.allocate_padded(layout) } {
-                return ptr;
+        let mut chunks = self.chunks.lock();
+
+        for chunk in &*chunks {
+            // We have the only pointer to the chunk. The chunk allocation
+            // can be reused.
+            if chunk.ref_count.load(Ordering::Relaxed) == 1 {
+                unsafe {
+                    chunk.reset();
+                }
+            }
+
+            if let Some(buf) = chunk.get(size) {
+                return buf;
             }
         }
 
-        let (chunk, chunk_ptr) = self.make_chunk();
+        let chunk = Chunk::new(self.chunk_size);
+        let buf = chunk.get(size).unwrap();
+        chunks.push(chunk);
 
-        drop(chunks);
-        let mut chunks = self.inner.chunks.write();
-        let ptr = unsafe { chunk.allocate_padded(layout).unwrap() };
+        buf
+    }
 
-        chunks.push((chunk, chunk_ptr));
-        ptr
+    pub fn chunks(&self) -> usize {
+        let chunks = self.chunks.lock();
+        chunks.len()
     }
+}
 
-    unsafe fn dealloc(&self, ptr: NonNull<u8>, layout: Layout) {
-        let mut chunks = self.inner.chunks.write();
-        for (i, (chunk, base)) in chunks.iter().enumerate() {
-            // Pointer within base + page_size; the pointer is a part of this chunk.
-            if ptr.as_ptr() as usize <= *base as usize + self.inner.page_size {
-                // SAFETY: `ptr` was previously allocated in this `chunk`.
-                unsafe {
-                    chunk.deallocate(layout, ptr);
-                }
+impl Default for Arena {
+    #[inline]
+    fn default() -> Self {
+        Self::new(PAGE)
+    }
+}
 
-                chunks.remove(i);
-                break;
-            }
+#[derive(Debug)]
+pub struct BytesMut {
+    chunk: NonNull<ChunkInner>,
+    ptr: NonNull<u8>,
+    len: usize,
+}
+
+impl BytesMut {
+    pub fn truncate(&mut self, len: usize) {
+        if len <= self.len() {
+            unsafe { self.set_len(len) };
         }
     }
 
+    pub fn freeze(self) -> Bytes {
+        let ptr = self.ptr.as_ptr();
+        let len = self.len;
+        let data = AtomicPtr::new(self.chunk.as_ptr().cast());
+
+        // Do not decrement the reference count.
+        std::mem::forget(self);
+
+        unsafe { Bytes::with_vtable(ptr, len, data, &SHARED_VTABLE) }
+    }
+
+    pub unsafe fn set_len(&mut self, len: usize) {
+        self.len = len;
+    }
+
     #[inline]
-    fn make_chunk(&self) -> (Chunk, *const u8) {
-        let layout = unsafe {
-            Layout::from_size_align_unchecked(
-                std::mem::size_of::<u8>() * self.inner.page_size,
-                std::mem::align_of::<usize>(),
-            )
-        };
+    fn as_slice(&self) -> &[u8] {
+        unsafe { slice::from_raw_parts(self.ptr.as_ptr(), self.len) }
+    }
 
-        let ptr = unsafe { std::alloc::alloc(layout) };
-        assert!(!ptr.is_null());
+    #[inline]
+    fn as_slice_mut(&mut self) -> &mut [u8] {
+        unsafe { slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) }
+    }
+}
 
-        let ptr = unsafe { NonNull::new_unchecked(ptr) };
+impl AsRef<[u8]> for BytesMut {
+    fn as_ref(&self) -> &[u8] {
+        self.as_slice()
+    }
+}
 
-        (
-            Chunk {
-                arena: self.clone(),
-                ptr,
-                layout,
-                head: AtomicUsize::new(0),
-                entries: AtomicUsize::new(0),
-            },
-            ptr.as_ptr(),
-        )
+impl AsMut<[u8]> for BytesMut {
+    fn as_mut(&mut self) -> &mut [u8] {
+        self.as_slice_mut()
     }
 }
 
-impl Default for Arena {
+impl Deref for BytesMut {
+    type Target = [u8];
+
     #[inline]
-    fn default() -> Self {
-        Self::new(PAGE)
+    fn deref(&self) -> &Self::Target {
+        self.as_slice()
     }
 }
 
-unsafe impl Allocator for Arena {
-    fn allocate(&self, layout: Layout) -> Result<NonNull<[u8]>, std::alloc::AllocError> {
-        unsafe { Ok(self.alloc(layout)) }
+impl DerefMut for BytesMut {
+    #[inline]
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.as_slice_mut()
     }
+}
 
-    unsafe fn deallocate(&self, ptr: NonNull<u8>, layout: Layout) {
-        unsafe { self.dealloc(ptr, layout) };
+impl Drop for BytesMut {
+    fn drop(&mut self) {
+        let chunk = unsafe { self.chunk.as_ref() };
+
+        let rc = chunk.ref_count.fetch_sub(1, Ordering::Release);
+
+        if rc != 1 {
+            return;
+        }
+
+        unsafe { Box::from_raw(self.ptr.as_ptr()) };
     }
 }
 
-/// A fixed size memory page.
+unsafe impl Send for BytesMut {}
+unsafe impl Sync for BytesMut {}
+
+/// A strong reference to a chunk.
 #[derive(Debug)]
 struct Chunk {
-    arena: Arena,
-    ptr: NonNull<u8>,
+    ptr: NonNull<ChunkInner>,
+}
+
+impl Chunk {
+    pub fn new(size: usize) -> Self {
+        let mut inner = Box::new(ChunkInner::new(size));
+        let ptr = Box::leak(inner).into();
+
+        Self { ptr }
+    }
+
+    pub fn get(&self, size: usize) -> Option<BytesMut> {
+        let inner = unsafe { self.ptr.as_ref() };
+        let ptr = inner.get(size)?;
+
+        unsafe {
+            Some(BytesMut {
+                chunk: self.ptr,
+                ptr: NonNull::new_unchecked(ptr),
+                len: size,
+            })
+        }
+    }
+}
+
+impl Deref for Chunk {
+    type Target = ChunkInner;
+
+    fn deref(&self) -> &Self::Target {
+        unsafe { self.ptr.as_ref() }
+    }
+}
+
+impl Drop for Chunk {
+    fn drop(&mut self) {
+        let chunk = unsafe { self.ptr.as_ref() };
+
+        let rc = chunk.ref_count.fetch_sub(1, Ordering::Release);
+
+        if rc != 1 {
+            return;
+        }
+
+        // Drop the allocation
+        unsafe { Box::from_raw(self.ptr.as_ptr()) };
+    }
+}
+
+unsafe impl Send for Chunk {}
+unsafe impl Sync for Chunk {}
+
+#[derive(Debug)]
+struct ChunkInner {
+    /// The layout used to allocate this chunk.
     layout: Layout,
+    /// The starting pointer of the allocated chunk.
+    ptr: *mut u8,
+    /// Where to put the next bytes block.
     head: AtomicUsize,
-    /// The number of active references to the buffer of this chunk.
-    entries: AtomicUsize,
+    /// Strong reference counts
+    ref_count: AtomicUsize,
 }
 
-impl Chunk {
-    /// # Safety
-    ///
-    /// The `Chunk` must have enough [`contiguous_capacity`] to allocate using `layout`.
-    ///
-    /// [`contiguous_capacity`]: Self::contiguous_capacity
-    unsafe fn allocate_padded(&self, layout: Layout) -> Result<NonNull<[u8]>, AllocError> {
-        // Acquire the head offset.
-        let mut head = self.head.load(Ordering::SeqCst);
-        if head > head + layout.size() {
-            return Err(AllocError);
+impl ChunkInner {
+    pub fn new(size: usize) -> Self {
+        let layout = Layout::array::<u8>(size).unwrap();
+
+        let ptr = unsafe { std::alloc::alloc(layout) };
+        assert!(!ptr.is_null());
+
+        Self {
+            layout,
+            ptr,
+            head: AtomicUsize::new(0),
+            ref_count: AtomicUsize::new(1),
         }
+    }
 
-        while let Err(current) = self.head.compare_exchange_weak(
-            head,
-            head + layout.size(),
-            Ordering::SeqCst,
-            Ordering::SeqCst,
-        ) {
-            head = current;
-
-            // Another thread won the race and we no longer have enough space to
-            // allocate using this `layout`.
-            if head > head + layout.size() {
-                return Err(AllocError);
-            }
+    pub fn get(&self, size: usize) -> Option<*mut u8> {
+        let mut head = self.head.load(Ordering::Acquire);
+
+        // Check that this chunk has enough free capacity to hold size bytes.
+        if head + size > self.layout.size() {
+            return None;
         }
 
-        self.entries.fetch_add(1, Ordering::SeqCst);
+        while let Err(curr) =
+            self.head
+                .compare_exchange_weak(head, head + size, Ordering::SeqCst, Ordering::SeqCst)
+        {
+            head = curr;
 
-        let ptr = unsafe { NonNull::new_unchecked(self.ptr.as_ptr().add(head)) };
-        let len = layout.size();
+            if head + size > self.layout.size() {
+                return None;
+            }
+        }
+
+        self.ref_count.fetch_add(1, Ordering::SeqCst);
 
-        // SAFETY: The caller guarantees that `end < buffer.len()`.
-        // let fatptr = [self.buffer.as_ptr() as usize + head, layout.size()];
-        // Ok(unsafe { std::mem::transmute::<[usize; 2], NonNull<[u8]>>(fatptr) })
-        // Ok(unsafe { self.buffer.get_unchecked(head..end).into() })
-        Ok(NonNull::from_raw_parts(ptr.cast(), len))
+        // SAFETY: We have exlusive access to ptr + head.
+        unsafe { Some(self.ptr.add(head)) }
     }
 
-    /// Deallocate given memory block owned by this `Chunk`.
-    ///
-    /// # Safety
-    ///
-    /// The given `ptr` must be owned by this `Chunk`, i.e. have been previously allocated using
-    /// `allocate` of the same `Chunk`. The `layout` must be same layout used to allocate.
-    unsafe fn deallocate(&self, layout: Layout, ptr: NonNull<u8>) {
-        self.entries.fetch_sub(1, Ordering::SeqCst);
+    unsafe fn reset(&self) {
+        // The chunk may immediately be reused with self.get() which uses
+        // acquire loads, hence the release store is necessary.
+        self.head.store(0, Ordering::Release);
     }
 }
 
-impl Drop for Chunk {
+impl Drop for ChunkInner {
     fn drop(&mut self) {
         unsafe {
-            std::alloc::dealloc(self.ptr.as_ptr(), self.layout);
+            std::alloc::dealloc(self.ptr, self.layout);
         }
     }
 }
 
+// ==== impl Vtable ====
+
+const SHARED_VTABLE: Vtable = Vtable {
+    clone: chunk_clone,
+    to_vec: chunk_to_vec,
+    drop: chunk_drop,
+};
+
+unsafe fn chunk_clone(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Bytes {
+    let shared = data.load(Ordering::Relaxed) as *mut ChunkInner;
+
+    let old = unsafe { (*shared).ref_count.fetch_add(1, Ordering::Relaxed) };
+
+    if old > usize::MAX >> 1 {
+        std::process::abort();
+    }
+
+    unsafe { Bytes::with_vtable(ptr, len, AtomicPtr::new(shared as *mut ()), &SHARED_VTABLE) }
+}
+
+unsafe fn chunk_to_vec(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8> {
+    let mut buf = Vec::with_capacity(len);
+    unsafe { std::ptr::copy_nonoverlapping(ptr, buf.as_mut_ptr(), len) };
+    buf
+}
+
+unsafe fn chunk_drop(data: &mut AtomicPtr<()>, ptr: *const u8, len: usize) {
+    let shared = data.load(Ordering::Relaxed) as *mut ChunkInner;
+
+    let rc = unsafe { (*shared).ref_count.fetch_sub(1, Ordering::Release) };
+
+    if rc != 1 {
+        return;
+    }
+
+    // Fence to prevent reordering of data access after deletion.
+    // Syncs with relase load (fetch_sub) above.
+    unsafe { (*shared).ref_count.load(Ordering::Acquire) };
+
+    unsafe { Box::from_raw(shared) };
+}
+
+// #[derive(Debug)]
+// struct ArenaInner {
+//     page_size: usize,
+//     chunks: RwLock<Vec<(Chunk, *const u8)>>,
+// }
+
+// impl Arena {
+//     #[inline]
+//     pub fn new(page_size: usize) -> Self {
+//         Self {
+//             inner: Arc::new(ArenaInner {
+//                 page_size,
+//                 chunks: RwLock::new(Vec::new()),
+//             }),
+//         }
+//     }
+
+//     unsafe fn alloc(&self, mut layout: Layout) -> NonNull<[u8]> {
+//         layout = layout.pad_to_align();
+
+//         if layout.size() > self.inner.page_size {
+//             panic!("exceeded page size");
+//             // The requested layout is bigger than the maximum chunk size. We can never
+//             // allocate this object with this arena. Forward to global allocator.
+//             // return unsafe { NonNull::new_unchecked(std::alloc::alloc(layout)) };
+//             // Equivalent to ptr.cast() but is unsized.
+//         }
+
+//         let chunks = self.inner.chunks.read();
+//         for (chunk, _) in &*chunks {
+//             if let Ok(ptr) = unsafe { chunk.allocate_padded(layout) } {
+//                 return ptr;
+//             }
+//         }
+
+//         let (chunk, chunk_ptr) = self.make_chunk();
+
+//         drop(chunks);
+//         let mut chunks = self.inner.chunks.write();
+//         let ptr = unsafe { chunk.allocate_padded(layout).unwrap() };
+
+//         chunks.push((chunk, chunk_ptr));
+//         ptr
+//     }
+
+//     unsafe fn dealloc(&self, ptr: NonNull<u8>, layout: Layout) {
+//         let mut chunks = self.inner.chunks.write();
+//         for (i, (chunk, base)) in chunks.iter().enumerate() {
+//             // Pointer within base + page_size; the pointer is a part of this chunk.
+//             if ptr.as_ptr() as usize <= *base as usize + self.inner.page_size {
+//                 // SAFETY: `ptr` was previously allocated in this `chunk`.
+//                 unsafe {
+//                     chunk.deallocate(layout, ptr);
+//                 }
+
+//                 chunks.remove(i);
+//                 break;
+//             }
+//         }
+//     }
+
+//     #[inline]
+//     fn make_chunk(&self) -> (Chunk, *const u8) {
+//         let layout = unsafe {
+//             Layout::from_size_align_unchecked(
+//                 std::mem::size_of::<u8>() * self.inner.page_size,
+//                 std::mem::align_of::<usize>(),
+//             )
+//         };
+
+//         let ptr = unsafe { std::alloc::alloc(layout) };
+//         assert!(!ptr.is_null());
+
+//         let ptr = unsafe { NonNull::new_unchecked(ptr) };
+
+//         (
+//             Chunk {
+//                 arena: self.clone(),
+//                 ptr,
+//                 layout,
+//                 head: AtomicUsize::new(0),
+//                 entries: AtomicUsize::new(0),
+//             },
+//             ptr.as_ptr(),
+//         )
+//     }
+// }
+
+// impl Default for Arena {
+//     #[inline]
+//     fn default() -> Self {
+//         Self::new(PAGE)
+//     }
+// }
+
+// unsafe impl Allocator for Arena {
+//     fn allocate(&self, layout: Layout) -> Result<NonNull<[u8]>, std::alloc::AllocError> {
+//         unsafe { Ok(self.alloc(layout)) }
+//     }
+
+//     unsafe fn deallocate(&self, ptr: NonNull<u8>, layout: Layout) {
+//         unsafe { self.dealloc(ptr, layout) };
+//     }
+// }
+
+// /// A fixed size memory page.
+// #[derive(Debug)]
+// struct Chunk {
+//     arena: Arena,
+//     ptr: NonNull<u8>,
+//     layout: Layout,
+//     head: AtomicUsize,
+//     /// The number of active references to the buffer of this chunk.
+//     entries: AtomicUsize,
+// }
+
+// impl Chunk {
+//     /// # Safety
+//     ///
+//     /// The `Chunk` must have enough [`contiguous_capacity`] to allocate using `layout`.
+//     ///
+//     /// [`contiguous_capacity`]: Self::contiguous_capacity
+//     unsafe fn allocate_padded(&self, layout: Layout) -> Result<NonNull<[u8]>, AllocError> {
+//         // Acquire the head offset.
+//         let mut head = self.head.load(Ordering::SeqCst);
+//         if head > head + layout.size() {
+//             return Err(AllocError);
+//         }
+
+//         while let Err(current) = self.head.compare_exchange_weak(
+//             head,
+//             head + layout.size(),
+//             Ordering::SeqCst,
+//             Ordering::SeqCst,
+//         ) {
+//             head = current;
+
+//             // Another thread won the race and we no longer have enough space to
+//             // allocate using this `layout`.
+//             if head > head + layout.size() {
+//                 return Err(AllocError);
+//             }
+//         }
+
+//         self.entries.fetch_add(1, Ordering::SeqCst);
+
+//         let ptr = unsafe { NonNull::new_unchecked(self.ptr.as_ptr().add(head)) };
+//         let len = layout.size();
+
+//         // SAFETY: The caller guarantees that `end < buffer.len()`.
+//         // let fatptr = [self.buffer.as_ptr() as usize + head, layout.size()];
+//         // Ok(unsafe { std::mem::transmute::<[usize; 2], NonNull<[u8]>>(fatptr) })
+//         // Ok(unsafe { self.buffer.get_unchecked(head..end).into() })
+//         Ok(NonNull::from_raw_parts(ptr.cast(), len))
+//     }
+
+//     /// Deallocate given memory block owned by this `Chunk`.
+//     ///
+//     /// # Safety
+//     ///
+//     /// The given `ptr` must be owned by this `Chunk`, i.e. have been previously allocated using
+//     /// `allocate` of the same `Chunk`. The `layout` must be same layout used to allocate.
+//     unsafe fn deallocate(&self, layout: Layout, ptr: NonNull<u8>) {
+//         self.entries.fetch_sub(1, Ordering::SeqCst);
+//     }
+// }
+
+// impl Drop for Chunk {
+//     fn drop(&mut self) {
+//         unsafe {
+//             std::alloc::dealloc(self.ptr.as_ptr(), self.layout);
+//         }
+//     }
+// }
+
+// #[cfg(test)]
+// mod tests {
+//     use std::alloc::{Allocator, Layout};
+//     use std::slice;
+
+//     use super::{Arena, PAGE};
+
+//     #[test]
+//     fn test_arena() {
+//         let arena = Arena::default();
+//         let ptr = arena.allocate(Layout::new::<u8>()).unwrap();
+
+//         unsafe { arena.deallocate(ptr.to_raw_parts().0.cast(), Layout::new::<u8>()) };
+
+//         let mut ptr = arena.allocate(Layout::array::<u8>(PAGE).unwrap()).unwrap();
+//         let slice = unsafe { ptr.as_mut() };
+//         for b in slice {
+//             *b = 25;
+//         }
+
+//         unsafe {
+//             arena.deallocate(
+//                 ptr.to_raw_parts().0.cast(),
+//                 Layout::array::<u8>(PAGE).unwrap(),
+//             );
+//         }
+//     }
+
+//     #[test]
+//     fn test_arena_vec() {
+//         let arena = Arena::default();
+
+//         let mut vec = Vec::new_in(arena);
+//         vec.push("Hello World");
+//         vec.push("str1");
+//         vec.push("str2");
+//         vec.push("str3");
+//         vec.truncate(0);
+//         vec.shrink_to_fit();
+//         drop(vec);
+//     }
+// }
+
 #[cfg(test)]
 mod tests {
-    use std::alloc::{Allocator, Layout};
-    use std::slice;
-
-    use super::{Arena, PAGE};
+    use super::Arena;
 
     #[test]
     fn test_arena() {
-        let arena = Arena::default();
-        let ptr = arena.allocate(Layout::new::<u8>()).unwrap();
+        let arena = Arena::new(30);
 
-        unsafe { arena.deallocate(ptr.to_raw_parts().0.cast(), Layout::new::<u8>()) };
+        let mut bufs = Vec::new();
+        for _ in 0..30 {
+            bufs.push(arena.get(1));
+        }
+        assert_eq!(arena.chunks(), 1);
 
-        let mut ptr = arena.allocate(Layout::array::<u8>(PAGE).unwrap()).unwrap();
-        let slice = unsafe { ptr.as_mut() };
-        for b in slice {
-            *b = 25;
+        for buf in &bufs {
+            assert_eq!(bufs[0].chunk, buf.chunk);
         }
 
-        unsafe {
-            arena.deallocate(
-                ptr.to_raw_parts().0.cast(),
-                Layout::array::<u8>(PAGE).unwrap(),
-            );
+        for buf in &mut bufs {
+            for b in buf.as_mut() {
+                *b = 42;
+            }
         }
+
+        let _c2 = arena.get(1);
+        assert_eq!(arena.chunks(), 2);
     }
 
     #[test]
-    fn test_arena_vec() {
-        let arena = Arena::default();
-
-        let mut vec = Vec::new_in(arena);
-        vec.push("Hello World");
-        vec.push("str1");
-        vec.push("str2");
-        vec.push("str3");
-        vec.truncate(0);
-        vec.shrink_to_fit();
-        drop(vec);
+    fn test_arena_shared() {
+        let arena = Arena::new(3000);
+
+        let buf = arena.get(1500);
+        let b1 = buf.freeze();
+        let b2 = b1.clone();
+        let b3 = b2.clone();
+        assert_eq!(arena.chunks(), 1);
+
+        drop(b2);
+        drop(b1);
+        drop(b3);
+    }
+
+    #[test]
+    fn test_chunk_reuse() {
+        let arena = Arena::new(3000);
+
+        let b1 = arena.get(1500);
+        let b2 = arena.get(1500);
+        assert_eq!(arena.chunks(), 1);
+
+        let ptr1 = b1.chunk;
+
+        let _b3 = arena.get(1500);
+        assert_eq!(arena.chunks(), 2);
+
+        drop(b1);
+        drop(b2);
+        let b4 = arena.get(1500);
+
+        assert_eq!(ptr1, b4.chunk);
     }
 }

From a8e7a9ed28bae2ec7c55f0131366c29897a82757 Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Thu, 12 Jan 2023 21:52:02 +0100
Subject: [PATCH 3/9] Cleanup and more docs

---
 stsync-proxy/src/utils/arena.rs | 325 ++++++++++----------------------
 1 file changed, 99 insertions(+), 226 deletions(-)

diff --git a/stsync-proxy/src/utils/arena.rs b/stsync-proxy/src/utils/arena.rs
index 4101c61..82b5a1a 100644
--- a/stsync-proxy/src/utils/arena.rs
+++ b/stsync-proxy/src/utils/arena.rs
@@ -7,31 +7,48 @@ use std::slice;
 use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
 
 use bytes::{Bytes, Vtable};
-use parking_lot::Mutex;
+use parking_lot::RwLock;
 
-const PAGE: usize = 65536;
+/// The default chunk size.
+pub const PAGE: usize = 65536;
 
+/// A fast arena-based byte allocator optimized for short lifetime objects.
+///
+/// `Arena` is optimized for objects with a short lifetime, e.g. network buffers. The allocator
+/// implementation is not stabilized and might change in the future.
+///
+/// The current implementation is based on a thread-shared bump allocator with reusable chunks.
 #[derive(Debug)]
 pub struct Arena {
     chunk_size: usize,
     // FIXME: Maybe this can be a lock-free intrusive doubly linked list.
-    chunks: Mutex<Vec<Chunk>>,
+    chunks: RwLock<Vec<Chunk>>,
+    max_chunks: usize,
 }
 
 impl Arena {
+    /// Creates a new `Arena` with the given `chunk_size`.
+    #[inline]
     pub fn new(chunk_size: usize) -> Self {
         Self {
             chunk_size,
-            chunks: Mutex::new(Vec::new()),
+            chunks: RwLock::new(Vec::new()),
+            max_chunks: usize::MAX,
         }
     }
 
+    #[inline]
+    pub const fn builder() -> ArenaBuilder {
+        ArenaBuilder::new()
+    }
+
+    /// Acquires a new, mutable memory block from `Arena` and returns it in form of a [`BytesMut`].
     pub fn get(&self, size: usize) -> BytesMut {
         if size > self.chunk_size {
             panic!("Cannot allocate larger than chunk size");
         }
 
-        let mut chunks = self.chunks.lock();
+        let chunks = self.chunks.read();
 
         for chunk in &*chunks {
             // We have the only pointer to the chunk. The chunk allocation
@@ -47,15 +64,23 @@ impl Arena {
             }
         }
 
+        if chunks.len() >= self.max_chunks {
+            panic!("Arena chunks oom");
+        }
+
+        drop(chunks);
         let chunk = Chunk::new(self.chunk_size);
         let buf = chunk.get(size).unwrap();
+
+        let mut chunks = self.chunks.write();
         chunks.push(chunk);
 
         buf
     }
 
+    #[inline]
     pub fn chunks(&self) -> usize {
-        let chunks = self.chunks.lock();
+        let chunks = self.chunks.read();
         chunks.len()
     }
 }
@@ -67,6 +92,63 @@ impl Default for Arena {
     }
 }
 
+#[derive(Copy, Clone, Debug)]
+pub struct ArenaBuilder {
+    chunk_size: usize,
+    min_chunks: usize,
+    max_chunks: usize,
+}
+
+impl ArenaBuilder {
+    #[inline]
+    pub const fn new() -> Self {
+        Self {
+            chunk_size: PAGE,
+            min_chunks: 0,
+            max_chunks: usize::MAX,
+        }
+    }
+
+    #[inline]
+    pub const fn chunk_size(mut self, chunk_size: usize) -> Self {
+        self.chunk_size = chunk_size;
+        self
+    }
+
+    #[inline]
+    pub const fn min_chunks(mut self, min_chunks: usize) -> Self {
+        self.min_chunks = min_chunks;
+        self
+    }
+
+    #[inline]
+    pub const fn max_chunks(mut self, max_chunks: usize) -> Self {
+        self.max_chunks = max_chunks;
+        self
+    }
+
+    pub fn build(self) -> Arena {
+        let mut chunks = Vec::with_capacity(self.min_chunks);
+        for _ in 0..self.min_chunks {
+            chunks.push(Chunk::new(self.chunk_size));
+        }
+
+        Arena {
+            chunk_size: self.chunk_size,
+            chunks: RwLock::new(chunks),
+            max_chunks: self.max_chunks,
+        }
+    }
+}
+
+impl Default for ArenaBuilder {
+    #[inline]
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// A fixed-size mutable continguous slice of memory.
 #[derive(Debug)]
 pub struct BytesMut {
     chunk: NonNull<ChunkInner>,
@@ -75,12 +157,18 @@ pub struct BytesMut {
 }
 
 impl BytesMut {
+    /// Shortens the length of the buffer to `len`.
+    ///
+    /// If `len` is greater than the current length of the buffer, this has no effect.
+    #[inline]
     pub fn truncate(&mut self, len: usize) {
         if len <= self.len() {
             unsafe { self.set_len(len) };
         }
     }
 
+    /// Converts `self` into an immutable [`Bytes`].
+    #[inline]
     pub fn freeze(self) -> Bytes {
         let ptr = self.ptr.as_ptr();
         let len = self.len;
@@ -92,7 +180,8 @@ impl BytesMut {
         unsafe { Bytes::with_vtable(ptr, len, data, &SHARED_VTABLE) }
     }
 
-    pub unsafe fn set_len(&mut self, len: usize) {
+    #[inline]
+    unsafe fn set_len(&mut self, len: usize) {
         self.len = len;
     }
 
@@ -108,12 +197,14 @@ impl BytesMut {
 }
 
 impl AsRef<[u8]> for BytesMut {
+    #[inline]
     fn as_ref(&self) -> &[u8] {
         self.as_slice()
     }
 }
 
 impl AsMut<[u8]> for BytesMut {
+    #[inline]
     fn as_mut(&mut self) -> &mut [u8] {
         self.as_slice_mut()
     }
@@ -160,7 +251,7 @@ struct Chunk {
 
 impl Chunk {
     pub fn new(size: usize) -> Self {
-        let mut inner = Box::new(ChunkInner::new(size));
+        let inner = Box::new(ChunkInner::new(size));
         let ptr = Box::leak(inner).into();
 
         Self { ptr }
@@ -315,224 +406,6 @@ unsafe fn chunk_drop(data: &mut AtomicPtr<()>, ptr: *const u8, len: usize) {
     unsafe { Box::from_raw(shared) };
 }
 
-// #[derive(Debug)]
-// struct ArenaInner {
-//     page_size: usize,
-//     chunks: RwLock<Vec<(Chunk, *const u8)>>,
-// }
-
-// impl Arena {
-//     #[inline]
-//     pub fn new(page_size: usize) -> Self {
-//         Self {
-//             inner: Arc::new(ArenaInner {
-//                 page_size,
-//                 chunks: RwLock::new(Vec::new()),
-//             }),
-//         }
-//     }
-
-//     unsafe fn alloc(&self, mut layout: Layout) -> NonNull<[u8]> {
-//         layout = layout.pad_to_align();
-
-//         if layout.size() > self.inner.page_size {
-//             panic!("exceeded page size");
-//             // The requested layout is bigger than the maximum chunk size. We can never
-//             // allocate this object with this arena. Forward to global allocator.
-//             // return unsafe { NonNull::new_unchecked(std::alloc::alloc(layout)) };
-//             // Equivalent to ptr.cast() but is unsized.
-//         }
-
-//         let chunks = self.inner.chunks.read();
-//         for (chunk, _) in &*chunks {
-//             if let Ok(ptr) = unsafe { chunk.allocate_padded(layout) } {
-//                 return ptr;
-//             }
-//         }
-
-//         let (chunk, chunk_ptr) = self.make_chunk();
-
-//         drop(chunks);
-//         let mut chunks = self.inner.chunks.write();
-//         let ptr = unsafe { chunk.allocate_padded(layout).unwrap() };
-
-//         chunks.push((chunk, chunk_ptr));
-//         ptr
-//     }
-
-//     unsafe fn dealloc(&self, ptr: NonNull<u8>, layout: Layout) {
-//         let mut chunks = self.inner.chunks.write();
-//         for (i, (chunk, base)) in chunks.iter().enumerate() {
-//             // Pointer within base + page_size; the pointer is a part of this chunk.
-//             if ptr.as_ptr() as usize <= *base as usize + self.inner.page_size {
-//                 // SAFETY: `ptr` was previously allocated in this `chunk`.
-//                 unsafe {
-//                     chunk.deallocate(layout, ptr);
-//                 }
-
-//                 chunks.remove(i);
-//                 break;
-//             }
-//         }
-//     }
-
-//     #[inline]
-//     fn make_chunk(&self) -> (Chunk, *const u8) {
-//         let layout = unsafe {
-//             Layout::from_size_align_unchecked(
-//                 std::mem::size_of::<u8>() * self.inner.page_size,
-//                 std::mem::align_of::<usize>(),
-//             )
-//         };
-
-//         let ptr = unsafe { std::alloc::alloc(layout) };
-//         assert!(!ptr.is_null());
-
-//         let ptr = unsafe { NonNull::new_unchecked(ptr) };
-
-//         (
-//             Chunk {
-//                 arena: self.clone(),
-//                 ptr,
-//                 layout,
-//                 head: AtomicUsize::new(0),
-//                 entries: AtomicUsize::new(0),
-//             },
-//             ptr.as_ptr(),
-//         )
-//     }
-// }
-
-// impl Default for Arena {
-//     #[inline]
-//     fn default() -> Self {
-//         Self::new(PAGE)
-//     }
-// }
-
-// unsafe impl Allocator for Arena {
-//     fn allocate(&self, layout: Layout) -> Result<NonNull<[u8]>, std::alloc::AllocError> {
-//         unsafe { Ok(self.alloc(layout)) }
-//     }
-
-//     unsafe fn deallocate(&self, ptr: NonNull<u8>, layout: Layout) {
-//         unsafe { self.dealloc(ptr, layout) };
-//     }
-// }
-
-// /// A fixed size memory page.
-// #[derive(Debug)]
-// struct Chunk {
-//     arena: Arena,
-//     ptr: NonNull<u8>,
-//     layout: Layout,
-//     head: AtomicUsize,
-//     /// The number of active references to the buffer of this chunk.
-//     entries: AtomicUsize,
-// }
-
-// impl Chunk {
-//     /// # Safety
-//     ///
-//     /// The `Chunk` must have enough [`contiguous_capacity`] to allocate using `layout`.
-//     ///
-//     /// [`contiguous_capacity`]: Self::contiguous_capacity
-//     unsafe fn allocate_padded(&self, layout: Layout) -> Result<NonNull<[u8]>, AllocError> {
-//         // Acquire the head offset.
-//         let mut head = self.head.load(Ordering::SeqCst);
-//         if head > head + layout.size() {
-//             return Err(AllocError);
-//         }
-
-//         while let Err(current) = self.head.compare_exchange_weak(
-//             head,
-//             head + layout.size(),
-//             Ordering::SeqCst,
-//             Ordering::SeqCst,
-//         ) {
-//             head = current;
-
-//             // Another thread won the race and we no longer have enough space to
-//             // allocate using this `layout`.
-//             if head > head + layout.size() {
-//                 return Err(AllocError);
-//             }
-//         }
-
-//         self.entries.fetch_add(1, Ordering::SeqCst);
-
-//         let ptr = unsafe { NonNull::new_unchecked(self.ptr.as_ptr().add(head)) };
-//         let len = layout.size();
-
-//         // SAFETY: The caller guarantees that `end < buffer.len()`.
-//         // let fatptr = [self.buffer.as_ptr() as usize + head, layout.size()];
-//         // Ok(unsafe { std::mem::transmute::<[usize; 2], NonNull<[u8]>>(fatptr) })
-//         // Ok(unsafe { self.buffer.get_unchecked(head..end).into() })
-//         Ok(NonNull::from_raw_parts(ptr.cast(), len))
-//     }
-
-//     /// Deallocate given memory block owned by this `Chunk`.
-//     ///
-//     /// # Safety
-//     ///
-//     /// The given `ptr` must be owned by this `Chunk`, i.e. have been previously allocated using
-//     /// `allocate` of the same `Chunk`. The `layout` must be same layout used to allocate.
-//     unsafe fn deallocate(&self, layout: Layout, ptr: NonNull<u8>) {
-//         self.entries.fetch_sub(1, Ordering::SeqCst);
-//     }
-// }
-
-// impl Drop for Chunk {
-//     fn drop(&mut self) {
-//         unsafe {
-//             std::alloc::dealloc(self.ptr.as_ptr(), self.layout);
-//         }
-//     }
-// }
-
-// #[cfg(test)]
-// mod tests {
-//     use std::alloc::{Allocator, Layout};
-//     use std::slice;
-
-//     use super::{Arena, PAGE};
-
-//     #[test]
-//     fn test_arena() {
-//         let arena = Arena::default();
-//         let ptr = arena.allocate(Layout::new::<u8>()).unwrap();
-
-//         unsafe { arena.deallocate(ptr.to_raw_parts().0.cast(), Layout::new::<u8>()) };
-
-//         let mut ptr = arena.allocate(Layout::array::<u8>(PAGE).unwrap()).unwrap();
-//         let slice = unsafe { ptr.as_mut() };
-//         for b in slice {
-//             *b = 25;
-//         }
-
-//         unsafe {
-//             arena.deallocate(
-//                 ptr.to_raw_parts().0.cast(),
-//                 Layout::array::<u8>(PAGE).unwrap(),
-//             );
-//         }
-//     }
-
-//     #[test]
-//     fn test_arena_vec() {
-//         let arena = Arena::default();
-
-//         let mut vec = Vec::new_in(arena);
-//         vec.push("Hello World");
-//         vec.push("str1");
-//         vec.push("str2");
-//         vec.push("str3");
-//         vec.truncate(0);
-//         vec.shrink_to_fit();
-//         drop(vec);
-//     }
-// }
-
 #[cfg(test)]
 mod tests {
     use super::Arena;

From febc4ca61e2539e03202cf735fad1f8ded11c602 Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Thu, 12 Jan 2023 21:52:27 +0100
Subject: [PATCH 4/9] Integrate Arena allocator into Worker pool

---
 stsync-proxy/src/config.rs     |  3 ++-
 stsync-proxy/src/srt/config.rs | 14 +++++++++++++-
 stsync-proxy/src/srt/server.rs |  9 +++------
 stsync-proxy/src/srt/state.rs  |  9 +++++++++
 stsync-proxy/src/state.rs      |  1 +
 5 files changed, 28 insertions(+), 8 deletions(-)

diff --git a/stsync-proxy/src/config.rs b/stsync-proxy/src/config.rs
index 85b6e86..8e973b3 100644
--- a/stsync-proxy/src/config.rs
+++ b/stsync-proxy/src/config.rs
@@ -6,6 +6,7 @@ use std::path::Path;
 use serde::{Deserialize, Serialize};
 
 use crate::srt;
+use crate::srt::config::Workers;
 
 #[derive(Serialize, Deserialize)]
 pub struct Config {
@@ -37,7 +38,7 @@ pub struct Http {
 pub struct Srt {
     pub enabled: bool,
     pub bind: SocketAddr,
-    pub workers: Option<usize>,
+    pub workers: Workers,
 
     pub rcvbuf: usize,
     pub sndbuf: usize,
diff --git a/stsync-proxy/src/srt/config.rs b/stsync-proxy/src/srt/config.rs
index 5d453a5..467034d 100644
--- a/stsync-proxy/src/srt/config.rs
+++ b/stsync-proxy/src/srt/config.rs
@@ -1,4 +1,5 @@
 use std::net::SocketAddr;
+use std::thread::available_parallelism;
 
 use serde::{Deserialize, Serialize};
 
@@ -7,7 +8,7 @@ pub struct Config {
     /// The tuple to bind the server to.
     pub bind: SocketAddr,
     /// The number of workers or ``
-    pub workers: Option<usize>,
+    pub workers: Workers,
 
     /// The size of `SO_RCVBUF` in bytes.
     pub rcvbuf: usize,
@@ -21,3 +22,14 @@ pub struct Config {
     /// Latency in millis
     pub latency: u16,
 }
+
+#[derive(Copy, Clone, Debug, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct Workers(Option<usize>);
+
+impl Workers {
+    pub fn get(self) -> usize {
+        self.0
+            .unwrap_or_else(|| available_parallelism().map(|n| n.get()).unwrap_or(1))
+    }
+}
diff --git a/stsync-proxy/src/srt/server.rs b/stsync-proxy/src/srt/server.rs
index e4bdc2f..aa9203d 100644
--- a/stsync-proxy/src/srt/server.rs
+++ b/stsync-proxy/src/srt/server.rs
@@ -45,11 +45,7 @@ where
         tracing::info!("Srt socket listening on {}", socket.local_addr()?);
         tracing::info!("Socket Recv-Q: {}, Send-Q: {}", rx, tx);
 
-        let num_workers = config.workers.unwrap_or_else(|| {
-            std::thread::available_parallelism()
-                .map(|n| n.get())
-                .unwrap_or(1)
-        });
+        let num_workers = config.workers.get();
 
         let socket = Arc::new(socket);
         let state = State::new(session_manager, config);
@@ -164,10 +160,11 @@ impl Worker {
             );
 
             loop {
-                let mut buf = BytesMut::zeroed(1500);
+                let mut buf = state.arena.get(1500);
                 let (len, addr) = socket.recv_from(&mut buf).await?;
                 tracing::trace!("[{}] Got {} bytes from {}", ident, len, addr);
                 buf.truncate(len);
+                let mut buf = buf.freeze();
 
                 let packet = match Packet::decode(&mut buf) {
                     Ok(packet) => packet,
diff --git a/stsync-proxy/src/srt/state.rs b/stsync-proxy/src/srt/state.rs
index 8d1946e..bbd9209 100644
--- a/stsync-proxy/src/srt/state.rs
+++ b/stsync-proxy/src/srt/state.rs
@@ -11,6 +11,7 @@ use rand::rngs::OsRng;
 use rand::RngCore;
 
 use crate::session::SessionManager;
+use crate::utils::arena::Arena;
 
 use super::config::Config;
 use super::conn::ConnectionHandle;
@@ -29,6 +30,12 @@ where
     S: SessionManager,
 {
     pub fn new(session_manager: S, config: Config) -> Self {
+        let arena = Arena::builder()
+            .chunk_size(config.mtu as usize * config.flow_window as usize)
+            .min_chunks(config.workers.get())
+            .max_chunks(usize::MAX << 1)
+            .build();
+
         Self {
             inner: Arc::new(StateInner {
                 config: config,
@@ -37,6 +44,7 @@ where
                 session_manager,
                 conn_metrics: Mutex::new(AHashMap::new()),
                 metrics: ServerMetrics::new(),
+                arena,
             }),
         }
     }
@@ -77,6 +85,7 @@ where
     pub session_manager: S,
     pub conn_metrics: Mutex<AHashMap<ConnectionId, Arc<ConnectionMetrics>>>,
     pub metrics: ServerMetrics,
+    pub arena: Arena,
 }
 
 impl<S> StateInner<S>
diff --git a/stsync-proxy/src/state.rs b/stsync-proxy/src/state.rs
index f801160..1c94a70 100644
--- a/stsync-proxy/src/state.rs
+++ b/stsync-proxy/src/state.rs
@@ -4,6 +4,7 @@ use std::sync::Arc;
 use crate::database::Database;
 use crate::session::buffer::BufferSessionManager;
 use crate::srt;
+use crate::utils::arena::Arena;
 
 #[derive(Clone, Debug)]
 pub struct State(Arc<StateInner>);

From c8b44cc56c94c88e467e5d5957482b10f94f43a7 Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Thu, 12 Jan 2023 21:55:42 +0100
Subject: [PATCH 5/9] cargo clippy

---
 stsync-proxy/src/utils/arena.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/stsync-proxy/src/utils/arena.rs b/stsync-proxy/src/utils/arena.rs
index 82b5a1a..c78267c 100644
--- a/stsync-proxy/src/utils/arena.rs
+++ b/stsync-proxy/src/utils/arena.rs
@@ -384,13 +384,13 @@ unsafe fn chunk_clone(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Bytes
     unsafe { Bytes::with_vtable(ptr, len, AtomicPtr::new(shared as *mut ()), &SHARED_VTABLE) }
 }
 
-unsafe fn chunk_to_vec(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8> {
+unsafe fn chunk_to_vec(_data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8> {
     let mut buf = Vec::with_capacity(len);
     unsafe { std::ptr::copy_nonoverlapping(ptr, buf.as_mut_ptr(), len) };
     buf
 }
 
-unsafe fn chunk_drop(data: &mut AtomicPtr<()>, ptr: *const u8, len: usize) {
+unsafe fn chunk_drop(data: &mut AtomicPtr<()>, _ptr: *const u8, _len: usize) {
     let shared = data.load(Ordering::Relaxed) as *mut ChunkInner;
 
     let rc = unsafe { (*shared).ref_count.fetch_sub(1, Ordering::Release) };

From a1606a84fc17472e219a56c74b76da52395e8abe Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Thu, 12 Jan 2023 23:07:23 +0100
Subject: [PATCH 6/9] Emit acquire fence when dropping chunk

---
 stsync-proxy/src/utils/arena.rs | 24 +++++++++++++++---------
 1 file changed, 15 insertions(+), 9 deletions(-)

diff --git a/stsync-proxy/src/utils/arena.rs b/stsync-proxy/src/utils/arena.rs
index c78267c..5a3ab73 100644
--- a/stsync-proxy/src/utils/arena.rs
+++ b/stsync-proxy/src/utils/arena.rs
@@ -236,6 +236,9 @@ impl Drop for BytesMut {
             return;
         }
 
+        // fence
+        chunk.ref_count.load(Ordering::Acquire);
+
         unsafe { Box::from_raw(self.ptr.as_ptr()) };
     }
 }
@@ -261,13 +264,11 @@ impl Chunk {
         let inner = unsafe { self.ptr.as_ref() };
         let ptr = inner.get(size)?;
 
-        unsafe {
-            Some(BytesMut {
-                chunk: self.ptr,
-                ptr: NonNull::new_unchecked(ptr),
-                len: size,
-            })
-        }
+        Some(BytesMut {
+            chunk: self.ptr,
+            ptr,
+            len: size,
+        })
     }
 }
 
@@ -289,6 +290,9 @@ impl Drop for Chunk {
             return;
         }
 
+        // fence
+        chunk.ref_count.load(Ordering::Acquire);
+
         // Drop the allocation
         unsafe { Box::from_raw(self.ptr.as_ptr()) };
     }
@@ -324,7 +328,9 @@ impl ChunkInner {
         }
     }
 
-    pub fn get(&self, size: usize) -> Option<*mut u8> {
+    /// Acquire a new buffer with the given `size` from this chunk. Returns `None` if this chunk
+    /// does not have enough empty capacity to allocate the requested buffer.
+    pub fn get(&self, size: usize) -> Option<NonNull<u8>> {
         let mut head = self.head.load(Ordering::Acquire);
 
         // Check that this chunk has enough free capacity to hold size bytes.
@@ -346,7 +352,7 @@ impl ChunkInner {
         self.ref_count.fetch_add(1, Ordering::SeqCst);
 
         // SAFETY: We have exlusive access to ptr + head.
-        unsafe { Some(self.ptr.add(head)) }
+        unsafe { Some(NonNull::new_unchecked(self.ptr.add(head))) }
     }
 
     unsafe fn reset(&self) {

From 834a2c2a0c83d9affb767a8cab2ecfcb46b8383c Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Sat, 14 Jan 2023 15:35:50 +0100
Subject: [PATCH 7/9] Use custom bytes crate

---
 stsync-proxy/Cargo.toml | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/stsync-proxy/Cargo.toml b/stsync-proxy/Cargo.toml
index a49f5fb..c1dac67 100644
--- a/stsync-proxy/Cargo.toml
+++ b/stsync-proxy/Cargo.toml
@@ -22,7 +22,6 @@ log = "0.4.17"
 tracing ={ version = "0.1.37", features = ["log-always"] }
 socket2 = "0.4.7"
 snowflaked = { version = "0.1.5", features = ["sync"] }
-bytes = "1.2.1"
 tokio-stream = { version = "0.1.11", features = ["sync"] }
 hyper = { version = "0.14.20", features = ["http1", "http2", "server"] }
 ahash = "0.8.0"
@@ -32,6 +31,10 @@ pin-project = "1.0.12"
 serde_json = "1.0.87"
 toml = "0.5.9"
 
+# Fork of bytes with a public bytes vtable until bytes::Vtable becomes pub.
+# Required for custom arena allocator.
+bytes = { git = "https://github.com/MrGunflame/bytes", branch = "pub-vtable" }
+
 [target.'cfg(unix)'.dependencies]
 nix = { version = "0.26.1", features = ["signal"] }
 

From a751c739b9b39480768278a59bd07d8ff704ff7f Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Wed, 15 Feb 2023 17:16:34 +0100
Subject: [PATCH 8/9] Implement arena allocator through polymock crate

---
 stsync-proxy/Cargo.toml         |   4 +-
 stsync-proxy/src/main.rs        |   4 +
 stsync-proxy/src/srt/mod.rs     |   3 +-
 stsync-proxy/src/srt/server.rs  |   6 +-
 stsync-proxy/src/srt/state.rs   |   8 +-
 stsync-proxy/src/state.rs       |   1 -
 stsync-proxy/src/utils/arena.rs | 477 --------------------------------
 stsync-proxy/src/utils/mod.rs   |   1 -
 8 files changed, 13 insertions(+), 491 deletions(-)
 delete mode 100644 stsync-proxy/src/utils/arena.rs

diff --git a/stsync-proxy/Cargo.toml b/stsync-proxy/Cargo.toml
index fdf5e01..a02d146 100644
--- a/stsync-proxy/Cargo.toml
+++ b/stsync-proxy/Cargo.toml
@@ -31,9 +31,7 @@ pin-project = "1.0.12"
 serde_json = "1.0.87"
 toml = "0.5.9"
 
-# Fork of bytes with a public bytes vtable until bytes::Vtable becomes pub.
-# Required for custom arena allocator.
-bytes = { git = "https://github.com/MrGunflame/bytes", branch = "pub-vtable" }
+polymock = { git = "https://github.com/MrGunflame/polymock-rs" }
 
 [target.'cfg(unix)'.dependencies]
 nix = { version = "0.26.1", features = ["signal"] }
diff --git a/stsync-proxy/src/main.rs b/stsync-proxy/src/main.rs
index 3105e3c..177982f 100644
--- a/stsync-proxy/src/main.rs
+++ b/stsync-proxy/src/main.rs
@@ -1,6 +1,10 @@
 #![deny(unsafe_op_in_unsafe_fn)]
 #![deny(unused_crate_dependencies)]
 
+// Use the bytes drop-in replacement types until the bytes
+// crate allows public vtable creation.
+extern crate polymock as bytes;
+
 // We only import log to remove trace and debug levels at compile time.
 use log as _;
 
diff --git a/stsync-proxy/src/srt/mod.rs b/stsync-proxy/src/srt/mod.rs
index fcb173a..bc9e1d1 100644
--- a/stsync-proxy/src/srt/mod.rs
+++ b/stsync-proxy/src/srt/mod.rs
@@ -173,7 +173,8 @@ impl Default for DataPacket {
 
         Self {
             header,
-            data: Bytes::new(),
+            // FIXME: Change back to `Bytes::new()` when possible.
+            data: Bytes::copy_from_slice(&[]),
         }
     }
 }
diff --git a/stsync-proxy/src/srt/server.rs b/stsync-proxy/src/srt/server.rs
index aa9203d..107d9ad 100644
--- a/stsync-proxy/src/srt/server.rs
+++ b/stsync-proxy/src/srt/server.rs
@@ -1,4 +1,4 @@
-use bytes::BytesMut;
+use bytes::Bytes;
 use futures::stream::FuturesUnordered;
 use futures::{FutureExt, StreamExt};
 use std::future::Future;
@@ -160,10 +160,12 @@ impl Worker {
             );
 
             loop {
-                let mut buf = state.arena.get(1500);
+                let mut buf = state.arena.alloc(1500);
+
                 let (len, addr) = socket.recv_from(&mut buf).await?;
                 tracing::trace!("[{}] Got {} bytes from {}", ident, len, addr);
                 buf.truncate(len);
+
                 let mut buf = buf.freeze();
 
                 let packet = match Packet::decode(&mut buf) {
diff --git a/stsync-proxy/src/srt/state.rs b/stsync-proxy/src/srt/state.rs
index bbd9209..3cffb5a 100644
--- a/stsync-proxy/src/srt/state.rs
+++ b/stsync-proxy/src/srt/state.rs
@@ -7,11 +7,11 @@ use std::sync::Arc;
 
 use ahash::{AHashMap, AHashSet};
 use parking_lot::{Mutex, RwLock};
+use polymock::Arena;
 use rand::rngs::OsRng;
 use rand::RngCore;
 
 use crate::session::SessionManager;
-use crate::utils::arena::Arena;
 
 use super::config::Config;
 use super::conn::ConnectionHandle;
@@ -30,11 +30,7 @@ where
     S: SessionManager,
 {
     pub fn new(session_manager: S, config: Config) -> Self {
-        let arena = Arena::builder()
-            .chunk_size(config.mtu as usize * config.flow_window as usize)
-            .min_chunks(config.workers.get())
-            .max_chunks(usize::MAX << 1)
-            .build();
+        let arena = Arena::new(config.mtu as usize * config.flow_window as usize);
 
         Self {
             inner: Arc::new(StateInner {
diff --git a/stsync-proxy/src/state.rs b/stsync-proxy/src/state.rs
index 1c94a70..f801160 100644
--- a/stsync-proxy/src/state.rs
+++ b/stsync-proxy/src/state.rs
@@ -4,7 +4,6 @@ use std::sync::Arc;
 use crate::database::Database;
 use crate::session::buffer::BufferSessionManager;
 use crate::srt;
-use crate::utils::arena::Arena;
 
 #[derive(Clone, Debug)]
 pub struct State(Arc<StateInner>);
diff --git a/stsync-proxy/src/utils/arena.rs b/stsync-proxy/src/utils/arena.rs
deleted file mode 100644
index 5a3ab73..0000000
--- a/stsync-proxy/src/utils/arena.rs
+++ /dev/null
@@ -1,477 +0,0 @@
-//! A fast arena allocator for network buffers.
-
-use std::alloc::Layout;
-use std::ops::{Deref, DerefMut};
-use std::ptr::NonNull;
-use std::slice;
-use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
-
-use bytes::{Bytes, Vtable};
-use parking_lot::RwLock;
-
-/// The default chunk size.
-pub const PAGE: usize = 65536;
-
-/// A fast arena-based byte allocator optimized for short lifetime objects.
-///
-/// `Arena` is optimized for objects with a short lifetime, e.g. network buffers. The allocator
-/// implementation is not stabilized and might change in the future.
-///
-/// The current implementation is based on a thread-shared bump allocator with reusable chunks.
-#[derive(Debug)]
-pub struct Arena {
-    chunk_size: usize,
-    // FIXME: Maybe this can be a lock-free intrusive doubly linked list.
-    chunks: RwLock<Vec<Chunk>>,
-    max_chunks: usize,
-}
-
-impl Arena {
-    /// Creates a new `Arena` with the given `chunk_size`.
-    #[inline]
-    pub fn new(chunk_size: usize) -> Self {
-        Self {
-            chunk_size,
-            chunks: RwLock::new(Vec::new()),
-            max_chunks: usize::MAX,
-        }
-    }
-
-    #[inline]
-    pub const fn builder() -> ArenaBuilder {
-        ArenaBuilder::new()
-    }
-
-    /// Acquires a new, mutable memory block from `Arena` and returns it in form of a [`BytesMut`].
-    pub fn get(&self, size: usize) -> BytesMut {
-        if size > self.chunk_size {
-            panic!("Cannot allocate larger than chunk size");
-        }
-
-        let chunks = self.chunks.read();
-
-        for chunk in &*chunks {
-            // We have the only pointer to the chunk. The chunk allocation
-            // can be reused.
-            if chunk.ref_count.load(Ordering::Relaxed) == 1 {
-                unsafe {
-                    chunk.reset();
-                }
-            }
-
-            if let Some(buf) = chunk.get(size) {
-                return buf;
-            }
-        }
-
-        if chunks.len() >= self.max_chunks {
-            panic!("Arena chunks oom");
-        }
-
-        drop(chunks);
-        let chunk = Chunk::new(self.chunk_size);
-        let buf = chunk.get(size).unwrap();
-
-        let mut chunks = self.chunks.write();
-        chunks.push(chunk);
-
-        buf
-    }
-
-    #[inline]
-    pub fn chunks(&self) -> usize {
-        let chunks = self.chunks.read();
-        chunks.len()
-    }
-}
-
-impl Default for Arena {
-    #[inline]
-    fn default() -> Self {
-        Self::new(PAGE)
-    }
-}
-
-#[derive(Copy, Clone, Debug)]
-pub struct ArenaBuilder {
-    chunk_size: usize,
-    min_chunks: usize,
-    max_chunks: usize,
-}
-
-impl ArenaBuilder {
-    #[inline]
-    pub const fn new() -> Self {
-        Self {
-            chunk_size: PAGE,
-            min_chunks: 0,
-            max_chunks: usize::MAX,
-        }
-    }
-
-    #[inline]
-    pub const fn chunk_size(mut self, chunk_size: usize) -> Self {
-        self.chunk_size = chunk_size;
-        self
-    }
-
-    #[inline]
-    pub const fn min_chunks(mut self, min_chunks: usize) -> Self {
-        self.min_chunks = min_chunks;
-        self
-    }
-
-    #[inline]
-    pub const fn max_chunks(mut self, max_chunks: usize) -> Self {
-        self.max_chunks = max_chunks;
-        self
-    }
-
-    pub fn build(self) -> Arena {
-        let mut chunks = Vec::with_capacity(self.min_chunks);
-        for _ in 0..self.min_chunks {
-            chunks.push(Chunk::new(self.chunk_size));
-        }
-
-        Arena {
-            chunk_size: self.chunk_size,
-            chunks: RwLock::new(chunks),
-            max_chunks: self.max_chunks,
-        }
-    }
-}
-
-impl Default for ArenaBuilder {
-    #[inline]
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-/// A fixed-size mutable continguous slice of memory.
-#[derive(Debug)]
-pub struct BytesMut {
-    chunk: NonNull<ChunkInner>,
-    ptr: NonNull<u8>,
-    len: usize,
-}
-
-impl BytesMut {
-    /// Shortens the length of the buffer to `len`.
-    ///
-    /// If `len` is greater than the current length of the buffer, this has no effect.
-    #[inline]
-    pub fn truncate(&mut self, len: usize) {
-        if len <= self.len() {
-            unsafe { self.set_len(len) };
-        }
-    }
-
-    /// Converts `self` into an immutable [`Bytes`].
-    #[inline]
-    pub fn freeze(self) -> Bytes {
-        let ptr = self.ptr.as_ptr();
-        let len = self.len;
-        let data = AtomicPtr::new(self.chunk.as_ptr().cast());
-
-        // Do not decrement the reference count.
-        std::mem::forget(self);
-
-        unsafe { Bytes::with_vtable(ptr, len, data, &SHARED_VTABLE) }
-    }
-
-    #[inline]
-    unsafe fn set_len(&mut self, len: usize) {
-        self.len = len;
-    }
-
-    #[inline]
-    fn as_slice(&self) -> &[u8] {
-        unsafe { slice::from_raw_parts(self.ptr.as_ptr(), self.len) }
-    }
-
-    #[inline]
-    fn as_slice_mut(&mut self) -> &mut [u8] {
-        unsafe { slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) }
-    }
-}
-
-impl AsRef<[u8]> for BytesMut {
-    #[inline]
-    fn as_ref(&self) -> &[u8] {
-        self.as_slice()
-    }
-}
-
-impl AsMut<[u8]> for BytesMut {
-    #[inline]
-    fn as_mut(&mut self) -> &mut [u8] {
-        self.as_slice_mut()
-    }
-}
-
-impl Deref for BytesMut {
-    type Target = [u8];
-
-    #[inline]
-    fn deref(&self) -> &Self::Target {
-        self.as_slice()
-    }
-}
-
-impl DerefMut for BytesMut {
-    #[inline]
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        self.as_slice_mut()
-    }
-}
-
-impl Drop for BytesMut {
-    fn drop(&mut self) {
-        let chunk = unsafe { self.chunk.as_ref() };
-
-        let rc = chunk.ref_count.fetch_sub(1, Ordering::Release);
-
-        if rc != 1 {
-            return;
-        }
-
-        // fence
-        chunk.ref_count.load(Ordering::Acquire);
-
-        unsafe { Box::from_raw(self.ptr.as_ptr()) };
-    }
-}
-
-unsafe impl Send for BytesMut {}
-unsafe impl Sync for BytesMut {}
-
-/// A strong reference to a chunk.
-#[derive(Debug)]
-struct Chunk {
-    ptr: NonNull<ChunkInner>,
-}
-
-impl Chunk {
-    pub fn new(size: usize) -> Self {
-        let inner = Box::new(ChunkInner::new(size));
-        let ptr = Box::leak(inner).into();
-
-        Self { ptr }
-    }
-
-    pub fn get(&self, size: usize) -> Option<BytesMut> {
-        let inner = unsafe { self.ptr.as_ref() };
-        let ptr = inner.get(size)?;
-
-        Some(BytesMut {
-            chunk: self.ptr,
-            ptr,
-            len: size,
-        })
-    }
-}
-
-impl Deref for Chunk {
-    type Target = ChunkInner;
-
-    fn deref(&self) -> &Self::Target {
-        unsafe { self.ptr.as_ref() }
-    }
-}
-
-impl Drop for Chunk {
-    fn drop(&mut self) {
-        let chunk = unsafe { self.ptr.as_ref() };
-
-        let rc = chunk.ref_count.fetch_sub(1, Ordering::Release);
-
-        if rc != 1 {
-            return;
-        }
-
-        // fence
-        chunk.ref_count.load(Ordering::Acquire);
-
-        // Drop the allocation
-        unsafe { Box::from_raw(self.ptr.as_ptr()) };
-    }
-}
-
-unsafe impl Send for Chunk {}
-unsafe impl Sync for Chunk {}
-
-#[derive(Debug)]
-struct ChunkInner {
-    /// The layout used to allocate this chunk.
-    layout: Layout,
-    /// The starting pointer of the allocated chunk.
-    ptr: *mut u8,
-    /// Where to put the next bytes block.
-    head: AtomicUsize,
-    /// Strong reference counts
-    ref_count: AtomicUsize,
-}
-
-impl ChunkInner {
-    pub fn new(size: usize) -> Self {
-        let layout = Layout::array::<u8>(size).unwrap();
-
-        let ptr = unsafe { std::alloc::alloc(layout) };
-        assert!(!ptr.is_null());
-
-        Self {
-            layout,
-            ptr,
-            head: AtomicUsize::new(0),
-            ref_count: AtomicUsize::new(1),
-        }
-    }
-
-    /// Acquire a new buffer with the given `size` from this chunk. Returns `None` if this chunk
-    /// does not have enough empty capacity to allocate the requested buffer.
-    pub fn get(&self, size: usize) -> Option<NonNull<u8>> {
-        let mut head = self.head.load(Ordering::Acquire);
-
-        // Check that this chunk has enough free capacity to hold size bytes.
-        if head + size > self.layout.size() {
-            return None;
-        }
-
-        while let Err(curr) =
-            self.head
-                .compare_exchange_weak(head, head + size, Ordering::SeqCst, Ordering::SeqCst)
-        {
-            head = curr;
-
-            if head + size > self.layout.size() {
-                return None;
-            }
-        }
-
-        self.ref_count.fetch_add(1, Ordering::SeqCst);
-
-        // SAFETY: We have exlusive access to ptr + head.
-        unsafe { Some(NonNull::new_unchecked(self.ptr.add(head))) }
-    }
-
-    unsafe fn reset(&self) {
-        // The chunk may immediately be reused with self.get() which uses
-        // acquire loads, hence the release store is necessary.
-        self.head.store(0, Ordering::Release);
-    }
-}
-
-impl Drop for ChunkInner {
-    fn drop(&mut self) {
-        unsafe {
-            std::alloc::dealloc(self.ptr, self.layout);
-        }
-    }
-}
-
-// ==== impl Vtable ====
-
-const SHARED_VTABLE: Vtable = Vtable {
-    clone: chunk_clone,
-    to_vec: chunk_to_vec,
-    drop: chunk_drop,
-};
-
-unsafe fn chunk_clone(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Bytes {
-    let shared = data.load(Ordering::Relaxed) as *mut ChunkInner;
-
-    let old = unsafe { (*shared).ref_count.fetch_add(1, Ordering::Relaxed) };
-
-    if old > usize::MAX >> 1 {
-        std::process::abort();
-    }
-
-    unsafe { Bytes::with_vtable(ptr, len, AtomicPtr::new(shared as *mut ()), &SHARED_VTABLE) }
-}
-
-unsafe fn chunk_to_vec(_data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8> {
-    let mut buf = Vec::with_capacity(len);
-    unsafe { std::ptr::copy_nonoverlapping(ptr, buf.as_mut_ptr(), len) };
-    buf
-}
-
-unsafe fn chunk_drop(data: &mut AtomicPtr<()>, _ptr: *const u8, _len: usize) {
-    let shared = data.load(Ordering::Relaxed) as *mut ChunkInner;
-
-    let rc = unsafe { (*shared).ref_count.fetch_sub(1, Ordering::Release) };
-
-    if rc != 1 {
-        return;
-    }
-
-    // Fence to prevent reordering of data access after deletion.
-    // Syncs with relase load (fetch_sub) above.
-    unsafe { (*shared).ref_count.load(Ordering::Acquire) };
-
-    unsafe { Box::from_raw(shared) };
-}
-
-#[cfg(test)]
-mod tests {
-    use super::Arena;
-
-    #[test]
-    fn test_arena() {
-        let arena = Arena::new(30);
-
-        let mut bufs = Vec::new();
-        for _ in 0..30 {
-            bufs.push(arena.get(1));
-        }
-        assert_eq!(arena.chunks(), 1);
-
-        for buf in &bufs {
-            assert_eq!(bufs[0].chunk, buf.chunk);
-        }
-
-        for buf in &mut bufs {
-            for b in buf.as_mut() {
-                *b = 42;
-            }
-        }
-
-        let _c2 = arena.get(1);
-        assert_eq!(arena.chunks(), 2);
-    }
-
-    #[test]
-    fn test_arena_shared() {
-        let arena = Arena::new(3000);
-
-        let buf = arena.get(1500);
-        let b1 = buf.freeze();
-        let b2 = b1.clone();
-        let b3 = b2.clone();
-        assert_eq!(arena.chunks(), 1);
-
-        drop(b2);
-        drop(b1);
-        drop(b3);
-    }
-
-    #[test]
-    fn test_chunk_reuse() {
-        let arena = Arena::new(3000);
-
-        let b1 = arena.get(1500);
-        let b2 = arena.get(1500);
-        assert_eq!(arena.chunks(), 1);
-
-        let ptr1 = b1.chunk;
-
-        let _b3 = arena.get(1500);
-        assert_eq!(arena.chunks(), 2);
-
-        drop(b1);
-        drop(b2);
-        let b4 = arena.get(1500);
-
-        assert_eq!(ptr1, b4.chunk);
-    }
-}
diff --git a/stsync-proxy/src/utils/mod.rs b/stsync-proxy/src/utils/mod.rs
index 5d63a72..86e0ab3 100644
--- a/stsync-proxy/src/utils/mod.rs
+++ b/stsync-proxy/src/utils/mod.rs
@@ -1,4 +1,3 @@
-pub mod arena;
 pub mod serial;
 
 use std::cmp::Ordering;

From 45f8e803f6c74a97d0398463ca0ae50c6b9921cc Mon Sep 17 00:00:00 2001
From: MrGunflame <mrgunflame@protonmail.com>
Date: Wed, 15 Feb 2023 17:17:17 +0100
Subject: [PATCH 9/9] cargo update

---
 Cargo.lock | 348 ++++++++++++++++++++++++++---------------------------
 1 file changed, 174 insertions(+), 174 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index f0b6980..1af3650 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4,9 +4,9 @@ version = 3
 
 [[package]]
 name = "ahash"
-version = "0.8.2"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf6ccdb167abbf410dcb915cabd428929d7f6a04980b54a11f26a39f1c7f7107"
+checksum = "2c99f64d1e06488f620f932677e24bc6e2897582980441ae90a671415bd7ec2f"
 dependencies = [
  "cfg-if",
  "getrandom",
@@ -42,9 +42,9 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
 
 [[package]]
 name = "base64"
-version = "0.13.1"
+version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8"
+checksum = "a4a4ddaa51a5bc52a6948f74c06d20aaaddb71924eab79b8c97a8c556e942d6a"
 
 [[package]]
 name = "bitflags"
@@ -54,21 +54,21 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
 [[package]]
 name = "bumpalo"
-version = "3.11.1"
+version = "3.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "572f695136211188308f16ad2ca5c851a712c464060ae6974944458eb83880ba"
+checksum = "0d261e256854913907f67ed06efbc3338dfe6179796deefc1ff763fc1aee5535"
 
 [[package]]
 name = "bytes"
-version = "1.3.0"
+version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dfb24e866b15a1af2a1b663f10c6b6b8f397a84aadb828f12e5b289ec23a3a3c"
+checksum = "89b2fd2a0dcf38d7971e2194b6b6eebab45ae01067456a7fd93d5547a61b70be"
 
 [[package]]
 name = "cc"
-version = "1.0.78"
+version = "1.0.79"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a20104e2335ce8a659d6dd92a51a767a0c062599c73b343fd152cb401e828c3d"
+checksum = "50d30906286121d95be3d479533b458f87493b30a4b5f79a607db8f5d11aa91f"
 
 [[package]]
 name = "cfg-if"
@@ -78,9 +78,9 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
 [[package]]
 name = "clap"
-version = "4.0.32"
+version = "4.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a7db700bc935f9e43e88d00b0850dae18a63773cfbec6d8e070fccf7fef89a39"
+checksum = "f13b9c79b5d1dd500d20ef541215a6423c75829ef43117e1b4d17fd8af0b5d76"
 dependencies = [
  "bitflags",
  "clap_derive",
@@ -93,9 +93,9 @@ dependencies = [
 
 [[package]]
 name = "clap_derive"
-version = "4.0.21"
+version = "4.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0177313f9f02afc995627906bbd8967e2be069f5261954222dac78290c2b9014"
+checksum = "684a277d672e91966334af371f1a7b5833f9aa00b07c84e92fbce95e00208ce8"
 dependencies = [
  "heck",
  "proc-macro-error",
@@ -106,9 +106,9 @@ dependencies = [
 
 [[package]]
 name = "clap_lex"
-version = "0.3.0"
+version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d4198f73e42b4936b35b5bb248d81d2b595ecb170da0bac7655c54eedfa8da8"
+checksum = "783fe232adfca04f90f56201b26d79682d4cd2625e0bc7290b95123afe558ade"
 dependencies = [
  "os_str_bytes",
 ]
@@ -131,9 +131,9 @@ checksum = "5827cebf4670468b8772dd191856768aedcb1b0278a04f989f7766351917b9dc"
 
 [[package]]
 name = "encoding_rs"
-version = "0.8.31"
+version = "0.8.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9852635589dc9f9ea1b6fe9f05b50ef208c85c834a562f0c6abb1c475736ec2b"
+checksum = "071a31f4ee85403370b58aca746f01041ede6f0da2730960ad001edc2b71b394"
 dependencies = [
  "cfg-if",
 ]
@@ -174,9 +174,9 @@ dependencies = [
 
 [[package]]
 name = "fastrand"
-version = "1.8.0"
+version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a7a407cfaa3385c4ae6b23e84623d48c2798d06e3e6a1878f7f59f17b3f86499"
+checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be"
 dependencies = [
  "instant",
 ]
@@ -213,9 +213,9 @@ dependencies = [
 
 [[package]]
 name = "futures"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "38390104763dc37a5145a53c29c63c1290b5d316d6086ec32c293f6736051bb0"
+checksum = "13e2792b0ff0340399d58445b88fd9770e3489eff258a4cbc1523418f12abf84"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -228,9 +228,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52ba265a92256105f45b719605a571ffe2d1f0fea3807304b522c1d778f79eed"
+checksum = "2e5317663a9089767a1ec00a487df42e0ca174b61b4483213ac24448e4664df5"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -238,15 +238,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04909a7a7e4633ae6c4a9ab280aeb86da1236243a77b694a49eacd659a4bd3ac"
+checksum = "ec90ff4d0fe1f57d600049061dc6bb68ed03c7d2fbd697274c41805dcb3f8608"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7acc85df6714c176ab5edf386123fafe217be88c0840ec11f199441134a074e2"
+checksum = "e8de0a35a6ab97ec8869e32a2473f4b1324459e14c29275d14b10cb1fd19b50e"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -255,15 +255,15 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00f5fb52a06bdcadeb54e8d3671f8888a39697dcb0b81b23b55174030427f4eb"
+checksum = "bfb8371b6fb2aeb2d280374607aeabfc99d95c72edfe51692e42d3d7f0d08531"
 
 [[package]]
 name = "futures-macro"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bdfb8ce053d86b91919aad980c220b1fb8401a9394410e1c289ed7e66b61835d"
+checksum = "95a73af87da33b5acf53acfebdc339fe592ecf5357ac7c0a7734ab9d8c876a70"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -272,21 +272,21 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "39c15cf1a4aa79df40f1bb462fb39676d0ad9e366c2a33b590d7c66f4f81fcf9"
+checksum = "f310820bb3e8cfd46c80db4d7fb8353e15dfff853a127158425f31e0be6c8364"
 
 [[package]]
 name = "futures-task"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ffb393ac5d9a6eaa9d3fdf37ae2776656b706e200c8e16b1bdb227f5198e6ea"
+checksum = "dcf79a1bf610b10f42aea489289c5a2c478a786509693b80cd39c44ccd936366"
 
 [[package]]
 name = "futures-util"
-version = "0.3.25"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "197676987abd2f9cadff84926f410af1c183608d36641465df73ae8211dc65d6"
+checksum = "9c1d6de3acfef38d2be4b1f543f553131788603495be83da675e180c8d6b7bd1"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -338,9 +338,9 @@ checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
 
 [[package]]
 name = "heck"
-version = "0.4.0"
+version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9"
+checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
 
 [[package]]
 name = "hermit-abi"
@@ -360,6 +360,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "hermit-abi"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fed44880c466736ef9a5c5b5facefb5ed0785676d0c02d612db14e54f0d84286"
+
 [[package]]
 name = "http"
 version = "0.2.8"
@@ -405,9 +411,9 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "0.14.23"
+version = "0.14.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "034711faac9d2166cb1baf1a2fb0b60b1f277f8492fd72176c17f3515e1abd3c"
+checksum = "5e011372fa0b68db8350aa7a248930ecc7839bf46d8485577d69f117a75f164c"
 dependencies = [
  "bytes",
  "futures-channel",
@@ -471,30 +477,30 @@ dependencies = [
 
 [[package]]
 name = "io-lifetimes"
-version = "1.0.3"
+version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "46112a93252b123d31a119a8d1a1ac19deac4fac6e0e8b0df58f0d4e5870e63c"
+checksum = "1abeb7a0dd0f8181267ff8adc397075586500b81b28a73e8a0208b00fc170fb3"
 dependencies = [
  "libc",
- "windows-sys 0.42.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
 name = "ipnet"
-version = "2.7.0"
+version = "2.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11b0d96e660696543b251e58030cf9787df56da39dab19ad60eae7353040917e"
+checksum = "30e22bd8629359895450b59ea7a776c850561b96a3b1d31321c1949d9e6c9146"
 
 [[package]]
 name = "is-terminal"
-version = "0.4.2"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "28dfb6c8100ccc63462345b67d1bbc3679177c75ee4bf59bf29c8b1d110b8189"
+checksum = "22e18b0a45d56fe973d6db23972bf5bc46f988a4a2385deac9cc29572f09daef"
 dependencies = [
- "hermit-abi 0.2.6",
+ "hermit-abi 0.3.1",
  "io-lifetimes",
  "rustix",
- "windows-sys 0.42.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -505,9 +511,9 @@ checksum = "fad582f4b9e86b6caa621cabeb0963332d92eea04729ab12892c2533951e6440"
 
 [[package]]
 name = "js-sys"
-version = "0.3.60"
+version = "0.3.61"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "49409df3e3bf0856b916e2ceaca09ee28e6871cf7d9ce97a692cacfdb2a25a47"
+checksum = "445dde2150c55e483f3d8416706b97ec8e8237c307e5b7b4b8dd15e6af2a0730"
 dependencies = [
  "wasm-bindgen",
 ]
@@ -572,14 +578,14 @@ checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d"
 
 [[package]]
 name = "mio"
-version = "0.8.5"
+version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e5d732bc30207a6423068df043e3d02e0735b155ad7ce1a6f76fe2baa5b158de"
+checksum = "5b9d9a46eff5b4ff64b45a9e316a6d1e0bc719ef429cbec4dc630684212bfdf9"
 dependencies = [
  "libc",
  "log",
  "wasi",
- "windows-sys 0.42.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -602,9 +608,9 @@ dependencies = [
 
 [[package]]
 name = "nix"
-version = "0.26.1"
+version = "0.26.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "46a58d1d356c6597d08cde02c2f09d785b09e28711837b1ed667dc652c08a694"
+checksum = "bfdda3d196821d6af13126e40375cdf7da646a96114af134d5f417a9a1dc8e1a"
 dependencies = [
  "bitflags",
  "cfg-if",
@@ -626,9 +632,9 @@ dependencies = [
 
 [[package]]
 name = "once_cell"
-version = "1.17.0"
+version = "1.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f61fba1741ea2b3d6a1e3178721804bb716a68a6aeba1149b5d52e3d464ea66"
+checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3"
 
 [[package]]
 name = "openssl"
@@ -693,15 +699,15 @@ dependencies = [
 
 [[package]]
 name = "parking_lot_core"
-version = "0.9.5"
+version = "0.9.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ff9f3fef3968a3ec5945535ed654cb38ff72d7495a25619e2247fb15a2ed9ba"
+checksum = "9069cbb9f99e3a5083476ccb29ceb1de18b9118cafa53e90c9551235de2b9521"
 dependencies = [
  "cfg-if",
  "libc",
  "redox_syscall",
  "smallvec",
- "windows-sys 0.42.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -748,6 +754,11 @@ version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6ac9a59f73473f1b8d852421e59e64809f025994837ef743615c6d0c5b305160"
 
+[[package]]
+name = "polymock"
+version = "0.2.1"
+source = "git+https://github.com/MrGunflame/polymock-rs#e8b8e00dbb57bd520679452bb566a1e0c45dddd3"
+
 [[package]]
 name = "ppv-lite86"
 version = "0.2.17"
@@ -790,9 +801,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.49"
+version = "1.0.51"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "57a8eca9f9c4ffde41714334dee777596264c7825420f521abc92b5b5deb63a5"
+checksum = "5d727cae5b39d21da60fa540906919ad737832fe0b1c165da3a34d6548c849d6"
 dependencies = [
  "unicode-ident",
 ]
@@ -853,9 +864,9 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.7.0"
+version = "1.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e076559ef8e241f2ae3479e36f97bd5741c0330689e217ad51ce2c76808b868a"
+checksum = "48aaa5748ba571fb95cd2c85c09f629215d3a6ece942baa100950af03a34f733"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -879,9 +890,9 @@ dependencies = [
 
 [[package]]
 name = "reqwest"
-version = "0.11.13"
+version = "0.11.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68cc60575865c7831548863cc02356512e3f1dc2f3f82cb837d7fc4cc8f3c97c"
+checksum = "21eed90ec8570952d53b772ecf8f206aa1ec9a3d76b2521c56c42973f2d91ee9"
 dependencies = [
  "base64",
  "bytes",
@@ -916,16 +927,16 @@ dependencies = [
 
 [[package]]
 name = "rustix"
-version = "0.36.6"
+version = "0.36.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4feacf7db682c6c329c4ede12649cd36ecab0f3be5b7d74e6a20304725db4549"
+checksum = "f43abb88211988493c1abb44a70efa56ff0ce98f233b7b276146f1f3f7ba9644"
 dependencies = [
  "bitflags",
  "errno",
  "io-lifetimes",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.42.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -936,12 +947,11 @@ checksum = "7b4b9743ed687d4b4bcedf9ff5eaa7398495ae14e61cba0a295704edbc7decde"
 
 [[package]]
 name = "schannel"
-version = "0.1.20"
+version = "0.1.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "88d6731146462ea25d9244b2ed5fd1d716d25c52e4d54aa4fb0f3c4e9854dbe2"
+checksum = "713cfb06c7059f3588fb8044c0fad1d09e3c01d225e25b9220dbfdcf16dbb1b3"
 dependencies = [
- "lazy_static",
- "windows-sys 0.36.1",
+ "windows-sys 0.42.0",
 ]
 
 [[package]]
@@ -952,9 +962,9 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
 
 [[package]]
 name = "security-framework"
-version = "2.7.0"
+version = "2.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2bc1bb97804af6631813c55739f771071e0f2ed33ee20b68c86ec505d906356c"
+checksum = "a332be01508d814fed64bf28f798a146d73792121129962fdf335bb3c49a4254"
 dependencies = [
  "bitflags",
  "core-foundation",
@@ -965,9 +975,9 @@ dependencies = [
 
 [[package]]
 name = "security-framework-sys"
-version = "2.6.1"
+version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0160a13a177a45bfb43ce71c01580998474f556ad854dcbca936dd2841a5c556"
+checksum = "31c9bb296072e961fcbd8853511dd39c2d8be2deb1e17c6860b1d30732b323b4"
 dependencies = [
  "core-foundation-sys",
  "libc",
@@ -995,9 +1005,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.91"
+version = "1.0.93"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "877c235533714907a8c2464236f5c4b2a17262ef1bd71f38f35ea592c8da6883"
+checksum = "cad406b69c91885b5107daf2c29572f6c8cdb3c66826821e286c533490c0bc76"
 dependencies = [
  "itoa",
  "ryu",
@@ -1018,9 +1028,9 @@ dependencies = [
 
 [[package]]
 name = "signal-hook-registry"
-version = "1.4.0"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e51e73328dc4ac0c7ccbda3a494dfa03df1de2f46018127f60c693f2648455b0"
+checksum = "d8229b473baa5980ac72ef434c4415e70c4b5e71b423043adb4ba059f89c99a1"
 dependencies = [
  "libc",
 ]
@@ -1098,6 +1108,15 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "streamsync-transmit"
+version = "0.1.0"
+dependencies = [
+ "clap",
+ "streamsync-api",
+ "tokio",
+]
+
 [[package]]
 name = "strsim"
 version = "0.10.0"
@@ -1106,10 +1125,9 @@ checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
 
 [[package]]
 name = "stsync-proxy"
-version = "0.2.0"
+version = "0.2.1"
 dependencies = [
  "ahash",
- "bytes",
  "clap",
  "futures",
  "hyper",
@@ -1117,6 +1135,7 @@ dependencies = [
  "nix",
  "parking_lot",
  "pin-project",
+ "polymock",
  "pretty_env_logger",
  "rand",
  "serde",
@@ -1158,9 +1177,9 @@ dependencies = [
 
 [[package]]
 name = "termcolor"
-version = "1.1.3"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bab24d30b911b2376f3a13cc2cd443142f0c81dda04c118693e35b3835757755"
+checksum = "be55cf8942feac5c765c2c993422806843c9a9a45d4d5c407ad6dd2ea95eb9b6"
 dependencies = [
  "winapi-util",
 ]
@@ -1196,15 +1215,15 @@ dependencies = [
 
 [[package]]
 name = "tinyvec_macros"
-version = "0.1.0"
+version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
+checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.24.1"
+version = "1.25.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d9f76183f91ecfb55e1d7d5602bd1d979e38a3a522fe900241cf195624d67ae"
+checksum = "c8e00990ebabbe4c14c08aca901caed183ecd5c09562a12c824bb53d3c3fd3af"
 dependencies = [
  "autocfg",
  "bytes",
@@ -1234,9 +1253,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-native-tls"
-version = "0.3.0"
+version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b"
+checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2"
 dependencies = [
  "native-tls",
  "tokio",
@@ -1256,9 +1275,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-util"
-version = "0.7.4"
+version = "0.7.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bb2e075f03b3d66d8d8785356224ba688d2906a371015e225beeb65ca92c740"
+checksum = "5427d89453009325de0d8f342c9490009f76e999cb7672d77e46267448f7e6b2"
 dependencies = [
  "bytes",
  "futures-core",
@@ -1270,9 +1289,9 @@ dependencies = [
 
 [[package]]
 name = "toml"
-version = "0.5.10"
+version = "0.5.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1333c76748e868a4d9d1017b5ab53171dfd095f70c712fdb4653a406547f598f"
+checksum = "f4f7f0dd8d50a853a531c426359045b1998f04219d88799810762cd4ad314234"
 dependencies = [
  "serde",
 ]
@@ -1318,15 +1337,15 @@ dependencies = [
 
 [[package]]
 name = "try-lock"
-version = "0.2.3"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642"
+checksum = "3528ecfd12c466c6f163363caf2d02a71161dd5e1cc6ae7b34207ea2d42d81ed"
 
 [[package]]
 name = "unicode-bidi"
-version = "0.3.8"
+version = "0.3.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "099b7128301d285f79ddd55b9a83d5e6b9e97c92e0ea0daebee7263e932de992"
+checksum = "d54675592c1dbefd78cbd98db9bacd89886e1ca50692a0692baefffdeb92dd58"
 
 [[package]]
 name = "unicode-ident"
@@ -1384,9 +1403,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.83"
+version = "0.2.84"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eaf9f5aceeec8be17c128b2e93e031fb8a4d469bb9c4ae2d7dc1888b26887268"
+checksum = "31f8dcbc21f30d9b8f2ea926ecb58f6b91192c17e9d33594b3df58b2007ca53b"
 dependencies = [
  "cfg-if",
  "wasm-bindgen-macro",
@@ -1394,9 +1413,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-backend"
-version = "0.2.83"
+version = "0.2.84"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c8ffb332579b0557b52d268b91feab8df3615f265d5270fec2a8c95b17c1142"
+checksum = "95ce90fd5bcc06af55a641a86428ee4229e44e07033963a2290a8e241607ccb9"
 dependencies = [
  "bumpalo",
  "log",
@@ -1409,9 +1428,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.33"
+version = "0.4.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23639446165ca5a5de86ae1d8896b737ae80319560fbaa4c2887b7da6e7ebd7d"
+checksum = "f219e0d211ba40266969f6dbdd90636da12f75bee4fc9d6c23d1260dadb51454"
 dependencies = [
  "cfg-if",
  "js-sys",
@@ -1421,9 +1440,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.83"
+version = "0.2.84"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "052be0f94026e6cbc75cdefc9bae13fd6052cdcaf532fa6c45e7ae33a1e6c810"
+checksum = "4c21f77c0bedc37fd5dc21f897894a5ca01e7bb159884559461862ae90c0b4c5"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -1431,9 +1450,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.83"
+version = "0.2.84"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "07bc0c051dc5f23e307b13285f9d75df86bfdf816c5721e573dec1f9b8aa193c"
+checksum = "2aff81306fcac3c7515ad4e177f521b5c9a15f2b08f4e32d823066102f35a5f6"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1444,15 +1463,15 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.83"
+version = "0.2.84"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c38c045535d93ec4f0b4defec448e4291638ee608530863b1e2ba115d4fff7f"
+checksum = "0046fef7e28c3804e5e38bfa31ea2a0f73905319b677e57ebe37e49358989b5d"
 
 [[package]]
 name = "web-sys"
-version = "0.3.60"
+version = "0.3.61"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bcda906d8be16e728fd5adc5b729afad4e444e106ab28cd1c7256e54fa61510f"
+checksum = "e33b99f4b23ba3eec1a53ac264e35a755f00e966e0065077d6027c0f575b0b97"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -1489,19 +1508,6 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
 
-[[package]]
-name = "windows-sys"
-version = "0.36.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2"
-dependencies = [
- "windows_aarch64_msvc 0.36.1",
- "windows_i686_gnu 0.36.1",
- "windows_i686_msvc 0.36.1",
- "windows_x86_64_gnu 0.36.1",
- "windows_x86_64_msvc 0.36.1",
-]
-
 [[package]]
 name = "windows-sys"
 version = "0.42.0"
@@ -1509,85 +1515,79 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5a3e1820f08b8513f676f7ab6c1f99ff312fb97b553d30ff4dd86f9f15728aa7"
 dependencies = [
  "windows_aarch64_gnullvm",
- "windows_aarch64_msvc 0.42.0",
- "windows_i686_gnu 0.42.0",
- "windows_i686_msvc 0.42.0",
- "windows_x86_64_gnu 0.42.0",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
  "windows_x86_64_gnullvm",
- "windows_x86_64_msvc 0.42.0",
+ "windows_x86_64_msvc",
 ]
 
 [[package]]
-name = "windows_aarch64_gnullvm"
-version = "0.42.0"
+name = "windows-sys"
+version = "0.45.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41d2aa71f6f0cbe00ae5167d90ef3cfe66527d6f613ca78ac8024c3ccab9a19e"
+checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
+dependencies = [
+ "windows-targets",
+]
 
 [[package]]
-name = "windows_aarch64_msvc"
-version = "0.36.1"
+name = "windows-targets"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47"
+checksum = "8e2522491fbfcd58cc84d47aeb2958948c4b8982e9a2d8a2a35bbaed431390e7"
+dependencies = [
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc",
+]
 
 [[package]]
-name = "windows_aarch64_msvc"
-version = "0.42.0"
+name = "windows_aarch64_gnullvm"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dd0f252f5a35cac83d6311b2e795981f5ee6e67eb1f9a7f64eb4500fbc4dcdb4"
+checksum = "8c9864e83243fdec7fc9c5444389dcbbfd258f745e7853198f365e3c4968a608"
 
 [[package]]
-name = "windows_i686_gnu"
-version = "0.36.1"
+name = "windows_aarch64_msvc"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6"
+checksum = "4c8b1b673ffc16c47a9ff48570a9d85e25d265735c503681332589af6253c6c7"
 
 [[package]]
 name = "windows_i686_gnu"
-version = "0.42.0"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbeae19f6716841636c28d695375df17562ca208b2b7d0dc47635a50ae6c5de7"
+checksum = "de3887528ad530ba7bdbb1faa8275ec7a1155a45ffa57c37993960277145d640"
 
 [[package]]
 name = "windows_i686_msvc"
-version = "0.36.1"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024"
-
-[[package]]
-name = "windows_i686_msvc"
-version = "0.42.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "84c12f65daa39dd2babe6e442988fc329d6243fdce47d7d2d155b8d874862246"
+checksum = "bf4d1122317eddd6ff351aa852118a2418ad4214e6613a50e0191f7004372605"
 
 [[package]]
 name = "windows_x86_64_gnu"
-version = "0.36.1"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1"
-
-[[package]]
-name = "windows_x86_64_gnu"
-version = "0.42.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf7b1b21b5362cbc318f686150e5bcea75ecedc74dd157d874d754a2ca44b0ed"
+checksum = "c1040f221285e17ebccbc2591ffdc2d44ee1f9186324dd3e84e99ac68d699c45"
 
 [[package]]
 name = "windows_x86_64_gnullvm"
-version = "0.42.0"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09d525d2ba30eeb3297665bd434a54297e4170c7f1a44cad4ef58095b4cd2028"
+checksum = "628bfdf232daa22b0d64fdb62b09fcc36bb01f05a3939e20ab73aaf9470d0463"
 
 [[package]]
 name = "windows_x86_64_msvc"
-version = "0.36.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
-
-[[package]]
-name = "windows_x86_64_msvc"
-version = "0.42.0"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f40009d85759725a34da6d89a94e63d7bdc50a862acf0dbc7c8e488f1edcb6f5"
+checksum = "447660ad36a13288b1db4d4248e857b510e8c3a225c822ba4fb748c0aafecffd"
 
 [[package]]
 name = "winreg"