Chris M Thomasson

#include "stdafx.h"
#include "../../relacy/relacy_std.hpp"
#include <cstdio>
#include <list>


namespace raii {
  template<typename T>
  class unguard {
    T& m_guard;

  public:
    unguard(T& guard)
      : m_guard(guard) {
      m_guard.unlock();
    }

    ~unguard() {
      m_guard.lock();
    }
  };

  template<typename T, typename Y>
  class guard {
    T& m_mtx;
    Y& m_tid;

    friend class unguard<guard>;

  public:
    guard(T& mtx, Y& tid)
      : m_mtx(mtx), m_tid(tid) {
      lock();
    }

    ~guard() {
      unlock();
    }

    Y& get_tid() const {
      return m_tid;
    }

  private:
    void lock() {
      m_mtx.lock(m_tid);
    }

    void unlock() {
      m_mtx.unlock(m_tid);
    }
  };
}


namespace pthread {
  namespace sys {
    static void spin_wait() {
      rl::backoff wait;
      wait.yield($);
    }


    class spinlock {
      std::atomic<class thread*> m_state;

    public:
      void before() {
        m_state($) = NULL;
      }

      void after() {
        RL_ASSERT(! m_state($).load(std::memory_order_relaxed));
      }

    public:
      typedef raii::guard<spinlock, class thread> guard;
      typedef raii::unguard<guard> unguard;

    public:
      void lock(thread& tid) {
        class thread* cmp = NULL;
        while (! m_state($).compare_swap(cmp, &tid, std::memory_order_acquire)) {
          spin_wait();
          cmp = NULL;
        }
      }

      void unlock(thread& tid) {
        class thread* cmp = m_state($).swap(NULL, std::memory_order_release);
        if (cmp != &tid) {
          RL_ASSERT(cmp == &tid);
        }
      }
    };


    class thread {
      unsigned const m_idx;
      std::atomic<bool> m_signal;

    public:
      thread(unsigned const idx)
        : m_idx(idx) {
        m_signal($) = false;
      }

      void wait() {
        // std::cout << "thread: " << this << " waiting\n";
        while (! m_signal($).swap(false, std::memory_order_acquire)) {
          spin_wait();
        }
        // std::cout << "thread: " << this << " signal consume\n";
      }

      void signal() {
        m_signal($).store(true, std::memory_order_release);
        // std::cout << "thread: " << this << " signalled\n";
      }
    };


    class waitset {
      spinlock m_mtx;
      rl::var<unsigned> m_count;
      std::list<thread*> m_threads;

    public:
      void before() {
        m_count($) = 0;
        m_mtx.before();
      }

      void after() {
        RL_ASSERT(! m_threads.size());
        m_mtx.after();
      }

    public:
      void push(thread& tid) {
        spinlock::guard lock(m_mtx, tid);
        m_threads.push_back(&tid);
      }

      void pop(thread& tid) {
        spinlock::guard lock(m_mtx, tid);
        m_threads.remove(&tid);
      }

      void wait(thread& tid) {
        spinlock::guard lock(m_mtx, tid);
        if (! m_count($)) {
          m_threads.push_back(&tid);
          {
            spinlock::unguard unlock(lock);
            tid.wait();
          }
          m_threads.remove(&tid);
          return;
        }
        --m_count($);
      }

      void signal(thread& tid, unsigned count = 0) {
        spinlock::guard lock(m_mtx, tid);
        m_count($) += count;
        if (! m_threads.empty()) {
          m_threads.front()->signal();
        }
      }

      void broadcast(thread& tid, unsigned count = 0) {
        spinlock::guard lock(m_mtx, tid);
        m_count($) += count;
        std::list<thread*>::iterator i;
        for (i = m_threads.begin(); i != m_threads.end(); ++i) {
          (*i)->signal();
        }
      }
    };
  } // namespace sys


  class event {
    std::atomic<bool> m_state;
    sys::waitset m_wset;

  public:
    void before() {
      m_state($) = false;
      m_wset.before();
    }

    void after() {
      m_wset.after();
    }

  public:
    void set(sys::thread& tid) {
      if (! m_state($).swap(true, std::memory_order_seq_cst)) {
        m_wset.signal(tid, 1);
      }
    }

    void wait(sys::thread& tid) {
      bool cmp = true;
      while (! m_state($).compare_swap(cmp, false, std::memory_order_acquire)) {
        m_wset.wait(tid);
        cmp = true;
      }
    }
  };


  class mutex {
    enum constant {
      UNLOCKED = 0,
      LOCKED = 1,
      CONTENTION = 2
    };

    std::atomic<int> m_state;
    event m_event;

  public:
    void before() {
      m_state($) = UNLOCKED;
      m_event.before();
    }

    void after() {
      m_event.after();
      RL_ASSERT(m_state($).load(std::memory_order_relaxed) == UNLOCKED);
    }

  public:
    typedef raii::guard<mutex, sys::thread> guard;
    typedef raii::unguard<guard> unguard;

  public:
    void lock(sys::thread& tid) {
      if (m_state($).swap(LOCKED, std::memory_order_acquire)) {
        while (m_state($).swap(CONTENTION, std::memory_order_acquire)) {
          m_event.wait(tid);
        }
      }
    }

    void unlock(sys::thread& tid) {
      if (m_state($).swap(UNLOCKED, std::memory_order_release) == CONTENTION) {
        m_event.set(tid);
      }
    }
  };


  class condvar {
    sys::waitset m_wset;
    std::atomic<unsigned> m_state;

  public:
    void before() {
      m_state($) = 0;
      m_wset.before();
    }

    void after() {
      RL_ASSERT(! m_state($).load(std::memory_order_relaxed));
      m_wset.after();
    }

  public:
    void wait(mutex::guard& lock) {
      sys::thread& tid = lock.get_tid();
      m_state($).fetch_add(1, std::memory_order_relaxed);
      m_wset.push(tid);
      mutex::unguard unlock(lock);
      tid.wait();
      m_wset.pop(tid);
    }

    void signal(sys::thread& tid) {
      unsigned cmp = m_state($).load(std::memory_order_relaxed);
      do {
        if (! cmp) { return; }
      } while(! m_state($).compare_swap(cmp, cmp - 1, std::memory_order_relaxed));
      m_wset.signal(tid);
    }

    void broadcast(sys::thread& tid) {
      unsigned cmp = m_state($).load(std::memory_order_relaxed);
      if (cmp) {
        if (m_state($).swap(0, std::memory_order_relaxed)) {
          m_wset.broadcast(tid);
        }
      }
    }
  };
} // namespace pthread




#define CONSUMER_COUNT 2

struct condvar_test : rl::test_suite<condvar_test, CONSUMER_COUNT + 1> {

  pthread::mutex m_mtx;
  pthread::condvar m_cond;
  rl::var<bool> m_signal;
  rl::var<bool> m_finish;


  void before() {
    m_mtx.before();
    m_cond.before();
    m_signal($) = false;
    m_finish($) = false;
  }


  void after() {
    RL_ASSERT(! m_signal($));
    RL_ASSERT(! m_finish($));
    m_cond.after();
    m_mtx.after();
  }


  void producer(pthread::sys::thread& tid) {
    unsigned count = 0;
    // std::cout << "producer " << &tid << "\n";
    do {
      {
        pthread::mutex::guard lock(m_mtx, tid);
        m_signal($) = true;
      }
      m_cond.signal(tid);
      pthread::mutex::guard lock(m_mtx, tid);
      while (! m_finish($)) {
        m_cond.wait(lock);
      }
      m_finish($) = false;
      RL_ASSERT(! m_signal($));
      ++count;
    } while (count < CONSUMER_COUNT);
  }


  void consumers(pthread::sys::thread& tid) {
    // std::cout << "consumer " << &tid << "\n";
    {
      pthread::mutex::guard lock(m_mtx, tid);
      while (! m_signal($)) {
        m_cond.wait(lock);
      }

      m_signal($) = false;
      m_finish($) = true;
    }

    m_cond.broadcast(tid);
  }


  void thread(unsigned tidx) {
    pthread::sys::thread tid(tidx);
    if (! tidx) { // producer
      producer(tid);
    } else { // consumers
      consumers(tid);
    }
  }
};


int main() {
  {
    rl::test_params params;
    params.search_type = rl::random_scheduler_type;
    params.iteration_count = 1000;
    rl::simulate<condvar_test>(params);
    std::cout << rl::test_result_str(params.test_result) << std::endl;
  }
  std::puts("\n\n\nFINISHED!\n___________________________________\n\
press <ENTER> to exit...");
  std::getchar();
  return 0;
}