lustre/ptlrpc/import.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002, 2003 Cluster File Systems, Inc.
   5  *   Author: Mike Shaver <shaver@clusterfs.com>
   6  *
   7  *   This file is part of the Lustre file system, http://www.lustre.org
   8  *   Lustre is a trademark of Cluster File Systems, Inc.
   9  *
  10  *   You may have signed or agreed to another license before downloading
  11  *   this software.  If so, you are bound by the terms and conditions
  12  *   of that agreement, and the following does not apply to you.  See the
  13  *   LICENSE file included with this distribution for more information.
  14  *
  15  *   If you did not agree to a different license, then this copy of Lustre
  16  *   is open source software; you can redistribute it and/or modify it
  17  *   under the terms of version 2 of the GNU General Public License as
  18  *   published by the Free Software Foundation.
  19  *
  20  *   In either case, Lustre is distributed in the hope that it will be
  21  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  22  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  23  *   license text for more details.
  24  */
  25
  26 #define DEBUG_SUBSYSTEM S_RPC
  27 #ifdef __KERNEL__
  28 # include <linux/config.h>
  29 # include <linux/module.h>
  30 # include <linux/kmod.h>
  31 #else
  32 # include <liblustre.h>
  33 #endif
  34
  35 #include <linux/obd_support.h>
  36 #include <linux/lustre_ha.h>
  37 #include <linux/lustre_net.h>
  38 #include <linux/lustre_import.h>
  39 #include <linux/lustre_export.h>
  40 #include <linux/obd.h>
  41 #include <linux/obd_class.h>
  42
  43 #include "ptlrpc_internal.h"
  44
  45 struct ptlrpc_connect_async_args {
  46          __u64 pcaa_peer_committed;
  47         int pcaa_initial_connect;
  48 };
  49
  50 /* A CLOSED import should remain so. */
  51 #define IMPORT_SET_STATE_NOLOCK(imp, state)                                    \
  52 do {                                                                           \
  53         if (imp->imp_state != LUSTRE_IMP_CLOSED) {                             \
  54                CDEBUG(D_HA, "%p %s: changing import state from %s to %s\n",    \
  55                       imp, imp->imp_target_uuid.uuid,                          \
  56                       ptlrpc_import_state_name(imp->imp_state),                \
  57                       ptlrpc_import_state_name(state));                        \
  58                imp->imp_state = state;                                         \
  59         }                                                                      \
  60 } while(0)
  61
  62 #define IMPORT_SET_STATE(imp, state)                    \
  63 do {                                                    \
  64         unsigned long flags;                            \
  65                                                         \
  66         spin_lock_irqsave(&imp->imp_lock, flags);       \
  67         IMPORT_SET_STATE_NOLOCK(imp, state);            \
  68         spin_unlock_irqrestore(&imp->imp_lock, flags);  \
  69 } while(0)
  70
  71
  72 static int ptlrpc_connect_interpret(struct ptlrpc_request *request,
  73                                     void * data, int rc);
  74 int ptlrpc_import_recovery_state_machine(struct obd_import *imp);
  75
  76 /* Only this function is allowed to change the import state when it is
  77  * CLOSED. I would rather refcount the import and free it after
  78  * disconnection like we do with exports. To do that, the client_obd
  79  * will need to save the peer info somewhere other than in the import,
  80  * though. */
  81 int ptlrpc_init_import(struct obd_import *imp)
  82 {
  83         unsigned long flags;
  84
  85         spin_lock_irqsave(&imp->imp_lock, flags);
  86
  87         imp->imp_generation++;
  88         imp->imp_state =  LUSTRE_IMP_NEW;
  89
  90         spin_unlock_irqrestore(&imp->imp_lock, flags);
  91
  92         return 0;
  93 }
  94
  95 #define UUID_STR "_UUID"
  96 static void deuuidify(char *uuid, const char *prefix, char **uuid_start, int *uuid_len)
  97 {
  98         *uuid_start = !prefix || strncmp(uuid, prefix, strlen(prefix))
  99                 ? uuid : uuid + strlen(prefix);
 100
 101         *uuid_len = strlen(*uuid_start);
 102
 103         if (*uuid_len < strlen(UUID_STR))
 104                 return;
 105
 106         if (!strncmp(*uuid_start + *uuid_len - strlen(UUID_STR),
 107                     UUID_STR, strlen(UUID_STR)))
 108                 *uuid_len -= strlen(UUID_STR);
 109 }
 110
 111 /* Returns true if import was FULL, false if import was already not
 112  * connected.
 113  */
 114 int ptlrpc_set_import_discon(struct obd_import *imp)
 115 {
 116         unsigned long flags;
 117         int rc = 0;
 118
 119         spin_lock_irqsave(&imp->imp_lock, flags);
 120
 121         if (imp->imp_state == LUSTRE_IMP_FULL) {
 122                 char *target_start;
 123                 int   target_len;
 124
 125                 deuuidify(imp->imp_target_uuid.uuid, NULL,
 126                           &target_start, &target_len);
 127
 128                 LCONSOLE_ERROR("Connection to service %.*s via nid %s was "
 129                                "lost; in progress operations using this "
 130                                "service will %s.\n",
 131                                target_len, target_start,
 132                                libcfs_nid2str(imp->imp_connection->c_peer.nid),
 133                                imp->imp_replayable
 134                                ? "wait for recovery to complete"
 135                                : "fail");
 136
 137                 if (obd_dump_on_timeout)
 138                         libcfs_debug_dumplog();
 139
 140                 CWARN("%s: connection lost to %s@%s\n",
 141                       imp->imp_obd->obd_name,
 142                       imp->imp_target_uuid.uuid,
 143                       imp->imp_connection->c_remote_uuid.uuid);
 144                 IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_DISCON);
 145                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 146                 obd_import_event(imp->imp_obd, imp, IMP_EVENT_DISCON);
 147                 rc = 1;
 148         } else {
 149                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 150                 CDEBUG(D_HA, "%p %s: import already not connected: %s\n",
 151                        imp,imp->imp_client->cli_name,
 152                        ptlrpc_import_state_name(imp->imp_state));
 153         }
 154
 155         return rc;
 156 }
 157
 158 /*
 159  * This acts as a barrier; all existing requests are rejected, and
 160  * no new requests will be accepted until the import is valid again.
 161  */
 162 void ptlrpc_deactivate_import(struct obd_import *imp)
 163 {
 164         unsigned long flags;
 165         ENTRY;
 166
 167         spin_lock_irqsave(&imp->imp_lock, flags);
 168         CDEBUG(D_HA, "setting import %s INVALID\n", imp->imp_target_uuid.uuid);
 169         imp->imp_invalid = 1;
 170         imp->imp_generation++;
 171         spin_unlock_irqrestore(&imp->imp_lock, flags);
 172
 173         ptlrpc_abort_inflight(imp);
 174         obd_import_event(imp->imp_obd, imp, IMP_EVENT_INACTIVE);
 175 }
 176
 177 /*
 178  * This function will invalidate the import, if necessary, then block
 179  * for all the RPC completions, and finally notify the obd to
 180  * invalidate its state (ie cancel locks, clear pending requests,
 181  * etc).
 182  */
 183 void ptlrpc_invalidate_import(struct obd_import *imp)
 184 {
 185         struct l_wait_info lwi;
 186         int rc;
 187
 188         if (!imp->imp_invalid)
 189                 ptlrpc_deactivate_import(imp);
 190
 191         LASSERT(imp->imp_invalid);
 192
 193         /* wait for all requests to error out and call completion callbacks */
 194         lwi = LWI_TIMEOUT_INTR(MAX(obd_timeout * HZ, 1), NULL,
 195                                NULL, NULL);
 196         rc = l_wait_event(imp->imp_recovery_waitq,
 197                           (atomic_read(&imp->imp_inflight) == 0),
 198                           &lwi);
 199
 200         if (rc)
 201                 CERROR("%s: rc = %d waiting for callback (%d != 0)\n",
 202                        imp->imp_target_uuid.uuid, rc,
 203                        atomic_read(&imp->imp_inflight));
 204
 205         obd_import_event(imp->imp_obd, imp, IMP_EVENT_INVALIDATE);
 206 }
 207
 208 void ptlrpc_activate_import(struct obd_import *imp)
 209 {
 210         struct obd_device *obd = imp->imp_obd;
 211         unsigned long flags;
 212
 213         spin_lock_irqsave(&imp->imp_lock, flags);
 214         imp->imp_invalid = 0;
 215         spin_unlock_irqrestore(&imp->imp_lock, flags);
 216
 217         obd_import_event(obd, imp, IMP_EVENT_ACTIVE);
 218 }
 219
 220 void ptlrpc_fail_import(struct obd_import *imp, int generation)
 221 {
 222         ENTRY;
 223
 224         LASSERT (!imp->imp_dlm_fake);
 225
 226         if (ptlrpc_set_import_discon(imp)) {
 227                 unsigned long flags;
 228
 229                 if (!imp->imp_replayable) {
 230                         CDEBUG(D_HA, "import %s@%s for %s not replayable, "
 231                                "auto-deactivating\n",
 232                                imp->imp_target_uuid.uuid,
 233                                imp->imp_connection->c_remote_uuid.uuid,
 234                                imp->imp_obd->obd_name);
 235                         ptlrpc_deactivate_import(imp);
 236                 }
 237
 238                 CDEBUG(D_HA, "%s: waking up pinger\n",
 239                        imp->imp_target_uuid.uuid);
 240
 241                 spin_lock_irqsave(&imp->imp_lock, flags);
 242                 imp->imp_force_verify = 1;
 243                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 244
 245                 ptlrpc_pinger_wake_up();
 246         }
 247         EXIT;
 248 }
 249
 250 static int import_select_connection(struct obd_import *imp)
 251 {
 252         struct obd_import_conn *imp_conn;
 253         struct obd_export *dlmexp;
 254         ENTRY;
 255
 256         spin_lock(&imp->imp_lock);
 257
 258         if (list_empty(&imp->imp_conn_list)) {
 259                 CERROR("%s: no connections available\n",
 260                         imp->imp_obd->obd_name);
 261                 spin_unlock(&imp->imp_lock);
 262                 RETURN(-EINVAL);
 263         }
 264
 265         if (imp->imp_conn_current &&
 266             imp->imp_conn_current->oic_item.next != &imp->imp_conn_list) {
 267                 imp_conn = list_entry(imp->imp_conn_current->oic_item.next,
 268                                       struct obd_import_conn, oic_item);
 269         } else {
 270                 imp_conn = list_entry(imp->imp_conn_list.next,
 271                                       struct obd_import_conn, oic_item);
 272         }
 273
 274         /* switch connection, don't mind if it's same as the current one */
 275         if (imp->imp_connection)
 276                 ptlrpc_put_connection(imp->imp_connection);
 277         imp->imp_connection = ptlrpc_connection_addref(imp_conn->oic_conn);
 278
 279         dlmexp =  class_conn2export(&imp->imp_dlm_handle);
 280         LASSERT(dlmexp != NULL);
 281         if (dlmexp->exp_connection)
 282                 ptlrpc_put_connection(dlmexp->exp_connection);
 283         dlmexp->exp_connection = ptlrpc_connection_addref(imp_conn->oic_conn);
 284         class_export_put(dlmexp);
 285
 286         if (imp->imp_conn_current && (imp->imp_conn_current != imp_conn)) {
 287                 LCONSOLE_WARN("Changing connection for %s to %s\n",
 288                               imp->imp_obd->obd_name, imp_conn->oic_uuid.uuid);
 289         }
 290         imp->imp_conn_current = imp_conn;
 291         CDEBUG(D_HA, "%s: import %p using connection %s\n",
 292                imp->imp_obd->obd_name, imp, imp_conn->oic_uuid.uuid);
 293         spin_unlock(&imp->imp_lock);
 294
 295         RETURN(0);
 296 }
 297
 298 int ptlrpc_connect_import(struct obd_import *imp, char * new_uuid)
 299 {
 300         struct obd_device *obd = imp->imp_obd;
 301         int initial_connect = 0;
 302         int rc;
 303         __u64 committed_before_reconnect = 0;
 304         struct ptlrpc_request *request;
 305         int size[] = {sizeof(imp->imp_target_uuid),
 306                       sizeof(obd->obd_uuid),
 307                       sizeof(imp->imp_dlm_handle),
 308                       sizeof(imp->imp_connect_data)};
 309         char *tmp[] = {imp->imp_target_uuid.uuid,
 310                        obd->obd_uuid.uuid,
 311                        (char *)&imp->imp_dlm_handle,
 312                        (char *)&imp->imp_connect_data};
 313         struct ptlrpc_connect_async_args *aa;
 314         unsigned long flags;
 315
 316         spin_lock_irqsave(&imp->imp_lock, flags);
 317         if (imp->imp_state == LUSTRE_IMP_CLOSED) {
 318                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 319                 CERROR("can't connect to a closed import\n");
 320                 RETURN(-EINVAL);
 321         } else if (imp->imp_state == LUSTRE_IMP_FULL) {
 322                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 323                 CERROR("already connected\n");
 324                 RETURN(0);
 325         } else if (imp->imp_state == LUSTRE_IMP_CONNECTING) {
 326                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 327                 CERROR("already connecting\n");
 328                 RETURN(-EALREADY);
 329         }
 330
 331         IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_CONNECTING);
 332
 333         imp->imp_conn_cnt++;
 334         imp->imp_resend_replay = 0;
 335
 336         if (imp->imp_remote_handle.cookie == 0) {
 337                 initial_connect = 1;
 338         } else {
 339                 committed_before_reconnect = imp->imp_peer_committed_transno;
 340         }
 341
 342         spin_unlock_irqrestore(&imp->imp_lock, flags);
 343
 344         if (new_uuid) {
 345                 struct obd_uuid uuid;
 346
 347                 obd_str2uuid(&uuid, new_uuid);
 348                 rc = import_set_conn_priority(imp, &uuid);
 349                 if (rc)
 350                         GOTO(out, rc);
 351         }
 352
 353         rc = import_select_connection(imp);
 354         if (rc)
 355                 GOTO(out, rc);
 356
 357         request = ptlrpc_prep_req(imp, imp->imp_connect_op, 4, size, tmp);
 358         if (!request)
 359                 GOTO(out, rc = -ENOMEM);
 360
 361 #ifndef __KERNEL__
 362         lustre_msg_add_op_flags(request->rq_reqmsg, MSG_CONNECT_LIBCLIENT);
 363 #endif
 364
 365         request->rq_send_state = LUSTRE_IMP_CONNECTING;
 366         /* Allow a slightly larger reply for future growth compatibility */
 367         size[0] = sizeof(struct obd_connect_data) + 16 * sizeof(__u64);
 368         request->rq_replen = lustre_msg_size(1, size);
 369         request->rq_interpret_reply = ptlrpc_connect_interpret;
 370
 371         LASSERT (sizeof (*aa) <= sizeof (request->rq_async_args));
 372         aa = (struct ptlrpc_connect_async_args *)&request->rq_async_args;
 373         memset(aa, 0, sizeof *aa);
 374
 375         aa->pcaa_peer_committed = committed_before_reconnect;
 376         aa->pcaa_initial_connect = initial_connect;
 377
 378         if (aa->pcaa_initial_connect) {
 379                 imp->imp_replayable = 1;
 380                 /* On an initial connect, we don't know which one of a
 381                    failover server pair is up.  Don't wait long. */
 382                 request->rq_timeout = max((int)(obd_timeout / 20), 5);
 383         }
 384
 385         DEBUG_REQ(D_RPCTRACE, request, "(re)connect request");
 386         ptlrpcd_add_req(request);
 387         rc = 0;
 388 out:
 389         if (rc != 0) {
 390                 IMPORT_SET_STATE(imp, LUSTRE_IMP_DISCON);
 391         }
 392
 393         RETURN(rc);
 394 }
 395
 396 static void ptlrpc_maybe_ping_import_soon(struct obd_import *imp)
 397 {
 398         struct obd_import_conn *imp_conn;
 399         unsigned long flags;
 400         int wake_pinger = 0;
 401
 402         ENTRY;
 403
 404         spin_lock_irqsave(&imp->imp_lock, flags);
 405         if (list_empty(&imp->imp_conn_list))
 406                 GOTO(unlock, 0);
 407
 408         imp_conn = list_entry(imp->imp_conn_list.prev,
 409                               struct obd_import_conn,
 410                               oic_item);
 411
 412         if (imp->imp_conn_current != imp_conn) {
 413                 ptlrpc_ping_import_soon(imp);
 414                 wake_pinger = 1;
 415         }
 416
 417  unlock:
 418         spin_unlock_irqrestore(&imp->imp_lock, flags);
 419
 420         if (wake_pinger)
 421                 ptlrpc_pinger_wake_up();
 422
 423         EXIT;
 424 }
 425
 426 static int ptlrpc_connect_interpret(struct ptlrpc_request *request,
 427                                     void * data, int rc)
 428 {
 429         struct ptlrpc_connect_async_args *aa = data;
 430         struct obd_import *imp = request->rq_import;
 431         struct lustre_handle old_hdl;
 432         unsigned long flags;
 433         int msg_flags;
 434         ENTRY;
 435
 436         spin_lock_irqsave(&imp->imp_lock, flags);
 437         if (imp->imp_state == LUSTRE_IMP_CLOSED) {
 438                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 439                 RETURN(0);
 440         }
 441         spin_unlock_irqrestore(&imp->imp_lock, flags);
 442
 443         if (rc)
 444                 GOTO(out, rc);
 445
 446         LASSERT(imp->imp_conn_current);
 447
 448         msg_flags = lustre_msg_get_op_flags(request->rq_repmsg);
 449
 450         /* All imports are pingable */
 451         imp->imp_pingable = 1;
 452
 453         if (aa->pcaa_initial_connect) {
 454                 if (msg_flags & MSG_CONNECT_REPLAYABLE) {
 455                         CDEBUG(D_HA, "connected to replayable target: %s\n",
 456                                imp->imp_target_uuid.uuid);
 457                         imp->imp_replayable = 1;
 458                 } else {
 459                         imp->imp_replayable = 0;
 460                 }
 461                 imp->imp_remote_handle = request->rq_repmsg->handle;
 462
 463                 IMPORT_SET_STATE(imp, LUSTRE_IMP_FULL);
 464                 GOTO(finish, rc = 0);
 465         }
 466
 467         /* Determine what recovery state to move the import to. */
 468         if (MSG_CONNECT_RECONNECT & msg_flags) {
 469                 memset(&old_hdl, 0, sizeof(old_hdl));
 470                 if (!memcmp(&old_hdl, &request->rq_repmsg->handle,
 471                             sizeof (old_hdl))) {
 472                         CERROR("%s@%s didn't like our handle "LPX64
 473                                ", failed\n", imp->imp_target_uuid.uuid,
 474                                imp->imp_connection->c_remote_uuid.uuid,
 475                                imp->imp_dlm_handle.cookie);
 476                         GOTO(out, rc = -ENOTCONN);
 477                 }
 478
 479                 if (memcmp(&imp->imp_remote_handle, &request->rq_repmsg->handle,
 480                            sizeof(imp->imp_remote_handle))) {
 481                         CERROR("%s@%s changed handle from "LPX64" to "LPX64
 482                                "; copying, but this may foreshadow disaster\n",
 483                                imp->imp_target_uuid.uuid,
 484                                imp->imp_connection->c_remote_uuid.uuid,
 485                                imp->imp_remote_handle.cookie,
 486                                request->rq_repmsg->handle.cookie);
 487                         imp->imp_remote_handle = request->rq_repmsg->handle;
 488                 } else {
 489                         CDEBUG(D_HA, "reconnected to %s@%s after partition\n",
 490                                imp->imp_target_uuid.uuid,
 491                                imp->imp_connection->c_remote_uuid.uuid);
 492                 }
 493
 494                 if (imp->imp_invalid) {
 495                         IMPORT_SET_STATE(imp, LUSTRE_IMP_EVICTED);
 496                 } else if (MSG_CONNECT_RECOVERING & msg_flags) {
 497                         CDEBUG(D_HA, "%s: reconnected to %s during replay\n",
 498                                imp->imp_obd->obd_name,
 499                                imp->imp_target_uuid.uuid);
 500                         imp->imp_resend_replay = 1;
 501                         IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY);
 502                 } else {
 503                         IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
 504                 }
 505         } else if ((MSG_CONNECT_RECOVERING & msg_flags) && !imp->imp_invalid) {
 506                 LASSERT(imp->imp_replayable);
 507                 imp->imp_remote_handle = request->rq_repmsg->handle;
 508                 imp->imp_last_replay_transno = 0;
 509                 IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY);
 510         } else {
 511                 imp->imp_remote_handle = request->rq_repmsg->handle;
 512                 IMPORT_SET_STATE(imp, LUSTRE_IMP_EVICTED);
 513         }
 514
 515         /* Sanity checks for a reconnected import. */
 516         if (!(imp->imp_replayable) != !(msg_flags & MSG_CONNECT_REPLAYABLE)) {
 517                 CERROR("imp_replayable flag does not match server "
 518                        "after reconnect. We should LBUG right here.\n");
 519         }
 520
 521         if (request->rq_repmsg->last_committed < aa->pcaa_peer_committed) {
 522                 CERROR("%s went back in time (transno "LPD64
 523                        " was previously committed, server now claims "LPD64
 524                        ")! is shared storage not coherent?\n",
 525                        imp->imp_target_uuid.uuid,
 526                        aa->pcaa_peer_committed,
 527                        request->rq_repmsg->last_committed);
 528         }
 529
 530 finish:
 531         rc = ptlrpc_import_recovery_state_machine(imp);
 532         if (rc != 0) {
 533                 if (rc == -ENOTCONN) {
 534                         CDEBUG(D_HA, "evicted/aborted by %s@%s during recovery;"
 535                                "invalidating and reconnecting\n",
 536                                imp->imp_target_uuid.uuid,
 537                                imp->imp_connection->c_remote_uuid.uuid);
 538                         ptlrpc_connect_import(imp, NULL);
 539                         RETURN(0);
 540                 }
 541         } else {
 542                 struct obd_connect_data *ocd;
 543
 544                 ocd = lustre_swab_repbuf(request, 0,
 545                                          sizeof *ocd, lustre_swab_connect);
 546                 if (ocd == NULL) {
 547                         CERROR("Wrong connect data from server\n");
 548                         rc = -EPROTO;
 549                         GOTO(out, rc);
 550                 }
 551                 spin_lock_irqsave(&imp->imp_lock, flags);
 552                 /*
 553                  * check that server granted subset of flags we asked for.
 554                  */
 555                 LASSERT((ocd->ocd_connect_flags &
 556                          imp->imp_connect_data.ocd_connect_flags) ==
 557                         ocd->ocd_connect_flags);
 558                 imp->imp_connect_data = *ocd;
 559                 if (imp->imp_conn_current != NULL) {
 560                         list_del(&imp->imp_conn_current->oic_item);
 561                         list_add(&imp->imp_conn_current->oic_item,
 562                                  &imp->imp_conn_list);
 563                         imp->imp_conn_current = NULL;
 564                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 565                 } else {
 566                         static int bug7269_dump = 0;
 567                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 568                         CERROR("this is bug 7269 - please attach log there\n");
 569                         if (bug7269_dump == 0)
 570                                 libcfs_debug_dumplog();
 571                         bug7269_dump = 1;
 572                 }
 573         }
 574
 575  out:
 576         if (rc != 0) {
 577                 IMPORT_SET_STATE(imp, LUSTRE_IMP_DISCON);
 578                 if (aa->pcaa_initial_connect && !imp->imp_initial_recov) {
 579                         ptlrpc_deactivate_import(imp);
 580                 }
 581
 582                 ptlrpc_maybe_ping_import_soon(imp);
 583
 584                 CDEBUG(D_HA, "recovery of %s on %s failed (%d)\n",
 585                        imp->imp_target_uuid.uuid,
 586                        (char *)imp->imp_connection->c_remote_uuid.uuid, rc);
 587         }
 588
 589         wake_up(&imp->imp_recovery_waitq);
 590         RETURN(rc);
 591 }
 592
 593 static int completed_replay_interpret(struct ptlrpc_request *req,
 594                                     void * data, int rc)
 595 {
 596         atomic_dec(&req->rq_import->imp_replay_inflight);
 597         if (req->rq_status == 0) {
 598                 ptlrpc_import_recovery_state_machine(req->rq_import);
 599         } else {
 600                 CDEBUG(D_HA, "%s: LAST_REPLAY message error: %d, "
 601                        "reconnecting\n",
 602                        req->rq_import->imp_obd->obd_name, req->rq_status);
 603                 ptlrpc_connect_import(req->rq_import, NULL);
 604         }
 605
 606         RETURN(0);
 607 }
 608
 609 static int signal_completed_replay(struct obd_import *imp)
 610 {
 611         struct ptlrpc_request *req;
 612         ENTRY;
 613
 614         LASSERT(atomic_read(&imp->imp_replay_inflight) == 0);
 615         atomic_inc(&imp->imp_replay_inflight);
 616
 617         req = ptlrpc_prep_req(imp, OBD_PING, 0, NULL, NULL);
 618         if (!req) {
 619                 atomic_dec(&imp->imp_replay_inflight);
 620                 RETURN(-ENOMEM);
 621         }
 622
 623         req->rq_replen = lustre_msg_size(0, NULL);
 624         req->rq_send_state = LUSTRE_IMP_REPLAY_WAIT;
 625         req->rq_reqmsg->flags |= MSG_LAST_REPLAY;
 626         req->rq_timeout *= 3;
 627         req->rq_interpret_reply = completed_replay_interpret;
 628
 629         ptlrpcd_add_req(req);
 630         RETURN(0);
 631 }
 632
 633 #ifdef __KERNEL__
 634 static int ptlrpc_invalidate_import_thread(void *data)
 635 {
 636         struct obd_import *imp = data;
 637         unsigned long flags;
 638
 639         ENTRY;
 640
 641         lock_kernel();
 642         ptlrpc_daemonize();
 643
 644         SIGNAL_MASK_LOCK(current, flags);
 645         sigfillset(&current->blocked);
 646         RECALC_SIGPENDING;
 647         SIGNAL_MASK_UNLOCK(current, flags);
 648         THREAD_NAME(current->comm, sizeof(current->comm), "ll_imp_inval");
 649         unlock_kernel();
 650
 651         CDEBUG(D_HA, "thread invalidate import %s to %s@%s\n",
 652                imp->imp_obd->obd_name, imp->imp_target_uuid.uuid,
 653                imp->imp_connection->c_remote_uuid.uuid);
 654
 655         ptlrpc_invalidate_import(imp);
 656
 657         IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
 658         ptlrpc_import_recovery_state_machine(imp);
 659
 660         RETURN(0);
 661 }
 662 #endif
 663
 664 int ptlrpc_import_recovery_state_machine(struct obd_import *imp)
 665 {
 666         int rc = 0;
 667         int inflight;
 668         char *target_start;
 669         int target_len;
 670
 671         if (imp->imp_state == LUSTRE_IMP_EVICTED) {
 672                 deuuidify(imp->imp_target_uuid.uuid, NULL,
 673                           &target_start, &target_len);
 674                 LCONSOLE_ERROR("This client was evicted by %.*s; in progress "
 675                                "operations using this service will fail.\n",
 676                                target_len, target_start);
 677                 CDEBUG(D_HA, "evicted from %s@%s; invalidating\n",
 678                        imp->imp_target_uuid.uuid,
 679                        imp->imp_connection->c_remote_uuid.uuid);
 680
 681 #ifdef __KERNEL__
 682                 rc = kernel_thread(ptlrpc_invalidate_import_thread, imp,
 683                                    CLONE_VM | CLONE_FILES);
 684                 if (rc < 0)
 685                         CERROR("error starting invalidate thread: %d\n", rc);
 686                 else
 687                         rc = 0;
 688                 RETURN(rc);
 689 #else
 690                 ptlrpc_invalidate_import(imp);
 691
 692                 IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
 693 #endif
 694         }
 695
 696         if (imp->imp_state == LUSTRE_IMP_REPLAY) {
 697                 CDEBUG(D_HA, "replay requested by %s\n",
 698                        imp->imp_target_uuid.uuid);
 699                 rc = ptlrpc_replay_next(imp, &inflight);
 700                 if (inflight == 0 &&
 701                     atomic_read(&imp->imp_replay_inflight) == 0) {
 702                         IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY_LOCKS);
 703                         rc = ldlm_replay_locks(imp);
 704                         if (rc)
 705                                 GOTO(out, rc);
 706                 }
 707                 rc = 0;
 708         }
 709
 710         if (imp->imp_state == LUSTRE_IMP_REPLAY_LOCKS) {
 711                 if (atomic_read(&imp->imp_replay_inflight) == 0) {
 712                         IMPORT_SET_STATE(imp, LUSTRE_IMP_REPLAY_WAIT);
 713                         rc = signal_completed_replay(imp);
 714                         if (rc)
 715                                 GOTO(out, rc);
 716                 }
 717
 718         }
 719
 720         if (imp->imp_state == LUSTRE_IMP_REPLAY_WAIT) {
 721                 if (atomic_read(&imp->imp_replay_inflight) == 0) {
 722                         IMPORT_SET_STATE(imp, LUSTRE_IMP_RECOVER);
 723                 }
 724         }
 725
 726         if (imp->imp_state == LUSTRE_IMP_RECOVER) {
 727                 char   *nidstr;
 728
 729                 CDEBUG(D_HA, "reconnected to %s@%s\n",
 730                        imp->imp_target_uuid.uuid,
 731                        imp->imp_connection->c_remote_uuid.uuid);
 732
 733                 rc = ptlrpc_resend(imp);
 734                 if (rc)
 735                         GOTO(out, rc);
 736                 IMPORT_SET_STATE(imp, LUSTRE_IMP_FULL);
 737                 ptlrpc_activate_import(imp);
 738
 739                 deuuidify(imp->imp_target_uuid.uuid, NULL,
 740                           &target_start, &target_len);
 741                 nidstr = libcfs_nid2str(imp->imp_connection->c_peer.nid);
 742
 743                 LCONSOLE_INFO("Connection restored to service %.*s using nid "
 744                               "%s.\n", target_len, target_start, nidstr);
 745
 746                 CWARN("%s: connection restored to %s@%s\n",
 747                       imp->imp_obd->obd_name,
 748                       imp->imp_target_uuid.uuid,
 749                       imp->imp_connection->c_remote_uuid.uuid);
 750         }
 751
 752         if (imp->imp_state == LUSTRE_IMP_FULL) {
 753                 wake_up(&imp->imp_recovery_waitq);
 754                 ptlrpc_wake_delayed(imp);
 755         }
 756
 757  out:
 758         RETURN(rc);
 759 }
 760
 761 static int back_to_sleep(void *unused)
 762 {
 763         return 0;
 764 }
 765
 766 int ptlrpc_disconnect_import(struct obd_import *imp)
 767 {
 768         struct ptlrpc_request *request;
 769         int rq_opc;
 770         int rc = 0;
 771         unsigned long flags;
 772         ENTRY;
 773
 774         switch (imp->imp_connect_op) {
 775         case OST_CONNECT: rq_opc = OST_DISCONNECT; break;
 776         case MDS_CONNECT: rq_opc = MDS_DISCONNECT; break;
 777         case MGMT_CONNECT:rq_opc = MGMT_DISCONNECT;break;
 778         default:
 779                 CERROR("don't know how to disconnect from %s (connect_op %d)\n",
 780                        imp->imp_target_uuid.uuid, imp->imp_connect_op);
 781                 RETURN(-EINVAL);
 782         }
 783
 784
 785         if (ptlrpc_import_in_recovery(imp)) {
 786                 struct l_wait_info lwi;
 787                 lwi = LWI_TIMEOUT_INTR(MAX(obd_timeout * HZ, 1), back_to_sleep,
 788                                        NULL, NULL);
 789                 rc = l_wait_event(imp->imp_recovery_waitq,
 790                                   !ptlrpc_import_in_recovery(imp), &lwi);
 791
 792         }
 793
 794         spin_lock_irqsave(&imp->imp_lock, flags);
 795         if (imp->imp_state != LUSTRE_IMP_FULL)
 796                 GOTO(out, 0);
 797
 798         spin_unlock_irqrestore(&imp->imp_lock, flags);
 799
 800         request = ptlrpc_prep_req(imp, rq_opc, 0, NULL, NULL);
 801         if (request) {
 802                 /* We are disconnecting, do not retry a failed DISCONNECT rpc if
 803                  * it fails.  We can get through the above with a down server
 804                  * if the client doesn't know the server is gone yet. */
 805                 request->rq_no_resend = 1;
 806                 IMPORT_SET_STATE(imp, LUSTRE_IMP_CONNECTING);
 807                 request->rq_send_state =  LUSTRE_IMP_CONNECTING;
 808                 request->rq_replen = lustre_msg_size(0, NULL);
 809                 rc = ptlrpc_queue_wait(request);
 810                 ptlrpc_req_finished(request);
 811         }
 812
 813         spin_lock_irqsave(&imp->imp_lock, flags);
 814 out:
 815         IMPORT_SET_STATE_NOLOCK(imp, LUSTRE_IMP_CLOSED);
 816         memset(&imp->imp_remote_handle, 0, sizeof(imp->imp_remote_handle));
 817         spin_unlock_irqrestore(&imp->imp_lock, flags);
 818
 819         RETURN(rc);
 820 }
 821