diff --git a/CHANGELOG.md b/CHANGELOG.md index 1919ea9a..ac61cb2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ The minor version will be incremented upon a breaking change and the patch versi - geyser: trigger end of startup when parent slot 0 seen in `update_slot_status` notification because `notify_end_of_startup` is not triggered when cluster started from genesis ([#207](https://github.com/rpcpool/yellowstone-grpc/pull/207)) - tools: correctly handle SIGINT in kafka ([#219](https://github.com/rpcpool/yellowstone-grpc/pull/219)) +- geyser: use Ordering::Relaxed instead of SeqCst ([#221](https://github.com/rpcpool/yellowstone-grpc/pull/221)) ### Features diff --git a/yellowstone-grpc-geyser/src/grpc.rs b/yellowstone-grpc-geyser/src/grpc.rs index a7e36d26..64e0ca34 100644 --- a/yellowstone-grpc-geyser/src/grpc.rs +++ b/yellowstone-grpc-geyser/src/grpc.rs @@ -1163,7 +1163,7 @@ impl Geyser for GrpcService { &self, mut request: Request>, ) -> TonicResult> { - let id = self.subscribe_id.fetch_add(1, Ordering::SeqCst); + let id = self.subscribe_id.fetch_add(1, Ordering::Relaxed); let filter = Filter::new( &SubscribeRequest { accounts: HashMap::new(), diff --git a/yellowstone-grpc-geyser/src/plugin.rs b/yellowstone-grpc-geyser/src/plugin.rs index 2004b536..8e0bcaa6 100644 --- a/yellowstone-grpc-geyser/src/plugin.rs +++ b/yellowstone-grpc-geyser/src/plugin.rs @@ -55,7 +55,7 @@ impl Plugin { { // Full block reconstruction will fail before first processed slot received let inner = self.inner.as_ref().expect("initialized"); - if inner.startup_status.load(Ordering::SeqCst) + if inner.startup_status.load(Ordering::Relaxed) == STARTUP_END_OF_RECEIVED | STARTUP_PROCESSED_RECEIVED { f(inner) @@ -153,6 +153,10 @@ impl GeyserPlugin for Plugin { fn notify_end_of_startup(&self) -> PluginResult<()> { let inner = self.inner.as_ref().expect("initialized"); + inner + .startup_status + .fetch_or(STARTUP_END_OF_RECEIVED, Ordering::Relaxed); + if let Some(channel) = &inner.snapshot_channel { match channel.send(None) { Ok(()) => MESSAGE_QUEUE_SIZE.inc(), @@ -160,10 +164,6 @@ impl GeyserPlugin for Plugin { } } - inner - .startup_status - .fetch_or(STARTUP_END_OF_RECEIVED, Ordering::SeqCst); - Ok(()) } @@ -179,14 +179,15 @@ impl GeyserPlugin for Plugin { if parent == Some(0) { inner .startup_status - .fetch_or(STARTUP_END_OF_RECEIVED, Ordering::SeqCst); + .fetch_or(STARTUP_END_OF_RECEIVED, Ordering::Relaxed); } - if inner.startup_status.load(Ordering::SeqCst) == STARTUP_END_OF_RECEIVED - && status == SlotStatus::Processed - { - inner - .startup_status - .fetch_or(STARTUP_PROCESSED_RECEIVED, Ordering::SeqCst); + if status == SlotStatus::Processed { + let _ = inner.startup_status.compare_exchange( + STARTUP_END_OF_RECEIVED, + STARTUP_END_OF_RECEIVED | STARTUP_PROCESSED_RECEIVED, + Ordering::Relaxed, + Ordering::Relaxed, + ); } self.with_inner(|inner| { diff --git a/yellowstone-grpc-tools/src/kafka/grpc.rs b/yellowstone-grpc-tools/src/kafka/grpc.rs index c8772e50..bb9997ca 100644 --- a/yellowstone-grpc-tools/src/kafka/grpc.rs +++ b/yellowstone-grpc-tools/src/kafka/grpc.rs @@ -103,7 +103,7 @@ impl Geyser for GrpcService { &self, mut request: Request>, ) -> TonicResult> { - let id = self.subscribe_id.fetch_add(1, Ordering::SeqCst); + let id = self.subscribe_id.fetch_add(1, Ordering::Relaxed); let (stream_tx, stream_rx) = mpsc::channel(self.channel_capacity); let notify_client = Arc::new(Notify::new()); let notify_exit1 = Arc::new(Notify::new());