Added documentation for aktoro-channel.

Author: r3v2d0g

aktoro-channel/src/builder.rs

@@ -10,72 +10,134 @@ use crate::queue::Queue;
 use crate::receiver::Receiver;
 use crate::sender::Sender;
 
+/// A configuration builder for a
+/// channel.
 pub struct Builder {
+    /// The capacity of the future
+    /// channel, or `None` if it
+    /// should be unbounded.
     cap: Option<usize>,
+    /// The limit of messages that
+    /// can be send over the channel,
+    /// or `None` if no limit should
+    /// be set.
     msgs: Option<usize>,
+    /// The limit of senders the
+    /// channel should be allowed to
+    /// have at the same time, or
+    /// `None` if no limit should
+    /// be set.
     senders: Option<usize>,
+    /// The limit of receivers the
+    /// channel should be allowed to
+    /// have at the same time, or
+    /// `None` if no limit should
+    /// be set.
     recvers: Option<usize>,
 }
 
 impl Builder {
+    /// Creates a new builder with
+    /// the default configuration:
+    /// - unbounded
+    /// - no limit of messages
+    /// - no limit of senders
+    /// - no limit of receivers
     pub fn new() -> Builder {
         Builder::default()
     }
 
+    /// Sets the channel to be created
+    /// as bounded with `cap` as its
+    /// buffer capacity.
     pub fn bounded(mut self, cap: usize) -> Builder {
         self.cap = Some(cap);
         self
     }
 
+    /// Sets the channel to be created
+    /// as unbounded.
     pub fn unbounded(mut self) -> Builder {
         self.cap = None;
         self
     }
 
+    /// Sets the maximum number of
+    /// messages that the channel will
+    /// be able to pass.
     pub fn limited_msgs(mut self, limit: usize) -> Builder {
         self.msgs = Some(limit);
         self
     }
 
+    /// Allows an infinite number of
+    /// messages to be sent over the
+    /// channel.
     pub fn unlimited_msgs(mut self) -> Builder {
         self.msgs = None;
         self
     }
 
+    /// Sets the maximum number of
+    /// senders that the channel will
+    /// be able to have at the same
+    /// time.
     pub fn limited_senders(mut self, limit: usize) -> Builder {
         self.senders = Some(limit);
         self
     }
 
+    /// Alows an inifinite number of
+    /// senders to be connected to
+    /// the channel at the same time.
     pub fn unlimited_senders(mut self) -> Builder {
         self.senders = None;
         self
     }
 
+    /// Sets the maximum number of
+    /// receivers that will be allowed
+    /// to be connected to the channel
+    /// at the same time.
     pub fn limited_receivers(mut self, limit: usize) -> Builder {
         self.recvers = Some(limit);
         self
     }
 
+    /// Allows an inifinite number of
+    /// receivers to be connected
+    /// to the channel at the same time.
     pub fn unlimited_receivers(mut self) -> Builder {
         self.recvers = None;
         self
     }
 
+    /// Builds the channel using the
+    /// specified configuration and
+    /// returning a sender and receiver
+    /// connected to it.
     pub fn build<T>(self) -> (Sender<T>, Receiver<T>) {
+        // We create either a bounded or
+        // unbounded queue.
         let queue = if let Some(cap) = self.cap {
             Queue::Bounded(ArrayQueue::new(cap))
         } else {
             Queue::Unbounded(SegQueue::new())
         };
 
+        // We create the channel and put
+        // it inside an atomically
+        // reference counted pointer.
         let channel = Arc::new(Channel {
             queue,
             closed: AtomicBool::new(false),
             counters: Counters::new(self.msgs, self.senders, self.recvers),
             wakers: SegQueue::new(),
         });
 
+        // We return a sender and a
+        // receiver containing a copy
+        // of the pointer.
         (Sender::new(channel.clone()), Receiver::new(channel))
     }
 }

aktoro-channel/src/channel.rs

@@ -9,6 +9,9 @@ use crate::error::*;
 use crate::message::Message;
 use crate::queue::Queue;
 
+/// A channel allowing senders to pass
+/// messages over it, and receivers to
+/// retrieve them.
 pub(crate) struct Channel<T> {
     pub(crate) queue: Queue<Message<T>>,
     pub(crate) closed: AtomicBool,
@@ -17,54 +20,87 @@ pub(crate) struct Channel<T> {
 }
 
 impl<T> Channel<T> {
+    /// Tries to send a message over the
+    /// channel.
     pub(crate) fn try_send(&self, msg: Message<T>) -> Result<(), TrySendError<T>> {
+        // If the channel has already
+        // been closed, we return an
+        // error.
         if self.is_closed() {
             return Err(TrySendError::closed(msg.msg));
         }
 
+        // If we couldn't increase the
+        // number of messages inside the
+        // inner counters, we return an
+        // error.
         if self.counters.add_msg().is_err() {
             return Err(TrySendError::limit(msg.msg));
         }
 
+        // We try to push the message
+        // over the queue.
         self.queue
             .push(msg)
             .map_err(|msg| TrySendError::full(msg.msg))?;
 
+        // We notify a receiver that a
+        // new message is available.
         self.notify();
 
         Ok(())
     }
 
+    /// Tries to receive a message from the
+    /// channel if one is available.
     pub(crate) fn try_recv(&self) -> Result<Option<Message<T>>, TryRecvError> {
+        // If the queue is empty, we
+        // return an error if it's closed.
         if self.queue.is_empty() {
             if self.check_is_closed() {
                 Err(TryRecvError::closed())
             } else {
                 Ok(None)
             }
+        // Otherwise, we pop try to
+        // pop a message from it (it
+        // could return `None` if the
+        // message was already poped).
         } else {
             Ok(self.queue.pop())
         }
     }
 
+    /// Registers a new waker to be
+    /// notified when a new message is
+    /// available.
     pub(crate) fn register(&self, waker: Waker) {
         self.wakers.push(waker);
     }
 
+    /// Notifies a waker if one is
+    /// available.
     fn notify(&self) {
         if let Ok(waker) = self.wakers.pop() {
             waker.wake();
         }
     }
 
+    /// Whether the queue is empty.
     pub(crate) fn is_empty(&self) -> bool {
         self.queue.is_empty()
     }
 
+    /// Whether the channel has been
+    /// closed.
     pub(crate) fn is_closed(&self) -> bool {
         self.closed.load(Ordering::SeqCst)
     }
 
+    /// Verifies whether the channel has
+    /// been closed and checks if senders
+    /// are still connected to it (closing
+    /// the channel if not).
     pub(crate) fn check_is_closed(&self) -> bool {
         if self.is_closed() {
             return true;
@@ -78,6 +114,7 @@ impl<T> Channel<T> {
         }
     }
 
+    /// Closes the channel.
     pub(crate) fn close(&self) {
         self.closed.store(true, Ordering::SeqCst);
     }

aktoro-channel/src/counters.rs

@@ -1,17 +1,37 @@
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 
+/// Counters used to store the number
+/// of senders, receivers, messages
+/// sent and the channel's limits.
 pub(crate) struct Counters {
+    /// The number of messages sent
+    /// over the channel.
     cmsgs: Option<AtomicUsize>,
+    /// The number of senders
+    /// connected to the channel.
     csenders: AtomicUsize,
+    /// The number of receivers
+    /// connected to the channel.
     crecvers: AtomicUsize,
 
+    /// The number of messages that
+    /// can be sent over the channel
+    /// (in total).
     lmsgs: Option<usize>,
+    /// The maximum number of senders
+    /// that can be connected to the
+    /// channel.
     lsenders: Option<usize>,
+    /// The maximum number of
+    /// receivers that can be connected
+    /// to the channel.
     lrecvers: Option<usize>,
 }
 
 impl Counters {
+    /// Creates counters with the
+    /// specified limits.
     pub(crate) fn new(msgs: Option<usize>, senders: Option<usize>, recvers: Option<usize>) -> Self {
         Counters {
             cmsgs: msgs.map(|_| AtomicUsize::new(0)),
@@ -24,14 +44,23 @@ impl Counters {
         }
     }
 
+    /// Gets the current number of
+    /// senders connected to the channel.
     pub(crate) fn senders(&self) -> usize {
         self.csenders.load(Ordering::SeqCst)
     }
 
+    /// Increases the total number of
+    /// messages sent over the channel if
+    /// necessary.
     pub(crate) fn add_msg(&self) -> Result<(), ()> {
         if let Some(counter) = &self.cmsgs {
             let limit = self.lmsgs.unwrap();
 
+            // We CAS the sent messages
+            // counter to increase it of
+            // 1 or return an error if it
+            // is above the limit.
             loop {
                 let cur = counter.load(Ordering::SeqCst);
                 let new = cur + 1;
@@ -49,7 +78,15 @@ impl Counters {
         Ok(())
     }
 
+    /// Increases the counter for the
+    /// number of senders connected to the
+    /// channel.
     pub(crate) fn add_sender(&self) -> Result<(), ()> {
+        // If there is a limit for the number
+        // of senders connected to the channel,
+        // we CAS the counter to increase it of
+        // 1 and return an error if it is
+        // above the limit.
         if let Some(limit) = &self.lsenders {
             loop {
                 let cur = self.csenders.load(Ordering::SeqCst);
@@ -70,7 +107,15 @@ impl Counters {
         Ok(())
     }
 
+    /// Increases the counter for the
+    /// number of receivers connected to
+    /// the channel.
     pub(crate) fn add_recver(&self) -> Result<(), ()> {
+        // If there is a limit for the number
+        // or receivers that can be connected
+        // to the channel, we CAS the counter
+        // to increase it of 1 and return an
+        // error if it is above the limit.
         if let Some(limit) = &self.lrecvers {
             loop {
                 let cur = self.crecvers.load(Ordering::SeqCst);
@@ -91,6 +136,8 @@ impl Counters {
         Ok(())
     }
 
+    /// Decreases the number of senders,
+    /// returning the updated number.
     pub(crate) fn sub_sender(&self) -> usize {
         loop {
             let cur = self.csenders.load(Ordering::SeqCst);
@@ -105,6 +152,8 @@ impl Counters {
         }
     }
 
+    /// Decreases the number of receivers,
+    /// returning the updated number.
     pub(crate) fn sub_recver(&self) -> usize {
         loop {
             let cur = self.crecvers.load(Ordering::SeqCst);