sipwise
/
rtpengine
mirror of https://github.com/sipwise/rtpengine.git
1
0
Fork 0
Code Issues Releases Wiki Activity

MT#55283 upgrade fd in struct to union

Browse Source 
Currently unused and transparent

Change-Id: I35604c9d5f724ae6a3bce714d708bed0fbed563f
(cherry picked from commit b2cf472a20)
mr26.0
Richard Fuchs 4 weeks ago
parent fef185612e
commit bc6a35656c
7 changed files with 57 additions and 47 deletions
Show Stats Download Patch File Download Diff File

2
daemon/media_socket.c
Unescape View File

@ -4006,7 +4006,7 @@ stream_fd *stream_fd_new(struct socket_port_link *spl, call_t *call, struct loca
__C_DBG("stream_fd_new localport=%d", sfd->socket.local.port);
ZERO(pi);
pi.fd = sfd->socket.fd;
pi.u.fd = sfd->socket.fd;
pi.obj = &sfd->obj;
pi.readable = stream_fd_readable;
pi.recv = stream_fd_recv;

4
daemon/tcp_listener.c
Unescape View File

@ -84,7 +84,7 @@ static int tcp_listener_init(socket_t *sock, const endpoint_t *ep,
goto fail;
ZERO(i);
i.fd = sock->fd;
i.u.fd = sock->fd;
i.closed = tcp_listener_closed;
i.readable = tcp_listener_incoming;
i.obj = &cb->obj;
@ -171,7 +171,7 @@ static void streambuf_listener_newconn(struct obj *p, socket_t *newsock, char *a
s->addr = strdup(addr);
ZERO(i);
i.fd = newsock->fd;
i.u.fd = newsock->fd;
i.closed = streambuf_stream_closed;
i.readable = streambuf_stream_readable;
i.writeable = streambuf_stream_writeable;

2
daemon/udp_listener.c
Unescape View File

@ -92,7 +92,7 @@ int udp_listener_init(socket_t *sock, const endpoint_t *ep,
socket_pktinfo(sock);
ZERO(i);
i.fd = sock->fd;
i.u.fd = sock->fd;
i.closed = udp_listener_closed;
i.readable = udp_listener_incoming;
i.obj = &cb->obj;

26
lib/poller.c
Unescape View File

@ -103,7 +103,7 @@ bool poller_add_item(struct poller *p, struct poller_item *i) {
return false;
if (!i)
return false;
if (i->fd < 0)
if (i->u.fd < 0)
return false;
if (!i->readable && !i->writeable)
return false;
@ -114,22 +114,22 @@ bool poller_add_item(struct poller *p, struct poller_item *i) {
LOCK(&p->lock);
if (i->fd < p->items->len && p->items->pdata[i->fd])
if (i->u.fd < p->items->len && p->items->pdata[i->u.fd])
return false;
ZERO(e);
e.events = epoll_events(i, NULL);
e.data.fd = i->fd;
if (epoll_ctl(p->fd, EPOLL_CTL_ADD, i->fd, &e))
e.data.fd = i->u.fd;
if (epoll_ctl(p->fd, EPOLL_CTL_ADD, i->u.fd, &e))
return false;
if (i->fd >= p->items->len)
g_ptr_array_set_size(p->items, i->fd + 1);
if (i->u.fd >= p->items->len)
g_ptr_array_set_size(p->items, i->u.fd + 1);
ip = obj_alloc0(struct poller_item_int, poller_item_free);
memcpy(&ip->item, i, sizeof(*i));
obj_hold_o(ip->item.obj); /* new ref in *ip */
p->items->pdata[i->fd] = obj_get(ip);
p->items->pdata[i->u.fd] = obj_get(ip);
} // unlock
@ -217,28 +217,28 @@ static int poller_poll(struct poller *p, int timeout, struct epoll_event *evs, i
mutex_unlock(&p->lock);
if (it->error) {
it->item.closed(it->item.fd, it->item.obj);
it->item.closed(it->item.u.fd, it->item.obj);
goto next;
}
if ((ev->events & (POLLERR | POLLHUP)))
it->item.closed(it->item.fd, it->item.obj);
it->item.closed(it->item.u.fd, it->item.obj);
else if ((ev->events & POLLOUT)) {
mutex_lock(&p->lock);
it->blocked = 0;
ZERO(e);
e.events = epoll_events(NULL, it);
e.data.fd = it->item.fd;
int eret = epoll_ctl(p->fd, EPOLL_CTL_MOD, it->item.fd, &e);
e.data.fd = it->item.u.fd;
int eret = epoll_ctl(p->fd, EPOLL_CTL_MOD, it->item.u.fd, &e);
mutex_unlock(&p->lock);
if (eret == 0 && it->item.writeable)
it->item.writeable(it->item.fd, it->item.obj);
it->item.writeable(it->item.u.fd, it->item.obj);
}
else if ((ev->events & POLLIN))
it->item.readable(it->item.fd, it->item.obj);
it->item.readable(it->item.u.fd, it->item.obj);
else if (!ev->events)
goto next;
else

8
lib/poller.h
Unescape View File

@ -24,8 +24,14 @@ struct sockaddr;
typedef void (*poller_func_t)(int, void *);
union poller_item_u64 {
struct {
int fd;
} __attribute__((packed));
};
struct poller_item {
int fd;
union poller_item_u64 u;
struct obj *obj;
poller_func_t readable;

58
lib/uring.c
Unescape View File

@ -225,7 +225,7 @@ bool uring_poller_add_item(struct poller *p, struct poller_item *i) {
return false;
if (!i)
return false;
if (i->fd < 0)
if (i->u.fd < 0)
return false;
if (!i->readable)
return false;
@ -315,29 +315,29 @@ static void uring_poll_event(struct uring_req *req, int32_t res, uint32_t flags)
struct uring_poll_event *ereq = (__typeof(ereq)) req;
bool closed = false;
//ilog(LOG_INFO, "uring poll event %i %i %i", ereq->it.fd, res, flags);
//ilog(LOG_INFO, "uring poll event %i %i %i", ereq->it.u.fd, res, flags);
if (res < 0) {
if (res != -ECANCELED)
ilog(LOG_WARNING | LOG_FLAG_LIMIT, "io_uring poll error on fd %i: %s",
ereq->it.fd, strerror(-res));
ereq->it.u.fd, strerror(-res));
closed = true;
}
else if ((res & (POLLERR | POLLHUP))) {
//ilog(LOG_INFO, "uring poll fd error %i %i", ereq->it.fd, res);
//ilog(LOG_INFO, "uring poll fd error %i %i", ereq->it.u.fd, res);
closed = true;
}
else if ((res & POLLIN))
ereq->it.readable(ereq->it.fd, ereq->it.obj);
ereq->it.readable(ereq->it.u.fd, ereq->it.obj);
else {
ilog(LOG_WARNING | LOG_FLAG_LIMIT, "unhandled io_uring poll event mask on fd %i: %i",
ereq->it.fd, res);
ereq->it.u.fd, res);
closed = true;
}
if (closed) {
if (!ereq->closed)
ereq->it.closed(ereq->it.fd, ereq->it.obj);
ereq->it.closed(ereq->it.u.fd, ereq->it.obj);
ereq->closed = true;
}
@ -347,8 +347,8 @@ static void uring_poll_event(struct uring_req *req, int32_t res, uint32_t flags)
struct poller *p = ereq->poller;
{
LOCK(&p->lock);
if (p->evs->len > ereq->it.fd && p->evs->pdata[ereq->it.fd] == ereq)
p->evs->pdata[ereq->it.fd] = NULL;
if (p->evs->len > ereq->it.u.fd && p->evs->pdata[ereq->it.u.fd] == ereq)
p->evs->pdata[ereq->it.u.fd] = NULL;
}
g_free(&ereq->req);
}
@ -381,25 +381,25 @@ static void uring_poll_unblocked(struct uring_req *req, int32_t res, uint32_t fl
bool closed = false;
if (res < 0) {
ilog(LOG_WARNING | LOG_FLAG_LIMIT, "io_uring poll write error on fd %i: %s",
ureq->it.fd, strerror(-res));
ureq->it.u.fd, strerror(-res));
closed = true;
}
else if (!(res & (POLLOUT))) {
ilog(LOG_WARNING | LOG_FLAG_LIMIT, "unhandled io_uring poll event write mask on fd %i: %i",
ureq->it.fd, res);
ureq->it.u.fd, res);
closed = true;
}
else {
struct poller *p = ureq->poller;
if (p->blocked->len > ureq->it.fd)
g_array_index(p->blocked, char, ureq->it.fd) = 0;
ureq->it.writeable(ureq->it.fd, ureq->it.obj);
if (p->blocked->len > ureq->it.u.fd)
g_array_index(p->blocked, char, ureq->it.u.fd) = 0;
ureq->it.writeable(ureq->it.u.fd, ureq->it.obj);
}
assert((flags & IORING_CQE_F_MORE) == 0);
if (closed)
ureq->it.closed(ureq->it.fd, ureq->it.obj);
ureq->it.closed(ureq->it.u.fd, ureq->it.obj);
if (ureq->it.obj)
obj_put_o(ureq->it.obj);
@ -427,12 +427,12 @@ static void uring_poll_recv(struct uring_req *req, int32_t res, uint32_t flags)
struct poller *p = rreq->poller;
bool closed = false;
//ilog(LOG_INFO, "uring recvmsg event %i %i %i", rreq->it.fd, res, flags);
//ilog(LOG_INFO, "uring recvmsg event %i %i %i", rreq->it.u.fd, res, flags);
if (res < 0) {
if (res != -ECANCELED)
ilog(LOG_WARNING | LOG_FLAG_LIMIT, "io_uring recvmsg error on fd %i: %s",
rreq->it.fd, strerror(-res));
rreq->it.u.fd, strerror(-res));
closed = true;
}
else {
@ -461,12 +461,12 @@ static void uring_poll_recv(struct uring_req *req, int32_t res, uint32_t flags)
if (closed) {
if (!rreq->closed)
rreq->it.closed(rreq->it.fd, rreq->it.obj);
rreq->it.closed(rreq->it.u.fd, rreq->it.obj);
rreq->closed = true;
}
if (!(flags & IORING_CQE_F_MORE)) {
//ilog(LOG_INFO, "last uring recv event for fd %i for %p (%i)", rreq->it.fd, rreq->it.obj, rreq->it.obj->ref);
//ilog(LOG_INFO, "last uring recv event for fd %i for %p (%i)", rreq->it.u.fd, rreq->it.obj, rreq->it.obj->ref);
if (rreq->it.obj)
obj_put_o(rreq->it.obj);
g_free(&rreq->req);
@ -475,19 +475,19 @@ static void uring_poll_recv(struct uring_req *req, int32_t res, uint32_t flags)
static void uring_poller_do_add(struct poller *p, struct poller_req *preq) {
// don't allow duplicates
if (p->evs->len > preq->it.fd && p->evs->pdata[preq->it.fd])
if (p->evs->len > preq->it.u.fd && p->evs->pdata[preq->it.u.fd])
abort(); // XXX handle gracefully?
struct uring_poll_event *ereq
= uring_alloc_req(struct uring_poll_event, uring_poll_event);
ereq->it = preq->it;
ereq->poller = p;
struct io_uring_sqe *sqe = io_uring_get_sqe(&rtpe_uring);
io_uring_prep_poll_multishot(sqe, ereq->it.fd, POLLHUP | POLLERR | POLLIN);
io_uring_prep_poll_multishot(sqe, ereq->it.u.fd, POLLHUP | POLLERR | POLLIN);
io_uring_sqe_set_data(sqe, ereq);
// save ereq for write blocks. no extra obj reference
if (p->evs->len <= ereq->it.fd)
g_ptr_array_set_size(p->evs, ereq->it.fd + 1);
p->evs->pdata[ereq->it.fd] = ereq;
if (p->evs->len <= ereq->it.u.fd)
g_ptr_array_set_size(p->evs, ereq->it.u.fd + 1);
p->evs->pdata[ereq->it.u.fd] = ereq;
}
static void uring_poller_do_blocked(struct poller *p, struct poller_req *preq) {
// valid fd?
@ -501,7 +501,7 @@ static void uring_poller_do_blocked(struct poller *p, struct poller_req *preq) {
if (ureq->it.obj)
obj_hold_o(ureq->it.obj);
struct io_uring_sqe *sqe = io_uring_get_sqe(&rtpe_uring);
io_uring_prep_poll_add(sqe, ureq->it.fd, POLLOUT);
io_uring_prep_poll_add(sqe, ureq->it.u.fd, POLLOUT);
io_uring_sqe_set_data(sqe, ureq);
}
static void uring_poller_do_error(struct poller *p, struct poller_req *preq) {
@ -528,7 +528,7 @@ static void uring_poller_do_buffers(struct poller *p, struct poller_req *preq) {
io_uring_sqe_set_data(sqe, breq); // XXX no content? not needed?
}
static void uring_poller_do_recv(struct poller *p, struct poller_req *preq) {
//ilog(LOG_INFO, "adding recv fd %i on %p for %p", preq->it.fd, p, preq->it.obj);
//ilog(LOG_INFO, "adding recv fd %i on %p for %p", preq->it.u.fd, p, preq->it.obj);
struct uring_poll_recv *rreq
= uring_alloc_req(struct uring_poll_recv, uring_poll_recv);
rreq->it = preq->it;
@ -543,7 +543,7 @@ static void uring_poller_do_recv(struct poller *p, struct poller_req *preq) {
.msg_namelen = sizeof(struct sockaddr_storage),
.msg_controllen = 64,
};
io_uring_prep_recvmsg_multishot(sqe, rreq->it.fd, &rreq->msg, 0);
io_uring_prep_recvmsg_multishot(sqe, rreq->it.u.fd, &rreq->msg, 0);
sqe->ioprio |= IORING_RECVSEND_POLL_FIRST;
io_uring_sqe_set_flags(sqe, IOSQE_BUFFER_SELECT);
sqe->buf_group = 0;
@ -597,7 +597,9 @@ void uring_poller_add_waker(struct poller *p) {
&(struct poller_item) {
.readable = uring_poller_waker_read,
.closed = uring_poller_waker_closed,
.fd = p->waker_fds[1],
.u = {
.fd = p->waker_fds[1],
},
});
}

4
perf-tester/main.c
Unescape View File

@ -543,7 +543,9 @@ static void new_stream_params(
abort();
struct poller_item pi = {
.fd = s->timer_fd,
.u = {
.fd = s->timer_fd,
},
.obj = &s->obj,
.readable = readable,
.closed = closed,

Write Preview
Loading…
Cancel
Save