msgr: introduce orig_src, forward_message()

diff --git a/src/kernel/messenger.c b/src/kernel/messenger.c
index 9f3c61fc585799521e97201ed48fa17b06ea7188..b30c25761a2b4174181bf6db83050162454c5785 100644 (file)
--- a/src/kernel/messenger.c
+++ b/src/kernel/messenger.c
@@ -1844,6 +1844,7 @@ int ceph_msg_send(struct ceph_messenger *msgr, struct ceph_msg *msg,
 
        /* set source */
        msg->hdr.src = msgr->inst;
+       msg->hdr.orig_src = msgr->inst;
 
        /* do we have the connection? */
        spin_lock(&msgr->con_lock);

diff --git a/src/msg/FakeMessenger.cc b/src/msg/FakeMessenger.cc
index bae18d3650cf4e6a28c7014b29a29d8d5c37caae..93c46889bd2267ed7e9a9edbe5c81772dc2aff71 100644 (file)
--- a/src/msg/FakeMessenger.cc
+++ b/src/msg/FakeMessenger.cc
@@ -364,12 +364,22 @@ void FakeMessenger::reset_myname(entity_name_t m)
 
 int FakeMessenger::send_message(Message *m, entity_inst_t inst)
 {
-  entity_name_t dest = inst.name;
-  
-  m->set_source(get_myname());
-  m->set_source_addr(get_myaddr());
+  m->set_source_inst(_myinst);
+  m->set_orig_source_inst(_myinst);
+  m->set_dest_inst(inst);
+  return submit_message(m, inst);
+}
 
+int FakeMessenger::forward_message(Message *m, entity_inst_t inst)
+{
+  m->set_source_inst(_myinst);
   m->set_dest_inst(inst);
+  return submit_message(m, inst);
+}
+
+int FakeMessenger::submit_message(Message *m, entity_inst_t inst)
+{
+  entity_name_t dest = inst.name;
 
   lock.Lock();
 

diff --git a/src/msg/FakeMessenger.h b/src/msg/FakeMessenger.h
index 698eee5184b6962540b063f612832ffcb3cca3d5..cbe73d3a852b6dfa2626cde091f8f84c6950f04f 100644 (file)
--- a/src/msg/FakeMessenger.h
+++ b/src/msg/FakeMessenger.h
@@ -44,6 +44,8 @@ class FakeMessenger : public Messenger {
 
   // msg interface
   int send_message(Message *m, entity_inst_t dest);
+  int forward_message(Message *m, entity_inst_t dest);
+  int submit_message(Message *m, entity_inst_t dest);
   
   int get_dispatch_queue_len() { return qlen; }
 

diff --git a/src/msg/Message.h b/src/msg/Message.h
index 383a52f9d2e918c1e71bba6af85afbca0e923429..7d84e54c8cad0ca18853283b90a4f5f4e9b4328f 100644 (file)
--- a/src/msg/Message.h
+++ b/src/msg/Message.h
@@ -160,19 +160,18 @@ public:
 
   // source/dest
   entity_inst_t get_dest_inst() { return entity_inst_t(env.dst); }
+  entity_name_t get_dest() { return entity_name_t(env.dst.name); }
   void set_dest_inst(entity_inst_t& inst) { env.dst = inst; }
 
   entity_inst_t get_source_inst() { return entity_inst_t(env.src); }
-  void set_source_inst(entity_inst_t& inst) { env.src = inst; }
-
-  entity_name_t get_dest() { return entity_name_t(env.dst.name); }
-  void set_dest(entity_name_t a) { env.dst.name = a; }
-  
   entity_name_t get_source() { return entity_name_t(env.src.name); }
-  void set_source(entity_name_t a) { env.src.name = a; }
-
   entity_addr_t get_source_addr() { return entity_addr_t(env.src.addr); }
-  void set_source_addr(const entity_addr_t &i) { env.src.addr = i; }
+  void set_source_inst(entity_inst_t& inst) { env.src = inst; }
+
+  entity_inst_t get_orig_source_inst() { return entity_inst_t(env.orig_src); }
+  entity_name_t get_orig_source() { return entity_name_t(env.orig_src.name); }
+  entity_addr_t get_orig_source_addr() { return entity_addr_t(env.orig_src.addr); }
+  void set_orig_source_inst(entity_inst_t &i) { env.orig_src = i; }
 
   // virtual bits
   virtual void decode_payload() = 0;

diff --git a/src/msg/Messenger.h b/src/msg/Messenger.h
index 79c63bf48a3a91deaecca57e84cc3e9e0f31ec58..e508fcb39913d0db0a181f983bc26292d5bb94df 100644 (file)
--- a/src/msg/Messenger.h
+++ b/src/msg/Messenger.h
@@ -80,6 +80,7 @@ protected:
   // send message
   virtual void prepare_dest(const entity_inst_t& inst) {}
   virtual int send_message(Message *m, entity_inst_t dest) = 0;
+  virtual int forward_message(Message *m, entity_inst_t dest) = 0;
   virtual int lazy_send_message(Message *m, entity_inst_t dest) {
     return send_message(m, dest);
   }

diff --git a/src/msg/SimpleMessenger.cc b/src/msg/SimpleMessenger.cc
index a2e218ed986e6976dd9e70895bc08eb3310f03c4..a0327a308667ae6e630f1775a85c3015b0156b6b 100644 (file)
--- a/src/msg/SimpleMessenger.cc
+++ b/src/msg/SimpleMessenger.cc
@@ -660,6 +660,7 @@ int Rank::EntityMessenger::send_message(Message *m, entity_inst_t dest)
 {
   // set envelope
   m->set_source_inst(_myinst);
+  m->set_orig_source_inst(_myinst);
   m->set_dest_inst(dest);
  
   dout(1) << m->get_source()
@@ -673,10 +674,30 @@ int Rank::EntityMessenger::send_message(Message *m, entity_inst_t dest)
   return 0;
 }
 
+int Rank::EntityMessenger::forward_message(Message *m, entity_inst_t dest)
+{
+  // set envelope
+  m->set_source_inst(_myinst);
+  m->set_dest_inst(dest);
+ 
+  dout(1) << m->get_source()
+          << " **> " << dest.name << " " << dest.addr
+          << " -- " << *m
+         << " -- " << m
+          << dendl;
+
+  rank.submit_message(m, dest.addr);
+
+  return 0;
+}
+
+
+
 int Rank::EntityMessenger::lazy_send_message(Message *m, entity_inst_t dest)
 {
   // set envelope
   m->set_source_inst(_myinst);
+  m->set_orig_source_inst(_myinst);
   m->set_dest_inst(dest);
  
   dout(1) << "lazy " << m->get_source()
@@ -1625,7 +1646,7 @@ Message *Rank::Pipe::read_message()
   
   if (env.src.addr.ipaddr.sin_addr.s_addr == htonl(INADDR_ANY)) {
     dout(10) << "reader munging src addr " << env.src << " to be " << peer_addr << dendl;
-    env.src.addr.ipaddr = peer_addr.ipaddr;
+    env.orig_src.addr.ipaddr = env.src.addr.ipaddr = peer_addr.ipaddr;
   }
 
   // read front

diff --git a/src/msg/SimpleMessenger.h b/src/msg/SimpleMessenger.h
index a579967f57432db3178b82e72fc27cda18d36a67..f85f676d48593e1cc3d8e1f8b1267753f90259ad 100644 (file)
--- a/src/msg/SimpleMessenger.h
+++ b/src/msg/SimpleMessenger.h
@@ -314,6 +314,7 @@ private:
     void suicide();
     void prepare_dest(const entity_inst_t& inst);
     int send_message(Message *m, entity_inst_t dest);
+    int forward_message(Message *m, entity_inst_t dest);
     int lazy_send_message(Message *m, entity_inst_t dest);
     
     void mark_down(entity_addr_t a);