WIP: haload sources

include/haproxy/haload.h

@@ -0,0 +1,65 @@
+#ifndef _HAPROXY_HALOAD_H
+#define _HAPROXY_HALOAD_H
+
+#include <import/ist.h>
+#include <haproxy/list-t.h>
+#include <haproxy/proxy-t.h>
+#include <haproxy/server-t.h>
+#include <haproxy/task-t.h>
+
+struct hld_path {
+	char *path;
+	struct hld_path *next;
+};
+
+struct hld_url_cfg {
+	int ssl;
+	int h2c;
+	char *addr;
+	char *raw_addr; // used only to set the host header value
+	char *srv_opts;
+	char *tls_opts;
+	struct server *srv;
+	struct hld_path *paths;
+	struct hld_url_cfg *next;
+};
+
+struct hld_url {
+	int mreqs;
+	int flags;
+	uint64_t tot_req;
+	uint64_t tot_rconn_done;
+	uint64_t tot_rconn_sent;
+	struct hld_url_cfg *cfg;
+	struct hld_url *next;
+};
+
+/* haload header */
+struct hld_hdr {
+	struct ist name;
+	struct ist value;
+	struct list list;
+};
+
+extern const char *arg_host;
+extern const char *arg_conn_hdr;
+extern const char *arg_uri;
+extern const char *arg_path; // TO REMOVE
+extern struct list hld_hdrs;
+extern struct hld_url_cfg *hld_url_cfgs;
+
+extern struct proxy hld_proxy;
+extern int arg_dura;
+extern int arg_head;
+extern int arg_hscd;
+extern int arg_long;
+extern int arg_mreqs;
+extern int arg_reqs;
+extern int arg_rcon;
+extern int arg_slow;
+extern int arg_serr;
+extern int arg_usr;
+extern int arg_nbthrds;
+extern int arg_wait;
+
+#endif /* _HAPROXY_HALOAD_H */

include/haproxy/hldstream-t.h

@@ -12,6 +12,7 @@
 struct hldstream {
 	enum obj_type obj_type;
 	struct connection *conn;
+	unsigned int expire;
 	int64_t hash;
 	struct hld_usr *usr;
 	struct hld_url *url;
@@ -23,6 +24,7 @@ struct hldstream {
 	int flags;
 	int state;
 	unsigned long long to_send; /* number of body data bytes to send */
+	struct timeval req_date;
 	struct list list;
 };
 

src/haload_init.c

@@ -0,0 +1,661 @@
+#include <errno.h>
+
+#include <haproxy/connection.h>
+#include <haproxy/errors.h>
+#include <haproxy/global.h>
+#include <haproxy/hbuf.h>
+#include <haproxy/haload.h>
+#include <haproxy/proxy.h>
+#include <haproxy/version.h>
+#include <haproxy/server.h>
+
+static int hld_debug;
+struct hld_url_cfg *hld_url_cfgs;
+char *srv_opts, *tls_ciphers, *tls_ciphersuites, *tls_curves, *alpn;
+int h2c;
+
+static void  hld_usage(char *name, int argc, int line)
+{
+	fprintf(stderr,
+		"Usage : %s [opts] [URL]\n"
+		"where <opts> may be any combination of:\n"
+		"        -d <time>        test duration in seconds (0)\n"
+		"        -e               stop upon first connection error\n"
+		"        -h(0|1|2|2c|3)   use h0 (hq-interop for QUIC), h1, h2, h2c or h3 (QUIC/TCP) protocols\n"
+		"        -(0|1|2|2c|3)    same as above\n"
+		"        -l               enable long output format; double for raw values\n"
+		"        -m <streams>     maximum concurrent streams (1)\n"
+		"        -n <reqs>        maximum total requests (-1)\n"
+		"        -r <reqs>        number of requests per connection (-1)\n"
+		"        -s <time>        soft start: time in sec to reach 100%% load\n"
+		"        -t <threads>     number of threads\n"
+		"        -u <users>       number of users (1)\n"
+		"        -w <time>        I/O timeout in milliseconds (10000)\n"
+		"        -C               dump the configuration and exit\n"
+		"        -H \"foo:bar\"   add this header name and value\n"
+		"        -I               use HEAD instead of GET\n"
+		"        -v               shows version\n"
+		"        --defaults <str> add a string to default section\n"
+		"        --global <str>   add a string to global section\n"
+		"        --server <opts>  set server <opt> options as defined for \"server\" haproxy keyword\n"
+		"        --show-status-codes show HTTP status codes distribution\n"
+		"        --traces         enable the traces for all the HTTP protocols\n"
+		"SSL options:\n"
+		"        --tls-ciphers <ciphers>       for TLS1.2 and below\n"
+		"        --tls-ciphersuites <ciphers>  for TLS1.3 and above\n"
+		"        --tls-curves <curves>\n"
+		"URL formats:\n"
+		"        (http|https|quic)://<addr>:<port>/<path>\n",
+		name);
+	exit(1);
+}
+
+static const char *hld_cfg_traces_str =
+	"traces\n"
+		"\ttrace haload sink stderr level developer start now verbosity clean\n"
+		"\ttrace quic sink stderr level developer start now\n"
+		"\ttrace ssl sink stderr level developer start now verbosity minimal\n"
+		"\ttrace h1 sink stderr level developer start now verbosity minimal\n"
+		"\ttrace h2 sink stderr level developer start now verbosity minimal\n"
+		"\ttrace h3 sink stderr level developer start now verbosity minimal\n"
+		"\ttrace qmux sink stderr level developer start now verbosity minimal\n";
+
+static struct hld_hdr *hld_parse_hdr(char *hdr_str)
+{
+	struct hld_hdr *hdr= NULL;
+	char *value = strchr(hdr_str, ':');
+
+	if (value) {
+		*value++ = '\0';
+		if (!*value)
+			value = NULL;
+	}
+
+	if (strcasecmp(hdr_str, "host") == 0)
+		arg_host = value;
+	else if (strcasecmp(hdr_str, "connection") == 0)
+		arg_conn_hdr = value;
+
+	hdr = malloc(sizeof(*hdr));
+	if (hdr) {
+		hdr->name = ist(hdr_str);
+		hdr->value = ist(value);
+	}
+
+	return hdr;
+}
+
+static int hld_add_opt_to_buf(struct hbuf *buf,
+                              const char *kw, const char *value)
+{
+	if (hbuf_is_null(buf)) {
+		if (hbuf_alloc(buf) == NULL) {
+			ha_alert("failed to allocate a buffer.\n");
+			return 0;
+		}
+	}
+	else
+		hbuf_appendf(buf, " ");
+
+	hbuf_appendf(buf, "%s", kw);
+	hbuf_appendf(buf, " ");
+	hbuf_appendf(buf, "%s", value);
+	return 1;
+}
+
+static inline void hld_free_url_cfg(struct hld_url_cfg *h)
+{
+	free(h->addr);
+	free(h->srv_opts);
+	free(h->tls_opts);
+	free(h);
+}
+
+static inline void hld_free_url_cfgs(void)
+{
+	struct hld_url_cfg *purl;
+
+	purl = hld_url_cfgs;
+
+	while (purl) {
+		struct hld_url_cfg *purl_next;
+		struct hld_path *path;
+
+		path = purl->paths;
+		while (path) {
+			struct hld_path *path_next;
+
+			path_next = path->next;
+			free(path->path);
+			free(path);
+			path = path_next;
+		}
+
+		purl_next = purl->next;
+
+		hld_free_url_cfg(purl);
+		purl = purl_next;
+	}
+}
+
+/* Allocate a URL from <url> command line argument without
+ * duplicating it, and append it to <hld_url_cfgs> list of URL.
+ * A URL is identified by its peer address and if it uses
+ * SSL or not. When a URL with the same peer address already
+ * exists, this function only add a new path to the list
+ * of paths attaached to the URL.
+ * Return the URL if succeeded, NULL if not.
+ */
+static struct hld_url_cfg *hld_alloc_url(char *url)
+{
+	int ssl = 0, is_quic = 0;
+	char *addr = NULL, *raw_addr = NULL, *path = NULL;
+	struct hld_url_cfg *hld_url_cfg = NULL;
+	struct hld_url_cfg *purl;
+	struct hld_path *p = NULL;
+	struct hbuf opts_buf = HBUF_NULL;
+	char quic_addr[128];
+
+	if (strncmp(url, "http://", 7) == 0)
+		addr = url + 7;
+	else if (strncmp(url, "https://", 8) == 0) {
+		ssl = 1;
+		addr = url + 8;
+	}
+	else if (strncmp(url, "quic://",  7) == 0) {
+		ssl = 1;
+		addr = url + 7;
+		is_quic = 1;
+	}
+	else
+		addr = url;
+
+	path = strchr(addr, '/');
+	if (path) {
+		char *new_path = strdup(path);
+		*path = '\0';
+		path = new_path;
+	}
+	else
+		path = strdup("/");
+
+	if (!path)
+		goto err;
+
+	for (purl = hld_url_cfgs; purl; purl = purl->next) {
+		/* XXX TODO: improve this check for QUIC XXX */
+		if (strcmp(purl->addr, addr) == 0 && purl->ssl == ssl) {
+			/* Already existing URL with the same address. */
+			hld_url_cfg = purl;
+
+			p = calloc(1, sizeof(*p));
+			if (!p)
+				goto err;
+
+			/* Append a new path to this URL */
+			p->path = path;
+			p->next = hld_url_cfg->paths;
+			hld_url_cfg->paths = p;
+
+			return hld_url_cfg;
+		}
+	}
+
+	if (!is_quic) {
+		addr = strdup(addr);
+	}
+	else {
+		snprintf(quic_addr, sizeof(quic_addr), "quic+%s", addr);
+		addr = strdup(quic_addr);
+	}
+
+	if (!addr)
+		goto err;
+
+	raw_addr = strchr(addr, '+');
+	if (!raw_addr)
+		raw_addr = strchr(addr, '@');
+
+	raw_addr = raw_addr ? raw_addr + 1: addr;
+
+	hld_url_cfg = calloc(1, sizeof(*hld_url_cfg));
+	p = malloc(sizeof(*p));
+	if (!hld_url_cfg || !p)
+		goto err;
+
+	p->path = path;
+	p->next = NULL;
+
+	hld_url_cfg->ssl = ssl;
+	hld_url_cfg->addr = addr;
+	hld_url_cfg->raw_addr = raw_addr;
+
+	if (srv_opts) {
+		hld_url_cfg->srv_opts = strdup(srv_opts);
+		if (!hld_url_cfg->srv_opts)
+			goto err;
+	}
+
+	if (tls_ciphers &&
+	    !hld_add_opt_to_buf(&opts_buf, "ciphers", tls_ciphers))
+		goto err;
+
+	if (tls_ciphersuites &&
+	    !hld_add_opt_to_buf(&opts_buf, "ciphersuites", tls_ciphersuites))
+		goto err;
+
+	if (tls_curves &&
+	    !hld_add_opt_to_buf(&opts_buf, "curves", tls_curves))
+		goto err;
+
+	if (alpn && !h2c &&
+	    !hld_add_opt_to_buf(&opts_buf, "alpn", alpn))
+		goto err;
+
+	if (!hbuf_is_null(&opts_buf))
+		hld_url_cfg->tls_opts = strdup(opts_buf.area);
+
+	hld_url_cfg->h2c = h2c;
+
+	free_hbuf(&opts_buf);
+	hld_url_cfg->srv = NULL;
+	hld_url_cfg->paths = p;
+	/* Append this new URL to the list */
+	hld_url_cfg->next = hld_url_cfgs;
+	hld_url_cfgs = hld_url_cfg;
+
+	return hld_url_cfg;
+ err:
+	hld_free_url_cfgs();
+	hld_free_url_cfg(hld_url_cfg);
+	free(p);
+	free(path);
+	return NULL;
+}
+
+static void hld_parse_long(int *val, char *opt, int *argc, char ***argv)
+{
+	char *endptr;
+
+	if (!*opt) {
+		++*argv; --*argc;
+		if (*argc <= 0 || ***argv == '-')
+			hld_usage(progname, *argc, __LINE__);
+
+		opt = **argv;
+	}
+
+	*val = strtol(opt, &endptr, 0);
+	if (endptr == opt || *val < 0)
+		hld_usage(progname, *argc, __LINE__);
+}
+
+static inline void hld_url_cfgs_inv(void)
+{
+	struct hld_url_cfg *urls = NULL, *url = hld_url_cfgs, *next_url;
+
+	/* inverse the URLs order */
+	while (url) {
+		struct hld_path *paths = NULL, *path = url->paths, *next_path;
+
+		/* inverse the paths order */
+		while (path) {
+			next_path = path->next;
+			path->next = paths;
+			paths = path;
+			path = next_path;
+		}
+		url->paths = paths;
+
+		next_url = url->next;
+		url->next = urls;
+		urls = url;
+		url = next_url;
+	}
+
+	hld_url_cfgs = urls;
+}
+
+
+void haproxy_init_args(int argc, char **argv)
+{
+	int err = 1, dump = 0;
+	struct hbuf buf = HBUF_NULL;  // cfgfile
+	struct hbuf gbuf = HBUF_NULL; // "global" section
+	struct hbuf tbuf = HBUF_NULL; // "traces" section
+	struct hbuf dbuf = HBUF_NULL; // "default" section
+
+	if (argc <= 1)
+		hld_usage(progname, argc, __LINE__);
+
+	if (hbuf_alloc(&gbuf) == NULL) {
+		ha_alert("failed to allocate a buffer.\n");
+		goto leave;
+	}
+
+	/* use 3MB of local cache per thread mainly for QUIC.
+	 * Also trust the server certificates
+	 */
+	hbuf_appendf(&gbuf, "global\n");
+	hbuf_appendf(&gbuf, "\ttune.memory.hot-size 3145728\n");
+	hbuf_appendf(&gbuf, "\tssl-server-verify none\n");
+
+	if (hbuf_alloc(&buf) == NULL) {
+		ha_alert("failed to allocate a buffer\n");
+		exit(1);
+	}
+
+	fileless_mode = 1;
+	client_mode = 1;
+	/* skip program name and start */
+	argc--; argv++;
+
+	while (argc > 0) {
+		if (**argv == '-') {
+			char *opt = *argv + 1;
+
+			//fprintf(stderr, "||||**argv='%c' argc=%d\n", **argv, argc);
+			//fprintf(stderr, "====> *opt='%c'\n", *opt);
+			if (*opt == '-') {
+				/* long option */
+				opt++;
+				if (strcmp(opt, "defaults") == 0) {
+					argv++; argc--;
+					if (argc <= 0 || **argv == '-')
+						hld_usage(progname, argc, __LINE__);
+
+					if (hbuf_is_null(&dbuf)) {
+						if (hbuf_alloc(&dbuf) == NULL) {
+							ha_alert("failed to allocate a buffer.\n");
+							goto leave;
+						}
+
+						hbuf_appendf(&dbuf, "defaults\n");
+					}
+
+					hbuf_str_append(&dbuf, *argv);
+				}
+				else if (strcmp(opt, "global") == 0) {
+					argv++; argc--;
+					if (argc <= 0 || **argv == '-')
+						hld_usage(progname, argc, __LINE__);
+
+					hbuf_str_append(&gbuf, *argv);
+				}
+				else if (strcmp(opt, "server") == 0) {
+					argv++, argc--;
+					if ((argc <= 0 || **argv == '-'))
+						hld_usage(progname, argc, __LINE__);
+
+					opt = *argv;
+					free(srv_opts);
+					srv_opts = strdup(opt);
+				}
+				else if (strcmp(opt, "show-status-codes") == 0) {
+					arg_hscd = 1;
+				}
+				else if (strcmp(opt, "tls-ciphers") == 0) {
+					argv++, argc--;
+					if ((argc <= 0 || **argv == '-'))
+						hld_usage(progname, argc, __LINE__);
+					opt = *argv;
+					free(tls_ciphers);
+					tls_ciphers = strdup(opt);
+				}
+				else if (strcmp(opt, "tls-ciphersuites") == 0) {
+					argv++, argc--;
+					if ((argc <= 0 || **argv == '-'))
+						hld_usage(progname, argc, __LINE__);
+					opt = *argv;
+					free(tls_ciphersuites);
+					tls_ciphersuites = strdup(opt);
+				}
+				else if (strcmp(opt, "tls-curves") == 0) {
+					argv++, argc--;
+					if ((argc <= 0 || **argv == '-'))
+						hld_usage(progname, argc, __LINE__);
+					opt = *argv;
+					free(tls_curves);
+					tls_curves = strdup(opt);
+				}
+				else if (strcmp(opt, "traces") == 0) {
+					hld_debug = 1;
+				}
+				else
+					hld_usage(progname, argc, __LINE__);
+			}
+			else if (strcmp(opt, "0") == 0 ||
+			         strcmp(opt, "h0") == 0) {
+				alpn = "hq-interop";
+				h2c = 0;
+			}
+			else if (strcmp(opt, "1") == 0 ||
+			         strcmp(opt, "h1") == 0) {
+				alpn = "http/1.1";
+				h2c = 0;
+			}
+			else if (strcmp(opt, "2") == 0 ||
+			         strcmp(opt, "h2") == 0) {
+				alpn = "h2";
+				h2c = 0;
+			}
+			else if (strcmp(opt, "2c") == 0 ||
+			         strcmp(opt, "h2c") == 0) {
+				alpn = NULL;
+				h2c = 1;
+			}
+			else if (strcmp(opt, "3") == 0 ||
+			         strcmp(opt, "h3") == 0) {
+				alpn = "h3";
+				h2c = 0;
+			}
+			else if (*opt == 'd') {
+				opt++;
+				hld_parse_long(&arg_dura, opt, &argc, &argv);
+			}
+			else if (*opt == 'e') {
+				/* empty option */
+				if (*(opt + 1))
+					hld_usage(progname, argc, __LINE__);
+
+				arg_serr = 1;
+			}
+			else if (*opt == 'l') {
+				arg_long++;
+				while (*++opt && *opt == 'l')
+					arg_long++;
+			}
+			else if (*opt == 'm') {
+				opt++;
+				hld_parse_long(&arg_mreqs, opt, &argc, &argv);
+			}
+			else if (*opt == 'n') {
+				opt++;
+				hld_parse_long(&arg_reqs, opt, &argc, &argv);
+			}
+			else if (*opt == 'r') {
+				opt++;
+				hld_parse_long(&arg_rcon, opt, &argc, &argv);
+			}
+			else if (*opt == 's') {
+				opt++;
+				hld_parse_long(&arg_slow, opt, &argc, &argv);
+				arg_slow *= 1000;
+			}
+			else if (*opt == 't') {
+				opt++;
+				hld_parse_long(&arg_nbthrds, opt, &argc, &argv);
+			}
+			else if (*opt == 'u') {
+				opt++;
+				hld_parse_long(&arg_usr, opt, &argc, &argv);
+			}
+			else if (*opt == 'w') {
+				opt++;
+				hld_parse_long(&arg_wait, opt, &argc, &argv);
+			}
+			else if (*opt == 'C') {
+				/* empty option */
+				if (*(opt + 1))
+					hld_usage(progname, argc, __LINE__);
+
+				dump = 1;
+			}
+			else if (*opt == 'H') {
+				char *hdr_str;
+				struct hld_hdr *hdr;
+
+				opt++;
+				if (!*opt) {
+					argv++; argc--;
+					if ((argc <= 0 || **argv == '-'))
+						hld_usage(progname, argc, __LINE__);
+
+					opt = *argv;
+				}
+
+				hdr_str = opt;
+				hdr = hld_parse_hdr(hdr_str);
+				if (!hdr) {
+					ha_alert("could not allocate a header\n");
+					goto leave;
+				}
+
+				LIST_APPEND(&hld_hdrs, &hdr->list);
+			}
+			else if (*opt == 'I') {
+				/* empty option */
+				if (*(opt + 1))
+					hld_usage(progname, argc, __LINE__);
+
+				arg_head = 1;
+			}
+			else if (*opt == 'v') {
+				/* empty option */
+				if (*(opt + 1))
+					hld_usage(progname, argc, __LINE__);
+
+				printf("haload version " HAPROXY_VERSION " released " HAPROXY_DATE "\n");
+				exit(0);
+			}
+			else
+				hld_usage(progname, argc, __LINE__);
+		}
+		else {
+			struct hld_url_cfg *url;
+
+			url = hld_alloc_url(*argv);
+			if (!url) {
+				ha_alert("could not parse a new URL\n");
+				goto leave;
+			}
+
+
+		}
+
+		argv++; argc--;
+		//fprintf(stderr, "///argc=%d argv='%s'\n", argc, *argv);
+	}
+
+	if (arg_rcon > 0 && arg_rcon < arg_mreqs) {
+		ha_warning("number of maximum concurrent streams is greater that number of requests by connection\n");
+		ha_warning("set both these parameters values to %d (number of requests by connection)\n", arg_rcon);
+		arg_mreqs = arg_rcon;
+	}
+
+	if (!hld_url_cfgs) {
+		ha_alert("no URL provided\n");
+		goto leave;
+	}
+
+	/* "global" section */
+	hbuf_appendf(&buf, "%.*s", (int)gbuf.data, gbuf.area);
+	if (arg_nbthrds != -1)
+		hbuf_appendf(&buf, "\tnbthread %d\n", arg_nbthrds);
+	if (arg_mreqs)
+		hbuf_appendf(&buf,
+		             "\ttune.h2.be.max-concurrent-streams %d\n"
+		             "\ttune.quic.be.stream.max-concurrent %d\n", arg_mreqs, arg_mreqs);
+	/* "traces" section */
+	if (hld_debug) {
+		hbuf_appendf(&buf, "%s", hld_cfg_traces_str);
+		if (!hbuf_is_null(&tbuf))
+			hbuf_appendf(&buf, "%.*s\n", (int)tbuf.data, tbuf.area);
+	}
+	/* "default section */
+	if (!hbuf_is_null(&dbuf))
+		hbuf_appendf(&buf, "%.*s\n", (int)dbuf.data, dbuf.area);
+
+	fileless_cfg.filename = strdup("haterm cfgfile");
+	fileless_cfg.content = strdup(buf.area);
+	if (!fileless_cfg.filename || !fileless_cfg.content) {
+		ha_alert("cfgfile strdup() failed.\n");
+		goto leave;
+	}
+
+	fileless_cfg.size = buf.data;
+
+	/* Config dump */
+	if (dump) {
+		fprintf(stdout, "%.*s", (int)fileless_cfg.size, fileless_cfg.content);
+		exit(0);
+	}
+
+#if 0
+	if (arg_reqs > 0 && arg_reqs < arg_usr) {
+		ha_alert("user count must not exceed request count\n");
+		goto leave;
+	}
+#endif
+
+	/* Inverse the URLs and their paths */
+	hld_url_cfgs_inv();
+	err = 0;
+leave:
+	free_hbuf(&dbuf);
+	free_hbuf(&gbuf);
+	free_hbuf(&buf);
+	if (err)
+		exit(1);
+}
+
+/* Dummy argv copier function */
+char **copy_argv(int argc, char **argv)
+{
+	char **ret = calloc(1, sizeof(*ret));
+
+	if (ret)
+		*ret = strdup("");
+
+	return ret;
+}
+
+static int hld_pre_check(void)
+{
+	char *errmsg = NULL;
+
+	if (!setup_new_proxy(&hld_proxy, "<HALOAD-BE>",
+	                     PR_CAP_FE | PR_CAP_BE | PR_CAP_INT, &errmsg)) {
+		ha_alert("could not setup internal proxy: %s\n", errmsg);
+		ha_free(&errmsg);
+		return ERR_FATAL;
+	}
+
+	hld_proxy.mode = PR_MODE_HTTP;
+	hld_proxy.next = proxies_list;
+    proxies_list = &hld_proxy;
+
+    hld_proxy.timeout.server = 60000;
+    hld_proxy.timeout.connect = 60000;
+
+	return ERR_NONE;
+}
+REGISTER_PRE_CHECK(hld_pre_check);
+
+static int hld_deinit(void)
+{
+	ha_free(&old_argv[0]);
+	ha_free(&old_argv);
+	return 1;
+}
+REGISTER_POST_DEINIT(hld_deinit);

src/stconn.c

@@ -29,6 +29,13 @@
 DECLARE_TYPED_POOL(pool_head_connstream, "stconn", struct stconn);
 DECLARE_TYPED_POOL(pool_head_sedesc, "sedesc", struct sedesc);
 
+/* Only used by haload */
+extern __attribute__((weak))
+struct task *hld_strm_task(struct task *t, void *context, unsigned int state)
+{
+       return NULL;
+}
+
 static int sc_conn_recv(struct stconn *sc);
 static int sc_conn_send(struct stconn *sc);
 
@@ -298,6 +305,17 @@ int sc_attach_mux(struct stconn *sc, void *sd, void *ctx)
 			sc->wait_event.events = 0;
 		}
 	}
+	else if (sc_hldstream(sc)) {
+		if (!sc->wait_event.tasklet) {
+			sc->wait_event.tasklet = tasklet_new();
+			if (!sc->wait_event.tasklet)
+				return -1;
+			sc->wait_event.tasklet->process = hld_strm_task;
+			sc->wait_event.tasklet->expire = TICK_ETERNITY;
+			sc->wait_event.tasklet->context = __sc_hldstream(sc);;
+			sc->wait_event.events = 0;
+		}
+	}
 
 	sedesc->se = sd;
 	sedesc->conn = ctx;