v1.16: geyser: use Ordering::Relaxed instead of SeqCst (backport of #221

) (#222) (cherry picked from commit f88fd64) Co-authored-by: Kirill Fomichev <[email protected]>
rpcpool · Oct 27, 2023 · 94f2330 · 94f2330
1 parent 60fd038
commit 94f2330
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 14 deletions.
diff --git 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
@@ -1163,7 +1163,7 @@ impl Geyser for GrpcService {
         &self,
         mut request: Request<Streaming<SubscribeRequest>>,
     ) -> TonicResult<Response<Self::SubscribeStream>> {
-        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
@@ -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,17 +153,17 @@ 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(),
                 Err(_) => panic!("failed to send message to startup queue: channel closed"),
             }
         }
 
-        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
@@ -103,7 +103,7 @@ impl Geyser for GrpcService {
         &self,
         mut request: Request<Streaming<SubscribeRequest>>,
     ) -> TonicResult<Response<Self::SubscribeStream>> {
-        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());