rumpuser_sp.c revision 1.35 1 /* $NetBSD: rumpuser_sp.c,v 1.35 2011/01/12 12:52:16 pooka Exp $ */
2
3 /*
4 * Copyright (c) 2010, 2011 Antti Kantee. All Rights Reserved.
5 *
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions
8 * are met:
9 * 1. Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * 2. Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 *
15 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS
16 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
17 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
18 * DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
19 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
20 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
21 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
22 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
23 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
24 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
25 * SUCH DAMAGE.
26 */
27
28 /*
29 * Sysproxy routines. This provides system RPC support over host sockets.
30 * The most notable limitation is that the client and server must share
31 * the same ABI. This does not mean that they have to be the same
32 * machine or that they need to run the same version of the host OS,
33 * just that they must agree on the data structures. This even *might*
34 * work correctly from one hardware architecture to another.
35 */
36
37 #include <sys/cdefs.h>
38 __RCSID("$NetBSD: rumpuser_sp.c,v 1.35 2011/01/12 12:52:16 pooka Exp $");
39
40 #include <sys/types.h>
41 #include <sys/atomic.h>
42 #include <sys/mman.h>
43 #include <sys/socket.h>
44
45 #include <arpa/inet.h>
46 #include <netinet/in.h>
47 #include <netinet/tcp.h>
48
49 #include <assert.h>
50 #include <errno.h>
51 #include <fcntl.h>
52 #include <poll.h>
53 #include <pthread.h>
54 #include <stdarg.h>
55 #include <stdio.h>
56 #include <stdlib.h>
57 #include <string.h>
58 #include <unistd.h>
59
60 #include <rump/rump.h> /* XXX: for rfork flags */
61 #include <rump/rumpuser.h>
62 #include "rumpuser_int.h"
63
64 #include "sp_common.c"
65
66 #ifndef MAXCLI
67 #define MAXCLI 256
68 #endif
69 #ifndef MAXWORKER
70 #define MAXWORKER 128
71 #endif
72 #ifndef IDLEWORKER
73 #define IDLEWORKER 16
74 #endif
75 int rumpsp_maxworker = MAXWORKER;
76 int rumpsp_idleworker = IDLEWORKER;
77
78 static struct pollfd pfdlist[MAXCLI];
79 static struct spclient spclist[MAXCLI];
80 static unsigned int disco;
81 static volatile int spfini;
82
83 static struct rumpuser_sp_ops spops;
84
85 static char banner[MAXBANNER];
86
87 #define PROTOMAJOR 0
88 #define PROTOMINOR 1
89
90 struct prefork {
91 uint32_t pf_auth[AUTHLEN];
92 struct lwp *pf_lwp;
93
94 LIST_ENTRY(prefork) pf_entries; /* global list */
95 LIST_ENTRY(prefork) pf_spcentries; /* linked from forking spc */
96 };
97 static LIST_HEAD(, prefork) preforks = LIST_HEAD_INITIALIZER(preforks);
98 static pthread_mutex_t pfmtx;
99
100 /*
101 * This version is for the server. It's optimized for multiple threads
102 * and is *NOT* reentrant wrt to signals.
103 */
104 static int
105 waitresp(struct spclient *spc, struct respwait *rw)
106 {
107 int spcstate;
108 int rv = 0;
109
110 pthread_mutex_lock(&spc->spc_mtx);
111 sendunlockl(spc);
112 while (!rw->rw_done && spc->spc_state != SPCSTATE_DYING) {
113 pthread_cond_wait(&rw->rw_cv, &spc->spc_mtx);
114 }
115 TAILQ_REMOVE(&spc->spc_respwait, rw, rw_entries);
116 spcstate = spc->spc_state;
117 pthread_mutex_unlock(&spc->spc_mtx);
118
119 pthread_cond_destroy(&rw->rw_cv);
120
121 if (rv)
122 return rv;
123 if (spcstate == SPCSTATE_DYING)
124 return ENOTCONN;
125 return rw->rw_error;
126 }
127
128 /*
129 * Manual wrappers, since librump does not have access to the
130 * user namespace wrapped interfaces.
131 */
132
133 static void
134 lwproc_switch(struct lwp *l)
135 {
136
137 spops.spop_schedule();
138 spops.spop_lwproc_switch(l);
139 spops.spop_unschedule();
140 }
141
142 static void
143 lwproc_release(void)
144 {
145
146 spops.spop_schedule();
147 spops.spop_lwproc_release();
148 spops.spop_unschedule();
149 }
150
151 static int
152 lwproc_rfork(struct spclient *spc, int flags)
153 {
154 int rv;
155
156 spops.spop_schedule();
157 rv = spops.spop_lwproc_rfork(spc, flags);
158 spops.spop_unschedule();
159
160 return rv;
161 }
162
163 static int
164 lwproc_newlwp(pid_t pid)
165 {
166 int rv;
167
168 spops.spop_schedule();
169 rv = spops.spop_lwproc_newlwp(pid);
170 spops.spop_unschedule();
171
172 return rv;
173 }
174
175 static struct lwp *
176 lwproc_curlwp(void)
177 {
178 struct lwp *l;
179
180 spops.spop_schedule();
181 l = spops.spop_lwproc_curlwp();
182 spops.spop_unschedule();
183
184 return l;
185 }
186
187 static pid_t
188 lwproc_getpid(void)
189 {
190 pid_t p;
191
192 spops.spop_schedule();
193 p = spops.spop_getpid();
194 spops.spop_unschedule();
195
196 return p;
197 }
198
199 static void
200 lwproc_procexit(void)
201 {
202
203 spops.spop_schedule();
204 spops.spop_procexit();
205 spops.spop_unschedule();
206 }
207
208 static int
209 rumpsyscall(int sysnum, void *data, register_t *retval)
210 {
211 int rv;
212
213 spops.spop_schedule();
214 rv = spops.spop_syscall(sysnum, data, retval);
215 spops.spop_unschedule();
216
217 return rv;
218 }
219
220 static uint64_t
221 nextreq(struct spclient *spc)
222 {
223 uint64_t nw;
224
225 pthread_mutex_lock(&spc->spc_mtx);
226 nw = spc->spc_nextreq++;
227 pthread_mutex_unlock(&spc->spc_mtx);
228
229 return nw;
230 }
231
232 static void
233 send_error_resp(struct spclient *spc, uint64_t reqno, int error)
234 {
235 struct rsp_hdr rhdr;
236
237 rhdr.rsp_len = sizeof(rhdr);
238 rhdr.rsp_reqno = reqno;
239 rhdr.rsp_class = RUMPSP_ERROR;
240 rhdr.rsp_type = 0;
241 rhdr.rsp_error = error;
242
243 sendlock(spc);
244 (void)dosend(spc, &rhdr, sizeof(rhdr));
245 sendunlock(spc);
246 }
247
248 static int
249 send_handshake_resp(struct spclient *spc, uint64_t reqno, int error)
250 {
251 struct rsp_hdr rhdr;
252 int rv;
253
254 rhdr.rsp_len = sizeof(rhdr) + sizeof(error);
255 rhdr.rsp_reqno = reqno;
256 rhdr.rsp_class = RUMPSP_RESP;
257 rhdr.rsp_type = RUMPSP_HANDSHAKE;
258 rhdr.rsp_error = 0;
259
260 sendlock(spc);
261 rv = dosend(spc, &rhdr, sizeof(rhdr));
262 rv = dosend(spc, &error, sizeof(error));
263 sendunlock(spc);
264
265 return rv;
266 }
267
268 static int
269 send_syscall_resp(struct spclient *spc, uint64_t reqno, int error,
270 register_t *retval)
271 {
272 struct rsp_hdr rhdr;
273 struct rsp_sysresp sysresp;
274 int rv;
275
276 rhdr.rsp_len = sizeof(rhdr) + sizeof(sysresp);
277 rhdr.rsp_reqno = reqno;
278 rhdr.rsp_class = RUMPSP_RESP;
279 rhdr.rsp_type = RUMPSP_SYSCALL;
280 rhdr.rsp_sysnum = 0;
281
282 sysresp.rsys_error = error;
283 memcpy(sysresp.rsys_retval, retval, sizeof(sysresp.rsys_retval));
284
285 sendlock(spc);
286 rv = dosend(spc, &rhdr, sizeof(rhdr));
287 rv = dosend(spc, &sysresp, sizeof(sysresp));
288 sendunlock(spc);
289
290 return rv;
291 }
292
293 static int
294 send_prefork_resp(struct spclient *spc, uint64_t reqno, uint32_t *auth)
295 {
296 struct rsp_hdr rhdr;
297 int rv;
298
299 rhdr.rsp_len = sizeof(rhdr) + AUTHLEN*sizeof(*auth);
300 rhdr.rsp_reqno = reqno;
301 rhdr.rsp_class = RUMPSP_RESP;
302 rhdr.rsp_type = RUMPSP_PREFORK;
303 rhdr.rsp_sysnum = 0;
304
305 sendlock(spc);
306 rv = dosend(spc, &rhdr, sizeof(rhdr));
307 rv = dosend(spc, auth, AUTHLEN*sizeof(*auth));
308 sendunlock(spc);
309
310 return rv;
311 }
312
313 static int
314 copyin_req(struct spclient *spc, const void *remaddr, size_t *dlen,
315 int wantstr, void **resp)
316 {
317 struct rsp_hdr rhdr;
318 struct rsp_copydata copydata;
319 struct respwait rw;
320 int rv;
321
322 DPRINTF(("copyin_req: %zu bytes from %p\n", *dlen, remaddr));
323
324 rhdr.rsp_len = sizeof(rhdr) + sizeof(copydata);
325 rhdr.rsp_class = RUMPSP_REQ;
326 if (wantstr)
327 rhdr.rsp_type = RUMPSP_COPYINSTR;
328 else
329 rhdr.rsp_type = RUMPSP_COPYIN;
330 rhdr.rsp_sysnum = 0;
331
332 copydata.rcp_addr = __UNCONST(remaddr);
333 copydata.rcp_len = *dlen;
334
335 putwait(spc, &rw, &rhdr);
336 rv = dosend(spc, &rhdr, sizeof(rhdr));
337 rv = dosend(spc, ©data, sizeof(copydata));
338 if (rv) {
339 unputwait(spc, &rw);
340 return rv;
341 }
342
343 rv = waitresp(spc, &rw);
344
345 DPRINTF(("copyin: response %d\n", rv));
346
347 *resp = rw.rw_data;
348 if (wantstr)
349 *dlen = rw.rw_dlen;
350
351 return rv;
352
353 }
354
355 static int
356 send_copyout_req(struct spclient *spc, const void *remaddr,
357 const void *data, size_t dlen)
358 {
359 struct rsp_hdr rhdr;
360 struct rsp_copydata copydata;
361 int rv;
362
363 DPRINTF(("copyout_req (async): %zu bytes to %p\n", dlen, remaddr));
364
365 rhdr.rsp_len = sizeof(rhdr) + sizeof(copydata) + dlen;
366 rhdr.rsp_reqno = nextreq(spc);
367 rhdr.rsp_class = RUMPSP_REQ;
368 rhdr.rsp_type = RUMPSP_COPYOUT;
369 rhdr.rsp_sysnum = 0;
370
371 copydata.rcp_addr = __UNCONST(remaddr);
372 copydata.rcp_len = dlen;
373
374 sendlock(spc);
375 rv = dosend(spc, &rhdr, sizeof(rhdr));
376 rv = dosend(spc, ©data, sizeof(copydata));
377 rv = dosend(spc, data, dlen);
378 sendunlock(spc);
379
380 return rv;
381 }
382
383 static int
384 anonmmap_req(struct spclient *spc, size_t howmuch, void **resp)
385 {
386 struct rsp_hdr rhdr;
387 struct respwait rw;
388 int rv;
389
390 DPRINTF(("anonmmap_req: %zu bytes\n", howmuch));
391
392 rhdr.rsp_len = sizeof(rhdr) + sizeof(howmuch);
393 rhdr.rsp_class = RUMPSP_REQ;
394 rhdr.rsp_type = RUMPSP_ANONMMAP;
395 rhdr.rsp_sysnum = 0;
396
397 putwait(spc, &rw, &rhdr);
398 rv = dosend(spc, &rhdr, sizeof(rhdr));
399 rv = dosend(spc, &howmuch, sizeof(howmuch));
400 if (rv) {
401 unputwait(spc, &rw);
402 return rv;
403 }
404
405 rv = waitresp(spc, &rw);
406
407 *resp = rw.rw_data;
408
409 DPRINTF(("anonmmap: mapped at %p\n", **(void ***)resp));
410
411 return rv;
412 }
413
414 static void
415 spcref(struct spclient *spc)
416 {
417
418 pthread_mutex_lock(&spc->spc_mtx);
419 spc->spc_refcnt++;
420 pthread_mutex_unlock(&spc->spc_mtx);
421 }
422
423 static void
424 spcrelease(struct spclient *spc)
425 {
426 int ref;
427
428 pthread_mutex_lock(&spc->spc_mtx);
429 ref = --spc->spc_refcnt;
430 pthread_mutex_unlock(&spc->spc_mtx);
431
432 if (ref > 0)
433 return;
434
435 DPRINTF(("rump_sp: spcrelease: spc %p fd %d\n", spc, spc->spc_fd));
436
437 _DIAGASSERT(TAILQ_EMPTY(&spc->spc_respwait));
438 _DIAGASSERT(spc->spc_buf == NULL);
439
440 if (spc->spc_mainlwp) {
441 lwproc_switch(spc->spc_mainlwp);
442 lwproc_release();
443 }
444 spc->spc_mainlwp = NULL;
445
446 close(spc->spc_fd);
447 spc->spc_fd = -1;
448 spc->spc_state = SPCSTATE_NEW;
449
450 atomic_inc_uint(&disco);
451 }
452
453 static void
454 serv_handledisco(unsigned int idx)
455 {
456 struct spclient *spc = &spclist[idx];
457
458 DPRINTF(("rump_sp: disconnecting [%u]\n", idx));
459
460 pfdlist[idx].fd = -1;
461 pfdlist[idx].revents = 0;
462 pthread_mutex_lock(&spc->spc_mtx);
463 spc->spc_state = SPCSTATE_DYING;
464 kickall(spc);
465 sendunlockl(spc);
466 pthread_mutex_unlock(&spc->spc_mtx);
467
468 if (spc->spc_mainlwp) {
469 lwproc_switch(spc->spc_mainlwp);
470 lwproc_procexit();
471 lwproc_switch(NULL);
472 }
473
474 /*
475 * Nobody's going to attempt to send/receive anymore,
476 * so reinit info relevant to that.
477 */
478 /*LINTED:pointer casts may be ok*/
479 memset((char *)spc + SPC_ZEROFF, 0, sizeof(*spc) - SPC_ZEROFF);
480
481 spcrelease(spc);
482 }
483
484 static void
485 serv_shutdown(void)
486 {
487 struct spclient *spc;
488 unsigned int i;
489
490 for (i = 1; i < MAXCLI; i++) {
491 spc = &spclist[i];
492 if (spc->spc_fd == -1)
493 continue;
494
495 shutdown(spc->spc_fd, SHUT_RDWR);
496 serv_handledisco(i);
497
498 spcrelease(spc);
499 }
500 }
501
502 static unsigned
503 serv_handleconn(int fd, connecthook_fn connhook, int busy)
504 {
505 struct sockaddr_storage ss;
506 socklen_t sl = sizeof(ss);
507 int newfd, flags;
508 unsigned i;
509
510 /*LINTED: cast ok */
511 newfd = accept(fd, (struct sockaddr *)&ss, &sl);
512 if (newfd == -1)
513 return 0;
514
515 if (busy) {
516 close(newfd); /* EBUSY */
517 return 0;
518 }
519
520 flags = fcntl(newfd, F_GETFL, 0);
521 if (fcntl(newfd, F_SETFL, flags | O_NONBLOCK) == -1) {
522 close(newfd);
523 return 0;
524 }
525
526 if (connhook(newfd) != 0) {
527 close(newfd);
528 return 0;
529 }
530
531 /* write out a banner for the client */
532 if (send(newfd, banner, strlen(banner), MSG_NOSIGNAL)
533 != (ssize_t)strlen(banner)) {
534 close(newfd);
535 return 0;
536 }
537
538 /* find empty slot the simple way */
539 for (i = 0; i < MAXCLI; i++) {
540 if (pfdlist[i].fd == -1 && spclist[i].spc_state == SPCSTATE_NEW)
541 break;
542 }
543
544 assert(i < MAXCLI);
545
546 pfdlist[i].fd = newfd;
547 spclist[i].spc_fd = newfd;
548 spclist[i].spc_istatus = SPCSTATUS_BUSY; /* dedicated receiver */
549 spclist[i].spc_refcnt = 1;
550
551 TAILQ_INIT(&spclist[i].spc_respwait);
552
553 DPRINTF(("rump_sp: added new connection fd %d at idx %u\n", newfd, i));
554
555 return i;
556 }
557
558 static void
559 serv_handlesyscall(struct spclient *spc, struct rsp_hdr *rhdr, uint8_t *data)
560 {
561 register_t retval[2] = {0, 0};
562 int rv, sysnum;
563
564 sysnum = (int)rhdr->rsp_sysnum;
565 DPRINTF(("rump_sp: handling syscall %d from client %d\n",
566 sysnum, spc->spc_pid));
567
568 lwproc_newlwp(spc->spc_pid);
569 rv = rumpsyscall(sysnum, data, retval);
570 lwproc_release();
571
572 DPRINTF(("rump_sp: got return value %d & %d/%d\n",
573 rv, retval[0], retval[1]));
574
575 send_syscall_resp(spc, rhdr->rsp_reqno, rv, retval);
576 }
577
578 struct sysbouncearg {
579 struct spclient *sba_spc;
580 struct rsp_hdr sba_hdr;
581 uint8_t *sba_data;
582
583 TAILQ_ENTRY(sysbouncearg) sba_entries;
584 };
585 static pthread_mutex_t sbamtx;
586 static pthread_cond_t sbacv;
587 static int nworker, idleworker;
588 static TAILQ_HEAD(, sysbouncearg) syslist = TAILQ_HEAD_INITIALIZER(syslist);
589
590 /*ARGSUSED*/
591 static void *
592 serv_syscallbouncer(void *arg)
593 {
594 struct sysbouncearg *sba;
595
596 for (;;) {
597 pthread_mutex_lock(&sbamtx);
598 if (idleworker >= rumpsp_idleworker) {
599 nworker--;
600 pthread_mutex_unlock(&sbamtx);
601 break;
602 }
603 idleworker++;
604 while (TAILQ_EMPTY(&syslist)) {
605 pthread_cond_wait(&sbacv, &sbamtx);
606 }
607
608 sba = TAILQ_FIRST(&syslist);
609 TAILQ_REMOVE(&syslist, sba, sba_entries);
610 idleworker--;
611 pthread_mutex_unlock(&sbamtx);
612
613 serv_handlesyscall(sba->sba_spc,
614 &sba->sba_hdr, sba->sba_data);
615 spcrelease(sba->sba_spc);
616 free(sba->sba_data);
617 free(sba);
618 }
619
620 return NULL;
621 }
622
623 static int
624 sp_copyin(void *arg, const void *raddr, void *laddr, size_t *len, int wantstr)
625 {
626 struct spclient *spc = arg;
627 void *rdata = NULL; /* XXXuninit */
628 int rv, nlocks;
629
630 rumpuser__kunlock(0, &nlocks, NULL);
631
632 rv = copyin_req(spc, raddr, len, wantstr, &rdata);
633 if (rv)
634 goto out;
635
636 memcpy(laddr, rdata, *len);
637 free(rdata);
638
639 out:
640 rumpuser__klock(nlocks, NULL);
641 if (rv)
642 return EFAULT;
643 return 0;
644 }
645
646 int
647 rumpuser_sp_copyin(void *arg, const void *raddr, void *laddr, size_t len)
648 {
649
650 return sp_copyin(arg, raddr, laddr, &len, 0);
651 }
652
653 int
654 rumpuser_sp_copyinstr(void *arg, const void *raddr, void *laddr, size_t *len)
655 {
656
657 return sp_copyin(arg, raddr, laddr, len, 1);
658 }
659
660 static int
661 sp_copyout(void *arg, const void *laddr, void *raddr, size_t dlen)
662 {
663 struct spclient *spc = arg;
664 int nlocks, rv;
665
666 rumpuser__kunlock(0, &nlocks, NULL);
667 rv = send_copyout_req(spc, raddr, laddr, dlen);
668 rumpuser__klock(nlocks, NULL);
669
670 if (rv)
671 return EFAULT;
672 return 0;
673 }
674
675 int
676 rumpuser_sp_copyout(void *arg, const void *laddr, void *raddr, size_t dlen)
677 {
678
679 return sp_copyout(arg, laddr, raddr, dlen);
680 }
681
682 int
683 rumpuser_sp_copyoutstr(void *arg, const void *laddr, void *raddr, size_t *dlen)
684 {
685
686 return sp_copyout(arg, laddr, raddr, *dlen);
687 }
688
689 int
690 rumpuser_sp_anonmmap(void *arg, size_t howmuch, void **addr)
691 {
692 struct spclient *spc = arg;
693 void *resp, *rdata;
694 int nlocks, rv;
695
696 rumpuser__kunlock(0, &nlocks, NULL);
697
698 rv = anonmmap_req(spc, howmuch, &rdata);
699 if (rv) {
700 rv = EFAULT;
701 goto out;
702 }
703
704 resp = *(void **)rdata;
705 free(rdata);
706
707 if (resp == NULL) {
708 rv = ENOMEM;
709 }
710
711 *addr = resp;
712
713 out:
714 rumpuser__klock(nlocks, NULL);
715
716 if (rv)
717 return rv;
718 return 0;
719 }
720
721 /*
722 *
723 * Startup routines and mainloop for server.
724 *
725 */
726
727 struct spservarg {
728 int sps_sock;
729 connecthook_fn sps_connhook;
730 };
731
732 static pthread_attr_t pattr_detached;
733 static void
734 handlereq(struct spclient *spc)
735 {
736 struct sysbouncearg *sba;
737 pthread_t pt;
738 int retries, error, i;
739
740 if (__predict_false(spc->spc_state == SPCSTATE_NEW)) {
741 if (spc->spc_hdr.rsp_type != RUMPSP_HANDSHAKE) {
742 send_error_resp(spc, spc->spc_hdr.rsp_reqno, EAUTH);
743 shutdown(spc->spc_fd, SHUT_RDWR);
744 spcfreebuf(spc);
745 return;
746 }
747
748 if (spc->spc_hdr.rsp_handshake == HANDSHAKE_GUEST) {
749 if ((error = lwproc_rfork(spc, RUMP_RFCFDG)) != 0) {
750 shutdown(spc->spc_fd, SHUT_RDWR);
751 }
752
753 spcfreebuf(spc);
754 if (error)
755 return;
756
757 spc->spc_mainlwp = lwproc_curlwp();
758
759 send_handshake_resp(spc, spc->spc_hdr.rsp_reqno, 0);
760 } else if (spc->spc_hdr.rsp_handshake == HANDSHAKE_FORK) {
761 struct lwp *tmpmain;
762 struct prefork *pf;
763 struct handshake_fork *rfp;
764 uint64_t reqno;
765 int cancel;
766
767 reqno = spc->spc_hdr.rsp_reqno;
768 if (spc->spc_off-HDRSZ != sizeof(*rfp)) {
769 send_error_resp(spc, reqno, EINVAL);
770 shutdown(spc->spc_fd, SHUT_RDWR);
771 spcfreebuf(spc);
772 return;
773 }
774
775 /*LINTED*/
776 rfp = (void *)spc->spc_buf;
777 cancel = rfp->rf_cancel;
778
779 pthread_mutex_lock(&pfmtx);
780 LIST_FOREACH(pf, &preforks, pf_entries) {
781 if (memcmp(rfp->rf_auth, pf->pf_auth,
782 sizeof(rfp->rf_auth)) == 0) {
783 LIST_REMOVE(pf, pf_entries);
784 LIST_REMOVE(pf, pf_spcentries);
785 break;
786 }
787 }
788 pthread_mutex_lock(&pfmtx);
789 spcfreebuf(spc);
790
791 if (!pf) {
792 send_error_resp(spc, reqno, ESRCH);
793 shutdown(spc->spc_fd, SHUT_RDWR);
794 return;
795 }
796
797 tmpmain = pf->pf_lwp;
798 free(pf);
799 lwproc_switch(tmpmain);
800 if (cancel) {
801 lwproc_release();
802 shutdown(spc->spc_fd, SHUT_RDWR);
803 return;
804 }
805
806 /*
807 * So, we forked already during "prefork" to save
808 * the file descriptors from a parent exit
809 * race condition. But now we need to fork
810 * a second time since the initial fork has
811 * the wrong spc pointer. (yea, optimize
812 * interfaces some day if anyone cares)
813 */
814 if ((error = lwproc_rfork(spc, 0)) != 0) {
815 send_error_resp(spc, reqno, error);
816 shutdown(spc->spc_fd, SHUT_RDWR);
817 lwproc_release();
818 return;
819 }
820 spc->spc_mainlwp = lwproc_curlwp();
821 lwproc_switch(tmpmain);
822 lwproc_release();
823 lwproc_switch(spc->spc_mainlwp);
824
825 send_handshake_resp(spc, reqno, 0);
826 }
827
828 spc->spc_pid = lwproc_getpid();
829
830 DPRINTF(("rump_sp: handshake for client %p complete, pid %d\n",
831 spc, spc->spc_pid));
832
833 lwproc_switch(NULL);
834 spc->spc_state = SPCSTATE_RUNNING;
835 return;
836 }
837
838 if (__predict_false(spc->spc_hdr.rsp_type == RUMPSP_PREFORK)) {
839 struct prefork *pf;
840 uint64_t reqno;
841 uint32_t auth[AUTHLEN];
842
843 DPRINTF(("rump_sp: prefork handler executing for %p\n", spc));
844 reqno = spc->spc_hdr.rsp_reqno;
845 spcfreebuf(spc);
846
847 pf = malloc(sizeof(*pf));
848 if (pf == NULL) {
849 send_error_resp(spc, reqno, ENOMEM);
850 return;
851 }
852
853 /*
854 * Use client main lwp to fork. this is never used by
855 * worker threads (except if spc refcount goes to 0),
856 * so we can safely use it here.
857 */
858 lwproc_switch(spc->spc_mainlwp);
859 if ((error = lwproc_rfork(spc, RUMP_RFFDG)) != 0) {
860 DPRINTF(("rump_sp: fork failed: %d (%p)\n",error, spc));
861 send_error_resp(spc, reqno, error);
862 lwproc_switch(NULL);
863 free(pf);
864 return;
865 }
866
867 /* Ok, we have a new process context and a new curlwp */
868 for (i = 0; i < AUTHLEN; i++) {
869 pf->pf_auth[i] = auth[i] = arc4random();
870 }
871 pf->pf_lwp = lwproc_curlwp();
872 lwproc_switch(NULL);
873
874 pthread_mutex_lock(&pfmtx);
875 LIST_INSERT_HEAD(&preforks, pf, pf_entries);
876 LIST_INSERT_HEAD(&spc->spc_pflist, pf, pf_spcentries);
877 pthread_mutex_unlock(&pfmtx);
878
879 DPRINTF(("rump_sp: prefork handler success %p\n", spc));
880
881 send_prefork_resp(spc, reqno, auth);
882 return;
883 }
884
885 if (__predict_false(spc->spc_hdr.rsp_type != RUMPSP_SYSCALL)) {
886 send_error_resp(spc, spc->spc_hdr.rsp_reqno, EINVAL);
887 spcfreebuf(spc);
888 return;
889 }
890
891 retries = 0;
892 while ((sba = malloc(sizeof(*sba))) == NULL) {
893 if (nworker == 0 || retries > 10) {
894 send_error_resp(spc, spc->spc_hdr.rsp_reqno, EAGAIN);
895 spcfreebuf(spc);
896 return;
897 }
898 /* slim chance of more memory? */
899 usleep(10000);
900 }
901
902 sba->sba_spc = spc;
903 sba->sba_hdr = spc->spc_hdr;
904 sba->sba_data = spc->spc_buf;
905 spcresetbuf(spc);
906
907 spcref(spc);
908
909 pthread_mutex_lock(&sbamtx);
910 TAILQ_INSERT_TAIL(&syslist, sba, sba_entries);
911 if (idleworker > 0) {
912 /* do we have a daemon's tool (i.e. idle threads)? */
913 pthread_cond_signal(&sbacv);
914 } else if (nworker < rumpsp_maxworker) {
915 /*
916 * Else, need to create one
917 * (if we can, otherwise just expect another
918 * worker to pick up the syscall)
919 */
920 if (pthread_create(&pt, &pattr_detached,
921 serv_syscallbouncer, NULL) == 0)
922 nworker++;
923 }
924 pthread_mutex_unlock(&sbamtx);
925 }
926
927 static void *
928 spserver(void *arg)
929 {
930 struct spservarg *sarg = arg;
931 struct spclient *spc;
932 unsigned idx;
933 int seen;
934 int rv;
935 unsigned int nfds, maxidx;
936
937 for (idx = 0; idx < MAXCLI; idx++) {
938 pfdlist[idx].fd = -1;
939 pfdlist[idx].events = POLLIN;
940
941 spc = &spclist[idx];
942 pthread_mutex_init(&spc->spc_mtx, NULL);
943 pthread_cond_init(&spc->spc_cv, NULL);
944 spc->spc_fd = -1;
945 }
946 pfdlist[0].fd = spclist[0].spc_fd = sarg->sps_sock;
947 pfdlist[0].events = POLLIN;
948 nfds = 1;
949 maxidx = 0;
950
951 pthread_attr_init(&pattr_detached);
952 pthread_attr_setdetachstate(&pattr_detached, PTHREAD_CREATE_DETACHED);
953 /* XXX: doesn't stacksize currently work on NetBSD */
954 pthread_attr_setstacksize(&pattr_detached, 32*1024);
955
956 pthread_mutex_init(&sbamtx, NULL);
957 pthread_cond_init(&sbacv, NULL);
958
959 DPRINTF(("rump_sp: server mainloop\n"));
960
961 for (;;) {
962 int discoed;
963
964 /* g/c hangarounds (eventually) */
965 discoed = atomic_swap_uint(&disco, 0);
966 while (discoed--) {
967 nfds--;
968 idx = maxidx;
969 while (idx) {
970 if (pfdlist[idx].fd != -1) {
971 maxidx = idx;
972 break;
973 }
974 idx--;
975 }
976 DPRINTF(("rump_sp: set maxidx to [%u]\n",
977 maxidx));
978 }
979
980 DPRINTF(("rump_sp: loop nfd %d\n", maxidx+1));
981 seen = 0;
982 rv = poll(pfdlist, maxidx+1, INFTIM);
983 assert(maxidx+1 <= MAXCLI);
984 assert(rv != 0);
985 if (rv == -1) {
986 if (errno == EINTR)
987 continue;
988 fprintf(stderr, "rump_spserver: poll returned %d\n",
989 errno);
990 break;
991 }
992
993 for (idx = 0; seen < rv && idx < MAXCLI; idx++) {
994 if ((pfdlist[idx].revents & POLLIN) == 0)
995 continue;
996
997 seen++;
998 DPRINTF(("rump_sp: activity at [%u] %d/%d\n",
999 idx, seen, rv));
1000 if (idx > 0) {
1001 spc = &spclist[idx];
1002 DPRINTF(("rump_sp: mainloop read [%u]\n", idx));
1003 switch (readframe(spc)) {
1004 case 0:
1005 break;
1006 case -1:
1007 serv_handledisco(idx);
1008 break;
1009 default:
1010 switch (spc->spc_hdr.rsp_class) {
1011 case RUMPSP_RESP:
1012 kickwaiter(spc);
1013 break;
1014 case RUMPSP_REQ:
1015 handlereq(spc);
1016 break;
1017 default:
1018 send_error_resp(spc,
1019 spc->spc_hdr.rsp_reqno,
1020 ENOENT);
1021 spcfreebuf(spc);
1022 break;
1023 }
1024 break;
1025 }
1026
1027 } else {
1028 DPRINTF(("rump_sp: mainloop new connection\n"));
1029
1030 if (__predict_false(spfini)) {
1031 close(spclist[0].spc_fd);
1032 serv_shutdown();
1033 goto out;
1034 }
1035
1036 idx = serv_handleconn(pfdlist[0].fd,
1037 sarg->sps_connhook, nfds == MAXCLI);
1038 if (idx)
1039 nfds++;
1040 if (idx > maxidx)
1041 maxidx = idx;
1042 DPRINTF(("rump_sp: maxid now %d\n", maxidx));
1043 }
1044 }
1045 }
1046
1047 out:
1048 return NULL;
1049 }
1050
1051 static unsigned cleanupidx;
1052 static struct sockaddr *cleanupsa;
1053 int
1054 rumpuser_sp_init(const char *url, const struct rumpuser_sp_ops *spopsp,
1055 const char *ostype, const char *osrelease, const char *machine)
1056 {
1057 pthread_t pt;
1058 struct spservarg *sarg;
1059 struct sockaddr *sap;
1060 char *p;
1061 unsigned idx;
1062 int error, s;
1063
1064 p = strdup(url);
1065 if (p == NULL)
1066 return ENOMEM;
1067 error = parseurl(p, &sap, &idx, 1);
1068 free(p);
1069 if (error)
1070 return error;
1071
1072 snprintf(banner, sizeof(banner), "RUMPSP-%d.%d-%s-%s/%s\n",
1073 PROTOMAJOR, PROTOMINOR, ostype, osrelease, machine);
1074
1075 s = socket(parsetab[idx].domain, SOCK_STREAM, 0);
1076 if (s == -1)
1077 return errno;
1078
1079 spops = *spopsp;
1080 sarg = malloc(sizeof(*sarg));
1081 if (sarg == NULL) {
1082 close(s);
1083 return ENOMEM;
1084 }
1085
1086 sarg->sps_sock = s;
1087 sarg->sps_connhook = parsetab[idx].connhook;
1088
1089 cleanupidx = idx;
1090 cleanupsa = sap;
1091
1092 /* sloppy error recovery */
1093
1094 /*LINTED*/
1095 if (bind(s, sap, sap->sa_len) == -1) {
1096 fprintf(stderr, "rump_sp: server bind failed\n");
1097 return errno;
1098 }
1099
1100 if (listen(s, MAXCLI) == -1) {
1101 fprintf(stderr, "rump_sp: server listen failed\n");
1102 return errno;
1103 }
1104
1105 if ((error = pthread_create(&pt, NULL, spserver, sarg)) != 0) {
1106 fprintf(stderr, "rump_sp: cannot create wrkr thread\n");
1107 return errno;
1108 }
1109 pthread_detach(pt);
1110
1111 return 0;
1112 }
1113
1114 void
1115 rumpuser_sp_fini()
1116 {
1117
1118 if (spclist[0].spc_fd) {
1119 parsetab[cleanupidx].cleanup(cleanupsa);
1120 shutdown(spclist[0].spc_fd, SHUT_RDWR);
1121 spfini = 1;
1122 }
1123 }
1124