rumpuser_sp.c revision 1.56 1 /* $NetBSD: rumpuser_sp.c,v 1.56 2013/04/29 14:51:40 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 "rumpuser_port.h"
38
39 #if !defined(lint)
40 __RCSID("$NetBSD: rumpuser_sp.c,v 1.56 2013/04/29 14:51:40 pooka Exp $");
41 #endif /* !lint */
42
43 #include <sys/types.h>
44 #include <sys/mman.h>
45 #include <sys/socket.h>
46
47 #include <arpa/inet.h>
48 #include <netinet/in.h>
49 #include <netinet/tcp.h>
50
51 #include <assert.h>
52 #include <errno.h>
53 #include <fcntl.h>
54 #include <poll.h>
55 #include <pthread.h>
56 #include <stdarg.h>
57 #include <stdio.h>
58 #include <stdlib.h>
59 #include <string.h>
60 #include <unistd.h>
61
62 #include <rump/rump.h> /* XXX: for rfork flags */
63 #include <rump/rumpuser.h>
64
65 #include "rumpuser_int.h"
66
67 #include "sp_common.c"
68
69 #ifndef MAXCLI
70 #define MAXCLI 256
71 #endif
72 #ifndef MAXWORKER
73 #define MAXWORKER 128
74 #endif
75 #ifndef IDLEWORKER
76 #define IDLEWORKER 16
77 #endif
78 int rumpsp_maxworker = MAXWORKER;
79 int rumpsp_idleworker = IDLEWORKER;
80
81 static struct pollfd pfdlist[MAXCLI];
82 static struct spclient spclist[MAXCLI];
83 static unsigned int disco;
84 static volatile int spfini;
85
86 static char banner[MAXBANNER];
87
88 #define PROTOMAJOR 0
89 #define PROTOMINOR 4
90
91
92 /* how to use atomic ops on Linux? */
93 #if defined(__linux__) || defined(__CYGWIN__)
94 static pthread_mutex_t discomtx = PTHREAD_MUTEX_INITIALIZER;
95
96 static void
97 signaldisco(void)
98 {
99
100 pthread_mutex_lock(&discomtx);
101 disco++;
102 pthread_mutex_unlock(&discomtx);
103 }
104
105 static unsigned int
106 getdisco(void)
107 {
108 unsigned int discocnt;
109
110 pthread_mutex_lock(&discomtx);
111 discocnt = disco;
112 disco = 0;
113 pthread_mutex_unlock(&discomtx);
114
115 return discocnt;
116 }
117
118 #elif defined(__FreeBSD__) || defined(__DragonFly__)
119
120 #include <machine/atomic.h>
121 #define signaldisco() atomic_add_int(&disco, 1)
122 #define getdisco() atomic_readandclear_int(&disco)
123
124 #else /* NetBSD */
125
126 #include <sys/atomic.h>
127 #define signaldisco() atomic_inc_uint(&disco)
128 #define getdisco() atomic_swap_uint(&disco, 0)
129
130 #endif
131
132
133 struct prefork {
134 uint32_t pf_auth[AUTHLEN];
135 struct lwp *pf_lwp;
136
137 LIST_ENTRY(prefork) pf_entries; /* global list */
138 LIST_ENTRY(prefork) pf_spcentries; /* linked from forking spc */
139 };
140 static LIST_HEAD(, prefork) preforks = LIST_HEAD_INITIALIZER(preforks);
141 static pthread_mutex_t pfmtx;
142
143 /*
144 * This version is for the server. It's optimized for multiple threads
145 * and is *NOT* reentrant wrt to signals.
146 */
147 static int
148 waitresp(struct spclient *spc, struct respwait *rw)
149 {
150 int spcstate;
151 int rv = 0;
152
153 pthread_mutex_lock(&spc->spc_mtx);
154 sendunlockl(spc);
155 while (!rw->rw_done && spc->spc_state != SPCSTATE_DYING) {
156 pthread_cond_wait(&rw->rw_cv, &spc->spc_mtx);
157 }
158 TAILQ_REMOVE(&spc->spc_respwait, rw, rw_entries);
159 spcstate = spc->spc_state;
160 pthread_mutex_unlock(&spc->spc_mtx);
161
162 pthread_cond_destroy(&rw->rw_cv);
163
164 if (rv)
165 return rv;
166 if (spcstate == SPCSTATE_DYING)
167 return ENOTCONN;
168 return rw->rw_error;
169 }
170
171 /*
172 * Manual wrappers, since librump does not have access to the
173 * user namespace wrapped interfaces.
174 */
175
176 static void
177 lwproc_switch(struct lwp *l)
178 {
179
180 rumpuser__hyp.hyp_schedule();
181 rumpuser__hyp.hyp_lwproc_switch(l);
182 rumpuser__hyp.hyp_unschedule();
183 }
184
185 static void
186 lwproc_release(void)
187 {
188
189 rumpuser__hyp.hyp_schedule();
190 rumpuser__hyp.hyp_lwproc_release();
191 rumpuser__hyp.hyp_unschedule();
192 }
193
194 static int
195 lwproc_rfork(struct spclient *spc, int flags, const char *comm)
196 {
197 int rv;
198
199 rumpuser__hyp.hyp_schedule();
200 rv = rumpuser__hyp.hyp_lwproc_rfork(spc, flags, comm);
201 rumpuser__hyp.hyp_unschedule();
202
203 return rv;
204 }
205
206 static int
207 lwproc_newlwp(pid_t pid)
208 {
209 int rv;
210
211 rumpuser__hyp.hyp_schedule();
212 rv = rumpuser__hyp.hyp_lwproc_newlwp(pid);
213 rumpuser__hyp.hyp_unschedule();
214
215 return rv;
216 }
217
218 static struct lwp *
219 lwproc_curlwp(void)
220 {
221 struct lwp *l;
222
223 rumpuser__hyp.hyp_schedule();
224 l = rumpuser__hyp.hyp_lwproc_curlwp();
225 rumpuser__hyp.hyp_unschedule();
226
227 return l;
228 }
229
230 static pid_t
231 lwproc_getpid(void)
232 {
233 pid_t p;
234
235 rumpuser__hyp.hyp_schedule();
236 p = rumpuser__hyp.hyp_getpid();
237 rumpuser__hyp.hyp_unschedule();
238
239 return p;
240 }
241
242 static void
243 lwproc_execnotify(const char *comm)
244 {
245
246 rumpuser__hyp.hyp_schedule();
247 rumpuser__hyp.hyp_execnotify(comm);
248 rumpuser__hyp.hyp_unschedule();
249 }
250
251 static void
252 lwproc_lwpexit(void)
253 {
254
255 rumpuser__hyp.hyp_schedule();
256 rumpuser__hyp.hyp_lwpexit();
257 rumpuser__hyp.hyp_unschedule();
258 }
259
260 static int
261 rumpsyscall(int sysnum, void *data, register_t *regrv)
262 {
263 long retval[2] = {0, 0};
264 int rv;
265
266 rumpuser__hyp.hyp_schedule();
267 rv = rumpuser__hyp.hyp_syscall(sysnum, data, retval);
268 rumpuser__hyp.hyp_unschedule();
269
270 regrv[0] = retval[0];
271 regrv[1] = retval[1];
272 return rv;
273 }
274
275 static uint64_t
276 nextreq(struct spclient *spc)
277 {
278 uint64_t nw;
279
280 pthread_mutex_lock(&spc->spc_mtx);
281 nw = spc->spc_nextreq++;
282 pthread_mutex_unlock(&spc->spc_mtx);
283
284 return nw;
285 }
286
287 /*
288 * XXX: we send responses with "blocking" I/O. This is not
289 * ok for the main thread. XXXFIXME
290 */
291
292 static void
293 send_error_resp(struct spclient *spc, uint64_t reqno, enum rumpsp_err error)
294 {
295 struct rsp_hdr rhdr;
296 struct iovec iov[1];
297
298 rhdr.rsp_len = sizeof(rhdr);
299 rhdr.rsp_reqno = reqno;
300 rhdr.rsp_class = RUMPSP_ERROR;
301 rhdr.rsp_type = 0;
302 rhdr.rsp_error = error;
303
304 IOVPUT(iov[0], rhdr);
305
306 sendlock(spc);
307 (void)SENDIOV(spc, iov);
308 sendunlock(spc);
309 }
310
311 static int
312 send_handshake_resp(struct spclient *spc, uint64_t reqno, int error)
313 {
314 struct rsp_hdr rhdr;
315 struct iovec iov[2];
316 int rv;
317
318 rhdr.rsp_len = sizeof(rhdr) + sizeof(error);
319 rhdr.rsp_reqno = reqno;
320 rhdr.rsp_class = RUMPSP_RESP;
321 rhdr.rsp_type = RUMPSP_HANDSHAKE;
322 rhdr.rsp_error = 0;
323
324 IOVPUT(iov[0], rhdr);
325 IOVPUT(iov[1], error);
326
327 sendlock(spc);
328 rv = SENDIOV(spc, iov);
329 sendunlock(spc);
330
331 return rv;
332 }
333
334 static int
335 send_syscall_resp(struct spclient *spc, uint64_t reqno, int error,
336 register_t *retval)
337 {
338 struct rsp_hdr rhdr;
339 struct rsp_sysresp sysresp;
340 struct iovec iov[2];
341 int rv;
342
343 rhdr.rsp_len = sizeof(rhdr) + sizeof(sysresp);
344 rhdr.rsp_reqno = reqno;
345 rhdr.rsp_class = RUMPSP_RESP;
346 rhdr.rsp_type = RUMPSP_SYSCALL;
347 rhdr.rsp_sysnum = 0;
348
349 sysresp.rsys_error = error;
350 memcpy(sysresp.rsys_retval, retval, sizeof(sysresp.rsys_retval));
351
352 IOVPUT(iov[0], rhdr);
353 IOVPUT(iov[1], sysresp);
354
355 sendlock(spc);
356 rv = SENDIOV(spc, iov);
357 sendunlock(spc);
358
359 return rv;
360 }
361
362 static int
363 send_prefork_resp(struct spclient *spc, uint64_t reqno, uint32_t *auth)
364 {
365 struct rsp_hdr rhdr;
366 struct iovec iov[2];
367 int rv;
368
369 rhdr.rsp_len = sizeof(rhdr) + AUTHLEN*sizeof(*auth);
370 rhdr.rsp_reqno = reqno;
371 rhdr.rsp_class = RUMPSP_RESP;
372 rhdr.rsp_type = RUMPSP_PREFORK;
373 rhdr.rsp_sysnum = 0;
374
375 IOVPUT(iov[0], rhdr);
376 IOVPUT_WITHSIZE(iov[1], auth, AUTHLEN*sizeof(*auth));
377
378 sendlock(spc);
379 rv = SENDIOV(spc, iov);
380 sendunlock(spc);
381
382 return rv;
383 }
384
385 static int
386 copyin_req(struct spclient *spc, const void *remaddr, size_t *dlen,
387 int wantstr, void **resp)
388 {
389 struct rsp_hdr rhdr;
390 struct rsp_copydata copydata;
391 struct respwait rw;
392 struct iovec iov[2];
393 int rv;
394
395 DPRINTF(("copyin_req: %zu bytes from %p\n", *dlen, remaddr));
396
397 rhdr.rsp_len = sizeof(rhdr) + sizeof(copydata);
398 rhdr.rsp_class = RUMPSP_REQ;
399 if (wantstr)
400 rhdr.rsp_type = RUMPSP_COPYINSTR;
401 else
402 rhdr.rsp_type = RUMPSP_COPYIN;
403 rhdr.rsp_sysnum = 0;
404
405 copydata.rcp_addr = __UNCONST(remaddr);
406 copydata.rcp_len = *dlen;
407
408 IOVPUT(iov[0], rhdr);
409 IOVPUT(iov[1], copydata);
410
411 putwait(spc, &rw, &rhdr);
412 rv = SENDIOV(spc, iov);
413 if (rv) {
414 unputwait(spc, &rw);
415 return rv;
416 }
417
418 rv = waitresp(spc, &rw);
419
420 DPRINTF(("copyin: response %d\n", rv));
421
422 *resp = rw.rw_data;
423 if (wantstr)
424 *dlen = rw.rw_dlen;
425
426 return rv;
427
428 }
429
430 static int
431 send_copyout_req(struct spclient *spc, const void *remaddr,
432 const void *data, size_t dlen)
433 {
434 struct rsp_hdr rhdr;
435 struct rsp_copydata copydata;
436 struct iovec iov[3];
437 int rv;
438
439 DPRINTF(("copyout_req (async): %zu bytes to %p\n", dlen, remaddr));
440
441 rhdr.rsp_len = sizeof(rhdr) + sizeof(copydata) + dlen;
442 rhdr.rsp_reqno = nextreq(spc);
443 rhdr.rsp_class = RUMPSP_REQ;
444 rhdr.rsp_type = RUMPSP_COPYOUT;
445 rhdr.rsp_sysnum = 0;
446
447 copydata.rcp_addr = __UNCONST(remaddr);
448 copydata.rcp_len = dlen;
449
450 IOVPUT(iov[0], rhdr);
451 IOVPUT(iov[1], copydata);
452 IOVPUT_WITHSIZE(iov[2], __UNCONST(data), dlen);
453
454 sendlock(spc);
455 rv = SENDIOV(spc, iov);
456 sendunlock(spc);
457
458 return rv;
459 }
460
461 static int
462 anonmmap_req(struct spclient *spc, size_t howmuch, void **resp)
463 {
464 struct rsp_hdr rhdr;
465 struct respwait rw;
466 struct iovec iov[2];
467 int rv;
468
469 DPRINTF(("anonmmap_req: %zu bytes\n", howmuch));
470
471 rhdr.rsp_len = sizeof(rhdr) + sizeof(howmuch);
472 rhdr.rsp_class = RUMPSP_REQ;
473 rhdr.rsp_type = RUMPSP_ANONMMAP;
474 rhdr.rsp_sysnum = 0;
475
476 IOVPUT(iov[0], rhdr);
477 IOVPUT(iov[1], howmuch);
478
479 putwait(spc, &rw, &rhdr);
480 rv = SENDIOV(spc, iov);
481 if (rv) {
482 unputwait(spc, &rw);
483 return rv;
484 }
485
486 rv = waitresp(spc, &rw);
487
488 *resp = rw.rw_data;
489
490 DPRINTF(("anonmmap: mapped at %p\n", **(void ***)resp));
491
492 return rv;
493 }
494
495 static int
496 send_raise_req(struct spclient *spc, int signo)
497 {
498 struct rsp_hdr rhdr;
499 struct iovec iov[1];
500 int rv;
501
502 rhdr.rsp_len = sizeof(rhdr);
503 rhdr.rsp_class = RUMPSP_REQ;
504 rhdr.rsp_type = RUMPSP_RAISE;
505 rhdr.rsp_signo = signo;
506
507 IOVPUT(iov[0], rhdr);
508
509 sendlock(spc);
510 rv = SENDIOV(spc, iov);
511 sendunlock(spc);
512
513 return rv;
514 }
515
516 static void
517 spcref(struct spclient *spc)
518 {
519
520 pthread_mutex_lock(&spc->spc_mtx);
521 spc->spc_refcnt++;
522 pthread_mutex_unlock(&spc->spc_mtx);
523 }
524
525 static void
526 spcrelease(struct spclient *spc)
527 {
528 int ref;
529
530 pthread_mutex_lock(&spc->spc_mtx);
531 ref = --spc->spc_refcnt;
532 if (__predict_false(spc->spc_inexec && ref <= 2))
533 pthread_cond_broadcast(&spc->spc_cv);
534 pthread_mutex_unlock(&spc->spc_mtx);
535
536 if (ref > 0)
537 return;
538
539 DPRINTF(("rump_sp: spcrelease: spc %p fd %d\n", spc, spc->spc_fd));
540
541 _DIAGASSERT(TAILQ_EMPTY(&spc->spc_respwait));
542 _DIAGASSERT(spc->spc_buf == NULL);
543
544 if (spc->spc_mainlwp) {
545 lwproc_switch(spc->spc_mainlwp);
546 lwproc_release();
547 }
548 spc->spc_mainlwp = NULL;
549
550 close(spc->spc_fd);
551 spc->spc_fd = -1;
552 spc->spc_state = SPCSTATE_NEW;
553
554 signaldisco();
555 }
556
557 static void
558 serv_handledisco(unsigned int idx)
559 {
560 struct spclient *spc = &spclist[idx];
561 int dolwpexit;
562
563 DPRINTF(("rump_sp: disconnecting [%u]\n", idx));
564
565 pfdlist[idx].fd = -1;
566 pfdlist[idx].revents = 0;
567 pthread_mutex_lock(&spc->spc_mtx);
568 spc->spc_state = SPCSTATE_DYING;
569 kickall(spc);
570 sendunlockl(spc);
571 /* exec uses mainlwp in another thread, but also nuked all lwps */
572 dolwpexit = !spc->spc_inexec;
573 pthread_mutex_unlock(&spc->spc_mtx);
574
575 if (dolwpexit && spc->spc_mainlwp) {
576 lwproc_switch(spc->spc_mainlwp);
577 lwproc_lwpexit();
578 lwproc_switch(NULL);
579 }
580
581 /*
582 * Nobody's going to attempt to send/receive anymore,
583 * so reinit info relevant to that.
584 */
585 /*LINTED:pointer casts may be ok*/
586 memset((char *)spc + SPC_ZEROFF, 0, sizeof(*spc) - SPC_ZEROFF);
587
588 spcrelease(spc);
589 }
590
591 static void
592 serv_shutdown(void)
593 {
594 struct spclient *spc;
595 unsigned int i;
596
597 for (i = 1; i < MAXCLI; i++) {
598 spc = &spclist[i];
599 if (spc->spc_fd == -1)
600 continue;
601
602 shutdown(spc->spc_fd, SHUT_RDWR);
603 serv_handledisco(i);
604
605 spcrelease(spc);
606 }
607 }
608
609 static unsigned
610 serv_handleconn(int fd, connecthook_fn connhook, int busy)
611 {
612 struct sockaddr_storage ss;
613 socklen_t sl = sizeof(ss);
614 int newfd, flags;
615 unsigned i;
616
617 /*LINTED: cast ok */
618 newfd = accept(fd, (struct sockaddr *)&ss, &sl);
619 if (newfd == -1)
620 return 0;
621
622 if (busy) {
623 close(newfd); /* EBUSY */
624 return 0;
625 }
626
627 flags = fcntl(newfd, F_GETFL, 0);
628 if (fcntl(newfd, F_SETFL, flags | O_NONBLOCK) == -1) {
629 close(newfd);
630 return 0;
631 }
632
633 if (connhook(newfd) != 0) {
634 close(newfd);
635 return 0;
636 }
637
638 /* write out a banner for the client */
639 if (send(newfd, banner, strlen(banner), MSG_NOSIGNAL)
640 != (ssize_t)strlen(banner)) {
641 close(newfd);
642 return 0;
643 }
644
645 /* find empty slot the simple way */
646 for (i = 0; i < MAXCLI; i++) {
647 if (pfdlist[i].fd == -1 && spclist[i].spc_state == SPCSTATE_NEW)
648 break;
649 }
650
651 assert(i < MAXCLI);
652
653 pfdlist[i].fd = newfd;
654 spclist[i].spc_fd = newfd;
655 spclist[i].spc_istatus = SPCSTATUS_BUSY; /* dedicated receiver */
656 spclist[i].spc_refcnt = 1;
657
658 TAILQ_INIT(&spclist[i].spc_respwait);
659
660 DPRINTF(("rump_sp: added new connection fd %d at idx %u\n", newfd, i));
661
662 return i;
663 }
664
665 static void
666 serv_handlesyscall(struct spclient *spc, struct rsp_hdr *rhdr, uint8_t *data)
667 {
668 register_t retval[2] = {0, 0};
669 int rv, sysnum;
670
671 sysnum = (int)rhdr->rsp_sysnum;
672 DPRINTF(("rump_sp: handling syscall %d from client %d\n",
673 sysnum, spc->spc_pid));
674
675 if (__predict_false((rv = lwproc_newlwp(spc->spc_pid)) != 0)) {
676 retval[0] = -1;
677 send_syscall_resp(spc, rhdr->rsp_reqno, rv, retval);
678 return;
679 }
680 spc->spc_syscallreq = rhdr->rsp_reqno;
681 rv = rumpsyscall(sysnum, data, retval);
682 spc->spc_syscallreq = 0;
683 lwproc_release();
684
685 DPRINTF(("rump_sp: got return value %d & %d/%d\n",
686 rv, retval[0], retval[1]));
687
688 send_syscall_resp(spc, rhdr->rsp_reqno, rv, retval);
689 }
690
691 static void
692 serv_handleexec(struct spclient *spc, struct rsp_hdr *rhdr, char *comm)
693 {
694 size_t commlen = rhdr->rsp_len - HDRSZ;
695
696 pthread_mutex_lock(&spc->spc_mtx);
697 /* one for the connection and one for us */
698 while (spc->spc_refcnt > 2)
699 pthread_cond_wait(&spc->spc_cv, &spc->spc_mtx);
700 pthread_mutex_unlock(&spc->spc_mtx);
701
702 /*
703 * ok, all the threads are dead (or one is still alive and
704 * the connection is dead, in which case this doesn't matter
705 * very much). proceed with exec.
706 */
707
708 /* ensure comm is 0-terminated */
709 /* TODO: make sure it contains sensible chars? */
710 comm[commlen] = '\0';
711
712 lwproc_switch(spc->spc_mainlwp);
713 lwproc_execnotify(comm);
714 lwproc_switch(NULL);
715
716 pthread_mutex_lock(&spc->spc_mtx);
717 spc->spc_inexec = 0;
718 pthread_mutex_unlock(&spc->spc_mtx);
719 send_handshake_resp(spc, rhdr->rsp_reqno, 0);
720 }
721
722 enum sbatype { SBA_SYSCALL, SBA_EXEC };
723
724 struct servbouncearg {
725 struct spclient *sba_spc;
726 struct rsp_hdr sba_hdr;
727 enum sbatype sba_type;
728 uint8_t *sba_data;
729
730 TAILQ_ENTRY(servbouncearg) sba_entries;
731 };
732 static pthread_mutex_t sbamtx;
733 static pthread_cond_t sbacv;
734 static int nworker, idleworker, nwork;
735 static TAILQ_HEAD(, servbouncearg) wrklist = TAILQ_HEAD_INITIALIZER(wrklist);
736
737 /*ARGSUSED*/
738 static void *
739 serv_workbouncer(void *arg)
740 {
741 struct servbouncearg *sba;
742
743 for (;;) {
744 pthread_mutex_lock(&sbamtx);
745 if (__predict_false(idleworker - nwork >= rumpsp_idleworker)) {
746 nworker--;
747 pthread_mutex_unlock(&sbamtx);
748 break;
749 }
750 idleworker++;
751 while (TAILQ_EMPTY(&wrklist)) {
752 _DIAGASSERT(nwork == 0);
753 pthread_cond_wait(&sbacv, &sbamtx);
754 }
755 idleworker--;
756
757 sba = TAILQ_FIRST(&wrklist);
758 TAILQ_REMOVE(&wrklist, sba, sba_entries);
759 nwork--;
760 pthread_mutex_unlock(&sbamtx);
761
762 if (__predict_true(sba->sba_type == SBA_SYSCALL)) {
763 serv_handlesyscall(sba->sba_spc,
764 &sba->sba_hdr, sba->sba_data);
765 } else {
766 _DIAGASSERT(sba->sba_type == SBA_EXEC);
767 serv_handleexec(sba->sba_spc, &sba->sba_hdr,
768 (char *)sba->sba_data);
769 }
770 spcrelease(sba->sba_spc);
771 free(sba->sba_data);
772 free(sba);
773 }
774
775 return NULL;
776 }
777
778 static int
779 sp_copyin(void *arg, const void *raddr, void *laddr, size_t *len, int wantstr)
780 {
781 struct spclient *spc = arg;
782 void *rdata = NULL; /* XXXuninit */
783 int rv, nlocks;
784
785 rumpkern_unsched(&nlocks, NULL);
786
787 rv = copyin_req(spc, raddr, len, wantstr, &rdata);
788 if (rv)
789 goto out;
790
791 memcpy(laddr, rdata, *len);
792 free(rdata);
793
794 out:
795 rumpkern_sched(nlocks, NULL);
796 if (rv)
797 return EFAULT;
798 return 0;
799 }
800
801 int
802 rumpuser_sp_copyin(void *arg, const void *raddr, void *laddr, size_t len)
803 {
804
805 return sp_copyin(arg, raddr, laddr, &len, 0);
806 }
807
808 int
809 rumpuser_sp_copyinstr(void *arg, const void *raddr, void *laddr, size_t *len)
810 {
811
812 return sp_copyin(arg, raddr, laddr, len, 1);
813 }
814
815 static int
816 sp_copyout(void *arg, const void *laddr, void *raddr, size_t dlen)
817 {
818 struct spclient *spc = arg;
819 int nlocks, rv;
820
821 rumpkern_unsched(&nlocks, NULL);
822 rv = send_copyout_req(spc, raddr, laddr, dlen);
823 rumpkern_sched(nlocks, NULL);
824
825 if (rv)
826 return EFAULT;
827 return 0;
828 }
829
830 int
831 rumpuser_sp_copyout(void *arg, const void *laddr, void *raddr, size_t dlen)
832 {
833
834 return sp_copyout(arg, laddr, raddr, dlen);
835 }
836
837 int
838 rumpuser_sp_copyoutstr(void *arg, const void *laddr, void *raddr, size_t *dlen)
839 {
840
841 return sp_copyout(arg, laddr, raddr, *dlen);
842 }
843
844 int
845 rumpuser_sp_anonmmap(void *arg, size_t howmuch, void **addr)
846 {
847 struct spclient *spc = arg;
848 void *resp, *rdata;
849 int nlocks, rv;
850
851 rumpkern_unsched(&nlocks, NULL);
852
853 rv = anonmmap_req(spc, howmuch, &rdata);
854 if (rv) {
855 rv = EFAULT;
856 goto out;
857 }
858
859 resp = *(void **)rdata;
860 free(rdata);
861
862 if (resp == NULL) {
863 rv = ENOMEM;
864 }
865
866 *addr = resp;
867
868 out:
869 rumpkern_sched(nlocks, NULL);
870
871 if (rv)
872 return rv;
873 return 0;
874 }
875
876 int
877 rumpuser_sp_raise(void *arg, int signo)
878 {
879 struct spclient *spc = arg;
880 int rv, nlocks;
881
882 rumpkern_unsched(&nlocks, NULL);
883 rv = send_raise_req(spc, signo);
884 rumpkern_sched(nlocks, NULL);
885
886 return rv;
887 }
888
889 static pthread_attr_t pattr_detached;
890 static void
891 schedulework(struct spclient *spc, enum sbatype sba_type)
892 {
893 struct servbouncearg *sba;
894 pthread_t pt;
895 uint64_t reqno;
896 int retries = 0;
897
898 reqno = spc->spc_hdr.rsp_reqno;
899 while ((sba = malloc(sizeof(*sba))) == NULL) {
900 if (nworker == 0 || retries > 10) {
901 send_error_resp(spc, reqno, RUMPSP_ERR_TRYAGAIN);
902 spcfreebuf(spc);
903 return;
904 }
905 /* slim chance of more memory? */
906 usleep(10000);
907 }
908
909 sba->sba_spc = spc;
910 sba->sba_type = sba_type;
911 sba->sba_hdr = spc->spc_hdr;
912 sba->sba_data = spc->spc_buf;
913 spcresetbuf(spc);
914
915 spcref(spc);
916
917 pthread_mutex_lock(&sbamtx);
918 TAILQ_INSERT_TAIL(&wrklist, sba, sba_entries);
919 nwork++;
920 if (nwork <= idleworker) {
921 /* do we have a daemon's tool (i.e. idle threads)? */
922 pthread_cond_signal(&sbacv);
923 } else if (nworker < rumpsp_maxworker) {
924 /*
925 * Else, need to create one
926 * (if we can, otherwise just expect another
927 * worker to pick up the syscall)
928 */
929 if (pthread_create(&pt, &pattr_detached,
930 serv_workbouncer, NULL) == 0) {
931 nworker++;
932 }
933 }
934 pthread_mutex_unlock(&sbamtx);
935 }
936
937 /*
938 *
939 * Startup routines and mainloop for server.
940 *
941 */
942
943 struct spservarg {
944 int sps_sock;
945 connecthook_fn sps_connhook;
946 };
947
948 static void
949 handlereq(struct spclient *spc)
950 {
951 uint64_t reqno;
952 int error;
953
954 reqno = spc->spc_hdr.rsp_reqno;
955 if (__predict_false(spc->spc_state == SPCSTATE_NEW)) {
956 if (spc->spc_hdr.rsp_type != RUMPSP_HANDSHAKE) {
957 send_error_resp(spc, reqno, RUMPSP_ERR_AUTH);
958 shutdown(spc->spc_fd, SHUT_RDWR);
959 spcfreebuf(spc);
960 return;
961 }
962
963 if (spc->spc_hdr.rsp_handshake == HANDSHAKE_GUEST) {
964 char *comm = (char *)spc->spc_buf;
965 size_t commlen = spc->spc_hdr.rsp_len - HDRSZ;
966
967 /* ensure it's 0-terminated */
968 /* XXX make sure it contains sensible chars? */
969 comm[commlen] = '\0';
970
971 if ((error = lwproc_rfork(spc,
972 RUMP_RFCFDG, comm)) != 0) {
973 shutdown(spc->spc_fd, SHUT_RDWR);
974 }
975
976 spcfreebuf(spc);
977 if (error)
978 return;
979
980 spc->spc_mainlwp = lwproc_curlwp();
981
982 send_handshake_resp(spc, reqno, 0);
983 } else if (spc->spc_hdr.rsp_handshake == HANDSHAKE_FORK) {
984 struct lwp *tmpmain;
985 struct prefork *pf;
986 struct handshake_fork *rfp;
987 int cancel;
988
989 if (spc->spc_off-HDRSZ != sizeof(*rfp)) {
990 send_error_resp(spc, reqno,
991 RUMPSP_ERR_MALFORMED_REQUEST);
992 shutdown(spc->spc_fd, SHUT_RDWR);
993 spcfreebuf(spc);
994 return;
995 }
996
997 /*LINTED*/
998 rfp = (void *)spc->spc_buf;
999 cancel = rfp->rf_cancel;
1000
1001 pthread_mutex_lock(&pfmtx);
1002 LIST_FOREACH(pf, &preforks, pf_entries) {
1003 if (memcmp(rfp->rf_auth, pf->pf_auth,
1004 sizeof(rfp->rf_auth)) == 0) {
1005 LIST_REMOVE(pf, pf_entries);
1006 LIST_REMOVE(pf, pf_spcentries);
1007 break;
1008 }
1009 }
1010 pthread_mutex_lock(&pfmtx);
1011 spcfreebuf(spc);
1012
1013 if (!pf) {
1014 send_error_resp(spc, reqno,
1015 RUMPSP_ERR_INVALID_PREFORK);
1016 shutdown(spc->spc_fd, SHUT_RDWR);
1017 return;
1018 }
1019
1020 tmpmain = pf->pf_lwp;
1021 free(pf);
1022 lwproc_switch(tmpmain);
1023 if (cancel) {
1024 lwproc_release();
1025 shutdown(spc->spc_fd, SHUT_RDWR);
1026 return;
1027 }
1028
1029 /*
1030 * So, we forked already during "prefork" to save
1031 * the file descriptors from a parent exit
1032 * race condition. But now we need to fork
1033 * a second time since the initial fork has
1034 * the wrong spc pointer. (yea, optimize
1035 * interfaces some day if anyone cares)
1036 */
1037 if ((error = lwproc_rfork(spc, 0, NULL)) != 0) {
1038 send_error_resp(spc, reqno,
1039 RUMPSP_ERR_RFORK_FAILED);
1040 shutdown(spc->spc_fd, SHUT_RDWR);
1041 lwproc_release();
1042 return;
1043 }
1044 spc->spc_mainlwp = lwproc_curlwp();
1045 lwproc_switch(tmpmain);
1046 lwproc_release();
1047 lwproc_switch(spc->spc_mainlwp);
1048
1049 send_handshake_resp(spc, reqno, 0);
1050 } else {
1051 send_error_resp(spc, reqno, RUMPSP_ERR_AUTH);
1052 shutdown(spc->spc_fd, SHUT_RDWR);
1053 spcfreebuf(spc);
1054 return;
1055 }
1056
1057 spc->spc_pid = lwproc_getpid();
1058
1059 DPRINTF(("rump_sp: handshake for client %p complete, pid %d\n",
1060 spc, spc->spc_pid));
1061
1062 lwproc_switch(NULL);
1063 spc->spc_state = SPCSTATE_RUNNING;
1064 return;
1065 }
1066
1067 if (__predict_false(spc->spc_hdr.rsp_type == RUMPSP_PREFORK)) {
1068 struct prefork *pf;
1069 uint32_t auth[AUTHLEN];
1070 int inexec;
1071
1072 DPRINTF(("rump_sp: prefork handler executing for %p\n", spc));
1073 spcfreebuf(spc);
1074
1075 pthread_mutex_lock(&spc->spc_mtx);
1076 inexec = spc->spc_inexec;
1077 pthread_mutex_unlock(&spc->spc_mtx);
1078 if (inexec) {
1079 send_error_resp(spc, reqno, RUMPSP_ERR_INEXEC);
1080 shutdown(spc->spc_fd, SHUT_RDWR);
1081 return;
1082 }
1083
1084 pf = malloc(sizeof(*pf));
1085 if (pf == NULL) {
1086 send_error_resp(spc, reqno, RUMPSP_ERR_NOMEM);
1087 return;
1088 }
1089
1090 /*
1091 * Use client main lwp to fork. this is never used by
1092 * worker threads (except in exec, but we checked for that
1093 * above) so we can safely use it here.
1094 */
1095 lwproc_switch(spc->spc_mainlwp);
1096 if ((error = lwproc_rfork(spc, RUMP_RFFDG, NULL)) != 0) {
1097 DPRINTF(("rump_sp: fork failed: %d (%p)\n",error, spc));
1098 send_error_resp(spc, reqno, RUMPSP_ERR_RFORK_FAILED);
1099 lwproc_switch(NULL);
1100 free(pf);
1101 return;
1102 }
1103
1104 /* Ok, we have a new process context and a new curlwp */
1105 rumpuser_getrandom(auth, sizeof(auth), 0);
1106 memcpy(pf->pf_auth, auth, sizeof(pf->pf_auth));
1107 pf->pf_lwp = lwproc_curlwp();
1108 lwproc_switch(NULL);
1109
1110 pthread_mutex_lock(&pfmtx);
1111 LIST_INSERT_HEAD(&preforks, pf, pf_entries);
1112 LIST_INSERT_HEAD(&spc->spc_pflist, pf, pf_spcentries);
1113 pthread_mutex_unlock(&pfmtx);
1114
1115 DPRINTF(("rump_sp: prefork handler success %p\n", spc));
1116
1117 send_prefork_resp(spc, reqno, auth);
1118 return;
1119 }
1120
1121 if (__predict_false(spc->spc_hdr.rsp_type == RUMPSP_HANDSHAKE)) {
1122 int inexec;
1123
1124 if (spc->spc_hdr.rsp_handshake != HANDSHAKE_EXEC) {
1125 send_error_resp(spc, reqno,
1126 RUMPSP_ERR_MALFORMED_REQUEST);
1127 shutdown(spc->spc_fd, SHUT_RDWR);
1128 spcfreebuf(spc);
1129 return;
1130 }
1131
1132 pthread_mutex_lock(&spc->spc_mtx);
1133 inexec = spc->spc_inexec;
1134 pthread_mutex_unlock(&spc->spc_mtx);
1135 if (inexec) {
1136 send_error_resp(spc, reqno, RUMPSP_ERR_INEXEC);
1137 shutdown(spc->spc_fd, SHUT_RDWR);
1138 spcfreebuf(spc);
1139 return;
1140 }
1141
1142 pthread_mutex_lock(&spc->spc_mtx);
1143 spc->spc_inexec = 1;
1144 pthread_mutex_unlock(&spc->spc_mtx);
1145
1146 /*
1147 * start to drain lwps. we will wait for it to finish
1148 * in another thread
1149 */
1150 lwproc_switch(spc->spc_mainlwp);
1151 lwproc_lwpexit();
1152 lwproc_switch(NULL);
1153
1154 /*
1155 * exec has to wait for lwps to drain, so finish it off
1156 * in another thread
1157 */
1158 schedulework(spc, SBA_EXEC);
1159 return;
1160 }
1161
1162 if (__predict_false(spc->spc_hdr.rsp_type != RUMPSP_SYSCALL)) {
1163 send_error_resp(spc, reqno, RUMPSP_ERR_MALFORMED_REQUEST);
1164 spcfreebuf(spc);
1165 return;
1166 }
1167
1168 schedulework(spc, SBA_SYSCALL);
1169 }
1170
1171 static void *
1172 spserver(void *arg)
1173 {
1174 struct spservarg *sarg = arg;
1175 struct spclient *spc;
1176 unsigned idx;
1177 int seen;
1178 int rv;
1179 unsigned int nfds, maxidx;
1180
1181 for (idx = 0; idx < MAXCLI; idx++) {
1182 pfdlist[idx].fd = -1;
1183 pfdlist[idx].events = POLLIN;
1184
1185 spc = &spclist[idx];
1186 pthread_mutex_init(&spc->spc_mtx, NULL);
1187 pthread_cond_init(&spc->spc_cv, NULL);
1188 spc->spc_fd = -1;
1189 }
1190 pfdlist[0].fd = spclist[0].spc_fd = sarg->sps_sock;
1191 pfdlist[0].events = POLLIN;
1192 nfds = 1;
1193 maxidx = 0;
1194
1195 pthread_attr_init(&pattr_detached);
1196 pthread_attr_setdetachstate(&pattr_detached, PTHREAD_CREATE_DETACHED);
1197 #if NOTYET
1198 pthread_attr_setstacksize(&pattr_detached, 32*1024);
1199 #endif
1200
1201 pthread_mutex_init(&sbamtx, NULL);
1202 pthread_cond_init(&sbacv, NULL);
1203
1204 DPRINTF(("rump_sp: server mainloop\n"));
1205
1206 for (;;) {
1207 int discoed;
1208
1209 /* g/c hangarounds (eventually) */
1210 discoed = getdisco();
1211 while (discoed--) {
1212 nfds--;
1213 idx = maxidx;
1214 while (idx) {
1215 if (pfdlist[idx].fd != -1) {
1216 maxidx = idx;
1217 break;
1218 }
1219 idx--;
1220 }
1221 DPRINTF(("rump_sp: set maxidx to [%u]\n",
1222 maxidx));
1223 }
1224
1225 DPRINTF(("rump_sp: loop nfd %d\n", maxidx+1));
1226 seen = 0;
1227 rv = poll(pfdlist, maxidx+1, INFTIM);
1228 assert(maxidx+1 <= MAXCLI);
1229 assert(rv != 0);
1230 if (rv == -1) {
1231 if (errno == EINTR)
1232 continue;
1233 fprintf(stderr, "rump_spserver: poll returned %d\n",
1234 errno);
1235 break;
1236 }
1237
1238 for (idx = 0; seen < rv && idx < MAXCLI; idx++) {
1239 if ((pfdlist[idx].revents & POLLIN) == 0)
1240 continue;
1241
1242 seen++;
1243 DPRINTF(("rump_sp: activity at [%u] %d/%d\n",
1244 idx, seen, rv));
1245 if (idx > 0) {
1246 spc = &spclist[idx];
1247 DPRINTF(("rump_sp: mainloop read [%u]\n", idx));
1248 switch (readframe(spc)) {
1249 case 0:
1250 break;
1251 case -1:
1252 serv_handledisco(idx);
1253 break;
1254 default:
1255 switch (spc->spc_hdr.rsp_class) {
1256 case RUMPSP_RESP:
1257 kickwaiter(spc);
1258 break;
1259 case RUMPSP_REQ:
1260 handlereq(spc);
1261 break;
1262 default:
1263 send_error_resp(spc,
1264 spc->spc_hdr.rsp_reqno,
1265 RUMPSP_ERR_MALFORMED_REQUEST);
1266 spcfreebuf(spc);
1267 break;
1268 }
1269 break;
1270 }
1271
1272 } else {
1273 DPRINTF(("rump_sp: mainloop new connection\n"));
1274
1275 if (__predict_false(spfini)) {
1276 close(spclist[0].spc_fd);
1277 serv_shutdown();
1278 goto out;
1279 }
1280
1281 idx = serv_handleconn(pfdlist[0].fd,
1282 sarg->sps_connhook, nfds == MAXCLI);
1283 if (idx)
1284 nfds++;
1285 if (idx > maxidx)
1286 maxidx = idx;
1287 DPRINTF(("rump_sp: maxid now %d\n", maxidx));
1288 }
1289 }
1290 }
1291
1292 out:
1293 return NULL;
1294 }
1295
1296 static unsigned cleanupidx;
1297 static struct sockaddr *cleanupsa;
1298 int
1299 rumpuser_sp_init(const char *url,
1300 const char *ostype, const char *osrelease, const char *machine)
1301 {
1302 pthread_t pt;
1303 struct spservarg *sarg;
1304 struct sockaddr *sap;
1305 char *p;
1306 unsigned idx;
1307 int error, s;
1308
1309 p = strdup(url);
1310 if (p == NULL)
1311 return ENOMEM;
1312 error = parseurl(p, &sap, &idx, 1);
1313 free(p);
1314 if (error)
1315 return error;
1316
1317 snprintf(banner, sizeof(banner), "RUMPSP-%d.%d-%s-%s/%s\n",
1318 PROTOMAJOR, PROTOMINOR, ostype, osrelease, machine);
1319
1320 s = socket(parsetab[idx].domain, SOCK_STREAM, 0);
1321 if (s == -1)
1322 return errno;
1323
1324 sarg = malloc(sizeof(*sarg));
1325 if (sarg == NULL) {
1326 close(s);
1327 return ENOMEM;
1328 }
1329
1330 sarg->sps_sock = s;
1331 sarg->sps_connhook = parsetab[idx].connhook;
1332
1333 cleanupidx = idx;
1334 cleanupsa = sap;
1335
1336 /* sloppy error recovery */
1337
1338 /*LINTED*/
1339 if (bind(s, sap, parsetab[idx].slen) == -1) {
1340 fprintf(stderr, "rump_sp: server bind failed\n");
1341 return errno;
1342 }
1343
1344 if (listen(s, MAXCLI) == -1) {
1345 fprintf(stderr, "rump_sp: server listen failed\n");
1346 return errno;
1347 }
1348
1349 if ((error = pthread_create(&pt, NULL, spserver, sarg)) != 0) {
1350 fprintf(stderr, "rump_sp: cannot create wrkr thread\n");
1351 return errno;
1352 }
1353 pthread_detach(pt);
1354
1355 return 0;
1356 }
1357
1358 void
1359 rumpuser_sp_fini(void *arg)
1360 {
1361 struct spclient *spc = arg;
1362 register_t retval[2] = {0, 0};
1363
1364 if (spclist[0].spc_fd) {
1365 parsetab[cleanupidx].cleanup(cleanupsa);
1366 }
1367
1368 /*
1369 * stuff response into the socket, since this process is just
1370 * about to exit
1371 */
1372 if (spc && spc->spc_syscallreq)
1373 send_syscall_resp(spc, spc->spc_syscallreq, 0, retval);
1374
1375 if (spclist[0].spc_fd) {
1376 shutdown(spclist[0].spc_fd, SHUT_RDWR);
1377 spfini = 1;
1378 }
1379 }
1380