Index: include/lustre_export.h
===================================================================
RCS file: /cvsroot/cfs/lustre-core/include/lustre_export.h,v
retrieving revision 1.4.6.9.2.7
diff -u -p -r1.4.6.9.2.7 lustre_export.h
--- include/lustre_export.h	28 May 2009 03:55:07 -0000	1.4.6.9.2.7
+++ include/lustre_export.h	25 Aug 2009 17:41:07 -0000
@@ -166,7 +166,8 @@ struct obd_export {
                                   exp_libclient:1, /* liblustre client? */
                                   /* client timed out and tried to reconnect,
                                    * but couldn't because of active rpcs */
-                                  exp_abort_active_req:1;
+                                  exp_abort_active_req:1,
+                                  exp_ctrl_shutdown:1;
         struct list_head          exp_queued_rpc;  /* RPC to be handled */
         /* VBR: per-export last committed */
         __u64                     exp_last_committed;
Index: include/lustre_lib.h
===================================================================
RCS file: /cvsroot/cfs/lustre-core/include/lustre_lib.h,v
retrieving revision 1.5.4.6.4.3
diff -u -p -r1.5.4.6.4.3 lustre_lib.h
--- include/lustre_lib.h	21 Aug 2009 01:29:00 -0000	1.5.4.6.4.3
+++ include/lustre_lib.h	25 Aug 2009 17:41:07 -0000
@@ -89,6 +89,7 @@ void target_destroy_export(struct obd_ex
 int target_handle_ping(struct ptlrpc_request *req);
 int target_pack_pool_reply(struct ptlrpc_request *req);
 void target_committed_to_req(struct ptlrpc_request *req);
+void target_ctrl_shutdown(struct obd_device *obd);
 
 /* quotacheck callback, dqacq/dqrel callback handler */
 int target_handle_qc_callback(struct ptlrpc_request *req);
@@ -478,6 +479,7 @@ static inline void obd_ioctl_freedata(ch
 #define OBD_IOC_NO_TRANSNO             _IOW ('f', 140, OBD_IOC_DATA_TYPE)
 #define OBD_IOC_SET_READONLY           _IOW ('f', 141, OBD_IOC_DATA_TYPE)
 #define OBD_IOC_ABORT_RECOVERY         _IOR ('f', 142, OBD_IOC_DATA_TYPE)
+#define OBD_IOC_CTRL_SHUTDOWN          _IOW ('f', 143, OBD_IOC_DATA_TYPE)
 
 #define OBD_GET_VERSION                _IOWR ('f', 144, OBD_IOC_DATA_TYPE)
 
Index: include/lustre_net.h
===================================================================
RCS file: /cvsroot/cfs/lustre-core/include/lustre_net.h,v
retrieving revision 1.9.6.28.2.14
diff -u -p -r1.9.6.28.2.14 lustre_net.h
--- include/lustre_net.h	16 Jul 2009 05:22:05 -0000	1.9.6.28.2.14
+++ include/lustre_net.h	25 Aug 2009 17:41:08 -0000
@@ -328,7 +328,9 @@ struct ptlrpc_request {
                 /* a copy of the request is queued to replay during recovery */
                 rq_copy_queued:1,
                 /* whether the rquest is a copy of another replay request */
-                rq_copy:1;
+                rq_copy:1,
+                /* request packed in 2.0 format */
+                rq_fmt_20:1;
         enum rq_phase rq_phase;     /* one of RQ_PHASE_* */
         enum rq_phase rq_next_phase; /* one of RQ_PHASE_* to be used next */
         atomic_t rq_refcount;   /* client-side refcount for SENT race,
@@ -994,6 +996,7 @@ void lustre_msg_set_conn_cnt(struct lust
 void lustre_msg_set_timeout(struct lustre_msg *msg, __u32 timeout);
 void lustre_msg_set_service_time(struct lustre_msg *msg, __u32 service_time);
 void lustre_msg_set_cksum(struct lustre_msg *msg, __u32 cksum);
+int lustre_reqmsg_is_mds_open(struct lustre_msg *msg);
 
 static inline void
 ptlrpc_rqphase_move(struct ptlrpc_request *req, enum rq_phase new_phase)
Index: include/obd.h
===================================================================
RCS file: /cvsroot/cfs/lustre-core/include/obd.h,v
retrieving revision 1.8.6.29.2.31
diff -u -p -r1.8.6.29.2.31 obd.h
--- include/obd.h	23 Jul 2009 10:14:20 -0000	1.8.6.29.2.31
+++ include/obd.h	25 Aug 2009 17:41:08 -0000
@@ -936,8 +936,10 @@ struct obd_device {
                       obd_fail:1,          /* cleanup with failover */
                       obd_async_recov:1,   /* allow asyncronous orphan cleanup */
                       obd_no_conn:1,       /* deny new connections */
-                      obd_inactive:1;      /* device active/inactive
+                      obd_inactive:1,      /* device active/inactive
                                             * (for /proc/status only!!) */
+                      obd_css_blocking:1;  /* block further api calls for
+                                            * controlled server shutdown */
         /* uuid-export hash body */
         struct lustre_hash     *obd_uuid_hash;
         /* nid-export hash body */
@@ -996,6 +998,10 @@ struct obd_device {
         time_t                           obd_recovery_end; /* seconds, for lprocfs_status */
         time_t                           obd_recovery_time_hard;
         int                              obd_recovery_timeout;
+        /* css: server: the number of exports to wait for */
+        atomic_t                         obd_css_nr_exp;
+        /* css: client: waitqueue for blocking requests */
+        cfs_waitq_t                      obd_css_waitq;
 
         union {
                 struct obd_device_target obt;
@@ -1271,6 +1277,8 @@ struct obd_ops {
         void (*o_getref)(struct obd_device *obd);
         void (*o_putref)(struct obd_device *obd);
 
+        int (*o_convert_rpc)(struct obd_device *obd,
+                             struct ptlrpc_request *req);
         /*
          * NOTE: If adding ops, add another LPROCFS_OBD_OP_INIT() line
          * to lprocfs_alloc_obd_stats() in obdclass/lprocfs_status.c.
Index: include/obd_class.h
===================================================================
RCS file: /cvsroot/cfs/lustre-core/include/obd_class.h,v
retrieving revision 1.6.4.17.2.10
diff -u -p -r1.6.4.17.2.10 obd_class.h
--- include/obd_class.h	7 Jul 2009 16:09:03 -0000	1.6.4.17.2.10
+++ include/obd_class.h	25 Aug 2009 17:41:08 -0000
@@ -103,6 +103,8 @@ int obd_zombie_impexp_init(void);
 void obd_zombie_impexp_stop(void);
 void obd_zombie_impexp_cull(void);
 
+int obd_wait_css(struct obd_device *obd, const char *func);
+
 /* obd_config.c */
 int class_process_config(struct lustre_cfg *lcfg);
 int class_process_proc_param(char *prefix, struct lprocfs_vars *lvars,
@@ -339,6 +341,15 @@ do {                                    
         }                                                       \
 } while (0)
 
+#define OBD_CHECK_CSS_BLOCKING(obd)                             \
+do {                                                            \
+        if (unlikely(obd->obd_css_blocking)) {                  \
+                int ret = obd_wait_css(obd, __FUNCTION__);      \
+                if (ret)                                        \
+                        RETURN(ret);                            \
+        }                                                       \
+} while (0)
+
 static inline int class_devno_max(void)
 {
         return MAX_OBD_DEVICES;
@@ -353,6 +364,7 @@ static inline int obd_get_info(struct ob
 
         EXP_CHECK_OP(exp, get_info);
         EXP_COUNTER_INCREMENT(exp, get_info);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, get_info)(exp, keylen, key, vallen, val, lsm);
         RETURN(rc);
@@ -367,6 +379,7 @@ static inline int obd_set_info_async(str
 
         EXP_CHECK_OP(exp, set_info_async);
         EXP_COUNTER_INCREMENT(exp, set_info_async);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, set_info_async)(exp, keylen, key, vallen, val,
                                                set);
@@ -528,6 +541,7 @@ static inline int obd_precreate(struct o
 
         EXP_CHECK_OP(exp, precreate);
         OBD_COUNTER_INCREMENT(exp->exp_obd, precreate);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, precreate)(exp);
         RETURN(rc);
@@ -543,6 +557,7 @@ static inline int obd_create_async(struc
 
         EXP_CHECK_OP(exp, create_async);
         EXP_COUNTER_INCREMENT(exp, create_async);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, create_async)(exp, oinfo, ea, oti);
         RETURN(rc);
@@ -557,6 +572,7 @@ static inline int obd_create(struct obd_
 
         EXP_CHECK_OP(exp, create);
         EXP_COUNTER_INCREMENT(exp, create);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, create)(exp, obdo, ea, oti);
         RETURN(rc);
@@ -572,6 +588,7 @@ static inline int obd_destroy(struct obd
 
         EXP_CHECK_OP(exp, destroy);
         EXP_COUNTER_INCREMENT(exp, destroy);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, destroy)(exp, obdo, ea, oti, md_exp);
         RETURN(rc);
@@ -584,6 +601,7 @@ static inline int obd_getattr(struct obd
 
         EXP_CHECK_OP(exp, getattr);
         EXP_COUNTER_INCREMENT(exp, getattr);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, getattr)(exp, oinfo);
         RETURN(rc);
@@ -598,6 +616,7 @@ static inline int obd_getattr_async(stru
 
         EXP_CHECK_OP(exp, getattr_async);
         EXP_COUNTER_INCREMENT(exp, getattr_async);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, getattr_async)(exp, oinfo, set);
         RETURN(rc);
@@ -611,6 +630,7 @@ static inline int obd_setattr(struct obd
 
         EXP_CHECK_OP(exp, setattr);
         EXP_COUNTER_INCREMENT(exp, setattr);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, setattr)(exp, oinfo, oti);
         RETURN(rc);
@@ -627,6 +647,7 @@ static inline int obd_setattr_rqset(stru
 
         EXP_CHECK_OP(exp, setattr_async);
         EXP_COUNTER_INCREMENT(exp, setattr_async);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         set =  ptlrpc_prep_set();
         if (set == NULL)
@@ -651,6 +672,7 @@ static inline int obd_setattr_async(stru
 
         EXP_CHECK_OP(exp, setattr_async);
         EXP_COUNTER_INCREMENT(exp, setattr_async);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, setattr_async)(exp, oinfo, oti, set);
         RETURN(rc);
@@ -902,6 +924,7 @@ static inline int obd_statfs_async(struc
 
         OBD_CHECK_OP(obd, statfs, -EOPNOTSUPP);
         OBD_COUNTER_INCREMENT(obd, statfs);
+        OBD_CHECK_CSS_BLOCKING(obd);
 
         CDEBUG(D_SUPER, "%s: osfs %p age "LPU64", max_age "LPU64"\n",
                obd->obd_name, &obd->obd_osfs, obd->obd_osfs_age, max_age);
@@ -959,6 +982,7 @@ static inline int obd_statfs(struct obd_
 
         OBD_CHECK_OP(obd, statfs, -EOPNOTSUPP);
         OBD_COUNTER_INCREMENT(obd, statfs);
+        OBD_CHECK_CSS_BLOCKING(obd);
 
         CDEBUG(D_SUPER, "osfs "LPU64", max_age "LPU64"\n",
                obd->obd_osfs_age, max_age);
@@ -992,6 +1016,7 @@ static inline int obd_sync_rqset(struct 
 
         OBD_CHECK_OP(exp->exp_obd, sync, -EOPNOTSUPP);
         EXP_COUNTER_INCREMENT(exp, sync);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         set =  ptlrpc_prep_set();
         if (set == NULL)
@@ -1013,6 +1038,7 @@ static inline int obd_sync(struct obd_ex
 
         OBD_CHECK_OP(exp->exp_obd, sync, -EOPNOTSUPP);
         EXP_COUNTER_INCREMENT(exp, sync);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, sync)(exp, oinfo, start, end, set);
         RETURN(rc);
@@ -1028,6 +1054,7 @@ static inline int obd_punch_rqset(struct
 
         EXP_CHECK_OP(exp, punch);
         EXP_COUNTER_INCREMENT(exp, punch);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         set =  ptlrpc_prep_set();
         if (set == NULL)
@@ -1049,6 +1076,7 @@ static inline int obd_punch(struct obd_e
 
         EXP_CHECK_OP(exp, punch);
         EXP_COUNTER_INCREMENT(exp, punch);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, punch)(exp, oinfo, oti, rqset);
         RETURN(rc);
@@ -1063,6 +1091,7 @@ static inline int obd_brw(int cmd, struc
 
         EXP_CHECK_OP(exp, brw);
         EXP_COUNTER_INCREMENT(exp, brw);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         if (!(cmd & (OBD_BRW_RWMASK | OBD_BRW_CHECK))) {
                 CERROR("obd_brw: cmd must be OBD_BRW_READ, OBD_BRW_WRITE, "
@@ -1084,6 +1113,7 @@ static inline int obd_brw_async(int cmd,
 
         EXP_CHECK_OP(exp, brw_async);
         EXP_COUNTER_INCREMENT(exp, brw_async);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         if (!(cmd & OBD_BRW_RWMASK)) {
                 CERROR("obd_brw: cmd must be OBD_BRW_READ or OBD_BRW_WRITE\n");
@@ -1173,6 +1203,7 @@ static inline int obd_queue_async_io(str
 
         OBD_CHECK_OP(exp->exp_obd, queue_async_io, -EOPNOTSUPP);
         EXP_COUNTER_INCREMENT(exp, queue_async_io);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
         LASSERT(cmd & OBD_BRW_RWMASK);
 
         rc = OBP(exp->exp_obd, queue_async_io)(exp, lsm, loi, cookie, cmd, off,
@@ -1209,6 +1240,7 @@ static inline int obd_queue_group_io(str
 
         OBD_CHECK_OP(exp->exp_obd, queue_group_io, -EOPNOTSUPP);
         EXP_COUNTER_INCREMENT(exp, queue_group_io);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
         LASSERT(cmd & OBD_BRW_RWMASK);
 
         rc = OBP(exp->exp_obd, queue_group_io)(exp, lsm, loi, oig, cookie,
@@ -1257,6 +1289,7 @@ static inline int obd_preprw(int cmd, st
 
         OBD_CHECK_OP(exp->exp_obd, preprw, -EOPNOTSUPP);
         EXP_COUNTER_INCREMENT(exp, preprw);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, preprw)(cmd, exp, oa, objcount, obj, remote,
                                        pages, local, oti);
@@ -1329,6 +1362,7 @@ static inline int obd_iocontrol(unsigned
 
         EXP_CHECK_OP(exp, iocontrol);
         EXP_COUNTER_INCREMENT(exp, iocontrol);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, iocontrol)(cmd, exp, len, karg, uarg);
         RETURN(rc);
@@ -1344,6 +1378,7 @@ static inline int obd_enqueue_rqset(stru
 
         EXP_CHECK_OP(exp, enqueue);
         EXP_COUNTER_INCREMENT(exp, enqueue);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         set =  ptlrpc_prep_set();
         if (set == NULL)
@@ -1366,6 +1401,7 @@ static inline int obd_enqueue(struct obd
 
         EXP_CHECK_OP(exp, enqueue);
         EXP_COUNTER_INCREMENT(exp, enqueue);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, enqueue)(exp, oinfo, einfo, set);
         RETURN(rc);
@@ -1381,6 +1417,7 @@ static inline int obd_match(struct obd_e
 
         EXP_CHECK_OP(exp, match);
         EXP_COUNTER_INCREMENT(exp, match);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, match)(exp, ea, type, policy, mode, flags, data,
                                       lockh, n_matches);
@@ -1450,6 +1487,7 @@ static inline int obd_pin(struct obd_exp
 
         EXP_CHECK_OP(exp, pin);
         EXP_COUNTER_INCREMENT(exp, pin);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, pin)(exp, fid, handle, flag);
         RETURN(rc);
@@ -1463,6 +1501,7 @@ static inline int obd_unpin(struct obd_e
 
         EXP_CHECK_OP(exp, unpin);
         EXP_COUNTER_INCREMENT(exp, unpin);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, unpin)(exp, handle, flag);
         RETURN(rc);
@@ -1542,6 +1581,7 @@ static inline int obd_quotacheck(struct 
 
         EXP_CHECK_OP(exp, quotacheck);
         EXP_COUNTER_INCREMENT(exp, quotacheck);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, quotacheck)(exp, oqctl);
         RETURN(rc);
@@ -1555,6 +1595,7 @@ static inline int obd_quotactl(struct ob
 
         EXP_CHECK_OP(exp, quotactl);
         EXP_COUNTER_INCREMENT(exp, quotactl);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, quotactl)(exp, oqctl);
         RETURN(rc);
@@ -1578,6 +1619,7 @@ static inline int obd_quota_adjust_qunit
 #endif
         EXP_CHECK_OP(exp, quota_adjust_qunit);
         EXP_COUNTER_INCREMENT(exp, quota_adjust_qunit);
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         rc = OBP(exp->exp_obd, quota_adjust_qunit)(exp, oqaq, qctxt);
 
@@ -1682,6 +1724,19 @@ static inline int obd_unregister_lock_ca
         RETURN(rc);
 }
 
+static inline int obd_convert_rpc(struct obd_device *obd,
+                                  struct ptlrpc_request *req)
+{
+        int rc;
+        ENTRY;
+
+        OBD_CHECK_OP(obd, convert_rpc, 0);
+        OBD_COUNTER_INCREMENT(obd, convert_rpc);
+
+        rc = OBP(obd, convert_rpc)(obd, req);
+        RETURN(rc);
+}
+
 /* OBD Metadata Support */
 
 extern int obd_init_caches(void);
Index: ldlm/ldlm_lib.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ldlm/ldlm_lib.c,v
retrieving revision 1.96.6.38.2.32
diff -u -p -r1.96.6.38.2.32 ldlm_lib.c
--- ldlm/ldlm_lib.c	28 Jul 2009 23:28:19 -0000	1.96.6.38.2.32
+++ ldlm/ldlm_lib.c	25 Aug 2009 17:41:08 -0000
@@ -1059,6 +1059,9 @@ int target_handle_disconnect(struct ptlr
         int rc;
         ENTRY;
 
+        if (req->rq_export->exp_ctrl_shutdown)
+                atomic_dec(&req->rq_export->exp_obd->obd_css_nr_exp);
+
         rc = lustre_pack_reply(req, 1, NULL, NULL);
         if (rc)
                 RETURN(rc);
@@ -2016,6 +2019,140 @@ int target_handle_qc_callback(struct ptl
         return 0;
 }
 
+/*
+ * FIXME we don't wait reply. if we do, better to use ptlrpc_set to
+ * send parallel notifications.
+ */
+static int target_css_send_notify(struct obd_import *imp)
+{
+        struct ptlrpc_request  *req;
+        int                     rc;
+
+        req = ptlrpc_prep_req(imp, LUSTRE_DLM_VERSION, 12345678, 1, 0, NULL);
+        if (req == NULL)
+                return -ENOMEM;
+
+        /* XXX rpc counters?? */
+        rc = ptl_send_rpc(req, 1);
+        ptlrpc_req_finished(req);
+        return rc;
+}
+
+static int target_ctrl_shutdown_notify(struct obd_export *exp)
+{
+        CWARN("notifying exp %p\n", exp);
+
+        if (exp->exp_disconnected) {
+                CWARN("exp %p: already started disconnecting, skip\n", exp);
+                return 1;
+        }
+
+        /*
+         * if the export is under connecting/reconnecting, it'd be unsafe
+         * to reference the reverse import. this should rarely happen.
+         */
+        if (exp->exp_connecting) {
+                CWARN("exp %p: during connecting/reconnecting, skip\n", exp);
+                return 1;
+        }
+
+        /* XXX
+         * there's still a race: we started just after a reconnect passed
+         * obd_stopping checking, but before exp_connecting get set.
+         */
+
+        if (exp->exp_imp_reverse == NULL) {
+                CWARN("exp %p: no reverse import, skip\n", exp);
+                return 1;
+        }
+
+        /* XXX take a ref on the import */
+        return target_css_send_notify(exp->exp_imp_reverse);
+}
+
+void target_ctrl_shutdown(struct obd_device *obd)
+{
+        wait_queue_head_t       waitq;
+        struct l_wait_info      lwi;
+        struct obd_export      *exp;
+        int                     nr_client = 0;
+        int                     notify_error = 0;
+        int                     waittime = 60;
+        ENTRY;
+
+        if (obd->obd_recovering) {
+                LCONSOLE_WARN("%s: in recovery, can't proceed to controlled "
+                              "shutdown!\n", obd->obd_name);
+                EXIT;
+                return;
+        }
+
+        LCONSOLE_INFO("%s: start controlled shutdown, syncing with "
+                      "clients...\n", obd->obd_name);
+        LASSERT(obd->obd_fail == 0);
+        /*
+         * from now on, no new client will be accepted, but existing
+         * export can be disconnected at any time.
+         */
+        obd->obd_no_conn = 1;
+
+startover:
+        spin_lock(&obd->obd_dev_lock);
+        list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
+                LASSERT(!exp->exp_delayed);
+                if (exp->exp_ctrl_shutdown)
+                        continue;
+
+                if (exp == obd->obd_self_export)
+                        continue;
+
+                exp->exp_ctrl_shutdown = 1;
+                nr_client++;
+                /*
+                 * add failover flag to preserve the export status.
+                 */
+                exp->exp_flags |= OBD_OPT_FAILOVER;
+
+                class_export_get(exp);
+                spin_unlock(&obd->obd_dev_lock);
+
+                atomic_inc(&obd->obd_css_nr_exp);
+                if (target_ctrl_shutdown_notify(exp)) {
+                        notify_error++;
+                        atomic_dec(&obd->obd_css_nr_exp);
+                }
+
+                /*
+                 * if exp_disconnected is set, there would be no guarantee
+                 * the export is still on the obd chain, we simply startover.
+                 * this won't happen usually, and it's not performance critical
+                 * here anyway.
+                 */
+                if (exp->exp_disconnected) {
+                        class_export_put(exp);
+                        goto startover;
+                }
+
+                spin_lock(&obd->obd_dev_lock);
+                class_export_put(exp);
+        }
+        spin_unlock(&obd->obd_dev_lock);
+
+        /* XXX
+         * perhaps we can adjust waiting time according to detected situations??
+         */
+        init_waitqueue_head(&waitq);
+        lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(waittime),
+                                   cfs_time_seconds(5), NULL, NULL);
+        l_wait_event(waitq, atomic_read(&obd->obd_css_nr_exp) <= 0, &lwi);
+
+        LCONSOLE_INFO("%s: controlled shutdown finished, total %d clients, "
+                      "%d couldn't be notified, %d didn't finish syncing\n",
+                      obd->obd_name, nr_client, notify_error,
+                      atomic_read(&obd->obd_css_nr_exp));
+        EXIT;
+}
+
 #ifdef HAVE_QUOTA_SUPPORT
 int target_handle_dqacq_callback(struct ptlrpc_request *req)
 {
Index: ldlm/ldlm_lockd.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ldlm/ldlm_lockd.c,v
retrieving revision 1.189.34.40.2.13
diff -u -p -r1.189.34.40.2.13 ldlm_lockd.c
--- ldlm/ldlm_lockd.c	16 Jul 2009 01:32:09 -0000	1.189.34.40.2.13
+++ ldlm/ldlm_lockd.c	25 Aug 2009 17:41:08 -0000
@@ -1638,6 +1638,170 @@ int ldlm_bl_to_thread_list(struct ldlm_n
 #endif
 }
 
+static int replay_queue_empty(struct obd_import *imp)
+{
+        struct ptlrpc_request  *req;
+        int                     rc = 1;
+
+        LASSERT(imp);
+
+        spin_lock(&imp->imp_lock);
+
+        list_for_each_entry(req, &imp->imp_replay_list, rq_replay_list) {
+                if (!lustre_reqmsg_is_mds_open(req->rq_reqmsg)) {
+                        CWARN("found replay opc %u, x%lld\n",
+                              lustre_msg_get_opc(req->rq_reqmsg), req->rq_xid);
+                        rc = 0;
+                        break;
+                }
+        }
+
+        spin_unlock(&imp->imp_lock);
+        return rc;
+}
+
+static int check_css_sync_status(struct obd_import *imp, int *rc)
+{
+        if (imp->imp_state != LUSTRE_IMP_FULL)
+                return 1;
+
+        if (atomic_read(&imp->imp_inflight) == 0 && replay_queue_empty(imp)) {
+                *rc = 0;
+                return 1;
+        }
+
+        return 0;
+}
+
+static int do_css_sync(struct obd_import *imp,
+                       struct ldlm_namespace *ns,
+                       int timeout)
+{
+        struct l_wait_info      lwi;
+        wait_queue_head_t       waitq;
+        int                     rc = 1;
+
+        /*
+         * NOTE grouplocks can't be flushed, we have to live with it.
+         */
+        CWARN("start cleaning namespace: %llu res, %d locks\n",
+              ns->ns_resources, atomic_read(&ns->ns_locks));
+        ldlm_namespace_cleanup(ns, 0);
+        CWARN("finish cleaning namespace: %llu res, %d locks\n",
+              ns->ns_resources, atomic_read(&ns->ns_locks));
+
+        init_waitqueue_head(&waitq);
+        /*
+         * NOTE: ping itself will increase imp_inflight.
+         */
+        lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(timeout),
+                                   cfs_time_seconds(3), NULL, NULL);
+        l_wait_event(waitq, check_css_sync_status(imp, &rc), &lwi);
+
+        if (imp->imp_state != LUSTRE_IMP_FULL)
+                CWARN("imp state changed to %s, abort syncing\n",
+                      ptlrpc_import_state_name(imp->imp_state));
+        else if (rc)
+                CWARN("syncing timed out, abort\n");
+
+        return rc;
+}
+
+/* XXX
+ * to start a separate thread to do syncing
+ */
+static void ldlm_handle_ctrl_shutdown(struct obd_device *obd)
+{
+        struct ldlm_namespace  *ns;
+        struct obd_import      *imp;
+        int                     start_time, timeout;
+        int                     rc;
+        ENTRY;
+
+        /*
+         * during CSS export ref is hold, thus the obd won't be
+         * freed. so obd->u.cli.cl_sem is always usable.
+         */
+
+        spin_lock(&obd->obd_dev_lock);
+        if (obd->obd_css_blocking) {
+                spin_unlock(&obd->obd_dev_lock);
+                CWARN("%s: already started blocking for CSS\n", obd->obd_name);
+                EXIT;
+                return;
+        }
+
+        LCONSOLE_INFO("%s: client start blocking requests for server "
+                      "shutdown\n", obd->obd_name);
+        obd->obd_css_blocking = 1;
+        spin_unlock(&obd->obd_dev_lock);
+
+        CWARN("%s: controlled server shutdown, start syncing with server\n",
+              obd->obd_name);
+
+        down_read(&obd->u.cli.cl_sem);
+        if (obd->u.cli.cl_import == NULL) {
+                CWARN("import already gone\n");
+                up_read(&obd->u.cli.cl_sem);
+                EXIT;
+                return;
+        }
+
+        imp = class_import_get(obd->u.cli.cl_import);
+        /*
+         * if import is not FULL, we can't proceed because server won't accept
+         * reconnect anyway. There's a small chance that our reconnect request
+         * arrives server before server start controlled shutdown and reply
+         * was heavily delayed, in that case we just accept the bad luck.
+         *
+         * but during following synching, import state still can be changed at
+         * any time, that's bad luck too. we just make sure don't wait forever.
+         *
+         * for controlled shutdown, we don't try too hard.
+         */
+        if (imp->imp_state != LUSTRE_IMP_FULL) {
+                CWARN("import state is %s, skip\n",
+                      ptlrpc_import_state_name(imp->imp_state));
+                up_read(&obd->u.cli.cl_sem);
+                GOTO(out_imp, rc = 1);
+        }
+
+        LASSERT(obd->obd_namespace);
+        ns = obd->obd_namespace;
+        ldlm_namespace_get(ns);
+        up_read(&obd->u.cli.cl_sem);
+
+        timeout = 40; /* XXX fix the timeout value */
+
+        /*
+         * we can't be abolutely sure the syncing is finished, as an extreme
+         * case: an rpc formed slowly, get sent after we cleaned up namespace,
+         * and then obtained a new lock or fill in replay queue thereafter.
+         * so we do the syncing twice to be more confident.
+         */
+        start_time = cfs_time_current_sec();
+        rc = do_css_sync(imp, ns, timeout);
+        if (rc == 0) {
+                timeout = cfs_time_current_sec() - start_time;
+                cfs_pause(HZ);
+                rc = do_css_sync(imp, ns, timeout);
+        }
+
+        ldlm_namespace_put(ns, 1);
+
+        /* notify server we've finished syncing
+         * XXX perhaps we should notify anyway??
+         */
+        if (rc == 0)
+                ptlrpc_disconnect_import(imp, 1);
+
+out_imp:
+        class_import_put(imp);
+        CWARN("%s: %s syncing with server\n",
+              obd->obd_name, rc ? "failed" : "successfully");
+        EXIT;
+}
+
 static int ldlm_callback_handler(struct ptlrpc_request *req)
 {
         struct ldlm_namespace *ns;
@@ -1660,6 +1824,9 @@ static int ldlm_callback_handler(struct 
         LASSERT(req->rq_export->exp_obd != NULL);
 
         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
+        case 12345678:
+                ldlm_handle_ctrl_shutdown(req->rq_export->exp_obd);
+                RETURN(0);
         case LDLM_BL_CALLBACK:
                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
                 break;
@@ -2431,6 +2598,7 @@ EXPORT_SYMBOL(target_handle_ping);
 EXPORT_SYMBOL(target_pack_pool_reply);
 EXPORT_SYMBOL(target_handle_disconnect);
 EXPORT_SYMBOL(target_handle_reply);
+EXPORT_SYMBOL(target_ctrl_shutdown);
 
 /* l_lock.c */
 EXPORT_SYMBOL(lock_res_and_lock);
Index: ldlm/ldlm_request.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ldlm/ldlm_request.c,v
retrieving revision 1.129.34.32.2.12
diff -u -p -r1.129.34.32.2.12 ldlm_request.c
--- ldlm/ldlm_request.c	16 Jul 2009 01:32:09 -0000	1.129.34.32.2.12
+++ ldlm/ldlm_request.c	25 Aug 2009 17:41:08 -0000
@@ -668,6 +668,8 @@ int ldlm_cli_enqueue(struct obd_export *
 
         LASSERT(exp != NULL);
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         /* If we're replaying this lock, just check some invariants.
          * If we're creating a new lock, get everything all setup nice. */
         if (is_replay) {
Index: llite/file.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/llite/file.c,v
retrieving revision 1.218.12.45.2.33
diff -u -p -r1.218.12.45.2.33 file.c
--- llite/file.c	21 Aug 2009 01:29:03 -0000	1.218.12.45.2.33
+++ llite/file.c	25 Aug 2009 17:41:09 -0000
@@ -183,6 +183,8 @@ int ll_mdc_close(struct obd_export *mdc_
         int rc = 0;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(mdc_exp->exp_obd);
+
         /* clear group lock, if present */
         if (unlikely(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
                 struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
Index: mdc/mdc_locks.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/mdc/mdc_locks.c,v
retrieving revision 1.45.2.16.2.10
diff -u -p -r1.45.2.16.2.10 mdc_locks.c
--- mdc/mdc_locks.c	20 Aug 2009 15:57:20 -0000	1.45.2.16.2.10
+++ mdc/mdc_locks.c	25 Aug 2009 17:41:09 -0000
@@ -577,6 +577,8 @@ int mdc_enqueue(struct obd_export *exp, 
         int rc;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(obddev);
+
         fid_build_reg_res_name((void *)&data->fid1, &res_id);
         LASSERTF(einfo->ei_type == LDLM_IBITS,"lock type %d\n", einfo->ei_type);
         if (it->it_op & (IT_UNLINK | IT_GETATTR | IT_READDIR))
@@ -643,6 +645,8 @@ int mdc_revalidate_lock(struct obd_expor
         ldlm_mode_t mode;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         fid_build_reg_res_name((struct lu_fid*)fid, &res_id);
         /* As not all attributes are kept under update lock, e.g. 
            owner/group/acls are under lookup lock, we need both 
@@ -813,6 +817,8 @@ int mdc_intent_lock(struct obd_export *e
                PFID(((void *)&op_data->fid1)),
                ldlm_it2str(it->it_op), it->it_flags);
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         lockh.cookie = 0;
         if (op_data->fid2.id &&
             (it->it_op == IT_LOOKUP || it->it_op == IT_GETATTR)) {
Index: mdc/mdc_reint.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/mdc/mdc_reint.c,v
retrieving revision 1.64.34.6.4.2
diff -u -p -r1.64.34.6.4.2 mdc_reint.c
--- mdc/mdc_reint.c	29 Jan 2009 08:34:55 -0000	1.64.34.6.4.2
+++ mdc/mdc_reint.c	25 Aug 2009 17:41:09 -0000
@@ -143,6 +143,8 @@ int mdc_setattr(struct obd_export *exp, 
 
         LASSERT(iattr != NULL);
 
+        OBD_CHECK_CSS_BLOCKING(obd);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 size[REQ_REC_OFF] = sizeof(struct mdt_rec_setattr);
                 size[REQ_REC_OFF + 1] = 0; /* capa */
@@ -209,6 +211,8 @@ int mdc_create(struct obd_export *exp, s
         int count;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(obd);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 size[REQ_REC_OFF] = sizeof(struct mdt_rec_create);
                 size[REQ_REC_OFF + 1] = 0; /* capa */
@@ -285,6 +289,8 @@ int mdc_unlink(struct obd_export *exp, s
         int offset = REQ_REC_OFF + 2;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(obd);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 size[REQ_REC_OFF] = sizeof(struct mdt_rec_unlink);
                 size[REQ_REC_OFF + 1] = 0 /* capa */;
@@ -338,6 +344,8 @@ int mdc_link(struct obd_export *exp, str
         int offset = REQ_REC_OFF + 2;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(obd);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 size[REQ_REC_OFF] = sizeof(struct mdt_rec_link);
                 size[REQ_REC_OFF + 1] = 0; /* capa */
@@ -389,6 +397,8 @@ int mdc_rename(struct obd_export *exp, s
         int offset = REQ_REC_OFF + 3;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(obd);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 size[REQ_REC_OFF] = sizeof(struct mdt_rec_rename);
                 size[REQ_REC_OFF + 1] = 0; /* capa */
Index: mdc/mdc_request.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/mdc/mdc_request.c,v
retrieving revision 1.235.2.18.2.15
diff -u -p -r1.235.2.18.2.15 mdc_request.c
--- mdc/mdc_request.c	28 Jul 2009 12:42:33 -0000	1.235.2.18.2.15
+++ mdc/mdc_request.c	25 Aug 2009 17:41:09 -0000
@@ -77,6 +77,8 @@ static int send_getstatus(struct obd_exp
         int rc;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION, MDS_GETSTATUS, 2, size,
                               NULL);
         if (!req)
@@ -187,6 +189,8 @@ int mdc_getattr(struct obd_export *exp, 
         int acl_size = 0, rc;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         /* XXX do we need to make another request here?  We just did a getattr
          *     to do the lookup in the first place.
          */
@@ -226,6 +230,8 @@ int mdc_getattr_name(struct obd_export *
         int nameoffset = REQ_REC_OFF + 1;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 size[REQ_REC_OFF + 1] = 0;
                 size[REQ_REC_OFF + 2] = namelen;
@@ -273,6 +279,8 @@ int mdc_xattr_common(struct obd_export *
         void *tmp;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(obddev);
+
         if (mdc_exp_is_2_0_server(exp)) {
                 bufcnt++;
                 offset++;
@@ -812,6 +820,13 @@ int mdc_close(struct obd_export *exp, st
         int bufcount = 2;
         ENTRY;
 
+        /* XXX
+         * OBD_CHECK_CSS_BLOCKING(obd);
+         *
+         * we can't check css blocking because lock flushing may reach here.
+         * so we have raise the blocking check to a higher level into llite.
+         */
+
         if (mdc_exp_is_2_0_server(exp)) {
                 reqsize[1] = sizeof(struct mdt_epoch);
                 reqsize[2] = sizeof(struct mdt_rec_create);
@@ -905,6 +920,8 @@ int mdc_done_writing(struct obd_export *
         int rc;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_DONE_WRITING, 2, size, NULL);
         if (req == NULL)
@@ -940,6 +957,8 @@ int mdc_readpage(struct obd_export *exp,
 
         CDEBUG(D_INODE, "inode: "LPU64"\n", fid->id);
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
+
         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_READPAGE, 2, size,
                               NULL);
         if (req == NULL)
@@ -1294,6 +1313,7 @@ int mdc_sync(struct obd_export *exp, str
         int rc;
         ENTRY;
 
+        OBD_CHECK_CSS_BLOCKING(exp->exp_obd);
 
         if (mdc_exp_is_2_0_server(exp))
                 bufcount = 3;
@@ -1614,6 +1634,32 @@ static int mdc_fid_fini(struct obd_expor
         RETURN(0);
 }
 
+static int mdc_convert_rpc(struct obd_device *obd,
+                           struct ptlrpc_request *req)
+{
+        ENTRY;
+
+        if (req->rq_fmt_20) {
+                DEBUG_REQ(D_ERROR, req, "do not support down-convertion");
+                RETURN(-ENOTSUPP);
+        }
+
+        if (!mdc_exp_is_2_0_server(obd->obd_self_export)) {
+                CERROR("server is not 2.0??\n");
+                RETURN(-ENOTSUPP);
+        }
+
+        if (!lustre_reqmsg_is_mds_open(req->rq_reqmsg)) {
+                DEBUG_REQ(D_ERROR, req, "do not support this up-convertion");
+                RETURN(-ENOTSUPP);
+        }
+
+        //XXX to do convertion
+        CWARN("faking doing convertion...\n");
+
+        RETURN(0);
+}
+
 struct obd_ops mdc_obd_ops = {
         .o_owner        = THIS_MODULE,
         .o_setup        = mdc_setup,
@@ -1635,6 +1681,7 @@ struct obd_ops mdc_obd_ops = {
         .o_llog_init    = mdc_llog_init,
         .o_llog_finish  = mdc_llog_finish,
         .o_process_config = mdc_process_config,
+        .o_convert_rpc  = mdc_convert_rpc,
 };
 
 int __init mdc_init(void)
Index: mds/mds_lov.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/mds/mds_lov.c,v
retrieving revision 1.90.18.18.4.13
diff -u -p -r1.90.18.18.4.13 mds_lov.c
--- mds/mds_lov.c	21 Aug 2009 01:29:04 -0000	1.90.18.18.4.13
+++ mds/mds_lov.c	25 Aug 2009 17:41:09 -0000
@@ -873,6 +873,12 @@ int mds_iocontrol(unsigned int cmd, stru
                 RETURN(rc);
         }
 
+        case OBD_IOC_CTRL_SHUTDOWN: {
+                if (1 /*XXX check proc entry*/)
+                        target_ctrl_shutdown(obd);
+                RETURN(0);
+        }
+
         case OBD_IOC_SET_READONLY: {
                 void *handle;
                 struct inode *inode = obd->u.obt.obt_sb->s_root->d_inode;
Index: obdclass/genops.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/obdclass/genops.c,v
retrieving revision 1.145.6.22.2.10
diff -u -p -r1.145.6.22.2.10 genops.c
--- obdclass/genops.c	2 Jul 2009 21:59:21 -0000	1.145.6.22.2.10
+++ obdclass/genops.c	25 Aug 2009 17:41:09 -0000
@@ -1579,3 +1579,18 @@ void obd_zombie_impexp_stop(void)
         liblustre_deregister_idle_callback(obd_zombi_impexp_idle_cb);
 #endif
 }
+
+int obd_wait_css(struct obd_device *obd, const char *func)
+{
+        struct l_wait_info lwi;
+        int                ret;
+
+        lwi = LWI_TIMEOUT_INTR(0, NULL, LWI_ON_SIGNAL_NOOP, NULL);
+        CWARN("%s: %s start blocking\n", obd->obd_name, func);
+        ret = l_wait_event(obd->obd_css_waitq,
+                           obd->obd_css_blocking == 0,
+                           &lwi);
+        CWARN("%s: %s finished blocking, ret %d\n", obd->obd_name, func, ret);
+        return ret;
+}
+EXPORT_SYMBOL(obd_wait_css);
Index: obdclass/lprocfs_status.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/obdclass/lprocfs_status.c,v
retrieving revision 1.43.34.29.2.20
diff -u -p -r1.43.34.29.2.20 lprocfs_status.c
--- obdclass/lprocfs_status.c	20 Aug 2009 15:53:50 -0000	1.43.34.29.2.20
+++ obdclass/lprocfs_status.c	25 Aug 2009 17:41:09 -0000
@@ -1441,6 +1441,7 @@ void lprocfs_init_ops_stats(int num_priv
         LPROCFS_OBD_OP_INIT(num_private_stats, stats, pool_del);
         LPROCFS_OBD_OP_INIT(num_private_stats, stats, getref);
         LPROCFS_OBD_OP_INIT(num_private_stats, stats, putref);
+        LPROCFS_OBD_OP_INIT(num_private_stats, stats, convert_rpc);
 }
 
 void lprocfs_init_ldlm_stats(struct lprocfs_stats *ldlm_stats)
Index: obdclass/obd_config.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/obdclass/obd_config.c,v
retrieving revision 1.42.6.12.2.11
diff -u -p -r1.42.6.12.2.11 obd_config.c
--- obdclass/obd_config.c	18 Jun 2009 05:39:33 -0000	1.42.6.12.2.11
+++ obdclass/obd_config.c	25 Aug 2009 17:41:09 -0000
@@ -229,6 +229,8 @@ int class_attach(struct lustre_cfg *lcfg
         init_mutex(&obd->obd_llog_cat_process);
         CFS_INIT_LIST_HEAD(&obd->obd_recovery_queue);
         CFS_INIT_LIST_HEAD(&obd->obd_delayed_reply_queue);
+        atomic_set(&obd->obd_css_nr_exp, 0);
+        cfs_waitq_init(&obd->obd_css_waitq);
 
         len = strlen(uuid);
         if (len >= sizeof(obd->obd_uuid)) {
@@ -460,7 +462,7 @@ int class_cleanup(struct obd_device *obd
                         case 'A':
                                 LCONSOLE_WARN("Failing over %s\n",
                                               obd->obd_name);
-                                obd->obd_fail = 1;
+
                                 obd->obd_no_transno = 1;
                                 obd->obd_no_recov = 1;
                                 /* Set the obd readonly if we can */
Index: obdclass/obd_mount.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/obdclass/obd_mount.c,v
retrieving revision 1.6.10.20.2.9
diff -u -p -r1.6.10.20.2.9 obd_mount.c
--- obdclass/obd_mount.c	22 Aug 2009 16:07:33 -0000	1.6.10.20.2.9
+++ obdclass/obd_mount.c	25 Aug 2009 17:41:10 -0000
@@ -1426,8 +1426,12 @@ static void server_put_super(struct supe
                 obd = class_name2obd(lsi->lsi_ldd->ldd_svname);
                 if (obd) {
                         CDEBUG(D_MOUNT, "stopping %s\n", obd->obd_name);
-                        if (lsi->lsi_flags & LSI_UMOUNT_FAILOVER)
+                        if (lsi->lsi_flags & LSI_UMOUNT_FAILOVER) {
+                                obd_iocontrol(OBD_IOC_CTRL_SHUTDOWN,
+                                              obd->obd_self_export,
+                                              0, NULL, NULL);
                                 obd->obd_fail = 1;
+                        }
                         /* We can't seem to give an error return code
                          * to .put_super, so we better make sure we clean up! */
                         obd->obd_force = 1;
Index: obdfilter/filter.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/obdfilter/filter.c,v
retrieving revision 1.247.2.45.2.39
diff -u -p -r1.247.2.45.2.39 filter.c
--- obdfilter/filter.c	21 Aug 2009 15:42:44 -0000	1.247.2.45.2.39
+++ obdfilter/filter.c	25 Aug 2009 17:41:10 -0000
@@ -3971,6 +3971,12 @@ int filter_iocontrol(unsigned int cmd, s
                 RETURN(rc);
         }
 
+        case OBD_IOC_CTRL_SHUTDOWN: {
+                if (1 /*XXX check proc entry*/)
+                        target_ctrl_shutdown(obd);
+                RETURN(0);
+        }
+
         case OBD_IOC_LLOG_CANCEL:
         case OBD_IOC_LLOG_REMOVE:
         case OBD_IOC_LLOG_INFO:
Index: ptlrpc/client.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ptlrpc/client.c,v
retrieving revision 1.171.2.36.2.12
diff -u -p -r1.171.2.36.2.12 client.c
--- ptlrpc/client.c	10 Jun 2009 18:26:36 -0000	1.171.2.36.2.12
+++ ptlrpc/client.c	25 Aug 2009 17:41:10 -0000
@@ -574,6 +574,8 @@ ptlrpc_prep_req_pool(struct obd_import *
         }
 
         lustre_msg_add_version(request->rq_reqmsg, version);
+        request->rq_fmt_20 = !!(imp->imp_connect_data.ocd_connect_flags &
+                                OBD_CONNECT_FID);
         request->rq_send_state = LUSTRE_IMP_FULL;
         request->rq_type = PTL_RPC_MSG_REQUEST;
         request->rq_import = class_import_get(imp);
@@ -2215,6 +2217,13 @@ restart:
         RETURN(rc);
 }
 
+int ptlrpc_req_need_conversion(struct ptlrpc_request *req)
+{
+        return (!!(req->rq_import->imp_connect_data.ocd_connect_flags &
+                   OBD_CONNECT_FID) !=
+                !!(req->rq_fmt_20));
+}
+
 struct ptlrpc_replay_async_args {
         int praa_old_state;
         int praa_old_status;
@@ -2294,10 +2303,20 @@ static int ptlrpc_replay_interpret(struc
 int ptlrpc_replay_req(struct ptlrpc_request *req)
 {
         struct ptlrpc_replay_async_args *aa;
+        int                              rc;
         ENTRY;
 
         LASSERT(req->rq_import->imp_state == LUSTRE_IMP_REPLAY);
 
+        if (ptlrpc_req_need_conversion(req)) {
+                CWARN("NEED CONVERT RPC!!!\n");
+                rc = obd_convert_rpc(req->rq_import->imp_obd, req);
+                if (rc) {
+                        ptlrpc_import_self_evict(req->rq_import);
+                        RETURN(rc);
+                }
+        }
+
         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
         aa = ptlrpc_req_async_args(req);
         memset(aa, 0, sizeof *aa);
Index: ptlrpc/import.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ptlrpc/import.c,v
retrieving revision 1.43.6.32.2.18
diff -u -p -r1.43.6.32.2.18 import.c
--- ptlrpc/import.c	23 Jul 2009 19:19:46 -0000	1.43.6.32.2.18
+++ ptlrpc/import.c	25 Aug 2009 17:41:10 -0000
@@ -396,6 +396,14 @@ void ptlrpc_fail_import(struct obd_impor
         EXIT;
 }
 
+void ptlrpc_import_self_evict(struct obd_import *imp)
+{               
+        CWARN("%s: evicting myself\n", imp->imp_obd->obd_name);
+        
+        IMPORT_SET_STATE(imp, LUSTRE_IMP_EVICTED);
+        ptlrpc_import_recovery_state_machine(imp); 
+}
+
 int ptlrpc_reconnect_import(struct obd_import *imp)
 {
 
@@ -940,6 +948,17 @@ static int ptlrpc_connect_interpret(stru
         }
 
 finish:
+        /* FIXME */
+        {
+                struct obd_connect_data *ocd;
+                ocd = lustre_swab_repbuf(request, REPLY_REC_OFF, sizeof(*ocd),
+                                         lustre_swab_connect);
+                if (ocd == NULL)
+                        LBUG();
+                imp->imp_connect_data = *ocd;
+        }
+        /* FIXME */
+
         rc = ptlrpc_import_recovery_state_machine(imp);
         if (rc != 0) {
                 if (rc == -ENOTCONN) {
@@ -1290,6 +1309,9 @@ int ptlrpc_import_recovery_state_machine
                 CDEBUG(D_HA, "replay requested by %s\n",
                        obd2cli_tgt(imp->imp_obd));
                 rc = ptlrpc_replay_next(imp, &inflight);
+                if (rc)
+                        GOTO(out, rc);
+
                 if (inflight == 0 &&
                     atomic_read(&imp->imp_replay_inflight) == 0) {
                         IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY_LOCKS);
@@ -1297,7 +1319,6 @@ int ptlrpc_import_recovery_state_machine
                         if (rc)
                                 GOTO(out, rc);
                 }
-                rc = 0;
         }
 
         if (imp->imp_state == LUSTRE_IMP_REPLAY_LOCKS) {
@@ -1336,6 +1357,14 @@ int ptlrpc_import_recovery_state_machine
         }
 
         if (imp->imp_state == LUSTRE_IMP_FULL) {
+                if (imp->imp_obd->obd_css_blocking) {
+                        LCONSOLE_INFO("%s: client unblocking requests after "
+                                      "server restart\n",
+                                      imp->imp_obd->obd_name);
+                        imp->imp_obd->obd_css_blocking = 0;
+                        cfs_waitq_broadcast(&imp->imp_obd->obd_css_waitq);
+                }
+
                 cfs_waitq_broadcast(&imp->imp_recovery_waitq);
                 ptlrpc_wake_delayed(imp);
         }
@@ -1426,7 +1455,16 @@ out:
                 IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_DISCON);
         else
                 IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_CLOSED);
-        memset(&imp->imp_remote_handle, 0, sizeof(imp->imp_remote_handle));
+
+        /*
+         * if it's called during CSS syncing, we don't cleanup the
+         * remote_handle, so next reconnect will look like a reconnect
+         * and lead to recovery.
+         */
+        if (imp->imp_obd->obd_css_blocking == 0)
+                memset(&imp->imp_remote_handle, 0,
+                       sizeof(imp->imp_remote_handle));
+
         /* Try all connections in the future - bz 12758 */
         imp->imp_last_recon = 0;
         spin_unlock(&imp->imp_lock);
Index: ptlrpc/pack_generic.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ptlrpc/pack_generic.c,v
retrieving revision 1.71.30.17.2.12
diff -u -p -r1.71.30.17.2.12 pack_generic.c
--- ptlrpc/pack_generic.c	24 Jul 2009 14:00:10 -0000	1.71.30.17.2.12
+++ ptlrpc/pack_generic.c	25 Aug 2009 17:41:10 -0000
@@ -1930,6 +1930,34 @@ void lustre_msg_set_cksum(struct lustre_
         }
 }
 
+int lustre_reqmsg_is_mds_open(struct lustre_msg *msg)
+{
+        struct ldlm_request *ldlm_req;
+        struct ldlm_intent  *ldlm_it;
+
+        /* XXX
+         * to confirm that all OPEN go through LDLM_ENQUEUE
+         */
+        if (lustre_msg_get_opc(msg) != LDLM_ENQUEUE)
+                return 0;
+
+        ldlm_req = lustre_msg_buf(msg, DLM_LOCKREQ_OFF, sizeof(*ldlm_req));
+        LASSERT(ldlm_req);
+        if ((ldlm_req->lock_flags & LDLM_FL_HAS_INTENT) == 0)
+                return 0;
+
+        ldlm_it = lustre_msg_buf(msg, DLM_INTENT_IT_OFF, sizeof(*ldlm_it));
+        LASSERT(ldlm_it);
+//XXX
+#define IT_OPEN (0x0001)
+        if ((ldlm_it->opc & IT_OPEN) == 0)
+                return 0;
+
+        CWARN("found a ENQUEUE-OPEN, t%lld\n", lustre_msg_get_transno(msg));
+        return 1;
+}
+EXPORT_SYMBOL(lustre_reqmsg_is_mds_open);
+
 
 /* byte flipping routines for all wire types declared in
  * lustre_idl.h implemented here.
Index: ptlrpc/ptlrpc_internal.h
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ptlrpc/ptlrpc_internal.h,v
retrieving revision 1.21.34.11.4.3
diff -u -p -r1.21.34.11.4.3 ptlrpc_internal.h
--- ptlrpc/ptlrpc_internal.h	6 Apr 2009 19:36:32 -0000	1.21.34.11.4.3
+++ ptlrpc/ptlrpc_internal.h	25 Aug 2009 17:41:10 -0000
@@ -49,8 +49,12 @@ extern int test_req_buffer_pressure;
 extern cfs_mem_cache_t *ptlrpc_cbdata_slab;
 
 /* client.c */
+int ptlrpc_req_need_conversion(struct ptlrpc_request *req);
 void ptlrpc_init_xid(void);
 
+/* import.c */
+void ptlrpc_import_self_evict(struct obd_import *imp);
+
 /* events.c */
 int ptlrpc_init_portals(void);
 void ptlrpc_exit_portals(void);
Index: ptlrpc/recover.c
===================================================================
RCS file: /cvsroot/cfs/lustre-core/ptlrpc/recover.c,v
retrieving revision 1.57.6.1.34.2
diff -u -p -r1.57.6.1.34.2 recover.c
--- ptlrpc/recover.c	24 Mar 2009 07:37:11 -0000	1.57.6.1.34.2
+++ ptlrpc/recover.c	25 Aug 2009 17:41:10 -0000
@@ -144,6 +144,7 @@ int ptlrpc_replay_next(struct obd_import
 int ptlrpc_resend(struct obd_import *imp)
 {
         struct ptlrpc_request *req, *next;
+        int                    evict = 0;
 
         ENTRY;
 
@@ -163,11 +164,24 @@ int ptlrpc_resend(struct obd_import *imp
                 LASSERTF((long)req > CFS_PAGE_SIZE && req != LP_POISON,
                          "req %p bad\n", req);
                 LASSERTF(req->rq_type != LI_POISON, "req %p freed\n", req);
+
+                if (ptlrpc_req_need_conversion(req)) {
+                        /*
+                         * with CSS this shouldn't happen, just evict myself.
+                         */
+                        DEBUG_REQ(D_ERROR, req, "still in resend queue");
+                        evict = 1;
+                        break;
+                }
+
                 if (!req->rq_no_resend)
                         ptlrpc_resend_req(req);
         }
         spin_unlock(&imp->imp_lock);
 
+        if (evict)
+                ptlrpc_import_self_evict(req->rq_import);
+
         RETURN(0);
 }
 
