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 "ldlm_internal.h"
  40
  41 /* @priority: if non-zero, move the selected to the list head
  42  * @create: if zero, only search in existed connections
  43  */
  44 static int import_set_conn(struct obd_import *imp, struct obd_uuid *uuid,
  45                            int priority, int create)
  46 {
  47         struct ptlrpc_connection *ptlrpc_conn;
  48         struct obd_import_conn *imp_conn = NULL, *item;
  49         int rc = 0;
  50         ENTRY;
  51
  52         if (!create && !priority) {
  53                 CDEBUG(D_HA, "Nothing to do\n");
  54                 RETURN(-EINVAL);
  55         }
  56
  57         ptlrpc_conn = ptlrpc_uuid_to_connection(uuid);
  58         if (!ptlrpc_conn) {
  59                 CDEBUG(D_HA, "can't find connection %s\n", uuid->uuid);
  60                 RETURN (-ENOENT);
  61         }
  62
  63         if (create) {
  64                 OBD_ALLOC(imp_conn, sizeof(*imp_conn));
  65                 if (!imp_conn) {
  66                         GOTO(out_put, rc = -ENOMEM);
  67                 }
  68         }
  69
  70         spin_lock(&imp->imp_lock);
  71         list_for_each_entry(item, &imp->imp_conn_list, oic_item) {
  72                 if (obd_uuid_equals(uuid, &item->oic_uuid)) {
  73                         if (priority) {
  74                                 list_del(&item->oic_item);
  75                                 list_add(&item->oic_item, &imp->imp_conn_list);
  76                                 item->oic_last_attempt = 0;
  77                         }
  78                         CDEBUG(D_HA, "imp %p@%s: found existing conn %s%s\n",
  79                                imp, imp->imp_obd->obd_name, uuid->uuid,
  80                                (priority ? ", moved to head" : ""));
  81                         spin_unlock(&imp->imp_lock);
  82                         GOTO(out_free, rc = 0);
  83                 }
  84         }
  85         /* not found */
  86         if (create) {
  87                 imp_conn->oic_conn = ptlrpc_conn;
  88                 imp_conn->oic_uuid = *uuid;
  89                 item->oic_last_attempt = 0;
  90                 if (priority)
  91                         list_add(&imp_conn->oic_item, &imp->imp_conn_list);
  92                 else
  93                         list_add_tail(&imp_conn->oic_item, &imp->imp_conn_list);
  94                 CDEBUG(D_HA, "imp %p@%s: add connection %s at %s\n",
  95                        imp, imp->imp_obd->obd_name, uuid->uuid,
  96                        (priority ? "head" : "tail"));
  97         } else {
  98                 spin_unlock(&imp->imp_lock);
  99                 GOTO(out_free, rc = -ENOENT);
 100
 101         }
 102
 103         spin_unlock(&imp->imp_lock);
 104         RETURN(0);
 105 out_free:
 106         if (imp_conn)
 107                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 108 out_put:
 109         ptlrpc_put_connection(ptlrpc_conn);
 110         RETURN(rc);
 111 }
 112
 113 int import_set_conn_priority(struct obd_import *imp, struct obd_uuid *uuid)
 114 {
 115         return import_set_conn(imp, uuid, 1, 0);
 116 }
 117
 118 int client_import_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
 119                            int priority)
 120 {
 121         return import_set_conn(imp, uuid, priority, 1);
 122 }
 123
 124 int client_import_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
 125 {
 126         struct obd_import_conn *imp_conn;
 127         struct obd_export *dlmexp;
 128         int rc = -ENOENT;
 129         ENTRY;
 130
 131         spin_lock(&imp->imp_lock);
 132         if (list_empty(&imp->imp_conn_list)) {
 133                 LASSERT(!imp->imp_connection);
 134                 GOTO(out, rc);
 135         }
 136
 137         list_for_each_entry(imp_conn, &imp->imp_conn_list, oic_item) {
 138                 if (!obd_uuid_equals(uuid, &imp_conn->oic_uuid))
 139                         continue;
 140                 LASSERT(imp_conn->oic_conn);
 141
 142                 /* is current conn? */
 143                 if (imp_conn == imp->imp_conn_current) {
 144                         LASSERT(imp_conn->oic_conn == imp->imp_connection);
 145
 146                         if (imp->imp_state != LUSTRE_IMP_CLOSED &&
 147                             imp->imp_state != LUSTRE_IMP_DISCON) {
 148                                 CERROR("can't remove current connection\n");
 149                                 GOTO(out, rc = -EBUSY);
 150                         }
 151
 152                         ptlrpc_put_connection(imp->imp_connection);
 153                         imp->imp_connection = NULL;
 154
 155                         dlmexp = class_conn2export(&imp->imp_dlm_handle);
 156                         if (dlmexp && dlmexp->exp_connection) {
 157                                 LASSERT(dlmexp->exp_connection ==
 158                                         imp_conn->oic_conn);
 159                                 ptlrpc_put_connection(dlmexp->exp_connection);
 160                                 dlmexp->exp_connection = NULL;
 161                         }
 162                 }
 163
 164                 list_del(&imp_conn->oic_item);
 165                 ptlrpc_put_connection(imp_conn->oic_conn);
 166                 OBD_FREE(imp_conn, sizeof(*imp_conn));
 167                 CDEBUG(D_HA, "imp %p@%s: remove connection %s\n",
 168                        imp, imp->imp_obd->obd_name, uuid->uuid);
 169                 rc = 0;
 170                 break;
 171         }
 172 out:
 173         spin_unlock(&imp->imp_lock);
 174         if (rc == -ENOENT)
 175                 CERROR("connection %s not found\n", uuid->uuid);
 176         RETURN(rc);
 177 }
 178
 179 /* configure an RPC client OBD device
 180  *
 181  * lcfg parameters:
 182  * 1 - client UUID
 183  * 2 - server UUID
 184  * 3 - inactive-on-startup
 185  */
 186 int client_obd_setup(struct obd_device *obddev, obd_count len, void *buf)
 187 {
 188         struct lustre_cfg* lcfg = buf;
 189         struct client_obd *cli = &obddev->u.cli;
 190         struct obd_import *imp;
 191         struct obd_uuid server_uuid;
 192         int rq_portal, rp_portal, connect_op;
 193         char *name = obddev->obd_type->typ_name;
 194         int rc;
 195         ENTRY;
 196
 197         /* In a more perfect world, we would hang a ptlrpc_client off of
 198          * obd_type and just use the values from there. */
 199         if (!strcmp(name, LUSTRE_OSC_NAME)) {
 200 #ifdef __KERNEL__
 201                 /* Can be removed in Lustre 1.8, for compatibility only */
 202                 rq_portal = OST_IO_PORTAL;
 203 #else
 204                 rq_portal = OST_REQUEST_PORTAL;
 205 #endif
 206                 rp_portal = OSC_REPLY_PORTAL;
 207                 connect_op = OST_CONNECT;
 208         } else if (!strcmp(name, LUSTRE_MDC_NAME)) {
 209                 rq_portal = MDS_REQUEST_PORTAL;
 210                 rp_portal = MDC_REPLY_PORTAL;
 211                 connect_op = MDS_CONNECT;
 212         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
 213                 rq_portal = MGS_REQUEST_PORTAL;
 214                 rp_portal = MGC_REPLY_PORTAL;
 215                 connect_op = MGS_CONNECT;
 216         } else {
 217                 CERROR("unknown client OBD type \"%s\", can't setup\n",
 218                        name);
 219                 RETURN(-EINVAL);
 220         }
 221
 222         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
 223                 CERROR("requires a TARGET UUID\n");
 224                 RETURN(-EINVAL);
 225         }
 226
 227         if (LUSTRE_CFG_BUFLEN(lcfg, 1) > 37) {
 228                 CERROR("client UUID must be less than 38 characters\n");
 229                 RETURN(-EINVAL);
 230         }
 231
 232         if (LUSTRE_CFG_BUFLEN(lcfg, 2) < 1) {
 233                 CERROR("setup requires a SERVER UUID\n");
 234                 RETURN(-EINVAL);
 235         }
 236
 237         if (LUSTRE_CFG_BUFLEN(lcfg, 2) > 37) {
 238                 CERROR("target UUID must be less than 38 characters\n");
 239                 RETURN(-EINVAL);
 240         }
 241
 242         sema_init(&cli->cl_sem, 1);
 243         sema_init(&cli->cl_mgc_sem, 1);
 244         cli->cl_conn_count = 0;
 245         memcpy(server_uuid.uuid, lustre_cfg_buf(lcfg, 2),
 246                min_t(unsigned int, LUSTRE_CFG_BUFLEN(lcfg, 2),
 247                      sizeof(server_uuid)));
 248
 249         cli->cl_dirty = 0;
 250         cli->cl_avail_grant = 0;
 251         /* FIXME: should limit this for the sum of all cl_dirty_max */
 252         cli->cl_dirty_max = OSC_MAX_DIRTY_DEFAULT * 1024 * 1024;
 253         if (cli->cl_dirty_max >> CFS_PAGE_SHIFT > num_physpages / 8)
 254                 cli->cl_dirty_max = num_physpages << (CFS_PAGE_SHIFT - 3);
 255         CFS_INIT_LIST_HEAD(&cli->cl_cache_waiters);
 256         CFS_INIT_LIST_HEAD(&cli->cl_loi_ready_list);
 257         CFS_INIT_LIST_HEAD(&cli->cl_loi_write_list);
 258         CFS_INIT_LIST_HEAD(&cli->cl_loi_read_list);
 259         client_obd_list_lock_init(&cli->cl_loi_list_lock);
 260         cli->cl_r_in_flight = 0;
 261         cli->cl_w_in_flight = 0;
 262         spin_lock_init(&cli->cl_read_rpc_hist.oh_lock);
 263         spin_lock_init(&cli->cl_write_rpc_hist.oh_lock);
 264         spin_lock_init(&cli->cl_read_page_hist.oh_lock);
 265         spin_lock_init(&cli->cl_write_page_hist.oh_lock);
 266         spin_lock_init(&cli->cl_read_offset_hist.oh_lock);
 267         spin_lock_init(&cli->cl_write_offset_hist.oh_lock);
 268         cfs_waitq_init(&cli->cl_destroy_waitq);
 269         atomic_set(&cli->cl_destroy_in_flight, 0);
 270 #ifdef ENABLE_CHECKSUM
 271         /* Turn on checksumming by default. */
 272         cli->cl_checksum = 1;
 273         /*
 274          * The supported checksum types will be worked out at connect time
 275          * Set cl_chksum* to CRC32 for now to avoid returning screwed info
 276          * through procfs.
 277          */
 278         cli->cl_cksum_type = cli->cl_supp_cksum_types = OBD_CKSUM_CRC32;
 279 #endif
 280         atomic_set(&cli->cl_resends, OSC_DEFAULT_RESENDS);
 281
 282         /* This value may be changed at connect time in
 283            ptlrpc_connect_interpret. */
 284         cli->cl_max_pages_per_rpc = min((int)PTLRPC_MAX_BRW_PAGES,
 285                                         (int)(1024 * 1024 >> CFS_PAGE_SHIFT));
 286
 287         if (!strcmp(name, LUSTRE_MDC_NAME)) {
 288                 cli->cl_max_rpcs_in_flight = MDC_MAX_RIF_DEFAULT;
 289         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 128 /* MB */) {
 290                 cli->cl_max_rpcs_in_flight = 2;
 291         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 256 /* MB */) {
 292                 cli->cl_max_rpcs_in_flight = 3;
 293         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 512 /* MB */) {
 294                 cli->cl_max_rpcs_in_flight = 4;
 295         } else {
 296                 cli->cl_max_rpcs_in_flight = OSC_MAX_RIF_DEFAULT;
 297         }
 298         rc = ldlm_get_ref();
 299         if (rc) {
 300                 CERROR("ldlm_get_ref failed: %d\n", rc);
 301                 GOTO(err, rc);
 302         }
 303
 304         ptlrpc_init_client(rq_portal, rp_portal, name,
 305                            &obddev->obd_ldlm_client);
 306
 307         imp = class_new_import(obddev);
 308         if (imp == NULL)
 309                 GOTO(err_ldlm, rc = -ENOENT);
 310         imp->imp_client = &obddev->obd_ldlm_client;
 311         imp->imp_connect_op = connect_op;
 312         imp->imp_initial_recov = 1;
 313         imp->imp_initial_recov_bk = 0;
 314         CFS_INIT_LIST_HEAD(&imp->imp_pinger_chain);
 315         memcpy(cli->cl_target_uuid.uuid, lustre_cfg_buf(lcfg, 1),
 316                LUSTRE_CFG_BUFLEN(lcfg, 1));
 317         class_import_put(imp);
 318
 319         rc = client_import_add_conn(imp, &server_uuid, 1);
 320         if (rc) {
 321                 CERROR("can't add initial connection\n");
 322                 GOTO(err_import, rc);
 323         }
 324
 325         cli->cl_import = imp;
 326         /* cli->cl_max_mds_{easize,cookiesize} updated by mdc_init_ea_size() */
 327         cli->cl_max_mds_easize = sizeof(struct lov_mds_md);
 328         cli->cl_max_mds_cookiesize = sizeof(struct llog_cookie);
 329
 330         if (LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
 331                 if (!strcmp(lustre_cfg_string(lcfg, 3), "inactive")) {
 332                         CDEBUG(D_HA, "marking %s %s->%s as inactive\n",
 333                                name, obddev->obd_name,
 334                                cli->cl_target_uuid.uuid);
 335                         spin_lock(&imp->imp_lock);
 336                         imp->imp_invalid = 1;
 337                         spin_unlock(&imp->imp_lock);
 338                 }
 339         }
 340
 341         cli->cl_qchk_stat = CL_NOT_QUOTACHECKED;
 342
 343         RETURN(rc);
 344
 345 err_import:
 346         class_destroy_import(imp);
 347 err_ldlm:
 348         ldlm_put_ref(0);
 349 err:
 350         RETURN(rc);
 351
 352 }
 353
 354 int client_obd_cleanup(struct obd_device *obddev)
 355 {
 356         ENTRY;
 357         ldlm_put_ref(obddev->obd_force);
 358         RETURN(0);
 359 }
 360
 361 /* ->o_connect() method for client side (OSC and MDC and MGC) */
 362 int client_connect_import(struct lustre_handle *dlm_handle,
 363                           struct obd_device *obd, struct obd_uuid *cluuid,
 364                           struct obd_connect_data *data, void *localdata)
 365 {
 366         struct client_obd *cli = &obd->u.cli;
 367         struct obd_import *imp = cli->cl_import;
 368         struct obd_export *exp;
 369         struct obd_connect_data *ocd;
 370         struct ldlm_namespace *to_be_freed = NULL;
 371         int rc;
 372         ENTRY;
 373
 374         mutex_down(&cli->cl_sem);
 375         rc = class_connect(dlm_handle, obd, cluuid);
 376         if (rc)
 377                 GOTO(out_sem, rc);
 378
 379         cli->cl_conn_count++;
 380         if (cli->cl_conn_count > 1)
 381                 GOTO(out_sem, rc);
 382         exp = class_conn2export(dlm_handle);
 383
 384         if (obd->obd_namespace != NULL)
 385                 CERROR("already have namespace!\n");
 386         obd->obd_namespace = ldlm_namespace_new(obd->obd_name,
 387                                                 LDLM_NAMESPACE_CLIENT,
 388                                                 LDLM_NAMESPACE_GREEDY);
 389         if (obd->obd_namespace == NULL)
 390                 GOTO(out_disco, rc = -ENOMEM);
 391
 392         imp->imp_dlm_handle = *dlm_handle;
 393         rc = ptlrpc_init_import(imp);
 394         if (rc != 0)
 395                 GOTO(out_ldlm, rc);
 396
 397         ocd = &imp->imp_connect_data;
 398         if (data) {
 399                 *ocd = *data;
 400                 imp->imp_connect_flags_orig = data->ocd_connect_flags;
 401         }
 402
 403         rc = ptlrpc_connect_import(imp, NULL);
 404         if (rc != 0) {
 405                 LASSERT (imp->imp_state == LUSTRE_IMP_DISCON);
 406                 GOTO(out_ldlm, rc);
 407         }
 408         LASSERT(exp->exp_connection);
 409
 410         if (data) {
 411                 LASSERT((ocd->ocd_connect_flags & data->ocd_connect_flags) ==
 412                         ocd->ocd_connect_flags);
 413                 data->ocd_connect_flags = ocd->ocd_connect_flags;
 414         }
 415
 416         ptlrpc_pinger_add_import(imp);
 417         EXIT;
 418
 419         if (rc) {
 420 out_ldlm:
 421                 ldlm_namespace_free_prior(obd->obd_namespace);
 422                 to_be_freed = obd->obd_namespace;
 423                 obd->obd_namespace = NULL;
 424 out_disco:
 425                 cli->cl_conn_count--;
 426                 class_disconnect(exp);
 427         } else {
 428                 class_export_put(exp);
 429         }
 430 out_sem:
 431         mutex_up(&cli->cl_sem);
 432         if (to_be_freed)
 433                 ldlm_namespace_free_post(to_be_freed, 0);
 434         return rc;
 435 }
 436
 437 int client_disconnect_export(struct obd_export *exp)
 438 {
 439         struct obd_device *obd = class_exp2obd(exp);
 440         struct client_obd *cli;
 441         struct obd_import *imp;
 442         struct ldlm_namespace *to_be_freed = NULL;
 443         int rc = 0, err;
 444         ENTRY;
 445
 446         if (!obd) {
 447                 CERROR("invalid export for disconnect: exp %p cookie "LPX64"\n",
 448                        exp, exp ? exp->exp_handle.h_cookie : -1);
 449                 RETURN(-EINVAL);
 450         }
 451
 452         cli = &obd->u.cli;
 453         imp = cli->cl_import;
 454
 455         mutex_down(&cli->cl_sem);
 456         if (!cli->cl_conn_count) {
 457                 CERROR("disconnecting disconnected device (%s)\n",
 458                        obd->obd_name);
 459                 GOTO(out_sem, rc = -EINVAL);
 460         }
 461
 462         cli->cl_conn_count--;
 463         if (cli->cl_conn_count)
 464                 GOTO(out_no_disconnect, rc = 0);
 465
 466         /* Mark import deactivated now, so we don't try to reconnect if any
 467          * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
 468          * fully deactivate the import, or that would drop all requests. */
 469         spin_lock(&imp->imp_lock);
 470         imp->imp_deactive = 1;
 471         spin_unlock(&imp->imp_lock);
 472
 473         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
 474          * delete it regardless.  (It's safe to delete an import that was
 475          * never added.) */
 476         (void)ptlrpc_pinger_del_import(imp);
 477
 478         if (obd->obd_namespace != NULL) {
 479                 /* obd_force == local only */
 480                 ldlm_cli_cancel_unused(obd->obd_namespace, NULL,
 481                                        obd->obd_force ? LDLM_FL_LOCAL_ONLY:0,
 482                                        NULL);
 483                 ldlm_namespace_free_prior(obd->obd_namespace);
 484                 to_be_freed = obd->obd_namespace;
 485         }
 486
 487         rc = ptlrpc_disconnect_import(imp, 0);
 488
 489         ptlrpc_invalidate_import(imp);
 490         /* set obd_namespace to NULL only after invalidate, because we can have
 491          * some connect requests in flight, and his need store a connect flags
 492          * in obd_namespace. bug 14260 */
 493         obd->obd_namespace = NULL;
 494
 495         ptlrpc_free_rq_pool(imp->imp_rq_pool);
 496         class_destroy_import(imp);
 497         cli->cl_import = NULL;
 498
 499         EXIT;
 500  out_no_disconnect:
 501         err = class_disconnect(exp);
 502         if (!rc && err)
 503                 rc = err;
 504  out_sem:
 505         mutex_up(&cli->cl_sem);
 506         if (to_be_freed)
 507                 ldlm_namespace_free_post(to_be_freed, obd->obd_force);
 508         RETURN(rc);
 509 }
 510
 511 /* --------------------------------------------------------------------------
 512  * from old lib/target.c
 513  * -------------------------------------------------------------------------- */
 514
 515 int target_handle_reconnect(struct lustre_handle *conn, struct obd_export *exp,
 516                             struct obd_uuid *cluuid)
 517 {
 518         ENTRY;
 519         if (exp->exp_connection && exp->exp_imp_reverse) {
 520                 struct lustre_handle *hdl;
 521                 hdl = &exp->exp_imp_reverse->imp_remote_handle;
 522                 /* Might be a re-connect after a partition. */
 523                 if (!memcmp(&conn->cookie, &hdl->cookie, sizeof conn->cookie)) {
 524                         CWARN("%s: %s reconnecting\n", exp->exp_obd->obd_name,
 525                               cluuid->uuid);
 526                         conn->cookie = exp->exp_handle.h_cookie;
 527                         /* target_handle_connect() treats EALREADY and
 528                          * -EALREADY differently.  EALREADY means we are
 529                          * doing a valid reconnect from the same client. */
 530                         RETURN(EALREADY);
 531                 } else {
 532                         CERROR("%s reconnecting from %s, "
 533                                "handle mismatch (ours "LPX64", theirs "
 534                                LPX64")\n", cluuid->uuid,
 535                                exp->exp_connection->c_remote_uuid.uuid,
 536                                hdl->cookie, conn->cookie);
 537                         memset(conn, 0, sizeof *conn);
 538                         /* target_handle_connect() treats EALREADY and
 539                          * -EALREADY differently.  -EALREADY is an error
 540                          * (same UUID, different handle). */
 541                         RETURN(-EALREADY);
 542                 }
 543         }
 544
 545         conn->cookie = exp->exp_handle.h_cookie;
 546         CDEBUG(D_HA, "connect export for UUID '%s' at %p, cookie "LPX64"\n",
 547                cluuid->uuid, exp, conn->cookie);
 548         RETURN(0);
 549 }
 550
 551 void target_client_add_cb(struct obd_device *obd, __u64 transno, void *cb_data,
 552                           int error)
 553 {
 554         struct obd_export *exp = cb_data;
 555
 556         CDEBUG(D_RPCTRACE, "%s: committing for initial connect of %s\n",
 557                obd->obd_name, exp->exp_client_uuid.uuid);
 558
 559         spin_lock(&exp->exp_lock);
 560         exp->exp_need_sync = 0;
 561         spin_unlock(&exp->exp_lock);
 562 }
 563 EXPORT_SYMBOL(target_client_add_cb);
 564
 565 int target_handle_connect(struct ptlrpc_request *req, svc_handler_t handler)
 566 {
 567         struct obd_device *target, *targref = NULL;
 568         struct obd_export *export = NULL;
 569         struct obd_import *revimp;
 570         struct lustre_handle conn;
 571         struct obd_uuid tgtuuid;
 572         struct obd_uuid cluuid;
 573         struct obd_uuid remote_uuid;
 574         char *str, *tmp;
 575         int rc = 0, abort_recovery;
 576         struct obd_connect_data *data;
 577         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*data) };
 578         lnet_nid_t client_nid = 0;
 579         ENTRY;
 580
 581         OBD_RACE(OBD_FAIL_TGT_CONN_RACE);
 582
 583         lustre_set_req_swabbed(req, REQ_REC_OFF);
 584         str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF, sizeof(tgtuuid)-1);
 585         if (str == NULL) {
 586                 DEBUG_REQ(D_ERROR, req, "bad target UUID for connect");
 587                 GOTO(out, rc = -EINVAL);
 588         }
 589
 590         obd_str2uuid (&tgtuuid, str);
 591         target = class_uuid2obd(&tgtuuid);
 592         /* COMPAT_146 */
 593         /* old (pre 1.6) lustre_process_log tries to connect to mdsname
 594            (eg. mdsA) instead of uuid. */
 595         if (!target) {
 596                 snprintf((char *)tgtuuid.uuid, sizeof(tgtuuid), "%s_UUID", str);
 597                 target = class_uuid2obd(&tgtuuid);
 598         }
 599         if (!target)
 600                 target = class_name2obd(str);
 601         /* end COMPAT_146 */
 602
 603         if (!target || target->obd_stopping || !target->obd_set_up) {
 604                 LCONSOLE_ERROR_MSG(0x137, "UUID '%s' is not available "
 605                                    " for connect (%s)\n", str,
 606                                    !target ? "no target" :
 607                                    (target->obd_stopping ? "stopping" :
 608                                    "not set up"));
 609                 GOTO(out, rc = -ENODEV);
 610         }
 611
 612         if (target->obd_no_conn) {
 613                 LCONSOLE_WARN("%s: temporarily refusing client connection "
 614                               "from %s\n", target->obd_name,
 615                               libcfs_nid2str(req->rq_peer.nid));
 616                 GOTO(out, rc = -EAGAIN);
 617         }
 618
 619         /* Make sure the target isn't cleaned up while we're here. Yes,
 620            there's still a race between the above check and our incref here.
 621            Really, class_uuid2obd should take the ref. */
 622         targref = class_incref(target);
 623
 624         lustre_set_req_swabbed(req, REQ_REC_OFF + 1);
 625         str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF + 1,
 626                                 sizeof(cluuid) - 1);
 627         if (str == NULL) {
 628                 DEBUG_REQ(D_ERROR, req, "bad client UUID for connect");
 629                 GOTO(out, rc = -EINVAL);
 630         }
 631
 632         obd_str2uuid (&cluuid, str);
 633
 634         /* XXX extract a nettype and format accordingly */
 635         switch (sizeof(lnet_nid_t)) {
 636                 /* NB the casts only avoid compiler warnings */
 637         case 8:
 638                 snprintf(remote_uuid.uuid, sizeof remote_uuid,
 639                          "NET_"LPX64"_UUID", (__u64)req->rq_peer.nid);
 640                 break;
 641         case 4:
 642                 snprintf(remote_uuid.uuid, sizeof remote_uuid,
 643                          "NET_%x_UUID", (__u32)req->rq_peer.nid);
 644                 break;
 645         default:
 646                 LBUG();
 647         }
 648
 649         spin_lock_bh(&target->obd_processing_task_lock);
 650         abort_recovery = target->obd_abort_recovery;
 651         spin_unlock_bh(&target->obd_processing_task_lock);
 652         if (abort_recovery)
 653                 target_abort_recovery(target);
 654
 655         tmp = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn);
 656         if (tmp == NULL)
 657                 GOTO(out, rc = -EPROTO);
 658
 659         memcpy(&conn, tmp, sizeof conn);
 660
 661         data = lustre_swab_reqbuf(req, REQ_REC_OFF + 3, sizeof(*data),
 662                                   lustre_swab_connect);
 663         rc = lustre_pack_reply(req, 2, size, NULL);
 664         if (rc)
 665                 GOTO(out, rc);
 666
 667         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
 668                 if (!data) {
 669                         DEBUG_REQ(D_WARNING, req, "Refusing old (unversioned) "
 670                                   "libclient connection attempt");
 671                         GOTO(out, rc = -EPROTO);
 672                 } else if (data->ocd_version < LUSTRE_VERSION_CODE -
 673                                                LUSTRE_VERSION_ALLOWED_OFFSET ||
 674                            data->ocd_version > LUSTRE_VERSION_CODE +
 675                                                LUSTRE_VERSION_ALLOWED_OFFSET) {
 676                         DEBUG_REQ(D_WARNING, req, "Refusing %s (%d.%d.%d.%d) "
 677                                   "libclient connection attempt",
 678                                   data->ocd_version < LUSTRE_VERSION_CODE ?
 679                                   "old" : "new",
 680                                   OBD_OCD_VERSION_MAJOR(data->ocd_version),
 681                                   OBD_OCD_VERSION_MINOR(data->ocd_version),
 682                                   OBD_OCD_VERSION_PATCH(data->ocd_version),
 683                                   OBD_OCD_VERSION_FIX(data->ocd_version));
 684                         data = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 685                                               offsetof(typeof(*data),
 686                                                        ocd_version) +
 687                                               sizeof(data->ocd_version));
 688                         if (data) {
 689                                 data->ocd_connect_flags = OBD_CONNECT_VERSION;
 690                                 data->ocd_version = LUSTRE_VERSION_CODE;
 691                         }
 692                         GOTO(out, rc = -EPROTO);
 693                 }
 694         }
 695
 696         /* lctl gets a backstage, all-access pass. */
 697         if (obd_uuid_equals(&cluuid, &target->obd_uuid))
 698                 goto dont_check_exports;
 699
 700         spin_lock(&target->obd_dev_lock);
 701         export = lustre_hash_get_object_by_key(target->obd_uuid_hash_body, &cluuid);
 702
 703         if (export != NULL && export->exp_connecting) { /* bug 9635, et. al. */
 704                 CWARN("%s: exp %p already connecting\n",
 705                       export->exp_obd->obd_name, export);
 706                 class_export_put(export);
 707                 export = NULL;
 708                 rc = -EALREADY;
 709         } else if (export != NULL && export->exp_connection != NULL &&
 710                    req->rq_peer.nid != export->exp_connection->c_peer.nid) {
 711                 /* make darn sure this is coming from the same peer
 712                  * if the UUIDs matched */
 713                   CWARN("%s: cookie %s seen on new NID %s when "
 714                           "existing NID %s is already connected\n",
 715                         target->obd_name, cluuid.uuid,
 716                   libcfs_nid2str(req->rq_peer.nid),
 717                   libcfs_nid2str(export->exp_connection->c_peer.nid));
 718                   class_export_put(export);
 719                   export = NULL;
 720                   rc = -EALREADY;
 721         } else if (export != NULL && export->exp_failed) { /* bug 11327 */
 722                 CDEBUG(D_HA, "%s: exp %p evict in progress - new cookie needed "
 723                       "for connect\n", export->exp_obd->obd_name, export);
 724                 class_export_put(export);
 725                 export = NULL;
 726                 rc = -ENODEV;
 727         } else if (export != NULL) {
 728                 spin_lock(&export->exp_lock);
 729                 export->exp_connecting = 1;
 730                 spin_unlock(&export->exp_lock);
 731                 class_export_put(export);
 732                 spin_unlock(&target->obd_dev_lock);
 733                 LASSERT(export->exp_obd == target);
 734
 735                 rc = target_handle_reconnect(&conn, export, &cluuid);
 736         }
 737
 738         /* If we found an export, we already unlocked. */
 739         if (!export) {
 740                 spin_unlock(&target->obd_dev_lock);
 741                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_CONNECT, 2 * obd_timeout);
 742         } else if (req->rq_export == NULL &&
 743                    atomic_read(&export->exp_rpc_count) > 0) {
 744                 CWARN("%s: refuse connection from %s/%s to 0x%p; still busy "
 745                       "with %d references\n", target->obd_name, cluuid.uuid,
 746                       libcfs_nid2str(req->rq_peer.nid),
 747                       export, atomic_read(&export->exp_refcount));
 748                 GOTO(out, rc = -EBUSY);
 749         } else if (req->rq_export != NULL &&
 750                    atomic_read(&export->exp_rpc_count) > 1) {
 751                 CWARN("%s: refuse reconnection from %s@%s to 0x%p; still busy "
 752                       "with %d active RPCs\n", target->obd_name, cluuid.uuid,
 753                       libcfs_nid2str(req->rq_peer.nid),
 754                       export, atomic_read(&export->exp_rpc_count));
 755                 GOTO(out, rc = -EBUSY);
 756         } else if (lustre_msg_get_conn_cnt(req->rq_reqmsg) == 1) {
 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         }
 764
 765         if (rc < 0)
 766                 GOTO(out, rc);
 767
 768         /* Tell the client if we're in recovery. */
 769         if (target->obd_recovering) {
 770                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
 771                 /* If this is the first time a client connects,
 772                    reset the recovery timer */
 773                 if (rc == 0)
 774                         target_start_recovery_timer(target, handler, req);
 775         }
 776
 777         /* We want to handle EALREADY but *not* -EALREADY from
 778          * target_handle_reconnect(), return reconnection state in a flag */
 779         if (rc == EALREADY) {
 780                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECONNECT);
 781                 rc = 0;
 782         } else {
 783                 LASSERT(rc == 0);
 784         }
 785
 786         /* Tell the client if we support replayable requests */
 787         if (target->obd_replayable)
 788                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
 789         client_nid = req->rq_peer.nid;
 790
 791         if (export == NULL) {
 792                 if (target->obd_recovering) {
 793                         CERROR("%s: denying connection for new client %s (%s): "
 794                                "%d clients in recovery for %lds\n",
 795                                target->obd_name,
 796                                libcfs_nid2str(req->rq_peer.nid), cluuid.uuid,
 797                                target->obd_recoverable_clients,
 798                                cfs_duration_sec(cfs_time_sub(cfs_timer_deadline(&target->obd_recovery_timer),
 799                                                              cfs_time_current())));
 800                         rc = -EBUSY;
 801                 } else {
 802  dont_check_exports:
 803                         rc = obd_connect(&conn, target, &cluuid, data,
 804                                          &client_nid);
 805                 }
 806         } else {
 807                 rc = obd_reconnect(export, target, &cluuid, data);
 808         }
 809
 810         if (rc)
 811                 GOTO(out, rc);
 812
 813         /* Return only the parts of obd_connect_data that we understand, so the
 814          * client knows that we don't understand the rest. */
 815         if (data)
 816                 memcpy(lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 817                                       sizeof(*data)),
 818                        data, sizeof(*data));
 819
 820         /* If all else goes well, this is our RPC return code. */
 821         req->rq_status = 0;
 822
 823         lustre_msg_set_handle(req->rq_repmsg, &conn);
 824
 825         /* ownership of this export ref transfers to the request AFTER we
 826          * drop any previous reference the request had, but we don't want
 827          * that to go to zero before we get our new export reference. */
 828         export = class_conn2export(&conn);
 829         if (!export) {
 830                 DEBUG_REQ(D_ERROR, req, "Missing export!");
 831                 GOTO(out, rc = -ENODEV);
 832         }
 833
 834         /* If the client and the server are the same node, we will already
 835          * have an export that really points to the client's DLM export,
 836          * because we have a shared handles table.
 837          *
 838          * XXX this will go away when shaver stops sending the "connect" handle
 839          * in the real "remote handle" field of the request --phik 24 Apr 2003
 840          */
 841         if (req->rq_export != NULL)
 842                 class_export_put(req->rq_export);
 843
 844         req->rq_export = export;
 845
 846         spin_lock(&export->exp_lock);
 847         if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
 848                 CERROR("%s: %s already connected at higher conn_cnt: %d > %d\n",
 849                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
 850                        export->exp_conn_cnt,
 851                        lustre_msg_get_conn_cnt(req->rq_reqmsg));
 852
 853                 spin_unlock(&export->exp_lock);
 854                 GOTO(out, rc = -EALREADY);
 855         }
 856         export->exp_conn_cnt = lustre_msg_get_conn_cnt(req->rq_reqmsg);
 857
 858         /* request from liblustre?  Don't evict it for not pinging. */
 859         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
 860                 export->exp_libclient = 1;
 861                 spin_unlock(&export->exp_lock);
 862
 863                 spin_lock(&target->obd_dev_lock);
 864                 list_del_init(&export->exp_obd_chain_timed);
 865                 spin_unlock(&target->obd_dev_lock);
 866         } else {
 867                 spin_unlock(&export->exp_lock);
 868         }
 869
 870         if (export->exp_connection != NULL)
 871                 ptlrpc_put_connection(export->exp_connection);
 872         export->exp_connection = ptlrpc_get_connection(req->rq_peer,
 873                                                        req->rq_self,
 874                                                        &remote_uuid);
 875
 876         spin_lock(&target->obd_dev_lock);
 877         /* Export might be hashed already, e.g. if this is reconnect */
 878         if (hlist_unhashed(&export->exp_nid_hash))
 879                 lustre_hash_additem(export->exp_obd->obd_nid_hash_body,
 880                                     &export->exp_connection->c_peer.nid,
 881                                     &export->exp_nid_hash);
 882         spin_unlock(&target->obd_dev_lock);
 883
 884         if (lustre_msg_get_op_flags(req->rq_repmsg) & MSG_CONNECT_RECONNECT) {
 885                 revimp = class_import_get(export->exp_imp_reverse);
 886                 GOTO(set_flags, rc = 0);
 887         }
 888
 889         if (target->obd_recovering)
 890                 target->obd_connected_clients++;
 891
 892         memcpy(&conn,
 893                lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn),
 894                sizeof conn);
 895
 896         if (export->exp_imp_reverse != NULL)
 897                 class_destroy_import(export->exp_imp_reverse);
 898         revimp = export->exp_imp_reverse = class_new_import(target);
 899         revimp->imp_connection = ptlrpc_connection_addref(export->exp_connection);
 900         revimp->imp_client = &export->exp_obd->obd_ldlm_client;
 901         revimp->imp_remote_handle = conn;
 902         revimp->imp_dlm_fake = 1;
 903         revimp->imp_state = LUSTRE_IMP_FULL;
 904
 905 set_flags:
 906         if (req->rq_reqmsg->lm_magic == LUSTRE_MSG_MAGIC_V1 &&
 907             lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_NEXT_VER) {
 908                 revimp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
 909                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_NEXT_VER);
 910         } else {
 911                 /* unknown versions will be caught in
 912                  * ptlrpc_handle_server_req_in->lustre_unpack_msg() */
 913                 revimp->imp_msg_magic = req->rq_reqmsg->lm_magic;
 914         }
 915
 916         if (revimp->imp_msg_magic != LUSTRE_MSG_MAGIC_V1) {
 917                 if (export->exp_connect_flags & OBD_CONNECT_AT)
 918                         revimp->imp_msghdr_flags |= MSGHDR_AT_SUPPORT;
 919         }
 920
 921         class_import_put(revimp);
 922 out:
 923         if (export) {
 924                 spin_lock(&export->exp_lock);
 925                 export->exp_connecting = 0;
 926                 spin_unlock(&export->exp_lock);
 927         }
 928         if (targref)
 929                 class_decref(targref);
 930         if (rc)
 931                 req->rq_status = rc;
 932         RETURN(rc);
 933 }
 934
 935 int target_handle_disconnect(struct ptlrpc_request *req)
 936 {
 937         int rc;
 938         ENTRY;
 939
 940         rc = lustre_pack_reply(req, 1, NULL, NULL);
 941         if (rc)
 942                 RETURN(rc);
 943
 944         /* keep the rq_export around so we can send the reply */
 945         req->rq_status = obd_disconnect(class_export_get(req->rq_export));
 946         RETURN(0);
 947 }
 948
 949 void target_destroy_export(struct obd_export *exp)
 950 {
 951         /* exports created from last_rcvd data, and "fake"
 952            exports created by lctl don't have an import */
 953         if (exp->exp_imp_reverse != NULL)
 954                 class_destroy_import(exp->exp_imp_reverse);
 955
 956         /* We cancel locks at disconnect time, but this will catch any locks
 957          * granted in a race with recovery-induced disconnect. */
 958         if (exp->exp_obd->obd_namespace != NULL)
 959                 ldlm_cancel_locks_for_export(exp);
 960 }
 961
 962 /*
 963  * Recovery functions
 964  */
 965
 966
 967 static void target_release_saved_req(struct ptlrpc_request *req)
 968 {
 969         ptlrpc_req_drop_rs(req);
 970         class_export_put(req->rq_export);
 971         OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
 972         OBD_FREE(req, sizeof *req);
 973 }
 974
 975 static void target_finish_recovery(struct obd_device *obd)
 976 {
 977         struct list_head *tmp, *n;
 978
 979         LCONSOLE_INFO("%s: sending delayed replies to recovered clients\n",
 980                       obd->obd_name);
 981
 982         ldlm_reprocess_all_ns(obd->obd_namespace);
 983
 984         /* when recovery finished, cleanup orphans on mds and ost */
 985         if (OBT(obd) && OBP(obd, postrecov)) {
 986                 int rc = OBP(obd, postrecov)(obd);
 987                 LCONSOLE_WARN("%s: recovery %s: rc %d\n", obd->obd_name,
 988                               rc < 0 ? "failed" : "complete", rc);
 989         }
 990
 991         list_for_each_safe(tmp, n, &obd->obd_delayed_reply_queue) {
 992                 struct ptlrpc_request *req;
 993                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 994                 list_del(&req->rq_list);
 995                 DEBUG_REQ(D_HA, req, "delayed:");
 996                 ptlrpc_reply(req);
 997                 target_release_saved_req(req);
 998         }
 999         obd->obd_recovery_end = cfs_time_current_sec();
1000 }
1001
1002 static void abort_recovery_queue(struct obd_device *obd)
1003 {
1004         struct ptlrpc_request *req;
1005         struct list_head *tmp, *n;
1006         int rc;
1007
1008         list_for_each_safe(tmp, n, &obd->obd_recovery_queue) {
1009                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
1010                 list_del(&req->rq_list);
1011                 DEBUG_REQ(D_ERROR, req, "aborted:");
1012                 req->rq_status = -ENOTCONN;
1013                 req->rq_type = PTL_RPC_MSG_ERR;
1014                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1015                 if (rc == 0)
1016                         ptlrpc_reply(req);
1017                 else
1018                         DEBUG_REQ(D_ERROR, req,
1019                                   "packing failed for abort-reply; skipping");
1020                 target_release_saved_req(req);
1021         }
1022 }
1023
1024 /* Called from a cleanup function if the device is being cleaned up
1025    forcefully.  The exports should all have been disconnected already,
1026    the only thing left to do is
1027      - clear the recovery flags
1028      - cancel the timer
1029      - free queued requests and replies, but don't send replies
1030    Because the obd_stopping flag is set, no new requests should be received.
1031
1032 */
1033 void target_cleanup_recovery(struct obd_device *obd)
1034 {
1035         struct list_head *tmp, *n;
1036         struct ptlrpc_request *req;
1037         ENTRY;
1038
1039         LASSERT(obd->obd_stopping);
1040
1041         spin_lock_bh(&obd->obd_processing_task_lock);
1042         if (!obd->obd_recovering) {
1043                 spin_unlock_bh(&obd->obd_processing_task_lock);
1044                 EXIT;
1045                 return;
1046         }
1047         obd->obd_recovering = obd->obd_abort_recovery = 0;
1048         target_cancel_recovery_timer(obd);
1049         spin_unlock_bh(&obd->obd_processing_task_lock);
1050
1051         list_for_each_safe(tmp, n, &obd->obd_delayed_reply_queue) {
1052                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
1053                 list_del(&req->rq_list);
1054                 target_release_saved_req(req);
1055         }
1056
1057         list_for_each_safe(tmp, n, &obd->obd_recovery_queue) {
1058                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
1059                 list_del(&req->rq_list);
1060                 target_release_saved_req(req);
1061         }
1062         EXIT;
1063 }
1064
1065 void target_abort_recovery(void *data)
1066 {
1067         struct obd_device *obd = data;
1068         ENTRY;
1069
1070         spin_lock_bh(&obd->obd_processing_task_lock);
1071         if (!obd->obd_recovering) {
1072                 spin_unlock_bh(&obd->obd_processing_task_lock);
1073                 EXIT;
1074                 return;
1075         }
1076         obd->obd_recovering = obd->obd_abort_recovery = 0;
1077         target_cancel_recovery_timer(obd);
1078         spin_unlock_bh(&obd->obd_processing_task_lock);
1079
1080         LCONSOLE_WARN("%s: recovery period over; %d clients never reconnected "
1081                       "after %lds (%d clients did)\n",
1082                       obd->obd_name, obd->obd_recoverable_clients,
1083                       cfs_time_current_sec()- obd->obd_recovery_start,
1084                       obd->obd_connected_clients);
1085         class_disconnect_stale_exports(obd);
1086         abort_recovery_queue(obd);
1087
1088         target_finish_recovery(obd);
1089         CDEBUG(D_HA, "%s: recovery complete\n", obd_uuid2str(&obd->obd_uuid));
1090         EXIT;
1091 }
1092
1093 static void target_recovery_expired(unsigned long castmeharder)
1094 {
1095         struct obd_device *obd = (struct obd_device *)castmeharder;
1096         CERROR("%s: recovery timed out, aborting\n", obd->obd_name);
1097         spin_lock_bh(&obd->obd_processing_task_lock);
1098         if (obd->obd_recovering)
1099                 obd->obd_abort_recovery = 1;
1100         cfs_waitq_signal(&obd->obd_next_transno_waitq);
1101         spin_unlock_bh(&obd->obd_processing_task_lock);
1102 }
1103
1104
1105 /* obd_processing_task_lock should be held */
1106 void target_cancel_recovery_timer(struct obd_device *obd)
1107 {
1108         CDEBUG(D_HA, "%s: cancel recovery timer\n", obd->obd_name);
1109         cfs_timer_disarm(&obd->obd_recovery_timer);
1110 }
1111
1112 static void reset_recovery_timer(struct obd_device *obd, int mintime)
1113 {
1114         spin_lock_bh(&obd->obd_processing_task_lock);
1115         if (!obd->obd_recovering) {
1116                 spin_unlock_bh(&obd->obd_processing_task_lock);
1117                 return;
1118         }
1119         /* Track the client's largest expected replay time */
1120         obd->obd_recovery_timeout = max(mintime, obd->obd_recovery_timeout);
1121 #ifdef CRAY_XT3
1122         if(cfs_time_current_sec() + obd->obd_recovery_timeout >
1123            obd->obd_recovery_start + obd->obd_recovery_max_time)
1124                 obd->obd_recovery_timeout = obd->obd_recovery_start +
1125                         obd->obd_recovery_max_time - cfs_time_current_sec();
1126 #endif
1127         obd->obd_recovery_end = cfs_time_current_sec() +
1128                 obd->obd_recovery_timeout;
1129         cfs_timer_arm(&obd->obd_recovery_timer,
1130                       cfs_time_shift(obd->obd_recovery_timeout));
1131         spin_unlock_bh(&obd->obd_processing_task_lock);
1132         CDEBUG(D_HA, "%s: recovery timer will expire in %u seconds\n",
1133                obd->obd_name, obd->obd_recovery_timeout);
1134 }
1135
1136
1137 /* Reset the timer with each new client connection */
1138 void target_start_recovery_timer(struct obd_device *obd, svc_handler_t handler,
1139                                  struct ptlrpc_request *req)
1140 {
1141         spin_lock_bh(&obd->obd_processing_task_lock);
1142         if (obd->obd_recovery_handler)
1143                 goto out;
1144         CWARN("%s: starting recovery timer\n", obd->obd_name);
1145         obd->obd_recovery_start = cfs_time_current_sec();
1146         /* minimum */
1147         obd->obd_recovery_timeout = OBD_RECOVERY_FACTOR * obd_timeout;
1148         obd->obd_recovery_handler = handler;
1149         cfs_timer_init(&obd->obd_recovery_timer, target_recovery_expired, obd);
1150 out:
1151         spin_unlock_bh(&obd->obd_processing_task_lock);
1152         reset_recovery_timer(obd, OBD_RECOVERY_FACTOR *
1153                              lustre_msg_get_timeout(req->rq_reqmsg));
1154 }
1155
1156 static int check_for_next_transno(struct obd_device *obd)
1157 {
1158         struct ptlrpc_request *req;
1159         int wake_up = 0, connected, completed, queue_len, max;
1160         __u64 next_transno, req_transno;
1161
1162         spin_lock_bh(&obd->obd_processing_task_lock);
1163         req = list_entry(obd->obd_recovery_queue.next,
1164                          struct ptlrpc_request, rq_list);
1165         max = obd->obd_max_recoverable_clients;
1166         req_transno = lustre_msg_get_transno(req->rq_reqmsg);
1167         connected = obd->obd_connected_clients;
1168         completed = max - obd->obd_recoverable_clients;
1169         queue_len = obd->obd_requests_queued_for_recovery;
1170         next_transno = obd->obd_next_recovery_transno;
1171
1172         CDEBUG(D_HA,"max: %d, connected: %d, completed: %d, queue_len: %d, "
1173                "req_transno: "LPU64", next_transno: "LPU64"\n",
1174                max, connected, completed, queue_len, req_transno, next_transno);
1175         if (obd->obd_abort_recovery) {
1176                 CDEBUG(D_HA, "waking for aborted recovery\n");
1177                 wake_up = 1;
1178         } else if (!obd->obd_recovering) {
1179                 CDEBUG(D_HA, "waking for completed recovery (?)\n");
1180                 wake_up = 1;
1181         } else if (req_transno == next_transno) {
1182                 CDEBUG(D_HA, "waking for next ("LPD64")\n", next_transno);
1183                 wake_up = 1;
1184         } else if (queue_len + completed == max) {
1185                 CDEBUG(D_ERROR,
1186                        "waking for skipped transno (skip: "LPD64
1187                        ", ql: %d, comp: %d, conn: %d, next: "LPD64")\n",
1188                        next_transno, queue_len, completed, max, req_transno);
1189                 obd->obd_next_recovery_transno = req_transno;
1190                 wake_up = 1;
1191         }
1192         spin_unlock_bh(&obd->obd_processing_task_lock);
1193         LASSERT(lustre_msg_get_transno(req->rq_reqmsg) >= next_transno);
1194         return wake_up;
1195 }
1196
1197 static void process_recovery_queue(struct obd_device *obd)
1198 {
1199         struct ptlrpc_request *req;
1200         int abort_recovery = 0;
1201         struct l_wait_info lwi = { 0 };
1202         ENTRY;
1203
1204         for (;;) {
1205                 spin_lock_bh(&obd->obd_processing_task_lock);
1206                 LASSERT(obd->obd_processing_task == cfs_curproc_pid());
1207                 req = list_entry(obd->obd_recovery_queue.next,
1208                                  struct ptlrpc_request, rq_list);
1209
1210                 if (lustre_msg_get_transno(req->rq_reqmsg) !=
1211                     obd->obd_next_recovery_transno) {
1212                         spin_unlock_bh(&obd->obd_processing_task_lock);
1213                         CDEBUG(D_HA, "Waiting for transno "LPD64" (1st is "
1214                                LPD64", x"LPU64")\n",
1215                                obd->obd_next_recovery_transno,
1216                                lustre_msg_get_transno(req->rq_reqmsg),
1217                                req->rq_xid);
1218                         l_wait_event(obd->obd_next_transno_waitq,
1219                                      check_for_next_transno(obd), &lwi);
1220                         spin_lock_bh(&obd->obd_processing_task_lock);
1221                         abort_recovery = obd->obd_abort_recovery;
1222                         spin_unlock_bh(&obd->obd_processing_task_lock);
1223                         if (abort_recovery) {
1224                                 target_abort_recovery(obd);
1225                                 return;
1226                         }
1227                         continue;
1228                 }
1229                 list_del_init(&req->rq_list);
1230                 obd->obd_requests_queued_for_recovery--;
1231                 spin_unlock_bh(&obd->obd_processing_task_lock);
1232
1233                 DEBUG_REQ(D_HA, req, "processing: ");
1234                 (void)obd->obd_recovery_handler(req);
1235                 obd->obd_replayed_requests++;
1236                 reset_recovery_timer(obd, 2 * /* safety */
1237                         at_get(&req->rq_rqbd->rqbd_service->srv_at_estimate));
1238                 /* bug 1580: decide how to properly sync() in recovery */
1239                 //mds_fsync_super(obd->u.obt.obt_sb);
1240                 class_export_put(req->rq_export);
1241                 ptlrpc_req_drop_rs(req);
1242                 OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
1243                 OBD_FREE(req, sizeof *req);
1244                 spin_lock_bh(&obd->obd_processing_task_lock);
1245                 obd->obd_next_recovery_transno++;
1246                 if (list_empty(&obd->obd_recovery_queue)) {
1247                         obd->obd_processing_task = 0;
1248                         spin_unlock_bh(&obd->obd_processing_task_lock);
1249                         break;
1250                 }
1251                 spin_unlock_bh(&obd->obd_processing_task_lock);
1252         }
1253         EXIT;
1254 }
1255
1256 int target_queue_recovery_request(struct ptlrpc_request *req,
1257                                   struct obd_device *obd)
1258 {
1259         struct list_head *tmp;
1260         int inserted = 0;
1261         __u64 transno = lustre_msg_get_transno(req->rq_reqmsg);
1262         struct ptlrpc_request *saved_req;
1263         struct lustre_msg *reqmsg;
1264
1265         /* CAVEAT EMPTOR: The incoming request message has been swabbed
1266          * (i.e. buflens etc are in my own byte order), but type-dependent
1267          * buffers (eg mds_body, ost_body etc) have NOT been swabbed. */
1268
1269         if (!transno) {
1270                 CFS_INIT_LIST_HEAD(&req->rq_list);
1271                 DEBUG_REQ(D_HA, req, "not queueing");
1272                 return 1;
1273         }
1274
1275         /* XXX If I were a real man, these LBUGs would be sane cleanups. */
1276         /* XXX just like the request-dup code in queue_final_reply */
1277         OBD_ALLOC(saved_req, sizeof *saved_req);
1278         if (!saved_req)
1279                 LBUG();
1280         OBD_ALLOC(reqmsg, req->rq_reqlen);
1281         if (!reqmsg)
1282                 LBUG();
1283
1284         spin_lock_bh(&obd->obd_processing_task_lock);
1285
1286         /* If we're processing the queue, we want don't want to queue this
1287          * message.
1288          *
1289          * Also, if this request has a transno less than the one we're waiting
1290          * for, we should process it now.  It could (and currently always will)
1291          * be an open request for a descriptor that was opened some time ago.
1292          *
1293          * Also, a resent, replayed request that has already been
1294          * handled will pass through here and be processed immediately.
1295          */
1296         if (obd->obd_processing_task == cfs_curproc_pid() ||
1297             transno < obd->obd_next_recovery_transno) {
1298                 /* Processing the queue right now, don't re-add. */
1299                 LASSERT(list_empty(&req->rq_list));
1300                 spin_unlock_bh(&obd->obd_processing_task_lock);
1301                 OBD_FREE(reqmsg, req->rq_reqlen);
1302                 OBD_FREE(saved_req, sizeof *saved_req);
1303                 return 1;
1304         }
1305
1306         /* A resent, replayed request that is still on the queue; just drop it.
1307            The queued request will handle this. */
1308         if ((lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT|MSG_REPLAY)) ==
1309             (MSG_RESENT | MSG_REPLAY)) {
1310                 DEBUG_REQ(D_ERROR, req, "dropping resent queued req");
1311                 spin_unlock_bh(&obd->obd_processing_task_lock);
1312                 OBD_FREE(reqmsg, req->rq_reqlen);
1313                 OBD_FREE(saved_req, sizeof *saved_req);
1314                 return 0;
1315         }
1316
1317         memcpy(saved_req, req, sizeof *req);
1318         memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
1319         req = saved_req;
1320         req->rq_reqmsg = reqmsg;
1321         class_export_get(req->rq_export);
1322         CFS_INIT_LIST_HEAD(&req->rq_list);
1323
1324         /* XXX O(n^2) */
1325         list_for_each(tmp, &obd->obd_recovery_queue) {
1326                 struct ptlrpc_request *reqiter =
1327                         list_entry(tmp, struct ptlrpc_request, rq_list);
1328
1329                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) > transno) {
1330                         list_add_tail(&req->rq_list, &reqiter->rq_list);
1331                         inserted = 1;
1332                         break;
1333                 }
1334         }
1335
1336         if (!inserted) {
1337                 list_add_tail(&req->rq_list, &obd->obd_recovery_queue);
1338         }
1339
1340         obd->obd_requests_queued_for_recovery++;
1341
1342         if (obd->obd_processing_task != 0) {
1343                 /* Someone else is processing this queue, we'll leave it to
1344                  * them.
1345                  */
1346                 cfs_waitq_signal(&obd->obd_next_transno_waitq);
1347                 spin_unlock_bh(&obd->obd_processing_task_lock);
1348                 return 0;
1349         }
1350
1351         /* Nobody is processing, and we know there's (at least) one to process
1352          * now, so we'll do the honours.
1353          */
1354         obd->obd_processing_task = cfs_curproc_pid();
1355         spin_unlock_bh(&obd->obd_processing_task_lock);
1356
1357         process_recovery_queue(obd);
1358         return 0;
1359 }
1360
1361 struct obd_device * target_req2obd(struct ptlrpc_request *req)
1362 {
1363         return req->rq_export->exp_obd;
1364 }
1365
1366 int target_queue_last_replay_reply(struct ptlrpc_request *req, int rc)
1367 {
1368         struct obd_device *obd = target_req2obd(req);
1369         struct ptlrpc_request *saved_req;
1370         struct lustre_msg *reqmsg;
1371         int recovery_done = 0;
1372
1373         LASSERT ((rc == 0) == req->rq_packed_final);
1374
1375         if (!req->rq_packed_final) {
1376                 /* Just like ptlrpc_error, but without the sending. */
1377                 rc = lustre_pack_reply(req, 1, NULL, NULL);
1378                 if (rc)
1379                         return rc;
1380                 req->rq_type = PTL_RPC_MSG_ERR;
1381         }
1382
1383         LASSERT(!req->rq_reply_state->rs_difficult);
1384         LASSERT(list_empty(&req->rq_list));
1385         /* XXX a bit like the request-dup code in queue_recovery_request */
1386         OBD_ALLOC(saved_req, sizeof *saved_req);
1387         if (!saved_req)
1388                 return -ENOMEM;
1389         OBD_ALLOC(reqmsg, req->rq_reqlen);
1390         if (!reqmsg) {
1391                 OBD_FREE(saved_req, sizeof *req);
1392                 return -ENOMEM;
1393         }
1394         *saved_req = *req;
1395         memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
1396
1397         /* Don't race cleanup */
1398         spin_lock_bh(&obd->obd_processing_task_lock);
1399         if (obd->obd_stopping) {
1400                 spin_unlock_bh(&obd->obd_processing_task_lock);
1401                 OBD_FREE(reqmsg, req->rq_reqlen);
1402                 OBD_FREE(saved_req, sizeof *req);
1403                 req->rq_status = -ENOTCONN;
1404                 /* rv is ignored anyhow */
1405                 return -ENOTCONN;
1406         }
1407         ptlrpc_rs_addref(req->rq_reply_state);  /* +1 ref for saved reply */
1408         req = saved_req;
1409         req->rq_reqmsg = reqmsg;
1410         class_export_get(req->rq_export);
1411         list_add(&req->rq_list, &obd->obd_delayed_reply_queue);
1412
1413         /* only count the first "replay over" request from each
1414            export */
1415         if (req->rq_export->exp_replay_needed) {
1416                 --obd->obd_recoverable_clients;
1417
1418                 spin_lock(&req->rq_export->exp_lock);
1419                 req->rq_export->exp_replay_needed = 0;
1420                 spin_unlock(&req->rq_export->exp_lock);
1421         }
1422         recovery_done = (obd->obd_recoverable_clients == 0);
1423         spin_unlock_bh(&obd->obd_processing_task_lock);
1424
1425         OBD_RACE(OBD_FAIL_LDLM_RECOV_CLIENTS);
1426         if (recovery_done) {
1427                 spin_lock_bh(&obd->obd_processing_task_lock);
1428                 obd->obd_recovering = obd->obd_abort_recovery = 0;
1429                 target_cancel_recovery_timer(obd);
1430                 spin_unlock_bh(&obd->obd_processing_task_lock);
1431
1432                 target_finish_recovery(obd);
1433                 CDEBUG(D_HA, "%s: recovery complete\n",
1434                        obd_uuid2str(&obd->obd_uuid));
1435         } else {
1436                 CWARN("%s: %d recoverable clients remain\n",
1437                        obd->obd_name, obd->obd_recoverable_clients);
1438                 cfs_waitq_signal(&obd->obd_next_transno_waitq);
1439         }
1440
1441         return 1;
1442 }
1443
1444 static inline struct ldlm_pool *ldlm_exp2pl(struct obd_export *exp)
1445 {
1446         LASSERT(exp != NULL);
1447         return &exp->exp_obd->obd_namespace->ns_pool;
1448 }
1449
1450 int target_pack_pool_reply(struct ptlrpc_request *req)
1451 {
1452         struct ldlm_pool *pl;
1453         ENTRY;
1454
1455         if (!req->rq_export || !req->rq_export->exp_obd ||
1456             !req->rq_export->exp_obd->obd_namespace ||
1457             !exp_connect_lru_resize(req->rq_export)) {
1458                 lustre_msg_set_slv(req->rq_repmsg, 0);
1459                 lustre_msg_set_limit(req->rq_repmsg, 0);
1460                 RETURN(0);
1461         }
1462
1463         pl = ldlm_exp2pl(req->rq_export);
1464
1465         spin_lock(&pl->pl_lock);
1466         LASSERT(ldlm_pool_get_slv(pl) != 0 && ldlm_pool_get_limit(pl) != 0);
1467         lustre_msg_set_slv(req->rq_repmsg, ldlm_pool_get_slv(pl));
1468         lustre_msg_set_limit(req->rq_repmsg, ldlm_pool_get_limit(pl));
1469         spin_unlock(&pl->pl_lock);
1470
1471         RETURN(0);
1472 }
1473
1474 int
1475 target_send_reply_msg (struct ptlrpc_request *req, int rc, int fail_id)
1476 {
1477         if (OBD_FAIL_CHECK(fail_id | OBD_FAIL_ONCE)) {
1478                 obd_fail_loc |= OBD_FAIL_ONCE | OBD_FAILED;
1479                 DEBUG_REQ(D_ERROR, req, "dropping reply");
1480                 return (-ECOMM);
1481         }
1482
1483         if (rc) {
1484                 DEBUG_REQ(D_ERROR, req, "processing error (%d)", rc);
1485                 req->rq_status = rc;
1486                 return (ptlrpc_error(req));
1487         } else {
1488                 DEBUG_REQ(D_NET, req, "sending reply");
1489         }
1490
1491         return (ptlrpc_send_reply(req, PTLRPC_REPLY_MAYBE_DIFFICULT));
1492 }
1493
1494 void
1495 target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
1496 {
1497         int                        netrc;
1498         struct ptlrpc_reply_state *rs;
1499         struct obd_device         *obd;
1500         struct obd_export         *exp;
1501         struct ptlrpc_service     *svc;
1502
1503         svc = req->rq_rqbd->rqbd_service;
1504         rs = req->rq_reply_state;
1505         if (rs == NULL || !rs->rs_difficult) {
1506                 /* no notifiers */
1507                 target_send_reply_msg (req, rc, fail_id);
1508                 return;
1509         }
1510
1511         /* must be an export if locks saved */
1512         LASSERT (req->rq_export != NULL);
1513         /* req/reply consistent */
1514         LASSERT (rs->rs_service == svc);
1515
1516         /* "fresh" reply */
1517         LASSERT (!rs->rs_scheduled);
1518         LASSERT (!rs->rs_scheduled_ever);
1519         LASSERT (!rs->rs_handled);
1520         LASSERT (!rs->rs_on_net);
1521         LASSERT (rs->rs_export == NULL);
1522         LASSERT (list_empty(&rs->rs_obd_list));
1523         LASSERT (list_empty(&rs->rs_exp_list));
1524
1525         exp = class_export_get (req->rq_export);
1526         obd = exp->exp_obd;
1527
1528         /* disable reply scheduling onto srv_reply_queue while I'm setting up */
1529         rs->rs_scheduled = 1;
1530         rs->rs_on_net    = 1;
1531         rs->rs_xid       = req->rq_xid;
1532         rs->rs_transno   = req->rq_transno;
1533         rs->rs_export    = exp;
1534
1535         spin_lock(&obd->obd_uncommitted_replies_lock);
1536
1537         if (rs->rs_transno > obd->obd_last_committed) {
1538                 /* not committed already */
1539                 list_add_tail (&rs->rs_obd_list,
1540                                &obd->obd_uncommitted_replies);
1541         }
1542
1543         spin_unlock (&obd->obd_uncommitted_replies_lock);
1544         spin_lock (&exp->exp_lock);
1545
1546         list_add_tail (&rs->rs_exp_list, &exp->exp_outstanding_replies);
1547
1548         spin_unlock(&exp->exp_lock);
1549
1550         netrc = target_send_reply_msg (req, rc, fail_id);
1551
1552         spin_lock(&svc->srv_lock);
1553
1554         svc->srv_n_difficult_replies++;
1555
1556         if (netrc != 0) {
1557                 /* error sending: reply is off the net.  Also we need +1
1558                  * reply ref until ptlrpc_server_handle_reply() is done
1559                  * with the reply state (if the send was successful, there
1560                  * would have been +1 ref for the net, which
1561                  * reply_out_callback leaves alone) */
1562                 rs->rs_on_net = 0;
1563                 ptlrpc_rs_addref(rs);
1564                 atomic_inc (&svc->srv_outstanding_replies);
1565         }
1566
1567         if (!rs->rs_on_net ||                   /* some notifier */
1568             list_empty(&rs->rs_exp_list) ||     /* completed already */
1569             list_empty(&rs->rs_obd_list)) {
1570                 list_add_tail (&rs->rs_list, &svc->srv_reply_queue);
1571                 cfs_waitq_signal (&svc->srv_waitq);
1572         } else {
1573                 list_add (&rs->rs_list, &svc->srv_active_replies);
1574                 rs->rs_scheduled = 0;           /* allow notifier to schedule */
1575         }
1576
1577         spin_unlock(&svc->srv_lock);
1578 }
1579
1580 int target_handle_ping(struct ptlrpc_request *req)
1581 {
1582         obd_ping(req->rq_export);
1583         return lustre_pack_reply(req, 1, NULL, NULL);
1584 }
1585
1586 void target_committed_to_req(struct ptlrpc_request *req)
1587 {
1588         struct obd_device *obd = req->rq_export->exp_obd;
1589
1590         if (!obd->obd_no_transno && req->rq_repmsg != NULL)
1591                 lustre_msg_set_last_committed(req->rq_repmsg,
1592                                               obd->obd_last_committed);
1593         else
1594                 DEBUG_REQ(D_IOCTL, req, "not sending last_committed update (%d/"
1595                           "%d)", obd->obd_no_transno, req->rq_repmsg == NULL);
1596
1597         CDEBUG(D_INFO, "last_committed x"LPU64", this req x"LPU64"\n",
1598                obd->obd_last_committed, req->rq_xid);
1599 }
1600
1601 EXPORT_SYMBOL(target_committed_to_req);
1602
1603 #ifdef HAVE_QUOTA_SUPPORT
1604 int target_handle_qc_callback(struct ptlrpc_request *req)
1605 {
1606         struct obd_quotactl *oqctl;
1607         struct client_obd *cli = &req->rq_export->exp_obd->u.cli;
1608
1609         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1610                                    lustre_swab_obd_quotactl);
1611         if (oqctl == NULL) {
1612                 CERROR("Can't unpack obd_quotactl\n");
1613                 RETURN(-EPROTO);
1614         }
1615
1616         cli->cl_qchk_stat = oqctl->qc_stat;
1617
1618         return 0;
1619 }
1620
1621 int target_handle_dqacq_callback(struct ptlrpc_request *req)
1622 {
1623 #ifdef __KERNEL__
1624         struct obd_device *obd = req->rq_export->exp_obd;
1625         struct obd_device *master_obd;
1626         struct lustre_quota_ctxt *qctxt;
1627         struct qunit_data *qdata = NULL;
1628         int rc = 0;
1629         int repsize[2] = { sizeof(struct ptlrpc_body), 0 };
1630         ENTRY;
1631
1632         if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_DROP_QUOTA_REQ))
1633                 RETURN(rc);
1634
1635         repsize[1] = quota_get_qunit_data_size(req->rq_export->
1636                                                exp_connect_flags);
1637
1638         rc = lustre_pack_reply(req, 2, repsize, NULL);
1639         if (rc)
1640                 RETURN(rc);
1641
1642         LASSERT(req->rq_export);
1643
1644         /* there are three forms of qunit(historic causes), so we need to
1645          * adjust qunits from slaves to the same form here */
1646         OBD_ALLOC(qdata, sizeof(struct qunit_data));
1647         if (!qdata)
1648                 RETURN(-ENOMEM);
1649         rc = quota_get_qdata(req, qdata, QUOTA_REQUEST, QUOTA_EXPORT);
1650         if (rc < 0) {
1651                 CDEBUG(D_ERROR, "Can't unpack qunit_data\n");
1652                 GOTO(out, rc = -EPROTO);
1653         }
1654
1655         /* we use the observer */
1656         LASSERT(obd->obd_observer && obd->obd_observer->obd_observer);
1657         master_obd = obd->obd_observer->obd_observer;
1658         qctxt = &master_obd->u.obt.obt_qctxt;
1659
1660         LASSERT(qctxt->lqc_handler);
1661         rc = qctxt->lqc_handler(master_obd, qdata,
1662                                 lustre_msg_get_opc(req->rq_reqmsg));
1663         if (rc && rc != -EDQUOT)
1664                 CDEBUG(rc == -EBUSY  ? D_QUOTA : D_ERROR,
1665                        "dqacq failed! (rc:%d)\n", rc);
1666         req->rq_status = rc;
1667
1668         /* there are three forms of qunit(historic causes), so we need to
1669          * adjust the same form to different forms slaves needed */
1670         rc = quota_copy_qdata(req, qdata, QUOTA_REPLY, QUOTA_EXPORT);
1671         if (rc < 0) {
1672                 CDEBUG(D_ERROR, "Can't pack qunit_data\n");
1673                 GOTO(out, rc = -EPROTO);
1674         }
1675
1676         /* Block the quota req. b=14840 */
1677         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_BLOCK_QUOTA_REQ, obd_timeout);
1678
1679         rc = ptlrpc_reply(req);
1680 out:
1681         OBD_FREE(qdata, sizeof(struct qunit_data));
1682         RETURN(rc);
1683 #else
1684         return 0;
1685 #endif /* !__KERNEL__ */
1686 }
1687 #endif /* HAVE_QUOTA_SUPPORT */
1688
1689 ldlm_mode_t lck_compat_array[] = {
1690         [LCK_EX] LCK_COMPAT_EX,
1691         [LCK_PW] LCK_COMPAT_PW,
1692         [LCK_PR] LCK_COMPAT_PR,
1693         [LCK_CW] LCK_COMPAT_CW,
1694         [LCK_CR] LCK_COMPAT_CR,
1695         [LCK_NL] LCK_COMPAT_NL,
1696         [LCK_GROUP] LCK_COMPAT_GROUP
1697 };