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};
8use crate::vfs::buffers::{InputBuffer, OutputBuffer};
9use crate::vfs::socket::SocketAddress;
10use starnix_uapi::error;
11use starnix_uapi::errors::Errno;
12use starnix_uapi::vfs::FdEvents;
13
14#[derive(Debug, Default, Clone)]
15pub struct MessageReadInfo {
16    pub bytes_read: usize,
17    pub message_length: usize,
18    pub address: Option<SocketAddress>,
19    pub ancillary_data: Vec<AncillaryData>,
20}
21
22impl MessageReadInfo {
23    /// Appends `info` to self.
24    pub fn append(&mut self, info: &mut MessageReadInfo) {
25        self.bytes_read += info.bytes_read;
26        self.message_length += info.message_length;
27        self.ancillary_data.append(&mut info.ancillary_data);
28    }
29}
30
31/// A `MessageQueue` stores a FIFO sequence of messages.
32#[derive(Debug)]
33pub struct MessageQueue<D: MessageData = Vec<u8>> {
34    /// The messages stored in the message queue.
35    ///
36    /// Writes are added at the end of the queue. Reads consume from the front of the queue.
37    messages: VecDeque<Message<D>>,
38
39    /// The total number of bytes currently in the message queue.
40    length: usize,
41
42    /// The maximum number of bytes that can be stored inside this pipe.
43    capacity: usize,
44}
45
46impl<D: MessageData> MessageQueue<D> {
47    pub fn new(capacity: usize) -> Self {
48        MessageQueue { messages: VecDeque::default(), length: 0, capacity }
49    }
50
51    /// Returns the number of bytes that can be written to the message queue before the buffer is
52    /// full.
53    pub fn available_capacity(&self) -> usize {
54        self.capacity - self.length
55    }
56
57    /// Returns the total number of bytes this message queue can store, regardless of the current
58    /// amount of data in the buffer.
59    pub fn capacity(&self) -> usize {
60        self.capacity
61    }
62
63    pub fn messages(&self) -> impl Iterator<Item = &Message<D>> {
64        self.messages.iter()
65    }
66
67    /// Sets the capacity of the message queue to the provided number of bytes.
68    ///
69    /// Reurns an error if the requested capacity could not be set (e.g., if the requested capacity
70    /// was less than the current number of bytes stored).
71    pub fn set_capacity(&mut self, requested_capacity: usize) -> Result<(), Errno> {
72        if requested_capacity < self.length {
73            return error!(EBUSY);
74        }
75        self.capacity = requested_capacity;
76        Ok(())
77    }
78
79    /// Returns true if the message queue is empty, or it only contains empty messages.
80    pub fn is_empty(&self) -> bool {
81        self.len() == 0
82    }
83
84    /// Returns the total length of all the messages in the message queue.
85    pub fn len(&self) -> usize {
86        self.length
87    }
88
89    fn update_address(message: &Message<D>, address: &mut Option<SocketAddress>) -> bool {
90        if message.address.is_some() && *address != message.address {
91            if address.is_some() {
92                return false;
93            }
94            *address = message.address.clone();
95        }
96        true
97    }
98
99    /// Reads messages until there are no more messages, a message with ancillary data is
100    /// encountered, or `data` are full.
101    ///
102    /// To read data from the queue without consuming the messages, see `peek_stream`.
103    ///
104    /// # Parameters
105    /// - `data`: The `OutputBuffer` to write the data to.
106    ///
107    /// Returns the message information containing the number of bytes read, the address, and any
108    /// ancillary data. Also returns a boolean indicating if any messages were read.
109    pub fn read_stream(
110        &mut self,
111        data: &mut dyn OutputBuffer,
112    ) -> Result<(MessageReadInfo, bool), Errno> {
113        let mut total_bytes_read = 0;
114        let mut address = None;
115        let mut ancillary_data = vec![];
116
117        let mut messages_read = 0;
118        loop {
119            let mut message = match self.read_message() {
120                Some(m) => m,
121                None => break,
122            };
123            if !Self::update_address(&message, &mut address) {
124                // We've already locked onto an address for this batch of messages, but we
125                // have found a message that doesn't match. We put it back for now and
126                // return the messages we have so far.
127                self.write_front(message);
128                break;
129            }
130            messages_read += 1;
131
132            let bytes_read = message.data.copy_to_user(data)?;
133            total_bytes_read += bytes_read;
134
135            if let Some(remaining_data) = message.data.split_off(bytes_read) {
136                // If not all the message data could fit, return the ancillary data now,
137                // and put the remaining data back without it.
138                ancillary_data = message.ancillary_data;
139                self.write_front(Message::new(remaining_data, message.address.clone(), vec![]));
140                break;
141            }
142
143            // TODO(https://fxbug.dev/542829111): Only break on credentials if SO_PASSCRED is enabled.
144            if !message.ancillary_data.is_empty() {
145                ancillary_data = message.ancillary_data;
146                break;
147            }
148
149            if data.available() == 0 {
150                break;
151            }
152        }
153
154        Ok((
155            MessageReadInfo {
156                bytes_read: total_bytes_read,
157                message_length: total_bytes_read,
158                address,
159                ancillary_data,
160            },
161            messages_read > 0,
162        ))
163    }
164
165    /// Peeks messages until there are no more messages, a message with ancillary data is
166    /// encountered, or `data` are full.
167    ///
168    /// Unlike `read_stream`, this function does not remove the messages from the queue.
169    ///
170    /// Used to implement MSG_PEEK.
171    ///
172    /// # Parameters
173    /// - `data`: The `OutputBuffer` to write the data to.
174    ///
175    /// Returns the message information containing the number of bytes read, the address, and any
176    /// ancillary data. Also returns a boolean indicating if any messages were read.
177    pub fn peek_stream(
178        &self,
179        data: &mut dyn OutputBuffer,
180    ) -> Result<(MessageReadInfo, bool), Errno> {
181        let mut total_bytes_read = 0;
182        let mut address = None;
183        let mut ancillary_data = vec![];
184
185        let mut messages_peeked = 0;
186        for (index, message) in self.messages.iter().enumerate() {
187            if index > 0 && data.available() == 0 {
188                break;
189            }
190
191            if !Self::update_address(message, &mut address) {
192                break;
193            }
194            messages_peeked += 1;
195
196            if !message.ancillary_data.is_empty() {
197                ancillary_data = message.ancillary_data.clone();
198            }
199
200            let bytes_read = message.data.copy_to_user(data)?;
201            total_bytes_read += bytes_read;
202
203            if bytes_read < message.len() {
204                break;
205            }
206
207            if !ancillary_data.is_empty() {
208                break;
209            }
210        }
211
212        Ok((
213            MessageReadInfo {
214                bytes_read: total_bytes_read,
215                message_length: total_bytes_read,
216                address,
217                ancillary_data,
218            },
219            messages_peeked > 0,
220        ))
221    }
222
223    pub fn read_datagram(
224        &mut self,
225        data: &mut dyn OutputBuffer,
226    ) -> Result<(MessageReadInfo, bool), Errno> {
227        if let Some(message) = self.read_message() {
228            Ok((
229                MessageReadInfo {
230                    bytes_read: message.data.copy_to_user(data)?,
231                    message_length: message.len(),
232                    address: message.address,
233                    ancillary_data: message.ancillary_data,
234                },
235                true,
236            ))
237        } else {
238            Ok((MessageReadInfo::default(), false))
239        }
240    }
241
242    pub fn peek_datagram(
243        &mut self,
244        data: &mut dyn OutputBuffer,
245    ) -> Result<(MessageReadInfo, bool), Errno> {
246        if let Some(message) = self.peek_message() {
247            Ok((
248                MessageReadInfo {
249                    bytes_read: message.data.copy_to_user(data)?,
250                    message_length: message.len(),
251                    address: message.address.clone(),
252                    ancillary_data: message.ancillary_data.clone(),
253                },
254                true,
255            ))
256        } else {
257            Ok((MessageReadInfo::default(), false))
258        }
259    }
260
261    /// Reads the next message in the buffer, if such a message exists.
262    pub fn read_message(&mut self) -> Option<Message<D>> {
263        self.messages.pop_front().map(|message| {
264            self.length -= message.len();
265            message
266        })
267    }
268
269    pub fn peek_queue(&self) -> &VecDeque<Message<D>> {
270        &self.messages
271    }
272
273    /// Peeks the next message in the buffer, if such a message exists.
274    fn peek_message(&self) -> Option<&Message<D>> {
275        self.messages.front()
276    }
277
278    /// Writes the the contents of `InputBuffer` into this socket.
279    /// Will return EAGAIN if not enough capacity is available.
280    ///
281    /// # Parameters
282    /// - `task`: The task to read memory from.
283    /// - `data`: The `InputBuffer` to read the data from.
284    ///
285    /// Returns the number of bytes that were written to the socket.
286    pub fn write_stream(
287        &mut self,
288        data: &mut dyn InputBuffer,
289        address: Option<SocketAddress>,
290        ancillary_data: &mut Vec<AncillaryData>,
291    ) -> Result<usize, Errno> {
292        self.write_stream_with_filter(data, address, ancillary_data, Some)
293    }
294
295    /// Writes the the contents of `InputBuffer` into this socket.
296    /// Will return EAGAIN if not enough capacity is available.
297    ///
298    /// # Parameters
299    /// - `task`: The task to read memory from.
300    /// - `data`: The `InputBuffer` to read the data from.
301    /// - `filter`: A filter to run on the message before inserting it into the queue. If it
302    ///             returns None, no message is enqueued.
303    ///
304    /// Returns the number of bytes that were written to the socket.
305    pub fn write_stream_with_filter(
306        &mut self,
307        data: &mut dyn InputBuffer,
308        address: Option<SocketAddress>,
309        ancillary_data: &mut Vec<AncillaryData>,
310        filter: impl FnOnce(Message<D>) -> Option<Message<D>>,
311    ) -> Result<usize, Errno> {
312        let actual = std::cmp::min(self.available_capacity(), data.available());
313        if actual == 0 && data.available() > 0 {
314            return error!(EAGAIN);
315        }
316        let data = MessageData::copy_from_user(data, actual)?;
317        let message = Message::new(data, address, std::mem::take(ancillary_data));
318        if let Some(message) = filter(message) {
319            self.write_message(message);
320        }
321        Ok(actual)
322    }
323
324    /// Writes the the contents of `InputBuffer` into this socket as
325    /// single message. Will return EAGAIN if not enough capacity is available.
326    ///
327    /// # Parameters
328    /// - `task`: The task to read memory from.
329    /// - `data`: The `InputBuffer` to read the data from.
330    ///
331    /// Returns the number of bytes that were written to the socket.
332    pub fn write_datagram(
333        &mut self,
334        data: &mut dyn InputBuffer,
335        address: Option<SocketAddress>,
336        ancillary_data: &mut Vec<AncillaryData>,
337    ) -> Result<usize, Errno> {
338        self.write_datagram_with_filter(data, address, ancillary_data, Some)
339    }
340
341    /// Writes the the contents of `InputBuffer` into this socket as
342    /// single message. Will return EAGAIN if not enough capacity is available.
343    ///
344    /// # Parameters
345    /// - `task`: The task to read memory from.
346    /// - `data`: The `InputBuffer` to read the data from.
347    /// - `filter`: A filter to run on the message before inserting it into the queue. If it
348    ///             returns None, no message is enqueued.
349    ///
350    /// Returns the number of bytes that were written to the socket.
351    pub fn write_datagram_with_filter(
352        &mut self,
353        data: &mut dyn InputBuffer,
354        address: Option<SocketAddress>,
355        ancillary_data: &mut Vec<AncillaryData>,
356        filter: impl FnOnce(Message<D>) -> Option<Message<D>>,
357    ) -> Result<usize, Errno> {
358        let actual = data.available();
359        if actual > self.capacity() {
360            return error!(EMSGSIZE);
361        }
362        if actual > self.available_capacity() {
363            return error!(EAGAIN);
364        }
365        let data = MessageData::copy_from_user(data, actual)?;
366        let message = Message::new(data, address, std::mem::take(ancillary_data));
367        if let Some(message) = filter(message) {
368            self.write_message(message);
369        }
370        Ok(actual)
371    }
372
373    /// Writes a message to the front of the message queue.
374    pub fn write_front(&mut self, message: Message<D>) {
375        self.length += message.len();
376        debug_assert!(self.length <= self.capacity);
377        self.messages.push_front(message);
378    }
379
380    /// Writes a message to the back of the message queue.
381    pub fn write_message(&mut self, message: Message<D>) {
382        self.length += message.len();
383        debug_assert!(self.length <= self.capacity);
384        self.messages.push_back(message);
385    }
386
387    pub fn query_events(&self) -> FdEvents {
388        let mut events = FdEvents::empty();
389        if self.available_capacity() > 0 {
390            events |= FdEvents::POLLOUT;
391        }
392        if !self.is_empty() {
393            events |= FdEvents::POLLIN;
394        }
395        events
396    }
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402    use crate::vfs::UnixControlData;
403
404    /// Tests that a write followed by a read returns the written message.
405    #[::fuchsia::test]
406    fn test_read_write() {
407        let mut message_queue = MessageQueue::new(usize::MAX);
408        let bytes: Vec<u8> = vec![1, 2, 3];
409        let message: Message = bytes.into();
410        message_queue.write_message(message.clone());
411        assert_eq!(message_queue.len(), 3);
412        assert_eq!(message_queue.read_message(), Some(message));
413        assert!(message_queue.is_empty());
414    }
415
416    /// Tests that ancillary data does not contribute to the message queue length.
417    #[::fuchsia::test]
418    fn test_control_len() {
419        let mut message_queue = MessageQueue::new(usize::MAX);
420        let bytes: Vec<u8> = vec![1, 2, 3];
421        let ancillary_data =
422            vec![AncillaryData::Unix(UnixControlData::Security(bytes.clone().into()))];
423        let message = Message::new(vec![].into(), None, ancillary_data);
424        message_queue.write_message(message);
425        assert_eq!(message_queue.len(), 0);
426        message_queue.write_message(bytes.clone().into());
427        assert_eq!(message_queue.len(), bytes.len());
428    }
429
430    /// Tests that multiple writes followed by multiple reads return the data in the correct order.
431    #[::fuchsia::test]
432    fn test_read_write_multiple() {
433        let mut message_queue = MessageQueue::new(usize::MAX);
434        let first_bytes: Vec<u8> = vec![1, 2, 3];
435        let second_bytes: Vec<u8> = vec![3, 4, 5];
436
437        for message in [first_bytes.clone().into(), second_bytes.clone().into()] {
438            message_queue.write_message(message);
439        }
440
441        assert_eq!(message_queue.len(), first_bytes.len() + second_bytes.len());
442        assert_eq!(message_queue.read_message(), Some(first_bytes.into()));
443        assert_eq!(message_queue.len(), second_bytes.len());
444        assert_eq!(message_queue.read_message(), Some(second_bytes.into()));
445        assert_eq!(message_queue.read_message(), None);
446    }
447}
starnix_core/vfs/buffers/message_queue.rs

starnix_core/vfs/buffers/
message_queue.rs
