apollo_storage: wrap FileHandlers in Arc to avoid cloning (#14445)

Author: dafnamatsry

crates/apollo_storage/src/lib.rs

@@ -301,7 +301,7 @@ fn open_storage_internal(
     let reader = StorageReader {
         db_reader,
         scope: storage_config.scope,
-        file_readers,
+        file_readers: Arc::new(file_readers),
         tables,
         open_readers_metric,
     };
@@ -505,7 +505,7 @@ pub enum StorageScope {
 #[derive(Clone)]
 pub struct StorageReader {
     db_reader: DbReader,
-    file_readers: FileHandlers<RO>,
+    file_readers: Arc<FileHandlers<RO>>,
     tables: Arc<Tables>,
     scope: StorageScope,
     open_readers_metric: Option<&'static MetricGauge>,
@@ -569,8 +569,7 @@ impl StorageReader {
     pub fn begin_ro_txn(&self) -> StorageResult<StorageTxn<'_, RO>> {
         Ok(StorageTxn {
             txn: self.db_reader.begin_ro_txn()?,
-            // TODO(Dean): Wrap FileHandlers with an Arc to avoid cloning.
-            file_handlers: self.file_readers.clone(),
+            file_handlers: Arc::clone(&self.file_readers),
             tables: self.tables.clone(),
             scope: self.scope,
             _metric_updater: MetricsHandler::new(self.open_readers_metric),
@@ -636,7 +635,7 @@ type SharedStorageState = Arc<Mutex<SharedState>>;
 /// There is a single non-clonable writer instance, to make sure there is only one write transaction
 /// at any given moment.
 pub struct StorageWriter {
-    file_writers: FileHandlers<RW>,
+    file_writers: Arc<FileHandlers<RW>>,
     scope: StorageScope,
     /// Shared state containing db_writer, tables, and batching information.
     shared_state: SharedStorageState,
@@ -649,7 +648,7 @@ impl StorageWriter {
         scope: StorageScope,
         shared_state: SharedStorageState,
     ) -> Self {
-        Self { file_writers, scope, shared_state }
+        Self { file_writers: Arc::new(file_writers), scope, shared_state }
     }
 
     /// Returns a [`StorageTxnRW`] for reading and modifying data in the storage.
@@ -663,7 +662,7 @@ impl StorageWriter {
         guard.ensure_active_txn()?;
         Ok(StorageTxnRW {
             guard,
-            file_handlers: self.file_writers.clone(),
+            file_handlers: Arc::clone(&self.file_writers),
             scope: self.scope,
             _metric_updater: MetricsHandler::new(None),
         })
@@ -817,7 +816,7 @@ pub(crate) trait StorageTransaction: Sized {
 /// The actual functionality is implemented on the transaction in multiple traits.
 pub struct StorageTxn<'env, Mode: TransactionKind + 'static> {
     txn: DbTransaction<'env, Mode>,
-    file_handlers: FileHandlers<Mode>,
+    file_handlers: Arc<FileHandlers<Mode>>,
     tables: Arc<Tables>,
     scope: StorageScope,
     // Do not remove this. It is used to automatically update metrics on create/drop.
@@ -829,7 +828,7 @@ pub struct StorageTxn<'env, Mode: TransactionKind + 'static> {
 /// to the transaction.
 pub struct StorageTxnRW<'a> {
     guard: MutexGuard<'a, SharedState>,
-    file_handlers: FileHandlers<RW>,
+    file_handlers: Arc<FileHandlers<RW>>,
     scope: StorageScope,
     _metric_updater: MetricsHandler,
 }