williamp/osmocom-analog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add libselect to use select instead of polling all file descriptors

Browse Source
This commit is contained in:
Andreas Eversberg
2023-01-20 18:13:11 +01:00
parent ffaed13648
commit 9f662d309f
35 changed files with 408 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/libosmocc/endpoint.c
View File

@@ -24,6 +24,7 @@
#include <arpa/inet.h>
#include <unistd.h>
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "../libdebug/debug.h"
#include "endpoint.h"

1
src/libosmocc/helper.c
View File

@@ -25,6 +25,7 @@
#include <sys/time.h>
#include <inttypes.h>
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "../libdebug/debug.h"
#include "endpoint.h"
#include "helper.h"

134
src/libosmocc/rtp.c
View File

@@ -28,6 +28,7 @@
#include <arpa/inet.h>
#include "../libdebug/debug.h"
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "endpoint.h"
#define RTP_VERSION 2
@@ -132,6 +133,22 @@ static int rtp_receive(int sock, uint8_t **payload_p, int *payload_len_p, uint8_
return 0;
}
static int rtcp_receive(int sock)
{
static uint8_t data[2048];
int len;
len = read(sock, data, sizeof(data));
if (len < 0) {
if (errno == EAGAIN)
return -EAGAIN;
PDEBUG(DCC, DEBUG_DEBUG, "Read errno = %d (%s)\n", errno, strerror(errno));
return -EIO;
}
return 0;
}
static void rtp_send(int sock, uint8_t *payload, int payload_len, uint8_t marker, uint8_t pt, uint16_t sequence, uint32_t timestamp, uint32_t ssrc)
{
struct rtp_hdr *rtph;
@@ -157,6 +174,9 @@ static void rtp_send(int sock, uint8_t *payload, int payload_len, uint8_t marker
PDEBUG(DCC, DEBUG_DEBUG, "Write errno = %d (%s)\n", errno, strerror(errno));
}
static int rtp_listen_cb(struct osmo_fd *ofd, unsigned int when);
static int rtcp_listen_cb(struct osmo_fd *ofd, unsigned int when);
/* open and bind RTP
* set local port to what we bound
*/
@@ -221,15 +241,23 @@ socket_error:
osmo_cc_rtp_close(media);
return -EIO;
}
media->rtp_socket = rc;
media->rtp_ofd.fd = rc;
media->rtp_ofd.cb = rtp_listen_cb;
media->rtp_ofd.data = media;
media->rtp_ofd.when = OSMO_FD_READ;
osmo_fd_register(&media->rtp_ofd);
rc = socket(domain, SOCK_DGRAM, IPPROTO_UDP);
if (rc < 0)
goto socket_error;
media->rtcp_socket = rc;
media->rtcp_ofd.fd = rc;
media->rtcp_ofd.cb = rtcp_listen_cb;
media->rtcp_ofd.data = media;
media->rtcp_ofd.when = OSMO_FD_READ;
osmo_fd_register(&media->rtcp_ofd);
/* bind sockets */
*sport = htons(conf->rtp_port_next);
rc = bind(media->rtp_socket, (struct sockaddr *)&sa, slen);
rc = bind(media->rtp_ofd.fd, (struct sockaddr *)&sa, slen);
if (rc < 0) {
bind_error:
osmo_cc_rtp_close(media);
@@ -241,18 +269,18 @@ bind_error:
continue;
}
*sport = htons(conf->rtp_port_next + 1);
rc = bind(media->rtcp_socket, (struct sockaddr *)&sa, slen);
rc = bind(media->rtcp_ofd.fd, (struct sockaddr *)&sa, slen);
if (rc < 0)
goto bind_error;
media->description.port_local = conf->rtp_port_next;
conf->rtp_port_next = (conf->rtp_port_next + 2 > conf->rtp_port_to) ? conf->rtp_port_from : conf->rtp_port_next + 2;
/* set nonblocking io */
flags = fcntl(media->rtp_socket, F_GETFL);
/* set nonblocking io, to prevent write to block */
flags = fcntl(media->rtp_ofd.fd, F_GETFL);
flags |= O_NONBLOCK;
fcntl(media->rtp_socket, F_SETFL, flags);
flags = fcntl(media->rtcp_socket, F_GETFL);
fcntl(media->rtp_ofd.fd, F_SETFL, flags);
flags = fcntl(media->rtcp_ofd.fd, F_GETFL);
flags |= O_NONBLOCK;
fcntl(media->rtcp_socket, F_SETFL, flags);
fcntl(media->rtcp_ofd.fd, F_SETFL, flags);
break;
}
@@ -305,7 +333,7 @@ pton_error:
}
*sport = htons(media->description.port_remote);
rc = connect(media->rtp_socket, (struct sockaddr *)&sa, slen);
rc = connect(media->rtp_ofd.fd, (struct sockaddr *)&sa, slen);
if (rc < 0) {
connect_error:
PDEBUG(DCC, DEBUG_NOTICE, "Cannot connect to address '%s'.\n", media->connection_data_remote.address);
@@ -313,7 +341,7 @@ connect_error:
return -EIO;
}
*sport = htons(media->description.port_remote + 1);
rc = connect(media->rtcp_socket, (struct sockaddr *)&sa, slen);
rc = connect(media->rtcp_ofd.fd, (struct sockaddr *)&sa, slen);
if (rc < 0)
goto connect_error;
@@ -326,7 +354,7 @@ void osmo_cc_rtp_send(osmo_cc_session_codec_t *codec, uint8_t *data, int len, ui
uint8_t *payload = NULL;
int payload_len = 0;
if (!codec || !codec->media->rtp_socket)
if (!codec || !codec->media->rtp_ofd.fd)
return;
if (codec->encoder)
@@ -336,7 +364,7 @@ void osmo_cc_rtp_send(osmo_cc_session_codec_t *codec, uint8_t *data, int len, ui
payload_len = len;
}
rtp_send(codec->media->rtp_socket, payload, payload_len, marker, codec->payload_type_remote, codec->media->tx_sequence, codec->media->tx_timestamp, codec->media->tx_ssrc);
rtp_send(codec->media->rtp_ofd.fd, payload, payload_len, marker, codec->payload_type_remote, codec->media->tx_sequence, codec->media->tx_timestamp, codec->media->tx_ssrc);
codec->media->tx_sequence += inc_sequence;
codec->media->tx_timestamp += inc_timestamp;
@@ -344,9 +372,9 @@ void osmo_cc_rtp_send(osmo_cc_session_codec_t *codec, uint8_t *data, int len, ui
free(payload);
}
/* receive rtp data for given media, return < 0, if there is nothing this time */
int osmo_cc_rtp_receive(osmo_cc_session_media_t *media, void *priv)
static int rtp_listen_cb(struct osmo_fd *ofd, unsigned int when)
{
osmo_cc_session_media_t *media = ofd->data;
int rc;
uint8_t *payload = NULL;
int payload_len = 0;
@@ -356,49 +384,63 @@ int osmo_cc_rtp_receive(osmo_cc_session_media_t *media, void *priv)
uint8_t *data;
int len;
if (!media || media->rtp_socket <= 0)
return -EIO;
if (when & OSMO_FD_READ) {
rc = rtp_receive(media->rtp_ofd.fd, &payload, &payload_len, &marker, &payload_type, &media->rx_sequence, &media->rx_timestamp, &media->rx_ssrc);
if (rc < 0)
return rc;
rc = rtp_receive(media->rtp_socket, &payload, &payload_len, &marker, &payload_type, &media->rx_sequence, &media->rx_timestamp, &media->rx_ssrc);
if (rc < 0)
return rc;
/* search for codec */
for (codec = media->codec_list; codec; codec = codec->next) {
if (codec->payload_type_local == payload_type)
break;
}
if (!codec) {
PDEBUG(DCC, DEBUG_NOTICE, "Received RTP frame for unknown codec (payload_type = %d).\n", payload_type);
return 0;
}
/* search for codec */
for (codec = media->codec_list; codec; codec = codec->next) {
if (codec->payload_type_local == payload_type)
break;
}
if (!codec) {
PDEBUG(DCC, DEBUG_NOTICE, "Received RTP frame for unknown codec (payload_type = %d).\n", payload_type);
return 0;
if (codec->decoder)
codec->decoder(payload, payload_len, &data, &len, media->session->priv);
else {
data = payload;
len = payload_len;
}
if (codec->media->receive)
codec->media->receiver(codec, marker, media->rx_sequence, media->rx_timestamp, media->rx_ssrc, data, len);
if (codec->decoder)
free(data);
}
if (codec->decoder)
codec->decoder(payload, payload_len, &data, &len, priv);
else {
data = payload;
len = payload_len;
return 0;
}
static int rtcp_listen_cb(struct osmo_fd *ofd, unsigned int when)
{
osmo_cc_session_media_t *media = ofd->data;
int rc;
if (when & OSMO_FD_READ) {
rc = rtcp_receive(media->rtp_ofd.fd);
if (rc < 0)
return rc;
}
if (codec->media->receive)
codec->media->receiver(codec, marker, media->rx_sequence, media->rx_timestamp, media->rx_ssrc, data, len);
if (codec->decoder)
free(data);
return 0;
}
void osmo_cc_rtp_close(osmo_cc_session_media_t *media)
{
if (media->rtp_socket) {
close(media->rtp_socket);
media->rtp_socket = 0;
if (media->rtp_ofd.fd) {
osmo_fd_unregister(&media->rtp_ofd);
close(media->rtp_ofd.fd);
media->rtp_ofd.fd = 0;
}
if (media->rtcp_socket) {
close(media->rtcp_socket);
media->rtcp_socket = 0;
if (media->rtcp_ofd.fd) {
osmo_fd_unregister(&media->rtcp_ofd);
close(media->rtcp_ofd.fd);
media->rtcp_ofd.fd = 0;
}
}

1
src/libosmocc/rtp.h
View File

@@ -3,6 +3,5 @@ void osmo_cc_set_rtp_ports(osmo_cc_session_config_t *conf, uint16_t from, uint16
int osmo_cc_rtp_open(osmo_cc_session_media_t *media);
int osmo_cc_rtp_connect(osmo_cc_session_media_t *media);
void osmo_cc_rtp_send(osmo_cc_session_codec_t *codec, uint8_t *data, int len, uint8_t marker, int inc_sequence, int inc_timestamp, void *priv);
int osmo_cc_rtp_receive(osmo_cc_session_media_t *media, void *priv);
void osmo_cc_rtp_close(osmo_cc_session_media_t *media);

1
src/libosmocc/screen.c
View File

@@ -24,6 +24,7 @@
#include <arpa/inet.h>
#include <unistd.h>
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "../libdebug/debug.h"
#include "endpoint.h"
#include "message.h"

1
src/libosmocc/sdp.c
View File

@@ -25,6 +25,7 @@
#include <errno.h>
#include "../libdebug/debug.h"
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "endpoint.h"
#include "sdp.h"

17
src/libosmocc/session.c
View File

@@ -25,6 +25,7 @@
#include <sys/time.h>
#include <inttypes.h>
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "../libdebug/debug.h"
#include "../liboptions/options.h"
#include "endpoint.h"
@@ -622,19 +623,3 @@ int osmo_cc_session_if_codec(osmo_cc_session_codec_t *codec, const char *name, u
&& codec->payload_channels == channels);
}
int osmo_cc_session_handle(osmo_cc_session_t *session, void *codec_priv)
{
osmo_cc_session_media_t *media;
int w = 0, rc;
osmo_cc_session_for_each_media(session->media_list, media) {
do {
rc = osmo_cc_rtp_receive(media, codec_priv);
if (rc >= 0)
w = 1;
} while (rc >= 0);
}
return w;
}

5
src/libosmocc/session.h
View File

@@ -80,8 +80,8 @@ typedef struct osmo_cc_session_media {
struct osmo_cc_session_codec *codec_list;
int send, receive;
void (*receiver)(struct osmo_cc_session_codec *codec, uint8_t marker, uint16_t sequence_number, uint32_t timestamp, uint32_t ssrc, uint8_t *data, int len);
int rtp_socket;
int rtcp_socket;
struct osmo_fd rtp_ofd;
struct osmo_fd rtcp_ofd;
uint32_t tx_ssrc, rx_ssrc;
uint16_t tx_sequence, rx_sequence;
uint32_t tx_timestamp, rx_timestamp;
@@ -126,5 +126,4 @@ const char *osmo_cc_session_addrtype2string(enum osmo_cc_session_addrtype addrty
const char *osmo_cc_session_media_type2string(enum osmo_cc_session_media_type media_type);
const char *osmo_cc_session_media_proto2string(enum osmo_cc_session_media_proto media_proto);
int osmo_cc_session_if_codec(osmo_cc_session_codec_t *codec, const char *name, uint32_t rate, int channels);
int osmo_cc_session_handle(osmo_cc_session_t *session, void *codec_priv);

128
src/libosmocc/socket.c
View File

@@ -28,6 +28,7 @@
#include <netdb.h>
#include "../libdebug/debug.h"
#include "../libtimer/timer.h"
#include "../libselect/select.h"
#include "message.h"
#include "cause.h"
#include "socket.h"
@@ -101,12 +102,14 @@ static void rx_keepalive_timeout(void *data)
close_conn(conn, OSMO_CC_SOCKET_CAUSE_TIMEOUT);
}
static int socket_listen_cb(struct osmo_fd *ofd, unsigned int when);
/* create socket process and bind socket */
int osmo_cc_open_socket(osmo_cc_socket_t *os, const char *host, uint16_t port, void *priv, void (*recv_msg_cb)(void *priv, uint32_t callref, osmo_cc_msg_t *msg), uint8_t location)
{
int try = 0, auto_port = 0;
struct addrinfo *result, *rp;
int rc, sock, flags;
int rc, sock;
memset(os, 0, sizeof(*os));
@@ -148,15 +151,16 @@ try_again:
rc = listen(sock, 10);
if (rc < 0) {
PDEBUG(DCC, DEBUG_ERROR, "Failed to listen on socket.\n");
close(sock);
return rc;
}
/* set nonblocking io */
flags = fcntl(sock, F_GETFL);
flags |= O_NONBLOCK;
fcntl(sock, F_SETFL, flags);
os->socket = sock;
/* register */
os->ofd.fd = sock;
os->ofd.cb = socket_listen_cb;
os->ofd.data = os;
os->ofd.when = OSMO_FD_READ;
osmo_fd_register(&os->ofd);
os->recv_msg_cb = recv_msg_cb;
os->priv = priv;
os->location = location;
@@ -164,6 +168,8 @@ try_again:
return port;
}
static int socket_conn_cb(struct osmo_fd *ofd, unsigned int when);
/* create a connection */
static osmo_cc_conn_t *open_conn(osmo_cc_socket_t *os, int sock, uint32_t callref, int read_setup)
{
@@ -176,7 +182,11 @@ static osmo_cc_conn_t *open_conn(osmo_cc_socket_t *os, int sock, uint32_t callre
abort();
}
conn->os = os;
conn->socket = sock;
conn->ofd.fd = sock;
conn->ofd.cb = socket_conn_cb;
conn->ofd.data = conn;
conn->ofd.when = OSMO_FD_READ;
osmo_fd_register(&conn->ofd);
conn->read_version = 1;
conn->write_version = 1;
conn->read_setup = read_setup;
@@ -221,8 +231,10 @@ static void close_conn(osmo_cc_conn_t *conn, uint8_t socket_cause)
PDEBUG(DCC, DEBUG_DEBUG, "Destroy socket connection (callref %d).\n", conn->callref);
/* close socket */
if (conn->socket)
close(conn->socket);
if (conn->ofd.fd) {
osmo_fd_unregister(&conn->ofd);
close(conn->ofd.fd);
}
/* free partly received message */
if (conn->read_msg)
osmo_cc_free_msg(conn->read_msg);
@@ -250,9 +262,10 @@ void osmo_cc_close_socket(osmo_cc_socket_t *os)
while (os->conn_list)
close_conn(os->conn_list, 0);
/* close socket */
if (os->socket > 0) {
close(os->socket);
os->socket = 0;
if (os->ofd.fd > 0) {
osmo_fd_unregister(&os->ofd);
close(os->ofd.fd);
os->ofd.fd = 0;
}
/* free send queue */
while ((ml = os->write_list)) {
@@ -293,7 +306,7 @@ static int receive_conn(osmo_cc_conn_t *conn)
/* get version from remote */
if (conn->read_version) {
rc = recv(conn->socket, conn->read_version_string + conn->read_version_pos, strlen(version_string) - conn->read_version_pos, 0);
rc = recv(conn->ofd.fd, conn->read_version_string + conn->read_version_pos, strlen(version_string) - conn->read_version_pos, 0);
if (rc < 0 && errno == EAGAIN)
return work;
work = 1;
@@ -320,7 +333,7 @@ static int receive_conn(osmo_cc_conn_t *conn)
try_next_message:
/* read message header from remote */
if (!conn->read_msg) {
rc = recv(conn->socket, ((uint8_t *)&conn->read_hdr) + conn->read_pos, sizeof(conn->read_hdr) - conn->read_pos, 0);
rc = recv(conn->ofd.fd, ((uint8_t *)&conn->read_hdr) + conn->read_pos, sizeof(conn->read_hdr) - conn->read_pos, 0);
if (rc < 0 && errno == EAGAIN)
return work;
work = 1;
@@ -344,7 +357,7 @@ try_next_message:
len = ntohs(msg->length_networkorder);
if (len == 0)
goto empty_message;
rc = recv(conn->socket, msg->data + conn->read_pos, len - conn->read_pos, 0);
rc = recv(conn->ofd.fd, msg->data + conn->read_pos, len - conn->read_pos, 0);
if (rc < 0 && errno == EAGAIN)
return work;
work = 1;
@@ -397,7 +410,7 @@ static int transmit_conn(osmo_cc_conn_t *conn)
/* send socket version to remote */
if (conn->write_version) {
rc = write(conn->socket, version_string, strlen(version_string));
rc = write(conn->ofd.fd, version_string, strlen(version_string));
if (rc < 0 && errno == EAGAIN)
return work;
work = 1;
@@ -416,7 +429,7 @@ static int transmit_conn(osmo_cc_conn_t *conn)
timer_stop(&conn->tx_keepalive_timer);
msg = conn->write_list->msg;
len = sizeof(*msg) + ntohs(msg->length_networkorder);
rc = write(conn->socket, msg, len);
rc = write(conn->ofd.fd, msg, len);
if (rc < 0 && errno == EAGAIN)
return work;
work = 1;
@@ -460,8 +473,6 @@ close:
*/
int osmo_cc_handle_socket(osmo_cc_socket_t *os)
{
struct sockaddr_storage sa;
socklen_t slen = sizeof(sa);
int sock;
osmo_cc_conn_t *conn;
osmo_cc_msg_list_t *ml, **mlp;
@@ -487,6 +498,7 @@ int osmo_cc_handle_socket(osmo_cc_socket_t *os)
while (*mlp)
mlp = &((*mlp)->next);
*mlp = ml;
conn->ofd.when |= OSMO_FD_WRITE;
/* done with message */
continue;
}
@@ -526,7 +538,7 @@ int osmo_cc_handle_socket(osmo_cc_socket_t *os)
sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
if (sock < 0)
continue;
/* set nonblocking io */
/* set nonblocking io, to prevent connect() and subsequent reads from blocking */
flags = fcntl(sock, F_GETFL);
flags |= O_NONBLOCK;
fcntl(sock, F_SETFL, flags);
@@ -550,34 +562,58 @@ int osmo_cc_handle_socket(osmo_cc_socket_t *os)
conn = open_conn(os, sock, ml->callref, 0);
/* attach to list */
conn->write_list = ml;
conn->ofd.when |= OSMO_FD_WRITE;
/* done with (setup) message */
}
/* handle new socket connection */
while ((sock = accept(os->socket, (struct sockaddr *)&sa, &slen)) > 0) {
work = 1;
/* set nonblocking io */
flags = fcntl(sock, F_GETFL);
flags |= O_NONBLOCK;
fcntl(sock, F_SETFL, flags);
/* create connection */
open_conn(os, sock, 0, 1);
}
/* start with list after each read/write, because while handling (the message), one or more connections may be destroyed */
for (conn = os->conn_list; conn; conn=conn->next) {
/* check for rx */
work = receive_conn(conn);
/* if "change" is set, connection list might have changed, so we restart processing the list */
if (work)
break;
/* check for tx */
work = transmit_conn(conn);
/* if "change" is set, connection list might have changed, so we restart processing the list */
if (work)
break;
}
return work;
}
static int socket_listen_cb(struct osmo_fd *ofd, unsigned int when)
{
osmo_cc_socket_t *os = ofd->data;
struct sockaddr_storage sa;
socklen_t slen = sizeof(sa);
int sock;
int flags;
if (when & OSMO_FD_READ) {
/* handle new socket connection */
if ((sock = accept(os->ofd.fd, (struct sockaddr *)&sa, &slen)) > 0) {
/* set nonblocking io, to prevent subsequent reads from blocking */
flags = fcntl(sock, F_GETFL);
flags |= O_NONBLOCK;
fcntl(sock, F_SETFL, flags);
/* create connection */
open_conn(os, sock, 0, 1);
}
}
return 0;
}
static int socket_conn_cb(struct osmo_fd *ofd, unsigned int when)
{
osmo_cc_conn_t *conn = ofd->data;
int work;
if (when & OSMO_FD_READ) {
/* check for rx */
work = receive_conn(conn);
/* if "change" is set, connection list might have changed, so we restart processing the list */
if (work)
return 0;
}
if (when & OSMO_FD_WRITE) {
/* check for tx */
work = transmit_conn(conn);
/* if "change" is set, connection list might have changed, so we restart processing the list */
if (work)
return 0;
else
conn->ofd.when &= ~OSMO_FD_WRITE;
}
return 0;
}

4
src/libosmocc/socket.h
View File

@@ -12,7 +12,7 @@ struct osmo_cc_socket;
typedef struct osmo_cc_conn {
struct osmo_cc_conn *next;
struct osmo_cc_socket *os;
int socket;
struct osmo_fd ofd;
uint32_t callref;
int read_setup;
int read_version;
@@ -28,7 +28,7 @@ typedef struct osmo_cc_conn {
} osmo_cc_conn_t;
typedef struct osmo_cc_socket {
int socket;
struct osmo_fd ofd;
osmo_cc_conn_t *conn_list;
osmo_cc_msg_list_t *write_list;
void (*recv_msg_cb)(void *priv, uint32_t callref, osmo_cc_msg_t *msg);