| 1 | use crate::{Doc, Origin, Store, Transaction, TransactionMut}; |
| 2 | use async_lock::futures::{Read, Write}; |
| 3 | use std::future::Future; |
| 4 | use std::pin::Pin; |
| 5 | use std::task::{Context, Poll}; |
| 6 | use thiserror::Error; |
| 7 | |
| 8 | /// Trait implemented by [Doc] and shared types, used for carrying over the responsibilities of |
| 9 | /// creating new transactions, used as a unit of work in Yrs. |
| 10 | pub trait Transact { |
| 11 | /// Creates and returns a lightweight read-only transaction. |
| 12 | /// |
| 13 | /// # Errors |
| 14 | /// |
| 15 | /// While it's possible to have multiple read-only transactions active at the same time, |
| 16 | /// this method will return a [TransactionAcqError::SharedAcqFailed] error whenever called |
| 17 | /// while a read-write transaction (see: [Self::try_transact_mut]) is active at the same time. |
| 18 | fn try_transact(&self) -> Result<Transaction, TransactionAcqError>; |
| 19 | |
| 20 | /// Creates and returns a read-write capable transaction. This transaction can be used to |
| 21 | /// mutate the contents of underlying document store and upon dropping or committing it may |
| 22 | /// subscription callbacks. |
| 23 | /// |
| 24 | /// # Errors |
| 25 | /// |
| 26 | /// Only one read-write transaction can be active at the same time. If any other transaction - |
| 27 | /// be it a read-write or read-only one - is active at the same time, this method will return |
| 28 | /// a [TransactionAcqError::ExclusiveAcqFailed] error. |
| 29 | fn try_transact_mut(&self) -> Result<TransactionMut, TransactionAcqError>; |
| 30 | |
| 31 | /// Creates and returns a read-write capable transaction with an `origin` classifier attached. |
| 32 | /// This transaction can be used to mutate the contents of underlying document store and upon |
| 33 | /// dropping or committing it may subscription callbacks. |
| 34 | /// |
| 35 | /// An `origin` may be used to identify context of operations made (example updates performed |
| 36 | /// locally vs. incoming from remote replicas) and it's used i.e. by [`UndoManager`][crate::undo::UndoManager]. |
| 37 | /// |
| 38 | /// # Errors |
| 39 | /// |
| 40 | /// Only one read-write transaction can be active at the same time. If any other transaction - |
| 41 | /// be it a read-write or read-only one - is active at the same time, this method will return |
| 42 | /// a [TransactionAcqError::ExclusiveAcqFailed] error. |
| 43 | fn try_transact_mut_with<T>(&self, origin: T) -> Result<TransactionMut, TransactionAcqError> |
| 44 | where |
| 45 | T: Into<Origin>; |
| 46 | |
| 47 | /// Creates and returns a read-write capable transaction with an `origin` classifier attached. |
| 48 | /// This transaction can be used to mutate the contents of underlying document store and upon |
| 49 | /// dropping or committing it may subscription callbacks. |
| 50 | /// |
| 51 | /// An `origin` may be used to identify context of operations made (example updates performed |
| 52 | /// locally vs. incoming from remote replicas) and it's used i.e. by [`UndoManager`][crate::undo::UndoManager]. |
| 53 | /// |
| 54 | /// # Errors |
| 55 | /// |
| 56 | /// Only one read-write transaction can be active at the same time. If any other transaction - |
| 57 | /// be it a read-write or read-only one - is active at the same time, this method will |
| 58 | /// block the thread until exclusive access can be acquired, unless building for wasm |
| 59 | /// which panics. If blocking is undesirable, use `try_transact_mut_with(origin).unwrap()` instead`. |
| 60 | fn transact_mut_with<T>(&self, origin: T) -> TransactionMut |
| 61 | where |
| 62 | T: Into<Origin>; |
| 63 | |
| 64 | /// Creates and returns a lightweight read-only transaction. |
| 65 | /// |
| 66 | /// # Errors |
| 67 | /// |
| 68 | /// While it's possible to have multiple read-only transactions active at the same time, |
| 69 | /// this method will block the thread until exclusive access can be acquired, unless building for wasm |
| 70 | /// which panics. If blocking is undesirable, use `try_transact(origin).unwrap()` instead`. |
| 71 | fn transact(&self) -> Transaction; |
| 72 | |
| 73 | /// Creates and returns a read-write capable transaction. This transaction can be used to |
| 74 | /// mutate the contents of underlying document store and upon dropping or committing it may |
| 75 | /// subscription callbacks. |
| 76 | /// |
| 77 | /// # Errors |
| 78 | /// |
| 79 | /// Only one read-write transaction can be active at the same time. If any other transaction - |
| 80 | /// be it a read-write or read-only one - is active at the same time, this method will block |
| 81 | /// the thread until exclusive access can be acquired, unless building for wasm |
| 82 | /// which panics. If blocking is undesirable, use `try_transact_mut(origin).unwrap()` instead`. |
| 83 | fn transact_mut(&self) -> TransactionMut; |
| 84 | } |
| 85 | |
| 86 | impl Transact for Doc { |
| 87 | fn try_transact(&self) -> Result<Transaction, TransactionAcqError> { |
| 88 | match self.store.try_read() { |
| 89 | Some(store) => Ok(Transaction::new(store)), |
| 90 | None => Err(TransactionAcqError::SharedAcqFailed), |
| 91 | } |
| 92 | } |
| 93 | |
| 94 | fn try_transact_mut(&self) -> Result<TransactionMut, TransactionAcqError> { |
| 95 | match self.store.try_write() { |
| 96 | Some(store) => Ok(TransactionMut::new(self.clone(), store, None)), |
| 97 | None => Err(TransactionAcqError::ExclusiveAcqFailed), |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | fn try_transact_mut_with<T>(&self, origin: T) -> Result<TransactionMut, TransactionAcqError> |
| 102 | where |
| 103 | T: Into<Origin>, |
| 104 | { |
| 105 | match self.store.try_write() { |
| 106 | Some(store) => Ok(TransactionMut::new( |
| 107 | self.clone(), |
| 108 | store, |
| 109 | Some(origin.into()), |
| 110 | )), |
| 111 | None => Err(TransactionAcqError::ExclusiveAcqFailed), |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | fn transact_mut_with<T>(&self, origin: T) -> TransactionMut |
| 116 | where |
| 117 | T: Into<Origin>, |
| 118 | { |
| 119 | let lock = self.store.write_blocking(); |
| 120 | TransactionMut::new(self.clone(), lock, Some(origin.into())) |
| 121 | } |
| 122 | |
| 123 | fn transact(&self) -> Transaction { |
| 124 | let lock = self.store.read_blocking(); |
| 125 | Transaction::new(lock) |
| 126 | } |
| 127 | |
| 128 | fn transact_mut(&self) -> TransactionMut { |
| 129 | let lock = self.store.write_blocking(); |
| 130 | TransactionMut::new(self.clone(), lock, None) |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | /// Trait implemented by [Doc] and shared types, used for carrying over the responsibilities of |
| 135 | /// creating new transactions, used as a unit of work in Yrs. |
| 136 | pub trait AsyncTransact<'doc> { |
| 137 | type Read: Future<Output = Transaction<'doc>>; |
| 138 | type Write: Future<Output = TransactionMut<'doc>>; |
| 139 | |
| 140 | fn transact(&'doc self) -> Self::Read; |
| 141 | fn transact_mut(&'doc self) -> Self::Write; |
| 142 | |
| 143 | /// Creates and returns a read-write capable transaction with an `origin` classifier attached. |
| 144 | /// This transaction can be used to mutate the contents of underlying document store and upon |
| 145 | /// dropping or committing it may subscription callbacks. |
| 146 | /// |
| 147 | /// An `origin` may be used to identify context of operations made (example updates performed |
| 148 | /// locally vs. incoming from remote replicas) and it's used i.e. by [`UndoManager`][crate::undo::UndoManager]. |
| 149 | fn transact_mut_with<T>(&'doc self, origin: T) -> Self::Write |
| 150 | where |
| 151 | T: Into<Origin>; |
| 152 | } |
| 153 | |
| 154 | impl<'doc> AsyncTransact<'doc> for Doc { |
| 155 | type Read = AcquireTransaction<'doc>; |
| 156 | type Write = AcquireTransactionMut<'doc>; |
| 157 | |
| 158 | fn transact(&'doc self) -> Self::Read { |
| 159 | let fut = self.store.read_async(); |
| 160 | AcquireTransaction { fut } |
| 161 | } |
| 162 | |
| 163 | fn transact_mut(&'doc self) -> Self::Write { |
| 164 | let fut = self.store.write_async(); |
| 165 | AcquireTransactionMut { |
| 166 | doc: self.clone(), |
| 167 | origin: None, |
| 168 | fut, |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | fn transact_mut_with<T>(&'doc self, origin: T) -> Self::Write |
| 173 | where |
| 174 | T: Into<Origin>, |
| 175 | { |
| 176 | let fut = self.store.write_async(); |
| 177 | AcquireTransactionMut { |
| 178 | doc: self.clone(), |
| 179 | origin: Some(origin.into()), |
| 180 | fut, |
| 181 | } |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | pub struct AcquireTransaction<'doc> { |
| 186 | fut: Read<'doc, Store>, |
| 187 | } |
| 188 | |
| 189 | impl<'doc> Unpin for AcquireTransaction<'doc> {} |
| 190 | |
| 191 | impl<'doc> Future for AcquireTransaction<'doc> { |
| 192 | type Output = Transaction<'doc>; |
| 193 | |
| 194 | fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |
| 195 | let pinned = unsafe { Pin::new_unchecked(&mut self.fut) }; |
| 196 | pinned.poll(cx).map(Transaction::new) |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | pub struct AcquireTransactionMut<'doc> { |
| 201 | doc: Doc, |
| 202 | origin: Option<Origin>, |
| 203 | fut: Write<'doc, Store>, |
| 204 | } |
| 205 | |
| 206 | impl<'doc> Unpin for AcquireTransactionMut<'doc> {} |
| 207 | |
| 208 | impl<'doc> Future for AcquireTransactionMut<'doc> { |
| 209 | type Output = TransactionMut<'doc>; |
| 210 | |
| 211 | fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |
| 212 | let pinned = unsafe { Pin::new_unchecked(&mut self.fut) }; |
| 213 | match pinned.poll(cx) { |
| 214 | Poll::Ready(store) => { |
| 215 | let doc = self.doc.clone(); |
| 216 | let origin = self.origin.take(); |
| 217 | Poll::Ready(TransactionMut::new(doc, store, origin)) |
| 218 | } |
| 219 | Poll::Pending => Poll::Pending, |
| 220 | } |
| 221 | } |
| 222 | } |
| 223 | |
| 224 | #[derive(Error, Debug)] |
| 225 | pub enum TransactionAcqError { |
| 226 | #[error("Failed to acquire read-only transaction. Drop read-write transaction and retry.")] |
| 227 | SharedAcqFailed, |
| 228 | #[error("Failed to acquire read-write transaction. Drop other transactions and retry.")] |
| 229 | ExclusiveAcqFailed, |
| 230 | #[error("All references to a parent document containing this structure has been dropped.")] |
| 231 | DocumentDropped, |
| 232 | } |
| 233 | |
| 234 | #[cfg(test)] |
| 235 | mod test { |
| 236 | use crate::{Doc, GetString, Text, Transact}; |
| 237 | use rand::random; |
| 238 | use std::sync::{Arc, Barrier}; |
| 239 | use std::time::{Duration, Instant}; |
| 240 | |
| 241 | #[test] |
| 242 | fn multi_thread_transact_mut() { |
| 243 | let doc = Doc::new(); |
| 244 | let txt = doc.get_or_insert_text("text"); |
| 245 | |
| 246 | const N: usize = 3; |
| 247 | let barrier = Arc::new(Barrier::new(N + 1)); |
| 248 | |
| 249 | let start = Instant::now(); |
| 250 | for _ in 0..N { |
| 251 | let d = doc.clone(); |
| 252 | let t = txt.clone(); |
| 253 | let b = barrier.clone(); |
| 254 | std::thread::spawn(move || { |
| 255 | // let mut txn = d.try_transact_mut().unwrap(); // this will hang forever |
| 256 | let mut txn = d.transact_mut(); |
| 257 | let n = random::<u64>() % 5; |
| 258 | std::thread::sleep(Duration::from_millis(n * 100)); |
| 259 | t.insert(&mut txn, 0, "a"); |
| 260 | drop(txn); |
| 261 | b.wait(); |
| 262 | }); |
| 263 | } |
| 264 | |
| 265 | barrier.wait(); |
| 266 | println!("{} threads executed in {:?}", N, Instant::now() - start); |
| 267 | |
| 268 | let expected: String = (0..N).map(|_| 'a').collect(); |
| 269 | let txn = doc.transact(); |
| 270 | let str = txt.get_string(&txn); |
| 271 | assert_eq!(str, expected); |
| 272 | } |
| 273 | } |
| 274 | |
amadan