starnix_core/vfs/buffers/
message_queue.rs

1// Copyright 2021 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use std::collections::VecDeque;
6
7use super::message_types::{AncillaryData, Message, MessageData, UnixControlData};
8use crate::vfs::buffers::{InputBuffer, OutputBuffer};
9use crate::vfs::socket::SocketAddress;
10use starnix_uapi::errors::Errno;
11use starnix_uapi::vfs::FdEvents;
12use starnix_uapi::{error, uapi};
13
14/// Controls whether `SO_PASSCRED` is enabled, which causes stream reads and peeks to break before
15/// messages with differing credentials.
16#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
17pub enum PasscredMode {
18    #[default]
19    Disabled,
20    Enabled,
21}
22
23impl PasscredMode {
24    pub fn is_enabled(self) -> bool {
25        matches!(self, Self::Enabled)
26    }
27}
28
29impl From<bool> for PasscredMode {
30    fn from(enabled: bool) -> Self {
31        if enabled { Self::Enabled } else { Self::Disabled }
32    }
33}
34
35#[derive(Debug, Default, Clone)]
36pub struct MessageReadInfo {
37    pub bytes_read: usize,
38    pub message_length: usize,
39    pub address: Option<SocketAddress>,
40    pub credentials: Option<uapi::ucred>,
41    pub ancillary_data: Vec<AncillaryData>,
42}
43
44impl MessageReadInfo {
45    /// Appends `info` to self.
46    pub fn append(&mut self, info: &mut MessageReadInfo) {
47        self.bytes_read += info.bytes_read;
48        self.message_length += info.message_length;
49        if self.address.is_none() {
50            self.address = info.address.take();
51        }
52        if self.credentials.is_none() {
53            self.credentials = info.credentials.take();
54        }
55        if self.ancillary_data.iter().any(AncillaryData::is_credentials) {
56            info.ancillary_data.retain(|d| !d.is_credentials());
57        }
58        self.ancillary_data.append(&mut info.ancillary_data);
59    }
60}
61
62/// A `MessageQueue` stores a FIFO sequence of messages.
63#[derive(Debug)]
64pub struct MessageQueue<D: MessageData = Vec<u8>> {
65    /// The messages stored in the message queue.
66    ///
67    /// Writes are added at the end of the queue. Reads consume from the front of the queue.
68    messages: VecDeque<Message<D>>,
69
70    /// The total number of bytes currently in the message queue.
71    length: usize,
72
73    /// The maximum number of bytes that can be stored inside this pipe.
74    capacity: usize,
75}
76
77impl<D: MessageData> MessageQueue<D> {
78    pub fn new(capacity: usize) -> Self {
79        MessageQueue { messages: VecDeque::default(), length: 0, capacity }
80    }
81
82    /// Returns the number of bytes that can be written to the message queue before the buffer is
83    /// full.
84    pub fn available_capacity(&self) -> usize {
85        self.capacity - self.length
86    }
87
88    /// Returns the total number of bytes this message queue can store, regardless of the current
89    /// amount of data in the buffer.
90    pub fn capacity(&self) -> usize {
91        self.capacity
92    }
93
94    pub fn messages(&self) -> impl Iterator<Item = &Message<D>> {
95        self.messages.iter()
96    }
97
98    /// Sets the capacity of the message queue to the provided number of bytes.
99    ///
100    /// Reurns an error if the requested capacity could not be set (e.g., if the requested capacity
101    /// was less than the current number of bytes stored).
102    pub fn set_capacity(&mut self, requested_capacity: usize) -> Result<(), Errno> {
103        if requested_capacity < self.length {
104            return error!(EBUSY);
105        }
106        self.capacity = requested_capacity;
107        Ok(())
108    }
109
110    /// Returns true if the message queue is empty, or it only contains empty messages.
111    pub fn is_empty(&self) -> bool {
112        self.len() == 0
113    }
114
115    /// Returns the total length of all the messages in the message queue.
116    pub fn len(&self) -> usize {
117        self.length
118    }
119
120    fn update_address(message: &Message<D>, address: &mut Option<SocketAddress>) -> bool {
121        if message.address.is_some() && *address != message.address {
122            if address.is_some() {
123                return false;
124            }
125            *address = message.address.clone();
126        }
127        true
128    }
129
130    fn update_creds(
131        passcred: PasscredMode,
132        message: &Message<D>,
133        creds: &mut Option<uapi::ucred>,
134    ) -> bool {
135        if passcred.is_enabled() {
136            let message_creds = message.credentials.unwrap_or_else(UnixControlData::unknown_ucred);
137            if let Some(existing_creds) = *creds {
138                if existing_creds != message_creds {
139                    return false;
140                }
141            } else {
142                *creds = Some(message_creds);
143            }
144        }
145        true
146    }
147
148    /// Returns whether the next message in the queue (if any) can be coalesced into `info`.
149    pub fn can_coalesce_with(&self, info: &MessageReadInfo, passcred: PasscredMode) -> bool {
150        if info.bytes_read == 0 {
151            return true;
152        }
153        if info.ancillary_data.iter().any(|d| !d.is_credentials()) {
154            return false;
155        }
156        if passcred.is_enabled() != info.credentials.is_some() {
157            return false;
158        }
159        let Some(message) = self.peek_message() else {
160            return true;
161        };
162        let mut address = info.address.clone();
163        let mut creds = info.credentials;
164        Self::update_address(message, &mut address)
165            && Self::update_creds(passcred, message, &mut creds)
166    }
167
168    /// Reads messages until there are no more messages, a message with ancillary data is
169    /// encountered, or `data` are full.
170    ///
171    /// To read data from the queue without consuming the messages, see `peek_stream`.
172    ///
173    /// # Parameters
174    /// - `data`: The `OutputBuffer` to write the data to.
175    ///
176    /// Returns the message information containing the number of bytes read, the address, and any
177    /// ancillary data. Also returns a boolean indicating if any messages were read.
178    pub fn read_stream(
179        &mut self,
180        data: &mut dyn OutputBuffer,
181    ) -> Result<(MessageReadInfo, bool), Errno> {
182        self.read_stream_with_passcred(data, PasscredMode::Disabled)
183    }
184
185    /// Reads messages from a stream until there are no more messages, a message with ancillary
186    /// data is encountered, a message with differing credentials is encountered (when `passcred`
187    /// is `PasscredMode::Enabled`), or `data` are full.
188    ///
189    /// # Parameters
190    /// - `data`: The `OutputBuffer` to write the data to.
191    /// - `passcred`: Whether `SO_PASSCRED` is enabled on the receiving socket.
192    pub fn read_stream_with_passcred(
193        &mut self,
194        data: &mut dyn OutputBuffer,
195        passcred: PasscredMode,
196    ) -> Result<(MessageReadInfo, bool), Errno> {
197        let mut total_bytes_read = 0;
198        let mut address = None;
199        let mut ancillary_data = vec![];
200        let mut creds = None;
201
202        let mut messages_read = 0;
203        loop {
204            let mut message = match self.read_message() {
205                Some(m) => m,
206                None => break,
207            };
208            if !Self::update_address(&message, &mut address)
209                || !Self::update_creds(passcred, &message, &mut creds)
210            {
211                // We've already locked onto an address or credentials for this batch of messages,
212                // but we have found a message that doesn't match. We put it back for now and
213                // return the messages we have so far.
214                self.write_front(message);
215                break;
216            }
217            let bytes_read = match message.data.copy_to_user(data) {
218                Ok(bytes_read) => bytes_read,
219                Err(e) => {
220                    self.write_front(message);
221                    if total_bytes_read > 0 {
222                        break;
223                    }
224                    return Err(e);
225                }
226            };
227            messages_read += 1;
228            total_bytes_read += bytes_read;
229
230            if let Some(remaining_message) = message.split_off(bytes_read) {
231                // If not all the message data could fit, return the ancillary data now,
232                // and put the remaining data back with its address and credentials preserved.
233                ancillary_data.append(&mut message.ancillary_data);
234                self.write_front(remaining_message);
235                break;
236            }
237
238            if !message.ancillary_data.is_empty() {
239                ancillary_data.append(&mut message.ancillary_data);
240                break;
241            }
242
243            if data.available() == 0 {
244                break;
245            }
246        }
247
248        Ok((
249            MessageReadInfo {
250                bytes_read: total_bytes_read,
251                message_length: total_bytes_read,
252                address,
253                credentials: creds,
254                ancillary_data,
255            },
256            messages_read > 0,
257        ))
258    }
259
260    /// Peeks messages until there are no more messages, a message with ancillary data is
261    /// encountered, a message with differing credentials is encountered (when `passcred` is
262    /// `PasscredMode::Enabled`), or `data` are full.
263    ///
264    /// Unlike `read_stream`, this function does not remove the messages from the queue.
265    ///
266    /// Used to implement MSG_PEEK.
267    ///
268    /// # Parameters
269    /// - `data`: The `OutputBuffer` to write the data to.
270    /// - `passcred`: Whether `SO_PASSCRED` is enabled on the receiving socket.
271    ///
272    /// Returns the message information containing the number of bytes read, the address, and any
273    /// ancillary data. Also returns a boolean indicating if any messages were read.
274    pub fn peek_stream(
275        &self,
276        data: &mut dyn OutputBuffer,
277        passcred: PasscredMode,
278    ) -> Result<(MessageReadInfo, bool), Errno> {
279        let mut total_bytes_read = 0;
280        let mut address = None;
281        let mut ancillary_data = vec![];
282        let mut creds = None;
283
284        let mut messages_peeked = 0;
285        for (index, message) in self.messages.iter().enumerate() {
286            if index > 0 && data.available() == 0 {
287                break;
288            }
289
290            if !Self::update_address(message, &mut address)
291                || !Self::update_creds(passcred, message, &mut creds)
292            {
293                break;
294            }
295            let bytes_read = match message.data.copy_to_user(data) {
296                Ok(bytes_read) => bytes_read,
297                Err(e) => {
298                    if total_bytes_read > 0 {
299                        break;
300                    }
301                    return Err(e);
302                }
303            };
304            messages_peeked += 1;
305            ancillary_data.extend(message.ancillary_data.iter().cloned());
306            total_bytes_read += bytes_read;
307
308            if bytes_read < message.len() || !message.ancillary_data.is_empty() {
309                break;
310            }
311        }
312
313        Ok((
314            MessageReadInfo {
315                bytes_read: total_bytes_read,
316                message_length: total_bytes_read,
317                address,
318                credentials: creds,
319                ancillary_data,
320            },
321            messages_peeked > 0,
322        ))
323    }
324
325    pub fn read_datagram(
326        &mut self,
327        data: &mut dyn OutputBuffer,
328    ) -> Result<(MessageReadInfo, bool), Errno> {
329        if let Some(message) = self.read_message() {
330            Ok((
331                MessageReadInfo {
332                    bytes_read: message.data.copy_to_user(data)?,
333                    message_length: message.len(),
334                    address: message.address,
335                    credentials: message.credentials,
336                    ancillary_data: message.ancillary_data,
337                },
338                true,
339            ))
340        } else {
341            Ok((MessageReadInfo::default(), false))
342        }
343    }
344
345    pub fn peek_datagram(
346        &mut self,
347        data: &mut dyn OutputBuffer,
348    ) -> Result<(MessageReadInfo, bool), Errno> {
349        if let Some(message) = self.peek_message() {
350            Ok((
351                MessageReadInfo {
352                    bytes_read: message.data.copy_to_user(data)?,
353                    message_length: message.len(),
354                    address: message.address.clone(),
355                    credentials: message.credentials,
356                    ancillary_data: message.ancillary_data.clone(),
357                },
358                true,
359            ))
360        } else {
361            Ok((MessageReadInfo::default(), false))
362        }
363    }
364
365    /// Reads the next message in the buffer, if such a message exists.
366    pub fn read_message(&mut self) -> Option<Message<D>> {
367        self.messages.pop_front().map(|message| {
368            self.length -= message.len();
369            message
370        })
371    }
372
373    pub fn peek_queue(&self) -> &VecDeque<Message<D>> {
374        &self.messages
375    }
376
377    /// Peeks the next message in the buffer, if such a message exists.
378    fn peek_message(&self) -> Option<&Message<D>> {
379        self.messages.front()
380    }
381
382    /// Writes the the contents of `InputBuffer` into this socket.
383    /// Will return EAGAIN if not enough capacity is available.
384    ///
385    /// # Parameters
386    /// - `task`: The task to read memory from.
387    /// - `data`: The `InputBuffer` to read the data from.
388    ///
389    /// Returns the number of bytes that were written to the socket.
390    pub fn write_stream(
391        &mut self,
392        data: &mut dyn InputBuffer,
393        address: Option<SocketAddress>,
394        ancillary_data: &mut Vec<AncillaryData>,
395    ) -> Result<usize, Errno> {
396        self.write_stream_with_filter(data, address, None, ancillary_data, Some)
397    }
398
399    /// Writes the the contents of `InputBuffer` into this socket.
400    /// Will return EAGAIN if not enough capacity is available.
401    ///
402    /// # Parameters
403    /// - `task`: The task to read memory from.
404    /// - `data`: The `InputBuffer` to read the data from.
405    /// - `filter`: A filter to run on the message before inserting it into the queue. If it
406    ///             returns None, no message is enqueued.
407    ///
408    /// Returns the number of bytes that were written to the socket.
409    pub fn write_stream_with_filter(
410        &mut self,
411        data: &mut dyn InputBuffer,
412        address: Option<SocketAddress>,
413        default_credentials: Option<uapi::ucred>,
414        ancillary_data: &mut Vec<AncillaryData>,
415        filter: impl FnOnce(Message<D>) -> Option<Message<D>>,
416    ) -> Result<usize, Errno> {
417        let actual = std::cmp::min(self.available_capacity(), data.available());
418        if actual == 0 && data.available() > 0 {
419            return error!(EAGAIN);
420        }
421        let message_data = MessageData::copy_from_user(data, actual)?;
422        let explicit_creds = ancillary_data.iter().rev().find_map(AncillaryData::credentials);
423        let mut message_ancillary_data = std::mem::take(ancillary_data);
424        message_ancillary_data.retain(|d| !d.is_credentials());
425        if data.available() > 0 {
426            if let Some(creds) = explicit_creds {
427                ancillary_data.push(AncillaryData::Unix(UnixControlData::Credentials(creds)));
428            }
429        }
430        let message = Message::new(
431            message_data,
432            address,
433            explicit_creds.or(default_credentials),
434            message_ancillary_data,
435        );
436        if let Some(message) = filter(message) {
437            if message.len() > 0
438                || !message.ancillary_data.is_empty()
439                || message.credentials.is_some()
440                || message.address.is_some()
441            {
442                self.write_message(message);
443            }
444        }
445        Ok(actual)
446    }
447
448    /// Writes the the contents of `InputBuffer` into this socket as
449    /// single message. Will return EAGAIN if not enough capacity is available.
450    ///
451    /// # Parameters
452    /// - `task`: The task to read memory from.
453    /// - `data`: The `InputBuffer` to read the data from.
454    ///
455    /// Returns the number of bytes that were written to the socket.
456    pub fn write_datagram(
457        &mut self,
458        data: &mut dyn InputBuffer,
459        address: Option<SocketAddress>,
460        default_credentials: Option<uapi::ucred>,
461        ancillary_data: &mut Vec<AncillaryData>,
462    ) -> Result<usize, Errno> {
463        self.write_datagram_with_filter(data, address, default_credentials, ancillary_data, Some)
464    }
465
466    /// Writes the the contents of `InputBuffer` into this socket as
467    /// single message. Will return EAGAIN if not enough capacity is available.
468    ///
469    /// # Parameters
470    /// - `task`: The task to read memory from.
471    /// - `data`: The `InputBuffer` to read the data from.
472    /// - `filter`: A filter to run on the message before inserting it into the queue. If it
473    ///             returns None, no message is enqueued.
474    ///
475    /// Returns the number of bytes that were written to the socket.
476    pub fn write_datagram_with_filter(
477        &mut self,
478        data: &mut dyn InputBuffer,
479        address: Option<SocketAddress>,
480        default_credentials: Option<uapi::ucred>,
481        ancillary_data: &mut Vec<AncillaryData>,
482        filter: impl FnOnce(Message<D>) -> Option<Message<D>>,
483    ) -> Result<usize, Errno> {
484        let actual = data.available();
485        if actual > self.capacity() {
486            return error!(EMSGSIZE);
487        }
488        if actual > self.available_capacity() {
489            return error!(EAGAIN);
490        }
491        let message_data = MessageData::copy_from_user(data, actual)?;
492        let explicit_creds = ancillary_data.iter().rev().find_map(AncillaryData::credentials);
493        let mut message_ancillary_data = std::mem::take(ancillary_data);
494        message_ancillary_data.retain(|d| !d.is_credentials());
495        let message = Message::new(
496            message_data,
497            address,
498            explicit_creds.or(default_credentials),
499            message_ancillary_data,
500        );
501        if let Some(message) = filter(message) {
502            self.write_message(message);
503        }
504        Ok(actual)
505    }
506
507    /// Writes a message to the front of the message queue.
508    pub fn write_front(&mut self, message: Message<D>) {
509        self.length += message.len();
510        debug_assert!(self.length <= self.capacity);
511        self.messages.push_front(message);
512    }
513
514    /// Writes a message to the back of the message queue.
515    pub fn write_message(&mut self, message: Message<D>) {
516        self.length += message.len();
517        debug_assert!(self.length <= self.capacity);
518        self.messages.push_back(message);
519    }
520
521    pub fn query_events(&self) -> FdEvents {
522        let mut events = FdEvents::empty();
523        if self.available_capacity() > 0 {
524            events |= FdEvents::POLLOUT;
525        }
526        if !self.is_empty() {
527            events |= FdEvents::POLLIN;
528        }
529        events
530    }
531}
532
533#[cfg(test)]
534mod tests {
535    use super::*;
536
537    /// Tests that a write followed by a read returns the written message.
538    #[::fuchsia::test]
539    fn test_read_write() {
540        let mut message_queue = MessageQueue::new(usize::MAX);
541        let bytes: Vec<u8> = vec![1, 2, 3];
542        let message: Message = bytes.into();
543        message_queue.write_message(message.clone());
544        assert_eq!(message_queue.len(), 3);
545        assert_eq!(message_queue.read_message(), Some(message));
546        assert!(message_queue.is_empty());
547    }
548
549    /// Tests that ancillary data does not contribute to the message queue length.
550    #[::fuchsia::test]
551    fn test_control_len() {
552        let mut message_queue = MessageQueue::new(usize::MAX);
553        let bytes: Vec<u8> = vec![1, 2, 3];
554        let ancillary_data =
555            vec![AncillaryData::Unix(UnixControlData::Security(bytes.clone().into()))];
556        let message = Message::new(vec![].into(), None, None, ancillary_data);
557        message_queue.write_message(message);
558        assert_eq!(message_queue.len(), 0);
559        message_queue.write_message(bytes.clone().into());
560        assert_eq!(message_queue.len(), bytes.len());
561    }
562
563    /// Tests that multiple writes followed by multiple reads return the data in the correct order.
564    #[::fuchsia::test]
565    fn test_read_write_multiple() {
566        let mut message_queue = MessageQueue::new(usize::MAX);
567        let first_bytes: Vec<u8> = vec![1, 2, 3];
568        let second_bytes: Vec<u8> = vec![3, 4, 5];
569
570        for message in [first_bytes.clone().into(), second_bytes.clone().into()] {
571            message_queue.write_message(message);
572        }
573
574        assert_eq!(message_queue.len(), first_bytes.len() + second_bytes.len());
575        assert_eq!(message_queue.read_message(), Some(first_bytes.into()));
576        assert_eq!(message_queue.len(), second_bytes.len());
577        assert_eq!(message_queue.read_message(), Some(second_bytes.into()));
578        assert_eq!(message_queue.read_message(), None);
579    }
580
581    #[::fuchsia::test]
582    fn test_can_coalesce_with_passcred_toggled() {
583        let mut message_queue: MessageQueue = MessageQueue::new(usize::MAX);
584        let creds = uapi::ucred { pid: 1, uid: 2, gid: 3 };
585        message_queue.write_message(Message::new(vec![1, 2].into(), None, Some(creds), Vec::new()));
586
587        let info_with_creds = MessageReadInfo {
588            bytes_read: 2,
589            message_length: 2,
590            address: None,
591            credentials: Some(creds),
592            ancillary_data: Vec::new(),
593        };
594        assert!(message_queue.can_coalesce_with(&info_with_creds, PasscredMode::Enabled));
595        assert!(!message_queue.can_coalesce_with(&info_with_creds, PasscredMode::Disabled));
596
597        let info_without_creds = MessageReadInfo {
598            bytes_read: 2,
599            message_length: 2,
600            address: None,
601            credentials: None,
602            ancillary_data: Vec::new(),
603        };
604        assert!(message_queue.can_coalesce_with(&info_without_creds, PasscredMode::Disabled));
605        assert!(!message_queue.can_coalesce_with(&info_without_creds, PasscredMode::Enabled));
606    }
607}
starnix_core/vfs/buffers/message_queue.rs

starnix_core/vfs/buffers/
message_queue.rs
