From: Sage Weil Date: Wed, 2 Sep 2009 22:55:43 +0000 (-0700) Subject: msgr: detect rank addr during connect handshake; clean up addr apis X-Git-Tag: v0.14~55 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=218e4ed11370bbee1cf6104bd1eab37e29768506;p=ceph.git msgr: detect rank addr during connect handshake; clean up addr apis --- diff --git a/src/client/SyntheticClient.cc b/src/client/SyntheticClient.cc index 6af0730fba0..01d3af68899 100644 --- a/src/client/SyntheticClient.cc +++ b/src/client/SyntheticClient.cc @@ -292,8 +292,8 @@ string SyntheticClient::get_sarg(int seq) sargs.pop_front(); } if (a.length() == 0 || a == "~") { - char s[20]; - sprintf(s,"/syn.%d.%d", client->whoami.v, seq); + char s[30]; + sprintf(s,"/syn.%lld.%d", (long long)client->whoami.v, seq); a = s; } return a; diff --git a/src/mds/MDS.cc b/src/mds/MDS.cc index 402445abc54..2dc9e1c2c78 100644 --- a/src/mds/MDS.cc +++ b/src/mds/MDS.cc @@ -387,7 +387,7 @@ int MDS::init() want_state = MDSMap::STATE_BOOT; beacon_start(); whoami = -1; - messenger->reset_myname(entity_name_t::MDS(whoami)); + messenger->set_myname(entity_name_t::MDS(whoami)); objecter->init(); @@ -641,7 +641,7 @@ void MDS::handle_mds_map(MMDSMap *m) if (oldwhoami != whoami) { // update messenger. dout(1) << "handle_mds_map i am now mds" << whoami << "." << incarnation << dendl; - messenger->reset_myname(entity_name_t::MDS(whoami)); + messenger->set_myname(entity_name_t::MDS(whoami)); // do i need an osdmap? if (oldwhoami < 0) { diff --git a/src/messages/MClientMountAck.h b/src/messages/MClientMountAck.h index e031200bc33..a526f0e3317 100644 --- a/src/messages/MClientMountAck.h +++ b/src/messages/MClientMountAck.h @@ -19,7 +19,6 @@ struct MClientMountAck : public Message { __s64 client; - entity_addr_t addr; __s32 result; cstring result_msg; bufferlist monmap_bl; @@ -40,14 +39,12 @@ struct MClientMountAck : public Message { void decode_payload() { bufferlist::iterator p = payload.begin(); ::decode(client, p); - ::decode(addr, p); ::decode(result, p); ::decode(result_msg, p); ::decode(monmap_bl, p); } void encode_payload() { ::encode(client, payload); - ::encode(addr, payload); ::encode(result, payload); ::encode(result_msg, payload); ::encode(monmap_bl, payload); diff --git a/src/messages/MMonMap.h b/src/messages/MMonMap.h index b2f5d966fc0..38e4280b763 100644 --- a/src/messages/MMonMap.h +++ b/src/messages/MMonMap.h @@ -19,24 +19,20 @@ class MMonMap : public Message { public: - entity_addr_t addr; bufferlist monmapbl; MMonMap() : Message(CEPH_MSG_MON_MAP) { } - MMonMap(entity_addr_t t, bufferlist &bl) : Message(CEPH_MSG_MON_MAP) { - addr = t; + MMonMap(bufferlist &bl) : Message(CEPH_MSG_MON_MAP) { monmapbl.claim(bl); } const char *get_type_name() { return "mon_map"; } void encode_payload() { - ::encode(addr, payload); ::encode(monmapbl, payload); } void decode_payload() { bufferlist::iterator p = payload.begin(); - ::decode(addr, p); ::decode(monmapbl, p); } }; diff --git a/src/mon/ClientMonitor.cc b/src/mon/ClientMonitor.cc index cffcaa66119..3fe2b985b09 100644 --- a/src/mon/ClientMonitor.cc +++ b/src/mon/ClientMonitor.cc @@ -236,7 +236,6 @@ void ClientMonitor::_mounted(__s64 client, MClientMount *m) // reply with client ticket MClientMountAck *ack = new MClientMountAck; ack->client = client; - ack->addr = to.addr; mon->monmap->encode(ack->monmap_bl); mon->send_reply(m, ack, to); diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc index 8c69ee8ffb6..e1d6d5e59bb 100644 --- a/src/mon/MonClient.cc +++ b/src/mon/MonClient.cc @@ -154,6 +154,9 @@ bool MonClient::ms_dispatch(Message *m) { dout(10) << "dispatch " << *m << dendl; + if (my_addr == entity_addr_t()) + my_addr = messenger->get_myaddr(); + switch (m->get_type()) { case CEPH_MSG_MON_MAP: handle_monmap((MMonMap*)m); @@ -176,10 +179,6 @@ void MonClient::handle_monmap(MMonMap *m) dout(10) << "handle_monmap " << *m << dendl; monc_lock.Lock(); - my_addr = m->addr; - messenger->_set_myaddr(m->addr); - dout(10) << " i am " << m->addr << dendl; - bufferlist::iterator p = m->monmapbl.begin(); ::decode(monmap, p); map_cond.Signal(); @@ -247,8 +246,7 @@ void MonClient::handle_mount_ack(MClientMountAck* m) bufferlist::iterator p = m->monmap_bl.begin(); ::decode(monmap, p); - messenger->_set_myaddr(m->addr); - messenger->reset_myname(entity_name_t::CLIENT(m->client)); + messenger->set_myname(entity_name_t::CLIENT(m->client)); // finish. timer.cancel_event(mount_timeout_event); diff --git a/src/mon/Monitor.cc b/src/mon/Monitor.cc index f8360d27951..89e6694cd49 100644 --- a/src/mon/Monitor.cc +++ b/src/mon/Monitor.cc @@ -525,7 +525,7 @@ void Monitor::handle_mon_get_map(MMonGetMap *m) dout(10) << "handle_mon_get_map" << dendl; bufferlist bl; monmap->encode(bl); - messenger->send_message(new MMonMap(m->get_orig_source_addr(), bl), m->get_orig_source_inst()); + messenger->send_message(new MMonMap(bl), m->get_orig_source_inst()); delete m; } diff --git a/src/msg/Messenger.h b/src/msg/Messenger.h index e37e4df701c..77b675bfea0 100644 --- a/src/msg/Messenger.h +++ b/src/msg/Messenger.h @@ -36,7 +36,7 @@ class Messenger { Dispatcher *dispatcher; protected: - entity_inst_t _myinst; + entity_name_t _my_name; int default_send_priority; atomic_t nref; @@ -45,7 +45,7 @@ protected: Messenger(entity_name_t w) : dispatcher(0), default_send_priority(CEPH_MSG_PRIO_DEFAULT), nref(1) { - _myinst.name = w; + _my_name = w; } virtual ~Messenger() { assert(nref.test() == 0); @@ -63,13 +63,11 @@ protected: } // accessors - entity_name_t get_myname() { return _myinst.name; } - const entity_addr_t& get_myaddr() { return _myinst.addr; } - const entity_inst_t& get_myinst() { return _myinst; } + entity_name_t get_myname() { return _my_name; } + virtual entity_addr_t get_myaddr() = 0; + entity_inst_t get_myinst() { return entity_inst_t(get_myname(), get_myaddr()); } - void _set_myname(entity_name_t m) { _myinst.name = m; } - virtual void _set_myaddr(entity_addr_t a) { _myinst.addr = a; } - virtual void reset_myname(entity_name_t m) = 0; + void set_myname(entity_name_t m) { _my_name = m; } void set_default_send_priority(int p) { default_send_priority = p; } int get_default_send_priority() { return default_send_priority; } diff --git a/src/msg/SimpleMessenger.cc b/src/msg/SimpleMessenger.cc index 4af974d9302..55f5d6649cd 100644 --- a/src/msg/SimpleMessenger.cc +++ b/src/msg/SimpleMessenger.cc @@ -144,9 +144,7 @@ int SimpleMessenger::Accepter::bind(int64_t force_nonce) } rank->rank_addr.erank = 0; - dout(1) << "accepter.bind rank_addr is " << rank->rank_addr - << " need_addr=" << rank->need_addr - << dendl; + dout(1) << "accepter.bind rank_addr is " << rank->rank_addr << " need_addr=" << rank->need_addr << dendl; return 0; } @@ -379,11 +377,12 @@ void SimpleMessenger::Endpoint::prepare_dest(const entity_inst_t& inst) int SimpleMessenger::Endpoint::send_message(Message *m, entity_inst_t dest) { // set envelope - m->set_source_inst(_myinst); - m->set_orig_source_inst(_myinst); + m->get_header().src = get_myinst(); + m->get_header().orig_src = m->get_header().src; + if (!m->get_priority()) m->set_priority(get_default_send_priority()); - dout(1) << m->get_source() + dout(1) << m->get_source_inst() << " --> " << dest.name << " " << dest.addr << " -- " << *m << " -- ?+" << m->get_data().length() @@ -398,7 +397,8 @@ int SimpleMessenger::Endpoint::send_message(Message *m, entity_inst_t dest) int SimpleMessenger::Endpoint::forward_message(Message *m, entity_inst_t dest) { // set envelope - m->set_source_inst(_myinst); + m->get_header().src = get_myinst(); + if (!m->get_priority()) m->set_priority(get_default_send_priority()); dout(1) << m->get_source() @@ -418,8 +418,9 @@ int SimpleMessenger::Endpoint::forward_message(Message *m, entity_inst_t dest) int SimpleMessenger::Endpoint::lazy_send_message(Message *m, entity_inst_t dest) { // set envelope - m->set_source_inst(_myinst); - m->set_orig_source_inst(_myinst); + m->get_header().src = get_myinst(); + m->get_header().orig_src = m->get_header().src; + if (!m->get_priority()) m->set_priority(get_default_send_priority()); dout(1) << "lazy " << m->get_source() @@ -442,31 +443,22 @@ int SimpleMessenger::Endpoint::send_keepalive(entity_inst_t dest) -void SimpleMessenger::Endpoint::_set_myaddr(entity_addr_t a) -{ - Messenger::_set_myaddr(a); // still call original - - dout(10) << "_set_myaddr " << a << dendl; - rank->rank_addr.ipaddr = a.ipaddr; -} - -void SimpleMessenger::Endpoint::reset_myname(entity_name_t newname) +void SimpleMessenger::Endpoint::mark_down(entity_addr_t a) { - entity_name_t oldname = get_myname(); - dout(10) << "reset_myname " << oldname << " to " << newname << dendl; - _set_myname(newname); + rank->mark_down(a); } -void SimpleMessenger::Endpoint::mark_down(entity_addr_t a) +entity_addr_t SimpleMessenger::Endpoint::get_myaddr() { - rank->mark_down(a); + entity_addr_t a = rank->rank_addr; + a.erank = my_rank; + return a; } - /************************************** * Pipe */ @@ -877,16 +869,8 @@ int SimpleMessenger::Pipe::connect() } dout(20) << "connect peer addr for me is " << peer_addr_for_me << dendl; - if (rank->need_addr) { - rank->lock.Lock(); - entity_addr_t was = rank->rank_addr; - rank->rank_addr.ipaddr = peer_addr_for_me.ipaddr; - rank->rank_addr.ipaddr.sin_port = was.ipaddr.sin_port; - dout(0) << "rank discovered i am " << rank->rank_addr - << " (was " << was << ", peer says i am " << peer_addr_for_me << ")" << dendl; - rank->need_addr = false; - rank->lock.Unlock(); - } + if (rank->need_addr) + rank->learned_addr(peer_addr_for_me); memset(&msg, 0, sizeof(msg)); msgvec[0].iov_base = (char*)&rank->rank_addr; @@ -1319,23 +1303,6 @@ void SimpleMessenger::Pipe::reader() // find entity entity = rank->local[erank]; entity->get(); - - // first message? - /* - if (entity->need_addr) { - entity->_set_myaddr(m->get_dest_inst().addr); - dout(2) << "reader entity addr is " << entity->get_myaddr() << dendl; - entity->need_addr = false; - } - - if (rank->need_addr) { - rank->rank_addr = m->get_dest_inst().addr; - rank->rank_addr.erank = 0; - dout(2) << "reader rank_addr is " << rank->rank_addr << dendl; - rank->need_addr = false; - } - */ - } else { derr(0) << "reader got message " << *m << ", which isn't local" << dendl; } @@ -2038,14 +2005,8 @@ SimpleMessenger::Endpoint *SimpleMessenger::register_entity(entity_name_t name) msgr->get(); local[erank] = msgr; stopped[erank] = false; - msgr->_myinst.addr = rank_addr; - if (msgr->_myinst.addr.ipaddr == entity_addr_t().ipaddr) - msgr->need_addr = true; - msgr->_myinst.addr.erank = erank; - dout(10) << "register_entity " << name << " at " << msgr->_myinst.addr - << " need_addr=" << need_addr - << dendl; + dout(10) << "register_entity " << name << " at " << msgr->get_myaddr() << dendl; num_local++; @@ -2261,3 +2222,13 @@ void SimpleMessenger::mark_down(entity_addr_t addr) lock.Unlock(); } +void SimpleMessenger::learned_addr(entity_addr_t peer_addr_for_me) +{ + lock.Lock(); + entity_addr_t was = rank_addr; + rank_addr.ipaddr = peer_addr_for_me.ipaddr; + rank_addr.ipaddr.sin_port = was.ipaddr.sin_port; + dout(1) << "learned my addr " << rank_addr << dendl; + need_addr = false; + lock.Unlock(); +} diff --git a/src/msg/SimpleMessenger.h b/src/msg/SimpleMessenger.h index d16b48d82d2..f33c2626e2b 100644 --- a/src/msg/SimpleMessenger.h +++ b/src/msg/SimpleMessenger.h @@ -291,10 +291,7 @@ private: bool stop; int qlen; int my_rank; - public: - bool need_addr; - private: class DispatchThread : public Thread { Endpoint *m; public: @@ -358,7 +355,6 @@ private: stop(false), qlen(0), my_rank(rn), - need_addr(false), dispatch_thread(this) { } ~Endpoint() { } @@ -379,8 +375,8 @@ private: int get_dispatch_queue_len() { return qlen; } - void _set_myaddr(entity_addr_t a); - void reset_myname(entity_name_t m); + entity_addr_t get_myaddr(); + int shutdown(); void suicide(); @@ -472,6 +468,8 @@ public: void prepare_dest(const entity_inst_t& inst); void send_keepalive(const entity_inst_t& addr); + void learned_addr(entity_addr_t peer_addr_for_me); + // create a new messenger Endpoint *new_entity(entity_name_t addr);