lustre/ldlm/ldlm_lib.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) 2010, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  */
  32
  33 /**
  34  * This file deals with various client/target related logic including recovery.
  35  *
  36  * TODO: This code more logically belongs in the ptlrpc module than in ldlm and
  37  * should be moved.
  38  */
  39
  40 #define DEBUG_SUBSYSTEM S_LDLM
  41
  42 #include <linux/jiffies.h>
  43 #include <linux/kthread.h>
  44 #include <libcfs/libcfs.h>
  45 #include <obd.h>
  46 #include <obd_class.h>
  47 #include <lustre_dlm.h>
  48 #include <lustre_net.h>
  49 #include <lustre_sec.h>
  50 #include "ldlm_internal.h"
  51
  52 /* @priority: If non-zero, move the selected connection to the list head.
  53  * @create: If zero, only search in existing connections.
  54  */
  55 static int import_set_conn(struct obd_import *imp, struct obd_uuid *uuid,
  56                            int priority, int create)
  57 {
  58         struct ptlrpc_connection *ptlrpc_conn;
  59         struct obd_import_conn *imp_conn = NULL, *item;
  60         lnet_nid_t nid4refnet = LNET_NID_ANY;
  61         int rc = 0;
  62         ENTRY;
  63
  64         if (!create && !priority) {
  65                 CDEBUG(D_HA, "Nothing to do\n");
  66                 RETURN(-EINVAL);
  67         }
  68
  69         if (imp->imp_connection &&
  70             imp->imp_connection->c_remote_uuid.uuid[0] == 0)
  71                 /* nid4refnet is used to restrict network connections */
  72                 nid4refnet = imp->imp_connection->c_self;
  73         ptlrpc_conn = ptlrpc_uuid_to_connection(uuid, nid4refnet);
  74         if (!ptlrpc_conn) {
  75                 CDEBUG(D_HA, "can't find connection %s\n", uuid->uuid);
  76                 RETURN(-ENOENT);
  77         }
  78
  79         if (create) {
  80                 OBD_ALLOC(imp_conn, sizeof(*imp_conn));
  81                 if (!imp_conn)
  82                         GOTO(out_put, rc = -ENOMEM);
  83         }
  84
  85         spin_lock(&imp->imp_lock);
  86         list_for_each_entry(item, &imp->imp_conn_list, oic_item) {
  87                 if (obd_uuid_equals(uuid, &item->oic_uuid)) {
  88                         if (priority) {
  89                                 list_del(&item->oic_item);
  90                                 list_add(&item->oic_item,
  91                                          &imp->imp_conn_list);
  92                                 item->oic_last_attempt = 0;
  93                         }
  94                         CDEBUG(D_HA, "imp %p@%s: found existing conn %s%s\n",
  95                                imp, imp->imp_obd->obd_name, uuid->uuid,
  96                                (priority ? ", moved to head" : ""));
  97                         spin_unlock(&imp->imp_lock);
  98                         GOTO(out_free, rc = 0);
  99                 }
 100         }
 101         /* No existing import connection found for \a uuid. */
 102         if (create) {
 103                 imp_conn->oic_conn = ptlrpc_conn;
 104                 imp_conn->oic_uuid = *uuid;
 105                 imp_conn->oic_last_attempt = 0;
 106                 if (priority)
 107                         list_add(&imp_conn->oic_item, &imp->imp_conn_list);
 108                 else
 109                         list_add_tail(&imp_conn->oic_item,
 110                                       &imp->imp_conn_list);
 111                 CDEBUG(D_HA, "imp %p@%s: add connection %s at %s\n",
 112                        imp, imp->imp_obd->obd_name, uuid->uuid,
 113                        (priority ? "head" : "tail"));
 114         } else {
 115                 spin_unlock(&imp->imp_lock);
 116                 GOTO(out_free, rc = -ENOENT);
 117         }
 118
 119         spin_unlock(&imp->imp_lock);
 120         RETURN(0);
 121 out_free:
 122         if (imp_conn)
 123                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 124 out_put:
 125         ptlrpc_connection_put(ptlrpc_conn);
 126         RETURN(rc);
 127 }
 128
 129 int import_set_conn_priority(struct obd_import *imp, struct obd_uuid *uuid)
 130 {
 131         return import_set_conn(imp, uuid, 1, 0);
 132 }
 133
 134 int client_import_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
 135                            int priority)
 136 {
 137         return import_set_conn(imp, uuid, priority, 1);
 138 }
 139 EXPORT_SYMBOL(client_import_add_conn);
 140
 141 int client_import_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
 142 {
 143         struct obd_import_conn *imp_conn;
 144         struct obd_export *dlmexp;
 145         int rc = -ENOENT;
 146         ENTRY;
 147
 148         spin_lock(&imp->imp_lock);
 149         if (list_empty(&imp->imp_conn_list)) {
 150                 LASSERT(!imp->imp_connection);
 151                 GOTO(out, rc);
 152         }
 153
 154         list_for_each_entry(imp_conn, &imp->imp_conn_list, oic_item) {
 155                 if (!obd_uuid_equals(uuid, &imp_conn->oic_uuid))
 156                         continue;
 157                 LASSERT(imp_conn->oic_conn);
 158
 159                 if (imp_conn == imp->imp_conn_current) {
 160                         LASSERT(imp_conn->oic_conn == imp->imp_connection);
 161
 162                         if (imp->imp_state != LUSTRE_IMP_CLOSED &&
 163                             imp->imp_state != LUSTRE_IMP_DISCON) {
 164                                 CERROR("can't remove current connection\n");
 165                                 GOTO(out, rc = -EBUSY);
 166                         }
 167
 168                         ptlrpc_connection_put(imp->imp_connection);
 169                         imp->imp_connection = NULL;
 170
 171                         dlmexp = class_conn2export(&imp->imp_dlm_handle);
 172                         if (dlmexp && dlmexp->exp_connection) {
 173                                 LASSERT(dlmexp->exp_connection ==
 174                                         imp_conn->oic_conn);
 175                                 ptlrpc_connection_put(dlmexp->exp_connection);
 176                                 dlmexp->exp_connection = NULL;
 177                         }
 178
 179                         if (dlmexp != NULL)
 180                                 class_export_put(dlmexp);
 181                 }
 182
 183                 list_del(&imp_conn->oic_item);
 184                 ptlrpc_connection_put(imp_conn->oic_conn);
 185                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 186                 CDEBUG(D_HA, "imp %p@%s: remove connection %s\n",
 187                        imp, imp->imp_obd->obd_name, uuid->uuid);
 188                 rc = 0;
 189                 break;
 190         }
 191 out:
 192         spin_unlock(&imp->imp_lock);
 193         if (rc == -ENOENT)
 194                 CERROR("connection %s not found\n", uuid->uuid);
 195         RETURN(rc);
 196 }
 197 EXPORT_SYMBOL(client_import_del_conn);
 198
 199 /**
 200  * Find conn UUID by peer NID. \a peer is a server NID. This function is used
 201  * to find a conn uuid of \a imp which can reach \a peer.
 202  */
 203 int client_import_find_conn(struct obd_import *imp, lnet_nid_t peer,
 204                             struct obd_uuid *uuid)
 205 {
 206         struct obd_import_conn *conn;
 207         int rc = -ENOENT;
 208         ENTRY;
 209
 210         spin_lock(&imp->imp_lock);
 211         list_for_each_entry(conn, &imp->imp_conn_list, oic_item) {
 212                 /* Check if conn UUID does have this peer NID. */
 213                 if (class_check_uuid(&conn->oic_uuid, peer)) {
 214                         *uuid = conn->oic_uuid;
 215                         rc = 0;
 216                         break;
 217                 }
 218         }
 219         spin_unlock(&imp->imp_lock);
 220         RETURN(rc);
 221 }
 222 EXPORT_SYMBOL(client_import_find_conn);
 223
 224 void client_destroy_import(struct obd_import *imp)
 225 {
 226         /* Drop security policy instance after all RPCs have finished/aborted
 227          * to let all busy contexts be released. */
 228         class_import_get(imp);
 229         class_destroy_import(imp);
 230         sptlrpc_import_sec_put(imp);
 231         class_import_put(imp);
 232 }
 233 EXPORT_SYMBOL(client_destroy_import);
 234
 235 /**
 236  * Check whether or not the OSC is on MDT.
 237  * In the config log,
 238  * osc on MDT
 239  *      setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
 240  * osc on client
 241  *      setup 0:{fsname}-OSTxxxx-osc 1:lustre-OST0000_UUID 2:NID
 242  *
 243  **/
 244 static int osc_on_mdt(char *obdname)
 245 {
 246         char *ptr;
 247
 248         ptr = strrchr(obdname, '-');
 249         if (ptr == NULL)
 250                 return 0;
 251
 252         if (strncmp(ptr + 1, "MDT", 3) == 0)
 253                 return 1;
 254
 255         return 0;
 256 }
 257
 258 /* Configure an RPC client OBD device.
 259  *
 260  * lcfg parameters:
 261  * 1 - client UUID
 262  * 2 - server UUID
 263  * 3 - inactive-on-startup
 264  * 4 - restrictive net
 265  */
 266 int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
 267 {
 268         struct client_obd *cli = &obddev->u.cli;
 269         struct obd_import *imp;
 270         struct obd_uuid server_uuid;
 271         int rq_portal, rp_portal, connect_op;
 272         char *name = obddev->obd_type->typ_name;
 273         enum ldlm_ns_type ns_type = LDLM_NS_TYPE_UNKNOWN;
 274         char *cli_name = lustre_cfg_buf(lcfg, 0);
 275         struct ptlrpc_connection fake_conn = { .c_self = 0,
 276                                                .c_remote_uuid.uuid[0] = 0 };
 277         int rc;
 278         ENTRY;
 279
 280         /* In a more perfect world, we would hang a ptlrpc_client off of
 281          * obd_type and just use the values from there. */
 282         if (!strcmp(name, LUSTRE_OSC_NAME)) {
 283                 rq_portal = OST_REQUEST_PORTAL;
 284                 rp_portal = OSC_REPLY_PORTAL;
 285                 connect_op = OST_CONNECT;
 286                 cli->cl_sp_me = LUSTRE_SP_CLI;
 287                 cli->cl_sp_to = LUSTRE_SP_OST;
 288                 ns_type = LDLM_NS_TYPE_OSC;
 289         } else if (!strcmp(name, LUSTRE_MDC_NAME) ||
 290                    !strcmp(name, LUSTRE_LWP_NAME)) {
 291                 rq_portal = MDS_REQUEST_PORTAL;
 292                 rp_portal = MDC_REPLY_PORTAL;
 293                 connect_op = MDS_CONNECT;
 294                 if (is_lwp_on_ost(cli_name))
 295                         cli->cl_sp_me = LUSTRE_SP_OST;
 296                 else if (is_lwp_on_mdt(cli_name))
 297                         cli->cl_sp_me = LUSTRE_SP_MDT;
 298                 else
 299                         cli->cl_sp_me = LUSTRE_SP_CLI;
 300                 cli->cl_sp_to = LUSTRE_SP_MDT;
 301                 ns_type = LDLM_NS_TYPE_MDC;
 302         } else if (!strcmp(name, LUSTRE_OSP_NAME)) {
 303                 if (strstr(lustre_cfg_buf(lcfg, 1), "OST") == NULL) {
 304                         /* OSP_on_MDT for other MDTs */
 305                         connect_op = MDS_CONNECT;
 306                         cli->cl_sp_to = LUSTRE_SP_MDT;
 307                         ns_type = LDLM_NS_TYPE_MDC;
 308                         rq_portal = OUT_PORTAL;
 309                 } else {
 310                         /* OSP on MDT for OST */
 311                         connect_op = OST_CONNECT;
 312                         cli->cl_sp_to = LUSTRE_SP_OST;
 313                         ns_type = LDLM_NS_TYPE_OSC;
 314                         rq_portal = OST_REQUEST_PORTAL;
 315                 }
 316                 rp_portal = OSC_REPLY_PORTAL;
 317                 cli->cl_sp_me = LUSTRE_SP_MDT;
 318         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
 319                 rq_portal = MGS_REQUEST_PORTAL;
 320                 rp_portal = MGC_REPLY_PORTAL;
 321                 connect_op = MGS_CONNECT;
 322                 cli->cl_sp_me = LUSTRE_SP_MGC;
 323                 cli->cl_sp_to = LUSTRE_SP_MGS;
 324                 cli->cl_flvr_mgc.sf_rpc = SPTLRPC_FLVR_INVALID;
 325                 ns_type = LDLM_NS_TYPE_MGC;
 326         } else {
 327                 CERROR("unknown client OBD type \"%s\", can't setup\n",
 328                        name);
 329                 RETURN(-EINVAL);
 330         }
 331
 332         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
 333                 CERROR("requires a TARGET UUID\n");
 334                 RETURN(-EINVAL);
 335         }
 336
 337         if (LUSTRE_CFG_BUFLEN(lcfg, 1) > 37) {
 338                 CERROR("client UUID must be less than 38 characters\n");
 339                 RETURN(-EINVAL);
 340         }
 341
 342         if (LUSTRE_CFG_BUFLEN(lcfg, 2) < 1) {
 343                 CERROR("setup requires a SERVER UUID\n");
 344                 RETURN(-EINVAL);
 345         }
 346
 347         if (LUSTRE_CFG_BUFLEN(lcfg, 2) > 37) {
 348                 CERROR("target UUID must be less than 38 characters\n");
 349                 RETURN(-EINVAL);
 350         }
 351
 352         init_rwsem(&cli->cl_sem);
 353         mutex_init(&cli->cl_mgc_mutex);
 354         cli->cl_seq = NULL;
 355         init_rwsem(&cli->cl_seq_rwsem);
 356         cli->cl_conn_count = 0;
 357         memcpy(server_uuid.uuid, lustre_cfg_buf(lcfg, 2),
 358                min_t(unsigned int, LUSTRE_CFG_BUFLEN(lcfg, 2),
 359                      sizeof(server_uuid)));
 360
 361         cli->cl_dirty_pages = 0;
 362         cli->cl_avail_grant = 0;
 363         /* FIXME: Should limit this for the sum of all cl_dirty_max_pages. */
 364         /* cl_dirty_max_pages may be changed at connect time in
 365          * ptlrpc_connect_interpret(). */
 366         client_adjust_max_dirty(cli);
 367         INIT_LIST_HEAD(&cli->cl_cache_waiters);
 368         INIT_LIST_HEAD(&cli->cl_loi_ready_list);
 369         INIT_LIST_HEAD(&cli->cl_loi_hp_ready_list);
 370         INIT_LIST_HEAD(&cli->cl_loi_write_list);
 371         INIT_LIST_HEAD(&cli->cl_loi_read_list);
 372         spin_lock_init(&cli->cl_loi_list_lock);
 373         atomic_set(&cli->cl_pending_w_pages, 0);
 374         atomic_set(&cli->cl_pending_r_pages, 0);
 375         cli->cl_r_in_flight = 0;
 376         cli->cl_w_in_flight = 0;
 377
 378         spin_lock_init(&cli->cl_read_rpc_hist.oh_lock);
 379         spin_lock_init(&cli->cl_write_rpc_hist.oh_lock);
 380         spin_lock_init(&cli->cl_read_page_hist.oh_lock);
 381         spin_lock_init(&cli->cl_write_page_hist.oh_lock);
 382         spin_lock_init(&cli->cl_read_offset_hist.oh_lock);
 383         spin_lock_init(&cli->cl_write_offset_hist.oh_lock);
 384
 385         /* lru for osc. */
 386         INIT_LIST_HEAD(&cli->cl_lru_osc);
 387         atomic_set(&cli->cl_lru_shrinkers, 0);
 388         atomic_long_set(&cli->cl_lru_busy, 0);
 389         atomic_long_set(&cli->cl_lru_in_list, 0);
 390         INIT_LIST_HEAD(&cli->cl_lru_list);
 391         spin_lock_init(&cli->cl_lru_list_lock);
 392         atomic_long_set(&cli->cl_unstable_count, 0);
 393         INIT_LIST_HEAD(&cli->cl_shrink_list);
 394
 395         INIT_LIST_HEAD(&cli->cl_flight_waiters);
 396         cli->cl_rpcs_in_flight = 0;
 397
 398         init_waitqueue_head(&cli->cl_destroy_waitq);
 399         atomic_set(&cli->cl_destroy_in_flight, 0);
 400
 401         cli->cl_supp_cksum_types = OBD_CKSUM_CRC32;
 402 #ifdef ENABLE_CHECKSUM
 403         /* Turn on checksumming by default. */
 404         cli->cl_checksum = 1;
 405         /*
 406          * The supported checksum types will be worked out at connect time
 407          * Set cl_chksum* to CRC32 for now to avoid returning screwed info
 408          * through procfs.
 409          */
 410         cli->cl_cksum_type = cli->cl_supp_cksum_types;
 411 #endif
 412         atomic_set(&cli->cl_resends, OSC_DEFAULT_RESENDS);
 413
 414         /* Set it to possible maximum size. It may be reduced by ocd_brw_size
 415          * from OFD after connecting. */
 416         cli->cl_max_pages_per_rpc = PTLRPC_MAX_BRW_PAGES;
 417
 418         cli->cl_short_io_bytes = OBD_MAX_SHORT_IO_BYTES;
 419
 420         /* set cl_chunkbits default value to PAGE_SHIFT,
 421          * it will be updated at OSC connection time. */
 422         cli->cl_chunkbits = PAGE_SHIFT;
 423
 424         if (!strcmp(name, LUSTRE_MDC_NAME)) {
 425                 cli->cl_max_rpcs_in_flight = OBD_MAX_RIF_DEFAULT;
 426         } else if (totalram_pages >> (20 - PAGE_SHIFT) <= 128 /* MB */) {
 427                 cli->cl_max_rpcs_in_flight = 2;
 428         } else if (totalram_pages >> (20 - PAGE_SHIFT) <= 256 /* MB */) {
 429                 cli->cl_max_rpcs_in_flight = 3;
 430         } else if (totalram_pages >> (20 - PAGE_SHIFT) <= 512 /* MB */) {
 431                 cli->cl_max_rpcs_in_flight = 4;
 432         } else {
 433                 if (osc_on_mdt(obddev->obd_name))
 434                         cli->cl_max_rpcs_in_flight = OBD_MAX_RIF_MAX;
 435                 else
 436                         cli->cl_max_rpcs_in_flight = OBD_MAX_RIF_DEFAULT;
 437         }
 438
 439         spin_lock_init(&cli->cl_mod_rpcs_lock);
 440         spin_lock_init(&cli->cl_mod_rpcs_hist.oh_lock);
 441         cli->cl_max_mod_rpcs_in_flight = 0;
 442         cli->cl_mod_rpcs_in_flight = 0;
 443         cli->cl_close_rpcs_in_flight = 0;
 444         init_waitqueue_head(&cli->cl_mod_rpcs_waitq);
 445         cli->cl_mod_tag_bitmap = NULL;
 446
 447         INIT_LIST_HEAD(&cli->cl_chg_dev_linkage);
 448
 449         if (connect_op == MDS_CONNECT) {
 450                 cli->cl_max_mod_rpcs_in_flight = cli->cl_max_rpcs_in_flight - 1;
 451                 OBD_ALLOC(cli->cl_mod_tag_bitmap,
 452                           BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long));
 453                 if (cli->cl_mod_tag_bitmap == NULL)
 454                         GOTO(err, rc = -ENOMEM);
 455         }
 456
 457         rc = ldlm_get_ref();
 458         if (rc) {
 459                 CERROR("ldlm_get_ref failed: %d\n", rc);
 460                 GOTO(err, rc);
 461         }
 462
 463         ptlrpc_init_client(rq_portal, rp_portal, name,
 464                            &obddev->obd_ldlm_client);
 465
 466         imp = class_new_import(obddev);
 467         if (imp == NULL)
 468                 GOTO(err_ldlm, rc = -ENOENT);
 469         imp->imp_client = &obddev->obd_ldlm_client;
 470         imp->imp_connect_op = connect_op;
 471         memcpy(cli->cl_target_uuid.uuid, lustre_cfg_buf(lcfg, 1),
 472                LUSTRE_CFG_BUFLEN(lcfg, 1));
 473         class_import_put(imp);
 474
 475         if (lustre_cfg_buf(lcfg, 4)) {
 476                 __u32 refnet = libcfs_str2net(lustre_cfg_string(lcfg, 4));
 477
 478                 if (refnet == LNET_NIDNET(LNET_NID_ANY)) {
 479                         rc = -EINVAL;
 480                         CERROR("%s: bad mount option 'network=%s': rc = %d\n",
 481                                obddev->obd_name, lustre_cfg_string(lcfg, 4),
 482                                rc);
 483                         GOTO(err_import, rc);
 484                 }
 485                 fake_conn.c_self = LNET_MKNID(refnet, 0);
 486                 imp->imp_connection = &fake_conn;
 487         }
 488
 489         rc = client_import_add_conn(imp, &server_uuid, 1);
 490         if (rc) {
 491                 CERROR("can't add initial connection\n");
 492                 GOTO(err_import, rc);
 493         }
 494         imp->imp_connection = NULL;
 495
 496         cli->cl_import = imp;
 497         /* cli->cl_max_mds_easize updated by mdc_init_ea_size() */
 498         cli->cl_max_mds_easize = sizeof(struct lov_mds_md_v3);
 499
 500         if (LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
 501                 if (!strcmp(lustre_cfg_string(lcfg, 3), "inactive")) {
 502                         CDEBUG(D_HA, "marking %s %s->%s as inactive\n",
 503                                name, obddev->obd_name,
 504                                cli->cl_target_uuid.uuid);
 505                         spin_lock(&imp->imp_lock);
 506                         imp->imp_deactive = 1;
 507                         spin_unlock(&imp->imp_lock);
 508                 }
 509         }
 510
 511         obddev->obd_namespace = ldlm_namespace_new(obddev, obddev->obd_name,
 512                                                    LDLM_NAMESPACE_CLIENT,
 513                                                    LDLM_NAMESPACE_GREEDY,
 514                                                    ns_type);
 515         if (obddev->obd_namespace == NULL) {
 516                 CERROR("Unable to create client namespace - %s\n",
 517                        obddev->obd_name);
 518                 GOTO(err_import, rc = -ENOMEM);
 519         }
 520
 521         RETURN(rc);
 522
 523 err_import:
 524         class_destroy_import(imp);
 525 err_ldlm:
 526         ldlm_put_ref();
 527 err:
 528         if (cli->cl_mod_tag_bitmap != NULL)
 529                 OBD_FREE(cli->cl_mod_tag_bitmap,
 530                          BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long));
 531         cli->cl_mod_tag_bitmap = NULL;
 532         RETURN(rc);
 533
 534 }
 535 EXPORT_SYMBOL(client_obd_setup);
 536
 537 int client_obd_cleanup(struct obd_device *obddev)
 538 {
 539         struct client_obd *cli = &obddev->u.cli;
 540         ENTRY;
 541
 542         ldlm_namespace_free_post(obddev->obd_namespace);
 543         obddev->obd_namespace = NULL;
 544
 545         obd_cleanup_client_import(obddev);
 546         LASSERT(obddev->u.cli.cl_import == NULL);
 547
 548         ldlm_put_ref();
 549
 550         if (cli->cl_mod_tag_bitmap != NULL)
 551                 OBD_FREE(cli->cl_mod_tag_bitmap,
 552                          BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long));
 553         cli->cl_mod_tag_bitmap = NULL;
 554
 555         RETURN(0);
 556 }
 557 EXPORT_SYMBOL(client_obd_cleanup);
 558
 559 /* ->o_connect() method for client side (OSC and MDC and MGC) */
 560 int client_connect_import(const struct lu_env *env,
 561                           struct obd_export **exp,
 562                           struct obd_device *obd, struct obd_uuid *cluuid,
 563                           struct obd_connect_data *data, void *localdata)
 564 {
 565         struct client_obd       *cli    = &obd->u.cli;
 566         struct obd_import       *imp    = cli->cl_import;
 567         struct obd_connect_data *ocd;
 568         struct lustre_handle    conn    = { 0 };
 569         int                     rc;
 570         ENTRY;
 571
 572         *exp = NULL;
 573         down_write(&cli->cl_sem);
 574         if (cli->cl_conn_count > 0)
 575                 GOTO(out_sem, rc = -EALREADY);
 576
 577         rc = class_connect(&conn, obd, cluuid);
 578         if (rc)
 579                 GOTO(out_sem, rc);
 580
 581         cli->cl_conn_count++;
 582         *exp = class_conn2export(&conn);
 583
 584         LASSERT(obd->obd_namespace);
 585
 586         imp->imp_dlm_handle = conn;
 587         rc = ptlrpc_init_import(imp);
 588         if (rc != 0)
 589                 GOTO(out_ldlm, rc);
 590
 591         ocd = &imp->imp_connect_data;
 592         if (data) {
 593                 *ocd = *data;
 594                 imp->imp_connect_flags_orig = data->ocd_connect_flags;
 595                 imp->imp_connect_flags2_orig = data->ocd_connect_flags2;
 596         }
 597
 598         rc = ptlrpc_connect_import(imp);
 599         if (rc != 0) {
 600                 LASSERT(imp->imp_state == LUSTRE_IMP_DISCON);
 601                 GOTO(out_ldlm, rc);
 602         }
 603         LASSERT(*exp != NULL && (*exp)->exp_connection);
 604
 605         if (data) {
 606                 LASSERTF((ocd->ocd_connect_flags & data->ocd_connect_flags) ==
 607                          ocd->ocd_connect_flags, "old %#llx, new %#llx\n",
 608                          data->ocd_connect_flags, ocd->ocd_connect_flags);
 609                 data->ocd_connect_flags = ocd->ocd_connect_flags;
 610                 data->ocd_connect_flags2 = ocd->ocd_connect_flags2;
 611         }
 612
 613         ptlrpc_pinger_add_import(imp);
 614
 615         EXIT;
 616
 617         if (rc) {
 618 out_ldlm:
 619                 cli->cl_conn_count--;
 620                 class_disconnect(*exp);
 621                 *exp = NULL;
 622         }
 623 out_sem:
 624         up_write(&cli->cl_sem);
 625
 626         return rc;
 627 }
 628 EXPORT_SYMBOL(client_connect_import);
 629
 630 int client_disconnect_export(struct obd_export *exp)
 631 {
 632         struct obd_device *obd = class_exp2obd(exp);
 633         struct client_obd *cli;
 634         struct obd_import *imp;
 635         int rc = 0, err;
 636         ENTRY;
 637
 638         if (!obd) {
 639                 CERROR("invalid export for disconnect: exp %p cookie %#llx\n",
 640                        exp, exp ? exp->exp_handle.h_cookie : -1);
 641                 RETURN(-EINVAL);
 642         }
 643
 644         cli = &obd->u.cli;
 645         imp = cli->cl_import;
 646
 647         down_write(&cli->cl_sem);
 648         CDEBUG(D_INFO, "disconnect %s - %zu\n", obd->obd_name,
 649                 cli->cl_conn_count);
 650
 651         if (cli->cl_conn_count == 0) {
 652                 CERROR("disconnecting disconnected device (%s)\n",
 653                        obd->obd_name);
 654                 GOTO(out_disconnect, rc = -EINVAL);
 655         }
 656
 657         cli->cl_conn_count--;
 658         if (cli->cl_conn_count != 0)
 659                 GOTO(out_disconnect, rc = 0);
 660
 661         /* Mark import deactivated now, so we don't try to reconnect if any
 662          * of the cleanup RPCs fails (e.g. LDLM cancel, etc).  We don't
 663          * fully deactivate the import, or that would drop all requests. */
 664         spin_lock(&imp->imp_lock);
 665         imp->imp_deactive = 1;
 666         spin_unlock(&imp->imp_lock);
 667
 668         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
 669          * delete it regardless.  (It's safe to delete an import that was
 670          * never added.) */
 671         (void)ptlrpc_pinger_del_import(imp);
 672
 673         if (obd->obd_namespace != NULL) {
 674                 /* obd_force == local only */
 675                 ldlm_cli_cancel_unused(obd->obd_namespace, NULL,
 676                                        obd->obd_force ? LCF_LOCAL : 0, NULL);
 677                 ldlm_namespace_free_prior(obd->obd_namespace, imp, obd->obd_force);
 678         }
 679
 680         /* There's no need to hold sem while disconnecting an import,
 681          * and it may actually cause deadlock in GSS. */
 682         up_write(&cli->cl_sem);
 683         rc = ptlrpc_disconnect_import(imp, 0);
 684         down_write(&cli->cl_sem);
 685
 686         ptlrpc_invalidate_import(imp);
 687
 688         EXIT;
 689
 690 out_disconnect:
 691         /* Use server style - class_disconnect should be always called for
 692          * o_disconnect. */
 693         err = class_disconnect(exp);
 694         if (!rc && err)
 695                 rc = err;
 696
 697         up_write(&cli->cl_sem);
 698
 699         RETURN(rc);
 700 }
 701 EXPORT_SYMBOL(client_disconnect_export);
 702
 703 #ifdef HAVE_SERVER_SUPPORT
 704 int server_disconnect_export(struct obd_export *exp)
 705 {
 706         int rc;
 707         ENTRY;
 708
 709         /* Disconnect early so that clients can't keep using export. */
 710         rc = class_disconnect(exp);
 711         /* Close import to avoid sending any requests. */
 712         if (exp->exp_imp_reverse)
 713                 ptlrpc_cleanup_imp(exp->exp_imp_reverse);
 714
 715         ldlm_bl_thread_wakeup();
 716
 717         /* complete all outstanding replies */
 718         spin_lock(&exp->exp_lock);
 719         while (!list_empty(&exp->exp_outstanding_replies)) {
 720                 struct ptlrpc_reply_state *rs =
 721                         list_entry(exp->exp_outstanding_replies.next,
 722                                        struct ptlrpc_reply_state, rs_exp_list);
 723                 struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
 724
 725                 spin_lock(&svcpt->scp_rep_lock);
 726
 727                 list_del_init(&rs->rs_exp_list);
 728
 729                 spin_lock(&rs->rs_lock);
 730                 /* clear rs_convert_lock to make sure rs is handled and put */
 731                 rs->rs_convert_lock = 0;
 732                 ptlrpc_schedule_difficult_reply(rs);
 733                 spin_unlock(&rs->rs_lock);
 734
 735                 spin_unlock(&svcpt->scp_rep_lock);
 736         }
 737         spin_unlock(&exp->exp_lock);
 738
 739         RETURN(rc);
 740 }
 741 EXPORT_SYMBOL(server_disconnect_export);
 742
 743 /* --------------------------------------------------------------------------
 744  * from old lib/target.c
 745  * -------------------------------------------------------------------------- */
 746
 747 static int target_handle_reconnect(struct lustre_handle *conn,
 748                                    struct obd_export *exp,
 749                                    struct obd_uuid *cluuid)
 750 {
 751         struct obd_device *target;
 752         struct lustre_handle *hdl;
 753         time64_t deadline;
 754         time64_t timeout;
 755         time64_t now;
 756         int rc = 0;
 757
 758         ENTRY;
 759         hdl = &exp->exp_imp_reverse->imp_remote_handle;
 760         if (!exp->exp_connection || !lustre_handle_is_used(hdl)) {
 761                 conn->cookie = exp->exp_handle.h_cookie;
 762                 CDEBUG(D_HA, "connect export for UUID '%s' at %p,"
 763                        " cookie %#llx\n", cluuid->uuid, exp, conn->cookie);
 764                 RETURN(0);
 765         }
 766
 767         target = exp->exp_obd;
 768
 769         /* Might be a re-connect after a partition. */
 770         if (memcmp(&conn->cookie, &hdl->cookie, sizeof conn->cookie)) {
 771                 LCONSOLE_WARN("%s: already connected client %s (at %s) "
 772                               "with handle %#llx. Rejecting client "
 773                               "with the same UUID trying to reconnect "
 774                               "with handle %#llx\n", target->obd_name,
 775                               obd_uuid2str(&exp->exp_client_uuid),
 776                               obd_export_nid2str(exp),
 777                               hdl->cookie, conn->cookie);
 778                 memset(conn, 0, sizeof *conn);
 779                 /* target_handle_connect() treats EALREADY and
 780                  * -EALREADY differently.  -EALREADY is an error
 781                  * (same UUID, different handle). */
 782                 RETURN(-EALREADY);
 783         }
 784
 785         if (!target->obd_recovering) {
 786                 LCONSOLE_WARN("%s: Client %s (at %s) reconnecting\n",
 787                         target->obd_name, obd_uuid2str(&exp->exp_client_uuid),
 788                         obd_export_nid2str(exp));
 789                 GOTO(out_already, rc);
 790         }
 791
 792         now = ktime_get_seconds();
 793         deadline = jiffies_to_msecs(target->obd_recovery_timer.expires) /
 794                    MSEC_PER_SEC;
 795         if (now < deadline) {
 796                 struct target_distribute_txn_data *tdtd;
 797                 int size = 0;
 798                 int count = 0;
 799                 char *buf = NULL;
 800
 801                 timeout = deadline - now;
 802                 tdtd = class_exp2tgt(exp)->lut_tdtd;
 803                 if (tdtd && tdtd->tdtd_show_update_logs_retrievers)
 804                         buf = tdtd->tdtd_show_update_logs_retrievers(
 805                                 tdtd->tdtd_show_retrievers_cbdata,
 806                                 &size, &count);
 807
 808                 if (count > 0)
 809                         LCONSOLE_WARN("%s: Recovery already passed deadline "
 810                                       "%lld:%.02lld. It is due to DNE recovery "
 811                                       "failed/stuck on the %d MDT(s):%s. "
 812                                       "Please wait until all MDTs recovered "
 813                                       "or abort the recovery by force.\n",
 814                                       target->obd_name, timeout / 60,
 815                                       timeout % 60, count,
 816                                       buf ? buf : "unknown (not enough RAM)");
 817                 else
 818                         LCONSOLE_WARN("%s: Recovery already passed deadline "
 819                                       "%lld:%.02lld. If you do not want to wait "
 820                                       "more, please abort the recovery by "
 821                                       "force.\n", target->obd_name,
 822                                       timeout / 60, timeout % 60);
 823
 824                 if (buf != NULL)
 825                         OBD_FREE(buf, size);
 826         } else {
 827                 timeout = now - deadline;
 828                 LCONSOLE_WARN("%s: Recovery already passed deadline"
 829                         " %lld:%.02lld, It is most likely due to DNE"
 830                         " recovery is failed or stuck, please wait a"
 831                         " few more minutes or abort the recovery.\n",
 832                         target->obd_name, timeout / 60, timeout % 60);
 833         }
 834
 835 out_already:
 836         conn->cookie = exp->exp_handle.h_cookie;
 837         /* target_handle_connect() treats EALREADY and
 838          * -EALREADY differently.  EALREADY means we are
 839          * doing a valid reconnect from the same client. */
 840         RETURN(EALREADY);
 841 }
 842
 843 static void
 844 check_and_start_recovery_timer(struct obd_device *obd,
 845                                struct ptlrpc_request *req, int new_client);
 846
 847 /**
 848  * update flags for import during reconnect process
 849  */
 850 static int rev_import_flags_update(struct obd_import *revimp,
 851                                    struct ptlrpc_request *req)
 852 {
 853         int rc;
 854         struct obd_connect_data *data;
 855
 856         data = req_capsule_client_get(&req->rq_pill, &RMF_CONNECT_DATA);
 857
 858         if (data->ocd_connect_flags & OBD_CONNECT_AT)
 859                 revimp->imp_msghdr_flags |= MSGHDR_AT_SUPPORT;
 860         else
 861                 revimp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;
 862
 863         revimp->imp_msghdr_flags |= MSGHDR_CKSUM_INCOMPAT18;
 864
 865         rc = sptlrpc_import_sec_adapt(revimp, req->rq_svc_ctx, &req->rq_flvr);
 866         if (rc) {
 867                 CERROR("%s: cannot get reverse import %s security: rc = %d\n",
 868                         revimp->imp_client->cli_name,
 869                         libcfs_id2str(req->rq_peer), rc);
 870                 return rc;
 871         }
 872
 873         return 0;
 874 }
 875
 876 /**
 877  * Allocate a new reverse import for an export.
 878  *
 879  * \retval -errno in case error hit
 880  * \retval 0 if reverse import correctly init
 881  **/
 882 int rev_import_init(struct obd_export *export)
 883 {
 884         struct obd_device *obd = export->exp_obd;
 885         struct obd_import *revimp;
 886
 887         LASSERT(export->exp_imp_reverse == NULL);
 888
 889         revimp = class_new_import(obd);
 890         if (revimp == NULL)
 891                 return -ENOMEM;
 892
 893         revimp->imp_remote_handle.cookie = 0ULL;
 894         revimp->imp_client = &obd->obd_ldlm_client;
 895         revimp->imp_dlm_fake = 1;
 896
 897         /* it is safe to connect import in new state as no sends possible */
 898         spin_lock(&export->exp_lock);
 899         export->exp_imp_reverse = revimp;
 900         spin_unlock(&export->exp_lock);
 901         class_import_put(revimp);
 902
 903         return 0;
 904 }
 905 EXPORT_SYMBOL(rev_import_init);
 906
 907 /**
 908  * Handle reconnect for an export.
 909  *
 910  * \param exp export to handle reconnect process
 911  * \param req client reconnect request
 912  *
 913  * \retval -rc in case securitfy flavor can't be changed
 914  * \retval 0 in case none problems
 915  */
 916 static int rev_import_reconnect(struct obd_export *exp,
 917                                 struct ptlrpc_request *req)
 918 {
 919         struct obd_import *revimp = exp->exp_imp_reverse;
 920         struct lustre_handle *lh;
 921         int rc;
 922
 923         /* avoid sending a request until import flags are changed */
 924         ptlrpc_import_enter_resend(revimp);
 925
 926         if (revimp->imp_connection != NULL)
 927                 ptlrpc_connection_put(revimp->imp_connection);
 928
 929         /*
 930          * client from recovery don't have a handle so we need to take from
 931          * request. it may produce situation when wrong client connected
 932          * to recovery as we trust a client uuid
 933          */
 934         lh = req_capsule_client_get(&req->rq_pill, &RMF_CONN);
 935         revimp->imp_remote_handle = *lh;
 936
 937         /* unknown versions will be caught in
 938          * ptlrpc_handle_server_req_in->lustre_unpack_msg() */
 939         revimp->imp_msg_magic = req->rq_reqmsg->lm_magic;
 940
 941         revimp->imp_connection = ptlrpc_connection_addref(exp->exp_connection);
 942
 943         rc = rev_import_flags_update(revimp, req);
 944         if (rc != 0) {
 945                 /* it is safe to still be in RECOVERY phase as we are not able
 946                  * to setup correct security flavor so requests are not able to
 947                  * be delivered correctly */
 948                 return rc;
 949         }
 950
 951         /* resend all rpc's via new connection */
 952         return ptlrpc_import_recovery_state_machine(revimp);
 953 }
 954
 955 int target_handle_connect(struct ptlrpc_request *req)
 956 {
 957         struct obd_device *target = NULL;
 958         struct obd_export *export = NULL;
 959         /* connect handle - filled from target_handle_reconnect in
 960          * reconnect case */
 961         struct lustre_handle conn;
 962         struct lustre_handle *tmp;
 963         struct obd_uuid cluuid;
 964         char *str;
 965         int rc = 0;
 966         char *target_start;
 967         int target_len;
 968         bool     mds_conn = false, lw_client = false, initial_conn = false;
 969         bool     mds_mds_conn = false;
 970         bool     new_mds_mds_conn = false;
 971         struct obd_connect_data *data, *tmpdata;
 972         int size, tmpsize;
 973         lnet_nid_t *client_nid = NULL;
 974         ENTRY;
 975
 976         OBD_RACE(OBD_FAIL_TGT_CONN_RACE);
 977
 978         str = req_capsule_client_get(&req->rq_pill, &RMF_TGTUUID);
 979         if (str == NULL) {
 980                 DEBUG_REQ(D_ERROR, req, "bad target UUID for connect");
 981                 GOTO(out, rc = -EINVAL);
 982         }
 983
 984         target = class_dev_by_str(str);
 985         if (!target) {
 986                 deuuidify(str, NULL, &target_start, &target_len);
 987                 LCONSOLE_ERROR_MSG(0x137, "%s: not available for connect "
 988                                    "from %s (no target). If you are running "
 989                                    "an HA pair check that the target is "
 990                                    "mounted on the other server.\n", str,
 991                                    libcfs_nid2str(req->rq_peer.nid));
 992                 GOTO(out, rc = -ENODEV);
 993         }
 994
 995         spin_lock(&target->obd_dev_lock);
 996
 997         target->obd_conn_inprogress++;
 998
 999         if (target->obd_stopping || !target->obd_set_up) {
1000                 spin_unlock(&target->obd_dev_lock);
1001
1002                 deuuidify(str, NULL, &target_start, &target_len);
1003                 LCONSOLE_INFO("%.*s: Not available for connect from %s (%s)\n",
1004                               target_len, target_start,
1005                               libcfs_nid2str(req->rq_peer.nid),
1006                               (target->obd_stopping ?
1007                                "stopping" : "not set up"));
1008                 GOTO(out, rc = -ENODEV);
1009         }
1010
1011         if (target->obd_no_conn) {
1012                 spin_unlock(&target->obd_dev_lock);
1013
1014                 CDEBUG(D_INFO, "%s: Temporarily refusing client connection "
1015                                "from %s\n", target->obd_name,
1016                                libcfs_nid2str(req->rq_peer.nid));
1017                 GOTO(out, rc = -EAGAIN);
1018         }
1019
1020         spin_unlock(&target->obd_dev_lock);
1021
1022         str = req_capsule_client_get(&req->rq_pill, &RMF_CLUUID);
1023         if (str == NULL) {
1024                 DEBUG_REQ(D_ERROR, req, "bad client UUID for connect");
1025                 GOTO(out, rc = -EINVAL);
1026         }
1027
1028         obd_str2uuid(&cluuid, str);
1029
1030         tmp = req_capsule_client_get(&req->rq_pill, &RMF_CONN);
1031         if (tmp == NULL)
1032                 GOTO(out, rc = -EPROTO);
1033
1034         conn = *tmp;
1035
1036         size = req_capsule_get_size(&req->rq_pill, &RMF_CONNECT_DATA,
1037                                     RCL_CLIENT);
1038         data = req_capsule_client_get(&req->rq_pill, &RMF_CONNECT_DATA);
1039         if (!data)
1040                 GOTO(out, rc = -EPROTO);
1041
1042         rc = req_capsule_server_pack(&req->rq_pill);
1043         if (rc)
1044                 GOTO(out, rc);
1045
1046 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(3, 0, 53, 0)
1047         /* Don't allow clients to connect that are using old 1.8 format
1048          * protocol conventions (LUSTRE_MSG_MAGIC_v1, !MSGHDR_CKSUM_INCOMPAT18,
1049          * ldlm_flock_policy_wire format, MDT_ATTR_xTIME_SET, etc).  The
1050          * FULL20 flag should be set on all connections since 2.0, but no
1051          * longer affects behaviour.
1052          *
1053          * Later this check will be disabled and the flag can be retired
1054          * completely once interop with 3.0 is no longer needed.
1055          */
1056         if (!(data->ocd_connect_flags & OBD_CONNECT_FULL20))
1057                 GOTO(out, rc = -EPROTO);
1058
1059         /* Don't allow liblustre clients to connect.
1060          * - testing was disabled in v2_2_50_0-61-g6a75d65
1061          * - building was disabled in v2_5_58_0-28-g7277179
1062          * - client code was deleted in v2_6_50_0-101-gcdfbc72,
1063          * - clients were refused connect for version difference > 0.0.1.32  */
1064         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
1065                 DEBUG_REQ(D_WARNING, req, "Refusing libclient connection");
1066                 GOTO(out, rc = -EPROTO);
1067         }
1068 #endif
1069
1070         /* Note: lw_client is needed in MDS-MDS failover during update log
1071          * processing, so we needs to allow lw_client to be connected at
1072          * anytime, instead of only the initial connection */
1073         lw_client = (data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT) != 0;
1074
1075         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_INITIAL) {
1076                 initial_conn = true;
1077                 mds_conn = (data->ocd_connect_flags & OBD_CONNECT_MDS) != 0;
1078                 mds_mds_conn = (data->ocd_connect_flags &
1079                                 OBD_CONNECT_MDS_MDS) != 0;
1080
1081                 /* OBD_CONNECT_MNE_SWAB is defined as OBD_CONNECT_MDS_MDS
1082                  * for Imperative Recovery connection from MGC to MGS.
1083                  *
1084                  * Via check OBD_CONNECT_FID, we can distinguish whether
1085                  * the OBD_CONNECT_MDS_MDS/OBD_CONNECT_MNE_SWAB is from
1086                  * MGC or MDT, since MGC does not use OBD_CONNECT_FID.
1087                  */
1088                 if (!lw_client &&
1089                     (data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
1090                     (data->ocd_connect_flags & OBD_CONNECT_FID) &&
1091                     (data->ocd_connect_flags & OBD_CONNECT_VERSION)) {
1092                         __u32 major = OBD_OCD_VERSION_MAJOR(data->ocd_version);
1093                         __u32 minor = OBD_OCD_VERSION_MINOR(data->ocd_version);
1094                         __u32 patch = OBD_OCD_VERSION_PATCH(data->ocd_version);
1095
1096                         /* We do not support the MDT-MDT interoperations with
1097                          * different version MDT because of protocol changes. */
1098                         if (unlikely(major != LUSTRE_MAJOR ||
1099                                      minor != LUSTRE_MINOR ||
1100                                      abs(patch - LUSTRE_PATCH) > 3)) {
1101                                 LCONSOLE_WARN("%s (%u.%u.%u.%u) refused the "
1102                                         "connection from different version MDT "
1103                                         "(%d.%d.%d.%d) %s %s\n",
1104                                         target->obd_name, LUSTRE_MAJOR,
1105                                         LUSTRE_MINOR, LUSTRE_PATCH, LUSTRE_FIX,
1106                                         major, minor, patch,
1107                                         OBD_OCD_VERSION_FIX(data->ocd_version),
1108                                         libcfs_nid2str(req->rq_peer.nid), str);
1109
1110                                 GOTO(out, rc = -EPROTO);
1111                         }
1112                 }
1113         }
1114
1115         /* lctl gets a backstage, all-access pass. */
1116         if (obd_uuid_equals(&cluuid, &target->obd_uuid))
1117                 goto dont_check_exports;
1118
1119         export = cfs_hash_lookup(target->obd_uuid_hash, &cluuid);
1120         if (!export)
1121                 goto no_export;
1122
1123         /* We've found an export in the hash. */
1124
1125         spin_lock(&export->exp_lock);
1126
1127         if (export->exp_connecting) { /* bug 9635, et. al. */
1128                 spin_unlock(&export->exp_lock);
1129                 LCONSOLE_WARN("%s: Export %p already connecting from %s\n",
1130                               export->exp_obd->obd_name, export,
1131                               libcfs_nid2str(req->rq_peer.nid));
1132                 class_export_put(export);
1133                 export = NULL;
1134                 rc = -EALREADY;
1135         } else if ((mds_conn || (lw_client && initial_conn) ||
1136                    data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
1137                    export->exp_connection != NULL) {
1138                 spin_unlock(&export->exp_lock);
1139                 if (req->rq_peer.nid != export->exp_connection->c_peer.nid) {
1140                         /* MDS or LWP reconnected after failover. */
1141                         LCONSOLE_WARN("%s: Received %s connection from "
1142                             "%s, removing former export from %s\n",
1143                             target->obd_name, mds_conn ? "MDS" : "LWP",
1144                             libcfs_nid2str(req->rq_peer.nid),
1145                             libcfs_nid2str(export->exp_connection->c_peer.nid));
1146                 } else {
1147                         /* New MDS connection from the same NID. */
1148                         LCONSOLE_WARN("%s: Received new %s connection from "
1149                                 "%s, removing former export from same NID\n",
1150                                 target->obd_name, mds_conn ? "MDS" : "LWP",
1151                                 libcfs_nid2str(req->rq_peer.nid));
1152                 }
1153
1154                 if (req->rq_peer.nid == export->exp_connection->c_peer.nid &&
1155                     data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) {
1156                         /* Because exports between MDTs will always be
1157                          * kept, let's do not fail such export if they
1158                          * come from the same NID, otherwise it might
1159                          * cause eviction between MDTs, which might
1160                          * cause namespace inconsistency */
1161                         spin_lock(&export->exp_lock);
1162                         export->exp_connecting = 1;
1163                         spin_unlock(&export->exp_lock);
1164                         conn.cookie = export->exp_handle.h_cookie;
1165                         rc = EALREADY;
1166                 } else {
1167                         class_fail_export(export);
1168                         class_export_put(export);
1169                         export = NULL;
1170                         rc = 0;
1171                 }
1172         } else if (export->exp_connection != NULL && initial_conn &&
1173                    req->rq_peer.nid != export->exp_connection->c_peer.nid) {
1174                 spin_unlock(&export->exp_lock);
1175                 /* In MDS failover we have static UUID but NID can change. */
1176                 LCONSOLE_WARN("%s: Client %s seen on new nid %s when "
1177                               "existing nid %s is already connected\n",
1178                               target->obd_name, cluuid.uuid,
1179                               libcfs_nid2str(req->rq_peer.nid),
1180                               libcfs_nid2str(
1181                                       export->exp_connection->c_peer.nid));
1182                 rc = -EALREADY;
1183                 class_export_put(export);
1184                 export = NULL;
1185         } else {
1186                 export->exp_connecting = 1;
1187                 spin_unlock(&export->exp_lock);
1188                 LASSERT(export->exp_obd == target);
1189
1190                 rc = target_handle_reconnect(&conn, export, &cluuid);
1191         }
1192
1193         /* If we found an export, we already unlocked. */
1194         if (!export) {
1195 no_export:
1196                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_CONNECT, 2 * obd_timeout);
1197         } else if (req->rq_export == NULL &&
1198                    atomic_read(&export->exp_rpc_count) > 0) {
1199                 LCONSOLE_WARN("%s: Client %s (at %s) refused connection, "
1200                               "still busy with %d references\n",
1201                               target->obd_name, cluuid.uuid,
1202                               libcfs_nid2str(req->rq_peer.nid),
1203                               atomic_read(&export->exp_refcount));
1204                 GOTO(out, rc = -EBUSY);
1205         } else if (lustre_msg_get_conn_cnt(req->rq_reqmsg) == 1) {
1206                 if (!strstr(cluuid.uuid, "mdt"))
1207                         LCONSOLE_WARN("%s: Rejecting reconnect from the "
1208                                       "known client %s (at %s) because it "
1209                                       "is indicating it is a new client",
1210                                       target->obd_name, cluuid.uuid,
1211                                       libcfs_nid2str(req->rq_peer.nid));
1212                 GOTO(out, rc = -EALREADY);
1213         } else {
1214                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_RECONNECT, 2 * obd_timeout);
1215         }
1216
1217         if (rc < 0) {
1218                 GOTO(out, rc);
1219         }
1220
1221         CDEBUG(D_HA, "%s: connection from %s@%s %st%llu exp %p cur %lld last %lld\n",
1222                target->obd_name, cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
1223                target->obd_recovering ? "recovering/" : "", data->ocd_transno,
1224                export, ktime_get_real_seconds(),
1225                export ? export->exp_last_request_time : 0);
1226
1227         /* If this is the first time a client connects, reset the recovery
1228          * timer. Discard lightweight connections which might be local. */
1229         if (!lw_client && rc == 0 && target->obd_recovering)
1230                 check_and_start_recovery_timer(target, req, export == NULL);
1231
1232         /* We want to handle EALREADY but *not* -EALREADY from
1233          * target_handle_reconnect(), return reconnection state in a flag. */
1234         if (rc == EALREADY) {
1235                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECONNECT);
1236                 rc = 0;
1237         } else {
1238                 LASSERT(rc == 0);
1239         }
1240
1241         /* Tell the client if we support replayable requests. */
1242         if (target->obd_replayable)
1243                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
1244         client_nid = &req->rq_peer.nid;
1245
1246         if (export == NULL) {
1247                 /* allow lightweight connections during recovery */
1248                 /* allow "new" MDT to be connected during recovery, since we
1249                  * need retrieve recovery update records from it */
1250                 if (target->obd_recovering && !lw_client && !mds_mds_conn) {
1251                         time64_t t;
1252                         int c; /* connected */
1253                         int i; /* in progress */
1254                         int k; /* known */
1255                         int s; /* stale/evicted */
1256
1257                         c = atomic_read(&target->obd_connected_clients);
1258                         i = atomic_read(&target->obd_lock_replay_clients);
1259                         k = target->obd_max_recoverable_clients;
1260                         s = target->obd_stale_clients;
1261                         t = jiffies_to_msecs(target->obd_recovery_timer.expires);
1262                         t /= MSEC_PER_SEC;
1263                         t -= ktime_get_seconds();
1264                         LCONSOLE_WARN("%s: Denying connection for new client %s"
1265                                       "(at %s), waiting for %d known clients "
1266                                       "(%d recovered, %d in progress, and %d "
1267                                       "evicted) to recover in %lld:%.02lld\n",
1268                                       target->obd_name, cluuid.uuid,
1269                                       libcfs_nid2str(req->rq_peer.nid), k,
1270                                       c - i, i, s, t / 60, t % 60);
1271                         rc = -EBUSY;
1272                 } else {
1273 dont_check_exports:
1274                         rc = obd_connect(req->rq_svc_thread->t_env,
1275                                          &export, target, &cluuid, data,
1276                                          client_nid);
1277                         if (mds_conn && OBD_FAIL_CHECK(OBD_FAIL_TGT_RCVG_FLAG))
1278                                 lustre_msg_add_op_flags(req->rq_repmsg,
1279                                                         MSG_CONNECT_RECOVERING);
1280                         if (rc == 0) {
1281                                 conn.cookie = export->exp_handle.h_cookie;
1282                                 rc = rev_import_init(export);
1283                         }
1284
1285                         if (mds_mds_conn)
1286                                 new_mds_mds_conn = true;
1287                 }
1288         } else {
1289                 rc = obd_reconnect(req->rq_svc_thread->t_env,
1290                                    export, target, &cluuid, data, client_nid);
1291         }
1292         if (rc)
1293                 GOTO(out, rc);
1294
1295         LASSERT(target->u.obt.obt_magic == OBT_MAGIC);
1296         data->ocd_instance = target->u.obt.obt_instance;
1297
1298         /* Return only the parts of obd_connect_data that we understand, so the
1299          * client knows that we don't understand the rest. */
1300         if (data) {
1301                 tmpsize = req_capsule_get_size(&req->rq_pill, &RMF_CONNECT_DATA,
1302                                                RCL_SERVER);
1303                 tmpdata = req_capsule_server_get(&req->rq_pill,
1304                                                  &RMF_CONNECT_DATA);
1305                 /* Don't use struct assignment here, because the client reply
1306                  * buffer may be smaller/larger than the local struct
1307                  * obd_connect_data. */
1308                 memcpy(tmpdata, data, min(tmpsize, size));
1309         }
1310
1311         /* If the client and the server are the same node, we will already
1312          * have an export that really points to the client's DLM export,
1313          * because we have a shared handles table.
1314          *
1315          * XXX this will go away when shaver stops sending the "connect" handle
1316          * in the real "remote handle" field of the request --phik 24 Apr 2003
1317          */
1318         ptlrpc_request_change_export(req, export);
1319
1320         spin_lock(&export->exp_lock);
1321         if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
1322                 spin_unlock(&export->exp_lock);
1323                 CDEBUG(D_RPCTRACE, "%s: %s already connected at greater "
1324                        "or equal conn_cnt: %d >= %d\n",
1325                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
1326                        export->exp_conn_cnt,
1327                        lustre_msg_get_conn_cnt(req->rq_reqmsg));
1328
1329                 GOTO(out, rc = -EALREADY);
1330         }
1331         LASSERT(lustre_msg_get_conn_cnt(req->rq_reqmsg) > 0);
1332         export->exp_conn_cnt = lustre_msg_get_conn_cnt(req->rq_reqmsg);
1333         spin_unlock(&export->exp_lock);
1334
1335         if (export->exp_connection != NULL) {
1336                 /* Check to see if connection came from another NID. */
1337                 if ((export->exp_connection->c_peer.nid != req->rq_peer.nid) &&
1338                     !hlist_unhashed(&export->exp_nid_hash))
1339                         cfs_hash_del(export->exp_obd->obd_nid_hash,
1340                                      &export->exp_connection->c_peer.nid,
1341                                      &export->exp_nid_hash);
1342
1343                 ptlrpc_connection_put(export->exp_connection);
1344         }
1345
1346         export->exp_connection = ptlrpc_connection_get(req->rq_peer,
1347                                                        req->rq_self,
1348                                                        &cluuid);
1349         if (hlist_unhashed(&export->exp_nid_hash))
1350                 cfs_hash_add(export->exp_obd->obd_nid_hash,
1351                              &export->exp_connection->c_peer.nid,
1352                              &export->exp_nid_hash);
1353
1354         lustre_msg_set_handle(req->rq_repmsg, &conn);
1355
1356         rc = rev_import_reconnect(export, req);
1357         if (rc != 0)
1358                 GOTO(out, rc);
1359
1360         if (target->obd_recovering && !export->exp_in_recovery && !lw_client) {
1361                 int has_transno;
1362                 __u64 transno = data->ocd_transno;
1363
1364                 spin_lock(&export->exp_lock);
1365                 /* possible race with class_disconnect_stale_exports,
1366                  * export may be already in the eviction process */
1367                 if (export->exp_failed) {
1368                         spin_unlock(&export->exp_lock);
1369                         GOTO(out, rc = -ENODEV);
1370                 }
1371                 export->exp_in_recovery = 1;
1372                 export->exp_req_replay_needed = 1;
1373                 export->exp_lock_replay_needed = 1;
1374                 spin_unlock(&export->exp_lock);
1375
1376                 has_transno = !!(lustre_msg_get_op_flags(req->rq_reqmsg) &
1377                                  MSG_CONNECT_TRANSNO);
1378                 if (has_transno && transno == 0)
1379                         CWARN("Connect with zero transno!\n");
1380
1381                 if (has_transno && transno > 0 &&
1382                     transno < target->obd_next_recovery_transno &&
1383                     transno > target->obd_last_committed) {
1384                         /* Another way is to use cmpxchg() to be lock-free. */
1385                         spin_lock(&target->obd_recovery_task_lock);
1386                         if (transno < target->obd_next_recovery_transno)
1387                                 target->obd_next_recovery_transno = transno;
1388                         spin_unlock(&target->obd_recovery_task_lock);
1389                 }
1390
1391                 atomic_inc(&target->obd_req_replay_clients);
1392                 atomic_inc(&target->obd_lock_replay_clients);
1393                 /* Note: MDS-MDS connection is allowed to be connected during
1394                  * recovery, no matter if the exports needs to be recoveried.
1395                  * Because we need retrieve updates logs from all other MDTs.
1396                  * So if the MDS-MDS export is new, obd_max_recoverable_clients
1397                  * also needs to be increased to match other recovery checking
1398                  * condition. */
1399                 if (new_mds_mds_conn)
1400                         target->obd_max_recoverable_clients++;
1401                 if (atomic_inc_return(&target->obd_connected_clients) ==
1402                     target->obd_max_recoverable_clients)
1403                         wake_up(&target->obd_next_transno_waitq);
1404         }
1405
1406         /* Tell the client we're in recovery, when client is involved in it. */
1407         if (target->obd_recovering && !lw_client)
1408                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
1409
1410 out:
1411         if (export) {
1412                 spin_lock(&export->exp_lock);
1413                 export->exp_connecting = 0;
1414                 spin_unlock(&export->exp_lock);
1415
1416                 class_export_put(export);
1417         }
1418         if (target != NULL) {
1419                 spin_lock(&target->obd_dev_lock);
1420                 target->obd_conn_inprogress--;
1421                 spin_unlock(&target->obd_dev_lock);
1422                 class_decref(target, "find", current);
1423         }
1424         req->rq_status = rc;
1425         RETURN(rc);
1426 }
1427
1428 int target_handle_disconnect(struct ptlrpc_request *req)
1429 {
1430         int rc;
1431         ENTRY;
1432
1433         rc = req_capsule_server_pack(&req->rq_pill);
1434         if (rc)
1435                 RETURN(rc);
1436
1437         /* Keep the rq_export around so we can send the reply. */
1438         req->rq_status = obd_disconnect(class_export_get(req->rq_export));
1439
1440         RETURN(0);
1441 }
1442
1443 void target_destroy_export(struct obd_export *exp)
1444 {
1445         struct obd_import       *imp = NULL;
1446         /* exports created from last_rcvd data, and "fake"
1447            exports created by lctl don't have an import */
1448         spin_lock(&exp->exp_lock);
1449         if (exp->exp_imp_reverse != NULL) {
1450                 imp = exp->exp_imp_reverse;
1451                 exp->exp_imp_reverse = NULL;
1452         }
1453         spin_unlock(&exp->exp_lock);
1454         if (imp != NULL)
1455                 client_destroy_import(imp);
1456
1457         LASSERT_ATOMIC_ZERO(&exp->exp_locks_count);
1458         LASSERT_ATOMIC_ZERO(&exp->exp_rpc_count);
1459         LASSERT_ATOMIC_ZERO(&exp->exp_cb_count);
1460         LASSERT_ATOMIC_ZERO(&exp->exp_replay_count);
1461 }
1462 EXPORT_SYMBOL(target_destroy_export);
1463
1464 /*
1465  * Recovery functions
1466  */
1467 static void target_request_copy_get(struct ptlrpc_request *req)
1468 {
1469         class_export_rpc_inc(req->rq_export);
1470         LASSERT(list_empty(&req->rq_list));
1471         INIT_LIST_HEAD(&req->rq_replay_list);
1472
1473         /* Increase refcount to keep request in queue. */
1474         atomic_inc(&req->rq_refcount);
1475         /* Let export know it has replays to be handled. */
1476         atomic_inc(&req->rq_export->exp_replay_count);
1477 }
1478
1479 static void target_request_copy_put(struct ptlrpc_request *req)
1480 {
1481         LASSERT(list_empty(&req->rq_replay_list));
1482         LASSERT_ATOMIC_POS(&req->rq_export->exp_replay_count);
1483
1484         atomic_dec(&req->rq_export->exp_replay_count);
1485         class_export_rpc_dec(req->rq_export);
1486         ptlrpc_server_drop_request(req);
1487 }
1488
1489 static int target_exp_enqueue_req_replay(struct ptlrpc_request *req)
1490 {
1491         __u64                  transno = lustre_msg_get_transno(req->rq_reqmsg);
1492         struct obd_export     *exp = req->rq_export;
1493         struct ptlrpc_request *reqiter;
1494         struct ptlrpc_request *dup_req = NULL;
1495         int                    dup = 0;
1496
1497         LASSERT(exp);
1498
1499         spin_lock(&exp->exp_lock);
1500         list_for_each_entry(reqiter, &exp->exp_req_replay_queue,
1501                                 rq_replay_list) {
1502                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) == transno) {
1503                         dup_req = reqiter;
1504                         dup = 1;
1505                         break;
1506                 }
1507         }
1508
1509         if (dup) {
1510                 /* We expect it with RESENT and REPLAY flags. */
1511                 if ((lustre_msg_get_flags(req->rq_reqmsg) &
1512                      (MSG_RESENT | MSG_REPLAY)) != (MSG_RESENT | MSG_REPLAY))
1513                         CERROR("invalid flags %x of resent replay\n",
1514                                lustre_msg_get_flags(req->rq_reqmsg));
1515
1516                 if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
1517                         __u32 new_conn;
1518
1519                         new_conn = lustre_msg_get_conn_cnt(req->rq_reqmsg);
1520                         if (new_conn >
1521                             lustre_msg_get_conn_cnt(dup_req->rq_reqmsg))
1522                                 lustre_msg_set_conn_cnt(dup_req->rq_reqmsg,
1523                                                         new_conn);
1524                 }
1525         } else {
1526                 list_add_tail(&req->rq_replay_list,
1527                                   &exp->exp_req_replay_queue);
1528         }
1529
1530         spin_unlock(&exp->exp_lock);
1531         return dup;
1532 }
1533
1534 static void target_exp_dequeue_req_replay(struct ptlrpc_request *req)
1535 {
1536         LASSERT(!list_empty(&req->rq_replay_list));
1537         LASSERT(req->rq_export);
1538
1539         spin_lock(&req->rq_export->exp_lock);
1540         list_del_init(&req->rq_replay_list);
1541         spin_unlock(&req->rq_export->exp_lock);
1542 }
1543
1544 static void target_finish_recovery(struct lu_target *lut)
1545 {
1546         struct obd_device *obd = lut->lut_obd;
1547         ENTRY;
1548
1549         /* Only log a recovery message when recovery has occurred. */
1550         if (obd->obd_recovery_start) {
1551                 time64_t now = ktime_get_real_seconds();
1552                 time64_t elapsed_time;
1553
1554                 elapsed_time = max_t(time64_t, now - obd->obd_recovery_start, 1);
1555                 LCONSOLE_INFO("%s: Recovery over after %lld:%.02lld, of %d clients "
1556                         "%d recovered and %d %s evicted.\n", obd->obd_name,
1557                         (s64)elapsed_time / 60, (s64)elapsed_time % 60,
1558                         obd->obd_max_recoverable_clients,
1559                         atomic_read(&obd->obd_connected_clients),
1560                         obd->obd_stale_clients,
1561                         obd->obd_stale_clients == 1 ? "was" : "were");
1562         }
1563
1564         ldlm_reprocess_recovery_done(obd->obd_namespace);
1565         spin_lock(&obd->obd_recovery_task_lock);
1566         if (!list_empty(&obd->obd_req_replay_queue) ||
1567             !list_empty(&obd->obd_lock_replay_queue) ||
1568             !list_empty(&obd->obd_final_req_queue)) {
1569                 CERROR("%s: Recovery queues ( %s%s%s) are not empty\n",
1570                        obd->obd_name,
1571                        list_empty(&obd->obd_req_replay_queue) ? "" : "req ",
1572                        list_empty(&obd->obd_lock_replay_queue) ? \
1573                                "" : "lock ",
1574                        list_empty(&obd->obd_final_req_queue) ? \
1575                                "" : "final ");
1576                 spin_unlock(&obd->obd_recovery_task_lock);
1577                 LBUG();
1578         }
1579         spin_unlock(&obd->obd_recovery_task_lock);
1580
1581         obd->obd_recovery_end = ktime_get_real_seconds();
1582
1583         /* When recovery finished, cleanup orphans on MDS and OST. */
1584         if (obd->obd_type && OBP(obd, postrecov)) {
1585                 int rc = OBP(obd, postrecov)(obd);
1586
1587                 if (rc < 0)
1588                         LCONSOLE_WARN("%s: Post recovery failed, rc %d\n",
1589                                       obd->obd_name, rc);
1590         }
1591         EXIT;
1592 }
1593
1594 static void abort_req_replay_queue(struct obd_device *obd)
1595 {
1596         struct ptlrpc_request *req, *n;
1597         struct list_head abort_list;
1598
1599         INIT_LIST_HEAD(&abort_list);
1600         spin_lock(&obd->obd_recovery_task_lock);
1601         list_splice_init(&obd->obd_req_replay_queue, &abort_list);
1602         spin_unlock(&obd->obd_recovery_task_lock);
1603         list_for_each_entry_safe(req, n, &abort_list, rq_list) {
1604                 DEBUG_REQ(D_WARNING, req, "aborted:");
1605                 req->rq_status = -ENOTCONN;
1606                 if (ptlrpc_error(req)) {
1607                         DEBUG_REQ(D_ERROR, req,
1608                                   "failed abort_req_reply; skipping");
1609                 }
1610                 target_exp_dequeue_req_replay(req);
1611                 target_request_copy_put(req);
1612         }
1613 }
1614
1615 static void abort_lock_replay_queue(struct obd_device *obd)
1616 {
1617         struct ptlrpc_request *req, *n;
1618         struct list_head abort_list;
1619
1620         INIT_LIST_HEAD(&abort_list);
1621         spin_lock(&obd->obd_recovery_task_lock);
1622         list_splice_init(&obd->obd_lock_replay_queue, &abort_list);
1623         spin_unlock(&obd->obd_recovery_task_lock);
1624         list_for_each_entry_safe(req, n, &abort_list, rq_list) {
1625                 DEBUG_REQ(D_ERROR, req, "aborted:");
1626                 req->rq_status = -ENOTCONN;
1627                 if (ptlrpc_error(req)) {
1628                         DEBUG_REQ(D_ERROR, req,
1629                                   "failed abort_lock_reply; skipping");
1630                 }
1631                 target_request_copy_put(req);
1632         }
1633 }
1634
1635 /* Called from a cleanup function if the device is being cleaned up
1636    forcefully.  The exports should all have been disconnected already,
1637    the only thing left to do is
1638      - clear the recovery flags
1639      - cancel the timer
1640      - free queued requests and replies, but don't send replies
1641    Because the obd_stopping flag is set, no new requests should be received.
1642
1643 */
1644 void target_cleanup_recovery(struct obd_device *obd)
1645 {
1646         struct ptlrpc_request *req, *n;
1647         struct list_head clean_list;
1648
1649         INIT_LIST_HEAD(&clean_list);
1650         spin_lock(&obd->obd_dev_lock);
1651         if (!obd->obd_recovering) {
1652                 spin_unlock(&obd->obd_dev_lock);
1653                 EXIT;
1654                 return;
1655         }
1656         obd->obd_recovering = obd->obd_abort_recovery = 0;
1657         spin_unlock(&obd->obd_dev_lock);
1658
1659         spin_lock(&obd->obd_recovery_task_lock);
1660         target_cancel_recovery_timer(obd);
1661         list_splice_init(&obd->obd_req_replay_queue, &clean_list);
1662         spin_unlock(&obd->obd_recovery_task_lock);
1663
1664         list_for_each_entry_safe(req, n, &clean_list, rq_list) {
1665                 LASSERT(req->rq_reply_state == NULL);
1666                 target_exp_dequeue_req_replay(req);
1667                 target_request_copy_put(req);
1668         }
1669
1670         spin_lock(&obd->obd_recovery_task_lock);
1671         list_splice_init(&obd->obd_lock_replay_queue, &clean_list);
1672         list_splice_init(&obd->obd_final_req_queue, &clean_list);
1673         spin_unlock(&obd->obd_recovery_task_lock);
1674
1675         list_for_each_entry_safe(req, n, &clean_list, rq_list) {
1676                 LASSERT(req->rq_reply_state == NULL);
1677                 target_request_copy_put(req);
1678         }
1679
1680         EXIT;
1681 }
1682 EXPORT_SYMBOL(target_cleanup_recovery);
1683
1684 /* obd_recovery_task_lock should be held */
1685 void target_cancel_recovery_timer(struct obd_device *obd)
1686 {
1687         CDEBUG(D_HA, "%s: cancel recovery timer\n", obd->obd_name);
1688         del_timer(&obd->obd_recovery_timer);
1689 }
1690
1691 static void target_start_recovery_timer(struct obd_device *obd)
1692 {
1693         if (obd->obd_recovery_start != 0)
1694                 return;
1695
1696         spin_lock(&obd->obd_dev_lock);
1697         if (!obd->obd_recovering || obd->obd_abort_recovery) {
1698                 spin_unlock(&obd->obd_dev_lock);
1699                 return;
1700         }
1701
1702         LASSERT(obd->obd_recovery_timeout != 0);
1703
1704         if (obd->obd_recovery_start != 0) {
1705                 spin_unlock(&obd->obd_dev_lock);
1706                 return;
1707         }
1708
1709         mod_timer(&obd->obd_recovery_timer,
1710                   jiffies + cfs_time_seconds(obd->obd_recovery_timeout));
1711         obd->obd_recovery_start = ktime_get_real_seconds();
1712         spin_unlock(&obd->obd_dev_lock);
1713
1714         LCONSOLE_WARN("%s: Will be in recovery for at least %llu:%02llu, or until %d client%s reconnect%s\n",
1715                       obd->obd_name,
1716                       obd->obd_recovery_timeout / 60,
1717                       obd->obd_recovery_timeout % 60,
1718                       obd->obd_max_recoverable_clients,
1719                       (obd->obd_max_recoverable_clients == 1) ? "" : "s",
1720                       (obd->obd_max_recoverable_clients == 1) ? "s": "");
1721 }
1722
1723 /**
1724  * extend recovery window.
1725  *
1726  * if @extend is true, extend recovery window to have @drt remaining at least;
1727  * otherwise, make sure the recovery timeout value is not less than @drt.
1728  */
1729 static void extend_recovery_timer(struct obd_device *obd, time64_t drt,
1730                                   bool extend)
1731 {
1732         time64_t now;
1733         time64_t end;
1734         time64_t left;
1735         time64_t to;
1736
1737         spin_lock(&obd->obd_dev_lock);
1738         if (!obd->obd_recovering || obd->obd_abort_recovery) {
1739                 spin_unlock(&obd->obd_dev_lock);
1740                 return;
1741         }
1742         LASSERT(obd->obd_recovery_start != 0);
1743
1744         now = ktime_get_real_seconds();
1745         to = obd->obd_recovery_timeout;
1746         end = obd->obd_recovery_start + to;
1747         left = end - now;
1748
1749         if (extend && (drt > left)) {
1750                 to += drt - left;
1751         } else if (!extend && (drt > to)) {
1752                 to = drt;
1753         }
1754
1755         if (to > obd->obd_recovery_time_hard) {
1756                 to = obd->obd_recovery_time_hard;
1757                 CWARN("%s: extended recovery timer reaching hard limit: %lld, extend: %d\n",
1758                       obd->obd_name, to, extend);
1759         }
1760
1761         if (obd->obd_recovery_timeout < to) {
1762                 obd->obd_recovery_timeout = to;
1763                 end = obd->obd_recovery_start + to;
1764                 mod_timer(&obd->obd_recovery_timer,
1765                           jiffies + cfs_time_seconds(end - now));
1766         }
1767         spin_unlock(&obd->obd_dev_lock);
1768
1769         CDEBUG(D_HA, "%s: recovery timer will expire in %lld seconds\n",
1770                 obd->obd_name, (s64)(end - now));
1771 }
1772
1773 /* Reset the timer with each new client connection */
1774 /*
1775  * This timer is actually reconnect_timer, which is for making sure
1776  * the total recovery window is at least as big as my reconnect
1777  * attempt timing. So the initial recovery time_out will be set to
1778  * OBD_RECOVERY_FACTOR * obd_timeout. If the timeout coming
1779  * from client is bigger than this, then the recovery time_out will
1780  * be extended to make sure the client could be reconnected, in the
1781  * process, the timeout from the new client should be ignored.
1782  */
1783
1784 static void
1785 check_and_start_recovery_timer(struct obd_device *obd,
1786                                struct ptlrpc_request *req,
1787                                int new_client)
1788 {
1789         time64_t service_time = lustre_msg_get_service_time(req->rq_reqmsg);
1790         struct obd_device_target *obt = &obd->u.obt;
1791
1792         if (!new_client && service_time)
1793                 /* Teach server about old server's estimates, as first guess
1794                  * at how long new requests will take. */
1795                 at_measured(&req->rq_rqbd->rqbd_svcpt->scp_at_estimate,
1796                             service_time);
1797
1798         target_start_recovery_timer(obd);
1799
1800         /* Convert the service time to RPC timeout,
1801          * and reuse service_time to limit stack usage.
1802          */
1803         service_time = at_est2timeout(service_time);
1804
1805         if (OBD_FAIL_CHECK(OBD_FAIL_TGT_SLUGGISH_NET) &&
1806             service_time < at_extra)
1807                 service_time = at_extra;
1808
1809         /* We expect other clients to timeout within service_time, then try
1810          * to reconnect, then try the failover server.  The max delay between
1811          * connect attempts is SWITCH_MAX + SWITCH_INC + INITIAL. */
1812         service_time += 2 * INITIAL_CONNECT_TIMEOUT;
1813
1814         LASSERT(obt->obt_magic == OBT_MAGIC);
1815         service_time += 2 * (CONNECTION_SWITCH_MAX + CONNECTION_SWITCH_INC);
1816         if (service_time > obd->obd_recovery_timeout && !new_client)
1817                 extend_recovery_timer(obd, service_time, false);
1818 }
1819
1820 /** Health checking routines */
1821 static inline int exp_connect_healthy(struct obd_export *exp)
1822 {
1823         return (exp->exp_in_recovery);
1824 }
1825
1826 /** if export done req_replay or has replay in queue */
1827 static inline int exp_req_replay_healthy(struct obd_export *exp)
1828 {
1829         return (!exp->exp_req_replay_needed ||
1830                 atomic_read(&exp->exp_replay_count) > 0);
1831 }
1832
1833
1834 static inline int exp_req_replay_healthy_or_from_mdt(struct obd_export *exp)
1835 {
1836         return (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) ||
1837                exp_req_replay_healthy(exp);
1838 }
1839
1840 /** if export done lock_replay or has replay in queue */
1841 static inline int exp_lock_replay_healthy(struct obd_export *exp)
1842 {
1843         return (!exp->exp_lock_replay_needed ||
1844                 atomic_read(&exp->exp_replay_count) > 0);
1845 }
1846
1847 static inline int exp_vbr_healthy(struct obd_export *exp)
1848 {
1849         return (!exp->exp_vbr_failed);
1850 }
1851
1852 static inline int exp_finished(struct obd_export *exp)
1853 {
1854         return (exp->exp_in_recovery && !exp->exp_lock_replay_needed);
1855 }
1856
1857 static inline int exp_finished_or_from_mdt(struct obd_export *exp)
1858 {
1859         return (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) ||
1860                 exp_finished(exp);
1861 }
1862
1863 static int check_for_next_transno(struct lu_target *lut)
1864 {
1865         struct ptlrpc_request *req = NULL;
1866         struct obd_device *obd = lut->lut_obd;
1867         struct target_distribute_txn_data *tdtd = lut->lut_tdtd;
1868         int wake_up = 0, connected, completed, queue_len;
1869         __u64 req_transno = 0;
1870         __u64 update_transno = 0;
1871         __u64 next_transno = 0;
1872         ENTRY;
1873
1874         spin_lock(&obd->obd_recovery_task_lock);
1875         if (!list_empty(&obd->obd_req_replay_queue)) {
1876                 req = list_entry(obd->obd_req_replay_queue.next,
1877                                      struct ptlrpc_request, rq_list);
1878                 req_transno = lustre_msg_get_transno(req->rq_reqmsg);
1879         }
1880
1881         if (tdtd != NULL)
1882                 update_transno = distribute_txn_get_next_transno(tdtd);
1883
1884         connected = atomic_read(&obd->obd_connected_clients);
1885         completed = connected - atomic_read(&obd->obd_req_replay_clients);
1886         queue_len = obd->obd_requests_queued_for_recovery;
1887         next_transno = obd->obd_next_recovery_transno;
1888
1889         CDEBUG(D_HA, "max: %d, connected: %d, completed: %d, queue_len: %d, "
1890                "req_transno: %llu, next_transno: %llu\n",
1891                obd->obd_max_recoverable_clients, connected, completed,
1892                queue_len, req_transno, next_transno);
1893
1894         if (obd->obd_abort_recovery) {
1895                 CDEBUG(D_HA, "waking for aborted recovery\n");
1896                 wake_up = 1;
1897         } else if (obd->obd_recovery_expired) {
1898                 CDEBUG(D_HA, "waking for expired recovery\n");
1899                 wake_up = 1;
1900         } else if (tdtd != NULL && req != NULL &&
1901                    is_req_replayed_by_update(req)) {
1902                 LASSERTF(req_transno < next_transno, "req_transno %llu"
1903                          "next_transno%llu\n", req_transno, next_transno);
1904                 CDEBUG(D_HA, "waking for duplicate req (%llu)\n",
1905                        req_transno);
1906                 wake_up = 1;
1907         } else if (req_transno == next_transno ||
1908                    (update_transno != 0 && update_transno <= next_transno)) {
1909                 CDEBUG(D_HA, "waking for next (%lld)\n", next_transno);
1910                 wake_up = 1;
1911         } else if (queue_len > 0 &&
1912                    queue_len == atomic_read(&obd->obd_req_replay_clients)) {
1913                 /** handle gaps occured due to lost reply or VBR */
1914                 LASSERTF(req_transno >= next_transno,
1915                          "req_transno: %llu, next_transno: %llu\n",
1916                          req_transno, next_transno);
1917                 CDEBUG(D_HA,
1918                        "%s: waking for gap in transno, VBR is %s (skip: "
1919                        "%lld, ql: %d, comp: %d, conn: %d, next: %lld"
1920                        ", next_update %lld last_committed: %lld)\n",
1921                        obd->obd_name, obd->obd_version_recov ? "ON" : "OFF",
1922                        next_transno, queue_len, completed, connected,
1923                        req_transno, update_transno, obd->obd_last_committed);
1924                 obd->obd_next_recovery_transno = req_transno;
1925                 wake_up = 1;
1926         } else if (atomic_read(&obd->obd_req_replay_clients) == 0) {
1927                 CDEBUG(D_HA, "waking for completed recovery\n");
1928                 wake_up = 1;
1929         } else if (OBD_FAIL_CHECK(OBD_FAIL_MDS_RECOVERY_ACCEPTS_GAPS)) {
1930                 CDEBUG(D_HA, "accepting transno gaps is explicitly allowed"
1931                        " by fail_lock, waking up (%lld)\n", next_transno);
1932                 obd->obd_next_recovery_transno = req_transno;
1933                 wake_up = 1;
1934         }
1935         spin_unlock(&obd->obd_recovery_task_lock);
1936         return wake_up;
1937 }
1938
1939 static int check_for_next_lock(struct lu_target *lut)
1940 {
1941         struct obd_device *obd = lut->lut_obd;
1942         int wake_up = 0;
1943
1944         spin_lock(&obd->obd_recovery_task_lock);
1945         if (!list_empty(&obd->obd_lock_replay_queue)) {
1946                 CDEBUG(D_HA, "waking for next lock\n");
1947                 wake_up = 1;
1948         } else if (atomic_read(&obd->obd_lock_replay_clients) == 0) {
1949                 CDEBUG(D_HA, "waking for completed lock replay\n");
1950                 wake_up = 1;
1951         } else if (obd->obd_abort_recovery) {
1952                 CDEBUG(D_HA, "waking for aborted recovery\n");
1953                 wake_up = 1;
1954         } else if (obd->obd_recovery_expired) {
1955                 CDEBUG(D_HA, "waking for expired recovery\n");
1956                 wake_up = 1;
1957         }
1958         spin_unlock(&obd->obd_recovery_task_lock);
1959
1960         return wake_up;
1961 }
1962
1963 /**
1964  * wait for recovery events,
1965  * check its status with help of check_routine
1966  * evict dead clients via health_check
1967  */
1968 static int target_recovery_overseer(struct lu_target *lut,
1969                                     int (*check_routine)(struct lu_target *),
1970                                     int (*health_check)(struct obd_export *))
1971 {
1972         struct obd_device       *obd = lut->lut_obd;
1973         struct target_distribute_txn_data *tdtd;
1974         time64_t last = 0;
1975         time64_t now;
1976 repeat:
1977         if (obd->obd_recovering && obd->obd_recovery_start == 0) {
1978                 now = ktime_get_seconds();
1979                 if (now - last > 600) {
1980                         LCONSOLE_INFO("%s: in recovery but waiting for "
1981                                       "the first client to connect\n",
1982                                       obd->obd_name);
1983                         last = now;
1984                 }
1985         }
1986         if (obd->obd_recovery_start != 0 && ktime_get_real_seconds() >=
1987               (obd->obd_recovery_start + obd->obd_recovery_time_hard)) {
1988                 __u64 next_update_transno = 0;
1989
1990                 /* Only abort the recovery if there are no update recovery
1991                  * left in the queue */
1992                 spin_lock(&obd->obd_recovery_task_lock);
1993                 if (lut->lut_tdtd != NULL) {
1994                         next_update_transno =
1995                                 distribute_txn_get_next_transno(lut->lut_tdtd);
1996
1997                         tdtd = lut->lut_tdtd;
1998                         /* If next_update_transno == 0, it probably because
1999                          * updatelog retrieve threads did not get any records
2000                          * yet, let's wait those threads stopped */
2001                         if (next_update_transno == 0) {
2002                                 struct l_wait_info lwi = { 0 };
2003
2004                                 l_wait_event(tdtd->tdtd_recovery_threads_waitq,
2005                                        atomic_read(
2006                                        &tdtd->tdtd_recovery_threads_count) == 0,
2007                                        &lwi);
2008
2009                                 next_update_transno =
2010                                         distribute_txn_get_next_transno(
2011                                                                 lut->lut_tdtd);
2012                         }
2013                 }
2014
2015                 if (next_update_transno != 0 && !obd->obd_abort_recovery) {
2016                         obd->obd_next_recovery_transno = next_update_transno;
2017                         spin_unlock(&obd->obd_recovery_task_lock);
2018                         /* Disconnect unfinished exports from clients, and
2019                          * keep connection from MDT to make sure the update
2020                          * recovery will still keep trying until some one
2021                          * manually abort the recovery */
2022                         class_disconnect_stale_exports(obd,
2023                                                 exp_finished_or_from_mdt);
2024                         /* Abort all of replay and replay lock req from
2025                          * clients */
2026                         abort_req_replay_queue(obd);
2027                         abort_lock_replay_queue(obd);
2028                         CDEBUG(D_HA, "%s: there are still update replay (%#llx"
2029                                ")in the queue.\n", obd->obd_name,
2030                                next_update_transno);
2031                 } else {
2032                         obd->obd_abort_recovery = 1;
2033                         spin_unlock(&obd->obd_recovery_task_lock);
2034                         CWARN("%s recovery is aborted by hard timeout\n",
2035                               obd->obd_name);
2036                 }
2037         }
2038
2039         while (wait_event_timeout(obd->obd_next_transno_waitq,
2040                                   check_routine(lut),
2041                                   msecs_to_jiffies(60 * MSEC_PER_SEC)) == 0)
2042                 /* wait indefinitely for event, but don't trigger watchdog */;
2043
2044         if (obd->obd_abort_recovery) {
2045                 CWARN("recovery is aborted, evict exports in recovery\n");
2046                 if (lut->lut_tdtd != NULL) {
2047                         struct l_wait_info lwi = { 0 };
2048
2049                         tdtd = lut->lut_tdtd;
2050                         /* Let's wait all of the update log recovery thread
2051                          * finished */
2052                         l_wait_event(tdtd->tdtd_recovery_threads_waitq,
2053                          atomic_read(&tdtd->tdtd_recovery_threads_count) == 0,
2054                              &lwi);
2055                         /* Then abort the update recovery list */
2056                         dtrq_list_destroy(lut->lut_tdtd);
2057                 }
2058
2059                 /** evict exports which didn't finish recovery yet */
2060                 class_disconnect_stale_exports(obd, exp_finished);
2061                 return 1;
2062         } else if (obd->obd_recovery_expired) {
2063                 obd->obd_recovery_expired = 0;
2064                 /** If some clients died being recovered, evict them */
2065                 LCONSOLE_WARN("%s: recovery is timed out, "
2066                               "evict stale exports\n", obd->obd_name);
2067                 /** evict cexports with no replay in queue, they are stalled */
2068                 class_disconnect_stale_exports(obd, health_check);
2069
2070                 /** continue with VBR */
2071                 spin_lock(&obd->obd_dev_lock);
2072                 obd->obd_version_recov = 1;
2073                 spin_unlock(&obd->obd_dev_lock);
2074                 /**
2075                  * reset timer, recovery will proceed with versions now,
2076                  * timeout is set just to handle reconnection delays
2077                  */
2078                 extend_recovery_timer(obd, RECONNECT_DELAY_MAX, true);
2079                 /** Wait for recovery events again, after evicting bad clients */
2080                 goto repeat;
2081         }
2082         return 0;
2083 }
2084
2085 static struct ptlrpc_request *target_next_replay_lock(struct lu_target *lut)
2086 {
2087         struct obd_device       *obd = lut->lut_obd;
2088         struct ptlrpc_request *req = NULL;
2089
2090         CDEBUG(D_HA, "Waiting for lock\n");
2091         if (target_recovery_overseer(lut, check_for_next_lock,
2092                                      exp_lock_replay_healthy))
2093                 abort_lock_replay_queue(obd);
2094
2095         spin_lock(&obd->obd_recovery_task_lock);
2096         if (!list_empty(&obd->obd_lock_replay_queue)) {
2097                 req = list_entry(obd->obd_lock_replay_queue.next,
2098                                      struct ptlrpc_request, rq_list);
2099                 list_del_init(&req->rq_list);
2100                 spin_unlock(&obd->obd_recovery_task_lock);
2101         } else {
2102                 spin_unlock(&obd->obd_recovery_task_lock);
2103                 LASSERT(list_empty(&obd->obd_lock_replay_queue));
2104                 LASSERT(atomic_read(&obd->obd_lock_replay_clients) == 0);
2105                 /** evict exports failed VBR */
2106                 class_disconnect_stale_exports(obd, exp_vbr_healthy);
2107         }
2108         return req;
2109 }
2110
2111 static struct ptlrpc_request *target_next_final_ping(struct obd_device *obd)
2112 {
2113         struct ptlrpc_request *req = NULL;
2114
2115         spin_lock(&obd->obd_recovery_task_lock);
2116         if (!list_empty(&obd->obd_final_req_queue)) {
2117                 req = list_entry(obd->obd_final_req_queue.next,
2118                                      struct ptlrpc_request, rq_list);
2119                 list_del_init(&req->rq_list);
2120                 spin_unlock(&obd->obd_recovery_task_lock);
2121                 if (req->rq_export->exp_in_recovery) {
2122                         spin_lock(&req->rq_export->exp_lock);
2123                         req->rq_export->exp_in_recovery = 0;
2124                         spin_unlock(&req->rq_export->exp_lock);
2125                 }
2126         } else {
2127                 spin_unlock(&obd->obd_recovery_task_lock);
2128         }
2129         return req;
2130 }
2131
2132 static void handle_recovery_req(struct ptlrpc_thread *thread,
2133                                 struct ptlrpc_request *req,
2134                                 svc_handler_t handler)
2135 {
2136         ENTRY;
2137
2138         /**
2139          * export can be evicted during recovery, no need to handle replays for
2140          * it after that, discard such request silently
2141          */
2142         if (req->rq_export->exp_disconnected)
2143                 RETURN_EXIT;
2144
2145         req->rq_session.lc_thread = thread;
2146         req->rq_svc_thread = thread;
2147         req->rq_svc_thread->t_env->le_ses = &req->rq_session;
2148
2149         /* thread context */
2150         lu_context_enter(&thread->t_env->le_ctx);
2151         (void)handler(req);
2152         lu_context_exit(&thread->t_env->le_ctx);
2153
2154         /* don't reset timer for final stage */
2155         if (!exp_finished(req->rq_export)) {
2156                 time64_t to = obd_timeout;
2157
2158                 /**
2159                  * Add request timeout to the recovery time so next request from
2160                  * this client may come in recovery time
2161                  */
2162                 if (!AT_OFF) {
2163                         struct ptlrpc_service_part *svcpt;
2164
2165                         svcpt = req->rq_rqbd->rqbd_svcpt;
2166                         /* If the server sent early reply for this request,
2167                          * the client will recalculate the timeout according to
2168                          * current server estimate service time, so we will
2169                          * use the maxium timeout here for waiting the client
2170                          * sending the next req */
2171                         to = max((int)at_est2timeout(
2172                                  at_get(&svcpt->scp_at_estimate)),
2173                                  (int)lustre_msg_get_timeout(req->rq_reqmsg));
2174                         /* Add 2 net_latency, one for balance rq_deadline
2175                          * (see ptl_send_rpc), one for resend the req to server,
2176                          * Note: client will pack net_latency in replay req
2177                          * (see ptlrpc_replay_req) */
2178                         to += 2 * lustre_msg_get_service_time(req->rq_reqmsg);
2179                 }
2180                 extend_recovery_timer(class_exp2obd(req->rq_export), to, true);
2181         }
2182         EXIT;
2183 }
2184
2185 /** Checking routines for recovery */
2186 static int check_for_recovery_ready(struct lu_target *lut)
2187 {
2188         struct obd_device *obd = lut->lut_obd;
2189         unsigned int clnts = atomic_read(&obd->obd_connected_clients);
2190
2191         CDEBUG(D_HA, "connected %d stale %d max_recoverable_clients %d"
2192                " abort %d expired %d\n", clnts, obd->obd_stale_clients,
2193                obd->obd_max_recoverable_clients, obd->obd_abort_recovery,
2194                obd->obd_recovery_expired);
2195
2196         if (!obd->obd_abort_recovery && !obd->obd_recovery_expired) {
2197                 LASSERT(clnts <= obd->obd_max_recoverable_clients);
2198                 if (clnts + obd->obd_stale_clients <
2199                     obd->obd_max_recoverable_clients)
2200                         return 0;
2201         }
2202
2203         if (lut->lut_tdtd != NULL) {
2204                 if (!lut->lut_tdtd->tdtd_replay_ready &&
2205                     !obd->obd_abort_recovery) {
2206                         /* Let's extend recovery timer, in case the recovery
2207                          * timer expired, and some clients got evicted */
2208                         extend_recovery_timer(obd, obd->obd_recovery_timeout,
2209                                               true);
2210                         CDEBUG(D_HA, "%s update recovery is not ready, extend recovery %llu\n",
2211                                obd->obd_name, obd->obd_recovery_timeout);
2212                         return 0;
2213                 }
2214         }
2215
2216         return 1;
2217 }
2218
2219 enum {
2220         REQUEST_RECOVERY = 1,
2221         UPDATE_RECOVERY = 2,
2222 };
2223
2224 static __u64 get_next_replay_req_transno(struct obd_device *obd)
2225 {
2226         __u64 transno = 0;
2227
2228         if (!list_empty(&obd->obd_req_replay_queue)) {
2229                 struct ptlrpc_request *req;
2230
2231                 req = list_entry(obd->obd_req_replay_queue.next,
2232                                  struct ptlrpc_request, rq_list);
2233                 transno = lustre_msg_get_transno(req->rq_reqmsg);
2234         }
2235
2236         return transno;
2237 }
2238
2239 static __u64 get_next_transno(struct lu_target *lut, int *type)
2240 {
2241         struct obd_device *obd = lut->lut_obd;
2242         struct target_distribute_txn_data *tdtd = lut->lut_tdtd;
2243         __u64 transno = 0;
2244         __u64 update_transno;
2245         ENTRY;
2246
2247         transno = get_next_replay_req_transno(obd);
2248         if (type != NULL)
2249                 *type = REQUEST_RECOVERY;
2250
2251         if (tdtd == NULL)
2252                 RETURN(transno);
2253
2254         update_transno = distribute_txn_get_next_transno(tdtd);
2255         if (transno == 0 || (transno >= update_transno &&
2256                              update_transno != 0)) {
2257                 transno = update_transno;
2258                 if (type != NULL)
2259                         *type = UPDATE_RECOVERY;
2260         }
2261
2262         RETURN(transno);
2263 }
2264
2265 /**
2266  * drop duplicate replay request
2267  *
2268  * Because the operation has been replayed by update recovery, the request
2269  * with the same transno will be dropped and also notify the client to send
2270  * next replay request.
2271  *
2272  * \param[in] env       execution environment
2273  * \param[in] obd       failover obd device
2274  * \param[in] req       request to be dropped
2275  */
2276 static void drop_duplicate_replay_req(struct lu_env *env,
2277                                       struct obd_device *obd,
2278                                       struct ptlrpc_request *req)
2279 {
2280         DEBUG_REQ(D_HA, req, "remove t%lld from %s because of duplicate"
2281                   " update records are found.\n",
2282                   lustre_msg_get_transno(req->rq_reqmsg),
2283                   libcfs_nid2str(req->rq_peer.nid));
2284
2285         /* Right now, only for MDS reint operation update replay and
2286          * normal request replay can have the same transno */
2287         if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_REINT) {
2288                 req_capsule_set(&req->rq_pill, &RQF_MDS_REINT);
2289                 req->rq_status = req_capsule_server_pack(&req->rq_pill);
2290                 if (likely(req->rq_export))
2291                         target_committed_to_req(req);
2292                 lustre_msg_set_transno(req->rq_repmsg, req->rq_transno);
2293                 target_send_reply(req, req->rq_status, 0);
2294         } else {
2295                 DEBUG_REQ(D_ERROR, req, "wrong opc" "from %s\n",
2296                 libcfs_nid2str(req->rq_peer.nid));
2297         }
2298         target_exp_dequeue_req_replay(req);
2299         target_request_copy_put(req);
2300         obd->obd_replayed_requests++;
2301 }
2302
2303 static void replay_request_or_update(struct lu_env *env,
2304                                      struct lu_target *lut,
2305                                      struct target_recovery_data *trd,
2306                                      struct ptlrpc_thread *thread)
2307 {
2308         struct obd_device *obd = lut->lut_obd;
2309         struct ptlrpc_request *req = NULL;
2310         int                     type;
2311         __u64                   transno;
2312         ENTRY;
2313
2314         CDEBUG(D_HA, "Waiting for transno %lld\n",
2315                obd->obd_next_recovery_transno);
2316
2317         /* Replay all of request and update by transno */
2318         do {
2319                 struct target_distribute_txn_data *tdtd = lut->lut_tdtd;
2320
2321                 CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLAY_DELAY2, cfs_fail_val);
2322
2323                 /** It is needed to extend recovery window above
2324                  *  recovery_time_soft. Extending is possible only in the
2325                  *  end of recovery window (see more details in
2326                  *  handle_recovery_req()).
2327                  */
2328                 CFS_FAIL_TIMEOUT_MS(OBD_FAIL_TGT_REPLAY_DELAY, 300);
2329
2330                 if (target_recovery_overseer(lut, check_for_next_transno,
2331                                         exp_req_replay_healthy_or_from_mdt)) {
2332                         abort_req_replay_queue(obd);
2333                         abort_lock_replay_queue(obd);
2334                         goto abort;
2335                 }
2336
2337                 spin_lock(&obd->obd_recovery_task_lock);
2338                 transno = get_next_transno(lut, &type);
2339                 if (type == REQUEST_RECOVERY && transno != 0) {
2340                         /* Drop replay request from client side, if the
2341                          * replay has been executed by update with the
2342                          * same transno */
2343                         req = list_entry(obd->obd_req_replay_queue.next,
2344                                         struct ptlrpc_request, rq_list);
2345
2346                         list_del_init(&req->rq_list);
2347                         obd->obd_requests_queued_for_recovery--;
2348                         spin_unlock(&obd->obd_recovery_task_lock);
2349
2350                         /* Let's check if the request has been redone by
2351                          * update replay */
2352                         if (is_req_replayed_by_update(req)) {
2353                                 struct distribute_txn_replay_req *dtrq;
2354
2355                                 dtrq = distribute_txn_lookup_finish_list(tdtd,
2356                                                                    req->rq_xid);
2357                                 LASSERT(dtrq != NULL);
2358                                 spin_lock(&tdtd->tdtd_replay_list_lock);
2359                                 list_del_init(&dtrq->dtrq_list);
2360                                 spin_unlock(&tdtd->tdtd_replay_list_lock);
2361                                 dtrq_destroy(dtrq);
2362
2363                                 drop_duplicate_replay_req(env, obd, req);
2364
2365                                 continue;
2366                         }
2367
2368                         LASSERT(trd->trd_processing_task == current_pid());
2369                         DEBUG_REQ(D_HA, req, "processing t%lld from %s",
2370                                   lustre_msg_get_transno(req->rq_reqmsg),
2371                                   libcfs_nid2str(req->rq_peer.nid));
2372
2373                         handle_recovery_req(thread, req,
2374                                             trd->trd_recovery_handler);
2375                         /**
2376                          * bz18031: increase next_recovery_transno before
2377                          * target_request_copy_put() will drop exp_rpc reference
2378                          */
2379                         spin_lock(&obd->obd_recovery_task_lock);
2380                         obd->obd_next_recovery_transno++;
2381                         spin_unlock(&obd->obd_recovery_task_lock);
2382                         target_exp_dequeue_req_replay(req);
2383                         target_request_copy_put(req);
2384                         obd->obd_replayed_requests++;
2385                 } else if (type == UPDATE_RECOVERY && transno != 0) {
2386                         struct distribute_txn_replay_req *dtrq;
2387                         int rc;
2388
2389                         spin_unlock(&obd->obd_recovery_task_lock);
2390
2391                         LASSERT(tdtd != NULL);
2392                         dtrq = distribute_txn_get_next_req(tdtd);
2393                         lu_context_enter(&thread->t_env->le_ctx);
2394                         rc = tdtd->tdtd_replay_handler(env, tdtd, dtrq);
2395                         lu_context_exit(&thread->t_env->le_ctx);
2396                         extend_recovery_timer(obd, obd_timeout, true);
2397
2398                         if (rc == 0 && dtrq->dtrq_xid != 0) {
2399                                 CDEBUG(D_HA, "Move x%llu t%llu"
2400                                        " to finish list\n", dtrq->dtrq_xid,
2401                                        dtrq->dtrq_master_transno);
2402
2403                                 /* Add it to the replay finish list */
2404                                 spin_lock(&tdtd->tdtd_replay_list_lock);
2405                                 list_add(&dtrq->dtrq_list,
2406                                          &tdtd->tdtd_replay_finish_list);
2407                                 spin_unlock(&tdtd->tdtd_replay_list_lock);
2408
2409                                 spin_lock(&obd->obd_recovery_task_lock);
2410                                 if (transno == obd->obd_next_recovery_transno)
2411                                         obd->obd_next_recovery_transno++;
2412                                 else if (transno >
2413                                          obd->obd_next_recovery_transno)
2414                                         obd->obd_next_recovery_transno =
2415                                                                 transno + 1;
2416                                 spin_unlock(&obd->obd_recovery_task_lock);
2417                         } else {
2418                                 dtrq_destroy(dtrq);
2419                         }
2420                 } else {
2421                         spin_unlock(&obd->obd_recovery_task_lock);
2422 abort:
2423                         LASSERT(list_empty(&obd->obd_req_replay_queue));
2424                         LASSERT(atomic_read(&obd->obd_req_replay_clients) == 0);
2425                         /** evict exports failed VBR */
2426                         class_disconnect_stale_exports(obd, exp_vbr_healthy);
2427                         break;
2428                 }
2429         } while (1);
2430 }
2431
2432 static int target_recovery_thread(void *arg)
2433 {
2434         struct lu_target *lut = arg;
2435         struct obd_device *obd = lut->lut_obd;
2436         struct ptlrpc_request *req;
2437         struct target_recovery_data *trd = &obd->obd_recovery_data;
2438         unsigned long delta;
2439         struct lu_env *env;
2440         struct ptlrpc_thread *thread = NULL;
2441         int rc = 0;
2442         ENTRY;
2443
2444         unshare_fs_struct();
2445         OBD_ALLOC_PTR(thread);
2446         if (thread == NULL)
2447                 RETURN(-ENOMEM);
2448
2449         OBD_ALLOC_PTR(env);
2450         if (env == NULL) {
2451                 OBD_FREE_PTR(thread);
2452                 RETURN(-ENOMEM);
2453         }
2454
2455         rc = lu_context_init(&env->le_ctx, LCT_MD_THREAD | LCT_DT_THREAD);
2456         if (rc) {
2457                 OBD_FREE_PTR(thread);
2458                 OBD_FREE_PTR(env);
2459                 RETURN(rc);
2460         }
2461
2462         thread->t_env = env;
2463         thread->t_id = -1; /* force filter_iobuf_get/put to use local buffers */
2464         env->le_ctx.lc_thread = thread;
2465         tgt_io_thread_init(thread); /* init thread_big_cache for IO requests */
2466         thread->t_watchdog = NULL;
2467
2468         CDEBUG(D_HA, "%s: started recovery thread pid %d\n", obd->obd_name,
2469                current_pid());
2470         trd->trd_processing_task = current_pid();
2471
2472         spin_lock(&obd->obd_dev_lock);
2473         obd->obd_recovering = 1;
2474         spin_unlock(&obd->obd_dev_lock);
2475         complete(&trd->trd_starting);
2476
2477         /* first of all, we have to know the first transno to replay */
2478         if (target_recovery_overseer(lut, check_for_recovery_ready,
2479                                      exp_connect_healthy)) {
2480                 abort_req_replay_queue(obd);
2481                 abort_lock_replay_queue(obd);
2482                 if (lut->lut_tdtd != NULL)
2483                         dtrq_list_destroy(lut->lut_tdtd);
2484         }
2485
2486         /* next stage: replay requests or update */
2487         delta = jiffies;
2488         CDEBUG(D_INFO, "1: request replay stage - %d clients from t%llu\n",
2489                atomic_read(&obd->obd_req_replay_clients),
2490                obd->obd_next_recovery_transno);
2491         replay_request_or_update(env, lut, trd, thread);
2492
2493         /**
2494          * The second stage: replay locks
2495          */
2496         CDEBUG(D_INFO, "2: lock replay stage - %d clients\n",
2497                atomic_read(&obd->obd_lock_replay_clients));
2498         while ((req = target_next_replay_lock(lut))) {
2499                 LASSERT(trd->trd_processing_task == current_pid());
2500                 DEBUG_REQ(D_HA, req, "processing lock from %s: ",
2501                           libcfs_nid2str(req->rq_peer.nid));
2502                 handle_recovery_req(thread, req,
2503                                     trd->trd_recovery_handler);
2504                 target_request_copy_put(req);
2505                 obd->obd_replayed_locks++;
2506         }
2507
2508         /**
2509          * The third stage: reply on final pings, at this moment all clients
2510          * must have request in final queue
2511          */
2512         CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLAY_RECONNECT, cfs_fail_val);
2513         CDEBUG(D_INFO, "3: final stage - process recovery completion pings\n");
2514         /** Update server last boot epoch */
2515         tgt_boot_epoch_update(lut);
2516         /* We drop recoverying flag to forward all new requests
2517          * to regular mds_handle() since now */
2518         spin_lock(&obd->obd_dev_lock);
2519         obd->obd_recovering = obd->obd_abort_recovery = 0;
2520         spin_unlock(&obd->obd_dev_lock);
2521         spin_lock(&obd->obd_recovery_task_lock);
2522         target_cancel_recovery_timer(obd);
2523         spin_unlock(&obd->obd_recovery_task_lock);
2524         while ((req = target_next_final_ping(obd))) {
2525                 LASSERT(trd->trd_processing_task == current_pid());
2526                 DEBUG_REQ(D_HA, req, "processing final ping from %s: ",
2527                           libcfs_nid2str(req->rq_peer.nid));
2528                 handle_recovery_req(thread, req,
2529                                     trd->trd_recovery_handler);
2530                 /* Because the waiting client can not send ping to server,
2531                  * so we need refresh the last_request_time, to avoid the
2532                  * export is being evicted */
2533                 ptlrpc_update_export_timer(req->rq_export, 0);
2534                 target_request_copy_put(req);
2535         }
2536
2537         delta = jiffies_to_msecs(jiffies - delta) / MSEC_PER_SEC;
2538         CDEBUG(D_INFO,"4: recovery completed in %lus - %d/%d reqs/locks\n",
2539               delta, obd->obd_replayed_requests, obd->obd_replayed_locks);
2540         if (delta > OBD_RECOVERY_TIME_SOFT) {
2541                 CWARN("too long recovery - read logs\n");
2542                 libcfs_debug_dumplog();
2543         }
2544
2545         target_finish_recovery(lut);
2546
2547         lu_context_fini(&env->le_ctx);
2548         trd->trd_processing_task = 0;
2549         complete(&trd->trd_finishing);
2550
2551         tgt_io_thread_done(thread);
2552         OBD_FREE_PTR(thread);
2553         OBD_FREE_PTR(env);
2554         RETURN(rc);
2555 }
2556
2557 static int target_start_recovery_thread(struct lu_target *lut,
2558                                         svc_handler_t handler)
2559 {
2560         struct obd_device *obd = lut->lut_obd;
2561         int rc = 0;
2562         struct target_recovery_data *trd = &obd->obd_recovery_data;
2563         int index;
2564
2565         memset(trd, 0, sizeof(*trd));
2566         init_completion(&trd->trd_starting);
2567         init_completion(&trd->trd_finishing);
2568         trd->trd_recovery_handler = handler;
2569
2570         rc = server_name2index(obd->obd_name, &index, NULL);
2571         if (rc < 0)
2572                 return rc;
2573
2574         if (!IS_ERR(kthread_run(target_recovery_thread,
2575                                 lut, "tgt_recover_%d", index))) {
2576                 wait_for_completion(&trd->trd_starting);
2577                 LASSERT(obd->obd_recovering != 0);
2578         } else {
2579                 rc = -ECHILD;
2580         }
2581
2582         return rc;
2583 }
2584
2585 void target_stop_recovery_thread(struct obd_device *obd)
2586 {
2587         if (obd->obd_recovery_data.trd_processing_task > 0) {
2588                 struct target_recovery_data *trd = &obd->obd_recovery_data;
2589                 /** recovery can be done but postrecovery is not yet */
2590                 spin_lock(&obd->obd_dev_lock);
2591                 if (obd->obd_recovering) {
2592                         CERROR("%s: Aborting recovery\n", obd->obd_name);
2593                         obd->obd_abort_recovery = 1;
2594                         wake_up(&obd->obd_next_transno_waitq);
2595                 }
2596                 spin_unlock(&obd->obd_dev_lock);
2597                 wait_for_completion(&trd->trd_finishing);
2598         }
2599 }
2600 EXPORT_SYMBOL(target_stop_recovery_thread);
2601
2602 void target_recovery_fini(struct obd_device *obd)
2603 {
2604         class_disconnect_exports(obd);
2605         target_stop_recovery_thread(obd);
2606         target_cleanup_recovery(obd);
2607 }
2608 EXPORT_SYMBOL(target_recovery_fini);
2609
2610 static void target_recovery_expired(cfs_timer_cb_arg_t data)
2611 {
2612         struct obd_device *obd = cfs_from_timer(obd, data, obd_recovery_timer);
2613         CDEBUG(D_HA, "%s: recovery timed out; %d clients are still in recovery"
2614                " after %llus (%d clients connected)\n",
2615                obd->obd_name, atomic_read(&obd->obd_lock_replay_clients),
2616                (s64)(ktime_get_real_seconds() - obd->obd_recovery_start),
2617                atomic_read(&obd->obd_connected_clients));
2618
2619         obd->obd_recovery_expired = 1;
2620         wake_up(&obd->obd_next_transno_waitq);
2621 }
2622
2623 void target_recovery_init(struct lu_target *lut, svc_handler_t handler)
2624 {
2625         struct obd_device *obd = lut->lut_obd;
2626
2627         if (lut->lut_bottom->dd_rdonly)
2628                 return;
2629
2630         if (obd->obd_max_recoverable_clients == 0) {
2631                 /** Update server last boot epoch */
2632                 tgt_boot_epoch_update(lut);
2633                 return;
2634         }
2635
2636         CDEBUG(D_HA, "RECOVERY: service %s, %d recoverable clients, "
2637                "last_transno %llu\n", obd->obd_name,
2638                obd->obd_max_recoverable_clients, obd->obd_last_committed);
2639         LASSERT(obd->obd_stopping == 0);
2640         obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
2641         obd->obd_recovery_start = 0;
2642         obd->obd_recovery_end = 0;
2643
2644         cfs_timer_setup(&obd->obd_recovery_timer, target_recovery_expired,
2645                         (unsigned long)obd, 0);
2646         target_start_recovery_thread(lut, handler);
2647 }
2648 EXPORT_SYMBOL(target_recovery_init);
2649
2650 static int target_process_req_flags(struct obd_device *obd,
2651                                     struct ptlrpc_request *req)
2652 {
2653         struct obd_export *exp = req->rq_export;
2654         LASSERT(exp != NULL);
2655         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
2656                 /* client declares he's ready to replay locks */
2657                 spin_lock(&exp->exp_lock);
2658                 if (exp->exp_req_replay_needed) {
2659                         exp->exp_req_replay_needed = 0;
2660                         spin_unlock(&exp->exp_lock);
2661
2662                         LASSERT_ATOMIC_POS(&obd->obd_req_replay_clients);
2663                         atomic_dec(&obd->obd_req_replay_clients);
2664                 } else {
2665                         spin_unlock(&exp->exp_lock);
2666                 }
2667         }
2668         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
2669                 /* client declares he's ready to complete recovery
2670                  * so, we put the request on th final queue */
2671                 spin_lock(&exp->exp_lock);
2672                 if (exp->exp_lock_replay_needed) {
2673                         exp->exp_lock_replay_needed = 0;
2674                         spin_unlock(&exp->exp_lock);
2675
2676                         LASSERT_ATOMIC_POS(&obd->obd_lock_replay_clients);
2677                         atomic_dec(&obd->obd_lock_replay_clients);
2678                 } else {
2679                         spin_unlock(&exp->exp_lock);
2680                 }
2681         }
2682         return 0;
2683 }
2684
2685 int target_queue_recovery_request(struct ptlrpc_request *req,
2686                                   struct obd_device *obd)
2687 {
2688         __u64 transno = lustre_msg_get_transno(req->rq_reqmsg);
2689         struct ptlrpc_request *reqiter;
2690         int inserted = 0;
2691         ENTRY;
2692
2693         if (obd->obd_recovery_data.trd_processing_task == current_pid()) {
2694                 /* Processing the queue right now, don't re-add. */
2695                 RETURN(1);
2696         }
2697
2698         target_process_req_flags(obd, req);
2699
2700         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
2701                 if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_TGT_RECOVERY_REQ_RACE))) {
2702                         if (cfs_fail_val == 1) {
2703                                 cfs_race_state = 1;
2704                                 cfs_fail_val = 0;
2705                                 wake_up(&cfs_race_waitq);
2706
2707                                 set_current_state(TASK_INTERRUPTIBLE);
2708                                 schedule_timeout(cfs_time_seconds(1));
2709                         }
2710                 }
2711
2712                 /* client declares he's ready to complete recovery
2713                  * so, we put the request on th final queue */
2714                 target_request_copy_get(req);
2715                 DEBUG_REQ(D_HA, req, "queue final req");
2716                 wake_up(&obd->obd_next_transno_waitq);
2717                 spin_lock(&obd->obd_recovery_task_lock);
2718                 if (obd->obd_recovering) {
2719                         struct ptlrpc_request *tmp;
2720                         struct ptlrpc_request *duplicate = NULL;
2721
2722                         if (likely(!req->rq_export->exp_replay_done)) {
2723                                 req->rq_export->exp_replay_done = 1;
2724                                 list_add_tail(&req->rq_list,
2725                                               &obd->obd_final_req_queue);
2726                                 spin_unlock(&obd->obd_recovery_task_lock);
2727                                 RETURN(0);
2728                         }
2729
2730                         /* XXX O(n), but only happens if final ping is
2731                          * timed out, probably reorganize the list as
2732                          * a hash list later */
2733                         list_for_each_entry_safe(reqiter, tmp,
2734                                                  &obd->obd_final_req_queue,
2735                                                  rq_list) {
2736                                 if (reqiter->rq_export == req->rq_export) {
2737                                         list_del_init(&reqiter->rq_list);
2738                                         duplicate = reqiter;
2739                                         break;
2740                                 }
2741                         }
2742
2743                         list_add_tail(&req->rq_list,
2744                                       &obd->obd_final_req_queue);
2745                         req->rq_export->exp_replay_done = 1;
2746                         spin_unlock(&obd->obd_recovery_task_lock);
2747
2748                         if (duplicate != NULL) {
2749                                 DEBUG_REQ(D_HA, duplicate,
2750                                           "put prev final req\n");
2751                                 target_request_copy_put(duplicate);
2752                         }
2753                         RETURN(0);
2754                 } else {
2755                         spin_unlock(&obd->obd_recovery_task_lock);
2756                         target_request_copy_put(req);
2757                         RETURN(obd->obd_stopping ? -ENOTCONN : 1);
2758                 }
2759         }
2760         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
2761                 /* client declares he's ready to replay locks */
2762                 target_request_copy_get(req);
2763                 DEBUG_REQ(D_HA, req, "queue lock replay req");
2764                 wake_up(&obd->obd_next_transno_waitq);
2765                 spin_lock(&obd->obd_recovery_task_lock);
2766                 LASSERT(obd->obd_recovering);
2767                 /* usually due to recovery abort */
2768                 if (!req->rq_export->exp_in_recovery) {
2769                         spin_unlock(&obd->obd_recovery_task_lock);
2770                         target_request_copy_put(req);
2771                         RETURN(-ENOTCONN);
2772                 }
2773                 LASSERT(req->rq_export->exp_lock_replay_needed);
2774                 list_add_tail(&req->rq_list, &obd->obd_lock_replay_queue);
2775                 spin_unlock(&obd->obd_recovery_task_lock);
2776                 RETURN(0);
2777         }
2778
2779         /* CAVEAT EMPTOR: The incoming request message has been swabbed
2780          * (i.e. buflens etc are in my own byte order), but type-dependent
2781          * buffers (eg mdt_body, ost_body etc) have NOT been swabbed. */
2782
2783         if (!transno) {
2784                 INIT_LIST_HEAD(&req->rq_list);
2785                 DEBUG_REQ(D_HA, req, "not queueing");
2786                 RETURN(1);
2787         }
2788
2789         /* If we're processing the queue, we want don't want to queue this
2790          * message.
2791          *
2792          * Also, if this request has a transno less than the one we're waiting
2793          * for, we should process it now.  It could (and currently always will)
2794          * be an open request for a descriptor that was opened some time ago.
2795          *
2796          * Also, a resent, replayed request that has already been
2797          * handled will pass through here and be processed immediately.
2798          */
2799         CDEBUG(D_HA, "Next recovery transno: %llu"
2800                ", current: %llu, replaying\n",
2801                obd->obd_next_recovery_transno, transno);
2802
2803         /* If the request has been replayed by update replay, then sends this
2804          * request to the recovery thread (replay_request_or_update()), where
2805          * it will be handled */
2806         spin_lock(&obd->obd_recovery_task_lock);
2807         if (transno < obd->obd_next_recovery_transno &&
2808             !is_req_replayed_by_update(req)) {
2809                 /* Processing the queue right now, don't re-add. */
2810                 LASSERT(list_empty(&req->rq_list));
2811                 spin_unlock(&obd->obd_recovery_task_lock);
2812                 RETURN(1);
2813         }
2814         spin_unlock(&obd->obd_recovery_task_lock);
2815
2816         if (OBD_FAIL_CHECK(OBD_FAIL_TGT_REPLAY_DROP))
2817                 RETURN(0);
2818
2819         target_request_copy_get(req);
2820         if (!req->rq_export->exp_in_recovery) {
2821                 target_request_copy_put(req);
2822                 RETURN(-ENOTCONN);
2823         }
2824         LASSERT(req->rq_export->exp_req_replay_needed);
2825
2826         if (target_exp_enqueue_req_replay(req)) {
2827                 DEBUG_REQ(D_ERROR, req, "dropping resent queued req");
2828                 target_request_copy_put(req);
2829                 RETURN(0);
2830         }
2831
2832         /* XXX O(n^2) */
2833         spin_lock(&obd->obd_recovery_task_lock);
2834         LASSERT(obd->obd_recovering);
2835         list_for_each_entry(reqiter, &obd->obd_req_replay_queue, rq_list) {
2836                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) > transno) {
2837                         list_add_tail(&req->rq_list, &reqiter->rq_list);
2838                         inserted = 1;
2839                         goto added;
2840                 }
2841
2842                 if (unlikely(lustre_msg_get_transno(reqiter->rq_reqmsg) ==
2843                              transno)) {
2844                         DEBUG_REQ(D_ERROR, req, "dropping replay: transno "
2845                                   "has been claimed by another client");
2846                         spin_unlock(&obd->obd_recovery_task_lock);
2847                         target_exp_dequeue_req_replay(req);
2848                         target_request_copy_put(req);
2849                         RETURN(0);
2850                 }
2851         }
2852 added:
2853         if (!inserted)
2854                 list_add_tail(&req->rq_list, &obd->obd_req_replay_queue);
2855
2856         obd->obd_requests_queued_for_recovery++;
2857         spin_unlock(&obd->obd_recovery_task_lock);
2858         wake_up(&obd->obd_next_transno_waitq);
2859         RETURN(0);
2860 }
2861
2862 int target_handle_ping(struct ptlrpc_request *req)
2863 {
2864         obd_ping(req->rq_svc_thread->t_env, req->rq_export);
2865         return req_capsule_server_pack(&req->rq_pill);
2866 }
2867
2868 void target_committed_to_req(struct ptlrpc_request *req)
2869 {
2870         struct obd_export *exp = req->rq_export;
2871
2872         if (!exp->exp_obd->obd_no_transno && req->rq_repmsg != NULL)
2873                 lustre_msg_set_last_committed(req->rq_repmsg,
2874                                               exp->exp_last_committed);
2875         else
2876                 DEBUG_REQ(D_IOCTL, req, "not sending last_committed update (%d/"
2877                           "%d)", exp->exp_obd->obd_no_transno,
2878                           req->rq_repmsg == NULL);
2879
2880         CDEBUG(D_INFO, "last_committed %llu, transno %llu, xid %llu\n",
2881                exp->exp_last_committed, req->rq_transno, req->rq_xid);
2882 }
2883
2884 #endif /* HAVE_SERVER_SUPPORT */
2885
2886 /**
2887  * Packs current SLV and Limit into \a req.
2888  */
2889 int target_pack_pool_reply(struct ptlrpc_request *req)
2890 {
2891         struct obd_device *obd;
2892         ENTRY;
2893
2894         /* Check that we still have all structures alive as this may
2895          * be some late RPC at shutdown time. */
2896         if (unlikely(!req->rq_export || !req->rq_export->exp_obd ||
2897                      !exp_connect_lru_resize(req->rq_export))) {
2898                 lustre_msg_set_slv(req->rq_repmsg, 0);
2899                 lustre_msg_set_limit(req->rq_repmsg, 0);
2900                 RETURN(0);
2901         }
2902
2903         /* OBD is alive here as export is alive, which we checked above. */
2904         obd = req->rq_export->exp_obd;
2905
2906         read_lock(&obd->obd_pool_lock);
2907         lustre_msg_set_slv(req->rq_repmsg, obd->obd_pool_slv);
2908         lustre_msg_set_limit(req->rq_repmsg, obd->obd_pool_limit);
2909         read_unlock(&obd->obd_pool_lock);
2910
2911         RETURN(0);
2912 }
2913
2914 static int target_send_reply_msg(struct ptlrpc_request *req,
2915                                  int rc, int fail_id)
2916 {
2917         if (OBD_FAIL_CHECK_ORSET(fail_id & ~OBD_FAIL_ONCE, OBD_FAIL_ONCE)) {
2918                 DEBUG_REQ(D_ERROR, req, "dropping reply");
2919                 return -ECOMM;
2920         }
2921         /* We can have a null rq_reqmsg in the event of bad signature or
2922          * no context when unwrapping */
2923         if (req->rq_reqmsg &&
2924             unlikely(lustre_msg_get_opc(req->rq_reqmsg) == MDS_REINT &&
2925             OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_MULTI_NET_REP)))
2926                 return -ECOMM;
2927
2928         if (unlikely(rc)) {
2929                 DEBUG_REQ(D_NET, req, "processing error (%d)", rc);
2930                 req->rq_status = rc;
2931                 return ptlrpc_send_error(req, 1);
2932         } else {
2933                 DEBUG_REQ(D_NET, req, "sending reply");
2934         }
2935
2936         return ptlrpc_send_reply(req, PTLRPC_REPLY_MAYBE_DIFFICULT);
2937 }
2938
2939 void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
2940 {
2941         struct ptlrpc_service_part *svcpt;
2942         int                        netrc;
2943         struct ptlrpc_reply_state *rs;
2944         struct obd_export         *exp;
2945         ENTRY;
2946
2947         if (req->rq_no_reply) {
2948                 EXIT;
2949                 return;
2950         }
2951
2952         svcpt = req->rq_rqbd->rqbd_svcpt;
2953         rs = req->rq_reply_state;
2954         if (rs == NULL || !rs->rs_difficult) {
2955                 /* no notifiers */
2956                 target_send_reply_msg (req, rc, fail_id);
2957                 EXIT;
2958                 return;
2959         }
2960
2961         /* must be an export if locks saved */
2962         LASSERT(req->rq_export != NULL);
2963         /* req/reply consistent */
2964         LASSERT(rs->rs_svcpt == svcpt);
2965
2966         /* "fresh" reply */
2967         LASSERT(!rs->rs_scheduled);
2968         LASSERT(!rs->rs_scheduled_ever);
2969         LASSERT(!rs->rs_handled);
2970         LASSERT(!rs->rs_on_net);
2971         LASSERT(rs->rs_export == NULL);
2972         LASSERT(list_empty(&rs->rs_obd_list));
2973         LASSERT(list_empty(&rs->rs_exp_list));
2974
2975         exp = class_export_get(req->rq_export);
2976
2977         /* disable reply scheduling while I'm setting up */
2978         rs->rs_scheduled = 1;
2979         rs->rs_on_net    = 1;
2980         rs->rs_xid       = req->rq_xid;
2981         rs->rs_transno   = req->rq_transno;
2982         rs->rs_export    = exp;
2983         rs->rs_opc       = lustre_msg_get_opc(req->rq_reqmsg);
2984
2985         spin_lock(&exp->exp_uncommitted_replies_lock);
2986         CDEBUG(D_NET, "rs transno = %llu, last committed = %llu\n",
2987                rs->rs_transno, exp->exp_last_committed);
2988         if (rs->rs_transno > exp->exp_last_committed) {
2989                 /* not committed already */
2990                 list_add_tail(&rs->rs_obd_list,
2991                                   &exp->exp_uncommitted_replies);
2992         }
2993         spin_unlock(&exp->exp_uncommitted_replies_lock);
2994
2995         spin_lock(&exp->exp_lock);
2996         list_add_tail(&rs->rs_exp_list, &exp->exp_outstanding_replies);
2997         spin_unlock(&exp->exp_lock);
2998
2999         netrc = target_send_reply_msg(req, rc, fail_id);
3000
3001         spin_lock(&svcpt->scp_rep_lock);
3002
3003         atomic_inc(&svcpt->scp_nreps_difficult);
3004
3005         if (netrc != 0) {
3006                 /* error sending: reply is off the net.  Also we need +1
3007                  * reply ref until ptlrpc_handle_rs() is done
3008                  * with the reply state (if the send was successful, there
3009                  * would have been +1 ref for the net, which
3010                  * reply_out_callback leaves alone) */
3011                 rs->rs_on_net = 0;
3012                 ptlrpc_rs_addref(rs);
3013         }
3014
3015         spin_lock(&rs->rs_lock);
3016         if (rs->rs_transno <= exp->exp_last_committed ||
3017             (!rs->rs_on_net && !rs->rs_no_ack) ||
3018             list_empty(&rs->rs_exp_list) ||     /* completed already */
3019             list_empty(&rs->rs_obd_list)) {
3020                 CDEBUG(D_HA, "Schedule reply immediately\n");
3021                 ptlrpc_dispatch_difficult_reply(rs);
3022         } else {
3023                 list_add(&rs->rs_list, &svcpt->scp_rep_active);
3024                 rs->rs_scheduled = 0;   /* allow notifier to schedule */
3025         }
3026         spin_unlock(&rs->rs_lock);
3027         spin_unlock(&svcpt->scp_rep_lock);
3028         EXIT;
3029 }
3030
3031 enum ldlm_mode lck_compat_array[] = {
3032         [LCK_EX]    = LCK_COMPAT_EX,
3033         [LCK_PW]    = LCK_COMPAT_PW,
3034         [LCK_PR]    = LCK_COMPAT_PR,
3035         [LCK_CW]    = LCK_COMPAT_CW,
3036         [LCK_CR]    = LCK_COMPAT_CR,
3037         [LCK_NL]    = LCK_COMPAT_NL,
3038         [LCK_GROUP] = LCK_COMPAT_GROUP,
3039         [LCK_COS]   = LCK_COMPAT_COS,
3040 };
3041
3042 /**
3043  * Rather arbitrary mapping from LDLM error codes to errno values. This should
3044  * not escape to the user level.
3045  */
3046 int ldlm_error2errno(enum ldlm_error error)
3047 {
3048         int result;
3049
3050         switch (error) {
3051         case ELDLM_OK:
3052         case ELDLM_LOCK_MATCHED:
3053                 result = 0;
3054                 break;
3055         case ELDLM_LOCK_CHANGED:
3056                 result = -ESTALE;
3057                 break;
3058         case ELDLM_LOCK_ABORTED:
3059                 result = -ENAVAIL;
3060                 break;
3061         case ELDLM_LOCK_REPLACED:
3062                 result = -ESRCH;
3063                 break;
3064         case ELDLM_NO_LOCK_DATA:
3065                 result = -ENOENT;
3066                 break;
3067         case ELDLM_NAMESPACE_EXISTS:
3068                 result = -EEXIST;
3069                 break;
3070         case ELDLM_BAD_NAMESPACE:
3071                 result = -EBADF;
3072                 break;
3073         default:
3074                 if (((int)error) < 0) { /* cast to signed type */
3075                         result = error; /* as ldlm_error can be unsigned */
3076                 } else {
3077                         CERROR("Invalid DLM result code: %d\n", error);
3078                         result = -EPROTO;
3079                 }
3080         }
3081         return result;
3082 }
3083 EXPORT_SYMBOL(ldlm_error2errno);
3084
3085 /**
3086  * Dual to ldlm_error2errno(): maps errno values back to enum ldlm_error.
3087  */
3088 enum ldlm_error ldlm_errno2error(int err_no)
3089 {
3090         int error;
3091
3092         switch (err_no) {
3093         case 0:
3094                 error = ELDLM_OK;
3095                 break;
3096         case -ESTALE:
3097                 error = ELDLM_LOCK_CHANGED;
3098                 break;
3099         case -ENAVAIL:
3100                 error = ELDLM_LOCK_ABORTED;
3101                 break;
3102         case -ESRCH:
3103                 error = ELDLM_LOCK_REPLACED;
3104                 break;
3105         case -ENOENT:
3106                 error = ELDLM_NO_LOCK_DATA;
3107                 break;
3108         case -EEXIST:
3109                 error = ELDLM_NAMESPACE_EXISTS;
3110                 break;
3111         case -EBADF:
3112                 error = ELDLM_BAD_NAMESPACE;
3113                 break;
3114         default:
3115                 error = err_no;
3116         }
3117         return error;
3118 }
3119
3120 #if LUSTRE_TRACKS_LOCK_EXP_REFS
3121 void ldlm_dump_export_locks(struct obd_export *exp)
3122 {
3123         spin_lock(&exp->exp_locks_list_guard);
3124         if (!list_empty(&exp->exp_locks_list)) {
3125                 struct ldlm_lock *lock;
3126
3127                 CERROR("dumping locks for export %p,"
3128                        "ignore if the unmount doesn't hang\n", exp);
3129                 list_for_each_entry(lock, &exp->exp_locks_list,
3130                                         l_exp_refs_link)
3131                         LDLM_ERROR(lock, "lock:");
3132         }
3133         spin_unlock(&exp->exp_locks_list_guard);
3134 }
3135 #endif
3136
3137 #ifdef HAVE_SERVER_SUPPORT
3138 static int target_bulk_timeout(void *data)
3139 {
3140         ENTRY;
3141         /* We don't fail the connection here, because having the export
3142          * killed makes the (vital) call to commitrw very sad.
3143          */
3144         RETURN(1);
3145 }
3146
3147 static inline const char *bulk2type(struct ptlrpc_request *req)
3148 {
3149         if (req->rq_bulk_read)
3150                 return "READ";
3151         if (req->rq_bulk_write)
3152                 return "WRITE";
3153         return "UNKNOWN";
3154 }
3155
3156 int target_bulk_io(struct obd_export *exp, struct ptlrpc_bulk_desc *desc,
3157                    struct l_wait_info *lwi)
3158 {
3159         struct ptlrpc_request *req = desc->bd_req;
3160         time64_t start = ktime_get_real_seconds();
3161         time64_t deadline;
3162         int rc = 0;
3163
3164         ENTRY;
3165
3166         /* If there is eviction in progress, wait for it to finish. */
3167         if (unlikely(atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
3168                 *lwi = LWI_INTR(NULL, NULL);
3169                 rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
3170                                   !atomic_read(&exp->exp_obd->
3171                                                    obd_evict_inprogress),
3172                                   lwi);
3173         }
3174
3175         /* Check if client was evicted or reconnected already. */
3176         if (exp->exp_failed ||
3177             exp->exp_conn_cnt > lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
3178                 rc = -ENOTCONN;
3179         } else {
3180                 if (req->rq_bulk_read)
3181                         rc = sptlrpc_svc_wrap_bulk(req, desc);
3182
3183                 if (OCD_HAS_FLAG(&exp->exp_connect_data, BULK_MBITS))
3184                         req->rq_mbits = lustre_msg_get_mbits(req->rq_reqmsg);
3185                 else /* old version, bulk matchbits is rq_xid */
3186                         req->rq_mbits = req->rq_xid;
3187
3188                 if (rc == 0)
3189                         rc = ptlrpc_start_bulk_transfer(desc);
3190         }
3191
3192         if (rc < 0) {
3193                 DEBUG_REQ(D_ERROR, req, "bulk %s failed: rc %d",
3194                           bulk2type(req), rc);
3195                 RETURN(rc);
3196         }
3197
3198         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) {
3199                 ptlrpc_abort_bulk(desc);
3200                 RETURN(0);
3201         }
3202
3203         /* limit actual bulk transfer to bulk_timeout seconds */
3204         deadline = start + bulk_timeout;
3205         if (deadline > req->rq_deadline)
3206                 deadline = req->rq_deadline;
3207
3208         do {
3209                 time64_t timeoutl = deadline - ktime_get_real_seconds();
3210                 long timeout_jiffies = timeoutl <= 0 ?
3211                                        1 : cfs_time_seconds(timeoutl);
3212                 time64_t rq_deadline;
3213
3214                 *lwi = LWI_TIMEOUT_INTERVAL(timeout_jiffies,
3215                                             cfs_time_seconds(1),
3216                                             target_bulk_timeout, desc);
3217                 rc = l_wait_event(desc->bd_waitq,
3218                                   !ptlrpc_server_bulk_active(desc) ||
3219                                   exp->exp_failed ||
3220                                   exp->exp_conn_cnt >
3221                                   lustre_msg_get_conn_cnt(req->rq_reqmsg),
3222                                   lwi);
3223                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
3224                 /* Wait again if we changed rq_deadline. */
3225                 rq_deadline = READ_ONCE(req->rq_deadline);
3226                 deadline = start + bulk_timeout;
3227                 if (deadline > rq_deadline)
3228                         deadline = rq_deadline;
3229         } while (rc == -ETIMEDOUT &&
3230                  deadline > ktime_get_real_seconds());
3231
3232         if (rc == -ETIMEDOUT) {
3233                 DEBUG_REQ(D_ERROR, req, "timeout on bulk %s after %lld%+llds",
3234                           bulk2type(req), deadline - start,
3235                           ktime_get_real_seconds() - deadline);
3236                 ptlrpc_abort_bulk(desc);
3237         } else if (exp->exp_failed) {
3238                 DEBUG_REQ(D_ERROR, req, "Eviction on bulk %s",
3239                           bulk2type(req));
3240                 rc = -ENOTCONN;
3241                 ptlrpc_abort_bulk(desc);
3242         } else if (exp->exp_conn_cnt >
3243                    lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
3244                 DEBUG_REQ(D_ERROR, req, "Reconnect on bulk %s",
3245                           bulk2type(req));
3246                 /* We don't reply anyway. */
3247                 rc = -ETIMEDOUT;
3248                 ptlrpc_abort_bulk(desc);
3249         } else if (desc->bd_failure) {
3250                 DEBUG_REQ(D_ERROR, req, "network error on bulk %s",
3251                           bulk2type(req));
3252                 /* XXX should this be a different errno? */
3253                 rc = -ETIMEDOUT;
3254         } else {
3255                 if (req->rq_bulk_write)
3256                         rc = sptlrpc_svc_unwrap_bulk(req, desc);
3257                 if (rc == 0 && desc->bd_nob_transferred != desc->bd_nob) {
3258                         DEBUG_REQ(D_ERROR, req, "truncated bulk %s %d(%d)",
3259                                   bulk2type(req), desc->bd_nob_transferred,
3260                                   desc->bd_nob);
3261                         /* XXX should this be a different errno? */
3262                         rc = -ETIMEDOUT;
3263                 }
3264         }
3265
3266         RETURN(rc);
3267 }
3268 EXPORT_SYMBOL(target_bulk_io);
3269
3270 #endif /* HAVE_SERVER_SUPPORT */