b=24255 move seq_client_alloc_seq out of lcs_sem

[fs/lustre-release.git] / lustre / fid / fid_request.c

diff --git a/lustre/fid/fid_request.c b/lustre/fid/fid_request.c
index 7b5d545..6e8d2a4 100644 (file)
--- a/lustre/fid/fid_request.c
+++ b/lustre/fid/fid_request.c
@@ -26,7 +26,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  */
 /*
@@ -89,17 +89,26 @@ static int seq_client_rpc(struct lu_client_seq *seq,
 
         ptlrpc_request_set_replen(req);
 
-        if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
-                req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
-                        SEQ_CONTROLLER_PORTAL : SEQ_METADATA_PORTAL;
-                /* update mdt field of *in, it is required for fld update
-                 * on super sequence allocator node. */
-                if (opc == SEQ_ALLOC_SUPER)
-                        in->lsr_mdt = seq->lcs_space.lsr_mdt;
+       if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
+                req->rq_request_portal = SEQ_METADATA_PORTAL;
+                in->lsr_flags = LU_SEQ_RANGE_MDT;
         } else {
-                LASSERT(opc == SEQ_ALLOC_META);
+                LASSERTF(seq->lcs_type == LUSTRE_SEQ_DATA,
+                         "unknown lcs_type %u\n", seq->lcs_type);
                 req->rq_request_portal = SEQ_DATA_PORTAL;
+                in->lsr_flags = LU_SEQ_RANGE_OST;
         }
+
+        if (opc == SEQ_ALLOC_SUPER) {
+                /* Update index field of *in, it is required for
+                 * FLD update on super sequence allocator node. */
+                in->lsr_index = seq->lcs_space.lsr_index;
+                req->rq_request_portal = SEQ_CONTROLLER_PORTAL;
+        } else {
+                LASSERTF(opc == SEQ_ALLOC_META,
+                         "unknown opcode %u\n, opc", opc);
+        }
+
         ptlrpc_at_set_req_timeout(req);
 
         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
@@ -211,22 +220,62 @@ static int seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
         RETURN(rc);
 }
 
+static int seq_fid_alloc_prep(struct lu_client_seq *seq,
+                              cfs_waitlink_t *link)
+{
+
+        cfs_waitq_add(&seq->lcs_waitq, link);
+        if (seq->lcs_update) {
+                cfs_up(&seq->lcs_sem);
+                cfs_set_current_state(CFS_TASK_UNINT);
+                cfs_waitq_wait(link, CFS_TASK_UNINT);
+                cfs_set_current_state(CFS_TASK_RUNNING);
+                cfs_down(&seq->lcs_sem);
+                return -EAGAIN;
+        }
+        ++seq->lcs_update;
+        cfs_up(&seq->lcs_sem);
+        return 0;
+}
+
+static void seq_fid_alloc_fini(struct lu_client_seq *seq,
+                               cfs_waitlink_t *link)
+{
+        LASSERT(seq->lcs_update == 1);
+        cfs_down(&seq->lcs_sem);
+        --seq->lcs_update;
+        cfs_waitq_del(&seq->lcs_waitq, link);
+        cfs_waitq_signal(&seq->lcs_waitq);
+}
+
 /* Allocate new fid on passed client @seq and save it to @fid. */
 int seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
 {
+        cfs_waitlink_t link;
         int rc;
         ENTRY;
 
         LASSERT(seq != NULL);
         LASSERT(fid != NULL);
 
+        cfs_waitlink_init(&link);
         cfs_down(&seq->lcs_sem);
 
-        if (fid_is_zero(&seq->lcs_fid) ||
-            fid_oid(&seq->lcs_fid) >= seq->lcs_width)
-        {
+        while (1) {
                 seqno_t seqnr;
 
+                if (!fid_is_zero(&seq->lcs_fid) &&
+                    fid_oid(&seq->lcs_fid) < seq->lcs_width) {
+                        /* Just bump last allocated fid and return to caller. */
+                        seq->lcs_fid.f_oid += 1;
+                        rc = 0;
+                        break;
+                }
+
+                rc = seq_fid_alloc_prep(seq, &link);
+                if (rc)
+                        continue;
+
                 rc = seq_client_alloc_seq(seq, &seqnr);
                 if (rc) {
                         CERROR("%s: Can't allocate new sequence, "
@@ -247,10 +296,9 @@ int seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
                  * to setup FLD for it.
                  */
                 rc = 1;
-        } else {
-                /* Just bump last allocated fid and return to caller. */
-                seq->lcs_fid.f_oid += 1;
-                rc = 0;
+
+                seq_fid_alloc_fini(seq, &link);
+                break;
         }
 
         *fid = seq->lcs_fid;
@@ -275,7 +323,7 @@ void seq_client_flush(struct lu_client_seq *seq)
          * set to -1 for dgb check.
          */
 
-        seq->lcs_space.lsr_mdt = -1;
+        seq->lcs_space.lsr_index = -1;
 
         range_init(&seq->lcs_space);
         cfs_up(&seq->lcs_sem);
@@ -355,6 +403,7 @@ int seq_client_init(struct lu_client_seq *seq,
         seq->lcs_type = type;
         cfs_sema_init(&seq->lcs_sem, 1);
         seq->lcs_width = LUSTRE_SEQ_MAX_WIDTH;
+        cfs_waitq_init(&seq->lcs_waitq);
 
         /* Make sure that things are clear before work is started. */
         seq_client_flush(seq);