Whamcloud - gitweb
e6bbf18842e37a100c7ab23e265c6479e6513b45
[fs/lustre-release.git] / lustre / ptlrpc / ptlrpcd.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/ptlrpc/ptlrpcd.c
37  */
38
39 #define DEBUG_SUBSYSTEM S_RPC
40
41 #ifdef __KERNEL__
42 # include <libcfs/libcfs.h>
43 #else /* __KERNEL__ */
44 # include <liblustre.h>
45 # include <ctype.h>
46 #endif
47
48 #include <lustre_net.h>
49 # include <lustre_lib.h>
50
51 #include <lustre_ha.h>
52 #include <obd_class.h>   /* for obd_zombie */
53 #include <obd_support.h> /* for OBD_FAIL_CHECK */
54 #include <cl_object.h> /* cl_env_{get,put}() */
55 #include <lprocfs_status.h>
56
57 enum pscope_thread {
58         PT_NORMAL,
59         PT_RECOVERY,
60         PT_NR
61 };
62
63 struct ptlrpcd_scope_ctl {
64         struct ptlrpcd_thread {
65                 const char        *pt_name;
66                 struct ptlrpcd_ctl pt_ctl;
67         } pscope_thread[PT_NR];
68 };
69
70 static struct ptlrpcd_scope_ctl ptlrpcd_scopes[PSCOPE_NR] = {
71         [PSCOPE_BRW] = {
72                 .pscope_thread = {
73                         [PT_NORMAL] = {
74                                 .pt_name = "ptlrpcd-brw"
75                         },
76                         [PT_RECOVERY] = {
77                                 .pt_name = "ptlrpcd-brw-rcv"
78                         }
79                 }
80         },
81         [PSCOPE_OTHER] = {
82                 .pscope_thread = {
83                         [PT_NORMAL] = {
84                                 .pt_name = "ptlrpcd"
85                         },
86                         [PT_RECOVERY] = {
87                                 .pt_name = "ptlrpcd-rcv"
88                         }
89                 }
90         }
91 };
92
93 cfs_semaphore_t ptlrpcd_sem;
94 static int ptlrpcd_users = 0;
95
96 void ptlrpcd_wake(struct ptlrpc_request *req)
97 {
98         struct ptlrpc_request_set *rq_set = req->rq_set;
99
100         LASSERT(rq_set != NULL);
101
102         cfs_waitq_signal(&rq_set->set_waitq);
103 }
104
105 /*
106  * Move all request from an existing request set to the ptlrpcd queue.
107  * All requests from the set must be in phase RQ_PHASE_NEW.
108  */
109 void ptlrpcd_add_rqset(struct ptlrpc_request_set *set)
110 {
111         cfs_list_t *tmp, *pos;
112
113         cfs_list_for_each_safe(pos, tmp, &set->set_requests) {
114                 struct ptlrpc_request *req =
115                         cfs_list_entry(pos, struct ptlrpc_request,
116                                        rq_set_chain);
117
118                 LASSERT(req->rq_phase == RQ_PHASE_NEW);
119                 cfs_list_del_init(&req->rq_set_chain);
120                 req->rq_set = NULL;
121                 ptlrpcd_add_req(req, PSCOPE_OTHER);
122                 set->set_remaining--;
123         }
124         LASSERT(set->set_remaining == 0);
125 }
126 EXPORT_SYMBOL(ptlrpcd_add_rqset);
127
128 /*
129  * Requests that are added to the ptlrpcd queue are sent via
130  * ptlrpcd_check->ptlrpc_check_set().
131  */
132 int ptlrpcd_add_req(struct ptlrpc_request *req, enum ptlrpcd_scope scope)
133 {
134         struct ptlrpcd_ctl *pc;
135         enum pscope_thread  pt;
136         int rc;
137
138         LASSERT(scope < PSCOPE_NR);
139         pt = req->rq_send_state == LUSTRE_IMP_FULL ? PT_NORMAL : PT_RECOVERY;
140         pc = &ptlrpcd_scopes[scope].pscope_thread[pt].pt_ctl;
141         rc = ptlrpc_set_add_new_req(pc, req);
142         /*
143          * XXX disable this for CLIO: environment is needed for interpreter.
144          *     add debug temporary to check rc.
145          */
146         LASSERTF(rc == 0, "ptlrpcd_add_req failed (rc = %d)\n", rc);
147         if (rc && 0) {
148                 /*
149                  * Thread is probably in stop now so we need to
150                  * kill this rpc as it was not added. Let's call
151                  * interpret for it to let know we're killing it
152                  * so that higher levels might free associated
153                  * resources.
154                  */
155                 ptlrpc_req_interpret(NULL, req, -EBADR);
156                 req->rq_set = NULL;
157                 ptlrpc_req_finished(req);
158         } else if (req->rq_send_state == LUSTRE_IMP_CONNECTING) {
159                 /*
160                  * The request is for recovery, should be sent ASAP.
161                  */
162                 cfs_waitq_signal(&pc->pc_set->set_waitq);
163         }
164
165         return rc;
166 }
167
168 static int ptlrpcd_check(const struct lu_env *env, struct ptlrpcd_ctl *pc)
169 {
170         cfs_list_t *tmp, *pos;
171         struct ptlrpc_request *req;
172         int rc = 0;
173         ENTRY;
174
175         cfs_spin_lock(&pc->pc_set->set_new_req_lock);
176         cfs_list_for_each_safe(pos, tmp, &pc->pc_set->set_new_requests) {
177                 req = cfs_list_entry(pos, struct ptlrpc_request, rq_set_chain);
178                 cfs_list_del_init(&req->rq_set_chain);
179                 ptlrpc_set_add_req(pc->pc_set, req);
180                 /*
181                  * Need to calculate its timeout.
182                  */
183                 rc = 1;
184         }
185         cfs_spin_unlock(&pc->pc_set->set_new_req_lock);
186
187         if (pc->pc_set->set_remaining) {
188                 rc = rc | ptlrpc_check_set(env, pc->pc_set);
189
190                 /*
191                  * XXX: our set never completes, so we prune the completed
192                  * reqs after each iteration. boy could this be smarter.
193                  */
194                 cfs_list_for_each_safe(pos, tmp, &pc->pc_set->set_requests) {
195                         req = cfs_list_entry(pos, struct ptlrpc_request,
196                                          rq_set_chain);
197                         if (req->rq_phase != RQ_PHASE_COMPLETE)
198                                 continue;
199
200                         cfs_list_del_init(&req->rq_set_chain);
201                         req->rq_set = NULL;
202                         ptlrpc_req_finished (req);
203                 }
204         }
205
206         if (rc == 0) {
207                 /*
208                  * If new requests have been added, make sure to wake up.
209                  */
210                 cfs_spin_lock(&pc->pc_set->set_new_req_lock);
211                 rc = !cfs_list_empty(&pc->pc_set->set_new_requests);
212                 cfs_spin_unlock(&pc->pc_set->set_new_req_lock);
213         }
214
215         RETURN(rc);
216 }
217
218 #ifdef __KERNEL__
219 /*
220  * ptlrpc's code paths like to execute in process context, so we have this
221  * thread which spins on a set which contains the io rpcs. llite specifies
222  * ptlrpcd's set when it pushes pages down into the oscs.
223  */
224 static int ptlrpcd(void *arg)
225 {
226         struct ptlrpcd_ctl *pc = arg;
227         struct lu_env env = { .le_ses = NULL };
228         int rc, exit = 0;
229         ENTRY;
230
231         rc = cfs_daemonize_ctxt(pc->pc_name);
232         if (rc == 0) {
233                 /*
234                  * XXX So far only "client" ptlrpcd uses an environment. In
235                  * the future, ptlrpcd thread (or a thread-set) has to given
236                  * an argument, describing its "scope".
237                  */
238                 rc = lu_context_init(&env.le_ctx,
239                                      LCT_CL_THREAD|LCT_REMEMBER|LCT_NOREF);
240         }
241
242         cfs_complete(&pc->pc_starting);
243
244         if (rc != 0)
245                 RETURN(rc);
246         env.le_ctx.lc_cookie = 0x7;
247
248         /*
249          * This mainloop strongly resembles ptlrpc_set_wait() except that our
250          * set never completes.  ptlrpcd_check() calls ptlrpc_check_set() when
251          * there are requests in the set. New requests come in on the set's
252          * new_req_list and ptlrpcd_check() moves them into the set.
253          */
254         do {
255                 struct l_wait_info lwi;
256                 int timeout;
257
258                 rc = lu_env_refill(&env);
259                 if (rc != 0) {
260                         /*
261                          * XXX This is very awkward situation, because
262                          * execution can neither continue (request
263                          * interpreters assume that env is set up), nor repeat
264                          * the loop (as this potentially results in a tight
265                          * loop of -ENOMEM's).
266                          *
267                          * Fortunately, refill only ever does something when
268                          * new modules are loaded, i.e., early during boot up.
269                          */
270                         CERROR("Failure to refill session: %d\n", rc);
271                         continue;
272                 }
273
274                 timeout = ptlrpc_set_next_timeout(pc->pc_set);
275                 lwi = LWI_TIMEOUT(cfs_time_seconds(timeout ? timeout : 1),
276                                   ptlrpc_expired_set, pc->pc_set);
277
278                 lu_context_enter(&env.le_ctx);
279                 l_wait_event(pc->pc_set->set_waitq,
280                              ptlrpcd_check(&env, pc), &lwi);
281                 lu_context_exit(&env.le_ctx);
282
283                 /*
284                  * Abort inflight rpcs for forced stop case.
285                  */
286                 if (cfs_test_bit(LIOD_STOP, &pc->pc_flags)) {
287                         if (cfs_test_bit(LIOD_FORCE, &pc->pc_flags))
288                                 ptlrpc_abort_set(pc->pc_set);
289                         exit++;
290                 }
291
292                 /*
293                  * Let's make one more loop to make sure that ptlrpcd_check()
294                  * copied all raced new rpcs into the set so we can kill them.
295                  */
296         } while (exit < 2);
297
298         /*
299          * Wait for inflight requests to drain.
300          */
301         if (!cfs_list_empty(&pc->pc_set->set_requests))
302                 ptlrpc_set_wait(pc->pc_set);
303         lu_context_fini(&env.le_ctx);
304         cfs_complete(&pc->pc_finishing);
305
306         cfs_clear_bit(LIOD_START, &pc->pc_flags);
307         cfs_clear_bit(LIOD_STOP, &pc->pc_flags);
308         cfs_clear_bit(LIOD_FORCE, &pc->pc_flags);
309         return 0;
310 }
311
312 #else /* !__KERNEL__ */
313
314 int ptlrpcd_check_async_rpcs(void *arg)
315 {
316         struct ptlrpcd_ctl *pc = arg;
317         int                 rc = 0;
318
319         /*
320          * Single threaded!!
321          */
322         pc->pc_recurred++;
323
324         if (pc->pc_recurred == 1) {
325                 rc = lu_env_refill(&pc->pc_env);
326                 if (rc == 0) {
327                         lu_context_enter(&pc->pc_env.le_ctx);
328                         rc = ptlrpcd_check(&pc->pc_env, pc);
329                         lu_context_exit(&pc->pc_env.le_ctx);
330                         if (!rc)
331                                 ptlrpc_expired_set(pc->pc_set);
332                         /*
333                          * XXX: send replay requests.
334                          */
335                         if (cfs_test_bit(LIOD_RECOVERY, &pc->pc_flags))
336                                 rc = ptlrpcd_check(&pc->pc_env, pc);
337                 }
338         }
339
340         pc->pc_recurred--;
341         return rc;
342 }
343
344 int ptlrpcd_idle(void *arg)
345 {
346         struct ptlrpcd_ctl *pc = arg;
347
348         return (cfs_list_empty(&pc->pc_set->set_new_requests) &&
349                 pc->pc_set->set_remaining == 0);
350 }
351
352 #endif
353
354 int ptlrpcd_start(const char *name, struct ptlrpcd_ctl *pc)
355 {
356         int rc;
357         ENTRY;
358
359         /*
360          * Do not allow start second thread for one pc.
361          */
362         if (cfs_test_and_set_bit(LIOD_START, &pc->pc_flags)) {
363                 CERROR("Starting second thread (%s) for same pc %p\n",
364                        name, pc);
365                 RETURN(-EALREADY);
366         }
367
368         cfs_init_completion(&pc->pc_starting);
369         cfs_init_completion(&pc->pc_finishing);
370         cfs_spin_lock_init(&pc->pc_lock);
371         strncpy(pc->pc_name, name, sizeof(pc->pc_name) - 1);
372         pc->pc_set = ptlrpc_prep_set();
373         if (pc->pc_set == NULL)
374                 GOTO(out, rc = -ENOMEM);
375         /*
376          * So far only "client" ptlrpcd uses an environment. In the future,
377          * ptlrpcd thread (or a thread-set) has to be given an argument,
378          * describing its "scope".
379          */
380         rc = lu_context_init(&pc->pc_env.le_ctx, LCT_CL_THREAD|LCT_REMEMBER);
381         if (rc != 0) {
382                 ptlrpc_set_destroy(pc->pc_set);
383                 GOTO(out, rc);
384         }
385
386 #ifdef __KERNEL__
387         rc = cfs_kernel_thread(ptlrpcd, pc, 0);
388         if (rc < 0)  {
389                 lu_context_fini(&pc->pc_env.le_ctx);
390                 ptlrpc_set_destroy(pc->pc_set);
391                 GOTO(out, rc);
392         }
393         rc = 0;
394         cfs_wait_for_completion(&pc->pc_starting);
395 #else
396         pc->pc_wait_callback =
397                 liblustre_register_wait_callback("ptlrpcd_check_async_rpcs",
398                                                  &ptlrpcd_check_async_rpcs, pc);
399         pc->pc_idle_callback =
400                 liblustre_register_idle_callback("ptlrpcd_check_idle_rpcs",
401                                                  &ptlrpcd_idle, pc);
402 #endif
403 out:
404         if (rc)
405                 cfs_clear_bit(LIOD_START, &pc->pc_flags);
406         RETURN(rc);
407 }
408
409 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force)
410 {
411         if (!cfs_test_bit(LIOD_START, &pc->pc_flags)) {
412                 CERROR("Thread for pc %p was not started\n", pc);
413                 return;
414         }
415
416         cfs_set_bit(LIOD_STOP, &pc->pc_flags);
417         if (force)
418                 cfs_set_bit(LIOD_FORCE, &pc->pc_flags);
419         cfs_waitq_signal(&pc->pc_set->set_waitq);
420 #ifdef __KERNEL__
421         cfs_wait_for_completion(&pc->pc_finishing);
422 #else
423         liblustre_deregister_wait_callback(pc->pc_wait_callback);
424         liblustre_deregister_idle_callback(pc->pc_idle_callback);
425 #endif
426         lu_context_fini(&pc->pc_env.le_ctx);
427         ptlrpc_set_destroy(pc->pc_set);
428 }
429
430 void ptlrpcd_fini(void)
431 {
432         int i;
433         int j;
434
435         ENTRY;
436
437         for (i = 0; i < PSCOPE_NR; ++i) {
438                 for (j = 0; j < PT_NR; ++j) {
439                         struct ptlrpcd_ctl *pc;
440
441                         pc = &ptlrpcd_scopes[i].pscope_thread[j].pt_ctl;
442
443                         if (cfs_test_bit(LIOD_START, &pc->pc_flags))
444                                 ptlrpcd_stop(pc, 0);
445                 }
446         }
447         EXIT;
448 }
449
450 int ptlrpcd_addref(void)
451 {
452         int rc = 0;
453         int i;
454         int j;
455         ENTRY;
456
457         cfs_mutex_down(&ptlrpcd_sem);
458         if (++ptlrpcd_users == 1) {
459                 for (i = 0; rc == 0 && i < PSCOPE_NR; ++i) {
460                         for (j = 0; rc == 0 && j < PT_NR; ++j) {
461                                 struct ptlrpcd_thread *pt;
462                                 struct ptlrpcd_ctl    *pc;
463
464                                 pt = &ptlrpcd_scopes[i].pscope_thread[j];
465                                 pc = &pt->pt_ctl;
466                                 if (j == PT_RECOVERY)
467                                         cfs_set_bit(LIOD_RECOVERY, &pc->pc_flags);
468                                 rc = ptlrpcd_start(pt->pt_name, pc);
469                         }
470                 }
471                 if (rc != 0) {
472                         --ptlrpcd_users;
473                         ptlrpcd_fini();
474                 }
475         }
476         cfs_mutex_up(&ptlrpcd_sem);
477         RETURN(rc);
478 }
479
480 void ptlrpcd_decref(void)
481 {
482         cfs_mutex_down(&ptlrpcd_sem);
483         if (--ptlrpcd_users == 0)
484                 ptlrpcd_fini();
485         cfs_mutex_up(&ptlrpcd_sem);
486 }