Skip to content

Commit e2f54ea

Browse files
ibraheemdevibraheemdev
committed
use blocking state in select macro
1 parent 86debba commit e2f54ea

File tree

13 files changed

+314
-108
lines changed

13 files changed

+314
-108
lines changed

‎crossbeam-channel/src/channel.rs‎

Lines changed: 42 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ use crate::err::{
1414
};
1515
use crate::flavors;
1616
use crate::select::{Operation, SelectHandle, Token};
17+
use crate::waker::BlockingState;
1718

1819
/// Creates a multi-producer multi-consumer channel of unbounded capacity.
1920
///
@@ -1358,6 +1359,14 @@ impl<T> fmt::Debug for IntoIter<T> {
13581359
}
13591360

13601361
impl<T> SelectHandle for Sender<T> {
1362+
fn start(&self) -> Option<BlockingState<'_>> {
1363+
match &self.flavor {
1364+
SenderFlavor::Array(chan) => chan.sender().start_ref(),
1365+
SenderFlavor::List(chan) => chan.sender().start_ref(),
1366+
SenderFlavor::Zero(chan) => chan.start(),
1367+
}
1368+
}
1369+
13611370
fn try_select(&self, token: &mut Token) -> bool {
13621371
match &self.flavor {
13631372
SenderFlavor::Array(chan) => chan.sender().try_select(token),
@@ -1370,11 +1379,11 @@ impl<T> SelectHandle for Sender<T> {
13701379
None
13711380
}
13721381

1373-
fn register(&self, oper: Operation, cx: &Context) -> bool {
1382+
fn register(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
13741383
match &self.flavor {
1375-
SenderFlavor::Array(chan) => chan.sender().register(oper, cx),
1376-
SenderFlavor::List(chan) => chan.sender().register(oper, cx),
1377-
SenderFlavor::Zero(chan) => chan.sender().register(oper, cx),
1384+
SenderFlavor::Array(chan) => chan.sender().register(oper, cx, state),
1385+
SenderFlavor::List(chan) => chan.sender().register(oper, cx, state),
1386+
SenderFlavor::Zero(chan) => chan.sender().register(oper, cx, state),
13781387
}
13791388
}
13801389

@@ -1402,11 +1411,11 @@ impl<T> SelectHandle for Sender<T> {
14021411
}
14031412
}
14041413

1405-
fn watch(&self, oper: Operation, cx: &Context) -> bool {
1414+
fn watch(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
14061415
match &self.flavor {
1407-
SenderFlavor::Array(chan) => chan.sender().watch(oper, cx),
1408-
SenderFlavor::List(chan) => chan.sender().watch(oper, cx),
1409-
SenderFlavor::Zero(chan) => chan.sender().watch(oper, cx),
1416+
SenderFlavor::Array(chan) => chan.sender().watch(oper, cx, state),
1417+
SenderFlavor::List(chan) => chan.sender().watch(oper, cx, state),
1418+
SenderFlavor::Zero(chan) => chan.sender().watch(oper, cx, state),
14101419
}
14111420
}
14121421

@@ -1420,6 +1429,17 @@ impl<T> SelectHandle for Sender<T> {
14201429
}
14211430

14221431
impl<T> SelectHandle for Receiver<T> {
1432+
fn start(&self) -> Option<BlockingState<'_>> {
1433+
match &self.flavor {
1434+
ReceiverFlavor::Array(chan) => chan.receiver().start_ref(),
1435+
ReceiverFlavor::List(chan) => chan.receiver().start_ref(),
1436+
ReceiverFlavor::Zero(chan) => chan.start(),
1437+
ReceiverFlavor::At(chan) => chan.start(),
1438+
ReceiverFlavor::Tick(chan) => chan.start(),
1439+
ReceiverFlavor::Never(chan) => chan.start(),
1440+
}
1441+
}
1442+
14231443
fn try_select(&self, token: &mut Token) -> bool {
14241444
match &self.flavor {
14251445
ReceiverFlavor::Array(chan) => chan.receiver().try_select(token),
@@ -1442,14 +1462,14 @@ impl<T> SelectHandle for Receiver<T> {
14421462
}
14431463
}
14441464

1445-
fn register(&self, oper: Operation, cx: &Context) -> bool {
1465+
fn register(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
14461466
match &self.flavor {
1447-
ReceiverFlavor::Array(chan) => chan.receiver().register(oper, cx),
1448-
ReceiverFlavor::List(chan) => chan.receiver().register(oper, cx),
1449-
ReceiverFlavor::Zero(chan) => chan.receiver().register(oper, cx),
1450-
ReceiverFlavor::At(chan) => chan.register(oper, cx),
1451-
ReceiverFlavor::Tick(chan) => chan.register(oper, cx),
1452-
ReceiverFlavor::Never(chan) => chan.register(oper, cx),
1467+
ReceiverFlavor::Array(chan) => chan.receiver().register(oper, cx, state),
1468+
ReceiverFlavor::List(chan) => chan.receiver().register(oper, cx, state),
1469+
ReceiverFlavor::Zero(chan) => chan.receiver().register(oper, cx, state),
1470+
ReceiverFlavor::At(chan) => chan.register(oper, cx, state),
1471+
ReceiverFlavor::Tick(chan) => chan.register(oper, cx, state),
1472+
ReceiverFlavor::Never(chan) => chan.register(oper, cx, state),
14531473
}
14541474
}
14551475

@@ -1486,14 +1506,14 @@ impl<T> SelectHandle for Receiver<T> {
14861506
}
14871507
}
14881508

1489-
fn watch(&self, oper: Operation, cx: &Context) -> bool {
1509+
fn watch(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
14901510
match &self.flavor {
1491-
ReceiverFlavor::Array(chan) => chan.receiver().watch(oper, cx),
1492-
ReceiverFlavor::List(chan) => chan.receiver().watch(oper, cx),
1493-
ReceiverFlavor::Zero(chan) => chan.receiver().watch(oper, cx),
1494-
ReceiverFlavor::At(chan) => chan.watch(oper, cx),
1495-
ReceiverFlavor::Tick(chan) => chan.watch(oper, cx),
1496-
ReceiverFlavor::Never(chan) => chan.watch(oper, cx),
1511+
ReceiverFlavor::Array(chan) => chan.receiver().watch(oper, cx, state),
1512+
ReceiverFlavor::List(chan) => chan.receiver().watch(oper, cx, state),
1513+
ReceiverFlavor::Zero(chan) => chan.receiver().watch(oper, cx, state),
1514+
ReceiverFlavor::At(chan) => chan.watch(oper, cx, state),
1515+
ReceiverFlavor::Tick(chan) => chan.watch(oper, cx, state),
1516+
ReceiverFlavor::Never(chan) => chan.watch(oper, cx, state),
14971517
}
14981518
}
14991519

‎crossbeam-channel/src/flavors/array.rs‎

Lines changed: 37 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ use crossbeam_utils::{Backoff, CachePadded};
2020
use crate::context::Context;
2121
use crate::err::{RecvTimeoutError, SendTimeoutError, TryRecvError, TrySendError};
2222
use crate::select::{Operation, SelectHandle, Selected, Token};
23-
use crate::waker::SyncWaker;
23+
use crate::waker::{BlockingState, SyncWaker};
2424

2525
/// A slot in a channel.
2626
struct Slot<T> {
@@ -401,7 +401,7 @@ impl<T> Channel<T> {
401401
Context::with(|cx| {
402402
// Prepare for blocking until a receiver wakes us up.
403403
let oper = Operation::hook(token);
404-
self.senders.register2(oper, cx, &state);
404+
self.senders.register(oper, cx, &state);
405405

406406
// Has the channel become ready just now?
407407
if !self.is_full() || self.is_disconnected() {
@@ -478,7 +478,7 @@ impl<T> Channel<T> {
478478
Context::with(|cx| {
479479
// Prepare for blocking until a sender wakes us up.
480480
let oper = Operation::hook(token);
481-
self.receivers.register2(oper, cx, &mut state);
481+
self.receivers.register(oper, cx, &state);
482482

483483
// Has the channel become ready just now?
484484
if !self.is_empty() || self.is_disconnected() {
@@ -629,7 +629,18 @@ pub(crate) struct Receiver<'a, T>(&'a Channel<T>);
629629
/// Sender handle to a channel.
630630
pub(crate) struct Sender<'a, T>(&'a Channel<T>);
631631

632+
impl<'a, T> Receiver<'a, T> {
633+
/// Same as `SelectHandle::start`, but with a more specific lifetime.
634+
pub(crate) fn start_ref(&self) -> Option<BlockingState<'a>> {
635+
Some(self.0.receivers.start())
636+
}
637+
}
638+
632639
impl<T> SelectHandle for Receiver<'_, T> {
640+
fn start(&self) -> Option<BlockingState<'_>> {
641+
self.start_ref()
642+
}
643+
633644
fn try_select(&self, token: &mut Token) -> bool {
634645
self.0.start_recv(token) == Status::Ready
635646
}
@@ -638,8 +649,9 @@ impl<T> SelectHandle for Receiver<'_, T> {
638649
None
639650
}
640651

641-
fn register(&self, oper: Operation, cx: &Context) -> bool {
642-
self.0.receivers.register(oper, cx);
652+
fn register(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
653+
let state = state.expect("Receiver::start returns blocking state");
654+
self.0.receivers.register(oper, cx, state);
643655
self.is_ready()
644656
}
645657

@@ -655,8 +667,9 @@ impl<T> SelectHandle for Receiver<'_, T> {
655667
!self.0.is_empty() || self.0.is_disconnected()
656668
}
657669

658-
fn watch(&self, oper: Operation, cx: &Context) -> bool {
659-
self.0.receivers.watch(oper, cx);
670+
fn watch(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
671+
let state = state.expect("Receiver::start returns blocking state");
672+
self.0.receivers.watch(oper, cx, state);
660673
self.is_ready()
661674
}
662675

@@ -665,7 +678,18 @@ impl<T> SelectHandle for Receiver<'_, T> {
665678
}
666679
}
667680

681+
impl<'a, T> Sender<'a, T> {
682+
/// Same as `SelectHandle::start`, but with a more specific lifetime.
683+
pub(crate) fn start_ref(&self) -> Option<BlockingState<'a>> {
684+
Some(self.0.senders.start())
685+
}
686+
}
687+
668688
impl<T> SelectHandle for Sender<'_, T> {
689+
fn start(&self) -> Option<BlockingState<'_>> {
690+
self.start_ref()
691+
}
692+
669693
fn try_select(&self, token: &mut Token) -> bool {
670694
self.0.start_send(token) == Status::Ready
671695
}
@@ -674,8 +698,9 @@ impl<T> SelectHandle for Sender<'_, T> {
674698
None
675699
}
676700

677-
fn register(&self, oper: Operation, cx: &Context) -> bool {
678-
self.0.senders.register(oper, cx);
701+
fn register(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
702+
let state = state.expect("Sender::start returns blocking state");
703+
self.0.senders.register(oper, cx, state);
679704
self.is_ready()
680705
}
681706

@@ -691,8 +716,9 @@ impl<T> SelectHandle for Sender<'_, T> {
691716
!self.0.is_full() || self.0.is_disconnected()
692717
}
693718

694-
fn watch(&self, oper: Operation, cx: &Context) -> bool {
695-
self.0.senders.watch(oper, cx);
719+
fn watch(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
720+
let state = state.expect("Sender::start returns blocking state");
721+
self.0.senders.watch(oper, cx, state);
696722
self.is_ready()
697723
}
698724

‎crossbeam-channel/src/flavors/at.rs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ use crate::context::Context;
1010
use crate::err::{RecvTimeoutError, TryRecvError};
1111
use crate::select::{Operation, SelectHandle, Token};
1212
use crate::utils;
13+
use crate::waker::BlockingState;
1314

1415
/// Result of a receive operation.
1516
pub(crate) type AtToken = Option<Instant>;
@@ -140,6 +141,10 @@ impl Channel {
140141
}
141142

142143
impl SelectHandle for Channel {
144+
fn start(&self) -> Option<BlockingState<'_>> {
145+
None
146+
}
147+
143148
#[inline]
144149
fn try_select(&self, token: &mut Token) -> bool {
145150
match self.try_recv() {
@@ -166,7 +171,12 @@ impl SelectHandle for Channel {
166171
}
167172

168173
#[inline]
169-
fn register(&self, _oper: Operation, _cx: &Context) -> bool {
174+
fn register(
175+
&self,
176+
_oper: Operation,
177+
_cx: &Context,
178+
_state: Option<&BlockingState<'_>>,
179+
) -> bool {
170180
self.is_ready()
171181
}
172182

@@ -184,7 +194,7 @@ impl SelectHandle for Channel {
184194
}
185195

186196
#[inline]
187-
fn watch(&self, _oper: Operation, _cx: &Context) -> bool {
197+
fn watch(&self, _oper: Operation, _cx: &Context, _state: Option<&BlockingState<'_>>) -> bool {
188198
self.is_ready()
189199
}
190200

‎crossbeam-channel/src/flavors/list.rs‎

Lines changed: 38 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ use crossbeam_utils::{Backoff, CachePadded};
1313
use crate::context::Context;
1414
use crate::err::{RecvTimeoutError, SendTimeoutError, TryRecvError, TrySendError};
1515
use crate::select::{Operation, SelectHandle, Selected, Token};
16-
use crate::waker::SyncWaker;
16+
use crate::waker::{BlockingState, SyncWaker};
1717

1818
// TODO(stjepang): Once we bump the minimum required Rust version to 1.28 or newer, re-apply the
1919
// following changes by @kleimkuhler:
@@ -510,7 +510,7 @@ impl<T> Channel<T> {
510510
// Prepare for blocking until a sender wakes us up.
511511
Context::with(|cx| {
512512
let oper = Operation::hook(token);
513-
self.receivers.register2(oper, cx, &state);
513+
self.receivers.register(oper, cx, &state);
514514

515515
// Has the channel become ready just now?
516516
if !self.is_empty() || self.is_disconnected() {
@@ -734,7 +734,18 @@ pub(crate) struct Receiver<'a, T>(&'a Channel<T>);
734734
/// Sender handle to a channel.
735735
pub(crate) struct Sender<'a, T>(&'a Channel<T>);
736736

737+
impl<'a, T> Receiver<'a, T> {
738+
/// Same as `SelectHandle::start`, but with a more specific lifetime.
739+
pub(crate) fn start_ref(&self) -> Option<BlockingState<'a>> {
740+
Some(self.0.receivers.start())
741+
}
742+
}
743+
737744
impl<T> SelectHandle for Receiver<'_, T> {
745+
fn start(&self) -> Option<BlockingState<'_>> {
746+
self.start_ref()
747+
}
748+
738749
fn try_select(&self, token: &mut Token) -> bool {
739750
self.0.start_recv(token) == Status::Ready
740751
}
@@ -743,8 +754,9 @@ impl<T> SelectHandle for Receiver<'_, T> {
743754
None
744755
}
745756

746-
fn register(&self, oper: Operation, cx: &Context) -> bool {
747-
self.0.receivers.register(oper, cx);
757+
fn register(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
758+
let state = state.expect("Receiver::start returns blocking state");
759+
self.0.receivers.register(oper, cx, state);
748760
self.is_ready()
749761
}
750762

@@ -760,8 +772,9 @@ impl<T> SelectHandle for Receiver<'_, T> {
760772
!self.0.is_empty() || self.0.is_disconnected()
761773
}
762774

763-
fn watch(&self, oper: Operation, cx: &Context) -> bool {
764-
self.0.receivers.watch(oper, cx);
775+
fn watch(&self, oper: Operation, cx: &Context, state: Option<&BlockingState<'_>>) -> bool {
776+
let state = state.expect("Receiver::start returns blocking state");
777+
self.0.receivers.watch(oper, cx, state);
765778
self.is_ready()
766779
}
767780

@@ -770,7 +783,18 @@ impl<T> SelectHandle for Receiver<'_, T> {
770783
}
771784
}
772785

773-
impl<T> SelectHandle for Sender<'_, T> {
786+
impl<'a, T> Sender<'a, T> {
787+
/// Same as `SelectHandle::start`, but with a more specific lifetime.
788+
pub(crate) fn start_ref(&self) -> Option<BlockingState<'a>> {
789+
None
790+
}
791+
}
792+
793+
impl<'a, T> SelectHandle for Sender<'a, T> {
794+
fn start(&self) -> Option<BlockingState<'a>> {
795+
None
796+
}
797+
774798
fn try_select(&self, token: &mut Token) -> bool {
775799
self.0.start_send(token)
776800
}
@@ -779,7 +803,12 @@ impl<T> SelectHandle for Sender<'_, T> {
779803
None
780804
}
781805

782-
fn register(&self, _oper: Operation, _cx: &Context) -> bool {
806+
fn register(
807+
&self,
808+
_oper: Operation,
809+
_cx: &Context,
810+
_state: Option<&BlockingState<'_>>,
811+
) -> bool {
783812
self.is_ready()
784813
}
785814

@@ -793,7 +822,7 @@ impl<T> SelectHandle for Sender<'_, T> {
793822
true
794823
}
795824

796-
fn watch(&self, _oper: Operation, _cx: &Context) -> bool {
825+
fn watch(&self, _oper: Operation, _cx: &Context, _state: Option<&BlockingState<'_>>) -> bool {
797826
self.is_ready()
798827
}
799828

0 commit comments

Comments
 (0)