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