[Turbopack] concurrent write batches and little endian encoding

turbopack/crates/turbo-tasks-backend/src/backend/mod.rs

@@ -741,7 +741,7 @@ impl<B: BackingStorage> TurboTasksBackendInner<B> {
                 persisted_storage_meta_log,
                 persisted_storage_data_log,
             ) {
-                println!("Persising failed: {:#?}", err);
+                println!("Persising failed: {}", err);
                 return None;
             }
         }

turbopack/crates/turbo-tasks-backend/src/database/fresh_db_optimization.rs

@@ -7,7 +7,10 @@ use std::{
 
 use anyhow::Result;
 
-use crate::database::key_value_database::{KeySpace, KeyValueDatabase, WriteBatch};
+use crate::database::{
+    key_value_database::{KeySpace, KeyValueDatabase},
+    write_batch::{BaseWriteBatch, ConcurrentWriteBatch, SerialWriteBatch, WriteBatch},
+};
 
 pub fn is_fresh(path: &Path) -> bool {
     fs::exists(path).map_or(false, |exists| !exists)
@@ -63,33 +66,42 @@ impl<T: KeyValueDatabase> KeyValueDatabase for FreshDbOptimization<T> {
         self.database.get(transaction, key_space, key)
     }
 
-    type WriteBatch<'l>
-        = FreshDbOptimizationWriteBatch<'l, T>
+    type SerialWriteBatch<'l>
+        = FreshDbOptimizationWriteBatch<'l, T::SerialWriteBatch<'l>>
+    where
+        Self: 'l;
+
+    type ConcurrentWriteBatch<'l>
+        = FreshDbOptimizationWriteBatch<'l, T::ConcurrentWriteBatch<'l>>
     where
         Self: 'l;
 
-    fn write_batch(&self) -> Result<Self::WriteBatch<'_>> {
-        Ok(FreshDbOptimizationWriteBatch {
-            write_batch: self.database.write_batch()?,
-            fresh_db: &self.fresh_db,
+    fn write_batch(
+        &self,
+    ) -> Result<WriteBatch<'_, Self::SerialWriteBatch<'_>, Self::ConcurrentWriteBatch<'_>>> {
+        Ok(match self.database.write_batch()? {
+            WriteBatch::Serial(write_batch) => WriteBatch::serial(FreshDbOptimizationWriteBatch {
+                write_batch,
+                fresh_db: &self.fresh_db,
+            }),
+            WriteBatch::Concurrent(write_batch, _) => {
+                WriteBatch::concurrent(FreshDbOptimizationWriteBatch {
+                    write_batch,
+                    fresh_db: &self.fresh_db,
+                })
+            }
         })
     }
 }
 
-pub struct FreshDbOptimizationWriteBatch<'a, T: KeyValueDatabase>
-where
-    T: 'a,
-{
-    write_batch: T::WriteBatch<'a>,
+pub struct FreshDbOptimizationWriteBatch<'a, B> {
+    write_batch: B,
     fresh_db: &'a AtomicBool,
 }
 
-impl<'a, T: KeyValueDatabase> WriteBatch<'a> for FreshDbOptimizationWriteBatch<'a, T>
-where
-    T: 'a,
-{
+impl<'a, B: BaseWriteBatch<'a>> BaseWriteBatch<'a> for FreshDbOptimizationWriteBatch<'a, B> {
     type ValueBuffer<'l>
-        = <T::WriteBatch<'a> as WriteBatch<'a>>::ValueBuffer<'l>
+        = B::ValueBuffer<'l>
     where
         Self: 'l,
         'a: 'l;
@@ -101,16 +113,30 @@ where
         self.write_batch.get(key_space, key)
     }
 
+    fn commit(self) -> Result<()> {
+        self.fresh_db.store(false, Ordering::Release);
+        self.write_batch.commit()
+    }
+}
+
+impl<'a, B: SerialWriteBatch<'a>> SerialWriteBatch<'a> for FreshDbOptimizationWriteBatch<'a, B> {
     fn put(&mut self, key_space: KeySpace, key: Cow<[u8]>, value: Cow<[u8]>) -> Result<()> {
         self.write_batch.put(key_space, key, value)
     }
 
     fn delete(&mut self, key_space: KeySpace, key: Cow<[u8]>) -> Result<()> {
         self.write_batch.delete(key_space, key)
     }
+}
 
-    fn commit(self) -> Result<()> {
-        self.fresh_db.store(false, Ordering::Release);
-        self.write_batch.commit()
+impl<'a, B: ConcurrentWriteBatch<'a>> ConcurrentWriteBatch<'a>
+    for FreshDbOptimizationWriteBatch<'a, B>
+{
+    fn put(&self, key_space: KeySpace, key: Cow<[u8]>, value: Cow<[u8]>) -> Result<()> {
+        self.write_batch.put(key_space, key, value)
+    }
+
+    fn delete(&self, key_space: KeySpace, key: Cow<[u8]>) -> Result<()> {
+        self.write_batch.delete(key_space, key)
     }
 }

turbopack/crates/turbo-tasks-backend/src/database/key_value_database.rs

@@ -1,7 +1,9 @@
-use std::borrow::Cow;
-
 use anyhow::Result;
 
+use crate::database::write_batch::{
+    ConcurrentWriteBatch, SerialWriteBatch, UnimplementedWriteBatch, WriteBatch,
+};
+
 #[derive(Debug, Clone, Copy)]
 pub enum KeySpace {
     Infra,
@@ -11,20 +13,6 @@ pub enum KeySpace {
     ReverseTaskCache,
 }
 
-pub trait WriteBatch<'a> {
-    type ValueBuffer<'l>: std::borrow::Borrow<[u8]>
-    where
-        Self: 'l,
-        'a: 'l;
-
-    fn get<'l>(&'l self, key_space: KeySpace, key: &[u8]) -> Result<Option<Self::ValueBuffer<'l>>>
-    where
-        'a: 'l;
-    fn put(&mut self, key_space: KeySpace, key: Cow<[u8]>, value: Cow<[u8]>) -> Result<()>;
-    fn delete(&mut self, key_space: KeySpace, key: Cow<[u8]>) -> Result<()>;
-    fn commit(self) -> Result<()>;
-}
-
 pub trait KeyValueDatabase {
     type ReadTransaction<'l>
     where
@@ -47,8 +35,16 @@ pub trait KeyValueDatabase {
         key: &[u8],
     ) -> Result<Option<Self::ValueBuffer<'l>>>;
 
-    type WriteBatch<'l>: WriteBatch<'l>
+    type SerialWriteBatch<'l>: SerialWriteBatch<'l>
+        = UnimplementedWriteBatch
     where
         Self: 'l;
-    fn write_batch(&self) -> Result<Self::WriteBatch<'_>>;
+    type ConcurrentWriteBatch<'l>: ConcurrentWriteBatch<'l>
+        = UnimplementedWriteBatch
+    where
+        Self: 'l;
+
+    fn write_batch(
+        &self,
+    ) -> Result<WriteBatch<'_, Self::SerialWriteBatch<'_>, Self::ConcurrentWriteBatch<'_>>>;
 }

turbopack/crates/turbo-tasks-backend/src/database/lmdb/mod.rs

@@ -6,7 +6,10 @@ use lmdb::{
     Transaction, WriteFlags,
 };
 
-use crate::database::key_value_database::{KeySpace, KeyValueDatabase, WriteBatch};
+use crate::database::{
+    key_value_database::{KeySpace, KeyValueDatabase},
+    write_batch::{BaseWriteBatch, SerialWriteBatch, WriteBatch},
+};
 
 mod extended_key;
 
@@ -111,16 +114,18 @@ impl KeyValueDatabase for LmbdKeyValueDatabase {
         Ok(Some(value))
     }
 
-    type WriteBatch<'l>
+    type SerialWriteBatch<'l>
         = LmbdWriteBatch<'l>
     where
         Self: 'l;
 
-    fn write_batch(&self) -> Result<Self::WriteBatch<'_>> {
-        Ok(LmbdWriteBatch {
+    fn write_batch(
+        &self,
+    ) -> Result<WriteBatch<'_, Self::SerialWriteBatch<'_>, Self::ConcurrentWriteBatch<'_>>> {
+        Ok(WriteBatch::serial(LmbdWriteBatch {
             tx: self.env.begin_rw_txn()?,
             this: self,
-        })
+        }))
     }
 }
 
@@ -129,28 +134,7 @@ pub struct LmbdWriteBatch<'l> {
     this: &'l LmbdKeyValueDatabase,
 }
 
-impl<'a> WriteBatch<'a> for LmbdWriteBatch<'a> {
-    fn put(&mut self, key_space: KeySpace, key: Cow<[u8]>, value: Cow<[u8]>) -> Result<()> {
-        extended_key::put(
-            &mut self.tx,
-            self.this.db(key_space),
-            &key,
-            &value,
-            WriteFlags::empty(),
-        )?;
-        Ok(())
-    }
-
-    fn delete(&mut self, key_space: KeySpace, key: Cow<[u8]>) -> Result<()> {
-        extended_key::delete(
-            &mut self.tx,
-            self.this.db(key_space),
-            &key,
-            WriteFlags::empty(),
-        )?;
-        Ok(())
-    }
-
+impl<'a> BaseWriteBatch<'a> for LmbdWriteBatch<'a> {
     type ValueBuffer<'l>
         = &'l [u8]
     where
@@ -178,3 +162,26 @@ impl<'a> WriteBatch<'a> for LmbdWriteBatch<'a> {
         Ok(())
     }
 }
+
+impl<'a> SerialWriteBatch<'a> for LmbdWriteBatch<'a> {
+    fn put(&mut self, key_space: KeySpace, key: Cow<[u8]>, value: Cow<[u8]>) -> Result<()> {
+        extended_key::put(
+            &mut self.tx,
+            self.this.db(key_space),
+            &key,
+            &value,
+            WriteFlags::empty(),
+        )?;
+        Ok(())
+    }
+
+    fn delete(&mut self, key_space: KeySpace, key: Cow<[u8]>) -> Result<()> {
+        extended_key::delete(
+            &mut self.tx,
+            self.this.db(key_space),
+            &key,
+            WriteFlags::empty(),
+        )?;
+        Ok(())
+    }
+}

turbopack/crates/turbo-tasks-backend/src/database/mod.rs

@@ -6,6 +6,7 @@ pub mod lmdb;
 pub mod noop_kv;
 pub mod read_transaction_cache;
 mod startup_cache;
+pub mod write_batch;
 
 pub use db_versioning::handle_db_versioning;
 pub use fresh_db_optimization::{is_fresh, FreshDbOptimization};

turbopack/crates/turbo-tasks-backend/src/database/noop_kv.rs

@@ -2,7 +2,10 @@ use std::borrow::Cow;
 
 use anyhow::Result;
 
-use crate::database::key_value_database::{KeySpace, KeyValueDatabase, WriteBatch};
+use crate::database::{
+    key_value_database::{KeySpace, KeyValueDatabase},
+    write_batch::{BaseWriteBatch, ConcurrentWriteBatch, SerialWriteBatch, WriteBatch},
+};
 
 pub struct NoopKvDb;
 
@@ -36,23 +39,26 @@ impl KeyValueDatabase for NoopKvDb {
         Ok(None)
     }
 
-    type WriteBatch<'l>
+    type SerialWriteBatch<'l>
         = NoopWriteBatch
     where
         Self: 'l;
 
-    fn write_batch(&self) -> Result<Self::WriteBatch<'_>> {
-        Ok(NoopWriteBatch)
+    type ConcurrentWriteBatch<'l>
+        = NoopWriteBatch
+    where
+        Self: 'l;
+
+    fn write_batch(
+        &self,
+    ) -> Result<WriteBatch<'_, Self::SerialWriteBatch<'_>, Self::ConcurrentWriteBatch<'_>>> {
+        Ok(WriteBatch::concurrent(NoopWriteBatch))
     }
 }
 
 pub struct NoopWriteBatch;
 
-impl<'a> WriteBatch<'a> for NoopWriteBatch {
-    fn put(&mut self, _key_space: KeySpace, _key: Cow<[u8]>, _value: Cow<[u8]>) -> Result<()> {
-        Ok(())
-    }
-
+impl<'a> BaseWriteBatch<'a> for NoopWriteBatch {
     type ValueBuffer<'l>
         = &'l [u8]
     where
@@ -66,11 +72,27 @@ impl<'a> WriteBatch<'a> for NoopWriteBatch {
         Ok(None)
     }
 
+    fn commit(self) -> Result<()> {
+        Ok(())
+    }
+}
+
+impl SerialWriteBatch<'_> for NoopWriteBatch {
+    fn put(&mut self, _key_space: KeySpace, _key: Cow<[u8]>, _value: Cow<[u8]>) -> Result<()> {
+        Ok(())
+    }
+
     fn delete(&mut self, _key_space: KeySpace, _key: Cow<[u8]>) -> Result<()> {
         Ok(())
     }
+}
 
-    fn commit(self) -> Result<()> {
+impl ConcurrentWriteBatch<'_> for NoopWriteBatch {
+    fn put(&self, _key_space: KeySpace, _key: Cow<[u8]>, _value: Cow<[u8]>) -> Result<()> {
+        Ok(())
+    }
+
+    fn delete(&self, _key_space: KeySpace, _key: Cow<[u8]>) -> Result<()> {
         Ok(())
     }
 }