]> git.proxmox.com Git - pxar.git/blame - src/encoder/mod.rs
rustfmt
[pxar.git] / src / encoder / mod.rs
CommitLineData
70acf637
WB
1//! The `pxar` encoder state machine.
2//!
3//! This is the implementation used by both the synchronous and async pxar wrappers.
4
5use std::io;
6use std::mem::{forget, size_of, size_of_val, take};
7use std::os::unix::ffi::OsStrExt;
8use std::path::Path;
9use std::pin::Pin;
a08b84b7 10use std::sync::{Arc, Mutex};
70acf637
WB
11use std::task::{Context, Poll};
12
13use endian_trait::Endian;
14
749855a4 15use crate::binary_tree_array;
951620f1 16use crate::decoder::{self, SeqRead};
70acf637
WB
17use crate::format::{self, GoodbyeItem};
18use crate::poll_fn::poll_fn;
19use crate::Metadata;
20
54109840 21pub mod aio;
70acf637
WB
22pub mod sync;
23
24#[doc(inline)]
25pub use sync::Encoder;
26
d6748862
WB
27/// File reference used to create hard links.
28#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)]
29pub struct LinkOffset(u64);
30
31impl LinkOffset {
32 #[inline]
33 pub fn raw(self) -> u64 {
34 self.0
35 }
36}
37
70acf637
WB
38/// Sequential write interface used by the encoder's state machine.
39///
40/// This is our internal writer trait which is available for `std::io::Write` types in the
41/// synchronous wrapper and for both `tokio` and `future` `AsyncWrite` types in the asynchronous
42/// wrapper.
43pub trait SeqWrite {
44 fn poll_seq_write(
45 self: Pin<&mut Self>,
46 cx: &mut Context,
47 buf: &[u8],
48 ) -> Poll<io::Result<usize>>;
49
327e33f2 50 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>>;
70acf637
WB
51}
52
53/// Allow using trait objects for generics taking a `SeqWrite`.
d8402433
SR
54impl<S> SeqWrite for &mut S
55where
56 S: SeqWrite + ?Sized,
57{
70acf637
WB
58 fn poll_seq_write(
59 self: Pin<&mut Self>,
60 cx: &mut Context,
61 buf: &[u8],
62 ) -> Poll<io::Result<usize>> {
63 unsafe {
64 self.map_unchecked_mut(|this| &mut **this)
65 .poll_seq_write(cx, buf)
66 }
67 }
68
05356884
WB
69 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
70 unsafe { self.map_unchecked_mut(|this| &mut **this).poll_flush(cx) }
71 }
70acf637
WB
72}
73
fc9c8c9d 74/// awaitable verison of `poll_seq_write`.
ec4a53ed
WB
75async fn seq_write<T: SeqWrite + ?Sized>(
76 output: &mut T,
77 buf: &[u8],
78 position: &mut u64,
79) -> io::Result<usize> {
80 let put =
81 poll_fn(|cx| unsafe { Pin::new_unchecked(&mut *output).poll_seq_write(cx, buf) }).await?;
82 *position += put as u64;
83 Ok(put)
fc9c8c9d 84}
70acf637 85
d995b531 86/// awaitable version of 'poll_flush'.
eae6dc06 87async fn flush<T: SeqWrite + ?Sized>(output: &mut T) -> io::Result<()> {
d995b531
DC
88 poll_fn(|cx| unsafe { Pin::new_unchecked(&mut *output).poll_flush(cx) }).await
89}
90
fc9c8c9d 91/// Write the entire contents of a buffer, handling short writes.
ec4a53ed
WB
92async fn seq_write_all<T: SeqWrite + ?Sized>(
93 output: &mut T,
94 mut buf: &[u8],
95 position: &mut u64,
96) -> io::Result<()> {
fc9c8c9d 97 while !buf.is_empty() {
ec4a53ed 98 let got = seq_write(&mut *output, buf, &mut *position).await?;
fc9c8c9d 99 buf = &buf[got..];
70acf637 100 }
fc9c8c9d
WB
101 Ok(())
102}
70acf637 103
fc9c8c9d 104/// Write an endian-swappable struct.
ec4a53ed
WB
105async fn seq_write_struct<E: Endian, T>(
106 output: &mut T,
107 data: E,
108 position: &mut u64,
109) -> io::Result<()>
fc9c8c9d
WB
110where
111 T: SeqWrite + ?Sized,
112{
113 let data = data.to_le();
a08b84b7
WB
114 let buf =
115 unsafe { std::slice::from_raw_parts(&data as *const E as *const u8, size_of_val(&data)) };
8bff2f7c 116 seq_write_all(output, buf, position).await
fc9c8c9d 117}
70acf637 118
fc9c8c9d 119/// Write a pxar entry.
ec4a53ed
WB
120async fn seq_write_pxar_entry<T>(
121 output: &mut T,
122 htype: u64,
123 data: &[u8],
124 position: &mut u64,
125) -> io::Result<()>
fc9c8c9d
WB
126where
127 T: SeqWrite + ?Sized,
128{
ec0761f9
FG
129 let header = format::Header::with_content_size(htype, data.len() as u64);
130 header.check_header_size()?;
131
ec4a53ed
WB
132 seq_write_struct(&mut *output, header, &mut *position).await?;
133 seq_write_all(output, data, position).await
fc9c8c9d 134}
70acf637 135
fc9c8c9d
WB
136/// Write a pxar entry terminated by an additional zero which is not contained in the provided
137/// data buffer.
ec4a53ed
WB
138async fn seq_write_pxar_entry_zero<T>(
139 output: &mut T,
140 htype: u64,
141 data: &[u8],
142 position: &mut u64,
143) -> io::Result<()>
fc9c8c9d
WB
144where
145 T: SeqWrite + ?Sized,
146{
ec0761f9
FG
147 let header = format::Header::with_content_size(htype, 1 + data.len() as u64);
148 header.check_header_size()?;
149
ec4a53ed
WB
150 seq_write_struct(&mut *output, header, &mut *position).await?;
151 seq_write_all(&mut *output, data, &mut *position).await?;
152 seq_write_all(output, &[0u8], position).await
fc9c8c9d 153}
70acf637 154
fc9c8c9d 155/// Write a pxar entry consiting of an endian-swappable struct.
ec4a53ed
WB
156async fn seq_write_pxar_struct_entry<E, T>(
157 output: &mut T,
158 htype: u64,
159 data: E,
160 position: &mut u64,
161) -> io::Result<()>
fc9c8c9d
WB
162where
163 T: SeqWrite + ?Sized,
164 E: Endian,
165{
166 let data = data.to_le();
a08b84b7
WB
167 let buf =
168 unsafe { std::slice::from_raw_parts(&data as *const E as *const u8, size_of_val(&data)) };
8bff2f7c 169 seq_write_pxar_entry(output, htype, buf, position).await
70acf637
WB
170}
171
59e6f679
WB
172/// Error conditions caused by wrong usage of this crate.
173#[derive(Clone, Copy, Debug, Eq, PartialEq)]
174pub enum EncodeError {
175 /// The user dropped a `File` without without finishing writing all of its contents.
176 ///
177 /// This is required because the payload lengths is written out at the begining and decoding
178 /// requires there to follow the right amount of data.
179 IncompleteFile,
180
181 /// The user dropped a directory without finalizing it.
182 ///
183 /// Finalizing is required to build the goodbye table at the end of a directory.
184 IncompleteDirectory,
185}
186
70acf637
WB
187#[derive(Default)]
188struct EncoderState {
189 /// Goodbye items for this directory, excluding the tail.
190 items: Vec<GoodbyeItem>,
191
59e6f679
WB
192 /// User caused error conditions.
193 encode_error: Option<EncodeError>,
70acf637
WB
194
195 /// Offset of this directory's ENTRY.
196 entry_offset: u64,
197
198 /// Offset to this directory's first FILENAME.
199 files_offset: u64,
200
201 /// If this is a subdirectory, this points to the this directory's FILENAME.
202 file_offset: Option<u64>,
749855a4
WB
203
204 /// If this is a subdirectory, this contains this directory's hash for the goodbye item.
205 file_hash: u64,
ec4a53ed
WB
206
207 /// We need to keep track how much we have written to get offsets.
208 write_position: u64,
70acf637
WB
209}
210
59e6f679
WB
211impl EncoderState {
212 fn merge_error(&mut self, error: Option<EncodeError>) {
213 // one error is enough:
214 if self.encode_error.is_none() {
215 self.encode_error = error;
216 }
217 }
218
219 fn add_error(&mut self, error: EncodeError) {
220 self.merge_error(Some(error));
221 }
222}
223
d8402433
SR
224pub(crate) enum EncoderOutput<'a, T> {
225 Owned(T),
226 Borrowed(&'a mut T),
227}
228
b8d79e5f
WB
229impl<'a, T> EncoderOutput<'a, T> {
230 #[inline]
231 fn to_borrowed<'s>(&'s mut self) -> EncoderOutput<'s, T>
232 where
233 'a: 's,
234 {
235 EncoderOutput::Borrowed(self.as_mut())
236 }
237}
238
d8402433
SR
239impl<'a, T> std::convert::AsMut<T> for EncoderOutput<'a, T> {
240 fn as_mut(&mut self) -> &mut T {
241 match self {
b8d79e5f 242 EncoderOutput::Owned(o) => o,
d8402433
SR
243 EncoderOutput::Borrowed(b) => b,
244 }
245 }
246}
247
248impl<'a, T> std::convert::From<T> for EncoderOutput<'a, T> {
249 fn from(t: T) -> Self {
250 EncoderOutput::Owned(t)
251 }
252}
253
254impl<'a, T> std::convert::From<&'a mut T> for EncoderOutput<'a, T> {
255 fn from(t: &'a mut T) -> Self {
256 EncoderOutput::Borrowed(t)
257 }
258}
259
70acf637
WB
260/// The encoder state machine implementation for a directory.
261///
262/// We use `async fn` to implement the encoder state machine so that we can easily plug in both
263/// synchronous or `async` I/O objects in as output.
264pub(crate) struct EncoderImpl<'a, T: SeqWrite + 'a> {
d8402433 265 output: EncoderOutput<'a, T>,
70acf637
WB
266 state: EncoderState,
267 parent: Option<&'a mut EncoderState>,
268 finished: bool,
79e3f621
WB
269
270 /// Since only the "current" entry can be actively writing files, we share the file copy
271 /// buffer.
a08b84b7 272 file_copy_buffer: Arc<Mutex<Vec<u8>>>,
70acf637
WB
273}
274
275impl<'a, T: SeqWrite + 'a> Drop for EncoderImpl<'a, T> {
276 fn drop(&mut self) {
277 if let Some(ref mut parent) = self.parent {
278 // propagate errors:
59e6f679
WB
279 parent.merge_error(self.state.encode_error);
280 if !self.finished {
281 parent.add_error(EncodeError::IncompleteDirectory);
70acf637
WB
282 }
283 } else if !self.finished {
284 // FIXME: how do we deal with this?
2f56c76c 285 // eprintln!("Encoder dropped without finishing!");
70acf637
WB
286 }
287 }
288}
289
290impl<'a, T: SeqWrite + 'a> EncoderImpl<'a, T> {
737f75cf
WB
291 pub async fn new(
292 output: EncoderOutput<'a, T>,
293 metadata: &Metadata,
294 ) -> io::Result<EncoderImpl<'a, T>> {
70acf637
WB
295 if !metadata.is_dir() {
296 io_bail!("directory metadata must contain the directory mode flag");
297 }
298 let mut this = Self {
d8402433 299 output,
70acf637
WB
300 state: EncoderState::default(),
301 parent: None,
302 finished: false,
a08b84b7 303 file_copy_buffer: Arc::new(Mutex::new(crate::util::vec_new(1024 * 1024))),
70acf637
WB
304 };
305
306 this.encode_metadata(metadata).await?;
ec4a53ed 307 this.state.files_offset = this.position();
70acf637
WB
308
309 Ok(this)
310 }
311
312 fn check(&self) -> io::Result<()> {
59e6f679
WB
313 match self.state.encode_error {
314 Some(EncodeError::IncompleteFile) => io_bail!("incomplete file"),
315 Some(EncodeError::IncompleteDirectory) => io_bail!("directory not finalized"),
a7149b09 316 None => Ok(()),
70acf637
WB
317 }
318 }
319
320 pub async fn create_file<'b>(
321 &'b mut self,
322 metadata: &Metadata,
323 file_name: &Path,
324 file_size: u64,
d8402433 325 ) -> io::Result<FileImpl<'b, T>>
70acf637
WB
326 where
327 'a: 'b,
328 {
329 self.create_file_do(metadata, file_name.as_os_str().as_bytes(), file_size)
330 .await
331 }
332
333 async fn create_file_do<'b>(
334 &'b mut self,
335 metadata: &Metadata,
336 file_name: &[u8],
337 file_size: u64,
d8402433 338 ) -> io::Result<FileImpl<'b, T>>
70acf637
WB
339 where
340 'a: 'b,
341 {
342 self.check()?;
343
ec4a53ed 344 let file_offset = self.position();
2ab25a17 345 self.start_file_do(Some(metadata), file_name).await?;
70acf637 346
ec0761f9
FG
347 let header = format::Header::with_content_size(format::PXAR_PAYLOAD, file_size);
348 header.check_header_size()?;
349
737f75cf 350 seq_write_struct(self.output.as_mut(), header, &mut self.state.write_position).await?;
70acf637 351
ec4a53ed 352 let payload_data_offset = self.position();
70acf637
WB
353
354 let meta_size = payload_data_offset - file_offset;
355
356 Ok(FileImpl {
d8402433 357 output: self.output.as_mut(),
70acf637
WB
358 goodbye_item: GoodbyeItem {
359 hash: format::hash_filename(file_name),
360 offset: file_offset,
361 size: file_size + meta_size,
362 },
363 remaining_size: file_size,
364 parent: &mut self.state,
365 })
366 }
367
d6748862 368 /// Return a file offset usable with `add_hardlink`.
70acf637
WB
369 pub async fn add_file(
370 &mut self,
371 metadata: &Metadata,
372 file_name: &Path,
373 file_size: u64,
374 content: &mut dyn SeqRead,
d6748862 375 ) -> io::Result<LinkOffset> {
a08b84b7 376 let buf = Arc::clone(&self.file_copy_buffer);
70acf637 377 let mut file = self.create_file(metadata, file_name, file_size).await?;
a08b84b7 378 let mut buf = buf.lock().expect("failed to lock temporary buffer mutex");
70acf637 379 loop {
79e3f621 380 let got = decoder::seq_read(&mut *content, &mut buf[..]).await?;
70acf637
WB
381 if got == 0 {
382 break;
383 } else {
384 file.write_all(&buf[..got]).await?;
385 }
386 }
d6748862 387 Ok(file.file_offset())
70acf637
WB
388 }
389
d6748862 390 /// Return a file offset usable with `add_hardlink`.
2fb73e7b
WB
391 pub async fn add_symlink(
392 &mut self,
393 metadata: &Metadata,
394 file_name: &Path,
395 target: &Path,
501ea220 396 ) -> io::Result<()> {
16f0464d
WB
397 let target = target.as_os_str().as_bytes();
398 let mut data = Vec::with_capacity(target.len() + 1);
399 data.extend(target);
400 data.push(0);
b0487d4f
WB
401 let _ofs: LinkOffset = self
402 .add_file_entry(
403 Some(metadata),
404 file_name,
16f0464d 405 Some((format::PXAR_SYMLINK, &data)),
b0487d4f
WB
406 )
407 .await?;
501ea220 408 Ok(())
0abc4121
WB
409 }
410
d6748862 411 /// Return a file offset usable with `add_hardlink`.
b0752929 412 pub async fn add_hardlink(
0abc4121 413 &mut self,
0abc4121
WB
414 file_name: &Path,
415 target: &Path,
63f2296f 416 target_offset: LinkOffset,
3fe26522 417 ) -> io::Result<()> {
ec4a53ed 418 let current_offset = self.position();
63f2296f
WB
419 if current_offset <= target_offset.0 {
420 io_bail!("invalid hardlink offset, can only point to prior files");
421 }
422
f3e50baa
WB
423 let offset_bytes = (current_offset - target_offset.0).to_le_bytes();
424 let target_bytes = target.as_os_str().as_bytes();
16f0464d 425 let mut hardlink = Vec::with_capacity(offset_bytes.len() + target_bytes.len() + 1);
f3e50baa
WB
426 hardlink.extend(&offset_bytes);
427 hardlink.extend(target_bytes);
16f0464d 428 hardlink.push(0);
b0487d4f 429 let _this_offset: LinkOffset = self
f3e50baa 430 .add_file_entry(None, file_name, Some((format::PXAR_HARDLINK, &hardlink)))
b0487d4f 431 .await?;
d6748862 432 Ok(())
3fe26522
WB
433 }
434
d6748862 435 /// Return a file offset usable with `add_hardlink`.
3fe26522
WB
436 pub async fn add_device(
437 &mut self,
438 metadata: &Metadata,
439 file_name: &Path,
440 device: format::Device,
501ea220 441 ) -> io::Result<()> {
a7149b09
WB
442 if !metadata.is_device() {
443 io_bail!("entry added via add_device must have a device mode in its metadata");
444 }
445
3fe26522
WB
446 let device = device.to_le();
447 let device = unsafe {
448 std::slice::from_raw_parts(
449 &device as *const format::Device as *const u8,
450 size_of::<format::Device>(),
451 )
452 };
b0487d4f
WB
453 let _ofs: LinkOffset = self
454 .add_file_entry(
455 Some(metadata),
456 file_name,
457 Some((format::PXAR_DEVICE, device)),
458 )
459 .await?;
501ea220 460 Ok(())
3fe26522
WB
461 }
462
d6748862 463 /// Return a file offset usable with `add_hardlink`.
501ea220 464 pub async fn add_fifo(&mut self, metadata: &Metadata, file_name: &Path) -> io::Result<()> {
a7149b09
WB
465 if !metadata.is_fifo() {
466 io_bail!("entry added via add_device must be of type fifo in its metadata");
467 }
468
501ea220
WB
469 let _ofs: LinkOffset = self.add_file_entry(Some(metadata), file_name, None).await?;
470 Ok(())
a7149b09
WB
471 }
472
d6748862 473 /// Return a file offset usable with `add_hardlink`.
501ea220 474 pub async fn add_socket(&mut self, metadata: &Metadata, file_name: &Path) -> io::Result<()> {
a7149b09
WB
475 if !metadata.is_socket() {
476 io_bail!("entry added via add_device must be of type socket in its metadata");
477 }
478
501ea220
WB
479 let _ofs: LinkOffset = self.add_file_entry(Some(metadata), file_name, None).await?;
480 Ok(())
a7149b09
WB
481 }
482
d6748862 483 /// Return a file offset usable with `add_hardlink`.
3fe26522
WB
484 async fn add_file_entry(
485 &mut self,
2ab25a17 486 metadata: Option<&Metadata>,
3fe26522 487 file_name: &Path,
a7149b09 488 entry_htype_data: Option<(u64, &[u8])>,
d6748862 489 ) -> io::Result<LinkOffset> {
05356884 490 self.check()?;
2fb73e7b 491
ec4a53ed 492 let file_offset = self.position();
2fb73e7b
WB
493
494 let file_name = file_name.as_os_str().as_bytes();
2fb73e7b
WB
495
496 self.start_file_do(metadata, file_name).await?;
a7149b09 497 if let Some((htype, entry_data)) = entry_htype_data {
ec4a53ed 498 seq_write_pxar_entry(
d8402433 499 self.output.as_mut(),
ec4a53ed
WB
500 htype,
501 entry_data,
502 &mut self.state.write_position,
503 )
504 .await?;
a7149b09 505 }
2fb73e7b 506
ec4a53ed 507 let end_offset = self.position();
2fb73e7b
WB
508
509 self.state.items.push(GoodbyeItem {
510 hash: format::hash_filename(file_name),
511 offset: file_offset,
512 size: end_offset - file_offset,
513 });
514
d6748862 515 Ok(LinkOffset(file_offset))
2fb73e7b
WB
516 }
517
70acf637 518 #[inline]
ec4a53ed
WB
519 fn position(&mut self) -> u64 {
520 self.state.write_position
70acf637
WB
521 }
522
d8402433
SR
523 pub async fn create_directory(
524 &mut self,
70acf637
WB
525 file_name: &Path,
526 metadata: &Metadata,
d8402433 527 ) -> io::Result<EncoderImpl<'_, T>> {
70acf637
WB
528 self.check()?;
529
530 if !metadata.is_dir() {
531 io_bail!("directory metadata must contain the directory mode flag");
532 }
533
534 let file_name = file_name.as_os_str().as_bytes();
749855a4 535 let file_hash = format::hash_filename(file_name);
70acf637 536
ec4a53ed 537 let file_offset = self.position();
70acf637
WB
538 self.encode_filename(file_name).await?;
539
ec4a53ed 540 let entry_offset = self.position();
70acf637
WB
541 self.encode_metadata(&metadata).await?;
542
ec4a53ed
WB
543 let files_offset = self.position();
544
545 // the child will write to OUR state now:
546 let write_position = self.position();
70acf637 547
d8402433
SR
548 let file_copy_buffer = Arc::clone(&self.file_copy_buffer);
549
70acf637 550 Ok(EncoderImpl {
d8402433 551 // always forward as Borrowed(), to avoid stacking references on nested calls
b8d79e5f 552 output: self.output.to_borrowed(),
70acf637
WB
553 state: EncoderState {
554 entry_offset,
555 files_offset,
556 file_offset: Some(file_offset),
1b25fc08 557 file_hash,
ec4a53ed 558 write_position,
70acf637
WB
559 ..Default::default()
560 },
561 parent: Some(&mut self.state),
562 finished: false,
d8402433 563 file_copy_buffer,
70acf637
WB
564 })
565 }
566
2ab25a17
WB
567 async fn start_file_do(
568 &mut self,
569 metadata: Option<&Metadata>,
570 file_name: &[u8],
571 ) -> io::Result<()> {
70acf637 572 self.encode_filename(file_name).await?;
2ab25a17
WB
573 if let Some(metadata) = metadata {
574 self.encode_metadata(&metadata).await?;
575 }
70acf637
WB
576 Ok(())
577 }
578
579 async fn encode_metadata(&mut self, metadata: &Metadata) -> io::Result<()> {
ec4a53ed 580 seq_write_pxar_struct_entry(
d8402433 581 self.output.as_mut(),
ec4a53ed
WB
582 format::PXAR_ENTRY,
583 metadata.stat.clone(),
584 &mut self.state.write_position,
585 )
586 .await?;
b5a471a3
WB
587
588 for xattr in &metadata.xattrs {
589 self.write_xattr(xattr).await?;
590 }
591
592 self.write_acls(&metadata.acl).await?;
593
594 if let Some(fcaps) = &metadata.fcaps {
595 self.write_file_capabilities(fcaps).await?;
596 }
597
598 if let Some(qpid) = &metadata.quota_project_id {
599 self.write_quota_project_id(qpid).await?;
600 }
601
602 Ok(())
603 }
604
605 async fn write_xattr(&mut self, xattr: &format::XAttr) -> io::Result<()> {
ec4a53ed 606 seq_write_pxar_entry(
d8402433 607 self.output.as_mut(),
ec4a53ed
WB
608 format::PXAR_XATTR,
609 &xattr.data,
610 &mut self.state.write_position,
611 )
612 .await
b5a471a3
WB
613 }
614
615 async fn write_acls(&mut self, acl: &crate::Acl) -> io::Result<()> {
616 for acl in &acl.users {
ec4a53ed 617 seq_write_pxar_struct_entry(
d8402433 618 self.output.as_mut(),
ec4a53ed
WB
619 format::PXAR_ACL_USER,
620 acl.clone(),
621 &mut self.state.write_position,
622 )
623 .await?;
b5a471a3
WB
624 }
625
626 for acl in &acl.groups {
ec4a53ed 627 seq_write_pxar_struct_entry(
d8402433 628 self.output.as_mut(),
ec4a53ed
WB
629 format::PXAR_ACL_GROUP,
630 acl.clone(),
631 &mut self.state.write_position,
632 )
633 .await?;
b5a471a3
WB
634 }
635
636 if let Some(acl) = &acl.group_obj {
ec4a53ed 637 seq_write_pxar_struct_entry(
d8402433 638 self.output.as_mut(),
ec4a53ed
WB
639 format::PXAR_ACL_GROUP_OBJ,
640 acl.clone(),
641 &mut self.state.write_position,
642 )
643 .await?;
b5a471a3
WB
644 }
645
646 if let Some(acl) = &acl.default {
ec4a53ed 647 seq_write_pxar_struct_entry(
d8402433 648 self.output.as_mut(),
ec4a53ed
WB
649 format::PXAR_ACL_DEFAULT,
650 acl.clone(),
651 &mut self.state.write_position,
652 )
653 .await?;
b5a471a3
WB
654 }
655
656 for acl in &acl.default_users {
fc9c8c9d 657 seq_write_pxar_struct_entry(
d8402433 658 self.output.as_mut(),
fc9c8c9d
WB
659 format::PXAR_ACL_DEFAULT_USER,
660 acl.clone(),
ec4a53ed 661 &mut self.state.write_position,
fc9c8c9d
WB
662 )
663 .await?;
b5a471a3
WB
664 }
665
666 for acl in &acl.default_groups {
fc9c8c9d 667 seq_write_pxar_struct_entry(
d8402433 668 self.output.as_mut(),
fc9c8c9d
WB
669 format::PXAR_ACL_DEFAULT_GROUP,
670 acl.clone(),
ec4a53ed 671 &mut self.state.write_position,
fc9c8c9d
WB
672 )
673 .await?;
b5a471a3
WB
674 }
675
70acf637
WB
676 Ok(())
677 }
678
b5a471a3 679 async fn write_file_capabilities(&mut self, fcaps: &format::FCaps) -> io::Result<()> {
ec4a53ed 680 seq_write_pxar_entry(
d8402433 681 self.output.as_mut(),
ec4a53ed
WB
682 format::PXAR_FCAPS,
683 &fcaps.data,
684 &mut self.state.write_position,
685 )
686 .await
b5a471a3
WB
687 }
688
689 async fn write_quota_project_id(
690 &mut self,
691 quota_project_id: &format::QuotaProjectId,
692 ) -> io::Result<()> {
fc9c8c9d 693 seq_write_pxar_struct_entry(
d8402433 694 self.output.as_mut(),
fc9c8c9d 695 format::PXAR_QUOTA_PROJID,
1b25fc08 696 *quota_project_id,
ec4a53ed 697 &mut self.state.write_position,
fc9c8c9d
WB
698 )
699 .await
b5a471a3
WB
700 }
701
70acf637 702 async fn encode_filename(&mut self, file_name: &[u8]) -> io::Result<()> {
f3ac1c51 703 crate::util::validate_filename(file_name)?;
ec4a53ed 704 seq_write_pxar_entry_zero(
d8402433 705 self.output.as_mut(),
ec4a53ed
WB
706 format::PXAR_FILENAME,
707 file_name,
708 &mut self.state.write_position,
709 )
710 .await
70acf637
WB
711 }
712
d8402433 713 pub async fn finish(mut self) -> io::Result<()> {
70acf637 714 let tail_bytes = self.finish_goodbye_table().await?;
ec4a53ed 715 seq_write_pxar_entry(
d8402433 716 self.output.as_mut(),
ec4a53ed
WB
717 format::PXAR_GOODBYE,
718 &tail_bytes,
719 &mut self.state.write_position,
720 )
721 .await?;
722
e827b69a
WB
723 if let EncoderOutput::Owned(output) = &mut self.output {
724 flush(output).await?;
725 }
d995b531 726
ec4a53ed
WB
727 // done up here because of the self-borrow and to propagate
728 let end_offset = self.position();
729
749855a4 730 if let Some(parent) = &mut self.parent {
ec4a53ed
WB
731 parent.write_position = end_offset;
732
749855a4
WB
733 let file_offset = self
734 .state
735 .file_offset
736 .expect("internal error: parent set but no file_offset?");
737
749855a4
WB
738 parent.items.push(GoodbyeItem {
739 hash: self.state.file_hash,
740 offset: file_offset,
741 size: end_offset - file_offset,
742 });
743 }
70acf637 744 self.finished = true;
d8402433 745 Ok(())
70acf637
WB
746 }
747
748 async fn finish_goodbye_table(&mut self) -> io::Result<Vec<u8>> {
ec4a53ed 749 let goodbye_offset = self.position();
70acf637
WB
750
751 // "take" out the tail (to not leave an array of endian-swapped structs in `self`)
752 let mut tail = take(&mut self.state.items);
753 let tail_size = (tail.len() + 1) * size_of::<GoodbyeItem>();
754 let goodbye_size = tail_size as u64 + size_of::<format::Header>() as u64;
755
749855a4
WB
756 // sort, then create a BST
757 tail.sort_unstable_by(|a, b| a.hash.cmp(&b.hash));
70acf637 758
749855a4
WB
759 let mut bst = Vec::with_capacity(tail.len() + 1);
760 unsafe {
761 bst.set_len(tail.len());
70acf637 762 }
749855a4
WB
763 binary_tree_array::copy(tail.len(), |src, dest| {
764 let mut item = tail[src].clone();
765 // fixup the goodbye table offsets to be relative and with the right endianess
766 item.offset = goodbye_offset - item.offset;
767 unsafe {
768 std::ptr::write(&mut bst[dest], item.to_le());
769 }
770 });
771 drop(tail);
772
773 bst.push(
774 GoodbyeItem {
775 hash: format::PXAR_GOODBYE_TAIL_MARKER,
776 offset: goodbye_offset - self.state.entry_offset,
777 size: goodbye_size,
778 }
779 .to_le(),
780 );
70acf637
WB
781
782 // turn this into a byte vector since after endian-swapping we can no longer guarantee that
783 // the items make sense:
749855a4
WB
784 let data = bst.as_mut_ptr() as *mut u8;
785 let capacity = bst.capacity() * size_of::<GoodbyeItem>();
786 forget(bst);
70acf637
WB
787 Ok(unsafe { Vec::from_raw_parts(data, tail_size, capacity) })
788 }
789}
790
791/// Writer for a file object in a directory.
d8402433
SR
792pub struct FileImpl<'a, S: SeqWrite> {
793 output: &'a mut S,
70acf637
WB
794
795 /// This file's `GoodbyeItem`. FIXME: We currently don't touch this, can we just push it
796 /// directly instead of on Drop of FileImpl?
797 goodbye_item: GoodbyeItem,
798
799 /// While writing data to this file, this is how much space we still have left, this must reach
800 /// exactly zero.
801 remaining_size: u64,
802
59e6f679 803 /// The directory containing this file. This is where we propagate the `IncompleteFile` error
70acf637
WB
804 /// to, and where we insert our `GoodbyeItem`.
805 parent: &'a mut EncoderState,
806}
807
d8402433 808impl<'a, S: SeqWrite> Drop for FileImpl<'a, S> {
70acf637
WB
809 fn drop(&mut self) {
810 if self.remaining_size != 0 {
59e6f679 811 self.parent.add_error(EncodeError::IncompleteFile);
70acf637
WB
812 }
813
814 self.parent.items.push(self.goodbye_item.clone());
815 }
816}
817
d8402433 818impl<'a, S: SeqWrite> FileImpl<'a, S> {
d6748862
WB
819 /// Get the file offset to be able to reference it with `add_hardlink`.
820 pub fn file_offset(&self) -> LinkOffset {
821 LinkOffset(self.goodbye_item.offset)
822 }
823
70acf637
WB
824 fn check_remaining(&self, size: usize) -> io::Result<()> {
825 if size as u64 > self.remaining_size {
826 io_bail!("attempted to write more than previously allocated");
827 } else {
828 Ok(())
829 }
830 }
831
05356884 832 /// Poll write interface to more easily connect to tokio/futures.
7aee9c1f 833 #[cfg(feature = "tokio-io")]
05356884
WB
834 pub fn poll_write(
835 self: Pin<&mut Self>,
836 cx: &mut Context,
837 data: &[u8],
838 ) -> Poll<io::Result<usize>> {
839 let this = self.get_mut();
840 this.check_remaining(data.len())?;
841 let output = unsafe { Pin::new_unchecked(&mut *this.output) };
842 match output.poll_seq_write(cx, data) {
843 Poll::Ready(Ok(put)) => {
844 this.remaining_size -= put as u64;
ec4a53ed 845 this.parent.write_position += put as u64;
05356884
WB
846 Poll::Ready(Ok(put))
847 }
848 other => other,
849 }
850 }
851
852 /// Poll flush interface to more easily connect to tokio/futures.
7aee9c1f 853 #[cfg(feature = "tokio-io")]
05356884 854 pub fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
737f75cf 855 unsafe { self.map_unchecked_mut(|this| this.output).poll_flush(cx) }
05356884
WB
856 }
857
858 /// Poll close/shutdown interface to more easily connect to tokio/futures.
ec4a53ed
WB
859 ///
860 /// This just calls flush, though, since we're just a virtual writer writing to the file
861 /// provided by our encoder.
7aee9c1f 862 #[cfg(feature = "tokio-io")]
05356884 863 pub fn poll_close(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
737f75cf 864 unsafe { self.map_unchecked_mut(|this| this.output).poll_flush(cx) }
05356884
WB
865 }
866
70acf637
WB
867 /// Write file data for the current file entry in a pxar archive.
868 ///
869 /// This forwards to the output's `SeqWrite::poll_seq_write` and may write fewer bytes than
870 /// requested. Check the return value for how many. There's also a `write_all` method available
871 /// for convenience.
872 pub async fn write(&mut self, data: &[u8]) -> io::Result<usize> {
873 self.check_remaining(data.len())?;
ec4a53ed
WB
874 let put =
875 poll_fn(|cx| unsafe { Pin::new_unchecked(&mut self.output).poll_seq_write(cx, data) })
876 .await?;
84a1926f 877 //let put = seq_write(self.output.as_mut().unwrap(), data).await?;
70acf637 878 self.remaining_size -= put as u64;
ec4a53ed 879 self.parent.write_position += put as u64;
70acf637
WB
880 Ok(put)
881 }
882
883 /// Completely write file data for the current file entry in a pxar archive.
884 pub async fn write_all(&mut self, data: &[u8]) -> io::Result<()> {
885 self.check_remaining(data.len())?;
d8402433 886 seq_write_all(self.output, data, &mut self.parent.write_position).await?;
70acf637
WB
887 self.remaining_size -= data.len() as u64;
888 Ok(())
889 }
890}
05356884
WB
891
892#[cfg(feature = "tokio-io")]
d8402433 893impl<'a, S: SeqWrite> tokio::io::AsyncWrite for FileImpl<'a, S> {
327e33f2 894 fn poll_write(self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<io::Result<usize>> {
05356884
WB
895 FileImpl::poll_write(self, cx, buf)
896 }
897
898 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
899 FileImpl::poll_flush(self, cx)
900 }
901
902 fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
903 FileImpl::poll_close(self, cx)
904 }
905}