Removing worker group

Author: rodoufu

Cargo.lock

@@ -73,10 +73,6 @@ key_store.pyth_price_store_program_key = "3m6sv6HGqEbuyLV84mD7rJn4MAC9LhUa1y1AUN
 # takes to fetch all symbols.
 # oracle.max_lookup_batch_size = 100
 
-# Number of workers used to wait for the handle_price_account_update
-# oracle.handle_price_account_update_worker_poll_size = 25
-# Channel size used to wait for the handle_price_account_update
-# oracle.handle_price_account_update_channel_size = 1000
 # Minimum time for a subscriber to run
 # oracle.subscriber_finished_min_time = "30s"
 # Time to sleep if the subscriber do not run for more than the minimum time

config/config.toml

@@ -70,9 +70,6 @@ where
     )));
 
     if config.oracle.subscriber_enabled {
-        let number_of_workers = config.oracle.handle_price_account_update_worker_poll_size;
-        let (sender, receiver) =
-            tokio::sync::mpsc::channel(config.oracle.handle_price_account_update_channel_size);
         let min_elapsed_time = config.oracle.subscriber_finished_min_time;
         let sleep_time = config.oracle.subscriber_finished_sleep_time;
 
@@ -84,7 +81,6 @@ where
                     network,
                     state.clone(),
                     key_store.pyth_oracle_program_key,
-                    sender.clone(),
                 )
                 .await
                 {
@@ -96,22 +92,6 @@ where
                 }
             }
         }));
-
-        let receiver = Arc::new(tokio::sync::Mutex::new(receiver));
-        for _ in 0..number_of_workers {
-            let receiver = receiver.clone();
-            handles.push(tokio::spawn(async move {
-                loop {
-                    let mut receiver = receiver.lock().await;
-                    if let Some(task) = receiver.recv().await {
-                        drop(receiver);
-                        if let Err(err) = task.await {
-                            tracing::error!(%err, "error running price update");
-                        }
-                    }
-                }
-            }));
-        }
     }
 
     handles
@@ -128,7 +108,6 @@ async fn subscriber<S>(
     network: Network,
     state: Arc<S>,
     program_key: Pubkey,
-    sender: tokio::sync::mpsc::Sender<tokio::task::JoinHandle<()>>,
 ) -> Result<()>
 where
     S: Oracle,
@@ -156,17 +135,14 @@ where
             Some(account) => {
                 let pubkey: Pubkey = update.value.pubkey.as_str().try_into()?;
                 let state = state.clone();
-                sender
-                    .send(tokio::spawn(async move {
-                        if let Err(err) =
-                            Oracle::handle_price_account_update(&*state, network, &pubkey, &account)
-                                .await
-                        {
-                            tracing::error!(?err, "Failed to handle account update");
-                        }
-                    }))
-                    .await
-                    .context("sending handle_price_account_update task to worker")?;
+                tokio::spawn(async move {
+                    if let Err(err) =
+                        Oracle::handle_price_account_update(&*state, network, &pubkey, &account)
+                            .await
+                    {
+                        tracing::error!(?err, "Failed to handle account update");
+                    }
+                });
             }
 
             None => {

src/agent/services/oracle.rs

@@ -138,14 +138,6 @@ pub struct Data {
     pub publisher_buffer_key:  Option<Pubkey>,
 }
 
-fn default_handle_price_account_update_channel_size() -> usize {
-    1000
-}
-
-fn default_handle_price_account_update_worker_poll_size() -> usize {
-    25
-}
-
 fn default_subscriber_finished_min_time() -> Duration {
     Duration::from_secs(30)
 }
@@ -176,35 +168,25 @@ pub struct Config {
     /// trading off overall time it takes to fetch all symbols.
     pub max_lookup_batch_size: usize,
 
-    /// Number of workers used to wait for the handle_price_account_update
-    #[serde(default = "default_handle_price_account_update_worker_poll_size")]
-    pub handle_price_account_update_worker_poll_size: usize,
-    /// Channel size used to wait for the handle_price_account_update
-    #[serde(default = "default_handle_price_account_update_channel_size")]
-    pub handle_price_account_update_channel_size:     usize,
     /// Minimum time for a subscriber to run
     #[serde(default = "default_subscriber_finished_min_time")]
-    pub subscriber_finished_min_time:                 Duration,
+    pub subscriber_finished_min_time:   Duration,
     /// Time to sleep if the subscriber do not run for more than the minimum time
     #[serde(default = "default_subscriber_finished_sleep_time")]
-    pub subscriber_finished_sleep_time:               Duration,
+    pub subscriber_finished_sleep_time: Duration,
 }
 
 impl Default for Config {
     fn default() -> Self {
         Self {
-            commitment:                                   CommitmentLevel::Confirmed,
-            poll_interval_duration:                       Duration::from_secs(5),
-            subscriber_enabled:                           true,
-            updates_channel_capacity:                     10000,
-            data_channel_capacity:                        10000,
-            max_lookup_batch_size:                        100,
-            handle_price_account_update_worker_poll_size:
-                default_handle_price_account_update_worker_poll_size(),
-            handle_price_account_update_channel_size:
-                default_handle_price_account_update_channel_size(),
-            subscriber_finished_min_time:                 default_subscriber_finished_min_time(),
-            subscriber_finished_sleep_time:               default_subscriber_finished_sleep_time(),
+            commitment:                     CommitmentLevel::Confirmed,
+            poll_interval_duration:         Duration::from_secs(5),
+            subscriber_enabled:             true,
+            updates_channel_capacity:       10000,
+            data_channel_capacity:          10000,
+            max_lookup_batch_size:          100,
+            subscriber_finished_min_time:   default_subscriber_finished_min_time(),
+            subscriber_finished_sleep_time: default_subscriber_finished_sleep_time(),
         }
     }
 }