Adding upstream version 1:2.3.21+dfsg1.upstream/1%2.3.21+dfsg1

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1383 insertions, 0 deletions

diff --git a/src/lib-http/http-client-peer.c b/src/lib-http/http-client-peer.c
new file mode 100644
index 0000000..eff40b4
--- /dev/null
+++ b/src/lib-http/http-client-peer.c
@@ -0,0 +1,1383 @@
+/* Copyright (c) 2013-2018 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "net.h"
+#include "time-util.h"
+#include "str.h"
+#include "hash.h"
+#include "array.h"
+#include "llist.h"
+#include "istream.h"
+#include "ostream.h"
+#include "iostream-ssl.h"
+#include "http-response-parser.h"
+
+#include "http-client-private.h"
+
+static void
+http_client_peer_connect_backoff(struct http_client_peer *peer);
+
+static void
+http_client_peer_shared_connection_success(
+	struct http_client_peer_shared *pshared);
+static void
+http_client_peer_shared_connection_failure(
+	struct http_client_peer_shared *pshared);
+static void
+http_client_peer_connection_succeeded_pool(struct http_client_peer *peer);
+static void
+http_client_peer_connection_failed_pool(struct http_client_peer *peer,
+					const char *reason);
+
+/*
+ * Peer address
+ */
+
+unsigned int ATTR_NO_SANITIZE_INTEGER
+http_client_peer_addr_hash(const struct http_client_peer_addr *peer)
+{
+	unsigned int hash = (unsigned int)peer->type;
+
+	switch (peer->type) {
+	case HTTP_CLIENT_PEER_ADDR_HTTPS:
+	case HTTP_CLIENT_PEER_ADDR_HTTPS_TUNNEL:
+		if (peer->a.tcp.https_name != NULL)
+			hash += str_hash(peer->a.tcp.https_name);
+		/* fall through */
+	case HTTP_CLIENT_PEER_ADDR_RAW:
+	case HTTP_CLIENT_PEER_ADDR_HTTP:
+		if (peer->a.tcp.ip.family != 0)
+			hash += net_ip_hash(&peer->a.tcp.ip);
+		hash += peer->a.tcp.port;
+		break;
+	case HTTP_CLIENT_PEER_ADDR_UNIX:
+		hash += str_hash(peer->a.un.path);
+		break;
+	}
+
+	return hash;
+}
+
+int http_client_peer_addr_cmp(const struct http_client_peer_addr *peer1,
+			      const struct http_client_peer_addr *peer2)
+{
+	int ret;
+
+	if (peer1->type != peer2->type)
+		return (peer1->type > peer2->type ? 1 : -1);
+	switch (peer1->type) {
+	case HTTP_CLIENT_PEER_ADDR_RAW:
+	case HTTP_CLIENT_PEER_ADDR_HTTP:
+	case HTTP_CLIENT_PEER_ADDR_HTTPS:
+	case HTTP_CLIENT_PEER_ADDR_HTTPS_TUNNEL:
+		/* Queues are created with peer addresses that have an
+		   uninitialized IP value, because that is assigned later when
+		   the host lookup completes. In all other other contexts, the
+		   IP is always initialized, so we do not compare IPs when one
+		   of them is unassigned. */
+		if (peer1->a.tcp.ip.family != 0 &&
+		    peer2->a.tcp.ip.family != 0 &&
+		    (ret = net_ip_cmp(&peer1->a.tcp.ip, &peer2->a.tcp.ip)) != 0)
+			return ret;
+		if (peer1->a.tcp.port != peer2->a.tcp.port)
+			return (peer1->a.tcp.port > peer2->a.tcp.port ? 1 : -1);
+		if (peer1->type != HTTP_CLIENT_PEER_ADDR_HTTPS &&
+			peer1->type != HTTP_CLIENT_PEER_ADDR_HTTPS_TUNNEL)
+			return 0;
+		return null_strcmp(peer1->a.tcp.https_name,
+				   peer2->a.tcp.https_name);
+	case HTTP_CLIENT_PEER_ADDR_UNIX:
+		return null_strcmp(peer1->a.un.path, peer2->a.un.path);
+	}
+	i_unreached();
+}
+
+/*
+ * Peer pool
+ */
+
+static struct http_client_peer_pool *
+http_client_peer_pool_create(struct http_client_peer_shared *pshared,
+			     struct ssl_iostream_context *ssl_ctx,
+			     const char *rawlog_dir)
+{
+	struct http_client_peer_pool *ppool;
+
+	ppool = i_new(struct http_client_peer_pool, 1);
+	ppool->refcount = 1;
+	ppool->peer = pshared;
+	ppool->event = event_create(pshared->cctx->event);
+	event_set_append_log_prefix(
+		ppool->event,
+		t_strdup_printf("peer %s: ",
+				http_client_peer_shared_label(pshared)));
+
+	http_client_peer_shared_ref(pshared);
+
+	i_array_init(&ppool->conns, 16);
+	i_array_init(&ppool->pending_conns, 16);
+	i_array_init(&ppool->idle_conns, 16);
+
+	DLLIST_PREPEND(&pshared->pools_list, ppool);
+
+	ppool->ssl_ctx = ssl_ctx;
+	ppool->rawlog_dir = i_strdup(rawlog_dir);
+
+	e_debug(ppool->event, "Peer pool created");
+
+	return ppool;
+}
+
+void http_client_peer_pool_ref(struct http_client_peer_pool *ppool)
+{
+	if (ppool->destroyed)
+		return;
+	ppool->refcount++;
+}
+
+void http_client_peer_pool_close(struct http_client_peer_pool **_ppool)
+{
+	struct http_client_peer_pool *ppool = *_ppool;
+	struct http_client_connection *conn;
+	ARRAY_TYPE(http_client_connection) conns;
+
+	http_client_peer_pool_ref(ppool);
+
+	/* Make a copy of the connection array; freed connections modify it */
+	t_array_init(&conns, array_count(&ppool->conns));
+	array_copy(&conns.arr, 0, &ppool->conns.arr, 0,
+		   array_count(&ppool->conns));
+	array_foreach_elem(&conns, conn)
+		http_client_connection_unref(&conn);
+	i_assert(array_count(&ppool->idle_conns) == 0);
+	i_assert(array_count(&ppool->pending_conns) == 0);
+	i_assert(array_count(&ppool->conns) == 0);
+
+	http_client_peer_pool_unref(_ppool);
+}
+
+void http_client_peer_pool_unref(struct http_client_peer_pool **_ppool)
+{
+	struct http_client_peer_pool *ppool = *_ppool;
+	struct http_client_peer_shared *pshared = ppool->peer;
+
+	*_ppool = NULL;
+
+	if (ppool->destroyed)
+		return;
+
+	i_assert(ppool->refcount > 0);
+	if (--ppool->refcount > 0)
+		return;
+
+	e_debug(ppool->event, "Peer pool destroy");
+	ppool->destroyed = TRUE;
+
+	i_assert(array_count(&ppool->idle_conns) == 0);
+	i_assert(array_count(&ppool->conns) == 0);
+	array_free(&ppool->idle_conns);
+	array_free(&ppool->pending_conns);
+	array_free(&ppool->conns);
+
+	DLLIST_REMOVE(&pshared->pools_list, ppool);
+
+	event_unref(&ppool->event);
+	i_free(ppool->rawlog_dir);
+	i_free(ppool);
+	http_client_peer_shared_unref(&pshared);
+}
+
+static struct http_client_peer_pool *
+http_client_peer_pool_get(struct http_client_peer_shared *pshared,
+			  struct http_client *client)
+{
+	struct http_client_peer_pool *ppool;
+	struct ssl_iostream_context *ssl_ctx = client->ssl_ctx;
+	const char *rawlog_dir = client->set.rawlog_dir;
+
+	i_assert(!http_client_peer_addr_is_https(&pshared->addr) ||
+		 ssl_ctx != NULL);
+
+	ppool = pshared->pools_list;
+	while (ppool != NULL) {
+		if (ppool->ssl_ctx == ssl_ctx &&
+		    null_strcmp(ppool->rawlog_dir, rawlog_dir) == 0)
+			break;
+		ppool = ppool->next;
+	}
+
+	if (ppool == NULL) {
+		ppool = http_client_peer_pool_create(pshared, ssl_ctx,
+						     rawlog_dir);
+	} else {
+		e_debug(ppool->event, "Peer pool reused");
+		http_client_peer_pool_ref(ppool);
+	}
+
+	return ppool;
+}
+
+static void
+http_client_peer_pool_connection_success(struct http_client_peer_pool *ppool)
+{
+	e_debug(ppool->event, "Successfully connected "
+		"(%u connections exist, %u pending)",
+		array_count(&ppool->conns), array_count(&ppool->pending_conns));
+
+	http_client_peer_shared_connection_success(ppool->peer);
+
+	if (array_count(&ppool->pending_conns) > 0) {
+		/* If there are other connections attempting to connect, wait
+		   for them before notifying other peer objects about the
+		   success (which may be premature). */
+	} else {
+		struct http_client_peer *peer;
+
+		/* This was the only/last connection and connecting to it
+		   succeeded. notify all interested peers in this pool about the
+		   success */
+		peer = ppool->peer->peers_list;
+		while (peer != NULL) {
+			struct http_client_peer *peer_next = peer->shared_next;
+			if (peer->ppool == ppool)
+				http_client_peer_connection_succeeded_pool(peer);
+			peer = peer_next;
+		}
+	}
+}
+
+static void
+http_client_peer_pool_connection_failure(struct http_client_peer_pool *ppool,
+					 const char *reason)
+{
+	e_debug(ppool->event,
+		"Failed to make connection "
+		"(%u connections exist, %u pending)",
+		array_count(&ppool->conns), array_count(&ppool->pending_conns));
+
+	http_client_peer_shared_connection_failure(ppool->peer);
+
+	if (array_count(&ppool->pending_conns) > 0) {
+		/* If there are other connections attempting to connect, wait
+		   for them before failing the requests. remember that we had
+		   trouble with connecting so in future we don't try to create
+		   more than one connection until connects work again. */
+	} else {
+		struct http_client_peer *peer;
+
+		/* This was the only/last connection and connecting to it
+		   failed. notify all interested peers in this pool about the
+		   failure */
+		peer = ppool->peer->peers_list;
+		while (peer != NULL) {
+			struct http_client_peer *peer_next = peer->shared_next;
+			if (peer->ppool == ppool)
+				http_client_peer_connection_failed_pool(peer, reason);
+			peer = peer_next;
+		}
+	}
+}
+
+/*
+ * Peer (shared)
+ */
+
+static struct http_client_peer_shared *
+http_client_peer_shared_create(struct http_client_context *cctx,
+			       const struct http_client_peer_addr *addr)
+{
+	struct http_client_peer_shared *pshared;
+
+	pshared = i_new(struct http_client_peer_shared, 1);
+	pshared->refcount = 1;
+	pshared->cctx = cctx;
+
+	pshared->addr = *addr;
+	switch (addr->type) {
+	case HTTP_CLIENT_PEER_ADDR_RAW:
+	case HTTP_CLIENT_PEER_ADDR_HTTP:
+		i_assert(pshared->addr.a.tcp.ip.family != 0);
+		break;
+	case HTTP_CLIENT_PEER_ADDR_HTTPS:
+	case HTTP_CLIENT_PEER_ADDR_HTTPS_TUNNEL:
+		i_assert(pshared->addr.a.tcp.ip.family != 0);
+		pshared->addr_name = i_strdup(addr->a.tcp.https_name);
+		pshared->addr.a.tcp.https_name = pshared->addr_name;
+		break;
+	case HTTP_CLIENT_PEER_ADDR_UNIX:
+		pshared->addr_name = i_strdup(addr->a.un.path);
+		pshared->addr.a.un.path = pshared->addr_name;
+		break;
+	default:
+		break;
+	}
+	pshared->event = event_create(cctx->event);
+	event_set_append_log_prefix(
+		pshared->event,
+		t_strdup_printf("peer %s (shared): ",
+				http_client_peer_shared_label(pshared)));
+
+	hash_table_insert(cctx->peers,
+			  (const struct http_client_peer_addr *)&pshared->addr,
+			  pshared);
+	DLLIST_PREPEND(&cctx->peers_list, pshared);
+
+	pshared->backoff_initial_time_msecs =
+		cctx->set.connect_backoff_time_msecs;
+	pshared->backoff_max_time_msecs =
+		cctx->set.connect_backoff_max_time_msecs;
+
+	e_debug(pshared->event, "Peer created");
+	return pshared;
+}
+
+void http_client_peer_shared_ref(struct http_client_peer_shared *pshared)
+{
+	pshared->refcount++;
+}
+
+void http_client_peer_shared_unref(struct http_client_peer_shared **_pshared)
+{
+	struct http_client_peer_shared *pshared = *_pshared;
+
+	*_pshared = NULL;
+
+	i_assert(pshared->refcount > 0);
+	if (--pshared->refcount > 0)
+		return;
+
+	e_debug(pshared->event, "Peer destroy");
+
+	i_assert(pshared->pools_list == NULL);
+
+	/* Unlist in client */
+	hash_table_remove(pshared->cctx->peers,
+			  (const struct http_client_peer_addr *)&pshared->addr);
+	DLLIST_REMOVE(&pshared->cctx->peers_list, pshared);
+
+	timeout_remove(&pshared->to_backoff);
+
+	event_unref(&pshared->event);
+	i_free(pshared->addr_name);
+	i_free(pshared->label);
+	i_free(pshared);
+}
+
+static struct http_client_peer_shared *
+http_client_peer_shared_get(struct http_client_context *cctx,
+			    const struct http_client_peer_addr *addr)
+{
+	struct http_client_peer_shared *pshared;
+
+	pshared = hash_table_lookup(cctx->peers, addr);
+	if (pshared == NULL) {
+		pshared = http_client_peer_shared_create(cctx, addr);
+	} else {
+		e_debug(pshared->event, "Peer reused");
+		http_client_peer_shared_ref(pshared);
+	}
+
+	return pshared;
+}
+
+void http_client_peer_shared_close(struct http_client_peer_shared **_pshared)
+{
+	struct http_client_peer_shared *pshared = *_pshared;
+	struct http_client_peer_pool *pool, *next;
+
+	http_client_peer_shared_ref(pshared);
+	pool = pshared->pools_list;
+	while (pool != NULL) {
+		next = pool->next;
+		http_client_peer_pool_close(&pool);
+		pool = next;
+	}
+	http_client_peer_shared_unref(_pshared);
+}
+
+const char *
+http_client_peer_shared_label(struct http_client_peer_shared *pshared)
+{
+	if (pshared->label == NULL) {
+		switch (pshared->addr.type) {
+		case HTTP_CLIENT_PEER_ADDR_HTTPS_TUNNEL:
+			pshared->label = i_strconcat(
+				http_client_peer_addr2str(&pshared->addr),
+				" (tunnel)", NULL);
+			break;
+		default:
+			pshared->label = i_strdup(
+				http_client_peer_addr2str(&pshared->addr));
+		}
+	}
+	return pshared->label;
+}
+
+static void
+http_client_peer_shared_connect_backoff(struct http_client_peer_shared *pshared)
+{
+	struct http_client_peer *peer;
+
+	i_assert(pshared->to_backoff != NULL);
+
+	e_debug(pshared->event, "Backoff timer expired");
+
+	timeout_remove(&pshared->to_backoff);
+
+	peer = pshared->peers_list;
+	while (peer != NULL) {
+		struct http_client_peer *peer_next = peer->shared_next;
+		
+		http_client_peer_connect_backoff(peer);
+		peer = peer_next;
+	}
+}
+
+static bool
+http_client_peer_shared_start_backoff_timer(
+	struct http_client_peer_shared *pshared)
+{
+	if (pshared->to_backoff != NULL)
+		return TRUE;
+
+	if (pshared->last_failure.tv_sec > 0) {
+		int backoff_time_spent =
+			timeval_diff_msecs(&ioloop_timeval,
+					   &pshared->last_failure);
+
+		if (backoff_time_spent < (int)pshared->backoff_current_time_msecs) {
+			unsigned int new_time = (unsigned int)
+				(pshared->backoff_current_time_msecs -
+				 backoff_time_spent);
+			e_debug(pshared->event,
+				"Starting backoff timer for %d msecs", new_time);
+			pshared->to_backoff = timeout_add_to(
+				pshared->cctx->ioloop, new_time,
+				http_client_peer_shared_connect_backoff, pshared);
+			return TRUE;
+		}
+
+		e_debug(pshared->event,
+			"Backoff time already exceeded by %d msecs",
+			(backoff_time_spent -
+			 pshared->backoff_current_time_msecs));
+	}
+	return FALSE;
+}
+
+static void
+http_client_peer_shared_increase_backoff_timer(
+	struct http_client_peer_shared *pshared)
+{
+	if (pshared->backoff_current_time_msecs == 0) {
+		pshared->backoff_current_time_msecs =
+			pshared->backoff_initial_time_msecs;
+	} else {
+		pshared->backoff_current_time_msecs *= 2;
+	}
+	if (pshared->backoff_current_time_msecs >
+	    pshared->backoff_max_time_msecs) {
+		pshared->backoff_current_time_msecs =
+			pshared->backoff_max_time_msecs;
+	}
+}
+
+static void
+http_client_peer_shared_reset_backoff_timer(
+	struct http_client_peer_shared *pshared)
+{
+	pshared->backoff_current_time_msecs = 0;
+
+	timeout_remove(&pshared->to_backoff);
+}
+
+static void
+http_client_peer_shared_connection_success(
+	struct http_client_peer_shared *pshared)
+{
+	pshared->last_failure.tv_sec = pshared->last_failure.tv_usec = 0;
+	http_client_peer_shared_reset_backoff_timer(pshared);
+}
+
+static void
+http_client_peer_shared_connection_failure(
+	struct http_client_peer_shared *pshared)
+{
+	struct http_client_peer_pool *ppool;
+	unsigned int pending = 0;
+
+	/* Determine the number of connections still pending */
+	ppool = pshared->pools_list;
+	while (ppool != NULL) {
+		pending += array_count(&ppool->pending_conns);
+		ppool = ppool->next;
+	}
+
+	pshared->last_failure = ioloop_timeval;
+
+	/* Manage backoff timer only when this was the only attempt */
+	if (pending == 0)
+		http_client_peer_shared_increase_backoff_timer(pshared);
+}
+
+static void
+http_client_peer_shared_connection_lost(struct http_client_peer_shared *pshared,
+					bool premature)
+{
+	/* Update backoff timer if the connection was lost prematurely. this
+	   prevents reconnecting immediately to a server that is misbehaving by
+	   disconnecting before sending a response.
+	 */
+	if (premature) {
+		pshared->last_failure = ioloop_timeval;
+		http_client_peer_shared_increase_backoff_timer(pshared);
+	}
+}
+
+void http_client_peer_shared_switch_ioloop(
+	struct http_client_peer_shared *pshared)
+{
+	if (pshared->to_backoff != NULL) {
+		pshared->to_backoff =
+			io_loop_move_timeout(&pshared->to_backoff);
+	}
+}
+
+unsigned int
+http_client_peer_shared_max_connections(struct http_client_peer_shared *pshared)
+{
+	struct http_client_peer *peer;
+	unsigned int max_conns = 0;
+
+	peer = pshared->peers_list;
+	while (peer != NULL) {
+		const struct http_client_settings *set = &peer->client->set;
+		unsigned int client_max_conns = set->max_parallel_connections;
+
+		if ((UINT_MAX - max_conns) <= client_max_conns)
+			return UINT_MAX;
+		max_conns += client_max_conns;
+		peer = peer->shared_next;
+	}
+
+	return max_conns;
+}
+
+/*
+ * Peer
+ */
+
+static void http_client_peer_drop(struct http_client_peer **_peer);
+
+static struct http_client_peer *
+http_client_peer_create(struct http_client *client,
+			struct http_client_peer_shared *pshared)
+{
+	struct http_client_peer *peer;
+
+	peer = i_new(struct http_client_peer, 1);
+	peer->refcount = 1;
+	peer->client = client;
+	peer->shared = pshared;
+
+	peer->event = event_create(client->event);
+	event_set_append_log_prefix(peer->event, t_strdup_printf(
+		"peer %s: ", http_client_peer_shared_label(pshared)));
+
+	i_array_init(&peer->queues, 16);
+	i_array_init(&peer->conns, 16);
+	i_array_init(&peer->pending_conns, 16);
+
+	DLLIST_PREPEND_FULL(&client->peers_list, peer,
+			    client_prev, client_next);
+	DLLIST_PREPEND_FULL(&pshared->peers_list, peer,
+			    shared_prev, shared_next);
+	pshared->peers_count++;
+
+	http_client_peer_shared_ref(pshared);
+	peer->ppool = http_client_peer_pool_get(pshared, client);
+
+	/* Choose backoff times */
+	if (pshared->peers_list == NULL ||
+	    (client->set.connect_backoff_time_msecs <
+	     pshared->backoff_initial_time_msecs)) {
+		pshared->backoff_initial_time_msecs =
+			client->set.connect_backoff_time_msecs;
+	}
+	if (pshared->peers_list == NULL ||
+	    (client->set.connect_backoff_max_time_msecs >
+	     pshared->backoff_max_time_msecs)) {
+		pshared->backoff_max_time_msecs =
+			client->set.connect_backoff_max_time_msecs;
+	}
+
+	e_debug(peer->event, "Peer created");
+	return peer;
+}
+
+void http_client_peer_ref(struct http_client_peer *peer)
+{
+	peer->refcount++;
+}
+
+static void http_client_peer_disconnect(struct http_client_peer *peer)
+{
+	struct http_client_queue *queue;
+	struct http_client *client = peer->client;
+	struct http_client_peer_shared *pshared = peer->shared;
+	struct http_client_connection *conn;
+	ARRAY_TYPE(http_client_connection) conns;
+
+	if (peer->disconnected)
+		return;
+	peer->disconnected = TRUE;
+
+	e_debug(peer->event, "Peer disconnect");
+
+	/* Make a copy of the connection array; freed connections modify it */
+	t_array_init(&conns, array_count(&peer->conns));
+	array_copy(&conns.arr, 0, &peer->conns.arr, 0,
+		   array_count(&peer->conns));
+	array_foreach_elem(&conns, conn)
+		http_client_connection_lost_peer(conn);
+	i_assert(array_count(&peer->conns) == 0);
+	array_clear(&peer->pending_conns);
+
+	timeout_remove(&peer->to_req_handling);
+
+	/* Unlist in client */
+	DLLIST_REMOVE_FULL(&client->peers_list, peer,
+			   client_prev, client_next);
+	/* Unlist in peer */
+	DLLIST_REMOVE_FULL(&pshared->peers_list, peer,
+			   shared_prev, shared_next);
+	pshared->peers_count--;
+
+	/* Unlink all queues */
+	array_foreach_elem(&peer->queues, queue)
+		http_client_queue_peer_disconnected(queue, peer);
+	array_clear(&peer->queues);
+}
+
+bool http_client_peer_unref(struct http_client_peer **_peer)
+{
+	struct http_client_peer *peer = *_peer;
+	struct http_client_peer_pool *ppool = peer->ppool;
+	struct http_client_peer_shared *pshared = peer->shared;
+
+	*_peer = NULL;
+
+	i_assert(peer->refcount > 0);
+	if (--peer->refcount > 0)
+		return TRUE;
+
+	e_debug(peer->event, "Peer destroy");
+
+	http_client_peer_disconnect(peer);
+
+	i_assert(array_count(&peer->queues) == 0);
+
+	event_unref(&peer->event);
+	array_free(&peer->conns);
+	array_free(&peer->pending_conns);
+	array_free(&peer->queues);
+	i_free(peer);
+
+	/* Choose new backoff times */
+	peer = pshared->peers_list;
+	while (peer != NULL) {
+		struct http_client *client = peer->client;
+
+		if (client->set.connect_backoff_time_msecs <
+		    pshared->backoff_initial_time_msecs) {
+			pshared->backoff_initial_time_msecs =
+				client->set.connect_backoff_time_msecs;
+		}
+		if (client->set.connect_backoff_max_time_msecs >
+		    pshared->backoff_max_time_msecs) {
+			pshared->backoff_max_time_msecs =
+				client->set.connect_backoff_max_time_msecs;
+		}
+		peer = peer->shared_next;
+	}
+
+	http_client_peer_pool_unref(&ppool);
+	http_client_peer_shared_unref(&pshared);
+	return FALSE;
+}
+
+void http_client_peer_close(struct http_client_peer **_peer)
+{
+	struct http_client_peer *peer = *_peer;
+
+	e_debug(peer->event, "Peer close");
+
+	http_client_peer_disconnect(peer);
+
+	(void)http_client_peer_unref(_peer);
+}
+
+static void http_client_peer_drop(struct http_client_peer **_peer)
+{
+	struct http_client_peer *peer = *_peer;
+	struct http_client_peer_shared *pshared = peer->shared;
+	unsigned int conns_active =
+		http_client_peer_active_connections(peer);
+
+	if (conns_active > 0) {
+		e_debug(peer->event,
+			"Not dropping peer (%d connections active)",
+			conns_active);
+		return;
+	}
+
+	if (pshared->to_backoff != NULL)
+		return;
+
+	if (http_client_peer_shared_start_backoff_timer(pshared)) {
+		e_debug(peer->event,
+			"Dropping peer (waiting for backof timeout)");
+
+		/* Will disconnect any pending connections */
+		http_client_peer_trigger_request_handler(peer);
+	} else {
+		e_debug(peer->event, "Dropping peer now");
+		/* Drop peer immediately */
+		http_client_peer_close(_peer);
+	}
+}
+
+struct http_client_peer *
+http_client_peer_get(struct http_client *client,
+		     const struct http_client_peer_addr *addr)
+{
+	struct http_client_peer *peer;
+	struct http_client_peer_shared *pshared;
+
+	pshared = http_client_peer_shared_get(client->cctx, addr);
+
+	peer = pshared->peers_list;
+	while (peer != NULL) {
+		if (peer->client == client)
+			break;
+		peer = peer->shared_next;
+	}
+
+	if (peer == NULL)
+		peer = http_client_peer_create(client, pshared);
+
+	http_client_peer_shared_unref(&pshared);
+	return peer;
+}
+
+static void
+http_client_peer_do_connect(struct http_client_peer *peer, unsigned int count)
+{
+	struct http_client_peer_pool *ppool = peer->ppool;
+	struct http_client_connection *const *idle_conns;
+	unsigned int i, idle_count;
+	bool claimed_existing = FALSE;
+
+	if (count == 0)
+		return;
+
+	idle_conns = array_get(&ppool->idle_conns, &idle_count);
+	for (i = 0; i < count && i < idle_count; i++) {
+		http_client_connection_claim_idle(idle_conns[i], peer);
+		claimed_existing = TRUE;
+
+		e_debug(peer->event,
+			"Claimed idle connection "
+			"(%u connections exist, %u pending)",
+			array_count(&peer->conns),
+			array_count(&peer->pending_conns));
+	}
+
+	for (; i < count; i++) {
+		e_debug(peer->event,
+			"Making new connection %u of %u "
+			"(%u connections exist, %u pending)",
+			i+1, count, array_count(&peer->conns),
+			array_count(&peer->pending_conns));
+
+		(void)http_client_connection_create(peer);
+	}
+
+	if (claimed_existing)
+		http_client_peer_connection_success(peer);
+}
+
+static void http_client_peer_connect_backoff(struct http_client_peer *peer)
+{
+	if (peer->connect_backoff && array_count(&peer->queues) == 0) {
+		http_client_peer_close(&peer);
+		return;
+	}
+
+	http_client_peer_do_connect(peer, 1);
+	peer->connect_backoff = FALSE;
+}
+
+static void
+http_client_peer_connect(struct http_client_peer *peer, unsigned int count)
+{
+	if (http_client_peer_shared_start_backoff_timer(peer->shared)) {
+		peer->connect_backoff = TRUE;
+		return;
+	}
+
+	http_client_peer_do_connect(peer, count);
+}
+
+bool http_client_peer_is_connected(struct http_client_peer *peer)
+{
+	struct http_client_connection *conn;
+
+	if (array_count(&peer->ppool->idle_conns) > 0)
+		return TRUE;
+
+	array_foreach_elem(&peer->conns, conn) {
+		if (conn->connected)
+			return TRUE;
+	}
+
+	return FALSE;
+}
+
+static void
+http_client_peer_cancel(struct http_client_peer *peer)
+{
+	struct http_client_connection *conn;
+	ARRAY_TYPE(http_client_connection) conns;
+
+	e_debug(peer->event, "Peer cancel");
+
+	/* Make a copy of the connection array; freed connections modify it */
+	t_array_init(&conns, array_count(&peer->conns));
+	array_copy(&conns.arr, 0, &peer->conns.arr, 0,
+		   array_count(&peer->conns));
+	array_foreach_elem(&conns, conn) {
+		if (!http_client_connection_is_active(conn))
+			http_client_connection_close(&conn);
+	}
+	i_assert(array_count(&peer->pending_conns) == 0);
+}
+
+static unsigned int
+http_client_peer_requests_pending(struct http_client_peer *peer,
+				  unsigned int *num_urgent_r)
+{
+	struct http_client_queue *queue;
+	unsigned int num_requests = 0, num_urgent = 0, requests, urgent;
+
+	array_foreach_elem(&peer->queues, queue) {
+		requests = http_client_queue_requests_pending(queue, &urgent);
+
+		num_requests += requests;
+		num_urgent += urgent;
+	}
+	*num_urgent_r = num_urgent;
+	return num_requests;
+}
+
+static void http_client_peer_check_idle(struct http_client_peer *peer)
+{
+	struct http_client_connection *conn;
+	unsigned int num_urgent = 0;
+
+	if (array_count(&peer->conns) == 0 &&
+	    http_client_peer_requests_pending(peer, &num_urgent) == 0) {
+		/* No connections or pending requests; disconnect immediately */
+		http_client_peer_drop(&peer);
+		return;
+	}
+
+	/* Check all connections for idle status */
+	array_foreach_elem(&peer->conns, conn)
+		http_client_connection_check_idle(conn);
+}
+
+static void
+http_client_peer_handle_requests_real(struct http_client_peer *peer)
+{
+	struct _conn_available {
+		struct http_client_connection *conn;
+		unsigned int pending_requests;
+	};
+	struct http_client_connection *const *conn_idx;
+	ARRAY(struct _conn_available) conns_avail;
+	struct _conn_available *conn_avail_idx;
+	struct http_client_peer_shared *pshared = peer->shared;
+	unsigned int connecting, closing, idle;
+	unsigned int num_pending, num_urgent, new_connections;
+	unsigned int working_conn_count;
+	struct http_client_peer *tmp_peer;
+	bool statistics_dirty = TRUE;
+
+	/* FIXME: limit the number of requests handled in one run to prevent
+	   I/O starvation. */
+
+	/* Disconnect pending connections if we're not linked to any queue
+	   anymore */
+	if (array_count(&peer->queues) == 0) {
+		if (array_count(&peer->conns) == 0 &&
+		    pshared->to_backoff == NULL) {
+			/* Peer is completely unused and inactive; drop it
+			   immediately */
+			http_client_peer_drop(&peer);
+			return;
+		}
+		e_debug(peer->event,
+			"Peer no longer used; will now cancel pending connections "
+			"(%u connections exist, %u pending)",
+			array_count(&peer->conns),
+			array_count(&peer->pending_conns));
+
+		http_client_peer_cancel(peer);
+		return;
+	}
+
+	/* Don't do anything unless we have pending requests */
+	num_pending = http_client_peer_requests_pending(peer, &num_urgent);
+	if (num_pending == 0) {
+		e_debug(peer->event,
+			"No requests to service for this peer "
+			"(%u connections exist, %u pending)",
+			array_count(&peer->conns),
+			array_count(&peer->pending_conns));
+		http_client_peer_check_idle(peer);
+		return;
+	}
+
+	http_client_peer_ref(peer);
+	peer->handling_requests = TRUE;
+	t_array_init(&conns_avail, array_count(&peer->conns));
+	do {
+		bool conn_lost = FALSE;
+
+		array_clear(&conns_avail);
+		closing = idle = 0;
+
+		/* Gather connection statistics */
+		array_foreach(&peer->conns, conn_idx) {
+			struct http_client_connection *conn = *conn_idx;
+			int ret;
+
+			ret = http_client_connection_check_ready(conn);
+			if (ret < 0) {
+				conn_lost = TRUE;
+				break;
+			} else if (ret > 0) {
+				struct _conn_available *conn_avail;
+				unsigned int insert_idx, pending_requests;
+
+				/* Compile sorted availability list */
+				pending_requests = http_client_connection_count_pending(conn);
+				if (array_count(&conns_avail) == 0) {
+					insert_idx = 0;
+				} else {
+					insert_idx = array_count(&conns_avail);
+					array_foreach_modifiable(&conns_avail, conn_avail_idx) {
+						if (conn_avail_idx->pending_requests > pending_requests) {
+							insert_idx = array_foreach_idx(&conns_avail, conn_avail_idx);
+							break;
+						}
+					}
+				}
+				conn_avail = array_insert_space(&conns_avail, insert_idx);
+				conn_avail->conn = conn;
+				conn_avail->pending_requests = pending_requests;
+				if (pending_requests == 0)
+					idle++;
+			}
+			/* Count the number of connecting and closing connections */
+			if (conn->closing)
+				closing++;
+		}
+
+		if (conn_lost) {
+			/* Connection array changed while iterating; retry */
+			continue;
+		}
+
+		working_conn_count = array_count(&peer->conns) - closing;
+		statistics_dirty = FALSE;
+
+		/* Use idle connections right away */
+		if (idle > 0) {
+			e_debug(peer->event,
+				"Using %u idle connections to handle %u requests "
+				"(%u total connections ready)",
+				idle, num_pending > idle ? idle : num_pending,
+				array_count(&conns_avail));
+
+			array_foreach_modifiable(&conns_avail, conn_avail_idx) {
+				if (num_pending == 0 ||
+				    conn_avail_idx->pending_requests > 0)
+					break;
+				idle--;
+				if (http_client_connection_next_request(conn_avail_idx->conn) <= 0) {
+					/* No longer available (probably connection error/closed) */
+					statistics_dirty = TRUE;
+					conn_avail_idx->conn = NULL;
+				} else {
+					/* Update statistics */
+					conn_avail_idx->pending_requests++;
+					if (num_urgent > 0)
+						num_urgent--;
+					num_pending--;
+				}
+			}
+		}
+
+		/* Don't continue unless we have more pending requests */
+		num_pending = http_client_peer_requests_pending(peer, &num_urgent);
+		if (num_pending == 0) {
+			e_debug(peer->event,
+				"No more requests to service for this peer "
+				"(%u connections exist, %u pending)",
+				array_count(&peer->conns),
+				array_count(&peer->pending_conns));
+			http_client_peer_check_idle(peer);
+			break;
+		}
+	} while (statistics_dirty);
+
+	tmp_peer = peer;
+	if (!http_client_peer_unref(&tmp_peer))
+		return;
+	peer->handling_requests = FALSE;
+
+	if (num_pending == 0)
+		return;
+
+	i_assert(idle == 0);
+	connecting = array_count(&peer->pending_conns);
+
+	/* Determine how many new connections we can set up */
+	if (pshared->last_failure.tv_sec > 0 && working_conn_count > 0 &&
+	    working_conn_count == connecting) {
+		/* Don't create new connections until the existing ones have
+		   finished connecting successfully. */
+		new_connections = 0;
+	} else {
+		if ((working_conn_count - connecting + num_urgent) >=
+		    peer->client->set.max_parallel_connections) {
+			/* Only create connections for urgent requests */
+			new_connections = (num_urgent > connecting ?
+					   num_urgent - connecting : 0);
+		} else if (num_pending <= connecting) {
+			/* There are already enough connections being made */
+			new_connections = 0;
+		} else if (working_conn_count == connecting) {
+			/* No connections succeeded so far, don't hammer the
+			   server with more than one connection attempt unless
+			   its urgent */
+			if (num_urgent > 0) {
+				new_connections = (num_urgent > connecting ?
+						   num_urgent - connecting : 0);
+			} else {
+				new_connections = (connecting == 0 ? 1 : 0);
+			}
+		} else if ((num_pending - connecting) >
+			   (peer->client->set.max_parallel_connections -
+			    working_conn_count)) {
+			/* Create maximum allowed connections */
+			new_connections =
+				(peer->client->set.max_parallel_connections -
+				 working_conn_count);
+		} else {
+			/* Create as many connections as we need */
+			new_connections = num_pending - connecting;
+		}
+	}
+
+	/* Create connections */
+	if (new_connections > 0) {
+		e_debug(peer->event,
+			"Creating %u new connections to handle requests "
+			"(already %u usable, connecting to %u, closing %u)",
+			new_connections, working_conn_count - connecting,
+			connecting, closing);
+		http_client_peer_connect(peer, new_connections);
+		return;
+	}
+
+	/* Cannot create new connections for normal request; attempt pipelining
+	 */
+	if ((working_conn_count - connecting) >=
+	    peer->client->set.max_parallel_connections) {
+		unsigned int pipeline_level = 0, total_handled = 0, handled;
+
+		if (!pshared->allows_pipelining) {
+			e_debug(peer->event,
+				"Will not pipeline until peer has shown support");
+			return;
+		}
+
+		/* Fill pipelines */
+		do {
+			handled = 0;
+			/* Fill smallest pipelines first,
+			   until all pipelines are filled to the same level */
+			array_foreach_modifiable(&conns_avail, conn_avail_idx) {
+				if (conn_avail_idx->conn == NULL)
+					continue;
+				if (pipeline_level == 0) {
+					pipeline_level = conn_avail_idx->pending_requests;
+				} else if (conn_avail_idx->pending_requests > pipeline_level) {
+					pipeline_level = conn_avail_idx->pending_requests;
+					break; /* restart from least busy connection */
+				}
+				/* Pipeline it */
+				if (http_client_connection_next_request(conn_avail_idx->conn) <= 0) {
+					/* connection now unavailable */
+					conn_avail_idx->conn = NULL;
+				} else {
+					/* Successfully pipelined */
+					conn_avail_idx->pending_requests++;
+					num_pending--;
+					handled++;
+				}
+			}
+
+			total_handled += handled;
+		} while (num_pending > num_urgent && handled > 0);
+
+		e_debug(peer->event,
+			"Pipelined %u requests "
+			"(filled pipelines up to %u requests)",
+			total_handled, pipeline_level);
+		return;
+	}
+
+	/* Still waiting for connections to finish */
+	e_debug(peer->event, "No request handled; "
+		"waiting for pending connections "
+		"(%u connections exist, %u pending)",
+		array_count(&peer->conns), connecting);
+	return;
+}
+
+static void http_client_peer_handle_requests(struct http_client_peer *peer)
+{
+	timeout_remove(&peer->to_req_handling);
+
+	T_BEGIN {
+		http_client_peer_handle_requests_real(peer);
+	} T_END;
+}
+
+void http_client_peer_trigger_request_handler(struct http_client_peer *peer)
+{
+	/* Trigger request handling through timeout */
+	if (peer->to_req_handling == NULL) {
+		peer->to_req_handling =	timeout_add_short_to(
+			peer->client->ioloop, 0,
+			http_client_peer_handle_requests, peer);
+	}
+}
+
+bool http_client_peer_have_queue(struct http_client_peer *peer,
+				 struct http_client_queue *queue)
+{
+	struct http_client_queue *const *queue_idx;
+
+	array_foreach(&peer->queues, queue_idx) {
+		if (*queue_idx == queue)
+			return TRUE;
+	}
+	return FALSE;
+}
+
+void http_client_peer_link_queue(struct http_client_peer *peer,
+				 struct http_client_queue *queue)
+{
+	if (!http_client_peer_have_queue(peer, queue)) {
+		array_push_back(&peer->queues, &queue);
+
+		e_debug(peer->event, "Linked queue %s (%d queues linked)",
+			queue->name, array_count(&peer->queues));
+	}
+}
+
+void http_client_peer_unlink_queue(struct http_client_peer *peer,
+				   struct http_client_queue *queue)
+{
+	struct http_client_queue *const *queue_idx;
+
+	array_foreach(&peer->queues, queue_idx) {
+		if (*queue_idx == queue) {
+			array_delete(&peer->queues,
+				array_foreach_idx(&peer->queues, queue_idx), 1);
+
+			e_debug(peer->event,
+				"Unlinked queue %s (%d queues linked)",
+				queue->name, array_count(&peer->queues));
+
+			if (array_count(&peer->queues) == 0)
+				http_client_peer_check_idle(peer);
+			return;
+		}
+	}
+}
+
+struct http_client_request *
+http_client_peer_claim_request(struct http_client_peer *peer, bool no_urgent)
+{
+	struct http_client_queue *queue;
+	struct http_client_request *req;
+
+	array_foreach_elem(&peer->queues, queue) {
+		req = http_client_queue_claim_request(
+			queue, &peer->shared->addr, no_urgent);
+		if (req != NULL) {
+			req->peer = peer;
+			return req;
+		}
+	}
+
+	return NULL;
+}
+
+void http_client_peer_connection_success(struct http_client_peer *peer)
+{
+	struct http_client_peer_pool *ppool = peer->ppool;
+	struct http_client_queue *queue;
+
+	e_debug(peer->event, "Successfully connected "
+		"(%u connections exist, %u pending)",
+		array_count(&peer->conns), array_count(&peer->pending_conns));
+
+	http_client_peer_pool_connection_success(ppool);
+
+	array_foreach_elem(&peer->queues, queue)
+		http_client_queue_connection_success(queue, peer);
+
+	http_client_peer_trigger_request_handler(peer);
+}
+
+void http_client_peer_connection_failure(struct http_client_peer *peer,
+					 const char *reason)
+{
+	struct http_client_peer_pool *ppool = peer->ppool;
+
+	e_debug(peer->event, "Connection failed "
+		"(%u connections exist, %u pending)",
+		array_count(&peer->conns), array_count(&peer->pending_conns));
+
+	http_client_peer_pool_connection_failure(ppool, reason);
+
+	peer->connect_failed = TRUE;
+}
+
+static void
+http_client_peer_connection_succeeded_pool(struct http_client_peer *peer)
+{
+	if (!peer->connect_failed)
+		return;
+	peer->connect_failed = FALSE;
+
+	e_debug(peer->event,
+		"A connection succeeded within our peer pool, "
+		"so this peer can retry connecting as well if needed "
+		"(%u connections exist, %u pending)",
+		array_count(&peer->conns), array_count(&peer->pending_conns));
+
+	/* If there are pending requests for this peer, try creating a new
+	   connection for them. if not, this peer will wind itself down. */
+	http_client_peer_trigger_request_handler(peer);
+}
+
+static void
+http_client_peer_connection_failed_pool(struct http_client_peer *peer,
+					const char *reason)
+{
+	struct http_client_queue *queue;
+	ARRAY_TYPE(http_client_queue) queues;
+
+	e_debug(peer->event,
+		"Failed to establish any connection within our peer pool: %s "
+		"(%u connections exist, %u pending)", reason,
+		array_count(&peer->conns), array_count(&peer->pending_conns));
+
+	peer->connect_failed = TRUE;
+
+	/* Make a copy of the queue array; queues get linked/unlinged while the
+	   connection failure is handled */
+	t_array_init(&queues, array_count(&peer->queues));
+	array_copy(&queues.arr, 0, &peer->queues.arr, 0,
+		   array_count(&peer->queues));
+
+	/* Failed to make any connection. a second connect will probably also
+	   fail, so just try another IP for the hosts(s) or abort all requests
+	   if this was the only/last option. */
+	array_foreach_elem(&queues, queue)
+		http_client_queue_connection_failure(queue, peer, reason);
+}
+
+void http_client_peer_connection_lost(struct http_client_peer *peer,
+				      bool premature)
+{
+	unsigned int num_pending, num_urgent;
+
+	/* We get here when an already connected connection fails. if the
+	   connect itself fails, http_client_peer_shared_connection_failure() is
+	   called instead. */
+
+	if (peer->disconnected)
+		return;
+
+	http_client_peer_shared_connection_lost(peer->shared, premature);
+
+	num_pending = http_client_peer_requests_pending(peer, &num_urgent);
+
+	e_debug(peer->event,
+		"Lost a connection%s "
+		"(%u queues linked, %u connections left, "
+		 "%u connections pending, %u requests pending, "
+		 "%u requests urgent)",
+		(premature ? " prematurely" : ""),
+		array_count(&peer->queues), array_count(&peer->conns),
+		array_count(&peer->pending_conns), num_pending, num_urgent);
+
+	if (peer->handling_requests) {
+		/* We got here from the request handler loop */
+		e_debug(peer->event,
+			"Lost a connection while handling requests");
+		return;
+	}
+
+	/* If there are pending requests for this peer, create a new connection
+	   for them. if not, this peer will wind itself down. */
+	http_client_peer_trigger_request_handler(peer);
+}
+
+unsigned int
+http_client_peer_active_connections(struct http_client_peer *peer)
+{
+	struct http_client_connection *conn;
+	unsigned int active = 0;
+
+	/* Find idle connections */
+	array_foreach_elem(&peer->conns, conn) {
+		if (http_client_connection_is_active(conn))
+			active++;
+	}
+
+	return active;
+}
+
+unsigned int
+http_client_peer_pending_connections(struct http_client_peer *peer)
+{
+	return array_count(&peer->pending_conns);
+}
+
+void http_client_peer_switch_ioloop(struct http_client_peer *peer)
+{
+	if (peer->to_req_handling != NULL) {
+		peer->to_req_handling =
+			io_loop_move_timeout(&peer->to_req_handling);
+	}
+}