From 91f2ff2608e685753973febd6b52258d75f2abec Mon Sep 17 00:00:00 2001
From: sageweil <sageweil@29311d96-e01e-0410-9327-a35deaab8ce9>
Date: Mon, 8 Oct 2007 23:48:22 +0000
Subject: [PATCH] prioritize monitor messages; move non-monitor messages to
 dispatch one by one

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@1895 29311d96-e01e-0410-9327-a35deaab8ce9
---
 branches/sage/mds/msg/SimpleMessenger.cc | 18 +++++++++++++++---
 branches/sage/mds/msg/SimpleMessenger.h  | 21 ++++++++++-----------
 2 files changed, 25 insertions(+), 14 deletions(-)

diff --git a/branches/sage/mds/msg/SimpleMessenger.cc b/branches/sage/mds/msg/SimpleMessenger.cc
index b885fd6f83c1f..116204e245fc8 100644
--- a/branches/sage/mds/msg/SimpleMessenger.cc
+++ b/branches/sage/mds/msg/SimpleMessenger.cc
@@ -1226,10 +1226,22 @@ void Rank::EntityMessenger::dispatch_entry()
 {
   lock.Lock();
   while (!stop) {
-    if (!dispatch_queue.empty()) {
+    if (!dispatch_queue.empty() || !prio_dispatch_queue.empty()) {
       list<Message*> ls;
-      ls.swap(dispatch_queue);
-      qlen = 0;
+      if (!prio_dispatch_queue.empty()) {
+	ls.swap(prio_dispatch_queue);
+	pqlen = 0;
+      } else {
+	if (0) {
+	  ls.swap(dispatch_queue);
+	  qlen = 0;
+	} else {
+	  // limit how much low-prio stuff we grab, to avoid starving high-prio messages!
+	  ls.push_back(dispatch_queue.front());
+	  dispatch_queue.pop_front();
+	  qlen--;
+	}
+      }
 
       lock.Unlock();
       {
diff --git a/branches/sage/mds/msg/SimpleMessenger.h b/branches/sage/mds/msg/SimpleMessenger.h
index 87aa9793144b3..3457952e73de1 100644
--- a/branches/sage/mds/msg/SimpleMessenger.h
+++ b/branches/sage/mds/msg/SimpleMessenger.h
@@ -164,8 +164,9 @@ private:
     Mutex lock;
     Cond cond;
     list<Message*> dispatch_queue;
+    list<Message*> prio_dispatch_queue;
     bool stop;
-    int qlen;
+    int qlen, pqlen;
 
     class DispatchThread : public Thread {
       EntityMessenger *m;
@@ -184,15 +185,13 @@ private:
       m->set_recv_stamp(g_clock.now());
       
       lock.Lock();
-      dispatch_queue.push_back(m);
-      qlen++;
-      cond.Signal();
-      lock.Unlock();
-    }
-    void queue_messages(list<Message*> ls) {
-      lock.Lock();
-      qlen += ls.size();
-      dispatch_queue.splice(dispatch_queue.end(), ls);
+      if (m->get_source().is_mon()) {
+	prio_dispatch_queue.push_back(m);
+	pqlen++;
+      } else {
+	qlen++;
+	dispatch_queue.push_back(m);
+      }
       cond.Signal();
       lock.Unlock();
     }
@@ -210,7 +209,7 @@ private:
     
     const entity_addr_t &get_myaddr();
 
-    int get_dispatch_queue_len() { return qlen; }
+    int get_dispatch_queue_len() { return qlen + pqlen; }
 
     void reset_myname(entity_name_t m);
 
-- 
2.39.5