LeftRight.h 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229
  1. #pragma once
  2. #include <c10/macros/Macros.h>
  3. #include <c10/util/Synchronized.h>
  4. #include <array>
  5. #include <atomic>
  6. #include <mutex>
  7. #include <thread>
  8. namespace c10 {
  9. namespace detail {
  10. struct IncrementRAII final {
  11. public:
  12. explicit IncrementRAII(std::atomic<int32_t>* counter) : _counter(counter) {
  13. _counter->fetch_add(1);
  14. }
  15. ~IncrementRAII() {
  16. _counter->fetch_sub(1);
  17. }
  18. IncrementRAII(IncrementRAII&&) = delete;
  19. IncrementRAII& operator=(IncrementRAII&&) = delete;
  20. private:
  21. std::atomic<int32_t>* _counter;
  22. C10_DISABLE_COPY_AND_ASSIGN(IncrementRAII);
  23. };
  24. } // namespace detail
  25. // LeftRight wait-free readers synchronization primitive
  26. // https://hal.archives-ouvertes.fr/hal-01207881/document
  27. //
  28. // LeftRight is quite easy to use (it can make an arbitrary
  29. // data structure permit wait-free reads), but it has some
  30. // particular performance characteristics you should be aware
  31. // of if you're deciding to use it:
  32. //
  33. // - Reads still incur an atomic write (this is how LeftRight
  34. // keeps track of how long it needs to keep around the old
  35. // data structure)
  36. //
  37. // - Writes get executed twice, to keep both the left and right
  38. // versions up to date. So if your write is expensive or
  39. // nondeterministic, this is also an inappropriate structure
  40. //
  41. // LeftRight is used fairly rarely in PyTorch's codebase. If you
  42. // are still not sure if you need it or not, consult your local
  43. // C++ expert.
  44. //
  45. template <class T>
  46. class LeftRight final {
  47. public:
  48. template <class... Args>
  49. explicit LeftRight(const Args&... args)
  50. : _counters{{{0}, {0}}},
  51. _foregroundCounterIndex(0),
  52. _foregroundDataIndex(0),
  53. _data{{T{args...}, T{args...}}} {}
  54. // Copying and moving would not be threadsafe.
  55. // Needs more thought and careful design to make that work.
  56. LeftRight(const LeftRight&) = delete;
  57. LeftRight(LeftRight&&) noexcept = delete;
  58. LeftRight& operator=(const LeftRight&) = delete;
  59. LeftRight& operator=(LeftRight&&) noexcept = delete;
  60. ~LeftRight() {
  61. // wait until any potentially running writers are finished
  62. {
  63. std::unique_lock<std::mutex> lock(_writeMutex);
  64. }
  65. // wait until any potentially running readers are finished
  66. while (_counters[0].load() != 0 || _counters[1].load() != 0) {
  67. std::this_thread::yield();
  68. }
  69. }
  70. template <typename F>
  71. auto read(F&& readFunc) const {
  72. detail::IncrementRAII _increment_counter(
  73. &_counters[_foregroundCounterIndex.load()]);
  74. return std::forward<F>(readFunc)(_data[_foregroundDataIndex.load()]);
  75. }
  76. // Throwing an exception in writeFunc is ok but causes the state to be either
  77. // the old or the new state, depending on if the first or the second call to
  78. // writeFunc threw.
  79. template <typename F>
  80. auto write(F&& writeFunc) {
  81. std::unique_lock<std::mutex> lock(_writeMutex);
  82. return _write(std::forward<F>(writeFunc));
  83. }
  84. private:
  85. template <class F>
  86. auto _write(const F& writeFunc) {
  87. /*
  88. * Assume, A is in background and B in foreground. In simplified terms, we
  89. * want to do the following:
  90. * 1. Write to A (old background)
  91. * 2. Switch A/B
  92. * 3. Write to B (new background)
  93. *
  94. * More detailed algorithm (explanations on why this is important are below
  95. * in code):
  96. * 1. Write to A
  97. * 2. Switch A/B data pointers
  98. * 3. Wait until A counter is zero
  99. * 4. Switch A/B counters
  100. * 5. Wait until B counter is zero
  101. * 6. Write to B
  102. */
  103. auto localDataIndex = _foregroundDataIndex.load();
  104. // 1. Write to A
  105. _callWriteFuncOnBackgroundInstance(writeFunc, localDataIndex);
  106. // 2. Switch A/B data pointers
  107. localDataIndex = localDataIndex ^ 1;
  108. _foregroundDataIndex = localDataIndex;
  109. /*
  110. * 3. Wait until A counter is zero
  111. *
  112. * In the previous write run, A was foreground and B was background.
  113. * There was a time after switching _foregroundDataIndex (B to foreground)
  114. * and before switching _foregroundCounterIndex, in which new readers could
  115. * have read B but incremented A's counter.
  116. *
  117. * In this current run, we just switched _foregroundDataIndex (A back to
  118. * foreground), but before writing to the new background B, we have to make
  119. * sure A's counter was zero briefly, so all these old readers are gone.
  120. */
  121. auto localCounterIndex = _foregroundCounterIndex.load();
  122. _waitForBackgroundCounterToBeZero(localCounterIndex);
  123. /*
  124. * 4. Switch A/B counters
  125. *
  126. * Now that we know all readers on B are really gone, we can switch the
  127. * counters and have new readers increment A's counter again, which is the
  128. * correct counter since they're reading A.
  129. */
  130. localCounterIndex = localCounterIndex ^ 1;
  131. _foregroundCounterIndex = localCounterIndex;
  132. /*
  133. * 5. Wait until B counter is zero
  134. *
  135. * This waits for all the readers on B that came in while both data and
  136. * counter for B was in foreground, i.e. normal readers that happened
  137. * outside of that brief gap between switching data and counter.
  138. */
  139. _waitForBackgroundCounterToBeZero(localCounterIndex);
  140. // 6. Write to B
  141. return _callWriteFuncOnBackgroundInstance(writeFunc, localDataIndex);
  142. }
  143. template <class F>
  144. auto _callWriteFuncOnBackgroundInstance(
  145. const F& writeFunc,
  146. uint8_t localDataIndex) {
  147. try {
  148. return writeFunc(_data[localDataIndex ^ 1]);
  149. } catch (...) {
  150. // recover invariant by copying from the foreground instance
  151. _data[localDataIndex ^ 1] = _data[localDataIndex];
  152. // rethrow
  153. throw;
  154. }
  155. }
  156. void _waitForBackgroundCounterToBeZero(uint8_t counterIndex) {
  157. while (_counters[counterIndex ^ 1].load() != 0) {
  158. std::this_thread::yield();
  159. }
  160. }
  161. mutable std::array<std::atomic<int32_t>, 2> _counters;
  162. std::atomic<uint8_t> _foregroundCounterIndex;
  163. std::atomic<uint8_t> _foregroundDataIndex;
  164. std::array<T, 2> _data;
  165. std::mutex _writeMutex;
  166. };
  167. // RWSafeLeftRightWrapper is API compatible with LeftRight and uses a
  168. // read-write lock to protect T (data).
  169. template <class T>
  170. class RWSafeLeftRightWrapper final {
  171. public:
  172. template <class... Args>
  173. explicit RWSafeLeftRightWrapper(const Args&... args) : data_{args...} {}
  174. // RWSafeLeftRightWrapper is not copyable or moveable since LeftRight
  175. // is not copyable or moveable.
  176. RWSafeLeftRightWrapper(const RWSafeLeftRightWrapper&) = delete;
  177. RWSafeLeftRightWrapper(RWSafeLeftRightWrapper&&) noexcept = delete;
  178. RWSafeLeftRightWrapper& operator=(const RWSafeLeftRightWrapper&) = delete;
  179. RWSafeLeftRightWrapper& operator=(RWSafeLeftRightWrapper&&) noexcept = delete;
  180. ~RWSafeLeftRightWrapper() = default;
  181. template <typename F>
  182. // NOLINTNEXTLINE(cppcoreguidelines-missing-std-forward)
  183. auto read(F&& readFunc) const {
  184. return data_.withLock(
  185. [&readFunc](T const& data) { return std::forward<F>(readFunc)(data); });
  186. }
  187. template <typename F>
  188. // NOLINTNEXTLINE(cppcoreguidelines-missing-std-forward)
  189. auto write(F&& writeFunc) {
  190. return data_.withLock(
  191. [&writeFunc](T& data) { return std::forward<F>(writeFunc)(data); });
  192. }
  193. private:
  194. c10::Synchronized<T> data_;
  195. };
  196. } // namespace c10