Skip to content

Commit

Permalink
XPRT - Simplify refcounting
Browse files Browse the repository at this point in the history 
The epoll loop needs a ref on the XPRT, so that it stays alive while in
epoll().  Previously, the ref was taken and dropped across individual
events.  Instead, take a ref when hooking, and keep it until it's
unhooked.  This means up to 2 refs for an XPRT (one for send and one for
receive).  This simplifies refcounting, making it symmetric everywhere,
rather than passing the refcount for an event off to the processor to be
released.

In addition, there is a race when unhooking events from epoll, where the
event could be ready for delivery (or even delivered, but the thread not
scheduled) and so the event is processed after the unhook, and therefore
after the XPRT has been freed.  To close this, stop putting a pointer to
the rec in the event data, and instead put the FD in there and use it to
look up the XPRT.  This ensures that, if we got the XPRT from lookup,
it's valid and ref'd for the duration of the event.

Signed-off-by: Daniel Gryniewicz <[email protected]>
  • Loading branch information
@dang
dang committed Mar 3, 2020
1 parent 7db112a commit 57aa974
Show file tree
Hide file tree
Showing 5 changed files with 46 additions and 29 deletions.
5 changes: 3 additions & 2 deletions src/svc_dg.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -94,8 +94,6 @@ svc_dg_xprt_zalloc(size_t iosz)

/* Init SVCXPRT locks, etc */
rpc_dplx_rec_init(&su->su_dr);
/* Extra ref to match TCP */
SVC_REF(&su->su_dr.xprt, SVC_REF_FLAG_NONE);
xdr_ioq_setup(&su->su_dr.ioq);
return (su);
}
Expand Down Expand Up @@ -496,6 +494,9 @@ svc_dg_unlink_it(SVCXPRT *xprt, u_int flags, const char *tag, const int line)
if (!xprt->xp_parent) {
/* only original parent is registered */
svc_rqst_xprt_unregister(xprt, flags);
} else {
/* Still need to unhook it */
svc_rqst_unhook(xprt);
}
}

Expand Down
1 change: 1 addition & 0 deletions src/svc_internal.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -173,5 +173,6 @@ int svc_rqst_xprt_register(SVCXPRT *, SVCXPRT *);
void svc_rqst_xprt_unregister(SVCXPRT *, uint32_t);
int svc_rqst_evchan_write(SVCXPRT *, struct xdr_ioq *, bool);
void svc_rqst_xprt_send_complete(SVCXPRT *);
void svc_rqst_unhook(SVCXPRT *);

#endif /* TIRPC_SVC_INTERNAL_H */
65 changes: 39 additions & 26 deletions src/svc_rqst.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -579,6 +579,8 @@ svc_rqst_unhook_events(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec,
&rec->xprt.xp_flags,
SVC_XPRT_FLAG_ADDED_RECV);
}
/* Release the hook ref */
SVC_RELEASE(&rec->xprt, SVC_RELEASE_FLAG_NONE);
}

if (ev_flags & SVC_XPRT_FLAG_ADDED_SEND) {
Expand Down Expand Up @@ -617,6 +619,8 @@ svc_rqst_unhook_events(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec,
close(rec->xprt.xp_fd_send);
rec->xprt.xp_fd_send = -1;
}
/* Release the hook ref */
SVC_RELEASE(&rec->xprt, SVC_RELEASE_FLAG_NONE);
}
break;
}
Expand Down Expand Up @@ -665,9 +669,8 @@ svc_rqst_rearm_events_locked(SVCXPRT *xprt, uint16_t ev_flags)
if (sr_rec->ev_flags & SVC_RQST_FLAG_SHUTDOWN)
return (0);

/* Currently, can only be called with one of ADDED_RECV or ADDED_SEND, so we
* only need to take one ref. */
SVC_REF(xprt, SVC_REF_FLAG_NONE);
/* Don't take a ref on the xprt. We take a ref in hook, and release it
* in unhook. */

/* assuming success */
atomic_set_uint16_t_bits(&xprt->xp_flags, ev_flags);
Expand Down Expand Up @@ -810,7 +813,7 @@ svc_rqst_hook_events(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec,
ev = &rec->ev_u.epoll.event_recv;

/* set up epoll user data */
ev->data.ptr = rec;
ev->data.fd = rec->xprt.xp_fd;

/* wait for read events, level triggered, oneshot */
ev->events = EPOLLONESHOT | EPOLLIN;
Expand Down Expand Up @@ -843,6 +846,8 @@ svc_rqst_hook_events(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec,
sr_rec, sr_rec->id_k, sr_rec->ev_refcnt,
sr_rec->ev_u.epoll.epoll_fd,
sr_rec->sv[0], sr_rec->sv[1], ev);
/* Get a ref for the epoll loop */
SVC_REF(&rec->xprt, SVC_REF_FLAG_NONE);
}
}

Expand Down Expand Up @@ -884,6 +889,8 @@ svc_rqst_hook_events(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec,
sr_rec, sr_rec->id_k, sr_rec->ev_refcnt,
sr_rec->ev_u.epoll.epoll_fd,
sr_rec->sv[0], sr_rec->sv[1], ev);
/* Get a ref for the epoll loop */
SVC_REF(&rec->xprt, SVC_REF_FLAG_NONE);
}
}
break;
Expand All @@ -904,20 +911,28 @@ svc_rqst_hook_events(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec,
return (code);
}

/*
* RPC_DPLX_LOCKED
*/
static void
svc_rqst_unreg(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec)
void
svc_rqst_unhook(SVCXPRT *xprt)
{
struct rpc_dplx_rec *rec = REC_XPRT(xprt);
struct svc_rqst_rec *sr_rec = (struct svc_rqst_rec *)rec->ev_p;
uint16_t xp_flags =
atomic_postclear_uint16_t_bits(&rec->xprt.xp_flags,
atomic_postclear_uint16_t_bits(&xprt->xp_flags,
SVC_XPRT_FLAG_ADDED_RECV |
SVC_XPRT_FLAG_ADDED_SEND);

/* clear events */
if (xp_flags & (SVC_XPRT_FLAG_ADDED_RECV | SVC_XPRT_FLAG_ADDED_SEND))
(void)svc_rqst_unhook_events(rec, sr_rec, xp_flags);
}

/*
* RPC_DPLX_LOCKED
*/
static void
svc_rqst_unreg(struct rpc_dplx_rec *rec, struct svc_rqst_rec *sr_rec)
{
svc_rqst_unhook(&rec->xprt);

/* Unlinking after debug message ensures both the xprt and the sr_rec
* are still present, as the xprt unregisters before release.
Expand Down Expand Up @@ -995,8 +1010,6 @@ svc_rqst_evchan_write(SVCXPRT *xprt, struct xdr_ioq *xioq, bool has_blocked)
code = svc_rqst_rearm_events_locked(xprt,
SVC_XPRT_FLAG_ADDED_SEND);
} else {
/* svc_rqst_hook_events doesn't take a ref, so take one here */
SVC_REF(xprt, SVC_REF_FLAG_NONE);
code = svc_rqst_hook_events(rec, sr_rec,
SVC_XPRT_FLAG_ADDED_SEND);
}
Expand Down Expand Up @@ -1156,7 +1169,6 @@ svc_rqst_xprt_task_recv(struct work_pool_entry *wpe)
struct xdr_ioq *ioq =
opr_containerof(wpe, struct xdr_ioq, ioq_wpe);
struct rpc_dplx_rec *rec = ioq->rec;
enum xprt_stat stat = XPRT_IDLE;

atomic_clear_uint16_t_bits(&ioq->ioq_s.qflags, IOQ_FLAG_WORKING);

Expand All @@ -1174,13 +1186,11 @@ svc_rqst_xprt_task_recv(struct work_pool_entry *wpe)
* xp_refcnt need more than 1 (this task).
*/
(void)clock_gettime(CLOCK_MONOTONIC_FAST, &rec->recv.ts);
stat = SVC_RECV(&rec->xprt);
(void)SVC_RECV(&rec->xprt);
}

if (stat != XPRT_SUSPEND) {
/* If tests fail, log non-fatal "WARNING! already destroying!" */
SVC_RELEASE(&rec->xprt, SVC_RELEASE_FLAG_NONE);
}
/* Release the ref taken on the event */
SVC_RELEASE(&rec->xprt, SVC_RELEASE_FLAG_NONE);
}

enum xprt_stat svc_request(SVCXPRT *xprt, XDR *xdrs)
Expand Down Expand Up @@ -1274,8 +1284,6 @@ svc_rqst_xprt_task_send(struct work_pool_entry *wpe)
svc_ioq_write(&rec->xprt);
}

/* If tests fail, log non-fatal "WARNING! already destroying!"
*/
SVC_RELEASE(&rec->xprt, SVC_RELEASE_FLAG_NONE);
}

Expand Down Expand Up @@ -1351,7 +1359,8 @@ svc_rqst_clean_idle(int timeout)
static struct xdr_ioq *
svc_rqst_epoll_event(struct svc_rqst_rec *sr_rec, struct epoll_event *ev)
{
struct rpc_dplx_rec *rec = (struct rpc_dplx_rec *) ev->data.ptr;
SVCXPRT *xprt;
struct rpc_dplx_rec *rec;
uint16_t xp_flags, ev_flag = 0;
struct xdr_ioq *ioq = NULL;
work_pool_fun_t fun;
Expand All @@ -1371,11 +1380,15 @@ svc_rqst_epoll_event(struct svc_rqst_rec *sr_rec, struct epoll_event *ev)
return (NULL);
}

/* Another task may release transport in parallel.
* We have a ref from being in epoll, but since epoll is one-shot, a new ref
* will be taken when we re-enter epoll. Use this ref for the processor
* without taking another one.
*/
xprt = svc_xprt_lookup(ev->data.fd, NULL);
if (!xprt) {
__warnx(TIRPC_DEBUG_FLAG_SVC_RQST,
"%s: fd %d no associated xprt",
__func__, ev->data.fd);
return (NULL);
}
/* At this point, we have a ref on the xprt, and know it's valid */
rec = REC_XPRT(xprt);

__warnx(TIRPC_DEBUG_FLAG_SVC_RQST,
"%s: event %p %08x%s%s rpc_dplx_rec %p (sr_rec %p)",
Expand Down
1 change: 1 addition & 0 deletions src/svc_vc.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -515,6 +515,7 @@ svc_vc_rendezvous(SVCXPRT *xprt)
SVC_DESTROY(newxprt);
/* Was never added to epoll */
SVC_RELEASE(newxprt, SVC_RELEASE_FLAG_NONE);
SVC_RELEASE(xprt, SVC_RELEASE_FLAG_NONE);
return (XPRT_DESTROYED);
}
return (XPRT_IDLE);
Expand Down
3 changes: 2 additions & 1 deletion src/svc_xprt.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -209,13 +209,14 @@ svc_xprt_lookup(int fd, svc_xprt_setup_t setup)
rpc_dplx_rli(rec);
xp_flags = atomic_clear_uint16_t_bits(&xprt->xp_flags,
SVC_XPRT_FLAG_INITIAL);
rpc_dplx_rui(rec);

if (!(xp_flags & SVC_XPRT_FLAG_DESTROYED)) {
/* do not return destroyed xprts */
return (xprt);
}

/* unlock before release permits releasing here after destroy */
rpc_dplx_rui(rec);
SVC_RELEASE(xprt, SVC_RELEASE_FLAG_NONE);
return (NULL);
}
Expand Down

1 comment on commit 57aa974

@malahal
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since we use "fd" for event hooking, I think it simplifies things. No need for holding any ref while hooking it or re-arming it (we already removed this). Since we get using lookup in svc_rqst_epoll_event().

Please sign in to comment.