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