summaryrefslogtreecommitdiffstats
path: root/tests/io/pq_rtp_test.c
blob: 2c1bd41e6aca255bfa52ee01df9d43a254874680 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
/*
 * This file is part of GAPK (GSM Audio Pocket Knife).
 *
 * (C) 2017 by Vadim Yanitskiy <axilirator@gmail.com>
 *
 * GAPK 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 3 of the License, or
 * (at your option) any later version.
 *
 * GAPK 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 GAPK.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <time.h>
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <talloc.h>
#include <assert.h>
#include <unistd.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <osmocom/core/socket.h>

#include <osmocom/gapk/procqueue.h>
#include <osmocom/gapk/common.h>

/**
 * This test is intended to check the RTP source / sink operability.
 * To do this, two processing queues are being allocated:
 *
 *     "generator": source/random -> sink/rtp
 *     "checker": source/rtp -> sink/checker
 *
 * The first one generates some amount of random bytes (payload),
 * and stores them inside a buffer that is shared between both
 * queues.
 *
 * After generation, a payload is being sent from the first
 * queue via an RTP sink, and then being received by the second
 * via an RTP source.
 *
 * As both queues do use a shared buffer, the last item of the
 * second queue (named 'sink/checker') is able to compare a
 * received payload with expected.
 */

static void talloc_ctx_walk_cb(const void *chunk, int depth,
	int max_depth, int is_ref, void *data)
{
	const char *chunk_name = talloc_get_name(chunk);
	int spaces_cnt;

	/* Hierarchical spacing */
	for (spaces_cnt = 0; spaces_cnt < depth; spaces_cnt++)
		printf("  ");

	/* Chunk info */
	printf("chunk %s: depth=%d\n", chunk_name, depth);
}

#define RTP_TEST_BUF_LEN 128

struct rtp_test_state {
	unsigned int payload_len;
	unsigned int rtp_port;
	int rtp_src_fd;
	int rtp_dst_fd;

	uint8_t data[RTP_TEST_BUF_LEN];
	uint8_t *ptr;
};

static int src_rand_proc(void *data, uint8_t *out, const uint8_t *in,
	unsigned int in_len)
{
	struct rtp_test_state *state = (struct rtp_test_state *) data;
	unsigned int i;

	/* Generate a random payload */
	for (i = 0; i < state->payload_len; i++) {
		uint8_t byte = rand() % 0xff;
		*(state->ptr + i) = byte;
		out[i] = byte;
	}

	return state->payload_len;
}

static void src_rand_exit(void *data)
{
	struct rtp_test_state *state = (struct rtp_test_state *) data;

	if (state->rtp_src_fd >= 0) {
		close(state->rtp_src_fd);
		state->rtp_src_fd = -1;
	}
}

static int sink_chk_proc(void *data, uint8_t *out, const uint8_t *in,
	unsigned int in_len)
{
	struct rtp_test_state *state = (struct rtp_test_state *) data;
	unsigned int i;

	/* Make sure we have all bytes transferred */
	if (in_len != state->payload_len) {
		printf("Data length mismatch!\n");
		return -EINVAL;
	}

	for (i = 0; i < in_len; i++) {
		if (in[i] != *(state->ptr + i)) {
			printf("Data mismatch!\n");
			return -EINVAL;
		}
	}

	return in_len;
}

static void sink_chk_exit(void *data)
{
	struct rtp_test_state *state = (struct rtp_test_state *) data;

	if (state->rtp_dst_fd >= 0) {
		close(state->rtp_dst_fd);
		state->rtp_dst_fd = -1;
	}
}

/* Allocates: source/random -> sink/rtp */
static int init_gen_queue(struct osmo_gapk_pq *pq,
	struct rtp_test_state *state, unsigned int payload_len)
{
	int rc;

	/* Allocate memory for the 'source/random' */
	struct osmo_gapk_pq_item *src_rand = osmo_gapk_pq_add_item(pq);
	if (!src_rand)
		return -ENOMEM;

	/* Fill in meta information */
	src_rand->type = OSMO_GAPK_ITEM_TYPE_SOURCE;
	src_rand->cat_name = OSMO_GAPK_CAT_NAME_SOURCE;
	src_rand->sub_name = "random";

	/* Set I/O buffer lengths */
	state->payload_len = payload_len;
	src_rand->len_out = payload_len;

	/* Set proc / exit callbacks and state */
	src_rand->proc = &src_rand_proc;
	src_rand->exit = &src_rand_exit;
	src_rand->state = state;


	/* Init connection socket */
	state->rtp_dst_fd = osmo_sock_init(AF_UNSPEC, SOCK_DGRAM,
		IPPROTO_UDP, "127.0.0.1", state->rtp_port, OSMO_SOCK_F_CONNECT);
	if (state->rtp_dst_fd < 0) {
		printf("Could not init connection socket\n");
		return -EINVAL;
	}

	/* Init an RTP sink */
	rc = osmo_gapk_pq_queue_rtp_output(pq, state->rtp_dst_fd, payload_len);
	if (rc) {
		printf("Could not init an RTP sink\n");
		return rc;
	}

	/* Check and prepare */
	rc = osmo_gapk_pq_check(pq, 1);
	if (rc) {
		printf("Queue check failed\n");
		return rc;
	}

	rc = osmo_gapk_pq_prepare(pq);
	if (rc) {
		printf("Queue preparation failed\n");
		return rc;
	}

	return 0;
}

/* Allocates: source/rtp -> sink/checker */
static int init_chk_queue(struct osmo_gapk_pq *pq,
	struct rtp_test_state *state, unsigned int payload_len)
{
	int rc;

	/* Init listening socket */
	state->rtp_src_fd = osmo_sock_init(AF_UNSPEC, SOCK_DGRAM,
		IPPROTO_UDP, "127.0.0.1", 0, OSMO_SOCK_F_BIND);
	if (state->rtp_src_fd < 0) {
		printf("Could not init listening socket\n");
		return -EINVAL;
	}

	/* Init an RTP source on any available port */
	rc = osmo_gapk_pq_queue_rtp_input(pq, state->rtp_src_fd, payload_len);
	if (rc) {
		printf("Could not init an RTP sink\n");
		return rc;
	}

	/* Determine on which port are we listening */
	struct sockaddr_in adr_inet;
	socklen_t len_inet;

	len_inet = sizeof(adr_inet);
	rc = getsockname(state->rtp_src_fd,
		(struct sockaddr *) &adr_inet, &len_inet);
	if (rc)
		return -EINVAL;

	/* Save assigned port to shared state */
	state->rtp_port = (unsigned int) ntohs(adr_inet.sin_port);


	/* Allocate memory for the 'sink/checker' */
	struct osmo_gapk_pq_item *sink_chk = osmo_gapk_pq_add_item(pq);
	if (!sink_chk)
		return -ENOMEM;

	/* Fill in meta information */
	sink_chk->type = OSMO_GAPK_ITEM_TYPE_SINK;
	sink_chk->cat_name = OSMO_GAPK_CAT_NAME_SINK;
	sink_chk->sub_name = "checker";

	/* Set I/O buffer lengths */
	sink_chk->len_in = payload_len;

	/* Set proc / exit callbacks and state */
	sink_chk->proc = &sink_chk_proc;
	sink_chk->exit = &sink_chk_exit;
	sink_chk->state = state;

	/* Check and prepare */
	rc = osmo_gapk_pq_check(pq, 1);
	if (rc) {
		printf("Queue check failed\n");
		return rc;
	}

	rc = osmo_gapk_pq_prepare(pq);
	if (rc) {
		printf("Queue preparation failed\n");
		return rc;
	}

	return 0;
}

static int rtp_test(struct rtp_test_state *state, unsigned int payload_len)
{
	struct osmo_gapk_pq *q_gen, *q_chk;
	unsigned int i, chunks;
	int rc;

	/* Allocate two queues */
	q_gen = osmo_gapk_pq_create("generator");
	q_chk = osmo_gapk_pq_create("checker");

	/* Make sure both queues are allocated */
	if (!q_gen || !q_chk) {
		rc = -ENOMEM;
		goto exit;
	}

	/* Init both queues: generator and checker */
	rc = init_chk_queue(q_chk, state, payload_len);
	if (rc)
		goto exit;

	rc = init_gen_queue(q_gen, state, payload_len);
	if (rc)
		goto exit;

	/* Calculate how much chunks do we have */
	chunks = RTP_TEST_BUF_LEN / payload_len;

	/* Execute both queues */
	for (i = 0; i < chunks; i++) {
		/* Move data pointer */
		state->ptr = state->data + i * payload_len;

		/* Generate and send a payload */
		rc = osmo_gapk_pq_execute(q_gen);
		if (rc) {
			printf("Queue '%s' execution aborted on chunk %u/%u\n",
				q_gen->name, i + 1, chunks);
			goto exit;
		}

		/* TODO: prevent test hang if nothing was being sent */

		/* Receive and check a payload */
		rc = osmo_gapk_pq_execute(q_chk);
		if (rc) {
			printf("Queue '%s' execution aborted on chunk %u/%u\n",
				q_gen->name, i + 1, chunks);
			goto exit;
		}
	}

	printf("Payload len=%u check ok\n", payload_len);

exit:
	/* Deallocate both queues and data */
	osmo_gapk_pq_destroy(q_gen);
	osmo_gapk_pq_destroy(q_chk);

	return rc;
}

int main(int argc, char **argv)
{
	struct rtp_test_state state;
	unsigned int len;

	/* Enable tracking the use of NULL memory contexts */
	talloc_enable_null_tracking();

	/* Init pseudo-random generator */
	srand(time(NULL));

	/* Perform testing with different payload size values */
	for (len = 1; len <= RTP_TEST_BUF_LEN; len *= 2)
		assert(rtp_test(&state, len) == 0);
	printf("\n");

	/* Memory leak detection test */
	talloc_report_depth_cb(NULL, 0, 10, &talloc_ctx_walk_cb, NULL);

	/* Make both Valgrind and LeakSanitizer happy */
	talloc_disable_null_tracking();

	return 0;
}