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