Introduce disconnection state for channels #1114

ryoqun · 2024-05-22T07:17:18Z

(hopefully) much improved pr than #959 ....

Currently, crossbeam-channel completely overlaps the channel connected status with rust sender/receiver instance aliveness. While this is uncontroversial and covers most use cases, it prevents advanced channel management.

Namely:

graceful receiver-initiated shutdown (notify disconnection from receiver to sender and drain all sent-not-received messages): see description of Add receiver.disconnect() for atomic graceful shutdown #959
temporal non-blocking faults (make {receiver,sender} disconnected until new {sender,receiver} is needed): see Add Receiver::new_sender to channels #750
off-loaded channel destruction (drop retired receivers outside performance sensitive code-path, which possibly could contain large number of not-received buffered messages while notifying disconnection to senders): see this comment: Add receiver.disconnect() for atomic graceful shutdown #959 (comment)

While these seem to be unrelated to each other, the underlying problem is the lack of explicit channel disconnection state as described above.

Thereby, this adds the following new methods:

impl Sender {
    fn disconnect(&self) -> bool;
    fn connect(&self) -> bool;
}
impl Receiver {
    fn disconnect(&self) -> bool;
    fn connect(&self) -> bool;
}

Fortunately, the implementation is quite simple by decoupling the already-existing disconnected bit (MARK_BIT in tail) into the two: one for ::drop() as was before, one for new ::disconnect(), although this pr touches the heart of the impl unlike #959 (i.e. more intrusive....). Also, the almost all of existing code is already robust enough to introduce the 2 kinds of disconnected status. Come to think of it, {receivers, senders} should be able to cope with disconnected {senders, receivers} at any given moment of time due to drops.

As for performance hit, it should be negligible because there's no additional operation at all. just << 2 instead of << 1 in the performance sensitive code-path. Also, no space increse. There were design considerations.

ryoqun · 2024-05-22T07:19:44Z

crossbeam-channel/src/channel.rs

+    pub fn disconnect(&self) -> bool {
+        match &self.flavor {
+            SenderFlavor::List(chan) => chan.disconnect(),
+            _ => todo!(),


obviously, i need to add support other flavors. this is true for other 3 todo!()s.

I'll do this once i get some confirmation of the intent of merging this pr without much change.

ryoqun · 2024-05-22T07:56:13Z

crossbeam-channel/src/flavors/list.rs


-        if tail & MARK_BIT == 0 {
+        if tail & TAIL_DISCONNECT_ANY == 0 {


the use of TAIL_DISCONNECT_ANY, instead of TAIL_DISCONNECT_EXPLICIT is intentional and needs a comment...

ryoqun · 2024-05-22T07:58:01Z

@taiki-e learning from #1040 , this pr has finished impl for the list flavor, added tests, also documented fully. CI should pass. Could you review this in your free time? Thanks a lot.

ryoqun · 2024-05-22T08:02:59Z

crossbeam-channel/src/flavors/list.rs

-const SHIFT: usize = 1;
-// Has two different purposes:
-// * If set in head, indicates that the block is not the last one.
-// * If set in tail, indicates that the channel is disconnected.
-const MARK_BIT: usize = 1;


btw, let me know if this flag dedup should be done as a preparatory pr.

ryoqun · 2024-05-22T08:06:36Z

crossbeam-channel/src/channel.rs

+    /// `false` as well, if this method is called on channels other than bounded-capacity,
+    /// unbounded-capacity or zero-capacity channels.


not coded yet. but i hope this isn't controversial behavior... ;)

ryoqun · 2024-05-22T08:07:04Z

crossbeam-channel/src/channel.rs

+    /// connected. Otherwise, this method does nothing and returns `false`. Also, this method does
+    /// nothing and returns `false` as well, if this method is called on channels other than
+    /// bounded-capacity, unbounded-capacity or zero-capacity channels.
+


oops, remove this empty line later....

ryoqun · 2024-05-22T08:08:00Z

crossbeam-channel/src/flavors/list.rs

@@ -628,13 +662,13 @@ impl<T> Channel<T> {
                drop(Box::from_raw(block));
            }
        }
-        head &= !MARK_BIT;
+        head &= !HEAD_NOT_LAST;
        self.head.index.store(head, Ordering::Release);
    }

    /// Returns `true` if the channel is disconnected.
    pub(crate) fn is_disconnected(&self) -> bool {


I guess it's a prime time to expose this to public api? (i know there were controversial in the past..)

ryoqun · 2024-05-23T01:11:35Z

crossbeam-channel/src/channel.rs

+    ///
+    /// Disconnected channels can be restored connected by calling [`connect`], as long as there
+    /// are at least a sender and a receiver for the channel.
+    ///


i guess i need to add some motivating example here for the justification of disconnect()...

ryoqun · 2024-05-27T01:13:09Z

crossbeam-channel/src/channel.rs

+    /// Connected channels can be disconnected again by calling [`disconnect`].
+    ///
+    /// [`disconnect`]: Self::disconnect
+    pub fn connect(&self) -> bool {


note to self: rename to reconnect()?

ryoqun · 2024-05-27T01:21:05Z

crossbeam-channel/tests/list.rs

@@ -580,3 +580,84 @@ fn channel_through_channel() {
    })
    .unwrap();
 }
+
+#[test]


maybe better off writing a test for possible race condition, where a sending thread disconnect/reconnect/send(T) successively , and the receiver thread may/may not see disconnected, or actual message.

ryoqun · 2024-05-27T12:33:06Z

crossbeam-channel/tests/list.rs

+    // connect should fail after all receivers has gone
+    assert!(!s.connect());


also disconnect should fail after all receivers has gone even if the channel isn't disconnected yet.

ryoqun · 2024-05-27T12:33:45Z

crossbeam-channel/src/channel.rs

+    ///
+    /// Disconnected channels can be restored connected by calling [`connect`], as long as there
+    /// are at least a sender and a receiver for the channel.
+    ///


maybe mention about race condition among other disconnect()s and drop of the last instance of sender/receiver...

ryoqun · 2024-05-28T06:05:32Z

crossbeam-channel/src/flavors/list.rs

-// * If set in head, indicates that the block is not the last one.
-// * If set in tail, indicates that the channel is disconnected.
-const MARK_BIT: usize = 1;
+const SHIFT: usize = 2;


(note to self): HEAD_SHIFT = 1 and TAIL_SHIFT = 2?

ryoqun · 2024-05-30T06:38:37Z

@taiki-e friendly ping. although i self-commented some after opening this pr, needing further work, I still think this pr is in some good standing for initial round of code-review. please let me know if i can do anything to moving forward this pr.

ryoqun · 2024-06-15T05:10:19Z

@taiki-e hey, another ping from me. just want to let you know I'm very eager to work on this pr.

esemeniuc · 2024-06-28T15:56:15Z

bump @taiki-e

Introduce disconnection state for channels

613fe7e

ryoqun commented May 22, 2024

View reviewed changes

ryoqun marked this pull request as ready for review May 22, 2024 07:55

ryoqun commented May 22, 2024

View reviewed changes

ryoqun mentioned this pull request May 22, 2024

PoH - Reserve Space anza-xyz/agave#1227

Open

ryoqun commented May 23, 2024

View reviewed changes

ryoqun mentioned this pull request May 23, 2024

Add receiver.disconnect() for atomic graceful shutdown #959

Closed

ryoqun commented May 27, 2024

View reviewed changes

ryoqun commented May 28, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce disconnection state for channels #1114

Introduce disconnection state for channels #1114

ryoqun commented May 22, 2024 •

edited

Loading

ryoqun May 22, 2024 •

edited

Loading

ryoqun May 22, 2024

ryoqun commented May 22, 2024 •

edited

Loading

ryoqun May 22, 2024

ryoqun May 22, 2024

ryoqun May 22, 2024

ryoqun May 22, 2024 •

edited

Loading

ryoqun May 23, 2024

ryoqun May 27, 2024

ryoqun May 27, 2024

ryoqun May 27, 2024

ryoqun May 27, 2024

ryoqun May 28, 2024

ryoqun commented May 30, 2024

ryoqun commented Jun 15, 2024

esemeniuc commented Jun 28, 2024


		if tail & MARK_BIT == 0 {
		if tail & TAIL_DISCONNECT_ANY == 0 {

		/// `false` as well, if this method is called on channels other than bounded-capacity,
		/// unbounded-capacity or zero-capacity channels.

		// connect should fail after all receivers has gone
		assert!(!s.connect());

Introduce disconnection state for channels #1114

Are you sure you want to change the base?

Introduce disconnection state for channels #1114

Conversation

ryoqun commented May 22, 2024 • edited Loading

ryoqun May 22, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryoqun commented May 22, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryoqun May 22, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryoqun commented May 30, 2024

ryoqun commented Jun 15, 2024

esemeniuc commented Jun 28, 2024

ryoqun commented May 22, 2024 •

edited

Loading

ryoqun May 22, 2024 •

edited

Loading

ryoqun commented May 22, 2024 •

edited

Loading

ryoqun May 22, 2024 •

edited

Loading