Whamcloud - gitweb
LU-10467 ptlrpc: convert use of l_wait_event_exclusive_head()
[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
1077 /**
1078  * to finish a request: stop sending more early replies, and release
1079  * the request.
1080  */
1081 static void ptlrpc_server_finish_request(struct ptlrpc_service_part *svcpt,
1082                                          struct ptlrpc_request *req)
1083 {
1084         ptlrpc_server_hpreq_fini(req);
1085
1086         ptlrpc_server_drop_request(req);
1087 }
1088
1089 /**
1090  * to finish an active request: stop sending more early replies, and release
1091  * the request. should be called after we finished handling the request.
1092  */
1093 static void ptlrpc_server_finish_active_request(
1094                                         struct ptlrpc_service_part *svcpt,
1095                                         struct ptlrpc_request *req)
1096 {
1097         spin_lock(&svcpt->scp_req_lock);
1098         ptlrpc_nrs_req_stop_nolock(req);
1099         svcpt->scp_nreqs_active--;
1100         if (req->rq_hp)
1101                 svcpt->scp_nhreqs_active--;
1102         spin_unlock(&svcpt->scp_req_lock);
1103
1104         ptlrpc_nrs_req_finalize(req);
1105
1106         if (req->rq_export != NULL)
1107                 class_export_rpc_dec(req->rq_export);
1108
1109         ptlrpc_server_finish_request(svcpt, req);
1110 }
1111
1112 /**
1113  * This function makes sure dead exports are evicted in a timely manner.
1114  * This function is only called when some export receives a message (i.e.,
1115  * the network is up.)
1116  */
1117 void ptlrpc_update_export_timer(struct obd_export *exp, time64_t extra_delay)
1118 {
1119         struct obd_export *oldest_exp;
1120         time64_t oldest_time, new_time;
1121
1122         ENTRY;
1123
1124         LASSERT(exp);
1125
1126         /*
1127          * Compensate for slow machines, etc, by faking our request time
1128          * into the future.  Although this can break the strict time-ordering
1129          * of the list, we can be really lazy here - we don't have to evict
1130          * at the exact right moment.  Eventually, all silent exports
1131          * will make it to the top of the list.
1132          */
1133
1134         /* Do not pay attention on 1sec or smaller renewals. */
1135         new_time = ktime_get_real_seconds() + extra_delay;
1136         if (exp->exp_last_request_time + 1 /*second */ >= new_time)
1137                 RETURN_EXIT;
1138
1139         exp->exp_last_request_time = new_time;
1140
1141         /*
1142          * exports may get disconnected from the chain even though the
1143          * export has references, so we must keep the spin lock while
1144          * manipulating the lists
1145          */
1146         spin_lock(&exp->exp_obd->obd_dev_lock);
1147
1148         if (list_empty(&exp->exp_obd_chain_timed)) {
1149                 /* this one is not timed */
1150                 spin_unlock(&exp->exp_obd->obd_dev_lock);
1151                 RETURN_EXIT;
1152         }
1153
1154         list_move_tail(&exp->exp_obd_chain_timed,
1155                        &exp->exp_obd->obd_exports_timed);
1156
1157         oldest_exp = list_entry(exp->exp_obd->obd_exports_timed.next,
1158                                 struct obd_export, exp_obd_chain_timed);
1159         oldest_time = oldest_exp->exp_last_request_time;
1160         spin_unlock(&exp->exp_obd->obd_dev_lock);
1161
1162         if (exp->exp_obd->obd_recovering) {
1163                 /* be nice to everyone during recovery */
1164                 EXIT;
1165                 return;
1166         }
1167
1168         /* Note - racing to start/reset the obd_eviction timer is safe */
1169         if (exp->exp_obd->obd_eviction_timer == 0) {
1170                 /* Check if the oldest entry is expired. */
1171                 if (ktime_get_real_seconds() >
1172                     oldest_time + PING_EVICT_TIMEOUT + extra_delay) {
1173                         /*
1174                          * We need a second timer, in case the net was down and
1175                          * it just came back. Since the pinger may skip every
1176                          * other PING_INTERVAL (see note in ptlrpc_pinger_main),
1177                          * we better wait for 3.
1178                          */
1179                         exp->exp_obd->obd_eviction_timer =
1180                                 ktime_get_real_seconds() + 3 * PING_INTERVAL;
1181                         CDEBUG(D_HA, "%s: Think about evicting %s from %lld\n",
1182                                exp->exp_obd->obd_name,
1183                                obd_export_nid2str(oldest_exp), oldest_time);
1184                 }
1185         } else {
1186                 if (ktime_get_real_seconds() >
1187                     (exp->exp_obd->obd_eviction_timer + extra_delay)) {
1188                         /*
1189                          * The evictor won't evict anyone who we've heard from
1190                          * recently, so we don't have to check before we start
1191                          * it.
1192                          */
1193                         if (!ping_evictor_wake(exp))
1194                                 exp->exp_obd->obd_eviction_timer = 0;
1195                 }
1196         }
1197
1198         EXIT;
1199 }
1200
1201 /**
1202  * Sanity check request \a req.
1203  * Return 0 if all is ok, error code otherwise.
1204  */
1205 static int ptlrpc_check_req(struct ptlrpc_request *req)
1206 {
1207         struct obd_device *obd = req->rq_export->exp_obd;
1208         int rc = 0;
1209
1210         if (unlikely(lustre_msg_get_conn_cnt(req->rq_reqmsg) <
1211                      req->rq_export->exp_conn_cnt)) {
1212                 DEBUG_REQ(D_RPCTRACE, req,
1213                           "DROPPING req from old connection %d < %d",
1214                           lustre_msg_get_conn_cnt(req->rq_reqmsg),
1215                           req->rq_export->exp_conn_cnt);
1216                 return -EEXIST;
1217         }
1218         if (unlikely(obd == NULL || obd->obd_fail)) {
1219                 /*
1220                  * Failing over, don't handle any more reqs,
1221                  * send error response instead.
1222                  */
1223                 CDEBUG(D_RPCTRACE, "Dropping req %p for failed obd %s\n",
1224                         req, (obd != NULL) ? obd->obd_name : "unknown");
1225                 rc = -ENODEV;
1226         } else if (lustre_msg_get_flags(req->rq_reqmsg) &
1227                    (MSG_REPLAY | MSG_REQ_REPLAY_DONE) &&
1228                    !obd->obd_recovering) {
1229                 DEBUG_REQ(D_ERROR, req,
1230                           "Invalid replay without recovery");
1231                 class_fail_export(req->rq_export);
1232                 rc = -ENODEV;
1233         } else if (lustre_msg_get_transno(req->rq_reqmsg) != 0 &&
1234                    !obd->obd_recovering) {
1235                 DEBUG_REQ(D_ERROR, req,
1236                           "Invalid req with transno %llu without recovery",
1237                           lustre_msg_get_transno(req->rq_reqmsg));
1238                 class_fail_export(req->rq_export);
1239                 rc = -ENODEV;
1240         }
1241
1242         if (unlikely(rc < 0)) {
1243                 req->rq_status = rc;
1244                 ptlrpc_error(req);
1245         }
1246         return rc;
1247 }
1248
1249 static void ptlrpc_at_set_timer(struct ptlrpc_service_part *svcpt)
1250 {
1251         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1252         time64_t next;
1253
1254         if (array->paa_count == 0) {
1255                 del_timer(&svcpt->scp_at_timer);
1256                 return;
1257         }
1258
1259         /* Set timer for closest deadline */
1260         next = array->paa_deadline - ktime_get_real_seconds() -
1261                at_early_margin;
1262         if (next <= 0) {
1263                 ptlrpc_at_timer(cfs_timer_cb_arg(svcpt, scp_at_timer));
1264         } else {
1265                 mod_timer(&svcpt->scp_at_timer,
1266                           jiffies + nsecs_to_jiffies(next * NSEC_PER_SEC));
1267                 CDEBUG(D_INFO, "armed %s at %+llds\n",
1268                        svcpt->scp_service->srv_name, next);
1269         }
1270 }
1271
1272 /* Add rpc to early reply check list */
1273 static int ptlrpc_at_add_timed(struct ptlrpc_request *req)
1274 {
1275         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1276         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1277         struct ptlrpc_request *rq = NULL;
1278         __u32 index;
1279
1280         if (AT_OFF)
1281                 return(0);
1282
1283         if (req->rq_no_reply)
1284                 return 0;
1285
1286         if ((lustre_msghdr_get_flags(req->rq_reqmsg) & MSGHDR_AT_SUPPORT) == 0)
1287                 return(-ENOSYS);
1288
1289         spin_lock(&svcpt->scp_at_lock);
1290         LASSERT(list_empty(&req->rq_timed_list));
1291
1292         div_u64_rem(req->rq_deadline, array->paa_size, &index);
1293         if (array->paa_reqs_count[index] > 0) {
1294                 /*
1295                  * latest rpcs will have the latest deadlines in the list,
1296                  * so search backward.
1297                  */
1298                 list_for_each_entry_reverse(rq, &array->paa_reqs_array[index],
1299                                             rq_timed_list) {
1300                         if (req->rq_deadline >= rq->rq_deadline) {
1301                                 list_add(&req->rq_timed_list,
1302                                          &rq->rq_timed_list);
1303                                 break;
1304                         }
1305                 }
1306         }
1307
1308         /* Add the request at the head of the list */
1309         if (list_empty(&req->rq_timed_list))
1310                 list_add(&req->rq_timed_list, &array->paa_reqs_array[index]);
1311
1312         spin_lock(&req->rq_lock);
1313         req->rq_at_linked = 1;
1314         spin_unlock(&req->rq_lock);
1315         req->rq_at_index = index;
1316         array->paa_reqs_count[index]++;
1317         array->paa_count++;
1318         if (array->paa_count == 1 || array->paa_deadline > req->rq_deadline) {
1319                 array->paa_deadline = req->rq_deadline;
1320                 ptlrpc_at_set_timer(svcpt);
1321         }
1322         spin_unlock(&svcpt->scp_at_lock);
1323
1324         return 0;
1325 }
1326
1327 static void ptlrpc_at_remove_timed(struct ptlrpc_request *req)
1328 {
1329         struct ptlrpc_at_array *array;
1330
1331         array = &req->rq_rqbd->rqbd_svcpt->scp_at_array;
1332
1333         /* NB: must call with hold svcpt::scp_at_lock */
1334         LASSERT(!list_empty(&req->rq_timed_list));
1335         list_del_init(&req->rq_timed_list);
1336
1337         spin_lock(&req->rq_lock);
1338         req->rq_at_linked = 0;
1339         spin_unlock(&req->rq_lock);
1340
1341         array->paa_reqs_count[req->rq_at_index]--;
1342         array->paa_count--;
1343 }
1344
1345 /*
1346  * Attempt to extend the request deadline by sending an early reply to the
1347  * client.
1348  */
1349 static int ptlrpc_at_send_early_reply(struct ptlrpc_request *req)
1350 {
1351         struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1352         struct ptlrpc_request *reqcopy;
1353         struct lustre_msg *reqmsg;
1354         time64_t olddl = req->rq_deadline - ktime_get_real_seconds();
1355         time64_t newdl;
1356         int rc;
1357
1358         ENTRY;
1359
1360         if (CFS_FAIL_CHECK(OBD_FAIL_TGT_REPLAY_RECONNECT)) {
1361                 /* don't send early reply */
1362                 RETURN(1);
1363         }
1364
1365         /*
1366          * deadline is when the client expects us to reply, margin is the
1367          * difference between clients' and servers' expectations
1368          */
1369         DEBUG_REQ(D_ADAPTTO, req,
1370                   "%ssending early reply (deadline %+llds, margin %+llds) for %d+%d",
1371                   AT_OFF ? "AT off - not " : "",
1372                   (s64)olddl, (s64)(olddl - at_get(&svcpt->scp_at_estimate)),
1373                   at_get(&svcpt->scp_at_estimate), at_extra);
1374
1375         if (AT_OFF)
1376                 RETURN(0);
1377
1378         if (olddl < 0) {
1379                 /* below message is checked in replay-ost-single.sh test_9 */
1380                 DEBUG_REQ(D_WARNING, req,
1381                           "Already past deadline (%+llds), not sending early reply. Consider increasing at_early_margin (%d)?",
1382                           (s64)olddl, at_early_margin);
1383
1384                 /* Return an error so we're not re-added to the timed list. */
1385                 RETURN(-ETIMEDOUT);
1386         }
1387
1388         if ((lustre_msghdr_get_flags(req->rq_reqmsg) &
1389              MSGHDR_AT_SUPPORT) == 0) {
1390                 DEBUG_REQ(D_INFO, req,
1391                           "Wanted to ask client for more time, but no AT support");
1392                 RETURN(-ENOSYS);
1393         }
1394
1395         if (req->rq_export &&
1396             lustre_msg_get_flags(req->rq_reqmsg) &
1397             (MSG_REPLAY | MSG_REQ_REPLAY_DONE | MSG_LOCK_REPLAY_DONE)) {
1398                 struct obd_device *obd_exp = req->rq_export->exp_obd;
1399
1400                 /*
1401                  * During recovery, we don't want to send too many early
1402                  * replies, but on the other hand we want to make sure the
1403                  * client has enough time to resend if the rpc is lost. So
1404                  * during the recovery period send at least 4 early replies,
1405                  * spacing them every at_extra if we can. at_estimate should
1406                  * always equal this fixed value during recovery.
1407                  */
1408
1409                 /*
1410                  * Don't account request processing time into AT history
1411                  * during recovery, it is not service time we need but
1412                  * includes also waiting time for recovering clients
1413                  */
1414                 newdl = min_t(time64_t, at_extra,
1415                               obd_exp->obd_recovery_timeout / 4) +
1416                         ktime_get_real_seconds();
1417         } else {
1418                 /*
1419                  * We want to extend the request deadline by at_extra seconds,
1420                  * so we set our service estimate to reflect how much time has
1421                  * passed since this request arrived plus an additional
1422                  * at_extra seconds. The client will calculate the new deadline
1423                  * based on this service estimate (plus some additional time to
1424                  * account for network latency). See ptlrpc_at_recv_early_reply
1425                  */
1426                 at_measured(&svcpt->scp_at_estimate, at_extra +
1427                             ktime_get_real_seconds() -
1428                             req->rq_arrival_time.tv_sec);
1429                 newdl = req->rq_arrival_time.tv_sec +
1430                         at_get(&svcpt->scp_at_estimate);
1431         }
1432
1433         /*
1434          * Check to see if we've actually increased the deadline -
1435          * we may be past adaptive_max
1436          */
1437         if (req->rq_deadline >= newdl) {
1438                 DEBUG_REQ(D_WARNING, req,
1439                           "Could not add any time (%lld/%lld), not sending early reply",
1440                           (s64)olddl, (s64)(newdl - ktime_get_real_seconds()));
1441                 RETURN(-ETIMEDOUT);
1442         }
1443
1444         reqcopy = ptlrpc_request_cache_alloc(GFP_NOFS);
1445         if (reqcopy == NULL)
1446                 RETURN(-ENOMEM);
1447         OBD_ALLOC_LARGE(reqmsg, req->rq_reqlen);
1448         if (!reqmsg)
1449                 GOTO(out_free, rc = -ENOMEM);
1450
1451         *reqcopy = *req;
1452         reqcopy->rq_reply_state = NULL;
1453         reqcopy->rq_rep_swab_mask = 0;
1454         reqcopy->rq_pack_bulk = 0;
1455         reqcopy->rq_pack_udesc = 0;
1456         reqcopy->rq_packed_final = 0;
1457         sptlrpc_svc_ctx_addref(reqcopy);
1458         /* We only need the reqmsg for the magic */
1459         reqcopy->rq_reqmsg = reqmsg;
1460         memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
1461
1462         /*
1463          * tgt_brw_read() and tgt_brw_write() may have decided not to reply.
1464          * Without this check, we would fail the rq_no_reply assertion in
1465          * ptlrpc_send_reply().
1466          */
1467         if (reqcopy->rq_no_reply)
1468                 GOTO(out, rc = -ETIMEDOUT);
1469
1470         LASSERT(atomic_read(&req->rq_refcount));
1471         /* if it is last refcount then early reply isn't needed */
1472         if (atomic_read(&req->rq_refcount) == 1) {
1473                 DEBUG_REQ(D_ADAPTTO, reqcopy,
1474                           "Normal reply already sent, abort early reply");
1475                 GOTO(out, rc = -EINVAL);
1476         }
1477
1478         /* Connection ref */
1479         reqcopy->rq_export = class_conn2export(
1480                         lustre_msg_get_handle(reqcopy->rq_reqmsg));
1481         if (reqcopy->rq_export == NULL)
1482                 GOTO(out, rc = -ENODEV);
1483
1484         /* RPC ref */
1485         class_export_rpc_inc(reqcopy->rq_export);
1486         if (reqcopy->rq_export->exp_obd &&
1487             reqcopy->rq_export->exp_obd->obd_fail)
1488                 GOTO(out_put, rc = -ENODEV);
1489
1490         rc = lustre_pack_reply_flags(reqcopy, 1, NULL, NULL, LPRFL_EARLY_REPLY);
1491         if (rc)
1492                 GOTO(out_put, rc);
1493
1494         rc = ptlrpc_send_reply(reqcopy, PTLRPC_REPLY_EARLY);
1495
1496         if (!rc) {
1497                 /* Adjust our own deadline to what we told the client */
1498                 req->rq_deadline = newdl;
1499                 req->rq_early_count++; /* number sent, server side */
1500         } else {
1501                 DEBUG_REQ(D_ERROR, req, "Early reply send failed: rc = %d", rc);
1502         }
1503
1504         /*
1505          * Free the (early) reply state from lustre_pack_reply.
1506          * (ptlrpc_send_reply takes it's own rs ref, so this is safe here)
1507          */
1508         ptlrpc_req_drop_rs(reqcopy);
1509
1510 out_put:
1511         class_export_rpc_dec(reqcopy->rq_export);
1512         class_export_put(reqcopy->rq_export);
1513 out:
1514         sptlrpc_svc_ctx_decref(reqcopy);
1515         OBD_FREE_LARGE(reqmsg, req->rq_reqlen);
1516 out_free:
1517         ptlrpc_request_cache_free(reqcopy);
1518         RETURN(rc);
1519 }
1520
1521 /*
1522  * Send early replies to everybody expiring within at_early_margin
1523  * asking for at_extra time
1524  */
1525 static int ptlrpc_at_check_timed(struct ptlrpc_service_part *svcpt)
1526 {
1527         struct ptlrpc_at_array *array = &svcpt->scp_at_array;
1528         struct ptlrpc_request *rq, *n;
1529         struct list_head work_list;
1530         __u32 index, count;
1531         time64_t deadline;
1532         time64_t now = ktime_get_real_seconds();
1533         s64 delay;
1534         int first, counter = 0;
1535
1536         ENTRY;
1537         spin_lock(&svcpt->scp_at_lock);
1538         if (svcpt->scp_at_check == 0) {
1539                 spin_unlock(&svcpt->scp_at_lock);
1540                 RETURN(0);
1541         }
1542         delay = ktime_ms_delta(ktime_get(), svcpt->scp_at_checktime);
1543         svcpt->scp_at_check = 0;
1544
1545         if (array->paa_count == 0) {
1546                 spin_unlock(&svcpt->scp_at_lock);
1547                 RETURN(0);
1548         }
1549
1550         /* The timer went off, but maybe the nearest rpc already completed. */
1551         first = array->paa_deadline - now;
1552         if (first > at_early_margin) {
1553                 /* We've still got plenty of time.  Reset the timer. */
1554                 ptlrpc_at_set_timer(svcpt);
1555                 spin_unlock(&svcpt->scp_at_lock);
1556                 RETURN(0);
1557         }
1558
1559         /*
1560          * We're close to a timeout, and we don't know how much longer the
1561          * server will take. Send early replies to everyone expiring soon.
1562          */
1563         INIT_LIST_HEAD(&work_list);
1564         deadline = -1;
1565         div_u64_rem(array->paa_deadline, array->paa_size, &index);
1566         count = array->paa_count;
1567         while (count > 0) {
1568                 count -= array->paa_reqs_count[index];
1569                 list_for_each_entry_safe(rq, n,
1570                                          &array->paa_reqs_array[index],
1571                                          rq_timed_list) {
1572                         if (rq->rq_deadline > now + at_early_margin) {
1573                                 /* update the earliest deadline */
1574                                 if (deadline == -1 ||
1575                                     rq->rq_deadline < deadline)
1576                                         deadline = rq->rq_deadline;
1577                                 break;
1578                         }
1579
1580                         /**
1581                          * ptlrpc_server_drop_request() may drop
1582                          * refcount to 0 already. Let's check this and
1583                          * don't add entry to work_list
1584                          */
1585                         if (likely(atomic_inc_not_zero(&rq->rq_refcount))) {
1586                                 ptlrpc_at_remove_timed(rq);
1587                                 list_add(&rq->rq_timed_list, &work_list);
1588                         } else {
1589                                 ptlrpc_at_remove_timed(rq);
1590                         }
1591
1592                         counter++;
1593                 }
1594
1595                 if (++index >= array->paa_size)
1596                         index = 0;
1597         }
1598         array->paa_deadline = deadline;
1599         /* we have a new earliest deadline, restart the timer */
1600         ptlrpc_at_set_timer(svcpt);
1601
1602         spin_unlock(&svcpt->scp_at_lock);
1603
1604         CDEBUG(D_ADAPTTO,
1605                "timeout in %+ds, asking for %d secs on %d early replies\n",
1606                first, at_extra, counter);
1607         if (first < 0) {
1608                 /*
1609                  * We're already past request deadlines before we even get a
1610                  * chance to send early replies
1611                  */
1612                 LCONSOLE_WARN("%s: This server is not able to keep up with request traffic (cpu-bound).\n",
1613                               svcpt->scp_service->srv_name);
1614                 CWARN("earlyQ=%d reqQ=%d recA=%d, svcEst=%d, delay=%lld\n",
1615                       counter, svcpt->scp_nreqs_incoming,
1616                       svcpt->scp_nreqs_active,
1617                       at_get(&svcpt->scp_at_estimate), delay);
1618         }
1619
1620         /*
1621          * we took additional refcount so entries can't be deleted from list, no
1622          * locking is needed
1623          */
1624         while (!list_empty(&work_list)) {
1625                 rq = list_entry(work_list.next, struct ptlrpc_request,
1626                                 rq_timed_list);
1627                 list_del_init(&rq->rq_timed_list);
1628
1629                 if (ptlrpc_at_send_early_reply(rq) == 0)
1630                         ptlrpc_at_add_timed(rq);
1631
1632                 ptlrpc_server_drop_request(rq);
1633         }
1634
1635         RETURN(1); /* return "did_something" for liblustre */
1636 }
1637
1638 /*
1639  * Check if we are already handling earlier incarnation of this request.
1640  * Called under &req->rq_export->exp_rpc_lock locked
1641  */
1642 static struct ptlrpc_request*
1643 ptlrpc_server_check_resend_in_progress(struct ptlrpc_request *req)
1644 {
1645         struct ptlrpc_request *tmp = NULL;
1646
1647         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT) ||
1648             (atomic_read(&req->rq_export->exp_rpc_count) == 0))
1649                 return NULL;
1650
1651         /*
1652          * bulk request are aborted upon reconnect, don't try to
1653          * find a match
1654          */
1655         if (req->rq_bulk_write || req->rq_bulk_read)
1656                 return NULL;
1657
1658         /*
1659          * This list should not be longer than max_requests in
1660          * flights on the client, so it is not all that long.
1661          * Also we only hit this codepath in case of a resent
1662          * request which makes it even more rarely hit
1663          */
1664         list_for_each_entry(tmp, &req->rq_export->exp_reg_rpcs,
1665                                 rq_exp_list) {
1666                 /* Found duplicate one */
1667                 if (tmp->rq_xid == req->rq_xid)
1668                         goto found;
1669         }
1670         list_for_each_entry(tmp, &req->rq_export->exp_hp_rpcs,
1671                                 rq_exp_list) {
1672                 /* Found duplicate one */
1673                 if (tmp->rq_xid == req->rq_xid)
1674                         goto found;
1675         }
1676         return NULL;
1677
1678 found:
1679         DEBUG_REQ(D_HA, req, "Found duplicate req in processing");
1680         DEBUG_REQ(D_HA, tmp, "Request being processed");
1681         return tmp;
1682 }
1683
1684 #ifdef HAVE_SERVER_SUPPORT
1685 static void ptlrpc_server_mark_obsolete(struct ptlrpc_request *req)
1686 {
1687         req->rq_obsolete = 1;
1688 }
1689
1690 static void
1691 ptlrpc_server_mark_in_progress_obsolete(struct ptlrpc_request *req)
1692 {
1693         struct ptlrpc_request   *tmp = NULL;
1694         __u16                   tag;
1695
1696         if (!tgt_is_increasing_xid_client(req->rq_export) ||
1697             req->rq_export->exp_used_slots == NULL)
1698                 return;
1699
1700         tag = lustre_msg_get_tag(req->rq_reqmsg);
1701         if (tag == 0)
1702                 return;
1703
1704         if (!test_bit(tag - 1, req->rq_export->exp_used_slots))
1705                 return;
1706
1707         /* This list should not be longer than max_requests in
1708          * flights on the client, so it is not all that long.
1709          * Also we only hit this codepath in case of a resent
1710          * request which makes it even more rarely hit */
1711         list_for_each_entry(tmp, &req->rq_export->exp_reg_rpcs, rq_exp_list) {
1712                 if (tag == lustre_msg_get_tag(tmp->rq_reqmsg) &&
1713                     req->rq_xid > tmp->rq_xid)
1714                         ptlrpc_server_mark_obsolete(tmp);
1715
1716         }
1717         list_for_each_entry(tmp, &req->rq_export->exp_hp_rpcs, rq_exp_list) {
1718                 if (tag == lustre_msg_get_tag(tmp->rq_reqmsg) &&
1719                     req->rq_xid > tmp->rq_xid)
1720                         ptlrpc_server_mark_obsolete(tmp);
1721         }
1722 }
1723 #endif
1724
1725 /**
1726  * Check if a request should be assigned with a high priority.
1727  *
1728  * \retval      < 0: error occurred
1729  *                0: normal RPC request
1730  *               +1: high priority request
1731  */
1732 static int ptlrpc_server_hpreq_init(struct ptlrpc_service_part *svcpt,
1733                                     struct ptlrpc_request *req)
1734 {
1735         int rc = 0;
1736
1737         ENTRY;
1738         if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL) {
1739                 rc = svcpt->scp_service->srv_ops.so_hpreq_handler(req);
1740                 if (rc < 0)
1741                         RETURN(rc);
1742
1743                 LASSERT(rc == 0);
1744         }
1745
1746         if (req->rq_export != NULL && req->rq_ops != NULL) {
1747                 /*
1748                  * Perform request specific check. We should do this
1749                  * check before the request is added into exp_hp_rpcs
1750                  * list otherwise it may hit swab race at LU-1044.
1751                  */
1752                 if (req->rq_ops->hpreq_check != NULL) {
1753                         rc = req->rq_ops->hpreq_check(req);
1754                         if (rc == -ESTALE) {
1755                                 req->rq_status = rc;
1756                                 ptlrpc_error(req);
1757                         }
1758                         /*
1759                          * can only return error,
1760                          * 0 for normal request,
1761                          * or 1 for high priority request
1762                          */
1763                         LASSERT(rc <= 1);
1764                 }
1765         }
1766
1767         RETURN(rc);
1768 }
1769
1770 /** Remove the request from the export list. */
1771 static void ptlrpc_server_hpreq_fini(struct ptlrpc_request *req)
1772 {
1773         ENTRY;
1774         if (req->rq_export) {
1775                 /*
1776                  * refresh lock timeout again so that client has more
1777                  * room to send lock cancel RPC.
1778                  */
1779                 if (req->rq_ops && req->rq_ops->hpreq_fini)
1780                         req->rq_ops->hpreq_fini(req);
1781
1782                 ptlrpc_del_exp_list(req);
1783         }
1784         EXIT;
1785 }
1786
1787 static int ptlrpc_hpreq_check(struct ptlrpc_request *req)
1788 {
1789         return 1;
1790 }
1791
1792 static struct ptlrpc_hpreq_ops ptlrpc_hpreq_common = {
1793         .hpreq_check       = ptlrpc_hpreq_check,
1794 };
1795
1796 /* Hi-Priority RPC check by RPC operation code. */
1797 int ptlrpc_hpreq_handler(struct ptlrpc_request *req)
1798 {
1799         int opc = lustre_msg_get_opc(req->rq_reqmsg);
1800
1801         /*
1802          * Check for export to let only reconnects for not yet evicted
1803          * export to become a HP rpc.
1804          */
1805         if ((req->rq_export != NULL) &&
1806             (opc == OBD_PING || opc == MDS_CONNECT || opc == OST_CONNECT))
1807                 req->rq_ops = &ptlrpc_hpreq_common;
1808
1809         return 0;
1810 }
1811 EXPORT_SYMBOL(ptlrpc_hpreq_handler);
1812
1813 static int ptlrpc_server_request_add(struct ptlrpc_service_part *svcpt,
1814                                      struct ptlrpc_request *req)
1815 {
1816         int rc;
1817         bool hp;
1818         struct ptlrpc_request *orig;
1819
1820         ENTRY;
1821
1822         rc = ptlrpc_server_hpreq_init(svcpt, req);
1823         if (rc < 0)
1824                 RETURN(rc);
1825
1826         hp = rc > 0;
1827         ptlrpc_nrs_req_initialize(svcpt, req, hp);
1828
1829         while (req->rq_export != NULL) {
1830                 struct obd_export *exp = req->rq_export;
1831
1832                 /*
1833                  * do search for duplicated xid and the adding to the list
1834                  * atomically
1835                  */
1836                 spin_lock_bh(&exp->exp_rpc_lock);
1837 #ifdef HAVE_SERVER_SUPPORT
1838                 ptlrpc_server_mark_in_progress_obsolete(req);
1839 #endif
1840                 orig = ptlrpc_server_check_resend_in_progress(req);
1841                 if (orig && OBD_FAIL_PRECHECK(OBD_FAIL_PTLRPC_RESEND_RACE)) {
1842                         spin_unlock_bh(&exp->exp_rpc_lock);
1843
1844                         OBD_RACE(OBD_FAIL_PTLRPC_RESEND_RACE);
1845                         msleep(4 * MSEC_PER_SEC);
1846                         continue;
1847                 }
1848
1849                 if (orig && likely(atomic_inc_not_zero(&orig->rq_refcount))) {
1850                         bool linked;
1851
1852                         spin_unlock_bh(&exp->exp_rpc_lock);
1853
1854                         /*
1855                          * When the client resend request and the server has
1856                          * the previous copy of it, we need to update deadlines,
1857                          * to be sure that the client and the server have equal
1858                          *  request deadlines.
1859                          */
1860
1861                         spin_lock(&orig->rq_rqbd->rqbd_svcpt->scp_at_lock);
1862                         linked = orig->rq_at_linked;
1863                         if (likely(linked))
1864                                 ptlrpc_at_remove_timed(orig);
1865                         spin_unlock(&orig->rq_rqbd->rqbd_svcpt->scp_at_lock);
1866                         orig->rq_deadline = req->rq_deadline;
1867                         if (likely(linked))
1868                                 ptlrpc_at_add_timed(orig);
1869                         ptlrpc_server_drop_request(orig);
1870                         ptlrpc_nrs_req_finalize(req);
1871
1872                         /* don't mark slot unused for resend in progress */
1873                         req->rq_obsolete = 1;
1874
1875                         RETURN(-EBUSY);
1876                 }
1877
1878                 ptlrpc_add_exp_list_nolock(req, exp, hp || req->rq_ops != NULL);
1879
1880                 spin_unlock_bh(&exp->exp_rpc_lock);
1881                 break;
1882         }
1883
1884         /*
1885          * the current thread is not the processing thread for this request
1886          * since that, but request is in exp_hp_list and can be find there.
1887          * Remove all relations between request and old thread.
1888          */
1889         req->rq_svc_thread->t_env->le_ses = NULL;
1890         req->rq_svc_thread = NULL;
1891         req->rq_session.lc_thread = NULL;
1892
1893         ptlrpc_nrs_req_add(svcpt, req, hp);
1894
1895         RETURN(0);
1896 }
1897
1898 /**
1899  * Allow to handle high priority request
1900  * User can call it w/o any lock but need to hold
1901  * ptlrpc_service_part::scp_req_lock to get reliable result
1902  */
1903 static bool ptlrpc_server_allow_high(struct ptlrpc_service_part *svcpt,
1904                                      bool force)
1905 {
1906         int running = svcpt->scp_nthrs_running;
1907
1908         if (!nrs_svcpt_has_hp(svcpt))
1909                 return false;
1910
1911         if (force)
1912                 return true;
1913
1914         if (ptlrpc_nrs_req_throttling_nolock(svcpt, true))
1915                 return false;
1916
1917         if (unlikely(svcpt->scp_service->srv_req_portal == MDS_REQUEST_PORTAL &&
1918                      CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND))) {
1919                 /* leave just 1 thread for normal RPCs */
1920                 running = PTLRPC_NTHRS_INIT;
1921                 if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL)
1922                         running += 1;
1923         }
1924
1925         if (svcpt->scp_nreqs_active >= running - 1)
1926                 return false;
1927
1928         if (svcpt->scp_nhreqs_active == 0)
1929                 return true;
1930
1931         return !ptlrpc_nrs_req_pending_nolock(svcpt, false) ||
1932                svcpt->scp_hreq_count < svcpt->scp_service->srv_hpreq_ratio;
1933 }
1934
1935 static bool ptlrpc_server_high_pending(struct ptlrpc_service_part *svcpt,
1936                                        bool force)
1937 {
1938         return ptlrpc_server_allow_high(svcpt, force) &&
1939                ptlrpc_nrs_req_pending_nolock(svcpt, true);
1940 }
1941
1942 /**
1943  * Only allow normal priority requests on a service that has a high-priority
1944  * queue if forced (i.e. cleanup), if there are other high priority requests
1945  * already being processed (i.e. those threads can service more high-priority
1946  * requests), or if there are enough idle threads that a later thread can do
1947  * a high priority request.
1948  * User can call it w/o any lock but need to hold
1949  * ptlrpc_service_part::scp_req_lock to get reliable result
1950  */
1951 static bool ptlrpc_server_allow_normal(struct ptlrpc_service_part *svcpt,
1952                                        bool force)
1953 {
1954         int running = svcpt->scp_nthrs_running;
1955
1956         if (unlikely(svcpt->scp_service->srv_req_portal == MDS_REQUEST_PORTAL &&
1957                      CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND))) {
1958                 /* leave just 1 thread for normal RPCs */
1959                 running = PTLRPC_NTHRS_INIT;
1960                 if (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL)
1961                         running += 1;
1962         }
1963
1964         if (force)
1965                 return true;
1966
1967         if (ptlrpc_nrs_req_throttling_nolock(svcpt, false))
1968                 return false;
1969
1970         if (svcpt->scp_nreqs_active < running - 2)
1971                 return true;
1972
1973         if (svcpt->scp_nreqs_active >= running - 1)
1974                 return false;
1975
1976         return svcpt->scp_nhreqs_active > 0 || !nrs_svcpt_has_hp(svcpt);
1977 }
1978
1979 static bool ptlrpc_server_normal_pending(struct ptlrpc_service_part *svcpt,
1980                                          bool force)
1981 {
1982         return ptlrpc_server_allow_normal(svcpt, force) &&
1983                ptlrpc_nrs_req_pending_nolock(svcpt, false);
1984 }
1985
1986 /**
1987  * Returns true if there are requests available in incoming
1988  * request queue for processing and it is allowed to fetch them.
1989  * User can call it w/o any lock but need to hold ptlrpc_service::scp_req_lock
1990  * to get reliable result
1991  * \see ptlrpc_server_allow_normal
1992  * \see ptlrpc_server_allow high
1993  */
1994 static inline
1995 bool ptlrpc_server_request_pending(struct ptlrpc_service_part *svcpt,
1996                                    bool force)
1997 {
1998         return ptlrpc_server_high_pending(svcpt, force) ||
1999                ptlrpc_server_normal_pending(svcpt, force);
2000 }
2001
2002 /**
2003  * Fetch a request for processing from queue of unprocessed requests.
2004  * Favors high-priority requests.
2005  * Returns a pointer to fetched request.
2006  */
2007 static struct ptlrpc_request *
2008 ptlrpc_server_request_get(struct ptlrpc_service_part *svcpt, bool force)
2009 {
2010         struct ptlrpc_request *req = NULL;
2011
2012         ENTRY;
2013
2014         spin_lock(&svcpt->scp_req_lock);
2015
2016         if (ptlrpc_server_high_pending(svcpt, force)) {
2017                 req = ptlrpc_nrs_req_get_nolock(svcpt, true, force);
2018                 if (req != NULL) {
2019                         svcpt->scp_hreq_count++;
2020                         goto got_request;
2021                 }
2022         }
2023
2024         if (ptlrpc_server_normal_pending(svcpt, force)) {
2025                 req = ptlrpc_nrs_req_get_nolock(svcpt, false, force);
2026                 if (req != NULL) {
2027                         svcpt->scp_hreq_count = 0;
2028                         goto got_request;
2029                 }
2030         }
2031
2032         spin_unlock(&svcpt->scp_req_lock);
2033         RETURN(NULL);
2034
2035 got_request:
2036         svcpt->scp_nreqs_active++;
2037         if (req->rq_hp)
2038                 svcpt->scp_nhreqs_active++;
2039
2040         spin_unlock(&svcpt->scp_req_lock);
2041
2042         if (likely(req->rq_export))
2043                 class_export_rpc_inc(req->rq_export);
2044
2045         RETURN(req);
2046 }
2047
2048 /**
2049  * Handle freshly incoming reqs, add to timed early reply list,
2050  * pass on to regular request queue.
2051  * All incoming requests pass through here before getting into
2052  * ptlrpc_server_handle_req later on.
2053  */
2054 static int ptlrpc_server_handle_req_in(struct ptlrpc_service_part *svcpt,
2055                                        struct ptlrpc_thread *thread)
2056 {
2057         struct ptlrpc_service *svc = svcpt->scp_service;
2058         struct ptlrpc_request *req;
2059         __u32 deadline;
2060         int rc;
2061
2062         ENTRY;
2063
2064         spin_lock(&svcpt->scp_lock);
2065         if (list_empty(&svcpt->scp_req_incoming)) {
2066                 spin_unlock(&svcpt->scp_lock);
2067                 RETURN(0);
2068         }
2069
2070         req = list_entry(svcpt->scp_req_incoming.next,
2071                              struct ptlrpc_request, rq_list);
2072         list_del_init(&req->rq_list);
2073         svcpt->scp_nreqs_incoming--;
2074         /*
2075          * Consider this still a "queued" request as far as stats are
2076          * concerned
2077          */
2078         spin_unlock(&svcpt->scp_lock);
2079
2080         /* go through security check/transform */
2081         rc = sptlrpc_svc_unwrap_request(req);
2082         switch (rc) {
2083         case SECSVC_OK:
2084                 break;
2085         case SECSVC_COMPLETE:
2086                 target_send_reply(req, 0, OBD_FAIL_MDS_ALL_REPLY_NET);
2087                 goto err_req;
2088         case SECSVC_DROP:
2089                 goto err_req;
2090         default:
2091                 LBUG();
2092         }
2093
2094         /*
2095          * for null-flavored rpc, msg has been unpacked by sptlrpc, although
2096          * redo it wouldn't be harmful.
2097          */
2098         if (SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc) != SPTLRPC_POLICY_NULL) {
2099                 rc = ptlrpc_unpack_req_msg(req, req->rq_reqlen);
2100                 if (rc != 0) {
2101                         CERROR("error unpacking request: ptl %d from %s x%llu\n",
2102                                svc->srv_req_portal, libcfs_id2str(req->rq_peer),
2103                                req->rq_xid);
2104                         goto err_req;
2105                 }
2106         }
2107
2108         rc = lustre_unpack_req_ptlrpc_body(req, MSG_PTLRPC_BODY_OFF);
2109         if (rc) {
2110                 CERROR("error unpacking ptlrpc body: ptl %d from %s x %llu\n",
2111                        svc->srv_req_portal, libcfs_id2str(req->rq_peer),
2112                        req->rq_xid);
2113                 goto err_req;
2114         }
2115
2116         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_REQ_OPC) &&
2117             lustre_msg_get_opc(req->rq_reqmsg) == cfs_fail_val) {
2118                 CERROR("drop incoming rpc opc %u, x%llu\n",
2119                        cfs_fail_val, req->rq_xid);
2120                 goto err_req;
2121         }
2122
2123         rc = -EINVAL;
2124         if (lustre_msg_get_type(req->rq_reqmsg) != PTL_RPC_MSG_REQUEST) {
2125                 CERROR("wrong packet type received (type=%u) from %s\n",
2126                        lustre_msg_get_type(req->rq_reqmsg),
2127                        libcfs_id2str(req->rq_peer));
2128                 goto err_req;
2129         }
2130
2131         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2132         case MDS_WRITEPAGE:
2133         case OST_WRITE:
2134         case OUT_UPDATE:
2135                 req->rq_bulk_write = 1;
2136                 break;
2137         case MDS_READPAGE:
2138         case OST_READ:
2139         case MGS_CONFIG_READ:
2140                 req->rq_bulk_read = 1;
2141                 break;
2142         }
2143
2144         CDEBUG(D_RPCTRACE, "got req x%llu\n", req->rq_xid);
2145
2146         req->rq_export = class_conn2export(
2147                 lustre_msg_get_handle(req->rq_reqmsg));
2148         if (req->rq_export) {
2149                 rc = ptlrpc_check_req(req);
2150                 if (rc == 0) {
2151                         rc = sptlrpc_target_export_check(req->rq_export, req);
2152                         if (rc)
2153                                 DEBUG_REQ(D_ERROR, req,
2154                                           "DROPPING req with illegal security flavor");
2155                 }
2156
2157                 if (rc)
2158                         goto err_req;
2159                 ptlrpc_update_export_timer(req->rq_export, 0);
2160         }
2161
2162         /* req_in handling should/must be fast */
2163         if (ktime_get_real_seconds() - req->rq_arrival_time.tv_sec > 5)
2164                 DEBUG_REQ(D_WARNING, req, "Slow req_in handling %llds",
2165                           (s64)(ktime_get_real_seconds() -
2166                                 req->rq_arrival_time.tv_sec));
2167
2168         /* Set rpc server deadline and add it to the timed list */
2169         deadline = (lustre_msghdr_get_flags(req->rq_reqmsg) &
2170                     MSGHDR_AT_SUPPORT) ?
2171                     /* The max time the client expects us to take */
2172                     lustre_msg_get_timeout(req->rq_reqmsg) : obd_timeout;
2173
2174         req->rq_deadline = req->rq_arrival_time.tv_sec + deadline;
2175         if (unlikely(deadline == 0)) {
2176                 DEBUG_REQ(D_ERROR, req, "Dropping request with 0 timeout");
2177                 goto err_req;
2178         }
2179
2180         /* Skip early reply */
2181         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_RESEND))
2182                 req->rq_deadline += obd_timeout;
2183
2184         req->rq_svc_thread = thread;
2185         if (thread != NULL) {
2186                 /*
2187                  * initialize request session, it is needed for request
2188                  * processing by target
2189                  */
2190                 rc = lu_context_init(&req->rq_session, LCT_SERVER_SESSION |
2191                                                        LCT_NOREF);
2192                 if (rc) {
2193                         CERROR("%s: failure to initialize session: rc = %d\n",
2194                                thread->t_name, rc);
2195                         goto err_req;
2196                 }
2197                 req->rq_session.lc_thread = thread;
2198                 lu_context_enter(&req->rq_session);
2199                 thread->t_env->le_ses = &req->rq_session;
2200         }
2201
2202         ptlrpc_at_add_timed(req);
2203
2204         /* Move it over to the request processing queue */
2205         rc = ptlrpc_server_request_add(svcpt, req);
2206         if (rc)
2207                 GOTO(err_req, rc);
2208
2209         wake_up(&svcpt->scp_waitq);
2210         RETURN(1);
2211
2212 err_req:
2213         ptlrpc_server_finish_request(svcpt, req);
2214
2215         RETURN(1);
2216 }
2217
2218 /**
2219  * Main incoming request handling logic.
2220  * Calls handler function from service to do actual processing.
2221  */
2222 static int ptlrpc_server_handle_request(struct ptlrpc_service_part *svcpt,
2223                                         struct ptlrpc_thread *thread)
2224 {
2225         struct ptlrpc_service *svc = svcpt->scp_service;
2226         struct ptlrpc_request *request;
2227         ktime_t work_start;
2228         ktime_t work_end;
2229         ktime_t arrived;
2230         s64 timediff_usecs;
2231         s64 arrived_usecs;
2232         int fail_opc = 0;
2233
2234         ENTRY;
2235
2236         request = ptlrpc_server_request_get(svcpt, false);
2237         if (request == NULL)
2238                 RETURN(0);
2239
2240         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT))
2241                 fail_opc = OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT;
2242         else if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
2243                 fail_opc = OBD_FAIL_PTLRPC_HPREQ_TIMEOUT;
2244
2245         if (unlikely(fail_opc)) {
2246                 if (request->rq_export && request->rq_ops)
2247                         OBD_FAIL_TIMEOUT(fail_opc, 4);
2248         }
2249
2250         ptlrpc_rqphase_move(request, RQ_PHASE_INTERPRET);
2251
2252         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DUMP_LOG))
2253                 libcfs_debug_dumplog();
2254
2255         work_start = ktime_get_real();
2256         arrived = timespec64_to_ktime(request->rq_arrival_time);
2257         timediff_usecs = ktime_us_delta(work_start, arrived);
2258         if (likely(svc->srv_stats != NULL)) {
2259                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQWAIT_CNTR,
2260                                     timediff_usecs);
2261                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQQDEPTH_CNTR,
2262                                     svcpt->scp_nreqs_incoming);
2263                 lprocfs_counter_add(svc->srv_stats, PTLRPC_REQACTIVE_CNTR,
2264                                     svcpt->scp_nreqs_active);
2265                 lprocfs_counter_add(svc->srv_stats, PTLRPC_TIMEOUT,
2266                                     at_get(&svcpt->scp_at_estimate));
2267         }
2268
2269         if (likely(request->rq_export)) {
2270                 if (unlikely(ptlrpc_check_req(request)))
2271                         goto put_conn;
2272                 ptlrpc_update_export_timer(request->rq_export,
2273                                            div_u64(timediff_usecs,
2274                                                    USEC_PER_SEC / 2));
2275         }
2276
2277         /*
2278          * Discard requests queued for longer than the deadline.
2279          * The deadline is increased if we send an early reply.
2280          */
2281         if (ktime_get_real_seconds() > request->rq_deadline) {
2282                 DEBUG_REQ(D_ERROR, request,
2283                           "Dropping timed-out request from %s: deadline %lld/%llds ago",
2284                           libcfs_id2str(request->rq_peer),
2285                           request->rq_deadline -
2286                           request->rq_arrival_time.tv_sec,
2287                           ktime_get_real_seconds() - request->rq_deadline);
2288                 goto put_conn;
2289         }
2290
2291         CDEBUG(D_RPCTRACE,
2292                "Handling RPC req@%p pname:cluuid+ref:pid:xid:nid:opc:job %s:%s+%d:%d:x%llu:%s:%d:%s\n",
2293                request, current_comm(),
2294                (request->rq_export ?
2295                 (char *)request->rq_export->exp_client_uuid.uuid : "0"),
2296                (request->rq_export ?
2297                 refcount_read(&request->rq_export->exp_handle.h_ref) : -99),
2298                lustre_msg_get_status(request->rq_reqmsg), request->rq_xid,
2299                libcfs_id2str(request->rq_peer),
2300                lustre_msg_get_opc(request->rq_reqmsg),
2301                lustre_msg_get_jobid(request->rq_reqmsg) ?: "");
2302
2303         if (lustre_msg_get_opc(request->rq_reqmsg) != OBD_PING)
2304                 CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_PAUSE_REQ, cfs_fail_val);
2305
2306         CDEBUG(D_NET, "got req %llu\n", request->rq_xid);
2307
2308         /* re-assign request and sesson thread to the current one */
2309         request->rq_svc_thread = thread;
2310         if (thread != NULL) {
2311                 LASSERT(request->rq_session.lc_thread == NULL);
2312                 request->rq_session.lc_thread = thread;
2313                 thread->t_env->le_ses = &request->rq_session;
2314         }
2315         svc->srv_ops.so_req_handler(request);
2316
2317         ptlrpc_rqphase_move(request, RQ_PHASE_COMPLETE);
2318
2319 put_conn:
2320         if (unlikely(ktime_get_real_seconds() > request->rq_deadline)) {
2321                 DEBUG_REQ(D_WARNING, request,
2322                           "Request took longer than estimated (%lld/%llds); client may timeout",
2323                           request->rq_deadline -
2324                           request->rq_arrival_time.tv_sec,
2325                           ktime_get_real_seconds() - request->rq_deadline);
2326         }
2327
2328         work_end = ktime_get_real();
2329         timediff_usecs = ktime_us_delta(work_end, work_start);
2330         arrived_usecs = ktime_us_delta(work_end, arrived);
2331         CDEBUG(D_RPCTRACE,
2332                "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",
2333                request, current_comm(),
2334                (request->rq_export ?
2335                (char *)request->rq_export->exp_client_uuid.uuid : "0"),
2336                (request->rq_export ?
2337                 refcount_read(&request->rq_export->exp_handle.h_ref) : -99),
2338                lustre_msg_get_status(request->rq_reqmsg),
2339                request->rq_xid,
2340                libcfs_id2str(request->rq_peer),
2341                lustre_msg_get_opc(request->rq_reqmsg),
2342                lustre_msg_get_jobid(request->rq_reqmsg) ?: "",
2343                timediff_usecs,
2344                arrived_usecs,
2345                (request->rq_repmsg ?
2346                lustre_msg_get_transno(request->rq_repmsg) :
2347                request->rq_transno),
2348                request->rq_status,
2349                (request->rq_repmsg ?
2350                lustre_msg_get_status(request->rq_repmsg) : -999));
2351         if (likely(svc->srv_stats != NULL && request->rq_reqmsg != NULL)) {
2352                 __u32 op = lustre_msg_get_opc(request->rq_reqmsg);
2353                 int opc = opcode_offset(op);
2354
2355                 if (opc > 0 && !(op == LDLM_ENQUEUE || op == MDS_REINT)) {
2356                         LASSERT(opc < LUSTRE_MAX_OPCODES);
2357                         lprocfs_counter_add(svc->srv_stats,
2358                                             opc + EXTRA_MAX_OPCODES,
2359                                             timediff_usecs);
2360                 }
2361         }
2362         if (unlikely(request->rq_early_count)) {
2363                 DEBUG_REQ(D_ADAPTTO, request,
2364                           "sent %d early replies before finishing in %llds",
2365                           request->rq_early_count,
2366                           div_u64(arrived_usecs, USEC_PER_SEC));
2367         }
2368
2369         ptlrpc_server_finish_active_request(svcpt, request);
2370
2371         RETURN(1);
2372 }
2373
2374 /**
2375  * An internal function to process a single reply state object.
2376  */
2377 static int ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
2378 {
2379         struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
2380         struct ptlrpc_service *svc = svcpt->scp_service;
2381         struct obd_export *exp;
2382         int nlocks;
2383         int been_handled;
2384
2385         ENTRY;
2386
2387         exp = rs->rs_export;
2388
2389         LASSERT(rs->rs_difficult);
2390         LASSERT(rs->rs_scheduled);
2391         LASSERT(list_empty(&rs->rs_list));
2392
2393         /*
2394          * The disk commit callback holds exp_uncommitted_replies_lock while it
2395          * iterates over newly committed replies, removing them from
2396          * exp_uncommitted_replies.  It then drops this lock and schedules the
2397          * replies it found for handling here.
2398          *
2399          * We can avoid contention for exp_uncommitted_replies_lock between the
2400          * HRT threads and further commit callbacks by checking rs_committed
2401          * which is set in the commit callback while it holds both
2402          * rs_lock and exp_uncommitted_reples.
2403          *
2404          * If we see rs_committed clear, the commit callback _may_ not have
2405          * handled this reply yet and we race with it to grab
2406          * exp_uncommitted_replies_lock before removing the reply from
2407          * exp_uncommitted_replies.  Note that if we lose the race and the
2408          * reply has already been removed, list_del_init() is a noop.
2409          *
2410          * If we see rs_committed set, we know the commit callback is handling,
2411          * or has handled this reply since store reordering might allow us to
2412          * see rs_committed set out of sequence.  But since this is done
2413          * holding rs_lock, we can be sure it has all completed once we hold
2414          * rs_lock, which we do right next.
2415          */
2416         if (!rs->rs_committed) {
2417                 /*
2418                  * if rs was commited, no need to convert locks, don't check
2419                  * rs_committed here because rs may never be added into
2420                  * exp_uncommitted_replies and this flag never be set, see
2421                  * target_send_reply()
2422                  */
2423                 if (rs->rs_convert_lock &&
2424                     rs->rs_transno > exp->exp_last_committed) {
2425                         struct ldlm_lock *lock;
2426                         struct ldlm_lock *ack_locks[RS_MAX_LOCKS] = { NULL };
2427
2428                         spin_lock(&rs->rs_lock);
2429                         if (rs->rs_convert_lock &&
2430                             rs->rs_transno > exp->exp_last_committed) {
2431                                 nlocks = rs->rs_nlocks;
2432                                 while (nlocks-- > 0) {
2433                                         /*
2434                                          * NB don't assume rs is always handled
2435                                          * by the same service thread (see
2436                                          * ptlrpc_hr_select, so REP-ACK hr may
2437                                          * race with trans commit, while the
2438                                          * latter will release locks, get locks
2439                                          * here early to convert to COS mode
2440                                          * safely.
2441                                          */
2442                                         lock = ldlm_handle2lock(
2443                                                         &rs->rs_locks[nlocks]);
2444                                         LASSERT(lock);
2445                                         ack_locks[nlocks] = lock;
2446                                         rs->rs_modes[nlocks] = LCK_COS;
2447                                 }
2448                                 nlocks = rs->rs_nlocks;
2449                                 rs->rs_convert_lock = 0;
2450                                 /*
2451                                  * clear rs_scheduled so that commit callback
2452                                  * can schedule again
2453                                  */
2454                                 rs->rs_scheduled = 0;
2455                                 spin_unlock(&rs->rs_lock);
2456
2457                                 while (nlocks-- > 0) {
2458                                         lock = ack_locks[nlocks];
2459                                         ldlm_lock_mode_downgrade(lock, LCK_COS);
2460                                         LDLM_LOCK_PUT(lock);
2461                                 }
2462                                 RETURN(0);
2463                         }
2464                         spin_unlock(&rs->rs_lock);
2465                 }
2466
2467                 spin_lock(&exp->exp_uncommitted_replies_lock);
2468                 list_del_init(&rs->rs_obd_list);
2469                 spin_unlock(&exp->exp_uncommitted_replies_lock);
2470         }
2471
2472         spin_lock(&exp->exp_lock);
2473         /* Noop if removed already */
2474         list_del_init(&rs->rs_exp_list);
2475         spin_unlock(&exp->exp_lock);
2476
2477         spin_lock(&rs->rs_lock);
2478
2479         been_handled = rs->rs_handled;
2480         rs->rs_handled = 1;
2481
2482         nlocks = rs->rs_nlocks; /* atomic "steal", but */
2483         rs->rs_nlocks = 0; /* locks still on rs_locks! */
2484
2485         if (nlocks == 0 && !been_handled) {
2486                 /*
2487                  * If we see this, we should already have seen the warning
2488                  * in mds_steal_ack_locks()
2489                  */
2490                 CDEBUG(D_HA,
2491                        "All locks stolen from rs %p x%lld.t%lld o%d NID %s\n",
2492                        rs, rs->rs_xid, rs->rs_transno, rs->rs_opc,
2493                        libcfs_nid2str(exp->exp_connection->c_peer.nid));
2494         }
2495
2496         if ((!been_handled && rs->rs_on_net) || nlocks > 0) {
2497                 spin_unlock(&rs->rs_lock);
2498
2499                 if (!been_handled && rs->rs_on_net) {
2500                         LNetMDUnlink(rs->rs_md_h);
2501                         /* Ignore return code; we're racing with completion */
2502                 }
2503
2504                 while (nlocks-- > 0)
2505                         ldlm_lock_decref(&rs->rs_locks[nlocks],
2506                                          rs->rs_modes[nlocks]);
2507
2508                 spin_lock(&rs->rs_lock);
2509         }
2510
2511         rs->rs_scheduled = 0;
2512         rs->rs_convert_lock = 0;
2513
2514         if (!rs->rs_on_net) {
2515                 /* Off the net */
2516                 spin_unlock(&rs->rs_lock);
2517
2518                 class_export_put(exp);
2519                 rs->rs_export = NULL;
2520                 ptlrpc_rs_decref(rs);
2521                 if (atomic_dec_and_test(&svcpt->scp_nreps_difficult) &&
2522                     svc->srv_is_stopping)
2523                         wake_up_all(&svcpt->scp_waitq);
2524                 RETURN(1);
2525         }
2526
2527         /* still on the net; callback will schedule */
2528         spin_unlock(&rs->rs_lock);
2529         RETURN(1);
2530 }
2531
2532
2533 static void ptlrpc_check_rqbd_pool(struct ptlrpc_service_part *svcpt)
2534 {
2535         int avail = svcpt->scp_nrqbds_posted;
2536         int low_water = test_req_buffer_pressure ? 0 :
2537                         svcpt->scp_service->srv_nbuf_per_group / 2;
2538
2539         /* NB I'm not locking; just looking. */
2540
2541         /*
2542          * CAVEAT EMPTOR: We might be allocating buffers here because we've
2543          * allowed the request history to grow out of control.  We could put a
2544          * sanity check on that here and cull some history if we need the
2545          * space.
2546          */
2547
2548         if (avail <= low_water)
2549                 ptlrpc_grow_req_bufs(svcpt, 1);
2550
2551         if (svcpt->scp_service->srv_stats) {
2552                 lprocfs_counter_add(svcpt->scp_service->srv_stats,
2553                                     PTLRPC_REQBUF_AVAIL_CNTR, avail);
2554         }
2555 }
2556
2557 static inline int ptlrpc_threads_enough(struct ptlrpc_service_part *svcpt)
2558 {
2559         return svcpt->scp_nreqs_active <
2560                svcpt->scp_nthrs_running - 1 -
2561                (svcpt->scp_service->srv_ops.so_hpreq_handler != NULL);
2562 }
2563
2564 /**
2565  * allowed to create more threads
2566  * user can call it w/o any lock but need to hold
2567  * ptlrpc_service_part::scp_lock to get reliable result
2568  */
2569 static inline int ptlrpc_threads_increasable(struct ptlrpc_service_part *svcpt)
2570 {
2571         return svcpt->scp_nthrs_running +
2572                svcpt->scp_nthrs_starting <
2573                svcpt->scp_service->srv_nthrs_cpt_limit;
2574 }
2575
2576 /**
2577  * too many requests and allowed to create more threads
2578  */
2579 static inline int ptlrpc_threads_need_create(struct ptlrpc_service_part *svcpt)
2580 {
2581         return !ptlrpc_threads_enough(svcpt) &&
2582                 ptlrpc_threads_increasable(svcpt);
2583 }
2584
2585 static inline int ptlrpc_thread_stopping(struct ptlrpc_thread *thread)
2586 {
2587         return thread_is_stopping(thread) ||
2588                thread->t_svcpt->scp_service->srv_is_stopping;
2589 }
2590
2591 /* stop the highest numbered thread if there are too many threads running */
2592 static inline bool ptlrpc_thread_should_stop(struct ptlrpc_thread *thread)
2593 {
2594         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2595
2596         return thread->t_id >= svcpt->scp_service->srv_nthrs_cpt_limit &&
2597                 thread->t_id == svcpt->scp_thr_nextid - 1;
2598 }
2599
2600 static void ptlrpc_stop_thread(struct ptlrpc_thread *thread)
2601 {
2602         CDEBUG(D_INFO, "Stopping thread %s #%u\n",
2603                thread->t_svcpt->scp_service->srv_thread_name, thread->t_id);
2604         thread_add_flags(thread, SVC_STOPPING);
2605 }
2606
2607 static inline void ptlrpc_thread_stop(struct ptlrpc_thread *thread)
2608 {
2609         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2610
2611         spin_lock(&svcpt->scp_lock);
2612         if (ptlrpc_thread_should_stop(thread)) {
2613                 ptlrpc_stop_thread(thread);
2614                 svcpt->scp_thr_nextid--;
2615         }
2616         spin_unlock(&svcpt->scp_lock);
2617 }
2618
2619 static inline int ptlrpc_rqbd_pending(struct ptlrpc_service_part *svcpt)
2620 {
2621         return !list_empty(&svcpt->scp_rqbd_idle) &&
2622                svcpt->scp_rqbd_timeout == 0;
2623 }
2624
2625 static inline int
2626 ptlrpc_at_check(struct ptlrpc_service_part *svcpt)
2627 {
2628         return svcpt->scp_at_check;
2629 }
2630
2631 /*
2632  * If a thread runs too long or spends to much time on a single request,
2633  * we want to know about it, so we set up a delayed work item as a watchdog.
2634  * If it fires, we display a stack trace of the delayed thread,
2635  * providing we aren't rate-limited
2636  *
2637  * Watchdog stack traces are limited to 3 per 'libcfs_watchdog_ratelimit'
2638  * seconds
2639  */
2640 static struct ratelimit_state watchdog_limit;
2641
2642 static void ptlrpc_watchdog_fire(struct work_struct *w)
2643 {
2644         struct ptlrpc_thread *thread = container_of(w, struct ptlrpc_thread,
2645                                                     t_watchdog.work);
2646         u64 ms_lapse = ktime_ms_delta(ktime_get(), thread->t_touched);
2647         u32 ms_frac = do_div(ms_lapse, MSEC_PER_SEC);
2648
2649         /* ___ratelimit() returns true if the action is NOT ratelimited */
2650         if (__ratelimit(&watchdog_limit)) {
2651                 /* below message is checked in sanity-quota.sh test_6,18 */
2652                 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",
2653                               thread->t_task->comm, thread->t_task->pid,
2654                               ms_lapse, ms_frac);
2655
2656                 libcfs_debug_dumpstack(thread->t_task);
2657         } else {
2658                 /* below message is checked in sanity-quota.sh test_6,18 */
2659                 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",
2660                               thread->t_task->comm, thread->t_task->pid,
2661                               ms_lapse, ms_frac, libcfs_watchdog_ratelimit);
2662         }
2663 }
2664
2665 static void ptlrpc_watchdog_init(struct delayed_work *work, time_t time)
2666 {
2667         INIT_DELAYED_WORK(work, ptlrpc_watchdog_fire);
2668         schedule_delayed_work(work, cfs_time_seconds(time));
2669 }
2670
2671 static void ptlrpc_watchdog_disable(struct delayed_work *work)
2672 {
2673         cancel_delayed_work_sync(work);
2674 }
2675
2676 static void ptlrpc_watchdog_touch(struct delayed_work *work, time_t time)
2677 {
2678         struct ptlrpc_thread *thread = container_of(&work->work,
2679                                                     struct ptlrpc_thread,
2680                                                     t_watchdog.work);
2681         thread->t_touched = ktime_get();
2682         mod_delayed_work(system_wq, work, cfs_time_seconds(time));
2683 }
2684
2685 /**
2686  * requests wait on preprocessing
2687  * user can call it w/o any lock but need to hold
2688  * ptlrpc_service_part::scp_lock to get reliable result
2689  */
2690 static inline int
2691 ptlrpc_server_request_incoming(struct ptlrpc_service_part *svcpt)
2692 {
2693         return !list_empty(&svcpt->scp_req_incoming);
2694 }
2695
2696 static __attribute__((__noinline__)) int
2697 ptlrpc_wait_event(struct ptlrpc_service_part *svcpt,
2698                   struct ptlrpc_thread *thread)
2699 {
2700         ptlrpc_watchdog_disable(&thread->t_watchdog);
2701
2702         cond_resched();
2703
2704         if (svcpt->scp_rqbd_timeout == 0)
2705                 /* Don't exit while there are replies to be handled */
2706                 wait_event_idle_exclusive_lifo(
2707                         svcpt->scp_waitq,
2708                         ptlrpc_thread_stopping(thread) ||
2709                         ptlrpc_server_request_incoming(svcpt) ||
2710                         ptlrpc_server_request_pending(svcpt, false) ||
2711                         ptlrpc_rqbd_pending(svcpt) ||
2712                         ptlrpc_at_check(svcpt));
2713         else if (wait_event_idle_exclusive_lifo_timeout(
2714                          svcpt->scp_waitq,
2715                          ptlrpc_thread_stopping(thread) ||
2716                          ptlrpc_server_request_incoming(svcpt) ||
2717                          ptlrpc_server_request_pending(svcpt, false) ||
2718                          ptlrpc_rqbd_pending(svcpt) ||
2719                          ptlrpc_at_check(svcpt),
2720                          svcpt->scp_rqbd_timeout) == 0)
2721                 svcpt->scp_rqbd_timeout = 0;
2722
2723         if (ptlrpc_thread_stopping(thread))
2724                 return -EINTR;
2725
2726         ptlrpc_watchdog_touch(&thread->t_watchdog,
2727                               ptlrpc_server_get_timeout(svcpt));
2728         return 0;
2729 }
2730
2731 /**
2732  * Main thread body for service threads.
2733  * Waits in a loop waiting for new requests to process to appear.
2734  * Every time an incoming requests is added to its queue, a waitq
2735  * is woken up and one of the threads will handle it.
2736  */
2737 static int ptlrpc_main(void *arg)
2738 {
2739         struct ptlrpc_thread *thread = (struct ptlrpc_thread *)arg;
2740         struct ptlrpc_service_part *svcpt = thread->t_svcpt;
2741         struct ptlrpc_service *svc = svcpt->scp_service;
2742         struct ptlrpc_reply_state *rs;
2743         struct group_info *ginfo = NULL;
2744         struct lu_env *env;
2745         int counter = 0, rc = 0;
2746
2747         ENTRY;
2748
2749         thread->t_task = current;
2750         thread->t_pid = current_pid();
2751         unshare_fs_struct();
2752
2753         if (svc->srv_cpt_bind) {
2754                 rc = cfs_cpt_bind(svc->srv_cptable, svcpt->scp_cpt);
2755                 if (rc != 0) {
2756                         CWARN("%s: failed to bind %s on CPT %d\n",
2757                               svc->srv_name, thread->t_name, svcpt->scp_cpt);
2758                 }
2759         }
2760
2761         ginfo = groups_alloc(0);
2762         if (!ginfo)
2763                 GOTO(out, rc = -ENOMEM);
2764
2765         set_current_groups(ginfo);
2766         put_group_info(ginfo);
2767
2768         if (svc->srv_ops.so_thr_init != NULL) {
2769                 rc = svc->srv_ops.so_thr_init(thread);
2770                 if (rc)
2771                         GOTO(out, rc);
2772         }
2773
2774         OBD_ALLOC_PTR(env);
2775         if (env == NULL)
2776                 GOTO(out_srv_fini, rc = -ENOMEM);
2777         rc = lu_env_add(env);
2778         if (rc)
2779                 GOTO(out_env, rc);
2780
2781         rc = lu_context_init(&env->le_ctx,
2782                              svc->srv_ctx_tags|LCT_REMEMBER|LCT_NOREF);
2783         if (rc)
2784                 GOTO(out_env_remove, rc);
2785
2786         thread->t_env = env;
2787         env->le_ctx.lc_thread = thread;
2788         env->le_ctx.lc_cookie = 0x6;
2789
2790         while (!list_empty(&svcpt->scp_rqbd_idle)) {
2791                 rc = ptlrpc_server_post_idle_rqbds(svcpt);
2792                 if (rc >= 0)
2793                         continue;
2794
2795                 CERROR("Failed to post rqbd for %s on CPT %d: %d\n",
2796                         svc->srv_name, svcpt->scp_cpt, rc);
2797                 GOTO(out_ctx_fini, rc);
2798         }
2799
2800         /* Alloc reply state structure for this one */
2801         OBD_ALLOC_LARGE(rs, svc->srv_max_reply_size);
2802         if (!rs)
2803                 GOTO(out_ctx_fini, rc = -ENOMEM);
2804
2805         spin_lock(&svcpt->scp_lock);
2806
2807         LASSERT(thread_is_starting(thread));
2808         thread_clear_flags(thread, SVC_STARTING);
2809
2810         LASSERT(svcpt->scp_nthrs_starting == 1);
2811         svcpt->scp_nthrs_starting--;
2812
2813         /*
2814          * SVC_STOPPING may already be set here if someone else is trying
2815          * to stop the service while this new thread has been dynamically
2816          * forked. We still set SVC_RUNNING to let our creator know that
2817          * we are now running, however we will exit as soon as possible
2818          */
2819         thread_add_flags(thread, SVC_RUNNING);
2820         svcpt->scp_nthrs_running++;
2821         spin_unlock(&svcpt->scp_lock);
2822
2823         /* wake up our creator in case he's still waiting. */
2824         wake_up(&thread->t_ctl_waitq);
2825
2826         thread->t_touched = ktime_get();
2827         ptlrpc_watchdog_init(&thread->t_watchdog,
2828                          ptlrpc_server_get_timeout(svcpt));
2829
2830         spin_lock(&svcpt->scp_rep_lock);
2831         list_add(&rs->rs_list, &svcpt->scp_rep_idle);
2832         wake_up(&svcpt->scp_rep_waitq);
2833         spin_unlock(&svcpt->scp_rep_lock);
2834
2835         CDEBUG(D_NET, "service thread %d (#%d) started\n", thread->t_id,
2836                svcpt->scp_nthrs_running);
2837
2838         /* XXX maintain a list of all managed devices: insert here */
2839         while (!ptlrpc_thread_stopping(thread)) {
2840                 if (ptlrpc_wait_event(svcpt, thread))
2841                         break;
2842
2843                 ptlrpc_check_rqbd_pool(svcpt);
2844
2845                 if (ptlrpc_threads_need_create(svcpt)) {
2846                         /* Ignore return code - we tried... */
2847                         ptlrpc_start_thread(svcpt, 0);
2848                 }
2849
2850                 /* reset le_ses to initial state */
2851                 env->le_ses = NULL;
2852                 /* Refill the context before execution to make sure
2853                  * all thread keys are allocated */
2854                 lu_env_refill(env);
2855                 /* Process all incoming reqs before handling any */
2856                 if (ptlrpc_server_request_incoming(svcpt)) {
2857                         lu_context_enter(&env->le_ctx);
2858                         ptlrpc_server_handle_req_in(svcpt, thread);
2859                         lu_context_exit(&env->le_ctx);
2860
2861                         /* but limit ourselves in case of flood */
2862                         if (counter++ < 100)
2863                                 continue;
2864                         counter = 0;
2865                 }
2866
2867                 if (ptlrpc_at_check(svcpt))
2868                         ptlrpc_at_check_timed(svcpt);
2869
2870                 if (ptlrpc_server_request_pending(svcpt, false)) {
2871                         lu_context_enter(&env->le_ctx);
2872                         ptlrpc_server_handle_request(svcpt, thread);
2873                         lu_context_exit(&env->le_ctx);
2874                 }
2875
2876                 if (ptlrpc_rqbd_pending(svcpt) &&
2877                     ptlrpc_server_post_idle_rqbds(svcpt) < 0) {
2878                         /*
2879                          * I just failed to repost request buffers.
2880                          * Wait for a timeout (unless something else
2881                          * happens) before I try again
2882                          */
2883                         svcpt->scp_rqbd_timeout = cfs_time_seconds(1) / 10;
2884                         CDEBUG(D_RPCTRACE, "Posted buffers: %d\n",
2885                                svcpt->scp_nrqbds_posted);
2886                 }
2887                 /*
2888                  * If the number of threads has been tuned downward and this
2889                  * thread should be stopped, then stop in reverse order so the
2890                  * the threads always have contiguous thread index values.
2891                  */
2892                 if (unlikely(ptlrpc_thread_should_stop(thread)))
2893                         ptlrpc_thread_stop(thread);
2894         }
2895
2896         ptlrpc_watchdog_disable(&thread->t_watchdog);
2897
2898 out_ctx_fini:
2899         lu_context_fini(&env->le_ctx);
2900 out_env_remove:
2901         lu_env_remove(env);
2902 out_env:
2903         OBD_FREE_PTR(env);
2904 out_srv_fini:
2905         /* deconstruct service thread state created by ptlrpc_start_thread() */
2906         if (svc->srv_ops.so_thr_done != NULL)
2907                 svc->srv_ops.so_thr_done(thread);
2908 out:
2909         CDEBUG(D_RPCTRACE, "%s: service thread [%p:%u] %d exiting: rc = %d\n",
2910                thread->t_name, thread, thread->t_pid, thread->t_id, rc);
2911         spin_lock(&svcpt->scp_lock);
2912         if (thread_test_and_clear_flags(thread, SVC_STARTING))
2913                 svcpt->scp_nthrs_starting--;
2914
2915         if (thread_test_and_clear_flags(thread, SVC_RUNNING)) {
2916                 /* must know immediately */
2917                 svcpt->scp_nthrs_running--;
2918         }
2919
2920         thread->t_id = rc;
2921         thread_add_flags(thread, SVC_STOPPED);
2922
2923         wake_up(&thread->t_ctl_waitq);
2924         spin_unlock(&svcpt->scp_lock);
2925
2926         return rc;
2927 }
2928
2929 static int hrt_dont_sleep(struct ptlrpc_hr_thread *hrt,
2930                           struct list_head *replies)
2931 {
2932         int result;
2933
2934         spin_lock(&hrt->hrt_lock);
2935
2936         list_splice_init(&hrt->hrt_queue, replies);
2937         result = ptlrpc_hr.hr_stopping || !list_empty(replies);
2938
2939         spin_unlock(&hrt->hrt_lock);
2940         return result;
2941 }
2942
2943 /**
2944  * Main body of "handle reply" function.
2945  * It processes acked reply states
2946  */
2947 static int ptlrpc_hr_main(void *arg)
2948 {
2949         struct ptlrpc_hr_thread *hrt = (struct ptlrpc_hr_thread *)arg;
2950         struct ptlrpc_hr_partition *hrp = hrt->hrt_partition;
2951         struct list_head replies;
2952         struct lu_env *env;
2953         int rc;
2954
2955         OBD_ALLOC_PTR(env);
2956         if (env == NULL)
2957                 RETURN(-ENOMEM);
2958
2959         INIT_LIST_HEAD(&replies);
2960         unshare_fs_struct();
2961
2962         rc = cfs_cpt_bind(ptlrpc_hr.hr_cpt_table, hrp->hrp_cpt);
2963         if (rc != 0) {
2964                 char threadname[20];
2965
2966                 snprintf(threadname, sizeof(threadname), "ptlrpc_hr%02d_%03d",
2967                          hrp->hrp_cpt, hrt->hrt_id);
2968                 CWARN("Failed to bind %s on CPT %d of CPT table %p: rc = %d\n",
2969                       threadname, hrp->hrp_cpt, ptlrpc_hr.hr_cpt_table, rc);
2970         }
2971
2972         rc = lu_context_init(&env->le_ctx, LCT_MD_THREAD | LCT_DT_THREAD |
2973                              LCT_REMEMBER | LCT_NOREF);
2974         if (rc)
2975                 GOTO(out_env, rc);
2976
2977         rc = lu_env_add(env);
2978         if (rc)
2979                 GOTO(out_ctx_fini, rc);
2980
2981         atomic_inc(&hrp->hrp_nstarted);
2982         wake_up(&ptlrpc_hr.hr_waitq);
2983
2984         while (!ptlrpc_hr.hr_stopping) {
2985                 l_wait_condition(hrt->hrt_waitq, hrt_dont_sleep(hrt, &replies));
2986
2987                 while (!list_empty(&replies)) {
2988                         struct ptlrpc_reply_state *rs;
2989
2990                         rs = list_entry(replies.prev,
2991                                         struct ptlrpc_reply_state,
2992                                         rs_list);
2993                         list_del_init(&rs->rs_list);
2994                         /* refill keys if needed */
2995                         lu_env_refill(env);
2996                         lu_context_enter(&env->le_ctx);
2997                         ptlrpc_handle_rs(rs);
2998                         lu_context_exit(&env->le_ctx);
2999                 }
3000         }
3001
3002         atomic_inc(&hrp->hrp_nstopped);
3003         wake_up(&ptlrpc_hr.hr_waitq);
3004
3005         lu_env_remove(env);
3006 out_ctx_fini:
3007         lu_context_fini(&env->le_ctx);
3008 out_env:
3009         OBD_FREE_PTR(env);
3010         return 0;
3011 }
3012
3013 static void ptlrpc_stop_hr_threads(void)
3014 {
3015         struct ptlrpc_hr_partition *hrp;
3016         int i;
3017         int j;
3018
3019         ptlrpc_hr.hr_stopping = 1;
3020
3021         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3022                 if (hrp->hrp_thrs == NULL)
3023                         continue; /* uninitialized */
3024                 for (j = 0; j < hrp->hrp_nthrs; j++)
3025                         wake_up_all(&hrp->hrp_thrs[j].hrt_waitq);
3026         }
3027
3028         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3029                 if (hrp->hrp_thrs == NULL)
3030                         continue; /* uninitialized */
3031                 wait_event(ptlrpc_hr.hr_waitq,
3032                                atomic_read(&hrp->hrp_nstopped) ==
3033                                atomic_read(&hrp->hrp_nstarted));
3034         }
3035 }
3036
3037 static int ptlrpc_start_hr_threads(void)
3038 {
3039         struct ptlrpc_hr_partition *hrp;
3040         int i;
3041         int j;
3042
3043         ENTRY;
3044
3045         cfs_percpt_for_each(hrp, i, ptlrpc_hr.hr_partitions) {
3046                 int     rc = 0;
3047
3048                 for (j = 0; j < hrp->hrp_nthrs; j++) {
3049                         struct ptlrpc_hr_thread *hrt = &hrp->hrp_thrs[j];
3050                         struct task_struct *task;
3051
3052                         task = kthread_run(ptlrpc_hr_main,
3053                                            &hrp->hrp_thrs[j],
3054                                            "ptlrpc_hr%02d_%03d",
3055                                            hrp->hrp_cpt,
3056                                            hrt->hrt_id);
3057                         if (IS_ERR(task)) {
3058                                 rc = PTR_ERR(task);
3059                                 break;
3060                         }
3061                 }
3062
3063                 wait_event(ptlrpc_hr.hr_waitq,
3064                            atomic_read(&hrp->hrp_nstarted) == j);
3065
3066                 if (rc < 0) {
3067                         CERROR("cannot start reply handler thread %d:%d: rc = %d\n",
3068                                i, j, rc);
3069                         ptlrpc_stop_hr_threads();
3070                         RETURN(rc);
3071                 }
3072         }
3073
3074         RETURN(0);
3075 }
3076
3077 static void ptlrpc_svcpt_stop_threads(struct ptlrpc_service_part *svcpt)
3078 {
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                 wait_event_idle(thread->t_ctl_waitq,
3107                                 thread_is_stopped(thread));
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 ptlrpc_thread *thread;
3176         struct ptlrpc_service *svc;
3177         struct task_struct *task;
3178         int rc;
3179
3180         ENTRY;
3181
3182         LASSERT(svcpt != NULL);
3183
3184         svc = svcpt->scp_service;
3185
3186         CDEBUG(D_RPCTRACE, "%s[%d] started %d min %d max %d\n",
3187                svc->srv_name, svcpt->scp_cpt, svcpt->scp_nthrs_running,
3188                svc->srv_nthrs_cpt_init, svc->srv_nthrs_cpt_limit);
3189
3190  again:
3191         if (unlikely(svc->srv_is_stopping))
3192                 RETURN(-ESRCH);
3193
3194         if (!ptlrpc_threads_increasable(svcpt) ||
3195             (OBD_FAIL_CHECK(OBD_FAIL_TGT_TOOMANY_THREADS) &&
3196              svcpt->scp_nthrs_running == svc->srv_nthrs_cpt_init - 1))
3197                 RETURN(-EMFILE);
3198
3199         OBD_CPT_ALLOC_PTR(thread, svc->srv_cptable, svcpt->scp_cpt);
3200         if (thread == NULL)
3201                 RETURN(-ENOMEM);
3202         init_waitqueue_head(&thread->t_ctl_waitq);
3203
3204         spin_lock(&svcpt->scp_lock);
3205         if (!ptlrpc_threads_increasable(svcpt)) {
3206                 spin_unlock(&svcpt->scp_lock);
3207                 OBD_FREE_PTR(thread);
3208                 RETURN(-EMFILE);
3209         }
3210
3211         if (svcpt->scp_nthrs_starting != 0) {
3212                 /*
3213                  * serialize starting because some modules (obdfilter)
3214                  * might require unique and contiguous t_id
3215                  */
3216                 LASSERT(svcpt->scp_nthrs_starting == 1);
3217                 spin_unlock(&svcpt->scp_lock);
3218                 OBD_FREE_PTR(thread);
3219                 if (wait) {
3220                         CDEBUG(D_INFO, "Waiting for creating thread %s #%d\n",
3221                                svc->srv_thread_name, svcpt->scp_thr_nextid);
3222                         schedule();
3223                         goto again;
3224                 }
3225
3226                 CDEBUG(D_INFO, "Creating thread %s #%d race, retry later\n",
3227                        svc->srv_thread_name, svcpt->scp_thr_nextid);
3228                 RETURN(-EAGAIN);
3229         }
3230
3231         svcpt->scp_nthrs_starting++;
3232         thread->t_id = svcpt->scp_thr_nextid++;
3233         thread_add_flags(thread, SVC_STARTING);
3234         thread->t_svcpt = svcpt;
3235
3236         list_add(&thread->t_link, &svcpt->scp_threads);
3237         spin_unlock(&svcpt->scp_lock);
3238
3239         if (svcpt->scp_cpt >= 0) {
3240                 snprintf(thread->t_name, PTLRPC_THR_NAME_LEN, "%s%02d_%03d",
3241                          svc->srv_thread_name, svcpt->scp_cpt, thread->t_id);
3242         } else {
3243                 snprintf(thread->t_name, PTLRPC_THR_NAME_LEN, "%s_%04d",
3244                          svc->srv_thread_name, thread->t_id);
3245         }
3246
3247         CDEBUG(D_RPCTRACE, "starting thread '%s'\n", thread->t_name);
3248         task = kthread_run(ptlrpc_main, thread, "%s", thread->t_name);
3249         if (IS_ERR(task)) {
3250                 rc = PTR_ERR(task);
3251                 CERROR("cannot start thread '%s': rc = %d\n",
3252                        thread->t_name, rc);
3253                 spin_lock(&svcpt->scp_lock);
3254                 --svcpt->scp_nthrs_starting;
3255                 if (thread_is_stopping(thread)) {
3256                         /*
3257                          * this ptlrpc_thread is being hanled
3258                          * by ptlrpc_svcpt_stop_threads now
3259                          */
3260                         thread_add_flags(thread, SVC_STOPPED);
3261                         wake_up(&thread->t_ctl_waitq);
3262                         spin_unlock(&svcpt->scp_lock);
3263                 } else {
3264                         list_del(&thread->t_link);
3265                         spin_unlock(&svcpt->scp_lock);
3266                         OBD_FREE_PTR(thread);
3267                 }
3268                 RETURN(rc);
3269