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            messages_read += 1;
218
219            let bytes_read = message.data.copy_to_user(data)?;
220            total_bytes_read += bytes_read;
221
222            if let Some(remaining_message) = message.split_off(bytes_read) {
223                // If not all the message data could fit, return the ancillary data now,
224                // and put the remaining data back with its address and credentials preserved.
225                ancillary_data.append(&mut message.ancillary_data);
226                self.write_front(remaining_message);
227                break;
228            }
229
230            if !message.ancillary_data.is_empty() {
231                ancillary_data.append(&mut message.ancillary_data);
232                break;
233            }
234
235            if data.available() == 0 {
236                break;
237            }
238        }
239
240        Ok((
241            MessageReadInfo {
242                bytes_read: total_bytes_read,
243                message_length: total_bytes_read,
244                address,
245                credentials: creds,
246                ancillary_data,
247            },
248            messages_read > 0,
249        ))
250    }
251
252    /// Peeks messages until there are no more messages, a message with ancillary data is
253    /// encountered, a message with differing credentials is encountered (when `passcred` is
254    /// `PasscredMode::Enabled`), or `data` are full.
255    ///
256    /// Unlike `read_stream`, this function does not remove the messages from the queue.
257    ///
258    /// Used to implement MSG_PEEK.
259    ///
260    /// # Parameters
261    /// - `data`: The `OutputBuffer` to write the data to.
262    /// - `passcred`: Whether `SO_PASSCRED` is enabled on the receiving socket.
263    ///
264    /// Returns the message information containing the number of bytes read, the address, and any
265    /// ancillary data. Also returns a boolean indicating if any messages were read.
266    pub fn peek_stream(
267        &self,
268        data: &mut dyn OutputBuffer,
269        passcred: PasscredMode,
270    ) -> Result<(MessageReadInfo, bool), Errno> {
271        let mut total_bytes_read = 0;
272        let mut address = None;
273        let mut ancillary_data = vec![];
274        let mut creds = None;
275
276        let mut messages_peeked = 0;
277        for (index, message) in self.messages.iter().enumerate() {
278            if index > 0 && data.available() == 0 {
279                break;
280            }
281
282            if !Self::update_address(message, &mut address)
283                || !Self::update_creds(passcred, message, &mut creds)
284            {
285                break;
286            }
287            messages_peeked += 1;
288
289            ancillary_data.extend(message.ancillary_data.iter().cloned());
290
291            let bytes_read = message.data.copy_to_user(data)?;
292            total_bytes_read += bytes_read;
293
294            if bytes_read < message.len() || !message.ancillary_data.is_empty() {
295                break;
296            }
297        }
298
299        Ok((
300            MessageReadInfo {
301                bytes_read: total_bytes_read,
302                message_length: total_bytes_read,
303                address,
304                credentials: creds,
305                ancillary_data,
306            },
307            messages_peeked > 0,
308        ))
309    }
310
311    pub fn read_datagram(
312        &mut self,
313        data: &mut dyn OutputBuffer,
314    ) -> Result<(MessageReadInfo, bool), Errno> {
315        if let Some(message) = self.read_message() {
316            Ok((
317                MessageReadInfo {
318                    bytes_read: message.data.copy_to_user(data)?,
319                    message_length: message.len(),
320                    address: message.address,
321                    credentials: message.credentials,
322                    ancillary_data: message.ancillary_data,
323                },
324                true,
325            ))
326        } else {
327            Ok((MessageReadInfo::default(), false))
328        }
329    }
330
331    pub fn peek_datagram(
332        &mut self,
333        data: &mut dyn OutputBuffer,
334    ) -> Result<(MessageReadInfo, bool), Errno> {
335        if let Some(message) = self.peek_message() {
336            Ok((
337                MessageReadInfo {
338                    bytes_read: message.data.copy_to_user(data)?,
339                    message_length: message.len(),
340                    address: message.address.clone(),
341                    credentials: message.credentials,
342                    ancillary_data: message.ancillary_data.clone(),
343                },
344                true,
345            ))
346        } else {
347            Ok((MessageReadInfo::default(), false))
348        }
349    }
350
351    /// Reads the next message in the buffer, if such a message exists.
352    pub fn read_message(&mut self) -> Option<Message<D>> {
353        self.messages.pop_front().map(|message| {
354            self.length -= message.len();
355            message
356        })
357    }
358
359    pub fn peek_queue(&self) -> &VecDeque<Message<D>> {
360        &self.messages
361    }
362
363    /// Peeks the next message in the buffer, if such a message exists.
364    fn peek_message(&self) -> Option<&Message<D>> {
365        self.messages.front()
366    }
367
368    /// Writes the the contents of `InputBuffer` into this socket.
369    /// Will return EAGAIN if not enough capacity is available.
370    ///
371    /// # Parameters
372    /// - `task`: The task to read memory from.
373    /// - `data`: The `InputBuffer` to read the data from.
374    ///
375    /// Returns the number of bytes that were written to the socket.
376    pub fn write_stream(
377        &mut self,
378        data: &mut dyn InputBuffer,
379        address: Option<SocketAddress>,
380        ancillary_data: &mut Vec<AncillaryData>,
381    ) -> Result<usize, Errno> {
382        self.write_stream_with_filter(data, address, None, ancillary_data, Some)
383    }
384
385    /// Writes the the contents of `InputBuffer` into this socket.
386    /// Will return EAGAIN if not enough capacity is available.
387    ///
388    /// # Parameters
389    /// - `task`: The task to read memory from.
390    /// - `data`: The `InputBuffer` to read the data from.
391    /// - `filter`: A filter to run on the message before inserting it into the queue. If it
392    ///             returns None, no message is enqueued.
393    ///
394    /// Returns the number of bytes that were written to the socket.
395    pub fn write_stream_with_filter(
396        &mut self,
397        data: &mut dyn InputBuffer,
398        address: Option<SocketAddress>,
399        default_credentials: Option<uapi::ucred>,
400        ancillary_data: &mut Vec<AncillaryData>,
401        filter: impl FnOnce(Message<D>) -> Option<Message<D>>,
402    ) -> Result<usize, Errno> {
403        let actual = std::cmp::min(self.available_capacity(), data.available());
404        if actual == 0 && data.available() > 0 {
405            return error!(EAGAIN);
406        }
407        let message_data = MessageData::copy_from_user(data, actual)?;
408        let explicit_creds = ancillary_data.iter().rev().find_map(AncillaryData::credentials);
409        let mut message_ancillary_data = std::mem::take(ancillary_data);
410        message_ancillary_data.retain(|d| !d.is_credentials());
411        if data.available() > 0 {
412            if let Some(creds) = explicit_creds {
413                ancillary_data.push(AncillaryData::Unix(UnixControlData::Credentials(creds)));
414            }
415        }
416        let message = Message::new(
417            message_data,
418            address,
419            explicit_creds.or(default_credentials),
420            message_ancillary_data,
421        );
422        if let Some(message) = filter(message) {
423            if message.len() > 0
424                || !message.ancillary_data.is_empty()
425                || message.credentials.is_some()
426                || message.address.is_some()
427            {
428                self.write_message(message);
429            }
430        }
431        Ok(actual)
432    }
433
434    /// Writes the the contents of `InputBuffer` into this socket as
435    /// single message. Will return EAGAIN if not enough capacity is available.
436    ///
437    /// # Parameters
438    /// - `task`: The task to read memory from.
439    /// - `data`: The `InputBuffer` to read the data from.
440    ///
441    /// Returns the number of bytes that were written to the socket.
442    pub fn write_datagram(
443        &mut self,
444        data: &mut dyn InputBuffer,
445        address: Option<SocketAddress>,
446        default_credentials: Option<uapi::ucred>,
447        ancillary_data: &mut Vec<AncillaryData>,
448    ) -> Result<usize, Errno> {
449        self.write_datagram_with_filter(data, address, default_credentials, ancillary_data, Some)
450    }
451
452    /// Writes the the contents of `InputBuffer` into this socket as
453    /// single message. Will return EAGAIN if not enough capacity is available.
454    ///
455    /// # Parameters
456    /// - `task`: The task to read memory from.
457    /// - `data`: The `InputBuffer` to read the data from.
458    /// - `filter`: A filter to run on the message before inserting it into the queue. If it
459    ///             returns None, no message is enqueued.
460    ///
461    /// Returns the number of bytes that were written to the socket.
462    pub fn write_datagram_with_filter(
463        &mut self,
464        data: &mut dyn InputBuffer,
465        address: Option<SocketAddress>,
466        default_credentials: Option<uapi::ucred>,
467        ancillary_data: &mut Vec<AncillaryData>,
468        filter: impl FnOnce(Message<D>) -> Option<Message<D>>,
469    ) -> Result<usize, Errno> {
470        let actual = data.available();
471        if actual > self.capacity() {
472            return error!(EMSGSIZE);
473        }
474        if actual > self.available_capacity() {
475            return error!(EAGAIN);
476        }
477        let message_data = MessageData::copy_from_user(data, actual)?;
478        let explicit_creds = ancillary_data.iter().rev().find_map(AncillaryData::credentials);
479        let mut message_ancillary_data = std::mem::take(ancillary_data);
480        message_ancillary_data.retain(|d| !d.is_credentials());
481        let message = Message::new(
482            message_data,
483            address,
484            explicit_creds.or(default_credentials),
485            message_ancillary_data,
486        );
487        if let Some(message) = filter(message) {
488            self.write_message(message);
489        }
490        Ok(actual)
491    }
492
493    /// Writes a message to the front of the message queue.
494    pub fn write_front(&mut self, message: Message<D>) {
495        self.length += message.len();
496        debug_assert!(self.length <= self.capacity);
497        self.messages.push_front(message);
498    }
499
500    /// Writes a message to the back of the message queue.
501    pub fn write_message(&mut self, message: Message<D>) {
502        self.length += message.len();
503        debug_assert!(self.length <= self.capacity);
504        self.messages.push_back(message);
505    }
506
507    pub fn query_events(&self) -> FdEvents {
508        let mut events = FdEvents::empty();
509        if self.available_capacity() > 0 {
510            events |= FdEvents::POLLOUT;
511        }
512        if !self.is_empty() {
513            events |= FdEvents::POLLIN;
514        }
515        events
516    }
517}
518
519#[cfg(test)]
520mod tests {
521    use super::*;
522
523    /// Tests that a write followed by a read returns the written message.
524    #[::fuchsia::test]
525    fn test_read_write() {
526        let mut message_queue = MessageQueue::new(usize::MAX);
527        let bytes: Vec<u8> = vec![1, 2, 3];
528        let message: Message = bytes.into();
529        message_queue.write_message(message.clone());
530        assert_eq!(message_queue.len(), 3);
531        assert_eq!(message_queue.read_message(), Some(message));
532        assert!(message_queue.is_empty());
533    }
534
535    /// Tests that ancillary data does not contribute to the message queue length.
536    #[::fuchsia::test]
537    fn test_control_len() {
538        let mut message_queue = MessageQueue::new(usize::MAX);
539        let bytes: Vec<u8> = vec![1, 2, 3];
540        let ancillary_data =
541            vec![AncillaryData::Unix(UnixControlData::Security(bytes.clone().into()))];
542        let message = Message::new(vec![].into(), None, None, ancillary_data);
543        message_queue.write_message(message);
544        assert_eq!(message_queue.len(), 0);
545        message_queue.write_message(bytes.clone().into());
546        assert_eq!(message_queue.len(), bytes.len());
547    }
548
549    /// Tests that multiple writes followed by multiple reads return the data in the correct order.
550    #[::fuchsia::test]
551    fn test_read_write_multiple() {
552        let mut message_queue = MessageQueue::new(usize::MAX);
553        let first_bytes: Vec<u8> = vec![1, 2, 3];
554        let second_bytes: Vec<u8> = vec![3, 4, 5];
555
556        for message in [first_bytes.clone().into(), second_bytes.clone().into()] {
557            message_queue.write_message(message);
558        }
559
560        assert_eq!(message_queue.len(), first_bytes.len() + second_bytes.len());
561        assert_eq!(message_queue.read_message(), Some(first_bytes.into()));
562        assert_eq!(message_queue.len(), second_bytes.len());
563        assert_eq!(message_queue.read_message(), Some(second_bytes.into()));
564        assert_eq!(message_queue.read_message(), None);
565    }
566
567    #[::fuchsia::test]
568    fn test_can_coalesce_with_passcred_toggled() {
569        let mut message_queue: MessageQueue = MessageQueue::new(usize::MAX);
570        let creds = uapi::ucred { pid: 1, uid: 2, gid: 3 };
571        message_queue.write_message(Message::new(vec![1, 2].into(), None, Some(creds), Vec::new()));
572
573        let info_with_creds = MessageReadInfo {
574            bytes_read: 2,
575            message_length: 2,
576            address: None,
577            credentials: Some(creds),
578            ancillary_data: Vec::new(),
579        };
580        assert!(message_queue.can_coalesce_with(&info_with_creds, PasscredMode::Enabled));
581        assert!(!message_queue.can_coalesce_with(&info_with_creds, PasscredMode::Disabled));
582
583        let info_without_creds = MessageReadInfo {
584            bytes_read: 2,
585            message_length: 2,
586            address: None,
587            credentials: None,
588            ancillary_data: Vec::new(),
589        };
590        assert!(message_queue.can_coalesce_with(&info_without_creds, PasscredMode::Disabled));
591        assert!(!message_queue.can_coalesce_with(&info_without_creds, PasscredMode::Enabled));
592    }
593}
starnix_core/vfs/buffers/message_queue.rs

starnix_core/vfs/buffers/
message_queue.rs
