lustre/ptlrpc/service.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) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2010, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  */
  31
  32 #define DEBUG_SUBSYSTEM S_RPC
  33
  34 #include <linux/fs_struct.h>
  35 #include <linux/kthread.h>
  36 #include <linux/ratelimit.h>
  37
  38 #include <obd_support.h>
  39 #include <obd_class.h>
  40 #include <lustre_net.h>
  41 #include <lu_object.h>
  42 #include <uapi/linux/lnet/lnet-types.h>
  43 #include "ptlrpc_internal.h"
  44 #include <linux/delay.h>
  45
  46 /* The following are visible and mutable through /sys/module/ptlrpc */
  47 int test_req_buffer_pressure = 0;
  48 module_param(test_req_buffer_pressure, int, 0444);
  49 MODULE_PARM_DESC(test_req_buffer_pressure, "set non-zero to put pressure on request buffer pools");
  50 module_param(at_min, int, 0644);
  51 MODULE_PARM_DESC(at_min, "Adaptive timeout minimum (sec)");
  52 module_param(at_max, int, 0644);
  53 MODULE_PARM_DESC(at_max, "Adaptive timeout maximum (sec)");
  54 module_param(at_history, int, 0644);
  55 MODULE_PARM_DESC(at_history,
  56                  "Adaptive timeouts remember the slowest event that took place within this period (sec)");
  57 module_param(at_early_margin, int, 0644);
  58 MODULE_PARM_DESC(at_early_margin, "How soon before an RPC deadline to send an early reply");
  59 module_param(at_extra, int, 0644);
  60 MODULE_PARM_DESC(at_extra, "How much extra time to give with each early reply");
  61
  62 /* forward ref */
  63 static int ptlrpc_server_post_idle_rqbds(struct ptlrpc_service_part *svcpt);
  64 static void ptlrpc_server_hpreq_fini(struct ptlrpc_request *req);
  65 static void ptlrpc_at_remove_timed(struct ptlrpc_request *req);
  66 static int ptlrpc_start_threads(struct ptlrpc_service *svc);
  67 static int ptlrpc_start_thread(struct ptlrpc_service_part *svcpt, int wait);
  68
  69 /** Holds a list of all PTLRPC services */
  70 LIST_HEAD(ptlrpc_all_services);
  71 /** Used to protect the \e ptlrpc_all_services list */
  72 struct mutex ptlrpc_all_services_mutex;
  73
  74 static struct ptlrpc_request_buffer_desc *
  75 ptlrpc_alloc_rqbd(struct ptlrpc_service_part *svcpt)
  76 {
  77         struct ptlrpc_service             *svc = svcpt->scp_service;
  78         struct ptlrpc_request_buffer_desc *rqbd;
  79
  80         OBD_CPT_ALLOC_PTR(rqbd, svc->srv_cptable, svcpt->scp_cpt);
  81         if (rqbd == NULL)
  82                 return NULL;
  83
  84         rqbd->rqbd_svcpt = svcpt;
  85         rqbd->rqbd_refcount = 0;
  86         rqbd->rqbd_cbid.cbid_fn = request_in_callback;
  87         rqbd->rqbd_cbid.cbid_arg = rqbd;
  88         INIT_LIST_HEAD(&rqbd->rqbd_reqs);
  89         OBD_CPT_ALLOC_LARGE(rqbd->rqbd_buffer, svc->srv_cptable,
  90                             svcpt->scp_cpt, svc->srv_buf_size);
  91         if (rqbd->rqbd_buffer == NULL) {
  92                 OBD_FREE_PTR(rqbd);
  93                 return NULL;
  94         }
  95
  96         spin_lock(&svcpt->scp_lock);
  97         list_add(&rqbd->rqbd_list, &svcpt->scp_rqbd_idle);
  98         svcpt->scp_nrqbds_total++;
  99         spin_unlock(&svcpt->scp_lock);
 100
 101         return rqbd;
 102 }
 103
 104 static void ptlrpc_free_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
 105 {
 106         struct ptlrpc_service_part *svcpt = rqbd->rqbd_svcpt;
 107
 108         LASSERT(rqbd->rqbd_refcount == 0);
 109         LASSERT(list_empty(&rqbd->rqbd_reqs));
 110
 111         spin_lock(&svcpt->scp_lock);
 112         list_del(&rqbd->rqbd_list);
 113         svcpt->scp_nrqbds_total--;
 114         spin_unlock(&svcpt->scp_lock);
 115
 116         OBD_FREE_LARGE(rqbd->rqbd_buffer, svcpt->scp_service->srv_buf_size);
 117         OBD_FREE_PTR(rqbd);
 118 }
 119
 120 static int ptlrpc_grow_req_bufs(struct ptlrpc_service_part *svcpt, int post)
 121 {
 122         struct ptlrpc_service *svc = svcpt->scp_service;
 123         struct ptlrpc_request_buffer_desc *rqbd;
 124         int rc = 0;
 125         int i;
 126
 127         if (svcpt->scp_rqbd_allocating)
 128                 goto try_post;
 129
 130         spin_lock(&svcpt->scp_lock);
 131         /* check again with lock */
 132         if (svcpt->scp_rqbd_allocating) {
 133                 /* NB: we might allow more than one thread in the future */
 134                 LASSERT(svcpt->scp_rqbd_allocating == 1);
 135                 spin_unlock(&svcpt->scp_lock);
 136                 goto try_post;
 137         }
 138
 139         svcpt->scp_rqbd_allocating++;
 140         spin_unlock(&svcpt->scp_lock);
 141
 142
 143         for (i = 0; i < svc->srv_nbuf_per_group; i++) {
 144                 /*
 145                  * NB: another thread might have recycled enough rqbds, we
 146                  * need to make sure it wouldn't over-allocate, see LU-1212.
 147                  */
 148                 if (svcpt->scp_nrqbds_posted >= svc->srv_nbuf_per_group ||
 149                     (svc->srv_nrqbds_max != 0 &&
 150                      svcpt->scp_nrqbds_total > svc->srv_nrqbds_max))
 151                         break;
 152
 153                 rqbd = ptlrpc_alloc_rqbd(svcpt);
 154
 155                 if (rqbd == NULL) {
 156                         CERROR("%s: Can't allocate request buffer\n",
 157                                svc->srv_name);
 158                         rc = -ENOMEM;
 159                         break;
 160                 }
 161         }
 162
 163         spin_lock(&svcpt->scp_lock);
 164
 165         LASSERT(svcpt->scp_rqbd_allocating == 1);
 166         svcpt->scp_rqbd_allocating--;
 167
 168         spin_unlock(&svcpt->scp_lock);
 169
 170         CDEBUG(D_RPCTRACE,
 171                "%s: allocate %d new %d-byte reqbufs (%d/%d left), rc = %d\n",
 172                svc->srv_name, i, svc->srv_buf_size, svcpt->scp_nrqbds_posted,
 173                svcpt->scp_nrqbds_total, rc);
 174
 175  try_post:
 176         if (post && rc == 0)
 177                 rc = ptlrpc_server_post_idle_rqbds(svcpt);
 178
 179         return rc;
 180 }
 181
 182 /**
 183  * Part of Rep-Ack logic.
 184  * Puts a lock and its mode into reply state assotiated to request reply.
 185  */
 186 void ptlrpc_save_lock(struct ptlrpc_request *req, struct lustre_handle *lock,
 187                       int mode, bool no_ack, bool convert_lock)
 188 {
 189         struct ptlrpc_reply_state *rs = req->rq_reply_state;
 190         int idx;
 191
 192         LASSERT(rs != NULL);
 193         LASSERT(rs->rs_nlocks < RS_MAX_LOCKS);
 194
 195         idx = rs->rs_nlocks++;
 196         rs->rs_locks[idx] = *lock;
 197         rs->rs_modes[idx] = mode;
 198         rs->rs_difficult = 1;
 199         rs->rs_no_ack = no_ack;
 200         rs->rs_convert_lock = convert_lock;
 201 }
 202 EXPORT_SYMBOL(ptlrpc_save_lock);
 203
 204
 205 struct ptlrpc_hr_partition;
 206
 207 struct ptlrpc_hr_thread {
 208         int                             hrt_id;         /* thread ID */
 209         spinlock_t                      hrt_lock;
 210         wait_queue_head_t               hrt_waitq;
 211         struct list_head                hrt_queue;
 212         struct ptlrpc_hr_partition      *hrt_partition;
 213 };
 214
 215 struct ptlrpc_hr_partition {
 216         /* # of started threads */
 217         atomic_t                        hrp_nstarted;
 218         /* # of stopped threads */
 219         atomic_t                        hrp_nstopped;
 220         /* cpu partition id */
 221         int                             hrp_cpt;
 222         /* round-robin rotor for choosing thread */
 223         int                             hrp_rotor;
 224         /* total number of threads on this partition */
 225         int                             hrp_nthrs;
 226         /* threads table */
 227         struct ptlrpc_hr_thread         *hrp_thrs;
 228 };
 229
 230 #define HRT_RUNNING 0
 231 #define HRT_STOPPING 1
 232
 233 struct ptlrpc_hr_service {
 234         /* CPU partition table, it's just cfs_cpt_tab for now */
 235         struct cfs_cpt_table            *hr_cpt_table;
 236         /** controller sleep waitq */
 237         wait_queue_head_t               hr_waitq;
 238         unsigned int                    hr_stopping;
 239         /** roundrobin rotor for non-affinity service */
 240         unsigned int                    hr_rotor;
 241         /* partition data */
 242         struct ptlrpc_hr_partition      **hr_partitions;
 243 };
 244
 245 struct rs_batch {
 246         struct list_head                        rsb_replies;
 247         unsigned int                    rsb_n_replies;
 248         struct ptlrpc_service_part      *rsb_svcpt;
 249 };
 250
 251 /** reply handling service. */
 252 static struct ptlrpc_hr_service         ptlrpc_hr;
 253
 254 /**
 255  * maximum mumber of replies scheduled in one batch
 256  */
 257 #define MAX_SCHEDULED 256
 258
 259 /**
 260  * Initialize a reply batch.
 261  *
 262  * \param b batch
 263  */
 264 static void rs_batch_init(struct rs_batch *b)
 265 {
 266         memset(b, 0, sizeof(*b));
 267         INIT_LIST_HEAD(&b->rsb_replies);
 268 }
 269
 270 /**
 271  * Choose an hr thread to dispatch requests to.
 272  */
 273 static
 274 struct ptlrpc_hr_thread *ptlrpc_hr_select(struct ptlrpc_service_part *svcpt)
 275 {
 276         struct ptlrpc_hr_partition      *hrp;
 277         unsigned int                    rotor;
 278
 279         if (svcpt->scp_cpt >= 0 &&
 280             svcpt->scp_service->srv_cptable == ptlrpc_hr.hr_cpt_table) {
 281                 /* directly match partition */
 282                 hrp = ptlrpc_hr.hr_partitions[svcpt->scp_cpt];
 283
 284         } else {
 285                 rotor = ptlrpc_hr.hr_rotor++;
 286                 rotor %= cfs_cpt_number(ptlrpc_hr.hr_cpt_table);
 287
 288                 hrp = ptlrpc_hr.hr_partitions[rotor];
 289         }
 290
 291         rotor = hrp->hrp_rotor++;
 292         return &hrp->hrp_thrs[rotor % hrp->hrp_nthrs];
 293 }
 294
 295 /**
 296  * Dispatch all replies accumulated in the batch to one from
 297  * dedicated reply handling threads.
 298  *
 299  * \param b batch
 300  */
 301 static void rs_batch_dispatch(struct rs_batch *b)
 302 {
 303         if (b->rsb_n_replies != 0) {
 304                 struct ptlrpc_hr_thread *hrt;
 305
 306                 hrt = ptlrpc_hr_select(b->rsb_svcpt);
 307
 308                 spin_lock(&hrt->hrt_lock);
 309                 list_splice_init(&b->rsb_replies, &hrt->hrt_queue);
 310                 spin_unlock(&hrt->hrt_lock);
 311
 312                 wake_up(&hrt->hrt_waitq);
 313                 b->rsb_n_replies = 0;
 314         }
 315 }
 316
 317 /**
 318  * Add a reply to a batch.
 319  * Add one reply object to a batch, schedule batched replies if overload.
 320  *
 321  * \param b batch
 322  * \param rs reply
 323  */
 324 static void rs_batch_add(struct rs_batch *b, struct ptlrpc_reply_state *rs)
 325 {
 326         struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
 327
 328         if (svcpt != b->rsb_svcpt || b->rsb_n_replies >= MAX_SCHEDULED) {
 329                 if (b->rsb_svcpt != NULL) {
 330                         rs_batch_dispatch(b);
 331                         spin_unlock(&b->rsb_svcpt->scp_rep_lock);
 332                 }
 333                 spin_lock(&svcpt->scp_rep_lock);
 334                 b->rsb_svcpt = svcpt;
 335         }
 336         spin_lock(&rs->rs_lock);
 337         rs->rs_scheduled_ever = 1;
 338         if (rs->rs_scheduled == 0) {
 339                 list_move(&rs->rs_list, &b->rsb_replies);
 340                 rs->rs_scheduled = 1;
 341                 b->rsb_n_replies++;
 342         }
 343         rs->rs_committed = 1;
 344         spin_unlock(&rs->rs_lock);
 345 }
 346
 347 /**
 348  * Reply batch finalization.
 349  * Dispatch remaining replies from the batch
 350  * and release remaining spinlock.
 351  *
 352  * \param b batch
 353  */
 354 static void rs_batch_fini(struct rs_batch *b)
 355 {
 356         if (b->rsb_svcpt != NULL) {
 357                 rs_batch_dispatch(b);
 358                 spin_unlock(&b->rsb_svcpt->scp_rep_lock);
 359         }
 360 }
 361
 362 #define DECLARE_RS_BATCH(b)     struct rs_batch b
 363
 364
 365 /**
 366  * Put reply state into a queue for processing because we received
 367  * ACK from the client
 368  */
 369 void ptlrpc_dispatch_difficult_reply(struct ptlrpc_reply_state *rs)
 370 {
 371         struct ptlrpc_hr_thread *hrt;
 372
 373         ENTRY;
 374
 375         LASSERT(list_empty(&rs->rs_list));
 376
 377         hrt = ptlrpc_hr_select(rs->rs_svcpt);
 378
 379         spin_lock(&hrt->hrt_lock);
 380         list_add_tail(&rs->rs_list, &hrt->hrt_queue);
 381         spin_unlock(&hrt->hrt_lock);
 382
 383         wake_up(&hrt->hrt_waitq);
 384         EXIT;
 385 }
 386
 387 void ptlrpc_schedule_difficult_reply(struct ptlrpc_reply_state *rs)
 388 {
 389         ENTRY;
 390
 391         assert_spin_locked(&rs->rs_svcpt->scp_rep_lock);
 392         assert_spin_locked(&rs->rs_lock);
 393         LASSERT(rs->rs_difficult);
 394         rs->rs_scheduled_ever = 1;  /* flag any notification attempt */
 395
 396         if (rs->rs_scheduled) {     /* being set up or already notified */
 397                 EXIT;
 398                 return;
 399         }
 400
 401         rs->rs_scheduled = 1;
 402         list_del_init(&rs->rs_list);
 403         ptlrpc_dispatch_difficult_reply(rs);
 404         EXIT;
 405 }
 406 EXPORT_SYMBOL(ptlrpc_schedule_difficult_reply);
 407
 408 void ptlrpc_commit_replies(struct obd_export *exp)
 409 {
 410         struct ptlrpc_reply_state *rs, *nxt;
 411         DECLARE_RS_BATCH(batch);
 412
 413         ENTRY;
 414
 415         rs_batch_init(&batch);
 416         /*
 417          * Find any replies that have been committed and get their service
 418          * to attend to complete them.
 419          */
 420
 421         /* CAVEAT EMPTOR: spinlock ordering!!! */
 422         spin_lock(&exp->exp_uncommitted_replies_lock);
 423         list_for_each_entry_safe(rs, nxt, &exp->exp_uncommitted_replies,
 424                                  rs_obd_list) {
 425                 LASSERT(rs->rs_difficult);
 426                 /* VBR: per-export last_committed */
 427                 LASSERT(rs->rs_export);
 428                 if (rs->rs_transno <= exp->exp_last_committed) {
 429                         list_del_init(&rs->rs_obd_list);
 430                         rs_batch_add(&batch, rs);
 431                 }
 432         }
 433         spin_unlock(&exp->exp_uncommitted_replies_lock);
 434         rs_batch_fini(&batch);
 435         EXIT;
 436 }
 437
 438 static int ptlrpc_server_post_idle_rqbds(struct ptlrpc_service_part *svcpt)
 439 {
 440         struct ptlrpc_request_buffer_desc *rqbd;
 441         int rc;
 442         int posted = 0;
 443
 444         for (;;) {
 445                 spin_lock(&svcpt->scp_lock);
 446
 447                 if (list_empty(&svcpt->scp_rqbd_idle)) {
 448                         spin_unlock(&svcpt->scp_lock);
 449                         return posted;
 450                 }
 451
 452                 rqbd = list_first_entry(&svcpt->scp_rqbd_idle,
 453                                         struct ptlrpc_request_buffer_desc,
 454                                         rqbd_list);
 455
 456                 /* assume we will post successfully */
 457                 svcpt->scp_nrqbds_posted++;
 458                 list_move(&rqbd->rqbd_list, &svcpt->scp_rqbd_posted);
 459
 460                 spin_unlock(&svcpt->scp_lock);
 461
 462                 rc = ptlrpc_register_rqbd(rqbd);
 463                 if (rc != 0)
 464                         break;
 465
 466                 posted = 1;
 467         }
 468
 469         spin_lock(&svcpt->scp_lock);
 470
 471         svcpt->scp_nrqbds_posted--;
 472         list_move_tail(&rqbd->rqbd_list, &svcpt->scp_rqbd_idle);
 473
 474         /*
 475          * Don't complain if no request buffers are posted right now; LNET
 476          * won't drop requests because we set the portal lazy!
 477          */
 478
 479         spin_unlock(&svcpt->scp_lock);
 480
 481         return -1;
 482 }
 483
 484 static void ptlrpc_at_timer(cfs_timer_cb_arg_t data)
 485 {
 486         struct ptlrpc_service_part *svcpt;
 487
 488         svcpt = cfs_from_timer(svcpt, data, scp_at_timer);
 489
 490         svcpt->scp_at_check = 1;
 491         svcpt->scp_at_checktime = ktime_get();
 492         wake_up(&svcpt->scp_waitq);
 493 }
 494
 495 static void ptlrpc_server_nthreads_check(struct ptlrpc_service *svc,
 496                                          struct ptlrpc_service_conf *conf)
 497 {
 498         struct ptlrpc_service_thr_conf *tc = &conf->psc_thr;
 499         unsigned int init;
 500         unsigned int total;
 501         unsigned int nthrs;
 502         int weight;
 503
 504         /*
 505          * Common code for estimating & validating threads number.
 506          * CPT affinity service could have percpt thread-pool instead
 507          * of a global thread-pool, which means user might not always
 508          * get the threads number they give it in conf::tc_nthrs_user
 509          * even they did set. It's because we need to validate threads
 510          * number for each CPT to guarantee each pool will have enough
 511          * threads to keep the service healthy.
 512          */
 513         init = PTLRPC_NTHRS_INIT + (svc->srv_ops.so_hpreq_handler != NULL);
 514         init = max_t(int, init, tc->tc_nthrs_init);
 515
 516         /*
 517          * NB: please see comments in lustre_lnet.h for definition
 518          * details of these members
 519          */
 520         LASSERT(tc->tc_nthrs_max != 0);
 521
 522         if (tc->tc_nthrs_user != 0) {
 523                 /*
 524                  * In case there is a reason to test a service with many
 525                  * threads, we give a less strict check here, it can
 526                  * be up to 8 * nthrs_max
 527                  */
 528                 total = min(tc->tc_nthrs_max * 8, tc->tc_nthrs_user);
 529                 nthrs = total / svc->srv_ncpts;
 530                 init  = max(init, nthrs);
 531                 goto out;
 532         }
 533
 534         total = tc->tc_nthrs_max;
 535         if (tc->tc_nthrs_base == 0) {
 536                 /*
 537                  * don't care about base threads number per partition,
 538                  * this is most for non-affinity service
 539                  */
 540                 nthrs = total / svc->srv_ncpts;
 541                 goto out;
 542         }
 543
 544         nthrs = tc->tc_nthrs_base;
 545         if (svc->srv_ncpts == 1) {
 546                 int     i;
 547
 548                 /*
 549                  * NB: Increase the base number if it's single partition
 550                  * and total number of cores/HTs is larger or equal to 4.
 551                  * result will always < 2 * nthrs_base
 552                  */
 553                 weight = cfs_cpt_weight(svc->srv_cptable, CFS_CPT_ANY);
 554                 for (i = 1; (weight >> (i + 1)) != 0 && /* >= 4 cores/HTs */
 555                             (tc->tc_nthrs_base >> i) != 0; i++)
 556                         nthrs += tc->tc_nthrs_base >> i;
 557         }
 558
 559         if (tc->tc_thr_factor != 0) {
 560                 int       factor = tc->tc_thr_factor;
 561                 const int fade = 4;
 562
 563                 /*
 564                  * User wants to increase number of threads with for
 565                  * each CPU core/HT, most likely the factor is larger than
 566                  * one thread/core because service threads are supposed to
 567                  * be blocked by lock or wait for IO.
 568                  */
 569                 /*
 570                  * Amdahl's law says that adding processors wouldn't give
 571                  * a linear increasing of parallelism, so it's nonsense to
 572                  * have too many threads no matter how many cores/HTs
 573                  * there are.
 574                  */
 575                 preempt_disable();
 576                 if (cpumask_weight
 577                     (topology_sibling_cpumask(smp_processor_id())) > 1) {
 578                         /* weight is # of HTs */
 579                         /* depress thread factor for hyper-thread */
 580                         factor = factor - (factor >> 1) + (factor >> 3);
 581                 }
 582                 preempt_enable();
 583
 584                 weight = cfs_cpt_weight(svc->srv_cptable, 0);
 585
 586                 for (; factor > 0 && weight > 0; factor--, weight -= fade)
 587                         nthrs += min(weight, fade) * factor;
 588         }
 589
 590         if (nthrs * svc->srv_ncpts > tc->tc_nthrs_max) {
 591                 nthrs = max(tc->tc_nthrs_base,
 592                             tc->tc_nthrs_max / svc->srv_ncpts);
 593         }
 594  out:
 595         nthrs = max(nthrs, tc->tc_nthrs_init);
 596         svc->srv_nthrs_cpt_limit = nthrs;
 597         svc->srv_nthrs_cpt_init = init;
 598
 599         if (nthrs * svc->srv_ncpts > tc->tc_nthrs_max) {
 600                 CDEBUG(D_OTHER,
 601                        "%s: This service may have more threads (%d) than the given soft limit (%d)\n",
 602                        svc->srv_name, nthrs * svc->srv_ncpts,
 603                        tc->tc_nthrs_max);
 604         }
 605 }
 606
 607 /**
 608  * Initialize percpt data for a service
 609  */
 610 static int ptlrpc_service_part_init(struct ptlrpc_service *svc,
 611                                     struct ptlrpc_service_part *svcpt, int cpt)
 612 {
 613         struct ptlrpc_at_array *array;
 614         int size;
 615         int index;
 616         int rc;
 617
 618         svcpt->scp_cpt = cpt;
 619         INIT_LIST_HEAD(&svcpt->scp_threads);
 620
 621         /* rqbd and incoming request queue */
 622         spin_lock_init(&svcpt->scp_lock);
 623         mutex_init(&svcpt->scp_mutex);
 624         INIT_LIST_HEAD(&svcpt->scp_rqbd_idle);
 625         INIT_LIST_HEAD(&svcpt->scp_rqbd_posted);
 626         INIT_LIST_HEAD(&svcpt->scp_req_incoming);
 627         init_waitqueue_head(&svcpt->scp_waitq);
 628         /* history request & rqbd list */
 629         INIT_LIST_HEAD(&svcpt->scp_hist_reqs);
 630         INIT_LIST_HEAD(&svcpt->scp_hist_rqbds);
 631
 632         /* acitve requests and hp requests */
 633         spin_lock_init(&svcpt->scp_req_lock);
 634
 635         /* reply states */
 636         spin_lock_init(&svcpt->scp_rep_lock);
 637         INIT_LIST_HEAD(&svcpt->scp_rep_active);
 638         INIT_LIST_HEAD(&svcpt->scp_rep_idle);
 639         init_waitqueue_head(&svcpt->scp_rep_waitq);
 640         atomic_set(&svcpt->scp_nreps_difficult, 0);
 641
 642         /* adaptive timeout */
 643         spin_lock_init(&svcpt->scp_at_lock);
 644         array = &svcpt->scp_at_array;
 645
 646         size = at_est2timeout(at_max);
 647         array->paa_size     = size;
 648         array->paa_count    = 0;
 649         array->paa_deadline = -1;
 650
 651         /* allocate memory for scp_at_array (ptlrpc_at_array) */
 652         OBD_CPT_ALLOC(array->paa_reqs_array,
 653                       svc->srv_cptable, cpt, sizeof(struct list_head) * size);
 654         if (array->paa_reqs_array == NULL)
 655                 return -ENOMEM;
 656
 657         for (index = 0; index < size; index++)
 658                 INIT_LIST_HEAD(&array->paa_reqs_array[index]);
 659
 660         OBD_CPT_ALLOC(array->paa_reqs_count,
 661                       svc->srv_cptable, cpt, sizeof(__u32) * size);
 662         if (array->paa_reqs_count == NULL)
 663                 goto failed;
 664
 665         cfs_timer_setup(&svcpt->scp_at_timer, ptlrpc_at_timer,
 666                         (unsigned long)svcpt, 0);
 667
 668         /*
 669          * At SOW, service time should be quick; 10s seems generous. If client
 670          * timeout is less than this, we'll be sending an early reply.
 671          */
 672         at_init(&svcpt->scp_at_estimate, 10, 0);
 673
 674         /* assign this before call ptlrpc_grow_req_bufs */
 675         svcpt->scp_service = svc;
 676         /* Now allocate the request buffers, but don't post them now */
 677         rc = ptlrpc_grow_req_bufs(svcpt, 0);
 678         /*
 679          * We shouldn't be under memory pressure at startup, so
 680          * fail if we can't allocate all our buffers at this time.
 681          */
 682         if (rc != 0)
 683                 goto failed;
 684
 685         return 0;
 686
 687  failed:
 688         if (array->paa_reqs_count != NULL) {
 689                 OBD_FREE_PTR_ARRAY(array->paa_reqs_count, size);
 690                 array->paa_reqs_count = NULL;
 691         }
 692
 693         if (array->paa_reqs_array != NULL) {
 694                 OBD_FREE_PTR_ARRAY(array->paa_reqs_array, array->paa_size);
 695                 array->paa_reqs_array = NULL;
 696         }
 697
 698         return -ENOMEM;
 699 }
 700
 701 /**
 702  * Initialize service on a given portal.
 703  * This includes starting serving threads , allocating and posting rqbds and
 704  * so on.
 705  */
 706 struct ptlrpc_service *ptlrpc_register_service(struct ptlrpc_service_conf *conf,
 707                                                struct kset *parent,
 708                                                struct dentry *debugfs_entry)
 709 {
 710         struct ptlrpc_service_cpt_conf *cconf = &conf->psc_cpt;
 711         struct ptlrpc_service *service;
 712         struct ptlrpc_service_part *svcpt;
 713         struct cfs_cpt_table *cptable;
 714         __u32 *cpts = NULL;
 715         int ncpts;
 716         int cpt;
 717         int rc;
 718         int i;
 719
 720         ENTRY;
 721
 722         LASSERT(conf->psc_buf.bc_nbufs > 0);
 723         LASSERT(conf->psc_buf.bc_buf_size >=
 724                 conf->psc_buf.bc_req_max_size + SPTLRPC_MAX_PAYLOAD);
 725         LASSERT(conf->psc_thr.tc_ctx_tags != 0);
 726
 727         cptable = cconf->cc_cptable;
 728         if (cptable == NULL)
 729                 cptable = cfs_cpt_tab;
 730
 731         if (conf->psc_thr.tc_cpu_bind > 1) {
 732                 CERROR("%s: Invalid cpu bind value %d, only 1 or 0 allowed\n",
 733                        conf->psc_name, conf->psc_thr.tc_cpu_bind);
 734                 RETURN(ERR_PTR(-EINVAL));
 735         }
 736
 737         if (!cconf->cc_affinity) {
 738                 ncpts = 1;
 739         } else {
 740                 ncpts = cfs_cpt_number(cptable);
 741                 if (cconf->cc_pattern != NULL) {
 742                         struct cfs_expr_list    *el;
 743
 744                         rc = cfs_expr_list_parse(cconf->cc_pattern,
 745                                                  strlen(cconf->cc_pattern),
 746                                                  0, ncpts - 1, &el);
 747                         if (rc != 0) {
 748                                 CERROR("%s: invalid CPT pattern string: %s\n",
 749                                        conf->psc_name, cconf->cc_pattern);
 750                                 RETURN(ERR_PTR(-EINVAL));
 751                         }
 752
 753                         rc = cfs_expr_list_values(el, ncpts, &cpts);
 754                         cfs_expr_list_free(el);
 755                         if (rc <= 0) {
 756                                 CERROR("%s: failed to parse CPT array %s: %d\n",
 757                                        conf->psc_name, cconf->cc_pattern, rc);
 758                                 if (cpts != NULL)
 759                                         OBD_FREE_PTR_ARRAY(cpts, ncpts);
 760                                 RETURN(ERR_PTR(rc < 0 ? rc : -EINVAL));
 761                         }
 762                         ncpts = rc;
 763                 }
 764         }
 765
 766         OBD_ALLOC(service, offsetof(struct ptlrpc_service, srv_parts[ncpts]));
 767         if (service == NULL) {
 768                 if (cpts != NULL)
 769                         OBD_FREE_PTR_ARRAY(cpts, ncpts);
 770                 RETURN(ERR_PTR(-ENOMEM));
 771         }
 772
 773         service->srv_cptable            = cptable;
 774         service->srv_cpts               = cpts;
 775         service->srv_ncpts              = ncpts;
 776         service->srv_cpt_bind           = conf->psc_thr.tc_cpu_bind;
 777
 778         service->srv_cpt_bits = 0; /* it's zero already, easy to read... */
 779         while ((1 << service->srv_cpt_bits) < cfs_cpt_number(cptable))
 780                 service->srv_cpt_bits++;
 781
 782         /* public members */
 783         spin_lock_init(&service->srv_lock);
 784         service->srv_name               = conf->psc_name;
 785         service->srv_watchdog_factor    = conf->psc_watchdog_factor;
 786         INIT_LIST_HEAD(&service->srv_list); /* for safty of cleanup */
 787
 788         /* buffer configuration */
 789         service->srv_nbuf_per_group     = test_req_buffer_pressure ?
 790                                           1 : conf->psc_buf.bc_nbufs;
 791         /* do not limit max number of rqbds by default */
 792         service->srv_nrqbds_max         = 0;
 793
 794         service->srv_max_req_size       = conf->psc_buf.bc_req_max_size +
 795                                           SPTLRPC_MAX_PAYLOAD;
 796         service->srv_buf_size           = conf->psc_buf.bc_buf_size;
 797         service->srv_rep_portal         = conf->psc_buf.bc_rep_portal;
 798         service->srv_req_portal         = conf->psc_buf.bc_req_portal;
 799
 800         /* With slab/alloc_pages buffer size will be rounded up to 2^n */
 801         if (service->srv_buf_size & (service->srv_buf_size - 1)) {
 802                 int round = size_roundup_power2(service->srv_buf_size);
 803
 804                 service->srv_buf_size = round;
 805         }
 806
 807         /* Increase max reply size to next power of two */
 808         service->srv_max_reply_size = 1;
 809         while (service->srv_max_reply_size <
 810                conf->psc_buf.bc_rep_max_size + SPTLRPC_MAX_PAYLOAD)
 811                 service->srv_max_reply_size <<= 1;
 812
 813         service->srv_thread_name        = conf->psc_thr.tc_thr_name;
 814         service->srv_ctx_tags           = conf->psc_thr.tc_ctx_tags;
 815         service->srv_hpreq_ratio        = PTLRPC_SVC_HP_RATIO;
 816         service->srv_ops                = conf->psc_ops;
 817
 818         for (i = 0; i < ncpts; i++) {
 819                 if (!cconf->cc_affinity)
 820                         cpt = CFS_CPT_ANY;
 821                 else
 822                         cpt = cpts != NULL ? cpts[i] : i;
 823
 824                 OBD_CPT_ALLOC(svcpt, cptable, cpt, sizeof(*svcpt));
 825                 if (svcpt == NULL)
 826                         GOTO(failed, rc = -ENOMEM);
 827
 828                 service->srv_parts[i] = svcpt;
 829                 rc = ptlrpc_service_part_init(service, svcpt, cpt);
 830                 if (rc != 0)
 831                         GOTO(failed, rc);
 832         }
 833
 834         ptlrpc_server_nthreads_check(service, conf);
 835
 836         rc = LNetSetLazyPortal(service->srv_req_portal);
 837         LASSERT(rc == 0);
 838
 839         mutex_lock(&ptlrpc_all_services_mutex);
 840         list_add(&service->srv_list, &ptlrpc_all_services);
 841         mutex_unlock(&ptlrpc_all_services_mutex);
 842
 843         if (parent) {
 844                 rc = ptlrpc_sysfs_register_service(parent, service);
 845                 if (rc)
 846                         GOTO(failed, rc);
 847         }
 848
 849         if (debugfs_entry != NULL)
 850                 ptlrpc_ldebugfs_register_service(debugfs_entry, service);
 851
 852         rc = ptlrpc_service_nrs_setup(service);
 853         if (rc != 0)
 854                 GOTO(failed, rc);
 855
 856         CDEBUG(D_NET, "%s: Started, listening on portal %d\n",
 857                service->srv_name, service->srv_req_portal);
 858
 859         rc = ptlrpc_start_threads(service);
 860         if (rc != 0) {
 861                 CERROR("Failed to start threads for service %s: %d\n",
 862                        service->srv_name, rc);
 863                 GOTO(failed, rc);
 864         }
 865
 866         RETURN(service);
 867 failed:
 868         ptlrpc_unregister_service(service);
 869         RETURN(ERR_PTR(rc));
 870 }
 871 EXPORT_SYMBOL(ptlrpc_register_service);
 872
 873 /**
 874  * to actually free the request, must be called without holding svc_lock.
 875  * note it's caller's responsibility to unlink req->rq_list.
 876  */
 877 static void ptlrpc_server_free_request(struct ptlrpc_request *req)
 878 {
 879         LASSERT(atomic_read(&req->rq_refcount) == 0);
 880         LASSERT(list_empty(&req->rq_timed_list));
 881
 882         /*
 883          * DEBUG_REQ() assumes the reply state of a request with a valid
 884          * ref will not be destroyed until that reference is dropped.
 885          */
 886         ptlrpc_req_drop_rs(req);
 887
 888         sptlrpc_svc_ctx_decref(req);
 889
 890         if (req != &req->rq_rqbd->rqbd_req) {
 891                 /*
 892                  * NB request buffers use an embedded
 893                  * req if the incoming req unlinked the
 894                  * MD; this isn't one of them!
 895                  */
 896                 ptlrpc_request_cache_free(req);
 897         }
 898 }
 899
 900 /**
 901  * drop a reference count of the request. if it reaches 0, we either
 902  * put it into history list, or free it immediately.
 903  */
 904 void ptlrpc_server_drop_request(struct ptlrpc_request *req)
 905 {
 906         struct ptlrpc_request_buffer_desc *rqbd = req->rq_rqbd;
 907         struct ptlrpc_service_part        *svcpt = rqbd->rqbd_svcpt;
 908         struct ptlrpc_service             *svc = svcpt->scp_service;
 909         int                                refcount;
 910
 911         if (!atomic_dec_and_test(&req->rq_refcount))
 912                 return;
 913
 914         if (req->rq_session.lc_state == LCS_ENTERED) {
 915                 lu_context_exit(&req->rq_session);
 916                 lu_context_fini(&req->rq_session);
 917         }
 918
 919         if (req->rq_at_linked) {
 920                 spin_lock(&svcpt->scp_at_lock);
 921                 /*
 922                  * recheck with lock, in case it's unlinked by
 923                  * ptlrpc_at_check_timed()
 924                  */
 925                 if (likely(req->rq_at_linked))
 926                         ptlrpc_at_remove_timed(req);
 927                 spin_unlock(&svcpt->scp_at_lock);
 928         }
 929
 930         LASSERT(list_empty(&req->rq_timed_list));
 931
 932         /* finalize request */
 933         if (req->rq_export) {
 934                 class_export_put(req->rq_export);
 935                 req->rq_export = NULL;
 936         }
 937
 938         spin_lock(&svcpt->scp_lock);
 939
 940         list_add(&req->rq_list, &rqbd->rqbd_reqs);
 941
 942         refcount = --(rqbd->rqbd_refcount);
 943         if (refcount == 0) {
 944                 /* request buffer is now idle: add to history */
 945                 list_move_tail(&rqbd->rqbd_list, &svcpt->scp_hist_rqbds);
 946                 svcpt->scp_hist_nrqbds++;
 947
 948                 /*
 949                  * cull some history?
 950                  * I expect only about 1 or 2 rqbds need to be recycled here
 951                  */
 952                 while (svcpt->scp_hist_nrqbds > svc->srv_hist_nrqbds_cpt_max) {
 953                         rqbd = list_first_entry(&svcpt->scp_hist_rqbds,
 954                                                 struct ptlrpc_request_buffer_desc,
 955                                                 rqbd_list);
 956
 957                         list_del(&rqbd->rqbd_list);
 958                         svcpt->scp_hist_nrqbds--;
 959
 960                         /*
 961                          * remove rqbd's reqs from svc's req history while
 962                          * I've got the service lock
 963                          */
 964                         list_for_each_entry(req, &rqbd->rqbd_reqs, rq_list) {
 965                                 /* Track the highest culled req seq */
 966                                 if (req->rq_history_seq >
 967                                     svcpt->scp_hist_seq_culled) {
 968                                         svcpt->scp_hist_seq_culled =
 969                                                 req->rq_history_seq;
 970                                 }
 971                                 list_del(&req->rq_history_list);
 972                         }
 973
 974                         spin_unlock(&svcpt->scp_lock);
 975
 976                         while ((req = list_first_entry_or_null(
 977                                         &rqbd->rqbd_reqs,
 978                                         struct ptlrpc_request, rq_list))) {
 979                                 list_del(&req->rq_list);
 980                                 ptlrpc_server_free_request(req);
 981                         }
 982
 983                         spin_lock(&svcpt->scp_lock);
 984                         /*
 985                          * now all reqs including the embedded req has been
 986                          * disposed, schedule request buffer for re-use
 987                          * or free it to drain some in excess.
 988                          */
 989                         LASSERT(atomic_read(&rqbd->rqbd_req.rq_refcount) == 0);
 990                         if (svcpt->scp_nrqbds_posted >=
 991                             svc->srv_nbuf_per_group ||
 992                             (svc->srv_nrqbds_max != 0 &&
 993                              svcpt->scp_nrqbds_total > svc->srv_nrqbds_max) ||
 994                             test_req_buffer_pressure) {
 995                                 /* like in ptlrpc_free_rqbd() */
 996                                 svcpt->scp_nrqbds_total--;
 997                                 OBD_FREE_LARGE(rqbd->rqbd_buffer,
 998                                                svc->srv_buf_size);
 999                                 OBD_FREE_PTR(rqbd);
1000                         } else {
1001                                 list_add_tail(&rqbd->rqbd_list,
1002                                               &svcpt->scp_rqbd_idle);
1003                         }
1004                 }
1005
1006                 spin_unlock(&svcpt->scp_lock);
1007         } else if (req->rq_reply_state && req->rq_reply_state->rs_prealloc) {
1008                 /* If we are low on memory, we are not interested in history */
1009                 list_del(&req->rq_list);
1010                 list_del_init(&req->rq_history_list);
1011
1012                 /* Track the highest culled req seq */
1013                 if (req->rq_history_seq > svcpt->scp_hist_seq_culled)
1014                         svcpt->scp_hist_seq_culled = req->rq_history_seq;
1015
1016                 spin_unlock(&svcpt->scp_lock);
1017
1018                 ptlrpc_server_free_request(req);
1019         } else {
1020                 spin_unlock(&svcpt->scp_lock);
1021         }
1022 }
1023
1024 static void ptlrpc_add_exp_list_nolock(struct ptlrpc_request *req,
1025                                        struct obd_export *export, bool hp)
1026 {
1027         __u16 tag = lustre_msg_get_tag(req->rq_reqmsg);
1028
1029         if (hp)
1030                 list_add(&req->rq_exp_list, &export->exp_hp_rpcs);
1031         else
1032                 list_add(&req->rq_exp_list, &export->exp_reg_rpcs);
1033         if (tag && export->exp_used_slots)
1034                 set_bit(tag - 1, export->exp_used_slots);
1035 }
1036
1037 static void ptlrpc_del_exp_list(struct ptlrpc_request *req)
1038 {
1039         __u16 tag = lustre_msg_get_tag(req->rq_reqmsg);
1040
1041         spin_lock(&req->rq_export->exp_rpc_lock);
1042         list_del_init(&req->rq_exp_list);
1043         if (tag && !req->rq_obsolete && req->rq_export->exp_used_slots)
1044                 clear_bit(tag - 1, req->rq_export->exp_used_slots);
1045         spin_unlock(&req->rq_export->exp_rpc_lock);
1046 }
1047
1048 /** Change request export and move hp request from old export to new */
1049 void ptlrpc_request_change_export(struct ptlrpc_request *req,
1050                                   struct obd_export *export)
1051 {
1052         if (req->rq_export != NULL) {
1053                 LASSERT(!list_empty(&req->rq_exp_list));
1054                 /* remove rq_exp_list from last export */
1055                 ptlrpc_del_exp_list(req);
1056                 /* export has one reference already, so it's safe to
1057                  * add req to export queue here and get another
1058                  * reference for request later
1059                  */
1060                 spin_lock(&export->exp_rpc_lock);
1061                 ptlrpc_add_exp_list_nolock(req, export, req->rq_ops != NULL);
1062                 spin_unlock(&export->exp_rpc_lock);
1063
1064                 class_export_rpc_dec(req->rq_export);
1065                 class_export_put(req->rq_export);
1066         }
1067
1068         /* request takes one export refcount */
1069         req->rq_export = class_export_get(export);
1070         class_export_rpc_inc(export);
1071 }
1072
1073 /**
1074  * to finish a request: stop sending more early replies, and release
1075  * the request.
1076  */
1077 static void ptlrpc_server_finish_request(struct ptlrpc_service_part *svcpt,
1078                                          struct ptlrpc_request *req)
1079 {
1080         ptlrpc_server_hpreq_fini(req);
1081
1082         ptlrpc_server_drop_request(req);
1083 }
1084
1085 /**
1086  * to finish an active request: stop sending more early replies, and release
1087  * the request. should be called after we finished handling the request.
1088  */
1089 static void ptlrpc_server_finish_active_request(
1090                                         struct ptlrpc_service_part *svcpt,
1091                                         struct ptlrpc_request *req)
1092 {
1093         spin_lock(&svcpt->scp_req_lock);
1094         ptlrpc_nrs_req_stop_nolock(req);
1095         svcpt->scp_nreqs_active--;
1096         if (req->rq_hp)
1097                 svcpt->scp_nhreqs_active--;
1098         spin_unlock(&svcpt->scp_req_lock);
1099
1100         ptlrpc_nrs_req_finalize(req);
1101
1102         if (req->rq_export != NULL)
1103                 class_export_rpc_dec(req->rq_export);
1104
1105         ptlrpc_server_finish_request(svcpt, req);
1106 }
1107
1108 /**
1109  * This function makes sure dead exports are evicted in a timely manner.
1110  * This function is only called when some export receives a message (i.e.,
1111  * the network is up.)
1112  */
1113 void ptlrpc_update_export_timer(struct obd_export *exp, time64_t extra_delay)
1114 {
1115         struct obd_export *oldest_exp, *newest_exp;
1116         time64_t oldest_time, current_time;
1117         bool    evict = false;
1118         ENTRY;
1119
1120         LASSERT(exp);
1121
1122         /*
1123          * Compensate for slow machines, etc, by faking our request time
1124          * into the future.  Although this can break the strict time-ordering
1125          * of the list, we can be really lazy here - we don't have to evict
1126          * at the exact right moment.  Eventually, all silent exports
1127          * will make it to the top of the list.
1128          */
1129
1130         /* Do not pay attention on 1sec or smaller renewals. */
1131         current_time = ktime_get_real_seconds();
1132         /* 1 seconds */
1133         if (exp->exp_last_request_time + 1 >= current_time + extra_delay)
1134                 RETURN_EXIT;
1135
1136         exp->exp_last_request_time = current_time + extra_delay;
1137
1138         /*
1139          * exports may get disconnected from the chain even though the
1140          * export has references, so we must keep the spin lock while
1141          * manipulating the lists
1142          */
1143         spin_lock(&exp->exp_obd->obd_dev_lock);
1144
1145         if (list_empty(&exp->exp_obd_chain_timed)) {
1146                 /* this one is not timed */
1147                 spin_unlock(&exp->exp_obd->obd_dev_lock);
1148                 RETURN_EXIT;
1149         }
1150
1151         newest_exp = list_last_entry(&exp->exp_obd->obd_exports_timed,
1152                                      struct obd_export, exp_obd_chain_timed);
1153
1154         list_move_tail(&exp->exp_obd_chain_timed,
1155                        &exp->exp_obd->obd_exports_timed);
1156
1157         if (exp->exp_obd->obd_recovering) {
1158                 /* be nice to everyone during recovery */
1159                 spin_unlock(&exp->exp_obd->obd_dev_lock);
1160                 RETURN_EXIT;
1161         }
1162
1163         oldest_exp = list_entry(exp->exp_obd->obd_exports_timed.next,
1164                                 struct obd_export, exp_obd_chain_timed);
1165
1166         oldest_time = oldest_exp->exp_last_request_time;
1167
1168         /* Check if the oldest entry is expired. */
1169         if (exp->exp_obd->obd_eviction_timer == 0 &&
1170             current_time > oldest_time + PING_EVICT_TIMEOUT + extra_delay) {
1171
1172                 if (current_time < newest_exp->exp_last_request_time +
1173                              PING_EVICT_TIMEOUT / 2) {
1174                         /* If import is active - evict stale clients */
1175                         evict = true;
1176                 } else {
1177                         /*
1178                          * We need a second timer, in case the net was down and
1179                          * it just came back. Since the pinger may skip every
1180                          * other PING_INTERVAL (see note in ptlrpc_pinger_main),
1181                          * we better wait for 3.
1182                          */
1183                         exp->exp_obd->obd_eviction_timer =
1184                                 ktime_get_real_seconds() + 3 * PING_INTERVAL;
1185                         CDEBUG(D_HA, "%s: Think about evicting %s from %lld\n",
1186                                exp->exp_obd->obd_name,
1187                                obd_export_nid2str(oldest_exp), oldest_time);
1188
1189                 }
1190         }
1191
1192         spin_unlock(&exp->exp_obd->obd_dev_lock);
1193
1194         if (evict) {
1195                 /* Evict stale clients */
1196                 ping_evictor_wake(exp);
1197         } else {
1198                 if (ktime_get_real_seconds() >
1199                     (exp->exp_obd->obd_eviction_timer + extra_delay)) {
1200                         /*
1201                          * The evictor won't evict anyone who we've heard from
1202                          * recently, so we don't have to check before we start
1203                          * it.
1204                          */
1205                         if (!ping_evictor_wake(exp))
1206                                 exp->exp_obd->obd_eviction_timer = 0;
1207                 }
1208         }
1209
1210         EXIT;
1211 }
1212
1213 /**
1214  * Sanity check request \a req.
1215  * Return 0 if all is ok, error code otherwise.
1216  */
1217 static int ptlrpc_check_req(struct ptlrpc_request *req)
1218 {
1219         struct obd_device *obd = req->rq_export->exp_obd;
1220         int rc = 0;
1221
1222         if (unlikely(lustre_msg_get_conn_cnt(req->rq_reqmsg) <
1223                      req->rq_export->exp_conn_cnt)) {
1224                 DEBUG_REQ(D_RPCTRACE, req,
1225                           "DROPPING req from old connection %d < %d",
1226                           lustre_msg_get_conn_cnt(req->rq_reqmsg),
1227                           req->rq_export->exp_conn_cnt);
1228                 return -EEXIST;
1229         }
1230         if (unlikely(obd == NULL || obd->obd_fail)) {
1231                 /*
1232                  * Failing over, don't handle any more reqs,
1233                  * send error response instead.
1234                  */
1235                 CDEBUG(D_RPCTRACE, "Dropping req %p for failed obd %s\n",
1236                         req, (obd != NULL) ? obd->obd_name : "unknown");
1237                 rc = -ENODEV;
1238         } else if (lustre_msg_get_flags(req->rq_reqmsg) &
1239                    (MSG_REPLAY | MSG_REQ_REPLAY_DONE) &&
1240                    !obd->obd_recovering) {
1241                 DEBUG_REQ(D_ERROR, req,
1242                           "Invalid replay without recovery");
1243                 class_fail_export(req->rq_export);
1244                 rc = -ENODEV;
1245         } else if (lustre_msg_get_transno(req->rq_reqmsg) != 0 &&
1246                    !obd->obd_recovering) {
1247                 DEBUG_REQ(D_ERROR, req,
1248                           "Invalid req with transno %llu without recovery",
1249                           lustre_msg_get_transno(req->rq_reqmsg));
1250                 class_fail_export(req->rq_export);
1251                 rc = -ENODEV;
1252         }
1253
1254         if (unlikely(rc < 0)) {
1255                 req->rq_status = rc;
1256                 ptlrpc_error(req);
1257         }
1258         return rc;
1259 }
1260
1261 static void ptlrpc_at_set_timer(struct ptlrpc_service_part *svcpt)
1262 {
1263         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1264         time64_t next;
1265
1266         if (array->paa_count == 0) {
1267                 timer_delete(&svcpt->scp_at_timer);
1268                 return;
1269         }
1270
1271         /* Set timer for closest deadline */
1272         next = array->paa_deadline - ktime_get_real_seconds() -
1273                at_early_margin;
1274         if (next <= 0) {
1275                 ptlrpc_at_timer(cfs_timer_cb_arg(svcpt, scp_at_timer));
1276         } else {
1277                 mod_timer(&svcpt->scp_at_timer,
1278                           jiffies + nsecs_to_jiffies(next * NSEC_PER_SEC));
1279                 CDEBUG(D_INFO, "armed %s at %+llds\n",
1280                        svcpt->scp_service->srv_name, next);
1281         }
1282 }
1283
1284 /* Add rpc to early reply check list */
1285 static int ptlrpc_at_add_timed(struct ptlrpc_request *req)
1286 {
1287         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1288         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1289         struct ptlrpc_request *rq = NULL;
1290         __u32 index;
1291
1292         if (AT_OFF)
1293                 return(0);
1294
1295         if (req->rq_no_reply)
1296                 return 0;
1297
1298         if ((lustre_msghdr_get_flags(req->rq_reqmsg) & MSGHDR_AT_SUPPORT) == 0)
1299                 return(-ENOSYS);
1300
1301         spin_lock(&svcpt->scp_at_lock);
1302         LASSERT(list_empty(&req->rq_timed_list));
1303
1304         div_u64_rem(req->rq_deadline, array->paa_size, &index);
1305         if (array->paa_reqs_count[index] > 0) {
1306                 /*
1307                  * latest rpcs will have the latest deadlines in the list,
1308                  * so search backward.
1309                  */
1310                 list_for_each_entry_reverse(rq, &array->paa_reqs_array[index],
1311                                             rq_timed_list) {
1312                         if (req->rq_deadline >= rq->rq_deadline) {
1313                                 list_add(&req->rq_timed_list,
1314                                          &rq->rq_timed_list);
1315                                 break;
1316                         }
1317                 }
1318         }
1319
1320         /* Add the request at the head of the list */
1321         if (list_empty(&req->rq_timed_list))
1322                 list_add(&req->rq_timed_list, &array->paa_reqs_array[index]);
1323
1324         spin_lock(&req->rq_lock);
1325         req->rq_at_linked = 1;
1326         spin_unlock(&req->rq_lock);
1327         req->rq_at_index = index;
1328         array->paa_reqs_count[index]++;
1329         array->paa_count++;
1330         if (array->paa_count == 1 || array->paa_deadline > req->rq_deadline) {
1331                 array->paa_deadline = req->rq_deadline;
1332                 ptlrpc_at_set_timer(svcpt);
1333         }
1334         spin_unlock(&svcpt->scp_at_lock);
1335
1336         return 0;
1337 }
1338
1339 static void ptlrpc_at_remove_timed(struct ptlrpc_request *req)
1340 {
1341         struct ptlrpc_at_array *array;
1342
1343         array = &req->rq_rqbd->rqbd_svcpt->scp_at_array;
1344
1345         /* NB: must call with hold svcpt::scp_at_lock */
1346         LASSERT(!list_empty(&req->rq_timed_list));
1347         list_del_init(&req->rq_timed_list);
1348
1349         spin_lock(&req->rq_lock);
1350         req->rq_at_linked = 0;
1351         spin_unlock(&req->rq_lock);
1352
1353         array->paa_reqs_count[req->rq_at_index]--;
1354         array->paa_count--;
1355 }
1356
1357 /*
1358  * Attempt to extend the request deadline by sending an early reply to the
1359  * client.
1360  */
1361 static int ptlrpc_at_send_early_reply(struct ptlrpc_request *req)
1362 {
1363         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1364         struct ptlrpc_request *reqcopy;
1365         struct lustre_msg *reqmsg;
1366         timeout_t olddl = req->rq_deadline - ktime_get_real_seconds();
1367         time64_t newdl;
1368         int rc;
1369
1370         ENTRY;
1371
1372         if (CFS_FAIL_CHECK(OBD_FAIL_TGT_REPLAY_RECONNECT) ||
1373             CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_ENQ_RESEND)) {
1374                 /* don't send early reply */
1375                 RETURN(1);
1376         }
1377
1378         /*
1379          * deadline is when the client expects us to reply, margin is the
1380          * difference between clients' and servers' expectations
1381          */
1382         DEBUG_REQ(D_ADAPTTO, req,
1383                   "%ssending early reply (deadline %+ds, margin %+ds) for %d+%d",
1384                   AT_OFF ? "AT off - not " : "",
1385                   olddl, olddl - at_get(&svcpt->scp_at_estimate),
1386                   at_get(&svcpt->scp_at_estimate), at_extra);
1387
1388         if (AT_OFF)
1389                 RETURN(0);
1390
1391         if (olddl < 0) {
1392                 /* below message is checked in replay-ost-single.sh test_9 */
1393                 DEBUG_REQ(D_WARNING, req,
1394                           "Already past deadline (%+ds), not sending early reply. Consider increasing at_early_margin (%d)?",
1395                           olddl, at_early_margin);
1396
1397                 /* Return an error so we're not re-added to the timed list. */
1398                 RETURN(-ETIMEDOUT);
1399         }
1400
1401         if ((lustre_msghdr_get_flags(req->rq_reqmsg) &
1402              MSGHDR_AT_SUPPORT) == 0) {
1403                 DEBUG_REQ(D_INFO, req,
1404                           "Wanted to ask client for more time, but no AT support");
1405                 RETURN(-ENOSYS);
1406         }
1407
1408         if (req->rq_export &&
1409             lustre_msg_get_flags(req->rq_reqmsg) &
1410             (MSG_REPLAY | MSG_REQ_REPLAY_DONE | MSG_LOCK_REPLAY_DONE)) {
1411                 struct obd_device *obd_exp = req->rq_export->exp_obd;
1412
1413                 /*
1414                  * During recovery, we don't want to send too many early
1415                  * replies, but on the other hand we want to make sure the
1416                  * client has enough time to resend if the rpc is lost. So
1417                  * during the recovery period send at least 4 early replies,
1418                  * spacing them every at_extra if we can. at_estimate should
1419                  * always equal this fixed value during recovery.
1420                  */
1421
1422                 /*
1423                  * Don't account request processing time into AT history
1424                  * during recovery, it is not service time we need but
1425                  * includes also waiting time for recovering clients
1426                  */
1427                 newdl = min_t(time64_t, at_extra,
1428                               obd_exp->obd_recovery_timeout / 4) +
1429                         ktime_get_real_seconds();
1430         } else {
1431                 /*
1432                  * We want to extend the request deadline by at_extra seconds,
1433                  * so we set our service estimate to reflect how much time has
1434                  * passed since this request arrived plus an additional
1435                  * at_extra seconds. The client will calculate the new deadline
1436                  * based on this service estimate (plus some additional time to
1437                  * account for network latency). See ptlrpc_at_recv_early_reply
1438                  */
1439                 at_measured(&svcpt->scp_at_estimate, at_extra +
1440                             ktime_get_real_seconds() -
1441                             req->rq_arrival_time.tv_sec);
1442                 newdl = req->rq_arrival_time.tv_sec +
1443                         at_get(&svcpt->scp_at_estimate);
1444         }
1445
1446         /*
1447          * Check to see if we've actually increased the deadline -
1448          * we may be past adaptive_max
1449          */
1450         if (req->rq_deadline >= newdl) {
1451                 DEBUG_REQ(D_WARNING, req,
1452                           "Could not add any time (%d/%lld), not sending early reply",
1453                           olddl, newdl - ktime_get_real_seconds());
1454                 RETURN(-ETIMEDOUT);
1455         }
1456
1457         reqcopy = ptlrpc_request_cache_alloc(GFP_NOFS);
1458         if (reqcopy == NULL)
1459                 RETURN(-ENOMEM);
1460         OBD_ALLOC_LARGE(reqmsg, req->rq_reqlen);
1461         if (!reqmsg)
1462                 GOTO(out_free, rc = -ENOMEM);
1463
1464         *reqcopy = *req;
1465         spin_lock_init(&reqcopy->rq_early_free_lock);
1466         reqcopy->rq_reply_state = NULL;
1467         reqcopy->rq_rep_swab_mask = 0;
1468         reqcopy->rq_pack_bulk = 0;
1469         reqcopy->rq_pack_udesc = 0;
1470         reqcopy->rq_packed_final = 0;
1471         sptlrpc_svc_ctx_addref(reqcopy);
1472         /* We only need the reqmsg for the magic */
1473         reqcopy->rq_reqmsg = reqmsg;
1474         memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
1475
1476         /*
1477          * tgt_brw_read() and tgt_brw_write() may have decided not to reply.
1478          * Without this check, we would fail the rq_no_reply assertion in
1479          * ptlrpc_send_reply().
1480          */
1481         if (reqcopy->rq_no_reply)
1482                 GOTO(out, rc = -ETIMEDOUT);
1483
1484         LASSERT(atomic_read(&req->rq_refcount));
1485         /* if it is last refcount then early reply isn't needed */
1486         if (atomic_read(&req->rq_refcount) == 1) {
1487                 DEBUG_REQ(D_ADAPTTO, reqcopy,
1488                           "Normal reply already sent, abort early reply");
1489                 GOTO(out, rc = -EINVAL);
1490         }
1491
1492         /* Connection ref */
1493         reqcopy->rq_export = class_conn2export(
1494                         lustre_msg_get_handle(reqcopy->rq_reqmsg));
1495         if (reqcopy->rq_export == NULL)
1496                 GOTO(out, rc = -ENODEV);
1497
1498         /* RPC ref */
1499         class_export_rpc_inc(reqcopy->rq_export);
1500         if (reqcopy->rq_export->exp_obd &&
1501             reqcopy->rq_export->exp_obd->obd_fail)
1502                 GOTO(out_put, rc = -ENODEV);
1503
1504         rc = lustre_pack_reply_flags(reqcopy, 1, NULL, NULL, LPRFL_EARLY_REPLY);
1505         if (rc)
1506                 GOTO(out_put, rc);
1507
1508         rc = ptlrpc_send_reply(reqcopy, PTLRPC_REPLY_EARLY);
1509
1510         if (!rc) {
1511                 /* Adjust our own deadline to what we told the client */
1512                 req->rq_deadline = newdl;
1513                 req->rq_early_count++; /* number sent, server side */
1514         } else {
1515                 DEBUG_REQ(D_ERROR, req, "Early reply send failed: rc = %d", rc);
1516         }
1517
1518         /*
1519          * Free the (early) reply state from lustre_pack_reply.
1520          * (ptlrpc_send_reply takes it's own rs ref, so this is safe here)
1521          */
1522         ptlrpc_req_drop_rs(reqcopy);
1523
1524 out_put:
1525         class_export_rpc_dec(reqcopy->rq_export);
1526         class_export_put(reqcopy->rq_export);
1527 out:
1528         sptlrpc_svc_ctx_decref(reqcopy);
1529         OBD_FREE_LARGE(reqmsg, req->rq_reqlen);
1530 out_free:
1531         ptlrpc_request_cache_free(reqcopy);
1532         RETURN(rc);
1533 }
1534
1535 /*
1536  * Send early replies to everybody expiring within at_early_margin
1537  * asking for at_extra time
1538  */
1539 static int ptlrpc_at_check_timed(struct ptlrpc_service_part *svcpt)
1540 {
1541         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1542         struct ptlrpc_request *rq, *n;
1543         LIST_HEAD(work_list);
1544         __u32 index, count;
1545         time64_t deadline;
1546         time64_t now = ktime_get_real_seconds();
1547         s64 delay_ms;
1548         int first, counter = 0;
1549
1550         ENTRY;
1551         spin_lock(&svcpt->scp_at_lock);
1552         if (svcpt->scp_at_check == 0) {
1553                 spin_unlock(&svcpt->scp_at_lock);
1554                 RETURN(0);
1555         }
1556         delay_ms = ktime_ms_delta(ktime_get(), svcpt->scp_at_checktime);
1557         svcpt->scp_at_check = 0;
1558
1559         if (array->paa_count == 0) {
1560                 spin_unlock(&svcpt->scp_at_lock);
1561                 RETURN(0);
1562         }
1563
1564         /* The timer went off, but maybe the nearest rpc already completed. */
1565         first = array->paa_deadline - now;
1566         if (first > at_early_margin) {
1567                 /* We've still got plenty of time.  Reset the timer. */
1568                 ptlrpc_at_set_timer(svcpt);
1569                 spin_unlock(&svcpt->scp_at_lock);
1570                 RETURN(0);
1571         }
1572
1573         /*
1574          * We're close to a timeout, and we don't know how much longer the
1575          * server will take. Send early replies to everyone expiring soon.
1576          */
1577         deadline = -1;
1578         div_u64_rem(array->paa_deadline, array->paa_size, &index);
1579         count = array->paa_count;
1580         while (count > 0) {
1581                 count -= array->paa_reqs_count[index];
1582                 list_for_each_entry_safe(rq, n,
1583                                          &array->paa_reqs_array[index],
1584                                          rq_timed_list) {
1585                         if (rq->rq_deadline > now + at_early_margin) {
1586                                 /* update the earliest deadline */
1587                                 if (deadline == -1 ||
1588                                     rq->rq_deadline < deadline)
1589                                         deadline = rq->rq_deadline;
1590                                 break;
1591                         }
1592
1593                         /**
1594                          * ptlrpc_server_drop_request() may drop
1595                          * refcount to 0 already. Let's check this and
1596                          * don't add entry to work_list
1597                          */
1598                         if (likely(atomic_inc_not_zero(&rq->rq_refcount))) {
1599                                 ptlrpc_at_remove_timed(rq);
1600                                 list_add(&rq->rq_timed_list, &work_list);
1601                         } else {
1602                                 ptlrpc_at_remove_timed(rq);
1603                         }
1604
1605                         counter++;
1606                 }
1607
1608                 if (++index >= array->paa_size)
1609                         index = 0;
1610         }
1611         array->paa_deadline = deadline;
1612         /* we have a new earliest deadline, restart the timer */
1613         ptlrpc_at_set_timer(svcpt);
1614
1615         spin_unlock(&svcpt->scp_at_lock);
1616
1617         CDEBUG(D_ADAPTTO,
1618                "timeout in %+ds, asking for %d secs on %d early replies\n",
1619                first, at_extra, counter);
1620         if (first < 0) {
1621                 /*
1622                  * We're already past request deadlines before we even get a
1623                  * chance to send early replies
1624                  */
1625                 LCONSOLE_WARN("'%s' is processing requests too slowly, client may timeout. Late by %ds, missed %d early replies (reqs waiting=%d active=%d, at_estimate=%d, delay=%lldms)\n",
1626                               svcpt->scp_service->srv_name, -first, counter,
1627                               svcpt->scp_nreqs_incoming,
1628                               svcpt->scp_nreqs_active,
1629                               at_get(&svcpt->scp_at_estimate), delay_ms);
1630         }
1631
1632         /*
1633          * we took additional refcount so entries can't be deleted from list, no
1634          * locking is needed
1635          */
1636         while ((rq = list_first_entry_or_null(&work_list,
1637                                               struct ptlrpc_request,
1638                                               rq_timed_list)) != NULL) {
1639                 list_del_init(&rq->rq_timed_list);
1640
1641                 if (ptlrpc_at_send_early_reply(rq) == 0)
1642                         ptlrpc_at_add_timed(rq);
1643
1644                 ptlrpc_server_drop_request(rq);
1645         }
1646
1647         RETURN(1); /* return "did_something" for liblustre */
1648 }
1649
1650 /*
1651  * Check if we are already handling earlier incarnation of this request.
1652  * Called under &req->rq_export->exp_rpc_lock locked
1653  */
1654 static struct ptlrpc_request*
1655 ptlrpc_server_check_resend_in_progress(struct ptlrpc_request *req)
1656 {
1657         struct ptlrpc_request *tmp = NULL;
1658
1659         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))
1660                 return NULL;
1661
1662         /*
1663          * This list should not be longer than max_requests in
1664          * flights on the client, so it is not all that long.
1665          * Also we only hit this codepath in case of a resent
1666          * request which makes it even more rarely hit
1667          */
1668         list_for_each_entry(tmp, &req->rq_export->exp_reg_rpcs,
1669                                 rq_exp_list) {
1670                 /* Found duplicate one */
1671                 if (tmp->rq_xid == req->rq_xid)
1672                         goto found;
1673         }
1674         list_for_each_entry(tmp, &req->rq_export->exp_hp_rpcs,
1675                                 rq_exp_list) {
1676                 /* Found duplicate one */
1677                 if (tmp->rq_xid == req->rq_xid)
1678                         goto found;
1679         }
1680         return NULL;
1681
1682 found:
1683         DEBUG_REQ(D_HA, req, "Found duplicate req in processing");
1684         DEBUG_REQ(D_HA, tmp, "Request being processed");
1685         return tmp;
1686 }
1687
1688 #ifdef HAVE_SERVER_SUPPORT
1689 static void ptlrpc_server_mark_obsolete(struct ptlrpc_request *req)
1690 {
1691         req->rq_obsolete = 1;
1692 }
1693
1694 static void
1695 ptlrpc_server_mark_in_progress_obsolete(struct ptlrpc_request *req)
1696 {
1697         struct ptlrpc_request   *tmp = NULL;
1698         __u16                   tag;
1699
1700         if (!tgt_is_increasing_xid_client(req->rq_export) ||
1701             req->rq_export->exp_used_slots == NULL)
1702                 return;
1703
1704         tag = lustre_msg_get_tag(req->rq_reqmsg);
1705         if (tag == 0)
1706                 return;
1707
1708         if (!test_bit(tag - 1, req->rq_export->exp_used_slots))
1709                 return;
1710
1711         /* This list should not be longer than max_requests in
1712          * flights on the client, so it is not all that long.
1713          * Also we only hit this codepath in case of a resent
1714          * request which makes it even more rarely hit */
1715         list_for_each_entry(tmp, &req->rq_export->exp_reg_rpcs, rq_exp_list) {
1716                 if (tag == lustre_msg_get_tag(tmp->rq_reqmsg) &&
1717                     req->rq_xid > tmp->rq_xid)
1718                         ptlrpc_server_mark_obsolete(tmp);
1719
1720         }
1721         list_for_each_entry(tmp, &req->rq_export->exp_hp_rpcs, rq_exp_list) {
1722                 if (tag == lustre_msg_get_tag(tmp->rq_reqmsg) &&
1723                     req->rq_xid > tmp->rq_xid)
1724                         ptlrpc_server_mark_obsolete(tmp);
1725         }
1726 }
1727 #endif
1728
1729 /**
1730  * Check if a request should be assigned with a high priority.
1731  *
1732  * \retval      < 0: error occurred
1733  *                0: normal RPC request
1734  *               +1: high priority request
1735  */
1736 static int ptlrpc_server_hpreq_init(struct ptlrpc_service_part *svcpt,
1737                                     struct ptlrpc_request *req)
1738 {
1739         int rc = 0;
1740
1741         ENTRY;
1742         if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL) {
1743                 rc = svcpt->scp_service->srv_ops.so_hpreq_handler(req);
1744                 if (rc < 0)
1745                         RETURN(rc);
1746
1747                 LASSERT(rc == 0);
1748         }
1749
1750         if (req->rq_export != NULL && req->rq_ops != NULL) {
1751                 /*
1752                  * Perform request specific check. We should do this
1753                  * check before the request is added into exp_hp_rpcs
1754                  * list otherwise it may hit swab race at LU-1044.
1755                  */
1756                 if (req->rq_ops->hpreq_check != NULL) {
1757                         rc = req->rq_ops->hpreq_check(req);
1758                         if (rc == -ESTALE) {
1759                                 req->rq_status = rc;
1760                                 ptlrpc_error(req);
1761                         }
1762                         /*
1763                          * can only return error,
1764                          * 0 for normal request,
1765                          * or 1 for high priority request
1766                          */
1767                         LASSERT(rc <= 1);
1768                 }
1769         }
1770
1771         RETURN(rc);
1772 }
1773
1774 /** Remove the request from the export list. */
1775 static void ptlrpc_server_hpreq_fini(struct ptlrpc_request *req)
1776 {
1777         ENTRY;
1778         if (req->rq_export) {
1779                 /*
1780                  * refresh lock timeout again so that client has more
1781                  * room to send lock cancel RPC.
1782                  */
1783                 if (req->rq_ops && req->rq_ops->hpreq_fini)
1784                         req->rq_ops->hpreq_fini(req);
1785
1786                 ptlrpc_del_exp_list(req);
1787         }
1788         EXIT;
1789 }
1790
1791 static int ptlrpc_hpreq_check(struct ptlrpc_request *req)
1792 {
1793         return 1;
1794 }
1795
1796 static struct ptlrpc_hpreq_ops ptlrpc_hpreq_common = {
1797         .hpreq_check       = ptlrpc_hpreq_check,
1798 };
1799
1800 /* Hi-Priority RPC check by RPC operation code. */
1801 int ptlrpc_hpreq_handler(struct ptlrpc_request *req)
1802 {
1803         int opc = lustre_msg_get_opc(req->rq_reqmsg);
1804
1805         /*
1806          * Check for export to let only reconnects for not yet evicted
1807          * export to become a HP rpc.
1808          */
1809         if ((req->rq_export != NULL) &&
1810             (opc == OBD_PING || opc == MDS_CONNECT || opc == OST_CONNECT))
1811                 req->rq_ops = &ptlrpc_hpreq_common;
1812
1813         return 0;
1814 }
1815 EXPORT_SYMBOL(ptlrpc_hpreq_handler);
1816
1817 static int ptlrpc_server_request_add(struct ptlrpc_service_part *svcpt,
1818                                      struct ptlrpc_request *req)
1819 {
1820         int rc;
1821         bool hp;
1822         struct ptlrpc_request *orig;
1823
1824         ENTRY;
1825
1826         rc = ptlrpc_server_hpreq_init(svcpt, req);
1827         if (rc < 0)
1828                 RETURN(rc);
1829
1830         hp = rc > 0;
1831         ptlrpc_nrs_req_initialize(svcpt, req, hp);
1832
1833         while (req->rq_export != NULL) {
1834                 struct obd_export *exp = req->rq_export;
1835
1836                 /*
1837                  * do search for duplicated xid and the adding to the list
1838                  * atomically
1839                  */
1840                 spin_lock_bh(&exp->exp_rpc_lock);
1841 #ifdef HAVE_SERVER_SUPPORT
1842                 ptlrpc_server_mark_in_progress_obsolete(req);
1843 #endif
1844                 orig = ptlrpc_server_check_resend_in_progress(req);
1845                 if (orig && CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_RESEND_RACE)) {
1846                         spin_unlock_bh(&exp->exp_rpc_lock);
1847
1848                         CFS_RACE(OBD_FAIL_PTLRPC_RESEND_RACE);
1849                         msleep(4 * MSEC_PER_SEC);
1850                         continue;
1851                 }
1852
1853                 if (orig && likely(atomic_inc_not_zero(&orig->rq_refcount))) {
1854                         bool linked;
1855
1856                         spin_unlock_bh(&exp->exp_rpc_lock);
1857
1858                         /*
1859                          * When the client resend request and the server has
1860                          * the previous copy of it, we need to update deadlines,
1861                          * to be sure that the client and the server have equal
1862                          *  request deadlines.
1863                          */
1864
1865                         spin_lock(&orig->rq_rqbd->rqbd_svcpt->scp_at_lock);
1866                         linked = orig->rq_at_linked;
1867                         if (likely(linked))
1868                                 ptlrpc_at_remove_timed(orig);
1869                         spin_unlock(&orig->rq_rqbd->rqbd_svcpt->scp_at_lock);
1870                         orig->rq_deadline = req->rq_deadline;
1871                         orig->rq_rep_mbits = req->rq_rep_mbits;
1872                         if (likely(linked))
1873                                 ptlrpc_at_add_timed(orig);
1874                         ptlrpc_server_drop_request(orig);
1875                         ptlrpc_nrs_req_finalize(req);
1876
1877                         /* don't mark slot unused for resend in progress */
1878                         req->rq_obsolete = 1;
1879
1880                         RETURN(-EBUSY);
1881                 }
1882
1883                 ptlrpc_add_exp_list_nolock(req, exp, hp || req->rq_ops != NULL);
1884
1885                 spin_unlock_bh(&exp->exp_rpc_lock);
1886                 break;
1887         }
1888
1889         /*
1890          * the current thread is not the processing thread for this request
1891          * since that, but request is in exp_hp_list and can be find there.
1892          * Remove all relations between request and old thread.
1893          */
1894         req->rq_svc_thread->t_env->le_ses = NULL;
1895         req->rq_svc_thread = NULL;
1896         req->rq_session.lc_thread = NULL;
1897
1898         ptlrpc_nrs_req_add(svcpt, req, hp);
1899
1900         RETURN(0);
1901 }
1902
1903 /**
1904  * Allow to handle high priority request
1905  * User can call it w/o any lock but need to hold
1906  * ptlrpc_service_part::scp_req_lock to get reliable result
1907  */
1908 static bool ptlrpc_server_allow_high(struct ptlrpc_service_part *svcpt,
1909                                      bool force)
1910 {
1911         int running = svcpt->scp_nthrs_running;
1912
1913         if (!nrs_svcpt_has_hp(svcpt))
1914                 return false;
1915
1916         if (force)
1917                 return true;
1918
1919         if (ptlrpc_nrs_req_throttling_nolock(svcpt, true))
1920                 return false;
1921
1922         if (unlikely(svcpt->scp_service->srv_req_portal == MDS_REQUEST_PORTAL &&
1923                      CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND))) {
1924                 /* leave just 1 thread for normal RPCs */
1925                 running = PTLRPC_NTHRS_INIT;
1926                 if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL)
1927                         running += 1;
1928         }
1929
1930         if (svcpt->scp_nreqs_active >= running - 1)
1931                 return false;
1932
1933         if (svcpt->scp_nhreqs_active == 0)
1934                 return true;
1935
1936         return !ptlrpc_nrs_req_pending_nolock(svcpt, false) ||
1937                svcpt->scp_hreq_count < svcpt->scp_service->srv_hpreq_ratio;
1938 }
1939
1940 static bool ptlrpc_server_high_pending(struct ptlrpc_service_part *svcpt,
1941                                        bool force)
1942 {
1943         return ptlrpc_server_allow_high(svcpt, force) &&
1944                ptlrpc_nrs_req_pending_nolock(svcpt, true);
1945 }
1946
1947 /**
1948  * Only allow normal priority requests on a service that has a high-priority
1949  * queue if forced (i.e. cleanup), if there are other high priority requests
1950  * already being processed (i.e. those threads can service more high-priority
1951  * requests), or if there are enough idle threads that a later thread can do
1952  * a high priority request.
1953  * User can call it w/o any lock but need to hold
1954  * ptlrpc_service_part::scp_req_lock to get reliable result
1955  */
1956 static bool ptlrpc_server_allow_normal(struct ptlrpc_service_part *svcpt,
1957                                        bool force)
1958 {
1959         int running = svcpt->scp_nthrs_running;
1960
1961         if (unlikely(svcpt->scp_service->srv_req_portal == MDS_REQUEST_PORTAL &&
1962                      CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND))) {
1963                 /* leave just 1 thread for normal RPCs */
1964                 running = PTLRPC_NTHRS_INIT;
1965                 if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL)
1966                         running += 1;
1967         }
1968
1969         if (force)
1970                 return true;
1971
1972         if (ptlrpc_nrs_req_throttling_nolock(svcpt, false))
1973                 return false;
1974
1975         if (svcpt->scp_nreqs_active < running - 2)
1976                 return true;
1977
1978         if (svcpt->scp_nreqs_active >= running - 1)
1979                 return false;
1980
1981         return svcpt->scp_nhreqs_active > 0 || !nrs_svcpt_has_hp(svcpt);
1982 }
1983
1984 static bool ptlrpc_server_normal_pending(struct ptlrpc_service_part *svcpt,
1985                                          bool force)
1986 {
1987         return ptlrpc_server_allow_normal(svcpt, force) &&
1988                ptlrpc_nrs_req_pending_nolock(svcpt, false);
1989 }
1990
1991 /**
1992  * Returns true if there are requests available in incoming
1993  * request queue for processing and it is allowed to fetch them.
1994  * User can call it w/o any lock but need to hold ptlrpc_service::scp_req_lock
1995  * to get reliable result
1996  * \see ptlrpc_server_allow_normal
1997  * \see ptlrpc_server_allow high
1998  */
1999 static inline
2000 bool ptlrpc_server_request_pending(struct ptlrpc_service_part *svcpt,
2001                                    bool force)
2002 {
2003         return ptlrpc_server_high_pending(svcpt, force) ||
2004                ptlrpc_server_normal_pending(svcpt, force);
2005 }
2006
2007 /**
2008  * Fetch a request for processing from queue of unprocessed requests.
2009  * Favors high-priority requests.
2010  * Returns a pointer to fetched request.
2011  */
2012 static struct ptlrpc_request *
2013 ptlrpc_server_request_get(struct ptlrpc_service_part *svcpt, bool force)
2014 {
2015         struct ptlrpc_request *req = NULL;
2016
2017         ENTRY;
2018
2019         spin_lock(&svcpt->scp_req_lock);
2020
2021         if (ptlrpc_server_high_pending(svcpt, force)) {
2022                 req = ptlrpc_nrs_req_get_nolock(svcpt, true, force);
2023                 if (req != NULL) {
2024                         svcpt->scp_hreq_count++;
2025                         goto got_request;
2026                 }
2027         }
2028
2029         if (ptlrpc_server_normal_pending(svcpt, force)) {
2030                 req = ptlrpc_nrs_req_get_nolock(svcpt, false, force);
2031                 if (req != NULL) {
2032                         svcpt->scp_hreq_count = 0;
2033                         goto got_request;
2034                 }
2035         }
2036
2037         spin_unlock(&svcpt->scp_req_lock);
2038         RETURN(NULL);
2039
2040 got_request:
2041         svcpt->scp_nreqs_active++;
2042         if (req->rq_hp)
2043                 svcpt->scp_nhreqs_active++;
2044
2045         spin_unlock(&svcpt->scp_req_lock);
2046
2047         if (likely(req->rq_export))
2048                 class_export_rpc_inc(req->rq_export);
2049
2050         RETURN(req);
2051 }
2052
2053 /**
2054  * Handle freshly incoming reqs, add to timed early reply list,
2055  * pass on to regular request queue.
2056  * All incoming requests pass through here before getting into
2057  * ptlrpc_server_handle_req later on.
2058  */
2059 static int ptlrpc_server_handle_req_in(struct ptlrpc_service_part *svcpt,
2060                                        struct ptlrpc_thread *thread)
2061 {
2062         struct ptlrpc_service *svc = svcpt->scp_service;
2063         struct ptlrpc_request *req;
2064         __u32 deadline;
2065         __u32 opc;
2066         int rc;
2067
2068         ENTRY;
2069
2070         spin_lock(&svcpt->scp_lock);
2071         if (list_empty(&svcpt->scp_req_incoming)) {
2072                 spin_unlock(&svcpt->scp_lock);
2073                 RETURN(0);
2074         }
2075
2076         req = list_first_entry(&svcpt->scp_req_incoming,
2077                                struct ptlrpc_request, rq_list);
2078         list_del_init(&req->rq_list);
2079         svcpt->scp_nreqs_incoming--;
2080         /*
2081          * Consider this still a "queued" request as far as stats are
2082          * concerned
2083          */
2084         spin_unlock(&svcpt->scp_lock);
2085
2086         /* go through security check/transform */
2087         rc = sptlrpc_svc_unwrap_request(req);
2088         switch (rc) {
2089         case SECSVC_OK:
2090                 break;
2091         case SECSVC_COMPLETE:
2092                 target_send_reply(req, 0, OBD_FAIL_MDS_ALL_REPLY_NET);
2093                 goto err_req;
2094         case SECSVC_DROP:
2095                 goto err_req;
2096         default:
2097                 LBUG();
2098         }
2099
2100         /*
2101          * for null-flavored rpc, msg has been unpacked by sptlrpc, although
2102          * redo it wouldn't be harmful.
2103          */
2104         if (SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc) != SPTLRPC_POLICY_NULL) {
2105                 rc = ptlrpc_unpack_req_msg(req, req->rq_reqlen);
2106                 if (rc != 0) {
2107                         CERROR("error unpacking request: ptl %d from %s x%llu\n",
2108                                svc->srv_req_portal, libcfs_idstr(&req->rq_peer),
2109                                req->rq_xid);
2110                         goto err_req;
2111                 }
2112         }
2113
2114         rc = lustre_unpack_req_ptlrpc_body(req, MSG_PTLRPC_BODY_OFF);
2115         if (rc) {
2116                 CERROR("error unpacking ptlrpc body: ptl %d from %s x %llu\n",
2117                        svc->srv_req_portal, libcfs_idstr(&req->rq_peer),
2118                        req->rq_xid);
2119                 goto err_req;
2120         }
2121
2122         opc = lustre_msg_get_opc(req->rq_reqmsg);
2123         if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_REQ_OPC) &&
2124             opc == cfs_fail_val) {
2125                 CERROR("drop incoming rpc opc %u, x%llu\n",
2126                        cfs_fail_val, req->rq_xid);
2127                 goto err_req;
2128         }
2129
2130         rc = -EINVAL;
2131         if (lustre_msg_get_type(req->rq_reqmsg) != PTL_RPC_MSG_REQUEST) {
2132                 CERROR("wrong packet type received (type=%u) from %s\n",
2133                        lustre_msg_get_type(req->rq_reqmsg),
2134                        libcfs_idstr(&req->rq_peer));
2135                 goto err_req;
2136         }
2137
2138         switch (opc) {
2139         case MDS_WRITEPAGE:
2140         case OST_WRITE:
2141         case OUT_UPDATE:
2142                 req->rq_bulk_write = 1;
2143                 break;
2144         case MDS_READPAGE:
2145         case OST_READ:
2146         case MGS_CONFIG_READ:
2147                 req->rq_bulk_read = 1;
2148                 break;
2149         }
2150
2151         CDEBUG(D_RPCTRACE, "got req x%llu\n", req->rq_xid);
2152
2153         req->rq_export = class_conn2export(
2154                 lustre_msg_get_handle(req->rq_reqmsg));
2155         if (req->rq_export) {
2156                 rc = ptlrpc_check_req(req);
2157                 if (rc == 0) {
2158                         rc = sptlrpc_target_export_check(req->rq_export, req);
2159                         if (rc)
2160                                 DEBUG_REQ(D_ERROR, req,
2161                                           "DROPPING req with illegal security flavor");
2162                 }
2163
2164                 if (rc)
2165                         goto err_req;
2166                 ptlrpc_update_export_timer(req->rq_export, 0);
2167         }
2168
2169         /* req_in handling should/must be fast */
2170         if (ktime_get_real_seconds() - req->rq_arrival_time.tv_sec > 5)
2171                 DEBUG_REQ(D_WARNING, req, "Slow req_in handling %llds",
2172                           ktime_get_real_seconds() -
2173                           req->rq_arrival_time.tv_sec);
2174
2175         /* Set rpc server deadline and add it to the timed list */
2176         deadline = (lustre_msghdr_get_flags(req->rq_reqmsg) &
2177                     MSGHDR_AT_SUPPORT) ?
2178                     /* The max time the client expects us to take */
2179                     lustre_msg_get_timeout(req->rq_reqmsg) : obd_timeout;
2180
2181         req->rq_deadline = req->rq_arrival_time.tv_sec + deadline;
2182         if (unlikely(deadline == 0)) {
2183                 DEBUG_REQ(D_ERROR, req, "Dropping request with 0 timeout");
2184                 goto err_req;
2185         }
2186
2187         /* Skip early reply */
2188         if (CFS_FAIL_PRECHECK(OBD_FAIL_MDS_RESEND))
2189                 req->rq_deadline += obd_timeout;
2190
2191         req->rq_svc_thread = thread;
2192         if (thread != NULL) {
2193                 /*
2194                  * initialize request session, it is needed for request
2195                  * processing by target
2196                  */
2197                 rc = lu_context_init(&req->rq_session, LCT_SERVER_SESSION |
2198                                                        LCT_NOREF);
2199                 if (rc) {
2200                         CERROR("%s: failure to initialize session: rc = %d\n",
2201                                thread->t_name, rc);
2202                         goto err_req;
2203                 }
2204                 req->rq_session.lc_thread = thread;
2205                 lu_context_enter(&req->rq_session);
2206                 thread->t_env->le_ses = &req->rq_session;
2207         }
2208
2209
2210         if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_ENQ_RESEND) &&
2211                      (opc == LDLM_ENQUEUE) &&
2212                      (lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)))
2213                 CFS_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_ENQ_RESEND, 6);
2214
2215         ptlrpc_at_add_timed(req);
2216
2217         if (opc != OST_CONNECT && opc != MDS_CONNECT &&
2218             opc != MGS_CONNECT && req->rq_export != NULL) {
2219                 if (exp_connect_flags2(req->rq_export) & OBD_CONNECT2_REP_MBITS)
2220                         req->rq_rep_mbits = lustre_msg_get_mbits(req->rq_reqmsg);
2221         }
2222
2223         /* Move it over to the request processing queue */
2224         rc = ptlrpc_server_request_add(svcpt, req);
2225         if (rc)
2226                 GOTO(err_req, rc);
2227
2228         wake_up(&svcpt->scp_waitq);
2229         RETURN(1);
2230
2231 err_req:
2232         ptlrpc_server_finish_request(svcpt, req);
2233
2234         RETURN(1);
2235 }
2236
2237 /**
2238  * Main incoming request handling logic.
2239  * Calls handler function from service to do actual processing.
2240  */
2241 static int ptlrpc_server_handle_request(struct ptlrpc_service_part *svcpt,
2242                                         struct ptlrpc_thread *thread)
2243 {
2244         struct ptlrpc_service *svc = svcpt->scp_service;
2245         struct ptlrpc_request *request;
2246         ktime_t work_start;
2247         ktime_t work_end;
2248         ktime_t arrived;
2249         s64 timediff_usecs;
2250         s64 arrived_usecs;
2251         int fail_opc = 0;
2252
2253         ENTRY;
2254
2255         request = ptlrpc_server_request_get(svcpt, false);
2256         if (request == NULL)
2257                 RETURN(0);
2258
2259         if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT))
2260                 fail_opc = OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT;
2261         else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
2262                 fail_opc = OBD_FAIL_PTLRPC_HPREQ_TIMEOUT;
2263
2264         if (unlikely(fail_opc)) {
2265                 if (request->rq_export && request->rq_ops)
2266                         CFS_FAIL_TIMEOUT(fail_opc, 4);
2267         }
2268
2269         ptlrpc_rqphase_move(request, RQ_PHASE_INTERPRET);
2270
2271         if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_DUMP_LOG))
2272                 libcfs_debug_dumplog();
2273
2274         work_start = ktime_get_real();
2275         arrived = timespec64_to_ktime(request->rq_arrival_time);
2276         timediff_usecs = ktime_us_delta(work_start, arrived);
2277         if (likely(svc->srv_stats != NULL)) {
2278                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
2279                                     timediff_usecs);
2280                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
2281                                     svcpt->scp_nreqs_incoming);
2282                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
2283                                     svcpt->scp_nreqs_active);
2284                 lprocfs_counter_add(svc->srv_stats, PTLRPC_TIMEOUT,
2285                                     at_get(&svcpt->scp_at_estimate));
2286         }
2287
2288         if (likely(request->rq_export)) {
2289                 if (unlikely(ptlrpc_check_req(request)))
2290                         goto put_conn;
2291                 ptlrpc_update_export_timer(request->rq_export,
2292                                            div_u64(timediff_usecs,
2293                                                    USEC_PER_SEC / 2));
2294         }
2295
2296         /*
2297          * Discard requests queued for longer than the deadline.
2298          * The deadline is increased if we send an early reply.
2299          */
2300         if (ktime_get_real_seconds() > request->rq_deadline) {
2301                 DEBUG_REQ(D_ERROR, request,
2302                           "Dropping timed-out request from %s: deadline %lld/%llds ago",
2303                           libcfs_idstr(&request->rq_peer),
2304                           request->rq_deadline -
2305                           request->rq_arrival_time.tv_sec,
2306                           ktime_get_real_seconds() - request->rq_deadline);
2307                 goto put_conn;
2308         }
2309
2310         CDEBUG(D_RPCTRACE,
2311                "Handling RPC req@%p pname:cluuid+ref:pid:xid:nid:opc:job %s:%s+%d:%d:x%llu:%s:%d:%s\n",
2312                request, current->comm,
2313                (request->rq_export ?
2314                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
2315                (request->rq_export ?
2316                 refcount_read(&request->rq_export->exp_handle.h_ref) : -99),
2317                lustre_msg_get_status(request->rq_reqmsg), request->rq_xid,
2318                libcfs_idstr(&request->rq_peer),
2319                lustre_msg_get_opc(request->rq_reqmsg),
2320                lustre_msg_get_jobid(request->rq_reqmsg) ?: "");
2321
2322         if (lustre_msg_get_opc(request->rq_reqmsg) != OBD_PING)
2323                 CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_PAUSE_REQ, cfs_fail_val);
2324
2325         CDEBUG(D_NET, "got req %llu\n", request->rq_xid);
2326
2327         /* re-assign request and sesson thread to the current one */
2328         request->rq_svc_thread = thread;
2329         if (thread != NULL) {
2330                 LASSERT(request->rq_session.lc_thread == NULL);
2331                 request->rq_session.lc_thread = thread;
2332                 thread->t_env->le_ses = &request->rq_session;
2333         }
2334         svc->srv_ops.so_req_handler(request);
2335
2336         ptlrpc_rqphase_move(request, RQ_PHASE_COMPLETE);
2337
2338 put_conn:
2339         if (unlikely(ktime_get_real_seconds() > request->rq_deadline)) {
2340                 DEBUG_REQ(D_WARNING, request,
2341                           "Request took longer than estimated (%lld/%llds); client may timeout",
2342                           request->rq_deadline -
2343                           request->rq_arrival_time.tv_sec,
2344                           ktime_get_real_seconds() - request->rq_deadline);
2345         }
2346
2347         work_end = ktime_get_real();
2348         timediff_usecs = ktime_us_delta(work_end, work_start);
2349         arrived_usecs = ktime_us_delta(work_end, arrived);
2350         CDEBUG(D_RPCTRACE,
2351                "Handled RPC req@%p pname:cluuid+ref:pid:xid:nid:opc:job %s:%s+%d:%d:x%llu:%s:%d:%s Request processed in %lldus (%lldus total) trans %llu rc %d/%d\n",
2352                request, current->comm,
2353                (request->rq_export ?
2354                (char *)request->rq_export->exp_client_uuid.uuid : "0"),
2355                (request->rq_export ?
2356                 refcount_read(&request->rq_export->exp_handle.h_ref) : -99),
2357                lustre_msg_get_status(request->rq_reqmsg),
2358                request->rq_xid,
2359                libcfs_idstr(&request->rq_peer),
2360                lustre_msg_get_opc(request->rq_reqmsg),
2361                lustre_msg_get_jobid(request->rq_reqmsg) ?: "",
2362                timediff_usecs,
2363                arrived_usecs,
2364                (request->rq_repmsg ?
2365                lustre_msg_get_transno(request->rq_repmsg) :
2366                request->rq_transno),
2367                request->rq_status,
2368                (request->rq_repmsg ?
2369                lustre_msg_get_status(request->rq_repmsg) : -999));
2370         if (likely(svc->srv_stats != NULL && request->rq_reqmsg != NULL)) {
2371                 __u32 op = lustre_msg_get_opc(request->rq_reqmsg);
2372                 int opc = opcode_offset(op);
2373
2374                 if (opc > 0 && !(op == LDLM_ENQUEUE || op == MDS_REINT)) {
2375                         LASSERT(opc < LUSTRE_MAX_OPCODES);
2376                         lprocfs_counter_add(svc->srv_stats,
2377                                             opc + EXTRA_MAX_OPCODES,
2378                                             timediff_usecs);
2379                 }
2380         }
2381         if (unlikely(request->rq_early_count)) {
2382                 DEBUG_REQ(D_ADAPTTO, request,
2383                           "sent %d early replies before finishing in %llds",
2384                           request->rq_early_count,
2385                           div_u64(arrived_usecs, USEC_PER_SEC));
2386         }
2387
2388         ptlrpc_server_finish_active_request(svcpt, request);
2389
2390         RETURN(1);
2391 }
2392
2393 /**
2394  * An internal function to process a single reply state object.
2395  */
2396 static int ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
2397 {
2398         struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
2399         struct ptlrpc_service *svc = svcpt->scp_service;
2400         struct obd_export *exp;
2401         int nlocks;
2402         int been_handled;
2403
2404         ENTRY;
2405
2406         exp = rs->rs_export;
2407
2408         LASSERT(rs->rs_difficult);
2409         LASSERT(rs->rs_scheduled);
2410         LASSERT(list_empty(&rs->rs_list));
2411
2412         /*
2413          * The disk commit callback holds exp_uncommitted_replies_lock while it
2414          * iterates over newly committed replies, removing them from
2415          * exp_uncommitted_replies.  It then drops this lock and schedules the
2416          * replies it found for handling here.
2417          *
2418          * We can avoid contention for exp_uncommitted_replies_lock between the
2419          * HRT threads and further commit callbacks by checking rs_committed
2420          * which is set in the commit callback while it holds both
2421          * rs_lock and exp_uncommitted_reples.
2422          *
2423          * If we see rs_committed clear, the commit callback _may_ not have
2424          * handled this reply yet and we race with it to grab
2425          * exp_uncommitted_replies_lock before removing the reply from
2426          * exp_uncommitted_replies.  Note that if we lose the race and the
2427          * reply has already been removed, list_del_init() is a noop.
2428          *
2429          * If we see rs_committed set, we know the commit callback is handling,
2430          * or has handled this reply since store reordering might allow us to
2431          * see rs_committed set out of sequence.  But since this is done
2432          * holding rs_lock, we can be sure it has all completed once we hold
2433          * rs_lock, which we do right next.
2434          */
2435         if (!rs->rs_committed) {
2436                 /*
2437                  * if rs was commited, no need to convert locks, don't check
2438                  * rs_committed here because rs may never be added into
2439                  * exp_uncommitted_replies and this flag never be set, see
2440                  * target_send_reply()
2441                  */
2442                 if (rs->rs_convert_lock &&
2443                     rs->rs_transno > exp->exp_last_committed) {
2444                         struct ldlm_lock *lock;
2445                         struct ldlm_lock *ack_locks[RS_MAX_LOCKS] = { NULL };
2446
2447                         spin_lock(&rs->rs_lock);
2448                         if (rs->rs_convert_lock &&
2449                             rs->rs_transno > exp->exp_last_committed) {
2450                                 nlocks = rs->rs_nlocks;
2451                                 while (nlocks-- > 0) {
2452                                         /*
2453                                          * NB don't assume rs is always handled
2454                                          * by the same service thread (see
2455                                          * ptlrpc_hr_select, so REP-ACK hr may
2456                                          * race with trans commit, while the
2457                                          * latter will release locks, get locks
2458                                          * here early to downgrade to TXN mode
2459                                          * safely.
2460                                          */
2461                                         lock = ldlm_handle2lock(
2462                                                         &rs->rs_locks[nlocks]);
2463                                         LASSERT(lock);
2464                                         ack_locks[nlocks] = lock;
2465                                         rs->rs_modes[nlocks] = LCK_TXN;
2466                                 }
2467                                 nlocks = rs->rs_nlocks;
2468                                 rs->rs_convert_lock = 0;
2469                                 /*
2470                                  * clear rs_scheduled so that commit callback
2471                                  * can schedule again
2472                                  */
2473                                 rs->rs_scheduled = 0;
2474                                 spin_unlock(&rs->rs_lock);
2475
2476                                 while (nlocks-- > 0) {
2477                                         lock = ack_locks[nlocks];
2478                                         ldlm_lock_mode_downgrade(lock, LCK_TXN);
2479                                         LDLM_LOCK_PUT(lock);
2480                                 }
2481                                 RETURN(0);
2482                         }
2483                         spin_unlock(&rs->rs_lock);
2484                 }
2485
2486                 spin_lock(&exp->exp_uncommitted_replies_lock);
2487                 list_del_init(&rs->rs_obd_list);
2488                 spin_unlock(&exp->exp_uncommitted_replies_lock);
2489         }
2490
2491         spin_lock(&exp->exp_lock);
2492         /* Noop if removed already */
2493         list_del_init(&rs->rs_exp_list);
2494         spin_unlock(&exp->exp_lock);
2495
2496         spin_lock(&rs->rs_lock);
2497
2498         been_handled = rs->rs_handled;
2499         rs->rs_handled = 1;
2500
2501         nlocks = rs->rs_nlocks; /* atomic "steal", but */
2502         rs->rs_nlocks = 0; /* locks still on rs_locks! */
2503
2504         if (nlocks == 0 && !been_handled) {
2505                 /*
2506                  * If we see this, we should already have seen the warning
2507                  * in mds_steal_ack_locks()
2508                  */
2509                 CDEBUG(D_HA,
2510                        "All locks stolen from rs %p x%lld.t%lld o%d NID %s\n",
2511                        rs, rs->rs_xid, rs->rs_transno, rs->rs_opc,
2512                        libcfs_nidstr(&exp->exp_connection->c_peer.nid));
2513         }
2514
2515         if ((rs->rs_sent && !rs->rs_unlinked) || nlocks > 0) {
2516                 spin_unlock(&rs->rs_lock);
2517
2518                 /* We can unlink if the LNET_EVENT_SEND has occurred.
2519                  * If rs_unlinked is set then MD is already unlinked and no
2520                  * need to do so here.
2521                  */
2522                 if ((rs->rs_sent && !rs->rs_unlinked)) {
2523                         LNetMDUnlink(rs->rs_md_h);
2524                         /* Ignore return code; we're racing with completion */
2525                 }
2526
2527                 while (nlocks-- > 0)
2528                         ldlm_lock_decref(&rs->rs_locks[nlocks],
2529                                          rs->rs_modes[nlocks]);
2530
2531                 spin_lock(&rs->rs_lock);
2532         }
2533
2534         rs->rs_scheduled = 0;
2535         rs->rs_convert_lock = 0;
2536
2537         if (rs->rs_unlinked) {
2538                 /* Off the net */
2539                 spin_unlock(&rs->rs_lock);
2540
2541                 class_export_put(exp);
2542                 rs->rs_export = NULL;
2543                 ptlrpc_rs_decref(rs);
2544                 if (atomic_dec_and_test(&svcpt->scp_nreps_difficult) &&
2545                     svc->srv_is_stopping)
2546                         wake_up_all(&svcpt->scp_waitq);
2547                 RETURN(1);
2548         }
2549
2550         /* still on the net; callback will schedule */
2551         spin_unlock(&rs->rs_lock);
2552         RETURN(1);
2553 }
2554
2555
2556 static void ptlrpc_check_rqbd_pool(struct ptlrpc_service_part *svcpt)
2557 {
2558         int avail = svcpt->scp_nrqbds_posted;
2559         int low_water = test_req_buffer_pressure ? 0 :
2560                         svcpt->scp_service->srv_nbuf_per_group / 2;
2561
2562         /* NB I'm not locking; just looking. */
2563
2564         /*
2565          * CAVEAT EMPTOR: We might be allocating buffers here because we've
2566          * allowed the request history to grow out of control.  We could put a
2567          * sanity check on that here and cull some history if we need the
2568          * space.
2569          */
2570
2571         if (avail <= low_water)
2572                 ptlrpc_grow_req_bufs(svcpt, 1);
2573
2574         if (svcpt->scp_service->srv_stats) {
2575                 lprocfs_counter_add(svcpt->scp_service->srv_stats,
2576                                     PTLRPC_REQBUF_AVAIL_CNTR, avail);
2577         }
2578 }
2579
2580 static inline int ptlrpc_threads_enough(struct ptlrpc_service_part *svcpt)
2581 {
2582         return svcpt->scp_nreqs_active <
2583                svcpt->scp_nthrs_running - 1 -
2584                (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL);
2585 }
2586
2587 /**
2588  * allowed to create more threads
2589  * user can call it w/o any lock but need to hold
2590  * ptlrpc_service_part::scp_lock to get reliable result
2591  */
2592 static inline int ptlrpc_threads_increasable(struct ptlrpc_service_part *svcpt)
2593 {
2594         return svcpt->scp_nthrs_running +
2595                svcpt->scp_nthrs_starting <
2596                svcpt->scp_service->srv_nthrs_cpt_limit;
2597 }
2598
2599 /**
2600  * too many requests and allowed to create more threads
2601  */
2602 static inline int ptlrpc_threads_need_create(struct ptlrpc_service_part *svcpt)
2603 {
2604         return !ptlrpc_threads_enough(svcpt) &&
2605                 ptlrpc_threads_increasable(svcpt);
2606 }
2607
2608 static inline int ptlrpc_thread_stopping(struct ptlrpc_thread *thread)
2609 {
2610         return thread_is_stopping(thread) ||
2611                thread->t_svcpt->scp_service->srv_is_stopping;
2612 }
2613
2614 /* stop the highest numbered thread if there are too many threads running */
2615 static inline bool ptlrpc_thread_should_stop(struct ptlrpc_thread *thread)
2616 {
2617         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2618
2619         return thread->t_id >= svcpt->scp_service->srv_nthrs_cpt_limit &&
2620                 thread->t_id == svcpt->scp_thr_nextid - 1;
2621 }
2622
2623 static void ptlrpc_stop_thread(struct ptlrpc_thread *thread)
2624 {
2625         CDEBUG(D_INFO, "Stopping thread %s #%u\n",
2626                thread->t_svcpt->scp_service->srv_thread_name, thread->t_id);
2627         thread_add_flags(thread, SVC_STOPPING);
2628 }
2629
2630 static inline void ptlrpc_thread_stop(struct ptlrpc_thread *thread)
2631 {
2632         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2633
2634         spin_lock(&svcpt->scp_lock);
2635         if (ptlrpc_thread_should_stop(thread)) {
2636                 ptlrpc_stop_thread(thread);
2637                 svcpt->scp_thr_nextid--;
2638         }
2639         spin_unlock(&svcpt->scp_lock);
2640 }
2641
2642 static inline int ptlrpc_rqbd_pending(struct ptlrpc_service_part *svcpt)
2643 {
2644         return !list_empty(&svcpt->scp_rqbd_idle) &&
2645                svcpt->scp_rqbd_timeout == 0;
2646 }
2647
2648 static inline int
2649 ptlrpc_at_check(struct ptlrpc_service_part *svcpt)
2650 {
2651         return svcpt->scp_at_check;
2652 }
2653
2654 /*
2655  * If a thread runs too long or spends to much time on a single request,
2656  * we want to know about it, so we set up a delayed work item as a watchdog.
2657  * If it fires, we display a stack trace of the delayed thread,
2658  * providing we aren't rate-limited
2659  *
2660  * Watchdog stack traces are limited to 3 per 'libcfs_watchdog_ratelimit'
2661  * seconds
2662  */
2663 static struct ratelimit_state watchdog_limit;
2664
2665 static void ptlrpc_watchdog_fire(struct work_struct *w)
2666 {
2667         struct ptlrpc_thread *thread = container_of(w, struct ptlrpc_thread,
2668                                                     t_watchdog.work);
2669         u64 ms_lapse = ktime_ms_delta(ktime_get(), thread->t_touched);
2670         u32 ms_frac = do_div(ms_lapse, MSEC_PER_SEC);
2671
2672         /* ___ratelimit() returns true if the action is NOT ratelimited */
2673         if (__ratelimit(&watchdog_limit)) {
2674                 /* below message is checked in sanity-quota.sh test_6,18 */
2675                 LCONSOLE_WARN("%s: service thread pid %u was inactive for %llu.%03u seconds. The thread might be hung, or it might only be slow and will resume later. Dumping the stack trace for debugging purposes:\n",
2676                               thread->t_task->comm, thread->t_task->pid,
2677                               ms_lapse, ms_frac);
2678
2679                 libcfs_debug_dumpstack(thread->t_task);
2680         } else {
2681                 /* below message is checked in sanity-quota.sh test_6,18 */
2682                 LCONSOLE_WARN("%s: service thread pid %u was inactive for %llu.%03u seconds. Watchdog stack traces are limited to 3 per %u seconds, skipping this one.\n",
2683                               thread->t_task->comm, thread->t_task->pid,
2684                               ms_lapse, ms_frac, libcfs_watchdog_ratelimit);
2685         }
2686 }
2687
2688 void ptlrpc_watchdog_init(struct delayed_work *work, timeout_t timeout)
2689 {
2690         INIT_DELAYED_WORK(work, ptlrpc_watchdog_fire);
2691         schedule_delayed_work(work, cfs_time_seconds(timeout));
2692 }
2693
2694 void ptlrpc_watchdog_disable(struct delayed_work *work)
2695 {
2696         cancel_delayed_work_sync(work);
2697 }
2698
2699 void ptlrpc_watchdog_touch(struct delayed_work *work, timeout_t timeout)
2700 {
2701         struct ptlrpc_thread *thread = container_of(&work->work,
2702                                                     struct ptlrpc_thread,
2703                                                     t_watchdog.work);
2704         thread->t_touched = ktime_get();
2705         mod_delayed_work(system_wq, work, cfs_time_seconds(timeout));
2706 }
2707
2708 /**
2709  * requests wait on preprocessing
2710  * user can call it w/o any lock but need to hold
2711  * ptlrpc_service_part::scp_lock to get reliable result
2712  */
2713 static inline int
2714 ptlrpc_server_request_incoming(struct ptlrpc_service_part *svcpt)
2715 {
2716         return !list_empty(&svcpt->scp_req_incoming);
2717 }
2718
2719 static __attribute__((__noinline__)) int
2720 ptlrpc_wait_event(struct ptlrpc_service_part *svcpt,
2721                   struct ptlrpc_thread *thread)
2722 {
2723         ptlrpc_watchdog_disable(&thread->t_watchdog);
2724
2725         cond_resched();
2726
2727         if (svcpt->scp_rqbd_timeout == 0)
2728                 /* Don't exit while there are replies to be handled */
2729                 wait_event_idle_exclusive_lifo(
2730                         svcpt->scp_waitq,
2731                         ptlrpc_thread_stopping(thread) ||
2732                         ptlrpc_server_request_incoming(svcpt) ||
2733                         ptlrpc_server_request_pending(svcpt, false) ||
2734                         ptlrpc_rqbd_pending(svcpt) ||
2735                         ptlrpc_at_check(svcpt));
2736         else if (wait_event_idle_exclusive_lifo_timeout(
2737                          svcpt->scp_waitq,
2738                          ptlrpc_thread_stopping(thread) ||
2739                          ptlrpc_server_request_incoming(svcpt) ||
2740                          ptlrpc_server_request_pending(svcpt, false) ||
2741                          ptlrpc_rqbd_pending(svcpt) ||
2742                          ptlrpc_at_check(svcpt),
2743                          svcpt->scp_rqbd_timeout) == 0)
2744                 svcpt->scp_rqbd_timeout = 0;
2745
2746         if (ptlrpc_thread_stopping(thread))
2747                 return -EINTR;
2748
2749         ptlrpc_watchdog_touch(&thread->t_watchdog,
2750                               ptlrpc_server_get_timeout(svcpt));
2751         return 0;
2752 }
2753
2754 /**
2755  * Main thread body for service threads.
2756  * Waits in a loop waiting for new requests to process to appear.
2757  * Every time an incoming requests is added to its queue, a waitq
2758  * is woken up and one of the threads will handle it.
2759  */
2760 static int ptlrpc_main(void *arg)
2761 {
2762         struct ptlrpc_thread *thread = (struct ptlrpc_thread *)arg;
2763         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2764         struct ptlrpc_service *svc = svcpt->scp_service;
2765         struct ptlrpc_reply_state *rs;
2766         struct group_info *ginfo = NULL;
2767         struct lu_env *env;
2768         int counter = 0, rc = 0;
2769
2770         ENTRY;
2771         unshare_fs_struct();
2772
2773         thread->t_task = current;
2774         thread->t_pid = current->pid;
2775
2776         if (svc->srv_cpt_bind) {
2777                 rc = cfs_cpt_bind(svc->srv_cptable, svcpt->scp_cpt);
2778                 if (rc != 0) {
2779                         CWARN("%s: failed to bind %s on CPT %d\n",
2780                               svc->srv_name, thread->t_name, svcpt->scp_cpt);
2781                 }
2782         }
2783
2784         ginfo = groups_alloc(0);
2785         if (!ginfo)
2786                 GOTO(out, rc = -ENOMEM);
2787
2788         set_current_groups(ginfo);
2789         put_group_info(ginfo);
2790
2791         if (svc->srv_ops.so_thr_init != NULL) {
2792                 rc = svc->srv_ops.so_thr_init(thread);
2793                 if (rc)
2794                         GOTO(out, rc);
2795         }
2796
2797         OBD_ALLOC_PTR(env);
2798         if (env == NULL)
2799                 GOTO(out_srv_fini, rc = -ENOMEM);
2800         rc = lu_env_add(env);
2801         if (rc)
2802                 GOTO(out_env, rc);
2803
2804         rc = lu_context_init(&env->le_ctx,
2805                              svc->srv_ctx_tags|LCT_REMEMBER|LCT_NOREF);
2806         if (rc)
2807                 GOTO(out_env_remove, rc);
2808
2809         thread->t_env = env;
2810         env->le_ctx.lc_thread = thread;
2811         env->le_ctx.lc_cookie = 0x6;
2812
2813         while (!list_empty(&svcpt->scp_rqbd_idle)) {
2814                 rc = ptlrpc_server_post_idle_rqbds(svcpt);
2815                 if (rc >= 0)
2816                         continue;
2817
2818                 CERROR("Failed to post rqbd for %s on CPT %d: %d\n",
2819                         svc->srv_name, svcpt->scp_cpt, rc);
2820                 GOTO(out_ctx_fini, rc);
2821         }
2822
2823         /* Alloc reply state structure for this one */
2824         OBD_ALLOC_LARGE(rs, svc->srv_max_reply_size);
2825         if (!rs)
2826                 GOTO(out_ctx_fini, rc = -ENOMEM);
2827
2828         spin_lock(&svcpt->scp_lock);
2829
2830         LASSERT(thread_is_starting(thread));
2831         thread_clear_flags(thread, SVC_STARTING);
2832
2833         LASSERT(svcpt->scp_nthrs_starting == 1);
2834         svcpt->scp_nthrs_starting--;
2835
2836         /*
2837          * SVC_STOPPING may already be set here if someone else is trying
2838          * to stop the service while this new thread has been dynamically
2839          * forked. We still set SVC_RUNNING to let our creator know that
2840          * we are now running, however we will exit as soon as possible
2841          */
2842         thread_add_flags(thread, SVC_RUNNING);
2843         svcpt->scp_nthrs_running++;
2844         spin_unlock(&svcpt->scp_lock);
2845
2846         /* wake up our creator in case he's still waiting. */
2847         wake_up(&thread->t_ctl_waitq);
2848
2849         thread->t_touched = ktime_get();
2850         ptlrpc_watchdog_init(&thread->t_watchdog,
2851                          ptlrpc_server_get_timeout(svcpt));
2852
2853         spin_lock(&svcpt->scp_rep_lock);
2854         list_add(&rs->rs_list, &svcpt->scp_rep_idle);
2855         wake_up(&svcpt->scp_rep_waitq);
2856         spin_unlock(&svcpt->scp_rep_lock);
2857
2858         CDEBUG(D_NET, "service thread %d (#%d) started\n", thread->t_id,
2859                svcpt->scp_nthrs_running);
2860
2861         /* XXX maintain a list of all managed devices: insert here */
2862         while (!ptlrpc_thread_stopping(thread)) {
2863                 if (ptlrpc_wait_event(svcpt, thread))
2864                         break;
2865
2866                 ptlrpc_check_rqbd_pool(svcpt);
2867
2868                 if (ptlrpc_threads_need_create(svcpt)) {
2869                         /* Ignore return code - we tried... */
2870                         ptlrpc_start_thread(svcpt, 0);
2871                 }
2872
2873                 /* reset le_ses to initial state */
2874                 env->le_ses = NULL;
2875                 /* Refill the context before execution to make sure
2876                  * all thread keys are allocated */
2877                 lu_env_refill(env);
2878                 /* Process all incoming reqs before handling any */
2879                 if (ptlrpc_server_request_incoming(svcpt)) {
2880                         lu_context_enter(&env->le_ctx);
2881                         ptlrpc_server_handle_req_in(svcpt, thread);
2882                         lu_context_exit(&env->le_ctx);
2883
2884                         /* but limit ourselves in case of flood */
2885                         if (counter++ < 100)
2886                                 continue;
2887                         counter = 0;
2888                 }
2889
2890                 if (ptlrpc_at_check(svcpt))
2891                         ptlrpc_at_check_timed(svcpt);
2892
2893                 if (ptlrpc_server_request_pending(svcpt, false)) {
2894                         lu_context_enter(&env->le_ctx);
2895                         ptlrpc_server_handle_request(svcpt, thread);
2896                         lu_context_exit(&env->le_ctx);
2897                 }
2898
2899                 if (ptlrpc_rqbd_pending(svcpt) &&
2900                     ptlrpc_server_post_idle_rqbds(svcpt) < 0) {
2901                         /*
2902                          * I just failed to repost request buffers.
2903                          * Wait for a timeout (unless something else
2904                          * happens) before I try again
2905                          */
2906                         svcpt->scp_rqbd_timeout = cfs_time_seconds(1) / 10;
2907                         CDEBUG(D_RPCTRACE, "Posted buffers: %d\n",
2908                                svcpt->scp_nrqbds_posted);
2909                 }
2910                 /*
2911                  * If the number of threads has been tuned downward and this
2912                  * thread should be stopped, then stop in reverse order so the
2913                  * the threads always have contiguous thread index values.
2914                  */
2915                 if (unlikely(ptlrpc_thread_should_stop(thread)))
2916                         ptlrpc_thread_stop(thread);
2917         }
2918
2919         ptlrpc_watchdog_disable(&thread->t_watchdog);
2920
2921 out_ctx_fini:
2922         lu_context_fini(&env->le_ctx);
2923 out_env_remove:
2924         lu_env_remove(env);
2925 out_env:
2926         OBD_FREE_PTR(env);
2927 out_srv_fini:
2928         /* deconstruct service thread state created by ptlrpc_start_thread() */
2929         if (svc->srv_ops.so_thr_done != NULL)
2930                 svc->srv_ops.so_thr_done(thread);
2931 out:
2932         CDEBUG(D_RPCTRACE, "%s: service thread [%p:%u] %d exiting: rc = %d\n",
2933                thread->t_name, thread, thread->t_pid, thread->t_id, rc);
2934         spin_lock(&svcpt->scp_lock);
2935         if (thread_test_and_clear_flags(thread, SVC_STARTING))
2936                 svcpt->scp_nthrs_starting--;
2937
2938         if (thread_test_and_clear_flags(thread, SVC_RUNNING)) {
2939                 /* must know immediately */
2940                 svcpt->scp_nthrs_running--;
2941         }
2942
2943         thread->t_id = rc;
2944         thread_add_flags(thread, SVC_STOPPED);
2945
2946         wake_up(&thread->t_ctl_waitq);
2947         spin_unlock(&svcpt->scp_lock);
2948
2949         return rc;
2950 }
2951
2952 static int hrt_dont_sleep(struct ptlrpc_hr_thread *hrt,
2953                           struct list_head *replies)
2954 {
2955         int result;
2956
2957         spin_lock(&hrt->hrt_lock);
2958
2959         list_splice_init(&hrt->hrt_queue, replies);
2960         result = ptlrpc_hr.hr_stopping || !list_empty(replies);
2961
2962         spin_unlock(&hrt->hrt_lock);
2963         return result;
2964 }
2965
2966 /**
2967  * Main body of "handle reply" function.
2968  * It processes acked reply states
2969  */
2970 static int ptlrpc_hr_main(void *arg)
2971 {
2972         struct ptlrpc_hr_thread *hrt = (struct ptlrpc_hr_thread *)arg;
2973         struct ptlrpc_hr_partition *hrp = hrt->hrt_partition;
2974         LIST_HEAD(replies);
2975         struct lu_env *env;
2976         int rc;
2977
2978         unshare_fs_struct();
2979         OBD_ALLOC_PTR(env);
2980         if (env == NULL)
2981                 RETURN(-ENOMEM);
2982
2983         rc = cfs_cpt_bind(ptlrpc_hr.hr_cpt_table, hrp->hrp_cpt);
2984         if (rc != 0) {
2985                 char threadname[20];
2986
2987                 snprintf(threadname, sizeof(threadname), "ptlrpc_hr%02d_%03d",
2988                          hrp->hrp_cpt, hrt->hrt_id);
2989                 CWARN("Failed to bind %s on CPT %d of CPT table %p: rc = %d\n",
2990                       threadname, hrp->hrp_cpt, ptlrpc_hr.hr_cpt_table, rc);
2991         }
2992
2993         rc = lu_context_init(&env->le_ctx, LCT_MD_THREAD | LCT_DT_THREAD |
2994                              LCT_REMEMBER | LCT_NOREF);
2995         if (rc)
2996                 GOTO(out_env, rc);
2997
2998         rc = lu_env_add(env);
2999         if (rc)
3000                 GOTO(out_ctx_fini, rc);
3001
3002         atomic_inc(&hrp->hrp_nstarted);
3003         wake_up(&ptlrpc_hr.hr_waitq);
3004
3005         while (!ptlrpc_hr.hr_stopping) {
3006                 wait_event_idle(hrt->hrt_waitq, hrt_dont_sleep(hrt, &replies));
3007
3008                 while (!list_empty(&replies)) {
3009                         struct ptlrpc_reply_state *rs;
3010
3011                         rs = list_entry(replies.prev,
3012                                         struct ptlrpc_reply_state,
3013                                         rs_list);
3014                         list_del_init(&rs->rs_list);
3015                         /* refill keys if needed */
3016                         lu_env_refill(env);
3017                         lu_context_enter(&env->le_ctx);
3018                         ptlrpc_handle_rs(rs);
3019                         lu_context_exit(&env->le_ctx);
3020                 }
3021         }
3022
3023         atomic_inc(&hrp->hrp_nstopped);
3024         wake_up(&ptlrpc_hr.hr_waitq);
3025
3026         lu_env_remove(env);
3027 out_ctx_fini:
3028         lu_context_fini(&env->le_ctx);
3029 out_env:
3030         OBD_FREE_PTR(env);
3031         return 0;
3032 }
3033
3034 static void ptlrpc_stop_hr_threads(void)
3035 {
3036         struct ptlrpc_hr_partition *hrp;
3037         int i;
3038         int j;
3039
3040         ptlrpc_hr.hr_stopping = 1;
3041
3042         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3043                 if (hrp->hrp_thrs == NULL)
3044                         continue; /* uninitialized */
3045                 for (j = 0; j < hrp->hrp_nthrs; j++)
3046                         wake_up(&hrp->hrp_thrs[j].hrt_waitq);
3047         }
3048
3049         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3050                 if (hrp->hrp_thrs == NULL)
3051                         continue; /* uninitialized */
3052                 wait_event(ptlrpc_hr.hr_waitq,
3053                                atomic_read(&hrp->hrp_nstopped) ==
3054                                atomic_read(&hrp->hrp_nstarted));
3055         }
3056 }
3057
3058 static int ptlrpc_start_hr_threads(void)
3059 {
3060         struct ptlrpc_hr_partition *hrp;
3061         int i;
3062         int j;
3063
3064         ENTRY;
3065
3066         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3067                 int     rc = 0;
3068
3069                 for (j = 0; j < hrp->hrp_nthrs; j++) {
3070                         struct ptlrpc_hr_thread *hrt = &hrp->hrp_thrs[j];
3071                         struct task_struct *task;
3072
3073                         task = kthread_run(ptlrpc_hr_main,
3074                                            &hrp->hrp_thrs[j],
3075                                            "ptlrpc_hr%02d_%03d",
3076                                            hrp->hrp_cpt,
3077                                            hrt->hrt_id);
3078                         if (IS_ERR(task)) {
3079                                 rc = PTR_ERR(task);
3080                                 break;
3081                         }
3082                 }
3083
3084                 wait_event(ptlrpc_hr.hr_waitq,
3085                            atomic_read(&hrp->hrp_nstarted) == j);
3086
3087                 if (rc < 0) {
3088                         CERROR("cannot start reply handler thread %d:%d: rc = %d\n",
3089                                i, j, rc);
3090                         ptlrpc_stop_hr_threads();
3091                         RETURN(rc);
3092                 }
3093         }
3094
3095         RETURN(0);
3096 }
3097
3098 static void ptlrpc_svcpt_stop_threads(struct ptlrpc_service_part *svcpt)
3099 {
3100         struct ptlrpc_thread *thread;
3101         LIST_HEAD(zombie);
3102
3103         ENTRY;
3104
3105         CDEBUG(D_INFO, "Stopping threads for service %s\n",
3106                svcpt->scp_service->srv_name);
3107
3108         spin_lock(&svcpt->scp_lock);
3109         /* let the thread know that we would like it to stop asap */
3110         list_for_each_entry(thread, &svcpt->scp_threads, t_link)
3111                 ptlrpc_stop_thread(thread);
3112
3113         wake_up_all(&svcpt->scp_waitq);
3114
3115         while ((thread = list_first_entry_or_null(&svcpt->scp_threads,
3116                                                   struct ptlrpc_thread,
3117                                                   t_link)) != NULL) {
3118                 if (thread_is_stopped(thread)) {
3119                         list_move(&thread->t_link, &zombie);
3120                         continue;
3121                 }
3122                 spin_unlock(&svcpt->scp_lock);
3123
3124                 CDEBUG(D_INFO, "waiting for stopping-thread %s #%u\n",
3125                        svcpt->scp_service->srv_thread_name, thread->t_id);
3126                 wait_event_idle(thread->t_ctl_waitq,
3127                                 thread_is_stopped(thread));
3128
3129                 spin_lock(&svcpt->scp_lock);
3130         }
3131
3132         spin_unlock(&svcpt->scp_lock);
3133
3134         while ((thread = list_first_entry_or_null(&zombie,
3135                                                   struct ptlrpc_thread,
3136                                                   t_link)) != NULL) {
3137                 list_del(&thread->t_link);
3138                 OBD_FREE_PTR(thread);
3139         }
3140         EXIT;
3141 }
3142
3143 /**
3144  * Stops all threads of a particular service \a svc
3145  */
3146 static void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
3147 {
3148         struct ptlrpc_service_part *svcpt;
3149         int i;
3150
3151         ENTRY;
3152
3153         ptlrpc_service_for_each_part(svcpt, i, svc) {
3154                 if (svcpt->scp_service != NULL)
3155                         ptlrpc_svcpt_stop_threads(svcpt);
3156         }
3157
3158         EXIT;
3159 }
3160
3161 static int ptlrpc_start_threads(struct ptlrpc_service *svc)
3162 {
3163         int rc = 0;
3164         int i;
3165         int j;
3166
3167         ENTRY;
3168
3169         /* We require 2 threads min, see note in ptlrpc_server_handle_request */
3170         LASSERT(svc->srv_nthrs_cpt_init >= PTLRPC_NTHRS_INIT);
3171
3172         for (i = 0; i < svc->srv_ncpts; i++) {
3173                 for (j = 0; j < svc->srv_nthrs_cpt_init; j++) {
3174                         rc = ptlrpc_start_thread(svc->srv_parts[i], 1);
3175                         if (rc == 0)
3176                                 continue;
3177
3178                         if (rc != -EMFILE)
3179                                 goto failed;
3180                         /* We have enough threads, don't start more. b=15759 */
3181                         break;
3182                 }
3183         }
3184
3185         RETURN(0);
3186  failed:
3187         CERROR("cannot start %s thread #%d_%d: rc %d\n",
3188                svc->srv_thread_name, i, j, rc);
3189         ptlrpc_stop_all_threads(svc);
3190         RETURN(rc);
3191 }
3192
3193 static int ptlrpc_start_thread(struct ptlrpc_service_part *svcpt, int wait)
3194 {
3195         struct ptlrpc_thread *thread;
3196         struct ptlrpc_service *svc;
3197         struct task_struct *task;
3198         int rc;
3199
3200         ENTRY;
3201
3202         LASSERT(svcpt != NULL);
3203
3204         svc = svcpt->scp_service;
3205
3206         CDEBUG(D_RPCTRACE, "%s[%d] started %d min %d max %d\n",
3207                svc->srv_name, svcpt->scp_cpt, svcpt->scp_nthrs_running,
3208                svc->srv_nthrs_cpt_init, svc->srv_nthrs_cpt_limit);
3209
3210  again:
3211         if (unlikely(svc->srv_is_stopping))
3212                 RETURN(-ESRCH);
3213
3214         if (!ptlrpc_threads_increasable(svcpt) ||
3215             (CFS_FAIL_CHECK(OBD_FAIL_TGT_TOOMANY_THREADS) &&
3216              svcpt->scp_nthrs_running == svc->srv_nthrs_cpt_init - 1))
3217                 RETURN(-EMFILE);
3218
3219         OBD_CPT_ALLOC_PTR(thread, svc->srv_cptable, svcpt->scp_cpt);
3220         if (thread == NULL)
3221                 RETURN(-ENOMEM);
3222         init_waitqueue_head(&thread->t_ctl_waitq);
3223
3224         spin_lock(&svcpt->scp_lock);
3225         if (!ptlrpc_threads_increasable(svcpt)) {
3226                 spin_unlock(&svcpt->scp_lock);
3227                 OBD_FREE_PTR(thread);
3228                 RETURN(-EMFILE);
3229         }
3230
3231         if (svcpt->scp_nthrs_starting != 0) {
3232                 /*
3233                  * serialize starting because some modules (obdfilter)
3234                  * might require unique and contiguous t_id
3235                  */
3236                 LASSERT(svcpt->scp_nthrs_starting == 1);
3237                 spin_unlock(&svcpt->scp_lock);
3238                 OBD_FREE_PTR(thread);
3239                 if (wait) {
3240                         CDEBUG(D_INFO, "Waiting for creating thread %s #%d\n",
3241                                svc->srv_thread_name, svcpt->scp_thr_nextid);
3242                         schedule();
3243                         goto again;
3244                 }
3245
3246                 CDEBUG(D_INFO, "Creating thread %s #%d race, retry later\n",
3247                        svc->srv_thread_name, svcpt->scp_thr_nextid);
3248                 RETURN(-EAGAIN);
3249         }
3250
3251         svcpt->scp_nthrs_starting++;
3252         thread->t_id = svcpt->scp_thr_nextid++;
3253         thread_add_flags(thread, SVC_STARTING);
3254         thread->t_svcpt = svcpt;
3255
3256         list_add(&thread->t_link, &svcpt->scp_threads);
3257         spin_unlock(&svcpt->scp_lock);
3258
3259         if (svcpt->scp_cpt >= 0) {
3260                 snprintf(thread->t_name, PTLRPC_THR_NAME_LEN, "%s%02d_%03d",
3261                          svc->srv_thread_name, svcpt->scp_cpt, thread->t_id);
3262         } else {
3263                 snprintf(thread->t_name, PTLRPC_THR_NAME_LEN, "%s_%04d",
3264                          svc->srv_thread_name, thread->t_id);
3265         }
3266
3267         CDEBUG(D_RPCTRACE, "starting thread '%s'\n", thread->t_name);
3268         task = kthread_run(ptlrpc_main, thread, "%s", thread->t_name);
3269         if (IS_ERR(task)) {
3270                 rc = PTR_ERR(task);
3271                 CERROR("cannot start thread '%s': rc = %d\n",
3272                        thread->t_name, rc);
3273                 spin_lock(&svcpt->scp_lock);
3274                 --svcpt->scp_nthrs_starting;
3275                 if (thread_is_stopping(thread)) {
3276                         /*
3277                          * this ptlrpc_thread is being hanled
3278                          * by ptlrpc_svcpt_stop_threads now
3279                          */
3280                         thread_add_flags(thread, SVC_STOPPED);
3281                         wake_up(&thread->t_ctl_waitq);
3282                         spin_unlock(&svcpt->scp_lock);
3283                 } else {
3284                         list_del(&thread->t_link);
3285                         spin_unlock(&svcpt->scp_lock);
3286                         OBD_FREE_PTR(thread);
3287                 }
3288                 RETURN(rc);
3289         }
3290
3291         if (!wait)
3292                 RETURN(0);
3293
3294         wait_event_idle(thread->t_ctl_waitq,
3295                         thread_is_running(thread) || thread_is_stopped(thread));
3296
3297         rc = thread_is_stopped(thread) ? thread->t_id : 0;
3298         RETURN(rc);
3299 }
3300
3301 int ptlrpc_hr_init(void)
3302 {
3303         struct ptlrpc_hr_partition *hrp;
3304         struct ptlrpc_hr_thread *hrt;
3305         int rc;
3306         int cpt;
3307         int i;
3308         int weight;
3309
3310         ENTRY;
3311
3312         memset(&ptlrpc_hr, 0, sizeof(ptlrpc_hr));
3313         ptlrpc_hr.hr_cpt_table = cfs_cpt_tab;
3314
3315         ptlrpc_hr.hr_partitions = cfs_percpt_alloc(ptlrpc_hr.hr_cpt_table,
3316                                                    sizeof(*hrp));
3317         if (ptlrpc_hr.hr_partitions == NULL)
3318                 RETURN(-ENOMEM);
3319
3320         ratelimit_state_init(&watchdog_limit,
3321                              cfs_time_seconds(libcfs_watchdog_ratelimit), 3);
3322
3323         init_waitqueue_head(&ptlrpc_hr.hr_waitq);
3324
3325         preempt_disable();
3326         weight = cpumask_weight(topology_sibling_cpumask(smp_processor_id()));
3327         preempt_enable();
3328
3329         cfs_percpt_for_each(hrp, cpt, ptlrpc_hr.hr_partitions) {
3330                 hrp->hrp_cpt = cpt;
3331
3332                 atomic_set(&hrp->hrp_nstarted, 0);
3333                 atomic_set(&hrp->hrp_nstopped, 0);
3334
3335                 hrp->hrp_nthrs = cfs_cpt_weight(ptlrpc_hr.hr_cpt_table, cpt);
3336                 hrp->hrp_nthrs /= weight;
3337                 if (hrp->hrp_nthrs == 0)
3338                         hrp->hrp_nthrs = 1;
3339
3340                 OBD_CPT_ALLOC(hrp->hrp_thrs, ptlrpc_hr.hr_cpt_table, cpt,
3341                               hrp->hrp_nthrs * sizeof(*hrt));
3342                 if (hrp->hrp_thrs == NULL)
3343                         GOTO(out, rc = -ENOMEM);
3344
3345                 for (i = 0; i < hrp->hrp_nthrs; i++) {
3346                         hrt = &hrp->hrp_thrs[i];
3347
3348                         hrt->hrt_id = i;
3349                         hrt->hrt_partition = hrp;
3350                         init_waitqueue_head(&hrt->hrt_waitq);
3351                         spin_lock_init(&hrt->hrt_lock);
3352                         INIT_LIST_HEAD(&hrt->hrt_queue);
3353                 }
3354         }
3355
3356         rc = ptlrpc_start_hr_threads();
3357 out:
3358         if (rc != 0)
3359                 ptlrpc_hr_fini();
3360         RETURN(rc);
3361 }
3362
3363 void ptlrpc_hr_fini(void)
3364 {
3365         struct ptlrpc_hr_partition *hrp;
3366         int cpt;
3367
3368         if (ptlrpc_hr.hr_partitions == NULL)
3369                 return;
3370
3371         ptlrpc_stop_hr_threads();
3372
3373         cfs_percpt_for_each(hrp, cpt, ptlrpc_hr.hr_partitions) {
3374                 if (hrp->hrp_thrs)
3375                         OBD_FREE_PTR_ARRAY(hrp->hrp_thrs, hrp->hrp_nthrs);
3376         }
3377
3378         cfs_percpt_free(ptlrpc_hr.hr_partitions);
3379         ptlrpc_hr.hr_partitions = NULL;
3380 }
3381
3382
3383 /**
3384  * Wait until all already scheduled replies are processed.
3385  */
3386 static void ptlrpc_wait_replies(struct ptlrpc_service_part *svcpt)
3387 {
3388         while (1) {
3389                 if (wait_event_idle_timeout(
3390                         svcpt->scp_waitq,
3391                         atomic_read(&svcpt->scp_nreps_difficult) == 0,
3392                         cfs_time_seconds(10)) > 0)
3393                         break;
3394                 CWARN("Unexpectedly long timeout %s %p\n",
3395                       svcpt->scp_service->srv_name, svcpt->scp_service);
3396         }
3397 }
3398
3399 static void
3400 ptlrpc_service_del_atimer(struct ptlrpc_service *svc)
3401 {
3402         struct ptlrpc_service_part *svcpt;
3403         int i;
3404
3405         /* early disarm AT timer... */
3406         ptlrpc_service_for_each_part(svcpt, i, svc) {
3407                 if (svcpt->scp_service != NULL)
3408                         timer_delete(&svcpt->scp_at_timer);
3409         }
3410 }
3411
3412 static void
3413 ptlrpc_service_unlink_rqbd(struct ptlrpc_service *svc)
3414 {
3415         struct ptlrpc_service_part *svcpt;
3416         struct ptlrpc_request_buffer_desc *rqbd;
3417         int rc;
3418         int i;
3419
3420         /*
3421          * All history will be culled when the next request buffer is
3422          * freed in ptlrpc_service_purge_all()
3423          */
3424         svc->srv_hist_nrqbds_cpt_max = 0;
3425
3426         rc = LNetClearLazyPortal(svc->srv_req_portal);
3427         LASSERT(rc == 0);
3428
3429         ptlrpc_service_for_each_part(svcpt, i, svc) {
3430                 if (svcpt->scp_service == NULL)
3431                         break;
3432
3433                 /*
3434                  * Unlink all the request buffers.  This forces a 'final'
3435                  * event with its 'unlink' flag set for each posted rqbd
3436                  */
3437                 list_for_each_entry(rqbd, &svcpt->scp_rqbd_posted,
3438                                         rqbd_list) {
3439                         rc = LNetMDUnlink(rqbd->rqbd_md_h);
3440                         LASSERT(rc == 0 || rc == -ENOENT);
3441                 }
3442         }
3443
3444         ptlrpc_service_for_each_part(svcpt, i, svc) {
3445                 if (svcpt->scp_service == NULL)
3446                         break;
3447
3448                 /*
3449                  * Wait for the network to release any buffers
3450                  * it's currently filling
3451                  */
3452                 spin_lock(&svcpt->scp_lock);
3453                 while (svcpt->scp_nrqbds_posted != 0) {
3454                         int seconds = PTLRPC_REQ_LONG_UNLINK;
3455
3456                         spin_unlock(&svcpt->scp_lock);
3457                         /*
3458                          * Network access will complete in finite time but
3459                          * the HUGE timeout lets us CWARN for visibility
3460                          * of sluggish NALs
3461                          */
3462                         while (seconds > 0 &&
3463                                wait_event_idle_timeout(
3464                                        svcpt->scp_waitq,
3465                                        svcpt->scp_nrqbds_posted == 0,
3466                                        cfs_time_seconds(1)) == 0)
3467                                 seconds -= 1;
3468                         if (seconds == 0) {
3469                                 CWARN("Service %s waiting for request buffers\n",
3470                                       svcpt->scp_service->srv_name);
3471                         }
3472                         spin_lock(&svcpt->scp_lock);
3473                 }
3474                 spin_unlock(&svcpt->scp_lock);
3475         }
3476 }
3477
3478 static void
3479 ptlrpc_service_purge_all(struct ptlrpc_service *svc)
3480 {
3481         struct ptlrpc_service_part *svcpt;
3482         struct ptlrpc_request_buffer_desc *rqbd;
3483         struct ptlrpc_request *req;
3484         struct ptlrpc_reply_state *rs;
3485         int i;
3486
3487         ptlrpc_service_for_each_part(svcpt, i, svc) {
3488                 if (svcpt->scp_service == NULL)
3489                         break;
3490
3491                 spin_lock(&svcpt->scp_rep_lock);
3492                 while ((rs = list_first_entry_or_null(&svcpt->scp_rep_active,
3493                                                       struct ptlrpc_reply_state,
3494                                                       rs_list)) != NULL) {
3495                         spin_lock(&rs->rs_lock);
3496                         ptlrpc_schedule_difficult_reply(rs);
3497                         spin_unlock(&rs->rs_lock);
3498                 }
3499                 spin_unlock(&svcpt->scp_rep_lock);
3500
3501                 /*
3502                  * purge the request queue.  NB No new replies (rqbds
3503                  * all unlinked) and no service threads, so I'm the only
3504                  * thread noodling the request queue now
3505                  */
3506                 while ((req = list_first_entry_or_null(&svcpt->scp_req_incoming,
3507                                                        struct ptlrpc_request,
3508                                                        rq_list)) != NULL) {
3509                         list_del(&req->rq_list);
3510                         svcpt->scp_nreqs_incoming--;
3511                         ptlrpc_server_finish_request(svcpt, req);
3512                 }
3513
3514                 while (ptlrpc_server_request_pending(svcpt, true)) {
3515                         req = ptlrpc_server_request_get(svcpt, true);
3516                         LASSERT(req);
3517                         ptlrpc_server_finish_active_request(svcpt, req);
3518                 }
3519
3520                 /*
3521                  * The portal may be shared by several services (eg:OUT_PORTAL).
3522                  * So the request could be referenced by other target. So we
3523                  * have to wait the ptlrpc_server_drop_request invoked.
3524                  *
3525                  * TODO: move the req_buffer as global rather than per service.
3526                  */
3527                 spin_lock(&svcpt->scp_lock);
3528                 while (!list_empty(&svcpt->scp_rqbd_posted)) {
3529                         spin_unlock(&svcpt->scp_lock);
3530                         wait_event_idle_timeout(svcpt->scp_waitq,
3531                                 list_empty(&svcpt->scp_rqbd_posted),
3532                                 cfs_time_seconds(1));
3533                         spin_lock(&svcpt->scp_lock);
3534                 }
3535                 spin_unlock(&svcpt->scp_lock);
3536
3537                 LASSERT(svcpt->scp_nreqs_incoming == 0);
3538                 LASSERT(svcpt->scp_nreqs_active == 0);
3539                 /*
3540                  * history should have been culled by
3541                  * ptlrpc_server_finish_request
3542                  */
3543                 LASSERT(svcpt->scp_hist_nrqbds == 0);
3544
3545                 /*
3546                  * Now free all the request buffers since nothing
3547                  * references them any more...
3548                  */
3549                 while ((rqbd = list_first_entry_or_null(&svcpt->scp_rqbd_idle,
3550                                                         struct ptlrpc_request_buffer_desc,
3551                                                         rqbd_list)) != NULL)
3552                         ptlrpc_free_rqbd(rqbd);
3553
3554                 ptlrpc_wait_replies(svcpt);
3555
3556                 while ((rs = list_first_entry_or_null(&svcpt->scp_rep_idle,
3557                                                       struct ptlrpc_reply_state,
3558                                                       rs_list)) != NULL) {
3559                         list_del(&rs->rs_list);
3560                         OBD_FREE_LARGE(rs, svc->srv_max_reply_size);
3561                 }
3562         }
3563 }
3564
3565 static void
3566 ptlrpc_service_free(struct ptlrpc_service *svc)
3567 {
3568         struct ptlrpc_service_part      *svcpt;
3569         struct ptlrpc_at_array          *array;
3570         int                             i;
3571
3572         ptlrpc_service_for_each_part(svcpt, i, svc) {
3573                 if (svcpt->scp_service == NULL)
3574                         break;
3575
3576                 /* In case somebody rearmed this in the meantime */
3577                 timer_delete(&svcpt->scp_at_timer);
3578                 array = &svcpt->scp_at_array;
3579
3580                 if (array->paa_reqs_array != NULL) {
3581                         OBD_FREE_PTR_ARRAY(array->paa_reqs_array,
3582                                            array->paa_size);
3583                         array->paa_reqs_array = NULL;
3584                 }
3585
3586                 if (array->paa_reqs_count != NULL) {
3587                         OBD_FREE_PTR_ARRAY(array->paa_reqs_count,
3588                                            array->paa_size);
3589                         array->paa_reqs_count = NULL;
3590                 }
3591         }
3592
3593         ptlrpc_service_for_each_part(svcpt, i, svc)
3594                 OBD_FREE_PTR(svcpt);
3595
3596         if (svc->srv_cpts != NULL)
3597                 cfs_expr_list_values_free(svc->srv_cpts, svc->srv_ncpts);
3598
3599         OBD_FREE(svc, offsetof(struct ptlrpc_service,
3600                                srv_parts[svc->srv_ncpts]));
3601 }
3602
3603 int ptlrpc_unregister_service(struct ptlrpc_service *service)
3604 {
3605         ENTRY;
3606
3607         CDEBUG(D_NET, "%s: tearing down\n", service->srv_name);
3608
3609         service->srv_is_stopping = 1;
3610
3611         mutex_lock(&ptlrpc_all_services_mutex);
3612         list_del_init(&service->srv_list);
3613         mutex_unlock(&ptlrpc_all_services_mutex);
3614
3615         ptlrpc_service_del_atimer(service);
3616         ptlrpc_stop_all_threads(service);
3617
3618         ptlrpc_service_unlink_rqbd(service);
3619         ptlrpc_service_purge_all(service);
3620         ptlrpc_service_nrs_cleanup(service);
3621
3622         ptlrpc_lprocfs_unregister_service(service);
3623         ptlrpc_sysfs_unregister_service(service);
3624
3625         ptlrpc_service_free(service);
3626
3627         RETURN(0);
3628 }
3629 EXPORT_SYMBOL(ptlrpc_unregister_service);
3630
3631 /**
3632  * Returns 0 if the service is healthy.
3633  *
3634  * Right now, it just checks to make sure that requests aren't languishing
3635  * in the queue.  We'll use this health check to govern whether a node needs
3636  * to be shot, so it's intentionally non-aggressive.
3637  */
3638 static int ptlrpc_svcpt_health_check(struct ptlrpc_service_part *svcpt)
3639 {
3640         struct ptlrpc_request *request = NULL;
3641         struct timespec64 right_now;
3642         struct timespec64 timediff;
3643
3644         ktime_get_real_ts64(&right_now);
3645
3646         spin_lock(&svcpt->scp_req_lock);
3647         /* How long has the next entry been waiting? */
3648         if (ptlrpc_server_high_pending(svcpt, true))
3649                 request = ptlrpc_nrs_req_peek_nolock(svcpt, true);
3650         else if (ptlrpc_server_normal_pending(svcpt, true))
3651                 request = ptlrpc_nrs_req_peek_nolock(svcpt, false);
3652
3653         if (request == NULL) {
3654                 spin_unlock(&svcpt->scp_req_lock);
3655                 return 0;
3656         }
3657
3658         timediff = timespec64_sub(right_now, request->rq_arrival_time);
3659         spin_unlock(&svcpt->scp_req_lock);
3660
3661         if ((timediff.tv_sec) >
3662             (AT_OFF ? obd_timeout * 3 / 2 : at_max)) {
3663                 CERROR("%s: unhealthy - request has been waiting %llds\n",
3664                        svcpt->scp_service->srv_name, (s64)timediff.tv_sec);
3665                 return -1;
3666         }
3667
3668         return 0;
3669 }
3670
3671 int
3672 ptlrpc_service_health_check(struct ptlrpc_service *svc)
3673 {
3674         struct ptlrpc_service_part      *svcpt;
3675         int                             i;
3676
3677         if (svc == NULL)
3678                 return 0;
3679
3680         ptlrpc_service_for_each_part(svcpt, i, svc) {
3681                 int rc = ptlrpc_svcpt_health_check(svcpt);
3682
3683                 if (rc != 0)
3684                         return rc;
3685         }
3686         return 0;
3687 }
3688 EXPORT_SYMBOL(ptlrpc_service_health_check);