lustre/ptlrpc/import.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/ptlrpc/import.c
  32  *
  33  * Author: Mike Shaver <shaver@clusterfs.com>
  34  */
  35
  36 #define DEBUG_SUBSYSTEM S_RPC
  37
  38 #include <linux/kthread.h>
  39 #include <linux/delay.h>
  40 #include <obd_support.h>
  41 #include <lustre_ha.h>
  42 #include <lustre_net.h>
  43 #include <lustre_import.h>
  44 #include <lustre_export.h>
  45 #include <obd.h>
  46 #include <obd_cksum.h>
  47 #include <obd_class.h>
  48
  49 #include "ptlrpc_internal.h"
  50
  51 struct ptlrpc_connect_async_args {
  52          __u64 pcaa_peer_committed;
  53         int pcaa_initial_connect;
  54 };
  55
  56 /**
  57  * Updates import \a imp current state to provided \a state value
  58  * Helper function.
  59  */
  60 static void import_set_state_nolock(struct obd_import *imp,
  61                                     enum lustre_imp_state state)
  62 {
  63         switch (state) {
  64         case LUSTRE_IMP_CLOSED:
  65         case LUSTRE_IMP_NEW:
  66         case LUSTRE_IMP_DISCON:
  67         case LUSTRE_IMP_CONNECTING:
  68                 break;
  69         case LUSTRE_IMP_REPLAY_WAIT:
  70                 imp->imp_replay_state = LUSTRE_IMP_REPLAY_LOCKS;
  71                 break;
  72         default:
  73                 imp->imp_replay_state = LUSTRE_IMP_REPLAY;
  74                 break;
  75         }
  76
  77         /* A CLOSED import should remain so. */
  78         if (imp->imp_state == LUSTRE_IMP_CLOSED)
  79                 return;
  80
  81         if (imp->imp_state != LUSTRE_IMP_NEW) {
  82                 CDEBUG(D_HA, "%p %s: changing import state from %s to %s\n",
  83                        imp, obd2cli_tgt(imp->imp_obd),
  84                        ptlrpc_import_state_name(imp->imp_state),
  85                        ptlrpc_import_state_name(state));
  86         }
  87
  88         imp->imp_state = state;
  89         imp->imp_state_hist[imp->imp_state_hist_idx].ish_state = state;
  90         imp->imp_state_hist[imp->imp_state_hist_idx].ish_time =
  91                 ktime_get_real_seconds();
  92         imp->imp_state_hist_idx = (imp->imp_state_hist_idx + 1) %
  93                 IMP_STATE_HIST_LEN;
  94 }
  95
  96 static void import_set_state(struct obd_import *imp,
  97                              enum lustre_imp_state new_state)
  98 {
  99         spin_lock(&imp->imp_lock);
 100         import_set_state_nolock(imp, new_state);
 101         spin_unlock(&imp->imp_lock);
 102 }
 103
 104 void ptlrpc_import_enter_resend(struct obd_import *imp)
 105 {
 106         import_set_state(imp, LUSTRE_IMP_RECOVER);
 107 }
 108 EXPORT_SYMBOL(ptlrpc_import_enter_resend);
 109
 110
 111 static int ptlrpc_connect_interpret(const struct lu_env *env,
 112                                     struct ptlrpc_request *request,
 113                                     void *args, int rc);
 114 int ptlrpc_import_recovery_state_machine(struct obd_import *imp);
 115
 116 /* Only this function is allowed to change the import state when it is
 117  * CLOSED. I would rather refcount the import and free it after
 118  * disconnection like we do with exports. To do that, the client_obd
 119  * will need to save the peer info somewhere other than in the import,
 120  * though. */
 121 int ptlrpc_init_import(struct obd_import *imp)
 122 {
 123         spin_lock(&imp->imp_lock);
 124
 125         imp->imp_generation++;
 126         imp->imp_state =  LUSTRE_IMP_NEW;
 127
 128         spin_unlock(&imp->imp_lock);
 129
 130         return 0;
 131 }
 132 EXPORT_SYMBOL(ptlrpc_init_import);
 133
 134 #define UUID_STR "_UUID"
 135 void deuuidify(char *uuid, const char *prefix, char **uuid_start, int *uuid_len)
 136 {
 137         *uuid_start = !prefix || strncmp(uuid, prefix, strlen(prefix))
 138                 ? uuid : uuid + strlen(prefix);
 139
 140         *uuid_len = strlen(*uuid_start);
 141
 142         if (*uuid_len < strlen(UUID_STR))
 143                 return;
 144
 145         if (!strncmp(*uuid_start + *uuid_len - strlen(UUID_STR),
 146                     UUID_STR, strlen(UUID_STR)))
 147                 *uuid_len -= strlen(UUID_STR);
 148 }
 149
 150 /* Must be called with imp_lock held! */
 151 static void ptlrpc_deactivate_import_nolock(struct obd_import *imp)
 152 {
 153         ENTRY;
 154
 155         assert_spin_locked(&imp->imp_lock);
 156         CDEBUG(D_HA, "setting import %s INVALID\n", obd2cli_tgt(imp->imp_obd));
 157         imp->imp_invalid = 1;
 158         imp->imp_generation++;
 159
 160         ptlrpc_abort_inflight(imp);
 161
 162         EXIT;
 163 }
 164
 165 /**
 166  * Returns true if import was FULL, false if import was already not
 167  * connected.
 168  * @imp - import to be disconnected
 169  * @conn_cnt - connection count (epoch) of the request that timed out
 170  *             and caused the disconnection.  In some cases, multiple
 171  *             inflight requests can fail to a single target (e.g. OST
 172  *             bulk requests) and if one has already caused a reconnection
 173  *             (increasing the import->conn_cnt) the older failure should
 174  *             not also cause a reconnection.  If zero it forces a reconnect.
 175  * @invalid - set import invalid flag
 176  */
 177 int ptlrpc_set_import_discon(struct obd_import *imp,
 178                              __u32 conn_cnt, bool invalid)
 179 {
 180         int rc = 0;
 181
 182         spin_lock(&imp->imp_lock);
 183
 184         if (imp->imp_state == LUSTRE_IMP_FULL &&
 185             (conn_cnt == 0 || conn_cnt == imp->imp_conn_cnt)) {
 186                 char *target_start;
 187                 int   target_len;
 188                 bool  inact = false;
 189
 190                 deuuidify(obd2cli_tgt(imp->imp_obd), NULL,
 191                           &target_start, &target_len);
 192
 193                 import_set_state_nolock(imp, LUSTRE_IMP_DISCON);
 194                 if (imp->imp_replayable) {
 195                         LCONSOLE_WARN("%s: Connection to %.*s (at %s) was "
 196                                "lost; in progress operations using this "
 197                                "service will wait for recovery to complete\n",
 198                                imp->imp_obd->obd_name, target_len, target_start,
 199                                obd_import_nid2str(imp));
 200                 } else {
 201                         LCONSOLE_ERROR_MSG(0x166, "%s: Connection to "
 202                                "%.*s (at %s) was lost; in progress "
 203                                "operations using this service will fail\n",
 204                                imp->imp_obd->obd_name, target_len, target_start,
 205                                obd_import_nid2str(imp));
 206                         if (invalid) {
 207                                 CDEBUG(D_HA, "import %s@%s for %s not "
 208                                        "replayable, auto-deactivating\n",
 209                                        obd2cli_tgt(imp->imp_obd),
 210                                        imp->imp_connection->c_remote_uuid.uuid,
 211                                        imp->imp_obd->obd_name);
 212                                 ptlrpc_deactivate_import_nolock(imp);
 213                                 inact = true;
 214                         }
 215                 }
 216                 spin_unlock(&imp->imp_lock);
 217
 218                 if (obd_dump_on_timeout)
 219                         libcfs_debug_dumplog();
 220
 221                 obd_import_event(imp->imp_obd, imp, IMP_EVENT_DISCON);
 222
 223                 if (inact)
 224                         obd_import_event(imp->imp_obd, imp, IMP_EVENT_INACTIVE);
 225                 rc = 1;
 226         } else {
 227                 spin_unlock(&imp->imp_lock);
 228                 CDEBUG(D_HA, "%s: import %p already %s (conn %u, was %u): %s\n",
 229                        imp->imp_client->cli_name, imp,
 230                        (imp->imp_state == LUSTRE_IMP_FULL &&
 231                         imp->imp_conn_cnt > conn_cnt) ?
 232                        "reconnected" : "not connected", imp->imp_conn_cnt,
 233                        conn_cnt, ptlrpc_import_state_name(imp->imp_state));
 234         }
 235
 236         return rc;
 237 }
 238
 239 /*
 240  * This acts as a barrier; all existing requests are rejected, and
 241  * no new requests will be accepted until the import is valid again.
 242  */
 243 void ptlrpc_deactivate_import(struct obd_import *imp)
 244 {
 245         spin_lock(&imp->imp_lock);
 246         ptlrpc_deactivate_import_nolock(imp);
 247         spin_unlock(&imp->imp_lock);
 248
 249         obd_import_event(imp->imp_obd, imp, IMP_EVENT_INACTIVE);
 250 }
 251 EXPORT_SYMBOL(ptlrpc_deactivate_import);
 252
 253 static time64_t ptlrpc_inflight_deadline(struct ptlrpc_request *req,
 254                                          time64_t now)
 255 {
 256         time64_t dl;
 257
 258         if (!(((req->rq_phase == RQ_PHASE_RPC) && !req->rq_waiting) ||
 259               (req->rq_phase == RQ_PHASE_BULK) ||
 260               (req->rq_phase == RQ_PHASE_NEW)))
 261                 return 0;
 262
 263         if (req->rq_timedout)
 264                 return 0;
 265
 266         if (req->rq_phase == RQ_PHASE_NEW)
 267                 dl = req->rq_sent;
 268         else
 269                 dl = req->rq_deadline;
 270
 271         if (dl <= now)
 272                 return 0;
 273
 274         return dl - now;
 275 }
 276
 277 static time64_t ptlrpc_inflight_timeout(struct obd_import *imp)
 278 {
 279         time64_t now = ktime_get_real_seconds();
 280         struct ptlrpc_request *req;
 281         time64_t timeout = 0;
 282
 283         spin_lock(&imp->imp_lock);
 284         list_for_each_entry(req, &imp->imp_sending_list, rq_list)
 285                 timeout = max(ptlrpc_inflight_deadline(req, now), timeout);
 286         spin_unlock(&imp->imp_lock);
 287         return timeout;
 288 }
 289
 290 /**
 291  * This function will invalidate the import, if necessary, then block
 292  * for all the RPC completions, and finally notify the obd to
 293  * invalidate its state (ie cancel locks, clear pending requests,
 294  * etc).
 295  */
 296 void ptlrpc_invalidate_import(struct obd_import *imp)
 297 {
 298         struct ptlrpc_request *req;
 299         time64_t timeout;
 300         int rc;
 301
 302         atomic_inc(&imp->imp_inval_count);
 303
 304         if (!imp->imp_invalid || imp->imp_obd->obd_no_recov)
 305                 ptlrpc_deactivate_import(imp);
 306
 307         if (OBD_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CONNECT_RACE)) {
 308                 OBD_RACE(OBD_FAIL_PTLRPC_CONNECT_RACE);
 309                 msleep(10 * MSEC_PER_SEC);
 310         }
 311         CFS_FAIL_TIMEOUT(OBD_FAIL_MGS_CONNECT_NET, 3 * cfs_fail_val / 2);
 312         LASSERT(imp->imp_invalid);
 313
 314         /* Wait forever until inflight == 0. We really can't do it another
 315          * way because in some cases we need to wait for very long reply
 316          * unlink. We can't do anything before that because there is really
 317          * no guarantee that some rdma transfer is not in progress right now.
 318          */
 319         do {
 320                 long timeout_jiffies;
 321
 322                 /* Calculate max timeout for waiting on rpcs to error
 323                  * out. Use obd_timeout if calculated value is smaller
 324                  * than it.
 325                  */
 326                 if (!OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK)) {
 327                         timeout = ptlrpc_inflight_timeout(imp);
 328                         timeout += div_u64(timeout, 3);
 329
 330                         if (timeout == 0)
 331                                 timeout = obd_timeout;
 332                 } else {
 333                         /* decrease the interval to increase race condition */
 334                         timeout = 1;
 335                 }
 336
 337                 CDEBUG(D_RPCTRACE, "Sleeping %llds for inflight to error out\n",
 338                        timeout);
 339
 340                 /* Wait for all requests to error out and call completion
 341                  * callbacks. Cap it at obd_timeout -- these should all
 342                  * have been locally cancelled by ptlrpc_abort_inflight.
 343                  */
 344                 timeout_jiffies = max_t(long, cfs_time_seconds(timeout), 1);
 345                 rc = wait_event_idle_timeout(
 346                                     imp->imp_recovery_waitq,
 347                                     (atomic_read(&imp->imp_inflight) == 0),
 348                                     timeout_jiffies);
 349
 350                 if (rc == 0) {
 351                         const char *cli_tgt = obd2cli_tgt(imp->imp_obd);
 352
 353                         CERROR("%s: timeout waiting for callback (%d != 0)\n",
 354                                cli_tgt, atomic_read(&imp->imp_inflight));
 355
 356                         spin_lock(&imp->imp_lock);
 357                         if (atomic_read(&imp->imp_inflight) == 0) {
 358                                 int count = atomic_read(&imp->imp_unregistering);
 359
 360                                 /* We know that "unregistering" rpcs only can
 361                                  * survive in sending or delaying lists (they
 362                                  * maybe waiting for long reply unlink in
 363                                  * sluggish nets). Let's check this. If there
 364                                  * is no inflight and unregistering != 0, this
 365                                  * is bug. */
 366                                 LASSERTF(count == 0, "Some RPCs are still "
 367                                          "unregistering: %d\n", count);
 368
 369                                 /* Let's save one loop as soon as inflight have
 370                                  * dropped to zero. No new inflights possible at
 371                                  * this point. */
 372                                 rc = 1;
 373                         } else {
 374                                 list_for_each_entry(req, &imp->imp_sending_list,
 375                                                     rq_list) {
 376                                         DEBUG_REQ(D_ERROR, req,
 377                                                   "still on sending list");
 378                                 }
 379                                 list_for_each_entry(req, &imp->imp_delayed_list,
 380                                                     rq_list) {
 381                                         DEBUG_REQ(D_ERROR, req,
 382                                                   "still on delayed list");
 383                                 }
 384
 385                                 CERROR("%s: Unregistering RPCs found (%d). "
 386                                        "Network is sluggish? Waiting for them "
 387                                        "to error out.\n", cli_tgt,
 388                                        atomic_read(&imp->imp_unregistering));
 389                         }
 390                         spin_unlock(&imp->imp_lock);
 391                 }
 392         } while (rc == 0);
 393
 394         /*
 395          * Let's additionally check that no new rpcs added to import in
 396          * "invalidate" state.
 397          */
 398         LASSERT(atomic_read(&imp->imp_inflight) == 0);
 399         obd_import_event(imp->imp_obd, imp, IMP_EVENT_INVALIDATE);
 400         sptlrpc_import_flush_all_ctx(imp);
 401
 402         atomic_dec(&imp->imp_inval_count);
 403         wake_up(&imp->imp_recovery_waitq);
 404 }
 405 EXPORT_SYMBOL(ptlrpc_invalidate_import);
 406
 407 /* unset imp_invalid */
 408 void ptlrpc_activate_import(struct obd_import *imp, bool set_state_full)
 409 {
 410         struct obd_device *obd = imp->imp_obd;
 411
 412         spin_lock(&imp->imp_lock);
 413         if (imp->imp_deactive != 0) {
 414                 LASSERT(imp->imp_state != LUSTRE_IMP_FULL);
 415                 if (imp->imp_state != LUSTRE_IMP_DISCON)
 416                         import_set_state_nolock(imp, LUSTRE_IMP_DISCON);
 417                 spin_unlock(&imp->imp_lock);
 418                 return;
 419         }
 420         if (set_state_full)
 421                 import_set_state_nolock(imp, LUSTRE_IMP_FULL);
 422
 423         imp->imp_invalid = 0;
 424
 425         spin_unlock(&imp->imp_lock);
 426         obd_import_event(obd, imp, IMP_EVENT_ACTIVE);
 427 }
 428 EXPORT_SYMBOL(ptlrpc_activate_import);
 429
 430 void ptlrpc_pinger_force(struct obd_import *imp)
 431 {
 432         CDEBUG(D_HA, "%s: waking up pinger s:%s\n", obd2cli_tgt(imp->imp_obd),
 433                ptlrpc_import_state_name(imp->imp_state));
 434
 435         spin_lock(&imp->imp_lock);
 436         imp->imp_force_verify = 1;
 437         spin_unlock(&imp->imp_lock);
 438
 439         if (imp->imp_state != LUSTRE_IMP_CONNECTING)
 440                 ptlrpc_pinger_wake_up();
 441 }
 442 EXPORT_SYMBOL(ptlrpc_pinger_force);
 443
 444 void ptlrpc_fail_import(struct obd_import *imp, __u32 conn_cnt)
 445 {
 446         ENTRY;
 447
 448         LASSERT(!imp->imp_dlm_fake);
 449
 450         if (ptlrpc_set_import_discon(imp, conn_cnt, true))
 451                 ptlrpc_pinger_force(imp);
 452
 453         EXIT;
 454 }
 455
 456 int ptlrpc_reconnect_import(struct obd_import *imp)
 457 {
 458 #ifdef CONFIG_LUSTRE_FS_PINGER
 459         long timeout_jiffies = cfs_time_seconds(obd_timeout);
 460         int rc;
 461
 462         ptlrpc_pinger_force(imp);
 463
 464         CDEBUG(D_HA, "%s: recovery started, waiting %u seconds\n",
 465                obd2cli_tgt(imp->imp_obd), obd_timeout);
 466
 467         rc = wait_event_idle_timeout(imp->imp_recovery_waitq,
 468                                      !ptlrpc_import_in_recovery(imp),
 469                                      timeout_jiffies);
 470         if (rc == 0)
 471                 rc = -ETIMEDOUT;
 472         else
 473                 rc = 0;
 474         CDEBUG(D_HA, "%s: recovery finished s:%s\n", obd2cli_tgt(imp->imp_obd),
 475                ptlrpc_import_state_name(imp->imp_state));
 476         return rc;
 477 #else
 478         ptlrpc_set_import_discon(imp, 0, false);
 479         /* Force a new connect attempt */
 480         ptlrpc_invalidate_import(imp);
 481         /* Do a fresh connect next time by zeroing the handle */
 482         ptlrpc_disconnect_import(imp, 1);
 483         /* Wait for all invalidate calls to finish */
 484         if (atomic_read(&imp->imp_inval_count) > 0) {
 485                 int rc;
 486
 487                 rc = l_wait_event_abortable(
 488                         imp->imp_recovery_waitq,
 489                         (atomic_read(&imp->imp_inval_count) == 0));
 490                 if (rc)
 491                         CERROR("Interrupted, inval=%d\n",
 492                                atomic_read(&imp->imp_inval_count));
 493         }
 494
 495         /* Allow reconnect attempts */
 496         imp->imp_obd->obd_no_recov = 0;
 497         /* Remove 'invalid' flag */
 498         ptlrpc_activate_import(imp, false);
 499         /* Attempt a new connect */
 500         ptlrpc_recover_import(imp, NULL, 0);
 501         return 0;
 502 #endif
 503 }
 504 EXPORT_SYMBOL(ptlrpc_reconnect_import);
 505
 506 /**
 507  * Connection on import \a imp is changed to another one (if more than one is
 508  * present). We typically chose connection that we have not tried to connect to
 509  * the longest
 510  */
 511 static int import_select_connection(struct obd_import *imp)
 512 {
 513         struct obd_import_conn *imp_conn = NULL, *conn;
 514         struct obd_export *dlmexp;
 515         char *target_start;
 516         int target_len, tried_all = 1;
 517         int rc = 0;
 518         ENTRY;
 519
 520         spin_lock(&imp->imp_lock);
 521
 522         if (list_empty(&imp->imp_conn_list)) {
 523                 rc = -EINVAL;
 524                 CERROR("%s: no connections available: rc = %d\n",
 525                        imp->imp_obd->obd_name, rc);
 526                 GOTO(out_unlock, rc);
 527         }
 528
 529         list_for_each_entry(conn, &imp->imp_conn_list, oic_item) {
 530                 CDEBUG(D_HA, "%s: connect to NID %s last attempt %lld\n",
 531                        imp->imp_obd->obd_name,
 532                        libcfs_nid2str(conn->oic_conn->c_peer.nid),
 533                        conn->oic_last_attempt);
 534
 535                 /* If we have not tried this connection since
 536                  * the last successful attempt, go with this one
 537                  */
 538                 if ((conn->oic_last_attempt == 0) ||
 539                     conn->oic_last_attempt <= imp->imp_last_success_conn) {
 540                         imp_conn = conn;
 541                         tried_all = 0;
 542                         break;
 543                 }
 544
 545                 /* If all of the connections have already been tried
 546                  * since the last successful connection; just choose the
 547                  * least recently used
 548                  */
 549                 if (!imp_conn)
 550                         imp_conn = conn;
 551                 else if (imp_conn->oic_last_attempt > conn->oic_last_attempt)
 552                         imp_conn = conn;
 553         }
 554
 555         /* if not found, simply choose the current one */
 556         if (!imp_conn || imp->imp_force_reconnect) {
 557                 LASSERT(imp->imp_conn_current);
 558                 imp_conn = imp->imp_conn_current;
 559                 tried_all = 0;
 560         }
 561         LASSERT(imp_conn->oic_conn);
 562
 563         /* If we've tried everything, and we're back to the beginning of the
 564          * list, increase our timeout and try again. It will be reset when
 565          * we do finally connect. (FIXME: really we should wait for all network
 566          * state associated with the last connection attempt to drain before
 567          * trying to reconnect on it.)
 568          */
 569         if (tried_all && (imp->imp_conn_list.next == &imp_conn->oic_item)) {
 570                 struct adaptive_timeout *at = &imp->imp_at.iat_net_latency;
 571
 572                 if (at_get(at) < CONNECTION_SWITCH_MAX) {
 573                         at_measured(at, at_get(at) + CONNECTION_SWITCH_INC);
 574                         if (at_get(at) > CONNECTION_SWITCH_MAX)
 575                                 at_reset(at, CONNECTION_SWITCH_MAX);
 576                 }
 577                 LASSERT(imp_conn->oic_last_attempt);
 578                 CDEBUG(D_HA,
 579                        "%s: tried all connections, increasing latency to %ds\n",
 580                        imp->imp_obd->obd_name, at_get(at));
 581         }
 582
 583         imp_conn->oic_last_attempt = ktime_get_seconds();
 584
 585         /* switch connection, don't mind if it's same as the current one */
 586         ptlrpc_connection_put(imp->imp_connection);
 587         imp->imp_connection = ptlrpc_connection_addref(imp_conn->oic_conn);
 588
 589         dlmexp = class_conn2export(&imp->imp_dlm_handle);
 590         if (!dlmexp)
 591                 GOTO(out_unlock, rc = -EINVAL);
 592         ptlrpc_connection_put(dlmexp->exp_connection);
 593         dlmexp->exp_connection = ptlrpc_connection_addref(imp_conn->oic_conn);
 594         class_export_put(dlmexp);
 595
 596         if (imp->imp_conn_current != imp_conn) {
 597                 if (imp->imp_conn_current) {
 598                         deuuidify(obd2cli_tgt(imp->imp_obd), NULL,
 599                                   &target_start, &target_len);
 600
 601                         CDEBUG(D_HA, "%s: Connection changing to"
 602                                " %.*s (at %s)\n",
 603                                imp->imp_obd->obd_name,
 604                                target_len, target_start,
 605                                libcfs_nid2str(imp_conn->oic_conn->c_peer.nid));
 606                 }
 607
 608                 imp->imp_conn_current = imp_conn;
 609         }
 610
 611         /* The below message is checked in conf-sanity.sh test_35[ab] */
 612         CDEBUG(D_HA, "%s: import %p using connection %s/%s\n",
 613                imp->imp_obd->obd_name, imp, imp_conn->oic_uuid.uuid,
 614                libcfs_nid2str(imp_conn->oic_conn->c_peer.nid));
 615
 616 out_unlock:
 617         spin_unlock(&imp->imp_lock);
 618         RETURN(rc);
 619 }
 620
 621 /*
 622  * must be called under imp_lock
 623  */
 624 static int ptlrpc_first_transno(struct obd_import *imp, __u64 *transno)
 625 {
 626         struct ptlrpc_request *req;
 627
 628         /* The requests in committed_list always have smaller transnos than
 629          * the requests in replay_list */
 630         if (!list_empty(&imp->imp_committed_list)) {
 631                 req = list_first_entry(&imp->imp_committed_list,
 632                                        struct ptlrpc_request, rq_replay_list);
 633                 *transno = req->rq_transno;
 634                 if (req->rq_transno == 0) {
 635                         DEBUG_REQ(D_ERROR, req,
 636                                   "zero transno in committed_list");
 637                         LBUG();
 638                 }
 639                 return 1;
 640         }
 641         if (!list_empty(&imp->imp_replay_list)) {
 642                 req = list_first_entry(&imp->imp_committed_list,
 643                                        struct ptlrpc_request, rq_replay_list);
 644                 *transno = req->rq_transno;
 645                 if (req->rq_transno == 0) {
 646                         DEBUG_REQ(D_ERROR, req, "zero transno in replay_list");
 647                         LBUG();
 648                 }
 649                 return 1;
 650         }
 651         return 0;
 652 }
 653
 654 int ptlrpc_connect_import(struct obd_import *imp)
 655 {
 656         spin_lock(&imp->imp_lock);
 657         return ptlrpc_connect_import_locked(imp);
 658 }
 659
 660 /**
 661  * Attempt to (re)connect import \a imp. This includes all preparations,
 662  * initializing CONNECT RPC request and passing it to ptlrpcd for
 663  * actual sending.
 664  *
 665  * Assumes imp->imp_lock is held, and releases it.
 666  *
 667  * Returns 0 on success or error code.
 668  */
 669 int ptlrpc_connect_import_locked(struct obd_import *imp)
 670 {
 671         struct obd_device *obd = imp->imp_obd;
 672         int initial_connect = 0;
 673         int set_transno = 0;
 674         __u64 committed_before_reconnect = 0;
 675         struct ptlrpc_request *request;
 676         struct obd_connect_data ocd;
 677         char *bufs[] = { NULL,
 678                          obd2cli_tgt(imp->imp_obd),
 679                          obd->obd_uuid.uuid,
 680                          (char *)&imp->imp_dlm_handle,
 681                          (char *)&ocd,
 682                          NULL };
 683         struct ptlrpc_connect_async_args *aa;
 684         int rc;
 685         ENTRY;
 686
 687         assert_spin_locked(&imp->imp_lock);
 688
 689         if (imp->imp_state == LUSTRE_IMP_CLOSED) {
 690                 spin_unlock(&imp->imp_lock);
 691                 CERROR("can't connect to a closed import\n");
 692                 RETURN(-EINVAL);
 693         } else if (imp->imp_state == LUSTRE_IMP_FULL) {
 694                 spin_unlock(&imp->imp_lock);
 695                 CERROR("already connected\n");
 696                 RETURN(0);
 697         } else if (imp->imp_state == LUSTRE_IMP_CONNECTING ||
 698                    imp->imp_state == LUSTRE_IMP_EVICTED ||
 699                    imp->imp_connected) {
 700                 spin_unlock(&imp->imp_lock);
 701                 CERROR("already connecting\n");
 702                 RETURN(-EALREADY);
 703         }
 704
 705         import_set_state_nolock(imp, LUSTRE_IMP_CONNECTING);
 706
 707         imp->imp_conn_cnt++;
 708         imp->imp_resend_replay = 0;
 709
 710         if (!lustre_handle_is_used(&imp->imp_remote_handle))
 711                 initial_connect = 1;
 712         else
 713                 committed_before_reconnect = imp->imp_peer_committed_transno;
 714
 715         set_transno = ptlrpc_first_transno(imp,
 716                                            &imp->imp_connect_data.ocd_transno);
 717         spin_unlock(&imp->imp_lock);
 718
 719         rc = import_select_connection(imp);
 720         if (rc)
 721                 GOTO(out, rc);
 722
 723         rc = sptlrpc_import_sec_adapt(imp, NULL, NULL);
 724         if (rc)
 725                 GOTO(out, rc);
 726
 727         /* Reset connect flags to the originally requested flags, in case
 728          * the server is updated on-the-fly we will get the new features. */
 729         ocd = imp->imp_connect_data;
 730         ocd.ocd_connect_flags = imp->imp_connect_flags_orig;
 731         ocd.ocd_connect_flags2 = imp->imp_connect_flags2_orig;
 732         /* Reset ocd_version each time so the server knows the exact versions */
 733         ocd.ocd_version = LUSTRE_VERSION_CODE;
 734         imp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;
 735         imp->imp_msghdr_flags &= ~MSGHDR_CKSUM_INCOMPAT18;
 736
 737         rc = obd_reconnect(NULL, imp->imp_obd->obd_self_export, obd,
 738                            &obd->obd_uuid, &ocd, NULL);
 739         if (rc)
 740                 GOTO(out, rc);
 741
 742         request = ptlrpc_request_alloc(imp, &RQF_MDS_CONNECT);
 743         if (request == NULL)
 744                 GOTO(out, rc = -ENOMEM);
 745
 746         /* get SELinux policy info if any */
 747         rc = sptlrpc_get_sepol(request);
 748         if (rc < 0) {
 749                 ptlrpc_request_free(request);
 750                 GOTO(out, rc);
 751         }
 752
 753         bufs[5] = request->rq_sepol;
 754
 755         req_capsule_set_size(&request->rq_pill, &RMF_SELINUX_POL, RCL_CLIENT,
 756                              strlen(request->rq_sepol) ?
 757                              strlen(request->rq_sepol) + 1 : 0);
 758
 759         rc = ptlrpc_request_bufs_pack(request, LUSTRE_OBD_VERSION,
 760                                       imp->imp_connect_op, bufs, NULL);
 761         if (rc) {
 762                 ptlrpc_request_free(request);
 763                 GOTO(out, rc);
 764         }
 765
 766         /* Report the rpc service time to the server so that it knows how long
 767          * to wait for clients to join recovery */
 768         lustre_msg_set_service_timeout(request->rq_reqmsg,
 769                                        at_timeout2est(request->rq_timeout));
 770
 771         /* The amount of time we give the server to process the connect req.
 772          * import_select_connection will increase the net latency on
 773          * repeated reconnect attempts to cover slow networks.
 774          * We override/ignore the server rpc completion estimate here,
 775          * which may be large if this is a reconnect attempt */
 776         request->rq_timeout = INITIAL_CONNECT_TIMEOUT;
 777         lustre_msg_set_timeout(request->rq_reqmsg, request->rq_timeout);
 778
 779         request->rq_no_resend = request->rq_no_delay = 1;
 780         request->rq_send_state = LUSTRE_IMP_CONNECTING;
 781         /* Allow a slightly larger reply for future growth compatibility */
 782         req_capsule_set_size(&request->rq_pill, &RMF_CONNECT_DATA, RCL_SERVER,
 783                              sizeof(struct obd_connect_data)+16*sizeof(__u64));
 784         ptlrpc_request_set_replen(request);
 785         request->rq_interpret_reply = ptlrpc_connect_interpret;
 786
 787         aa = ptlrpc_req_async_args(aa, request);
 788         memset(aa, 0, sizeof *aa);
 789
 790         aa->pcaa_peer_committed = committed_before_reconnect;
 791         aa->pcaa_initial_connect = initial_connect;
 792
 793         if (aa->pcaa_initial_connect) {
 794                 spin_lock(&imp->imp_lock);
 795                 imp->imp_replayable = 1;
 796                 spin_unlock(&imp->imp_lock);
 797                 lustre_msg_add_op_flags(request->rq_reqmsg,
 798                                         MSG_CONNECT_INITIAL);
 799         }
 800
 801         if (set_transno)
 802                 lustre_msg_add_op_flags(request->rq_reqmsg,
 803                                         MSG_CONNECT_TRANSNO);
 804
 805         DEBUG_REQ(D_RPCTRACE, request, "(re)connect request (timeout %d)",
 806                   request->rq_timeout);
 807         ptlrpcd_add_req(request);
 808         rc = 0;
 809 out:
 810         if (rc != 0)
 811                 import_set_state(imp, LUSTRE_IMP_DISCON);
 812
 813         RETURN(rc);
 814 }
 815 EXPORT_SYMBOL(ptlrpc_connect_import);
 816
 817 static void ptlrpc_maybe_ping_import_soon(struct obd_import *imp)
 818 {
 819         int force_verify;
 820
 821         spin_lock(&imp->imp_lock);
 822         force_verify = imp->imp_force_verify != 0;
 823         spin_unlock(&imp->imp_lock);
 824
 825         if (force_verify)
 826                 ptlrpc_pinger_wake_up();
 827 }
 828
 829 static int ptlrpc_busy_reconnect(int rc)
 830 {
 831         return (rc == -EBUSY) || (rc == -EAGAIN);
 832 }
 833
 834 static int ptlrpc_connect_set_flags(struct obd_import *imp,
 835                                     struct obd_connect_data *ocd,
 836                                     __u64 old_connect_flags,
 837                                     struct obd_export *exp, int init_connect)
 838 {
 839         static bool warned;
 840         struct client_obd *cli = &imp->imp_obd->u.cli;
 841
 842         spin_lock(&imp->imp_lock);
 843         list_move(&imp->imp_conn_current->oic_item,
 844                   &imp->imp_conn_list);
 845         imp->imp_last_success_conn =
 846                 imp->imp_conn_current->oic_last_attempt;
 847
 848         spin_unlock(&imp->imp_lock);
 849
 850         if (!warned && (ocd->ocd_connect_flags & OBD_CONNECT_VERSION) &&
 851             (ocd->ocd_version > LUSTRE_VERSION_CODE +
 852                                 LUSTRE_VERSION_OFFSET_WARN ||
 853              ocd->ocd_version < LUSTRE_VERSION_CODE -
 854                                 LUSTRE_VERSION_OFFSET_WARN)) {
 855                 /* Sigh, some compilers do not like #ifdef in the middle
 856                    of macro arguments */
 857                 const char *older = "older than client. "
 858                                     "Consider upgrading server";
 859                 const char *newer = "newer than client. "
 860                                     "Consider upgrading client";
 861
 862                 LCONSOLE_WARN("Server %s version (%d.%d.%d.%d) "
 863                               "is much %s (%s)\n",
 864                               obd2cli_tgt(imp->imp_obd),
 865                               OBD_OCD_VERSION_MAJOR(ocd->ocd_version),
 866                               OBD_OCD_VERSION_MINOR(ocd->ocd_version),
 867                               OBD_OCD_VERSION_PATCH(ocd->ocd_version),
 868                               OBD_OCD_VERSION_FIX(ocd->ocd_version),
 869                               ocd->ocd_version > LUSTRE_VERSION_CODE ?
 870                               newer : older, LUSTRE_VERSION_STRING);
 871                 warned = true;
 872         }
 873
 874         if (ocd->ocd_connect_flags & OBD_CONNECT_CKSUM) {
 875                 /* We sent to the server ocd_cksum_types with bits set
 876                  * for algorithms we understand. The server masked off
 877                  * the checksum types it doesn't support */
 878                 if ((ocd->ocd_cksum_types &
 879                      obd_cksum_types_supported_client()) == 0) {
 880                         LCONSOLE_ERROR("The negotiation of the checksum "
 881                                        "alogrithm to use with server %s "
 882                                        "failed (%x/%x)\n",
 883                                        obd2cli_tgt(imp->imp_obd),
 884                                        ocd->ocd_cksum_types,
 885                                        obd_cksum_types_supported_client());
 886                         return -EPROTO;
 887                 } else {
 888                         cli->cl_supp_cksum_types = ocd->ocd_cksum_types;
 889                 }
 890         } else {
 891                 /* The server does not support OBD_CONNECT_CKSUM.
 892                  * Enforce ADLER for backward compatibility*/
 893                 cli->cl_supp_cksum_types = OBD_CKSUM_ADLER;
 894         }
 895         cli->cl_cksum_type = obd_cksum_type_select(imp->imp_obd->obd_name,
 896                                                   cli->cl_supp_cksum_types,
 897                                                   cli->cl_preferred_cksum_type);
 898
 899         if (ocd->ocd_connect_flags & OBD_CONNECT_BRW_SIZE)
 900                 cli->cl_max_pages_per_rpc =
 901                         min(ocd->ocd_brw_size >> PAGE_SHIFT,
 902                             cli->cl_max_pages_per_rpc);
 903         else if (imp->imp_connect_op == MDS_CONNECT ||
 904                  imp->imp_connect_op == MGS_CONNECT)
 905                 cli->cl_max_pages_per_rpc = 1;
 906
 907         LASSERT((cli->cl_max_pages_per_rpc <= PTLRPC_MAX_BRW_PAGES) &&
 908                 (cli->cl_max_pages_per_rpc > 0));
 909
 910         client_adjust_max_dirty(cli);
 911
 912         /* Update client max modify RPCs in flight with value returned
 913          * by the server */
 914         if (ocd->ocd_connect_flags & OBD_CONNECT_MULTIMODRPCS)
 915                 cli->cl_max_mod_rpcs_in_flight = min(
 916                                         cli->cl_max_mod_rpcs_in_flight,
 917                                         ocd->ocd_maxmodrpcs);
 918         else
 919                 cli->cl_max_mod_rpcs_in_flight = 1;
 920
 921         /* Reset ns_connect_flags only for initial connect. It might be
 922          * changed in while using FS and if we reset it in reconnect
 923          * this leads to losing user settings done before such as
 924          * disable lru_resize, etc. */
 925         if (old_connect_flags != exp_connect_flags(exp) || init_connect) {
 926                 struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
 927                 __u64 changed_flags;
 928
 929                 changed_flags =
 930                         ns->ns_connect_flags ^ ns->ns_orig_connect_flags;
 931                 CDEBUG(D_HA, "%s: Resetting ns_connect_flags to server "
 932                              "flags: %#llx\n", imp->imp_obd->obd_name,
 933                              ocd->ocd_connect_flags);
 934                 ns->ns_connect_flags = (ns->ns_connect_flags & changed_flags) |
 935                                       (ocd->ocd_connect_flags & ~changed_flags);
 936                 ns->ns_orig_connect_flags = ocd->ocd_connect_flags;
 937         }
 938
 939         if (ocd->ocd_connect_flags & OBD_CONNECT_AT)
 940                 /* We need a per-message support flag, because
 941                  * a. we don't know if the incoming connect reply
 942                  *    supports AT or not (in reply_in_callback)
 943                  *    until we unpack it.
 944                  * b. failovered server means export and flags are gone
 945                  *    (in ptlrpc_send_reply).
 946                  *    Can only be set when we know AT is supported at
 947                  *    both ends */
 948                 imp->imp_msghdr_flags |= MSGHDR_AT_SUPPORT;
 949         else
 950                 imp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;
 951
 952         imp->imp_msghdr_flags |= MSGHDR_CKSUM_INCOMPAT18;
 953
 954         return 0;
 955 }
 956
 957 /**
 958  * Add all replay requests back to unreplied list before start replay,
 959  * so that we can make sure the known replied XID is always increased
 960  * only even if when replaying requests.
 961  */
 962 static void ptlrpc_prepare_replay(struct obd_import *imp)
 963 {
 964         struct ptlrpc_request *req;
 965
 966         if (imp->imp_state != LUSTRE_IMP_REPLAY ||
 967             imp->imp_resend_replay)
 968                 return;
 969
 970         /* If the server was restart during repaly, the requests may
 971          * have been added to the unreplied list in former replay. */
 972         spin_lock(&imp->imp_lock);
 973
 974         list_for_each_entry(req, &imp->imp_committed_list, rq_replay_list) {
 975                 if (list_empty(&req->rq_unreplied_list))
 976                         ptlrpc_add_unreplied(req);
 977         }
 978
 979         list_for_each_entry(req, &imp->imp_replay_list, rq_replay_list) {
 980                 if (list_empty(&req->rq_unreplied_list))
 981                         ptlrpc_add_unreplied(req);
 982         }
 983
 984         imp->imp_known_replied_xid = ptlrpc_known_replied_xid(imp);
 985         spin_unlock(&imp->imp_lock);
 986 }
 987
 988 /**
 989  * interpret_reply callback for connect RPCs.
 990  * Looks into returned status of connect operation and decides
 991  * what to do with the import - i.e enter recovery, promote it to
 992  * full state for normal operations of disconnect it due to an error.
 993  */
 994 static int ptlrpc_connect_interpret(const struct lu_env *env,
 995                                     struct ptlrpc_request *request,
 996                                     void *data, int rc)
 997 {
 998         struct ptlrpc_connect_async_args *aa = data;
 999         struct obd_import *imp = request->rq_import;
1000         struct lustre_handle old_hdl;
1001         __u64 old_connect_flags;
1002         timeout_t service_timeout;
1003         int msg_flags;
1004         struct obd_connect_data *ocd;
1005         struct obd_export *exp = NULL;
1006         int ret;
1007         ENTRY;
1008
1009         spin_lock(&imp->imp_lock);
1010         if (imp->imp_state == LUSTRE_IMP_CLOSED) {
1011                 imp->imp_connect_tried = 1;
1012                 spin_unlock(&imp->imp_lock);
1013                 RETURN(0);
1014         }
1015
1016         imp->imp_connect_error = rc;
1017         if (rc) {
1018                 struct ptlrpc_request *free_req;
1019                 struct ptlrpc_request *tmp;
1020
1021                 /* abort all delayed requests initiated connection */
1022                 list_for_each_entry_safe(free_req, tmp, &imp->imp_delayed_list,
1023                                          rq_list) {
1024                         spin_lock(&free_req->rq_lock);
1025                         if (free_req->rq_no_resend) {
1026                                 free_req->rq_err = 1;
1027                                 free_req->rq_status = -EIO;
1028                                 ptlrpc_client_wake_req(free_req);
1029                         }
1030                         spin_unlock(&free_req->rq_lock);
1031                 }
1032
1033                 /* if this reconnect to busy export - not need select new target
1034                  * for connecting*/
1035                 imp->imp_force_reconnect = ptlrpc_busy_reconnect(rc);
1036                 spin_unlock(&imp->imp_lock);
1037                 GOTO(out, rc);
1038         }
1039
1040         /* LU-7558: indicate that we are interpretting connect reply,
1041          * pltrpc_connect_import() will not try to reconnect until
1042          * interpret will finish. */
1043         imp->imp_connected = 1;
1044         spin_unlock(&imp->imp_lock);
1045
1046         LASSERT(imp->imp_conn_current);
1047
1048         msg_flags = lustre_msg_get_op_flags(request->rq_repmsg);
1049
1050         ret = req_capsule_get_size(&request->rq_pill, &RMF_CONNECT_DATA,
1051                                    RCL_SERVER);
1052         /* server replied obd_connect_data is always bigger */
1053         ocd = req_capsule_server_sized_get(&request->rq_pill,
1054                                            &RMF_CONNECT_DATA, ret);
1055
1056         if (ocd == NULL) {
1057                 CERROR("%s: no connect data from server\n",
1058                        imp->imp_obd->obd_name);
1059                 rc = -EPROTO;
1060                 GOTO(out, rc);
1061         }
1062
1063         spin_lock(&imp->imp_lock);
1064
1065         /* All imports are pingable */
1066         imp->imp_pingable = 1;
1067         imp->imp_force_reconnect = 0;
1068         imp->imp_force_verify = 0;
1069
1070         imp->imp_connect_data = *ocd;
1071
1072         CDEBUG(D_HA, "%s: connect to target with instance %u\n",
1073                imp->imp_obd->obd_name, ocd->ocd_instance);
1074         exp = class_conn2export(&imp->imp_dlm_handle);
1075
1076         spin_unlock(&imp->imp_lock);
1077
1078         if (!exp) {
1079                 /* This could happen if export is cleaned during the
1080                    connect attempt */
1081                 CERROR("%s: missing export after connect\n",
1082                        imp->imp_obd->obd_name);
1083                 GOTO(out, rc = -ENODEV);
1084         }
1085
1086         /* check that server granted subset of flags we asked for. */
1087         if ((ocd->ocd_connect_flags & imp->imp_connect_flags_orig) !=
1088             ocd->ocd_connect_flags) {
1089                 CERROR("%s: Server didn't grant requested subset of flags: "
1090                        "asked=%#llx granted=%#llx\n",
1091                        imp->imp_obd->obd_name, imp->imp_connect_flags_orig,
1092                        ocd->ocd_connect_flags);
1093                 GOTO(out, rc = -EPROTO);
1094         }
1095
1096         if ((ocd->ocd_connect_flags2 & imp->imp_connect_flags2_orig) !=
1097             ocd->ocd_connect_flags2) {
1098                 CERROR("%s: Server didn't grant requested subset of flags2: "
1099                        "asked=%#llx granted=%#llx\n",
1100                        imp->imp_obd->obd_name, imp->imp_connect_flags2_orig,
1101                        ocd->ocd_connect_flags2);
1102                 GOTO(out, rc = -EPROTO);
1103         }
1104
1105         if (!(imp->imp_connect_flags_orig & OBD_CONNECT_LIGHTWEIGHT) &&
1106             (imp->imp_connect_flags_orig & OBD_CONNECT_MDS_MDS) &&
1107             (imp->imp_connect_flags_orig & OBD_CONNECT_FID) &&
1108             (ocd->ocd_connect_flags & OBD_CONNECT_VERSION)) {
1109                 __u32 major = OBD_OCD_VERSION_MAJOR(ocd->ocd_version);
1110                 __u32 minor = OBD_OCD_VERSION_MINOR(ocd->ocd_version);
1111                 __u32 patch = OBD_OCD_VERSION_PATCH(ocd->ocd_version);
1112
1113                 /* We do not support the MDT-MDT interoperations with
1114                  * different version MDT because of protocol changes. */
1115                 if (unlikely(major != LUSTRE_MAJOR ||
1116                              minor != LUSTRE_MINOR ||
1117                              abs(patch - LUSTRE_PATCH) > 3)) {
1118                         LCONSOLE_WARN("%s: import %p (%u.%u.%u.%u) tried the "
1119                                       "connection to different version MDT "
1120                                       "(%d.%d.%d.%d) %s\n",
1121                                       imp->imp_obd->obd_name, imp, LUSTRE_MAJOR,
1122                                       LUSTRE_MINOR, LUSTRE_PATCH, LUSTRE_FIX,
1123                                       major, minor, patch,
1124                                       OBD_OCD_VERSION_FIX(ocd->ocd_version),
1125                                       imp->imp_connection->c_remote_uuid.uuid);
1126
1127                         GOTO(out, rc = -EPROTO);
1128                 }
1129         }
1130
1131         old_connect_flags = exp_connect_flags(exp);
1132         exp->exp_connect_data = *ocd;
1133         imp->imp_obd->obd_self_export->exp_connect_data = *ocd;
1134
1135         /* The net statistics after (re-)connect is not valid anymore,
1136          * because may reflect other routing, etc.
1137          */
1138         service_timeout = lustre_msg_get_service_timeout(request->rq_repmsg);
1139         at_reinit(&imp->imp_at.iat_net_latency, 0, 0);
1140         ptlrpc_at_adj_net_latency(request, service_timeout);
1141
1142         /* Import flags should be updated before waking import at FULL state */
1143         rc = ptlrpc_connect_set_flags(imp, ocd, old_connect_flags, exp,
1144                                       aa->pcaa_initial_connect);
1145         class_export_put(exp);
1146         exp = NULL;
1147
1148         if (rc != 0)
1149                 GOTO(out, rc);
1150
1151         obd_import_event(imp->imp_obd, imp, IMP_EVENT_OCD);
1152
1153         if (aa->pcaa_initial_connect) {
1154                 spin_lock(&imp->imp_lock);
1155                 if (msg_flags & MSG_CONNECT_REPLAYABLE) {
1156                         imp->imp_replayable = 1;
1157                         CDEBUG(D_HA, "connected to replayable target: %s\n",
1158                                obd2cli_tgt(imp->imp_obd));
1159                 } else {
1160                         imp->imp_replayable = 0;
1161                 }
1162
1163                 /* if applies, adjust the imp->imp_msg_magic here
1164                  * according to reply flags
1165                  */
1166
1167                 imp->imp_remote_handle =
1168                         *lustre_msg_get_handle(request->rq_repmsg);
1169
1170                 /* Initial connects are allowed for clients with non-random
1171                  * uuids when servers are in recovery.  Simply signal the
1172                  * servers replay is complete and wait in REPLAY_WAIT.
1173                  */
1174                 if (msg_flags & MSG_CONNECT_RECOVERING) {
1175                         CDEBUG(D_HA, "connect to %s during recovery\n",
1176                                obd2cli_tgt(imp->imp_obd));
1177                         import_set_state_nolock(imp, LUSTRE_IMP_REPLAY_LOCKS);
1178                         spin_unlock(&imp->imp_lock);
1179                 } else {
1180                         spin_unlock(&imp->imp_lock);
1181                         ptlrpc_activate_import(imp, true);
1182                 }
1183
1184                 GOTO(finish, rc = 0);
1185         }
1186
1187         /* Determine what recovery state to move the import to. */
1188         if (MSG_CONNECT_RECONNECT & msg_flags) {
1189                 memset(&old_hdl, 0, sizeof(old_hdl));
1190                 if (!memcmp(&old_hdl, lustre_msg_get_handle(request->rq_repmsg),
1191                             sizeof(old_hdl))) {
1192                         LCONSOLE_WARN("Reconnect to %s (at @%s) failed due "
1193                                       "bad handle %#llx\n",
1194                                       obd2cli_tgt(imp->imp_obd),
1195                                       imp->imp_connection->c_remote_uuid.uuid,
1196                                       imp->imp_dlm_handle.cookie);
1197                         GOTO(out, rc = -ENOTCONN);
1198                 }
1199
1200                 if (memcmp(&imp->imp_remote_handle,
1201                            lustre_msg_get_handle(request->rq_repmsg),
1202                            sizeof(imp->imp_remote_handle))) {
1203                         int level = msg_flags & MSG_CONNECT_RECOVERING ?
1204                                 D_HA : D_WARNING;
1205
1206                         /* Bug 16611/14775: if server handle have changed,
1207                          * that means some sort of disconnection happened.
1208                          * If the server is not in recovery, that also means it
1209                          * already erased all of our state because of previous
1210                          * eviction. If it is in recovery - we are safe to
1211                          * participate since we can reestablish all of our state
1212                          * with server again
1213                          */
1214                         if ((MSG_CONNECT_RECOVERING & msg_flags)) {
1215                                 CDEBUG_LIMIT(level,
1216                                        "%s@%s changed server handle from "
1217                                        "%#llx to %#llx"
1218                                        " but is still in recovery\n",
1219                                        obd2cli_tgt(imp->imp_obd),
1220                                        imp->imp_connection->c_remote_uuid.uuid,
1221                                        imp->imp_remote_handle.cookie,
1222                                        lustre_msg_get_handle(
1223                                                request->rq_repmsg)->cookie);
1224                         } else {
1225                                 LCONSOLE_WARN("Evicted from %s (at %s) "
1226                                               "after server handle changed from "
1227                                               "%#llx to %#llx\n",
1228                                               obd2cli_tgt(imp->imp_obd),
1229                                               imp->imp_connection->
1230                                               c_remote_uuid.uuid,
1231                                               imp->imp_remote_handle.cookie,
1232                                               lustre_msg_get_handle(
1233                                                       request->rq_repmsg)->cookie);
1234                         }
1235
1236                         imp->imp_remote_handle =
1237                                 *lustre_msg_get_handle(request->rq_repmsg);
1238
1239                         if (!(MSG_CONNECT_RECOVERING & msg_flags)) {
1240                                 import_set_state(imp, LUSTRE_IMP_EVICTED);
1241                                 GOTO(finish, rc = 0);
1242                         }
1243                 } else {
1244                         CDEBUG(D_HA, "reconnected to %s@%s after partition\n",
1245                                obd2cli_tgt(imp->imp_obd),
1246                                imp->imp_connection->c_remote_uuid.uuid);
1247                 }
1248
1249                 if (imp->imp_invalid) {
1250                         CDEBUG(D_HA, "%s: reconnected but import is invalid; "
1251                                "marking evicted\n", imp->imp_obd->obd_name);
1252                         import_set_state(imp, LUSTRE_IMP_EVICTED);
1253                 } else if (MSG_CONNECT_RECOVERING & msg_flags) {
1254                         CDEBUG(D_HA, "%s: reconnected to %s during replay\n",
1255                                imp->imp_obd->obd_name,
1256                                obd2cli_tgt(imp->imp_obd));
1257
1258                         spin_lock(&imp->imp_lock);
1259                         imp->imp_resend_replay = 1;
1260                         spin_unlock(&imp->imp_lock);
1261
1262                         import_set_state(imp, imp->imp_replay_state);
1263                 } else {
1264                         import_set_state(imp, LUSTRE_IMP_RECOVER);
1265                 }
1266         } else if ((MSG_CONNECT_RECOVERING & msg_flags) && !imp->imp_invalid) {
1267                 LASSERT(imp->imp_replayable);
1268                 imp->imp_remote_handle =
1269                         *lustre_msg_get_handle(request->rq_repmsg);
1270                 imp->imp_last_replay_transno = 0;
1271                 imp->imp_replay_cursor = &imp->imp_committed_list;
1272                 import_set_state(imp, LUSTRE_IMP_REPLAY);
1273         } else if ((ocd->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT) != 0 &&
1274                    !imp->imp_invalid) {
1275
1276                 obd_import_event(imp->imp_obd, imp, IMP_EVENT_INVALIDATE);
1277                 /* The below message is checked in recovery-small.sh test_106 */
1278                 DEBUG_REQ(D_HA, request, "%s: lwp recover",
1279                           imp->imp_obd->obd_name);
1280                 imp->imp_remote_handle =
1281                         *lustre_msg_get_handle(request->rq_repmsg);
1282                 import_set_state(imp, LUSTRE_IMP_RECOVER);
1283         } else {
1284                 DEBUG_REQ(D_HA, request,
1285                           "%s: evicting (reconnect/recover flags not set: %x)",
1286                           imp->imp_obd->obd_name, msg_flags);
1287                 imp->imp_remote_handle =
1288                         *lustre_msg_get_handle(request->rq_repmsg);
1289                 import_set_state(imp, LUSTRE_IMP_EVICTED);
1290         }
1291
1292         /* Sanity checks for a reconnected import. */
1293         if (!(imp->imp_replayable) != !(msg_flags & MSG_CONNECT_REPLAYABLE))
1294                 CERROR("imp_replayable flag does not match server after reconnect. We should LBUG right here.\n");
1295
1296         if (lustre_msg_get_last_committed(request->rq_repmsg) > 0 &&
1297             lustre_msg_get_last_committed(request->rq_repmsg) <
1298             aa->pcaa_peer_committed) {
1299                 static bool printed;
1300
1301                 /* The below message is checked in recovery-small.sh test_54 */
1302                 CERROR("%s: went back in time (transno %lld was previously committed, server now claims %lld)!\n",
1303                        obd2cli_tgt(imp->imp_obd), aa->pcaa_peer_committed,
1304                        lustre_msg_get_last_committed(request->rq_repmsg));
1305                 if (!printed) {
1306                         CERROR("For further information, see http://doc.lustre.org/lustre_manual.xhtml#went_back_in_time\n");
1307                         printed = true;
1308                 }
1309         }
1310
1311 finish:
1312         ptlrpc_prepare_replay(imp);
1313         rc = ptlrpc_import_recovery_state_machine(imp);
1314         if (rc == -ENOTCONN) {
1315                 CDEBUG(D_HA,
1316                        "evicted/aborted by %s@%s during recovery; invalidating and reconnecting\n",
1317                        obd2cli_tgt(imp->imp_obd),
1318                        imp->imp_connection->c_remote_uuid.uuid);
1319                 ptlrpc_connect_import(imp);
1320                 spin_lock(&imp->imp_lock);
1321                 imp->imp_connected = 0;
1322                 imp->imp_connect_tried = 1;
1323                 spin_unlock(&imp->imp_lock);
1324                 RETURN(0);
1325         }
1326
1327 out:
1328         if (exp != NULL)
1329                 class_export_put(exp);
1330
1331         spin_lock(&imp->imp_lock);
1332         imp->imp_connected = 0;
1333         imp->imp_connect_tried = 1;
1334
1335         if (rc != 0) {
1336                 bool inact = false;
1337                 time64_t now = ktime_get_seconds();
1338                 time64_t next_connect;
1339
1340                 import_set_state_nolock(imp, LUSTRE_IMP_DISCON);
1341                 if (rc == -EACCES) {
1342                         /*
1343                          * Give up trying to reconnect
1344                          * EACCES means client has no permission for connection
1345                          */
1346                         imp->imp_obd->obd_no_recov = 1;
1347                         ptlrpc_deactivate_import_nolock(imp);
1348                         inact = true;
1349                 } else if (rc == -EPROTO) {
1350                         struct obd_connect_data *ocd;
1351
1352                         /* reply message might not be ready */
1353                         if (request->rq_repmsg == NULL) {
1354                                 spin_unlock(&imp->imp_lock);
1355                                 RETURN(-EPROTO);
1356                         }
1357
1358                         ocd = req_capsule_server_get(&request->rq_pill,
1359                                                      &RMF_CONNECT_DATA);
1360                         /* Servers are not supposed to refuse connections from
1361                          * clients based on version, only connection feature
1362                          * flags.  We should never see this from llite, but it
1363                          * may be useful for debugging in the future. */
1364                         if (ocd &&
1365                             (ocd->ocd_connect_flags & OBD_CONNECT_VERSION) &&
1366                             (ocd->ocd_version != LUSTRE_VERSION_CODE)) {
1367                                 LCONSOLE_ERROR_MSG(0x16a, "Server %s version "
1368                                                    "(%d.%d.%d.%d)"
1369                                                    " refused connection from this client "
1370                                                    "with an incompatible version (%s).  "
1371                                                    "Client must be recompiled\n",
1372                                                    obd2cli_tgt(imp->imp_obd),
1373                                                    OBD_OCD_VERSION_MAJOR(ocd->ocd_version),
1374                                                    OBD_OCD_VERSION_MINOR(ocd->ocd_version),
1375                                                    OBD_OCD_VERSION_PATCH(ocd->ocd_version),
1376                                                    OBD_OCD_VERSION_FIX(ocd->ocd_version),
1377                                                    LUSTRE_VERSION_STRING);
1378                                 ptlrpc_deactivate_import_nolock(imp);
1379                                 import_set_state_nolock(imp, LUSTRE_IMP_CLOSED);
1380                                 inact = true;
1381                         }
1382                 } else if (rc == -ENODEV || rc == -ETIMEDOUT) {
1383                         /* ENODEV means there is no service, force reconnection
1384                          * to a pair if attempt happen ptlrpc_next_reconnect
1385                          * before now. ETIMEDOUT could be set during network
1386                          * error and do not guarantee request deadline happened.
1387                          */
1388                         struct obd_import_conn *conn;
1389                         time64_t reconnect_time;
1390
1391                         /* Same as ptlrpc_next_reconnect, but in past */
1392                         reconnect_time = now - INITIAL_CONNECT_TIMEOUT;
1393                         list_for_each_entry(conn, &imp->imp_conn_list,
1394                                             oic_item) {
1395                                 if (conn->oic_last_attempt <= reconnect_time) {
1396                                         imp->imp_force_verify = 1;
1397                                         break;
1398                                 }
1399                         }
1400                 }
1401
1402                 next_connect = imp->imp_conn_current->oic_last_attempt +
1403                                (request->rq_deadline - request->rq_sent);
1404                 spin_unlock(&imp->imp_lock);
1405
1406                 if (inact)
1407                         obd_import_event(imp->imp_obd, imp, IMP_EVENT_INACTIVE);
1408
1409                 if (rc == -EPROTO)
1410                         RETURN(rc);
1411
1412                 /* adjust imp_next_ping to request deadline + 1 and reschedule
1413                  * a pinger if import lost processing during CONNECTING or far
1414                  * away from request deadline. It could happen when connection
1415                  * was initiated outside of pinger, like
1416                  * ptlrpc_set_import_discon().
1417                  */
1418                 if (!imp->imp_force_verify && (imp->imp_next_ping <= now ||
1419                     imp->imp_next_ping > next_connect)) {
1420                         imp->imp_next_ping = max(now, next_connect) + 1;
1421                         ptlrpc_pinger_wake_up();
1422                 }
1423
1424                 ptlrpc_maybe_ping_import_soon(imp);
1425
1426                 CDEBUG(D_HA, "recovery of %s on %s failed (%d)\n",
1427                        obd2cli_tgt(imp->imp_obd),
1428                        (char *)imp->imp_connection->c_remote_uuid.uuid, rc);
1429         } else {
1430                 spin_unlock(&imp->imp_lock);
1431         }
1432
1433         wake_up(&imp->imp_recovery_waitq);
1434         RETURN(rc);
1435 }
1436
1437 /**
1438  * interpret callback for "completed replay" RPCs.
1439  * \see signal_completed_replay
1440  */
1441 static int completed_replay_interpret(const struct lu_env *env,
1442                                       struct ptlrpc_request *req,
1443                                       void *args, int rc)
1444 {
1445         ENTRY;
1446         atomic_dec(&req->rq_import->imp_replay_inflight);
1447         if (req->rq_status == 0 && !req->rq_import->imp_vbr_failed) {
1448                 ptlrpc_import_recovery_state_machine(req->rq_import);
1449         } else {
1450                 if (req->rq_import->imp_vbr_failed) {
1451                         CDEBUG(D_WARNING,
1452                                "%s: version recovery fails, reconnecting\n",
1453                                req->rq_import->imp_obd->obd_name);
1454                 } else {
1455                         CDEBUG(D_HA, "%s: LAST_REPLAY message error: %d, "
1456                                      "reconnecting\n",
1457                                req->rq_import->imp_obd->obd_name,
1458                                req->rq_status);
1459                 }
1460                 ptlrpc_connect_import(req->rq_import);
1461         }
1462
1463         RETURN(0);
1464 }
1465
1466 /**
1467  * Let server know that we have no requests to replay anymore.
1468  * Achieved by just sending a PING request
1469  */
1470 static int signal_completed_replay(struct obd_import *imp)
1471 {
1472         struct ptlrpc_request *req;
1473         ENTRY;
1474
1475         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_FINISH_REPLAY)))
1476                 RETURN(0);
1477
1478         if (!atomic_add_unless(&imp->imp_replay_inflight, 1, 1))
1479                 RETURN(0);
1480
1481         req = ptlrpc_request_alloc_pack(imp, &RQF_OBD_PING, LUSTRE_OBD_VERSION,
1482                                         OBD_PING);
1483         if (req == NULL) {
1484                 atomic_dec(&imp->imp_replay_inflight);
1485                 RETURN(-ENOMEM);
1486         }
1487
1488         ptlrpc_request_set_replen(req);
1489         req->rq_send_state = LUSTRE_IMP_REPLAY_WAIT;
1490         lustre_msg_add_flags(req->rq_reqmsg,
1491                              MSG_LOCK_REPLAY_DONE | MSG_REQ_REPLAY_DONE);
1492         if (AT_OFF)
1493                 req->rq_timeout *= 3;
1494         req->rq_interpret_reply = completed_replay_interpret;
1495
1496         ptlrpcd_add_req(req);
1497         RETURN(0);
1498 }
1499
1500 /**
1501  * In kernel code all import invalidation happens in its own
1502  * separate thread, so that whatever application happened to encounter
1503  * a problem could still be killed or otherwise continue
1504  */
1505 static int ptlrpc_invalidate_import_thread(void *data)
1506 {
1507         struct obd_import *imp = data;
1508
1509         ENTRY;
1510         CDEBUG(D_HA, "thread invalidate import %s to %s@%s\n",
1511                imp->imp_obd->obd_name, obd2cli_tgt(imp->imp_obd),
1512                imp->imp_connection->c_remote_uuid.uuid);
1513
1514         if (do_dump_on_eviction(imp->imp_obd)) {
1515                 CERROR("dump the log upon eviction\n");
1516                 libcfs_debug_dumplog();
1517         }
1518
1519         ptlrpc_invalidate_import(imp);
1520         import_set_state(imp, LUSTRE_IMP_RECOVER);
1521         ptlrpc_import_recovery_state_machine(imp);
1522
1523         class_import_put(imp);
1524         RETURN(0);
1525 }
1526
1527 /**
1528  * This is the state machine for client-side recovery on import.
1529  *
1530  * Typicaly we have two possibly paths. If we came to server and it is not
1531  * in recovery, we just enter IMP_EVICTED state, invalidate our import
1532  * state and reconnect from scratch.
1533  * If we came to server that is in recovery, we enter IMP_REPLAY import state.
1534  * We go through our list of requests to replay and send them to server one by
1535  * one.
1536  * After sending all request from the list we change import state to
1537  * IMP_REPLAY_LOCKS and re-request all the locks we believe we have from server
1538  * and also all the locks we don't yet have and wait for server to grant us.
1539  * After that we send a special "replay completed" request and change import
1540  * state to IMP_REPLAY_WAIT.
1541  * Upon receiving reply to that "replay completed" RPC we enter IMP_RECOVER
1542  * state and resend all requests from sending list.
1543  * After that we promote import to FULL state and send all delayed requests
1544  * and import is fully operational after that.
1545  *
1546  */
1547 int ptlrpc_import_recovery_state_machine(struct obd_import *imp)
1548 {
1549         int rc = 0;
1550         int inflight;
1551         char *target_start;
1552         int target_len;
1553
1554         ENTRY;
1555         if (imp->imp_state == LUSTRE_IMP_EVICTED) {
1556                 struct task_struct *task;
1557
1558                 deuuidify(obd2cli_tgt(imp->imp_obd), NULL,
1559                           &target_start, &target_len);
1560                 /* Don't care about MGC eviction */
1561                 if (strcmp(imp->imp_obd->obd_type->typ_name,
1562                            LUSTRE_MGC_NAME) != 0) {
1563                         LCONSOLE_ERROR_MSG(0x167, "%s: This client was evicted "
1564                                            "by %.*s; in progress operations "
1565                                            "using this service will fail.\n",
1566                                            imp->imp_obd->obd_name, target_len,
1567                                            target_start);
1568                         LASSERTF(!obd_lbug_on_eviction, "LBUG upon eviction\n");
1569                 }
1570                 CDEBUG(D_HA, "evicted from %s@%s; invalidating\n",
1571                        obd2cli_tgt(imp->imp_obd),
1572                        imp->imp_connection->c_remote_uuid.uuid);
1573                 /* reset vbr_failed flag upon eviction */
1574                 spin_lock(&imp->imp_lock);
1575                 imp->imp_vbr_failed = 0;
1576                 spin_unlock(&imp->imp_lock);
1577
1578                 /* bug 17802:  XXX client_disconnect_export vs connect request
1579                  * race. if client is evicted at this time then we start
1580                  * invalidate thread without reference to import and import can
1581                  * be freed at same time. */
1582                 class_import_get(imp);
1583                 task = kthread_run(ptlrpc_invalidate_import_thread, imp,
1584                                    "ll_imp_inval");
1585                 if (IS_ERR(task)) {
1586                         class_import_put(imp);
1587                         rc = PTR_ERR(task);
1588                         CERROR("%s: can't start invalidate thread: rc = %d\n",
1589                                imp->imp_obd->obd_name, rc);
1590                 } else {
1591                         rc = 0;
1592                 }
1593                 RETURN(rc);
1594         }
1595
1596         if (imp->imp_state == LUSTRE_IMP_REPLAY) {
1597                 CDEBUG(D_HA, "replay requested by %s\n",
1598                        obd2cli_tgt(imp->imp_obd));
1599                 rc = ptlrpc_replay_next(imp, &inflight);
1600                 if (inflight == 0 &&
1601                     atomic_read(&imp->imp_replay_inflight) == 0) {
1602                         import_set_state(imp, LUSTRE_IMP_REPLAY_LOCKS);
1603                         rc = ldlm_replay_locks(imp);
1604                         if (rc)
1605                                 GOTO(out, rc);
1606                 }
1607                 rc = 0;
1608         }
1609
1610         if (imp->imp_state == LUSTRE_IMP_REPLAY_LOCKS) {
1611                 if (atomic_read(&imp->imp_replay_inflight) == 0) {
1612                         import_set_state(imp, LUSTRE_IMP_REPLAY_WAIT);
1613                         rc = signal_completed_replay(imp);
1614                         if (rc)
1615                                 GOTO(out, rc);
1616                 }
1617         }
1618
1619         if (imp->imp_state == LUSTRE_IMP_REPLAY_WAIT) {
1620                 if (atomic_read(&imp->imp_replay_inflight) == 0) {
1621                         import_set_state(imp, LUSTRE_IMP_RECOVER);
1622                 }
1623         }
1624
1625         if (imp->imp_state == LUSTRE_IMP_RECOVER) {
1626                 struct ptlrpc_connection *conn = imp->imp_connection;
1627
1628                 rc = ptlrpc_resend(imp);
1629                 if (rc)
1630                         GOTO(out, rc);
1631                 ptlrpc_activate_import(imp, true);
1632
1633                 /* Reverse import are flagged with dlm_fake == 1.
1634                  * They do not do recovery and connection are not "restored".
1635                  */
1636                 if (!imp->imp_dlm_fake)
1637                         CDEBUG_LIMIT(imp->imp_was_idle ?
1638                                         imp->imp_idle_debug : D_CONSOLE,
1639                                      "%s: Connection restored to %s (at %s)\n",
1640                                      imp->imp_obd->obd_name,
1641                                      obd_uuid2str(&conn->c_remote_uuid),
1642                                      obd_import_nid2str(imp));
1643                 spin_lock(&imp->imp_lock);
1644                 imp->imp_was_idle = 0;
1645                 spin_unlock(&imp->imp_lock);
1646         }
1647
1648         if (imp->imp_state == LUSTRE_IMP_FULL) {
1649                 wake_up(&imp->imp_recovery_waitq);
1650                 ptlrpc_wake_delayed(imp);
1651         }
1652
1653 out:
1654         RETURN(rc);
1655 }
1656
1657 static struct ptlrpc_request *ptlrpc_disconnect_prep_req(struct obd_import *imp)
1658 {
1659         struct ptlrpc_request *req;
1660         int rq_opc, rc = 0;
1661         ENTRY;
1662
1663         switch (imp->imp_connect_op) {
1664         case OST_CONNECT:
1665                 rq_opc = OST_DISCONNECT;
1666                 break;
1667         case MDS_CONNECT:
1668                 rq_opc = MDS_DISCONNECT;
1669                 break;
1670         case MGS_CONNECT:
1671                 rq_opc = MGS_DISCONNECT;
1672                 break;
1673         default:
1674                 rc = -EINVAL;
1675                 CERROR("%s: don't know how to disconnect from %s "
1676                        "(connect_op %d): rc = %d\n",
1677                        imp->imp_obd->obd_name, obd2cli_tgt(imp->imp_obd),
1678                        imp->imp_connect_op, rc);
1679                 RETURN(ERR_PTR(rc));
1680         }
1681
1682         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_DISCONNECT,
1683                                         LUSTRE_OBD_VERSION, rq_opc);
1684         if (req == NULL)
1685                 RETURN(ERR_PTR(-ENOMEM));
1686
1687         /* We are disconnecting, do not retry a failed DISCONNECT rpc if
1688          * it fails.  We can get through the above with a down server
1689          * if the client doesn't know the server is gone yet. */
1690         req->rq_no_resend = 1;
1691
1692         /* We want client umounts to happen quickly, no matter the
1693            server state... */
1694         req->rq_timeout = min_t(timeout_t, req->rq_timeout,
1695                                 INITIAL_CONNECT_TIMEOUT);
1696
1697         req->rq_send_state =  LUSTRE_IMP_CONNECTING;
1698         ptlrpc_request_set_replen(req);
1699
1700         RETURN(req);
1701 }
1702
1703 int ptlrpc_disconnect_import(struct obd_import *imp, int noclose)
1704 {
1705         struct ptlrpc_request *req;
1706         int rc = 0;
1707         ENTRY;
1708
1709         if (imp->imp_obd->obd_force)
1710                 GOTO(set_state, rc);
1711
1712         /* probably the import has been disconnected already being idle */
1713         spin_lock(&imp->imp_lock);
1714         if (imp->imp_state == LUSTRE_IMP_IDLE)
1715                 GOTO(out, rc);
1716         spin_unlock(&imp->imp_lock);
1717
1718         if (ptlrpc_import_in_recovery(imp)) {
1719                 long timeout_jiffies;
1720                 time64_t timeout;
1721
1722                 if (AT_OFF) {
1723                         if (imp->imp_server_timeout)
1724                                 timeout = obd_timeout >> 1;
1725                         else
1726                                 timeout = obd_timeout;
1727                 } else {
1728                         u32 req_portal;
1729                         int idx;
1730
1731                         req_portal = imp->imp_client->cli_request_portal;
1732                         idx = import_at_get_index(imp, req_portal);
1733                         timeout = at_get(&imp->imp_at.iat_service_estimate[idx]);
1734                 }
1735
1736                 timeout_jiffies = cfs_time_seconds(timeout);
1737                 if (wait_event_idle_timeout(imp->imp_recovery_waitq,
1738                                             !ptlrpc_import_in_recovery(imp),
1739                                             timeout_jiffies) == 0 &&
1740                     l_wait_event_abortable(imp->imp_recovery_waitq,
1741                                            !ptlrpc_import_in_recovery(imp)) < 0)
1742                         rc = -EINTR;
1743         }
1744
1745         req = ptlrpc_disconnect_prep_req(imp);
1746         if (IS_ERR(req))
1747                 GOTO(set_state, rc = PTR_ERR(req));
1748
1749         spin_lock(&imp->imp_lock);
1750         if (imp->imp_state != LUSTRE_IMP_FULL) {
1751                 ptlrpc_req_finished_with_imp_lock(req);
1752                 GOTO(out, rc);
1753         }
1754         import_set_state_nolock(imp, LUSTRE_IMP_CONNECTING);
1755         spin_unlock(&imp->imp_lock);
1756
1757         rc = ptlrpc_queue_wait(req);
1758         ptlrpc_req_finished(req);
1759
1760 set_state:
1761         spin_lock(&imp->imp_lock);
1762 out:
1763         if (noclose)
1764                 import_set_state_nolock(imp, LUSTRE_IMP_DISCON);
1765         else
1766                 import_set_state_nolock(imp, LUSTRE_IMP_CLOSED);
1767         memset(&imp->imp_remote_handle, 0, sizeof(imp->imp_remote_handle));
1768         spin_unlock(&imp->imp_lock);
1769
1770         if (rc == -ETIMEDOUT || rc == -ENOTCONN || rc == -ESHUTDOWN)
1771                 rc = 0;
1772         RETURN(rc);
1773 }
1774 EXPORT_SYMBOL(ptlrpc_disconnect_import);
1775
1776 static void ptlrpc_reset_reqs_generation(struct obd_import *imp)
1777 {
1778         struct ptlrpc_request *old, *tmp;
1779
1780         /* tag all resendable requests generated before disconnection
1781          * notice this code is part of disconnect-at-idle path only */
1782         list_for_each_entry_safe(old, tmp, &imp->imp_delayed_list,
1783                         rq_list) {
1784                 spin_lock(&old->rq_lock);
1785                 if (old->rq_import_generation == imp->imp_generation - 1 &&
1786                     ((imp->imp_initiated_at == imp->imp_generation) ||
1787                      !old->rq_no_resend))
1788                         old->rq_import_generation = imp->imp_generation;
1789                 spin_unlock(&old->rq_lock);
1790         }
1791 }
1792
1793 static int ptlrpc_disconnect_idle_interpret(const struct lu_env *env,
1794                                             struct ptlrpc_request *req,
1795                                             void *args, int rc)
1796 {
1797         struct obd_import *imp = req->rq_import;
1798         int connect = 0;
1799
1800         DEBUG_REQ(D_HA, req, "inflight=%d, refcount=%d: rc = %d",
1801                   atomic_read(&imp->imp_inflight),
1802                   refcount_read(&imp->imp_refcount), rc);
1803
1804         spin_lock(&imp->imp_lock);
1805         /* DISCONNECT reply can be late and another connection can just
1806          * be initiated. so we have to abort disconnection. */
1807         if (req->rq_import_generation == imp->imp_generation &&
1808             imp->imp_state != LUSTRE_IMP_CLOSED) {
1809                 LASSERTF(imp->imp_state == LUSTRE_IMP_CONNECTING,
1810                          "%s\n", ptlrpc_import_state_name(imp->imp_state));
1811                 memset(&imp->imp_remote_handle, 0,
1812                        sizeof(imp->imp_remote_handle));
1813                 /* take our DISCONNECT into account */
1814                 if (atomic_read(&imp->imp_reqs) > 1) {
1815                         imp->imp_generation++;
1816                         imp->imp_initiated_at = imp->imp_generation;
1817                         import_set_state_nolock(imp, LUSTRE_IMP_NEW);
1818                         ptlrpc_reset_reqs_generation(imp);
1819                         connect = 1;
1820                 } else {
1821                         /* do not expose transient IDLE state */
1822                         import_set_state_nolock(imp, LUSTRE_IMP_IDLE);
1823                 }
1824         }
1825
1826         if (connect) {
1827                 rc = ptlrpc_connect_import_locked(imp);
1828                 if (rc >= 0)
1829                         ptlrpc_pinger_add_import(imp);
1830         } else {
1831                 spin_unlock(&imp->imp_lock);
1832         }
1833
1834         return 0;
1835 }
1836
1837 static bool ptlrpc_can_idle(struct obd_import *imp)
1838 {
1839         struct ldlm_namespace *ns = imp->imp_obd->obd_namespace;
1840
1841         /* one request for disconnect rpc */
1842         if (atomic_read(&imp->imp_reqs) > 1)
1843                 return false;
1844
1845         /* any lock increases ns_bref being a resource holder */
1846         if (ns && atomic_read(&ns->ns_bref) > 0)
1847                 return false;
1848
1849         return true;
1850 }
1851
1852 int ptlrpc_disconnect_and_idle_import(struct obd_import *imp)
1853 {
1854         struct ptlrpc_request *req;
1855         ENTRY;
1856
1857         if (imp->imp_obd->obd_force)
1858                 RETURN(0);
1859
1860         if (ptlrpc_import_in_recovery(imp))
1861                 RETURN(0);
1862
1863         req = ptlrpc_disconnect_prep_req(imp);
1864         if (IS_ERR(req))
1865                 RETURN(PTR_ERR(req));
1866
1867         req->rq_interpret_reply = ptlrpc_disconnect_idle_interpret;
1868
1869         if (OBD_FAIL_PRECHECK(OBD_FAIL_PTLRPC_IDLE_RACE)) {
1870                 __u32 idx;
1871
1872                 server_name2index(imp->imp_obd->obd_name, &idx, NULL);
1873                 if (idx == 0)
1874                         OBD_RACE(OBD_FAIL_PTLRPC_IDLE_RACE);
1875         }
1876
1877         spin_lock(&imp->imp_lock);
1878         if (imp->imp_state != LUSTRE_IMP_FULL || !ptlrpc_can_idle(imp)) {
1879                 ptlrpc_req_finished_with_imp_lock(req);
1880                 spin_unlock(&imp->imp_lock);
1881                 RETURN(0);
1882         }
1883         import_set_state_nolock(imp, LUSTRE_IMP_CONNECTING);
1884         /* don't make noise at reconnection */
1885         imp->imp_was_idle = 1;
1886         spin_unlock(&imp->imp_lock);
1887
1888         CDEBUG_LIMIT(imp->imp_idle_debug, "%s: disconnect after %llus idle\n",
1889                      imp->imp_obd->obd_name,
1890                      ktime_get_real_seconds() - imp->imp_last_reply_time);
1891
1892         ptlrpcd_add_req(req);
1893
1894         RETURN(1);
1895 }
1896 EXPORT_SYMBOL(ptlrpc_disconnect_and_idle_import);
1897
1898 void ptlrpc_cleanup_imp(struct obd_import *imp)
1899 {
1900         ENTRY;
1901
1902         spin_lock(&imp->imp_lock);
1903
1904         import_set_state_nolock(imp, LUSTRE_IMP_CLOSED);
1905         imp->imp_generation++;
1906         ptlrpc_abort_inflight(imp);
1907
1908         spin_unlock(&imp->imp_lock);
1909
1910         EXIT;
1911 }
1912
1913 /* Adaptive Timeout utils */
1914
1915 /* Update at_current_timeout with the specified value (bounded by at_min and
1916  * at_max), as well as the AT history "bins".
1917  *  - Bin into timeslices using AT_BINS bins.
1918  *  - This gives us a max of the last at_history seconds without the storage,
1919  *    but still smoothing out a return to normalcy from a slow response.
1920  *  - (E.g. remember the maximum latency in each minute of the last 4 minutes.)
1921  */
1922 timeout_t at_measured(struct adaptive_timeout *at, timeout_t timeout)
1923 {
1924         timeout_t old_timeout = at->at_current_timeout;
1925         time64_t now = ktime_get_real_seconds();
1926         long binlimit = max_t(long, at_history / AT_BINS, 1);
1927
1928         LASSERT(at);
1929         CDEBUG(D_OTHER, "add %u to %p time=%lld v=%u (%u %u %u %u)\n",
1930                timeout, at, now - at->at_binstart, at->at_current_timeout,
1931                at->at_hist[0], at->at_hist[1], at->at_hist[2], at->at_hist[3]);
1932
1933         if (timeout <= 0)
1934                 /* Negative timeouts and 0's don't count, because we never
1935                  * want our timeout to drop to 0 or below, and because 0 could
1936                  * mean an error
1937                  */
1938                 return 0;
1939
1940         spin_lock(&at->at_lock);
1941
1942         if (unlikely(at->at_binstart == 0)) {
1943                 /* Special case to remove default from history */
1944                 at->at_current_timeout = timeout;
1945                 at->at_worst_timeout_ever = timeout;
1946                 at->at_worst_timestamp = now;
1947                 at->at_hist[0] = timeout;
1948                 at->at_binstart = now;
1949         } else if (now - at->at_binstart < binlimit ) {
1950                 /* in bin 0 */
1951                 at->at_hist[0] = max_t(timeout_t, timeout, at->at_hist[0]);
1952                 at->at_current_timeout = max_t(timeout_t, timeout,
1953                                                at->at_current_timeout);
1954         } else {
1955                 int i, shift;
1956                 timeout_t maxv = timeout;
1957
1958                 /* move bins over */
1959                 shift = (u32)(now - at->at_binstart) / binlimit;
1960                 LASSERT(shift > 0);
1961                 for(i = AT_BINS - 1; i >= 0; i--) {
1962                         if (i >= shift) {
1963                                 at->at_hist[i] = at->at_hist[i - shift];
1964                                 maxv = max_t(timeout_t, maxv, at->at_hist[i]);
1965                         } else {
1966                                 at->at_hist[i] = 0;
1967                         }
1968                 }
1969                 at->at_hist[0] = timeout;
1970                 at->at_current_timeout = maxv;
1971                 at->at_binstart += shift * binlimit;
1972         }
1973
1974         if (at->at_current_timeout > at->at_worst_timeout_ever) {
1975                 at->at_worst_timeout_ever = at->at_current_timeout;
1976                 at->at_worst_timestamp = now;
1977         }
1978
1979         if (at->at_flags & AT_FLG_NOHIST)
1980                 /* Only keep last reported val; keeping the rest of the history
1981                  * for debugfs only
1982                  */
1983                 at->at_current_timeout = timeout;
1984
1985         if (at_max > 0)
1986                 at->at_current_timeout = min_t(timeout_t,
1987                                                at->at_current_timeout, at_max);
1988         at->at_current_timeout = max_t(timeout_t, at->at_current_timeout,
1989                                        at_min);
1990         if (at->at_current_timeout != old_timeout)
1991                 CDEBUG(D_OTHER,
1992                        "AT %p change: old=%u new=%u delta=%d (val=%d) hist %u %u %u %u\n",
1993                        at, old_timeout, at->at_current_timeout,
1994                        at->at_current_timeout - old_timeout, timeout,
1995                        at->at_hist[0], at->at_hist[1], at->at_hist[2],
1996                        at->at_hist[3]);
1997
1998         /* if we changed, report the old timeout value */
1999         old_timeout = (at->at_current_timeout != old_timeout) ? old_timeout : 0;
2000
2001         spin_unlock(&at->at_lock);
2002         return old_timeout;
2003 }
2004
2005 /* Find the imp_at index for a given portal; assign if space available */
2006 int import_at_get_index(struct obd_import *imp, int portal)
2007 {
2008         struct imp_at *at = &imp->imp_at;
2009         int i;
2010
2011         for (i = 0; i < IMP_AT_MAX_PORTALS; i++) {
2012                 if (at->iat_portal[i] == portal)
2013                         return i;
2014                 if (at->iat_portal[i] == 0)
2015                         /* unused */
2016                         break;
2017         }
2018
2019         /* Not found in list, add it under a lock */
2020         spin_lock(&imp->imp_lock);
2021
2022         /* Check unused under lock */
2023         for (; i < IMP_AT_MAX_PORTALS; i++) {
2024                 if (at->iat_portal[i] == portal)
2025                         goto out;
2026                 if (at->iat_portal[i] == 0)
2027                         /* unused */
2028                         break;
2029         }
2030
2031         /* Not enough portals? */
2032         LASSERT(i < IMP_AT_MAX_PORTALS);
2033
2034         at->iat_portal[i] = portal;
2035 out:
2036         spin_unlock(&imp->imp_lock);
2037         return i;
2038 }