From 0ef2a74097823121dad29a66799ef6837b5d156e Mon Sep 17 00:00:00 2001 From: sageweil Date: Mon, 12 Nov 2007 17:49:02 +0000 Subject: [PATCH] client mount instance cleanup, ceph_fsid checks git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@2049 29311d96-e01e-0410-9327-a35deaab8ce9 --- trunk/ceph/client/Client.cc | 6 ++-- trunk/ceph/include/types.h | 6 ++++ trunk/ceph/messages/MClientMount.h | 10 ++++--- trunk/ceph/messages/MClientUnmount.h | 10 +++---- trunk/ceph/messages/MMDSGetMap.h | 3 +- trunk/ceph/mon/ClientMonitor.cc | 45 +++++++++++----------------- trunk/ceph/mon/ClientMonitor.h | 14 ++++----- trunk/ceph/mon/MonMap.h | 13 +++++++- trunk/ceph/mon/OSDMonitor.cc | 8 +++-- trunk/ceph/msg/SimpleMessenger.cc | 12 ++++---- 10 files changed, 68 insertions(+), 59 deletions(-) diff --git a/trunk/ceph/client/Client.cc b/trunk/ceph/client/Client.cc index 14615d2a203fe..70ee46f4acf8b 100644 --- a/trunk/ceph/client/Client.cc +++ b/trunk/ceph/client/Client.cc @@ -1374,8 +1374,7 @@ void Client::_try_mount() int mon = monmap->pick_mon(); dout(2) << "sending client_mount to mon" << mon << " as instance " << my_instance << dendl; messenger->set_dispatcher(this); - messenger->send_message(new MClientMount(messenger->get_myaddr(), my_instance), - monmap->get_inst(mon)); + messenger->send_message(new MClientMount, monmap->get_inst(mon)); // schedule timeout? assert(mount_timeout_event == 0); @@ -1533,8 +1532,7 @@ int Client::unmount() // send unmount! int mon = monmap->pick_mon(); dout(2) << "sending client_unmount to mon" << mon << dendl; - messenger->send_message(new MClientUnmount(messenger->get_myinst()), - monmap->get_inst(mon)); + messenger->send_message(new MClientUnmount, monmap->get_inst(mon)); while (mounted) mount_cond.Wait(client_lock); diff --git a/trunk/ceph/include/types.h b/trunk/ceph/include/types.h index cf8374d329a77..c21c267ec0fdd 100644 --- a/trunk/ceph/include/types.h +++ b/trunk/ceph/include/types.h @@ -231,6 +231,12 @@ inline unsigned char MODE_TO_DT(int mode) { #define MAX_DENTRY_LEN 255 +// -- + +inline ostream& operator<<(ostream& out, ceph_fsid_t& f) { + return out << hex << f.major << '.' << f.minor << dec; +} + // -- io helpers -- diff --git a/trunk/ceph/messages/MClientMount.h b/trunk/ceph/messages/MClientMount.h index 44c943561f275..66033d0fc0360 100644 --- a/trunk/ceph/messages/MClientMount.h +++ b/trunk/ceph/messages/MClientMount.h @@ -19,24 +19,26 @@ class MClientMount : public Message { public: - entity_addr_t addr; - int32_t instance; // on this node - MClientMount() : Message(CEPH_MSG_CLIENT_MOUNT) { } - MClientMount(entity_addr_t a, int i = 0) : + /*MClientMount(entity_addr_t a, int i = 0) : Message(CEPH_MSG_CLIENT_MOUNT), addr(a), instance(i) { } + */ char *get_type_name() { return "client_mount"; } void decode_payload() { + /* int off = 0; ::_decode(addr, payload, off); ::_decode(instance, payload, off); + */ } void encode_payload() { + /* ::_encode(addr, payload); ::_encode(instance, payload); + */ } }; diff --git a/trunk/ceph/messages/MClientUnmount.h b/trunk/ceph/messages/MClientUnmount.h index b089cd86030d6..25f82dec8597a 100644 --- a/trunk/ceph/messages/MClientUnmount.h +++ b/trunk/ceph/messages/MClientUnmount.h @@ -19,21 +19,21 @@ class MClientUnmount : public Message { public: - entity_inst_t inst; - MClientUnmount() : Message(CEPH_MSG_CLIENT_UNMOUNT) { } + /* MClientUnmount(entity_inst_t i) : Message(CEPH_MSG_CLIENT_UNMOUNT), inst(i) { } + */ char *get_type_name() { return "client_unmount"; } void decode_payload() { - int off = 0; - ::_decode(inst, payload, off); + //int off = 0; + //::_decode(inst, payload, off); } void encode_payload() { - ::_encode(inst, payload); + //::_encode(inst, payload); } }; diff --git a/trunk/ceph/messages/MMDSGetMap.h b/trunk/ceph/messages/MMDSGetMap.h index 3658b86f18f50..80f753bd365eb 100644 --- a/trunk/ceph/messages/MMDSGetMap.h +++ b/trunk/ceph/messages/MMDSGetMap.h @@ -21,8 +21,7 @@ class MMDSGetMap : public Message { public: - MMDSGetMap() : Message(CEPH_MSG_MDS_GETMAP) { - } + MMDSGetMap() : Message(CEPH_MSG_MDS_GETMAP) { } char *get_type_name() { return "mdsgetmap"; } diff --git a/trunk/ceph/mon/ClientMonitor.cc b/trunk/ceph/mon/ClientMonitor.cc index 8c76a82a2fb84..9082db0b2db8f 100644 --- a/trunk/ceph/mon/ClientMonitor.cc +++ b/trunk/ceph/mon/ClientMonitor.cc @@ -126,11 +126,9 @@ bool ClientMonitor::preprocess_query(Message *m) case CEPH_MSG_CLIENT_MOUNT: { // already mounted? - MClientMount *mount = (MClientMount*)m; entity_addr_t addr = m->get_source_addr(); - pair addrinst(addr, mount->instance); - if (client_map.addr_client.count(addrinst)) { - int client = client_map.addr_client[addrinst]; + if (client_map.addr_client.count(addr)) { + int client = client_map.addr_client[addr]; dout(7) << " client" << client << " already mounted" << dendl; _mounted(client, (MClientMount*)m); return true; @@ -165,45 +163,38 @@ bool ClientMonitor::prepare_update(Message *m) switch (m->get_type()) { case CEPH_MSG_CLIENT_MOUNT: { - MClientMount *mount = (MClientMount*)m; - pair addrinst(mount->addr, mount->instance); + entity_addr_t addr = m->get_source_addr(); int client = -1; - if (mount->get_source().is_client()) - client = mount->get_source().num(); + if (m->get_source().is_client()) + client = m->get_source().num(); // choose a client id if (client < 0) { client = pending_inc.next_client; - dout(10) << "mount: assigned client" << client << " to " << mount->addr << dendl; + dout(10) << "mount: assigned client" << client << " to " << addr << dendl; } else { - dout(10) << "mount: client" << client << " requested by " - << mount->addr << "i" << mount->instance - << dendl; + dout(10) << "mount: client" << client << " requested by " << addr << dendl; if (client_map.client_addr.count(client)) { - assert(client_map.client_addr[client] != addrinst); - dout(0) << "mount: WARNING: client" << client << " requested by " - << mount->addr << "." << mount->instance - << ", which used to be " - << client_map.client_addr[client].first << "i" << client_map.client_addr[client].second - << dendl; + assert(client_map.client_addr[client] != addr); + dout(0) << "mount: WARNING: client" << client << " requested by " << addr + << ", which used to be " << client_map.client_addr[client] << dendl; } } - pending_inc.add_mount(client, mount->addr, mount->instance); - paxos->wait_for_commit(new C_Mounted(this, client, mount)); + pending_inc.add_mount(client, addr); + paxos->wait_for_commit(new C_Mounted(this, client, (MClientMount*)m)); } return true; case CEPH_MSG_CLIENT_UNMOUNT: { - MClientUnmount *unmount = (MClientUnmount*)m; - assert(unmount->inst.name.is_client()); - int client = unmount->inst.name.num(); + assert(m->get_source().is_client()); + int client = m->get_source().num(); assert(client_map.client_addr.count(client)); pending_inc.add_unmount(client); - paxos->wait_for_commit(new C_Unmounted(this, unmount)); + paxos->wait_for_commit(new C_Unmounted(this, (MClientUnmount*)m)); } return true; @@ -222,7 +213,7 @@ bool ClientMonitor::prepare_update(Message *m) void ClientMonitor::_mounted(int client, MClientMount *m) { entity_inst_t to; - to.addr = m->addr; + to.addr = m->get_source_addr(); to.name = entity_name_t::CLIENT(client); dout(10) << "_mounted client" << client << " at " << to << dendl; @@ -236,10 +227,10 @@ void ClientMonitor::_mounted(int client, MClientMount *m) void ClientMonitor::_unmounted(MClientUnmount *m) { - dout(10) << "_unmounted " << m->inst << dendl; + dout(10) << "_unmounted " << m->get_source_inst() << dendl; // reply with (same) unmount message - mon->messenger->send_message(m, m->inst); + mon->messenger->send_message(m, m->get_source_inst()); // auto-shutdown? // (hack for fakesyn/newsyn, mostly) diff --git a/trunk/ceph/mon/ClientMonitor.h b/trunk/ceph/mon/ClientMonitor.h index f36ee9f7c18bd..bd4dc4bc523d5 100644 --- a/trunk/ceph/mon/ClientMonitor.h +++ b/trunk/ceph/mon/ClientMonitor.h @@ -37,15 +37,15 @@ public: struct Incremental { version_t version; uint32_t next_client; - map > mount; + map mount; set unmount; Incremental() : version(0), next_client() {} bool is_empty() { return mount.empty() && unmount.empty(); } - void add_mount(uint32_t client, entity_addr_t addr, int instance) { + void add_mount(uint32_t client, entity_addr_t addr) { next_client = MAX(next_client, client+1); - mount[client] = pair(addr, instance); + mount[client] = addr; } void add_unmount(uint32_t client) { assert(client < next_client); @@ -72,14 +72,14 @@ public: struct Map { version_t version; uint32_t next_client; - map > client_addr; - map,uint32_t> addr_client; + map client_addr; + map addr_client; Map() : version(0), next_client(0) {} void reverse() { addr_client.clear(); - for (map >::iterator p = client_addr.begin(); + for (map::iterator p = client_addr.begin(); p != client_addr.end(); ++p) { addr_client[p->second] = p->first; @@ -89,7 +89,7 @@ public: assert(inc.version == version+1); version = inc.version; next_client = inc.next_client; - for (map >::iterator p = inc.mount.begin(); + for (map::iterator p = inc.mount.begin(); p != inc.mount.end(); ++p) { client_addr[p->first] = p->second; diff --git a/trunk/ceph/mon/MonMap.h b/trunk/ceph/mon/MonMap.h index dbe9c9b5ac5e9..c13e41472d608 100644 --- a/trunk/ceph/mon/MonMap.h +++ b/trunk/ceph/mon/MonMap.h @@ -25,12 +25,15 @@ class MonMap { public: epoch_t epoch; // what epoch/version of the monmap + ceph_fsid_t fsid; int32_t num_mon; vector mon_inst; int last_mon; // last mon i talked to - MonMap(int s=0) : epoch(0), num_mon(s), mon_inst(s), last_mon(-1) {} + MonMap(int s=0) : epoch(0), num_mon(s), mon_inst(s), last_mon(-1) { + generate_fsid(); + } void add_mon(entity_inst_t inst) { mon_inst.push_back(inst); @@ -53,6 +56,7 @@ class MonMap { void encode(bufferlist& blist) { ::_encode(epoch, blist); + ::_encode(fsid, blist); ::_encode(num_mon, blist); ::_encode(mon_inst, blist); } @@ -60,10 +64,17 @@ class MonMap { void decode(bufferlist& blist) { int off = 0; ::_decode(epoch, blist, off); + ::_decode(fsid, blist, off); ::_decode(num_mon, blist, off); ::_decode(mon_inst, blist, off); } + + void generate_fsid() { + fsid.major = ((uint64_t)rand() << 32) + rand(); + fsid.minor = ((uint64_t)rand() << 32) + rand(); + } + // read from/write to a file int write(char *fn) { // encode diff --git a/trunk/ceph/mon/OSDMonitor.cc b/trunk/ceph/mon/OSDMonitor.cc index da5c30d8a9faf..322011b7b3039 100644 --- a/trunk/ceph/mon/OSDMonitor.cc +++ b/trunk/ceph/mon/OSDMonitor.cc @@ -102,10 +102,11 @@ void OSDMonitor::create_initial() assert(mon->is_leader()); assert(paxos->get_version() == 0); - dout(1) << "create_initial -- creating initial osdmap from g_conf" << dendl; + dout(1) << "create_initial for " << mon->monmap->fsid << " from g_conf" << dendl; // OSDMap newmap; + newmap.set_fsid(mon->monmap->fsid); newmap.mon_epoch = mon->mon_epoch; newmap.ctime = g_clock.now(); @@ -316,8 +317,9 @@ bool OSDMonitor::update_from_paxos() void OSDMonitor::create_pending() { pending_inc = OSDMap::Incremental(osdmap.epoch+1); - dout(10) << "create_pending e " << pending_inc.epoch - << dendl; + pending_inc.fsid = mon->monmap->fsid; + + dout(10) << "create_pending e " << pending_inc.epoch << dendl; } void OSDMonitor::encode_pending(bufferlist &bl) diff --git a/trunk/ceph/msg/SimpleMessenger.cc b/trunk/ceph/msg/SimpleMessenger.cc index 670aca1ec1f24..2842ab393846b 100644 --- a/trunk/ceph/msg/SimpleMessenger.cc +++ b/trunk/ceph/msg/SimpleMessenger.cc @@ -827,7 +827,7 @@ int Rank::Pipe::connect() dout(10) << "connecting to " << peer_addr.v.ipaddr << dendl; rc = ::connect(newsd, (sockaddr*)&peer_addr.v.ipaddr, sizeof(peer_addr.v.ipaddr)); if (rc < 0) { - dout(10) << "connect error " << peer_addr.v.ipaddr + dout(2) << "connect error " << peer_addr.v.ipaddr << ", " << errno << ": " << strerror(errno) << dendl; goto fail; } @@ -843,7 +843,7 @@ int Rank::Pipe::connect() // identify peer rc = tcp_read(newsd, (char*)&paddr, sizeof(paddr)); if (rc < 0) { - dout(0) << "connect couldn't read peer addr, " << strerror(errno) << dendl; + dout(2) << "connect couldn't read peer addr, " << strerror(errno) << dendl; goto fail; } dout(20) << "connect read peer addr " << paddr << dendl; @@ -885,7 +885,7 @@ int Rank::Pipe::connect() // FINISH if (state != STATE_CONNECTING) { dout(2) << "connect hmm, race durring connect(), not connecting anymore, failing" << dendl; - goto fail2; // hmm! + goto fail_locked; // hmm! } if (tag == CEPH_MSGR_TAG_REJECT) { if (connect_seq != cseq) { @@ -895,7 +895,7 @@ int Rank::Pipe::connect() } else { dout(0) << "connect got REJECT, connection race (harmless), connect_seq=" << connect_seq << dendl; } - goto fail2; + goto fail_locked; } assert(tag == CEPH_MSGR_TAG_READY); state = STATE_OPEN; @@ -912,9 +912,9 @@ int Rank::Pipe::connect() fail: lock.Lock(); - fail2: + fail_locked: if (newsd > 0) ::close(newsd); - fault(tag == CEPH_MSGR_TAG_REJECT); + fault(tag == CEPH_MSGR_TAG_REJECT); // quiet if reject (not socket error) return -1; } -- 2.39.5