Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Send publisher permissions using a watch channel to address heap growth #118

Merged
merged 1 commit into from
Apr 25, 2024
+41 −58
Merged

Send publisher permissions using a watch channel to address heap growth #118

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions src/agent/solana.rs
View file
Original file line number Diff line number Diff line change
@@ -8,9 +8,9 @@ pub mod network {

use {
super::{
super::{
store,
store::global,
super::store::{
self,
global,
},
exporter,
key_store::{
@@ -29,8 +29,8 @@ pub mod network {
std::time::Duration,
tokio::{
sync::{
mpsc,
mpsc::Sender,
watch,
},
task::JoinHandle,
},
@@ -86,8 +86,7 @@ pub mod network {
logger: Logger,
) -> Result<Vec<JoinHandle<()>>> {
// Publisher permissions updates between oracle and exporter
let (publisher_permissions_tx, publisher_permissions_rx) =
mpsc::channel(config.oracle.updates_channel_capacity);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe this was the only use of the updates_channel_capacity config. Let's remove the config as well if that's the case.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updates_channel_capacity is still used for the "updates" channel (updates_tx,updates_rx).

let (publisher_permissions_tx, publisher_permissions_rx) = watch::channel(<_>::default());

// Spawn the Oracle
let mut jhs = oracle::spawn_oracle(
73 changes: 28 additions & 45 deletions src/agent/solana/exporter.rs
View file
Original file line number Diff line number Diff line change
@@ -172,7 +172,7 @@ pub fn spawn_exporter(
network: Network,
rpc_url: &str,
rpc_timeout: Duration,
publisher_permissions_rx: mpsc::Receiver<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
publisher_permissions_rx: watch::Receiver<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
key_store: KeyStore,
local_store_tx: Sender<store::local::Message>,
global_store_tx: Sender<store::global::Lookup>,
@@ -260,7 +260,7 @@ pub struct Exporter {
inflight_transactions_tx: Sender<Signature>,

/// publisher => { permissioned_price => market hours } as read by the oracle module
publisher_permissions_rx: mpsc::Receiver<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
publisher_permissions_rx: watch::Receiver<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,

/// Currently known permissioned prices of this publisher along with their market hours
our_prices: HashMap<Pubkey, MarketSchedule>,
@@ -287,7 +287,7 @@ impl Exporter {
global_store_tx: Sender<store::global::Lookup>,
network_state_rx: watch::Receiver<NetworkState>,
inflight_transactions_tx: Sender<Signature>,
publisher_permissions_rx: mpsc::Receiver<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
publisher_permissions_rx: watch::Receiver<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
keypair_request_tx: mpsc::Sender<KeypairRequest>,
logger: Logger,
) -> Self {
@@ -556,51 +556,34 @@ impl Exporter {
/// The loop ensures that we clear the channel and use
/// only the final, latest message; try_recv() is
/// non-blocking.
///
/// Note: This behavior is similar to
/// tokio::sync::watch::channel(), which was not appropriate here
/// because its internal RwLock would complain about not being
/// Send with the HashMap<HashSet<Pubkey>> inside.
/// TODO(2023-05-05): Debug the watch::channel() compilation errors
fn update_our_prices(&mut self, publish_pubkey: &Pubkey) {
loop {
match self.publisher_permissions_rx.try_recv() {
Ok(publisher_permissions) => {
self.our_prices = publisher_permissions.get(publish_pubkey) .cloned()
.unwrap_or_else( || {
warn!(
self.logger,
"Exporter: No permissioned prices were found for the publishing keypair on-chain. This is expected only on startup.";
"publish_pubkey" => publish_pubkey.to_string(),
);
HashMap::new()
});
trace!(
self.logger,
"Exporter: read permissioned price accounts from channel";
"new_value" => format!("{:?}", self.our_prices),
);
}
// Expected failures when channel is empty
Err(TryRecvError::Empty) => {
trace!(
self.logger,
"Exporter: No more permissioned price accounts in channel, using cached value";
);
break;
}
// Unexpected failures (channel closed, internal errors etc.)
Err(other) => {
warn!(
self.logger,
"Exporter: Updating permissioned price accounts failed unexpectedly, using cached value";
"cached_value" => format!("{:?}", self.our_prices),
"error" => other.to_string(),
);
break;
}
match self.publisher_permissions_rx.has_changed() {
Ok(true) => {}
Ok(false) => return,
Err(other) => {
warn!(
self.logger,
"Exporter: Updating permissioned price accounts failed unexpectedly, using cached value";
"cached_value" => format!("{:?}", self.our_prices),
"error" => other.to_string(),
);
return;
}
}

self.our_prices = self
.publisher_permissions_rx
.borrow_and_update()
.get(publish_pubkey)
.cloned()
.unwrap_or_else(|| {
warn!(
self.logger,
"Exporter: No permissioned prices were found for the publishing keypair on-chain. This is expected only on startup.";
"publish_pubkey" => publish_pubkey.to_string(),
);
HashMap::new()
});
}

async fn fetch_local_store_contents(&self) -> Result<HashMap<PriceIdentifier, PriceInfo>> {
15 changes: 8 additions & 7 deletions src/agent/solana/oracle.rs
View file
Original file line number Diff line number Diff line change
@@ -43,7 +43,10 @@ use {
time::Duration,
},
tokio::{
sync::mpsc,
sync::{
mpsc,
watch,
},
task::JoinHandle,
time::Interval,
},
@@ -204,7 +207,7 @@ pub fn spawn_oracle(
wss_url: &str,
rpc_timeout: Duration,
global_store_update_tx: mpsc::Sender<global::Update>,
publisher_permissions_tx: mpsc::Sender<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
publisher_permissions_tx: watch::Sender<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
key_store: KeyStore,
logger: Logger,
) -> Vec<JoinHandle<()>> {
@@ -419,7 +422,7 @@ struct Poller {
data_tx: mpsc::Sender<Data>,

/// Updates about permissioned price accounts from oracle to exporter
publisher_permissions_tx: mpsc::Sender<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
publisher_permissions_tx: watch::Sender<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,

/// The RPC client to use to poll data from the RPC node
rpc_client: RpcClient,
@@ -439,7 +442,7 @@ struct Poller {
impl Poller {
pub fn new(
data_tx: mpsc::Sender<Data>,
publisher_permissions_tx: mpsc::Sender<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
publisher_permissions_tx: watch::Sender<HashMap<Pubkey, HashMap<Pubkey, MarketSchedule>>>,
rpc_url: &str,
rpc_timeout: Duration,
commitment: CommitmentLevel,
@@ -481,9 +484,7 @@ impl Poller {
let fresh_data = self.poll().await?;

self.publisher_permissions_tx
.send(fresh_data.publisher_permissions.clone())
.await
.context("Updating permissioned price accounts for exporter")?;
.send_replace(fresh_data.publisher_permissions.clone());

self.data_tx
.send(fresh_data)