MEDIUM: threads/server: Add a lock per server and atomically update server vars
The server's lock is use, among other things, to lock acces to the active
connection list of a server.
diff --git a/src/backend.c b/src/backend.c
index 5e51f39..d17635f 100644
--- a/src/backend.c
+++ b/src/backend.c
@@ -702,7 +702,7 @@
}
else if (srv != prev_srv) {
HA_ATOMIC_ADD(&s->be->be_counters.cum_lbconn, 1);
- srv->counters.cum_lbconn++;
+ HA_ATOMIC_ADD(&srv->counters.cum_lbconn, 1);
}
s->target = &srv->obj_type;
}
@@ -879,10 +879,10 @@
s->txn->flags |= TX_CK_DOWN;
}
s->flags |= SF_REDISP;
- prev_srv->counters.redispatches++;
+ HA_ATOMIC_ADD(&prev_srv->counters.redispatches, 1);
HA_ATOMIC_ADD(&s->be->be_counters.redispatches, 1);
} else {
- prev_srv->counters.retries++;
+ HA_ATOMIC_ADD(&prev_srv->counters.retries, 1);
HA_ATOMIC_ADD(&s->be->be_counters.retries, 1);
}
}
@@ -1197,10 +1197,11 @@
s->si[1].exp = tick_add_ifset(now_ms, s->be->timeout.connect);
if (srv) {
+ int count;
+
s->flags |= SF_CURR_SESS;
- srv->cur_sess++;
- if (srv->cur_sess > srv->counters.cur_sess_max)
- srv->counters.cur_sess_max = srv->cur_sess;
+ count = HA_ATOMIC_ADD(&srv->cur_sess, 1);
+ HA_ATOMIC_UPDATE_MAX(&srv->counters.cur_sess_max, count);
if (s->be->lbprm.server_take_conn)
s->be->lbprm.server_take_conn(srv);
@@ -1277,7 +1278,7 @@
s->si[1].err_type = SI_ET_QUEUE_ERR;
}
- srv->counters.failed_conns++;
+ HA_ATOMIC_ADD(&srv->counters.failed_conns, 1);
HA_ATOMIC_ADD(&s->be->be_counters.failed_conns, 1);
return 1;
@@ -1307,7 +1308,7 @@
if (srv)
srv_set_sess_last(srv);
if (srv)
- srv->counters.failed_conns++;
+ HA_ATOMIC_ADD(&srv->counters.failed_conns, 1);
HA_ATOMIC_ADD(&s->be->be_counters.failed_conns, 1);
/* release other streams waiting for this server */
diff --git a/src/checks.c b/src/checks.c
index 3d60237..ed99bb5 100644
--- a/src/checks.c
+++ b/src/checks.c
@@ -242,7 +242,7 @@
if ((!(check->state & CHK_ST_AGENT) ||
(check->status >= HCHK_STATUS_L57DATA)) &&
(check->health >= check->rise)) {
- s->counters.failed_checks++;
+ HA_ATOMIC_ADD(&s->counters.failed_checks, 1);
report = 1;
check->health--;
if (check->health < check->rise)
@@ -410,7 +410,7 @@
return;
}
- s->consecutive_errors++;
+ HA_ATOMIC_ADD(&s->consecutive_errors, 1);
if (s->consecutive_errors < s->consecutive_errors_limit)
return;
@@ -449,7 +449,7 @@
}
s->consecutive_errors = 0;
- s->counters.failed_hana++;
+ HA_ATOMIC_ADD(&s->counters.failed_hana, 1);
if (s->check.fastinter) {
expire = tick_add(now_ms, MS_TO_TICKS(s->check.fastinter));
diff --git a/src/haproxy.c b/src/haproxy.c
index f5eebbe..f905473 100644
--- a/src/haproxy.c
+++ b/src/haproxy.c
@@ -2080,6 +2080,7 @@
if (xprt_get(XPRT_SSL) && xprt_get(XPRT_SSL)->destroy_srv)
xprt_get(XPRT_SSL)->destroy_srv(s);
}
+ SPIN_DESTROY(&s->lock);
free(s);
s = s_next;
}/* end while(s) */
diff --git a/src/proto_http.c b/src/proto_http.c
index cf1cc7a..f60f8ed 100644
--- a/src/proto_http.c
+++ b/src/proto_http.c
@@ -4277,7 +4277,7 @@
if (objt_server(s->target)) {
if (s->flags & SF_CURR_SESS) {
s->flags &= ~SF_CURR_SESS;
- objt_server(s->target)->cur_sess--;
+ HA_ATOMIC_SUB(&objt_server(s->target)->cur_sess, 1);
}
if (may_dequeue_tasks(objt_server(s->target), be))
process_srv_queue(objt_server(s->target));
@@ -4605,7 +4605,7 @@
txn->rsp.msg_state = HTTP_MSG_ERROR;
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.cli_aborts, 1);
}
goto wait_other_side;
}
@@ -4881,7 +4881,7 @@
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.cli_aborts, 1);
goto return_bad_req_stats_ok;
}
@@ -4953,7 +4953,7 @@
HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.srv_aborts, 1);
if (!(s->flags & SF_ERR_MASK))
s->flags |= SF_ERR_SRVCL;
@@ -5079,7 +5079,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (objt_server(s->target)) {
- objt_server(s->target)->counters.failed_resp++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_resp, 1);
health_adjust(objt_server(s->target), HANA_STATUS_HTTP_HDRRSP);
}
abort_response:
@@ -5114,7 +5114,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (objt_server(s->target)) {
- objt_server(s->target)->counters.failed_resp++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_resp, 1);
health_adjust(objt_server(s->target), HANA_STATUS_HTTP_READ_ERROR);
}
@@ -5139,7 +5139,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (objt_server(s->target)) {
- objt_server(s->target)->counters.failed_resp++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_resp, 1);
health_adjust(objt_server(s->target), HANA_STATUS_HTTP_READ_TIMEOUT);
}
@@ -5162,7 +5162,7 @@
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.cli_aborts, 1);
rep->analysers &= AN_RES_FLT_END;
channel_auto_close(rep);
@@ -5189,7 +5189,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (objt_server(s->target)) {
- objt_server(s->target)->counters.failed_resp++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_resp, 1);
health_adjust(objt_server(s->target), HANA_STATUS_HTTP_BROKEN_PIPE);
}
@@ -5255,7 +5255,7 @@
stream_inc_http_err_ctr(s);
if (objt_server(s->target))
- objt_server(s->target)->counters.p.http.rsp[n]++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.p.http.rsp[n], 1);
/* RFC7230#2.6 has enforced the format of the HTTP version string to be
* exactly one digit "." one digit. This check may be disabled using
@@ -5673,7 +5673,7 @@
if (apply_filters_to_response(s, rep, rule_set) < 0) {
return_bad_resp:
if (objt_server(s->target)) {
- objt_server(s->target)->counters.failed_resp++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_resp, 1);
health_adjust(objt_server(s->target), HANA_STATUS_HTTP_RSP);
}
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
@@ -5695,7 +5695,7 @@
/* has the response been denied ? */
if (txn->flags & TX_SVDENY) {
if (objt_server(s->target))
- objt_server(s->target)->counters.failed_secu++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_secu, 1);
HA_ATOMIC_ADD(&s->be->be_counters.denied_resp, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.denied_resp, 1);
@@ -5845,7 +5845,7 @@
* the 'checkcache' option, and send an alert.
*/
if (objt_server(s->target))
- objt_server(s->target)->counters.failed_secu++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_secu, 1);
HA_ATOMIC_ADD(&s->be->be_counters.denied_resp, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.denied_resp, 1);
@@ -6038,7 +6038,7 @@
s->flags |= SF_ERR_SRVCL;
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.srv_aborts, 1);
goto return_bad_res_stats_ok;
}
}
@@ -6076,7 +6076,7 @@
return_bad_res: /* let's centralize all bad responses */
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.failed_resp++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_resp, 1);
return_bad_res_stats_ok:
txn->rsp.err_state = txn->rsp.msg_state;
@@ -6105,7 +6105,7 @@
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
if (objt_server(s->target))
- objt_server(s->target)->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.cli_aborts, 1);
if (!(s->flags & SF_ERR_MASK))
s->flags |= SF_ERR_CLICL;
diff --git a/src/queue.c b/src/queue.c
index ba2a9f7..95b8eda 100644
--- a/src/queue.c
+++ b/src/queue.c
@@ -122,7 +122,7 @@
strm->flags |= SF_ASSIGNED;
strm->target = &srv->obj_type;
stream_add_srv_conn(strm, srv);
- srv->served++;
+ HA_ATOMIC_ADD(&srv->served, 1);
HA_ATOMIC_ADD(&srv->proxy->served, 1);
if (px->lbprm.server_take_conn)
px->lbprm.server_take_conn(srv);
diff --git a/src/server.c b/src/server.c
index 4c70ac4..b6986a9 100644
--- a/src/server.c
+++ b/src/server.c
@@ -761,9 +761,11 @@
{
struct stream *stream, *stream_bck;
+ SPIN_LOCK(SERVER_LOCK, &srv->lock);
list_for_each_entry_safe(stream, stream_bck, &srv->actconns, by_srv)
if (stream->srv_conn == srv)
stream_shutdown(stream, why);
+ SPIN_UNLOCK(SERVER_LOCK, &srv->lock);
}
/* Shutdown all connections of all backup servers of a proxy. The caller must
@@ -1029,7 +1031,6 @@
/* Register changes to be applied asynchronously */
if (LIST_ISEMPTY(&s->update_status))
LIST_ADDQ(&updated_servers, &s->update_status);
-
/* stop going down if the equivalent flag is still present (forced or inherited) */
if (((mode & SRV_ADMF_MAINT) && (s->next_admin & SRV_ADMF_MAINT)) ||
((mode & SRV_ADMF_DRAIN) && (s->next_admin & SRV_ADMF_DRAIN)))
@@ -2021,6 +2022,7 @@
/* Copy default server settings to new server settings. */
srv_settings_cpy(newsrv, &curproxy->defsrv, 0);
+ SPIN_INIT(&newsrv->lock);
cur_arg++;
} else {
newsrv = &curproxy->defsrv;
diff --git a/src/stats.c b/src/stats.c
index 66e9be4..e026f36 100644
--- a/src/stats.c
+++ b/src/stats.c
@@ -2880,9 +2880,11 @@
if (px->state != PR_STSTOPPED) {
struct stream *sess, *sess_bck;
+ SPIN_LOCK(SERVER_LOCK, &sv->lock);
list_for_each_entry_safe(sess, sess_bck, &sv->actconns, by_srv)
if (sess->srv_conn == sv)
stream_shutdown(sess, SF_ERR_KILLED);
+ SPIN_UNLOCK(SERVER_LOCK, &sv->lock);
altered_servers++;
total_servers++;
diff --git a/src/stream.c b/src/stream.c
index 2703f41..7bfe786 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -298,7 +298,7 @@
if (objt_server(s->target)) { /* there may be requests left pending in queue */
if (s->flags & SF_CURR_SESS) {
s->flags &= ~SF_CURR_SESS;
- objt_server(s->target)->cur_sess--;
+ HA_ATOMIC_SUB(&objt_server(s->target)->cur_sess, 1);
}
if (may_dequeue_tasks(objt_server(s->target), s->be))
process_srv_queue(objt_server(s->target));
@@ -474,7 +474,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.bytes_in, bytes);
if (objt_server(s->target))
- objt_server(s->target)->counters.bytes_in += bytes;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.bytes_in, bytes);
if (sess->listener && sess->listener->counters)
HA_ATOMIC_ADD(&sess->listener->counters->bytes_in, bytes);
@@ -510,7 +510,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.bytes_out, bytes);
if (objt_server(s->target))
- objt_server(s->target)->counters.bytes_out += bytes;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.bytes_out, bytes);
if (sess->listener && sess->listener->counters)
HA_ATOMIC_ADD(&sess->listener->counters->bytes_out, bytes);
@@ -630,7 +630,7 @@
if (s->flags & SF_CURR_SESS) {
s->flags &= ~SF_CURR_SESS;
- objt_server(s->target)->cur_sess--;
+ HA_ATOMIC_SUB(&objt_server(s->target)->cur_sess, 1);
}
if ((si->flags & SI_FL_ERR) &&
@@ -663,7 +663,7 @@
}
if (objt_server(s->target))
- objt_server(s->target)->counters.failed_conns++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.failed_conns, 1);
HA_ATOMIC_ADD(&s->be->be_counters.failed_conns, 1);
sess_change_server(s, NULL);
if (may_dequeue_tasks(objt_server(s->target), s->be))
@@ -711,7 +711,7 @@
si->state = SI_ST_REQ;
} else {
if (objt_server(s->target))
- objt_server(s->target)->counters.retries++;
+ HA_ATOMIC_ADD(&objt_server(s->target)->counters.retries, 1);
HA_ATOMIC_ADD(&s->be->be_counters.retries, 1);
si->state = SI_ST_ASS;
}
@@ -860,7 +860,7 @@
if (srv)
srv_set_sess_last(srv);
if (srv)
- srv->counters.failed_conns++;
+ HA_ATOMIC_ADD(&srv->counters.failed_conns, 1);
HA_ATOMIC_ADD(&s->be->be_counters.failed_conns, 1);
/* release other streams waiting for this server */
@@ -916,7 +916,7 @@
si->exp = TICK_ETERNITY;
s->logs.t_queue = tv_ms_elapsed(&s->logs.tv_accept, &now);
if (srv)
- srv->counters.failed_conns++;
+ HA_ATOMIC_ADD(&srv->counters.failed_conns, 1);
HA_ATOMIC_ADD(&s->be->be_counters.failed_conns, 1);
si_shutr(si);
si_shutw(si);
@@ -1703,7 +1703,7 @@
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
if (srv)
- srv->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
if (!(s->flags & SF_ERR_MASK))
s->flags |= SF_ERR_CLICL;
if (!(s->flags & SF_FINST_MASK))
@@ -1719,12 +1719,12 @@
stream_int_report_error(si_b);
HA_ATOMIC_ADD(&s->be->be_counters.failed_resp, 1);
if (srv)
- srv->counters.failed_resp++;
+ HA_ATOMIC_ADD(&srv->counters.failed_resp, 1);
if (!(req->analysers) && !(res->analysers)) {
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
if (srv)
- srv->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
if (!(s->flags & SF_ERR_MASK))
s->flags |= SF_ERR_SRVCL;
if (!(s->flags & SF_FINST_MASK))
@@ -1782,7 +1782,7 @@
if (srv) {
if (s->flags & SF_CURR_SESS) {
s->flags &= ~SF_CURR_SESS;
- srv->cur_sess--;
+ HA_ATOMIC_SUB(&srv->cur_sess, 1);
}
sess_change_server(s, NULL);
if (may_dequeue_tasks(srv, s->be))
@@ -1980,28 +1980,28 @@
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
if (srv)
- srv->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
s->flags |= SF_ERR_CLICL;
}
else if (req->flags & CF_READ_TIMEOUT) {
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
if (srv)
- srv->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
s->flags |= SF_ERR_CLITO;
}
else if (req->flags & CF_WRITE_ERROR) {
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
if (srv)
- srv->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
s->flags |= SF_ERR_SRVCL;
}
else {
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
if (srv)
- srv->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
s->flags |= SF_ERR_SRVTO;
}
sess_set_term_flags(s);
@@ -2013,28 +2013,28 @@
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
if (srv)
- srv->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
s->flags |= SF_ERR_SRVCL;
}
else if (res->flags & CF_READ_TIMEOUT) {
HA_ATOMIC_ADD(&s->be->be_counters.srv_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.srv_aborts, 1);
if (srv)
- srv->counters.srv_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.srv_aborts, 1);
s->flags |= SF_ERR_SRVTO;
}
else if (res->flags & CF_WRITE_ERROR) {
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
if (srv)
- srv->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
s->flags |= SF_ERR_CLICL;
}
else {
HA_ATOMIC_ADD(&s->be->be_counters.cli_aborts, 1);
HA_ATOMIC_ADD(&sess->fe->fe_counters.cli_aborts, 1);
if (srv)
- srv->counters.cli_aborts++;
+ HA_ATOMIC_ADD(&srv->counters.cli_aborts, 1);
s->flags |= SF_ERR_CLITO;
}
sess_set_term_flags(s);
@@ -2532,7 +2532,7 @@
return;
if (sess->srv_conn) {
- sess->srv_conn->served--;
+ HA_ATOMIC_SUB(&sess->srv_conn->served, 1);
HA_ATOMIC_SUB(&sess->srv_conn->proxy->served, 1);
if (sess->srv_conn->proxy->lbprm.server_drop_conn)
sess->srv_conn->proxy->lbprm.server_drop_conn(sess->srv_conn);
@@ -2540,7 +2540,7 @@
}
if (newsrv) {
- newsrv->served++;
+ HA_ATOMIC_ADD(&newsrv->served, 1);
HA_ATOMIC_ADD(&newsrv->proxy->served, 1);
if (newsrv->proxy->lbprm.server_take_conn)
newsrv->proxy->lbprm.server_take_conn(newsrv);
@@ -3263,9 +3263,11 @@
return 1;
/* kill all the stream that are on this server */
+ SPIN_LOCK(SERVER_LOCK, &sv->lock);
list_for_each_entry_safe(strm, strm_bck, &sv->actconns, by_srv)
if (strm->srv_conn == sv)
stream_shutdown(strm, SF_ERR_KILLED);
+ SPIN_UNLOCK(SERVER_LOCK, &sv->lock);
return 1;
}