LU-3534 osp: move RPC pack from declare to execution phase

[fs/lustre-release.git] / lustre / osp / osp_dev.c

diff --git a/lustre/osp/osp_dev.c b/lustre/osp/osp_dev.c
index 61e697f..445dfd2 100644 (file)
--- a/lustre/osp/osp_dev.c
+++ b/lustre/osp/osp_dev.c
@@ -27,7 +27,7 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, 2013, Intel Corporation.
+ * Copyright (c) 2012, 2014, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -78,7 +78,6 @@
 #include <lustre_ioctl.h>
 #include <lustre_param.h>
 #include <lustre_log.h>
-#include <lustre_mdc.h>
 
 #include "osp_internal.h"
 
@@ -111,9 +110,9 @@ static struct lu_kmem_descr osp_caches[] = {
  * \retval object      object being created if the creation succeed.
  * \retval NULL                NULL if the creation failed.
  */
-struct lu_object *osp_object_alloc(const struct lu_env *env,
-                                  const struct lu_object_header *hdr,
-                                  struct lu_device *d)
+static struct lu_object *osp_object_alloc(const struct lu_env *env,
+                                         const struct lu_object_header *hdr,
+                                         struct lu_device *d)
 {
        struct lu_object_header *h = NULL;
        struct osp_object       *o;
@@ -181,7 +180,7 @@ static struct dt_object
                RETURN(dto);
 
        /* Get attributes of the local object. */
-       rc = dt_attr_get(env, dto, attr, NULL);
+       rc = dt_attr_get(env, dto, attr);
        if (rc) {
                CERROR("%s: can't be initialized: rc = %d\n",
                       osp->opd_obd->obd_name, rc);
@@ -502,9 +501,10 @@ static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
 
        rc = osp_disconnect(d);
 
-       osp_sync_fini(d);
-
        if (!d->opd_connect_mdt) {
+               /* stop sync thread */
+               osp_sync_fini(d);
+
                /* stop precreate thread */
                osp_precreate_fini(d);
 
@@ -690,12 +690,30 @@ static int osp_sync(const struct lu_env *env, struct dt_device *dev)
        unsigned long      start = cfs_time_current();
        ENTRY;
 
+       /* No Sync between MDTs yet. */
+       if (d->opd_connect_mdt)
+               RETURN(0);
+
        if (unlikely(d->opd_imp_active == 0))
                RETURN(-ENOTCONN);
 
        id = d->opd_syn_last_used_id;
+       down_write(&d->opd_async_updates_rwsem);
+
+       CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
+              atomic_read(&d->opd_async_updates_count));
+
+       /* make sure the connection is fine */
+       expire = cfs_time_shift(obd_timeout);
+       lwi = LWI_TIMEOUT(expire - cfs_time_current(), osp_sync_timeout, d);
+       rc = l_wait_event(d->opd_syn_barrier_waitq,
+                         atomic_read(&d->opd_async_updates_count) == 0,
+                         &lwi);
+       up_write(&d->opd_async_updates_rwsem);
+       if (rc != 0)
+               GOTO(out, rc);
 
-       CDEBUG(D_OTHER, "%s: id: used %lu, processed %lu\n",
+       CDEBUG(D_CACHE, "%s: id: used %lu, processed %lu\n",
               d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);
 
        /* wait till all-in-line are processed */
@@ -727,7 +745,7 @@ static int osp_sync(const struct lu_env *env, struct dt_device *dev)
        /* block new processing (barrier>0 - few callers are possible */
        atomic_inc(&d->opd_syn_barrier);
 
-       CDEBUG(D_OTHER, "%s: %u in flight\n", d->opd_obd->obd_name,
+       CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
               d->opd_syn_rpc_in_flight);
 
        /* wait till all-in-flight are replied, so executed by the target */
@@ -756,13 +774,14 @@ static int osp_sync(const struct lu_env *env, struct dt_device *dev)
                GOTO(out, rc = -ETIMEDOUT);
        }
 
-       CDEBUG(D_OTHER, "%s: done in %lu\n", d->opd_obd->obd_name,
-              cfs_time_current() - start);
 out:
        /* resume normal processing (barrier=0) */
        atomic_dec(&d->opd_syn_barrier);
        __osp_sync_check_for_work(d);
 
+       CDEBUG(D_CACHE, "%s: done in %lu: rc = %d\n", d->opd_obd->obd_name,
+              cfs_time_current() - start, rc);
+
        RETURN(rc);
 }
 
@@ -867,6 +886,9 @@ static int osp_init0(const struct lu_env *env, struct osp_device *osp,
        ENTRY;
 
        mutex_init(&osp->opd_async_requests_mutex);
+       INIT_LIST_HEAD(&osp->opd_async_updates);
+       init_rwsem(&osp->opd_async_updates_rwsem);
+       atomic_set(&osp->opd_async_updates_count, 0);
 
        obd = class_name2obd(lustre_cfg_string(cfg, 0));
        if (obd == NULL) {
@@ -1029,16 +1051,16 @@ static int osp_init0(const struct lu_env *env, struct osp_device *osp,
                rc = osp_init_precreate(osp);
                if (rc)
                        GOTO(out_last_used, rc);
-       }
 
-       /*
-        * Initialize synhronization mechanism taking
-        * care of propogating changes to OST in near
-        * transactional manner.
-        */
-       rc = osp_sync_init(env, osp);
-       if (rc)
-               GOTO(out_precreat, rc);
+               /*
+                * Initialize synhronization mechanism taking
+                * care of propogating changes to OST in near
+                * transactional manner.
+                */
+               rc = osp_sync_init(env, osp);
+               if (rc < 0)
+                       GOTO(out_precreat, rc);
+       }
 
        /*
         * Initiate connect to OST
@@ -1056,8 +1078,9 @@ static int osp_init0(const struct lu_env *env, struct osp_device *osp,
        RETURN(0);
 
 out:
-       /* stop sync thread */
-       osp_sync_fini(osp);
+       if (!osp->opd_connect_mdt)
+               /* stop sync thread */
+               osp_sync_fini(osp);
 out_precreat:
        /* stop precreate thread */
        if (!osp->opd_connect_mdt)
@@ -1633,8 +1656,8 @@ static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
  *                      requested from seq meta server
  * \retval negative    negative errno if FID allocation failed.
  */
-int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
-                 struct lu_fid *fid, struct md_op_data *unused)
+static int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
+                        struct lu_fid *fid, struct md_op_data *unused)
 {
        struct client_obd       *cli = &exp->exp_obd->u.cli;
        struct osp_device       *osp = lu2osp_dev(exp->exp_obd->obd_lu_dev);
@@ -1762,8 +1785,8 @@ static int __init osp_mod_init(void)
                return rc;
 
        type = class_search_type(LUSTRE_OSP_NAME);
-       type->typ_procsym = lprocfs_seq_register("osc", proc_lustre_root,
-                                                NULL, NULL);
+       type->typ_procsym = lprocfs_register("osc", proc_lustre_root,
+                                            NULL, NULL);
        if (IS_ERR(type->typ_procsym)) {
                CERROR("osp: can't create compat entry \"osc\": %d\n",
                       (int) PTR_ERR(type->typ_procsym));