Branch b1_6

b=16260
i=tianzy
i=panda

The quota master must return EAGAIN to quota request until the config is
processed.

lustre/include/lustre_quota.h

@@ -230,8 +230,12 @@ struct lustre_quota_ctxt {
         dqacq_handler_t lqc_handler;    /* dqacq/dqrel RPC handler, only for quota master */
         unsigned long lqc_flags;        /* quota flags */
         unsigned long lqc_recovery:1,   /* Doing recovery */
-                      lqc_switch_qs:1;  /* the function of change qunit size
+                      lqc_switch_qs:1,  /* the function of change qunit size
                                          * 0:Off, 1:On */
+                      lqc_setup:1;      /* tell whether of not quota_type has
+                                         * been processed, so that the master
+                                         * knows when it can start processing
+                                         * incoming acq/rel quota requests */
         unsigned long lqc_iunit_sz;     /* original unit size of file quota and
                                          * upper limitation for adjust file
                                          * qunit */

lustre/ldlm/ldlm_lib.c

@@ -1787,6 +1787,17 @@ int target_handle_dqacq_callback(struct ptlrpc_request *req)
         master_obd = obd->obd_observer->obd_observer;
         qctxt = &master_obd->u.obt.obt_qctxt;
 
+        if (!qctxt->lqc_setup) {
+                /* quota_type has not been processed yet, return EAGAIN
+                 * until we know whether or not quotas are supposed to
+                 * be enabled */
+                CDEBUG(D_QUOTA, "quota_type not processed yet, return "
+                                "-EAGAIN\n");
+                req->rq_status = -EAGAIN;
+                rc = ptlrpc_reply(req);
+                GOTO(out, rc);
+        }
+
         LASSERT(qctxt->lqc_handler);
         rc = qctxt->lqc_handler(master_obd, qdata,
                                 lustre_msg_get_opc(req->rq_reqmsg));

lustre/mds/mds_lov.c

@@ -1054,6 +1054,10 @@ int mds_notify(struct obd_device *obd, struct obd_device *watched,
                 break;
         case OBD_NOTIFY_CONFIG:
                 mds_allow_cli(obd, (unsigned long)data);
+
+                /* quota_type has been processed, we can now handle
+                 * incoming quota requests */
+                obd->u.obt.obt_qctxt.lqc_setup = 1;
         default:
                 RETURN(0);
         }

lustre/quota/quota_context.c

@@ -1072,6 +1072,7 @@ qctxt_init(struct obd_device *obd, dqacq_handler_t handler)
         qctxt->lqc_cqs_least_iunit = 2;
         qctxt->lqc_cqs_qs_factor = 2;
         qctxt->lqc_flags = 0;
+        qctxt->lqc_setup = 0;
         qctxt->lqc_bunit_sz = default_bunit_sz;
         qctxt->lqc_btune_sz = default_bunit_sz / 100 * default_btune_ratio;
         qctxt->lqc_iunit_sz = default_iunit_sz;