Rework socket upcalls to close some races with setup/teardown of upcalls.

- Each socket upcall is now invoked with the appropriate socket buffer locked. It is not permissible to call soisconnected() with this lock held; however, so socket upcalls now return an integer value. The two possible values are SU_OK and SU_ISCONNECTED. If an upcall returns SU_ISCONNECTED, then the soisconnected() will be invoked on the socket after the socket buffer lock is dropped. - A new API is provided for setting and clearing socket upcalls. The API consists of soupcall_set() and soupcall_clear(). - To simplify locking, each socket buffer now has a separate upcall. - When a socket upcall returns SU_ISCONNECTED, the upcall is cleared from the receive socket buffer automatically. Note that a SO_SND upcall should never return SU_ISCONNECTED. - All this means that accept filters should now return SU_ISCONNECTED instead of calling soisconnected() directly. They also no longer need to explicitly clear the upcall on the new socket. - The HTTP accept filter still uses soupcall_set() to manage its internal state machine, but other accept filters no longer have any explicit knowlege of socket upcall internals aside from their return value. - The various RPC client upcalls currently drop the socket buffer lock while invoking soreceive() as a temporary band-aid. The plan for the future is to add a new flag to allow soreceive() to be called with the socket buffer locked. - The AIO callback for socket I/O is now also invoked with the socket buffer locked. Previously sowakeup() would drop the socket buffer lock only to call aio_swake() which immediately re-acquired the socket buffer lock for the duration of the function call. Discussed with: rwatson, rmacklem
svn path=/head/; revision=193272
2009-06-01 21:17:03 +00:00 · 2009-06-01 21:17:03 +00:00 · 74fb0ba732 · 2020-12-20 02:59:44 +00:00
commit 74fb0ba732
parent b2bc853659
20 changed files with 234 additions and 195 deletions
--- a/sys/dev/cxgb/ulp/iw_cxgb/iw_cxgb_cm.c
+++ b/sys/dev/cxgb/ulp/iw_cxgb/iw_cxgb_cm.c
@ -141,7 +141,7 @@ SYSCTL_UINT(_hw_cxgb, OID_AUTO, cong_flavor, CTLFLAG_RDTUN, &cong_flavor, 0,

 static void ep_timeout(void *arg);
 static void connect_reply_upcall(struct iwch_ep *ep, int status);
-static void iwch_so_upcall(struct socket *so, void *arg, int waitflag);
+static int iwch_so_upcall(struct socket *so, void *arg, int waitflag);

 /*
 * Cruft to offload socket upcalls onto thread.
@ -335,9 +335,7 @@ close_socket(struct iwch_ep_common *epc)
 {
 	CTR4(KTR_IW_CXGB, "%s ep %p so %p state %s", __FUNCTION__, epc, epc->so, states[epc->state]);
 	SOCK_LOCK(epc->so);
-	epc->so->so_upcall = NULL;
-	epc->so->so_upcallarg = NULL;
-	epc->so->so_rcv.sb_flags &= ~SB_UPCALL;
+	soupcall_clear(epc->so, SO_RCV);
 	SOCK_UNLOCK(epc->so);
 	soshutdown(epc->so, SHUT_WR|SHUT_RD);
 	epc->so = NULL;
@ -1108,7 +1106,7 @@ terminate(struct t3cdev *tdev, struct mbuf *m, void *ctx)
 {
 	struct toepcb *toep = (struct toepcb *)ctx;
 	struct socket *so = toeptoso(toep);
-	struct iwch_ep *ep = so->so_upcallarg;
+	struct iwch_ep *ep = so->so_rcv.sb_upcallarg;

 	CTR2(KTR_IW_CXGB, "%s ep %p", __FUNCTION__, ep);
 	m_adj(m, sizeof(struct cpl_rdma_terminate));
@ -1129,7 +1127,7 @@ ec_status(struct t3cdev *tdev, struct mbuf *m, void *ctx)
 	struct iwch_qp_attributes attrs;
 	int release = 0;

-	ep = so->so_upcallarg;
+	ep = so->so_rcv.sb_upcallarg;
 	CTR5(KTR_IW_CXGB, "%s ep %p so %p state %s ec_status %d", __FUNCTION__, ep, ep->com.so, states[ep->com.state], rep->status);
 	if (!so || !ep) {
 		panic("bogosity ep %p state %d, so %p state %x\n", ep, ep ? ep->com.state : -1, so, so ? so->so_state : -1); 
@ -1309,10 +1307,10 @@ static int init_sock(struct iwch_ep_common *epc)
 	struct sockopt sopt;
 	int on=1;

-	epc->so->so_upcall = iwch_so_upcall;
-	epc->so->so_upcallarg = epc;
-	epc->so->so_rcv.sb_flags |= SB_UPCALL;
+	SOCK_LOCK(epc->so);
+	soupcall_set(epc->so, SO_RCV, iwch_so_upcall, epc);
 	epc->so->so_state |= SS_NBIO;
+	SOCK_UNLOCK(epc->so);
 	sopt.sopt_dir = SOPT_SET;
 	sopt.sopt_level = SOL_SOCKET;
 	sopt.sopt_name = SO_NO_DDP;
@ -1611,10 +1609,8 @@ dequeue_socket(struct socket *head, struct sockaddr_in **remote, struct iwch_ep
 	so->so_qstate &= ~SQ_COMP;
 	so->so_head = NULL;
 	soref(so);
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, iwch_so_upcall, child_ep);
 	so->so_state |= SS_NBIO;
-	so->so_upcall = iwch_so_upcall;
-	so->so_upcallarg = child_ep;
 	PANIC_IF(!(so->so_state & SS_ISCONNECTED));
 	PANIC_IF(so->so_error);
 	SOCK_UNLOCK(so);
@ -1661,7 +1657,7 @@ process_newconn(struct iwch_ep *parent_ep)
 	process_mpa_request(child_ep);
 }

-static void
+static int
 iwch_so_upcall(struct socket *so, void *arg, int waitflag)
 {
 	struct iwch_ep *ep = arg;
@ -1674,6 +1670,7 @@ iwch_so_upcall(struct socket *so, void *arg, int waitflag)
 		taskqueue_enqueue(iw_cxgb_taskq, &iw_cxgb_task);
 	}
 	mtx_unlock(&req_lock);
+	return (SU_OK);
 }

 static void
--- a/sys/kern/uipc_sockbuf.c
+++ b/sys/kern/uipc_sockbuf.c
@ -175,6 +175,7 @@ sbunlock(struct sockbuf *sb)
 void
 sowakeup(struct socket *so, struct sockbuf *sb)
 {
+	int ret;

 	SOCKBUF_LOCK_ASSERT(sb);

@ -186,13 +187,22 @@ sowakeup(struct socket *so, struct sockbuf *sb)
 		wakeup(&sb->sb_cc);
 	}
 	KNOTE_LOCKED(&sb->sb_sel.si_note, 0);
-	SOCKBUF_UNLOCK(sb);
-	if ((so->so_state & SS_ASYNC) && so->so_sigio != NULL)
-		pgsigio(&so->so_sigio, SIGIO, 0);
-	if (sb->sb_flags & SB_UPCALL)
-		(*so->so_upcall)(so, so->so_upcallarg, M_DONTWAIT);
+	if (sb->sb_upcall != NULL) {
+		ret = sb->sb_upcall(so, sb->sb_upcallarg, M_DONTWAIT);
+		if (ret == SU_ISCONNECTED) {
+			KASSERT(sb == &so->so_rcv,
+			    ("SO_SND upcall returned SU_ISCONNECTED"));
+			soupcall_clear(so, SO_RCV);
+		}
+	} else
+		ret = SU_OK;
 	if (sb->sb_flags & SB_AIO)
 		aio_swake(so, sb);
+	SOCKBUF_UNLOCK(sb);
+	if (ret == SU_ISCONNECTED)
+		soisconnected(so);
+	if ((so->so_state & SS_ASYNC) && so->so_sigio != NULL)
+		pgsigio(&so->so_sigio, SIGIO, 0);
 	mtx_assert(SOCKBUF_MTX(sb), MA_NOTOWNED);
 }

--- a/sys/kern/uipc_socket.c
+++ b/sys/kern/uipc_socket.c
@ -3054,8 +3054,10 @@ soisconnecting(struct socket *so)
 void
 soisconnected(struct socket *so)
 {
-	struct socket *head;
+	struct socket *head;	
+	int ret;

+restart:
 	ACCEPT_LOCK();
 	SOCK_LOCK(so);
 	so->so_state &= ~(SS_ISCONNECTING|SS_ISDISCONNECTING|SS_ISCONFIRMING);
@ -3075,13 +3077,17 @@ soisconnected(struct socket *so)
 			wakeup_one(&head->so_timeo);
 		} else {
 			ACCEPT_UNLOCK();
-			so->so_upcall =
-			    head->so_accf->so_accept_filter->accf_callback;
-			so->so_upcallarg = head->so_accf->so_accept_filter_arg;
-			so->so_rcv.sb_flags |= SB_UPCALL;
+			soupcall_set(so, SO_RCV,
+			    head->so_accf->so_accept_filter->accf_callback,
+			    head->so_accf->so_accept_filter_arg);
 			so->so_options &= ~SO_ACCEPTFILTER;
+			ret = head->so_accf->so_accept_filter->accf_callback(so,
+			    head->so_accf->so_accept_filter_arg, M_DONTWAIT);
+			if (ret == SU_ISCONNECTED)
+				soupcall_clear(so, SO_RCV);
 			SOCK_UNLOCK(so);
-			so->so_upcall(so, so->so_upcallarg, M_DONTWAIT);
+			if (ret == SU_ISCONNECTED)
+				goto restart;
 		}
 		return;
 	}
@ -3145,6 +3151,57 @@ sodupsockaddr(const struct sockaddr *sa, int mflags)
 	return sa2;
 }

+/*
+ * Register per-socket buffer upcalls.
+ */
+void
+soupcall_set(struct socket *so, int which,
+    int (*func)(struct socket *, void *, int), void *arg)
+{
+	struct sockbuf *sb;
+	
+	switch (which) {
+	case SO_RCV:
+		sb = &so->so_rcv;
+		break;
+	case SO_SND:
+		sb = &so->so_snd;
+		break;
+	default:
+		panic("soupcall_set: bad which");
+	}
+	SOCKBUF_LOCK_ASSERT(sb);
+#if 0
+	/* XXX: accf_http actually wants to do this on purpose. */
+	KASSERT(sb->sb_upcall == NULL, ("soupcall_set: overwriting upcall"));
+#endif
+	sb->sb_upcall = func;
+	sb->sb_upcallarg = arg;
+	sb->sb_flags |= SB_UPCALL;
+}
+
+void
+soupcall_clear(struct socket *so, int which)
+{
+	struct sockbuf *sb;
+
+	switch (which) {
+	case SO_RCV:
+		sb = &so->so_rcv;
+		break;
+	case SO_SND:
+		sb = &so->so_snd;
+		break;
+	default:
+		panic("soupcall_clear: bad which");
+	}
+	SOCKBUF_LOCK_ASSERT(sb);
+	KASSERT(sb->sb_upcall != NULL, ("soupcall_clear: no upcall to clear"));
+	sb->sb_upcall = NULL;
+	sb->sb_upcallarg = NULL;
+	sb->sb_flags &= ~SB_UPCALL;
+}
+
 /*
 * Create an external-format (``xsocket'') structure using the information in
 * the kernel-format socket structure pointed to by so.  This is done to
--- a/sys/kern/vfs_aio.c
+++ b/sys/kern/vfs_aio.c
@ -1313,12 +1313,12 @@ aio_swake_cb(struct socket *so, struct sockbuf *sb)
 	struct aiocblist *cb, *cbn;
 	int opcode;

+	SOCKBUF_LOCK_ASSERT(sb);
 	if (sb == &so->so_snd)
 		opcode = LIO_WRITE;
 	else
 		opcode = LIO_READ;

-	SOCKBUF_LOCK(sb);
 	sb->sb_flags &= ~SB_AIO;
 	mtx_lock(&aio_job_mtx);
 	TAILQ_FOREACH_SAFE(cb, &so->so_aiojobq, list, cbn) {
@ -1336,7 +1336,6 @@ aio_swake_cb(struct socket *so, struct sockbuf *sb)
 		}
 	}
 	mtx_unlock(&aio_job_mtx);
-	SOCKBUF_UNLOCK(sb);
 }

 static int
--- a/sys/netgraph/bluetooth/socket/ng_btsocket_rfcomm.c
+++ b/sys/netgraph/bluetooth/socket/ng_btsocket_rfcomm.c
@ -93,7 +93,7 @@ MALLOC_DEFINE(M_NETGRAPH_BTSOCKET_RFCOMM, "netgraph_btsocks_rfcomm",
 #define	ALOT	0x7fff

 /* Local prototypes */
-static void ng_btsocket_rfcomm_upcall
+static int ng_btsocket_rfcomm_upcall
 	(struct socket *so, void *arg, int waitflag);
 static void ng_btsocket_rfcomm_sessions_task
 	(void *ctx, int pending);
@ -1007,7 +1007,7 @@ ng_btsocket_rfcomm_sockaddr(struct socket *so, struct sockaddr **nam)
 * Upcall function for L2CAP sockets. Enqueue RFCOMM task.
 */

-static void
+static int
 ng_btsocket_rfcomm_upcall(struct socket *so, void *arg, int waitflag)
 {
 	int	error;
@ -1018,6 +1018,7 @@ ng_btsocket_rfcomm_upcall(struct socket *so, void *arg, int waitflag)
 	if ((error = ng_btsocket_rfcomm_task_wakeup()) != 0)
 		NG_BTSOCKET_RFCOMM_ALERT(
 "%s: Could not enqueue RFCOMM task, error=%d\n", __func__, error);
+	return (SU_OK);
 } /* ng_btsocket_rfcomm_upcall */

 /*
@ -1047,13 +1048,11 @@ ng_btsocket_rfcomm_sessions_task(void *ctx, int pending)
 				panic("%s: DLC list is not empty\n", __func__);

 			/* Close L2CAP socket */
-			s->l2so->so_upcallarg = NULL;
-			s->l2so->so_upcall = NULL;
 			SOCKBUF_LOCK(&s->l2so->so_rcv);
-			s->l2so->so_rcv.sb_flags &= ~SB_UPCALL;
+			soupcall_clear(s->l2so, SO_RCV);
 			SOCKBUF_UNLOCK(&s->l2so->so_rcv);
 			SOCKBUF_LOCK(&s->l2so->so_snd);
-			s->l2so->so_snd.sb_flags &= ~SB_UPCALL;
+			soupcall_clear(s->l2so, SO_SND);
 			SOCKBUF_UNLOCK(&s->l2so->so_snd);
 			soclose(s->l2so);

@ -1286,13 +1285,11 @@ ng_btsocket_rfcomm_session_create(ng_btsocket_rfcomm_session_p *sp,
 	LIST_INIT(&s->dlcs);

 	/* Prepare L2CAP socket */
-	l2so->so_upcallarg = NULL;
-	l2so->so_upcall = ng_btsocket_rfcomm_upcall;
 	SOCKBUF_LOCK(&l2so->so_rcv);
-	l2so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(l2so, SO_RCV, ng_btsocket_rfcomm_upcall, NULL);
 	SOCKBUF_UNLOCK(&l2so->so_rcv);
 	SOCKBUF_LOCK(&l2so->so_snd);
-	l2so->so_snd.sb_flags |= SB_UPCALL;
+	soupcall_set(l2so, SO_SND, ng_btsocket_rfcomm_upcall, NULL);
 	SOCKBUF_UNLOCK(&l2so->so_snd);
 	l2so->so_state |= SS_NBIO;
 	s->l2so = l2so;
@ -1370,13 +1367,11 @@ ng_btsocket_rfcomm_session_create(ng_btsocket_rfcomm_session_p *sp,
 	mtx_unlock(&s->session_mtx);

 	/* Return L2CAP socket back to its original state */
-	l2so->so_upcallarg = NULL;
-	l2so->so_upcall = NULL;
 	SOCKBUF_LOCK(&l2so->so_rcv);
-	l2so->so_rcv.sb_flags &= ~SB_UPCALL;
+	soupcall_clear(s->l2so, SO_RCV);
 	SOCKBUF_UNLOCK(&l2so->so_rcv);
 	SOCKBUF_LOCK(&l2so->so_snd);
-	l2so->so_snd.sb_flags &= ~SB_UPCALL;
+	soupcall_clear(s->l2so, SO_SND);
 	SOCKBUF_UNLOCK(&l2so->so_snd);
 	l2so->so_state &= ~SS_NBIO;

--- a/sys/netgraph/ng_ksocket.c
+++ b/sys/netgraph/ng_ksocket.c
@ -158,7 +158,7 @@ static const struct ng_ksocket_alias ng_ksocket_protos[] = {
 /* Helper functions */
 static int	ng_ksocket_check_accept(priv_p);
 static void	ng_ksocket_finish_accept(priv_p);
-static void	ng_ksocket_incoming(struct socket *so, void *arg, int waitflag);
+static int	ng_ksocket_incoming(struct socket *so, void *arg, int waitflag);
 static int	ng_ksocket_parse(const struct ng_ksocket_alias *aliases,
 			const char *s, int family);
 static void	ng_ksocket_incoming2(node_p node, hook_p hook,
@ -616,13 +616,11 @@ ng_ksocket_connect(hook_p hook)
 	struct socket *const so = priv->so;

 	/* Add our hook for incoming data and other events */
-	priv->so->so_upcallarg = (caddr_t)node;
-	priv->so->so_upcall = ng_ksocket_incoming;
 	SOCKBUF_LOCK(&priv->so->so_rcv);
-	priv->so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(priv->so, SO_RCV, ng_ksocket_incoming, node);
 	SOCKBUF_UNLOCK(&priv->so->so_rcv);
 	SOCKBUF_LOCK(&priv->so->so_snd);
-	priv->so->so_snd.sb_flags |= SB_UPCALL;
+	soupcall_set(priv->so, SO_SND, ng_ksocket_incoming, node);
 	SOCKBUF_UNLOCK(&priv->so->so_snd);
 	SOCK_LOCK(priv->so);
 	priv->so->so_state |= SS_NBIO;
@ -941,12 +939,11 @@ ng_ksocket_shutdown(node_p node)
 	/* Close our socket (if any) */
 	if (priv->so != NULL) {
 		SOCKBUF_LOCK(&priv->so->so_rcv);
-		priv->so->so_rcv.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(priv->so, SO_RCV);
 		SOCKBUF_UNLOCK(&priv->so->so_rcv);
 		SOCKBUF_LOCK(&priv->so->so_snd);
-		priv->so->so_snd.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(priv->so, SO_SND);
 		SOCKBUF_UNLOCK(&priv->so->so_snd);
-		priv->so->so_upcall = NULL;
 		soclose(priv->so);
 		priv->so = NULL;
 	}
@ -1000,7 +997,7 @@ ng_ksocket_disconnect(hook_p hook)
 * To decouple stack, we use queue version of ng_send_fn().
 */

-static void
+static int
 ng_ksocket_incoming(struct socket *so, void *arg, int waitflag)
 {
 	const node_p node = arg;
@ -1017,6 +1014,7 @@ ng_ksocket_incoming(struct socket *so, void *arg, int waitflag)
 	    ng_send_fn1(node, NULL, &ng_ksocket_incoming2, so, 0, wait)) {
 		atomic_store_rel_int(&priv->fn_sent, 0);
 	}
+	return (SU_OK);
 }


@ -1258,13 +1256,11 @@ ng_ksocket_finish_accept(priv_p priv)
 	 */
 	LIST_INSERT_HEAD(&priv->embryos, priv2, siblings);

-	so->so_upcallarg = (caddr_t)node;
-	so->so_upcall = ng_ksocket_incoming;
 	SOCKBUF_LOCK(&so->so_rcv);
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, ng_ksocket_incoming, node);
 	SOCKBUF_UNLOCK(&so->so_rcv);
 	SOCKBUF_LOCK(&so->so_snd);
-	so->so_snd.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, ng_ksocket_incoming, node);
 	SOCKBUF_UNLOCK(&so->so_snd);

 	/* Fill in the response data and send it or return it to the caller */
--- a/sys/netinet/accf_data.c
+++ b/sys/netinet/accf_data.c
@ -38,7 +38,7 @@ __FBSDID("$FreeBSD$");

 /* accept filter that holds a socket until data arrives */

-static void	sohasdata(struct socket *so, void *arg, int waitflag);
+static int	sohasdata(struct socket *so, void *arg, int waitflag);

 static struct accept_filter accf_data_filter = {
 	"dataready",
@ -55,15 +55,12 @@ static moduledata_t accf_data_mod = {

 DECLARE_MODULE(accf_data, accf_data_mod, SI_SUB_DRIVERS, SI_ORDER_MIDDLE);

-static void
+static int
 sohasdata(struct socket *so, void *arg, int waitflag)
 {

 	if (!soreadable(so))
-		return;
+		return (SU_OK);

-	so->so_upcall = NULL;
-	so->so_rcv.sb_flags &= ~SB_UPCALL;
-	soisconnected(so);
-	return;
+	return (SU_ISCONNECTED);
 }
--- a/sys/netinet/accf_dns.c
+++ b/sys/netinet/accf_dns.c
@ -37,7 +37,7 @@
 #include <sys/socketvar.h>

 /* check for full DNS request */
-static void sohasdns(struct socket *so, void *arg, int waitflag);
+static int sohasdns(struct socket *so, void *arg, int waitflag);

 struct packet {
 	struct mbuf *m;		/* Current mbuf. */
@ -69,7 +69,7 @@ static moduledata_t accf_dns_mod = {

 DECLARE_MODULE(accf_dns, accf_dns_mod, SI_SUB_DRIVERS, SI_ORDER_MIDDLE);

-static void
+static int
 sohasdns(struct socket *so, void *arg, int waitflag)
 {
 	struct sockbuf *sb = &so->so_rcv;
@ -80,13 +80,10 @@ sohasdns(struct socket *so, void *arg, int waitflag)

 	/* Check to see if we have a request. */
 	if (skippacket(sb) == DNS_WAIT)
-		return;
+		return (SU_OK);

 ready:
-	so->so_upcall = NULL;
-	so->so_rcv.sb_flags &= ~SB_UPCALL;
-	soisconnected(so);
-	return;
+	return (SU_ISCONNECTED);
 }

 #define GET8(p, val) do { \
--- a/sys/netinet/accf_http.c
+++ b/sys/netinet/accf_http.c
@ -39,11 +39,11 @@ __FBSDID("$FreeBSD$");
 #include <sys/socketvar.h>

 /* check for GET/HEAD */
-static void sohashttpget(struct socket *so, void *arg, int waitflag);
+static int sohashttpget(struct socket *so, void *arg, int waitflag);
 /* check for HTTP/1.0 or HTTP/1.1 */
-static void soparsehttpvers(struct socket *so, void *arg, int waitflag);
+static int soparsehttpvers(struct socket *so, void *arg, int waitflag);
 /* check for end of HTTP/1.x request */
-static void soishttpconnected(struct socket *so, void *arg, int waitflag);
+static int soishttpconnected(struct socket *so, void *arg, int waitflag);
 /* strcmp on an mbuf chain */
 static int mbufstrcmp(struct mbuf *m, struct mbuf *npkt, int offset, char *cmp);
 /* strncmp on an mbuf chain */
@ -158,7 +158,7 @@ mbufstrncmp(struct mbuf *m, struct mbuf *npkt, int offset, int max, char *cmp)
 		slen = sizeof(str) - 1;					\
 	} while(0)

-static void
+static int
 sohashttpget(struct socket *so, void *arg, int waitflag)
 {

@ -170,7 +170,7 @@ sohashttpget(struct socket *so, void *arg, int waitflag)
 		m = so->so_rcv.sb_mb;
 		cc = so->so_rcv.sb_cc - 1;
 		if (cc < 1)
-			return;
+			return (SU_OK);
 		switch (*mtod(m, char *)) {
 		case 'G':
 			STRSETUP(cmp, cmplen, "ET ");
@ -184,7 +184,7 @@ sohashttpget(struct socket *so, void *arg, int waitflag)
 		if (cc < cmplen) {
 			if (mbufstrncmp(m, m->m_nextpkt, 1, cc, cmp) == 1) {
 				DPRINT("short cc (%d) but mbufstrncmp ok", cc);
-				return;
+				return (SU_OK);
 			} else {
 				DPRINT("short cc (%d) mbufstrncmp failed", cc);
 				goto fallout;
@ -193,23 +193,19 @@ sohashttpget(struct socket *so, void *arg, int waitflag)
 		if (mbufstrcmp(m, m->m_nextpkt, 1, cmp) == 1) {
 			DPRINT("mbufstrcmp ok");
 			if (parse_http_version == 0)
-				soishttpconnected(so, arg, waitflag);
+				return (soishttpconnected(so, arg, waitflag));
 			else
-				soparsehttpvers(so, arg, waitflag);
-			return;
+				return (soparsehttpvers(so, arg, waitflag));
 		}
 		DPRINT("mbufstrcmp bad");
 	}

 fallout:
 	DPRINT("fallout");
-	so->so_upcall = NULL;
-	so->so_rcv.sb_flags &= ~SB_UPCALL;
-	soisconnected(so);
-	return;
+	return (SU_ISCONNECTED);
 }

-static void
+static int
 soparsehttpvers(struct socket *so, void *arg, int waitflag)
 {
 	struct mbuf *m, *n;
@ -261,10 +257,9 @@ soparsehttpvers(struct socket *so, void *arg, int waitflag)
 					} else if (
 					    mbufstrcmp(m, n, i, "HTTP/1.0") ||
 					    mbufstrcmp(m, n, i, "HTTP/1.1")) {
-							DPRINT("ok");
-							soishttpconnected(so,
-							    arg, waitflag);
-							return;
+						DPRINT("ok");
+						return (soishttpconnected(so,
+						    arg, waitflag));
 					} else {
 						DPRINT("bad");
 						goto fallout;
@ -279,22 +274,18 @@ soparsehttpvers(struct socket *so, void *arg, int waitflag)
 	 * if we hit here we haven't hit something
 	 * we don't understand or a newline, so try again
 	 */
-	so->so_upcall = soparsehttpvers;
-	so->so_rcv.sb_flags |= SB_UPCALL;
-	return;
+	soupcall_set(so, SO_RCV, soparsehttpvers, arg);
+	return (SU_OK);

 fallout:
 	DPRINT("fallout");
-	so->so_upcall = NULL;
-	so->so_rcv.sb_flags &= ~SB_UPCALL;
-	soisconnected(so);
-	return;
+	return (SU_ISCONNECTED);
 }


 #define NCHRS 3

-static void
+static int
 soishttpconnected(struct socket *so, void *arg, int waitflag)
 {
 	char a, b, c;
@ -350,13 +341,9 @@ soishttpconnected(struct socket *so, void *arg, int waitflag)
 	}

 readmore:
-	so->so_upcall = soishttpconnected;
-	so->so_rcv.sb_flags |= SB_UPCALL;
-	return;
+	soupcall_set(so, SO_RCV, soishttpconnected, arg);
+	return (SU_OK);

 gotit:
-	so->so_upcall = NULL;
-	so->so_rcv.sb_flags &= ~SB_UPCALL;
-	soisconnected(so);
-	return;
+	return (SU_ISCONNECTED);
 }
--- a/sys/netsmb/smb_trantcp.c
+++ b/sys/netsmb/smb_trantcp.c
@ -100,14 +100,15 @@ nb_intr(struct nbpcb *nbp, struct proc *p)
 	return 0;
 }

-static void
+static int
 nb_upcall(struct socket *so, void *arg, int waitflag)
 {
 	struct nbpcb *nbp = arg;

 	if (arg == NULL || nbp->nbp_selectid == NULL)
-		return;
+		return (SU_OK);
 	wakeup(nbp->nbp_selectid);
+	return (SU_OK);
 }

 static int
@ -152,10 +153,8 @@ nb_connect_in(struct nbpcb *nbp, struct sockaddr_in *to, struct thread *td)
 	if (error)
 		return error;
 	nbp->nbp_tso = so;
-	so->so_upcallarg = (caddr_t)nbp;
-	so->so_upcall = nb_upcall;
 	SOCKBUF_LOCK(&so->so_rcv);
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, nb_upcall, nbp);
 	SOCKBUF_UNLOCK(&so->so_rcv);
 	so->so_rcv.sb_timeo = (5 * hz);
 	so->so_snd.sb_timeo = (5 * hz);
--- a/sys/nfsclient/nfs_socket.c
+++ b/sys/nfsclient/nfs_socket.c
@ -122,8 +122,8 @@ static int	nfs_realign(struct mbuf **pm, int hsiz);
 static int	nfs_reply(struct nfsreq *);
 static void	nfs_softterm(struct nfsreq *rep);
 static int	nfs_reconnect(struct nfsreq *rep);
-static void nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag);
-static void nfs_clnt_udp_soupcall(struct socket *so, void *arg, int waitflag);
+static int	nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag);
+static int	nfs_clnt_udp_soupcall(struct socket *so, void *arg, int waitflag);

 extern struct mtx nfs_reqq_mtx;

@ -457,12 +457,10 @@ nfs_connect(struct nfsmount *nmp, struct nfsreq *rep)
 		goto bad;
 	SOCKBUF_LOCK(&so->so_rcv);
 	so->so_rcv.sb_flags |= SB_NOINTR;
- 	so->so_upcallarg = (caddr_t)nmp;
 	if (so->so_type == SOCK_STREAM)
- 		so->so_upcall = nfs_clnt_tcp_soupcall;
- 	else	
- 		so->so_upcall = nfs_clnt_udp_soupcall;
- 	so->so_rcv.sb_flags |= SB_UPCALL;
+		soupcall_set(so, SO_RCV, nfs_clnt_tcp_soupcall, nmp);
+ 	else
+		soupcall_set(so, SO_RCV, nfs_clnt_udp_soupcall, nmp);
 	SOCKBUF_UNLOCK(&so->so_rcv);
 	SOCKBUF_LOCK(&so->so_snd);
 	so->so_snd.sb_flags |= SB_NOINTR;
@ -600,9 +598,7 @@ nfs_disconnect(struct nfsmount *nmp)
 		nmp->nm_so = NULL;
 		mtx_unlock(&nmp->nm_mtx);
 		SOCKBUF_LOCK(&so->so_rcv);
- 		so->so_upcallarg = NULL;
- 		so->so_upcall = NULL;
- 		so->so_rcv.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(so, SO_RCV);
 		SOCKBUF_UNLOCK(&so->so_rcv);
 		soshutdown(so, SHUT_WR);
 		soclose(so);
@ -811,7 +807,7 @@ nfs_reply(struct nfsreq *rep)
 * XXX TO DO
 * Make nfs_realign() non-blocking. Also make nfsm_dissect() nonblocking.
 */
-static void
+static int
 nfs_clnt_match_xid(struct socket *so, 
 		   struct nfsmount *nmp, 
 		   struct mbuf *mrep)
@ -928,11 +924,10 @@ nfstcp_readable(struct socket *so, int bytes)
 {
 	int retval;
 	
-	SOCKBUF_LOCK(&so->so_rcv);
+	SOCKBUF_LOCK_ASSERT(&so->so_rcv);
 	retval = (so->so_rcv.sb_cc >= (bytes) ||
 		  (so->so_rcv.sb_state & SBS_CANTRCVMORE) ||
 		  so->so_error);
-	SOCKBUF_UNLOCK(&so->so_rcv);
 	return (retval);
 }

@ -969,7 +964,7 @@ nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag)
 	mtx_lock(&nmp->nm_mtx);
 	if (nmp->nm_nfstcpstate.flags & NFS_TCP_FORCE_RECONNECT) {
 		mtx_unlock(&nmp->nm_mtx);		
-		return;
+		return (SU_OK);
 	} else			
 		mtx_unlock(&nmp->nm_mtx);
 	auio.uio_td = curthread;
@ -983,8 +978,9 @@ nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag)
 			mtx_unlock(&nmp->nm_mtx);
 			if (!nfstcp_marker_readable(so)) {
 				/* Marker is not readable */
-				return;
+				return (SU_OK);
 			}
+			SOCKBUF_UNLOCK(&so->so_rcv);
 			auio.uio_resid = sizeof(u_int32_t);
 			auio.uio_iov = NULL;
 			auio.uio_iovcnt = 0;
@ -992,6 +988,7 @@ nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag)
 			rcvflg = (MSG_DONTWAIT | MSG_SOCALLBCK);
 			error =  soreceive(so, (struct sockaddr **)0, &auio,
 			    &mp, (struct mbuf **)0, &rcvflg);
+			SOCKBUF_LOCK(&so->so_rcv);
 			/*
 			 * We've already tested that the socket is readable. 2 cases 
 			 * here, we either read 0 bytes (client closed connection), 
@ -1052,8 +1049,9 @@ nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag)
 			mtx_unlock(&nmp->nm_mtx);
 			if (!nfstcp_readable(so, nmp->nm_nfstcpstate.rpcresid)) {
 				/* All data not readable */
-				return;
+				return (SU_OK);
 			}
+			SOCKBUF_UNLOCK(&so->so_rcv);
 			auio.uio_resid = nmp->nm_nfstcpstate.rpcresid;
 			auio.uio_iov = NULL;
 			auio.uio_iovcnt = 0;
@ -1061,6 +1059,7 @@ nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag)
 			rcvflg = (MSG_DONTWAIT | MSG_SOCALLBCK);
 			error =  soreceive(so, (struct sockaddr **)0, &auio,
 			    &mp, (struct mbuf **)0, &rcvflg);
+			SOCKBUF_LOCK(&so->so_rcv);
 			if (error || auio.uio_resid > 0) {
 				if (error && error != ECONNRESET) {
 					log(LOG_ERR, 
@ -1083,6 +1082,7 @@ nfs_clnt_tcp_soupcall(struct socket *so, void *arg, int waitflag)

 mark_reconnect:
 	nfs_mark_for_reconnect(nmp);
+	return (SU_OK);
 }

 static void
@ -1094,6 +1094,7 @@ nfs_clnt_udp_soupcall(struct socket *so, void *arg, int waitflag)
 	struct mbuf *control = NULL;
 	int error, rcvflag;

+	SOCKBUF_UNLOCK(&so->so_rcv);
 	auio.uio_resid = 1000000;
 	auio.uio_td = curthread;
 	rcvflag = MSG_DONTWAIT;
@ -1106,6 +1107,8 @@ nfs_clnt_udp_soupcall(struct socket *so, void *arg, int waitflag)
 		if (mp)
 			nfs_clnt_match_xid(so, nmp, mp);
 	} while (mp && !error);
+	SOCKBUF_LOCK(&so->so_rcv);
+	return (SU_OK);
 }

 /*
--- a/sys/nfsserver/nfs.h
+++ b/sys/nfsserver/nfs.h
@ -367,7 +367,7 @@ void	nfsrv_timer(void *);
 int	nfsrv_getcache(struct nfsrv_descript *, struct mbuf **);
 void	nfsrv_updatecache(struct nfsrv_descript *, int, struct mbuf *);
 void	nfsrv_cleancache(void);
-void	nfsrv_rcv(struct socket *so, void *arg, int waitflag);
+int	nfsrv_rcv(struct socket *so, void *arg, int waitflag);
 void	nfsrv_slpderef(struct nfssvc_sock *slp);
 void	nfsrv_wakenfsd(struct nfssvc_sock *slp);
 int	nfsrv_writegather(struct nfsrv_descript **, struct nfssvc_sock *,
--- a/sys/nfsserver/nfs_srvsock.c
+++ b/sys/nfsserver/nfs_srvsock.c
@ -406,7 +406,7 @@ nfs_getreq(struct nfsrv_descript *nd, struct nfsd *nfsd, int has_header)
 * Essentially do as much as possible non-blocking, else punt and it will
 * be called with M_WAIT from an nfsd.
 */
-void
+int
 nfsrv_rcv(struct socket *so, void *arg, int waitflag)
 {
 	struct nfssvc_sock *slp = (struct nfssvc_sock *)arg;
@ -420,7 +420,7 @@ nfsrv_rcv(struct socket *so, void *arg, int waitflag)

 	/* XXXRW: Unlocked read. */
 	if ((slp->ns_flag & SLP_VALID) == 0)
-		return;
+		return (SU_OK);

 	/*
 	 * We can't do this in the context of a socket callback
--- a/sys/nfsserver/nfs_syscalls.c
+++ b/sys/nfsserver/nfs_syscalls.c
@ -263,11 +263,11 @@ nfssvc_addsock(struct file *fp, struct sockaddr *mynam)
 	slp->ns_nam = mynam;
 	fhold(fp);
 	slp->ns_fp = fp;
+	NFSD_UNLOCK();
 	SOCKBUF_LOCK(&so->so_rcv);
-	so->so_upcallarg = (caddr_t)slp;
-	so->so_upcall = nfsrv_rcv;
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, nfsrv_rcv, slp);
 	SOCKBUF_UNLOCK(&so->so_rcv);
+	NFSD_LOCK();
 	slp->ns_flag = (SLP_VALID | SLP_NEEDQ);
 	nfsrv_wakenfsd(slp);
 	NFSD_UNLOCK();
@ -585,9 +585,7 @@ nfsrv_zapsock(struct nfssvc_sock *slp)
 		slp->ns_fp = NULL;
 		so = slp->ns_so;
 		SOCKBUF_LOCK(&so->so_rcv);
-		so->so_rcv.sb_flags &= ~SB_UPCALL;
-		so->so_upcall = NULL;
-		so->so_upcallarg = NULL;
+		soupcall_clear(so, SO_RCV);
 		SOCKBUF_UNLOCK(&so->so_rcv);
 		soshutdown(so, SHUT_RDWR);
 		closef(fp, NULL);
--- a/sys/rpc/clnt_dg.c
+++ b/sys/rpc/clnt_dg.c
@ -79,7 +79,7 @@ static void clnt_dg_abort(CLIENT *);
 static bool_t clnt_dg_control(CLIENT *, u_int, void *);
 static void clnt_dg_close(CLIENT *);
 static void clnt_dg_destroy(CLIENT *);
-static void clnt_dg_soupcall(struct socket *so, void *arg, int waitflag);
+static int clnt_dg_soupcall(struct socket *so, void *arg, int waitflag);

 static struct clnt_ops clnt_dg_ops = {
 	.cl_call =	clnt_dg_call,
@ -112,7 +112,7 @@ TAILQ_HEAD(cu_request_list, cu_request);
 #define MCALL_MSG_SIZE 24

 /*
- * This structure is pointed to by the socket's so_upcallarg
+ * This structure is pointed to by the socket buffer's sb_upcallarg
 * member. It is separate from the client private data to facilitate
 * multiple clients sharing the same socket. The cs_lock mutex is used
 * to protect all fields of this structure, the socket's receive
@ -183,6 +183,7 @@ clnt_dg_create(
 	CLIENT *cl = NULL;		/* client handle */
 	struct cu_data *cu = NULL;	/* private data */
 	struct cu_socket *cs = NULL;
+	struct sockbuf *sb;
 	struct timeval now;
 	struct rpc_msg call_msg;
 	struct __rpc_sockinfo si;
@ -260,15 +261,16 @@ clnt_dg_create(
 	cu->cu_socket = so;
 	soreserve(so, 256*1024, 256*1024);

+	sb = &so->so_rcv;
 	SOCKBUF_LOCK(&so->so_rcv);
 recheck_socket:
-	if (so->so_upcall) {
-		if (so->so_upcall != clnt_dg_soupcall) {
+	if (sb->sb_upcall) {
+		if (sb->sb_upcall != clnt_dg_soupcall) {
 			SOCKBUF_UNLOCK(&so->so_rcv);
 			printf("clnt_dg_create(): socket already has an incompatible upcall\n");
 			goto err2;
 		}
-		cs = (struct cu_socket *) so->so_upcallarg;
+		cs = (struct cu_socket *) sb->sb_upcallarg;
 		mtx_lock(&cs->cs_lock);
 		cs->cs_refs++;
 		mtx_unlock(&cs->cs_lock);
@ -277,10 +279,10 @@ clnt_dg_create(
 		 * We are the first on this socket - allocate the
 		 * structure and install it in the socket.
 		 */
-		SOCKBUF_UNLOCK(&cu->cu_socket->so_rcv);
+		SOCKBUF_UNLOCK(&so->so_rcv);
 		cs = mem_alloc(sizeof(*cs));
-		SOCKBUF_LOCK(&cu->cu_socket->so_rcv);
-		if (so->so_upcall) {
+		SOCKBUF_LOCK(&so->so_rcv);
+		if (sb->sb_upcall) {
 			/*
 			 * We have lost a race with some other client.
 			 */
@ -290,9 +292,7 @@ clnt_dg_create(
 		mtx_init(&cs->cs_lock, "cs->cs_lock", NULL, MTX_DEF);
 		cs->cs_refs = 1;
 		TAILQ_INIT(&cs->cs_pending);
-		so->so_upcallarg = cs;
-		so->so_upcall = clnt_dg_soupcall;
-		so->so_rcv.sb_flags |= SB_UPCALL;
+		soupcall_set(so, SO_RCV, clnt_dg_soupcall, cs);
 	}
 	SOCKBUF_UNLOCK(&so->so_rcv);

@ -322,7 +322,7 @@ clnt_dg_call(
 	struct timeval	utimeout)	/* seconds to wait before giving up */
 {
 	struct cu_data *cu = (struct cu_data *)cl->cl_private;
-	struct cu_socket *cs = (struct cu_socket *) cu->cu_socket->so_upcallarg;
+	struct cu_socket *cs;
 	struct rpc_timers *rt;
 	AUTH *auth;
 	struct rpc_err *errp;
@ -343,6 +343,7 @@ clnt_dg_call(
 	struct cu_request *cr;
 	int error;

+	cs = cu->cu_socket->so_rcv.sb_upcallarg;
 	cr = malloc(sizeof(struct cu_request), M_RPC, M_WAITOK);

 	mtx_lock(&cs->cs_lock);
@ -797,9 +798,10 @@ static bool_t
 clnt_dg_control(CLIENT *cl, u_int request, void *info)
 {
 	struct cu_data *cu = (struct cu_data *)cl->cl_private;
-	struct cu_socket *cs = (struct cu_socket *) cu->cu_socket->so_upcallarg;
+	struct cu_socket *cs;
 	struct sockaddr *addr;

+	cs = cu->cu_socket->so_rcv.sb_upcallarg;
 	mtx_lock(&cs->cs_lock);

 	switch (request) {
@ -929,9 +931,10 @@ static void
 clnt_dg_close(CLIENT *cl)
 {
 	struct cu_data *cu = (struct cu_data *)cl->cl_private;
-	struct cu_socket *cs = (struct cu_socket *) cu->cu_socket->so_upcallarg;
+	struct cu_socket *cs;
 	struct cu_request *cr;

+	cs = cu->cu_socket->so_rcv.sb_upcallarg;
 	mtx_lock(&cs->cs_lock);

 	if (cu->cu_closed) {
@ -974,10 +977,11 @@ static void
 clnt_dg_destroy(CLIENT *cl)
 {
 	struct cu_data *cu = (struct cu_data *)cl->cl_private;
-	struct cu_socket *cs = (struct cu_socket *) cu->cu_socket->so_upcallarg;
+	struct cu_socket *cs;
 	struct socket *so = NULL;
 	bool_t lastsocketref;

+	cs = cu->cu_socket->so_rcv.sb_upcallarg;
 	clnt_dg_close(cl);

 	mtx_lock(&cs->cs_lock);
@ -986,9 +990,7 @@ clnt_dg_destroy(CLIENT *cl)
 	if (cs->cs_refs == 0) {
 		mtx_destroy(&cs->cs_lock);
 		SOCKBUF_LOCK(&cu->cu_socket->so_rcv);
-		cu->cu_socket->so_upcallarg = NULL;
-		cu->cu_socket->so_upcall = NULL;
-		cu->cu_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(cu->cu_socket, SO_RCV);
 		SOCKBUF_UNLOCK(&cu->cu_socket->so_rcv);
 		mem_free(cs, sizeof(*cs));
 		lastsocketref = TRUE;
@ -1023,7 +1025,7 @@ time_not_ok(struct timeval *t)
 		t->tv_usec < -1 || t->tv_usec > 1000000);
 }

-void
+int
 clnt_dg_soupcall(struct socket *so, void *arg, int waitflag)
 {
 	struct cu_socket *cs = (struct cu_socket *) arg;
@ -1037,12 +1039,14 @@ clnt_dg_soupcall(struct socket *so, void *arg, int waitflag)
 	uio.uio_resid = 1000000000;
 	uio.uio_td = curthread;
 	do {
+		SOCKBUF_UNLOCK(&so->so_rcv);
 		m = NULL;
 		control = NULL;
 		rcvflag = MSG_DONTWAIT;
 		error = soreceive(so, NULL, &uio, &m, &control, &rcvflag);
 		if (control)
 			m_freem(control);
+		SOCKBUF_LOCK(&so->so_rcv);

 		if (error == EWOULDBLOCK)
 			break;
@ -1107,5 +1111,6 @@ clnt_dg_soupcall(struct socket *so, void *arg, int waitflag)
 		if (!foundreq)
 			m_freem(m);
 	} while (m);
+	return (SU_OK);
 }

--- a/sys/rpc/clnt_vc.c
+++ b/sys/rpc/clnt_vc.c
@ -91,7 +91,7 @@ static bool_t clnt_vc_control(CLIENT *, u_int, void *);
 static void clnt_vc_close(CLIENT *);
 static void clnt_vc_destroy(CLIENT *);
 static bool_t time_not_ok(struct timeval *);
-static void clnt_vc_soupcall(struct socket *so, void *arg, int waitflag);
+static int clnt_vc_soupcall(struct socket *so, void *arg, int waitflag);

 static struct clnt_ops clnt_vc_ops = {
 	.cl_call =	clnt_vc_call,
@ -286,9 +286,7 @@ clnt_vc_create(
 	soreserve(ct->ct_socket, sendsz, recvsz);

 	SOCKBUF_LOCK(&ct->ct_socket->so_rcv);
-	ct->ct_socket->so_upcallarg = ct;
-	ct->ct_socket->so_upcall = clnt_vc_soupcall;
-	ct->ct_socket->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(ct->ct_socket, SO_RCV, clnt_vc_soupcall, ct);
 	SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv);

 	ct->ct_record = NULL;
@ -750,17 +748,18 @@ clnt_vc_close(CLIENT *cl)
 	}

 	if (ct->ct_socket) {
+		ct->ct_closing = TRUE;
+		mtx_unlock(&ct->ct_lock);
+
 		SOCKBUF_LOCK(&ct->ct_socket->so_rcv);
-		ct->ct_socket->so_upcallarg = NULL;
-		ct->ct_socket->so_upcall = NULL;
-		ct->ct_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(ct->ct_socket, SO_RCV);
 		SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv);

 		/*
 		 * Abort any pending requests and wait until everyone
 		 * has finished with clnt_vc_call.
 		 */
-		ct->ct_closing = TRUE;
+		mtx_lock(&ct->ct_lock);
 		TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
 			cr->cr_xid = 0;
 			cr->cr_error = ESHUTDOWN;
@ -815,7 +814,7 @@ time_not_ok(struct timeval *t)
 		t->tv_usec <= -1 || t->tv_usec > 1000000);
 }

-void
+int
 clnt_vc_soupcall(struct socket *so, void *arg, int waitflag)
 {
 	struct ct_data *ct = (struct ct_data *) arg;
@ -840,20 +839,20 @@ clnt_vc_soupcall(struct socket *so, void *arg, int waitflag)
 			 * error condition
 			 */
 			do_read = FALSE;
-			SOCKBUF_LOCK(&so->so_rcv);
 			if (so->so_rcv.sb_cc >= sizeof(uint32_t)
 			    || (so->so_rcv.sb_state & SBS_CANTRCVMORE)
 			    || so->so_error)
 				do_read = TRUE;
-			SOCKBUF_UNLOCK(&so->so_rcv);

 			if (!do_read)
-				return;
+				return (SU_OK);

+			SOCKBUF_UNLOCK(&so->so_rcv);
 			uio.uio_resid = sizeof(uint32_t);
 			m = NULL;
 			rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK;
 			error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag);
+			SOCKBUF_LOCK(&so->so_rcv);

 			if (error == EWOULDBLOCK)
 				break;
@ -893,25 +892,25 @@ clnt_vc_soupcall(struct socket *so, void *arg, int waitflag)
 			 * buffered.
 			 */
 			do_read = FALSE;
-			SOCKBUF_LOCK(&so->so_rcv);
 			if (so->so_rcv.sb_cc >= ct->ct_record_resid
 			    || (so->so_rcv.sb_state & SBS_CANTRCVMORE)
 			    || so->so_error)
 				do_read = TRUE;
-			SOCKBUF_UNLOCK(&so->so_rcv);

 			if (!do_read)
-				return;
+				return (SU_OK);

 			/*
 			 * We have the record mark. Read as much as
 			 * the socket has buffered up to the end of
 			 * this record.
 			 */
+			SOCKBUF_UNLOCK(&so->so_rcv);
 			uio.uio_resid = ct->ct_record_resid;
 			m = NULL;
 			rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK;
 			error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag);
+			SOCKBUF_LOCK(&so->so_rcv);

 			if (error == EWOULDBLOCK)
 				break;
@ -980,4 +979,5 @@ clnt_vc_soupcall(struct socket *so, void *arg, int waitflag)
 			}
 		}
 	} while (m);
+	return (SU_OK);
 }
--- a/sys/rpc/svc_dg.c
+++ b/sys/rpc/svc_dg.c
@ -68,7 +68,7 @@ static bool_t svc_dg_reply(SVCXPRT *, struct rpc_msg *,
    struct sockaddr *, struct mbuf *);
 static void svc_dg_destroy(SVCXPRT *);
 static bool_t svc_dg_control(SVCXPRT *, const u_int, void *);
-static void svc_dg_soupcall(struct socket *so, void *arg, int waitflag);
+static int svc_dg_soupcall(struct socket *so, void *arg, int waitflag);

 static struct xp_ops svc_dg_ops = {
 	.xp_recv =	svc_dg_recv,
@ -133,9 +133,7 @@ svc_dg_create(SVCPOOL *pool, struct socket *so, size_t sendsize,
 	xprt_register(xprt);

 	SOCKBUF_LOCK(&so->so_rcv);
-	so->so_upcallarg = xprt;
-	so->so_upcall = svc_dg_soupcall;
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, svc_dg_soupcall, xprt);
 	SOCKBUF_UNLOCK(&so->so_rcv);

 	return (xprt);
@ -205,9 +203,7 @@ svc_dg_recv(SVCXPRT *xprt, struct rpc_msg *msg,

 	if (error) {
 		SOCKBUF_LOCK(&xprt->xp_socket->so_rcv);
-		xprt->xp_socket->so_upcallarg = NULL;
-		xprt->xp_socket->so_upcall = NULL;
-		xprt->xp_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(xprt->xp_socket, SO_RCV);
 		SOCKBUF_UNLOCK(&xprt->xp_socket->so_rcv);
 		xprt_inactive(xprt);
 		sx_xunlock(&xprt->xp_lock);
@ -275,9 +271,7 @@ svc_dg_destroy(SVCXPRT *xprt)
 {

 	SOCKBUF_LOCK(&xprt->xp_socket->so_rcv);
-	xprt->xp_socket->so_upcallarg = NULL;
-	xprt->xp_socket->so_upcall = NULL;
-	xprt->xp_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+	soupcall_clear(xprt->xp_socket, SO_RCV);
 	SOCKBUF_UNLOCK(&xprt->xp_socket->so_rcv);

 	sx_destroy(&xprt->xp_lock);
@ -300,10 +294,11 @@ svc_dg_control(xprt, rq, in)
 	return (FALSE);
 }

-static void
+static int
 svc_dg_soupcall(struct socket *so, void *arg, int waitflag)
 {
 	SVCXPRT *xprt = (SVCXPRT *) arg;

 	xprt_active(xprt);
+	return (SU_OK);
 }
--- a/sys/rpc/svc_vc.c
+++ b/sys/rpc/svc_vc.c
@ -80,7 +80,7 @@ static bool_t svc_vc_rendezvous_control (SVCXPRT *xprt, const u_int rq,
 static SVCXPRT *svc_vc_create_conn(SVCPOOL *pool, struct socket *so,
    struct sockaddr *raddr);
 static int svc_vc_accept(struct socket *head, struct socket **sop);
-static void svc_vc_soupcall(struct socket *so, void *arg, int waitflag);
+static int svc_vc_soupcall(struct socket *so, void *arg, int waitflag);

 static struct xp_ops svc_vc_rendezvous_ops = {
 	.xp_recv =	svc_vc_rendezvous_recv,
@ -160,9 +160,7 @@ svc_vc_create(SVCPOOL *pool, struct socket *so, size_t sendsize,
 	solisten(so, SOMAXCONN, curthread);

 	SOCKBUF_LOCK(&so->so_rcv);
-	so->so_upcallarg = xprt;
-	so->so_upcall = svc_vc_soupcall;
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, svc_vc_soupcall, xprt);
 	SOCKBUF_UNLOCK(&so->so_rcv);

 	return (xprt);
@ -236,9 +234,7 @@ svc_vc_create_conn(SVCPOOL *pool, struct socket *so, struct sockaddr *raddr)
 	xprt_register(xprt);

 	SOCKBUF_LOCK(&so->so_rcv);
-	so->so_upcallarg = xprt;
-	so->so_upcall = svc_vc_soupcall;
-	so->so_rcv.sb_flags |= SB_UPCALL;
+	soupcall_set(so, SO_RCV, svc_vc_soupcall, xprt);
 	SOCKBUF_UNLOCK(&so->so_rcv);

 	/*
@ -358,9 +354,7 @@ svc_vc_rendezvous_recv(SVCXPRT *xprt, struct rpc_msg *msg,

 	if (error) {
 		SOCKBUF_LOCK(&xprt->xp_socket->so_rcv);
-		xprt->xp_socket->so_upcallarg = NULL;
-		xprt->xp_socket->so_upcall = NULL;
-		xprt->xp_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+		soupcall_clear(xprt->xp_socket, SO_RCV);
 		SOCKBUF_UNLOCK(&xprt->xp_socket->so_rcv);
 		xprt_inactive(xprt);
 		sx_xunlock(&xprt->xp_lock);
@ -405,9 +399,7 @@ static void
 svc_vc_destroy_common(SVCXPRT *xprt)
 {
 	SOCKBUF_LOCK(&xprt->xp_socket->so_rcv);
-	xprt->xp_socket->so_upcallarg = NULL;
-	xprt->xp_socket->so_upcall = NULL;
-	xprt->xp_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+	soupcall_clear(xprt->xp_socket, SO_RCV);
 	SOCKBUF_UNLOCK(&xprt->xp_socket->so_rcv);

 	sx_destroy(&xprt->xp_lock);
@ -642,9 +634,7 @@ svc_vc_recv(SVCXPRT *xprt, struct rpc_msg *msg,

 		if (error) {
 			SOCKBUF_LOCK(&xprt->xp_socket->so_rcv);
-			xprt->xp_socket->so_upcallarg = NULL;
-			xprt->xp_socket->so_upcall = NULL;
-			xprt->xp_socket->so_rcv.sb_flags &= ~SB_UPCALL;
+			soupcall_clear(xprt->xp_socket, SO_RCV);
 			SOCKBUF_UNLOCK(&xprt->xp_socket->so_rcv);
 			xprt_inactive(xprt);
 			cd->strm_stat = XPRT_DIED;
@ -729,12 +719,13 @@ svc_vc_null()
 	return (FALSE);
 }

-static void
+static int
 svc_vc_soupcall(struct socket *so, void *arg, int waitflag)
 {
 	SVCXPRT *xprt = (SVCXPRT *) arg;

 	xprt_active(xprt);
+	return (SU_OK);
 }

 #if 0
--- a/sys/sys/sockbuf.h
+++ b/sys/sys/sockbuf.h
@ -99,6 +99,8 @@ struct	sockbuf {
 	int	sb_lowat;	/* (c/d) low water mark */
 	int	sb_timeo;	/* (c/d) timeout for read/write */
 	short	sb_flags;	/* (c/d) flags, see below */
+	int	(*sb_upcall)(struct socket *, void *, int); /* (c/d) */
+	void	*sb_upcallarg;	/* (c/d) */
 };

 #ifdef _KERNEL
--- a/sys/sys/socketvar.h
+++ b/sys/sys/socketvar.h
@ -55,6 +55,8 @@ struct vnet;
 */
 typedef	u_quad_t so_gen_t;

+struct socket;
+
 /*-
 * Locking key to struct socket:
 * (a) constant after allocation, no locking required.
@ -104,8 +106,6 @@ struct socket {

 	struct sockbuf so_rcv, so_snd;

-	void	(*so_upcall)(struct socket *, void *, int);
-	void	*so_upcallarg;
 	struct	ucred *so_cred;		/* (a) user credentials */
 	struct	label *so_label;	/* (b) MAC label for socket */
 	struct	label *so_peerlabel;	/* (b) cached MAC label for peer */
@ -280,7 +280,7 @@ struct xsocket {

 struct accept_filter {
 	char	accf_name[16];
-	void	(*accf_callback)
+	int	(*accf_callback)
 		(struct socket *so, void *arg, int waitflag);
 	void *	(*accf_create)
 		(struct socket *so, char *arg);
@ -305,6 +305,14 @@ struct sockaddr;
 struct ucred;
 struct uio;

+/* 'which' values for socket upcalls. */
+#define	SO_RCV		1
+#define	SO_SND		2
+
+/* Return values for socket upcalls. */
+#define	SU_OK		0
+#define	SU_ISCONNECTED	1
+
 /*
 * From uipc_socket and friends
 */
@ -356,6 +364,9 @@ int	sosend_generic(struct socket *so, struct sockaddr *addr,
 	    int flags, struct thread *td);
 int	soshutdown(struct socket *so, int how);
 void	sotoxsocket(struct socket *so, struct xsocket *xso);
+void	soupcall_clear(struct socket *so, int which);
+void	soupcall_set(struct socket *so, int which,
+	    int (*func)(struct socket *, void *, int), void *arg);
 void	sowakeup(struct socket *so, struct sockbuf *sb);
 int	selsocket(struct socket *so, int events, struct timeval *tv,
 	    struct thread *td);