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

MT#56374 obsolete non-global pollers

Browse Source 
We have long switched to having a single primary global poller, so
there's no point in passing the poller pointer around and storing it in
associated objects. Remove all remnants of non-global pollers.

Change-Id: I5a3fd217d5de51e839e2b04fec7e23643ee83631
pull/1694/head
Richard Fuchs 2 years ago
parent 16c08efe62
commit 73489fc556
15 changed files with 41 additions and 82 deletions
Show Stats Download Patch File Download Diff File

9
daemon/cli.c
Unescape View File

@ -1255,15 +1255,12 @@ static void cli_free(void *p) {
streambuf_listener_shutdown(&c->listener);
}
struct cli *cli_new(struct poller *p, endpoint_t *ep) {
struct cli *cli_new(endpoint_t *ep) {
struct cli *c;
if (!p)
return NULL;
c = obj_alloc0("cli", sizeof(*c), cli_free);
if (streambuf_listener_init(&c->listener, p, ep,
if (streambuf_listener_init(&c->listener, ep,
cli_incoming, cli_stream_readable,
NULL,
&c->obj))
@ -1272,8 +1269,6 @@ struct cli *cli_new(struct poller *p, endpoint_t *ep) {
goto fail;
}
c->poller = p;
obj_put(c);
return c;

19
daemon/control_ng.c
Unescape View File

@ -526,25 +526,21 @@ void control_ng_free(void *p) {
g_hash_table_destroy(rtpe_cngs_hash);
rtpe_cngs_hash = NULL;
}
poller_del_item(c->poller, c->udp_listener.fd);
poller_del_item(rtpe_poller, c->udp_listener.fd);
close_socket(&c->udp_listener);
streambuf_listener_shutdown(&c->tcp_listener);
if (tcp_connections_hash)
g_hash_table_destroy(tcp_connections_hash);
}
struct control_ng *control_ng_new(struct poller *p, endpoint_t *ep, unsigned char tos) {
struct control_ng *control_ng_new(endpoint_t *ep, unsigned char tos) {
struct control_ng *c;
if (!p)
return NULL;
c = obj_alloc0("control_ng", sizeof(*c), control_ng_free);
c->udp_listener.fd = -1;
c->poller = p;
if (udp_listener_init(&c->udp_listener, p, ep, control_ng_incoming, &c->obj))
if (udp_listener_init(&c->udp_listener, ep, control_ng_incoming, &c->obj))
goto fail2;
if (tos)
set_tos(&c->udp_listener, tos);
@ -558,16 +554,11 @@ fail2:
return NULL;
}
struct control_ng *control_ng_tcp_new(struct poller *p, endpoint_t *ep) {
if (!p)
return NULL;
struct control_ng *control_ng_tcp_new(endpoint_t *ep) {
struct control_ng * ctrl_ng = obj_alloc0("control_ng", sizeof(*ctrl_ng), NULL);
ctrl_ng->udp_listener.fd = -1;
ctrl_ng->poller = p;
if (streambuf_listener_init(&ctrl_ng->tcp_listener, p, ep,
if (streambuf_listener_init(&ctrl_ng->tcp_listener, ep,
control_incoming, control_stream_readable,
control_closed,
&ctrl_ng->obj)) {

13
daemon/control_tcp.c
Unescape View File

@ -30,8 +30,6 @@ struct control_tcp {
pcre *parse_re;
pcre_extra *parse_ree;
struct poller *poller;
};
@ -44,7 +42,7 @@ static void control_stream_closed(struct streambuf_stream *s) {
static void control_list(struct control_tcp *c, struct streambuf_stream *s) {
if (!c->listener.listener.family || !c->listener.poller)
if (!c->listener.listener.family)
return;
mutex_lock(&c->listener.lock);
@ -153,17 +151,14 @@ static void control_tcp_free(void *p) {
pcre_free_study(c->parse_ree);
}
struct control_tcp *control_tcp_new(struct poller *p, endpoint_t *ep) {
struct control_tcp *control_tcp_new(endpoint_t *ep) {
struct control_tcp *c;
const char *errptr;
int erroff;
if (!p)
return NULL;
c = obj_alloc0("control", sizeof(*c), control_tcp_free);
if (streambuf_listener_init(&c->listener, p, ep,
if (streambuf_listener_init(&c->listener, ep,
control_incoming, control_stream_readable,
control_stream_closed,
&c->obj))
@ -178,8 +173,6 @@ struct control_tcp *control_tcp_new(struct poller *p, endpoint_t *ep) {
PCRE_DOLLAR_ENDONLY | PCRE_DOTALL, &errptr, &erroff, NULL);
c->parse_ree = pcre_study(c->parse_re, 0, &errptr);
c->poller = p;
obj_put(c);
return c;

7
daemon/control_udp.c
Unescape View File

@ -142,14 +142,11 @@ void control_udp_free(void *p) {
cookie_cache_cleanup(&u->cookie_cache);
}
struct control_udp *control_udp_new(struct poller *p, endpoint_t *ep) {
struct control_udp *control_udp_new(endpoint_t *ep) {
struct control_udp *c;
const char *errptr;
int erroff;
if (!p)
return NULL;
c = obj_alloc0("control_udp", sizeof(*c), control_udp_free);
c->parse_re = pcre_compile(
@ -173,7 +170,7 @@ struct control_udp *control_udp_new(struct poller *p, endpoint_t *ep) {
cookie_cache_init(&c->cookie_cache);
if (udp_listener_init(&c->udp_listener, p, ep, control_udp_incoming, &c->obj))
if (udp_listener_init(&c->udp_listener, ep, control_udp_incoming, &c->obj))
goto fail2;
return c;

20
daemon/main.c
Unescape View File

@ -1152,13 +1152,13 @@ no_kernel:
rtpe_config.redis_num_threads = num_cpu_cores(REDIS_RESTORE_NUM_THREADS);
if (rtpe_config.tcp_listen_ep[0].port) {
rtpe_tcp[0] = control_tcp_new(rtpe_poller, &rtpe_config.tcp_listen_ep[0]);
rtpe_tcp[0] = control_tcp_new(&rtpe_config.tcp_listen_ep[0]);
if (!rtpe_tcp[0])
die("Failed to open TCP control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.tcp_listen_ep[0]),
strerror(errno));
if (rtpe_config.tcp_listen_ep[1].port) {
rtpe_tcp[1] = control_tcp_new(rtpe_poller, &rtpe_config.tcp_listen_ep[1]);
rtpe_tcp[1] = control_tcp_new(&rtpe_config.tcp_listen_ep[1]);
if (!rtpe_tcp[1])
die("Failed to open TCP control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.tcp_listen_ep[1]),
@ -1168,13 +1168,13 @@ no_kernel:
if (rtpe_config.udp_listen_ep[0].port) {
interfaces_exclude_port(rtpe_config.udp_listen_ep[0].port);
rtpe_udp[0] = control_udp_new(rtpe_poller, &rtpe_config.udp_listen_ep[0]);
rtpe_udp[0] = control_udp_new(&rtpe_config.udp_listen_ep[0]);
if (!rtpe_udp[0])
die("Failed to open UDP control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.udp_listen_ep[0]),
strerror(errno));
if (rtpe_config.udp_listen_ep[1].port) {
rtpe_udp[1] = control_udp_new(rtpe_poller, &rtpe_config.udp_listen_ep[1]);
rtpe_udp[1] = control_udp_new(&rtpe_config.udp_listen_ep[1]);
if (!rtpe_udp[1])
die("Failed to open UDP control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.udp_listen_ep[1]),
@ -1184,14 +1184,14 @@ no_kernel:
if (rtpe_config.ng_listen_ep[0].port) {
interfaces_exclude_port(rtpe_config.ng_listen_ep[0].port);
rtpe_control_ng[0] = control_ng_new(rtpe_poller, &rtpe_config.ng_listen_ep[0],
rtpe_control_ng[0] = control_ng_new(&rtpe_config.ng_listen_ep[0],
rtpe_config.control_tos);
if (!rtpe_control_ng[0])
die("Failed to open UDP NG control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.ng_listen_ep[0]),
strerror(errno));
if (rtpe_config.ng_listen_ep[1].port) {
rtpe_control_ng[1] = control_ng_new(rtpe_poller, &rtpe_config.ng_listen_ep[1],
rtpe_control_ng[1] = control_ng_new(&rtpe_config.ng_listen_ep[1],
rtpe_config.control_tos);
if (!rtpe_control_ng[1])
die("Failed to open UDP NG control connection port (%s): %s",
@ -1201,13 +1201,13 @@ no_kernel:
}
if (rtpe_config.ng_tcp_listen_ep[0].port) {
rtpe_control_ng_tcp[0] = control_ng_tcp_new(rtpe_poller, &rtpe_config.ng_tcp_listen_ep[0]);
rtpe_control_ng_tcp[0] = control_ng_tcp_new(&rtpe_config.ng_tcp_listen_ep[0]);
if (!rtpe_control_ng_tcp[0])
die("Failed to open TCP NG control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.ng_tcp_listen_ep[0]),
strerror(errno));
if (rtpe_config.ng_tcp_listen_ep[1].port) {
rtpe_control_ng_tcp[1] = control_ng_tcp_new(rtpe_poller, &rtpe_config.ng_tcp_listen_ep[1]);
rtpe_control_ng_tcp[1] = control_ng_tcp_new(&rtpe_config.ng_tcp_listen_ep[1]);
if (!rtpe_control_ng_tcp[1])
die("Failed to open TCP NG control connection port (%s): %s",
endpoint_print_buf(&rtpe_config.ng_tcp_listen_ep[1]),
@ -1217,13 +1217,13 @@ no_kernel:
if (rtpe_config.cli_listen_ep[0].port) {
interfaces_exclude_port(rtpe_config.cli_listen_ep[0].port);
rtpe_cli[0] = cli_new(rtpe_poller, &rtpe_config.cli_listen_ep[0]);
rtpe_cli[0] = cli_new(&rtpe_config.cli_listen_ep[0]);
if (!rtpe_cli[0])
die("Failed to open CLI connection port (%s): %s",
endpoint_print_buf(&rtpe_config.cli_listen_ep[0]),
strerror(errno));
if (rtpe_config.cli_listen_ep[1].port) {
rtpe_cli[1] = cli_new(rtpe_poller, &rtpe_config.cli_listen_ep[1]);
rtpe_cli[1] = cli_new(&rtpe_config.cli_listen_ep[1]);
if (!rtpe_cli[1])
die("Failed to open CLI connection port (%s): %s",
endpoint_print_buf(&rtpe_config.cli_listen_ep[1]),

19
daemon/tcp_listener.c
Unescape View File

@ -63,7 +63,7 @@ static void __tlc_free(void *p) {
obj_put_o(cb->p);
}
int tcp_listener_init(socket_t *sock, struct poller *p, const endpoint_t *ep,
int tcp_listener_init(socket_t *sock, const endpoint_t *ep,
tcp_listener_callback_t func, struct obj *obj)
{
struct poller_item i;
@ -84,7 +84,7 @@ int tcp_listener_init(socket_t *sock, struct poller *p, const endpoint_t *ep,
i.closed = tcp_listener_closed;
i.readable = tcp_listener_incoming;
i.obj = &cb->obj;
if (poller_add_item(p, &i))
if (poller_add_item(rtpe_poller, &i))
goto fail;
obj_put(cb);
@ -119,7 +119,7 @@ static void streambuf_stream_closed(int fd, void *p, uintptr_t u) {
mutex_lock(&l->lock);
int ret = g_hash_table_remove(l->streams, s);
mutex_unlock(&l->lock);
poller_del_item(s->listener->poller, s->sock.fd);
poller_del_item(rtpe_poller, s->sock.fd);
if (ret)
obj_put(s);
}
@ -160,8 +160,8 @@ static void streambuf_listener_newconn(struct obj *p, socket_t *newsock, char *a
s = obj_alloc0("streambuf_stream", sizeof(*s), streambuf_stream_free);
s->sock = *newsock;
s->inbuf = streambuf_new(listener->poller, newsock->fd);
s->outbuf = streambuf_new(listener->poller, newsock->fd);
s->inbuf = streambuf_new(rtpe_poller, newsock->fd);
s->outbuf = streambuf_new(rtpe_poller, newsock->fd);
s->listener = listener;
s->cb = obj_get(cb);
s->parent = obj_get_o(cb->parent);
@ -183,7 +183,7 @@ static void streambuf_listener_newconn(struct obj *p, socket_t *newsock, char *a
g_hash_table_insert(listener->streams, s, s); // hand over ref
mutex_unlock(&listener->lock);
if (poller_add_item(listener->poller, &i))
if (poller_add_item(rtpe_poller, &i))
goto fail;
obj_put(s);
@ -206,7 +206,7 @@ static void __sb_free(void *p) {
obj_put_o(cb->parent);
}
int streambuf_listener_init(struct streambuf_listener *listener, struct poller *p, const endpoint_t *ep,
int streambuf_listener_init(struct streambuf_listener *listener, const endpoint_t *ep,
streambuf_callback_t newconn_func,
streambuf_callback_t newdata_func,
streambuf_callback_t closed_func,
@ -216,7 +216,6 @@ int streambuf_listener_init(struct streambuf_listener *listener, struct poller *
ZERO(*listener);
listener->poller = p;
mutex_init(&listener->lock);
listener->streams = g_hash_table_new(g_direct_hash, g_direct_equal);
@ -227,7 +226,7 @@ int streambuf_listener_init(struct streambuf_listener *listener, struct poller *
cb->parent = obj_get_o(obj);
cb->listener = listener;
if (tcp_listener_init(&listener->listener, p, ep, streambuf_listener_newconn, &cb->obj))
if (tcp_listener_init(&listener->listener, ep, streambuf_listener_newconn, &cb->obj))
goto fail;
obj_put(cb);
@ -240,7 +239,7 @@ fail:
void streambuf_listener_shutdown(struct streambuf_listener *listener) {
if (!listener)
return;
poller_del_item(listener->poller, listener->listener.fd);
poller_del_item(rtpe_poller, listener->listener.fd);
close_socket(&listener->listener);
if (listener->streams)
g_hash_table_destroy(listener->streams);

4
daemon/udp_listener.c
Unescape View File

@ -73,7 +73,7 @@ static void __ulc_free(void *p) {
obj_put_o(cb->p);
}
int udp_listener_init(socket_t *sock, struct poller *p, const endpoint_t *ep,
int udp_listener_init(socket_t *sock, const endpoint_t *ep,
udp_listener_callback_t func, struct obj *obj)
{
struct poller_item i;
@ -92,7 +92,7 @@ int udp_listener_init(socket_t *sock, struct poller *p, const endpoint_t *ep,
i.closed = udp_listener_closed;
i.readable = udp_listener_incoming;
i.obj = &cb->obj;
if (poller_add_item(p, &i))
if (poller_add_item(rtpe_poller, &i))
goto fail;
obj_put(cb);

1
include/call.h
Unescape View File

@ -232,7 +232,6 @@ enum block_dtmf_mode {
#include "dtls.h"
struct poller;
struct control_stream;
struct call;
struct redis;

4
include/cli.h
Unescape View File

@ -8,8 +8,6 @@
struct cli {
struct obj obj;
struct poller *poller;
struct streambuf_listener listener;
};
@ -24,7 +22,7 @@ struct cli_writer {
struct call_monologue *ml;
};
struct cli *cli_new(struct poller *p, endpoint_t *);
struct cli *cli_new(endpoint_t *);
void cli_handle(str *instr, struct cli_writer *);

7
include/control_ng.h
Unescape View File

@ -1,8 +1,6 @@
#ifndef _CONTROL_NG_H_
#define _CONTROL_NG_H_
struct poller;
enum ng_command {
NGC_PING = 0,
NGC_OFFER,
@ -56,7 +54,6 @@ struct control_ng {
struct obj obj;
socket_t udp_listener;
struct streambuf_listener tcp_listener;
struct poller *poller;
};
struct ng_buffer {
@ -68,8 +65,8 @@ struct ng_buffer {
extern const char *ng_command_strings[NGC_COUNT];
extern const char *ng_command_strings_short[NGC_COUNT];
struct control_ng *control_ng_new(struct poller *, endpoint_t *, unsigned char);
struct control_ng *control_ng_tcp_new(struct poller *, endpoint_t *);
struct control_ng *control_ng_new(endpoint_t *, unsigned char);
struct control_ng *control_ng_tcp_new(endpoint_t *);
void notify_ng_tcp_clients(str *);
void control_ng_init(void);
void control_ng_cleanup(void);

3
include/control_tcp.h
Unescape View File

@ -29,13 +29,12 @@
#define RE_TCP_D_INFO 13
#define RE_TCP_DIV_CMD 14
struct poller;
struct control_tcp;
struct streambuf_stream;
struct control_tcp *control_tcp_new(struct poller *, endpoint_t *);
struct control_tcp *control_tcp_new(endpoint_t *);

6
include/control_udp.h
Unescape View File

@ -38,10 +38,6 @@
#define RE_UDP_V_FLAGS 19
#define RE_UDP_V_PARMS 20
struct poller;
struct control_udp {
@ -59,7 +55,7 @@ struct control_udp {
struct control_udp *control_udp_new(struct poller *, endpoint_t *);
struct control_udp *control_udp_new(endpoint_t *);

1
include/dtls.h
Unescape View File

@ -23,7 +23,6 @@
struct packet_stream;
struct sockaddr_in6;
struct poller;
struct stream_fd;

6
include/tcp_listener.h
Unescape View File

@ -6,7 +6,6 @@
#include "helpers.h"
struct poller;
struct obj;
struct streambuf_callback;
struct streambuf_stream;
@ -16,7 +15,6 @@ typedef void (*streambuf_callback_t)(struct streambuf_stream *);
struct streambuf_listener {
socket_t listener;
struct poller *poller;
mutex_t lock;
GHashTable *streams;
};
@ -32,9 +30,9 @@ struct streambuf_stream {
};
int tcp_listener_init(socket_t *, struct poller *p, const endpoint_t *, tcp_listener_callback_t, struct obj *);
int tcp_listener_init(socket_t *, const endpoint_t *, tcp_listener_callback_t, struct obj *);
int streambuf_listener_init(struct streambuf_listener *listener, struct poller *p, const endpoint_t *ep,
int streambuf_listener_init(struct streambuf_listener *listener, const endpoint_t *ep,
streambuf_callback_t newconn_func,
streambuf_callback_t newdata_func,
streambuf_callback_t closed_func,

4
include/udp_listener.h
Unescape View File

@ -10,8 +10,6 @@
#define MAX_UDP_LENGTH 0xffff
struct poller;
struct udp_buffer {
struct obj obj;
char buf[MAX_UDP_LENGTH + RTP_BUFFER_TAIL_ROOM + RTP_BUFFER_HEAD_ROOM + 1];
@ -23,6 +21,6 @@ struct udp_buffer {
typedef void (*udp_listener_callback_t)(struct obj *p, struct udp_buffer *);
int udp_listener_init(socket_t *, struct poller *p, const endpoint_t *, udp_listener_callback_t, struct obj *);
int udp_listener_init(socket_t *, const endpoint_t *, udp_listener_callback_t, struct obj *);
#endif

Write Preview
Loading…
Cancel
Save