From f8543e2d4a4d723d0f690d2ed12c7bbcc9f03075 Mon Sep 17 00:00:00 2001
From: tqchen <tianqi.tchen@gmail.com>
Date: Sun, 13 Sep 2015 23:13:34 -0700
Subject: [PATCH 1/2] move write trigger out correctly

---
 src/engine/threaded_engine.cc | 47 ++++++++++++++++++++---------------
 1 file changed, 27 insertions(+), 20 deletions(-)

diff --git a/src/engine/threaded_engine.cc b/src/engine/threaded_engine.cc
index dd7662095097..09946f8d3ee2 100644
--- a/src/engine/threaded_engine.cc
+++ b/src/engine/threaded_engine.cc
@@ -83,7 +83,8 @@ void ThreadedVar::CompleteReadDependency(Dispatcher dispatcher) {
 
 template <typename Dispatcher>
 bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {
-  VersionedVarBlock *old_pending_write, *end_of_dispatch_chain;
+  VersionedVarBlock *old_pending_write, *end_of_read_chain;
+  bool trigger_write = false;
   {
     // this is lock scope
     std::lock_guard<std::mutex> lock{m_};
@@ -91,32 +92,31 @@ bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {
     // detach pending write
     old_pending_write = pending_write_;
     // search for chains to trigger
-    VersionedVarBlock *p = old_pending_write->next;
+    end_of_read_chain = old_pending_write->next;
     assert(num_pending_reads_ == 0);
-    while (p->next != nullptr && p->write == false) {
-      ++num_pending_reads_;
-      p = p->next;
+    while (end_of_read_chain->next != nullptr &&
+           end_of_read_chain->write == false) {
+      ++num_pending_reads;
+      end_of_read_chain = end_of_read_chain->next;
     }
-    // mark end of dispatch chain
-    end_of_dispatch_chain = p;
-
-    if (p->next == nullptr) {
+    // check the states
+    if (end_of_read_chain->next == nullptr) {
       ready_to_read_ = true;
       pending_write_ = nullptr;
-      assert(p->trigger == nullptr);
-      assert(p->write ==false);
     } else {
-      assert(p->write == true);
-      pending_write_ = p;
-      if (num_pending_reads_ == 0) {
-        if (--pending_write_->trigger->wait == 0) {
-          dispatcher(pending_write_->trigger);
-        }
+      assert(end_of_read_chain->write == true);
+      pending_write_ = end_of_read_chain;
+      if (num_pending_reads == 0) {
+        trigger_write = true;
       }
     }
   }
-  // this is outside of lock scope
-  // the linked list is detached from variable
+  // This is outside of lock scope
+  // Be very carful, pending_write_ and num_pending_reads_
+  // can change now, do not reply ont the two variables.
+  // The linked list \in [old_pending_write, end_of_read_chain)
+  // is already detached from this Var.
+  // So it is safe to modify these
   VersionedVarBlock *cur_head = old_pending_write->next;
   VersionedVarBlock::Delete(old_pending_write);
   if (to_delete_) {
@@ -125,7 +125,7 @@ bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {
     return true;
   }
   // dispatch all the events
-  while (cur_head != end_of_dispatch_chain) {
+  while (cur_head != end_of_read_chain) {
     if (--cur_head->trigger->wait == 0) {
       dispatcher(cur_head->trigger);
     }
@@ -134,6 +134,13 @@ bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {
     assert(cur_head != nullptr);
     VersionedVarBlock::Delete(prev);
   }
+  // Be careful, do not use pending_write_  or num_pending_reads_ here.
+  // As they can change, use end_of_read_chain
+  if (trigger_write) {
+    if (--end_of_read_chain->trigger->wait == 0) {
+      dispatcher(end_of_read_chain->trigger);
+    }
+  }
   return false;
 }
 

From 2e4594fda1321a78b3a51ed348f1043ace743ba2 Mon Sep 17 00:00:00 2001
From: tqchen <tianqi.tchen@gmail.com>
Date: Sun, 13 Sep 2015 23:14:05 -0700
Subject: [PATCH 2/2] fix

---
 src/engine/threaded_engine.cc | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/engine/threaded_engine.cc b/src/engine/threaded_engine.cc
index 09946f8d3ee2..8e59c59ab30e 100644
--- a/src/engine/threaded_engine.cc
+++ b/src/engine/threaded_engine.cc
@@ -96,7 +96,7 @@ bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {
     assert(num_pending_reads_ == 0);
     while (end_of_read_chain->next != nullptr &&
            end_of_read_chain->write == false) {
-      ++num_pending_reads;
+      ++num_pending_reads_;
       end_of_read_chain = end_of_read_chain->next;
     }
     // check the states
@@ -106,7 +106,7 @@ bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {
     } else {
       assert(end_of_read_chain->write == true);
       pending_write_ = end_of_read_chain;
-      if (num_pending_reads == 0) {
+      if (num_pending_reads_ == 0) {
         trigger_write = true;
       }
     }