Whamcloud - gitweb
LU-12501 utils: fix 'lfs df' printing loop
[fs/lustre-release.git] / libcfs / libcfs / workitem.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) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2013, 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  * libcfs/libcfs/workitem.c
33  *
34  * Author: Isaac Huang <isaac@clusterfs.com>
35  *         Liang Zhen  <zhen.liang@sun.com>
36  */
37
38 #define DEBUG_SUBSYSTEM S_LNET
39
40 #include <linux/kthread.h>
41 #include <libcfs/libcfs.h>
42
43 #define CFS_WS_NAME_LEN         16
44
45 struct cfs_wi_sched {
46         struct list_head                ws_list;        /* chain on global list */
47         /** serialised workitems */
48         spinlock_t                      ws_lock;
49         /** where schedulers sleep */
50         wait_queue_head_t               ws_waitq;
51         /** concurrent workitems */
52         struct list_head                ws_runq;
53         /** rescheduled running-workitems, a workitem can be rescheduled
54          * while running in wi_action(), but we don't to execute it again
55          * unless it returns from wi_action(), so we put it on ws_rerunq
56          * while rescheduling, and move it to runq after it returns
57          * from wi_action() */
58         struct list_head                ws_rerunq;
59         /** CPT-table for this scheduler */
60         struct cfs_cpt_table    *ws_cptab;
61         /** CPT id for affinity */
62         int                     ws_cpt;
63         /** number of scheduled workitems */
64         int                     ws_nscheduled;
65         /** started scheduler thread, protected by cfs_wi_data::wi_glock */
66         unsigned int            ws_nthreads:30;
67         /** shutting down, protected by cfs_wi_data::wi_glock */
68         unsigned int            ws_stopping:1;
69         /** serialize starting thread, protected by cfs_wi_data::wi_glock */
70         unsigned int            ws_starting:1;
71         /** scheduler name */
72         char                    ws_name[CFS_WS_NAME_LEN];
73 };
74
75 static struct cfs_workitem_data {
76         /** serialize */
77         spinlock_t              wi_glock;
78         /** list of all schedulers */
79         struct list_head        wi_scheds;
80         /** WI module is initialized */
81         int                     wi_init;
82         /** shutting down the whole WI module */
83         int                     wi_stopping;
84 } cfs_wi_data;
85
86 static inline int
87 cfs_wi_sched_cansleep(struct cfs_wi_sched *sched)
88 {
89         spin_lock(&sched->ws_lock);
90         if (sched->ws_stopping) {
91                 spin_unlock(&sched->ws_lock);
92                 return 0;
93         }
94
95         if (!list_empty(&sched->ws_runq)) {
96                 spin_unlock(&sched->ws_lock);
97                 return 0;
98         }
99         spin_unlock(&sched->ws_lock);
100         return 1;
101 }
102
103 /* XXX:
104  * 0. it only works when called from wi->wi_action.
105  * 1. when it returns no one shall try to schedule the workitem.
106  */
107 void
108 cfs_wi_exit(struct cfs_wi_sched *sched, struct cfs_workitem *wi)
109 {
110         LASSERT(!in_interrupt()); /* because we use plain spinlock */
111         LASSERT(!sched->ws_stopping);
112
113         spin_lock(&sched->ws_lock);
114
115         LASSERT(wi->wi_running);
116
117         if (wi->wi_scheduled) { /* cancel pending schedules */
118                 LASSERT(!list_empty(&wi->wi_list));
119                 list_del_init(&wi->wi_list);
120
121                 LASSERT(sched->ws_nscheduled > 0);
122                 sched->ws_nscheduled--;
123         }
124
125         LASSERT(list_empty(&wi->wi_list));
126
127         wi->wi_scheduled = 1; /* LBUG future schedule attempts */
128         spin_unlock(&sched->ws_lock);
129
130         return;
131 }
132 EXPORT_SYMBOL(cfs_wi_exit);
133
134 /**
135  * cancel schedule request of workitem \a wi
136  */
137 int
138 cfs_wi_deschedule(struct cfs_wi_sched *sched, struct cfs_workitem *wi)
139 {
140         int     rc;
141
142         LASSERT(!in_interrupt()); /* because we use plain spinlock */
143         LASSERT(!sched->ws_stopping);
144
145         /*
146          * return 0 if it's running already, otherwise return 1, which
147          * means the workitem will not be scheduled and will not have
148          * any race with wi_action.
149          */
150         spin_lock(&sched->ws_lock);
151
152         rc = !(wi->wi_running);
153
154         if (wi->wi_scheduled) { /* cancel pending schedules */
155                 LASSERT(!list_empty(&wi->wi_list));
156                 list_del_init(&wi->wi_list);
157
158                 LASSERT(sched->ws_nscheduled > 0);
159                 sched->ws_nscheduled--;
160
161                 wi->wi_scheduled = 0;
162         }
163
164         LASSERT (list_empty(&wi->wi_list));
165
166         spin_unlock(&sched->ws_lock);
167         return rc;
168 }
169 EXPORT_SYMBOL(cfs_wi_deschedule);
170
171 /*
172  * Workitem scheduled with (serial == 1) is strictly serialised not only with
173  * itself, but also with others scheduled this way.
174  *
175  * Now there's only one static serialised queue, but in the future more might
176  * be added, and even dynamic creation of serialised queues might be supported.
177  */
178 void
179 cfs_wi_schedule(struct cfs_wi_sched *sched, struct cfs_workitem *wi)
180 {
181         LASSERT(!in_interrupt()); /* because we use plain spinlock */
182         LASSERT(!sched->ws_stopping);
183
184         spin_lock(&sched->ws_lock);
185
186         if (!wi->wi_scheduled) {
187                 LASSERT (list_empty(&wi->wi_list));
188
189                 wi->wi_scheduled = 1;
190                 sched->ws_nscheduled++;
191                 if (!wi->wi_running) {
192                         list_add_tail(&wi->wi_list, &sched->ws_runq);
193                         wake_up(&sched->ws_waitq);
194                 } else {
195                         list_add(&wi->wi_list, &sched->ws_rerunq);
196                 }
197         }
198
199         LASSERT (!list_empty(&wi->wi_list));
200         spin_unlock(&sched->ws_lock);
201         return;
202 }
203 EXPORT_SYMBOL(cfs_wi_schedule);
204
205 static int
206 cfs_wi_scheduler(void *arg)
207 {
208         struct cfs_wi_sched *sched = (struct cfs_wi_sched *)arg;
209
210         cfs_block_allsigs();
211
212         /* CPT affinity scheduler? */
213         if (sched->ws_cptab != NULL)
214                 if (cfs_cpt_bind(sched->ws_cptab, sched->ws_cpt) != 0)
215                         CWARN("Unable to bind %s on CPU partition %d\n",
216                                 sched->ws_name, sched->ws_cpt);
217
218         spin_lock(&cfs_wi_data.wi_glock);
219
220         LASSERT(sched->ws_starting == 1);
221         sched->ws_starting--;
222         sched->ws_nthreads++;
223
224         spin_unlock(&cfs_wi_data.wi_glock);
225
226         spin_lock(&sched->ws_lock);
227
228         while (!sched->ws_stopping) {
229                 int             nloops = 0;
230                 int             rc;
231                 struct cfs_workitem *wi;
232
233                 while (!list_empty(&sched->ws_runq) &&
234                        nloops < CFS_WI_RESCHED) {
235                         wi = list_entry(sched->ws_runq.next,
236                                         struct cfs_workitem, wi_list);
237                         LASSERT(wi->wi_scheduled && !wi->wi_running);
238
239                         list_del_init(&wi->wi_list);
240
241                         LASSERT(sched->ws_nscheduled > 0);
242                         sched->ws_nscheduled--;
243
244                         wi->wi_running   = 1;
245                         wi->wi_scheduled = 0;
246
247                         spin_unlock(&sched->ws_lock);
248                         nloops++;
249
250                         rc = (*wi->wi_action) (wi);
251
252                         spin_lock(&sched->ws_lock);
253                         if (rc != 0) /* WI should be dead, even be freed! */
254                                 continue;
255
256                         wi->wi_running = 0;
257                         if (list_empty(&wi->wi_list))
258                                 continue;
259
260                         LASSERT(wi->wi_scheduled);
261                         /* wi is rescheduled, should be on rerunq now, we
262                          * move it to runq so it can run action now */
263                         list_move_tail(&wi->wi_list, &sched->ws_runq);
264                 }
265
266                 if (!list_empty(&sched->ws_runq)) {
267                         spin_unlock(&sched->ws_lock);
268                         /* don't sleep because some workitems still
269                          * expect me to come back soon */
270                         cond_resched();
271                         spin_lock(&sched->ws_lock);
272                         continue;
273                 }
274
275                 spin_unlock(&sched->ws_lock);
276                 rc = wait_event_interruptible_exclusive(sched->ws_waitq,
277                                 !cfs_wi_sched_cansleep(sched));
278                 spin_lock(&sched->ws_lock);
279         }
280
281         spin_unlock(&sched->ws_lock);
282
283         spin_lock(&cfs_wi_data.wi_glock);
284         sched->ws_nthreads--;
285         spin_unlock(&cfs_wi_data.wi_glock);
286
287         return 0;
288 }
289
290 void
291 cfs_wi_sched_destroy(struct cfs_wi_sched *sched)
292 {
293         LASSERT(cfs_wi_data.wi_init);
294         LASSERT(!cfs_wi_data.wi_stopping);
295
296         spin_lock(&cfs_wi_data.wi_glock);
297         if (sched->ws_stopping) {
298                 CDEBUG(D_INFO, "%s is in progress of stopping\n",
299                        sched->ws_name);
300                 spin_unlock(&cfs_wi_data.wi_glock);
301                 return;
302         }
303
304         LASSERT(!list_empty(&sched->ws_list));
305         sched->ws_stopping = 1;
306
307         spin_unlock(&cfs_wi_data.wi_glock);
308
309         wake_up_all(&sched->ws_waitq);
310
311         spin_lock(&cfs_wi_data.wi_glock);
312         {
313                 int i = 2;
314
315                 while (sched->ws_nthreads > 0) {
316                         CDEBUG(is_power_of_2(++i / 20) ? D_WARNING : D_NET,
317                                "waiting %us for %d %s worker threads to exit\n",
318                                i / 20, sched->ws_nthreads, sched->ws_name);
319
320                         spin_unlock(&cfs_wi_data.wi_glock);
321                         set_current_state(TASK_UNINTERRUPTIBLE);
322                         schedule_timeout(cfs_time_seconds(1) / 20);
323                         spin_lock(&cfs_wi_data.wi_glock);
324                 }
325         }
326
327         list_del(&sched->ws_list);
328
329         spin_unlock(&cfs_wi_data.wi_glock);
330
331         LASSERT(sched->ws_nscheduled == 0);
332
333         LIBCFS_FREE(sched, sizeof(*sched));
334 }
335 EXPORT_SYMBOL(cfs_wi_sched_destroy);
336
337 int
338 cfs_wi_sched_create(char *name, struct cfs_cpt_table *cptab,
339                     int cpt, int nthrs, struct cfs_wi_sched **sched_pp)
340 {
341         struct cfs_wi_sched     *sched;
342
343         LASSERT(cfs_wi_data.wi_init);
344         LASSERT(!cfs_wi_data.wi_stopping);
345         LASSERT(cptab == NULL || cpt == CFS_CPT_ANY ||
346                 (cpt >= 0 && cpt < cfs_cpt_number(cptab)));
347
348         LIBCFS_ALLOC(sched, sizeof(*sched));
349         if (sched == NULL)
350                 return -ENOMEM;
351
352         if (strlen(name) > sizeof(sched->ws_name)-1) {
353                 LIBCFS_FREE(sched, sizeof(*sched));
354                 return -E2BIG;
355         }
356         strlcpy(sched->ws_name, name, sizeof(sched->ws_name));
357
358         sched->ws_cptab = cptab;
359         sched->ws_cpt = cpt;
360
361         spin_lock_init(&sched->ws_lock);
362         init_waitqueue_head(&sched->ws_waitq);
363
364         INIT_LIST_HEAD(&sched->ws_runq);
365         INIT_LIST_HEAD(&sched->ws_rerunq);
366         INIT_LIST_HEAD(&sched->ws_list);
367
368         for (; nthrs > 0; nthrs--)  {
369                 char                    name[16];
370                 struct task_struct      *task;
371
372                 spin_lock(&cfs_wi_data.wi_glock);
373                 while (sched->ws_starting > 0) {
374                         spin_unlock(&cfs_wi_data.wi_glock);
375                         schedule();
376                         spin_lock(&cfs_wi_data.wi_glock);
377                 }
378
379                 sched->ws_starting++;
380                 spin_unlock(&cfs_wi_data.wi_glock);
381
382                 if (sched->ws_cptab != NULL && sched->ws_cpt >= 0) {
383                         snprintf(name, sizeof(name), "%s_%02d_%02d",
384                                  sched->ws_name, sched->ws_cpt,
385                                  sched->ws_nthreads);
386                 } else {
387                         snprintf(name, sizeof(name), "%s_%02d",
388                                  sched->ws_name, sched->ws_nthreads);
389                 }
390
391                 task = kthread_run(cfs_wi_scheduler, sched, name);
392                 if (IS_ERR(task)) {
393                         int rc = PTR_ERR(task);
394
395                         CERROR("Failed to create thread for "
396                                 "WI scheduler %s: %d\n", name, rc);
397
398                         spin_lock(&cfs_wi_data.wi_glock);
399
400                         /* make up for cfs_wi_sched_destroy */
401                         list_add(&sched->ws_list, &cfs_wi_data.wi_scheds);
402                         sched->ws_starting--;
403
404                         spin_unlock(&cfs_wi_data.wi_glock);
405
406                         cfs_wi_sched_destroy(sched);
407                         return rc;
408                 }
409         }
410
411         spin_lock(&cfs_wi_data.wi_glock);
412         list_add(&sched->ws_list, &cfs_wi_data.wi_scheds);
413         spin_unlock(&cfs_wi_data.wi_glock);
414
415         *sched_pp = sched;
416         return 0;
417 }
418 EXPORT_SYMBOL(cfs_wi_sched_create);
419
420 int
421 cfs_wi_startup(void)
422 {
423         memset(&cfs_wi_data, 0, sizeof(struct cfs_workitem_data));
424
425         spin_lock_init(&cfs_wi_data.wi_glock);
426         INIT_LIST_HEAD(&cfs_wi_data.wi_scheds);
427         cfs_wi_data.wi_init = 1;
428
429         return 0;
430 }
431
432 void
433 cfs_wi_shutdown (void)
434 {
435         struct cfs_wi_sched     *sched;
436
437         spin_lock(&cfs_wi_data.wi_glock);
438         cfs_wi_data.wi_stopping = 1;
439         spin_unlock(&cfs_wi_data.wi_glock);
440
441         /* nobody should contend on this list */
442         list_for_each_entry(sched, &cfs_wi_data.wi_scheds, ws_list) {
443                 sched->ws_stopping = 1;
444                 wake_up_all(&sched->ws_waitq);
445         }
446
447         list_for_each_entry(sched, &cfs_wi_data.wi_scheds, ws_list) {
448                 spin_lock(&cfs_wi_data.wi_glock);
449
450                 while (sched->ws_nthreads != 0) {
451                         spin_unlock(&cfs_wi_data.wi_glock);
452                         set_current_state(TASK_UNINTERRUPTIBLE);
453                         schedule_timeout(cfs_time_seconds(1) / 20);
454                         spin_lock(&cfs_wi_data.wi_glock);
455                 }
456                 spin_unlock(&cfs_wi_data.wi_glock);
457         }
458
459         while (!list_empty(&cfs_wi_data.wi_scheds)) {
460                 sched = list_entry(cfs_wi_data.wi_scheds.next,
461                                        struct cfs_wi_sched, ws_list);
462                 list_del(&sched->ws_list);
463                 LIBCFS_FREE(sched, sizeof(*sched));
464         }
465
466         cfs_wi_data.wi_stopping = 0;
467         cfs_wi_data.wi_init = 0;
468 }