httpclient: Don't store finished transfers internally

[pakfire.git] / src / libpakfire / httpclient.c

/*#############################################################################
#                                                                             #
# Pakfire - The IPFire package management system                              #
# Copyright (C) 2021 Pakfire development team                                 #
#                                                                             #
# 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 3 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 <errno.h>
#include <stdlib.h>
#include <sys/queue.h>

#include <curl/curl.h>

#include <pakfire/httpclient.h>
#include <pakfire/logging.h>
#include <pakfire/progress.h>
#include <pakfire/xfer.h>

// The number of concurrent downloads
#define MAX_PARALLEL			4

struct pakfire_xfer_element {
	TAILQ_ENTRY(pakfire_xfer_element) nodes;

	struct pakfire_xfer* xfer;
};

struct pakfire_httpclient {
	struct pakfire_ctx* ctx;
	int nrefs;

	unsigned int parallel;

	// cURL share handle
	CURLSH* share;

	// cURL multi handle
	CURLM* curl;

	// Transfers
	TAILQ_HEAD(xfers_queued, pakfire_xfer_element) xfers_queued;
	TAILQ_HEAD(xfers_running, pakfire_xfer_element) xfers_running;
};

static int pakfire_httpclient_setup_curl(struct pakfire_httpclient* client) {
	int r;

	// Initialize cURL
	r = curl_global_init(CURL_GLOBAL_DEFAULT);
	if (r) {
		CTX_ERROR(client->ctx, "Could not initialize cURL: %d\n", r);
		return r;
	}

	// Create a new share handle
	client->share = curl_share_init();
	if (!client->share) {
		CTX_ERROR(client->ctx, "Could not setup cURL share handle\n");
		return 1;
	}

	// Create a new multi handle
	client->curl = curl_multi_init();
	if (!client->curl) {
		CTX_ERROR(client->ctx, "Could not create cURL multi handle\n");
		return 1;
	}

	// Share connections between handles
	r = curl_share_setopt(client->share, CURLSHOPT_SHARE, CURL_LOCK_DATA_CONNECT);
	if (r) {
		CTX_ERROR(client->ctx, "Could not configure the share handle: %s\n",
			curl_share_strerror(r));
		return r;
	}

	return 0;
}

static struct pakfire_xfer_element* pakfire_httpclient_xfer_create(struct pakfire_xfer* xfer) {
	struct pakfire_xfer_element* x = NULL;

	// Allocate a new element
	x = calloc(1, sizeof(*x));
	if (!x)
		return NULL;

	// Store a reference to the xfer
	x->xfer = pakfire_xfer_ref(xfer);

	return x;
}

static struct pakfire_xfer_element* pakfire_httpclient_xfer_find_running(
		struct pakfire_httpclient* client, struct pakfire_xfer* xfer) {
	struct pakfire_xfer_element* x = NULL;

	TAILQ_FOREACH(x, &client->xfers_running, nodes) {
		if (x->xfer == xfer)
			return x;
	}

	return NULL;
}

static void pakfire_httpclient_xfer_free(struct pakfire_xfer_element* x) {
	if (x->xfer)
		pakfire_xfer_unref(x->xfer);

	free(x);
}

static void pakfire_httpclient_free(struct pakfire_httpclient* client) {
	struct pakfire_xfer_element* x = NULL;

	// Free any queued transfers
	while (!TAILQ_EMPTY(&client->xfers_queued)) {
		x = TAILQ_LAST(&client->xfers_queued, xfers_queued);
		TAILQ_REMOVE(&client->xfers_queued, x, nodes);

		pakfire_httpclient_xfer_free(x);
	}

	// Free any running transfers
	while (!TAILQ_EMPTY(&client->xfers_running)) {
		x = TAILQ_LAST(&client->xfers_running, xfers_running);
		TAILQ_REMOVE(&client->xfers_running, x, nodes);

		pakfire_httpclient_xfer_free(x);
	}

	if (client->share)
		curl_share_cleanup(client->share);
	if (client->curl)
		curl_multi_cleanup(client->curl);
	if (client->ctx)
		pakfire_ctx_unref(client->ctx);
	free(client);
}

int pakfire_httpclient_create(struct pakfire_httpclient** client, struct pakfire_ctx* ctx) {
	struct pakfire_httpclient* c = NULL;
	int r;

	// Fail if the context is flagged as offline
	if (pakfire_ctx_has_flag(ctx, PAKFIRE_CTX_OFFLINE)) {
		CTX_ERROR(ctx, "Cannot initialize downloader in offline mode\n");
		return -EPERM;
	}

	// Allocate a new object
	c = calloc(1, sizeof(*c));
	if (!c)
		return -ENOMEM;

	// Store reference to the context
	c->ctx = pakfire_ctx_ref(ctx);

	// Initialize reference counting
	c->nrefs = 1;

	// Set parallelism
	c->parallel = MAX_PARALLEL;

	// Init transfer queues
	TAILQ_INIT(&c->xfers_queued);
	TAILQ_INIT(&c->xfers_running);

	// Setup cURL
	r = pakfire_httpclient_setup_curl(c);
	if (r)
		goto ERROR;

	// Success
	*client = c;

	return 0;

ERROR:
	pakfire_httpclient_free(c);

	return r;
}

struct pakfire_httpclient* pakfire_httpclient_ref(struct pakfire_httpclient* client) {
	++client->nrefs;

	return client;
}

struct pakfire_httpclient* pakfire_httpclient_unref(struct pakfire_httpclient* client) {
	if (--client->nrefs > 0)
		return client;

	pakfire_httpclient_free(client);
	return NULL;
}

CURLSH* pakfire_httpclient_share(struct pakfire_httpclient* client) {
	return client->share;
}

int pakfire_httpclient_create_xfer(struct pakfire_xfer** xfer,
		struct pakfire_httpclient* client, const char* url) {
	return pakfire_xfer_create(xfer, client->ctx, client, url);
}

int pakfire_httpclient_enqueue_xfer(struct pakfire_httpclient* client,
		struct pakfire_xfer* xfer) {
	struct pakfire_xfer_element* x = NULL;

	// Create a new queueable object
	x = pakfire_httpclient_xfer_create(xfer);
	if (!x)
		return -errno;

	// Push this transfer onto the queue
	TAILQ_INSERT_HEAD(&client->xfers_queued, x, nodes);

	return 0;
}

static size_t pakfire_httpclient_total_downloadsize(struct pakfire_httpclient* client) {
	struct pakfire_xfer_element* x = NULL;
	size_t size;

	size_t total_size = 0;

	TAILQ_FOREACH(x, &client->xfers_queued, nodes) {
		size = pakfire_xfer_get_size(x->xfer);

		// Return zero if the size isn't known
		if (!size)
			return 0;

		total_size += size;
	}

	return total_size;
}

static unsigned int pakfire_httpclient_total_queued_xfers(struct pakfire_httpclient* client) {
	struct pakfire_xfer_element* x = NULL;
	unsigned int counter = 0;

	TAILQ_FOREACH(x, &client->xfers_queued, nodes)
		counter++;

	return counter;
}

static int pakfire_httpclient_start_transfers(struct pakfire_httpclient* client,
		struct pakfire_progress* progress, unsigned int* running_transfers) {
	struct pakfire_xfer_element* x = NULL;
	int r;

	// Keep running until we have reached our ceiling
	while (*running_transfers < client->parallel) {
		// We are done if there are no more transfers in the queue
		if (TAILQ_EMPTY(&client->xfers_queued))
			break;

		// Fetch the next transfer
		x = TAILQ_LAST(&client->xfers_queued, xfers_queued);
		TAILQ_REMOVE(&client->xfers_queued, x, nodes);

		// Prepare the xfer
		r = pakfire_xfer_prepare(x->xfer, progress, 0);
		if (r)
			goto ERROR;

		// Add the handle to cURL
		r = curl_multi_add_handle(client->curl, pakfire_xfer_handle(x->xfer));
		if (r) {
			CTX_ERROR(client->ctx, "Adding handle failed: %s\n", curl_multi_strerror(r));
			goto ERROR;
		}

		TAILQ_INSERT_TAIL(&client->xfers_running, x, nodes);
		(*running_transfers)++;
	}

	return 0;

ERROR:
	pakfire_httpclient_xfer_free(x);

	return r;
}

int pakfire_httpclient_run(struct pakfire_httpclient* client, const char* title) {
	struct pakfire_progress* progress = NULL;
	struct pakfire_xfer* xfer = NULL;
	unsigned int running_xfers = 0;
	int progress_flags =
		PAKFIRE_PROGRESS_SHOW_PERCENTAGE |
		PAKFIRE_PROGRESS_SHOW_ETA;
	int r = 0;

	struct pakfire_xfer_element* x = NULL;

	CURLMsg* msg = NULL;
	int still_running;
	int msgs_left = -1;

	// Fetch the total downloadsize
	const size_t downloadsize = pakfire_httpclient_total_downloadsize(client);

	// If we know the downloadsize, we can show more details
	if (downloadsize) {
		progress_flags |=
			PAKFIRE_PROGRESS_SHOW_BYTES_TRANSFERRED |
			PAKFIRE_PROGRESS_SHOW_TRANSFER_SPEED;
	} else {
		progress_flags |=
			PAKFIRE_PROGRESS_SHOW_COUNTER;
	}

	// Fetch the total number of queued transfers
	const unsigned int num_queued_xfers = pakfire_httpclient_total_queued_xfers(client);

	// Create a new progress indicator
	r = pakfire_progress_create(&progress, client->ctx, progress_flags, NULL);
	if (r)
		goto ERROR;

	// Set the title
	if (title) {
		r = pakfire_progress_set_title(progress, title);
		if (r)
			goto ERROR;
	}

	// Start the progress
	r = pakfire_progress_start(progress, (downloadsize) ? downloadsize : num_queued_xfers);
	if (r)
		goto ERROR;

	do {
		// Make sure that we have up to parallel transfers active
		if (!r) {
			r = pakfire_httpclient_start_transfers(client, progress, &running_xfers);
			if (r)
				goto ERROR;
		}

		// Run cURL
		r = curl_multi_perform(client->curl, &still_running);
		if (r) {
			CTX_ERROR(client->ctx, "cURL error: %s\n", curl_easy_strerror(r));
			goto ERROR;
		}

		for (;;) {
			// Read the next message
			msg = curl_multi_info_read(client->curl, &msgs_left);
			if (!msg)
				break;

			switch (msg->msg) {
				case CURLMSG_DONE:
					// Update reference to transfer
					curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &xfer);

					// Remove the handle
					curl_multi_remove_handle(client->curl, pakfire_xfer_handle(xfer));

					// Find the matching xfer element
					x = pakfire_httpclient_xfer_find_running(client, xfer);

					// Remove the transfer from the running list
					if (x)
						TAILQ_REMOVE(&client->xfers_running, x, nodes);
					running_xfers--;

					// Call the done callback
					r = pakfire_xfer_done(xfer, msg->data.result);
					switch (-r) {
						// If we are asked to try again we will re-queue the transfer
						case EAGAIN:
							if (x)
								TAILQ_INSERT_TAIL(&client->xfers_queued, x, nodes);
							break;

						// Otherwise this transfer has finished
						default:
							if (x)
								pakfire_httpclient_xfer_free(x);
							if (r)
								goto ERROR;
							break;
					}

					// Reset transfer
					xfer = NULL;
					break;

				default:
					CTX_ERROR(client->ctx, "Received unhandled cURL message %u\n", msg->msg);
					break;
			}
		}

		// Wait a little before going through the loop again
		if (still_running)
			curl_multi_wait(client->curl, NULL, 0, 100, NULL);
	} while (still_running || !TAILQ_EMPTY(&client->xfers_queued));

	// We are finished!
	r = pakfire_progress_finish(progress);
	if (r)
		goto ERROR;

ERROR:
	// If the client was aborted, we need to fail all remaining running transfers
	while (!TAILQ_EMPTY(&client->xfers_running)) {
		x = TAILQ_LAST(&client->xfers_running, xfers_running);
		TAILQ_REMOVE(&client->xfers_running, x, nodes);

		// Fail the transfer
		pakfire_xfer_fail(x->xfer);
	}

	if (progress)
		pakfire_progress_unref(progress);

	return r;
}