lustre/ptlrpc/pinger.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2015, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/ptlrpc/pinger.c
  33  *
  34  * Portal-RPC reconnection and replay operations, for use in recovery.
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_RPC
  38
  39 #include <linux/kthread.h>
  40 #include <obd_support.h>
  41 #include <obd_class.h>
  42 #include "ptlrpc_internal.h"
  43
  44 static int suppress_pings;
  45 module_param(suppress_pings, int, 0644);
  46 MODULE_PARM_DESC(suppress_pings, "Suppress pings");
  47
  48 struct mutex pinger_mutex;
  49 static struct list_head pinger_imports =
  50                 LIST_HEAD_INIT(pinger_imports);
  51 static struct list_head timeout_list =
  52                 LIST_HEAD_INIT(timeout_list);
  53
  54 int ptlrpc_pinger_suppress_pings()
  55 {
  56         return suppress_pings;
  57 }
  58 EXPORT_SYMBOL(ptlrpc_pinger_suppress_pings);
  59
  60 struct ptlrpc_request *
  61 ptlrpc_prep_ping(struct obd_import *imp)
  62 {
  63         struct ptlrpc_request *req;
  64
  65         req = ptlrpc_request_alloc_pack(imp, &RQF_OBD_PING,
  66                                         LUSTRE_OBD_VERSION, OBD_PING);
  67         if (req) {
  68                 ptlrpc_request_set_replen(req);
  69                 req->rq_no_resend = req->rq_no_delay = 1;
  70         }
  71         return req;
  72 }
  73
  74 int ptlrpc_obd_ping(struct obd_device *obd)
  75 {
  76         int rc;
  77         struct ptlrpc_request *req;
  78         ENTRY;
  79
  80         req = ptlrpc_prep_ping(obd->u.cli.cl_import);
  81         if (req == NULL)
  82                 RETURN(-ENOMEM);
  83
  84         req->rq_send_state = LUSTRE_IMP_FULL;
  85
  86         rc = ptlrpc_queue_wait(req);
  87
  88         ptlrpc_req_finished(req);
  89
  90         RETURN(rc);
  91 }
  92 EXPORT_SYMBOL(ptlrpc_obd_ping);
  93
  94 static int ptlrpc_ping(struct obd_import *imp)
  95 {
  96         struct ptlrpc_request   *req;
  97         ENTRY;
  98
  99         req = ptlrpc_prep_ping(imp);
 100         if (req == NULL) {
 101                 CERROR("OOM trying to ping %s->%s\n",
 102                        imp->imp_obd->obd_uuid.uuid,
 103                        obd2cli_tgt(imp->imp_obd));
 104                 RETURN(-ENOMEM);
 105         }
 106
 107         DEBUG_REQ(D_INFO, req, "pinging %s->%s",
 108                   imp->imp_obd->obd_uuid.uuid, obd2cli_tgt(imp->imp_obd));
 109         ptlrpcd_add_req(req);
 110
 111         RETURN(0);
 112 }
 113
 114 static void ptlrpc_update_next_ping(struct obd_import *imp, int soon)
 115 {
 116 #ifdef ENABLE_PINGER
 117         time64_t time = soon ? PING_INTERVAL_SHORT : PING_INTERVAL;
 118
 119         if (imp->imp_state == LUSTRE_IMP_DISCON) {
 120                 time64_t dtime = max_t(time64_t, CONNECTION_SWITCH_MIN,
 121                                        AT_OFF ? 0 :
 122                                        at_get(&imp->imp_at.iat_net_latency));
 123                 time = min(time, dtime);
 124         }
 125         imp->imp_next_ping = ktime_get_seconds() + time;
 126 #endif /* ENABLE_PINGER */
 127 }
 128
 129 void ptlrpc_ping_import_soon(struct obd_import *imp)
 130 {
 131         imp->imp_next_ping = ktime_get_seconds();
 132 }
 133
 134 static inline int imp_is_deactive(struct obd_import *imp)
 135 {
 136         return (imp->imp_deactive ||
 137                 OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_IMP_DEACTIVE));
 138 }
 139
 140 static inline time64_t ptlrpc_next_reconnect(struct obd_import *imp)
 141 {
 142         if (imp->imp_server_timeout)
 143                 return ktime_get_seconds() + (obd_timeout >> 1);
 144         else
 145                 return ktime_get_seconds() + obd_timeout;
 146 }
 147
 148 static time64_t pinger_check_timeout(time64_t time)
 149 {
 150         struct timeout_item *item;
 151         time64_t timeout = PING_INTERVAL;
 152
 153         /* This list is sorted in increasing timeout order */
 154         mutex_lock(&pinger_mutex);
 155         list_for_each_entry(item, &timeout_list, ti_chain) {
 156                 time64_t ti_timeout = item->ti_timeout;
 157
 158                 if (timeout > ti_timeout)
 159                         timeout = ti_timeout;
 160                 break;
 161         }
 162         mutex_unlock(&pinger_mutex);
 163
 164         return time + timeout - ktime_get_seconds();
 165 }
 166
 167 static bool ir_up;
 168
 169 void ptlrpc_pinger_ir_up(void)
 170 {
 171         CDEBUG(D_HA, "IR up\n");
 172         ir_up = true;
 173 }
 174 EXPORT_SYMBOL(ptlrpc_pinger_ir_up);
 175
 176 void ptlrpc_pinger_ir_down(void)
 177 {
 178         CDEBUG(D_HA, "IR down\n");
 179         ir_up = false;
 180 }
 181 EXPORT_SYMBOL(ptlrpc_pinger_ir_down);
 182
 183 static void ptlrpc_pinger_process_import(struct obd_import *imp,
 184                                          time64_t this_ping)
 185 {
 186         int level;
 187         int force;
 188         int force_next;
 189         int suppress;
 190
 191         spin_lock(&imp->imp_lock);
 192
 193         level = imp->imp_state;
 194         force = imp->imp_force_verify;
 195         force_next = imp->imp_force_next_verify;
 196         /*
 197          * This will be used below only if the import is "FULL".
 198          */
 199         suppress = ir_up && OCD_HAS_FLAG(&imp->imp_connect_data, PINGLESS);
 200
 201         imp->imp_force_verify = 0;
 202
 203         if (imp->imp_next_ping - 5 >= this_ping && !force) {
 204                 spin_unlock(&imp->imp_lock);
 205                 return;
 206         }
 207
 208         imp->imp_force_next_verify = 0;
 209
 210         spin_unlock(&imp->imp_lock);
 211
 212         CDEBUG(level == LUSTRE_IMP_FULL ? D_INFO : D_HA, "%s->%s: level %s/%u "
 213                "force %u force_next %u deactive %u pingable %u suppress %u\n",
 214                imp->imp_obd->obd_uuid.uuid, obd2cli_tgt(imp->imp_obd),
 215                ptlrpc_import_state_name(level), level, force, force_next,
 216                imp->imp_deactive, imp->imp_pingable, suppress);
 217
 218         if (level == LUSTRE_IMP_DISCON && !imp_is_deactive(imp)) {
 219                 /* wait for a while before trying recovery again */
 220                 imp->imp_next_ping = ptlrpc_next_reconnect(imp);
 221                 if (!imp->imp_no_pinger_recover)
 222                         ptlrpc_initiate_recovery(imp);
 223         } else if (level != LUSTRE_IMP_FULL ||
 224                    imp->imp_obd->obd_no_recov ||
 225                    imp_is_deactive(imp)) {
 226                 CDEBUG(D_HA, "%s->%s: not pinging (in recovery "
 227                        "or recovery disabled: %s)\n",
 228                        imp->imp_obd->obd_uuid.uuid, obd2cli_tgt(imp->imp_obd),
 229                        ptlrpc_import_state_name(level));
 230                 if (force) {
 231                         spin_lock(&imp->imp_lock);
 232                         imp->imp_force_verify = 1;
 233                         spin_unlock(&imp->imp_lock);
 234                 }
 235         } else if ((imp->imp_pingable && !suppress) || force_next || force) {
 236                 ptlrpc_ping(imp);
 237         }
 238 }
 239
 240 static int ptlrpc_pinger_main(void *arg)
 241 {
 242         struct ptlrpc_thread *thread = (struct ptlrpc_thread *)arg;
 243         ENTRY;
 244
 245         /* Record that the thread is running */
 246         thread_set_flags(thread, SVC_RUNNING);
 247         wake_up(&thread->t_ctl_waitq);
 248
 249         /* And now, loop forever, pinging as needed. */
 250         while (1) {
 251                 time64_t this_ping = ktime_get_seconds();
 252                 struct l_wait_info lwi;
 253                 time64_t time_to_next_wake;
 254                 struct timeout_item *item;
 255                 struct list_head *iter;
 256
 257                 mutex_lock(&pinger_mutex);
 258                 list_for_each_entry(item, &timeout_list, ti_chain)
 259                         item->ti_cb(item, item->ti_cb_data);
 260
 261                 list_for_each(iter, &pinger_imports) {
 262                         struct obd_import *imp = list_entry(iter,
 263                                                             struct obd_import,
 264                                                             imp_pinger_chain);
 265
 266                         ptlrpc_pinger_process_import(imp, this_ping);
 267                         /* obd_timeout might have changed */
 268                         if (imp->imp_pingable && imp->imp_next_ping &&
 269                             imp->imp_next_ping > this_ping + PING_INTERVAL)
 270                                 ptlrpc_update_next_ping(imp, 0);
 271                 }
 272                 mutex_unlock(&pinger_mutex);
 273                 /* update memory usage info */
 274                 obd_update_maxusage();
 275
 276                 /* Wait until the next ping time, or until we're stopped. */
 277                 time_to_next_wake = pinger_check_timeout(this_ping);
 278                 /* The ping sent by ptlrpc_send_rpc may get sent out
 279                    say .01 second after this.
 280                    ptlrpc_pinger_sending_on_import will then set the
 281                    next ping time to next_ping + .01 sec, which means
 282                    we will SKIP the next ping at next_ping, and the
 283                    ping will get sent 2 timeouts from now!  Beware. */
 284                 CDEBUG(D_INFO, "next wakeup in %lld (%lld)\n",
 285                        time_to_next_wake, this_ping + PING_INTERVAL);
 286                 if (time_to_next_wake > 0) {
 287                         time64_t time_max = max(time_to_next_wake, 1LL);
 288
 289                         lwi = LWI_TIMEOUT(cfs_time_seconds(time_max),
 290                                           NULL, NULL);
 291                         l_wait_event(thread->t_ctl_waitq,
 292                                      thread_is_stopping(thread) ||
 293                                      thread_is_event(thread),
 294                                      &lwi);
 295                         if (thread_test_and_clear_flags(thread, SVC_STOPPING)) {
 296                                 EXIT;
 297                                 break;
 298                         } else {
 299                                 /* woken after adding import to reset timer */
 300                                 thread_test_and_clear_flags(thread, SVC_EVENT);
 301                         }
 302                 }
 303         }
 304
 305         thread_set_flags(thread, SVC_STOPPED);
 306         wake_up(&thread->t_ctl_waitq);
 307
 308         CDEBUG(D_NET, "pinger thread exiting, process %d\n", current_pid());
 309         return 0;
 310 }
 311
 312 static struct ptlrpc_thread pinger_thread;
 313
 314 int ptlrpc_start_pinger(void)
 315 {
 316         struct l_wait_info lwi = { 0 };
 317         struct task_struct *task;
 318         int rc;
 319 #ifndef ENABLE_PINGER
 320         return 0;
 321 #endif
 322         ENTRY;
 323
 324         if (!thread_is_init(&pinger_thread) &&
 325             !thread_is_stopped(&pinger_thread))
 326                 RETURN(-EALREADY);
 327
 328         init_waitqueue_head(&pinger_thread.t_ctl_waitq);
 329
 330         strcpy(pinger_thread.t_name, "ll_ping");
 331
 332         task = kthread_run(ptlrpc_pinger_main, &pinger_thread,
 333                            pinger_thread.t_name);
 334         if (IS_ERR(task)) {
 335                 rc = PTR_ERR(task);
 336                 CERROR("cannot start pinger thread: rc = %d\n", rc);
 337                 RETURN(rc);
 338         }
 339
 340         l_wait_event(pinger_thread.t_ctl_waitq,
 341                      thread_is_running(&pinger_thread), &lwi);
 342
 343         if (suppress_pings)
 344                 CWARN("Pings will be suppressed at the request of the "
 345                       "administrator.  The configuration shall meet the "
 346                       "additional requirements described in the manual.  "
 347                       "(Search for the \"suppress_pings\" kernel module "
 348                       "parameter.)\n");
 349
 350         RETURN(0);
 351 }
 352
 353 int ptlrpc_pinger_remove_timeouts(void);
 354
 355 int ptlrpc_stop_pinger(void)
 356 {
 357         struct l_wait_info lwi = { 0 };
 358 #ifndef ENABLE_PINGER
 359         return 0;
 360 #endif
 361         ENTRY;
 362
 363         if (thread_is_init(&pinger_thread) ||
 364             thread_is_stopped(&pinger_thread))
 365                 RETURN(-EALREADY);
 366
 367         ptlrpc_pinger_remove_timeouts();
 368
 369         thread_set_flags(&pinger_thread, SVC_STOPPING);
 370         wake_up(&pinger_thread.t_ctl_waitq);
 371
 372         l_wait_event(pinger_thread.t_ctl_waitq,
 373                      thread_is_stopped(&pinger_thread), &lwi);
 374         RETURN(0);
 375 }
 376
 377 void ptlrpc_pinger_sending_on_import(struct obd_import *imp)
 378 {
 379         ptlrpc_update_next_ping(imp, 0);
 380 }
 381
 382 void ptlrpc_pinger_commit_expected(struct obd_import *imp)
 383 {
 384         ptlrpc_update_next_ping(imp, 1);
 385         assert_spin_locked(&imp->imp_lock);
 386         /*
 387          * Avoid reading stale imp_connect_data.  When not sure if pings are
 388          * expected or not on next connection, we assume they are not and force
 389          * one anyway to guarantee the chance of updating
 390          * imp_peer_committed_transno.
 391          */
 392         if (imp->imp_state != LUSTRE_IMP_FULL ||
 393             OCD_HAS_FLAG(&imp->imp_connect_data, PINGLESS))
 394                 imp->imp_force_next_verify = 1;
 395 }
 396
 397 int ptlrpc_pinger_add_import(struct obd_import *imp)
 398 {
 399         ENTRY;
 400         if (!list_empty(&imp->imp_pinger_chain))
 401                 RETURN(-EALREADY);
 402
 403         mutex_lock(&pinger_mutex);
 404         CDEBUG(D_HA, "adding pingable import %s->%s\n",
 405                imp->imp_obd->obd_uuid.uuid, obd2cli_tgt(imp->imp_obd));
 406         /* if we add to pinger we want recovery on this import */
 407         imp->imp_obd->obd_no_recov = 0;
 408         ptlrpc_update_next_ping(imp, 0);
 409         /* XXX sort, blah blah */
 410         list_add_tail(&imp->imp_pinger_chain, &pinger_imports);
 411         class_import_get(imp);
 412
 413         ptlrpc_pinger_wake_up();
 414         mutex_unlock(&pinger_mutex);
 415
 416         RETURN(0);
 417 }
 418 EXPORT_SYMBOL(ptlrpc_pinger_add_import);
 419
 420 int ptlrpc_pinger_del_import(struct obd_import *imp)
 421 {
 422         ENTRY;
 423
 424         if (list_empty(&imp->imp_pinger_chain))
 425                 RETURN(-ENOENT);
 426
 427         mutex_lock(&pinger_mutex);
 428         list_del_init(&imp->imp_pinger_chain);
 429         CDEBUG(D_HA, "removing pingable import %s->%s\n",
 430                imp->imp_obd->obd_uuid.uuid, obd2cli_tgt(imp->imp_obd));
 431         /* if we remove from pinger we don't want recovery on this import */
 432         imp->imp_obd->obd_no_recov = 1;
 433         class_import_put(imp);
 434         mutex_unlock(&pinger_mutex);
 435         RETURN(0);
 436 }
 437 EXPORT_SYMBOL(ptlrpc_pinger_del_import);
 438
 439 /**
 440  * Register a timeout callback to the pinger list, and the callback will
 441  * be called when timeout happens.
 442  */
 443 static struct timeout_item *ptlrpc_new_timeout(time64_t time,
 444                                                enum timeout_event event,
 445                                                timeout_cb_t cb, void *data)
 446 {
 447         struct timeout_item *ti;
 448
 449         OBD_ALLOC_PTR(ti);
 450         if (!ti)
 451                 return(NULL);
 452
 453         INIT_LIST_HEAD(&ti->ti_obd_list);
 454         INIT_LIST_HEAD(&ti->ti_chain);
 455         ti->ti_timeout = time;
 456         ti->ti_event = event;
 457         ti->ti_cb = cb;
 458         ti->ti_cb_data = data;
 459
 460         return ti;
 461 }
 462
 463 /**
 464  * Register timeout event on the the pinger thread.
 465  * Note: the timeout list is an sorted list with increased timeout value.
 466  */
 467 static struct timeout_item*
 468 ptlrpc_pinger_register_timeout(time64_t time, enum timeout_event event,
 469                                timeout_cb_t cb, void *data)
 470 {
 471         struct timeout_item *item, *tmp;
 472
 473         LASSERT(mutex_is_locked(&pinger_mutex));
 474
 475         list_for_each_entry(item, &timeout_list, ti_chain)
 476                 if (item->ti_event == event)
 477                         goto out;
 478
 479         item = ptlrpc_new_timeout(time, event, cb, data);
 480         if (item) {
 481                 list_for_each_entry_reverse(tmp, &timeout_list, ti_chain) {
 482                         if (tmp->ti_timeout < time) {
 483                                 list_add(&item->ti_chain, &tmp->ti_chain);
 484                                 goto out;
 485                         }
 486                 }
 487                 list_add(&item->ti_chain, &timeout_list);
 488         }
 489 out:
 490         return item;
 491 }
 492
 493 /* Add a client_obd to the timeout event list, when timeout(@time)
 494  * happens, the callback(@cb) will be called.
 495  */
 496 int ptlrpc_add_timeout_client(time64_t time, enum timeout_event event,
 497                               timeout_cb_t cb, void *data,
 498                               struct list_head *obd_list)
 499 {
 500         struct timeout_item *ti;
 501
 502         mutex_lock(&pinger_mutex);
 503         ti = ptlrpc_pinger_register_timeout(time, event, cb, data);
 504         if (!ti) {
 505                 mutex_unlock(&pinger_mutex);
 506                 return (-EINVAL);
 507         }
 508         list_add(obd_list, &ti->ti_obd_list);
 509         mutex_unlock(&pinger_mutex);
 510         return 0;
 511 }
 512 EXPORT_SYMBOL(ptlrpc_add_timeout_client);
 513
 514 int ptlrpc_del_timeout_client(struct list_head *obd_list,
 515                               enum timeout_event event)
 516 {
 517         struct timeout_item *ti = NULL, *item;
 518
 519         if (list_empty(obd_list))
 520                 return 0;
 521         mutex_lock(&pinger_mutex);
 522         list_del_init(obd_list);
 523         /**
 524          * If there are no obd attached to the timeout event
 525          * list, remove this timeout event from the pinger
 526          */
 527         list_for_each_entry(item, &timeout_list, ti_chain) {
 528                 if (item->ti_event == event) {
 529                         ti = item;
 530                         break;
 531                 }
 532         }
 533         LASSERTF(ti != NULL, "ti is NULL !\n");
 534         if (list_empty(&ti->ti_obd_list)) {
 535                 list_del(&ti->ti_chain);
 536                 OBD_FREE_PTR(ti);
 537         }
 538         mutex_unlock(&pinger_mutex);
 539         return 0;
 540 }
 541 EXPORT_SYMBOL(ptlrpc_del_timeout_client);
 542
 543 int ptlrpc_pinger_remove_timeouts(void)
 544 {
 545         struct timeout_item *item, *tmp;
 546
 547         mutex_lock(&pinger_mutex);
 548         list_for_each_entry_safe(item, tmp, &timeout_list, ti_chain) {
 549                 LASSERT(list_empty(&item->ti_obd_list));
 550                 list_del(&item->ti_chain);
 551                 OBD_FREE_PTR(item);
 552         }
 553         mutex_unlock(&pinger_mutex);
 554         return 0;
 555 }
 556
 557 void ptlrpc_pinger_wake_up()
 558 {
 559 #ifdef ENABLE_PINGER
 560         thread_add_flags(&pinger_thread, SVC_EVENT);
 561         wake_up(&pinger_thread.t_ctl_waitq);
 562 #endif
 563 }
 564
 565 /* Ping evictor thread */
 566 #define PET_READY     1
 567 #define PET_TERMINATE 2
 568
 569 static int               pet_refcount = 0;
 570 static int               pet_state;
 571 static wait_queue_head_t pet_waitq;
 572 static struct list_head  pet_list;
 573 static DEFINE_SPINLOCK(pet_lock);
 574
 575 int ping_evictor_wake(struct obd_export *exp)
 576 {
 577         struct obd_device *obd;
 578
 579         spin_lock(&pet_lock);
 580         if (pet_state != PET_READY) {
 581                 /* eventually the new obd will call here again. */
 582                 spin_unlock(&pet_lock);
 583                 return 1;
 584         }
 585
 586         obd = class_exp2obd(exp);
 587         if (list_empty(&obd->obd_evict_list)) {
 588                 class_incref(obd, "evictor", obd);
 589                 list_add(&obd->obd_evict_list, &pet_list);
 590         }
 591         spin_unlock(&pet_lock);
 592
 593         wake_up(&pet_waitq);
 594         return 0;
 595 }
 596
 597 static int ping_evictor_main(void *arg)
 598 {
 599         struct obd_device *obd;
 600         struct obd_export *exp;
 601         struct l_wait_info lwi = { 0 };
 602         time64_t expire_time;
 603
 604         ENTRY;
 605         unshare_fs_struct();
 606
 607         CDEBUG(D_HA, "Starting Ping Evictor\n");
 608         pet_state = PET_READY;
 609         while (1) {
 610                 l_wait_event(pet_waitq, (!list_empty(&pet_list)) ||
 611                             (pet_state == PET_TERMINATE), &lwi);
 612
 613                 /* loop until all obd's will be removed */
 614                 if ((pet_state == PET_TERMINATE) && list_empty(&pet_list))
 615                         break;
 616
 617                 /* we only get here if pet_exp != NULL, and the end of this
 618                  * loop is the only place which sets it NULL again, so lock
 619                  * is not strictly necessary. */
 620                 spin_lock(&pet_lock);
 621                 obd = list_entry(pet_list.next, struct obd_device,
 622                                  obd_evict_list);
 623                 spin_unlock(&pet_lock);
 624
 625                 expire_time = ktime_get_real_seconds() - PING_EVICT_TIMEOUT;
 626
 627                 CDEBUG(D_HA, "evicting all exports of obd %s older than %lld\n",
 628                        obd->obd_name, expire_time);
 629
 630                 /* Exports can't be deleted out of the list while we hold
 631                  * the obd lock (class_unlink_export), which means we can't
 632                  * lose the last ref on the export.  If they've already been
 633                  * removed from the list, we won't find them here. */
 634                 spin_lock(&obd->obd_dev_lock);
 635                 while (!list_empty(&obd->obd_exports_timed)) {
 636                         exp = list_entry(obd->obd_exports_timed.next,
 637                                          struct obd_export,
 638                                          exp_obd_chain_timed);
 639                         if (expire_time > exp->exp_last_request_time) {
 640                                 class_export_get(exp);
 641                                 spin_unlock(&obd->obd_dev_lock);
 642                                 LCONSOLE_WARN("%s: haven't heard from client %s"
 643                                               " (at %s) in %lld seconds. I think"
 644                                               " it's dead, and I am evicting"
 645                                               " it. exp %p, cur %lld expire %lld"
 646                                               " last %lld\n",
 647                                               obd->obd_name,
 648                                               obd_uuid2str(&exp->exp_client_uuid),
 649                                               obd_export_nid2str(exp),
 650                                               ktime_get_real_seconds() -
 651                                               exp->exp_last_request_time,
 652                                               exp, ktime_get_real_seconds(),
 653                                               expire_time,
 654                                               exp->exp_last_request_time);
 655                                 CDEBUG(D_HA, "Last request was at %lld\n",
 656                                        exp->exp_last_request_time);
 657                                 class_fail_export(exp);
 658                                 class_export_put(exp);
 659                                 spin_lock(&obd->obd_dev_lock);
 660                         } else {
 661                                 /* List is sorted, so everyone below is ok */
 662                                 break;
 663                         }
 664                 }
 665                 spin_unlock(&obd->obd_dev_lock);
 666
 667                 spin_lock(&pet_lock);
 668                 list_del_init(&obd->obd_evict_list);
 669                 spin_unlock(&pet_lock);
 670
 671                 class_decref(obd, "evictor", obd);
 672         }
 673         CDEBUG(D_HA, "Exiting Ping Evictor\n");
 674
 675         RETURN(0);
 676 }
 677
 678 void ping_evictor_start(void)
 679 {
 680         struct task_struct *task;
 681
 682         if (++pet_refcount > 1)
 683                 return;
 684
 685         INIT_LIST_HEAD(&pet_list);
 686         init_waitqueue_head(&pet_waitq);
 687
 688         task = kthread_run(ping_evictor_main, NULL, "ll_evictor");
 689         if (IS_ERR(task)) {
 690                 pet_refcount--;
 691                 CERROR("Cannot start ping evictor thread: %ld\n",
 692                         PTR_ERR(task));
 693         }
 694 }
 695 EXPORT_SYMBOL(ping_evictor_start);
 696
 697 void ping_evictor_stop(void)
 698 {
 699         if (--pet_refcount > 0)
 700                 return;
 701
 702         pet_state = PET_TERMINATE;
 703         wake_up(&pet_waitq);
 704 }
 705 EXPORT_SYMBOL(ping_evictor_stop);