lustre/ldlm/ldlm_lib.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  */
  24
  25 #ifndef EXPORT_SYMTAB
  26 # define EXPORT_SYMTAB
  27 #endif
  28 #define DEBUG_SUBSYSTEM S_LDLM
  29
  30 #ifdef __KERNEL__
  31 # include <libcfs/libcfs.h>
  32 #else
  33 # include <liblustre.h>
  34 #endif
  35 #include <obd.h>
  36 #include <lustre_mds.h>
  37 #include <lustre_dlm.h>
  38 #include <lustre_net.h>
  39 #include <lustre_sec.h>
  40 #include "ldlm_internal.h"
  41
  42
  43 /* @priority: if non-zero, move the selected to the list head
  44  * @create: if zero, only search in existed connections
  45  */
  46 static int import_set_conn(struct obd_import *imp, struct obd_uuid *uuid,
  47                            int priority, int create)
  48 {
  49         struct ptlrpc_connection *ptlrpc_conn;
  50         struct obd_import_conn *imp_conn = NULL, *item;
  51         int rc = 0;
  52         ENTRY;
  53
  54         if (!create && !priority) {
  55                 CDEBUG(D_HA, "Nothing to do\n");
  56                 RETURN(-EINVAL);
  57         }
  58
  59         ptlrpc_conn = ptlrpc_uuid_to_connection(uuid);
  60         if (!ptlrpc_conn) {
  61                 CDEBUG(D_HA, "can't find connection %s\n", uuid->uuid);
  62                 RETURN (-ENOENT);
  63         }
  64
  65         if (create) {
  66                 OBD_ALLOC(imp_conn, sizeof(*imp_conn));
  67                 if (!imp_conn) {
  68                         GOTO(out_put, rc = -ENOMEM);
  69                 }
  70         }
  71
  72         spin_lock(&imp->imp_lock);
  73         list_for_each_entry(item, &imp->imp_conn_list, oic_item) {
  74                 if (obd_uuid_equals(uuid, &item->oic_uuid)) {
  75                         if (priority) {
  76                                 list_del(&item->oic_item);
  77                                 list_add(&item->oic_item, &imp->imp_conn_list);
  78                                 item->oic_last_attempt = 0;
  79                         }
  80                         CDEBUG(D_HA, "imp %p@%s: found existing conn %s%s\n",
  81                                imp, imp->imp_obd->obd_name, uuid->uuid,
  82                                (priority ? ", moved to head" : ""));
  83                         spin_unlock(&imp->imp_lock);
  84                         GOTO(out_free, rc = 0);
  85                 }
  86         }
  87         /* not found */
  88         if (create) {
  89                 imp_conn->oic_conn = ptlrpc_conn;
  90                 imp_conn->oic_uuid = *uuid;
  91                 item->oic_last_attempt = 0;
  92                 if (priority)
  93                         list_add(&imp_conn->oic_item, &imp->imp_conn_list);
  94                 else
  95                         list_add_tail(&imp_conn->oic_item, &imp->imp_conn_list);
  96                 CDEBUG(D_HA, "imp %p@%s: add connection %s at %s\n",
  97                        imp, imp->imp_obd->obd_name, uuid->uuid,
  98                        (priority ? "head" : "tail"));
  99         } else {
 100                 spin_unlock(&imp->imp_lock);
 101                 GOTO(out_free, rc = -ENOENT);
 102         }
 103
 104         spin_unlock(&imp->imp_lock);
 105         RETURN(0);
 106 out_free:
 107         if (imp_conn)
 108                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 109 out_put:
 110         ptlrpc_put_connection(ptlrpc_conn);
 111         RETURN(rc);
 112 }
 113
 114 int import_set_conn_priority(struct obd_import *imp, struct obd_uuid *uuid)
 115 {
 116         return import_set_conn(imp, uuid, 1, 0);
 117 }
 118
 119 int client_import_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
 120                            int priority)
 121 {
 122         return import_set_conn(imp, uuid, priority, 1);
 123 }
 124
 125 int client_import_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
 126 {
 127         struct obd_import_conn *imp_conn;
 128         struct obd_export *dlmexp;
 129         int rc = -ENOENT;
 130         ENTRY;
 131
 132         spin_lock(&imp->imp_lock);
 133         if (list_empty(&imp->imp_conn_list)) {
 134                 LASSERT(!imp->imp_connection);
 135                 GOTO(out, rc);
 136         }
 137
 138         list_for_each_entry(imp_conn, &imp->imp_conn_list, oic_item) {
 139                 if (!obd_uuid_equals(uuid, &imp_conn->oic_uuid))
 140                         continue;
 141                 LASSERT(imp_conn->oic_conn);
 142
 143                 /* is current conn? */
 144                 if (imp_conn == imp->imp_conn_current) {
 145                         LASSERT(imp_conn->oic_conn == imp->imp_connection);
 146
 147                         if (imp->imp_state != LUSTRE_IMP_CLOSED &&
 148                             imp->imp_state != LUSTRE_IMP_DISCON) {
 149                                 CERROR("can't remove current connection\n");
 150                                 GOTO(out, rc = -EBUSY);
 151                         }
 152
 153                         ptlrpc_put_connection(imp->imp_connection);
 154                         imp->imp_connection = NULL;
 155
 156                         dlmexp = class_conn2export(&imp->imp_dlm_handle);
 157                         if (dlmexp && dlmexp->exp_connection) {
 158                                 LASSERT(dlmexp->exp_connection ==
 159                                         imp_conn->oic_conn);
 160                                 ptlrpc_put_connection(dlmexp->exp_connection);
 161                                 dlmexp->exp_connection = NULL;
 162                         }
 163                 }
 164
 165                 list_del(&imp_conn->oic_item);
 166                 ptlrpc_put_connection(imp_conn->oic_conn);
 167                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 168                 CDEBUG(D_HA, "imp %p@%s: remove connection %s\n",
 169                        imp, imp->imp_obd->obd_name, uuid->uuid);
 170                 rc = 0;
 171                 break;
 172         }
 173 out:
 174         spin_unlock(&imp->imp_lock);
 175         if (rc == -ENOENT)
 176                 CERROR("connection %s not found\n", uuid->uuid);
 177         RETURN(rc);
 178 }
 179
 180 static void destroy_import(struct obd_import *imp)
 181 {
 182         /* drop security policy instance after all rpc finished/aborted
 183          * to let all busy credentials be released.
 184          */
 185         class_import_get(imp);
 186         class_destroy_import(imp);
 187         sptlrpc_import_put_sec(imp);
 188         class_import_put(imp);
 189 }
 190
 191 /* configure an RPC client OBD device
 192  *
 193  * lcfg parameters:
 194  * 1 - client UUID
 195  * 2 - server UUID
 196  * 3 - inactive-on-startup
 197  */
 198 int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
 199 {
 200         struct client_obd *cli = &obddev->u.cli;
 201         struct obd_import *imp;
 202         struct obd_uuid server_uuid;
 203         int rq_portal, rp_portal, connect_op;
 204         char *name = obddev->obd_type->typ_name;
 205         int rc;
 206         ENTRY;
 207
 208         /* In a more perfect world, we would hang a ptlrpc_client off of
 209          * obd_type and just use the values from there. */
 210         if (!strcmp(name, LUSTRE_OSC_NAME)) {
 211                 rq_portal = OST_REQUEST_PORTAL;
 212                 rp_portal = OSC_REPLY_PORTAL;
 213                 connect_op = OST_CONNECT;
 214         } else if (!strcmp(name, LUSTRE_MDC_NAME)) {
 215                 rq_portal = MDS_REQUEST_PORTAL;
 216                 rp_portal = MDC_REPLY_PORTAL;
 217                 connect_op = MDS_CONNECT;
 218         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
 219                 rq_portal = MGS_REQUEST_PORTAL;
 220                 rp_portal = MGC_REPLY_PORTAL;
 221                 connect_op = MGS_CONNECT;
 222         } else {
 223                 CERROR("unknown client OBD type \"%s\", can't setup\n",
 224                        name);
 225                 RETURN(-EINVAL);
 226         }
 227
 228         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
 229                 CERROR("requires a TARGET UUID\n");
 230                 RETURN(-EINVAL);
 231         }
 232
 233         if (LUSTRE_CFG_BUFLEN(lcfg, 1) > 37) {
 234                 CERROR("client UUID must be less than 38 characters\n");
 235                 RETURN(-EINVAL);
 236         }
 237
 238         if (LUSTRE_CFG_BUFLEN(lcfg, 2) < 1) {
 239                 CERROR("setup requires a SERVER UUID\n");
 240                 RETURN(-EINVAL);
 241         }
 242
 243         if (LUSTRE_CFG_BUFLEN(lcfg, 2) > 37) {
 244                 CERROR("target UUID must be less than 38 characters\n");
 245                 RETURN(-EINVAL);
 246         }
 247
 248         sema_init(&cli->cl_sem, 1);
 249         sema_init(&cli->cl_mgc_sem, 1);
 250         cli->cl_sec_conf.sfc_rpc_flavor = SPTLRPC_FLVR_NULL;
 251         cli->cl_sec_conf.sfc_bulk_csum = BULK_CSUM_ALG_NULL;
 252         cli->cl_sec_conf.sfc_bulk_priv = BULK_PRIV_ALG_NULL;
 253         cli->cl_sec_conf.sfc_flags = 0;
 254         cli->cl_conn_count = 0;
 255         memcpy(server_uuid.uuid, lustre_cfg_buf(lcfg, 2),
 256                min_t(unsigned int, LUSTRE_CFG_BUFLEN(lcfg, 2),
 257                      sizeof(server_uuid)));
 258
 259         cli->cl_dirty = 0;
 260         cli->cl_avail_grant = 0;
 261         /* FIXME: should limit this for the sum of all cl_dirty_max */
 262         cli->cl_dirty_max = OSC_MAX_DIRTY_DEFAULT * 1024 * 1024;
 263         if (cli->cl_dirty_max >> CFS_PAGE_SHIFT > num_physpages / 8)
 264                 cli->cl_dirty_max = num_physpages << (CFS_PAGE_SHIFT - 3);
 265         CFS_INIT_LIST_HEAD(&cli->cl_cache_waiters);
 266         CFS_INIT_LIST_HEAD(&cli->cl_loi_ready_list);
 267         CFS_INIT_LIST_HEAD(&cli->cl_loi_write_list);
 268         CFS_INIT_LIST_HEAD(&cli->cl_loi_read_list);
 269         client_obd_list_lock_init(&cli->cl_loi_list_lock);
 270         cli->cl_r_in_flight = 0;
 271         cli->cl_w_in_flight = 0;
 272
 273         spin_lock_init(&cli->cl_read_rpc_hist.oh_lock);
 274         spin_lock_init(&cli->cl_write_rpc_hist.oh_lock);
 275         spin_lock_init(&cli->cl_read_page_hist.oh_lock);
 276         spin_lock_init(&cli->cl_write_page_hist.oh_lock);
 277         spin_lock_init(&cli->cl_read_offset_hist.oh_lock);
 278         spin_lock_init(&cli->cl_write_offset_hist.oh_lock);
 279 #ifdef ENABLE_CHECKSUM
 280         cli->cl_checksum = 1;
 281 #endif
 282         atomic_set(&cli->cl_resends, OSC_DEFAULT_RESENDS);
 283
 284         /* This value may be changed at connect time in
 285            ptlrpc_connect_interpret. */
 286         cli->cl_max_pages_per_rpc = min((int)PTLRPC_MAX_BRW_PAGES,
 287                                         (int)(1024 * 1024 >> CFS_PAGE_SHIFT));
 288
 289         if (!strcmp(name, LUSTRE_MDC_NAME)) {
 290                 cli->cl_max_rpcs_in_flight = MDC_MAX_RIF_DEFAULT;
 291         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 128 /* MB */) {
 292                 cli->cl_max_rpcs_in_flight = 2;
 293         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 256 /* MB */) {
 294                 cli->cl_max_rpcs_in_flight = 3;
 295         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 512 /* MB */) {
 296                 cli->cl_max_rpcs_in_flight = 4;
 297         } else {
 298                 cli->cl_max_rpcs_in_flight = OSC_MAX_RIF_DEFAULT;
 299         }
 300
 301         rc = ldlm_get_ref();
 302         if (rc) {
 303                 CERROR("ldlm_get_ref failed: %d\n", rc);
 304                 GOTO(err, rc);
 305         }
 306
 307         ptlrpc_init_client(rq_portal, rp_portal, name,
 308                            &obddev->obd_ldlm_client);
 309
 310         imp = class_new_import(obddev);
 311         if (imp == NULL)
 312                 GOTO(err_ldlm, rc = -ENOENT);
 313         imp->imp_client = &obddev->obd_ldlm_client;
 314         imp->imp_connect_op = connect_op;
 315         imp->imp_initial_recov = 1;
 316         imp->imp_initial_recov_bk = 0;
 317         CFS_INIT_LIST_HEAD(&imp->imp_pinger_chain);
 318         memcpy(cli->cl_target_uuid.uuid, lustre_cfg_buf(lcfg, 1),
 319                LUSTRE_CFG_BUFLEN(lcfg, 1));
 320         class_import_put(imp);
 321
 322         rc = client_import_add_conn(imp, &server_uuid, 1);
 323         if (rc) {
 324                 CERROR("can't add initial connection\n");
 325                 GOTO(err_import, rc);
 326         }
 327
 328         cli->cl_import = imp;
 329         /* cli->cl_max_mds_{easize,cookiesize} updated by mdc_init_ea_size() */
 330         cli->cl_max_mds_easize = sizeof(struct lov_mds_md);
 331         cli->cl_max_mds_cookiesize = sizeof(struct llog_cookie);
 332
 333         if (LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
 334                 if (!strcmp(lustre_cfg_string(lcfg, 3), "inactive")) {
 335                         CDEBUG(D_HA, "marking %s %s->%s as inactive\n",
 336                                name, obddev->obd_name,
 337                                cli->cl_target_uuid.uuid);
 338                         spin_lock(&imp->imp_lock);
 339                         imp->imp_invalid = 1;
 340                         spin_unlock(&imp->imp_lock);
 341                 }
 342         }
 343
 344         cli->cl_qchk_stat = CL_NOT_QUOTACHECKED;
 345
 346         RETURN(rc);
 347
 348 err_import:
 349         class_destroy_import(imp);
 350 err_ldlm:
 351         ldlm_put_ref(0);
 352 err:
 353         RETURN(rc);
 354
 355 }
 356
 357 int client_obd_cleanup(struct obd_device *obddev)
 358 {
 359         ENTRY;
 360         ldlm_put_ref(obddev->obd_force);
 361         RETURN(0);
 362 }
 363
 364 /* ->o_connect() method for client side (OSC and MDC and MGC) */
 365 int client_connect_import(const struct lu_env *env,
 366                           struct lustre_handle *dlm_handle,
 367                           struct obd_device *obd, struct obd_uuid *cluuid,
 368                           struct obd_connect_data *data)
 369 {
 370         struct client_obd *cli = &obd->u.cli;
 371         struct obd_import *imp = cli->cl_import;
 372         struct obd_export *exp;
 373         struct obd_connect_data *ocd;
 374         struct ldlm_namespace *to_be_freed = NULL;
 375         int rc;
 376         ENTRY;
 377
 378         mutex_down(&cli->cl_sem);
 379         rc = class_connect(dlm_handle, obd, cluuid);
 380         if (rc)
 381                 GOTO(out_sem, rc);
 382
 383         cli->cl_conn_count++;
 384         if (cli->cl_conn_count > 1)
 385                 GOTO(out_sem, rc);
 386         exp = class_conn2export(dlm_handle);
 387
 388         if (obd->obd_namespace != NULL)
 389                 CERROR("already have namespace!\n");
 390         obd->obd_namespace = ldlm_namespace_new(obd->obd_name,
 391                                                 LDLM_NAMESPACE_CLIENT,
 392                                                 LDLM_NAMESPACE_GREEDY);
 393         if (obd->obd_namespace == NULL)
 394                 GOTO(out_disco, rc = -ENOMEM);
 395
 396         imp->imp_dlm_handle = *dlm_handle;
 397         rc = ptlrpc_init_import(imp);
 398         if (rc != 0)
 399                 GOTO(out_ldlm, rc);
 400
 401         rc = sptlrpc_import_get_sec(imp, NULL, cli->cl_sec_conf.sfc_rpc_flavor,
 402                                     cli->cl_sec_conf.sfc_flags);
 403         if (rc)
 404                 GOTO(out_ldlm, rc);
 405
 406         ocd = &imp->imp_connect_data;
 407         if (data) {
 408                 *ocd = *data;
 409                 imp->imp_connect_flags_orig = data->ocd_connect_flags;
 410         }
 411
 412         rc = ptlrpc_connect_import(imp, NULL);
 413         if (rc != 0) {
 414                 LASSERT (imp->imp_state == LUSTRE_IMP_DISCON);
 415                 GOTO(out_ldlm, rc);
 416         }
 417         LASSERT(exp->exp_connection);
 418
 419         if (data) {
 420                 LASSERT((ocd->ocd_connect_flags & data->ocd_connect_flags) ==
 421                         ocd->ocd_connect_flags);
 422                 data->ocd_connect_flags = ocd->ocd_connect_flags;
 423         }
 424
 425         ptlrpc_pinger_add_import(imp);
 426
 427         EXIT;
 428
 429         if (rc) {
 430 out_ldlm:
 431                 ldlm_namespace_free_prior(obd->obd_namespace);
 432                 to_be_freed = obd->obd_namespace;
 433                 obd->obd_namespace = NULL;
 434 out_disco:
 435                 cli->cl_conn_count--;
 436                 class_disconnect(exp);
 437         } else {
 438                 class_export_put(exp);
 439         }
 440 out_sem:
 441         mutex_up(&cli->cl_sem);
 442         if (to_be_freed)
 443                 ldlm_namespace_free_post(to_be_freed, 1);
 444
 445         return rc;
 446 }
 447
 448 int client_disconnect_export(struct obd_export *exp)
 449 {
 450         struct obd_device *obd = class_exp2obd(exp);
 451         struct client_obd *cli;
 452         struct obd_import *imp;
 453         int rc = 0, err;
 454         struct ldlm_namespace *to_be_freed = NULL;
 455         ENTRY;
 456
 457         if (!obd) {
 458                 CERROR("invalid export for disconnect: exp %p cookie "LPX64"\n",
 459                        exp, exp ? exp->exp_handle.h_cookie : -1);
 460                 RETURN(-EINVAL);
 461         }
 462
 463         cli = &obd->u.cli;
 464         imp = cli->cl_import;
 465
 466         mutex_down(&cli->cl_sem);
 467         if (!cli->cl_conn_count) {
 468                 CERROR("disconnecting disconnected device (%s)\n",
 469                        obd->obd_name);
 470                 GOTO(out_sem, rc = -EINVAL);
 471         }
 472
 473         cli->cl_conn_count--;
 474         if (cli->cl_conn_count)
 475                 GOTO(out_no_disconnect, rc = 0);
 476
 477         /* Mark import deactivated now, so we don't try to reconnect if any
 478          * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
 479          * fully deactivate the import, or that would drop all requests. */
 480         spin_lock(&imp->imp_lock);
 481         imp->imp_deactive = 1;
 482         spin_unlock(&imp->imp_lock);
 483
 484         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
 485          * delete it regardless.  (It's safe to delete an import that was
 486          * never added.) */
 487         (void)ptlrpc_pinger_del_import(imp);
 488
 489         if (obd->obd_namespace != NULL) {
 490                 /* obd_force == local only */
 491                 ldlm_cli_cancel_unused(obd->obd_namespace, NULL,
 492                                        obd->obd_force ? LDLM_FL_LOCAL_ONLY:0,
 493                                        NULL);
 494                 ldlm_namespace_free_prior(obd->obd_namespace);
 495                 to_be_freed = obd->obd_namespace;
 496         }
 497
 498         rc = ptlrpc_disconnect_import(imp, 0);
 499
 500         ptlrpc_invalidate_import(imp);
 501         /* set obd_namespace to NULL only after invalidate, because we can have
 502          * some connect requests in flight, and his need store a connect flags
 503          * in obd_namespace. bug 14260 */
 504         obd->obd_namespace = NULL;
 505
 506         ptlrpc_free_rq_pool(imp->imp_rq_pool);
 507         destroy_import(imp);
 508         cli->cl_import = NULL;
 509
 510         EXIT;
 511  out_no_disconnect:
 512         err = class_disconnect(exp);
 513         if (!rc && err)
 514                 rc = err;
 515  out_sem:
 516         mutex_up(&cli->cl_sem);
 517         if (to_be_freed)
 518                 ldlm_namespace_free_post(to_be_freed, obd->obd_force);
 519
 520         RETURN(rc);
 521 }
 522
 523 /* --------------------------------------------------------------------------
 524  * from old lib/target.c
 525  * -------------------------------------------------------------------------- */
 526
 527 int target_handle_reconnect(struct lustre_handle *conn, struct obd_export *exp,
 528                             struct obd_uuid *cluuid, int initial_conn)
 529 {
 530         ENTRY;
 531         if (exp->exp_connection && exp->exp_imp_reverse && !initial_conn) {
 532                 struct lustre_handle *hdl;
 533                 hdl = &exp->exp_imp_reverse->imp_remote_handle;
 534                 /* Might be a re-connect after a partition. */
 535                 if (!memcmp(&conn->cookie, &hdl->cookie, sizeof conn->cookie)) {
 536                         CWARN("%s: %s reconnecting\n", exp->exp_obd->obd_name,
 537                               cluuid->uuid);
 538                         conn->cookie = exp->exp_handle.h_cookie;
 539                         /* target_handle_connect() treats EALREADY and
 540                          * -EALREADY differently.  EALREADY means we are
 541                          * doing a valid reconnect from the same client. */
 542                         RETURN(EALREADY);
 543                 } else {
 544                         CERROR("%s reconnecting from %s, "
 545                                "handle mismatch (ours "LPX64", theirs "
 546                                LPX64")\n", cluuid->uuid,
 547                                exp->exp_connection->c_remote_uuid.uuid,
 548                                hdl->cookie, conn->cookie);
 549                         memset(conn, 0, sizeof *conn);
 550                         /* target_handle_connect() treats EALREADY and
 551                          * -EALREADY differently.  -EALREADY is an error
 552                          * (same UUID, different handle). */
 553                         RETURN(-EALREADY);
 554                 }
 555         }
 556
 557         conn->cookie = exp->exp_handle.h_cookie;
 558         CDEBUG(D_HA, "connect export for UUID '%s' at %p, cookie "LPX64"\n",
 559                cluuid->uuid, exp, conn->cookie);
 560         RETURN(0);
 561 }
 562
 563 void target_client_add_cb(struct obd_device *obd, __u64 transno, void *cb_data,
 564                           int error)
 565 {
 566         struct obd_export *exp = cb_data;
 567
 568         CDEBUG(D_RPCTRACE, "%s: committing for initial connect of %s\n",
 569                obd->obd_name, exp->exp_client_uuid.uuid);
 570
 571         spin_lock(&exp->exp_lock);
 572         exp->exp_need_sync = 0;
 573         spin_unlock(&exp->exp_lock);
 574 }
 575 EXPORT_SYMBOL(target_client_add_cb);
 576
 577 int target_handle_connect(struct ptlrpc_request *req)
 578 {
 579         struct obd_device *target, *targref = NULL;
 580         struct obd_export *export = NULL;
 581         struct obd_import *revimp;
 582         struct lustre_handle conn;
 583         struct obd_uuid tgtuuid;
 584         struct obd_uuid cluuid;
 585         struct obd_uuid remote_uuid;
 586         char *str, *tmp;
 587         int rc = 0;
 588         int initial_conn = 0;
 589         struct obd_connect_data *data;
 590         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*data) };
 591         ENTRY;
 592
 593         OBD_RACE(OBD_FAIL_TGT_CONN_RACE);
 594
 595         lustre_set_req_swabbed(req, REQ_REC_OFF);
 596         str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF, sizeof(tgtuuid)-1);
 597         if (str == NULL) {
 598                 DEBUG_REQ(D_ERROR, req, "bad target UUID for connect");
 599                 GOTO(out, rc = -EINVAL);
 600         }
 601
 602         obd_str2uuid (&tgtuuid, str);
 603         target = class_uuid2obd(&tgtuuid);
 604         if (!target)
 605                 target = class_name2obd(str);
 606
 607         if (!target || target->obd_stopping || !target->obd_set_up) {
 608                 LCONSOLE_ERROR_MSG(0x137, "UUID '%s' is not available "
 609                                    " for connect (%s)\n", str,
 610                                    !target ? "no target" :
 611                                    (target->obd_stopping ? "stopping" :
 612                                    "not set up"));
 613                 GOTO(out, rc = -ENODEV);
 614         }
 615
 616         if (target->obd_no_conn) {
 617                 LCONSOLE_WARN("%s: temporarily refusing client connection "
 618                               "from %s\n", target->obd_name,
 619                               libcfs_nid2str(req->rq_peer.nid));
 620                 GOTO(out, rc = -EAGAIN);
 621         }
 622
 623         /* Make sure the target isn't cleaned up while we're here. Yes,
 624            there's still a race between the above check and our incref here.
 625            Really, class_uuid2obd should take the ref. */
 626         targref = class_incref(target);
 627
 628         lustre_set_req_swabbed(req, REQ_REC_OFF + 1);
 629         str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF + 1,
 630                                 sizeof(cluuid) - 1);
 631         if (str == NULL) {
 632                 DEBUG_REQ(D_ERROR, req, "bad client UUID for connect");
 633                 GOTO(out, rc = -EINVAL);
 634         }
 635
 636         obd_str2uuid (&cluuid, str);
 637
 638         /* XXX extract a nettype and format accordingly */
 639         switch (sizeof(lnet_nid_t)) {
 640                 /* NB the casts only avoid compiler warnings */
 641         case 8:
 642                 snprintf(remote_uuid.uuid, sizeof remote_uuid,
 643                          "NET_"LPX64"_UUID", (__u64)req->rq_peer.nid);
 644                 break;
 645         case 4:
 646                 snprintf(remote_uuid.uuid, sizeof remote_uuid,
 647                          "NET_%x_UUID", (__u32)req->rq_peer.nid);
 648                 break;
 649         default:
 650                 LBUG();
 651         }
 652
 653         tmp = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn);
 654         if (tmp == NULL)
 655                 GOTO(out, rc = -EPROTO);
 656
 657         memcpy(&conn, tmp, sizeof conn);
 658
 659         data = lustre_swab_reqbuf(req, REQ_REC_OFF + 3, sizeof(*data),
 660                                   lustre_swab_connect);
 661
 662         if (!data)
 663                 GOTO(out, rc = -EPROTO);
 664
 665         rc = lustre_pack_reply(req, 2, size, NULL);
 666         if (rc)
 667                 GOTO(out, rc);
 668
 669         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
 670                 if (!data) {
 671                         DEBUG_REQ(D_WARNING, req, "Refusing old (unversioned) "
 672                                   "libclient connection attempt");
 673                         GOTO(out, rc = -EPROTO);
 674                 } else if (data->ocd_version < LUSTRE_VERSION_CODE -
 675                                                LUSTRE_VERSION_ALLOWED_OFFSET ||
 676                            data->ocd_version > LUSTRE_VERSION_CODE +
 677                                                LUSTRE_VERSION_ALLOWED_OFFSET) {
 678                         DEBUG_REQ(D_WARNING, req, "Refusing %s (%d.%d.%d.%d) "
 679                                   "libclient connection attempt",
 680                                   data->ocd_version < LUSTRE_VERSION_CODE ?
 681                                   "old" : "new",
 682                                   OBD_OCD_VERSION_MAJOR(data->ocd_version),
 683                                   OBD_OCD_VERSION_MINOR(data->ocd_version),
 684                                   OBD_OCD_VERSION_PATCH(data->ocd_version),
 685                                   OBD_OCD_VERSION_FIX(data->ocd_version));
 686                         data = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 687                                               offsetof(typeof(*data),
 688                                                        ocd_version) +
 689                                               sizeof(data->ocd_version));
 690                         if (data) {
 691                                 data->ocd_connect_flags = OBD_CONNECT_VERSION;
 692                                 data->ocd_version = LUSTRE_VERSION_CODE;
 693                         }
 694                         GOTO(out, rc = -EPROTO);
 695                 }
 696         }
 697
 698         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_INITIAL)
 699                 initial_conn = 1;
 700
 701         /* lctl gets a backstage, all-access pass. */
 702         if (obd_uuid_equals(&cluuid, &target->obd_uuid))
 703                 goto dont_check_exports;
 704
 705         spin_lock(&target->obd_dev_lock);
 706         export = lustre_hash_get_object_by_key(target->obd_uuid_hash_body, &cluuid);
 707
 708         if (export != NULL && export->exp_connecting) { /* bug 9635, et. al. */
 709                 CWARN("%s: exp %p already connecting\n",
 710                       export->exp_obd->obd_name, export);
 711                 class_export_put(export);
 712                 export = NULL;
 713                 rc = -EALREADY;
 714         } else if (export != NULL && export->exp_connection != NULL &&
 715                    req->rq_peer.nid != export->exp_connection->c_peer.nid) {
 716                 /* make darn sure this is coming from the same peer
 717                  * if the UUIDs matched */
 718                   CWARN("%s: cookie %s seen on new NID %s when "
 719                           "existing NID %s is already connected\n",
 720                         target->obd_name, cluuid.uuid,
 721                   libcfs_nid2str(req->rq_peer.nid),
 722                   libcfs_nid2str(export->exp_connection->c_peer.nid));
 723                   class_export_put(export);
 724                   export = NULL;
 725                   rc = -EALREADY;
 726         } else if (export != NULL) {
 727                 spin_lock(&export->exp_lock);
 728                 export->exp_connecting = 1;
 729                 spin_unlock(&export->exp_lock);
 730                 class_export_put(export);
 731                 spin_unlock(&target->obd_dev_lock);
 732                 LASSERT(export->exp_obd == target);
 733
 734                 rc = target_handle_reconnect(&conn, export, &cluuid, initial_conn);
 735         }
 736
 737         /* If we found an export, we already unlocked. */
 738         if (!export) {
 739                 spin_unlock(&target->obd_dev_lock);
 740                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_CONNECT, 2 * obd_timeout);
 741         } else if (req->rq_export == NULL &&
 742                    atomic_read(&export->exp_rpc_count) > 0) {
 743                 CWARN("%s: refuse connection from %s/%s to 0x%p/%d\n",
 744                       target->obd_name, cluuid.uuid,
 745                       libcfs_nid2str(req->rq_peer.nid),
 746                       export, atomic_read(&export->exp_refcount));
 747                 GOTO(out, rc = -EBUSY);
 748         } else if (req->rq_export != NULL &&
 749                    (atomic_read(&export->exp_rpc_count) > 1)) {
 750                 CWARN("%s: refuse reconnection from %s@%s to 0x%p/%d\n",
 751                       target->obd_name, cluuid.uuid,
 752                       libcfs_nid2str(req->rq_peer.nid),
 753                       export, atomic_read(&export->exp_rpc_count));
 754                 GOTO(out, rc = -EBUSY);
 755         } else if (lustre_msg_get_conn_cnt(req->rq_reqmsg) == 1 &&
 756                    !initial_conn) {
 757                 CERROR("%s: NID %s (%s) reconnected with 1 conn_cnt; "
 758                        "cookies not random?\n", target->obd_name,
 759                        libcfs_nid2str(req->rq_peer.nid), cluuid.uuid);
 760                 GOTO(out, rc = -EALREADY);
 761         } else {
 762                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_RECONNECT, 2 * obd_timeout);
 763                 if (req->rq_export == NULL && initial_conn)
 764                        export->exp_last_request_time =
 765                                max(export->exp_last_request_time,
 766                                    (time_t)CURRENT_SECONDS);
 767         }
 768
 769         /* We want to handle EALREADY but *not* -EALREADY from
 770          * target_handle_reconnect(), return reconnection state in a flag */
 771         if (rc == EALREADY) {
 772                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECONNECT);
 773                 rc = 0;
 774         } else if (rc) {
 775                 GOTO(out, rc);
 776         }
 777         /* Tell the client if we're in recovery. */
 778         /* If this is the first client, start the recovery timer */
 779         CWARN("%s: connection from %s@%s %st"LPU64" exp %p cur %ld last %ld\n",
 780                target->obd_name, cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
 781               target->obd_recovering ? "recovering/" : "", data->ocd_transno,
 782               export, (long)CURRENT_SECONDS,
 783               export ? (long)export->exp_last_request_time : 0);
 784
 785
 786         if (target->obd_recovering) {
 787                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
 788                 target_start_recovery_timer(target);
 789         }
 790
 791         /* Tell the client if we support replayable requests */
 792         if (target->obd_replayable)
 793                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
 794
 795         if (export == NULL) {
 796                 if (target->obd_recovering) {
 797                         CERROR("%s: denying connection for new client %s (%s): "
 798                                "%d clients in recovery for %lds\n",
 799                                target->obd_name,
 800                                libcfs_nid2str(req->rq_peer.nid), cluuid.uuid,
 801                                target->obd_recoverable_clients,
 802                                cfs_duration_sec(cfs_time_sub(cfs_timer_deadline(&target->obd_recovery_timer),
 803                                                              cfs_time_current())));
 804                         rc = -EBUSY;
 805                 } else {
 806 dont_check_exports:
 807                         rc = obd_connect(req->rq_svc_thread->t_env,
 808                                          &conn, target, &cluuid, data);
 809                 }
 810         } else {
 811                 rc = obd_reconnect(export, target, &cluuid, data);
 812         }
 813         if (rc)
 814                 GOTO(out, rc);
 815         /* Return only the parts of obd_connect_data that we understand, so the
 816          * client knows that we don't understand the rest. */
 817         if (data)
 818                 memcpy(lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 819                                       sizeof(*data)),
 820                        data, sizeof(*data));
 821
 822         /* If all else goes well, this is our RPC return code. */
 823         req->rq_status = 0;
 824
 825         lustre_msg_set_handle(req->rq_repmsg, &conn);
 826
 827         /* ownership of this export ref transfers to the request AFTER we
 828          * drop any previous reference the request had, but we don't want
 829          * that to go to zero before we get our new export reference. */
 830         export = class_conn2export(&conn);
 831         if (!export) {
 832                 DEBUG_REQ(D_ERROR, req, "Missing export!");
 833                 GOTO(out, rc = -ENODEV);
 834         }
 835
 836         /* If the client and the server are the same node, we will already
 837          * have an export that really points to the client's DLM export,
 838          * because we have a shared handles table.
 839          *
 840          * XXX this will go away when shaver stops sending the "connect" handle
 841          * in the real "remote handle" field of the request --phik 24 Apr 2003
 842          */
 843         if (req->rq_export != NULL)
 844                 class_export_put(req->rq_export);
 845
 846         req->rq_export = export;
 847
 848         spin_lock(&export->exp_lock);
 849         if (initial_conn) {
 850                 lustre_msg_set_conn_cnt(req->rq_repmsg, export->exp_conn_cnt + 1);
 851         } else if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
 852                 spin_unlock(&export->exp_lock);
 853                 CERROR("%s: %s already connected at higher conn_cnt: %d > %d\n",
 854                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
 855                        export->exp_conn_cnt,
 856                        lustre_msg_get_conn_cnt(req->rq_reqmsg));
 857
 858                 GOTO(out, rc = -EALREADY);
 859         }
 860         export->exp_conn_cnt = lustre_msg_get_conn_cnt(req->rq_reqmsg);
 861
 862         /* request from liblustre?  Don't evict it for not pinging. */
 863         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
 864                 export->exp_libclient = 1;
 865                 spin_unlock(&export->exp_lock);
 866
 867                 spin_lock(&target->obd_dev_lock);
 868                 list_del_init(&export->exp_obd_chain_timed);
 869                 spin_unlock(&target->obd_dev_lock);
 870         } else {
 871                 spin_unlock(&export->exp_lock);
 872         }
 873
 874         if (export->exp_connection != NULL)
 875                 ptlrpc_put_connection(export->exp_connection);
 876         export->exp_connection = ptlrpc_get_connection(req->rq_peer,
 877                                                        req->rq_self,
 878                                                        &remote_uuid);
 879
 880         spin_lock(&target->obd_dev_lock);
 881         /* Export might be hashed already, e.g. if this is reconnect */
 882         if (hlist_unhashed(&export->exp_nid_hash))
 883                 lustre_hash_additem(export->exp_obd->obd_nid_hash_body,
 884                                     &export->exp_connection->c_peer.nid,
 885                                     &export->exp_nid_hash);
 886         spin_unlock(&target->obd_dev_lock);
 887
 888         spin_lock_bh(&target->obd_processing_task_lock);
 889         if (target->obd_recovering && !export->exp_in_recovery) {
 890                 spin_lock(&export->exp_lock);
 891                 export->exp_in_recovery = 1;
 892                 export->exp_req_replay_needed = 1;
 893                 export->exp_lock_replay_needed = 1;
 894                 spin_unlock(&export->exp_lock);
 895                 if ((lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_TRANSNO)
 896                      && data->ocd_transno < target->obd_next_recovery_transno)
 897                         target->obd_next_recovery_transno = data->ocd_transno;
 898                 target->obd_connected_clients++;
 899                 /* each connected client is counted as recoverable */
 900                 target->obd_recoverable_clients++;
 901                 atomic_inc(&target->obd_req_replay_clients);
 902                 atomic_inc(&target->obd_lock_replay_clients);
 903                 if (target->obd_connected_clients ==
 904                     target->obd_max_recoverable_clients)
 905                         wake_up(&target->obd_next_transno_waitq);
 906         }
 907         spin_unlock_bh(&target->obd_processing_task_lock);
 908         memcpy(&conn,
 909                lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn),
 910                sizeof conn);
 911
 912         if (export->exp_imp_reverse != NULL) {
 913                 /* destroyed import can be still referenced in ctxt */
 914                 obd_set_info_async(export, strlen(KEY_REVIMP_UPD),
 915                                    KEY_REVIMP_UPD, 0, NULL, NULL);
 916                 destroy_import(export->exp_imp_reverse);
 917         }
 918
 919         /* for the rest part, we return -ENOTCONN in case of errors
 920          * in order to let client initialize connection again.
 921          */
 922         revimp = export->exp_imp_reverse = class_new_import(target);
 923         if (!revimp) {
 924                 CERROR("fail to alloc new reverse import.\n");
 925                 GOTO(out, rc = -ENOTCONN);
 926         }
 927
 928         revimp->imp_connection = ptlrpc_connection_addref(export->exp_connection);
 929         revimp->imp_client = &export->exp_obd->obd_ldlm_client;
 930         revimp->imp_remote_handle = conn;
 931         revimp->imp_dlm_fake = 1;
 932         revimp->imp_state = LUSTRE_IMP_FULL;
 933
 934         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_NEXT_VER) {
 935                 revimp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
 936                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_NEXT_VER);
 937         }
 938
 939         rc = sptlrpc_import_get_sec(revimp, req->rq_svc_ctx,
 940                                     req->rq_sec_flavor, 0);
 941         if (rc) {
 942                 CERROR("Failed to get sec for reverse import: %d\n", rc);
 943                 export->exp_imp_reverse = NULL;
 944                 class_destroy_import(revimp);
 945         }
 946
 947         class_import_put(revimp);
 948 out:
 949         if (export) {
 950                 spin_lock(&export->exp_lock);
 951                 export->exp_connecting = 0;
 952                 spin_unlock(&export->exp_lock);
 953         }
 954         if (targref)
 955                 class_decref(targref);
 956         if (rc)
 957                 req->rq_status = rc;
 958         RETURN(rc);
 959 }
 960
 961 int target_handle_disconnect(struct ptlrpc_request *req)
 962 {
 963         int rc;
 964         ENTRY;
 965
 966         rc = lustre_pack_reply(req, 1, NULL, NULL);
 967         if (rc)
 968                 RETURN(rc);
 969
 970         /* keep the rq_export around so we can send the reply */
 971         req->rq_status = obd_disconnect(class_export_get(req->rq_export));
 972
 973         RETURN(0);
 974 }
 975
 976 void target_destroy_export(struct obd_export *exp)
 977 {
 978         /* exports created from last_rcvd data, and "fake"
 979            exports created by lctl don't have an import */
 980         if (exp->exp_imp_reverse != NULL)
 981                 destroy_import(exp->exp_imp_reverse);
 982
 983         /* We cancel locks at disconnect time, but this will catch any locks
 984          * granted in a race with recovery-induced disconnect. */
 985         if (exp->exp_obd->obd_namespace != NULL)
 986                 ldlm_cancel_locks_for_export(exp);
 987 }
 988
 989 /*
 990  * Recovery functions
 991  */
 992
 993 struct ptlrpc_request *ptlrpc_clone_req( struct ptlrpc_request *orig_req)
 994 {
 995         struct ptlrpc_request *copy_req;
 996         struct lustre_msg *copy_reqmsg;
 997         struct ptlrpc_user_desc *udesc = NULL;
 998
 999         OBD_ALLOC_PTR(copy_req);
1000         if (!copy_req)
1001                 return NULL;
1002         OBD_ALLOC(copy_reqmsg, orig_req->rq_reqlen);
1003         if (!copy_reqmsg){
1004                 OBD_FREE_PTR(copy_req);
1005                 return NULL;
1006         }
1007
1008         if (orig_req->rq_user_desc) {
1009                 int ngroups = orig_req->rq_user_desc->pud_ngroups;
1010
1011                 OBD_ALLOC(udesc, sptlrpc_user_desc_size(ngroups));
1012                 if (!udesc) {
1013                         OBD_FREE(copy_reqmsg, orig_req->rq_reqlen);
1014                         OBD_FREE_PTR(copy_req);
1015                         return NULL;
1016                 }
1017                 memcpy(udesc, orig_req->rq_user_desc,
1018                        sptlrpc_user_desc_size(ngroups));
1019         }
1020
1021         *copy_req = *orig_req;
1022         memcpy(copy_reqmsg, orig_req->rq_reqmsg, orig_req->rq_reqlen);
1023         copy_req->rq_reqmsg = copy_reqmsg;
1024         copy_req->rq_user_desc = udesc;
1025
1026         class_export_get(copy_req->rq_export);
1027         CFS_INIT_LIST_HEAD(&copy_req->rq_list);
1028         sptlrpc_svc_ctx_addref(copy_req);
1029
1030         if (copy_req->rq_reply_state) {
1031                 /* the copied req takes over the reply state */
1032                 orig_req->rq_reply_state = NULL;
1033                 /* to catch further access */
1034                 orig_req->rq_repmsg = NULL;
1035                 orig_req->rq_replen = 0;
1036         }
1037
1038         return copy_req;
1039 }
1040
1041 void ptlrpc_free_clone( struct ptlrpc_request *req)
1042 {
1043         if (req->rq_reply_state) {
1044                 ptlrpc_rs_decref(req->rq_reply_state);
1045                 req->rq_reply_state = NULL;
1046         }
1047
1048         sptlrpc_svc_ctx_decref(req);
1049         class_export_put(req->rq_export);
1050         list_del(&req->rq_list);
1051
1052         if (req->rq_user_desc) {
1053                 int ngroups = req->rq_user_desc->pud_ngroups;
1054                 OBD_FREE(req->rq_user_desc, sptlrpc_user_desc_size(ngroups));
1055         }
1056         OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
1057         OBD_FREE_PTR(req);
1058 }
1059
1060 #ifdef __KERNEL__
1061 static void target_finish_recovery(struct obd_device *obd)
1062 {
1063         ENTRY;
1064         LCONSOLE_INFO("%s: sending delayed replies to recovered clients\n",
1065                       obd->obd_name);
1066
1067         ldlm_reprocess_all_ns(obd->obd_namespace);
1068
1069         /* when recovery finished, cleanup orphans on mds and ost */
1070         if (OBT(obd) && OBP(obd, postrecov)) {
1071                 int rc = OBP(obd, postrecov)(obd);
1072                 LCONSOLE_WARN("%s: recovery %s: rc %d\n", obd->obd_name,
1073                               rc < 0 ? "failed" : "complete", rc);
1074         }
1075
1076         obd->obd_recovery_end = CURRENT_SECONDS;
1077         EXIT;
1078 }
1079
1080 static void abort_req_replay_queue(struct obd_device *obd)
1081 {
1082         struct ptlrpc_request *req, *n;
1083
1084         list_for_each_entry_safe(req, n, &obd->obd_req_replay_queue, rq_list) {
1085                 DEBUG_REQ(D_WARNING, req, "aborted:");
1086                 req->rq_status = -ENOTCONN;
1087                 if (ptlrpc_error(req)) {
1088                         DEBUG_REQ(D_ERROR, req,
1089                                   "failed abort_req_reply; skipping");
1090                 }
1091                 ptlrpc_free_clone(req);
1092         }
1093 }
1094
1095 static void abort_lock_replay_queue(struct obd_device *obd)
1096 {
1097         struct ptlrpc_request *req, *n;
1098
1099         list_for_each_entry_safe(req, n, &obd->obd_lock_replay_queue, rq_list){
1100                 DEBUG_REQ(D_ERROR, req, "aborted:");
1101                 req->rq_status = -ENOTCONN;
1102                 if (ptlrpc_error(req)) {
1103                         DEBUG_REQ(D_ERROR, req,
1104                                   "failed abort_lock_reply; skipping");
1105                 }
1106                 ptlrpc_free_clone(req);
1107         }
1108 }
1109 #endif
1110
1111 /* Called from a cleanup function if the device is being cleaned up
1112    forcefully.  The exports should all have been disconnected already,
1113    the only thing left to do is
1114      - clear the recovery flags
1115      - cancel the timer
1116      - free queued requests and replies, but don't send replies
1117    Because the obd_stopping flag is set, no new requests should be received.
1118
1119 */
1120 void target_cleanup_recovery(struct obd_device *obd)
1121 {
1122         struct ptlrpc_request *req, *n;
1123         ENTRY;
1124
1125         LASSERT(obd->obd_stopping);
1126
1127         spin_lock_bh(&obd->obd_processing_task_lock);
1128         if (!obd->obd_recovering) {
1129                 spin_unlock_bh(&obd->obd_processing_task_lock);
1130                 EXIT;
1131                 return;
1132         }
1133         obd->obd_recovering = obd->obd_abort_recovery = 0;
1134         target_cancel_recovery_timer(obd);
1135         spin_unlock_bh(&obd->obd_processing_task_lock);
1136
1137         list_for_each_entry_safe(req, n, &obd->obd_req_replay_queue, rq_list) {
1138                 LASSERT (req->rq_reply_state == 0);
1139                 ptlrpc_free_clone(req);
1140         }
1141         list_for_each_entry_safe(req, n, &obd->obd_lock_replay_queue, rq_list){
1142                 LASSERT (req->rq_reply_state == 0);
1143                 ptlrpc_free_clone(req);
1144         }
1145         list_for_each_entry_safe(req, n, &obd->obd_final_req_queue, rq_list) {
1146                 LASSERT (req->rq_reply_state == 0);
1147                 ptlrpc_free_clone(req);
1148         }
1149
1150         EXIT;
1151 }
1152
1153 static void target_recovery_expired(unsigned long castmeharder)
1154 {
1155         struct obd_device *obd = (struct obd_device *)castmeharder;
1156         CERROR("%s: recovery timed out, aborting\n", obd->obd_name);
1157         spin_lock_bh(&obd->obd_processing_task_lock);
1158         if (obd->obd_recovering)
1159                 obd->obd_abort_recovery = 1;
1160         cfs_waitq_signal(&obd->obd_next_transno_waitq);
1161         spin_unlock_bh(&obd->obd_processing_task_lock);
1162 }
1163
1164
1165 /* obd_processing_task_lock should be held */
1166 void target_cancel_recovery_timer(struct obd_device *obd)
1167 {
1168         CDEBUG(D_HA, "%s: cancel recovery timer\n", obd->obd_name);
1169         cfs_timer_disarm(&obd->obd_recovery_timer);
1170 }
1171
1172 static void reset_recovery_timer(struct obd_device *obd)
1173 {
1174         spin_lock_bh(&obd->obd_processing_task_lock);
1175         if (!obd->obd_recovering) {
1176                 spin_unlock_bh(&obd->obd_processing_task_lock);
1177                 return;
1178         }
1179         cfs_timer_arm(&obd->obd_recovery_timer,
1180                       cfs_time_shift(OBD_RECOVERY_TIMEOUT));
1181         spin_unlock_bh(&obd->obd_processing_task_lock);
1182         CDEBUG(D_HA, "%s: timer will expire in %u seconds\n", obd->obd_name,
1183                OBD_RECOVERY_TIMEOUT);
1184         /* Only used for lprocfs_status */
1185         obd->obd_recovery_end = CURRENT_SECONDS + OBD_RECOVERY_TIMEOUT;
1186 }
1187
1188
1189 /* Only start it the first time called */
1190 void target_start_recovery_timer(struct obd_device *obd)
1191 {
1192         spin_lock_bh(&obd->obd_processing_task_lock);
1193         if (obd->obd_recovery_handler
1194             || timer_pending((struct timer_list *)&obd->obd_recovery_timer)) {
1195                 spin_unlock_bh(&obd->obd_processing_task_lock);
1196                 return;
1197         }
1198         CWARN("%s: starting recovery timer (%us)\n", obd->obd_name,
1199               OBD_RECOVERY_TIMEOUT);
1200         cfs_timer_init(&obd->obd_recovery_timer, target_recovery_expired, obd);
1201         spin_unlock_bh(&obd->obd_processing_task_lock);
1202
1203         reset_recovery_timer(obd);
1204 }
1205
1206 #ifdef __KERNEL__
1207 static int check_for_next_transno(struct obd_device *obd)
1208 {
1209         struct ptlrpc_request *req = NULL;
1210         int wake_up = 0, connected, completed, queue_len, max;
1211         __u64 next_transno, req_transno;
1212         ENTRY;
1213         spin_lock_bh(&obd->obd_processing_task_lock);
1214
1215         if (!list_empty(&obd->obd_req_replay_queue)) {
1216                 req = list_entry(obd->obd_req_replay_queue.next,
1217                                  struct ptlrpc_request, rq_list);
1218                 req_transno = lustre_msg_get_transno(req->rq_reqmsg);
1219         } else {
1220                 req_transno = 0;
1221         }
1222
1223         max = obd->obd_max_recoverable_clients;
1224         connected = obd->obd_connected_clients;
1225         completed = connected - obd->obd_recoverable_clients;
1226         queue_len = obd->obd_requests_queued_for_recovery;
1227         next_transno = obd->obd_next_recovery_transno;
1228
1229         CDEBUG(D_HA, "max: %d, connected: %d, completed: %d, queue_len: %d, "
1230                "req_transno: "LPU64", next_transno: "LPU64"\n",
1231                max, connected, completed, queue_len, req_transno, next_transno);
1232
1233         if (obd->obd_abort_recovery) {
1234                 CDEBUG(D_HA, "waking for aborted recovery\n");
1235                 wake_up = 1;
1236         } else if (atomic_read(&obd->obd_req_replay_clients) == 0) {
1237                 CDEBUG(D_HA, "waking for completed recovery\n");
1238                 wake_up = 1;
1239         } else if (req_transno == next_transno) {
1240                 CDEBUG(D_HA, "waking for next ("LPD64")\n", next_transno);
1241                 wake_up = 1;
1242         } else if (queue_len + completed == max) {
1243                 /* handle gaps occured due to lost reply. It is allowed gaps
1244                  * because all clients are connected and there will be resend
1245                  * for missed transaction */
1246                 LASSERTF(req_transno >= next_transno,
1247                          "req_transno: "LPU64", next_transno: "LPU64"\n",
1248                          req_transno, next_transno);
1249
1250                 CDEBUG(req_transno > obd->obd_last_committed ? D_ERROR : D_HA,
1251                        "waking for skipped transno (skip: "LPD64
1252                        ", ql: %d, comp: %d, conn: %d, next: "LPD64")\n",
1253                        next_transno, queue_len, completed, connected, req_transno);
1254                 obd->obd_next_recovery_transno = req_transno;
1255                 wake_up = 1;
1256         } else if (queue_len == atomic_read(&obd->obd_req_replay_clients)) {
1257                 /* some clients haven't connected in time, but we can try
1258                  * to replay requests that demand on already committed ones
1259                  * also, we can replay first non-committed transation */
1260                 LASSERT(req_transno != 0);
1261                 if (req_transno == obd->obd_last_committed + 1) {
1262                         obd->obd_next_recovery_transno = req_transno;
1263                 } else if (req_transno > obd->obd_last_committed) {
1264                         /* can't continue recovery: have no needed transno */
1265                         obd->obd_abort_recovery = 1;
1266                         CDEBUG(D_ERROR, "abort due to missed clients. max: %d, "
1267                                "connected: %d, completed: %d, queue_len: %d, "
1268                                "req_transno: "LPU64", next_transno: "LPU64"\n",
1269                                max, connected, completed, queue_len,
1270                                req_transno, next_transno);
1271                 }
1272                 wake_up = 1;
1273         }
1274
1275         spin_unlock_bh(&obd->obd_processing_task_lock);
1276         return wake_up;
1277 }
1278
1279 static struct ptlrpc_request *target_next_replay_req(struct obd_device *obd)
1280 {
1281         struct l_wait_info lwi = { 0 };
1282         struct ptlrpc_request *req;
1283
1284         CDEBUG(D_HA, "Waiting for transno "LPD64"\n",
1285                obd->obd_next_recovery_transno);
1286         l_wait_event(obd->obd_next_transno_waitq,
1287                      check_for_next_transno(obd), &lwi);
1288
1289         spin_lock_bh(&obd->obd_processing_task_lock);
1290         if (obd->obd_abort_recovery) {
1291                 req = NULL;
1292         } else if (!list_empty(&obd->obd_req_replay_queue)) {
1293                 req = list_entry(obd->obd_req_replay_queue.next,
1294                                  struct ptlrpc_request, rq_list);
1295                 list_del_init(&req->rq_list);
1296                 obd->obd_requests_queued_for_recovery--;
1297         } else {
1298                 req = NULL;
1299         }
1300         spin_unlock_bh(&obd->obd_processing_task_lock);
1301         RETURN(req);
1302 }
1303
1304 static int check_for_next_lock(struct obd_device *obd)
1305 {
1306         struct ptlrpc_request *req = NULL;
1307         int wake_up = 0;
1308
1309         spin_lock_bh(&obd->obd_processing_task_lock);
1310         if (!list_empty(&obd->obd_lock_replay_queue)) {
1311                 req = list_entry(obd->obd_lock_replay_queue.next,
1312                                  struct ptlrpc_request, rq_list);
1313                 CDEBUG(D_HA, "waking for next lock\n");
1314                 wake_up = 1;
1315         } else if (atomic_read(&obd->obd_lock_replay_clients) == 0) {
1316                 CDEBUG(D_HA, "waking for completed lock replay\n");
1317                 wake_up = 1;
1318         } else if (obd->obd_abort_recovery) {
1319                 CDEBUG(D_HA, "waking for aborted recovery\n");
1320                 wake_up = 1;
1321         }
1322         spin_unlock_bh(&obd->obd_processing_task_lock);
1323
1324         return wake_up;
1325 }
1326
1327 static struct ptlrpc_request *target_next_replay_lock(struct obd_device *obd)
1328 {
1329         struct l_wait_info lwi = { 0 };
1330         struct ptlrpc_request *req;
1331
1332         CDEBUG(D_HA, "Waiting for lock\n");
1333         l_wait_event(obd->obd_next_transno_waitq,
1334                      check_for_next_lock(obd), &lwi);
1335
1336         spin_lock_bh(&obd->obd_processing_task_lock);
1337         if (obd->obd_abort_recovery) {
1338                 req = NULL;
1339         } else if (!list_empty(&obd->obd_lock_replay_queue)) {
1340                 req = list_entry(obd->obd_lock_replay_queue.next,
1341                                  struct ptlrpc_request, rq_list);
1342                 list_del_init(&req->rq_list);
1343         } else {
1344                 req = NULL;
1345         }
1346         spin_unlock_bh(&obd->obd_processing_task_lock);
1347         return req;
1348 }
1349
1350 static struct ptlrpc_request *target_next_final_ping(struct obd_device *obd)
1351 {
1352         struct ptlrpc_request *req;
1353
1354         spin_lock_bh(&obd->obd_processing_task_lock);
1355         if (!list_empty(&obd->obd_final_req_queue)) {
1356                 req = list_entry(obd->obd_final_req_queue.next,
1357                                  struct ptlrpc_request, rq_list);
1358                 list_del_init(&req->rq_list);
1359         } else {
1360                 req = NULL;
1361         }
1362         spin_unlock_bh(&obd->obd_processing_task_lock);
1363         return req;
1364 }
1365
1366 static inline int req_replay_done(struct obd_export *exp)
1367 {
1368         return (exp->exp_req_replay_needed == 0);
1369 }
1370
1371 static inline int lock_replay_done(struct obd_export *exp)
1372 {
1373         return (exp->exp_lock_replay_needed == 0);
1374 }
1375
1376 static inline int connect_done(struct obd_export *exp)
1377 {
1378         return (exp->exp_in_recovery != 0);
1379 }
1380
1381 static int check_for_clients(struct obd_device *obd)
1382 {
1383         if (obd->obd_abort_recovery)
1384                 return 1;
1385         LASSERT(obd->obd_connected_clients <= obd->obd_max_recoverable_clients);
1386         if (obd->obd_no_conn == 0 &&
1387             obd->obd_connected_clients == obd->obd_max_recoverable_clients)
1388                 return 1;
1389         return 0;
1390 }
1391
1392 static int handle_recovery_req(struct ptlrpc_thread *thread,
1393                                struct ptlrpc_request *req,
1394                                svc_handler_t handler)
1395 {
1396         int rc;
1397         ENTRY;
1398
1399         rc = lu_context_init(&req->rq_session, LCT_SESSION);
1400         if (rc) {
1401                 CERROR("Failure to initialize session: %d\n", rc);
1402                 return rc;
1403         }
1404         req->rq_session.lc_thread = thread;
1405         lu_context_enter(&req->rq_session);
1406         req->rq_svc_thread = thread;
1407         req->rq_svc_thread->t_env->le_ses = &req->rq_session;
1408
1409         (void)handler(req);
1410
1411         lu_context_exit(&req->rq_session);
1412         lu_context_fini(&req->rq_session);
1413         /* don't reset timer for final stage */
1414         if (!req_replay_done(req->rq_export) ||
1415             !lock_replay_done(req->rq_export))
1416                 reset_recovery_timer(class_exp2obd(req->rq_export));
1417         ptlrpc_free_clone(req);
1418         RETURN(0);
1419 }
1420
1421 static int target_recovery_thread(void *arg)
1422 {
1423         struct obd_device *obd = arg;
1424         struct ptlrpc_request *req;
1425         struct target_recovery_data *trd = &obd->obd_recovery_data;
1426         struct l_wait_info lwi = { 0 };
1427         unsigned long delta;
1428         unsigned long flags;
1429         struct lu_env env;
1430         struct ptlrpc_thread fake_svc_thread, *thread = &fake_svc_thread;
1431         __u32 recov_ctx_tags = LCT_MD_THREAD;
1432         int rc = 0;
1433         ENTRY;
1434
1435         cfs_daemonize("tgt_recov");
1436
1437         SIGNAL_MASK_LOCK(current, flags);
1438         sigfillset(&current->blocked);
1439         RECALC_SIGPENDING;
1440         SIGNAL_MASK_UNLOCK(current, flags);
1441
1442         rc = lu_context_init(&env.le_ctx, recov_ctx_tags);
1443         if (rc)
1444                 RETURN(rc);
1445
1446         thread->t_env = &env;
1447         env.le_ctx.lc_thread = thread;
1448
1449         CERROR("%s: started recovery thread pid %d\n", obd->obd_name,
1450                current->pid);
1451         trd->trd_processing_task = current->pid;
1452
1453         obd->obd_recovering = 1;
1454         complete(&trd->trd_starting);
1455
1456         /* first of all, we have to know the first transno to replay */
1457         obd->obd_abort_recovery = 0;
1458         l_wait_event(obd->obd_next_transno_waitq,
1459                      check_for_clients(obd), &lwi);
1460
1461         spin_lock_bh(&obd->obd_processing_task_lock);
1462         target_cancel_recovery_timer(obd);
1463         spin_unlock_bh(&obd->obd_processing_task_lock);
1464
1465         /* If some clients haven't connected in time, evict them */
1466         if (obd->obd_abort_recovery) {
1467                 CWARN("Some clients haven't connect in time (%d/%d),"
1468                        "evict them\n", obd->obd_connected_clients,
1469                        obd->obd_max_recoverable_clients);
1470                 obd->obd_abort_recovery = obd->obd_stopping;
1471                 class_disconnect_stale_exports(obd, connect_done);
1472         }
1473         /* next stage: replay requests */
1474         delta = jiffies;
1475         obd->obd_req_replaying = 1;
1476         CDEBUG(D_INFO, "1: request replay stage - %d clients from t"LPU64"\n",
1477               atomic_read(&obd->obd_req_replay_clients),
1478               obd->obd_next_recovery_transno);
1479         while ((req = target_next_replay_req(obd))) {
1480                 LASSERT(trd->trd_processing_task == current->pid);
1481                 DEBUG_REQ(D_HA, req, "processing t"LPD64" from %s",
1482                           lustre_msg_get_transno(req->rq_reqmsg),
1483                           libcfs_nid2str(req->rq_peer.nid));
1484
1485                 handle_recovery_req(thread, req,
1486                                     trd->trd_recovery_handler);
1487                 obd->obd_replayed_requests++;
1488                 spin_lock_bh(&obd->obd_processing_task_lock);
1489                 obd->obd_next_recovery_transno++;
1490                 spin_unlock_bh(&obd->obd_processing_task_lock);
1491         }
1492
1493         spin_lock_bh(&obd->obd_processing_task_lock);
1494         target_cancel_recovery_timer(obd);
1495         spin_unlock_bh(&obd->obd_processing_task_lock);
1496
1497         /* If some clients haven't replayed requests in time, evict them */
1498         if (obd->obd_abort_recovery) {
1499                 CDEBUG(D_ERROR, "req replay timed out, aborting ...\n");
1500                 obd->obd_abort_recovery = obd->obd_stopping;
1501                 class_disconnect_stale_exports(obd, req_replay_done);
1502                 abort_req_replay_queue(obd);
1503         }
1504         /* The second stage: replay locks */
1505         CDEBUG(D_INFO, "2: lock replay stage - %d clients\n",
1506                atomic_read(&obd->obd_lock_replay_clients));
1507         while ((req = target_next_replay_lock(obd))) {
1508                 LASSERT(trd->trd_processing_task == current->pid);
1509                 DEBUG_REQ(D_HA|D_WARNING, req, "processing lock from %s: ",
1510                           libcfs_nid2str(req->rq_peer.nid));
1511                 handle_recovery_req(thread, req,
1512                                     trd->trd_recovery_handler);
1513                 obd->obd_replayed_locks++;
1514         }
1515
1516         spin_lock_bh(&obd->obd_processing_task_lock);
1517         target_cancel_recovery_timer(obd);
1518         spin_unlock_bh(&obd->obd_processing_task_lock);
1519         /* If some clients haven't replayed requests in time, evict them */
1520         if (obd->obd_abort_recovery) {
1521                 int stale;
1522                 CERROR("lock replay timed out, aborting ...\n");
1523                 obd->obd_abort_recovery = obd->obd_stopping;
1524                 stale = class_disconnect_stale_exports(obd, lock_replay_done);
1525                 abort_lock_replay_queue(obd);
1526         }
1527
1528         /* We drop recoverying flag to forward all new requests
1529          * to regular mds_handle() since now */
1530         spin_lock_bh(&obd->obd_processing_task_lock);
1531         obd->obd_recovering = obd->obd_abort_recovery = 0;
1532         spin_unlock_bh(&obd->obd_processing_task_lock);
1533         /* The third stage: reply on final pings */
1534         CDEBUG(D_INFO, "3: final stage - process recovery completion pings\n");
1535         while ((req = target_next_final_ping(obd))) {
1536                 LASSERT(trd->trd_processing_task == current->pid);
1537                 DEBUG_REQ(D_HA, req, "processing final ping from %s: ",
1538                           libcfs_nid2str(req->rq_peer.nid));
1539                 handle_recovery_req(thread, req,
1540                                     trd->trd_recovery_handler);
1541         }
1542
1543         delta = (jiffies - delta) / HZ;
1544         CDEBUG(D_INFO,"4: recovery completed in %lus - %d/%d reqs/locks\n",
1545               delta, obd->obd_replayed_requests, obd->obd_replayed_locks);
1546         LASSERT(atomic_read(&obd->obd_req_replay_clients) == 0);
1547         LASSERT(atomic_read(&obd->obd_lock_replay_clients) == 0);
1548         if (delta > obd_timeout * 2) {
1549                 CWARN("too long recovery - read logs\n");
1550                 libcfs_debug_dumplog();
1551         }
1552
1553         target_finish_recovery(obd);
1554
1555         lu_env_fini(&env);
1556         trd->trd_processing_task = 0;
1557         complete(&trd->trd_finishing);
1558         RETURN(rc);
1559 }
1560
1561 int target_start_recovery_thread(struct obd_device *obd, svc_handler_t handler)
1562 {
1563         int rc = 0;
1564         struct target_recovery_data *trd = &obd->obd_recovery_data;
1565
1566         memset(trd, 0, sizeof(*trd));
1567         init_completion(&trd->trd_starting);
1568         init_completion(&trd->trd_finishing);
1569         trd->trd_recovery_handler = handler;
1570
1571         if (kernel_thread(target_recovery_thread, obd, 0) > 0) {
1572                 wait_for_completion(&trd->trd_starting);
1573                 LASSERT(obd->obd_recovering != 0);
1574         } else
1575                 rc = -ECHILD;
1576
1577         return rc;
1578 }
1579
1580 void target_stop_recovery_thread(struct obd_device *obd)
1581 {
1582         spin_lock_bh(&obd->obd_processing_task_lock);
1583         if (obd->obd_recovery_data.trd_processing_task > 0) {
1584                 struct target_recovery_data *trd = &obd->obd_recovery_data;
1585                 CERROR("%s: Aborting recovery\n", obd->obd_name);
1586                 obd->obd_abort_recovery = 1;
1587                 wake_up(&obd->obd_next_transno_waitq);
1588                 spin_unlock_bh(&obd->obd_processing_task_lock);
1589                 wait_for_completion(&trd->trd_finishing);
1590         } else {
1591                 spin_unlock_bh(&obd->obd_processing_task_lock);
1592         }
1593 }
1594
1595 void target_recovery_fini(struct obd_device *obd)
1596 {
1597         class_disconnect_exports(obd);
1598         target_stop_recovery_thread(obd);
1599         target_cleanup_recovery(obd);
1600 }
1601 EXPORT_SYMBOL(target_recovery_fini);
1602
1603 void target_recovery_init(struct obd_device *obd, svc_handler_t handler)
1604 {
1605         if (obd->obd_max_recoverable_clients == 0)
1606                 return;
1607
1608         CWARN("RECOVERY: service %s, %d recoverable clients, "
1609               "last_transno "LPU64"\n", obd->obd_name,
1610               obd->obd_max_recoverable_clients, obd->obd_last_committed);
1611         obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
1612         target_start_recovery_thread(obd, handler);
1613         obd->obd_recovery_start = CURRENT_SECONDS;
1614         /* Only used for lprocfs_status */
1615         obd->obd_recovery_end = obd->obd_recovery_start + OBD_RECOVERY_TIMEOUT;
1616 }
1617 EXPORT_SYMBOL(target_recovery_init);
1618
1619 #endif
1620
1621 int target_process_req_flags(struct obd_device *obd, struct ptlrpc_request *req)
1622 {
1623         struct obd_export *exp = req->rq_export;
1624         LASSERT(exp != NULL);
1625         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
1626                 /* client declares he's ready to replay locks */
1627                 spin_lock_bh(&obd->obd_processing_task_lock);
1628                 if (exp->exp_req_replay_needed) {
1629                         LASSERT(atomic_read(&obd->obd_req_replay_clients) > 0);
1630                         spin_lock(&exp->exp_lock);
1631                         exp->exp_req_replay_needed = 0;
1632                         spin_unlock(&exp->exp_lock);
1633                         atomic_dec(&obd->obd_req_replay_clients);
1634                         LASSERT(obd->obd_recoverable_clients > 0);
1635                         obd->obd_recoverable_clients--;
1636                         if (atomic_read(&obd->obd_req_replay_clients) == 0)
1637                                 CDEBUG(D_HA, "all clients have replayed reqs\n");
1638                         wake_up(&obd->obd_next_transno_waitq);
1639                 }
1640                 spin_unlock_bh(&obd->obd_processing_task_lock);
1641         }
1642         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
1643                 /* client declares he's ready to complete recovery
1644                  * so, we put the request on th final queue */
1645                 spin_lock_bh(&obd->obd_processing_task_lock);
1646                 if (exp->exp_lock_replay_needed) {
1647                         LASSERT(atomic_read(&obd->obd_lock_replay_clients) > 0);
1648                         spin_lock(&exp->exp_lock);
1649                         exp->exp_lock_replay_needed = 0;
1650                         spin_unlock(&exp->exp_lock);
1651                         atomic_dec(&obd->obd_lock_replay_clients);
1652                         if (atomic_read(&obd->obd_lock_replay_clients) == 0)
1653                                 CDEBUG(D_HA, "all clients have replayed locks\n");
1654                         wake_up(&obd->obd_next_transno_waitq);
1655                 }
1656                 spin_unlock_bh(&obd->obd_processing_task_lock);
1657         }
1658
1659         return 0;
1660 }
1661
1662 int target_queue_recovery_request(struct ptlrpc_request *req,
1663                                   struct obd_device *obd)
1664 {
1665         struct list_head *tmp;
1666         int inserted = 0;
1667         __u64 transno = lustre_msg_get_transno(req->rq_reqmsg);
1668
1669         ENTRY;
1670
1671         if (obd->obd_recovery_data.trd_processing_task == cfs_curproc_pid()) {
1672                 /* Processing the queue right now, don't re-add. */
1673                 RETURN(1);
1674         }
1675
1676         target_process_req_flags(obd, req);
1677
1678         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
1679                 /* client declares he's ready to complete recovery
1680                  * so, we put the request on th final queue */
1681                 req = ptlrpc_clone_req(req);
1682                 if (req == NULL)
1683                         RETURN(-ENOMEM);
1684                 DEBUG_REQ(D_HA, req, "queue final req");
1685                 spin_lock_bh(&obd->obd_processing_task_lock);
1686                 if (obd->obd_recovering)
1687                         list_add_tail(&req->rq_list, &obd->obd_final_req_queue);
1688                 else {
1689                         spin_unlock_bh(&obd->obd_processing_task_lock);
1690                         ptlrpc_free_clone(req);
1691                         if (obd->obd_stopping) {
1692                                 RETURN(-ENOTCONN);
1693                         } else {
1694                                 RETURN(1);
1695                         }
1696                 }
1697                 spin_unlock_bh(&obd->obd_processing_task_lock);
1698                 RETURN(0);
1699         }
1700         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
1701                 /* client declares he's ready to replay locks */
1702                 req = ptlrpc_clone_req(req);
1703                 if (req == NULL)
1704                         RETURN(-ENOMEM);
1705                 DEBUG_REQ(D_HA, req, "queue lock replay req");
1706                 spin_lock_bh(&obd->obd_processing_task_lock);
1707                 LASSERT(obd->obd_recovering);
1708                 /* usually due to recovery abort */
1709                 if (!req->rq_export->exp_in_recovery) {
1710                         spin_unlock_bh(&obd->obd_processing_task_lock);
1711                         ptlrpc_free_clone(req);
1712                         RETURN(-ENOTCONN);
1713                 }
1714                 LASSERT(req->rq_export->exp_lock_replay_needed);
1715                 list_add_tail(&req->rq_list, &obd->obd_lock_replay_queue);
1716                 spin_unlock_bh(&obd->obd_processing_task_lock);
1717                 wake_up(&obd->obd_next_transno_waitq);
1718                 RETURN(0);
1719         }
1720
1721         /* CAVEAT EMPTOR: The incoming request message has been swabbed
1722          * (i.e. buflens etc are in my own byte order), but type-dependent
1723          * buffers (eg mds_body, ost_body etc) have NOT been swabbed. */
1724
1725         if (!transno) {
1726                 CFS_INIT_LIST_HEAD(&req->rq_list);
1727                 DEBUG_REQ(D_HA, req, "not queueing");
1728                 RETURN(1);
1729         }
1730
1731         spin_lock_bh(&obd->obd_processing_task_lock);
1732
1733         /* If we're processing the queue, we want don't want to queue this
1734          * message.
1735          *
1736          * Also, if this request has a transno less than the one we're waiting
1737          * for, we should process it now.  It could (and currently always will)
1738          * be an open request for a descriptor that was opened some time ago.
1739          *
1740          * Also, a resent, replayed request that has already been
1741          * handled will pass through here and be processed immediately.
1742          */
1743         CWARN("Next recovery transno: "LPU64", current: "LPU64", replaying: %i\n",
1744               obd->obd_next_recovery_transno, transno, obd->obd_req_replaying);
1745         if (transno < obd->obd_next_recovery_transno && obd->obd_req_replaying) {
1746                 /* Processing the queue right now, don't re-add. */
1747                 LASSERT(list_empty(&req->rq_list));
1748                 spin_unlock_bh(&obd->obd_processing_task_lock);
1749                 RETURN(1);
1750         }
1751         spin_unlock_bh(&obd->obd_processing_task_lock);
1752
1753         /* A resent, replayed request that is still on the queue; just drop it.
1754            The queued request will handle this. */
1755         if ((lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT|MSG_REPLAY)) ==
1756             (MSG_RESENT | MSG_REPLAY)) {
1757                 DEBUG_REQ(D_ERROR, req, "dropping resent queued req");
1758                 RETURN(0);
1759         }
1760
1761         req = ptlrpc_clone_req(req);
1762         if (req == NULL)
1763                 RETURN(-ENOMEM);
1764
1765         spin_lock_bh(&obd->obd_processing_task_lock);
1766         LASSERT(obd->obd_recovering);
1767         if (!req->rq_export->exp_in_recovery) {
1768                 spin_unlock_bh(&obd->obd_processing_task_lock);
1769                 ptlrpc_free_clone(req);
1770                 RETURN(-ENOTCONN);
1771         }
1772         LASSERT(req->rq_export->exp_req_replay_needed);
1773
1774         /* XXX O(n^2) */
1775         list_for_each(tmp, &obd->obd_req_replay_queue) {
1776                 struct ptlrpc_request *reqiter =
1777                         list_entry(tmp, struct ptlrpc_request, rq_list);
1778
1779                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) > transno) {
1780                         list_add_tail(&req->rq_list, &reqiter->rq_list);
1781                         inserted = 1;
1782                         break;
1783                 }
1784         }
1785
1786         if (!inserted)
1787                 list_add_tail(&req->rq_list, &obd->obd_req_replay_queue);
1788
1789         obd->obd_requests_queued_for_recovery++;
1790         wake_up(&obd->obd_next_transno_waitq);
1791         spin_unlock_bh(&obd->obd_processing_task_lock);
1792         RETURN(0);
1793
1794 }
1795
1796 struct obd_device * target_req2obd(struct ptlrpc_request *req)
1797 {
1798         return req->rq_export->exp_obd;
1799 }
1800
1801 static inline struct ldlm_pool *ldlm_exp2pl(struct obd_export *exp)
1802 {
1803         LASSERT(exp != NULL);
1804         return &exp->exp_obd->obd_namespace->ns_pool;
1805 }
1806
1807 int target_pack_pool_reply(struct ptlrpc_request *req)
1808 {
1809         struct ldlm_pool *pl;
1810         ENTRY;
1811
1812         if (!req->rq_export || !exp_connect_lru_resize(req->rq_export)) {
1813                 lustre_msg_set_slv(req->rq_repmsg, 0);
1814                 lustre_msg_set_limit(req->rq_repmsg, 0);
1815                 RETURN(0);
1816         }
1817
1818         pl = ldlm_exp2pl(req->rq_export);
1819
1820         spin_lock(&pl->pl_lock);
1821         LASSERT(ldlm_pool_get_slv(pl) != 0 && ldlm_pool_get_limit(pl) != 0);
1822         lustre_msg_set_slv(req->rq_repmsg, ldlm_pool_get_slv(pl));
1823         lustre_msg_set_limit(req->rq_repmsg, ldlm_pool_get_limit(pl));
1824         spin_unlock(&pl->pl_lock);
1825
1826         RETURN(0);
1827 }
1828
1829 int target_send_reply_msg(struct ptlrpc_request *req, int rc, int fail_id)
1830 {
1831         if (OBD_FAIL_CHECK_ORSET(fail_id & ~OBD_FAIL_ONCE, OBD_FAIL_ONCE)) {
1832                 DEBUG_REQ(D_ERROR, req, "dropping reply");
1833                 return (-ECOMM);
1834         }
1835
1836         if (unlikely(rc)) {
1837                 DEBUG_REQ(D_ERROR, req, "processing error (%d)", rc);
1838                 req->rq_status = rc;
1839                 return (ptlrpc_error(req));
1840         } else {
1841                 DEBUG_REQ(D_NET, req, "sending reply");
1842         }
1843
1844         return (ptlrpc_send_reply(req, 1));
1845 }
1846
1847 void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
1848 {
1849         int                        netrc;
1850         struct ptlrpc_reply_state *rs;
1851         struct obd_device         *obd;
1852         struct obd_export         *exp;
1853         struct ptlrpc_service     *svc;
1854
1855         if (req->rq_no_reply)
1856                 return;
1857
1858         svc = req->rq_rqbd->rqbd_service;
1859         rs = req->rq_reply_state;
1860         if (rs == NULL || !rs->rs_difficult) {
1861                 /* no notifiers */
1862                 target_send_reply_msg (req, rc, fail_id);
1863                 return;
1864         }
1865
1866         /* must be an export if locks saved */
1867         LASSERT (req->rq_export != NULL);
1868         /* req/reply consistent */
1869         LASSERT (rs->rs_service == svc);
1870
1871         /* "fresh" reply */
1872         LASSERT (!rs->rs_scheduled);
1873         LASSERT (!rs->rs_scheduled_ever);
1874         LASSERT (!rs->rs_handled);
1875         LASSERT (!rs->rs_on_net);
1876         LASSERT (rs->rs_export == NULL);
1877         LASSERT (list_empty(&rs->rs_obd_list));
1878         LASSERT (list_empty(&rs->rs_exp_list));
1879
1880         exp = class_export_get (req->rq_export);
1881         obd = exp->exp_obd;
1882
1883         /* disable reply scheduling onto srv_reply_queue while I'm setting up */
1884         rs->rs_scheduled = 1;
1885         rs->rs_on_net    = 1;
1886         rs->rs_xid       = req->rq_xid;
1887         rs->rs_transno   = req->rq_transno;
1888         rs->rs_export    = exp;
1889
1890         spin_lock(&obd->obd_uncommitted_replies_lock);
1891
1892         if (rs->rs_transno > obd->obd_last_committed) {
1893                 /* not committed already */
1894                 list_add_tail (&rs->rs_obd_list,
1895                                &obd->obd_uncommitted_replies);
1896         }
1897
1898         spin_unlock (&obd->obd_uncommitted_replies_lock);
1899         spin_lock (&exp->exp_lock);
1900
1901         list_add_tail (&rs->rs_exp_list, &exp->exp_outstanding_replies);
1902
1903         spin_unlock(&exp->exp_lock);
1904
1905         netrc = target_send_reply_msg (req, rc, fail_id);
1906
1907         spin_lock(&svc->srv_lock);
1908
1909         svc->srv_n_difficult_replies++;
1910
1911         if (netrc != 0) {
1912                 /* error sending: reply is off the net.  Also we need +1
1913                  * reply ref until ptlrpc_server_handle_reply() is done
1914                  * with the reply state (if the send was successful, there
1915                  * would have been +1 ref for the net, which
1916                  * reply_out_callback leaves alone) */
1917                 rs->rs_on_net = 0;
1918                 ptlrpc_rs_addref(rs);
1919                 atomic_inc (&svc->srv_outstanding_replies);
1920         }
1921
1922         if (!rs->rs_on_net ||                   /* some notifier */
1923             list_empty(&rs->rs_exp_list) ||     /* completed already */
1924             list_empty(&rs->rs_obd_list)) {
1925                 list_add_tail (&rs->rs_list, &svc->srv_reply_queue);
1926                 cfs_waitq_signal (&svc->srv_waitq);
1927         } else {
1928                 list_add (&rs->rs_list, &svc->srv_active_replies);
1929                 rs->rs_scheduled = 0;           /* allow notifier to schedule */
1930         }
1931
1932         spin_unlock(&svc->srv_lock);
1933 }
1934
1935 int target_handle_ping(struct ptlrpc_request *req)
1936 {
1937         obd_ping(req->rq_export);
1938         return lustre_pack_reply(req, 1, NULL, NULL);
1939 }
1940
1941 void target_committed_to_req(struct ptlrpc_request *req)
1942 {
1943         struct obd_device *obd;
1944
1945         if (req == NULL || req->rq_export == NULL)
1946                 return;
1947
1948         obd = req->rq_export->exp_obd;
1949         if (obd == NULL)
1950                 return;
1951
1952         if (!obd->obd_no_transno && req->rq_repmsg != NULL)
1953                 lustre_msg_set_last_committed(req->rq_repmsg,
1954                                               obd->obd_last_committed);
1955         else
1956                 DEBUG_REQ(D_IOCTL, req, "not sending last_committed update");
1957
1958         CDEBUG(D_INFO, "last_committed "LPU64", transno "LPU64", xid "LPU64"\n",
1959                obd->obd_last_committed, req->rq_transno, req->rq_xid);
1960 }
1961
1962 EXPORT_SYMBOL(target_committed_to_req);
1963
1964 #ifdef HAVE_QUOTA_SUPPORT
1965 int target_handle_qc_callback(struct ptlrpc_request *req)
1966 {
1967         struct obd_quotactl *oqctl;
1968         struct client_obd *cli = &req->rq_export->exp_obd->u.cli;
1969
1970         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1971                                    lustre_swab_obd_quotactl);
1972         if (oqctl == NULL) {
1973                 CERROR("Can't unpack obd_quotactl\n");
1974                 RETURN(-EPROTO);
1975         }
1976
1977         cli->cl_qchk_stat = oqctl->qc_stat;
1978
1979         return 0;
1980 }
1981
1982 int target_handle_dqacq_callback(struct ptlrpc_request *req)
1983 {
1984 #ifdef __KERNEL__
1985         struct obd_device *obd = req->rq_export->exp_obd;
1986         struct obd_device *master_obd;
1987         struct lustre_quota_ctxt *qctxt;
1988         struct qunit_data *qdata;
1989         void* rep;
1990         struct qunit_data_old *qdata_old;
1991         int rc = 0;
1992         int repsize[2] = { sizeof(struct ptlrpc_body),
1993                            sizeof(struct qunit_data) };
1994         ENTRY;
1995
1996         rc = lustre_pack_reply(req, 2, repsize, NULL);
1997         if (rc)
1998                 RETURN(rc);
1999
2000         LASSERT(req->rq_export);
2001
2002         /* fixed for bug10707 */
2003         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_QUOTA64) &&
2004             !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT)) {
2005                 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
2006                 rep = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
2007                                      sizeof(struct qunit_data));
2008                 LASSERT(rep);
2009                 qdata = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*qdata),
2010                                            lustre_swab_qdata);
2011         } else {
2012                 CDEBUG(D_QUOTA, "qd_count is 32bit!\n");
2013                 rep = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
2014                                      sizeof(struct qunit_data_old));
2015                 LASSERT(rep);
2016                 qdata_old = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*qdata_old),
2017                                                lustre_swab_qdata_old);
2018                 qdata = lustre_quota_old_to_new(qdata_old);
2019         }
2020
2021         if (qdata == NULL) {
2022                 CERROR("Can't unpack qunit_data\n");
2023                 RETURN(-EPROTO);
2024         }
2025
2026         /* we use the observer */
2027         LASSERT(obd->obd_observer && obd->obd_observer->obd_observer);
2028         master_obd = obd->obd_observer->obd_observer;
2029         qctxt = &master_obd->u.obt.obt_qctxt;
2030
2031         LASSERT(qctxt->lqc_handler);
2032         rc = qctxt->lqc_handler(master_obd, qdata,
2033                                 lustre_msg_get_opc(req->rq_reqmsg));
2034         if (rc && rc != -EDQUOT)
2035                 CDEBUG(rc == -EBUSY  ? D_QUOTA : D_ERROR,
2036                        "dqacq failed! (rc:%d)\n", rc);
2037
2038         /* the qd_count might be changed in lqc_handler */
2039         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_QUOTA64) &&
2040             !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT)) {
2041                 memcpy(rep,qdata,sizeof(*qdata));
2042         } else {
2043                 qdata_old = lustre_quota_new_to_old(qdata);
2044                 memcpy(rep,qdata_old,sizeof(*qdata_old));
2045         }
2046         req->rq_status = rc;
2047         rc = ptlrpc_reply(req);
2048
2049         RETURN(rc);
2050 #else
2051         return 0;
2052 #endif /* !__KERNEL__ */
2053 }
2054 #endif /* HAVE_QUOTA_SUPPORT */
2055
2056 ldlm_mode_t lck_compat_array[] = {
2057         [LCK_EX] LCK_COMPAT_EX,
2058         [LCK_PW] LCK_COMPAT_PW,
2059         [LCK_PR] LCK_COMPAT_PR,
2060         [LCK_CW] LCK_COMPAT_CW,
2061         [LCK_CR] LCK_COMPAT_CR,
2062         [LCK_NL] LCK_COMPAT_NL,
2063         [LCK_GROUP] LCK_COMPAT_GROUP
2064 };