tiptorrent-client/src/main.c

/*
 * Copyright (C) 2021 Soleta Networks <info@soleta.eu>
 *
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License as published by the
 * Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 */

#include <stdlib.h>
#include <stdio.h>
#include <ev.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <errno.h>
#include <string.h>
#include <sys/fcntl.h>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <syslog.h>
#include <stddef.h>
#include <stdbool.h>
#include <sys/time.h>
#include <limits.h>
#include <syslog.h>
#include <netinet/tcp.h>

#define TIP_TORRENT_PORT	9999

/* number of chunks for files. */
#define MAX_CHUNKS		4

#define container_of(ptr, type, member) ({			\
	typeof( ((type *)0)->member ) *__mptr = (ptr);		\
	(type *)( (char *)__mptr - offsetof(type,member) );})

static const char *filename;
static const char *addr;
struct ev_loop *tip_main_loop;

enum {
	TIP_CLIENT_RECEIVING_HEADER,
	TIP_CLIENT_RECEIVING_PAYLOAD,
	TIP_CLIENT_NOTIFY_REDIRECT,
	TIP_CLIENT_DONE,
};

struct tip_client {
	ev_io io;
	struct sockaddr_in addr;
	char buf[10240000];
	uint32_t buf_len;
	uint64_t data_len;
	uint64_t content_len;
	int state;
	int fd;
	bool error;
	bool redirected;
	const char *payload;
};

static struct tip_client _cli = {
	.fd = -1,
};

struct {
	uint32_t	direct_from_server;
	uint32_t	redirects;
} tip_client_stats;

static int tip_client_recv(struct tip_client *cli, int events)
{
	struct ev_io *io = &cli->io;
	int ret;

	ret = recv(io->fd, cli->buf + cli->buf_len,
		   sizeof(cli->buf) - cli->buf_len, 0);
	if (ret < 0) {
		syslog(LOG_ERR, "error reading from server %s:%hu (%s)\n",
		       inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port),
		       strerror(errno));
	}

	return ret;
}

static void tip_client_close(struct tip_client *cli)
{
	ev_io_stop(tip_main_loop, &cli->io);
	shutdown(cli->io.fd, SHUT_RDWR);
	close(cli->io.fd);
	cli->buf_len = 0;
}

static void tip_client_error(struct tip_client *cli)
{
	cli->error = true;
	tip_client_close(cli);
}

static int tip_client_connect(const char *addr);

static int tip_client_state_recv_hdr(struct tip_client *cli)
{
	char *ptr, *trailer, *payload;
	char redirect_addr[32];
	uint32_t payload_len;
	uint32_t header_len;
	int ret;

	ptr = strstr(cli->buf, "\r\n\r\n");
	if (!ptr)
		return 0;

	if (!strncmp(cli->buf, "HTTP/1.1 404 Not Found", strlen("HTTP/1.1 404 Not Found"))) {
		syslog(LOG_ERR, "server says file `%s' not found\n", filename);
		return -1;
	}
	if (!strncmp(cli->buf, "HTTP/1.1 301 Moves Permanently", strlen("HTTP/1.1 301 Moves Permanently"))) {
		ptr = strstr(cli->buf, "Location:");
		if (!ptr)
			return -1;

		ret = sscanf(ptr, "Location: http://%31s[^\r\n]",
			     redirect_addr);
		if (ret != 1)
			return -1;

		ptr = strchr(redirect_addr, ':');
		if (!ptr)
			return -1;

		ptr[0] = '\0';

		syslog(LOG_INFO, "Redirected to %s to fetch file %s\n",
		       redirect_addr, filename);

		cli->redirected = true;
		tip_client_close(cli);
		tip_client_connect(redirect_addr);
		cli->state = TIP_CLIENT_RECEIVING_HEADER;

		return 0;
	}

	trailer = ptr + 4;

	ptr = strstr(cli->buf, "Content-Length: ");
	if (!ptr)
		return -1;

	if (sscanf(ptr, "Content-Length: %lu[^\r\n]", &cli->content_len) != 1)
		return -1;
	if (cli->content_len < 0)
		return -1;

	if (cli->content_len == 0) {
		cli->buf_len = 0;
		return 1;
	}

	if (cli->redirected)
		tip_client_stats.redirects++;
	else
		tip_client_stats.direct_from_server++;

	cli->fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0600);
	if (cli->fd < 0)
		return ret;

	header_len = trailer - cli->buf;
	payload = cli->buf + header_len;
	payload_len = cli->buf_len - header_len;
	cli->data_len += cli->buf_len;
	cli->buf_len = 0;

	if (payload_len > 0) {
		ret = write(cli->fd, payload, payload_len);
		if (ret < 0)
			return ret;
	}

	if (payload_len >= cli->content_len)
		return 0;

	return 1;
}

static int tip_client_state_recv_payload(struct tip_client *cli)
{
	int ret;

	cli->data_len += cli->buf_len;

	ret = write(cli->fd, cli->buf, cli->buf_len);
	if (ret < 0)
		return ret;

	cli->buf_len = 0;

	if (cli->data_len >= cli->content_len) {
		if (cli->redirected) {
			tip_client_close(cli);
			tip_client_connect(addr);
			cli->state = TIP_CLIENT_NOTIFY_REDIRECT;

			return 1;
		}
		cli->state = TIP_CLIENT_DONE;
		return 0;
	}

	return 1;
}

static int tip_client_state_notify_redirect(struct tip_client *cli)
{
	char *ptr;

	ptr = strstr(cli->buf, "\r\n\r\n");
	if (!ptr)
		return 0;

	if (strncmp(cli->buf, "HTTP/1.1 200 OK", strlen("HTTP/1.1 200 OK")))
		return -1;

	ptr = strstr(cli->buf, "Content-Length: ");
	if (!ptr)
		return -1;

	if (sscanf(ptr, "Content-Length: %lu[^\r\n]", &cli->content_len) != 1)
		return -1;
	if (cli->content_len < 0)
		return -1;

	if (cli->content_len != 0)
		return -1;

	cli->state = TIP_CLIENT_DONE;

	return 0;
}

static void tip_client_read_cb(struct ev_loop *loop, struct ev_io *io, int events)
{
	struct tip_client *cli;
	int ret;

	cli = container_of(io, struct tip_client, io);

	ret = tip_client_recv(cli, events);
	if (ret < 0)
		goto error;

	cli->buf_len += ret;

	switch (cli->state) {
	case TIP_CLIENT_RECEIVING_HEADER:
		ret = tip_client_state_recv_hdr(cli);
		if (ret < 0)
			goto error;
		if (!ret)
			return;

		cli->state = TIP_CLIENT_RECEIVING_PAYLOAD;
		/* Fall through. */
	case TIP_CLIENT_RECEIVING_PAYLOAD:
		ret = tip_client_state_recv_payload(cli);
		if (ret < 0)
			goto error;
		if (ret == 0)
			goto close;
		break;
	case TIP_CLIENT_NOTIFY_REDIRECT:
		ret = tip_client_state_notify_redirect(cli);
		if (ret < 0)
			goto error;
		if (ret == 0)
			goto close;
		break;
	}
	return;
error:
	tip_client_error(cli);
	return;
close:
	tip_client_close(cli);
	return;
}

static void tip_client_connect_cb(struct ev_loop *loop, struct ev_io *io, int events)
{
	struct tip_client *cli;
	char buf[PATH_MAX + 1];
	int ret, len;

	cli = container_of(io, struct tip_client, io);

	if (events & EV_ERROR)
		return;

	len = sizeof(cli->addr);
	ret = connect(cli->io.fd, (struct sockaddr *)&cli->addr, len);
	if (ret < 0) {
		syslog(LOG_ERR, "failed to connect to server to fetch %s", filename);
		tip_client_error(cli);
		return;
	}

	if (cli->state == TIP_CLIENT_NOTIFY_REDIRECT)
		snprintf(buf, sizeof(buf), "POST /%s HTTP/1.1\r\n\r\n", filename);
	else
		snprintf(buf, sizeof(buf), "GET /%s HTTP/1.1\r\n\r\n", filename);

	ret = send(cli->io.fd, buf, strlen(buf), 0);
	if (ret < 0) {
		tip_client_error(cli);
		return;
	}

	ev_io_stop(tip_main_loop, &cli->io);
	ev_io_init(&cli->io, tip_client_read_cb, cli->io.fd, EV_READ);
	ev_io_start(tip_main_loop, &cli->io);
}

#define TIP_TCP_KEEPALIVE_IDLE	60
#define TIP_TCP_KEEPALIVE_INTL	30
#define TIP_TCP_KEEPALIVE_CNT	4

static int tip_client_connect(const char *addr)
{
	int intl = TIP_TCP_KEEPALIVE_INTL, cnt = TIP_TCP_KEEPALIVE_CNT;
	int on = 1, idle = TIP_TCP_KEEPALIVE_IDLE;
	struct tip_client *cli = &_cli;
	int remote_fd;
	int flags;
	int len;
	int ret;

	remote_fd = socket(AF_INET, SOCK_STREAM, 0);
	if (remote_fd < 0)
		return -1;

	setsockopt(remote_fd, SOL_SOCKET, SO_KEEPALIVE, &on, sizeof(int));
	setsockopt(remote_fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(int));
	setsockopt(remote_fd, IPPROTO_TCP, TCP_KEEPINTVL, &intl, sizeof(int));
	setsockopt(remote_fd, IPPROTO_TCP, TCP_KEEPCNT, &cnt, sizeof(int));

	flags = fcntl(remote_fd, F_GETFL);
	flags |= O_NONBLOCK;
	ret = fcntl(remote_fd, F_SETFL, flags);
	if (ret < 0)
		return ret;

	cli->addr.sin_family = AF_INET;
	cli->addr.sin_addr.s_addr = inet_addr(addr);
	cli->addr.sin_port = htons(TIP_TORRENT_PORT);

	len = sizeof(cli->addr);
	ret = connect(remote_fd, (struct sockaddr *)&cli->addr, len);
	if (ret < 0 && errno != EINPROGRESS) {
		syslog(LOG_ERR, "failed to connect to server to fetch %s", filename);
		tip_client_error(cli);
		return ret;
	}

	ev_io_init(&cli->io, tip_client_connect_cb, remote_fd, EV_WRITE);
	ev_io_start(tip_main_loop, &cli->io);

	syslog(LOG_INFO, "connecting to %s to fetch file %s\n", addr, filename);

	return 0;
}

static uint32_t select_file_chunk(bool *file_chunk)
{
	struct timeval tv;
	uint32_t k;
	int i;

	gettimeofday(&tv, NULL);
	srand(tv.tv_usec);

	k = rand() % MAX_CHUNKS;
	for (i = 0; i < MAX_CHUNKS; i++) {
		if (!file_chunk[k])
			break;

		k++;
		if (k == MAX_CHUNKS)
			k = 0;
	}

	return k;
}

static char _filename[PATH_MAX + 1];

int main(int argc, char *argv[])
{
	struct timeval tv_start, tv_stop, tv;
	bool file_chunk[MAX_CHUNKS] = {};
	int i, k;

	if (argc != 3) {
		printf("%s [ip] [file]\n", argv[0]);
		return EXIT_FAILURE;
	}
	addr = argv[1];

	openlog("tiptorrent-client", LOG_PID, LOG_DAEMON);

	signal(SIGPIPE, SIG_IGN);

	tip_main_loop = ev_default_loop(0);

	gettimeofday(&tv_start, NULL);

	for (i = 0; i < MAX_CHUNKS; i++) {
		memset(&_cli, 0, sizeof(_cli));

		k = select_file_chunk(file_chunk);
		snprintf(_filename, sizeof(_filename), "%s.%u", argv[2], k);
		filename = _filename;

		syslog(LOG_INFO, "Requesting file %s to server\n", filename);

		tip_client_connect(argv[1]);
		_cli.state = TIP_CLIENT_RECEIVING_HEADER;

		while (_cli.state != TIP_CLIENT_DONE && !_cli.error)
			ev_loop(tip_main_loop, 0);

		file_chunk[k] = true;
	}

	if (_cli.state == TIP_CLIENT_DONE) {
		gettimeofday(&tv_stop, NULL);
		timersub(&tv_stop, &tv_start, &tv);
		printf("Done in %lu.%06lu seconds (%lu Mbytes/second). "
		       "Direct from server: %u Redirected: %u\n",
		       tv.tv_sec, tv.tv_usec,
		       _cli.data_len / 1024000 / tv.tv_sec,
		       tip_client_stats.direct_from_server,
		       tip_client_stats.redirects);
		return EXIT_SUCCESS;
	}

	printf("Failure, see syslog for details.\n");
	return EXIT_FAILURE;
}