mds: wait for client's mdsmap when specifying data pool

The client may have a newer map than we do; make sure we wait for it lest
we inadvertantly reply because we think the pool doesn't exist.

Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index a677bd2c2aae6510cf5b748a7cf3f407e9860f8c..1cc0d1b0727a3483bf63c01409ad964cf3709a52 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -2615,9 +2615,16 @@ void Server::handle_client_openc(MDRequest *mdr)
   if (req->head.args.open.object_size)
     layout.fl_object_size = req->head.args.open.object_size;
   if (req->get_connection()->has_feature(CEPH_FEATURE_CREATEPOOLID) &&
-      (__s32)req->head.args.open.pool >= 0)
+      (__s32)req->head.args.open.pool >= 0) {
     layout.fl_pg_pool = req->head.args.open.pool;
 
+    // make sure we have as new a map as the client
+    if (req->get_mdsmap_epoch() > mds->mdsmap->get_epoch()) {
+      mds->wait_for_mdsmap(req->get_mdsmap_epoch(), new C_MDS_RetryRequest(mdcache, mdr));
+      return;
+    }
+  }
+
   if (!ceph_file_layout_is_valid(&layout)) {
     dout(10) << " invalid initial file layout" << dendl;
     reply_request(mdr, -EINVAL);
@@ -3286,8 +3293,15 @@ void Server::handle_client_setlayout(MDRequest *mdr)
     layout.fl_cas_hash = req->head.args.setlayout.layout.fl_cas_hash;
   if (req->head.args.setlayout.layout.fl_object_stripe_unit > 0)
     layout.fl_object_stripe_unit = req->head.args.setlayout.layout.fl_object_stripe_unit;
-  if (req->head.args.setlayout.layout.fl_pg_pool > 0)
+  if (req->head.args.setlayout.layout.fl_pg_pool > 0) {
     layout.fl_pg_pool = req->head.args.setlayout.layout.fl_pg_pool;
+
+    // make sure we have as new a map as the client
+    if (req->get_mdsmap_epoch() > mds->mdsmap->get_epoch()) {
+      mds->wait_for_mdsmap(req->get_mdsmap_epoch(), new C_MDS_RetryRequest(mdcache, mdr));
+      return;
+    }
+  }
   if (!ceph_file_layout_is_valid(&layout)) {
     dout(10) << "bad layout" << dendl;
     reply_request(mdr, -EINVAL);
@@ -3361,8 +3375,16 @@ void Server::handle_client_setdirlayout(MDRequest *mdr)
     layout->layout.fl_cas_hash = req->head.args.setlayout.layout.fl_cas_hash;
   if (req->head.args.setlayout.layout.fl_object_stripe_unit > 0)
     layout->layout.fl_object_stripe_unit = req->head.args.setlayout.layout.fl_object_stripe_unit;
-  if (req->head.args.setlayout.layout.fl_pg_pool > 0)
+  if (req->head.args.setlayout.layout.fl_pg_pool > 0) {
     layout->layout.fl_pg_pool = req->head.args.setlayout.layout.fl_pg_pool;
+
+    // make sure we have as new a map as the client
+    if (req->get_mdsmap_epoch() > mds->mdsmap->get_epoch()) {
+      delete layout;
+      mds->wait_for_mdsmap(req->get_mdsmap_epoch(), new C_MDS_RetryRequest(mdcache, mdr));
+      return;
+    }
+  }
   if (!ceph_file_layout_is_valid(&layout->layout)) {
     dout(10) << "bad layout" << dendl;
     reply_request(mdr, -EINVAL);