LU-11760 ofd: limit num of objects to create in 1 transaction

Set flag th_sync when the number of objects to create per
sequence reaches OST_MAX_PRECREATE in one transaction.
It is needed to avoid gaps after OST failover.
See details in LU-11760.

Lustre-change: https://review.whamcloud.com/35373
Lustre-commit: 4485ee8be4cf224e2543f6344efc6e1cb295a0a7

Change-Id: Ie29de5a42e757b07561749982359c01df999e798
Signed-off-by: Sergey Cheremencev <c17829@cray.com>
Tested-by: jenkins <devops@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Alexander Zarochentsev <c17826@cray.com>
Signed-off-by: Minh Diep <mdiep@whamcloud.com>
Reviewed-on: https://review.whamcloud.com/35951
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/ofd/ofd_fs.c b/lustre/ofd/ofd_fs.c
index 8be08a3..d964029 100644 (file)
--- a/lustre/ofd/ofd_fs.c
+++ b/lustre/ofd/ofd_fs.c
@@ -409,6 +409,7 @@ struct ofd_seq *ofd_seq_load(const struct lu_env *env, struct ofd_device *ofd,
        ostid_set_seq(&oseq->os_oi, seq);
 
        atomic_set(&oseq->os_refc, 1);
+       atomic_set(&oseq->os_precreate_in_progress, 0);
 
        rc = dt_attr_get(env, dob, &info->fti_attr);
        if (rc)

diff --git a/lustre/ofd/ofd_internal.h b/lustre/ofd/ofd_internal.h
index 3b3291c..f810ddb 100644 (file)
--- a/lustre/ofd/ofd_internal.h
+++ b/lustre/ofd/ofd_internal.h
@@ -93,6 +93,7 @@ struct ofd_seq {
        spinlock_t              os_last_oid_lock;
        struct mutex            os_create_lock;
        atomic_t                os_refc;
+       atomic_t                os_precreate_in_progress;
        struct dt_object        *os_lastid_obj;
        unsigned long           os_destroys_in_progress:1;
 };

diff --git a/lustre/ofd/ofd_objects.c b/lustre/ofd/ofd_objects.c
index 3e71d4b..64dcb7e 100644 (file)
--- a/lustre/ofd/ofd_objects.c
+++ b/lustre/ofd/ofd_objects.c
@@ -160,6 +160,68 @@ int ofd_object_ff_load(const struct lu_env *env, struct ofd_object *fo)
        return 0;
 }
 
+struct ofd_precreate_cb {
+       struct dt_txn_commit_cb  opc_cb;
+       struct ofd_seq          *opc_oseq;
+       int                      opc_objects;
+};
+
+static void ofd_cb_precreate(struct lu_env *env, struct thandle *th,
+                            struct dt_txn_commit_cb *cb, int err)
+{
+       struct ofd_precreate_cb *opc;
+       struct ofd_seq *oseq;
+
+       opc = container_of(cb, struct ofd_precreate_cb, opc_cb);
+       oseq = opc->opc_oseq;
+
+       CDEBUG(D_OTHER, "Sub %d from %d for "DFID", th_sync %d\n",
+              opc->opc_objects, atomic_read(&oseq->os_precreate_in_progress),
+              PFID(&oseq->os_oi.oi_fid), th->th_sync);
+       atomic_sub(opc->opc_objects, &oseq->os_precreate_in_progress);
+       ofd_seq_put(env, opc->opc_oseq);
+       OBD_FREE_PTR(opc);
+}
+
+static int ofd_precreate_cb_add(const struct lu_env *env, struct thandle *th,
+                               struct ofd_seq *oseq, int objects)
+{
+       struct ofd_precreate_cb *opc;
+       struct dt_txn_commit_cb *dcb;
+       int precreate, rc;
+
+       OBD_ALLOC_PTR(opc);
+       if (!opc)
+               return -ENOMEM;
+
+       precreate = atomic_read(&oseq->os_precreate_in_progress);
+       atomic_inc(&oseq->os_refc);
+       opc->opc_oseq = oseq;
+       opc->opc_objects = objects;
+       CDEBUG(D_OTHER, "Add %d to %d for "DFID", th_sync %d\n",
+              opc->opc_objects, precreate,
+              PFID(&oseq->os_oi.oi_fid), th->th_sync);
+
+       if ((precreate + objects) >= (5 * OST_MAX_PRECREATE))
+               th->th_sync = 1;
+
+       dcb = &opc->opc_cb;
+       dcb->dcb_func = ofd_cb_precreate;
+       INIT_LIST_HEAD(&dcb->dcb_linkage);
+       strlcpy(dcb->dcb_name, "ofd_cb_precreate", sizeof(dcb->dcb_name));
+
+       rc = dt_trans_cb_add(th, dcb);
+       if (rc) {
+               ofd_seq_put(env, oseq);
+               OBD_FREE_PTR(opc);
+               return rc;
+       }
+
+       atomic_add(objects, &oseq->os_precreate_in_progress);
+
+       return 0;
+}
+
 /**
  * Precreate the given number \a nr of objects in the given sequence \a oseq.
  *
@@ -379,6 +441,8 @@ int ofd_precreate_objects(const struct lu_env *env, struct ofd_device *ofd,
                               ofd_seq_last_oid(oseq));
        }
 
+       if (objects)
+               ofd_precreate_cb_add(env, th, oseq, objects);
 trans_stop:
        rc2 = ofd_trans_stop(env, ofd, th, rc);
        if (rc2)