1use crate::error_code::ErrorCode;
2use abi_stable::StableAbi;
3use std::io::Cursor;
4use std::ptr::NonNull;
5use tarantool::error::BoxError;
6use tarantool::error::TarantoolErrorCode;
7use tarantool::ffi::tarantool as ffi;
8
9#[repr(C)]
15#[derive(StableAbi, Clone, Copy, Debug)]
16pub struct FfiSafeBytes {
17 pointer: NonNull<u8>,
18 len: usize,
19}
20
21impl FfiSafeBytes {
22 #[inline(always)]
23 pub fn len(self) -> usize {
24 self.len
25 }
26
27 #[inline(always)]
28 pub fn is_empty(self) -> bool {
29 self.len == 0
30 }
31
32 #[inline(always)]
36 pub unsafe fn from_raw_parts(pointer: NonNull<u8>, len: usize) -> Self {
37 Self { pointer, len }
38 }
39
40 #[inline(always)]
41 pub fn into_raw_parts(self) -> (*mut u8, usize) {
42 (self.pointer.as_ptr(), self.len)
43 }
44
45 pub unsafe fn as_bytes<'a>(self) -> &'a [u8] {
55 std::slice::from_raw_parts(self.pointer.as_ptr(), self.len)
56 }
57}
58
59impl Default for FfiSafeBytes {
60 #[inline(always)]
61 fn default() -> Self {
62 Self {
63 pointer: NonNull::dangling(),
64 len: 0,
65 }
66 }
67}
68
69impl<'a> From<&'a [u8]> for FfiSafeBytes {
70 #[inline(always)]
71 fn from(value: &'a [u8]) -> Self {
72 Self {
73 pointer: as_non_null_ptr(value),
74 len: value.len(),
75 }
76 }
77}
78
79impl<'a> From<&'a str> for FfiSafeBytes {
80 #[inline(always)]
81 fn from(value: &'a str) -> Self {
82 Self {
83 pointer: as_non_null_ptr(value.as_bytes()),
84 len: value.len(),
85 }
86 }
87}
88
89#[repr(C)]
98#[derive(StableAbi, Clone, Copy, Debug)]
99pub struct FfiSafeStr {
100 pointer: NonNull<u8>,
101 len: usize,
102}
103
104impl FfiSafeStr {
105 #[inline(always)]
106 pub fn len(self) -> usize {
107 self.len
108 }
109
110 #[inline(always)]
111 pub fn is_empty(self) -> bool {
112 self.len == 0
113 }
114
115 #[inline(always)]
119 pub unsafe fn from_raw_parts(pointer: NonNull<u8>, len: usize) -> Self {
120 Self { pointer, len }
121 }
122
123 pub unsafe fn from_utf8_unchecked(bytes: &[u8]) -> Self {
126 let pointer = as_non_null_ptr(bytes);
127 let len = bytes.len();
128 Self { pointer, len }
129 }
130
131 #[inline(always)]
132 pub fn into_raw_parts(self) -> (*mut u8, usize) {
133 (self.pointer.as_ptr(), self.len)
134 }
135
136 #[inline]
146 pub unsafe fn as_str<'a>(self) -> &'a str {
147 if cfg!(debug_assertions) {
148 std::str::from_utf8(self.as_bytes()).expect("should only be used with valid utf8")
149 } else {
150 std::str::from_utf8_unchecked(self.as_bytes())
151 }
152 }
153
154 #[inline(always)]
164 pub unsafe fn as_bytes<'a>(self) -> &'a [u8] {
165 std::slice::from_raw_parts(self.pointer.as_ptr(), self.len)
166 }
167}
168
169impl Default for FfiSafeStr {
170 #[inline(always)]
171 fn default() -> Self {
172 Self {
173 pointer: NonNull::dangling(),
174 len: 0,
175 }
176 }
177}
178
179impl<'a> From<&'a str> for FfiSafeStr {
180 #[inline(always)]
181 fn from(value: &'a str) -> Self {
182 Self {
183 pointer: as_non_null_ptr(value.as_bytes()),
184 len: value.len(),
185 }
186 }
187}
188
189#[derive(Debug)]
197pub struct RegionGuard {
198 save_point: usize,
199}
200
201impl RegionGuard {
202 #[inline(always)]
209 #[allow(clippy::new_without_default)]
210 pub fn new() -> Self {
211 let save_point = unsafe { ffi::box_region_used() };
214 Self { save_point }
215 }
216
217 #[inline(always)]
222 pub fn used_at_creation(&self) -> usize {
223 self.save_point
224 }
225}
226
227impl Drop for RegionGuard {
228 fn drop(&mut self) {
229 unsafe { ffi::box_region_truncate(self.save_point) }
232 }
233}
234
235#[inline]
242fn allocate_on_region(size: usize) -> Result<&'static mut [u8], BoxError> {
243 let pointer = unsafe { ffi::box_region_alloc(size).cast::<u8>() };
245 if pointer.is_null() {
246 return Err(BoxError::last());
247 }
248 let region_slice = unsafe { std::slice::from_raw_parts_mut(pointer, size) };
250 Ok(region_slice)
251}
252
253#[inline]
263pub fn copy_to_region(data: &[u8]) -> Result<&'static [u8], BoxError> {
264 let region_slice = allocate_on_region(data.len())?;
265 region_slice.copy_from_slice(data);
266 Ok(region_slice)
267}
268
269#[derive(Debug)]
283pub struct RegionBuffer {
284 guard: RegionGuard,
285
286 start: *mut u8,
287 count: usize,
288}
289
290impl RegionBuffer {
291 #[inline(always)]
295 #[allow(clippy::new_without_default)]
296 pub fn new() -> Self {
297 Self {
298 guard: RegionGuard::new(),
299 start: std::ptr::null_mut(),
300 count: 0,
301 }
302 }
303
304 #[track_caller]
312 pub fn push(&mut self, data: &[u8]) -> Result<(), BoxError> {
313 let added_count = data.len();
314 unsafe {
315 let pointer: *mut u8 = ffi::box_region_alloc(added_count) as _;
316
317 if pointer.is_null() {
318 #[rustfmt::skip]
319 return Err(BoxError::new(TarantoolErrorCode::MemoryIssue, format!("failed to allocate {added_count} bytes on the region allocator")));
320 }
321
322 memcpy(pointer, data.as_ptr(), added_count);
323 self.count += added_count;
324 if self.start.is_null() {
325 self.start = pointer;
326 }
327 }
328
329 Ok(())
330 }
331
332 #[inline]
348 pub fn into_raw_parts(self) -> (&'static [u8], usize) {
349 self.try_into_raw_parts().unwrap()
350 }
351
352 pub fn try_into_raw_parts(self) -> Result<(&'static [u8], usize), (BoxError, Self)> {
366 let res = unsafe { self.join() };
369 let slice = match res {
370 Ok(v) => v,
371 Err(e) => {
372 return Err((e, self));
373 }
374 };
375
376 let save_point = self.guard.used_at_creation();
377 std::mem::forget(self.guard);
378 Ok((slice, save_point))
379 }
380
381 #[inline]
385 unsafe fn join(&self) -> Result<&'static [u8], BoxError> {
386 use crate::internal::ffi;
387
388 if self.count == 0 {
389 return Ok(&[]);
390 }
391
392 if !ffi::has_box_region_join() {
393 return Err(BoxError::new(
394 TarantoolErrorCode::Unsupported,
395 "box_region_join is not supported in this version of picodata",
396 ));
397 }
398
399 let start = unsafe { ffi::box_region_join(self.count) };
402 if start.is_null() {
403 return Err(BoxError::last());
404 }
405 let slice = unsafe { std::slice::from_raw_parts(start.cast(), self.count) };
407 Ok(slice)
408 }
409
410 pub fn try_into_vec(self) -> Result<Vec<u8>, BoxError> {
413 let res = unsafe { self.join() };
416 let slice = match res {
417 Ok(v) => v,
418 Err(e) => {
419 return Err(e);
420 }
421 };
422
423 let res = Vec::from(slice);
424
425 drop(self);
427
428 Ok(res)
429 }
430}
431
432impl std::io::Write for RegionBuffer {
433 #[inline(always)]
434 fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
435 if let Err(e) = self.push(data) {
436 #[rustfmt::skip]
437 return Err(std::io::Error::new(std::io::ErrorKind::OutOfMemory, e.message()));
438 }
439
440 Ok(data.len())
441 }
442
443 #[inline(always)]
444 fn flush(&mut self) -> std::io::Result<()> {
445 Ok(())
446 }
447}
448
449#[inline(always)]
450unsafe fn memcpy(destination: *mut u8, source: *const u8, count: usize) {
451 let to = std::slice::from_raw_parts_mut(destination, count);
452 let from = std::slice::from_raw_parts(source, count);
453 to.copy_from_slice(from)
454}
455
456pub struct DisplayErrorLocation<'a>(pub &'a BoxError);
462
463impl std::fmt::Display for DisplayErrorLocation<'_> {
464 #[inline]
465 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
466 if let Some((file, line)) = self.0.file().zip(self.0.line()) {
467 write!(f, "{file}:{line}: ")?;
468 }
469 Ok(())
470 }
471}
472
473pub struct DisplayAsHexBytesLimitted<'a>(pub &'a [u8]);
479
480impl std::fmt::Display for DisplayAsHexBytesLimitted<'_> {
481 #[inline]
482 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
483 if self.0.len() > 512 {
484 f.write_str("<too-big-to-display>")
485 } else {
486 tarantool::util::DisplayAsHexBytes(self.0).fmt(f)
487 }
488 }
489}
490
491#[track_caller]
498#[inline]
499pub fn msgpack_decode_str(data: &[u8]) -> Result<&str, BoxError> {
500 let mut cursor = Cursor::new(data);
501 let length = rmp::decode::read_str_len(&mut cursor).map_err(invalid_msgpack)? as usize;
502
503 let res = str_from_cursor(length, &mut cursor)?;
504 let (_, tail) = cursor_split(&cursor);
505 if !tail.is_empty() {
506 return Err(invalid_msgpack(format!(
507 "unexpected data after msgpack value: {}",
508 DisplayAsHexBytesLimitted(tail)
509 )));
510 }
511
512 Ok(res)
513}
514
515#[track_caller]
518pub fn msgpack_read_str<'a>(cursor: &mut Cursor<&'a [u8]>) -> Result<&'a str, BoxError> {
519 let length = rmp::decode::read_str_len(cursor).map_err(invalid_msgpack)? as usize;
520
521 str_from_cursor(length, cursor)
522}
523
524#[track_caller]
534pub fn msgpack_read_rest_of_str<'a>(
535 marker: rmp::Marker,
536 cursor: &mut Cursor<&'a [u8]>,
537) -> Result<Option<&'a str>, BoxError> {
538 use rmp::decode::RmpRead as _;
539
540 let length = match marker {
541 rmp::Marker::FixStr(v) => v as usize,
542 rmp::Marker::Str8 => cursor.read_data_u8().map_err(invalid_msgpack)? as usize,
543 rmp::Marker::Str16 => cursor.read_data_u16().map_err(invalid_msgpack)? as usize,
544 rmp::Marker::Str32 => cursor.read_data_u32().map_err(invalid_msgpack)? as usize,
545 _ => return Ok(None),
546 };
547
548 str_from_cursor(length, cursor).map(Some)
549}
550
551#[inline]
552#[track_caller]
553fn str_from_cursor<'a>(length: usize, cursor: &mut Cursor<&'a [u8]>) -> Result<&'a str, BoxError> {
554 let start_index = cursor.position() as usize;
555 let data = *cursor.get_ref();
556 let remaining_length = data.len() - start_index;
557 if remaining_length < length {
558 return Err(invalid_msgpack(format!(
559 "expected a string of length {length}, got {remaining_length}"
560 )));
561 }
562
563 let end_index = start_index + length;
564 let res = std::str::from_utf8(&data[start_index..end_index]).map_err(invalid_msgpack)?;
565 cursor.set_position(end_index as _);
566 Ok(res)
567}
568
569#[track_caller]
571pub fn msgpack_decode_bin(data: &[u8]) -> Result<&[u8], BoxError> {
572 let mut cursor = Cursor::new(data);
573 let length = rmp::decode::read_bin_len(&mut cursor).map_err(invalid_msgpack)? as usize;
574
575 let res = bin_from_cursor(length, &mut cursor)?;
576 let (_, tail) = cursor_split(&cursor);
577 if !tail.is_empty() {
578 return Err(invalid_msgpack(format!(
579 "unexpected data after msgpack value: {}",
580 DisplayAsHexBytesLimitted(tail)
581 )));
582 }
583
584 Ok(res)
585}
586
587#[track_caller]
590pub fn msgpack_read_bin<'a>(cursor: &mut Cursor<&'a [u8]>) -> Result<&'a [u8], BoxError> {
591 let length = rmp::decode::read_bin_len(cursor).map_err(invalid_msgpack)? as usize;
592
593 bin_from_cursor(length, cursor)
594}
595
596#[track_caller]
605pub fn msgpack_read_rest_of_bin<'a>(
606 marker: rmp::Marker,
607 cursor: &mut Cursor<&'a [u8]>,
608) -> Result<Option<&'a [u8]>, BoxError> {
609 use rmp::decode::RmpRead as _;
610
611 let length = match marker {
612 rmp::Marker::Bin8 => cursor.read_data_u8().map_err(invalid_msgpack)? as usize,
613 rmp::Marker::Bin16 => cursor.read_data_u16().map_err(invalid_msgpack)? as usize,
614 rmp::Marker::Bin32 => cursor.read_data_u32().map_err(invalid_msgpack)? as usize,
615 _ => return Ok(None),
616 };
617
618 bin_from_cursor(length, cursor).map(Some)
619}
620
621#[inline]
622#[track_caller]
623fn bin_from_cursor<'a>(length: usize, cursor: &mut Cursor<&'a [u8]>) -> Result<&'a [u8], BoxError> {
624 let start_index = cursor.position() as usize;
625 let data = *cursor.get_ref();
626 let remaining_length = data.len() - start_index;
627 if remaining_length < length {
628 return Err(invalid_msgpack(format!(
629 "expected binary data of length {length}, got {remaining_length}"
630 )));
631 }
632
633 let end_index = start_index + length;
634 let res = &data[start_index..end_index];
635 cursor.set_position(end_index as _);
636 Ok(res)
637}
638
639fn cursor_split<'a>(cursor: &Cursor<&'a [u8]>) -> (&'a [u8], &'a [u8]) {
641 let slice = cursor.get_ref();
642 let pos = cursor.position().min(slice.len() as u64);
643 slice.split_at(pos as usize)
644}
645
646#[inline(always)]
647#[track_caller]
648fn invalid_msgpack(error: impl ToString) -> BoxError {
649 BoxError::new(TarantoolErrorCode::InvalidMsgpack, error.to_string())
650}
651
652#[inline(always)]
657fn as_non_null_ptr<T>(data: &[T]) -> NonNull<T> {
658 let pointer = data.as_ptr();
659 unsafe { NonNull::new_unchecked(pointer as *mut _) }
663}
664
665pub fn tarantool_error_to_box_error(e: tarantool::error::Error) -> BoxError {
667 match e {
668 tarantool::error::Error::Tarantool(e) => e,
669 other => BoxError::new(ErrorCode::Other, other.to_string()),
670 }
671}
672
673#[cfg(all(feature = "internal_test", not(test)))]
686mod test {
687 use super::*;
688
689 #[tarantool::test]
690 fn region_buffer() {
691 #[derive(serde::Serialize, Debug)]
692 struct S {
693 name: String,
694 x: f32,
695 y: f32,
696 array: Vec<(i32, i32, bool)>,
697 }
698
699 let s = S {
700 name: "foo".into(),
701 x: 4.2,
702 y: 6.9,
703 array: vec![(1, 2, true), (3, 4, false)],
704 };
705
706 let vec = rmp_serde::to_vec(&s).unwrap();
707 let mut buffer = RegionBuffer::new();
708 rmp_serde::encode::write(&mut buffer, &s).unwrap();
709 let data = buffer.try_into_vec().unwrap();
710 assert_eq!(vec, data);
711 }
712
713 #[tarantool::test(skip_if(
714 cfg(asan),
715 reason = "RegionBuffer built with ASAN uses different region_join() implementation that does not uphold the checked invariants"
716 ))]
717 fn region_buffer_tiny_allocation() {
718 let _guard = RegionGuard::new();
720
721 let mut buffer = RegionBuffer::new();
722 buffer.push(&[1, 2, 3]).unwrap();
723 let data = unsafe { buffer.join().unwrap() };
724 assert_eq!(data, &[1, 2, 3]);
725 assert_eq!(data.as_ptr(), buffer.start);
728
729 let data2 = unsafe { buffer.join().unwrap() };
732 assert_eq!(data2, data);
733 assert_eq!(data2.as_ptr(), buffer.start);
734
735 let (data3, _) = buffer.into_raw_parts();
737 assert_eq!(data3, data);
738 assert_eq!(data3.as_ptr(), data.as_ptr());
739 }
740
741 #[tarantool::test]
742 fn region_buffer_big_allocation() {
743 const N: usize = 4923;
744 const M: usize = 85;
745 const K: usize = 10;
746
747 const {
748 const SLAB_SIZE: usize = u16::MAX as usize + 1;
750 assert!(N * M * K > SLAB_SIZE);
751 };
752
753 let t0 = std::time::Instant::now();
754
755 let mut input = Vec::with_capacity(N);
757 for i in 0..N {
758 let mut row = Vec::with_capacity(M);
759 for j in 0..M {
760 let mut col = Vec::with_capacity(K);
761 for k in 0..K {
762 let v = (1 + i + j) * (1 + k);
764 col.push(v as u8);
765 }
766 row.push(col);
767 }
768 input.push(row);
769 }
770
771 tarantool::say_info!("generating data took {:?}", t0.elapsed());
772
773 let t0 = std::time::Instant::now();
774
775 let mut buffer = RegionBuffer::new();
776 rmp_serde::encode::write(&mut buffer, &input).unwrap();
779 let data = unsafe { buffer.join().unwrap() };
780
781 tarantool::say_info!(
782 "serializing data to region allocator took {:?}",
783 t0.elapsed()
784 );
785
786 assert_ne!(data.as_ptr(), buffer.start);
789
790 let t0 = std::time::Instant::now();
791
792 let control = rmp_serde::to_vec(&input).unwrap();
794
795 tarantool::say_info!("serializing data to rust allocator took {:?}", t0.elapsed());
796
797 assert_eq!(control, data);
798 }
799}