mds: use intrusive set for ClientLease tracking

This allows to avoid additional redundant lookups in CDentry::client_leases
for some scenarios, e.g.:
* CDentry::remove_client_lease() is called from CDentry::remove_client_leases().
* CDentry::remove_client_lease() is called from Locker::remove_stale_leases()
* CDentry::remove_client_lease() is called from Locker::process_request_cap_release()

And a few similar cases.
In all of them a caller has a pointer to ClientLease object but has to
perform another lookup to remove that lease.

Signed-off-by: Garry Drankovich <[email protected]>

Author: gardran

src/mds/CDentry.cc

@@ -500,6 +500,8 @@ void CDentry::decode_lock_state(int type, const bufferlist& bl)
 }
 
 
+MEMPOOL_DEFINE_OBJECT_FACTORY(ClientLease, mds_client_lease, mds_co);
+
 client_t ClientLease::get_client() const
 {
   return session->get_client();
@@ -508,17 +510,19 @@ client_t ClientLease::get_client() const
 ClientLease *CDentry::add_client_lease(Session *session)
 {
   client_t client = session->get_client();
-  auto em = client_leases.emplace(std::piecewise_construct,
-				  std::forward_as_tuple(client),
-				  std::forward_as_tuple(this, session));
-  ClientLease *l = &em.first->second;
-  if (em.second) {
+  ClientLease* l = nullptr;
+  auto it = client_leases.lower_bound(client);
+  if (it == client_leases.end() || it->get_client() != client) {
+    l = new ClientLease(this, session);
     dout(20) << __func__ << " client." << client << " on " << lock << dendl;
-    if (client_leases.size() == 1) {
+    if (client_leases.empty()) {
       get(PIN_CLIENTLEASE);
       lock.get_client_lease();
     }
+    client_leases.insert_before(it, *l);
     l->seq = ++session->lease_seq;
+  } else {
+    l = &(*it);
   }
   return l;
 }
@@ -528,12 +532,12 @@ void CDentry::remove_client_lease(ClientLease *l, Locker *locker)
   ceph_assert(l->parent == this);
 
   bool gather = false;
-  client_t client = l->get_client();
-  dout(20) << __func__ << " client." << client << " on " << lock << dendl;
+  dout(20) << __func__ << " client." << l->get_client() << " on " << lock << dendl;
 
   l->item_lease.remove_myself();
   l->item_session_lease.remove_myself();
-  client_leases.erase(client);
+  client_leases.erase(client_leases.iterator_to(*l));
+  delete l;
 
   if (client_leases.empty()) {
     gather = !lock.is_stable();
@@ -548,7 +552,7 @@ void CDentry::remove_client_lease(ClientLease *l, Locker *locker)
 void CDentry::remove_client_leases(Locker *locker)
 {
   while (!client_leases.empty())
-    remove_client_lease(&client_leases.begin()->second, locker);
+    remove_client_lease(&(*client_leases.begin()), locker);
 }
 
 void CDentry::_put()

src/mds/CDentry.h

@@ -17,14 +17,14 @@
 
 #include <string>
 #include <string_view>
-#include <set>
 
 #include "include/counter.h"
 #include "include/types.h"
 #include "include/buffer_fwd.h"
 #include "include/lru.h"
 #include "include/elist.h"
 #include "include/filepath.h"
+#include <boost/intrusive/set.hpp>
 
 #include "BatchOp.h"
 #include "MDSCacheObject.h"
@@ -34,14 +34,16 @@
 #include "ScrubHeader.h"
 
 class CInode;
-class CDentry;
 class CDir;
 class Locker;
 class CDentry;
 class LogSegment;
 class Session;
 
-struct ClientLease {
+struct ClientLease : public boost::intrusive::set_base_hook<>
+{
+  MEMPOOL_CLASS_HELPERS();
+
   ClientLease(CDentry *p, Session *s) :
     parent(p), session(s),
     item_session_lease(this),
@@ -57,6 +59,13 @@ struct ClientLease {
   xlist<ClientLease*>::item item_session_lease; // per-session list
   xlist<ClientLease*>::item item_lease;         // global list
 };
+struct client_is_key
+{
+  typedef client_t type;
+  const type operator() (const ClientLease& l) const {
+    return l.get_client();
+  }
+};
 
 // define an ordering
 bool operator<(const CDentry& l, const CDentry& r);
@@ -346,13 +355,13 @@ class CDentry : public MDSCacheObject, public LRUObject, public Counter<CDentry>
   const ClientLease *get_client_lease(client_t c) const {
     auto it = client_leases.find(c);
     if (it != client_leases.end())
-      return &it->second;
+      return &(*it);
     return nullptr;
   }
   ClientLease *get_client_lease(client_t c) {
     auto it = client_leases.find(c);
     if (it != client_leases.end())
-      return &it->second;
+      return &(*it);
     return nullptr;
   }
   bool have_client_lease(client_t c) const {
@@ -388,7 +397,10 @@ class CDentry : public MDSCacheObject, public LRUObject, public Counter<CDentry>
   SimpleLock lock; // FIXME referenced containers not in mempool
   LocalLockC versionlock; // FIXME referenced containers not in mempool
 
-  mempool::mds_co::map<client_t, ClientLease> client_leases;
+  typedef boost::intrusive::set<
+    ClientLease, boost::intrusive::key_of_value<client_is_key>> ClientLeaseMap;
+  ClientLeaseMap client_leases;
+
   std::map<int, std::unique_ptr<BatchOp>> batch_ops;
 
   ceph_tid_t reintegration_reqid = 0;

src/mds/Locker.cc

@@ -4529,8 +4529,7 @@ void Locker::issue_client_lease(CDentry *dn, CInode *in, const MDRequestRef& mdr
 void Locker::revoke_client_leases(SimpleLock *lock)
 {
   CDentry *dn = static_cast<CDentry*>(lock->get_parent());
-  for (auto& p : dn->client_leases) {
-    ClientLease *l = &p.second;
+  for (ClientLease& l : dn->client_leases) {
 
     ceph_assert(lock->get_type() == CEPH_LOCK_DN);
 
@@ -4539,8 +4538,8 @@ void Locker::revoke_client_leases(SimpleLock *lock)
 
     // i should also revoke the dir ICONTENT lease, if they have it!
     CInode *diri = dn->get_dir()->get_inode();
-    auto lease = make_message<MClientLease>(CEPH_MDS_LEASE_REVOKE, l->seq, mask, diri->ino(), diri->first, CEPH_NOSNAP, dn->get_name());
-    mds->send_message_client_counted(lease, l->session);
+    auto lease = make_message<MClientLease>(CEPH_MDS_LEASE_REVOKE, l.seq, mask, diri->ino(), diri->first, CEPH_NOSNAP, dn->get_name());
+    mds->send_message_client_counted(lease, l.session);
   }
 }