commit ff8d9dc9f6ff3cf3b3ccd74cbe603cab8517dae0
author Alexander Afanasyev <alexander.afanasyev@ucla.edu> Sat Jan 26 00:45:08 2013 -0800
committer Alexander Afanasyev <alexander.afanasyev@ucla.edu> Sat Jan 26 00:45:08 2013 -0800
tree 320f83116fe1ff9c8787298ac37d32617df4870d
parent 0c545ed1cafdf953a50d7cf4e3f21ff28842d6e0

Adding executor to fetcher: there was a deadlock otherwise

src/fetcher.cc

diff --git a/src/fetcher.cc b/src/fetcher.cc
index e9dec91..3fe7c63 100644
--- a/src/fetcher.cc
+++ b/src/fetcher.cc
@@ -62,11 +62,15 @@
 
   , m_pipeline (6) // initial "congestion window"
   , m_activePipeline (0)
+
+  , m_executor (1)
 {
+  m_executor.start ();
 }
 
 Fetcher::~Fetcher ()
 {
+  m_executor.shutdown ();
 }
 
 void
@@ -77,7 +81,7 @@
   // cout << "Restart: " << m_minSendSeqNo << endl;
   m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
 
-  FillPipeline ();
+  m_executor.execute (bind (&Fetcher::FillPipeline, this));
 }
 
 void
@@ -101,6 +105,7 @@
                             Closure (bind(&Fetcher::OnData, this, m_minSendSeqNo+1, _1, _2),
                                      bind(&Fetcher::OnTimeout, this, m_minSendSeqNo+1, _1)),
                             Selectors().interestLifetime (1)); // Alex: this lifetime should be changed to RTO
+      _LOG_DEBUG (" >>> i ok");
 
       m_activePipeline ++;
     }
@@ -170,13 +175,15 @@
     }
   else
     {
-      FillPipeline ();
+      m_executor.execute (bind (&Fetcher::FillPipeline, this));
     }
 }
 
 Closure::TimeoutCallbackReturnValue
 Fetcher::OnTimeout (uint64_t seqno, const Ccnx::Name &name)
 {
+  _LOG_DEBUG (" <<< :( timeout " << name.getPartialName (0, name.size () - 1) << ", seq = " << seqno);
+  
   // cout << "Fetcher::OnTimeout: " << name << endl;
   // cout << "Last: " << m_lastPositiveActivity << ", config: " << m_maximumNoActivityPeriod
   //      << ", now: " << date_time::second_clock<boost::posix_time::ptime>::universal_time()
@@ -188,6 +195,7 @@
       m_activePipeline --;
       if (m_activePipeline == 0)
         {
+          _LOG_DEBUG ("Telling that fetch failed");
           m_active = false;
           m_onFetchFailed (*this);
           // this is not valid anymore, but we still should be able finish work
@@ -195,5 +203,8 @@
       return Closure::RESULT_OK;
     }
   else
-    return Closure::RESULT_REEXPRESS;
+    {
+      _LOG_DEBUG ("Asking to reexpress");
+      return Closure::RESULT_REEXPRESS;
+    }
 }