1 use crate::sync::batch_semaphore::{Semaphore, TryAcquireError};
2 use crate::sync::mutex::TryLockError;
3 #[cfg(all(tokio_unstable, feature = "tracing"))]
4 use crate::util::trace;
5 use std::cell::UnsafeCell;
6 use std::marker;
7 use std::marker::PhantomData;
8 use std::mem::ManuallyDrop;
9 use std::sync::Arc;
10
11 pub(crate) mod owned_read_guard;
12 pub(crate) mod owned_write_guard;
13 pub(crate) mod owned_write_guard_mapped;
14 pub(crate) mod read_guard;
15 pub(crate) mod write_guard;
16 pub(crate) mod write_guard_mapped;
17 pub(crate) use owned_read_guard::OwnedRwLockReadGuard;
18 pub(crate) use owned_write_guard::OwnedRwLockWriteGuard;
19 pub(crate) use owned_write_guard_mapped::OwnedRwLockMappedWriteGuard;
20 pub(crate) use read_guard::RwLockReadGuard;
21 pub(crate) use write_guard::RwLockWriteGuard;
22 pub(crate) use write_guard_mapped::RwLockMappedWriteGuard;
23
24 #[cfg(not(loom))]
25 const MAX_READS: u32 = std::u32::MAX >> 3;
26
27 #[cfg(loom)]
28 const MAX_READS: u32 = 10;
29
30 /// An asynchronous reader-writer lock.
31 ///
32 /// This type of lock allows a number of readers or at most one writer at any
33 /// point in time. The write portion of this lock typically allows modification
34 /// of the underlying data (exclusive access) and the read portion of this lock
35 /// typically allows for read-only access (shared access).
36 ///
37 /// In comparison, a [`Mutex`] does not distinguish between readers or writers
38 /// that acquire the lock, therefore causing any tasks waiting for the lock to
39 /// become available to yield. An `RwLock` will allow any number of readers to
40 /// acquire the lock as long as a writer is not holding the lock.
41 ///
42 /// The priority policy of Tokio's read-write lock is _fair_ (or
43 /// [_write-preferring_]), in order to ensure that readers cannot starve
44 /// writers. Fairness is ensured using a first-in, first-out queue for the tasks
45 /// awaiting the lock; if a task that wishes to acquire the write lock is at the
46 /// head of the queue, read locks will not be given out until the write lock has
47 /// been released. This is in contrast to the Rust standard library's
48 /// `std::sync::RwLock`, where the priority policy is dependent on the
49 /// operating system's implementation.
50 ///
51 /// The type parameter `T` represents the data that this lock protects. It is
52 /// required that `T` satisfies [`Send`] to be shared across threads. The RAII guards
53 /// returned from the locking methods implement [`Deref`](trait@std::ops::Deref)
54 /// (and [`DerefMut`](trait@std::ops::DerefMut)
55 /// for the `write` methods) to allow access to the content of the lock.
56 ///
57 /// # Examples
58 ///
59 /// ```
60 /// use tokio::sync::RwLock;
61 ///
62 /// #[tokio::main]
63 /// async fn main() {
64 /// let lock = RwLock::new(5);
65 ///
66 /// // many reader locks can be held at once
67 /// {
68 /// let r1 = lock.read().await;
69 /// let r2 = lock.read().await;
70 /// assert_eq!(*r1, 5);
71 /// assert_eq!(*r2, 5);
72 /// } // read locks are dropped at this point
73 ///
74 /// // only one write lock may be held, however
75 /// {
76 /// let mut w = lock.write().await;
77 /// *w += 1;
78 /// assert_eq!(*w, 6);
79 /// } // write lock is dropped here
80 /// }
81 /// ```
82 ///
83 /// [`Mutex`]: struct@super::Mutex
84 /// [`RwLock`]: struct@RwLock
85 /// [`RwLockReadGuard`]: struct@RwLockReadGuard
86 /// [`RwLockWriteGuard`]: struct@RwLockWriteGuard
87 /// [`Send`]: trait@std::marker::Send
88 /// [_write-preferring_]: https://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock#Priority_policies
89 #[derive(Debug)]
90 pub struct RwLock<T: ?Sized> {
91 #[cfg(all(tokio_unstable, feature = "tracing"))]
92 resource_span: tracing::Span,
93
94 // maximum number of concurrent readers
95 mr: u32,
96
97 //semaphore to coordinate read and write access to T
98 s: Semaphore,
99
100 //inner data T
101 c: UnsafeCell<T>,
102 }
103
104 #[test]
105 #[cfg(not(loom))]
bounds()106 fn bounds() {
107 fn check_send<T: Send>() {}
108 fn check_sync<T: Sync>() {}
109 fn check_unpin<T: Unpin>() {}
110 // This has to take a value, since the async fn's return type is unnameable.
111 fn check_send_sync_val<T: Send + Sync>(_t: T) {}
112
113 check_send::<RwLock<u32>>();
114 check_sync::<RwLock<u32>>();
115 check_unpin::<RwLock<u32>>();
116
117 check_send::<RwLockReadGuard<'_, u32>>();
118 check_sync::<RwLockReadGuard<'_, u32>>();
119 check_unpin::<RwLockReadGuard<'_, u32>>();
120
121 check_send::<OwnedRwLockReadGuard<u32, i32>>();
122 check_sync::<OwnedRwLockReadGuard<u32, i32>>();
123 check_unpin::<OwnedRwLockReadGuard<u32, i32>>();
124
125 check_send::<RwLockWriteGuard<'_, u32>>();
126 check_sync::<RwLockWriteGuard<'_, u32>>();
127 check_unpin::<RwLockWriteGuard<'_, u32>>();
128
129 check_send::<RwLockMappedWriteGuard<'_, u32>>();
130 check_sync::<RwLockMappedWriteGuard<'_, u32>>();
131 check_unpin::<RwLockMappedWriteGuard<'_, u32>>();
132
133 check_send::<OwnedRwLockWriteGuard<u32>>();
134 check_sync::<OwnedRwLockWriteGuard<u32>>();
135 check_unpin::<OwnedRwLockWriteGuard<u32>>();
136
137 check_send::<OwnedRwLockMappedWriteGuard<u32, i32>>();
138 check_sync::<OwnedRwLockMappedWriteGuard<u32, i32>>();
139 check_unpin::<OwnedRwLockMappedWriteGuard<u32, i32>>();
140
141 let rwlock = Arc::new(RwLock::new(0));
142 check_send_sync_val(rwlock.read());
143 check_send_sync_val(Arc::clone(&rwlock).read_owned());
144 check_send_sync_val(rwlock.write());
145 check_send_sync_val(Arc::clone(&rwlock).write_owned());
146 }
147
148 // As long as T: Send + Sync, it's fine to send and share RwLock<T> between threads.
149 // If T were not Send, sending and sharing a RwLock<T> would be bad, since you can access T through
150 // RwLock<T>.
151 unsafe impl<T> Send for RwLock<T> where T: ?Sized + Send {}
152 unsafe impl<T> Sync for RwLock<T> where T: ?Sized + Send + Sync {}
153 // NB: These impls need to be explicit since we're storing a raw pointer.
154 // Safety: Stores a raw pointer to `T`, so if `T` is `Sync`, the lock guard over
155 // `T` is `Send`.
156 unsafe impl<T> Send for RwLockReadGuard<'_, T> where T: ?Sized + Sync {}
157 unsafe impl<T> Sync for RwLockReadGuard<'_, T> where T: ?Sized + Send + Sync {}
158 // T is required to be `Send` because an OwnedRwLockReadGuard can be used to drop the value held in
159 // the RwLock, unlike RwLockReadGuard.
160 unsafe impl<T, U> Send for OwnedRwLockReadGuard<T, U>
161 where
162 T: ?Sized + Send + Sync,
163 U: ?Sized + Sync,
164 {
165 }
166 unsafe impl<T, U> Sync for OwnedRwLockReadGuard<T, U>
167 where
168 T: ?Sized + Send + Sync,
169 U: ?Sized + Send + Sync,
170 {
171 }
172 unsafe impl<T> Sync for RwLockWriteGuard<'_, T> where T: ?Sized + Send + Sync {}
173 unsafe impl<T> Sync for OwnedRwLockWriteGuard<T> where T: ?Sized + Send + Sync {}
174 unsafe impl<T> Sync for RwLockMappedWriteGuard<'_, T> where T: ?Sized + Send + Sync {}
175 unsafe impl<T, U> Sync for OwnedRwLockMappedWriteGuard<T, U>
176 where
177 T: ?Sized + Send + Sync,
178 U: ?Sized + Send + Sync,
179 {
180 }
181 // Safety: Stores a raw pointer to `T`, so if `T` is `Sync`, the lock guard over
182 // `T` is `Send` - but since this is also provides mutable access, we need to
183 // make sure that `T` is `Send` since its value can be sent across thread
184 // boundaries.
185 unsafe impl<T> Send for RwLockWriteGuard<'_, T> where T: ?Sized + Send + Sync {}
186 unsafe impl<T> Send for OwnedRwLockWriteGuard<T> where T: ?Sized + Send + Sync {}
187 unsafe impl<T> Send for RwLockMappedWriteGuard<'_, T> where T: ?Sized + Send + Sync {}
188 unsafe impl<T, U> Send for OwnedRwLockMappedWriteGuard<T, U>
189 where
190 T: ?Sized + Send + Sync,
191 U: ?Sized + Send + Sync,
192 {
193 }
194
195 impl<T: ?Sized> RwLock<T> {
196 /// Creates a new instance of an `RwLock<T>` which is unlocked.
197 ///
198 /// # Examples
199 ///
200 /// ```
201 /// use tokio::sync::RwLock;
202 ///
203 /// let lock = RwLock::new(5);
204 /// ```
205 #[track_caller]
new(value: T) -> RwLock<T> where T: Sized,206 pub fn new(value: T) -> RwLock<T>
207 where
208 T: Sized,
209 {
210 #[cfg(all(tokio_unstable, feature = "tracing"))]
211 let resource_span = {
212 let location = std::panic::Location::caller();
213 let resource_span = tracing::trace_span!(
214 "runtime.resource",
215 concrete_type = "RwLock",
216 kind = "Sync",
217 loc.file = location.file(),
218 loc.line = location.line(),
219 loc.col = location.column(),
220 );
221
222 resource_span.in_scope(|| {
223 tracing::trace!(
224 target: "runtime::resource::state_update",
225 max_readers = MAX_READS,
226 );
227
228 tracing::trace!(
229 target: "runtime::resource::state_update",
230 write_locked = false,
231 );
232
233 tracing::trace!(
234 target: "runtime::resource::state_update",
235 current_readers = 0,
236 );
237 });
238
239 resource_span
240 };
241
242 #[cfg(all(tokio_unstable, feature = "tracing"))]
243 let s = resource_span.in_scope(|| Semaphore::new(MAX_READS as usize));
244
245 #[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
246 let s = Semaphore::new(MAX_READS as usize);
247
248 RwLock {
249 mr: MAX_READS,
250 c: UnsafeCell::new(value),
251 s,
252 #[cfg(all(tokio_unstable, feature = "tracing"))]
253 resource_span,
254 }
255 }
256
257 /// Creates a new instance of an `RwLock<T>` which is unlocked
258 /// and allows a maximum of `max_reads` concurrent readers.
259 ///
260 /// # Examples
261 ///
262 /// ```
263 /// use tokio::sync::RwLock;
264 ///
265 /// let lock = RwLock::with_max_readers(5, 1024);
266 /// ```
267 ///
268 /// # Panics
269 ///
270 /// Panics if `max_reads` is more than `u32::MAX >> 3`.
271 #[track_caller]
with_max_readers(value: T, max_reads: u32) -> RwLock<T> where T: Sized,272 pub fn with_max_readers(value: T, max_reads: u32) -> RwLock<T>
273 where
274 T: Sized,
275 {
276 assert!(
277 max_reads <= MAX_READS,
278 "a RwLock may not be created with more than {} readers",
279 MAX_READS
280 );
281
282 #[cfg(all(tokio_unstable, feature = "tracing"))]
283 let resource_span = {
284 let location = std::panic::Location::caller();
285
286 let resource_span = tracing::trace_span!(
287 "runtime.resource",
288 concrete_type = "RwLock",
289 kind = "Sync",
290 loc.file = location.file(),
291 loc.line = location.line(),
292 loc.col = location.column(),
293 );
294
295 resource_span.in_scope(|| {
296 tracing::trace!(
297 target: "runtime::resource::state_update",
298 max_readers = max_reads,
299 );
300
301 tracing::trace!(
302 target: "runtime::resource::state_update",
303 write_locked = false,
304 );
305
306 tracing::trace!(
307 target: "runtime::resource::state_update",
308 current_readers = 0,
309 );
310 });
311
312 resource_span
313 };
314
315 #[cfg(all(tokio_unstable, feature = "tracing"))]
316 let s = resource_span.in_scope(|| Semaphore::new(max_reads as usize));
317
318 #[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
319 let s = Semaphore::new(max_reads as usize);
320
321 RwLock {
322 mr: max_reads,
323 c: UnsafeCell::new(value),
324 s,
325 #[cfg(all(tokio_unstable, feature = "tracing"))]
326 resource_span,
327 }
328 }
329
330 /// Creates a new instance of an `RwLock<T>` which is unlocked.
331 ///
332 /// # Examples
333 ///
334 /// ```
335 /// use tokio::sync::RwLock;
336 ///
337 /// static LOCK: RwLock<i32> = RwLock::const_new(5);
338 /// ```
339 #[cfg(all(feature = "parking_lot", not(all(loom, test))))]
340 #[cfg_attr(docsrs, doc(cfg(feature = "parking_lot")))]
const_new(value: T) -> RwLock<T> where T: Sized,341 pub const fn const_new(value: T) -> RwLock<T>
342 where
343 T: Sized,
344 {
345 RwLock {
346 mr: MAX_READS,
347 c: UnsafeCell::new(value),
348 s: Semaphore::const_new(MAX_READS as usize),
349 #[cfg(all(tokio_unstable, feature = "tracing"))]
350 resource_span: tracing::Span::none(),
351 }
352 }
353
354 /// Creates a new instance of an `RwLock<T>` which is unlocked
355 /// and allows a maximum of `max_reads` concurrent readers.
356 ///
357 /// # Examples
358 ///
359 /// ```
360 /// use tokio::sync::RwLock;
361 ///
362 /// static LOCK: RwLock<i32> = RwLock::const_with_max_readers(5, 1024);
363 /// ```
364 #[cfg(all(feature = "parking_lot", not(all(loom, test))))]
365 #[cfg_attr(docsrs, doc(cfg(feature = "parking_lot")))]
const_with_max_readers(value: T, mut max_reads: u32) -> RwLock<T> where T: Sized,366 pub const fn const_with_max_readers(value: T, mut max_reads: u32) -> RwLock<T>
367 where
368 T: Sized,
369 {
370 max_reads &= MAX_READS;
371 RwLock {
372 mr: max_reads,
373 c: UnsafeCell::new(value),
374 s: Semaphore::const_new(max_reads as usize),
375 #[cfg(all(tokio_unstable, feature = "tracing"))]
376 resource_span: tracing::Span::none(),
377 }
378 }
379
380 /// Locks this `RwLock` with shared read access, causing the current task
381 /// to yield until the lock has been acquired.
382 ///
383 /// The calling task will yield until there are no writers which hold the
384 /// lock. There may be other readers inside the lock when the task resumes.
385 ///
386 /// Note that under the priority policy of [`RwLock`], read locks are not
387 /// granted until prior write locks, to prevent starvation. Therefore
388 /// deadlock may occur if a read lock is held by the current task, a write
389 /// lock attempt is made, and then a subsequent read lock attempt is made
390 /// by the current task.
391 ///
392 /// Returns an RAII guard which will drop this read access of the `RwLock`
393 /// when dropped.
394 ///
395 /// # Cancel safety
396 ///
397 /// This method uses a queue to fairly distribute locks in the order they
398 /// were requested. Cancelling a call to `read` makes you lose your place in
399 /// the queue.
400 ///
401 /// # Examples
402 ///
403 /// ```
404 /// use std::sync::Arc;
405 /// use tokio::sync::RwLock;
406 ///
407 /// #[tokio::main]
408 /// async fn main() {
409 /// let lock = Arc::new(RwLock::new(1));
410 /// let c_lock = lock.clone();
411 ///
412 /// let n = lock.read().await;
413 /// assert_eq!(*n, 1);
414 ///
415 /// tokio::spawn(async move {
416 /// // While main has an active read lock, we acquire one too.
417 /// let r = c_lock.read().await;
418 /// assert_eq!(*r, 1);
419 /// }).await.expect("The spawned task has panicked");
420 ///
421 /// // Drop the guard after the spawned task finishes.
422 /// drop(n);
423 /// }
424 /// ```
read(&self) -> RwLockReadGuard<'_, T>425 pub async fn read(&self) -> RwLockReadGuard<'_, T> {
426 #[cfg(all(tokio_unstable, feature = "tracing"))]
427 let inner = trace::async_op(
428 || self.s.acquire(1),
429 self.resource_span.clone(),
430 "RwLock::read",
431 "poll",
432 false,
433 );
434
435 #[cfg(not(all(tokio_unstable, feature = "tracing")))]
436 let inner = self.s.acquire(1);
437
438 inner.await.unwrap_or_else(|_| {
439 // The semaphore was closed. but, we never explicitly close it, and we have a
440 // handle to it through the Arc, which means that this can never happen.
441 unreachable!()
442 });
443
444 #[cfg(all(tokio_unstable, feature = "tracing"))]
445 self.resource_span.in_scope(|| {
446 tracing::trace!(
447 target: "runtime::resource::state_update",
448 current_readers = 1,
449 current_readers.op = "add",
450 )
451 });
452
453 RwLockReadGuard {
454 s: &self.s,
455 data: self.c.get(),
456 marker: marker::PhantomData,
457 #[cfg(all(tokio_unstable, feature = "tracing"))]
458 resource_span: self.resource_span.clone(),
459 }
460 }
461
462 /// Blockingly locks this `RwLock` with shared read access.
463 ///
464 /// This method is intended for use cases where you
465 /// need to use this rwlock in asynchronous code as well as in synchronous code.
466 ///
467 /// Returns an RAII guard which will drop the read access of this `RwLock` when dropped.
468 ///
469 /// # Panics
470 ///
471 /// This function panics if called within an asynchronous execution context.
472 ///
473 /// - If you find yourself in an asynchronous execution context and needing
474 /// to call some (synchronous) function which performs one of these
475 /// `blocking_` operations, then consider wrapping that call inside
476 /// [`spawn_blocking()`][crate::runtime::Handle::spawn_blocking]
477 /// (or [`block_in_place()`][crate::task::block_in_place]).
478 ///
479 /// # Examples
480 ///
481 /// ```
482 /// use std::sync::Arc;
483 /// use tokio::sync::RwLock;
484 ///
485 /// #[tokio::main]
486 /// async fn main() {
487 /// let rwlock = Arc::new(RwLock::new(1));
488 /// let mut write_lock = rwlock.write().await;
489 ///
490 /// let blocking_task = tokio::task::spawn_blocking({
491 /// let rwlock = Arc::clone(&rwlock);
492 /// move || {
493 /// // This shall block until the `write_lock` is released.
494 /// let read_lock = rwlock.blocking_read();
495 /// assert_eq!(*read_lock, 0);
496 /// }
497 /// });
498 ///
499 /// *write_lock -= 1;
500 /// drop(write_lock); // release the lock.
501 ///
502 /// // Await the completion of the blocking task.
503 /// blocking_task.await.unwrap();
504 ///
505 /// // Assert uncontended.
506 /// assert!(rwlock.try_write().is_ok());
507 /// }
508 /// ```
509 #[track_caller]
510 #[cfg(feature = "sync")]
blocking_read(&self) -> RwLockReadGuard<'_, T>511 pub fn blocking_read(&self) -> RwLockReadGuard<'_, T> {
512 crate::future::block_on(self.read())
513 }
514
515 /// Locks this `RwLock` with shared read access, causing the current task
516 /// to yield until the lock has been acquired.
517 ///
518 /// The calling task will yield until there are no writers which hold the
519 /// lock. There may be other readers inside the lock when the task resumes.
520 ///
521 /// This method is identical to [`RwLock::read`], except that the returned
522 /// guard references the `RwLock` with an [`Arc`] rather than by borrowing
523 /// it. Therefore, the `RwLock` must be wrapped in an `Arc` to call this
524 /// method, and the guard will live for the `'static` lifetime, as it keeps
525 /// the `RwLock` alive by holding an `Arc`.
526 ///
527 /// Note that under the priority policy of [`RwLock`], read locks are not
528 /// granted until prior write locks, to prevent starvation. Therefore
529 /// deadlock may occur if a read lock is held by the current task, a write
530 /// lock attempt is made, and then a subsequent read lock attempt is made
531 /// by the current task.
532 ///
533 /// Returns an RAII guard which will drop this read access of the `RwLock`
534 /// when dropped.
535 ///
536 /// # Cancel safety
537 ///
538 /// This method uses a queue to fairly distribute locks in the order they
539 /// were requested. Cancelling a call to `read_owned` makes you lose your
540 /// place in the queue.
541 ///
542 /// # Examples
543 ///
544 /// ```
545 /// use std::sync::Arc;
546 /// use tokio::sync::RwLock;
547 ///
548 /// #[tokio::main]
549 /// async fn main() {
550 /// let lock = Arc::new(RwLock::new(1));
551 /// let c_lock = lock.clone();
552 ///
553 /// let n = lock.read_owned().await;
554 /// assert_eq!(*n, 1);
555 ///
556 /// tokio::spawn(async move {
557 /// // While main has an active read lock, we acquire one too.
558 /// let r = c_lock.read_owned().await;
559 /// assert_eq!(*r, 1);
560 /// }).await.expect("The spawned task has panicked");
561 ///
562 /// // Drop the guard after the spawned task finishes.
563 /// drop(n);
564 ///}
565 /// ```
read_owned(self: Arc<Self>) -> OwnedRwLockReadGuard<T>566 pub async fn read_owned(self: Arc<Self>) -> OwnedRwLockReadGuard<T> {
567 #[cfg(all(tokio_unstable, feature = "tracing"))]
568 let inner = trace::async_op(
569 || self.s.acquire(1),
570 self.resource_span.clone(),
571 "RwLock::read_owned",
572 "poll",
573 false,
574 );
575
576 #[cfg(not(all(tokio_unstable, feature = "tracing")))]
577 let inner = self.s.acquire(1);
578
579 inner.await.unwrap_or_else(|_| {
580 // The semaphore was closed. but, we never explicitly close it, and we have a
581 // handle to it through the Arc, which means that this can never happen.
582 unreachable!()
583 });
584
585 #[cfg(all(tokio_unstable, feature = "tracing"))]
586 self.resource_span.in_scope(|| {
587 tracing::trace!(
588 target: "runtime::resource::state_update",
589 current_readers = 1,
590 current_readers.op = "add",
591 )
592 });
593
594 #[cfg(all(tokio_unstable, feature = "tracing"))]
595 let resource_span = self.resource_span.clone();
596
597 OwnedRwLockReadGuard {
598 data: self.c.get(),
599 lock: ManuallyDrop::new(self),
600 _p: PhantomData,
601 #[cfg(all(tokio_unstable, feature = "tracing"))]
602 resource_span,
603 }
604 }
605
606 /// Attempts to acquire this `RwLock` with shared read access.
607 ///
608 /// If the access couldn't be acquired immediately, returns [`TryLockError`].
609 /// Otherwise, an RAII guard is returned which will release read access
610 /// when dropped.
611 ///
612 /// [`TryLockError`]: TryLockError
613 ///
614 /// # Examples
615 ///
616 /// ```
617 /// use std::sync::Arc;
618 /// use tokio::sync::RwLock;
619 ///
620 /// #[tokio::main]
621 /// async fn main() {
622 /// let lock = Arc::new(RwLock::new(1));
623 /// let c_lock = lock.clone();
624 ///
625 /// let v = lock.try_read().unwrap();
626 /// assert_eq!(*v, 1);
627 ///
628 /// tokio::spawn(async move {
629 /// // While main has an active read lock, we acquire one too.
630 /// let n = c_lock.read().await;
631 /// assert_eq!(*n, 1);
632 /// }).await.expect("The spawned task has panicked");
633 ///
634 /// // Drop the guard when spawned task finishes.
635 /// drop(v);
636 /// }
637 /// ```
try_read(&self) -> Result<RwLockReadGuard<'_, T>, TryLockError>638 pub fn try_read(&self) -> Result<RwLockReadGuard<'_, T>, TryLockError> {
639 match self.s.try_acquire(1) {
640 Ok(permit) => permit,
641 Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
642 Err(TryAcquireError::Closed) => unreachable!(),
643 }
644
645 #[cfg(all(tokio_unstable, feature = "tracing"))]
646 self.resource_span.in_scope(|| {
647 tracing::trace!(
648 target: "runtime::resource::state_update",
649 current_readers = 1,
650 current_readers.op = "add",
651 )
652 });
653
654 Ok(RwLockReadGuard {
655 s: &self.s,
656 data: self.c.get(),
657 marker: marker::PhantomData,
658 #[cfg(all(tokio_unstable, feature = "tracing"))]
659 resource_span: self.resource_span.clone(),
660 })
661 }
662
663 /// Attempts to acquire this `RwLock` with shared read access.
664 ///
665 /// If the access couldn't be acquired immediately, returns [`TryLockError`].
666 /// Otherwise, an RAII guard is returned which will release read access
667 /// when dropped.
668 ///
669 /// This method is identical to [`RwLock::try_read`], except that the
670 /// returned guard references the `RwLock` with an [`Arc`] rather than by
671 /// borrowing it. Therefore, the `RwLock` must be wrapped in an `Arc` to
672 /// call this method, and the guard will live for the `'static` lifetime,
673 /// as it keeps the `RwLock` alive by holding an `Arc`.
674 ///
675 /// [`TryLockError`]: TryLockError
676 ///
677 /// # Examples
678 ///
679 /// ```
680 /// use std::sync::Arc;
681 /// use tokio::sync::RwLock;
682 ///
683 /// #[tokio::main]
684 /// async fn main() {
685 /// let lock = Arc::new(RwLock::new(1));
686 /// let c_lock = lock.clone();
687 ///
688 /// let v = lock.try_read_owned().unwrap();
689 /// assert_eq!(*v, 1);
690 ///
691 /// tokio::spawn(async move {
692 /// // While main has an active read lock, we acquire one too.
693 /// let n = c_lock.read_owned().await;
694 /// assert_eq!(*n, 1);
695 /// }).await.expect("The spawned task has panicked");
696 ///
697 /// // Drop the guard when spawned task finishes.
698 /// drop(v);
699 /// }
700 /// ```
try_read_owned(self: Arc<Self>) -> Result<OwnedRwLockReadGuard<T>, TryLockError>701 pub fn try_read_owned(self: Arc<Self>) -> Result<OwnedRwLockReadGuard<T>, TryLockError> {
702 match self.s.try_acquire(1) {
703 Ok(permit) => permit,
704 Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
705 Err(TryAcquireError::Closed) => unreachable!(),
706 }
707
708 #[cfg(all(tokio_unstable, feature = "tracing"))]
709 self.resource_span.in_scope(|| {
710 tracing::trace!(
711 target: "runtime::resource::state_update",
712 current_readers = 1,
713 current_readers.op = "add",
714 )
715 });
716
717 #[cfg(all(tokio_unstable, feature = "tracing"))]
718 let resource_span = self.resource_span.clone();
719
720 Ok(OwnedRwLockReadGuard {
721 data: self.c.get(),
722 lock: ManuallyDrop::new(self),
723 _p: PhantomData,
724 #[cfg(all(tokio_unstable, feature = "tracing"))]
725 resource_span,
726 })
727 }
728
729 /// Locks this `RwLock` with exclusive write access, causing the current
730 /// task to yield until the lock has been acquired.
731 ///
732 /// The calling task will yield while other writers or readers currently
733 /// have access to the lock.
734 ///
735 /// Returns an RAII guard which will drop the write access of this `RwLock`
736 /// when dropped.
737 ///
738 /// # Cancel safety
739 ///
740 /// This method uses a queue to fairly distribute locks in the order they
741 /// were requested. Cancelling a call to `write` makes you lose your place
742 /// in the queue.
743 ///
744 /// # Examples
745 ///
746 /// ```
747 /// use tokio::sync::RwLock;
748 ///
749 /// #[tokio::main]
750 /// async fn main() {
751 /// let lock = RwLock::new(1);
752 ///
753 /// let mut n = lock.write().await;
754 /// *n = 2;
755 ///}
756 /// ```
write(&self) -> RwLockWriteGuard<'_, T>757 pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
758 #[cfg(all(tokio_unstable, feature = "tracing"))]
759 let inner = trace::async_op(
760 || self.s.acquire(self.mr),
761 self.resource_span.clone(),
762 "RwLock::write",
763 "poll",
764 false,
765 );
766
767 #[cfg(not(all(tokio_unstable, feature = "tracing")))]
768 let inner = self.s.acquire(self.mr);
769
770 inner.await.unwrap_or_else(|_| {
771 // The semaphore was closed. but, we never explicitly close it, and we have a
772 // handle to it through the Arc, which means that this can never happen.
773 unreachable!()
774 });
775
776 #[cfg(all(tokio_unstable, feature = "tracing"))]
777 self.resource_span.in_scope(|| {
778 tracing::trace!(
779 target: "runtime::resource::state_update",
780 write_locked = true,
781 write_locked.op = "override",
782 )
783 });
784
785 RwLockWriteGuard {
786 permits_acquired: self.mr,
787 s: &self.s,
788 data: self.c.get(),
789 marker: marker::PhantomData,
790 #[cfg(all(tokio_unstable, feature = "tracing"))]
791 resource_span: self.resource_span.clone(),
792 }
793 }
794
795 /// Blockingly locks this `RwLock` with exclusive write access.
796 ///
797 /// This method is intended for use cases where you
798 /// need to use this rwlock in asynchronous code as well as in synchronous code.
799 ///
800 /// Returns an RAII guard which will drop the write access of this `RwLock` when dropped.
801 ///
802 /// # Panics
803 ///
804 /// This function panics if called within an asynchronous execution context.
805 ///
806 /// - If you find yourself in an asynchronous execution context and needing
807 /// to call some (synchronous) function which performs one of these
808 /// `blocking_` operations, then consider wrapping that call inside
809 /// [`spawn_blocking()`][crate::runtime::Handle::spawn_blocking]
810 /// (or [`block_in_place()`][crate::task::block_in_place]).
811 ///
812 /// # Examples
813 ///
814 /// ```
815 /// use std::sync::Arc;
816 /// use tokio::{sync::RwLock};
817 ///
818 /// #[tokio::main]
819 /// async fn main() {
820 /// let rwlock = Arc::new(RwLock::new(1));
821 /// let read_lock = rwlock.read().await;
822 ///
823 /// let blocking_task = tokio::task::spawn_blocking({
824 /// let rwlock = Arc::clone(&rwlock);
825 /// move || {
826 /// // This shall block until the `read_lock` is released.
827 /// let mut write_lock = rwlock.blocking_write();
828 /// *write_lock = 2;
829 /// }
830 /// });
831 ///
832 /// assert_eq!(*read_lock, 1);
833 /// // Release the last outstanding read lock.
834 /// drop(read_lock);
835 ///
836 /// // Await the completion of the blocking task.
837 /// blocking_task.await.unwrap();
838 ///
839 /// // Assert uncontended.
840 /// let read_lock = rwlock.try_read().unwrap();
841 /// assert_eq!(*read_lock, 2);
842 /// }
843 /// ```
844 #[track_caller]
845 #[cfg(feature = "sync")]
blocking_write(&self) -> RwLockWriteGuard<'_, T>846 pub fn blocking_write(&self) -> RwLockWriteGuard<'_, T> {
847 crate::future::block_on(self.write())
848 }
849
850 /// Locks this `RwLock` with exclusive write access, causing the current
851 /// task to yield until the lock has been acquired.
852 ///
853 /// The calling task will yield while other writers or readers currently
854 /// have access to the lock.
855 ///
856 /// This method is identical to [`RwLock::write`], except that the returned
857 /// guard references the `RwLock` with an [`Arc`] rather than by borrowing
858 /// it. Therefore, the `RwLock` must be wrapped in an `Arc` to call this
859 /// method, and the guard will live for the `'static` lifetime, as it keeps
860 /// the `RwLock` alive by holding an `Arc`.
861 ///
862 /// Returns an RAII guard which will drop the write access of this `RwLock`
863 /// when dropped.
864 ///
865 /// # Cancel safety
866 ///
867 /// This method uses a queue to fairly distribute locks in the order they
868 /// were requested. Cancelling a call to `write_owned` makes you lose your
869 /// place in the queue.
870 ///
871 /// # Examples
872 ///
873 /// ```
874 /// use std::sync::Arc;
875 /// use tokio::sync::RwLock;
876 ///
877 /// #[tokio::main]
878 /// async fn main() {
879 /// let lock = Arc::new(RwLock::new(1));
880 ///
881 /// let mut n = lock.write_owned().await;
882 /// *n = 2;
883 ///}
884 /// ```
write_owned(self: Arc<Self>) -> OwnedRwLockWriteGuard<T>885 pub async fn write_owned(self: Arc<Self>) -> OwnedRwLockWriteGuard<T> {
886 #[cfg(all(tokio_unstable, feature = "tracing"))]
887 let inner = trace::async_op(
888 || self.s.acquire(self.mr),
889 self.resource_span.clone(),
890 "RwLock::write_owned",
891 "poll",
892 false,
893 );
894
895 #[cfg(not(all(tokio_unstable, feature = "tracing")))]
896 let inner = self.s.acquire(self.mr);
897
898 inner.await.unwrap_or_else(|_| {
899 // The semaphore was closed. but, we never explicitly close it, and we have a
900 // handle to it through the Arc, which means that this can never happen.
901 unreachable!()
902 });
903
904 #[cfg(all(tokio_unstable, feature = "tracing"))]
905 self.resource_span.in_scope(|| {
906 tracing::trace!(
907 target: "runtime::resource::state_update",
908 write_locked = true,
909 write_locked.op = "override",
910 )
911 });
912
913 #[cfg(all(tokio_unstable, feature = "tracing"))]
914 let resource_span = self.resource_span.clone();
915
916 OwnedRwLockWriteGuard {
917 permits_acquired: self.mr,
918 data: self.c.get(),
919 lock: ManuallyDrop::new(self),
920 _p: PhantomData,
921 #[cfg(all(tokio_unstable, feature = "tracing"))]
922 resource_span,
923 }
924 }
925
926 /// Attempts to acquire this `RwLock` with exclusive write access.
927 ///
928 /// If the access couldn't be acquired immediately, returns [`TryLockError`].
929 /// Otherwise, an RAII guard is returned which will release write access
930 /// when dropped.
931 ///
932 /// [`TryLockError`]: TryLockError
933 ///
934 /// # Examples
935 ///
936 /// ```
937 /// use tokio::sync::RwLock;
938 ///
939 /// #[tokio::main]
940 /// async fn main() {
941 /// let rw = RwLock::new(1);
942 ///
943 /// let v = rw.read().await;
944 /// assert_eq!(*v, 1);
945 ///
946 /// assert!(rw.try_write().is_err());
947 /// }
948 /// ```
try_write(&self) -> Result<RwLockWriteGuard<'_, T>, TryLockError>949 pub fn try_write(&self) -> Result<RwLockWriteGuard<'_, T>, TryLockError> {
950 match self.s.try_acquire(self.mr) {
951 Ok(permit) => permit,
952 Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
953 Err(TryAcquireError::Closed) => unreachable!(),
954 }
955
956 #[cfg(all(tokio_unstable, feature = "tracing"))]
957 self.resource_span.in_scope(|| {
958 tracing::trace!(
959 target: "runtime::resource::state_update",
960 write_locked = true,
961 write_locked.op = "override",
962 )
963 });
964
965 Ok(RwLockWriteGuard {
966 permits_acquired: self.mr,
967 s: &self.s,
968 data: self.c.get(),
969 marker: marker::PhantomData,
970 #[cfg(all(tokio_unstable, feature = "tracing"))]
971 resource_span: self.resource_span.clone(),
972 })
973 }
974
975 /// Attempts to acquire this `RwLock` with exclusive write access.
976 ///
977 /// If the access couldn't be acquired immediately, returns [`TryLockError`].
978 /// Otherwise, an RAII guard is returned which will release write access
979 /// when dropped.
980 ///
981 /// This method is identical to [`RwLock::try_write`], except that the
982 /// returned guard references the `RwLock` with an [`Arc`] rather than by
983 /// borrowing it. Therefore, the `RwLock` must be wrapped in an `Arc` to
984 /// call this method, and the guard will live for the `'static` lifetime,
985 /// as it keeps the `RwLock` alive by holding an `Arc`.
986 ///
987 /// [`TryLockError`]: TryLockError
988 ///
989 /// # Examples
990 ///
991 /// ```
992 /// use std::sync::Arc;
993 /// use tokio::sync::RwLock;
994 ///
995 /// #[tokio::main]
996 /// async fn main() {
997 /// let rw = Arc::new(RwLock::new(1));
998 ///
999 /// let v = Arc::clone(&rw).read_owned().await;
1000 /// assert_eq!(*v, 1);
1001 ///
1002 /// assert!(rw.try_write_owned().is_err());
1003 /// }
1004 /// ```
try_write_owned(self: Arc<Self>) -> Result<OwnedRwLockWriteGuard<T>, TryLockError>1005 pub fn try_write_owned(self: Arc<Self>) -> Result<OwnedRwLockWriteGuard<T>, TryLockError> {
1006 match self.s.try_acquire(self.mr) {
1007 Ok(permit) => permit,
1008 Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
1009 Err(TryAcquireError::Closed) => unreachable!(),
1010 }
1011
1012 #[cfg(all(tokio_unstable, feature = "tracing"))]
1013 self.resource_span.in_scope(|| {
1014 tracing::trace!(
1015 target: "runtime::resource::state_update",
1016 write_locked = true,
1017 write_locked.op = "override",
1018 )
1019 });
1020
1021 #[cfg(all(tokio_unstable, feature = "tracing"))]
1022 let resource_span = self.resource_span.clone();
1023
1024 Ok(OwnedRwLockWriteGuard {
1025 permits_acquired: self.mr,
1026 data: self.c.get(),
1027 lock: ManuallyDrop::new(self),
1028 _p: PhantomData,
1029 #[cfg(all(tokio_unstable, feature = "tracing"))]
1030 resource_span,
1031 })
1032 }
1033
1034 /// Returns a mutable reference to the underlying data.
1035 ///
1036 /// Since this call borrows the `RwLock` mutably, no actual locking needs to
1037 /// take place -- the mutable borrow statically guarantees no locks exist.
1038 ///
1039 /// # Examples
1040 ///
1041 /// ```
1042 /// use tokio::sync::RwLock;
1043 ///
1044 /// fn main() {
1045 /// let mut lock = RwLock::new(1);
1046 ///
1047 /// let n = lock.get_mut();
1048 /// *n = 2;
1049 /// }
1050 /// ```
get_mut(&mut self) -> &mut T1051 pub fn get_mut(&mut self) -> &mut T {
1052 unsafe {
1053 // Safety: This is https://github.com/rust-lang/rust/pull/76936
1054 &mut *self.c.get()
1055 }
1056 }
1057
1058 /// Consumes the lock, returning the underlying data.
into_inner(self) -> T where T: Sized,1059 pub fn into_inner(self) -> T
1060 where
1061 T: Sized,
1062 {
1063 self.c.into_inner()
1064 }
1065 }
1066
1067 impl<T> From<T> for RwLock<T> {
from(s: T) -> Self1068 fn from(s: T) -> Self {
1069 Self::new(s)
1070 }
1071 }
1072
1073 impl<T: ?Sized> Default for RwLock<T>
1074 where
1075 T: Default,
1076 {
default() -> Self1077 fn default() -> Self {
1078 Self::new(T::default())
1079 }
1080 }
1081