Home | History | Annotate | Line # | Download | only in librumpuser
rumpuser_sp.c revision 1.14
      1 /*      $NetBSD: rumpuser_sp.c,v 1.14 2010/11/24 20:29:13 pooka Exp $	*/
      2 
      3 /*
      4  * Copyright (c) 2010 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  * Not finished yet, i.e. don't use in production.  Lacks locking plus
     37  * handling of multiple clients and unexpected connection closes.
     38  */
     39 
     40 #include <sys/cdefs.h>
     41 __RCSID("$NetBSD: rumpuser_sp.c,v 1.14 2010/11/24 20:29:13 pooka Exp $");
     42 
     43 #include <sys/types.h>
     44 #include <sys/atomic.h>
     45 #include <sys/mman.h>
     46 #include <sys/socket.h>
     47 
     48 #include <arpa/inet.h>
     49 #include <netinet/in.h>
     50 #include <netinet/tcp.h>
     51 
     52 #include <assert.h>
     53 #include <errno.h>
     54 #include <fcntl.h>
     55 #include <poll.h>
     56 #include <pthread.h>
     57 #include <stdarg.h>
     58 #include <stdio.h>
     59 #include <stdlib.h>
     60 #include <string.h>
     61 #include <unistd.h>
     62 
     63 #include <rump/rumpuser.h>
     64 #include "rumpuser_int.h"
     65 
     66 #include "sp_common.c"
     67 
     68 #define MAXCLI 4
     69 
     70 static struct pollfd pfdlist[MAXCLI];
     71 static struct spclient spclist[MAXCLI];
     72 static unsigned int disco;
     73 
     74 static struct rumpuser_sp_ops spops;
     75 
     76 /*
     77  * Manual wrappers, since librump does not have access to the
     78  * user namespace wrapped interfaces.
     79  */
     80 
     81 static void
     82 lwproc_switch(struct lwp *l)
     83 {
     84 
     85 	spops.spop_schedule();
     86 	spops.spop_lwproc_switch(l);
     87 	spops.spop_unschedule();
     88 }
     89 
     90 static void
     91 lwproc_release(void)
     92 {
     93 
     94 	spops.spop_schedule();
     95 	spops.spop_lwproc_release();
     96 	spops.spop_unschedule();
     97 }
     98 
     99 static int
    100 lwproc_newproc(struct spclient *spc)
    101 {
    102 	int rv;
    103 
    104 	spops.spop_schedule();
    105 	rv = spops.spop_lwproc_newproc(spc);
    106 	spops.spop_unschedule();
    107 
    108 	return rv;
    109 }
    110 
    111 static int
    112 lwproc_newlwp(pid_t pid)
    113 {
    114 	int rv;
    115 
    116 	spops.spop_schedule();
    117 	rv = spops.spop_lwproc_newlwp(pid);
    118 	spops.spop_unschedule();
    119 
    120 	return rv;
    121 }
    122 
    123 static struct lwp *
    124 lwproc_curlwp(void)
    125 {
    126 	struct lwp *l;
    127 
    128 	spops.spop_schedule();
    129 	l = spops.spop_lwproc_curlwp();
    130 	spops.spop_unschedule();
    131 
    132 	return l;
    133 }
    134 
    135 static pid_t
    136 lwproc_getpid(void)
    137 {
    138 	pid_t p;
    139 
    140 	spops.spop_schedule();
    141 	p = spops.spop_getpid();
    142 	spops.spop_unschedule();
    143 
    144 	return p;
    145 }
    146 
    147 static int
    148 rumpsyscall(int sysnum, void *data, register_t *retval)
    149 {
    150 	int rv;
    151 
    152 	spops.spop_schedule();
    153 	rv = spops.spop_syscall(sysnum, data, retval);
    154 	spops.spop_unschedule();
    155 
    156 	return rv;
    157 }
    158 
    159 static uint64_t
    160 nextreq(struct spclient *spc)
    161 {
    162 	uint64_t nw;
    163 
    164 	pthread_mutex_lock(&spc->spc_mtx);
    165 	nw = spc->spc_nextreq++;
    166 	pthread_mutex_unlock(&spc->spc_mtx);
    167 
    168 	return nw;
    169 }
    170 
    171 static int
    172 send_syscall_resp(struct spclient *spc, uint64_t reqno, int error,
    173 	register_t *retval)
    174 {
    175 	struct rsp_hdr rhdr;
    176 	struct rsp_sysresp sysresp;
    177 	int rv;
    178 
    179 	rhdr.rsp_len = sizeof(rhdr) + sizeof(sysresp);
    180 	rhdr.rsp_reqno = reqno;
    181 	rhdr.rsp_class = RUMPSP_RESP;
    182 	rhdr.rsp_type = RUMPSP_SYSCALL;
    183 	rhdr.rsp_sysnum = 0;
    184 
    185 	sysresp.rsys_error = error;
    186 	memcpy(sysresp.rsys_retval, retval, sizeof(sysresp.rsys_retval));
    187 
    188 	sendlock(spc);
    189 	rv = dosend(spc, &rhdr, sizeof(rhdr));
    190 	rv = dosend(spc, &sysresp, sizeof(sysresp));
    191 	sendunlock(spc);
    192 
    193 	return rv;
    194 }
    195 
    196 static int
    197 copyin_req(struct spclient *spc, const void *remaddr, size_t dlen, void **resp)
    198 {
    199 	struct rsp_hdr rhdr;
    200 	struct rsp_copydata copydata;
    201 	struct respwait rw;
    202 	int rv;
    203 
    204 	DPRINTF(("copyin_req: %zu bytes from %p\n", dlen, remaddr));
    205 
    206 	rhdr.rsp_len = sizeof(rhdr) + sizeof(copydata);
    207 	rhdr.rsp_class = RUMPSP_REQ;
    208 	rhdr.rsp_type = RUMPSP_COPYIN;
    209 	rhdr.rsp_sysnum = 0;
    210 
    211 	copydata.rcp_addr = __UNCONST(remaddr);
    212 	copydata.rcp_len = dlen;
    213 
    214 	putwait(spc, &rw, &rhdr);
    215 	rv = dosend(spc, &rhdr, sizeof(rhdr));
    216 	rv = dosend(spc, &copydata, sizeof(copydata));
    217 	if (rv) {
    218 		unputwait(spc, &rw);
    219 		return rv;
    220 	}
    221 
    222 	rv = waitresp(spc, &rw);
    223 
    224 	DPRINTF(("copyin: response %d\n", rv));
    225 
    226 	*resp = rw.rw_data;
    227 	return rv;
    228 
    229 }
    230 
    231 static int
    232 send_copyout_req(struct spclient *spc, const void *remaddr,
    233 	const void *data, size_t dlen)
    234 {
    235 	struct rsp_hdr rhdr;
    236 	struct rsp_copydata copydata;
    237 	int rv;
    238 
    239 	DPRINTF(("copyout_req (async): %zu bytes to %p\n", dlen, remaddr));
    240 
    241 	rhdr.rsp_len = sizeof(rhdr) + sizeof(copydata) + dlen;
    242 	rhdr.rsp_reqno = nextreq(spc);
    243 	rhdr.rsp_class = RUMPSP_REQ;
    244 	rhdr.rsp_type = RUMPSP_COPYOUT;
    245 	rhdr.rsp_sysnum = 0;
    246 
    247 	copydata.rcp_addr = __UNCONST(remaddr);
    248 	copydata.rcp_len = dlen;
    249 
    250 	sendlock(spc);
    251 	rv = dosend(spc, &rhdr, sizeof(rhdr));
    252 	rv = dosend(spc, &copydata, sizeof(copydata));
    253 	rv = dosend(spc, data, dlen);
    254 	sendunlock(spc);
    255 
    256 	return rv;
    257 }
    258 
    259 static int
    260 anonmmap_req(struct spclient *spc, size_t howmuch, void **resp)
    261 {
    262 	struct rsp_hdr rhdr;
    263 	struct respwait rw;
    264 	int rv;
    265 
    266 	DPRINTF(("anonmmap_req: %zu bytes\n", howmuch));
    267 
    268 	rhdr.rsp_len = sizeof(rhdr) + sizeof(howmuch);
    269 	rhdr.rsp_class = RUMPSP_REQ;
    270 	rhdr.rsp_type = RUMPSP_ANONMMAP;
    271 	rhdr.rsp_sysnum = 0;
    272 
    273 	putwait(spc, &rw, &rhdr);
    274 	rv = dosend(spc, &rhdr, sizeof(rhdr));
    275 	rv = dosend(spc, &howmuch, sizeof(howmuch));
    276 	if (rv) {
    277 		unputwait(spc, &rw);
    278 		return rv;
    279 	}
    280 
    281 	rv = waitresp(spc, &rw);
    282 
    283 	*resp = rw.rw_data;
    284 
    285 	DPRINTF(("anonmmap: mapped at %p\n", **(void ***)resp));
    286 
    287 	return rv;
    288 }
    289 
    290 static void
    291 spcref(struct spclient *spc)
    292 {
    293 
    294 	pthread_mutex_lock(&spc->spc_mtx);
    295 	spc->spc_refcnt++;
    296 	pthread_mutex_unlock(&spc->spc_mtx);
    297 }
    298 
    299 static void
    300 spcrelease(struct spclient *spc)
    301 {
    302 	int ref;
    303 
    304 	pthread_mutex_lock(&spc->spc_mtx);
    305 	ref = --spc->spc_refcnt;
    306 	pthread_mutex_unlock(&spc->spc_mtx);
    307 
    308 	if (ref > 0)
    309 		return;
    310 
    311 	DPRINTF(("spcrelease: spc %p fd %d\n", spc, spc->spc_fd));
    312 
    313 	_DIAGASSERT(TAILQ_EMPTY(&spc->spc_respwait));
    314 	_DIAGASSERT(spc->spc_buf == NULL);
    315 
    316 	lwproc_switch(spc->spc_mainlwp);
    317 	lwproc_release();
    318 	spc->spc_mainlwp = NULL;
    319 
    320 	close(spc->spc_fd);
    321 	spc->spc_fd = -1;
    322 	spc->spc_dying = 0;
    323 
    324 	atomic_inc_uint(&disco);
    325 
    326 }
    327 
    328 static void
    329 serv_handledisco(unsigned int idx)
    330 {
    331 	struct spclient *spc = &spclist[idx];
    332 
    333 	DPRINTF(("rump_sp: disconnecting [%u]\n", idx));
    334 
    335 	pfdlist[idx].fd = -1;
    336 	pfdlist[idx].revents = 0;
    337 	pthread_mutex_lock(&spc->spc_mtx);
    338 	spc->spc_dying = 1;
    339 	kickall(spc);
    340 	pthread_mutex_unlock(&spc->spc_mtx);
    341 	spcrelease(spc);
    342 }
    343 
    344 static unsigned
    345 serv_handleconn(int fd, connecthook_fn connhook, int busy)
    346 {
    347 	struct sockaddr_storage ss;
    348 	socklen_t sl = sizeof(ss);
    349 	int newfd, flags;
    350 	unsigned i;
    351 
    352 	/*LINTED: cast ok */
    353 	newfd = accept(fd, (struct sockaddr *)&ss, &sl);
    354 	if (newfd == -1)
    355 		return 0;
    356 
    357 	if (busy) {
    358 		close(newfd); /* EBUSY */
    359 		return 0;
    360 	}
    361 
    362 	/* XXX: should do some sort of handshake too */
    363 
    364 	flags = fcntl(newfd, F_GETFL, 0);
    365 	if (fcntl(newfd, F_SETFL, flags | O_NONBLOCK) == -1) {
    366 		close(newfd);
    367 		return 0;
    368 	}
    369 
    370 	if (connhook(newfd) != 0) {
    371 		close(newfd);
    372 		return 0;
    373 	}
    374 
    375 	/* find empty slot the simple way */
    376 	for (i = 0; i < MAXCLI; i++) {
    377 		if (pfdlist[i].fd == -1 && spclist[i].spc_dying == 0)
    378 			break;
    379 	}
    380 
    381 	if (lwproc_newproc(&spclist[i]) != 0) {
    382 		close(newfd);
    383 		return 0;
    384 	}
    385 
    386 	assert(i < MAXCLI);
    387 
    388 	pfdlist[i].fd = newfd;
    389 	spclist[i].spc_fd = newfd;
    390 	spclist[i].spc_mainlwp = lwproc_curlwp();
    391 	spclist[i].spc_istatus = SPCSTATUS_BUSY; /* dedicated receiver */
    392 	spclist[i].spc_pid = lwproc_getpid();
    393 	spclist[i].spc_refcnt = 1;
    394 
    395 	TAILQ_INIT(&spclist[i].spc_respwait);
    396 
    397 	DPRINTF(("rump_sp: added new connection fd %d at idx %u, pid %d\n",
    398 	    newfd, i, lwproc_getpid()));
    399 
    400 	lwproc_switch(NULL);
    401 
    402 	return i;
    403 }
    404 
    405 static void
    406 serv_handlesyscall(struct spclient *spc, struct rsp_hdr *rhdr, uint8_t *data)
    407 {
    408 	register_t retval[2] = {0, 0};
    409 	int rv, sysnum;
    410 
    411 	sysnum = (int)rhdr->rsp_sysnum;
    412 	DPRINTF(("rump_sp: handling syscall %d from client %d\n",
    413 	    sysnum, 0));
    414 
    415 	lwproc_newlwp(spc->spc_pid);
    416 	rv = rumpsyscall(sysnum, data, retval);
    417 	lwproc_switch(NULL);
    418 	free(data);
    419 
    420 	DPRINTF(("rump_sp: got return value %d & %d/%d\n",
    421 	    rv, retval[0], retval[1]));
    422 
    423 	send_syscall_resp(spc, rhdr->rsp_reqno, rv, retval);
    424 }
    425 
    426 struct sysbouncearg {
    427 	struct spclient *sba_spc;
    428 	struct rsp_hdr sba_hdr;
    429 	uint8_t *sba_data;
    430 };
    431 static void *
    432 serv_syscallbouncer(void *arg)
    433 {
    434 	struct sysbouncearg *barg = arg;
    435 
    436 	serv_handlesyscall(barg->sba_spc, &barg->sba_hdr, barg->sba_data);
    437 	spcrelease(barg->sba_spc);
    438 	free(arg);
    439 	return NULL;
    440 }
    441 
    442 int
    443 rumpuser_sp_copyin(void *arg, const void *uaddr, void *kaddr, size_t len)
    444 {
    445 	struct spclient *spc = arg;
    446 	void *rdata = NULL; /* XXXuninit */
    447 	int rv, nlocks;
    448 
    449 	rumpuser__kunlock(0, &nlocks, NULL);
    450 
    451 	rv = copyin_req(spc, uaddr, len, &rdata);
    452 	if (rv)
    453 		goto out;
    454 
    455 	memcpy(kaddr, rdata, len);
    456 	free(rdata);
    457 
    458  out:
    459 	rumpuser__klock(nlocks, NULL);
    460 	if (rv)
    461 		return EFAULT;
    462 	return 0;
    463 }
    464 
    465 int
    466 rumpuser_sp_copyout(void *arg, const void *kaddr, void *uaddr, size_t dlen)
    467 {
    468 	struct spclient *spc = arg;
    469 	int nlocks, rv;
    470 
    471 	rumpuser__kunlock(0, &nlocks, NULL);
    472 	rv = send_copyout_req(spc, uaddr, kaddr, dlen);
    473 	rumpuser__klock(nlocks, NULL);
    474 
    475 	if (rv)
    476 		return EFAULT;
    477 	return 0;
    478 }
    479 
    480 int
    481 rumpuser_sp_anonmmap(void *arg, size_t howmuch, void **addr)
    482 {
    483 	struct spclient *spc = arg;
    484 	void *resp, *rdata;
    485 	int nlocks, rv;
    486 
    487 	rumpuser__kunlock(0, &nlocks, NULL);
    488 
    489 	rv = anonmmap_req(spc, howmuch, &rdata);
    490 	if (rv) {
    491 		rv = EFAULT;
    492 		goto out;
    493 	}
    494 
    495 	resp = *(void **)rdata;
    496 	free(rdata);
    497 
    498 	if (resp == NULL) {
    499 		rv = ENOMEM;
    500 	}
    501 
    502 	*addr = resp;
    503 
    504  out:
    505 	rumpuser__klock(nlocks, NULL);
    506 
    507 	if (rv)
    508 		return rv;
    509 	return 0;
    510 }
    511 
    512 /*
    513  *
    514  * Startup routines and mainloop for server.
    515  *
    516  */
    517 
    518 struct spservarg {
    519 	int sps_sock;
    520 	connecthook_fn sps_connhook;
    521 };
    522 
    523 static pthread_attr_t pattr_detached;
    524 static void
    525 handlereq(struct spclient *spc)
    526 {
    527 	struct sysbouncearg *sba;
    528 	pthread_t pt;
    529 	int rv;
    530 
    531 	/* XXX: check that it's a syscall */
    532 
    533 	sba = malloc(sizeof(*sba));
    534 	if (sba == NULL) {
    535 		/* panic */
    536 		abort();
    537 	}
    538 
    539 	sba->sba_spc = spc;
    540 	sba->sba_hdr = spc->spc_hdr;
    541 	sba->sba_data = spc->spc_buf;
    542 
    543 	spc->spc_buf = NULL;
    544 	spc->spc_off = 0;
    545 
    546 	spcref(spc);
    547 	if ((rv = pthread_create(&pt, &pattr_detached,
    548 	    serv_syscallbouncer, sba)) != 0) {
    549 		/* panic */
    550 		abort();
    551 	}
    552 }
    553 
    554 static void *
    555 spserver(void *arg)
    556 {
    557 	struct spservarg *sarg = arg;
    558 	struct spclient *spc;
    559 	unsigned idx;
    560 	int seen;
    561 	int rv;
    562 	unsigned int nfds, maxidx;
    563 
    564 	for (idx = 0; idx < MAXCLI; idx++) {
    565 		pfdlist[idx].fd = -1;
    566 		pfdlist[idx].events = POLLIN;
    567 
    568 		spc = &spclist[idx];
    569 		pthread_mutex_init(&spc->spc_mtx, NULL);
    570 		pthread_cond_init(&spc->spc_cv, NULL);
    571 	}
    572 	pfdlist[0].fd = sarg->sps_sock;
    573 	pfdlist[0].events = POLLIN;
    574 	nfds = 1;
    575 	maxidx = 0;
    576 
    577 	pthread_attr_init(&pattr_detached);
    578 	pthread_attr_setdetachstate(&pattr_detached, PTHREAD_CREATE_DETACHED);
    579 
    580 	DPRINTF(("rump_sp: server mainloop\n"));
    581 
    582 	for (;;) {
    583 		/* g/c hangarounds (eventually) */
    584 		if (disco) {
    585 			int discoed;
    586 
    587 			discoed = atomic_swap_uint(&disco, 0);
    588 			while (discoed--) {
    589 				nfds--;
    590 				idx = maxidx;
    591 				while (idx) {
    592 					if (pfdlist[idx].fd != -1) {
    593 						maxidx = idx;
    594 						break;
    595 					}
    596 					idx--;
    597 				}
    598 				DPRINTF(("rump_sp: set maxidx to [%u]\n",
    599 				    maxidx));
    600 				assert(maxidx+1 >= nfds);
    601 			}
    602 		}
    603 
    604 		DPRINTF(("rump_sp: loop nfd %d\n", maxidx+1));
    605 		seen = 0;
    606 		rv = poll(pfdlist, maxidx+1, INFTIM);
    607 		assert(maxidx+1 <= MAXCLI);
    608 		assert(rv != 0);
    609 		if (rv == -1) {
    610 			if (errno == EINTR)
    611 				continue;
    612 			fprintf(stderr, "rump_spserver: poll returned %d\n",
    613 			    errno);
    614 			break;
    615 		}
    616 
    617 		for (idx = 0; seen < rv && idx < MAXCLI; idx++) {
    618 			if ((pfdlist[idx].revents & POLLIN) == 0)
    619 				continue;
    620 
    621 			seen++;
    622 			DPRINTF(("rump_sp: activity at [%u] %d/%d\n",
    623 			    idx, seen, rv));
    624 			if (idx > 0) {
    625 				spc = &spclist[idx];
    626 				DPRINTF(("rump_sp: mainloop read [%u]\n", idx));
    627 				switch (readframe(spc)) {
    628 				case 0:
    629 					break;
    630 				case -1:
    631 					serv_handledisco(idx);
    632 					break;
    633 				default:
    634 					switch (spc->spc_hdr.rsp_class) {
    635 					case RUMPSP_RESP:
    636 						kickwaiter(spc);
    637 						break;
    638 					case RUMPSP_REQ:
    639 						handlereq(spc);
    640 						break;
    641 					default:
    642 						printf("PANIC\n");
    643 						abort();
    644 						break;
    645 					}
    646 					break;
    647 				}
    648 
    649 			} else {
    650 				DPRINTF(("rump_sp: mainloop new connection\n"));
    651 
    652 				idx = serv_handleconn(pfdlist[0].fd,
    653 				    sarg->sps_connhook, nfds == MAXCLI);
    654 				if (idx)
    655 					nfds++;
    656 				if (idx > maxidx)
    657 					maxidx = idx;
    658 				DPRINTF(("rump_sp: maxid now %d\n", maxidx));
    659 			}
    660 		}
    661 	}
    662 
    663 	return NULL;
    664 }
    665 
    666 int
    667 rumpuser_sp_init(const struct rumpuser_sp_ops *spopsp, const char *url)
    668 {
    669 	pthread_t pt;
    670 	struct spservarg *sarg;
    671 	struct sockaddr *sap;
    672 	char *p;
    673 	unsigned idx;
    674 	int error, s;
    675 
    676 	p = strdup(url);
    677 	if (p == NULL)
    678 		return ENOMEM;
    679 	error = parseurl(p, &sap, &idx, 1);
    680 	free(p);
    681 	if (error)
    682 		return error;
    683 
    684 	s = socket(parsetab[idx].domain, SOCK_STREAM, 0);
    685 	if (s == -1)
    686 		return errno;
    687 
    688 	spops = *spopsp;
    689 	sarg = malloc(sizeof(*sarg));
    690 	if (sarg == NULL) {
    691 		close(s);
    692 		return ENOMEM;
    693 	}
    694 
    695 	sarg->sps_sock = s;
    696 	sarg->sps_connhook = parsetab[idx].connhook;
    697 
    698 	/* sloppy error recovery */
    699 
    700 	/*LINTED*/
    701 	if (bind(s, sap, sap->sa_len) == -1) {
    702 		fprintf(stderr, "rump_sp: server bind failed\n");
    703 		return errno;
    704 	}
    705 	if (listen(s, 20) == -1) {
    706 		fprintf(stderr, "rump_sp: server listen failed\n");
    707 		return errno;
    708 	}
    709 
    710 	if ((error = pthread_create(&pt, NULL, spserver, sarg)) != 0) {
    711 		fprintf(stderr, "rump_sp: cannot create wrkr thread\n");
    712 		return errno;
    713 	}
    714 	pthread_detach(pt);
    715 
    716 	return 0;
    717 }
    718