rtp-load-gen: Add CTRL interface

Change-Id: I1122e5c125f03b615d587f734fc1b7eadf827a27

contrib/rtp-load-gen/Makefile

@@ -0,0 +1,11 @@
+CFLAGS=-g -Wall $(shell pkg-config --cflags libosmocore libosmoctrl libosmo-netif liburing)
+LIBS=-lpthread $(shell pkg-config --libs libosmocore libosmoctrl libosmo-netif liburing)
+
+rtp-load-gen: rtp-load-gen.o rtp_provider.o rtp_provider_static.o ctrl_if.o
+	$(CC) -o $@ $^ $(LIBS)
+
+%.o: %.c
+	$(CC) $(CFLAGS) -o $@ -c $^
+
+clean:
+	@rm rtp-load-gen *.o

contrib/rtp-load-gen/ctrl_if.c

@@ -0,0 +1,214 @@
+/* CTRL interface of rtpsource program
+ *
+ * (C) 2020 by Harald Welte <laforge@gnumonks.org>
+ *
+ * All Rights Reserved
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <osmocom/ctrl/control_cmd.h>
+
+#include "internal.h"
+#include "rtp_provider.h"
+
+static struct rtpsim_connection *find_connection_by_cname(const char *cname)
+{
+	struct rtpsim_connection *rtpc;
+	struct rtpsim_instance *ri;
+
+	pthread_rwlock_rdlock(&g_rtpsim->rwlock);
+	llist_for_each_entry(ri, &g_rtpsim->instances, list) {
+		rtpc = rtpsim_conn_find(ri, cname);
+		if (rtpc) {
+			pthread_rwlock_unlock(&g_rtpsim->rwlock);
+			return rtpc;
+		}
+	}
+	pthread_rwlock_unlock(&g_rtpsim->rwlock);
+	return NULL;
+}
+
+static struct rtpsim_connection *create_connection(const char *cname, enum codec_type codec)
+{
+	struct rtpsim_connection *rtpc;
+	struct rtpsim_instance *ri;
+
+	pthread_rwlock_rdlock(&g_rtpsim->rwlock);
+	llist_for_each_entry(ri, &g_rtpsim->instances, list) {
+		rtpc = rtpsim_conn_reserve(ri, cname, codec);
+		if (rtpc) {
+			pthread_rwlock_unlock(&g_rtpsim->rwlock);
+			return rtpc;
+		}
+	}
+	pthread_rwlock_unlock(&g_rtpsim->rwlock);
+	return NULL;
+}
+
+static int connect_connection(struct rtpsim_connection *rtpc, const char *remote_host,
+			      uint16_t remote_port, uint8_t pt)
+{
+	int rc;
+
+	osmo_sockaddr_str_from_str(&rtpc->cfg.remote, remote_host, remote_port);
+	rtpc->cfg.pt = pt;
+
+	rc = rtpsim_conn_connect(rtpc);
+	if (rc < 0)
+		return rc;
+
+	rc = rtpsim_conn_start(rtpc);
+
+	return rc;
+}
+
+static int delete_connection(struct rtpsim_connection *rtpc)
+{
+	rtpsim_conn_stop(rtpc);
+	rtpsim_conn_unreserve(rtpc);
+	return 0;
+}
+
+CTRL_CMD_DEFINE_WO_NOVRF(rtp_create, "rtp_create");
+static int set_rtp_create(struct ctrl_cmd *cmd, void *data)
+{
+	struct rtpsim_connection *conn;
+	const char *cname, *codec_str;
+	char *tmp, *saveptr;
+	enum codec_type codec;
+
+	tmp = talloc_strdup(cmd, cmd->value);
+	OSMO_ASSERT(tmp);
+
+	cname = strtok_r(tmp, ",", &saveptr);
+	codec_str = strtok_r(NULL, ",", &saveptr);
+
+	if (!cname || !codec_str) {
+		cmd->reply = "Format is cname,codec";
+		goto error;
+	}
+
+	if (find_connection_by_cname(cname)) {
+		cmd->reply = "Connection already exists for cname";
+		goto error;
+	}
+
+	codec = get_string_value(codec_type_names, codec_str);
+	if (codec < 0) {
+		cmd->reply = "Invalid codec name (try GSM_FR, GSM_EFR etc.)";
+		goto error;
+	}
+
+	conn = create_connection(cname, codec);
+	if (!conn) {
+		cmd->reply = "Error creating RTP connection";
+		goto error;
+	}
+
+	/* Respond */
+	cmd->reply = talloc_asprintf(cmd, "%s,%s,%d", conn->cname, conn->cfg.local.ip, conn->cfg.local.port);
+	talloc_free(tmp);
+	return CTRL_CMD_REPLY;
+
+error:
+	talloc_free(tmp);
+	return CTRL_CMD_ERROR;
+}
+
+CTRL_CMD_DEFINE_WO_NOVRF(rtp_connect, "rtp_connect");
+static int set_rtp_connect(struct ctrl_cmd *cmd, void *data)
+{
+	struct rtpsim_connection *conn;
+	const char *cname, *remote_host, *remote_port, *pt;
+	char *tmp, *saveptr;
+	int rc;
+
+	tmp = talloc_strdup(cmd, cmd->value);
+	OSMO_ASSERT(tmp);
+
+	/* FIXME: parse command */
+	cname = strtok_r(tmp, ",", &saveptr);
+	remote_host = strtok_r(NULL, ",", &saveptr);
+	remote_port = strtok_r(NULL, ",", &saveptr);
+	pt = strtok_r(NULL, ",", &saveptr);
+
+	if (!cname || !remote_host || !remote_port || !pt) {
+		cmd->reply = "Format is cname,remote_host,remote_port,pt";
+		talloc_free(tmp);
+		return CTRL_CMD_ERROR;
+	}
+
+	conn = find_connection_by_cname(cname);
+	if (!conn) {
+		cmd->reply = "Error finding RTP connection for connect";
+		talloc_free(tmp);
+		return CTRL_CMD_ERROR;
+	}
+
+	rc = connect_connection(conn, remote_host, atoi(remote_port), atoi(pt));
+	if (rc < 0) {
+		cmd->reply = "Error binding RTP connection";
+		talloc_free(tmp);
+		return CTRL_CMD_ERROR;
+	}
+
+	/* Respond */
+	talloc_free(tmp);
+	cmd->reply = talloc_asprintf(cmd, "%s,%s,%d,%d", conn->cname, conn->cfg.remote.ip,
+					conn->cfg.remote.port, conn->cfg.pt);
+	return CTRL_CMD_REPLY;
+}
+
+CTRL_CMD_DEFINE_WO_NOVRF(rtp_delete, "rtp_delete");
+static int set_rtp_delete(struct ctrl_cmd *cmd, void *data)
+{
+	struct rtpsim_connection *conn;
+	const char *cname = cmd->value;
+
+	conn = find_connection_by_cname(cname);
+	if (!conn) {
+		cmd->reply = "Error finding RTP connection for delete";
+		return CTRL_CMD_ERROR;
+	}
+	cmd->reply = talloc_asprintf(cmd, "%s", conn->cname);
+
+	delete_connection(conn);
+
+	/* Respond */
+	return CTRL_CMD_REPLY;
+}
+
+
+
+
+
+int rtpsource_ctrl_cmds_install(void)
+{
+	int rc;
+
+	rc = ctrl_cmd_install(CTRL_NODE_ROOT, &cmd_rtp_create);
+	if (rc)
+		goto end;
+
+	rc = ctrl_cmd_install(CTRL_NODE_ROOT, &cmd_rtp_connect);
+	if (rc)
+		goto end;
+
+	rc = ctrl_cmd_install(CTRL_NODE_ROOT, &cmd_rtp_delete);
+	if (rc)
+		goto end;
+end:
+	return rc;
+}

contrib/rtp-load-gen/internal.h

@@ -0,0 +1,112 @@
+#pragma once
+#include <stdint.h>
+#include <pthread.h>
+#include <liburing.h>
+
+#include <osmocom/core/linuxlist.h>
+#include <osmocom/core/sockaddr_str.h>
+#include <osmocom/core/rate_ctr.h>
+#include <osmocom/core/select.h>
+#include <osmocom/ctrl/control_if.h>
+
+struct rtp_provider_instance;
+
+/* configuration of one RTP connection/socket */
+struct rtpsim_connection_cfg {
+	struct osmo_sockaddr_str local;
+	struct osmo_sockaddr_str remote;
+	uint8_t pt;
+	uint32_t ssrc;
+	uint32_t duration;
+};
+
+/* TX side state of RTP connection/socket */
+struct rtpsim_connection_tx {
+	bool enabled;
+	uint32_t timestamp;
+	uint16_t seq;
+
+	struct rtp_provider_instance *rtp_prov_inst;
+
+	/* transmit buffer for outgoing messages */
+	uint8_t *buf;
+	/* used part of buffer */
+	size_t buf_len;
+};
+
+/* RX side state of RTP connection/socket */
+struct rtpsim_connection_rx {
+	bool enabled;
+	/* receive buffer for incoming messages */
+	uint8_t *buf;
+	/* used length of buffer */
+	size_t buf_len;
+};
+
+struct rtpsim_instance;
+
+/* One RTP connection/socket */
+struct rtpsim_connection {
+	/* index in rtp_instance.connections */
+	unsigned int idx;
+	/* back-pointer */
+	struct rtpsim_instance *inst;
+
+	struct rtpsim_connection_cfg cfg;
+	struct rtpsim_connection_tx tx;
+	struct rtpsim_connection_rx rx;
+	struct rate_ctr_group *ctrg;
+
+	/* socket file descriptor */
+	int fd;
+	char *cname;
+};
+
+struct rtpsim_instance_cfg {
+	int num;
+	uint16_t base_port;
+	unsigned int num_flows;
+};
+
+/* one instance of the RTP simulator; typically one per worker thread */
+struct rtpsim_instance {
+	/* element in application global list of instances */
+	struct llist_head list;
+	struct rtpsim_instance_cfg cfg;
+	/* per-instance io_uring */
+	struct io_uring ring;
+	/* per-instance timerfd */
+	int timerfd;
+	/* counter group of per-instance counters */
+	struct rate_ctr_group *ctrg;
+
+	struct rtpsim_connection **connections;
+	/* size of 'connections' in number of pointers */
+	unsigned int connections_size;
+};
+
+struct rtpsim_global {
+	/* global list of instances */
+	struct llist_head instances;
+	pthread_rwlock_t rwlock;
+
+	struct ctrl_handle *ctrl;
+};
+
+
+enum {
+	DMAIN,
+};
+
+enum codec_type;
+
+extern struct rtpsim_global *g_rtpsim;
+
+struct rtpsim_connection *rtpsim_conn_find(struct rtpsim_instance *ri, const char *cname);
+struct rtpsim_connection *rtpsim_conn_reserve(struct rtpsim_instance *ri, const char *cname, enum codec_type codec);
+int rtpsim_conn_start(struct rtpsim_connection *rtpc);
+void rtpsim_conn_stop(struct rtpsim_connection *rtpc);
+void rtpsim_conn_unreserve(struct rtpsim_connection *rtpc);
+int rtpsim_conn_connect(struct rtpsim_connection *rtpc);
+
+int rtpsource_ctrl_cmds_install(void);

contrib/rtp-load-gen/rtp-load-gen.c

@@ -0,0 +1,560 @@
+#include <liburing.h>
+#include <string.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <sys/timerfd.h>
+
+#include <osmocom/core/talloc.h>
+#include <osmocom/core/utils.h>
+#include <osmocom/core/linuxlist.h>
+#include <osmocom/core/sockaddr_str.h>
+#include <osmocom/core/socket.h>
+#include <osmocom/core/rate_ctr.h>
+#include <osmocom/core/application.h>
+#include <osmocom/core/stats.h>
+#include <osmocom/netif/rtp.h>
+
+#include "internal.h"
+#include "rtp_provider.h"
+#include "internal.h"
+
+/* use a separate rx-completion thread: submit from main, reap from completion */
+//#define USE_CQ_THREAD
+
+/* use registered files: Doesn't seem to work with sockets? */
+//#define USE_REGISTERED_FILES
+
+/* use registered buffers (mapped once into kernel, rather than at every write */
+#define USE_REGISTERED_BUFFERS
+
+/* number of sockets/flows to create */
+#define NUM_FLOWS	4096
+
+/* number of workers to spawn.  Each worker will get an equal share of NR_FLOWS to work on */
+#define NR_WORKERS	4
+
+/* size of rx/tx buffer for one RTP frame */
+#define BUF_SIZE	256
+
+#define NUM_FLOWS_PER_WORKER	(NUM_FLOWS/NR_WORKERS)
+
+#define TX_BUF_IDX	0
+#define RX_BUF_IDX	1
+
+struct rtpsim_global *g_rtpsim;
+
+enum rtpsim_conn_ctr {
+	RTP_CONN_CTR_TX_PKTS,
+	RTP_CONN_CTR_TX_BYTES,
+	RTP_CONN_CTR_RX_PKTS,
+	RTP_CONN_CTR_RX_BYTES,
+	RTP_CONN_CTR_RX_INVALID,
+};
+
+static const struct rate_ctr_desc rtpsim_conn_ctrs[] = {
+	[RTP_CONN_CTR_TX_PKTS] = { "tx_pkts:total", "Transmitted packets" },
+	[RTP_CONN_CTR_TX_BYTES] = { "tx_bytes:total", "Transmitted bytes" },
+	[RTP_CONN_CTR_RX_PKTS] = { "rx_pkts:total", "Received packets (total)" },
+	[RTP_CONN_CTR_RX_BYTES] = { "rx_bytes:total", "Transmitted bytes" },
+	[RTP_CONN_CTR_RX_INVALID] = { "rx_pkts:invalid", "Received packets (invalidl)" },
+};
+static const struct rate_ctr_group_desc rtpsim_conn_ctrg_desc = {
+	.group_name_prefix = "rtpsim_conn",
+	.group_description = "RTP Simulator Connection",
+	.class_id = 0,
+	.num_ctr = ARRAY_SIZE(rtpsim_conn_ctrs),
+	.ctr_desc = rtpsim_conn_ctrs,
+};
+
+enum rtpsim_ctr {
+	RTP_INST_TIMERS_TOTAL,
+	RTP_INST_TIMERS_LATE,
+};
+static const struct rate_ctr_desc rtpsim_ctrs[] = {
+	[RTP_INST_TIMERS_TOTAL] = { "timers:total", "Timers expiring (total)" },
+	[RTP_INST_TIMERS_LATE] = { "timers:late", "Timers expiring (late)" },
+};
+static const struct rate_ctr_group_desc rtpsim_ctrg_desc = {
+	.group_name_prefix = "rtpsim",
+	.group_description = "RTP Simulator Instance",
+	.class_id = 0,
+	.num_ctr = ARRAY_SIZE(rtpsim_ctrs),
+	.ctr_desc = rtpsim_ctrs,
+};
+
+struct rtpsim_instance *rtpsim_instance_init(void *ctx, const struct rtpsim_instance_cfg *rmp)
+{
+	struct rtpsim_instance *ri = talloc_zero(ctx, struct rtpsim_instance);
+	int rc;
+
+	if (!ri)
+		return NULL;
+	ri->connections_size = NUM_FLOWS_PER_WORKER;
+	ri->connections = talloc_zero_size(ri, sizeof(struct rtpsim_connection *)*ri->connections_size);
+	if (!ri->connections) {
+		talloc_free(ri);
+		return NULL;
+	}
+
+	ri->cfg = *rmp;
+	rc = io_uring_queue_init(NUM_FLOWS_PER_WORKER*2, &ri->ring, 0);
+	if (rc < 0) {
+		talloc_free(ri);
+		return NULL;
+	}
+	ri->ctrg = rate_ctr_group_alloc(ri, &rtpsim_ctrg_desc, rmp->num);
+	OSMO_ASSERT(ri->ctrg);
+	return ri;
+}
+
+static int rtpsim_instance_conn_add(struct rtpsim_instance *ri, struct rtpsim_connection *rtpc)
+{
+	unsigned int i;
+
+	for (i = 0; i < ri->connections_size; i++) {
+		if (ri->connections[i] == NULL) {
+			ri->connections[i] = rtpc;
+			rtpc->idx = i;
+			return i;
+		}
+	}
+	return -ENOSPC;
+}
+
+static struct rtpsim_connection *
+rtpsim_conn_open_bind(struct rtpsim_instance *ri, const struct rtpsim_connection_cfg *rcfg)
+{
+	struct rtpsim_connection *rtpc = talloc_zero(ri, struct rtpsim_connection);
+	struct osmo_sockaddr sa_local;
+	int rc;
+
+	if (!rtpc)
+		return NULL;
+
+	rtpc->inst = ri;
+	rtpc->cfg = *rcfg;
+
+	osmo_sockaddr_str_to_sockaddr(&rtpc->cfg.local, &sa_local.u.sas);
+
+	rc = osmo_sock_init_osa(SOCK_DGRAM, IPPROTO_UDP, &sa_local, NULL, OSMO_SOCK_F_BIND);
+	if (rc < 0) {
+		talloc_free(rtpc);
+		return NULL;
+	}
+
+	rtpc->fd = rc;
+	rtpc->ctrg = rate_ctr_group_alloc(rtpc, &rtpsim_conn_ctrg_desc, rtpc->cfg.local.port);
+	OSMO_ASSERT(rtpc->ctrg);
+#ifndef USE_REGISTERED_BUFFERS
+	rtpc->tx.buf = talloc_zero_size(rtpc, BUF_SIZE);
+	rtpc->rx.buf = talloc_zero_size(rtpc, BUF_SIZE);
+#endif
+	OSMO_ASSERT(rtpsim_instance_conn_add(ri, rtpc) >= 0);
+
+	return rtpc;
+}
+
+/* find a connection for given cname (may be NULL to find unused connection) */
+struct rtpsim_connection *rtpsim_conn_find(struct rtpsim_instance *ri, const char *cname)
+{
+	int i;
+
+	for (i = 0; i < ri->connections_size; i++) {
+		struct rtpsim_connection *rtpc = ri->connections[i];
+		if (!rtpc)
+			continue;
+		if (!rtpc->cname) {
+			if (!cname)
+				return rtpc;
+		} else {
+			continue;
+		}
+		if (!strcmp(rtpc->cname, cname))
+			return rtpc;
+	}
+	return NULL;
+}
+
+/* reserve a connection; associates cname with it */
+struct rtpsim_connection *rtpsim_conn_reserve(struct rtpsim_instance *ri, const char *cname,
+					      enum codec_type codec)
+{
+	struct rtpsim_connection *rtpc;
+	const struct rtp_provider *rtp_prov;
+
+	rtp_prov = rtp_provider_find("static"); // TODO: configurable */
+	OSMO_ASSERT(rtp_prov);
+
+	rtpc = rtpsim_conn_find(ri, NULL);
+	if (!rtpc)
+		return NULL;
+
+	/* this is called from main thread, we cannot use per-thread talloc contexts
+	 * such as ri or rtpc */
+	rtpc->cname = talloc_strdup(g_rtpsim, cname);
+	rtpc->tx.rtp_prov_inst = rtp_provider_instance_alloc(g_rtpsim, rtp_prov, codec);
+	OSMO_ASSERT(rtpc->tx.rtp_prov_inst);
+
+	/* re-start from zero transmit sequence number */
+	rtpc->tx.seq = 0;
+
+	return rtpc;
+}
+
+int rtpsim_conn_start(struct rtpsim_connection *rtpc)
+{
+	rtpc->tx.enabled = true;
+	rtpc->rx.enabled = true;
+
+	return 0;
+}
+
+void rtpsim_conn_stop(struct rtpsim_connection *rtpc)
+{
+	/* disable Rx and Tx */
+	rtpc->tx.enabled = false;
+	rtpc->rx.enabled = false;
+}
+
+/* unreserve a connection; stops all rx/tx and removes cname */
+void rtpsim_conn_unreserve(struct rtpsim_connection *rtpc)
+{
+	rtp_provider_instance_free(rtpc->tx.rtp_prov_inst);
+	rtpc->tx.rtp_prov_inst = NULL;
+
+	talloc_free(rtpc->cname);
+	rtpc->cname = NULL;
+}
+
+/* connect a RTP connection to its remote peer (as in rtpc->cfg.remote) */
+int rtpsim_conn_connect(struct rtpsim_connection *rtpc)
+{
+	struct osmo_sockaddr sa_remote;
+	int rc;
+
+	osmo_sockaddr_str_to_sockaddr(&rtpc->cfg.remote, &sa_remote.u.sas);
+	rc = connect(rtpc->fd, &sa_remote.u.sa, sizeof(struct osmo_sockaddr));
+	return rc;
+}
+
+/* transmit one RTP frame for given connection */
+static int rtpsim_conn_tx_frame(struct rtpsim_connection *rtpc)
+{
+	struct rtp_hdr *rtph = (struct rtp_hdr *) rtpc->tx.buf;
+	struct io_uring_sqe *sqe;
+	uint8_t *payload;
+	int rc;
+
+	rtph->version = RTP_VERSION;
+	rtph->padding = 0;
+	rtph->extension = 0;
+	rtph->csrc_count = 0;
+	rtph->marker = 0;
+	rtph->payload_type = rtpc->cfg.pt;
+	rtph->sequence = htons(rtpc->tx.seq++);
+	rtph->timestamp = htonl(rtpc->tx.timestamp);
+	rtpc->tx.timestamp += rtpc->cfg.duration;
+	rtph->ssrc = htonl(rtpc->cfg.ssrc);
+	payload = rtpc->tx.buf + sizeof(*rtph);
+	/* add payload data */
+
+	rc = rtp_provider_instance_gen_frame(rtpc->tx.rtp_prov_inst, payload, BUF_SIZE-sizeof(*rtph));
+	OSMO_ASSERT(rc >= 0);
+	rtpc->tx.buf_len = sizeof(*rtph) + rc;
+
+	sqe = io_uring_get_sqe(&rtpc->inst->ring);
+	OSMO_ASSERT(sqe);
+	sqe->user_data = rtpc->idx;
+#ifdef USE_REGISTERED_FILES
+	io_uring_sqe_set_flags(sqe, IOSQE_FIXED_FILE);
+#ifdef USE_REGISTERED_BUFFERS
+	io_uring_prep_write_fixed(sqe, rtpc->idx, rtpc->tx.buf, rtpc->tx.buf_len, 0, TX_BUF_IDX);
+#else
+	io_uring_prep_write(sqe, rtpc->idx, rtpc->tx.buf, rtpc->tx.buf_len, 0);
+#endif
+#else /* REGISTERED FILES */
+#ifdef USE_REGISTERED_BUFFERS
+	io_uring_prep_write_fixed(sqe, rtpc->fd, rtpc->tx.buf, rtpc->tx.buf_len, 0, TX_BUF_IDX);
+#else
+	io_uring_prep_write(sqe, rtpc->fd, rtpc->tx.buf, rtpc->tx.buf_len, 0);
+#endif
+#endif /* REGISTERED_FILES */
+
+	return 0;
+}
+
+/* submit RX buffer for a RTP frame on given connection */
+static int rtpsim_conn_rx_prep(struct rtpsim_connection *rtpc)
+{
+	struct io_uring_sqe *sqe;
+
+	sqe = io_uring_get_sqe(&rtpc->inst->ring);
+	OSMO_ASSERT(sqe);
+	sqe->user_data = 0x8000000 | rtpc->idx;
+#ifdef USE_REGISTERED_FILES
+	io_uring_sqe_set_flags(sqe, IOSQE_FIXED_FILE);
+	/* FIXME */
+#else /* REGISTERED FILES */
+#ifdef USE_REGISTERED_BUFFERS
+	io_uring_prep_read_fixed(sqe, rtpc->fd, rtpc->rx.buf, BUF_SIZE, 0, RX_BUF_IDX);
+#else
+	io_uring_prep_read(sqe, rtpc->fd, rtpc->rx.buf, BUF_SIZE, 0);
+#endif
+#endif /* REGISTERED_FILES */
+	return 0;
+}
+
+/* process one completion entry */
+static void handle_completion(struct rtpsim_instance *ri, struct io_uring_cqe *cqe)
+{
+	struct rtpsim_connection *rtpc = ri->connections[cqe->user_data & 0x7fffffff];
+	OSMO_ASSERT(rtpc);
+
+	if (cqe->user_data & 0x80000000) {
+		/* read */
+		rate_ctr_inc(&rtpc->ctrg->ctr[RTP_CONN_CTR_RX_PKTS]);
+		rate_ctr_add(&rtpc->ctrg->ctr[RTP_CONN_CTR_RX_BYTES], cqe->res);
+		OSMO_ASSERT(cqe->res >= 0);
+	} else {
+		/* write */
+		rate_ctr_inc(&rtpc->ctrg->ctr[RTP_CONN_CTR_TX_PKTS]);
+		rate_ctr_add(&rtpc->ctrg->ctr[RTP_CONN_CTR_TX_BYTES], cqe->res);
+		OSMO_ASSERT(cqe->res == sizeof(struct rtp_hdr) + 33);
+	}
+	io_uring_cqe_seen(&ri->ring, cqe);
+}
+
+#ifdef USE_CQ_THREAD
+/* 'main' function for separate completion queue reaping thread */
+static void *reap_completion(void *_ri)
+{
+	struct rtpsim_instance *ri = _ri;
+	while (1) {
+		struct io_uring_cqe *cqe;
+		int rc;
+
+		rc = io_uring_wait_cqe(&ri->ring, &cqe);
+		OSMO_ASSERT(rc >= 0);
+		handle_completion(ri, cqe);
+	}
+}
+#endif
+
+extern int osmo_ctx_init(const char *id);
+
+static void rtpsim_main(const struct rtpsim_instance_cfg *rmp)
+{
+	struct rtpsim_instance *ri;
+	struct rtpsim_connection *rtpc;
+	int i, rc;
+
+	char namebuf[32];
+	snprintf(namebuf, sizeof(namebuf), "rtpsim_worker%d", rmp->num);
+	osmo_ctx_init(namebuf);
+
+	ri = rtpsim_instance_init(OTC_GLOBAL, rmp);
+	OSMO_ASSERT(ri);
+
+	pthread_rwlock_wrlock(&g_rtpsim->rwlock);
+	llist_add_tail(&ri->list, &g_rtpsim->instances);
+	pthread_rwlock_unlock(&g_rtpsim->rwlock);
+
+	/* create desired number of sockets */
+	printf("binding sockets\n");
+	for (i = 0; i < rmp->num_flows; i++) {
+		struct rtpsim_connection *rtpc;
+		struct rtpsim_connection_cfg rcfg = {};
+		rcfg.local = (struct osmo_sockaddr_str) {
+			.af = AF_INET,
+			.ip = "127.0.0.1",
+			.port = rmp->base_port + 2*i,
+		};
+		rcfg.remote = (struct osmo_sockaddr_str) {
+			.af = AF_INET,
+			.ip = "127.0.0.1",
+			.port = rmp->base_port + 2*i,
+		};
+		rcfg.pt = 3;
+		rcfg.ssrc = 0x80000000 + rmp->base_port + i;
+		rcfg.duration = 160;	/* 8000 Hz sampling rate / 50 Hz RTP rate */
+
+		rtpc = rtpsim_conn_open_bind(ri, &rcfg);
+		OSMO_ASSERT(rtpc);
+	}
+
+#if 1
+	/* HACK */
+	printf("connecting sockets\n");
+	for (i = 0; i < rmp->num_flows; i++) {
+		char namebuf[32];
+		snprintf(namebuf, sizeof(namebuf), "conn%d", i);
+		struct rtpsim_connection *rtpc = rtpsim_conn_reserve(ri, namebuf, CODEC_GSM_FR);
+		OSMO_ASSERT(rtpc);
+		OSMO_ASSERT(rtpsim_conn_connect(rtpc) == 0);
+		OSMO_ASSERT(rtpsim_conn_start(rtpc) == 0);
+	}
+#endif
+
+#ifdef USE_REGISTERED_FILES
+	/* register all our file descriptors; seems to fail on 5.8.x ? */
+	int fds[NUM_FLOWS_PER_WORKER];
+	for (i = 0; i < ri->connections_size; i++) {
+		if (!rtpc) {
+			fds[i] = -1;
+			continue;
+		}
+		rtpc = ri->connections[i];
+		fds[i] = rtpc->fd;
+	}
+	printf("Registering %d files\n", i);
+	rc = io_uring_register_files(&ri->ring, fds, i);
+	printf("rc = %d: %s\n", rc, strerror(-rc));
+	OSMO_ASSERT(rc == 0);
+#endif
+
+#ifdef USE_REGISTERED_BUFFERS
+	/* register two large buffers for Rx and Tx; assign per-connection
+	 * buffers within those two registered buffers */
+	void *largebuf_tx = talloc_zero_size(ri, rmp->num_flows * BUF_SIZE);
+	void *largebuf_rx = talloc_zero_size(ri, rmp->num_flows * BUF_SIZE);
+	struct iovec iov[2] = {
+		[TX_BUF_IDX] = {
+			.iov_base = largebuf_tx,
+			.iov_len = rmp->num_flows * BUF_SIZE,
+		},
+		[RX_BUF_IDX] = {
+			.iov_base = largebuf_rx,
+			.iov_len = rmp->num_flows * BUF_SIZE,
+		},
+	};
+	printf("Registering buffers for %d sockets\n", i);
+	rc = io_uring_register_buffers(&ri->ring, iov, ARRAY_SIZE(iov));
+	printf("rc = %d: %s\n", rc, strerror(-rc));
+	OSMO_ASSERT(rc == 0);
+	for (i = 0; i < ri->connections_size; i++) {
+		rtpc = ri->connections[i];
+		if (!rtpc)
+			continue;
+		rtpc->tx.buf = largebuf_tx + (i * BUF_SIZE);
+		rtpc->rx.buf = largebuf_rx + (i * BUF_SIZE);
+	}
+#endif
+
+#ifdef USE_CQ_THREAD
+	/* start a separate completion thread instead of handling completions in-line */
+	pthread_t complete;
+	rc = pthread_create(&complete, NULL, reap_completion, ri);
+	OSMO_ASSERT(rc >= 0);
+#endif
+
+	/* start timerfd every 20ms */
+	ri->timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
+	OSMO_ASSERT(ri->timerfd >= 0);
+	struct itimerspec its = (struct itimerspec) {
+		.it_interval = { 0, 20*1000*1000 },
+		.it_value = { 0, 20*1000*1000 },
+	};
+	rc = timerfd_settime(ri->timerfd, 0, &its, NULL);
+	OSMO_ASSERT(rc == 0);
+
+	/* start transmitting */
+
+	while (1) {
+		/* the assumption here is that every flow wants to write 50
+		 * packets per second, so we try try to submit one write for each
+		 * flow every 20ms */
+		unsigned int submit_granularity = rmp->num_flows/50;
+		/* number of 20ms timer expirations */
+		uint64_t num_exp;
+		unsigned int t;
+
+		if (submit_granularity <= 0)
+			submit_granularity = 1;
+
+		/* read from timerfd to pace the 20ms inter packet interval */
+		rc = read(ri->timerfd, &num_exp, sizeof(num_exp));
+		OSMO_ASSERT(rc == sizeof(num_exp));
+
+		rate_ctr_add(&ri->ctrg->ctr[RTP_INST_TIMERS_TOTAL], num_exp);
+		if (num_exp != 1) {
+			fputc('X', stdout);
+			rate_ctr_add(&ri->ctrg->ctr[RTP_INST_TIMERS_LATE], num_exp-1);
+		} else {
+			fputc('.', stdout);
+		}
+		fflush(stdout);
+
+		for (t = 0; t < num_exp; t++) {
+			for (i = 0; i < ri->connections_size; i++) {
+				rtpc = ri->connections[i];
+				if (!rtpc)
+					continue;
+				if (rtpc->tx.enabled)
+					rtpsim_conn_tx_frame(rtpc);
+				if (rtpc->rx.enabled)
+					rtpsim_conn_rx_prep(rtpc);
+				if ((i % submit_granularity) == 0) {
+					int pending = io_uring_submit(&ri->ring);
+
+#ifndef USE_CQ_THREAD
+					for (int j = 0; j < pending; j++) {
+						struct io_uring_cqe *cqe;
+						int rc;
+
+						rc = io_uring_wait_cqe(&ri->ring, &cqe);
+						OSMO_ASSERT(rc >= 0);
+						handle_completion(ri, cqe);
+					}
+#endif /* USE_CQ_THREAD */
+				}
+			}
+		}
+	}
+
+}
+
+static void *rtpsim_worker_thread(void *_rmp)
+{
+	rtpsim_main((struct rtpsim_instance_cfg *)_rmp);
+	return NULL;
+}
+
+int main(int argc, char **argv)
+{
+	pthread_t worker[NR_WORKERS];
+	struct rtpsim_instance_cfg rmp[NR_WORKERS];
+	int i, rc;
+
+	g_rtpsim = talloc_zero(NULL, struct rtpsim_global);
+	OSMO_ASSERT(g_rtpsim);
+	INIT_LLIST_HEAD(&g_rtpsim->instances);
+	pthread_rwlock_init(&g_rtpsim->rwlock, NULL);
+
+	osmo_init_logging2(g_rtpsim, NULL);
+//	osmo_stats_init(g_rtpsim);
+
+	/* Create worker threads */
+	for (i = 0; i < NR_WORKERS; i++) {
+		int rc;
+		rmp[i].num = i;
+		rmp[i].num_flows = NUM_FLOWS_PER_WORKER;
+		rmp[i].base_port = 10000 + i * (2 * rmp[i].num_flows);
+		rc = pthread_create(&worker[i], NULL, rtpsim_worker_thread, &rmp[i]);
+		OSMO_ASSERT(rc >= 0);
+	}
+
+	/* CTRL interface */
+	//g_rtpsim->ctrl = ctrl_interface_setup_dynip(g_rss, ctrl_vty_get_bind_addr(), 11111, NULL);
+	g_rtpsim->ctrl = ctrl_interface_setup_dynip(g_rtpsim, "127.0.0.1", 11111, NULL);
+	OSMO_ASSERT(g_rtpsim->ctrl);
+	rc = rtpsource_ctrl_cmds_install();
+	OSMO_ASSERT(rc == 0);
+
+	while (1) {
+		osmo_select_main(0);
+	}
+
+	for (i = 0; i < NR_WORKERS; i++) {
+		pthread_join(worker[i], NULL);
+	}
+}

contrib/rtp-load-gen/rtp_provider.c

@@ -0,0 +1,71 @@
+#include <osmocom/core/talloc.h>
+#include <osmocom/core/utils.h>
+#include <osmocom/core/linuxlist.h>
+#include <osmocom/core/logging.h>
+
+#include "rtp_provider.h"
+#include "internal.h"
+
+
+static LLIST_HEAD(g_providers);
+
+const struct value_string codec_type_names[] = {
+	{ CODEC_ULAW,		"ULAW" },
+	{ CODEC_ALAW,		"ALAW" },
+	{ CODEC_GSM_FR,		"GSM_FR" },
+	{ CODEC_GSM_EFR,	"GSM_EFR" },
+	{ CODEC_GSM_HR,		"GSM_HR" },
+	{ CODEC_AMR_4_75,	"AMR_4_75" },
+	{ CODEC_AMR_5_15,	"AMR_5_15" },
+	{ CODEC_AMR_5_90,	"AMR_5_90" },
+	{ CODEC_AMR_6_70,	"AMR_6_70" },
+	{ CODEC_AMR_7_40,	"AMR_7_40" },
+	{ CODEC_AMR_7_95,	"AMR_7_95" },
+	{ CODEC_AMR_10_2,	"AMR_10_2" },
+	{ CODEC_AMR_12_2,	"AMR_12_2" },
+	{ CODEC_AMR_SID,	"AMR_SID" },
+	{ 0, NULL }
+};
+
+void rtp_provider_register(struct rtp_provider *prov)
+{
+	llist_add_tail(&prov->list, &g_providers);
+}
+
+const struct rtp_provider *rtp_provider_find(const char *name)
+{
+	struct rtp_provider *p;
+	llist_for_each_entry(p, &g_providers, list) {
+		if (!strcmp(name, p->name))
+			return p;
+	}
+	LOGP(DMAIN, LOGL_ERROR, "Couldn't find RTP provider '%s'\n", name);
+	return NULL;
+}
+
+struct rtp_provider_instance *
+rtp_provider_instance_alloc(void *ctx, const struct rtp_provider *provider, enum codec_type codec)
+{
+	struct rtp_provider_instance *pi;
+
+	pi = talloc_zero(ctx, struct rtp_provider_instance);
+	if (!pi)
+		return NULL;
+
+	pi->provider = provider;
+	pi->codec = codec;
+
+	return pi;
+}
+
+void rtp_provider_instance_free(struct rtp_provider_instance *pi)
+{
+	llist_del(&pi->list);
+	talloc_free(pi);
+}
+
+int rtp_provider_instance_gen_frame(struct rtp_provider_instance *pi, uint8_t *out, size_t out_size)
+{
+	OSMO_ASSERT(pi->provider);
+	return pi->provider->rtp_gen(pi, out, out_size);
+}

contrib/rtp-load-gen/rtp_provider.h

@@ -0,0 +1,57 @@
+#pragma once
+#include <stdint.h>
+#include <osmocom/core/linuxlist.h>
+#include <osmocom/core/utils.h>
+
+enum codec_type {
+	CODEC_ULAW,
+	CODEC_ALAW,
+	CODEC_GSM_FR,
+	CODEC_GSM_EFR,
+	CODEC_GSM_HR,
+	CODEC_AMR_4_75,
+	CODEC_AMR_5_15,
+	CODEC_AMR_5_90,
+	CODEC_AMR_6_70,
+	CODEC_AMR_7_40,
+	CODEC_AMR_7_95,
+	CODEC_AMR_10_2,
+	CODEC_AMR_12_2,
+	CODEC_AMR_SID,
+	_NUM_CODECS
+};
+
+extern const struct value_string codec_type_names[];
+
+struct rtp_provider_instance;
+
+struct rtp_provider {
+	/* global list of RTP providers */
+	struct llist_head list;
+	const char *name;
+
+	/* create/initialie a RTP provider with specified argument string */
+	int (*setup)(struct rtp_provider_instance *inst, const char *arg);
+
+	/* generate the next RTP packet; return length in octests or negative on error */
+	int (*rtp_gen)(struct rtp_provider_instance *inst, uint8_t *out, size_t out_size);
+};
+
+struct rtp_provider_instance {
+	/* entry in global list of RTP provider instances */
+	struct llist_head list;
+	/* pointer to provider of which we are an instance */
+	const struct rtp_provider *provider;
+	/* codec payload we are to generate */
+	enum codec_type codec;
+
+	/* private user data */
+	void *priv;
+};
+
+void rtp_provider_register(struct rtp_provider *prov);
+const struct rtp_provider *rtp_provider_find(const char *name);
+
+struct rtp_provider_instance *rtp_provider_instance_alloc(void *ctx, const struct rtp_provider *provider, enum codec_type codec);
+void rtp_provider_instance_free(struct rtp_provider_instance *pi);
+int rtp_provider_instance_gen_frame(struct rtp_provider_instance *pi, uint8_t *out, size_t out_size);

contrib/rtp-load-gen/rtp_provider_static.c

@@ -0,0 +1,108 @@
+
+#include <errno.h>
+#include <osmocom/codec/codec.h>
+#include <osmocom/core/utils.h>
+#include <osmocom/core/logging.h>
+
+#include "rtp_provider.h"
+#include "internal.h"
+
+static struct rtp_provider static_provider;
+
+static const uint8_t len4codec[_NUM_CODECS] = {
+    [CODEC_ULAW] = 160,
+    [CODEC_ALAW] = 160,
+    [CODEC_GSM_FR] = GSM_FR_BYTES,
+    [CODEC_GSM_EFR] = GSM_EFR_BYTES,
+    [CODEC_GSM_HR] = GSM_HR_BYTES,
+    [CODEC_AMR_4_75] = 12,
+    [CODEC_AMR_5_15] = 13,
+    [CODEC_AMR_5_90] = 15,
+    [CODEC_AMR_6_70] = 17,
+    [CODEC_AMR_7_40] = 19,
+    [CODEC_AMR_7_95] = 20,
+    [CODEC_AMR_10_2] = 26,
+    [CODEC_AMR_12_2] = 31,
+    [CODEC_AMR_SID] = 5,
+};
+
+/* generate a static / fixed RTP payload of matching codec/mode */
+static int rtp_gen_static(struct rtp_provider_instance *pi, uint8_t *out, size_t out_size)
+{
+	uint8_t len;
+
+	OSMO_ASSERT(pi->provider == &static_provider);
+
+	len = len4codec[pi->codec];
+	if (out_size < len) {
+		LOGP(DMAIN, LOGL_ERROR, "out_size %zu < %u\n", out_size, len);
+		return -EINVAL;
+	}
+
+	memset(out, 0, len);
+
+	switch (pi->codec) {
+	case CODEC_ULAW:
+	case CODEC_ALAW:
+		break;
+	case CODEC_GSM_FR:
+		out[0] = (out[0] & 0x0f) | 0xD0; /* mask in first four bit for FR */
+		break;
+	case CODEC_GSM_EFR:
+		out[0] = (out[0] & 0x0f) | 0xC0; /* mask in first four bit for EFR */
+		break;
+	case CODEC_GSM_HR:
+		break;
+	case CODEC_AMR_4_75:
+		out[0] = 0 << 4;
+		out[1] = 0 << 3;
+		break;
+	case CODEC_AMR_5_15:
+		out[0] = 1 << 4;
+		out[1] = 1 << 3;
+		break;
+	case CODEC_AMR_5_90:
+		out[0] = 2 << 4;
+		out[1] = 2 << 3;
+		break;
+	case CODEC_AMR_6_70:
+		out[0] = 3 << 4;
+		out[1] = 3 << 3;
+		break;
+	case CODEC_AMR_7_40:
+		out[0] = 4 << 4;
+		out[1] = 4 << 3;
+		break;
+	case CODEC_AMR_7_95:
+		out[0] = 5 << 4;
+		out[1] = 5 << 3;
+		break;
+	case CODEC_AMR_10_2:
+		out[0] = 6 << 4;
+		out[1] = 6 << 3;
+		break;
+	case CODEC_AMR_12_2:
+		out[0] = 7 << 4;
+		out[1] = 7 << 3;
+		break;
+	case CODEC_AMR_SID:
+		out[0] = 2 << 4; /* CMR: 5.90 */
+		out[0] = 8 << 3;
+		break;
+	default:
+		OSMO_ASSERT(0);
+	}
+
+	return len;
+}
+
+
+static struct rtp_provider static_provider = {
+	.name = "static",
+	.rtp_gen = &rtp_gen_static,
+};
+
+static void __attribute__((constructor)) rtp_provider_static_constr(void)
+{
+	rtp_provider_register(&static_provider);
+}

contrib/simcom2rtp/Makefile

@@ -0,0 +1,11 @@
+CFLAGS:= -O2 -g -Wall $(shell pkg-config --cflags libosmocore libosmotrau)
+LIBS:= $(shell pkg-config --libs libosmocore libosmotrau)
+
+all: osmo-simcom2rtp
+
+osmo-simcom2rtp: g711.o g711_table.o simcom2rtp.o
+	$(CC) $(LDFLAGS) -o $@ $^ $(LIBS)
+
+%.o: %.c
+	$(CC) $(CFLAGS) -o $@ -c $^
+

contrib/simcom2rtp/g711.c

@@ -0,0 +1,313 @@
+/*
+ * This source code is a product of Sun Microsystems, Inc. and is provided
+ * for unrestricted use.  Users may copy or modify this source code without
+ * charge.
+ *
+ * SUN SOURCE CODE IS PROVIDED AS IS WITH NO WARRANTIES OF ANY KIND INCLUDING
+ * THE WARRANTIES OF DESIGN, MERCHANTIBILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE, OR ARISING FROM A COURSE OF DEALING, USAGE OR TRADE PRACTICE.
+ *
+ * Sun source code is provided with no support and without any obligation on
+ * the part of Sun Microsystems, Inc. to assist in its use, correction,
+ * modification or enhancement.
+ *
+ * SUN MICROSYSTEMS, INC. SHALL HAVE NO LIABILITY WITH RESPECT TO THE
+ * INFRINGEMENT OF COPYRIGHTS, TRADE SECRETS OR ANY PATENTS BY THIS SOFTWARE
+ * OR ANY PART THEREOF.
+ *
+ * In no event will Sun Microsystems, Inc. be liable for any lost revenue
+ * or profits or other special, indirect and consequential damages, even if
+ * Sun has been advised of the possibility of such damages.
+ *
+ * Sun Microsystems, Inc.
+ * 2550 Garcia Avenue
+ * Mountain View, California  94043
+ */
+/*
+ * December 30, 1994:
+ * Functions linear2alaw, linear2ulaw have been updated to correctly
+ * convert unquantized 16 bit values.
+ * Tables for direct u- to A-law and A- to u-law conversions have been
+ * corrected.
+ * Borge Lindberg, Center for PersonKommunikation, Aalborg University.
+ * bli@cpk.auc.dk
+ *
+ */
+/*
+ * Downloaded from comp.speech site in Cambridge.
+ *
+ */
+
+#include "g711.h"
+
+/*
+ * g711.c
+ *
+ * u-law, A-law and linear PCM conversions.
+ * Source: http://www.speech.kth.se/cost250/refsys/latest/src/g711.c
+ */
+#define	SIGN_BIT	(0x80)		/* Sign bit for a A-law byte. */
+#define	QUANT_MASK	(0xf)		/* Quantization field mask. */
+#define	NSEGS		(8)		/* Number of A-law segments. */
+#define	SEG_SHIFT	(4)		/* Left shift for segment number. */
+#define	SEG_MASK	(0x70)		/* Segment field mask. */
+
+static short seg_aend[8] = {0x1F, 0x3F, 0x7F, 0xFF,
+			    0x1FF, 0x3FF, 0x7FF, 0xFFF};
+static short seg_uend[8] = {0x3F, 0x7F, 0xFF, 0x1FF,
+			    0x3FF, 0x7FF, 0xFFF, 0x1FFF};
+
+/* copy from CCITT G.711 specifications */
+unsigned char _u2a[128] = {			/* u- to A-law conversions */
+	1,	1,	2,	2,	3,	3,	4,	4,
+	5,	5,	6,	6,	7,	7,	8,	8,
+	9,	10,	11,	12,	13,	14,	15,	16,
+	17,	18,	19,	20,	21,	22,	23,	24,
+	25,	27,	29,	31,	33,	34,	35,	36,
+	37,	38,	39,	40,	41,	42,	43,	44,
+	46,	48,	49,	50,	51,	52,	53,	54,
+	55,	56,	57,	58,	59,	60,	61,	62,
+	64,	65,	66,	67,	68,	69,	70,	71,
+	72,	73,	74,	75,	76,	77,	78,	79,
+/* corrected:
+	81,	82,	83,	84,	85,	86,	87,	88, 
+   should be: */
+	80,	82,	83,	84,	85,	86,	87,	88,
+	89,	90,	91,	92,	93,	94,	95,	96,
+	97,	98,	99,	100,	101,	102,	103,	104,
+	105,	106,	107,	108,	109,	110,	111,	112,
+	113,	114,	115,	116,	117,	118,	119,	120,
+	121,	122,	123,	124,	125,	126,	127,	128};
+
+unsigned char _a2u[128] = {			/* A- to u-law conversions */
+	1,	3,	5,	7,	9,	11,	13,	15,
+	16,	17,	18,	19,	20,	21,	22,	23,
+	24,	25,	26,	27,	28,	29,	30,	31,
+	32,	32,	33,	33,	34,	34,	35,	35,
+	36,	37,	38,	39,	40,	41,	42,	43,
+	44,	45,	46,	47,	48,	48,	49,	49,
+	50,	51,	52,	53,	54,	55,	56,	57,
+	58,	59,	60,	61,	62,	63,	64,	64,
+	65,	66,	67,	68,	69,	70,	71,	72,
+/* corrected:
+	73,	74,	75,	76,	77,	78,	79,	79,
+   should be: */
+	73,	74,	75,	76,	77,	78,	79,	80,
+	80,	81,	82,	83,	84,	85,	86,	87,
+	88,	89,	90,	91,	92,	93,	94,	95,
+	96,	97,	98,	99,	100,	101,	102,	103,
+	104,	105,	106,	107,	108,	109,	110,	111,
+	112,	113,	114,	115,	116,	117,	118,	119,
+	120,	121,	122,	123,	124,	125,	126,	127};
+
+static short search(
+   short val,
+   short *table,
+   short size)
+{
+   short i;
+   
+   for (i = 0; i < size; i++) {
+      if (val <= *table++)
+	 return (i);
+   }
+   return (size);
+}
+
+/*
+ * linear2alaw() - Convert a 16-bit linear PCM value to 8-bit A-law
+ *
+ * linear2alaw() accepts an 16-bit integer and encodes it as A-law data.
+ *
+ *		Linear Input Code	Compressed Code
+ *	------------------------	---------------
+ *	0000000wxyza			000wxyz
+ *	0000001wxyza			001wxyz
+ *	000001wxyzab			010wxyz
+ *	00001wxyzabc			011wxyz
+ *	0001wxyzabcd			100wxyz
+ *	001wxyzabcde			101wxyz
+ *	01wxyzabcdef			110wxyz
+ *	1wxyzabcdefg			111wxyz
+ *
+ * For further information see John C. Bellamy's Digital Telephony, 1982,
+ * John Wiley & Sons, pps 98-111 and 472-476.
+ */
+unsigned char
+linear2alaw(short pcm_val)	/* 2's complement (16-bit range) */
+{
+   short	 mask;
+   short	 seg;
+   unsigned char aval;
+   
+   pcm_val = pcm_val >> 3;
+
+   if (pcm_val >= 0) {
+      mask = 0xD5;		/* sign (7th) bit = 1 */
+   } else {
+      mask = 0x55;		/* sign bit = 0 */
+      pcm_val = -pcm_val - 1;
+   }
+   
+   /* Convert the scaled magnitude to segment number. */
+   seg = search(pcm_val, seg_aend, 8);
+   
+   /* Combine the sign, segment, and quantization bits. */
+   
+   if (seg >= 8)		/* out of range, return maximum value. */
+      return (unsigned char) (0x7F ^ mask);
+   else {
+      aval = (unsigned char) seg << SEG_SHIFT;
+      if (seg < 2)
+	 aval |= (pcm_val >> 1) & QUANT_MASK;
+      else
+	 aval |= (pcm_val >> seg) & QUANT_MASK;
+      return (aval ^ mask);
+   }
+}
+
+/*
+ * alaw2linear() - Convert an A-law value to 16-bit linear PCM
+ *
+ */
+short
+alaw2linear(
+   unsigned char	a_val)
+{
+   short t;
+   short seg;
+   
+   a_val ^= 0x55;
+   
+   t = (a_val & QUANT_MASK) << 4;
+   seg = ((unsigned)a_val & SEG_MASK) >> SEG_SHIFT;
+   switch (seg) {
+   case 0:
+      t += 8;
+      break;
+   case 1:
+      t += 0x108;
+      break;
+   default:
+      t += 0x108;
+      t <<= seg - 1;
+   }
+   return ((a_val & SIGN_BIT) ? t : -t);
+}
+
+#define	BIAS		(0x84)		/* Bias for linear code. */
+#define CLIP            8159
+
+/*
+* linear2ulaw() - Convert a linear PCM value to u-law
+*
+* In order to simplify the encoding process, the original linear magnitude
+* is biased by adding 33 which shifts the encoding range from (0 - 8158) to
+* (33 - 8191). The result can be seen in the following encoding table:
+*
+*	Biased Linear Input Code	Compressed Code
+*	------------------------	---------------
+*	00000001wxyza			000wxyz
+*	0000001wxyzab			001wxyz
+*	000001wxyzabc			010wxyz
+*	00001wxyzabcd			011wxyz
+*	0001wxyzabcde			100wxyz
+*	001wxyzabcdef			101wxyz
+*	01wxyzabcdefg			110wxyz
+*	1wxyzabcdefgh			111wxyz
+*
+* Each biased linear code has a leading 1 which identifies the segment
+* number. The value of the segment number is equal to 7 minus the number
+* of leading 0's. The quantization interval is directly available as the
+* four bits wxyz.  * The trailing bits (a - h) are ignored.
+*
+* Ordinarily the complement of the resulting code word is used for
+* transmission, and so the code word is complemented before it is returned.
+*
+* For further information see John C. Bellamy's Digital Telephony, 1982,
+* John Wiley & Sons, pps 98-111 and 472-476.
+*/
+unsigned char
+linear2ulaw(
+   short pcm_val)	/* 2's complement (16-bit range) */
+{
+   short         mask;
+   short	 seg;
+   unsigned char uval;
+   
+   /* Get the sign and the magnitude of the value. */
+   pcm_val = pcm_val >> 2;
+   if (pcm_val < 0) {
+      pcm_val = -pcm_val;
+      mask = 0x7F;
+   } else {
+      mask = 0xFF;
+   }
+   if ( pcm_val > CLIP ) pcm_val = CLIP;		/* clip the magnitude */
+   pcm_val += (BIAS >> 2);
+   
+   /* Convert the scaled magnitude to segment number. */
+   seg = search(pcm_val, seg_uend, 8);
+   
+   /*
+   * Combine the sign, segment, quantization bits;
+   * and complement the code word.
+   */
+   if (seg >= 8)		/* out of range, return maximum value. */
+      return (unsigned char) (0x7F ^ mask);
+   else {
+      uval = (unsigned char) (seg << 4) | ((pcm_val >> (seg + 1)) & 0xF);
+      return (uval ^ mask);
+   }
+   
+}
+
+/*
+ * ulaw2linear() - Convert a u-law value to 16-bit linear PCM
+ *
+ * First, a biased linear code is derived from the code word. An unbiased
+ * output can then be obtained by subtracting 33 from the biased code.
+ *
+ * Note that this function expects to be passed the complement of the
+ * original code word. This is in keeping with ISDN conventions.
+ */
+short
+ulaw2linear(
+   unsigned char	u_val)
+{
+   short t;
+   
+   /* Complement to obtain normal u-law value. */
+   u_val = ~u_val;
+   
+   /*
+    * Extract and bias the quantization bits. Then
+    * shift up by the segment number and subtract out the bias.
+    */
+   t = ((u_val & QUANT_MASK) << 3) + BIAS;
+   t <<= ((unsigned)u_val & SEG_MASK) >> SEG_SHIFT;
+   
+   return ((u_val & SIGN_BIT) ? (BIAS - t) : (t - BIAS));
+}
+
+/* A-law to u-law conversion */
+unsigned char
+alaw2ulaw(
+   unsigned char	aval)
+{
+   aval &= 0xff;
+   return (unsigned char) ((aval & 0x80) ? (0xFF ^ _a2u[aval ^ 0xD5]) :
+	   (0x7F ^ _a2u[aval ^ 0x55]));
+}
+
+/* u-law to A-law conversion */
+unsigned char
+ulaw2alaw(
+   unsigned char	uval)
+{
+   uval &= 0xff;
+   return (unsigned char) ((uval & 0x80) ? (0xD5 ^ (_u2a[0xFF ^ uval] - 1)) :
+			   (0x55 ^ (_u2a[0x7F ^ uval] - 1)));
+}
+
+/* ---------- end of g711.c ----------------------------------------------------- */

contrib/simcom2rtp/g711.h

@@ -0,0 +1,27 @@
+/*
+ * g711.h
+ *
+ * u-law, A-law and linear PCM conversions.
+ * Source: http://www.speech.kth.se/cost250/refsys/latest/src/g711.h
+ */
+
+#ifndef _G711_H_
+#define _G711_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+
+unsigned char	linear2alaw(short pcm_val);
+short		    alaw2linear(unsigned char a_val);
+unsigned char	linear2ulaw(short pcm_val);
+short		    ulaw2linear(unsigned char u_val);
+unsigned char	alaw2ulaw(unsigned char aval);
+unsigned char	ulaw2alaw(unsigned char uval);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _G711_H_ */

contrib/simcom2rtp/g711_table.c

@@ -0,0 +1,102 @@
+#ifndef G711_TABLE_H
+#define G711_TABLE_H
+
+#include "g711.h"
+
+/* 16384 entries per table (16 bit) */
+unsigned char linear_to_alaw[65536];
+unsigned char linear_to_ulaw[65536];
+
+/* 16384 entries per table (8 bit) */
+unsigned short alaw_to_linear[256];
+unsigned short ulaw_to_linear[256];
+
+static void build_linear_to_xlaw_table(unsigned char *linear_to_xlaw,
+                                       unsigned char (*linear2xlaw)(short))
+{
+    int i;
+
+    for (i=0; i<65536;i++){
+        linear_to_xlaw[i] = linear2xlaw((short) i);
+    }
+}
+
+static void build_xlaw_to_linear_table(unsigned short *xlaw_to_linear,
+                                       short (*xlaw2linear)(unsigned char))
+{
+    int i;
+
+    for (i=0; i<256;i++){
+        xlaw_to_linear[i] = (unsigned short) xlaw2linear(i);
+    }
+}
+
+static void pcm16_to_xlaw(unsigned char *linear_to_xlaw, int src_length, const char *src_samples, char *dst_samples)
+{
+    int i;
+    const unsigned short *s_samples;
+
+    s_samples = (const unsigned short *)src_samples;
+
+    for (i=0; i < src_length / 2; i++)
+    {
+        dst_samples[i] = linear_to_xlaw[s_samples[i]];
+    }
+}
+
+static void xlaw_to_pcm16(unsigned short *xlaw_to_linear, int src_length, const char *src_samples, char *dst_samples)
+{
+    int i;
+    unsigned char *s_samples;
+    unsigned short *d_samples;
+
+    s_samples = (unsigned char *) src_samples;
+    d_samples = (unsigned short *)dst_samples;
+
+    for (i=0; i < src_length; i++)
+    {
+        d_samples[i] = xlaw_to_linear[s_samples[i]];
+    }
+}
+
+void pcm16_to_alaw(int src_length, const char *src_samples, char *dst_samples)
+{
+    pcm16_to_xlaw(linear_to_alaw, src_length, src_samples, dst_samples);
+}
+
+void pcm16_to_ulaw(int src_length, const char *src_samples, char *dst_samples)
+{
+    pcm16_to_xlaw(linear_to_ulaw, src_length, src_samples, dst_samples);
+}
+
+void alaw_to_pcm16(int src_length, const char *src_samples, char *dst_samples)
+{
+    xlaw_to_pcm16(alaw_to_linear, src_length, src_samples, dst_samples);
+}
+
+void ulaw_to_pcm16(int src_length, const char *src_samples, char *dst_samples)
+{
+    xlaw_to_pcm16(ulaw_to_linear, src_length, src_samples, dst_samples);
+}
+
+void pcm16_alaw_tableinit()
+{
+    build_linear_to_xlaw_table(linear_to_alaw, linear2alaw);
+}
+
+void pcm16_ulaw_tableinit()
+{
+    build_linear_to_xlaw_table(linear_to_ulaw, linear2ulaw);
+}
+
+void alaw_pcm16_tableinit()
+{
+    build_xlaw_to_linear_table(alaw_to_linear, alaw2linear);
+}
+
+void ulaw_pcm16_tableinit()
+{
+    build_xlaw_to_linear_table(ulaw_to_linear, ulaw2linear);
+}
+
+#endif // G711_TABLE_H

contrib/simcom2rtp/g711_table.h

@@ -0,0 +1,14 @@
+#ifndef G711_TABLE_H
+#define G711_TABLE_H
+
+void pcm16_to_alaw(int length, const char *src_samples, char *dst_samples);
+void pcm16_to_ulaw(int length, const char *src_samples, char *dst_samples);
+void alaw_to_pcm16(int length, const char *src_samples, char *dst_samples);
+void ulaw_to_pcm16(int length, const char *src_samples, char *dst_samples);
+
+void pcm16_alaw_tableinit();
+void pcm16_ulaw_tableinit();
+void alaw_pcm16_tableinit();
+void ulaw_pcm16_tableinit();
+
+#endif // G711_TABLE_H

contrib/simcom2rtp/simcom2rtp.c

@@ -0,0 +1,218 @@
+#include <stdio.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <errno.h>
+#include <limits.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+
+
+#include <osmocom/core/select.h>
+#include <osmocom/core/msgb.h>
+#include <osmocom/core/fsm.h>
+#include <osmocom/core/logging.h>
+#include <osmocom/core/application.h>
+#include <osmocom/core/serial.h>
+
+#include <osmocom/trau/osmo_ortp.h>
+
+#include "g711.h"
+
+#define RTP_PT_PCMU	0
+#define RTP_PT_PCMA	8
+
+struct modem_state {
+	struct osmo_fd data_fd;
+	struct osmo_rtp_socket *rtp;
+	/* queue of linear PCM audio in RTP -> modem direction */
+	struct llist_head rtp2modem;
+	/* message buffer used if samples insufficient for next RTP frame were received */
+	struct msgb *modem2rtp;
+};
+
+static void *g_tall_ctx;
+
+
+/* call-back on received RTP data */
+static void ortp_rx_cb(struct osmo_rtp_socket *rs, const uint8_t *payload,
+			unsigned int payload_len, uint16_t seq_nr, uint32_t timestamp, bool marker)
+{
+	/* we received a RTP frame */
+	struct modem_state *ms = rs->priv;
+	struct msgb *msg = msgb_alloc(payload_len*2, "RTP Rx");
+	unsigned int i;
+	int16_t *out;
+
+	OSMO_ASSERT(msg);
+
+	out = (int16_t *) msgb_put(msg, payload_len*2);
+
+	if (payload_len != 160) {
+		fprintf(stderr, "RTP payload length %d != 160, dropping\n", payload_len);
+		msgb_free(msg);
+		return;
+	}
+
+	/* convert from Alaw to linear PCM (160 -> 320 bytes) */
+	for (i = 0; i < payload_len; i++)
+		out[i] = alaw2linear(payload[i]);
+
+	/* append to the write queue */
+	msgb_enqueue(&ms->rtp2modem, msg);
+	ms->data_fd.when |= OSMO_FD_WRITE;
+}
+
+static void modem2rtp(struct modem_state *ms, const uint8_t *data, unsigned int len)
+{
+	const int16_t *data16 = (const int16_t *)data;
+	unsigned int samples = len / 2;
+	unsigned int offset = 0;
+	unsigned int i;
+
+	/* samples are always 16bit, we cannot read half a sample */
+	OSMO_ASSERT((len & 1) == 0);
+
+	/* first complete any pending incomplete RTP frame */
+	if (ms->modem2rtp) {
+		struct msgb *msg = ms->modem2rtp;
+		unsigned int missing_samples = 160 - msgb_length(msg);
+		for (i = 0; i < missing_samples; i++) {
+			if (i >= samples)
+				break;
+			msgb_put_u8(msg, linear2alaw(data16[i]));
+		}
+		offset = i;
+		if (msgb_length(msg) == 160) {
+			osmo_rtp_send_frame_ext(ms->rtp, msgb_data(msg), msgb_length(msg), 160, false);
+			msgb_free(msg);
+		}
+	}
+
+	/* then send as many RTP frames as we have samples */
+	for (offset = offset; offset + 160 <= samples; offset += 160) {
+		uint8_t buf[160];
+		for (i = 0; i < sizeof(buf); i++)
+			buf[i] = linear2alaw(data16[offset + i]);
+		osmo_rtp_send_frame_ext(ms->rtp, buf, sizeof(buf), 160, false);
+	}
+
+	/* store remainder in msgb */
+	if (offset < samples) {
+		struct msgb *msg = msgb_alloc_c(ms, 160, "modem2rtp");
+		OSMO_ASSERT(msg);
+		OSMO_ASSERT(len - offset < 160);
+		for (i = 0; i < len - offset; i++)
+			msgb_put_u8(msg, linear2alaw(data16[offset + i]));
+		ms->modem2rtp = msg;
+	}
+}
+
+
+/* call back on file descriptor events of the modem DATA ttyUSB device */
+static int modem_data_fd_cb(struct osmo_fd *ofd, unsigned int what)
+{
+	struct modem_state *ms = ofd->data;
+	int rc;
+
+	if (what & OSMO_FD_READ) {
+		/* SIM5360 USB AUDIO Application Note v1.01 states 1600 bytes every 100ms */
+		uint8_t rx_buf[1600];
+		rc = read(ofd->fd, rx_buf, sizeof(rx_buf));
+		OSMO_ASSERT(rc > 0);
+		modem2rtp(ms, rx_buf, rc);
+	}
+
+	if (what & OSMO_FD_WRITE) {
+		struct msgb *msg = msgb_dequeue(&ms->rtp2modem);
+		if (!msg)
+			ofd->when &= ~OSMO_FD_WRITE;
+		else {
+			/* SIM5300 USB AUDIO Application Note v1.01 states 640 bytes every 40ms;
+			 * we simply write every RTP frame individually (320 bytes every 20ms) */
+			rc = write(ofd->fd, msgb_data(msg), msgb_length(msg));
+			if (rc != msgb_length(msg))
+				fprintf(stderr, "Short write: %d < %u\n", rc, msgb_length(msg));
+			msgb_free(msg);
+		}
+	}
+
+	return 0;
+}
+
+
+static int modem_data_open(struct modem_state *ms, const char *basepath)
+{
+	char fname[PATH_MAX+1];
+	int fd;
+
+	/* the assumption is that the caller provides something like
+	 * "/dev/serial/by-path/pci-0000:00:14.0-usb-0:2:1" */
+	snprintf(fname, sizeof(fname), "%s.0-port0", basepath);
+
+	fd = osmo_serial_init(fname, 921600);
+	if (fd < 0) {
+		fprintf(stderr, "failed to open device '%s': %s\n", fname, strerror(errno));
+		return -1;
+	}
+	osmo_fd_setup(&ms->data_fd, fd, OSMO_FD_READ, modem_data_fd_cb, ms, 0);
+	osmo_fd_register(&ms->data_fd);
+
+	return 0;
+}
+
+static struct modem_state *modem_create(void *ctx)
+{
+	struct modem_state *ms = talloc_zero(ctx, struct modem_state);
+	int rc;
+
+	INIT_LLIST_HEAD(&ms->rtp2modem);
+
+	ms->rtp = osmo_rtp_socket_create(ms, 0);
+	OSMO_ASSERT(ms->rtp);
+	osmo_rtp_socket_set_pt(ms->rtp, RTP_PT_PCMA);
+	ms->rtp->priv = ms;
+	ms->rtp->rx_cb = ortp_rx_cb;
+
+	rc = osmo_rtp_socket_bind(ms->rtp, "0.0.0.0", 1111);
+	OSMO_ASSERT(rc == 0);
+
+	rc = osmo_rtp_socket_connect(ms->rtp, "127.0.0.1", 2222);
+	//rc = osmo_rtp_socket_autoconnect(ms->rtp);
+	OSMO_ASSERT(rc == 0);
+
+	osmo_rtp_set_source_desc(ms->rtp, "cname", "simcom2rtp", NULL, NULL, NULL,
+				 "osmo-simcom2rtp", NULL);
+
+	return ms;
+}
+
+
+
+
+int main(int argc, char **argv)
+{
+
+	talloc_enable_null_tracking();
+	g_tall_ctx = talloc_named_const(NULL, 1, "simcom2rtp");
+
+	msgb_talloc_ctx_init(g_tall_ctx, 0);
+	osmo_init_logging2(g_tall_ctx, NULL);
+	osmo_fsm_log_timeouts(true);
+	osmo_fsm_log_addr(true);
+	//osmo_stats_init(g_tall_ctx);
+	osmo_rtp_init(g_tall_ctx);
+
+	struct modem_state *ms = modem_create(g_tall_ctx);
+	int rc;
+
+	OSMO_ASSERT(ms);
+	rc = modem_data_open(ms, "/dev/serial/by-path/pci-0000:00:14.0-usb-0:2:1");
+	OSMO_ASSERT(rc == 0);
+
+	while (1) {
+		osmo_select_main(0);
+	}
+
+}