graph: Introduce CheapClone and use it a bit

Author: leoyvens

chain/ethereum/src/ethereum_adapter.rs

@@ -12,8 +12,8 @@ use graph::components::ethereum::{EthereumAdapter as EthereumAdapterTrait, *};
 use graph::prelude::{
     debug, err_msg, error, ethabi, format_err,
     futures03::{self, compat::Future01CompatExt, FutureExt, StreamExt, TryStreamExt},
-    hex, retry, stream, tiny_keccak, trace, warn, web3, ChainStore, DynTryFuture, Error,
-    EthereumCallCache, Logger, TimeoutError,
+    hex, retry, stream, tiny_keccak, trace, warn, web3, ChainStore, CheapClone, DynTryFuture,
+    Error, EthereumCallCache, Logger, TimeoutError,
 };
 use web3::api::Web3;
 use web3::transports::batch::Batch;
@@ -62,6 +62,15 @@ lazy_static! {
             .expect("invalid GRAPH_ETHEREUM_REQUEST_RETRIES env var");
 }
 
+impl<T: web3::Transport> CheapClone for EthereumAdapter<T> {
+    fn cheap_clone(&self) -> Self {
+        Self {
+            web3: self.web3.cheap_clone(),
+            metrics: self.metrics.cheap_clone(),
+        }
+    }
+}
+
 impl<T> EthereumAdapter<T>
 where
     T: web3::BatchTransport + Send + Sync + 'static,
@@ -168,7 +177,7 @@ where
         subgraph_metrics: Arc<SubgraphEthRpcMetrics>,
         from: u64,
         to: u64,
-        filter: EthGetLogsFilter,
+        filter: Arc<EthGetLogsFilter>,
         too_many_logs_fingerprints: &'static [&'static str],
     ) -> impl Future<Item = Vec<Log>, Error = TimeoutError<web3::error::Error>> {
         let eth_adapter = self.clone();
@@ -276,7 +285,9 @@ where
         }
 
         // Collect all event sigs
-        let eth = self.clone();
+        let eth = self.cheap_clone();
+        let filter = Arc::new(filter);
+
         let step = match filter.contracts.is_empty() {
             // `to - from + 1`  blocks will be scanned.
             false => to - from,
@@ -287,10 +298,10 @@ where
         // node returns an error that signifies the request is to heavy to process, the range will
         // be broken down to smaller steps.
         futures03::stream::try_unfold((from, step), move |(start, step)| {
-            let logger = logger.clone();
-            let filter = filter.clone();
-            let eth = eth.clone();
-            let subgraph_metrics = subgraph_metrics.clone();
+            let logger = logger.cheap_clone();
+            let filter = filter.cheap_clone();
+            let eth = eth.cheap_clone();
+            let subgraph_metrics = subgraph_metrics.cheap_clone();
 
             async move {
                 if start > to {
@@ -305,10 +316,10 @@ where
                 let res = eth
                     .logs_with_sigs(
                         &logger,
-                        subgraph_metrics.clone(),
+                        subgraph_metrics.cheap_clone(),
                         start,
                         end,
-                        filter.clone(),
+                        filter.cheap_clone(),
                         TOO_MANY_LOGS_FINGERPRINTS,
                     )
                     .compat()
@@ -1035,12 +1046,18 @@ where
         to: u64,
         log_filter: EthereumLogFilter,
     ) -> DynTryFuture<'static, Vec<Log>, Error> {
-        let eth: Self = self.clone();
+        let eth: Self = self.cheap_clone();
         let logger = logger.clone();
 
         futures03::stream::iter(log_filter.eth_get_logs_filters().map(move |filter| {
-            eth.clone()
-                .log_stream(logger.clone(), subgraph_metrics.clone(), from, to, filter)
+            eth.cheap_clone()
+                .log_stream(
+                    logger.cheap_clone(),
+                    subgraph_metrics.cheap_clone(),
+                    from,
+                    to,
+                    filter,
+                )
                 .into_stream()
         }))
         .flatten()

graph/src/cheap_clone.rs

@@ -0,0 +1,14 @@
+use slog::Logger;
+use std::rc::Rc;
+use std::sync::Arc;
+
+/// Things that, in the context of an application such as Graph Node, are fast to clone.
+pub trait CheapClone: Clone {
+    fn cheap_clone(&self) -> Self {
+        self.clone()
+    }
+}
+
+impl<T: ?Sized> CheapClone for Rc<T> {}
+impl<T: ?Sized> CheapClone for Arc<T> {}
+impl CheapClone for Logger {}

graph/src/lib.rs

@@ -13,6 +13,9 @@ pub mod ext;
 /// Logging utilities
 pub mod log;
 
+/// `CheapClone` trait.
+pub mod cheap_clone;
+
 /// Module with mocks for different parts of the system.
 pub mod mock {
     pub use crate::components::ethereum::MockEthereumAdapter;
@@ -99,6 +102,7 @@ pub mod prelude {
     };
     pub use crate::components::{EventConsumer, EventProducer};
 
+    pub use crate::cheap_clone::CheapClone;
     pub use crate::data::graphql::{SerializableValue, TryFromValue, ValueMap};
     pub use crate::data::query::{
         Query, QueryError, QueryExecutionError, QueryResult, QueryVariables,