geyser: use runtime instaed of unconstrained

rpcpool · May 5, 2024 · 6c21de5 · 6c21de5
1 parent 4629f17
commit 6c21de5
Show file tree

Hide file tree

Showing 5 changed files with 46 additions and 29 deletions.
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -78,6 +78,24 @@ jobs:
       - name: Build release tarball
         run: ./ci/create-tarball.sh
 
+      - name: Deleteing directories to avoid upload conflict
+        run: |
+          rm -rf \
+            target/release/client.d \
+            target/release/config-check.d \
+            target/release/grpc-google-pubsub.d \
+            target/release/grpc-kafka.d \
+            target/release/grpc-scylladb.d
+
+      - name: Remove debug information
+        run: |
+          strip \
+            target/release/client \
+            target/release/config-check \
+            target/release/grpc-google-pubsub \
+            target/release/grpc-kafka \
+            target/release/grpc-scylladb
+
       - name: Rename binaries
         run: |
           mv target/release/client target/release/client-${{ matrix.os }}
@@ -92,15 +110,6 @@ jobs:
           mv ${{ env.GEYSER_PLUGIN_NAME }}-release-x86_64-unknown-linux-gnu.tar.bz2 ${{ env.GEYSER_PLUGIN_NAME }}-release22-x86_64-unknown-linux-gnu.tar.bz2
           mv ${{ env.GEYSER_PLUGIN_NAME }}-release-x86_64-unknown-linux-gnu.yml ${{ env.GEYSER_PLUGIN_NAME }}-release22-x86_64-unknown-linux-gnu.yml
 
-      - name: Deleteing directories to avoid upload conflict
-        run: |
-          rm -rf \
-            target/release/client.d \
-            target/release/config-check.d \
-            target/release/grpc-google-pubsub.d \
-            target/release/grpc-kafka.d \
-            target/release/grpc-scylladb.d
-
       - name: Release
         if: startsWith(github.ref, 'refs/tags/')
         uses: softprops/action-gh-release@v2
@@ -120,7 +129,6 @@ jobs:
             target/release/grpc-scylladb*
 
       - uses: actions/upload-artifact@v4
-        if: matrix.os == 'ubuntu-22.04'
         with:
           name: yellowstone-grpc-${{ github.sha }}-${{ matrix.os }}
           path: |

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,8 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 ### Features
 
+- geyser: use runtime instaed of unconstrained ([#331](https://github.com/rpcpool/yellowstone-grpc/pull/331))
+
 ### Breaking
 
 ## 2024-04-30

diff --git a/yellowstone-grpc-geyser/src/grpc.rs b/yellowstone-grpc-geyser/src/grpc.rs
@@ -27,7 +27,9 @@ use {
     },
     tokio::{
         fs,
+        runtime::Builder,
         sync::{broadcast, mpsc, Mutex, Notify, RwLock, Semaphore},
+        task::spawn_blocking,
         time::{sleep, Duration, Instant},
     },
     tokio_stream::wrappers::ReceiverStream,
@@ -787,12 +789,20 @@ impl GrpcService {
 
         // Run geyser message loop
         let (messages_tx, messages_rx) = mpsc::unbounded_channel();
-        tokio::spawn(tokio::task::unconstrained(Self::geyser_loop(
-            messages_rx,
-            blocks_meta_tx,
-            broadcast_tx,
-            block_fail_action,
-        )));
+        spawn_blocking(move || {
+            Builder::new_multi_thread()
+                .thread_name_fn(crate::get_thread_name)
+                .worker_threads(4)
+                .enable_all()
+                .build()
+                .expect("Failed to create a new runtime for geyser loop")
+                .block_on(Self::geyser_loop(
+                    messages_rx,
+                    blocks_meta_tx,
+                    broadcast_tx,
+                    block_fail_action,
+                ));
+        });
 
         // Run Server
         let shutdown = Arc::new(Notify::new());

diff --git a/yellowstone-grpc-geyser/src/lib.rs b/yellowstone-grpc-geyser/src/lib.rs
@@ -8,3 +8,11 @@ pub mod grpc;
 pub mod plugin;
 pub mod prom;
 pub mod version;
+
+pub fn get_thread_name() -> String {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
+    let id = ATOMIC_ID.fetch_add(1, Ordering::Relaxed);
+    format!("solGeyserGrpc{id:02}")
+}
diff --git a/yellowstone-grpc-geyser/src/plugin.rs b/yellowstone-grpc-geyser/src/plugin.rs
@@ -9,14 +9,7 @@ use {
         ReplicaEntryInfoVersions, ReplicaTransactionInfoVersions, Result as PluginResult,
         SlotStatus,
     },
-    std::{
-        concat, env,
-        sync::{
-            atomic::{AtomicUsize, Ordering},
-            Arc,
-        },
-        time::Duration,
-    },
+    std::{concat, env, sync::Arc, time::Duration},
     tokio::{
         runtime::{Builder, Runtime},
         sync::{mpsc, Notify},
@@ -68,11 +61,7 @@ impl GeyserPlugin for Plugin {
 
         // Create inner
         let runtime = Builder::new_multi_thread()
-            .thread_name_fn(|| {
-                static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
-                let id = ATOMIC_ID.fetch_add(1, Ordering::Relaxed);
-                format!("solGeyserGrpc{id:02}")
-            })
+            .thread_name_fn(crate::get_thread_name)
             .enable_all()
             .build()
             .map_err(|error| GeyserPluginError::Custom(Box::new(error)))?;