geyser: use Ordering::Relaxed instead of SeqCst (#221)

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
 

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(),

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| {

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());