MEDIUM: add set-priority-class and set-priority-offset

This adds the set-priority-class and set-priority-offset actions to
http-request and tcp-request content. At this point they are not used
yet, which is the purpose of the next commit, but all the logic to
set and clear the values is there.

doc/configuration.txt

@@ -3917,6 +3917,7 @@ http-request { allow | auth [realm <realm>] | redirect <rule> | reject |
               replace-value <name> <match-regex> <replace-fmt> |
               set-method <fmt> | set-path <fmt> | set-query <fmt> |
               set-uri <fmt> | set-tos <tos> | set-mark <mark> |
+              set-priority-class <expr> | set-priority-offset <expr>
               add-acl(<file name>) <key fmt> |
               del-acl(<file name>) <key fmt> |
               del-map(<file name>) <key fmt> |
@@ -4113,6 +4114,24 @@ http-request { allow | auth [realm <realm>] | redirect <rule> | reject |
       downloads). This works on Linux kernels 2.6.32 and above and requires
       admin privileges.
 
+    - "set-priority-class" is used to set the queue priority class of the
+      current request. The value must be a sample expression which converts to
+      an integer in the range -2047..2047. Results outside this range will be
+      truncated. The priority class determines the order in which queued
+      requests are processed. Lower values have higher priority.
+
+    - "set-priority-offset" is used to set the queue priority timestamp offset
+      of the current request. The value must be a sample expression which
+      converts to an integer in the range -524287..524287. Results outside this
+      range will be truncated. When a request is queued, it is ordered first by
+      the priority class, then by the current timestamp adjusted by the given
+      offset in milliseconds. Lower values have higher priority.
+      Note that the resulting timestamp is is only tracked with enough precision
+      for 524,287ms (8m44s287ms). If the request is queued long enough to where
+      the adjusted timestamp exceeds this value, it will be misidentified as
+      highest priority. Thus it is important to set "timeout queue" to a value,
+      where when combined with the offset, does not exceed this limit.
+
     - "add-acl" is used to add a new entry into an ACL. The ACL must be loaded
       from a file (even a dummy empty file). The file name of the ACL to be
       updated is passed between parentheses. It takes one argument: <key fmt>,
@@ -9452,6 +9471,7 @@ tcp-request content <action> [{if | unless} <condition>]
     - accept : the request is accepted
     - reject : the request is rejected and the connection is closed
     - capture : the specified sample expression is captured
+    - set-priority-class <expr> | set-priority-offset <expr>
     - { track-sc0 | track-sc1 | track-sc2 } <key> [table <table>]
     - sc-inc-gpc0(<sc-id>)
     - sc-inc-gpc1(<sc-id>)
@@ -9513,6 +9533,24 @@ tcp-request content <action> [{if | unless} <condition>]
   The "unset-var" is used to unset a variable. See above for details about
   <var-name>.
 
+  The "set-priority-class" is used to set the queue priority class of the
+  current request. The value must be a sample expression which converts to an
+  integer in the range -2047..2047. Results outside this range will be
+  truncated. The priority class determines the order in which queued requests
+  are processed. Lower values have higher priority.
+
+  The "set-priority-offset" is used to set the queue priority timestamp offset
+  of the current request. The value must be a sample expression which converts
+  to an integer in the range -524287..524287. Results outside this range will be
+  truncated. When a request is queued, it is ordered first by the priority
+  class, then by the current timestamp adjusted by the given offset in
+  milliseconds. Lower values have higher priority.
+  Note that the resulting timestamp is is only tracked with enough precision for
+  524,287ms (8m44s287ms). If the request is queued long enough to where the
+  adjusted timestamp exceeds this value, it will be misidentified as highest
+  priority.  Thus it is important to set "timeout queue" to a value, where when
+  combined with the offset, does not exceed this limit.
+
   The "send-spoe-group" is used to trigger sending of a group of SPOE
   messages. To do so, the SPOE engine used to send messages must be defined, as
   well as the SPOE group to send. Of course, the SPOE engine must refer to an

doc/lua-api/index.rst

@@ -1769,6 +1769,24 @@ TXN class
   :param class_txn txn: The class txn object containing the data.
   :param integer mark: The mark value.
 
+.. js:function:: TXN.set_priority_class(txn, prio)
+
+  This function adjusts the priority class of the transaction. The value should
+  be within the range -2047..2047. Values outside this range will be
+  truncated.
+
+  See the HAProxy configuration.txt file keyword "http-request" action
+  "set-priority-class" for details.
+
+.. js:function:: TXN.set_priority_offset(txn, prio)
+
+  This function adjusts the priority offset of the transaction. The value
+  should be within the range -524287..524287. Values outside this range will be
+  truncated.
+
+  See the HAProxy configuration.txt file keyword "http-request" action
+  "set-priority-offset" for details.
+
 .. _socket_class:
 
 Socket class

include/proto/queue.h

@@ -90,6 +90,25 @@ static inline int may_dequeue_tasks(const struct server *s, const struct proxy *
 		(!s->maxconn || s->cur_sess < srv_dynamic_maxconn(s)));
 }
 
+static inline int queue_limit_class(int class)
+{
+	if (class < -0x7ff)
+		return -0x7ff;
+	if (class > 0x7ff)
+		return 0x7ff;
+	return class;
+}
+
+static inline int queue_limit_offset(int offset)
+{
+	if (offset < -0x7ffff)
+		return -0x7ffff;
+	if (offset > 0x7ffff)
+		return 0x7ffff;
+	return offset;
+}
+
+
 #endif /* _PROTO_QUEUE_H */
 
 /*

include/types/stream.h

@@ -131,6 +131,8 @@ struct stream {
 	struct task *task;              /* the task associated with this stream */
 	unsigned short pending_events;	/* the pending events not yet processed by the stream.
 					 * This is a bit field of TASK_WOKEN_* */
+	int16_t priority_class;         /* priority class of the stream for the pending queue */
+	int32_t priority_offset;        /* priority offset of the stream for the pending queue */
 
 	struct list list;               /* position in global streams list */
 	struct list by_srv;             /* position in server stream list */

src/hlua.c

@@ -44,6 +44,7 @@
 #include <proto/hlua_fcn.h>
 #include <proto/map.h>
 #include <proto/obj_type.h>
+#include <proto/queue.h>
 #include <proto/pattern.h>
 #include <proto/payload.h>
 #include <proto/proto_http.h>
@@ -5388,6 +5389,26 @@ __LJMP static int hlua_txn_set_mark(lua_State *L)
 	return 0;
 }
 
+__LJMP static int hlua_txn_set_priority_class(lua_State *L)
+{
+	struct hlua_txn *htxn;
+
+	MAY_LJMP(check_args(L, 2, "set_priority_class"));
+	htxn = MAY_LJMP(hlua_checktxn(L, 1));
+	htxn->s->priority_class = queue_limit_class(MAY_LJMP(luaL_checkinteger(L, 2)));
+	return 0;
+}
+
+__LJMP static int hlua_txn_set_priority_offset(lua_State *L)
+{
+	struct hlua_txn *htxn;
+
+	MAY_LJMP(check_args(L, 2, "set_priority_offset"));
+	htxn = MAY_LJMP(hlua_checktxn(L, 1));
+	htxn->s->priority_offset = queue_limit_offset(MAY_LJMP(luaL_checkinteger(L, 2)));
+	return 0;
+}
+
 /* This function is an Lua binding that send pending data
  * to the client, and close the stream interface.
  */
@@ -7931,21 +7952,23 @@ void hlua_init(void)
 	lua_newtable(gL.T);
 
 	/* Register Lua functions. */
-	hlua_class_function(gL.T, "set_priv",    hlua_set_priv);
-	hlua_class_function(gL.T, "get_priv",    hlua_get_priv);
-	hlua_class_function(gL.T, "set_var",     hlua_set_var);
-	hlua_class_function(gL.T, "unset_var",   hlua_unset_var);
-	hlua_class_function(gL.T, "get_var",     hlua_get_var);
-	hlua_class_function(gL.T, "done",        hlua_txn_done);
-	hlua_class_function(gL.T, "set_loglevel",hlua_txn_set_loglevel);
-	hlua_class_function(gL.T, "set_tos",     hlua_txn_set_tos);
-	hlua_class_function(gL.T, "set_mark",    hlua_txn_set_mark);
-	hlua_class_function(gL.T, "deflog",      hlua_txn_deflog);
-	hlua_class_function(gL.T, "log",         hlua_txn_log);
-	hlua_class_function(gL.T, "Debug",       hlua_txn_log_debug);
-	hlua_class_function(gL.T, "Info",        hlua_txn_log_info);
-	hlua_class_function(gL.T, "Warning",     hlua_txn_log_warning);
-	hlua_class_function(gL.T, "Alert",       hlua_txn_log_alert);
+	hlua_class_function(gL.T, "set_priv",            hlua_set_priv);
+	hlua_class_function(gL.T, "get_priv",            hlua_get_priv);
+	hlua_class_function(gL.T, "set_var",             hlua_set_var);
+	hlua_class_function(gL.T, "unset_var",           hlua_unset_var);
+	hlua_class_function(gL.T, "get_var",             hlua_get_var);
+	hlua_class_function(gL.T, "done",                hlua_txn_done);
+	hlua_class_function(gL.T, "set_loglevel",        hlua_txn_set_loglevel);
+	hlua_class_function(gL.T, "set_tos",             hlua_txn_set_tos);
+	hlua_class_function(gL.T, "set_mark",            hlua_txn_set_mark);
+	hlua_class_function(gL.T, "set_priority_class",  hlua_txn_set_priority_class);
+	hlua_class_function(gL.T, "set_priority_offset", hlua_txn_set_priority_offset);
+	hlua_class_function(gL.T, "deflog",              hlua_txn_deflog);
+	hlua_class_function(gL.T, "log",                 hlua_txn_log);
+	hlua_class_function(gL.T, "Debug",               hlua_txn_log_debug);
+	hlua_class_function(gL.T, "Info",                hlua_txn_log_info);
+	hlua_class_function(gL.T, "Warning",             hlua_txn_log_warning);
+	hlua_class_function(gL.T, "Alert",               hlua_txn_log_alert);
 
 	lua_rawset(gL.T, -3);
 

src/queue.c

@@ -75,11 +75,14 @@
 #include <common/hathreads.h>
 #include <eb32tree.h>
 
+#include <proto/proto_http.h>
 #include <proto/queue.h>
+#include <proto/sample.h>
 #include <proto/server.h>
 #include <proto/stream.h>
 #include <proto/stream_interface.h>
 #include <proto/task.h>
+#include <proto/tcp_rules.h>
 
 
 struct pool_head *pool_head_pendconn;
@@ -456,6 +459,140 @@ int pendconn_dequeue(struct stream *strm)
 	return 0;
 }
 
+static enum act_return action_set_priority_class(struct act_rule *rule, struct proxy *px,
+                                                 struct session *sess, struct stream *s, int flags)
+{
+	struct sample *smp;
+
+	smp = sample_fetch_as_type(px, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL, rule->arg.expr, SMP_T_SINT);
+	if (!smp)
+		return ACT_RET_CONT;
+
+	s->priority_class = queue_limit_class(smp->data.u.sint);
+	return ACT_RET_CONT;
+}
+
+static enum act_return action_set_priority_offset(struct act_rule *rule, struct proxy *px,
+                                                  struct session *sess, struct stream *s, int flags)
+{
+	struct sample *smp;
+
+	smp = sample_fetch_as_type(px, sess, s, SMP_OPT_DIR_REQ|SMP_OPT_FINAL, rule->arg.expr, SMP_T_SINT);
+	if (!smp)
+		return ACT_RET_CONT;
+
+	s->priority_offset = queue_limit_offset(smp->data.u.sint);
+
+	return ACT_RET_CONT;
+}
+
+static enum act_parse_ret parse_set_priority_class(const char **args, int *arg, struct proxy *px,
+                                                   struct act_rule *rule, char **err)
+{
+	unsigned int where = 0;
+
+	rule->arg.expr = sample_parse_expr((char **)args, arg, px->conf.args.file,
+	                                   px->conf.args.line, err, &px->conf.args);
+	if (!rule->arg.expr)
+		return ACT_RET_PRS_ERR;
+
+	if (px->cap & PR_CAP_FE)
+		where |= SMP_VAL_FE_HRQ_HDR;
+	if (px->cap & PR_CAP_BE)
+		where |= SMP_VAL_BE_HRQ_HDR;
+
+	if (!(rule->arg.expr->fetch->val & where)) {
+		memprintf(err,
+			  "fetch method '%s' extracts information from '%s', none of which is available here",
+			  args[0], sample_src_names(rule->arg.expr->fetch->use));
+		free(rule->arg.expr);
+		return ACT_RET_PRS_ERR;
+	}
+
+	rule->action     = ACT_CUSTOM;
+	rule->action_ptr = action_set_priority_class;
+	return ACT_RET_PRS_OK;
+}
+
+static enum act_parse_ret parse_set_priority_offset(const char **args, int *arg, struct proxy *px,
+                                                    struct act_rule *rule, char **err)
+{
+	unsigned int where = 0;
+
+	rule->arg.expr = sample_parse_expr((char **)args, arg, px->conf.args.file,
+	                                   px->conf.args.line, err, &px->conf.args);
+	if (!rule->arg.expr)
+		return ACT_RET_PRS_ERR;
+
+	if (px->cap & PR_CAP_FE)
+		where |= SMP_VAL_FE_HRQ_HDR;
+	if (px->cap & PR_CAP_BE)
+		where |= SMP_VAL_BE_HRQ_HDR;
+
+	if (!(rule->arg.expr->fetch->val & where)) {
+		memprintf(err,
+			  "fetch method '%s' extracts information from '%s', none of which is available here",
+			  args[0], sample_src_names(rule->arg.expr->fetch->use));
+		free(rule->arg.expr);
+		return ACT_RET_PRS_ERR;
+	}
+
+	rule->action     = ACT_CUSTOM;
+	rule->action_ptr = action_set_priority_offset;
+	return ACT_RET_PRS_OK;
+}
+
+static struct action_kw_list tcp_cont_kws = {ILH, {
+	{ "set-priority-class", parse_set_priority_class },
+	{ "set-priority-offset", parse_set_priority_offset },
+	{ /* END */ }
+}};
+
+static struct action_kw_list http_req_kws = {ILH, {
+	{ "set-priority-class", parse_set_priority_class },
+	{ "set-priority-offset", parse_set_priority_offset },
+	{ /* END */ }
+}};
+
+static int
+smp_fetch_priority_class(const struct arg *args, struct sample *smp, const char *kw, void *private)
+{
+	if (!smp->strm)
+		return 0;
+
+	smp->data.type = SMP_T_SINT;
+	smp->data.u.sint = smp->strm->priority_class;
+
+	return 1;
+}
+
+static int
+smp_fetch_priority_offset(const struct arg *args, struct sample *smp, const char *kw, void *private)
+{
+	if (!smp->strm)
+		return 0;
+
+	smp->data.type = SMP_T_SINT;
+	smp->data.u.sint = smp->strm->priority_offset;
+
+	return 1;
+}
+
+
+static struct sample_fetch_kw_list smp_kws = {ILH, {
+	{ "prio_class", smp_fetch_priority_class, 0, NULL, SMP_T_SINT, SMP_USE_INTRN, },
+	{ "prio_offset", smp_fetch_priority_offset, 0, NULL, SMP_T_SINT, SMP_USE_INTRN, },
+	{ /* END */},
+}};
+
+__attribute__((constructor))
+static void __queue_init(void)
+{
+	tcp_req_cont_keywords_register(&tcp_cont_kws);
+	http_req_keywords_register(&http_req_kws);
+	sample_register_fetches(&smp_kws);
+}
+
 /*
  * Local variables:
  *  c-indent-level: 8

src/stream.c

@@ -221,6 +221,8 @@ struct stream *stream_new(struct session *sess, enum obj_type *origin)
 	s->target = sess->listener ? sess->listener->default_target : NULL;
 
 	s->pend_pos = NULL;
+	s->priority_class = 0;
+	s->priority_offset = 0;
 
 	/* init store persistence */
 	s->store_count = 0;