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