ntex_bytes/
pages.rs

1#![allow(clippy::missing_panics_doc, clippy::box_collection)]
2use std::{borrow::Borrow, cell::Cell, cmp, collections::VecDeque, fmt, io, mem, ops};
3
4use crate::{Buf, BufMut, BytePageSize, ByteString, Bytes, BytesMut};
5use crate::{buf::UninitSlice, stvec::StorageVec};
6
7/// A growable sequence of byte pages.
8///
9/// Data is stored in fixed-capacity pages selected by [`BytePageSize`]. This
10/// avoids reallocating and copying one large contiguous buffer as data grows.
11pub struct BytePages {
12    st: Option<Box<Inner>>,
13    current: Option<StorageVec>,
14}
15
16#[derive(Debug)]
17struct Inner {
18    size: BytePageSize,
19    /// Total length of `pages`, so `len()` does not walk them.
20    len: usize,
21    pages: VecDeque<BytePage>,
22}
23
24thread_local! {
25    static CACHE: Cell<Option<Box<Vec<Box<Inner>>>>> = Cell::new(Some(Box::default()));
26}
27const CACHE_SIZE: usize = 128;
28
29/// Appended data up to this size is copied into the current page.
30const APPEND_COPY_LIMIT: usize = 4096;
31
32impl BytePages {
33    /// Creates a new `BytePages` with the specified page size.
34    ///
35    /// Pages are allocated lazily using the specified capacity category.
36    ///
37    /// # Panics
38    ///
39    /// Panics if `size` is [`BytePageSize::Unset`].
40    pub fn new(size: BytePageSize) -> Self {
41        assert!(size != BytePageSize::Unset, "Page size cannot be Unset");
42
43        // the cache is unavailable while the thread-local is being destroyed
44        let cached = CACHE
45            .try_with(|c| {
46                let mut cache = c.take()?;
47                let item = cache.pop();
48                c.set(Some(cache));
49                item
50            })
51            .ok()
52            .flatten();
53
54        let st = if let Some(mut item) = cached {
55            item.size = size;
56            item
57        } else {
58            Box::new(Inner {
59                size,
60                len: 0,
61                pages: VecDeque::with_capacity(8),
62            })
63        };
64
65        BytePages {
66            st: Some(st),
67            current: None,
68        }
69    }
70
71    fn pages(&self) -> &VecDeque<BytePage> {
72        &self.st.as_ref().unwrap().pages
73    }
74
75    // Pages are only added and removed through these methods, which keep
76    // `Inner::len` up to date. A page in the list is never modified in place.
77    fn push_front(&mut self, page: BytePage) {
78        let st = self.st.as_mut().unwrap();
79        st.len += page.len();
80        st.pages.push_front(page);
81    }
82
83    fn pop_front(&mut self) -> Option<BytePage> {
84        let st = self.st.as_mut().unwrap();
85        let page = st.pages.pop_front()?;
86        st.len -= page.len();
87        Some(page)
88    }
89
90    fn push_back(&mut self, page: BytePage) {
91        let st = self.st.as_mut().unwrap();
92        st.len += page.len();
93        let pages = &mut st.pages;
94        pages.push_back(page);
95
96        #[cfg(feature = "overuse")]
97        if pages.len() == 128 {
98            log::debug!(
99                "Number of pages {}\n{:?}",
100                pages.len(),
101                backtrace::Backtrace::new()
102            );
103        }
104    }
105
106    /// Returns the capacity category used for new pages.
107    pub fn page_size(&self) -> BytePageSize {
108        self.st.as_ref().unwrap().size
109    }
110
111    /// Sets the page size for new pages.
112    ///
113    /// # Panics
114    ///
115    /// Panics if `size` is [`BytePageSize::Unset`].
116    pub fn set_page_size(&mut self, size: BytePageSize) {
117        assert!(size != BytePageSize::Unset, "Page size cannot be Unset");
118        self.st.as_mut().unwrap().size = size;
119    }
120
121    /// Inserts a non-empty page at the front of the collection.
122    ///
123    /// Returns whether a page was inserted.
124    pub fn prepend<T>(&mut self, buf: T) -> bool
125    where
126        BytePage: From<T>,
127    {
128        let p = BytePage::from(buf);
129        if p.is_empty() {
130            false
131        } else {
132            self.push_front(p);
133            true
134        }
135    }
136
137    /// Appends a page to the back of the collection.
138    ///
139    /// Empty pages are ignored. If the current page holds no data and `buf` is
140    /// a unique buffer with spare capacity, it becomes the new current page.
141    /// Data of up to 4 KiB is copied into the current page, and into new
142    /// pages as needed. Larger data is added as a separate page without
143    /// copying: the filled part of the current page is split off in front of
144    /// it, and the spare capacity of the current page stays available for
145    /// later writes.
146    pub fn append<T>(&mut self, buf: T)
147    where
148        BytePage: From<T>,
149    {
150        let p = BytePage::from(buf);
151        if !p.is_empty() {
152            if self.current_len() == 0 {
153                match p.into_storage() {
154                    Ok(st) => {
155                        self.current = Some(st);
156                    }
157                    Err(page) => {
158                        // add buffer to the page list
159                        self.push_back(page);
160                    }
161                }
162            } else if p.len() <= APPEND_COPY_LIMIT {
163                self.put_slice(p.as_ref());
164            } else {
165                // the current page is never full, its spare capacity is kept
166                if let Some(st) = self.current.as_mut() {
167                    let head = st.split_to(st.len());
168                    self.push_back(<BytePage as From<Bytes>>::from(Bytes { storage: head }));
169                }
170                self.push_back(p);
171            }
172        }
173    }
174
175    #[inline]
176    /// Appends the given bytes to this page object.
177    ///
178    /// Tries to write the data into the current page first. If there
179    /// is insufficient space, one or more new pages are allocated as
180    /// needed, and the remaining data is copied into them.
181    pub fn extend_from_slice(&mut self, extend: &[u8]) {
182        self.put_slice(extend);
183    }
184
185    #[inline]
186    /// Returns the total number of buffered bytes.
187    pub fn len(&self) -> usize {
188        self.st.as_ref().unwrap().len + self.current_len()
189    }
190
191    fn current_len(&self) -> usize {
192        self.current
193            .as_ref()
194            .map(StorageVec::len)
195            .unwrap_or_default()
196    }
197
198    // spare capacity of the current page
199    fn spare(&self) -> usize {
200        self.current
201            .as_ref()
202            .map(StorageVec::remaining)
203            .unwrap_or_default()
204    }
205
206    #[inline]
207    /// Returns `true` if no bytes are buffered.
208    pub fn is_empty(&self) -> bool {
209        self.len() == 0
210    }
211
212    #[inline]
213    /// Returns the number of allocated pages containing buffered data.
214    pub fn num_pages(&self) -> usize {
215        if self.current_len() == 0 {
216            self.pages().len()
217        } else {
218            self.pages().len() + 1
219        }
220    }
221
222    /// Removes and returns the first page from the collection.
223    ///
224    /// The current writable page is returned last. Returns `None` if there are
225    /// no pages with data, an empty current page keeps its spare capacity for
226    /// later writes.
227    pub fn take(&mut self) -> Option<BytePage> {
228        if let Some(page) = self.pop_front() {
229            Some(page)
230        } else if self.current_len() == 0 {
231            None
232        } else {
233            self.current.take().map(BytePage::from)
234        }
235    }
236
237    #[inline]
238    /// Appends all buffered data to another [`BytePages`] value, `self` is
239    /// left unchanged.
240    ///
241    /// Pages are shared with `pages` rather than copied, unless they are small
242    /// enough to be copied into the current page of `pages`, see
243    /// [`append`](Self::append).
244    pub fn copy_to(&self, pages: &mut BytePages) {
245        for p in self.pages() {
246            pages.append(p.clone());
247        }
248
249        if let Some(st) = &self.current {
250            // an immutable view, `st` stays the only handle that can write
251            // to the spare capacity
252            pages.append(Bytes {
253                storage: st.shallow_freeze(),
254            });
255        }
256    }
257
258    #[inline]
259    /// Moves all buffered data to the back of another [`BytePages`] value,
260    /// leaving `self` empty.
261    ///
262    /// Pages are moved according to the rules of [`append`](Self::append).
263    pub fn move_to(&mut self, pages: &mut BytePages) {
264        while let Some(page) = self.take() {
265            pages.append(page);
266        }
267    }
268
269    /// Splits the buffer into two at the given index.
270    ///
271    /// Afterwards, `self` contains elements `[at, len)`, and the returned [`BytePages`]
272    /// contains elements `[0, at)`. If `at > len`, all data is moved.
273    ///
274    /// Depending on the underlying storage, this operation might be `O(1)` or could
275    /// involve a memory copy.
276    #[must_use]
277    pub fn split_to(&mut self, at: usize) -> BytePages {
278        let mut pages = BytePages::new(self.page_size());
279        self.split_into(at, &mut pages);
280        pages
281    }
282
283    /// Splits the buffer, adding the resulting items to the supplied pages object.
284    ///
285    /// Afterwards, `self` contains elements `[at, len)`, and elements `[0, at)`
286    /// are appended to `to`. If `at > len`, all data is moved.
287    ///
288    /// Depending on the underlying storage, this operation might be `O(1)` or could
289    /// involve a memory copy.
290    pub fn split_into(&mut self, mut at: usize, to: &mut BytePages) {
291        while let Some(mut page) = self.pop_front() {
292            let len = cmp::min(page.len(), at);
293            to.append(page.split_to(len));
294
295            if !page.is_empty() {
296                self.push_front(page);
297                return;
298            }
299            at -= len;
300        }
301        if at > 0
302            && let Some(mut st) = self.current.take()
303        {
304            if at < st.len() {
305                // the remainder stays writable, so its spare capacity is kept
306                to.append(Bytes {
307                    storage: st.split_to(at),
308                });
309                self.current = Some(st);
310            } else if st.len() == 0 {
311                self.current = Some(st);
312            } else {
313                to.append(BytePage::from(st));
314            }
315        }
316    }
317
318    /// Clears the buffer, removing all data.
319    #[inline]
320    pub fn clear(&mut self) {
321        while self.take().is_some() {}
322    }
323
324    /// Drains all pages into one immutable [`Bytes`] value.
325    #[inline]
326    #[must_use]
327    pub fn freeze(&mut self) -> Bytes {
328        let pages = self.num_pages();
329        if pages == 0 || self.is_empty() {
330            Bytes::new()
331        } else if pages == 1 {
332            self.take().unwrap().freeze()
333        } else {
334            let mut buf = BytesMut::with_capacity(self.len());
335            while let Some(p) = self.take() {
336                buf.extend_from_slice(&p);
337            }
338            buf.freeze()
339        }
340    }
341
342    #[inline]
343    /// Moves the current writable page from `pages` if this value is empty.
344    pub fn try_get_current_from(&mut self, pages: &mut BytePages) {
345        if self.pages().is_empty()
346            && self.current.is_none()
347            && let Some(st) = pages.current.take()
348        {
349            self.current = Some(st);
350        }
351    }
352
353    /// Passes a new mutable buffer to `f` and appends its contents.
354    ///
355    /// After `f` returns, the buffer is appended according to the rules of
356    /// [`append`](Self::append). Existing pages and the current writable page
357    /// are not exposed to `f`. If `f` panics, the temporary buffer is dropped
358    /// and this collection is unchanged.
359    ///
360    /// This is a low-level API intended for ntex internals and may change
361    /// without notice.
362    #[doc(hidden)]
363    #[deprecated(
364        since = "1.10.0",
365        note = "use the `BufMut` methods of `BytePages` instead"
366    )]
367    pub fn with_bytes_mut<F, R>(&mut self, f: F) -> R
368    where
369        F: FnOnce(&mut BytesMut) -> R,
370    {
371        let mut buf = BytesMut::new();
372        let res = f(&mut buf);
373        self.append(buf);
374        res
375    }
376
377    fn with_current<F, R>(&mut self, f: F) -> R
378    where
379        F: FnOnce(&mut StorageVec) -> R,
380    {
381        let mut st = self
382            .current
383            .take()
384            .unwrap_or_else(|| StorageVec::sized(self.page_size()));
385        let result = f(&mut st);
386
387        // a full page moves to the page list
388        if st.is_full() {
389            self.push_back(BytePage::from(st));
390        } else {
391            self.current = Some(st);
392        }
393
394        result
395    }
396}
397
398impl Drop for BytePages {
399    fn drop(&mut self) {
400        if let Some(mut st) = self.st.take() {
401            st.pages.clear();
402            // a large write must not pin its page list in the cache
403            st.pages.shrink_to(8);
404            st.len = 0;
405            // the cache is unavailable while the thread-local is being destroyed
406            let _ = CACHE.try_with(move |c| {
407                if let Some(mut cache) = c.take() {
408                    if cache.len() < CACHE_SIZE {
409                        cache.push(st);
410                    }
411                    c.set(Some(cache));
412                }
413            });
414        }
415    }
416}
417
418impl fmt::Debug for BytePages {
419    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
420        let mut f = fmt.debug_tuple("BytePages");
421        for p in self.pages() {
422            f.field(p);
423        }
424        if let Some(st) = &self.current {
425            f.field(&crate::debug::BsDebug(st.as_ref()));
426        }
427        f.finish()
428    }
429}
430
431impl Default for BytePages {
432    fn default() -> Self {
433        BytePages::new(BytePageSize::Size16)
434    }
435}
436
437/// Pages are allocated on demand, so `remaining_mut()` reports
438/// `usize::MAX - len` and `chunk_mut()` is never empty. The chunk covers the
439/// spare capacity of the current page only.
440impl BufMut for BytePages {
441    #[inline]
442    fn remaining_mut(&self) -> usize {
443        usize::MAX - self.len()
444    }
445
446    #[inline]
447    unsafe fn advance_mut(&mut self, cnt: usize) {
448        if cnt == 0 {
449            return;
450        }
451        let spare = self.spare();
452        assert!(
453            cnt <= spare,
454            "cannot advance past the current page: {cnt:?} <= {spare:?}"
455        );
456        let st = self.current.as_mut().unwrap();
457        st.set_len(st.len() + cnt);
458    }
459
460    #[inline]
461    fn chunk_mut(&mut self) -> &mut UninitSlice {
462        if self.spare() == 0 {
463            if let Some(st) = self.current.take() {
464                self.push_back(BytePage::from(st));
465            }
466            self.current = Some(StorageVec::sized(self.page_size()));
467        }
468        // `current` is set, a new page is allocated above if there is no spare capacity
469        self.current.as_mut().unwrap().spare_mut()
470    }
471
472    fn put<T: Buf>(&mut self, mut src: T)
473    where
474        Self: Sized,
475    {
476        while src.has_remaining() {
477            let chunk = src.chunk();
478            let len = chunk.len();
479            self.put_slice(chunk);
480            src.advance(len);
481        }
482    }
483
484    fn put_slice(&mut self, mut src: &[u8]) {
485        while !src.is_empty() {
486            let amount = self.with_current(|st| st.put_slice_partial(src));
487
488            src = &src[amount..];
489        }
490    }
491
492    #[inline]
493    fn put_u8(&mut self, n: u8) {
494        self.with_current(|st| st.put_u8(n));
495    }
496
497    #[inline]
498    fn put_i8(&mut self, n: i8) {
499        self.put_u8(n as u8);
500    }
501}
502
503impl Clone for BytePages {
504    fn clone(&self) -> Self {
505        let size = self.page_size();
506        let mut pages = BytePages::new(size);
507        self.copy_to(&mut pages);
508        pages
509    }
510}
511
512impl io::Write for BytePages {
513    fn write(&mut self, src: &[u8]) -> Result<usize, io::Error> {
514        self.put_slice(src);
515        Ok(src.len())
516    }
517
518    fn flush(&mut self) -> Result<(), io::Error> {
519        Ok(())
520    }
521}
522
523impl From<BytePages> for Bytes {
524    /// A single page is converted without copying.
525    fn from(mut pages: BytePages) -> Bytes {
526        pages.freeze()
527    }
528}
529
530impl From<BytePages> for BytesMut {
531    /// A single page is converted without copying if nothing else refers to
532    /// its buffer.
533    fn from(mut pages: BytePages) -> BytesMut {
534        if pages.num_pages() == 1 {
535            return BytesMut::from(pages.take().unwrap());
536        }
537
538        let mut buf = BytesMut::with_capacity(pages.len());
539        while let Some(p) = pages.take() {
540            buf.extend_from_slice(&p);
541        }
542        buf
543    }
544}
545
546/// A contiguous chunk stored by [`BytePages`].
547pub struct BytePage {
548    inner: StorageType,
549}
550
551enum StorageType {
552    Bytes(Bytes),
553    Storage(StorageVec),
554    Vec(Vec<u8>),
555}
556
557impl BytePage {
558    #[inline]
559    /// Returns the number of bytes contained in this `BytePage`.
560    pub fn len(&self) -> usize {
561        self.as_ref().len()
562    }
563
564    #[inline]
565    /// Returns `true` if the page is empty.
566    pub fn is_empty(&self) -> bool {
567        self.len() == 0
568    }
569
570    #[inline]
571    /// Returns a raw pointer to the data.
572    ///
573    /// # Safety
574    ///
575    /// The returned pointer may only be dereferenced while the page is neither
576    /// moved, modified nor dropped, and only for [`len`](Self::len) bytes. An
577    /// inline page stores its data inside the `BytePage` itself, so moving the
578    /// page invalidates the pointer, see [`is_inline`](Self::is_inline).
579    pub unsafe fn as_ptr(&self) -> *const u8 {
580        unsafe {
581            match &self.inner {
582                StorageType::Bytes(b) => b.storage.as_ptr(),
583                StorageType::Storage(b) => b.as_ptr(),
584                StorageType::Vec(b) => b.as_ptr(),
585            }
586        }
587    }
588
589    #[inline]
590    #[doc(hidden)]
591    /// Returns the kind of storage backing this page.
592    pub fn info(&self) -> crate::info::PageKind {
593        match &self.inner {
594            StorageType::Bytes(_) => crate::info::PageKind::Bytes,
595            StorageType::Storage(_) => crate::info::PageKind::Storage,
596            StorageType::Vec(_) => crate::info::PageKind::Vec,
597        }
598    }
599
600    #[inline]
601    #[doc(hidden)]
602    /// Returns `true` if the data is stored inside the `BytePage` itself.
603    ///
604    /// Moving an inline page moves its data, pointers from `as_ptr()` do not
605    /// survive the move.
606    pub fn is_inline(&self) -> bool {
607        matches!(&self.inner, StorageType::Bytes(b) if b.is_inline())
608    }
609
610    /// Splits the buffer into two at the given index.
611    ///
612    /// Afterwards, `self` contains elements `[at, len)`, and the returned `BytePage`
613    /// contains elements `[0, at)`. If `at > len`, all data is moved.
614    ///
615    /// Depending on the underlying storage, this operation might be `O(1)` or could
616    /// involve a memory copy.
617    #[must_use]
618    pub fn split_to(&mut self, at: usize) -> BytePage {
619        match &mut self.inner {
620            StorageType::Bytes(b) => {
621                let buf = b.split_to(cmp::min(at, b.len()));
622                BytePage {
623                    inner: StorageType::Bytes(buf),
624                }
625            }
626            StorageType::Storage(_) => {
627                let inner = mem::replace(&mut self.inner, StorageType::Bytes(Bytes::new()));
628                if let StorageType::Storage(st) = inner {
629                    self.inner = StorageType::Bytes(Bytes {
630                        storage: st.freeze(),
631                    });
632                    self.split_to(at)
633                } else {
634                    unreachable!()
635                }
636            }
637            StorageType::Vec(_) => {
638                let inner = mem::replace(&mut self.inner, StorageType::Bytes(Bytes::new()));
639                if let StorageType::Vec(b) = inner {
640                    self.inner = StorageType::Bytes(Bytes::copy_from_slice(&b));
641                    self.split_to(at)
642                } else {
643                    unreachable!()
644                }
645            }
646        }
647    }
648
649    /// Advance the internal cursor.
650    ///
651    /// Afterwards `self` contains elements `[cnt, len)`.
652    /// This is an `O(1)` operation, except for pages backed by a `Vec<u8>`,
653    /// whose remaining data is copied.
654    ///
655    /// # Panics
656    ///
657    /// Panics if `cnt > len`.
658    #[inline]
659    pub fn advance_to(&mut self, cnt: usize) {
660        match &mut self.inner {
661            StorageType::Bytes(b) => b.advance_to(cnt),
662            StorageType::Storage(b) => unsafe { b.set_start(cnt) },
663            StorageType::Vec(b) => {
664                assert!(
665                    cnt <= b.len(),
666                    "cannot advance past the end of the buffer, cnt:{cnt} len:{}",
667                    b.len()
668                );
669                self.inner = StorageType::Bytes(Bytes::copy_from_slice(&b[cnt..]));
670            }
671        }
672    }
673
674    /// Converts `self` into an immutable `Bytes`.
675    #[inline]
676    #[must_use]
677    pub fn freeze(self) -> Bytes {
678        match self.inner {
679            StorageType::Bytes(b) => b,
680            StorageType::Storage(st) => Bytes {
681                storage: st.freeze(),
682            },
683            StorageType::Vec(v) => Bytes::from(v),
684        }
685    }
686
687    fn into_storage(self) -> Result<StorageVec, Self> {
688        if let StorageType::Storage(st) = self.inner {
689            // SAFETY: Converting back to `StorageVec` requires uniqueness.
690            if !st.is_full() && st.is_unique() {
691                Ok(st)
692            } else {
693                Err(Self {
694                    inner: StorageType::Storage(st),
695                })
696            }
697        } else {
698            Err(self)
699        }
700    }
701}
702
703impl Clone for BytePage {
704    fn clone(&self) -> Self {
705        let inner = match &self.inner {
706            StorageType::Bytes(b) => StorageType::Bytes(b.clone()),
707            // The clone is an immutable view, `st` must stay the only
708            // handle that can modify the shared header and spare capacity
709            StorageType::Storage(st) => StorageType::Bytes(Bytes {
710                storage: st.shallow_freeze(),
711            }),
712            StorageType::Vec(b) => StorageType::Bytes(Bytes::copy_from_slice(b)),
713        };
714
715        Self { inner }
716    }
717}
718
719impl AsRef<[u8]> for BytePage {
720    #[inline]
721    fn as_ref(&self) -> &[u8] {
722        match &self.inner {
723            StorageType::Bytes(b) => b.as_ref(),
724            StorageType::Storage(b) => b.as_ref(),
725            StorageType::Vec(b) => b.as_ref(),
726        }
727    }
728}
729
730impl Borrow<[u8]> for BytePage {
731    #[inline]
732    fn borrow(&self) -> &[u8] {
733        self.as_ref()
734    }
735}
736
737impl From<Bytes> for BytePage {
738    fn from(buf: Bytes) -> Self {
739        BytePage {
740            inner: StorageType::Bytes(buf),
741        }
742    }
743}
744
745impl<'a> From<&'a Bytes> for BytePage {
746    fn from(buf: &'a Bytes) -> Self {
747        BytePage {
748            inner: StorageType::Bytes(buf.clone()),
749        }
750    }
751}
752
753impl From<BytesMut> for BytePage {
754    fn from(buf: BytesMut) -> Self {
755        BytePage {
756            inner: StorageType::Storage(buf.storage),
757        }
758    }
759}
760
761impl From<ByteString> for BytePage {
762    fn from(s: ByteString) -> Self {
763        s.into_bytes().into()
764    }
765}
766
767impl<'a> From<&'a ByteString> for BytePage {
768    fn from(s: &'a ByteString) -> Self {
769        s.clone().into_bytes().into()
770    }
771}
772
773impl From<StorageVec> for BytePage {
774    fn from(buf: StorageVec) -> Self {
775        BytePage {
776            inner: StorageType::Storage(buf),
777        }
778    }
779}
780
781impl From<Vec<u8>> for BytePage {
782    fn from(buf: Vec<u8>) -> Self {
783        BytePage {
784            inner: StorageType::Vec(buf),
785        }
786    }
787}
788
789impl From<&'static str> for BytePage {
790    fn from(buf: &'static str) -> Self {
791        Bytes::from_static(buf.as_bytes()).into()
792    }
793}
794
795impl From<&'static [u8]> for BytePage {
796    fn from(buf: &'static [u8]) -> Self {
797        Bytes::from_static(buf).into()
798    }
799}
800
801impl<const N: usize> From<&'static [u8; N]> for BytePage {
802    fn from(buf: &'static [u8; N]) -> Self {
803        Bytes::from_static(buf).into()
804    }
805}
806
807impl From<BytePage> for Bytes {
808    fn from(page: BytePage) -> Self {
809        match page.inner {
810            StorageType::Bytes(b) => b,
811            StorageType::Storage(storage) => BytesMut { storage }.freeze(),
812            StorageType::Vec(v) => Bytes::copy_from_slice(&v),
813        }
814    }
815}
816
817impl From<BytePage> for BytesMut {
818    fn from(page: BytePage) -> Self {
819        match page.inner {
820            StorageType::Bytes(b) => b.into(),
821            // clones of the page may still read the data
822            StorageType::Storage(storage) => {
823                if storage.is_unique() {
824                    BytesMut { storage }
825                } else {
826                    BytesMut::copy_from_slice(storage.as_ref())
827                }
828            }
829            StorageType::Vec(v) => BytesMut::copy_from_slice(&v),
830        }
831    }
832}
833
834impl PartialEq for BytePage {
835    fn eq(&self, other: &BytePage) -> bool {
836        self.as_ref() == other.as_ref()
837    }
838}
839
840impl_partial_eq!(BytePage);
841
842impl_read!(BytePage);
843
844impl ops::Deref for BytePage {
845    type Target = [u8];
846
847    #[inline]
848    fn deref(&self) -> &[u8] {
849        self.as_ref()
850    }
851}
852
853impl fmt::Debug for BytePage {
854    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
855        fmt::Debug::fmt(&crate::debug::BsDebug(self.as_ref()), fmt)
856    }
857}
858
859#[cfg(test)]
860mod tests {
861    use rand::Rng;
862
863    use super::*;
864
865    #[test]
866    #[allow(clippy::op_ref, clippy::cmp_owned)]
867    fn page_eq_and_read() {
868        use std::io::Read;
869
870        let mut page = BytePage::from(Vec::from(&b"hello"[..]));
871        assert_eq!(page, b"hello"[..]);
872        assert_eq!(page, *b"hello");
873        assert_eq!(page, b"hello");
874        assert_eq!(page, &b"hello"[..]);
875        assert_eq!(page, "hello");
876        assert_eq!(page, *"hello");
877        assert_eq!(page, b"hello".to_vec());
878        assert_eq!(page, String::from("hello"));
879        assert_eq!(b"hello"[..], page);
880        assert_eq!(*b"hello", page);
881        assert_eq!(b"hello", page);
882        assert_eq!(&b"hello"[..], page);
883        assert_eq!("hello", page);
884        assert_eq!(*"hello", page);
885        assert_eq!(b"hello".to_vec(), page);
886        assert_eq!(String::from("hello"), page);
887        assert_ne!(page, "hell");
888
889        let mut buf = [0u8; 3];
890        assert_eq!(page.read(&mut buf).unwrap(), 3);
891        assert_eq!(&buf, b"hel");
892        assert_eq!(page, "lo");
893        assert_eq!(page.read(&mut buf).unwrap(), 2);
894        assert_eq!(page.read(&mut buf).unwrap(), 0);
895        assert!(page.is_empty());
896    }
897
898    #[test]
899    fn page_info() {
900        use crate::info::PageKind;
901
902        let p = BytePage::from(Bytes::copy_from_slice(&[1; 64]));
903        assert_eq!(p.info(), PageKind::Bytes);
904        let p = BytePage::from(BytesMut::copy_from_slice(&[1; 64][..]));
905        assert_eq!(p.info(), PageKind::Storage);
906        let p = BytePage::from(vec![1; 64]);
907        assert_eq!(p.info(), PageKind::Vec);
908        assert!(!p.is_inline());
909
910        assert!(BytePage::from(Bytes::copy_from_slice(&[1; 4])).is_inline());
911        assert!(!BytePage::from(Bytes::copy_from_slice(&[1; 64])).is_inline());
912        assert!(!BytePage::from(BytesMut::copy_from_slice(&[1; 4][..])).is_inline());
913    }
914
915    #[test]
916    fn append_copies_small_and_splits_for_large() {
917        let cap = BytePageSize::Size16.capacity();
918        let mut pages = BytePages::new(BytePageSize::Size16);
919        pages.extend_from_slice(b"head\r\n");
920
921        // small data is copied into the current page
922        pages.append(Bytes::copy_from_slice(&[1; APPEND_COPY_LIMIT]));
923        assert_eq!(pages.num_pages(), 1);
924        assert_eq!(pages.current_len(), 6 + APPEND_COPY_LIMIT);
925
926        // large data is not copied, the filled part is split off in front
927        let body = Bytes::copy_from_slice(&[2; APPEND_COPY_LIMIT + 1]);
928        let body_ptr = body.as_ptr();
929        pages.append(body.clone());
930        assert_eq!(pages.num_pages(), 2);
931        assert_eq!(pages.current_len(), 0);
932        assert_eq!(pages.spare(), cap - 6 - APPEND_COPY_LIMIT);
933
934        // later writes use the spare capacity of the current page
935        let page_ptr = pages.pages()[0].as_ref().as_ptr();
936        pages.extend_from_slice(b"\r\n");
937        assert_eq!(pages.num_pages(), 3);
938        let tail_ptr = pages.current.as_ref().unwrap().as_ref().as_ptr();
939        assert_eq!(tail_ptr, page_ptr.wrapping_add(6 + APPEND_COPY_LIMIT));
940
941        let mut expected = b"head\r\n".to_vec();
942        expected.extend_from_slice(&[1; APPEND_COPY_LIMIT]);
943        expected.extend_from_slice(&body);
944        expected.extend_from_slice(b"\r\n");
945        assert_eq!(pages.len(), expected.len());
946
947        let first = pages.take().unwrap();
948        let second = pages.take().unwrap();
949        assert_eq!(second.as_ref().as_ptr(), body_ptr);
950        let third = pages.take().unwrap();
951        assert!(pages.take().is_none());
952        let data = [first.as_ref(), second.as_ref(), third.as_ref()].concat();
953        assert_eq!(data, expected);
954
955        // draining the pages keeps an empty current page for later writes
956        pages.extend_from_slice(b"x");
957        pages.append(body);
958        assert_eq!(pages.take().unwrap().as_ref(), b"x");
959        assert_eq!(pages.take().unwrap().len(), APPEND_COPY_LIMIT + 1);
960        assert!(pages.take().is_none());
961        assert_eq!(pages.num_pages(), 0);
962        assert_eq!(pages.spare(), cap - 1);
963    }
964
965    #[test]
966    fn pages() {
967        let cap = BytePageSize::Size8.capacity();
968        unsafe {
969            // pages
970            let mut pages = BytePages::new(BytePageSize::Size8);
971            assert!(pages.is_empty());
972            assert_eq!(pages.len(), 0);
973            assert_eq!(pages.num_pages(), 0);
974            pages.extend_from_slice(b"b");
975            assert_eq!(pages.len(), 1);
976            assert_eq!(pages.num_pages(), 1);
977            pages.extend_from_slice("a".repeat(9 * 1024).as_bytes());
978            assert_eq!(pages.len(), 9217);
979            assert_eq!(pages.num_pages(), 2);
980            assert!(!pages.is_empty());
981
982            let mut pgs = BytePages::new(BytePageSize::Size8);
983            pgs.put_i8(b'a' as i8);
984            let p = pgs.take().unwrap();
985            assert_eq!(p.len(), 1);
986            assert_eq!(p.as_ref(), b"a");
987
988            pgs.extend_from_slice("a".repeat(cap - 1).as_bytes());
989            assert_eq!(pgs.num_pages(), 1);
990            pgs.put_u8(b'a');
991            assert_eq!(pgs.num_pages(), 1);
992            assert!(pgs.current.is_none());
993
994            pgs.put_u8(b'a');
995            assert_eq!(pgs.num_pages(), 2);
996
997            pgs.append(Bytes::copy_from_slice("a".repeat(cap).as_bytes()));
998            assert_eq!(pgs.num_pages(), 3);
999            assert_eq!(pgs.current_len(), 0);
1000            assert_eq!(pgs.spare(), cap - 1);
1001
1002            // page
1003            let p = pages.take().unwrap();
1004            assert_eq!(p.len(), cap);
1005            let p = pages.take().unwrap();
1006            assert_eq!(p.len(), 9217 - cap);
1007            assert!(!p.is_empty());
1008            assert_eq!(p.as_ref().as_ptr(), p.as_ptr());
1009            assert_eq!(p.as_ref(), "a".repeat(9217 - cap).as_bytes());
1010            assert!(pages.take().is_none());
1011
1012            let p = BytePage::from(Bytes::copy_from_slice(b"123"));
1013            assert_eq!(p.len(), 3);
1014            assert!(!p.is_empty());
1015            assert_eq!(p.as_ref(), b"123");
1016            assert_eq!(p.as_ref().as_ptr(), p.as_ptr());
1017
1018            let p = BytePage::from(&b"123"[..]);
1019            assert_eq!(p.len(), 3);
1020            assert!(!p.is_empty());
1021            assert_eq!(p.as_ref(), b"123");
1022            assert_eq!(p.as_ref().as_ptr(), p.as_ptr());
1023
1024            let p = BytePage::from(b"123");
1025            assert_eq!(p.len(), 3);
1026            assert!(!p.is_empty());
1027            assert_eq!(p.as_ref(), b"123");
1028            assert_eq!(p.as_ref().as_ptr(), p.as_ptr());
1029
1030            let p = BytePage::from("123");
1031            assert_eq!(p.len(), 3);
1032            assert!(!p.is_empty());
1033            assert_eq!(p.as_ref(), b"123");
1034            assert_eq!(p.as_ref().as_ptr(), p.as_ptr());
1035            assert_eq!(p.freeze(), b"123");
1036
1037            let p = BytePage::from(vec![b'1', b'2', b'3']);
1038            assert_eq!(p.len(), 3);
1039            assert!(!p.is_empty());
1040            assert_eq!(p.as_ref(), b"123");
1041            assert_eq!(p.as_ref().as_ptr(), p.as_ptr());
1042            assert_eq!(p.freeze(), b"123");
1043
1044            let mut p = BytePage::from(vec![b'1', b'2', b'3']);
1045            p.advance_to(1);
1046            assert_eq!(p.len(), 2);
1047            assert!(!p.is_empty());
1048            assert_eq!(p.as_ref(), b"23");
1049
1050            // debug
1051            let mut pages = BytePages::new(BytePageSize::Size8);
1052            pages.extend_from_slice(b"b");
1053            assert_eq!(format!("{pages:?}"), "BytePages(b\"b\")");
1054            let p = pages.take().unwrap();
1055            assert_eq!(p.as_ref(), b"b");
1056
1057            let mut pages = BytePages::new(BytePageSize::Size8);
1058            pages.extend_from_slice(b"a");
1059            pages.append(Bytes::copy_from_slice(b"123"));
1060            pages.push_back(p);
1061            assert_eq!(format!("{pages:?}"), "BytePages(b\"b\", b\"a123\")");
1062
1063            assert_eq!(pages.len(), 5);
1064            pages.clear();
1065            assert_eq!(pages.len(), 0);
1066        }
1067    }
1068
1069    /// Checks the tracked length against the pages.
1070    fn assert_len(pages: &BytePages) {
1071        let len = pages.pages().iter().map(BytePage::len).sum::<usize>() + pages.current_len();
1072        assert_eq!(pages.len(), len);
1073        assert_eq!(pages.is_empty(), len == 0);
1074    }
1075
1076    #[test]
1077    fn pages_len_tracking() {
1078        let mut rng = rand::rng();
1079        let mut pages = BytePages::new(BytePageSize::Size8);
1080        let mut other = BytePages::new(BytePageSize::Size8);
1081        let mut expected = 0;
1082
1083        let (iters, max) = if cfg!(miri) { (100, 256) } else { (2000, 12 * 1024) };
1084        for _ in 0..iters {
1085            let n = rng.random_range(0..max);
1086            match rng.random_range(0..9) {
1087                0 => {
1088                    pages.extend_from_slice(&vec![1; n]);
1089                    expected += n;
1090                }
1091                1 => {
1092                    pages.append(Bytes::copy_from_slice(&vec![2; n]));
1093                    expected += n;
1094                }
1095                2 => {
1096                    if pages.prepend(BytesMut::copy_from_slice(vec![3; n])) {
1097                        expected += n;
1098                    }
1099                }
1100                3 => {
1101                    if let Some(p) = pages.take() {
1102                        expected -= p.len();
1103                    }
1104                }
1105                4 => {
1106                    let at = cmp::min(n, expected);
1107                    let split = pages.split_to(at);
1108                    assert_len(&split);
1109                    assert_eq!(split.len(), at);
1110                    expected -= at;
1111                }
1112                5 => {
1113                    let at = cmp::min(n, expected);
1114                    pages.split_into(at, &mut other);
1115                    expected -= at;
1116                }
1117                6 => {
1118                    other.move_to(&mut pages);
1119                    assert_len(&other);
1120                    assert!(other.is_empty());
1121                    expected = pages.len();
1122                }
1123                7 => {
1124                    let mut copy = BytePages::new(BytePageSize::Size8);
1125                    pages.copy_to(&mut copy);
1126                    assert_len(&copy);
1127                    assert_eq!(copy.len(), expected);
1128                }
1129                _ => {
1130                    pages.put_u8(4);
1131                    expected += 1;
1132                }
1133            }
1134            assert_len(&pages);
1135            assert_len(&other);
1136            assert_eq!(pages.len(), expected);
1137        }
1138
1139        let len = pages.len();
1140        assert_eq!(pages.freeze().len(), len);
1141        assert_len(&pages);
1142        assert!(pages.is_empty());
1143        pages.extend_from_slice(b"123");
1144        pages.clear();
1145        assert_len(&pages);
1146        assert!(pages.is_empty());
1147    }
1148
1149    #[test]
1150    fn pages_len_after_cache_reuse() {
1151        let mut pages = BytePages::new(BytePageSize::Size8);
1152        pages.append(Bytes::copy_from_slice(&[1; 64]));
1153        pages.append(Bytes::copy_from_slice(&[1; 64]));
1154        drop(pages);
1155
1156        let pages = BytePages::new(BytePageSize::Size8);
1157        assert_len(&pages);
1158        assert!(pages.is_empty());
1159    }
1160
1161    #[test]
1162    fn pages_copy_to_shares_current() {
1163        let mut pages = BytePages::new(BytePageSize::Size8);
1164        pages.put_slice(&[1; 64]);
1165        let ptr = unsafe { pages.current.as_ref().unwrap().as_ptr() };
1166
1167        let mut copy = BytePages::new(BytePageSize::Size8);
1168        pages.copy_to(&mut copy);
1169        let page = copy.take().unwrap();
1170        assert_eq!(unsafe { page.as_ptr() }, ptr.cast_const());
1171        assert_eq!(page.as_ref(), &[1; 64][..]);
1172
1173        // the source keeps writing to the page it shares with the copy
1174        pages.put_slice(&[2; 64]);
1175        assert_eq!(unsafe { pages.current.as_ref().unwrap().as_ptr() }, ptr);
1176        assert_eq!(page.as_ref(), &[1; 64][..]);
1177        assert_eq!(pages.len(), 128);
1178
1179        drop(pages);
1180        assert_eq!(page.as_ref(), &[1; 64][..]);
1181    }
1182
1183    #[test]
1184    fn pages_split_keeps_current_writable() {
1185        let mut pages = BytePages::new(BytePageSize::Size8);
1186        pages.put_slice(&[1; 64]);
1187        let ptr = unsafe { pages.current.as_ref().unwrap().as_ptr() };
1188        let remaining = pages.spare();
1189
1190        let mut head = pages.split_to(40);
1191        assert_eq!(head.len(), 40);
1192        assert_eq!(pages.len(), 24);
1193        assert_eq!(pages.num_pages(), 1);
1194        assert_eq!(pages.spare(), remaining);
1195
1196        // new data goes into the same page
1197        pages.put_slice(&[2; 16]);
1198        assert_eq!(pages.num_pages(), 1);
1199        assert_eq!(
1200            unsafe { pages.current.as_ref().unwrap().as_ptr() },
1201            ptr.wrapping_add(40)
1202        );
1203        let mut expected = vec![1; 24];
1204        expected.extend_from_slice(&[2; 16]);
1205        assert_eq!(&pages.freeze()[..], &expected[..]);
1206        assert_eq!(&head.freeze()[..], &[1; 40][..]);
1207
1208        // an inline head
1209        let mut pages = BytePages::new(BytePageSize::Size8);
1210        pages.put_slice(&[1; 64]);
1211        let mut to = BytePages::new(BytePageSize::Size8);
1212        pages.split_into(2, &mut to);
1213        assert_eq!(&to.freeze()[..], &[1; 2][..]);
1214        assert_eq!(pages.spare(), remaining);
1215        assert_eq!(pages.len(), 62);
1216
1217        // the whole current page moves with its spare capacity
1218        let mut pages = BytePages::new(BytePageSize::Size8);
1219        pages.put_slice(&[1; 64]);
1220        let ptr = unsafe { pages.current.as_ref().unwrap().as_ptr() };
1221        let to = pages.split_to(64);
1222        assert!(pages.is_empty());
1223        assert_eq!(pages.num_pages(), 0);
1224        assert_eq!(to.spare(), remaining);
1225        assert_eq!(unsafe { to.current.as_ref().unwrap().as_ptr() }, ptr);
1226
1227        // an empty current page stays in place
1228        let mut pages = BytePages::new(BytePageSize::Size8);
1229        let _ = pages.chunk_mut();
1230        let head = pages.split_to(10);
1231        assert!(head.is_empty());
1232        assert_eq!(pages.spare(), remaining + 64);
1233    }
1234
1235    #[test]
1236    fn pages_buf_mut_grows() {
1237        let mut pages = BytePages::new(BytePageSize::Size8);
1238        assert!(pages.has_remaining_mut());
1239        assert_eq!(pages.remaining_mut(), usize::MAX);
1240        unsafe { pages.advance_mut(0) };
1241
1242        // filling the current page through `chunk_mut` starts a new one
1243        let n = pages.chunk_mut().len();
1244        unsafe {
1245            std::ptr::write_bytes(pages.chunk_mut().as_mut_ptr(), 1, n);
1246            pages.advance_mut(n);
1247        }
1248        assert!(pages.chunk_mut().len() > 0);
1249        // the new current page holds no data yet
1250        assert_eq!(pages.num_pages(), 1);
1251        unsafe {
1252            *pages.chunk_mut().as_mut_ptr() = 2;
1253            pages.advance_mut(1);
1254        }
1255        assert_eq!(pages.len(), n + 1);
1256        assert_eq!(pages.remaining_mut(), usize::MAX - n - 1);
1257
1258        // `put` and `io::Write` are not limited to the current page
1259        let mut pages = BytePages::new(BytePageSize::Size8);
1260        pages.put(&[3u8; 200][..]);
1261        pages.put(Bytes::from_static(b"abcd"));
1262        io::Write::write_all(&mut pages, &[4; 100]).unwrap();
1263        let mut expected = vec![3; 200];
1264        expected.extend_from_slice(b"abcd");
1265        expected.extend_from_slice(&[4; 100]);
1266        assert_eq!(&pages.freeze()[..], &expected[..]);
1267    }
1268
1269    #[test]
1270    #[should_panic(expected = "cannot advance past the current page")]
1271    fn pages_advance_past_current_page() {
1272        let mut pages = BytePages::new(BytePageSize::Size8);
1273        let n = pages.chunk_mut().len();
1274        unsafe { pages.advance_mut(n + 1) };
1275    }
1276
1277    #[test]
1278    #[should_panic(expected = "Page size cannot be Unset")]
1279    fn pages_new_unset() {
1280        let _ = BytePages::new(BytePageSize::Unset);
1281    }
1282
1283    #[test]
1284    #[should_panic(expected = "Page size cannot be Unset")]
1285    fn pages_set_page_size_unset() {
1286        let mut pages = BytePages::new(BytePageSize::Size8);
1287        pages.set_page_size(BytePageSize::Unset);
1288    }
1289
1290    #[test]
1291    fn cached_pages_list_is_shrunk() {
1292        let mut pages = BytePages::new(BytePageSize::Size4);
1293        for _ in 0..200 {
1294            pages.append(Bytes::from_static(b"page"));
1295        }
1296        assert!(pages.pages().capacity() >= 200);
1297        drop(pages);
1298
1299        let pages = BytePages::new(BytePageSize::Size4);
1300        assert!(
1301            pages.pages().capacity() < 200,
1302            "{}",
1303            pages.pages().capacity()
1304        );
1305    }
1306
1307    #[test]
1308    fn pages_into_bytes_single_page() {
1309        // the current page
1310        let mut pages = BytePages::new(BytePageSize::Size8);
1311        pages.put_slice(&[1; 64]);
1312        let ptr = unsafe { pages.current.as_ref().unwrap().as_ptr() };
1313        let mut buf = BytesMut::from(pages);
1314        assert_eq!(buf.as_ptr(), ptr.cast_const());
1315        assert_eq!(&buf[..], &[1; 64][..]);
1316        // the rest of the page is spare capacity
1317        assert_eq!(buf.capacity(), BytePageSize::Size8.capacity());
1318        buf.extend_from_slice(&[2; 64]);
1319        assert_eq!(buf.as_ptr(), ptr.cast_const());
1320
1321        let mut pages = BytePages::new(BytePageSize::Size8);
1322        pages.put_slice(&[1; 64]);
1323        let ptr = unsafe { pages.current.as_ref().unwrap().as_ptr() };
1324        let b = Bytes::from(pages);
1325        assert_eq!(b.as_ptr(), ptr.cast_const());
1326        assert_eq!(&b[..], &[1; 64][..]);
1327
1328        // a `Bytes` page
1329        let src = Bytes::copy_from_slice(&[3; 64]);
1330        let ptr = src.as_ptr();
1331        let mut pages = BytePages::new(BytePageSize::Size8);
1332        pages.prepend(src);
1333        assert_eq!(BytesMut::from(pages).as_ptr(), ptr);
1334
1335        // a shared page is copied
1336        let src = Bytes::copy_from_slice(&[3; 64]);
1337        let mut pages = BytePages::new(BytePageSize::Size8);
1338        pages.prepend(&src);
1339        let mut buf = BytesMut::from(pages);
1340        assert_ne!(buf.as_ptr(), src.as_ptr());
1341        buf[0] = 4;
1342        assert_eq!(&src[..], &[3; 64][..]);
1343    }
1344
1345    #[test]
1346    fn pages_into_bytes_multiple_pages() {
1347        let mut pages = BytePages::new(BytePageSize::Size8);
1348        pages.prepend(Bytes::copy_from_slice(&[1; 64]));
1349        pages.put_slice(&[2; 64]);
1350        assert_eq!(pages.num_pages(), 2);
1351        let buf = BytesMut::from(pages);
1352        assert_eq!(&buf[..64], &[1; 64][..]);
1353        assert_eq!(&buf[64..], &[2; 64][..]);
1354
1355        let mut pages = BytePages::new(BytePageSize::Size8);
1356        pages.prepend(Bytes::copy_from_slice(&[1; 64]));
1357        pages.put_slice(&[2; 64]);
1358        let b = Bytes::from(pages);
1359        assert_eq!(&b[..64], &[1; 64][..]);
1360        assert_eq!(&b[64..], &[2; 64][..]);
1361
1362        assert!(BytesMut::from(BytePages::new(BytePageSize::Size8)).is_empty());
1363        assert!(Bytes::from(BytePages::new(BytePageSize::Size8)).is_empty());
1364    }
1365
1366    #[test]
1367    fn pages_copy_to() {
1368        let mut pages = BytePages::default();
1369        let mut pages2 = BytePages::default();
1370        pages.put_slice(b"456");
1371        pages.prepend(BytePage::from(Bytes::copy_from_slice(b"123")));
1372        pages.copy_to(&mut pages2);
1373        let p = pages.freeze();
1374        assert_eq!(p, b"123456");
1375        let p2 = pages2.freeze();
1376        assert_eq!(p2, b"123456");
1377
1378        let mut pages = BytePages::default();
1379        let mut pages2 = BytePages::default();
1380        pages.put_slice(b"456");
1381        pages.prepend(BytePage::from(Bytes::copy_from_slice(b"123")));
1382        pages.copy_to(&mut pages2);
1383        pages.put_u8(b'7');
1384        let p = pages.freeze();
1385        assert_eq!(p, b"1234567");
1386        let p2 = pages2.freeze();
1387        assert_eq!(p2, b"123456");
1388
1389        let mut pages = BytePages::default();
1390        pages.put_slice(b"456");
1391        pages.prepend(BytePage::from(Bytes::copy_from_slice(b"123")));
1392        let mut pages2 = pages.clone();
1393        pages.put_u8(b'7');
1394        let p = pages.freeze();
1395        assert_eq!(p, b"1234567");
1396        let p2 = pages2.freeze();
1397        assert_eq!(p2, b"123456");
1398    }
1399
1400    #[test]
1401    fn pages_methods() {
1402        // .split_to()
1403        let mut pages = BytePages::default();
1404        pages.put_slice(b"456");
1405        pages.prepend(BytePage::from(&Bytes::copy_from_slice(b"123")));
1406        let mut pages2 = pages.split_to(1);
1407        let p = pages.freeze();
1408        assert_eq!(p, b"23456");
1409        let p2 = pages2.freeze();
1410        assert_eq!(p2, b"1");
1411
1412        let mut pages = BytePages::default();
1413        pages.put_slice(b"456");
1414        pages.prepend(BytePage::from(Bytes::copy_from_slice(b"123")));
1415        let mut pages2 = pages.split_to(4);
1416        let p = pages.freeze();
1417        assert_eq!(p, b"56");
1418        let p2 = pages2.freeze();
1419        assert_eq!(p2, b"1234");
1420
1421        // .split_into()
1422        let mut pages = BytePages::default();
1423        pages.put_slice(b"456");
1424        pages.prepend(BytePage::from(crate::ByteString::from_static("123")));
1425        let mut pages2 = BytePages::default();
1426        pages.split_into(1, &mut pages2);
1427        let p = pages.freeze();
1428        assert_eq!(p, b"23456");
1429        let p2 = pages2.freeze();
1430        assert_eq!(p2, b"1");
1431
1432        // .with_bytes_mut()
1433        let mut pages = BytePages::default();
1434        #[allow(deprecated)]
1435        pages.with_bytes_mut(|buf| buf.extend_from_slice(b"123"));
1436        assert_eq!(pages.len(), 3);
1437        let p = pages.freeze();
1438        assert_eq!(p, b"123");
1439
1440        let data = rand::rng()
1441            .sample_iter(&rand::distr::Alphanumeric)
1442            .take(65_536)
1443            .map(char::from)
1444            .collect::<String>();
1445
1446        let mut pages = BytePages::default();
1447        #[allow(deprecated)]
1448        pages.with_bytes_mut(|buf| buf.extend_from_slice(data.as_bytes()));
1449        assert_eq!(pages.len(), 65_536);
1450        let p = pages.freeze();
1451        assert_eq!(p, data.as_bytes());
1452
1453        // into bytes
1454        let page = BytePage::from(Bytes::copy_from_slice(b"123"));
1455        assert_eq!(page, b"123");
1456        assert_eq!(<BytePage as Borrow<[u8]>>::borrow(&page), b"123");
1457        let b = Bytes::from(page);
1458        assert_eq!(b, b"123");
1459    }
1460
1461    #[test]
1462    fn page_clone() {
1463        // Bytes storage
1464        let p = BytePage::from(Bytes::copy_from_slice(b"123"));
1465        let p2 = p.clone();
1466        assert_eq!(p, p2);
1467
1468        // StorageVec
1469        let mut p = BytePage::from(BytesMut::copy_from_slice(b"123"));
1470        if let StorageType::Storage(ref mut st) = p.inner {
1471            assert!(st.is_unique());
1472        } else {
1473            panic!()
1474        }
1475        let p2 = p.clone();
1476        assert_eq!(p, p2);
1477        // short data is copied into an inline view
1478        assert!(matches!(p2.inner, StorageType::Bytes(_)));
1479        if let StorageType::Storage(st) = p.inner {
1480            assert!(st.is_unique());
1481        } else {
1482            panic!()
1483        }
1484
1485        let mut p = BytePage::from(BytesMut::copy_from_slice([b'1'; 64]));
1486        let p2 = p.clone();
1487        assert_eq!(p, p2);
1488        assert!(matches!(p2.inner, StorageType::Bytes(_)));
1489        if let StorageType::Storage(ref mut st) = p.inner {
1490            assert!(!st.is_unique());
1491        } else {
1492            panic!()
1493        }
1494        drop(p2);
1495        if let StorageType::Storage(st) = p.inner {
1496            assert!(st.is_unique());
1497        } else {
1498            panic!()
1499        }
1500
1501        // Vec<u8> storage
1502        let p = BytePage::from(vec![b'1', b'2', b'3']);
1503        let p2 = p.clone();
1504        assert_eq!(p, p2);
1505        if let StorageType::Bytes(_) = p2.inner {
1506        } else {
1507            panic!()
1508        }
1509    }
1510
1511    #[test]
1512    fn page_split_to() {
1513        // Bytes storage
1514        let mut p = BytePage::from(Bytes::copy_from_slice(b"123"));
1515        let p2 = p.split_to(1);
1516        assert_eq!(p, b"23");
1517        assert_eq!(p2, b"1");
1518
1519        // StorageVec
1520        let mut p = BytePage::from(BytesMut::copy_from_slice(b"123"));
1521        let p2 = p.split_to(1);
1522        assert_eq!(p, b"23");
1523        assert_eq!(p2, b"1");
1524
1525        // Vec<u8> storage
1526        let mut p = BytePage::from(vec![b'1', b'2', b'3']);
1527        let p2 = p.split_to(1);
1528        assert_eq!(p, b"23");
1529        assert_eq!(p2, b"1");
1530    }
1531
1532    #[test]
1533    fn page_read() {
1534        use std::io::Read;
1535
1536        let mut page = BytePage::from(Bytes::copy_from_slice(b"123"));
1537
1538        let mut buf = [0; 10];
1539        assert_eq!(page.read(&mut buf).unwrap(), 3);
1540        assert_eq!(page.len(), 0);
1541        assert_eq!(buf, [49, 50, 51, 0, 0, 0, 0, 0, 0, 0]);
1542    }
1543
1544    #[test]
1545    fn pages_misc() {
1546        let mut pages = BytePages::new(BytePageSize::Size4);
1547        pages.set_page_size(BytePageSize::Size8);
1548        assert_eq!(pages.page_size(), BytePageSize::Size8);
1549        assert!(!pages.prepend(Bytes::new()));
1550        assert!(pages.prepend(Bytes::from_static(b"a")));
1551
1552        io::Write::write_all(&mut pages, b"bc").unwrap();
1553        io::Write::flush(&mut pages).unwrap();
1554        assert_eq!(pages.freeze(), "abc");
1555
1556        let s = ByteString::from_static("str");
1557        pages.append(&s);
1558        assert_eq!(pages.freeze(), "str");
1559    }
1560
1561    #[test]
1562    fn pages_try_get_current_from() {
1563        let mut src = BytePages::new(BytePageSize::Size4);
1564        src.extend_from_slice(b"data");
1565
1566        // the target already holds data
1567        let mut dst = BytePages::new(BytePageSize::Size4);
1568        dst.append(Bytes::from_static(b"x"));
1569        dst.try_get_current_from(&mut src);
1570        assert_eq!(src.len(), 4);
1571        assert_eq!(dst.len(), 1);
1572
1573        let mut dst = BytePages::new(BytePageSize::Size4);
1574        dst.try_get_current_from(&mut src);
1575        assert_eq!(src.len(), 0);
1576        assert_eq!(dst.len(), 4);
1577        dst.extend_from_slice(b"!");
1578        assert_eq!(dst.freeze(), "data!");
1579    }
1580
1581    #[test]
1582    fn pages_drop_cache() {
1583        CACHE.with(|c| c.set(Some(Box::default())));
1584        let cached = || {
1585            CACHE.with(|c| {
1586                let cache = c.take().unwrap();
1587                let len = cache.len();
1588                c.set(Some(cache));
1589                len
1590            })
1591        };
1592
1593        // the cache is full
1594        let pages: Vec<_> = (0..=CACHE_SIZE)
1595            .map(|_| BytePages::new(BytePageSize::Size4))
1596            .collect();
1597        drop(pages);
1598        assert_eq!(cached(), CACHE_SIZE);
1599
1600        // the cache is in use
1601        CACHE.with(|c| c.set(Some(Box::default())));
1602        let pages = BytePages::new(BytePageSize::Size4);
1603        let cache = CACHE.with(Cell::take);
1604        drop(pages);
1605        CACHE.with(|c| c.set(cache));
1606        assert_eq!(cached(), 0);
1607    }
1608
1609    #[test]
1610    fn page_conversions() {
1611        let page = BytePage::from(BytesMut::copy_from_slice([1; 64]));
1612        assert_eq!(page.info(), crate::info::PageKind::Storage);
1613        assert_eq!(Bytes::from(page), &[1; 64][..]);
1614
1615        let page = BytePage::from(vec![2; 64]);
1616        assert_eq!(page.info(), crate::info::PageKind::Vec);
1617        assert_eq!(Bytes::from(page), &[2; 64][..]);
1618
1619        let page = BytePage::from(vec![3; 64]);
1620        assert_eq!(BytesMut::from(page), &[3; 64][..]);
1621
1622        let s = ByteString::from_static("string");
1623        assert_eq!(BytePage::from(&s), "string");
1624    }
1625
1626    #[test]
1627    fn append_storage_page() {
1628        // a page with spare capacity becomes the current page
1629        let mut pages = BytePages::new(BytePageSize::Size4);
1630        let mut buf = BytesMut::with_capacity(64);
1631        buf.extend_from_slice(b"a");
1632        pages.append(buf);
1633        pages.extend_from_slice(b"b");
1634        assert_eq!(pages.num_pages(), 1);
1635        assert_eq!(pages.freeze(), "ab");
1636
1637        // full and shared pages are added to the page list
1638        let mut buf = BytesMut::with_capacity(64);
1639        let cap = buf.capacity();
1640        buf.resize(cap, b'x');
1641        pages.append(buf);
1642        let page = BytePage::from(BytesMut::copy_from_slice([b'y'; 64]));
1643        let shared = page.clone();
1644        pages.append(page);
1645        pages.extend_from_slice(b"z");
1646        assert_eq!(pages.num_pages(), 3);
1647        assert_eq!(pages.len(), cap + 65);
1648        assert_eq!(shared, &[b'y'; 64][..]);
1649    }
1650}
ntex_bytes/pages.rs

ntex_bytes/
pages.rs
