Chris M Thomasson

/*
Unbounded Dynamic Work-Stealing Deque
Copyright (C) 2009  Christopher Michael Thomasson

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.


You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
_______________________________________________________________________*/






#include <relacy/relacy_std.hpp>
#include <cstdio>




#if ! defined (NDEBUG)
#  define DBG_PRINTF(mp_exp) std::printf mp_exp
#else
#  define DBG_PRINTF(mp_exp) ((void)0)
#endif




#define WSDEQUE_SPLIT 4


class wsdeque {
public:
  struct node {
    rl::atomic<node*> m_next;
    rl::atomic<node*> m_prev;
  };


private:
  rl::atomic<node*> m_head;
  rl::atomic<node*> m_tail;
  rl::atomic<unsigned> m_lcount;
  rl::atomic<unsigned> m_rcount;
  rl::mutex m_mutex;


public:
  wsdeque() {
    m_head($).store(NULL);
    m_tail($).store(NULL);
    m_lcount($).store(0);
    m_rcount($).store(0);
  }


  ~wsdeque() {
    RL_ASSERT(! m_head($).load());
    RL_ASSERT(! m_tail($).load());
    RL_ASSERT(! (m_lcount($).load() - m_rcount($).load()));
  }


  void push(node* n) {
    unsigned lcount = m_lcount($).load();
    node* head = m_head($).load();
    n->m_prev($).store(NULL);
    n->m_next($).store(head);
    if (! head) {
      m_tail($).store(n);
    } else {
      head->m_prev($).store(n);
    }
    m_head($).store(n);
    m_lcount($).store(lcount + 1);
  }


  node* pop() {
   retry:
    node* head = m_head($).load();
    if (head) {
      bool unlock = false;
      unsigned lcount = m_lcount($).load();
      unsigned rcount = m_rcount($).load();
      if (lcount - rcount <= WSDEQUE_SPLIT) {
        m_mutex.lock($);
        lcount = m_lcount($).load();
        rcount = m_rcount($).load();
        if (lcount - rcount > WSDEQUE_SPLIT ||
          m_head($).load() != head) {
          m_mutex.unlock($);
          goto retry;
        }
        unlock = true;
      }
      node* next = head->m_next($).load();
      m_head($).store(next);
      if (! next) {
        m_tail($).store(NULL);
      } else {
        next->m_prev($).store(NULL);
      }
      m_lcount($).store(lcount - 1);
      if (unlock) {
        m_mutex.unlock($);
      }
    }
    return head;
  }


  node* steal() {
   retry:
    node* tail = m_tail($).load();
    if (tail) {
      unsigned lcount = m_lcount($).load();
      unsigned rcount = m_rcount($).load();
      if (lcount - rcount < WSDEQUE_SPLIT) {
        return NULL;
      }
      if (! m_mutex.try_lock($)) {
        return NULL;
      }
      lcount = m_lcount($).load();
      rcount = m_rcount($).load();
      if (lcount - rcount < WSDEQUE_SPLIT ||
        m_tail($).load() != tail) {
        m_mutex.unlock($);
        goto retry;
      }
      node* prev = tail->m_prev($).load();
      prev->m_next($) = NULL;
      m_tail($).store(prev);
      m_rcount($).store(rcount + 1);
      m_mutex.unlock($);
    }
    return tail;
  }
};




#define THREADS 7
#define ITERS 10


struct wsdeque_test : rl::test_suite<wsdeque_test, THREADS> {
  wsdeque m_wsdeque;

  void thread(unsigned tidx) {
    unsigned i;
    wsdeque::node* n;
    if (tidx < 1) {
      rl::backoff b;
      for (i = 0; i < ITERS; ++i) {
        n = RL_NEW wsdeque::node;
        m_wsdeque.push(n);
        DBG_PRINTF(("Thread(%u) has locally pushed node(%p)\n",
          tidx, (void*)n));
      }
      for (i = 0; i < ITERS; ++i) {
        if ((n = m_wsdeque.pop())) {
          DBG_PRINTF(("Thread(%u) has locally popped node(%p)\n",
            tidx, (void*)n));
          RL_DELETE(n);
          b.yield($);
        }
      }
    } else {
      for (i = 0; i < ITERS; ++i) {
        if ((n = m_wsdeque.steal())) {
          DBG_PRINTF(("Thread(%u) has remotely stolen node(%p)!\n",
            tidx, (void*)n));
          RL_DELETE(n);
        }
      }
    }
  }
};




int main() {
  rl::test_params params;
  params.iteration_count = 99999999;
  //params.search_type = rl::fair_full_search_scheduler_type;
  //params.search_type = rl::fair_context_bound_scheduler_type;
  rl::simulate<wsdeque_test>(params);
  std::puts("\n\n\n____________________________________\n"
            "DONE!!!!\npress <ENTER> to exit...");
  std::fflush(stdin);
  std::fflush(stdout);
  std::getchar();
  return 0;
}