Decouple code for thrmgr and conn handling threads, so create pxythr.c/h

pull/48/head
Soner Tari 4 years ago
parent aab56d42a9
commit 8b27cfce95

@ -435,6 +435,7 @@ protoautossl_free(pxy_conn_ctx_t *ctx)
protossl_free(ctx);
}
// @attention Called by thrmgr thread
protocol_t
protoautossl_setup(pxy_conn_ctx_t *ctx)
{

@ -978,6 +978,7 @@ protohttp_free_child(pxy_conn_child_ctx_t *ctx)
protohttp_free_ctx(http_ctx);
}
// @attention Called by thrmgr thread
protocol_t
protohttp_setup(pxy_conn_ctx_t *ctx)
{
@ -996,6 +997,7 @@ protohttp_setup(pxy_conn_ctx_t *ctx)
return PROTO_HTTP;
}
// @attention Called by thrmgr thread
protocol_t
protohttps_setup(pxy_conn_ctx_t *ctx)
{

@ -90,6 +90,7 @@ protopop3_validate(pxy_conn_ctx_t *ctx, char *packet
return 0;
}
// @attention Called by thrmgr thread
protocol_t
protopop3_setup(pxy_conn_ctx_t *ctx)
{
@ -106,6 +107,7 @@ protopop3_setup(pxy_conn_ctx_t *ctx)
return PROTO_POP3;
}
// @attention Called by thrmgr thread
protocol_t
protopop3s_setup(pxy_conn_ctx_t *ctx)
{

@ -302,6 +302,7 @@ protosmtps_bev_eventcb(struct bufferevent *bev, short events, void *arg)
}
}
// @attention Called by thrmgr thread
protocol_t
protosmtp_setup(pxy_conn_ctx_t *ctx)
{
@ -323,6 +324,7 @@ protosmtp_setup(pxy_conn_ctx_t *ctx)
return PROTO_SMTP;
}
// @attention Called by thrmgr thread
protocol_t
protosmtps_setup(pxy_conn_ctx_t *ctx)
{

@ -1136,7 +1136,7 @@ protossl_fd_readcb(evutil_socket_t fd, UNUSED short what, void *arg)
event_free(ctx->ev);
ctx->ev = NULL;
pxy_thrmgr_remove_pending_ssl_conn(ctx);
pxy_thr_remove_pending_ssl_conn(ctx);
// Child connections will use the sni info obtained by the parent conn
/* for SSL, peek ClientHello and parse SNI from it */
@ -1239,7 +1239,7 @@ protossl_init_conn(evutil_socket_t fd, UNUSED short what, void *arg)
// @attention Add the conn to pending ssl conns list before adding (activating) the event
// Because the event may (and does) fire before this thread adds the conn to pending ssl conns list,
// and since the pending flag is not set yet, the conn remains in the pending list
pxy_thrmgr_add_pending_ssl_conn(ctx);
pxy_thr_add_pending_ssl_conn(ctx);
if (event_add(ctx->ev, NULL) == -1) {
log_finest("event_add failed");
@ -1596,6 +1596,7 @@ protossl_bev_eventcb_child(struct bufferevent *bev, short events, void *arg)
}
}
// @attention Called by thrmgr thread
protocol_t
protossl_setup(pxy_conn_ctx_t *ctx)
{

@ -97,6 +97,14 @@ proxy_listener_ctx_free(proxy_listener_ctx_t *ctx)
/*
* Callback for accept events on the socket listener bufferevent.
* Called when a new incoming connection has been accepted.
* Initiates the connection to the server. The incoming connection
* from the client is not being activated until we have a successful
* connection to the server, because we need the server's certificate
* in order to set up the SSL session to the client.
* For consistency, plain TCP works the same way, even if we could
* start reading from the client while waiting on the connection to
* the server to connect.
*/
static void
proxy_listener_acceptcb(UNUSED struct evconnlistener *listener,
@ -105,8 +113,39 @@ proxy_listener_acceptcb(UNUSED struct evconnlistener *listener,
void *arg)
{
proxy_listener_ctx_t *lctx = arg;
log_finest_main_va("ENTER, fd=%d", fd);
pxy_conn_setup(fd, peeraddr, peeraddrlen, lctx->thrmgr, lctx->spec, lctx->global, lctx->clisock);
/* create per connection state and attach to thread */
pxy_conn_ctx_t *ctx = pxy_conn_ctx_new(fd, lctx->thrmgr, lctx->spec, lctx->global, lctx->clisock);
if (!ctx) {
log_err_level_printf(LOG_CRIT, "Error allocating ctx memory\n");
evutil_closesocket(fd);
return;
}
pxy_thrmgr_attach(ctx);
/* prepare logging part 1 and user auth */
ctx->srcaddrlen = peeraddrlen;
memcpy(&ctx->srcaddr, peeraddr, ctx->srcaddrlen);
// Switch from thrmgr to connection handling thread, i.e. change the event base, asap
// This prevents possible multithreading issues between thrmgr and conn handling threads
ctx->ev = event_new(ctx->evbase, -1, 0, ctx->protoctx->init_conn, ctx);
if (!ctx->ev) {
log_err_level_printf(LOG_CRIT, "Error creating initial event, aborting connection\n");
log_fine("Error creating initial event, aborting connection");
goto out;
}
// The only purpose of this event is to change the event base, so it is a one-shot event
if (event_add(ctx->ev, NULL) == -1)
goto out;
event_active(ctx->ev, 0, 0);
return;
out:
evutil_closesocket(fd);
pxy_conn_ctx_free(ctx, 1);
}
/*

@ -88,6 +88,7 @@ char *protocol_names[] = {
"SSL",
};
// @attention Called by thrmgr thread
static protocol_t NONNULL(1)
pxy_setup_proto(pxy_conn_ctx_t *ctx)
{
@ -178,7 +179,8 @@ pxy_setup_proto_child(pxy_conn_child_ctx_t *ctx)
return proto;
}
static pxy_conn_ctx_t * MALLOC NONNULL(2,3,4)
// @attention Called by thrmgr thread
pxy_conn_ctx_t *
pxy_conn_ctx_new(evutil_socket_t fd,
pxy_thrmgr_ctx_t *thrmgr,
proxyspec_t *spec, global_t *global,
@ -216,8 +218,6 @@ pxy_conn_ctx_new(evutil_socket_t fd,
ctx->next = NULL;
pxy_thrmgr_attach(ctx);
#ifdef HAVE_LOCAL_PROCINFO
ctx->lproc.pid = -1;
#endif /* HAVE_LOCAL_PROCINFO */
@ -247,7 +247,7 @@ pxy_conn_ctx_new_child(evutil_socket_t fd, pxy_conn_ctx_t *ctx)
}
// @attention Child connections use the parent's event bases, otherwise we would get multithreading issues
pxy_thrmgr_attach_child(ctx);
pxy_thr_inc_load(ctx->thr);
return child_ctx;
}
@ -256,11 +256,7 @@ pxy_conn_ctx_free_child(pxy_conn_child_ctx_t *ctx)
{
log_finest("ENTER");
if (ctx->conn->thr_locked) {
pxy_thrmgr_detach_child_unlocked(ctx->conn);
} else {
pxy_thrmgr_detach_child(ctx->conn);
}
pxy_thr_dec_load(ctx->conn->thr);
// If the proto doesn't have special args, proto_free() callback is NULL
if (ctx->protoctx->proto_free) {
@ -386,13 +382,7 @@ pxy_conn_ctx_free(pxy_conn_ctx_t *ctx, int by_requestor)
}
}
if (ctx->in_thr_conns) {
if (ctx->thr_locked) {
pxy_thrmgr_detach_unlocked(ctx);
} else {
pxy_thrmgr_detach(ctx);
}
}
pxy_thr_detach(ctx);
if (ctx->srchost_str) {
free(ctx->srchost_str);
@ -1060,7 +1050,7 @@ pxy_listener_acceptcb_child(UNUSED struct evconnlistener *listener, evutil_socke
pxy_conn_term(ctx, 1);
goto out;
}
ctx->thr->max_load = MAX(ctx->thr->max_load, ctx->thr->load);
ctx->thr->max_load = MAX(ctx->thr->max_load, pxy_thr_get_load(ctx->thr));
ctx->child_count++;
// Prepend child ctx to parent ctx child list
@ -1449,7 +1439,7 @@ pxy_bev_eventcb_postexec_logging_and_stats(struct bufferevent *bev, short events
}
if (bev == ctx->srvdst.bev) {
ctx->thr->max_load = MAX(ctx->thr->max_load, ctx->thr->load);
ctx->thr->max_load = MAX(ctx->thr->max_load, pxy_thr_get_load(ctx->thr));
ctx->thr->max_fd = MAX(ctx->thr->max_fd, ctx->fd);
// src and other fd stats are collected in acceptcb functions
@ -1848,7 +1838,8 @@ pxy_conn_init(pxy_conn_ctx_t *ctx)
log_finest("ENTER");
// Always keep thr load and conns list in sync
ctx->thr->load++;
pxy_thr_inc_load(ctx->thr);
ctx->next = ctx->thr->conns;
ctx->thr->conns = ctx;
ctx->in_thr_conns = 1;
@ -1896,55 +1887,4 @@ out:
return -1;
}
/*
* Callback for accept events on the socket listener bufferevent.
* Called when a new incoming connection has been accepted.
* Initiates the connection to the server. The incoming connection
* from the client is not being activated until we have a successful
* connection to the server, because we need the server's certificate
* in order to set up the SSL session to the client.
* For consistency, plain TCP works the same way, even if we could
* start reading from the client while waiting on the connection to
* the server to connect.
*/
void
pxy_conn_setup(evutil_socket_t fd,
struct sockaddr *peeraddr, int peeraddrlen,
pxy_thrmgr_ctx_t *thrmgr,
proxyspec_t *spec, global_t *global,
evutil_socket_t clisock)
{
log_finest_main_va("ENTER, fd=%d", fd);
/* create per connection state and attach to thread */
pxy_conn_ctx_t *ctx = pxy_conn_ctx_new(fd, thrmgr, spec, global, clisock);
if (!ctx) {
log_err_level_printf(LOG_CRIT, "Error allocating ctx memory\n");
evutil_closesocket(fd);
return;
}
/* prepare logging part 1 and user auth */
ctx->srcaddrlen = peeraddrlen;
memcpy(&ctx->srcaddr, peeraddr, ctx->srcaddrlen);
// Switch from thrmgr to connection handling thread, i.e. change the event base, asap
// This prevents possible multithreading issues between thrmgr and conn handling threads
ctx->ev = event_new(ctx->evbase, -1, 0, ctx->protoctx->init_conn, ctx);
if (!ctx->ev) {
log_err_level_printf(LOG_CRIT, "Error creating initial event, aborting connection\n");
log_fine("Error creating initial event, aborting connection");
goto out;
}
// The only purpose of this event is to change the event base, so it is a one-shot event
if (event_add(ctx->ev, NULL) == -1)
goto out;
event_active(ctx->ev, 0, 0);
return;
out:
evutil_closesocket(fd);
pxy_conn_ctx_free(ctx, 1);
}
/* vim: set noet ft=c: */

@ -244,7 +244,6 @@ struct pxy_conn_ctx {
// Thread that the conn is attached to
pxy_thr_ctx_t *thr;
unsigned int thr_locked : 1; /* 1 to prevent double locking */
unsigned int in_thr_conns : 1; /* 1 if conn in thr conns list */
// Unique id of the conn
@ -382,7 +381,7 @@ int pxy_try_consume_last_input_child(struct bufferevent *, pxy_conn_child_ctx_t
void pxy_discard_inbuf(struct bufferevent *) NONNULL(1);
int pxy_conn_init(pxy_conn_ctx_t *) NONNULL(1);
pxy_conn_ctx_t *pxy_conn_ctx_new(evutil_socket_t, pxy_thrmgr_ctx_t *, proxyspec_t *, global_t *, evutil_socket_t) MALLOC NONNULL(2,3,4);
void pxy_conn_ctx_free(pxy_conn_ctx_t *, int) NONNULL(1);
void pxy_conn_free(pxy_conn_ctx_t *, int) NONNULL(1);
void pxy_conn_term(pxy_conn_ctx_t *, int) NONNULL(1);

@ -0,0 +1,409 @@
/*-
* SSLsplit - transparent SSL/TLS interception
* https://www.roe.ch/SSLsplit
*
* Copyright (c) 2017-2020, Soner Tari <sonertari@gmail.com>.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* 1. Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDER AND CONTRIBUTORS ``AS IS''
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include "pxythr.h"
#include "log.h"
#include "pxyconn.h"
#include <assert.h>
#include <sys/param.h>
size_t
pxy_thr_get_load(pxy_thr_ctx_t *tctx)
{
size_t load;
pthread_mutex_lock(&tctx->mutex);
load = tctx->load;
pthread_mutex_unlock(&tctx->mutex);
return load;
}
void
pxy_thr_inc_load(pxy_thr_ctx_t *tctx)
{
pthread_mutex_lock(&tctx->mutex);
tctx->load++;
pthread_mutex_unlock(&tctx->mutex);
}
void
pxy_thr_dec_load(pxy_thr_ctx_t *tctx)
{
pthread_mutex_lock(&tctx->mutex);
tctx->load--;
pthread_mutex_unlock(&tctx->mutex);
}
void
pxy_thr_add_pending_ssl_conn(pxy_conn_ctx_t *ctx)
{
if (!ctx->sslctx->pending) {
log_finest("Adding conn");
ctx->sslctx->pending = 1;
ctx->thr->pending_ssl_conn_count++;
ctx->sslctx->next_pending = ctx->thr->pending_ssl_conns;
ctx->thr->pending_ssl_conns = ctx;
}
}
void
pxy_thr_remove_pending_ssl_conn(pxy_conn_ctx_t *ctx)
{
if (ctx->sslctx && ctx->sslctx->pending) {
log_finest("Removing conn");
// Thr pending_ssl_conns list cannot be empty, if the sslctx->pending flag of a conn is set
assert(ctx->thr->pending_ssl_conns != NULL);
ctx->sslctx->pending = 0;
ctx->thr->pending_ssl_conn_count--;
// @attention We may get multiple conns with the same fd combinations, so fds cannot uniquely define a conn; hence the need for unique ids.
if (ctx->id == ctx->thr->pending_ssl_conns->id) {
ctx->thr->pending_ssl_conns = ctx->thr->pending_ssl_conns->sslctx->next_pending;
return;
} else {
pxy_conn_ctx_t *current = ctx->thr->pending_ssl_conns->sslctx->next_pending;
pxy_conn_ctx_t *previous = ctx->thr->pending_ssl_conns;
while (current != NULL && previous != NULL) {
if (ctx->id == current->id) {
previous->sslctx->next_pending = current->sslctx->next_pending;
return;
}
previous = current;
current = current->sslctx->next_pending;
}
// This should never happen
log_err_level_printf(LOG_CRIT, "Cannot find conn in thrmgr pending_conns\n");
log_fine("Cannot find conn in thrmgr pending_conns");
assert(0);
}
}
}
/*
* Detach a connection from a thread by index.
* This function cannot fail.
*/
void
pxy_thr_detach(pxy_conn_ctx_t *ctx)
{
assert(ctx != NULL);
assert(ctx->children == NULL);
log_finest("ENTER");
pxy_thr_remove_pending_ssl_conn(ctx);
if (!ctx->in_thr_conns) {
log_fine("Not in thr conns");
return;
}
// Thr conns list cannot be empty
assert(ctx->thr->conns != NULL);
log_finest("Removing conn");
// We increment thr load in pxy_conn_init() only (for parent conns)
pxy_thr_dec_load(ctx->thr);
// No need to reset the ctx->in_thr_conns flag, as we free the ctx right after calling this function
// @attention We may get multiple conns with the same fd combinations, so fds cannot uniquely identify a conn; hence the need for unique ids.
if (ctx->id == ctx->thr->conns->id) {
ctx->thr->conns = ctx->thr->conns->next;
return;
} else {
pxy_conn_ctx_t *current = ctx->thr->conns->next;
pxy_conn_ctx_t *previous = ctx->thr->conns;
while (current != NULL && previous != NULL) {
if (ctx->id == current->id) {
previous->next = current->next;
return;
}
previous = current;
current = current->next;
}
// This should never happen
log_err_level_printf(LOG_CRIT, "Cannot find conn in thr conns\n");
log_fine("Cannot find conn in thr conns");
assert(0);
}
}
static void
pxy_thr_get_expired_conns(pxy_thr_ctx_t *tctx, pxy_conn_ctx_t **expired_conns)
{
*expired_conns = NULL;
if (tctx->conns) {
time_t now = time(NULL);
pxy_conn_ctx_t *ctx = tctx->conns;
while (ctx) {
time_t elapsed_time = now - ctx->atime;
if (elapsed_time > (time_t)tctx->thrmgr->global->conn_idle_timeout) {
ctx->next_expired = *expired_conns;
*expired_conns = ctx;
}
ctx = ctx->next;
}
ctx = tctx->pending_ssl_conns;
while (ctx) {
time_t elapsed_time = now - ctx->atime;
if (elapsed_time > (time_t)tctx->thrmgr->global->conn_idle_timeout) {
ctx->next_expired = *expired_conns;
*expired_conns = ctx;
}
ctx = ctx->sslctx->next_pending;
}
if (tctx->thrmgr->global->statslog) {
ctx = *expired_conns;
while (ctx) {
log_finest_main_va("thr=%d, fd=%d, child_fd=%d, time=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d, pc=%d",
ctx->thr->thridx, ctx->fd, ctx->child_fd, (long long)(now - ctx->atime),
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid, ctx->sslctx ? ctx->sslctx->pending : 0);
char *msg;
if (asprintf(&msg, "EXPIRED: thr=%d, time=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d\n",
ctx->thr->thridx, (long long)(now - ctx->atime),
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid) < 0) {
break;
}
if (log_conn(msg) == -1) {
log_err_level_printf(LOG_WARNING, "Expired conn logging failed\n");
}
free(msg);
ctx = ctx->next_expired;
}
}
}
}
static evutil_socket_t
pxy_thr_print_children(pxy_conn_child_ctx_t *ctx,
#ifdef DEBUG_PROXY
unsigned int parent_idx,
#endif /* DEBUG_PROXY */
evutil_socket_t max_fd)
{
while (ctx) {
// @attention No need to log child stats
log_finest_main_va("CHILD CONN: thr=%d, id=%d, pid=%u, src=%d, dst=%d, c=%d-%d",
ctx->conn->thr->thridx, ctx->conn->child_count, parent_idx, ctx->fd, ctx->dst_fd, ctx->src.closed, ctx->dst.closed);
max_fd = MAX(max_fd, MAX(ctx->fd, ctx->dst_fd));
ctx = ctx->next;
}
return max_fd;
}
static void
pxy_thr_print_info(pxy_thr_ctx_t *tctx)
{
log_finest_main_va("thr=%d, load=%zu", tctx->thridx, pxy_thr_get_load(tctx));
unsigned int idx = 1;
evutil_socket_t max_fd = 0;
time_t max_atime = 0;
time_t max_ctime = 0;
char *smsg = NULL;
if (tctx->conns || tctx->pending_ssl_conns) {
time_t now = time(NULL);
int conns_list = 1;
pxy_conn_ctx_t *ctx = tctx->conns;
if (!ctx) {
ctx = tctx->pending_ssl_conns;
conns_list = 0;
}
while (ctx) {
time_t atime = now - ctx->atime;
time_t ctime = now - ctx->ctime;
log_finest_main_va("PARENT CONN: thr=%d, id=%u, fd=%d, child_fd=%d, dst=%d, srvdst=%d, child_src=%d, child_dst=%d, p=%d-%d-%d c=%d-%d, ce=%d cc=%d, at=%lld ct=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d, pc=%d",
tctx->thridx, idx, ctx->fd, ctx->child_fd, ctx->dst_fd, ctx->srvdst_fd, ctx->child_src_fd, ctx->child_dst_fd,
ctx->src.closed, ctx->dst.closed, ctx->srvdst.closed, ctx->children ? ctx->children->src.closed : 0, ctx->children ? ctx->children->dst.closed : 0,
ctx->children ? 1:0, ctx->child_count, (long long)atime, (long long)ctime,
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid, ctx->sslctx ? ctx->sslctx->pending : 0);
// @attention Report idle connections only, i.e. the conns which have been idle since the last time we checked for expired conns
if (atime >= (time_t)tctx->thrmgr->global->expired_conn_check_period) {
if (asprintf(&smsg, "IDLE: thr=%d, id=%u, ce=%d cc=%d, at=%lld ct=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d, pc=%d\n",
tctx->thridx, idx, ctx->children ? 1:0, ctx->child_count, (long long)atime, (long long)ctime,
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid, ctx->sslctx ? ctx->sslctx->pending : 0) < 0) {
return;
}
if (log_conn(smsg) == -1) {
log_err_level_printf(LOG_WARNING, "Idle conn logging failed\n");
}
free(smsg);
smsg = NULL;
}
// child_src_fd and child_dst_fd fields are mostly for debugging purposes, used in debug printing parent conns.
// However, while an ssl child is closing, the children list may be empty, but child's ssl fd may be still open,
// hence we include those fields in this max comparisons too
max_fd = MAX(max_fd, MAX(ctx->fd, MAX(ctx->dst_fd, MAX(ctx->srvdst_fd, MAX(ctx->child_fd, MAX(ctx->child_src_fd, ctx->child_dst_fd))))));
max_atime = MAX(max_atime, atime);
max_ctime = MAX(max_ctime, ctime);
if (ctx->children) {
max_fd = pxy_thr_print_children(ctx->children,
#ifdef DEBUG_PROXY
idx,
#endif /* DEBUG_PROXY */
max_fd);
}
idx++;
if (conns_list) {
ctx = ctx->next;
if (!ctx) {
// Switch to pending ssl conns list
ctx = tctx->pending_ssl_conns;
conns_list = 0;
}
} else {
ctx = ctx->sslctx->next_pending;
}
}
}
log_finest_main_va("thr=%d, mld=%zu, mfd=%d, mat=%lld, mct=%lld, iib=%llu, iob=%llu, eib=%llu, eob=%llu, swm=%zu, uwm=%zu, to=%zu, err=%zu, pc=%llu, si=%u",
tctx->thridx, tctx->max_load, tctx->max_fd, (long long)max_atime, (long long)max_ctime, tctx->intif_in_bytes, tctx->intif_out_bytes, tctx->extif_in_bytes, tctx->extif_out_bytes,
tctx->set_watermarks, tctx->unset_watermarks, tctx->timedout_conns, tctx->errors, tctx->pending_ssl_conn_count, tctx->stats_id);
if (asprintf(&smsg, "STATS: thr=%d, mld=%zu, mfd=%d, mat=%lld, mct=%lld, iib=%llu, iob=%llu, eib=%llu, eob=%llu, swm=%zu, uwm=%zu, to=%zu, err=%zu, pc=%llu, si=%u\n",
tctx->thridx, tctx->max_load, tctx->max_fd, (long long)max_atime, (long long)max_ctime, tctx->intif_in_bytes, tctx->intif_out_bytes, tctx->extif_in_bytes, tctx->extif_out_bytes,
tctx->set_watermarks, tctx->unset_watermarks, tctx->timedout_conns, tctx->errors, tctx->pending_ssl_conn_count, tctx->stats_id) < 0) {
return;
}
if (log_stats(smsg) == -1) {
log_err_level_printf(LOG_WARNING, "Stats logging failed\n");
}
free(smsg);
tctx->stats_id++;
tctx->timedout_conns = 0;
tctx->errors = 0;
tctx->set_watermarks = 0;
tctx->unset_watermarks = 0;
tctx->intif_in_bytes = 0;
tctx->intif_out_bytes = 0;
tctx->extif_in_bytes = 0;
tctx->extif_out_bytes = 0;
// Reset these stats with the current values (do not reset to 0 directly, there may be active conns)
tctx->max_fd = max_fd;
tctx->max_load = pxy_thr_get_load(tctx);
}
/*
* Recurring timer event to prevent the event loops from exiting when
* they run out of events.
*/
static void
pxy_thr_timer_cb(UNUSED evutil_socket_t fd, UNUSED short what, UNUSED void *arg)
{
pxy_thr_ctx_t *ctx = arg;
log_finest_main_va("thr=%d, load=%zu, to=%u", ctx->thridx, pxy_thr_get_load(ctx), ctx->timeout_count);
pxy_conn_ctx_t *expired = NULL;
pxy_thr_get_expired_conns(ctx, &expired);
#ifdef DEBUG_PROXY
if (expired) {
time_t now = time(NULL);
#endif /* DEBUG_PROXY */
while (expired) {
pxy_conn_ctx_t *next = expired->next_expired;
log_fine_main_va("Delete timed out conn thr=%d, fd=%d, child_fd=%d, at=%lld ct=%lld",
expired->thr->thridx, expired->fd, expired->child_fd, (long long)(now - expired->atime), (long long)(now - expired->ctime));
// @attention Do not call the term function here, free the conn directly
pxy_conn_free(expired, 1);
ctx->timedout_conns++;
expired = next;
}
#ifdef DEBUG_PROXY
}
#endif /* DEBUG_PROXY */
// @attention Print thread info only if stats logging is enabled, if disabled debug logs are not printed either
if (ctx->thrmgr->global->statslog) {
ctx->timeout_count++;
if (ctx->timeout_count >= ctx->thrmgr->global->stats_period) {
ctx->timeout_count = 0;
pxy_thr_print_info(ctx);
}
}
}
/*
* Thread entry point; runs the event loop of the event base.
* Does not exit until the libevent loop is broken explicitly.
*/
void *
pxy_thr(void *arg)
{
pxy_thr_ctx_t *ctx = arg;
struct timeval timer_delay = {ctx->thrmgr->global->expired_conn_check_period, 0};
struct event *ev;
ev = event_new(ctx->evbase, -1, EV_PERSIST, pxy_thr_timer_cb, ctx);
if (!ev)
return NULL;
evtimer_add(ev, &timer_delay);
ctx->running = 1;
event_base_dispatch(ctx->evbase);
event_free(ev);
return NULL;
}
/* vim: set noet ft=c: */

@ -0,0 +1,98 @@
/*-
* SSLsplit - transparent SSL/TLS interception
* https://www.roe.ch/SSLsplit
*
* Copyright (c) 2017-2020, Soner Tari <sonertari@gmail.com>.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* 1. Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDER AND CONTRIBUTORS ``AS IS''
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef PXYTHR_H
#define PXYTHR_H
#include "attrib.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <event2/event.h>
#include <event2/dns.h>
#include <pthread.h>
typedef struct pxy_conn_ctx pxy_conn_ctx_t;
typedef struct pxy_thrmgr_ctx pxy_thrmgr_ctx_t;
typedef struct pxy_thr_ctx {
pthread_t thr;
int thridx;
pxy_thrmgr_ctx_t *thrmgr;
size_t load;
struct event_base *evbase;
struct evdns_base *dnsbase;
int running;
// Per-thread locking is necessary during connection setup and termination
// to prevent multithreading issues between thrmgr thread and conn handling threads
pthread_mutex_t mutex;
// Statistics
evutil_socket_t max_fd;
size_t max_load;
size_t timedout_conns;
size_t errors;
size_t set_watermarks;
size_t unset_watermarks;
long long unsigned int intif_in_bytes;
long long unsigned int intif_out_bytes;
long long unsigned int extif_in_bytes;
long long unsigned int extif_out_bytes;
// Each stats has an id, incremented on each stats print
unsigned short stats_id;
// Used to print statistics, compared against stats_period
unsigned int timeout_count;
// List of active connections on the thread
pxy_conn_ctx_t *conns;
// Per-thread sqlite stmt is necessary to prevent multithreading issues between threads
struct sqlite3_stmt *get_user;
// SSL conns wait for the first readcb to complete connection setup
// We keep track of conns at that stage using this list, to close them if they time out
pxy_conn_ctx_t *pending_ssl_conns;
long long unsigned int pending_ssl_conn_count;
} pxy_thr_ctx_t;
size_t pxy_thr_get_load(pxy_thr_ctx_t *) NONNULL(1);
void pxy_thr_inc_load(pxy_thr_ctx_t *) NONNULL(1);
void pxy_thr_dec_load(pxy_thr_ctx_t *) NONNULL(1);
void pxy_thr_add_pending_ssl_conn(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thr_remove_pending_ssl_conn(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thr_detach(pxy_conn_ctx_t *) NONNULL(1);
void *pxy_thr(void *);
#endif /* !PXYTHR_H */
/* vim: set noet ft=c: */

@ -35,272 +35,13 @@
#include <string.h>
#include <event2/bufferevent.h>
#include <pthread.h>
#include <assert.h>
#include <sys/param.h>
/*
* Proxy thread manager: manages the connection handling worker threads
* and the per-thread resources (i.e. event bases). The load is shared
* across num_cpu * 2 connection handling threads, using the number of
* currently assigned connections as the sole metric.
*
* The attach and detach functions are thread-safe.
*/
static void
pxy_thrmgr_get_thr_expired_conns(pxy_thr_ctx_t *tctx, pxy_conn_ctx_t **expired_conns)
{
*expired_conns = NULL;
if (tctx->conns) {
time_t now = time(NULL);
pxy_conn_ctx_t *ctx = tctx->conns;
while (ctx) {
time_t elapsed_time = now - ctx->atime;
if (elapsed_time > (time_t)tctx->thrmgr->global->conn_idle_timeout) {
ctx->next_expired = *expired_conns;
*expired_conns = ctx;
}
ctx = ctx->next;
}
ctx = tctx->pending_ssl_conns;
while (ctx) {
time_t elapsed_time = now - ctx->atime;
if (elapsed_time > (time_t)tctx->thrmgr->global->conn_idle_timeout) {
ctx->next_expired = *expired_conns;
*expired_conns = ctx;
}
ctx = ctx->sslctx->next_pending;
}
if (tctx->thrmgr->global->statslog) {
ctx = *expired_conns;
while (ctx) {
log_finest_main_va("thr=%d, fd=%d, child_fd=%d, time=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d, pc=%d",
ctx->thr->thridx, ctx->fd, ctx->child_fd, (long long)(now - ctx->atime),
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid, ctx->sslctx ? ctx->sslctx->pending : 0);
char *msg;
if (asprintf(&msg, "EXPIRED: thr=%d, time=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d\n",
ctx->thr->thridx, (long long)(now - ctx->atime),
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid) < 0) {
break;
}
if (log_conn(msg) == -1) {
log_err_level_printf(LOG_WARNING, "Expired conn logging failed\n");
}
free(msg);
ctx = ctx->next_expired;
}
}
}
}
static evutil_socket_t
pxy_thrmgr_print_children(pxy_conn_child_ctx_t *ctx,
#ifdef DEBUG_PROXY
unsigned int parent_idx,
#endif /* DEBUG_PROXY */
evutil_socket_t max_fd)
{
while (ctx) {
// @attention No need to log child stats
log_finest_main_va("CHILD CONN: thr=%d, id=%d, pid=%u, src=%d, dst=%d, c=%d-%d",
ctx->conn->thr->thridx, ctx->conn->child_count, parent_idx, ctx->fd, ctx->dst_fd, ctx->src.closed, ctx->dst.closed);
max_fd = MAX(max_fd, MAX(ctx->fd, ctx->dst_fd));
ctx = ctx->next;
}
return max_fd;
}
static void
pxy_thrmgr_print_thr_info(pxy_thr_ctx_t *tctx)
{
log_finest_main_va("thr=%d, load=%lu", tctx->thridx, tctx->load);
unsigned int idx = 1;
evutil_socket_t max_fd = 0;
time_t max_atime = 0;
time_t max_ctime = 0;
char *smsg = NULL;
if (tctx->conns || tctx->pending_ssl_conns) {
time_t now = time(NULL);
int conns_list = 1;
pxy_conn_ctx_t *ctx = tctx->conns;
if (!ctx) {
ctx = tctx->pending_ssl_conns;
conns_list = 0;
}
while (ctx) {
time_t atime = now - ctx->atime;
time_t ctime = now - ctx->ctime;
log_finest_main_va("PARENT CONN: thr=%d, id=%u, fd=%d, child_fd=%d, dst=%d, srvdst=%d, child_src=%d, child_dst=%d, p=%d-%d-%d c=%d-%d, ce=%d cc=%d, at=%lld ct=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d, pc=%d",
tctx->thridx, idx, ctx->fd, ctx->child_fd, ctx->dst_fd, ctx->srvdst_fd, ctx->child_src_fd, ctx->child_dst_fd,
ctx->src.closed, ctx->dst.closed, ctx->srvdst.closed, ctx->children ? ctx->children->src.closed : 0, ctx->children ? ctx->children->dst.closed : 0,
ctx->children ? 1:0, ctx->child_count, (long long)atime, (long long)ctime,
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid, ctx->sslctx ? ctx->sslctx->pending : 0);
// @attention Report idle connections only, i.e. the conns which have been idle since the last time we checked for expired conns
if (atime >= (time_t)tctx->thrmgr->global->expired_conn_check_period) {
if (asprintf(&smsg, "IDLE: thr=%d, id=%u, ce=%d cc=%d, at=%lld ct=%lld, src_addr=%s:%s, dst_addr=%s:%s, user=%s, valid=%d, pc=%d\n",
tctx->thridx, idx, ctx->children ? 1:0, ctx->child_count, (long long)atime, (long long)ctime,
STRORDASH(ctx->srchost_str), STRORDASH(ctx->srcport_str), STRORDASH(ctx->dsthost_str), STRORDASH(ctx->dstport_str),
STRORDASH(ctx->user), ctx->protoctx->is_valid, ctx->sslctx ? ctx->sslctx->pending : 0) < 0) {
return;
}
if (log_conn(smsg) == -1) {
log_err_level_printf(LOG_WARNING, "Idle conn logging failed\n");
}
free(smsg);
smsg = NULL;
}
// child_src_fd and child_dst_fd fields are mostly for debugging purposes, used in debug printing parent conns.
// However, while an ssl child is closing, the children list may be empty, but child's ssl fd may be still open,
// hence we include those fields in this max comparisons too
max_fd = MAX(max_fd, MAX(ctx->fd, MAX(ctx->dst_fd, MAX(ctx->srvdst_fd, MAX(ctx->child_fd, MAX(ctx->child_src_fd, ctx->child_dst_fd))))));
max_atime = MAX(max_atime, atime);
max_ctime = MAX(max_ctime, ctime);
if (ctx->children) {
max_fd = pxy_thrmgr_print_children(ctx->children,
#ifdef DEBUG_PROXY
idx,
#endif /* DEBUG_PROXY */
max_fd);
}
idx++;
if (conns_list) {
ctx = ctx->next;
if (!ctx) {
// Switch to pending ssl conns list
ctx = tctx->pending_ssl_conns;
conns_list = 0;
}
} else {
ctx = ctx->sslctx->next_pending;
}
}
}
log_finest_main_va("thr=%d, mld=%zu, mfd=%d, mat=%lld, mct=%lld, iib=%llu, iob=%llu, eib=%llu, eob=%llu, swm=%zu, uwm=%zu, to=%zu, err=%zu, pc=%llu, si=%u",
tctx->thridx, tctx->max_load, tctx->max_fd, (long long)max_atime, (long long)max_ctime, tctx->intif_in_bytes, tctx->intif_out_bytes, tctx->extif_in_bytes, tctx->extif_out_bytes,
tctx->set_watermarks, tctx->unset_watermarks, tctx->timedout_conns, tctx->errors, tctx->pending_ssl_conn_count, tctx->stats_id);
if (asprintf(&smsg, "STATS: thr=%d, mld=%zu, mfd=%d, mat=%lld, mct=%lld, iib=%llu, iob=%llu, eib=%llu, eob=%llu, swm=%zu, uwm=%zu, to=%zu, err=%zu, pc=%llu, si=%u\n",
tctx->thridx, tctx->max_load, tctx->max_fd, (long long)max_atime, (long long)max_ctime, tctx->intif_in_bytes, tctx->intif_out_bytes, tctx->extif_in_bytes, tctx->extif_out_bytes,
tctx->set_watermarks, tctx->unset_watermarks, tctx->timedout_conns, tctx->errors, tctx->pending_ssl_conn_count, tctx->stats_id) < 0) {
return;
}
if (log_stats(smsg) == -1) {
log_err_level_printf(LOG_WARNING, "Stats logging failed\n");
}
free(smsg);
tctx->stats_id++;
tctx->timedout_conns = 0;
tctx->errors = 0;
tctx->set_watermarks = 0;
tctx->unset_watermarks = 0;
tctx->intif_in_bytes = 0;
tctx->intif_out_bytes = 0;
tctx->extif_in_bytes = 0;
tctx->extif_out_bytes = 0;
// Reset these stats with the current values (do not reset to 0 directly, there may be active conns)
tctx->max_fd = max_fd;
tctx->max_load = tctx->load;
}
/*
* Recurring timer event to prevent the event loops from exiting when
* they run out of events.
*/
static void
pxy_thrmgr_timer_cb(UNUSED evutil_socket_t fd, UNUSED short what, UNUSED void *arg)
{
pxy_thr_ctx_t *ctx = arg;
pthread_mutex_lock(&ctx->mutex);
log_finest_main_va("thr=%d, load=%lu, to=%u", ctx->thridx, ctx->load, ctx->timeout_count);
pxy_conn_ctx_t *expired = NULL;
pxy_thrmgr_get_thr_expired_conns(ctx, &expired);
#ifdef DEBUG_PROXY
if (expired) {
time_t now = time(NULL);
#endif /* DEBUG_PROXY */
while (expired) {
pxy_conn_ctx_t *next = expired->next_expired;
log_fine_main_va("Delete timed out conn thr=%d, fd=%d, child_fd=%d, at=%lld ct=%lld",
expired->thr->thridx, expired->fd, expired->child_fd, (long long)(now - expired->atime), (long long)(now - expired->ctime));
// We have already locked the thr mutex above, do not lock again while detaching, otherwise we get signal 6 crash
// When detach_unlocked is set, *_ctx_free() functions call non-thread-safe detach functions
expired->thr_locked = 1;
// @attention Do not call the term function here, free the conn directly
pxy_conn_free(expired, 1);
ctx->timedout_conns++;
expired = next;
}
#ifdef DEBUG_PROXY
}
#endif /* DEBUG_PROXY */
// @attention Print thread info only if stats logging is enabled, if disabled debug logs are not printed either
if (ctx->thrmgr->global->statslog) {
ctx->timeout_count++;
if (ctx->timeout_count >= ctx->thrmgr->global->stats_period) {
ctx->timeout_count = 0;
pxy_thrmgr_print_thr_info(ctx);
}
}
pthread_mutex_unlock(&ctx->mutex);
}
/*
* Thread entry point; runs the event loop of the event base.
* Does not exit until the libevent loop is broken explicitly.
*/
static void *
pxy_thrmgr_thr(void *arg)
{
pxy_thr_ctx_t *ctx = arg;
struct timeval timer_delay = {ctx->thrmgr->global->expired_conn_check_period, 0};
struct event *ev;
ev = event_new(ctx->evbase, -1, EV_PERSIST, pxy_thrmgr_timer_cb, ctx);
if (!ev)
return NULL;
evtimer_add(ev, &timer_delay);
ctx->running = 1;
event_base_dispatch(ctx->evbase);
event_free(ev);
return NULL;
}
/*
* Create new thread manager but do not start any threads yet.
@ -381,8 +122,7 @@ pxy_thrmgr_run(pxy_thrmgr_ctx_t *ctx)
ctx->num_thr);
for (idx = 0; idx < ctx->num_thr; idx++) {
if (pthread_create(&ctx->thr[idx]->thr, NULL,
pxy_thrmgr_thr, ctx->thr[idx]))
if (pthread_create(&ctx->thr[idx]->thr, NULL, pxy_thr, ctx->thr[idx]))
goto leave_thr;
while (!ctx->thr[idx]->running) {
sched_yield();
@ -459,105 +199,12 @@ pxy_thrmgr_free(pxy_thrmgr_ctx_t *ctx)
free(ctx);
}
void
pxy_thrmgr_add_pending_ssl_conn(pxy_conn_ctx_t *ctx)
{
pthread_mutex_lock(&ctx->thr->mutex);
if (!ctx->sslctx->pending) {
log_finest("Adding conn");
ctx->sslctx->pending = 1;
ctx->thr->pending_ssl_conn_count++;
ctx->sslctx->next_pending = ctx->thr->pending_ssl_conns;
ctx->thr->pending_ssl_conns = ctx;
}
pthread_mutex_unlock(&ctx->thr->mutex);
}
static void NONNULL(1)
pxy_thrmgr_remove_pending_ssl_conn_unlocked(pxy_conn_ctx_t *ctx)
{
if (ctx->sslctx && ctx->sslctx->pending) {
log_finest("Removing conn");
// Thr pending_ssl_conns list cannot be empty, if the sslctx->pending flag of a conn is set
assert(ctx->thr->pending_ssl_conns != NULL);
ctx->sslctx->pending = 0;
ctx->thr->pending_ssl_conn_count--;
// @attention We may get multiple conns with the same fd combinations, so fds cannot uniquely define a conn; hence the need for unique ids.
if (ctx->id == ctx->thr->pending_ssl_conns->id) {
ctx->thr->pending_ssl_conns = ctx->thr->pending_ssl_conns->sslctx->next_pending;
return;
} else {
pxy_conn_ctx_t *current = ctx->thr->pending_ssl_conns->sslctx->next_pending;
pxy_conn_ctx_t *previous = ctx->thr->pending_ssl_conns;
while (current != NULL && previous != NULL) {
if (ctx->id == current->id) {
previous->sslctx->next_pending = current->sslctx->next_pending;
return;
}
previous = current;
current = current->sslctx->next_pending;
}
// This should never happen
log_err_level_printf(LOG_CRIT, "Cannot find conn in thrmgr pending_conns\n");
log_fine("Cannot find conn in thrmgr pending_conns");
assert(0);
}
}
}
void
pxy_thrmgr_remove_pending_ssl_conn(pxy_conn_ctx_t *ctx)
{
pthread_mutex_lock(&ctx->thr->mutex);
pxy_thrmgr_remove_pending_ssl_conn_unlocked(ctx);
pthread_mutex_unlock(&ctx->thr->mutex);
}
static void NONNULL(1)
pxy_thrmgr_remove_conn_unlocked(pxy_conn_ctx_t *ctx)
{
assert(ctx != NULL);
assert(ctx->children == NULL);
// Thr conns list cannot be empty
assert(ctx->thr->conns != NULL);
log_finest("Removing conn");
// We increment thr load in pxy_conn_init() only (for parent conns)
ctx->thr->load--;
// No need to reset the ctx->in_thr_conns flag, as we free the ctx right after calling this function
// @attention We may get multiple conns with the same fd combinations, so fds cannot uniquely identify a conn; hence the need for unique ids.
if (ctx->id == ctx->thr->conns->id) {
ctx->thr->conns = ctx->thr->conns->next;
return;
} else {
pxy_conn_ctx_t *current = ctx->thr->conns->next;
pxy_conn_ctx_t *previous = ctx->thr->conns;
while (current != NULL && previous != NULL) {
if (ctx->id == current->id) {
previous->next = current->next;
return;
}
previous = current;
current = current->next;
}
// This should never happen
log_err_level_printf(LOG_CRIT, "Cannot find conn in thr conns\n");
log_fine("Cannot find conn in thr conns");
assert(0);
}
}
/*
* Attach a new connection to a thread. Chooses the thread with the fewest
* currently active connections, returns the appropriate event bases.
* No need to be so accurate about balancing thread loads, so uses
* thread-level mutexes, instead of a thrmgr level mutex.
* Returns the index of the chosen thread (for passing to _detach later).
* Returns the index of the chosen thread.
* This function cannot fail.
*/
void
@ -565,28 +212,25 @@ pxy_thrmgr_attach(pxy_conn_ctx_t *ctx)
{
log_finest("ENTER");
int thridx = 0;
size_t minload;
pxy_thrmgr_ctx_t *tmctx = ctx->thrmgr;
pthread_mutex_lock(&tmctx->thr[0]->mutex);
minload = tmctx->thr[0]->load;
pthread_mutex_unlock(&tmctx->thr[0]->mutex);
size_t minload = pxy_thr_get_load(tmctx->thr[0]);
#ifdef DEBUG_THREAD
log_dbg_printf("===> Proxy connection handler thread status:\n"
"thr[0]: %zu\n", minload);
log_dbg_printf("===> Proxy connection handler thread status:\nthr[0]: %zu\n", minload);
#endif /* DEBUG_THREAD */
int thridx = 0;
for (int idx = 1; idx < tmctx->num_thr; idx++) {
pthread_mutex_lock(&tmctx->thr[idx]->mutex);
#ifdef DEBUG_THREAD
log_dbg_printf("thr[%d]: %zu\n", idx, tmctx->thr[idx]->load);
#endif /* DEBUG_THREAD */
if (minload > tmctx->thr[idx]->load) {
minload = tmctx->thr[idx]->load;
size_t thrload = pxy_thr_get_load(tmctx->thr[idx]);
if (minload > thrload) {
minload = thrload;
thridx = idx;
}
pthread_mutex_unlock(&tmctx->thr[idx]->mutex);
#ifdef DEBUG_THREAD
log_dbg_printf("thr[%d]: %zu\n", idx, thrload);
#endif /* DEBUG_THREAD */
}
ctx->thr = tmctx->thr[thridx];
@ -598,48 +242,4 @@ pxy_thrmgr_attach(pxy_conn_ctx_t *ctx)
#endif /* DEBUG_THREAD */
}
void
pxy_thrmgr_attach_child(pxy_conn_ctx_t *ctx)
{
log_finest("ENTER");
pthread_mutex_lock(&ctx->thr->mutex);
ctx->thr->load++;
pthread_mutex_unlock(&ctx->thr->mutex);
}
/*
* Detach a connection from a thread by index.
* This function cannot fail.
*/
void
pxy_thrmgr_detach_unlocked(pxy_conn_ctx_t *ctx)
{
log_finest("ENTER");
pxy_thrmgr_remove_pending_ssl_conn_unlocked(ctx);
pxy_thrmgr_remove_conn_unlocked(ctx);
}
void
pxy_thrmgr_detach(pxy_conn_ctx_t *ctx)
{
pthread_mutex_lock(&ctx->thr->mutex);
pxy_thrmgr_detach_unlocked(ctx);
pthread_mutex_unlock(&ctx->thr->mutex);
}
void
pxy_thrmgr_detach_child_unlocked(pxy_conn_ctx_t *ctx)
{
log_finest("ENTER");
ctx->thr->load--;
}
void
pxy_thrmgr_detach_child(pxy_conn_ctx_t *ctx)
{
pthread_mutex_lock(&ctx->thr->mutex);
pxy_thrmgr_detach_child_unlocked(ctx);
pthread_mutex_unlock(&ctx->thr->mutex);
}
/* vim: set noet ft=c: */

@ -32,61 +32,11 @@
#include "opts.h"
#include "attrib.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <event2/event.h>
#include <event2/dns.h>
#include <pthread.h>
#include "pxythr.h"
extern int descriptor_table_size;
#define FD_RESERVE 10
typedef struct pxy_conn_ctx pxy_conn_ctx_t;
typedef struct pxy_thrmgr_ctx pxy_thrmgr_ctx_t;
typedef struct pxy_thr_ctx {
pthread_t thr;
int thridx;
pxy_thrmgr_ctx_t *thrmgr;
size_t load;
struct event_base *evbase;
struct evdns_base *dnsbase;
int running;
// Per-thread locking is necessary during connection setup and termination
// to prevent multithreading issues between thrmgr thread and conn handling threads
pthread_mutex_t mutex;
// Statistics
evutil_socket_t max_fd;
size_t max_load;
size_t timedout_conns;
size_t errors;
size_t set_watermarks;
size_t unset_watermarks;
long long unsigned int intif_in_bytes;
long long unsigned int intif_out_bytes;
long long unsigned int extif_in_bytes;
long long unsigned int extif_out_bytes;
// Each stats has an id, incremented on each stats print
unsigned short stats_id;
// Used to print statistics, compared against stats_period
unsigned int timeout_count;
// List of active connections on the thread
pxy_conn_ctx_t *conns;
// Per-thread sqlite stmt is necessary to prevent multithreading issues between threads
struct sqlite3_stmt *get_user;
// SSL conns wait for the first readcb to complete connection setup
// We keep track of conns at that stage using this list, to close them if they time out
pxy_conn_ctx_t *pending_ssl_conns;
long long unsigned int pending_ssl_conn_count;
} pxy_thr_ctx_t;
struct pxy_thrmgr_ctx {
int num_thr;
global_t *global;
@ -100,15 +50,7 @@ pxy_thrmgr_ctx_t * pxy_thrmgr_new(global_t *) MALLOC;
int pxy_thrmgr_run(pxy_thrmgr_ctx_t *) NONNULL(1) WUNRES;
void pxy_thrmgr_free(pxy_thrmgr_ctx_t *) NONNULL(1);
void pxy_thrmgr_add_pending_ssl_conn(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_remove_pending_ssl_conn(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_attach(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_attach_child(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_detach_unlocked(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_detach(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_detach_child_unlocked(pxy_conn_ctx_t *) NONNULL(1);
void pxy_thrmgr_detach_child(pxy_conn_ctx_t *) NONNULL(1);
#endif /* !PXYTHRMGR_H */

Loading…
Cancel
Save