1 /* 2 Copyright (c) 2005-2021 Intel Corporation 3 4 Licensed under the Apache License, Version 2.0 (the "License"); 5 you may not use this file except in compliance with the License. 6 You may obtain a copy of the License at 7 8 http://www.apache.org/licenses/LICENSE-2.0 9 10 Unless required by applicable law or agreed to in writing, software 11 distributed under the License is distributed on an "AS IS" BASIS, 12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 See the License for the specific language governing permissions and 14 limitations under the License. 15 */ 16 17 /** Before making any changes in the implementation, please emulate algorithmic changes 18 with SPIN tool using <TBB directory>/tools/spin_models/ReaderWriterMutex.pml. 19 There could be some code looking as "can be restructured" but its structure does matter! */ 20 21 #include "oneapi/tbb/queuing_rw_mutex.h" 22 #include "oneapi/tbb/detail/_assert.h" 23 #include "oneapi/tbb/detail/_utils.h" 24 #include "itt_notify.h" 25 26 namespace tbb { 27 namespace detail { 28 namespace r1 { 29 30 #if defined(_MSC_VER) && !defined(__INTEL_COMPILER) 31 // Workaround for overzealous compiler warnings 32 #pragma warning (push) 33 #pragma warning (disable: 4311 4312) 34 #endif 35 36 //! A view of a T* with additional functionality for twiddling low-order bits. 37 template<typename T> 38 class tricky_atomic_pointer { 39 public: 40 using word = uintptr_t; 41 42 static T* fetch_add( std::atomic<word>& location, word addend, std::memory_order memory_order ) { 43 return reinterpret_cast<T*>(location.fetch_add(addend, memory_order)); 44 } 45 46 static T* exchange( std::atomic<word>& location, T* value, std::memory_order memory_order ) { 47 return reinterpret_cast<T*>(location.exchange(reinterpret_cast<word>(value), memory_order)); 48 } 49 50 static T* compare_exchange_strong( std::atomic<word>& obj, const T* expected, const T* desired, std::memory_order memory_order ) { 51 word expd = reinterpret_cast<word>(expected); 52 obj.compare_exchange_strong(expd, reinterpret_cast<word>(desired), memory_order); 53 return reinterpret_cast<T*>(expd); 54 } 55 56 static void store( std::atomic<word>& location, const T* value, std::memory_order memory_order ) { 57 location.store(reinterpret_cast<word>(value), memory_order); 58 } 59 60 static T* load( std::atomic<word>& location, std::memory_order memory_order ) { 61 return reinterpret_cast<T*>(location.load(memory_order)); 62 } 63 64 static void spin_wait_while_eq(const std::atomic<word>& location, const T* value) { 65 tbb::detail::d0::spin_wait_while_eq(location, reinterpret_cast<word>(value) ); 66 } 67 68 T* & ref; 69 tricky_atomic_pointer( T*& original ) : ref(original) {}; 70 tricky_atomic_pointer(const tricky_atomic_pointer&) = delete; 71 tricky_atomic_pointer& operator=(const tricky_atomic_pointer&) = delete; 72 T* operator&( const word operand2 ) const { 73 return reinterpret_cast<T*>( reinterpret_cast<word>(ref) & operand2 ); 74 } 75 T* operator|( const word operand2 ) const { 76 return reinterpret_cast<T*>( reinterpret_cast<word>(ref) | operand2 ); 77 } 78 }; 79 80 using tricky_pointer = tricky_atomic_pointer<queuing_rw_mutex::scoped_lock>; 81 82 #if defined(_MSC_VER) && !defined(__INTEL_COMPILER) 83 // Workaround for overzealous compiler warnings 84 #pragma warning (pop) 85 #endif 86 87 //! Flag bits in a state_t that specify information about a locking request. 88 enum state_t_flags : unsigned char { 89 STATE_NONE = 0, 90 STATE_WRITER = 1<<0, 91 STATE_READER = 1<<1, 92 STATE_READER_UNBLOCKNEXT = 1<<2, 93 STATE_ACTIVEREADER = 1<<3, 94 STATE_UPGRADE_REQUESTED = 1<<4, 95 STATE_UPGRADE_WAITING = 1<<5, 96 STATE_UPGRADE_LOSER = 1<<6, 97 STATE_COMBINED_WAITINGREADER = STATE_READER | STATE_READER_UNBLOCKNEXT, 98 STATE_COMBINED_READER = STATE_COMBINED_WAITINGREADER | STATE_ACTIVEREADER, 99 STATE_COMBINED_UPGRADING = STATE_UPGRADE_WAITING | STATE_UPGRADE_LOSER 100 }; 101 102 static const unsigned char RELEASED = 0; 103 static const unsigned char ACQUIRED = 1; 104 105 struct queuing_rw_mutex_impl { 106 //! Try to acquire the internal lock 107 /** Returns true if lock was successfully acquired. */ 108 static bool try_acquire_internal_lock(d1::queuing_rw_mutex::scoped_lock& s) 109 { 110 auto expected = RELEASED; 111 return s.my_internal_lock.compare_exchange_strong(expected, ACQUIRED); 112 } 113 114 //! Acquire the internal lock 115 static void acquire_internal_lock(d1::queuing_rw_mutex::scoped_lock& s) 116 { 117 // Usually, we would use the test-test-and-set idiom here, with exponential backoff. 118 // But so far, experiments indicate there is no value in doing so here. 119 while( !try_acquire_internal_lock(s) ) { 120 machine_pause(1); 121 } 122 } 123 124 //! Release the internal lock 125 static void release_internal_lock(d1::queuing_rw_mutex::scoped_lock& s) 126 { 127 s.my_internal_lock.store(RELEASED, std::memory_order_release); 128 } 129 130 //! Wait for internal lock to be released 131 static void wait_for_release_of_internal_lock(d1::queuing_rw_mutex::scoped_lock& s) 132 { 133 spin_wait_until_eq(s.my_internal_lock, RELEASED); 134 } 135 136 //! A helper function 137 static void unblock_or_wait_on_internal_lock(d1::queuing_rw_mutex::scoped_lock& s, uintptr_t flag ) { 138 if( flag ) { 139 wait_for_release_of_internal_lock(s); 140 } 141 else { 142 release_internal_lock(s); 143 } 144 } 145 146 //! Mask for low order bit of a pointer. 147 static const tricky_pointer::word FLAG = 0x1; 148 149 static uintptr_t get_flag( d1::queuing_rw_mutex::scoped_lock* ptr ) { 150 return reinterpret_cast<uintptr_t>(ptr) & FLAG; 151 } 152 153 //------------------------------------------------------------------------ 154 // Methods of queuing_rw_mutex::scoped_lock 155 //------------------------------------------------------------------------ 156 157 //! A method to acquire queuing_rw_mutex lock 158 static void acquire(d1::queuing_rw_mutex& m, d1::queuing_rw_mutex::scoped_lock& s, bool write) 159 { 160 __TBB_ASSERT( !s.my_mutex, "scoped_lock is already holding a mutex"); 161 162 // Must set all fields before the exchange, because once the 163 // exchange executes, *this becomes accessible to other threads. 164 s.my_mutex = &m; 165 s.my_prev.store(0U, std::memory_order_relaxed); 166 s.my_next.store(0U, std::memory_order_relaxed); 167 s.my_going.store(0U, std::memory_order_relaxed); 168 s.my_state.store(d1::queuing_rw_mutex::scoped_lock::state_t(write ? STATE_WRITER : STATE_READER), std::memory_order_relaxed); 169 s.my_internal_lock.store(RELEASED, std::memory_order_relaxed); 170 171 queuing_rw_mutex::scoped_lock* predecessor = m.q_tail.exchange(&s, std::memory_order_release); 172 173 if( write ) { // Acquiring for write 174 175 if( predecessor ) { 176 ITT_NOTIFY(sync_prepare, s.my_mutex); 177 predecessor = tricky_pointer(predecessor) & ~FLAG; 178 __TBB_ASSERT( !predecessor->my_next, "the predecessor has another successor!"); 179 tricky_pointer::store(predecessor->my_next, &s, std::memory_order_release); 180 spin_wait_until_eq(s.my_going, 1U); 181 } 182 183 } else { // Acquiring for read 184 #if __TBB_USE_ITT_NOTIFY 185 bool sync_prepare_done = false; 186 #endif 187 if( predecessor ) { 188 unsigned char pred_state{}; 189 __TBB_ASSERT( !s.my_prev.load(std::memory_order_relaxed), "the predecessor is already set" ); 190 if( tricky_pointer(predecessor) & FLAG ) { 191 /* this is only possible if predecessor is an upgrading reader and it signals us to wait */ 192 pred_state = STATE_UPGRADE_WAITING; 193 predecessor = tricky_pointer(predecessor) & ~FLAG; 194 } else { 195 // Load predecessor->my_state now, because once predecessor->my_next becomes 196 // non-NULL, we must assume that *predecessor might be destroyed. 197 pred_state = STATE_READER; 198 predecessor->my_state.compare_exchange_strong(pred_state, STATE_READER_UNBLOCKNEXT, std::memory_order_acq_rel); 199 } 200 tricky_pointer::store(s.my_prev, predecessor, std::memory_order_relaxed); // TODO: order? 201 __TBB_ASSERT( !( tricky_pointer(predecessor) & FLAG ), "use of corrupted pointer!" ); 202 __TBB_ASSERT( !predecessor->my_next.load(std::memory_order_relaxed), "the predecessor has another successor!"); 203 tricky_pointer::store(predecessor->my_next, &s, std::memory_order_release); 204 if( pred_state != STATE_ACTIVEREADER ) { 205 #if __TBB_USE_ITT_NOTIFY 206 sync_prepare_done = true; 207 ITT_NOTIFY(sync_prepare, s.my_mutex); 208 #endif 209 spin_wait_until_eq(s.my_going, 1U, std::memory_order_acquire); 210 } 211 } 212 213 // The protected state must have been acquired here before it can be further released to any other reader(s): 214 unsigned char old_state = STATE_READER; 215 s.my_state.compare_exchange_strong(old_state, STATE_ACTIVEREADER, std::memory_order_acq_rel); 216 if( old_state!=STATE_READER ) { 217 #if __TBB_USE_ITT_NOTIFY 218 if( !sync_prepare_done ) 219 ITT_NOTIFY(sync_prepare, s.my_mutex); 220 #endif 221 // Failed to become active reader -> need to unblock the next waiting reader first 222 __TBB_ASSERT( s.my_state.load(std::memory_order_relaxed)==STATE_READER_UNBLOCKNEXT, "unexpected state" ); 223 spin_wait_while_eq(s.my_next, 0U); 224 /* my_state should be changed before unblocking the next otherwise it might finish 225 and another thread can get our old state and left blocked */ 226 s.my_state.store(STATE_ACTIVEREADER, std::memory_order_relaxed); 227 tricky_pointer::load(s.my_next, std::memory_order_relaxed)->my_going.store(1U, std::memory_order_release); 228 } 229 __TBB_ASSERT( s.my_state.load(std::memory_order_relaxed) ==STATE_ACTIVEREADER, "unlocked reader is active reader" ); 230 } 231 232 ITT_NOTIFY(sync_acquired, s.my_mutex); 233 234 // Force acquire so that user's critical section receives correct values 235 // from processor that was previously in the user's critical section. 236 atomic_fence(std::memory_order_acquire); // TODO: consider exchange on line 171 237 } 238 239 //! A method to acquire queuing_rw_mutex if it is free 240 static bool try_acquire(d1::queuing_rw_mutex& m, d1::queuing_rw_mutex::scoped_lock& s, bool write) 241 { 242 __TBB_ASSERT( !s.my_mutex, "scoped_lock is already holding a mutex"); 243 244 if( m.q_tail.load(std::memory_order_relaxed) ) 245 return false; // Someone already took the lock 246 247 // Must set all fields before the exchange, because once the 248 // exchange executes, *this becomes accessible to other threads. 249 s.my_prev.store(0U, std::memory_order_relaxed); 250 s.my_next.store(0U, std::memory_order_relaxed); 251 s.my_going.store(0U, std::memory_order_relaxed); // TODO: remove dead assignment? 252 s.my_state.store(d1::queuing_rw_mutex::scoped_lock::state_t(write ? STATE_WRITER : STATE_ACTIVEREADER), std::memory_order_relaxed); 253 s.my_internal_lock.store(RELEASED, std::memory_order_relaxed); 254 255 // The CAS must have release semantics, because we are 256 // "sending" the fields initialized above to other processors. 257 d1::queuing_rw_mutex::scoped_lock* expected = nullptr; 258 if (!m.q_tail.compare_exchange_strong(expected, &s, std::memory_order_release)) 259 return false; // Someone already took the lock 260 // Force acquire so that user's critical section receives correct values 261 // from processor that was previously in the user's critical section. 262 atomic_fence(std::memory_order_acquire); 263 s.my_mutex = &m; 264 ITT_NOTIFY(sync_acquired, s.my_mutex); 265 return true; 266 } 267 268 //! A method to release queuing_rw_mutex lock 269 static void release(d1::queuing_rw_mutex::scoped_lock& s) { 270 __TBB_ASSERT(s.my_mutex!=nullptr, "no lock acquired"); 271 272 ITT_NOTIFY(sync_releasing, s.my_mutex); 273 274 if( s.my_state.load(std::memory_order_relaxed) == STATE_WRITER ) { // Acquired for write 275 276 // The logic below is the same as "writerUnlock", but elides 277 // "return" from the middle of the routine. 278 // In the statement below, acquire semantics of reading my_next is required 279 // so that following operations with fields of my_next are safe. 280 d1::queuing_rw_mutex::scoped_lock* next = tricky_pointer::load(s.my_next, std::memory_order_acquire); 281 if( !next ) { 282 d1::queuing_rw_mutex::scoped_lock* expected = &s; 283 if( s.my_mutex->q_tail.compare_exchange_strong(expected, nullptr, std::memory_order_release) ) { 284 // this was the only item in the queue, and the queue is now empty. 285 goto done; 286 } 287 spin_wait_while_eq(s.my_next, 0U); 288 next = tricky_pointer::load(s.my_next, std::memory_order_acquire); 289 } 290 next->my_going.store(2U, std::memory_order_relaxed); // protect next queue node from being destroyed too early 291 if( next->my_state.load()==STATE_UPGRADE_WAITING ) { 292 // the next waiting for upgrade means this writer was upgraded before. 293 acquire_internal_lock(s); 294 // Responsibility transition, the one who reads uncorrupted my_prev will do release. 295 d1::queuing_rw_mutex::scoped_lock* tmp = tricky_pointer::exchange(next->my_prev, nullptr, std::memory_order_release); 296 next->my_state.store(STATE_UPGRADE_LOSER, std::memory_order_relaxed); 297 next->my_going.store(1U, std::memory_order_release); 298 unblock_or_wait_on_internal_lock(s, get_flag(tmp)); 299 } else { 300 // next->state cannot be STATE_UPGRADE_REQUESTED 301 __TBB_ASSERT( next->my_state.load(std::memory_order_relaxed) & (STATE_COMBINED_WAITINGREADER | STATE_WRITER), "unexpected state" ); 302 __TBB_ASSERT( !( next->my_prev.load(std::memory_order_relaxed) & FLAG ), "use of corrupted pointer!" ); 303 tricky_pointer::store(next->my_prev, nullptr, std::memory_order_relaxed); 304 next->my_going.store(1U, std::memory_order_release); 305 } 306 307 } else { // Acquired for read 308 309 queuing_rw_mutex::scoped_lock *tmp = nullptr; 310 retry: 311 // Addition to the original paper: Mark my_prev as in use 312 queuing_rw_mutex::scoped_lock *predecessor = tricky_pointer::fetch_add(s.my_prev, FLAG, std::memory_order_acquire); 313 314 if( predecessor ) { 315 if( !(try_acquire_internal_lock(*predecessor)) ) 316 { 317 // Failed to acquire the lock on predecessor. The predecessor either unlinks or upgrades. 318 // In the second case, it could or could not know my "in use" flag - need to check 319 // Responsibility transition, the one who reads uncorrupted my_prev will do release. 320 tmp = tricky_pointer::compare_exchange_strong(s.my_prev, tricky_pointer(predecessor) | FLAG, predecessor, std::memory_order_release); 321 if( !(tricky_pointer(tmp) & FLAG) ) { 322 // Wait for the predecessor to change my_prev (e.g. during unlink) 323 // TODO: spin_wait condition seems never reachable 324 __TBB_ASSERT(tricky_pointer::load(s.my_prev, std::memory_order_relaxed) != (tricky_pointer(predecessor) | FLAG), nullptr); 325 tricky_pointer::spin_wait_while_eq( s.my_prev, tricky_pointer(predecessor)|FLAG ); 326 // Now owner of predecessor is waiting for _us_ to release its lock 327 release_internal_lock(*predecessor); 328 } 329 // else the "in use" flag is back -> the predecessor didn't get it and will release itself; nothing to do 330 331 tmp = nullptr; 332 goto retry; 333 } 334 __TBB_ASSERT(predecessor && predecessor->my_internal_lock.load(std::memory_order_relaxed)==ACQUIRED, "predecessor's lock is not acquired"); 335 tricky_pointer::store(s.my_prev, predecessor, std::memory_order_relaxed); 336 acquire_internal_lock(s); 337 338 tricky_pointer::store(predecessor->my_next, nullptr, std::memory_order_release); 339 340 d1::queuing_rw_mutex::scoped_lock* expected = &s; 341 if( !tricky_pointer::load(s.my_next, std::memory_order_relaxed) && !s.my_mutex->q_tail.compare_exchange_strong(expected, predecessor, std::memory_order_release) ) { 342 spin_wait_while_eq( s.my_next, 0U, std::memory_order_acquire ); 343 } 344 __TBB_ASSERT( !(s.my_next.load(std::memory_order_relaxed) & FLAG), "use of corrupted pointer" ); 345 346 // ensure acquire semantics of reading 'my_next' 347 if(d1::queuing_rw_mutex::scoped_lock *const l_next = tricky_pointer::load(s.my_next, std::memory_order_acquire) ) { // I->next != nil, TODO: rename to next after clearing up and adapting the n in the comment two lines below 348 // Equivalent to I->next->prev = I->prev but protected against (prev[n]&FLAG)!=0 349 tmp = tricky_pointer::exchange(l_next->my_prev, predecessor, std::memory_order_release); 350 // I->prev->next = I->next; 351 __TBB_ASSERT(tricky_pointer::load(s.my_prev, std::memory_order_relaxed)==predecessor, nullptr); 352 predecessor->my_next.store(s.my_next.load(std::memory_order_relaxed), std::memory_order_release); 353 } 354 // Safe to release in the order opposite to acquiring which makes the code simpler 355 release_internal_lock(*predecessor); 356 357 } else { // No predecessor when we looked 358 acquire_internal_lock(s); // "exclusiveLock(&I->EL)" 359 d1::queuing_rw_mutex::scoped_lock* next = tricky_pointer::load(s.my_next, std::memory_order_acquire); 360 if( !next ) { 361 d1::queuing_rw_mutex::scoped_lock* expected = &s; 362 if( !s.my_mutex->q_tail.compare_exchange_strong(expected, nullptr, std::memory_order_release) ) { 363 spin_wait_while_eq( s.my_next, 0U ); 364 next = tricky_pointer::load(s.my_next, std::memory_order_relaxed); 365 } else { 366 goto unlock_self; 367 } 368 } 369 next->my_going.store(2U, std::memory_order_relaxed); 370 // Responsibility transition, the one who reads uncorrupted my_prev will do release. 371 tmp = tricky_pointer::exchange(next->my_prev, nullptr, std::memory_order_release); 372 next->my_going.store(1U, std::memory_order_release); 373 } 374 unlock_self: 375 unblock_or_wait_on_internal_lock(s, get_flag(tmp)); 376 } 377 done: 378 spin_wait_while_eq( s.my_going, 2U ); 379 380 s.initialize(); 381 } 382 383 static bool downgrade_to_reader(d1::queuing_rw_mutex::scoped_lock& s) { 384 if ( s.my_state.load(std::memory_order_relaxed) == STATE_ACTIVEREADER ) return true; // Already a reader 385 386 ITT_NOTIFY(sync_releasing, s.my_mutex); 387 s.my_state.store(STATE_READER, std::memory_order_relaxed); 388 if( ! tricky_pointer::load(s.my_next, std::memory_order_relaxed)) { 389 // the following load of q_tail must not be reordered with setting STATE_READER above 390 if( &s==s.my_mutex->q_tail.load() ) { 391 unsigned char old_state = STATE_READER; 392 s.my_state.compare_exchange_strong(old_state, STATE_ACTIVEREADER, std::memory_order_release); 393 if( old_state==STATE_READER ) 394 return true; // Downgrade completed 395 } 396 /* wait for the next to register */ 397 spin_wait_while_eq( s.my_next, 0U ); 398 } 399 d1::queuing_rw_mutex::scoped_lock *const next = tricky_pointer::load(s.my_next, std::memory_order_acquire); 400 __TBB_ASSERT( next, "still no successor at this point!" ); 401 if( next->my_state & STATE_COMBINED_WAITINGREADER ) 402 next->my_going.store(1U, std::memory_order_release); 403 else if( next->my_state==STATE_UPGRADE_WAITING ) 404 // the next waiting for upgrade means this writer was upgraded before. 405 next->my_state.store(STATE_UPGRADE_LOSER, std::memory_order_relaxed); 406 s.my_state.store(STATE_ACTIVEREADER, std::memory_order_relaxed); 407 return true; 408 } 409 410 static bool upgrade_to_writer(d1::queuing_rw_mutex::scoped_lock& s) { 411 if (s.my_state.load(std::memory_order_relaxed) == STATE_WRITER) { 412 // Already a writer 413 return true; 414 } 415 416 __TBB_ASSERT(s.my_state.load(std::memory_order_relaxed) == STATE_ACTIVEREADER, "only active reader can be updated"); 417 418 queuing_rw_mutex::scoped_lock* tmp{}; 419 queuing_rw_mutex::scoped_lock* me = &s; 420 421 ITT_NOTIFY(sync_releasing, s.my_mutex); 422 s.my_state.store(STATE_UPGRADE_REQUESTED, std::memory_order_relaxed); 423 requested: 424 __TBB_ASSERT( !(s.my_next.load(std::memory_order_relaxed) & FLAG), "use of corrupted pointer!" ); 425 acquire_internal_lock(s); 426 d1::queuing_rw_mutex::scoped_lock* expected = &s; 427 if( !s.my_mutex->q_tail.compare_exchange_strong(expected, tricky_pointer(me)|FLAG, std::memory_order_release) ) { 428 spin_wait_while_eq( s.my_next, 0U ); 429 queuing_rw_mutex::scoped_lock * next; 430 next = tricky_pointer::fetch_add(s.my_next, FLAG, std::memory_order_acquire); 431 unsigned short n_state = next->my_state; 432 /* the next reader can be blocked by our state. the best thing to do is to unblock it */ 433 if( n_state & STATE_COMBINED_WAITINGREADER ) 434 next->my_going.store(1U, std::memory_order_release); 435 // Responsibility transition, the one who reads uncorrupted my_prev will do release. 436 tmp = tricky_pointer::exchange(next->my_prev, &s, std::memory_order_release); 437 unblock_or_wait_on_internal_lock(s, get_flag(tmp)); 438 if( n_state & (STATE_COMBINED_READER | STATE_UPGRADE_REQUESTED) ) { 439 // save next|FLAG for simplicity of following comparisons 440 tmp = tricky_pointer(next)|FLAG; 441 for( atomic_backoff b; tricky_pointer::load(s.my_next, std::memory_order_relaxed)==tmp; b.pause() ) { 442 if( s.my_state & STATE_COMBINED_UPGRADING ) { 443 if( tricky_pointer::load(s.my_next, std::memory_order_acquire)==tmp ) 444 tricky_pointer::store(s.my_next, next, std::memory_order_relaxed); 445 goto waiting; 446 } 447 } 448 __TBB_ASSERT(tricky_pointer::load(s.my_next, std::memory_order_relaxed) != (tricky_pointer(next)|FLAG), nullptr); 449 goto requested; 450 } else { 451 __TBB_ASSERT( n_state & (STATE_WRITER | STATE_UPGRADE_WAITING), "unexpected state"); 452 __TBB_ASSERT( (tricky_pointer(next)|FLAG) == tricky_pointer::load(s.my_next, std::memory_order_relaxed), nullptr); 453 tricky_pointer::store(s.my_next, next, std::memory_order_relaxed); 454 } 455 } else { 456 /* We are in the tail; whoever comes next is blocked by q_tail&FLAG */ 457 release_internal_lock(s); 458 } // if( this != my_mutex->q_tail... ) 459 { 460 unsigned char old_state = STATE_UPGRADE_REQUESTED; 461 s.my_state.compare_exchange_strong(old_state, STATE_UPGRADE_WAITING, std::memory_order_acquire); 462 } 463 waiting: 464 __TBB_ASSERT( !( s.my_next.load(std::memory_order_relaxed) & FLAG ), "use of corrupted pointer!" ); 465 __TBB_ASSERT( s.my_state & STATE_COMBINED_UPGRADING, "wrong state at upgrade waiting_retry" ); 466 __TBB_ASSERT( me==&s, nullptr ); 467 ITT_NOTIFY(sync_prepare, s.my_mutex); 468 /* if no one was blocked by the "corrupted" q_tail, turn it back */ 469 expected = tricky_pointer(me)|FLAG; 470 s.my_mutex->q_tail.compare_exchange_strong(expected, &s, std::memory_order_release); 471 queuing_rw_mutex::scoped_lock * predecessor; 472 // Mark my_prev as 'in use' to prevent predecessor from releasing 473 predecessor = tricky_pointer::fetch_add(s.my_prev, FLAG, std::memory_order_acquire); 474 if( predecessor ) { 475 bool success = try_acquire_internal_lock(*predecessor); 476 { 477 // While the predecessor pointer (my_prev) is in use (FLAG is set), we can safely update the node`s state. 478 // Corrupted pointer transitions responsibility to release the predecessor`s node on us. 479 unsigned char old_state = STATE_UPGRADE_REQUESTED; 480 predecessor->my_state.compare_exchange_strong(old_state, STATE_UPGRADE_WAITING, std::memory_order_release); 481 } 482 if( !success ) { 483 // Responsibility transition, the one who reads uncorrupted my_prev will do release. 484 tmp = tricky_pointer::compare_exchange_strong(s.my_prev, tricky_pointer(predecessor)|FLAG, predecessor, std::memory_order_release); 485 if( tricky_pointer(tmp) & FLAG ) { 486 tricky_pointer::spin_wait_while_eq(s.my_prev, predecessor); 487 predecessor = tricky_pointer::load(s.my_prev, std::memory_order_relaxed); 488 } else { 489 // TODO: spin_wait condition seems never reachable 490 tricky_pointer::spin_wait_while_eq(s.my_prev, tricky_pointer(predecessor)|FLAG); 491 release_internal_lock(*predecessor); 492 } 493 } else { 494 tricky_pointer::store(s.my_prev, predecessor, std::memory_order_relaxed); 495 release_internal_lock(*predecessor); 496 tricky_pointer::spin_wait_while_eq(s.my_prev, predecessor); 497 predecessor = tricky_pointer::load(s.my_prev, std::memory_order_relaxed); 498 } 499 if( predecessor ) 500 goto waiting; 501 } else { 502 tricky_pointer::store(s.my_prev, nullptr, std::memory_order_relaxed); 503 } 504 __TBB_ASSERT( !predecessor && !s.my_prev, nullptr ); 505 506 // additional lifetime issue prevention checks 507 // wait for the successor to finish working with my fields 508 wait_for_release_of_internal_lock(s); 509 // now wait for the predecessor to finish working with my fields 510 spin_wait_while_eq( s.my_going, 2U ); 511 512 // Acquire critical section indirectly from previous owner or directly from predecessor (TODO: not clear). 513 atomic_fence(std::memory_order_acquire); // on either "my_mutex->q_tail" or "my_going" (TODO: not clear) 514 515 bool result = ( s.my_state != STATE_UPGRADE_LOSER ); 516 s.my_state.store(STATE_WRITER, std::memory_order_relaxed); 517 s.my_going.store(1U, std::memory_order_relaxed); 518 519 ITT_NOTIFY(sync_acquired, s.my_mutex); 520 return result; 521 } 522 523 static bool is_writer(const d1::queuing_rw_mutex::scoped_lock& m) { 524 return m.my_state.load(std::memory_order_relaxed) == STATE_WRITER; 525 } 526 527 static void construct(d1::queuing_rw_mutex& m) { 528 suppress_unused_warning(m); 529 ITT_SYNC_CREATE(&m, _T("tbb::queuing_rw_mutex"), _T("")); 530 } 531 }; 532 533 void __TBB_EXPORTED_FUNC acquire(d1::queuing_rw_mutex& m, d1::queuing_rw_mutex::scoped_lock& s, bool write) { 534 queuing_rw_mutex_impl::acquire(m, s, write); 535 } 536 537 bool __TBB_EXPORTED_FUNC try_acquire(d1::queuing_rw_mutex& m, d1::queuing_rw_mutex::scoped_lock& s, bool write) { 538 return queuing_rw_mutex_impl::try_acquire(m, s, write); 539 } 540 541 void __TBB_EXPORTED_FUNC release(d1::queuing_rw_mutex::scoped_lock& s) { 542 queuing_rw_mutex_impl::release(s); 543 } 544 545 bool __TBB_EXPORTED_FUNC upgrade_to_writer(d1::queuing_rw_mutex::scoped_lock& s) { 546 return queuing_rw_mutex_impl::upgrade_to_writer(s); 547 } 548 549 bool __TBB_EXPORTED_FUNC is_writer(const d1::queuing_rw_mutex::scoped_lock& s) { 550 return queuing_rw_mutex_impl::is_writer(s); 551 } 552 553 bool __TBB_EXPORTED_FUNC downgrade_to_reader(d1::queuing_rw_mutex::scoped_lock& s) { 554 return queuing_rw_mutex_impl::downgrade_to_reader(s); 555 } 556 557 void __TBB_EXPORTED_FUNC construct(d1::queuing_rw_mutex& m) { 558 queuing_rw_mutex_impl::construct(m); 559 } 560 561 } // namespace r1 562 } // namespace detail 563 } // namespace tbb 564