mds: disallow slave requests when MDS is readonly

Signed-off-by: Yan, Zheng <zyan@redhat.com>

diff --git a/src/mds/Mutation.h b/src/mds/Mutation.h
index a334a7b071d51960bbaf1f57395b31804ea7f18b..9b6569536410951e9b5df97d39d924fe26cf8fa5 100644 (file)
--- a/src/mds/Mutation.h
+++ b/src/mds/Mutation.h
@@ -224,6 +224,7 @@ struct MDRequestImpl : public MutationImpl, public TrackedOp {
   // break rarely-used fields into a separately allocated structure 
   // to save memory for most ops
   struct More {
+    int slave_error;
     set<mds_rank_t> slaves;           // mds nodes that have slave requests to me (implies client_request)
     set<mds_rank_t> waiting_on_slave; // peers i'm waiting for slavereq replies from. 
 
@@ -271,6 +272,7 @@ struct MDRequestImpl : public MutationImpl, public TrackedOp {
     filepath filepath2;
 
     More() : 
+      slave_error(0),
       has_journaled_slaves(false), slave_update_journaled(false),
       srcdn_auth_mds(-1), inode_import_v(0), rename_inode(0),
       is_freeze_authpin(false), is_ambiguous_auth(false),

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 25400eba6443d921760933b6818477d35087b6fe..c2e2185332e5537cb2bb61fbf2ee01e8cd9c2f1b 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -1296,6 +1296,11 @@ void Server::dispatch_client_request(MDRequestRef& mdr)
       respond_to_request(mdr, -EROFS);
       return;
     }
+    if (mdr->has_more() && mdr->more()->slave_error) {
+      dout(10) << " got error from slaves" << dendl;
+      respond_to_request(mdr, mdr->more()->slave_error);
+      return;
+    }
   }
   
   switch (req->get_op()) {
@@ -1725,21 +1730,29 @@ void Server::handle_slave_auth_pin(MDRequestRef& mdr)
   // build list of objects
   list<MDSCacheObject*> objects;
   CInode *auth_pin_freeze = NULL;
-  bool fail = false, wouldblock = false;
+  bool fail = false, wouldblock = false, readonly = false;
 
-  for (vector<MDSCacheObjectInfo>::iterator p = mdr->slave_request->get_authpins().begin();
-       p != mdr->slave_request->get_authpins().end();
-       ++p) {
-    MDSCacheObject *object = mdcache->get_object(*p);
-    if (!object) {
-      dout(10) << " don't have " << *p << dendl;
-      fail = true;
-      break;
-    }
+  if (mdcache->is_readonly()) {
+    dout(10) << " read-only FS" << dendl;
+    readonly = true;
+    fail = true;
+  }
 
-    objects.push_back(object);
-    if (*p == mdr->slave_request->get_authpin_freeze())
-      auth_pin_freeze = static_cast<CInode*>(object);
+  if (!fail) {
+    for (vector<MDSCacheObjectInfo>::iterator p = mdr->slave_request->get_authpins().begin();
+       p != mdr->slave_request->get_authpins().end();
+       ++p) {
+      MDSCacheObject *object = mdcache->get_object(*p);
+      if (!object) {
+       dout(10) << " don't have " << *p << dendl;
+       fail = true;
+       break;
+      }
+
+      objects.push_back(object);
+      if (*p == mdr->slave_request->get_authpin_freeze())
+       auth_pin_freeze = static_cast<CInode*>(object);
+    }
   }
   
   // can we auth pin them?
@@ -1839,6 +1852,8 @@ void Server::handle_slave_auth_pin(MDRequestRef& mdr)
 
   if (wouldblock)
     reply->mark_error_wouldblock();
+  if (readonly)
+    reply->mark_error_rofs();
 
   mds->send_message_mds(reply, mdr->slave_to_mds);
   
@@ -1881,8 +1896,13 @@ void Server::handle_slave_auth_pin_ack(MDRequestRef& mdr, MMDSSlaveRequest *ack)
     }
   }
 
-  if (ack->is_error_wouldblock())
+  if (ack->is_error_rofs()) {
+    mdr->more()->slave_error = -EROFS;
+    mdr->aborted = true;
+  } else if (ack->is_error_wouldblock()) {
+    mdr->more()->slave_error = -EWOULDBLOCK;
     mdr->aborted = true;
+  }
   
   // note slave
   mdr->more()->slaves.insert(from);

diff --git a/src/messages/MMDSSlaveRequest.h b/src/messages/MMDSSlaveRequest.h
index a11337c91910face3e480aad97094f19577701c7..ae2badcc6f9c2578a07054f20f2a81a85307b7fe 100644 (file)
--- a/src/messages/MMDSSlaveRequest.h
+++ b/src/messages/MMDSSlaveRequest.h
@@ -99,6 +99,7 @@ class MMDSSlaveRequest : public Message {
   static const unsigned FLAG_NONBLOCK  = 1;
   static const unsigned FLAG_WOULDBLOCK        = 2;
   static const unsigned FLAG_NOTJOURNALED = 4;
+  static const unsigned FLAG_EROFS = 8;
 
   // for locking
   __u16 lock_type;  // lock object type
@@ -136,6 +137,8 @@ public:
   bool is_error_wouldblock() { return (flags & FLAG_WOULDBLOCK); }
   void mark_not_journaled() { flags |= FLAG_NOTJOURNALED; }
   bool is_not_journaled() { return (flags & FLAG_NOTJOURNALED); }
+  void mark_error_rofs() { flags |= FLAG_EROFS; }
+  bool is_error_rofs() { return (flags & FLAG_EROFS); }
 
   void set_lock_type(int t) { lock_type = t; }