[PATCH] knfsd: add a callback for when last rpc thread finishes
[linux-2.6.git] / net / sunrpc / svc.c
1 /*
2  * linux/net/sunrpc/svc.c
3  *
4  * High-level RPC service routines
5  *
6  * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
7  */
8
9 #include <linux/linkage.h>
10 #include <linux/sched.h>
11 #include <linux/errno.h>
12 #include <linux/net.h>
13 #include <linux/in.h>
14 #include <linux/mm.h>
15
16 #include <linux/sunrpc/types.h>
17 #include <linux/sunrpc/xdr.h>
18 #include <linux/sunrpc/stats.h>
19 #include <linux/sunrpc/svcsock.h>
20 #include <linux/sunrpc/clnt.h>
21
22 #define RPCDBG_FACILITY RPCDBG_SVCDSP
23 #define RPC_PARANOIA 1
24
25 /*
26  * Create an RPC service
27  */
28 struct svc_serv *
29 svc_create(struct svc_program *prog, unsigned int bufsize,
30            void (*shutdown)(struct svc_serv *serv))
31 {
32         struct svc_serv *serv;
33         int vers;
34         unsigned int xdrsize;
35
36         if (!(serv = kzalloc(sizeof(*serv), GFP_KERNEL)))
37                 return NULL;
38         serv->sv_name      = prog->pg_name;
39         serv->sv_program   = prog;
40         serv->sv_nrthreads = 1;
41         serv->sv_stats     = prog->pg_stats;
42         serv->sv_bufsz     = bufsize? bufsize : 4096;
43         serv->sv_shutdown  = shutdown;
44         xdrsize = 0;
45         while (prog) {
46                 prog->pg_lovers = prog->pg_nvers-1;
47                 for (vers=0; vers<prog->pg_nvers ; vers++)
48                         if (prog->pg_vers[vers]) {
49                                 prog->pg_hivers = vers;
50                                 if (prog->pg_lovers > vers)
51                                         prog->pg_lovers = vers;
52                                 if (prog->pg_vers[vers]->vs_xdrsize > xdrsize)
53                                         xdrsize = prog->pg_vers[vers]->vs_xdrsize;
54                         }
55                 prog = prog->pg_next;
56         }
57         serv->sv_xdrsize   = xdrsize;
58         INIT_LIST_HEAD(&serv->sv_threads);
59         INIT_LIST_HEAD(&serv->sv_sockets);
60         INIT_LIST_HEAD(&serv->sv_tempsocks);
61         INIT_LIST_HEAD(&serv->sv_permsocks);
62         spin_lock_init(&serv->sv_lock);
63
64         /* Remove any stale portmap registrations */
65         svc_register(serv, 0, 0);
66
67         return serv;
68 }
69
70 /*
71  * Destroy an RPC service
72  */
73 void
74 svc_destroy(struct svc_serv *serv)
75 {
76         struct svc_sock *svsk;
77
78         dprintk("RPC: svc_destroy(%s, %d)\n",
79                                 serv->sv_program->pg_name,
80                                 serv->sv_nrthreads);
81
82         if (serv->sv_nrthreads) {
83                 if (--(serv->sv_nrthreads) != 0) {
84                         svc_sock_update_bufs(serv);
85                         return;
86                 }
87         } else
88                 printk("svc_destroy: no threads for serv=%p!\n", serv);
89
90         while (!list_empty(&serv->sv_tempsocks)) {
91                 svsk = list_entry(serv->sv_tempsocks.next,
92                                   struct svc_sock,
93                                   sk_list);
94                 svc_delete_socket(svsk);
95         }
96         if (serv->sv_shutdown)
97                 serv->sv_shutdown(serv);
98
99         while (!list_empty(&serv->sv_permsocks)) {
100                 svsk = list_entry(serv->sv_permsocks.next,
101                                   struct svc_sock,
102                                   sk_list);
103                 svc_delete_socket(svsk);
104         }
105         
106         cache_clean_deferred(serv);
107
108         /* Unregister service with the portmapper */
109         svc_register(serv, 0, 0);
110         kfree(serv);
111 }
112
113 /*
114  * Allocate an RPC server's buffer space.
115  * We allocate pages and place them in rq_argpages.
116  */
117 static int
118 svc_init_buffer(struct svc_rqst *rqstp, unsigned int size)
119 {
120         int pages;
121         int arghi;
122         
123         if (size > RPCSVC_MAXPAYLOAD)
124                 size = RPCSVC_MAXPAYLOAD;
125         pages = 2 + (size+ PAGE_SIZE -1) / PAGE_SIZE;
126         rqstp->rq_argused = 0;
127         rqstp->rq_resused = 0;
128         arghi = 0;
129         BUG_ON(pages > RPCSVC_MAXPAGES);
130         while (pages) {
131                 struct page *p = alloc_page(GFP_KERNEL);
132                 if (!p)
133                         break;
134                 rqstp->rq_argpages[arghi++] = p;
135                 pages--;
136         }
137         rqstp->rq_arghi = arghi;
138         return ! pages;
139 }
140
141 /*
142  * Release an RPC server buffer
143  */
144 static void
145 svc_release_buffer(struct svc_rqst *rqstp)
146 {
147         while (rqstp->rq_arghi)
148                 put_page(rqstp->rq_argpages[--rqstp->rq_arghi]);
149         while (rqstp->rq_resused) {
150                 if (rqstp->rq_respages[--rqstp->rq_resused] == NULL)
151                         continue;
152                 put_page(rqstp->rq_respages[rqstp->rq_resused]);
153         }
154         rqstp->rq_argused = 0;
155 }
156
157 /*
158  * Create a server thread
159  */
160 int
161 svc_create_thread(svc_thread_fn func, struct svc_serv *serv)
162 {
163         struct svc_rqst *rqstp;
164         int             error = -ENOMEM;
165
166         rqstp = kzalloc(sizeof(*rqstp), GFP_KERNEL);
167         if (!rqstp)
168                 goto out;
169
170         init_waitqueue_head(&rqstp->rq_wait);
171
172         if (!(rqstp->rq_argp = kmalloc(serv->sv_xdrsize, GFP_KERNEL))
173          || !(rqstp->rq_resp = kmalloc(serv->sv_xdrsize, GFP_KERNEL))
174          || !svc_init_buffer(rqstp, serv->sv_bufsz))
175                 goto out_thread;
176
177         serv->sv_nrthreads++;
178         rqstp->rq_server = serv;
179         error = kernel_thread((int (*)(void *)) func, rqstp, 0);
180         if (error < 0)
181                 goto out_thread;
182         svc_sock_update_bufs(serv);
183         error = 0;
184 out:
185         return error;
186
187 out_thread:
188         svc_exit_thread(rqstp);
189         goto out;
190 }
191
192 /*
193  * Destroy an RPC server thread
194  */
195 void
196 svc_exit_thread(struct svc_rqst *rqstp)
197 {
198         struct svc_serv *serv = rqstp->rq_server;
199
200         svc_release_buffer(rqstp);
201         kfree(rqstp->rq_resp);
202         kfree(rqstp->rq_argp);
203         kfree(rqstp->rq_auth_data);
204         kfree(rqstp);
205
206         /* Release the server */
207         if (serv)
208                 svc_destroy(serv);
209 }
210
211 /*
212  * Register an RPC service with the local portmapper.
213  * To unregister a service, call this routine with 
214  * proto and port == 0.
215  */
216 int
217 svc_register(struct svc_serv *serv, int proto, unsigned short port)
218 {
219         struct svc_program      *progp;
220         unsigned long           flags;
221         int                     i, error = 0, dummy;
222
223         progp = serv->sv_program;
224
225         dprintk("RPC: svc_register(%s, %s, %d)\n",
226                 progp->pg_name, proto == IPPROTO_UDP? "udp" : "tcp", port);
227
228         if (!port)
229                 clear_thread_flag(TIF_SIGPENDING);
230
231         for (i = 0; i < progp->pg_nvers; i++) {
232                 if (progp->pg_vers[i] == NULL)
233                         continue;
234                 error = rpc_register(progp->pg_prog, i, proto, port, &dummy);
235                 if (error < 0)
236                         break;
237                 if (port && !dummy) {
238                         error = -EACCES;
239                         break;
240                 }
241         }
242
243         if (!port) {
244                 spin_lock_irqsave(&current->sighand->siglock, flags);
245                 recalc_sigpending();
246                 spin_unlock_irqrestore(&current->sighand->siglock, flags);
247         }
248
249         return error;
250 }
251
252 /*
253  * Process the RPC request.
254  */
255 int
256 svc_process(struct svc_serv *serv, struct svc_rqst *rqstp)
257 {
258         struct svc_program      *progp;
259         struct svc_version      *versp = NULL;  /* compiler food */
260         struct svc_procedure    *procp = NULL;
261         struct kvec *           argv = &rqstp->rq_arg.head[0];
262         struct kvec *           resv = &rqstp->rq_res.head[0];
263         kxdrproc_t              xdr;
264         __be32                  *statp;
265         u32                     dir, prog, vers, proc;
266         __be32                  auth_stat, rpc_stat;
267         int                     auth_res;
268         __be32                  *accept_statp;
269
270         rpc_stat = rpc_success;
271
272         if (argv->iov_len < 6*4)
273                 goto err_short_len;
274
275         /* setup response xdr_buf.
276          * Initially it has just one page 
277          */
278         svc_take_page(rqstp); /* must succeed */
279         resv->iov_base = page_address(rqstp->rq_respages[0]);
280         resv->iov_len = 0;
281         rqstp->rq_res.pages = rqstp->rq_respages+1;
282         rqstp->rq_res.len = 0;
283         rqstp->rq_res.page_base = 0;
284         rqstp->rq_res.page_len = 0;
285         rqstp->rq_res.buflen = PAGE_SIZE;
286         rqstp->rq_res.tail[0].iov_base = NULL;
287         rqstp->rq_res.tail[0].iov_len = 0;
288         /* Will be turned off only in gss privacy case: */
289         rqstp->rq_sendfile_ok = 1;
290         /* tcp needs a space for the record length... */
291         if (rqstp->rq_prot == IPPROTO_TCP)
292                 svc_putnl(resv, 0);
293
294         rqstp->rq_xid = svc_getu32(argv);
295         svc_putu32(resv, rqstp->rq_xid);
296
297         dir  = svc_getnl(argv);
298         vers = svc_getnl(argv);
299
300         /* First words of reply: */
301         svc_putnl(resv, 1);             /* REPLY */
302
303         if (dir != 0)           /* direction != CALL */
304                 goto err_bad_dir;
305         if (vers != 2)          /* RPC version number */
306                 goto err_bad_rpc;
307
308         /* Save position in case we later decide to reject: */
309         accept_statp = resv->iov_base + resv->iov_len;
310
311         svc_putnl(resv, 0);             /* ACCEPT */
312
313         rqstp->rq_prog = prog = svc_getnl(argv);        /* program number */
314         rqstp->rq_vers = vers = svc_getnl(argv);        /* version number */
315         rqstp->rq_proc = proc = svc_getnl(argv);        /* procedure number */
316
317         progp = serv->sv_program;
318
319         for (progp = serv->sv_program; progp; progp = progp->pg_next)
320                 if (prog == progp->pg_prog)
321                         break;
322
323         /*
324          * Decode auth data, and add verifier to reply buffer.
325          * We do this before anything else in order to get a decent
326          * auth verifier.
327          */
328         auth_res = svc_authenticate(rqstp, &auth_stat);
329         /* Also give the program a chance to reject this call: */
330         if (auth_res == SVC_OK && progp) {
331                 auth_stat = rpc_autherr_badcred;
332                 auth_res = progp->pg_authenticate(rqstp);
333         }
334         switch (auth_res) {
335         case SVC_OK:
336                 break;
337         case SVC_GARBAGE:
338                 rpc_stat = rpc_garbage_args;
339                 goto err_bad;
340         case SVC_SYSERR:
341                 rpc_stat = rpc_system_err;
342                 goto err_bad;
343         case SVC_DENIED:
344                 goto err_bad_auth;
345         case SVC_DROP:
346                 goto dropit;
347         case SVC_COMPLETE:
348                 goto sendit;
349         }
350
351         if (progp == NULL)
352                 goto err_bad_prog;
353
354         if (vers >= progp->pg_nvers ||
355           !(versp = progp->pg_vers[vers]))
356                 goto err_bad_vers;
357
358         procp = versp->vs_proc + proc;
359         if (proc >= versp->vs_nproc || !procp->pc_func)
360                 goto err_bad_proc;
361         rqstp->rq_server   = serv;
362         rqstp->rq_procinfo = procp;
363
364         /* Syntactic check complete */
365         serv->sv_stats->rpccnt++;
366
367         /* Build the reply header. */
368         statp = resv->iov_base +resv->iov_len;
369         svc_putnl(resv, RPC_SUCCESS);
370
371         /* Bump per-procedure stats counter */
372         procp->pc_count++;
373
374         /* Initialize storage for argp and resp */
375         memset(rqstp->rq_argp, 0, procp->pc_argsize);
376         memset(rqstp->rq_resp, 0, procp->pc_ressize);
377
378         /* un-reserve some of the out-queue now that we have a 
379          * better idea of reply size
380          */
381         if (procp->pc_xdrressize)
382                 svc_reserve(rqstp, procp->pc_xdrressize<<2);
383
384         /* Call the function that processes the request. */
385         if (!versp->vs_dispatch) {
386                 /* Decode arguments */
387                 xdr = procp->pc_decode;
388                 if (xdr && !xdr(rqstp, argv->iov_base, rqstp->rq_argp))
389                         goto err_garbage;
390
391                 *statp = procp->pc_func(rqstp, rqstp->rq_argp, rqstp->rq_resp);
392
393                 /* Encode reply */
394                 if (*statp == rpc_success && (xdr = procp->pc_encode)
395                  && !xdr(rqstp, resv->iov_base+resv->iov_len, rqstp->rq_resp)) {
396                         dprintk("svc: failed to encode reply\n");
397                         /* serv->sv_stats->rpcsystemerr++; */
398                         *statp = rpc_system_err;
399                 }
400         } else {
401                 dprintk("svc: calling dispatcher\n");
402                 if (!versp->vs_dispatch(rqstp, statp)) {
403                         /* Release reply info */
404                         if (procp->pc_release)
405                                 procp->pc_release(rqstp, NULL, rqstp->rq_resp);
406                         goto dropit;
407                 }
408         }
409
410         /* Check RPC status result */
411         if (*statp != rpc_success)
412                 resv->iov_len = ((void*)statp)  - resv->iov_base + 4;
413
414         /* Release reply info */
415         if (procp->pc_release)
416                 procp->pc_release(rqstp, NULL, rqstp->rq_resp);
417
418         if (procp->pc_encode == NULL)
419                 goto dropit;
420
421  sendit:
422         if (svc_authorise(rqstp))
423                 goto dropit;
424         return svc_send(rqstp);
425
426  dropit:
427         svc_authorise(rqstp);   /* doesn't hurt to call this twice */
428         dprintk("svc: svc_process dropit\n");
429         svc_drop(rqstp);
430         return 0;
431
432 err_short_len:
433 #ifdef RPC_PARANOIA
434         printk("svc: short len %Zd, dropping request\n", argv->iov_len);
435 #endif
436         goto dropit;                    /* drop request */
437
438 err_bad_dir:
439 #ifdef RPC_PARANOIA
440         printk("svc: bad direction %d, dropping request\n", dir);
441 #endif
442         serv->sv_stats->rpcbadfmt++;
443         goto dropit;                    /* drop request */
444
445 err_bad_rpc:
446         serv->sv_stats->rpcbadfmt++;
447         svc_putnl(resv, 1);     /* REJECT */
448         svc_putnl(resv, 0);     /* RPC_MISMATCH */
449         svc_putnl(resv, 2);     /* Only RPCv2 supported */
450         svc_putnl(resv, 2);
451         goto sendit;
452
453 err_bad_auth:
454         dprintk("svc: authentication failed (%d)\n", ntohl(auth_stat));
455         serv->sv_stats->rpcbadauth++;
456         /* Restore write pointer to location of accept status: */
457         xdr_ressize_check(rqstp, accept_statp);
458         svc_putnl(resv, 1);     /* REJECT */
459         svc_putnl(resv, 1);     /* AUTH_ERROR */
460         svc_putnl(resv, ntohl(auth_stat));      /* status */
461         goto sendit;
462
463 err_bad_prog:
464         dprintk("svc: unknown program %d\n", prog);
465         serv->sv_stats->rpcbadfmt++;
466         svc_putnl(resv, RPC_PROG_UNAVAIL);
467         goto sendit;
468
469 err_bad_vers:
470 #ifdef RPC_PARANOIA
471         printk("svc: unknown version (%d)\n", vers);
472 #endif
473         serv->sv_stats->rpcbadfmt++;
474         svc_putnl(resv, RPC_PROG_MISMATCH);
475         svc_putnl(resv, progp->pg_lovers);
476         svc_putnl(resv, progp->pg_hivers);
477         goto sendit;
478
479 err_bad_proc:
480 #ifdef RPC_PARANOIA
481         printk("svc: unknown procedure (%d)\n", proc);
482 #endif
483         serv->sv_stats->rpcbadfmt++;
484         svc_putnl(resv, RPC_PROC_UNAVAIL);
485         goto sendit;
486
487 err_garbage:
488 #ifdef RPC_PARANOIA
489         printk("svc: failed to decode args\n");
490 #endif
491         rpc_stat = rpc_garbage_args;
492 err_bad:
493         serv->sv_stats->rpcbadfmt++;
494         svc_putnl(resv, ntohl(rpc_stat));
495         goto sendit;
496 }