text: /**  
   * Copyright (c) 2009 Carnegie Mellon University. 
   *     All rights reserved.
   *
   *  Licensed under the Apache License, Version 2.0 (the "License");
   *  you may not use this file except in compliance with the License.
   *  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   *  Unless required by applicable law or agreed to in writing,
   *  software distributed under the License is distributed on an "AS
   *  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
   *  express or implied.  See the License for the specific language
   *  governing permissions and limitations under the License.
   *
   * For more about this software visit:
   *
   *      http://www.graphlab.ml.cmu.edu
   *
   */
  
  
  #ifndef QUEUED_RWLOCK_HPP
  #define QUEUED_RWLOCK_HPP
  
  
  namespace graphlab {
    
    
  #define QUEUED_RW_LOCK_REQUEST_READ 0
  #define QUEUED_RW_LOCK_REQUEST_WRITE 1
  #define QUEUED_RW_LOCK_REQUEST_NONE 2
  
  /**
   * Fair rw-lock with local-only spinning implemented and
   * modified from 
   * Scalable Reader-Writer Synchronization for Shared-Memory Multiprocessors.
   * John M. Mellor-Crummey and Michael L. Scott
   */
  class queued_rw_lock{
   public:
    
    union state_union {
      volatile uint32_t stateu;
      struct {
        volatile uint16_t successor_class;
        volatile bool blocked;
      } state;
    };
    
    struct request{
      void* id;  
      volatile request* volatile next;
      volatile state_union s;
      volatile char lockclass;
    };
   private:
    request* volatile tail;
    atomic<size_t> reader_count;
    request* volatile next_writer;
   public:
    queued_rw_lock(): tail(NULL), reader_count(0), next_writer(NULL) { }
    
    inline void writelock(request *I) {
      I->lockclass = QUEUED_RW_LOCK_REQUEST_WRITE;
      I->next = NULL;
      I->s.stateu = 0;
      I->s.state.blocked = true;
      I->s.state.successor_class = QUEUED_RW_LOCK_REQUEST_NONE;
      __sync_synchronize();
      request* predecessor = __sync_lock_test_and_set(&tail, I);
  
      if (predecessor == NULL) {
        next_writer = I;
        __sync_synchronize();
        if (reader_count.value == 0) {
          if (__sync_lock_test_and_set(&next_writer, (request*)NULL) == I) {
            I->s.state.blocked = false;
          }
        }
      }
      else {
        predecessor->s.state.successor_class = QUEUED_RW_LOCK_REQUEST_WRITE;
      __sync_synchronize();      
        predecessor->next = I;
      }
      // while I->blocked. continue
      volatile state_union& is = I->s;
      while (is.state.blocked) sched_yield();
      assert(reader_count.value == 0);
    }
  
    inline void wrunlock(request *I) {
      __sync_synchronize(); 
      if (I->next != NULL || !__sync_bool_compare_and_swap(&tail, I, (request*)NULL)) {
        // wait
        while(I->next == NULL) sched_yield();
       __sync_synchronize(); 
     
        if (I->next->lockclass == QUEUED_RW_LOCK_REQUEST_READ) {
          reader_count.inc();
        }
        I->next->s.state.blocked = false;
      }
    }
  
    inline void readlock(request *I)  {
      I->lockclass =QUEUED_RW_LOCK_REQUEST_READ;
      I->next = NULL;
      I->s.stateu = 0;
      I->s.state.successor_class = QUEUED_RW_LOCK_REQUEST_NONE;
      I->s.state.blocked = true;
      __sync_synchronize(); 
      request* predecessor = __sync_lock_test_and_set(&tail, I);
      if (predecessor == NULL) {
        reader_count.inc();
        I->s.state.blocked = false;
      }
      else {
        
        state_union tempold, tempnew;
        tempold.state.blocked = true;
        tempold.state.successor_class = QUEUED_RW_LOCK_REQUEST_NONE;
        tempnew.state.blocked = true;
        tempnew.state.successor_class = QUEUED_RW_LOCK_REQUEST_READ;
        __sync_synchronize();
        if (predecessor->lockclass == QUEUED_RW_LOCK_REQUEST_WRITE ||
            atomic_compare_and_swap(predecessor->s.stateu,
                                    tempold.stateu,
                                    tempnew.stateu)) {
          
          predecessor->next = I;
          // wait
          __sync_synchronize(); 
          volatile state_union& is = I->s;
          while(is.state.blocked) sched_yield();
        }
        else {
          reader_count.inc();
          predecessor->next = I;
          __sync_synchronize();
          I->s.state.blocked = false;
        }
      }
      __sync_synchronize();
      if (I->s.state.successor_class == QUEUED_RW_LOCK_REQUEST_READ) {
        
        // wait
        while(I->next == NULL) sched_yield();
        reader_count.inc();
        I->next->s.state.blocked = false;
      }
    }
  
    inline void rdunlock(request *I)  {
      __sync_synchronize();
      if (I->next != NULL || !__sync_bool_compare_and_swap(&tail, I, (request*)NULL)) {
        while(I->next == NULL) sched_yield();
        if (I->s.state.successor_class == QUEUED_RW_LOCK_REQUEST_WRITE) {
          next_writer = (request*)(I->next);
          __sync_synchronize();
        }
      }
      if (reader_count.dec() == 0) {
        __sync_synchronize(); 
        request * w = __sync_lock_test_and_set(&next_writer, (request*)NULL);
        if (w != NULL) {
          w->s.state.blocked = false;
          __sync_synchronize(); 
        }
      }
    }
  };
  
  }
  #endif
  
  