From e4de9a8bc71ec1f723fda5de452273b56c73f34c Mon Sep 17 00:00:00 2001 From: Roberto De Ioris Date: Fri, 9 Nov 2012 20:04:27 +0100 Subject: [PATCH] improved offload threads subsystem --- core/init.c | 2 + core/master_utils.c | 7 +- core/offload.c | 358 ++++++++++++++++++++++++------ core/protocol.c | 2 +- core/routing.c | 2 + core/socket.c | 4 +- core/uwsgi.c | 9 +- plugins/python/uwsgi_pymodule.c | 2 +- plugins/router_http/router_http.c | 23 +- uwsgi.h | 22 +- 10 files changed, 336 insertions(+), 95 deletions(-) diff --git a/core/init.c b/core/init.c index 5a9f60d2..fa56949a 100644 --- a/core/init.c +++ b/core/init.c @@ -36,6 +36,8 @@ void uwsgi_init_default() { uwsgi.cores = 1; uwsgi.threads = 1; + uwsgi.offload_threads_events = 64; + uwsgi.default_app = -1; uwsgi.buffer_size = 4096; diff --git a/core/master_utils.c b/core/master_utils.c index b5caca33..266ed44a 100644 --- a/core/master_utils.c +++ b/core/master_utils.c @@ -933,10 +933,6 @@ struct uwsgi_stats *uwsgi_master_generate_stats() { if (uwsgi_stats_keylong_comma(us, "signal_queue", (unsigned long long) signal_queue)) goto end; -/* - if (uwsgi_stats_keylong_comma(us, "static_offload_threads", (unsigned long long) uwsgi.workers[i + 1].static_offload_threads)) - goto end; -*/ if (uwsgi.workers[i + 1].cheaped) { if (uwsgi_stats_keyval_comma(us, "status", "cheap")) @@ -1050,6 +1046,9 @@ struct uwsgi_stats *uwsgi_master_generate_stats() { if (uwsgi_stats_keylong_comma(us, "static_requests", (unsigned long long) uc->static_requests)) goto end; + if (uwsgi_stats_keylong_comma(us, "routed_requests", (unsigned long long) uc->routed_requests)) + goto end; + if (uwsgi_stats_keylong(us, "in_request", (unsigned long long) uc->in_request)) goto end; diff --git a/core/offload.c b/core/offload.c index ed3174ce..07c2b80f 100644 --- a/core/offload.c +++ b/core/offload.c @@ -1,47 +1,92 @@ #include "uwsgi.h" -extern struct uwsgi_server uwsgi; - /* -enqueue a file transfer to the offload thread + uWSGI offloading subsystem */ -int uwsgi_offload_request_do(struct wsgi_request *wsgi_req, char *filename, size_t len) { - struct stat st; +extern struct uwsgi_server uwsgi; + +#define uwsgi_offload_retry if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) return 0; +#define uwsgi_offload_0r_1w(x, y) if (event_queue_del_fd(ut->queue, x, event_queue_read())) return -1;\ + if (event_queue_fd_read_to_write(ut->queue, y)) return -1; + +static int uwsgi_offload_net_transfer(struct uwsgi_thread *, struct uwsgi_offload_request *, int); +static int uwsgi_offload_sendfile_transfer(struct uwsgi_thread *, struct uwsgi_offload_request *, int); + +static void uwsgi_offload_setup(struct uwsgi_offload_request *uor, struct wsgi_request *wsgi_req, + int (*func)(struct uwsgi_thread *, struct uwsgi_offload_request *, int)) { - // avoid closing the connection wsgi_req->fd_closed = 1; + memset(uor, 0, sizeof(struct uwsgi_offload_request)); + uor->s = wsgi_req->poll.fd; + uor->func = func; + // put socket in non-blocking mode + uwsgi_socket_nb(uor->s); + +} + +static int uwsgi_offload_enqueue(struct uwsgi_offload_request *uor) { + if (write(uwsgi.offload_thread->pipe[0], uor, sizeof(struct uwsgi_offload_request)) != sizeof(struct uwsgi_offload_request)) { + return -1; + } + return 0; +} + +int uwsgi_offload_request_net_do(struct wsgi_request *wsgi_req, char *socket, struct uwsgi_buffer *ubuf) { // fill offload request struct uwsgi_offload_request uor; + uwsgi_offload_setup(&uor, wsgi_req, uwsgi_offload_net_transfer); + + uor.fd = uwsgi_connect(socket, 0, 1); + if (uor.fd < 0) { + uwsgi_error("uwsgi_offload_request_net_do() -> connect()"); + goto error; + } + + uor.ubuf = ubuf; + + if (uwsgi_offload_enqueue(&uor)) { + goto error2; + } + + return 0; + +error2: + close(uor.fd); +error: + wsgi_req->fd_closed = 0; + return -1; +} + +int uwsgi_offload_request_sendfile_do(struct wsgi_request *wsgi_req, char *filename, size_t len) { + + // fill offload request + struct uwsgi_offload_request uor; + uwsgi_offload_setup(&uor, wsgi_req, uwsgi_offload_sendfile_transfer); + uor.fd = open(filename, O_RDONLY | O_NONBLOCK); if (uor.fd < 0) { uwsgi_error_open(filename); goto error; } - uor.s = wsgi_req->poll.fd; + // make a fstat to get the file size if (!len) { + struct stat st; if (fstat(uor.fd, &st)) { uwsgi_error("fstat()"); goto error2; } len = st.st_size; } - uor.pos = 0; + uor.len = len; - uor.written = 0; - uor.buf = NULL; - uor.prev = NULL; - uor.next = NULL; - // put socket in non-blocking mode - uwsgi_socket_nb(uor.s); - - if (write(uwsgi.offload_thread->pipe[0], &uor, sizeof(struct uwsgi_offload_request)) != sizeof(struct uwsgi_offload_request)) { + if (uwsgi_offload_enqueue(&uor)) { goto error2; } @@ -82,6 +127,10 @@ static void uwsgi_offload_close(struct uwsgi_offload_request *uor) { free(uor->buf); } + if (uor->ubuf) { + uwsgi_buffer_destroy(uor->ubuf); + } + free(uor); } @@ -99,10 +148,10 @@ static void uwsgi_offload_append(struct uwsgi_offload_request *uor) { uwsgi.offload_requests_tail = uor; } -static struct uwsgi_offload_request *uwsgi_offload_get_by_socket(int s) { +static struct uwsgi_offload_request *uwsgi_offload_get_by_fd(int s) { struct uwsgi_offload_request *uor = uwsgi.offload_requests_head; while (uor) { - if (uor->s == s) { + if (uor->s == s || uor->fd == s) { return uor; } uor = uor->next; @@ -114,10 +163,10 @@ static struct uwsgi_offload_request *uwsgi_offload_get_by_socket(int s) { static void uwsgi_offload_loop(struct uwsgi_thread *ut) { int i; - void *events = event_queue_alloc(uwsgi.static_offload_to_thread); + void *events = event_queue_alloc(uwsgi.offload_threads_events); for (;;) { - int nevents = event_queue_wait_multi(ut->queue, -1, events, uwsgi.static_offload_to_thread); + int nevents = event_queue_wait_multi(ut->queue, -1, events, uwsgi.offload_threads_events); for (i = 0; i < nevents; i++) { int interesting_fd = event_queue_interesting_fd(events, i); if (interesting_fd == uwsgi.offload_thread->pipe[1]) { @@ -129,68 +178,22 @@ static void uwsgi_offload_loop(struct uwsgi_thread *ut) { continue; } // start monitoring socket for write - if (event_queue_add_fd_write(ut->queue, uor->s)) { - free(uor); + if (uor->func(ut, uor, -1)) { + uwsgi_offload_close(uor); continue; } uwsgi_offload_append(uor); continue; } - // ok check for socket writability - struct uwsgi_offload_request *uor = uwsgi_offload_get_by_socket(interesting_fd); + + // get the task from the interesting fd + struct uwsgi_offload_request *uor = uwsgi_offload_get_by_fd(interesting_fd); if (!uor) continue; - // sendfile() in chunks (128k is a good size...) -#if defined(__linux__) - ssize_t len = sendfile(uor->s, uor->fd, &uor->pos, 128 * 1024); - if (len > 0) { - uor->written += len; - if (uor->written >= uor->len) { - uwsgi_offload_close(uor); - } - continue; - } - else if (len < 0) { - if (errno == EAGAIN) - continue; - uwsgi_error("sendfile()"); - } -#else - if (!uor->buf) { - uor->buf = uwsgi_malloc(32768); - uor->to_write = 0; - } - if (uor->to_write == 0) { - ssize_t len = read(uor->fd, uor->buf, 32768); - if (len > 0) { - uor->to_write = len; - uor->buf_pos = 0; - continue; - } - else if (len < 0) { - uwsgi_error("read()"); - } + // run the hook + if (uor->func(ut, uor, interesting_fd)) { uwsgi_offload_close(uor); - continue; } - ssize_t len = write(uor->s, uor->buf + uor->buf_pos, uor->to_write); - if (len > 0) { - uor->written += len; - uor->to_write -= len; - uor->buf_pos += len; - if (uor->written >= uor->len) { - uwsgi_offload_close(uor); - } - continue; - } - else if (len < 0) { - if (errno == EAGAIN) - continue; - uwsgi_error("write()"); - } - -#endif - uwsgi_offload_close(uor); } } } @@ -198,3 +201,212 @@ static void uwsgi_offload_loop(struct uwsgi_thread *ut) { struct uwsgi_thread *uwsgi_offload_thread_start() { return uwsgi_thread_new(uwsgi_offload_loop); } + +/* the offload task starts after having acquired the file fd + + status: + 0 -> read + 1 -> write + +*/ + +/* +int uwsgi_offload_readwrite_transfer(struct uwsgi_offload_request *uor, int fd) { + + if (!uor->buf) { + uor->buf = uwsgi_malloc(32768); + uor->to_write = 0; + } + if (uor->to_write == 0) { + ssize_t len = read(uor->fd, uor->buf, 32768); + if (len > 0) { + uor->to_write = len; + uor->buf_pos = 0; + continue; + } + else if (len < 0) { + uwsgi_error("read()"); + } + uwsgi_offload_close(uor); + continue; + } + ssize_t len = write(uor->s, uor->buf + uor->buf_pos, uor->to_write); + if (len > 0) { + uor->written += len; + uor->to_write -= len; + uor->buf_pos += len; + if (uor->written >= uor->len) { + uwsgi_offload_close(uor); + } + continue; + } + else if (len < 0) { + if (errno == EAGAIN) + continue; + uwsgi_error("write()"); + } + + return -1; + +} +*/ + +/* + +the offload task starts after having acquired the file fd + + uor->len -> the size of the file + uor->pos -> start writing from pos (default 0) + + status: none + +*/ + +static int uwsgi_offload_sendfile_transfer(struct uwsgi_thread *ut, struct uwsgi_offload_request *uor, int fd) { + + if (fd == -1) { + if (event_queue_add_fd_write(ut->queue, uor->s)) return -1; + return 0; + } +#if defined(__linux__) + ssize_t len = sendfile(uor->s, uor->fd, &uor->pos, 128 * 1024); + if (len > 0) { + uor->written += len; + if (uor->written >= uor->len) { + return -1; + } + } + else if (len < 0) { + uwsgi_offload_retry + uwsgi_error("sendfile()"); + } +#endif + return -1; + +} + +/* +the offload task starts soon after the call to connect() + + status: + 0 -> waiting for connection on fd + 1 -> sending request to fd (write event) + 2 -> start waiting for read on s and fd + 3/a -> if read on s, stop all and wait for write on fd + 3/b -> if read on fd, stop all and wait for write on s + 3/c -> if one of the peer fail, just destroy the task +*/ +static int uwsgi_offload_net_transfer(struct uwsgi_thread *ut, struct uwsgi_offload_request *uor, int fd) { + + ssize_t rlen; + + // setup + if (fd == -1) { + event_queue_add_fd_write(ut->queue, uor->fd); + return 0; + } + + switch(uor->status) { + // waiting for connection + case 0: + if (fd == uor->fd) { + uor->status = 1; + // ok try to send the request right now... + return uwsgi_offload_net_transfer(ut, uor, fd); + } + return -1; + // write event (or just connected) + case 1: + if (fd == uor->fd) { + rlen = write(uor->fd, uor->ubuf->buf + uor->written, uor->ubuf->pos-uor->written); + if (rlen > 0) { + uor->written += rlen; + if (uor->written >= (size_t)uor->ubuf->pos) { + uor->status = 2; + if (event_queue_add_fd_read(ut->queue, uor->s)) return -1; + if (event_queue_fd_write_to_read(ut->queue, uor->fd)) return -1; + } + return 0; + } + else if (rlen < 0) { + uwsgi_offload_retry + uwsgi_error("uwsgi_offload_net_transfer() -> write()"); + } + } + return -1; + // read event from s or fd + case 2: + if (!uor->buf) { + uor->buf = uwsgi_malloc(4096); + } + if (fd == uor->fd) { + rlen = read(uor->fd, uor->buf, 4096); + if (rlen > 0) { + uor->to_write = rlen; + uor->pos = 0; + uwsgi_offload_0r_1w(uor->fd, uor->s) + uor->status = 3; + return 0; + } + if (rlen < 0) { + uwsgi_offload_retry + uwsgi_error("uwsgi_offload_net_transfer() -> read()/fd"); + } + } + else if (fd == uor->s) { + rlen = read(uor->fd, uor->buf, 4096); + if (rlen > 0) { + uor->to_write = rlen; + uor->pos = 0; + uwsgi_offload_0r_1w(uor->s, uor->fd) + uor->status = 4; + return 0; + } + if (rlen < 0) { + uwsgi_offload_retry + uwsgi_error("uwsgi_offload_net_transfer() -> read()/s"); + } + } + return -1; + // write event on s + case 3: + rlen = write(uor->s, uor->buf + uor->pos, uor->to_write); + if (rlen > 0) { + uor->to_write -= rlen; + uor->pos += rlen; + if (uor->to_write == 0) { + if (event_queue_fd_write_to_read(ut->queue, uor->s)) return -1; + if (event_queue_add_fd_read(ut->queue, uor->fd)) return -1; + uor->status = 2; + } + return 0; + } + else if (rlen < 0) { + uwsgi_offload_retry + uwsgi_error("uwsgi_offload_net_transfer() -> write()/s"); + } + return -1; + // write event on fd + case 4: + rlen = write(uor->fd, uor->buf + uor->pos, uor->to_write); + if (rlen > 0) { + uor->to_write -= rlen; + uor->pos += rlen; + if (uor->to_write == 0) { + if (event_queue_fd_write_to_read(ut->queue, uor->fd)) return -1; + if (event_queue_add_fd_read(ut->queue, uor->s)) return -1; + uor->status = 2; + } + return 0; + } + else if (rlen < 0) { + uwsgi_offload_retry + uwsgi_error("uwsgi_offload_net_transfer() -> write()/fd"); + } + return -1; + default: + break; + } + + return -1; +} diff --git a/core/protocol.c b/core/protocol.c index cd4f5455..1fe43206 100644 --- a/core/protocol.c +++ b/core/protocol.c @@ -1969,7 +1969,7 @@ int uwsgi_real_file_serve(struct wsgi_request *wsgi_req, char *real_filename, si // Ok, the file must be transferred from uWSGI if (wsgi_req->socket->can_offload) { - if (!uwsgi_offload_request_do(wsgi_req, real_filename, st->st_size)) { + if (!uwsgi_offload_request_sendfile_do(wsgi_req, real_filename, st->st_size)) { wsgi_req->status = -30; return 0; } diff --git a/core/routing.c b/core/routing.c index 329e21fa..3b907dc8 100644 --- a/core/routing.c +++ b/core/routing.c @@ -23,6 +23,7 @@ int uwsgi_apply_routes(struct wsgi_request *wsgi_req) { int n = uwsgi_regexp_match_ovec(routes->pattern, routes->pattern_extra, *subject, *subject_len, routes->ovector, routes->ovn); if (n >= 0) { int ret = routes->func(wsgi_req, routes); + if (ret == UWSGI_ROUTE_BREAK) uwsgi.workers[uwsgi.mywid].cores[wsgi_req->async_id].routed_requests++; if (ret != UWSGI_ROUTE_NEXT) { return ret; } @@ -44,6 +45,7 @@ int uwsgi_apply_routes_fast(struct wsgi_request *wsgi_req, char *uri, int len) { int n = uwsgi_regexp_match_ovec(routes->pattern, routes->pattern_extra, uri, len, routes->ovector, routes->ovn); if (n >= 0) { int ret = routes->func(wsgi_req, routes); + if (ret == UWSGI_ROUTE_BREAK) uwsgi.workers[uwsgi.mywid].cores[wsgi_req->async_id].routed_requests++; if (ret != UWSGI_ROUTE_NEXT) { return ret; } diff --git a/core/socket.c b/core/socket.c index ddc3ac3b..9bab6e0e 100644 --- a/core/socket.c +++ b/core/socket.c @@ -1785,7 +1785,7 @@ setup_proto: uwsgi_sock->proto_writev_header = uwsgi_proto_uwsgi_writev_header; uwsgi_sock->proto_sendfile = NULL; uwsgi_sock->proto_close = uwsgi_proto_base_close; - if (uwsgi.static_offload_to_thread) + if (uwsgi.offload_threads > 0) uwsgi_sock->can_offload = 1; } else if (requested_protocol && (!strcmp("fastcgi", requested_protocol) || !strcmp("fcgi", requested_protocol))) { @@ -1815,7 +1815,7 @@ setup_proto: uwsgi_sock->proto_writev_header = uwsgi_proto_uwsgi_writev_header; uwsgi_sock->proto_sendfile = NULL; uwsgi_sock->proto_close = uwsgi_proto_base_close; - if (uwsgi.static_offload_to_thread) + if (uwsgi.offload_threads > 0) uwsgi_sock->can_offload = 1; } nextsock: diff --git a/core/uwsgi.c b/core/uwsgi.c index 6050819f..05c98e0a 100644 --- a/core/uwsgi.c +++ b/core/uwsgi.c @@ -468,7 +468,7 @@ static struct uwsgi_option uwsgi_base_options[] = { {"static-expires-path-info-mtime", required_argument, 0, "set the Expires header based on PATH_INFO regexp and file mtime", uwsgi_opt_add_regexp_dyn_dict, &uwsgi.static_expires_path_info_mtime, UWSGI_OPT_MIME}, #endif - {"static-offload-to-thread", required_argument, 0, "offload static file serving to a thread (upto the specified number of threads)", uwsgi_opt_set_int, &uwsgi.static_offload_to_thread, UWSGI_OPT_MIME}, + {"offload-threads", required_argument, 0, "set the number of offload threads to spawn (per-worker, default 0)", uwsgi_opt_set_int, &uwsgi.offload_threads, 0}, {"file-serve-mode", required_argument, 0, "set static file serving mode", uwsgi_opt_fileserve_mode, NULL, UWSGI_OPT_MIME}, {"fileserve-mode", required_argument, 0, "set static file serving mode", uwsgi_opt_fileserve_mode, NULL, UWSGI_OPT_MIME}, @@ -513,6 +513,7 @@ static struct uwsgi_option uwsgi_base_options[] = { {"remap-modifier", required_argument, 0, "remap request modifier from one id to another", uwsgi_opt_set_str, &uwsgi.remap_modifier, 0}, {"app", required_argument, 0, "*** deprecated ***", uwsgi_opt_deprecated, (void *) "use the more advanced \"mount\" option", 0}, + {"static-offload-to-thread", required_argument, 0, "*** deprecated ***", uwsgi_opt_deprecated, (void *) "use the more advanced \"offload-threads\" option", 0}, {"dump-options", no_argument, 0, "dump the full list of available options", uwsgi_opt_true, &uwsgi.dump_options, 0}, {"show-config", no_argument, 0, "show the current config reformatted as ini", uwsgi_opt_true, &uwsgi.show_config, 0}, @@ -2684,11 +2685,11 @@ next2: // set default wsgi_req (for loading apps); uwsgi.wsgi_req = &uwsgi.workers[uwsgi.mywid].cores[0].req; - if (uwsgi.static_offload_to_thread) { + if (uwsgi.offload_threads > 0) { uwsgi.offload_thread = uwsgi_offload_thread_start(); if (!uwsgi.offload_thread) { - uwsgi_log("unable to offload static file serving !!!\n"); - uwsgi.static_offload_to_thread = 0; + uwsgi_log("unable to start offload thread for worker %d !!!\n", uwsgi.mywid); + uwsgi.offload_threads = 0; } } diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index bd48bca0..2f20c3d9 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -804,7 +804,7 @@ PyObject *py_uwsgi_offload_transfer(PyObject * self, PyObject * args) { UWSGI_RELEASE_GIL - if (uwsgi_offload_request_do(wsgi_req, filename, len)) { + if (uwsgi_offload_request_sendfile_do(wsgi_req, filename, len)) { UWSGI_GET_GIL return PyErr_Format(PyExc_ValueError, "Unable to offload the request"); } diff --git a/plugins/router_http/router_http.c b/plugins/router_http/router_http.c index 3073def6..5c5c75c0 100644 --- a/plugins/router_http/router_http.c +++ b/plugins/router_http/router_http.c @@ -12,6 +12,21 @@ int uwsgi_routing_func_http(struct wsgi_request *wsgi_req, struct uwsgi_route *u // get the http address from the route char *addr = ur->data; + // convert the wsgi_request to an http proxy request + struct uwsgi_buffer *ub = uwsgi_to_http(wsgi_req, ur->data2, ur->data2_len); + if (!ub) { + uwsgi_log("unable to generate http request for %s\n", addr); + return UWSGI_ROUTE_NEXT; + } + + // ok now if have offload threads, directly use them + if (wsgi_req->socket->can_offload) { + if (!uwsgi_offload_request_net_do(wsgi_req, addr, ub)) { + wsgi_req->status = -30; + return UWSGI_ROUTE_BREAK; + } + } + // connect to the http server int http_fd = uwsgi_connect(addr, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0); if (http_fd < 0) { @@ -19,14 +34,6 @@ int uwsgi_routing_func_http(struct wsgi_request *wsgi_req, struct uwsgi_route *u return UWSGI_ROUTE_NEXT; } - // convert the wsgi_request to an http proxy request - struct uwsgi_buffer *ub = uwsgi_to_http(wsgi_req, ur->data2, ur->data2_len); - if (!ub) { - uwsgi_log("unable to generate http request for %s\n", addr); - close(http_fd); - return UWSGI_ROUTE_NEXT; - } - // send the request if (uwsgi_buffer_send(ub, http_fd)) { uwsgi_log("error routing request to http server %s\n", addr); diff --git a/uwsgi.h b/uwsgi.h index 491937cf..1c70a47f 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1518,8 +1518,11 @@ struct uwsgi_server { struct uwsgi_dyn_dict *static_expires_path_info; struct uwsgi_dyn_dict *static_expires_path_info_mtime; + int offload_threads; + int offload_threads_events; + int check_static_docroot; - int static_offload_to_thread; + struct uwsgi_thread *offload_thread; // linked list for offloaded requests struct uwsgi_offload_request *offload_requests_head; @@ -2056,6 +2059,7 @@ struct uwsgi_core { uint64_t requests; uint64_t failed_requests; uint64_t static_requests; + uint64_t routed_requests; #ifdef UWSGI_THREADING pthread_t thread_id; @@ -3401,20 +3405,34 @@ struct uwsgi_thread { struct uwsgi_thread *uwsgi_thread_new(void (*)(struct uwsgi_thread *)); struct uwsgi_offload_request { + // the request socket int s; + // the peer int fd; + + // internal state + int status; + off_t pos; char *buf; off_t buf_pos; + size_t to_write; size_t len; size_t written; + + // a uwsgi_buffer (will be destroyed at the end of the task) + struct uwsgi_buffer *ubuf; + + int (*func)(struct uwsgi_thread *, struct uwsgi_offload_request *, int); + struct uwsgi_offload_request *prev; struct uwsgi_offload_request *next; }; struct uwsgi_thread *uwsgi_offload_thread_start(void); -int uwsgi_offload_request_do(struct wsgi_request *, char *, size_t); +int uwsgi_offload_request_sendfile_do(struct wsgi_request *, char *, size_t); +int uwsgi_offload_request_net_do(struct wsgi_request *, char *, struct uwsgi_buffer *); void uwsgi_subscription_set_algo(char *); struct uwsgi_subscribe_slot **uwsgi_subscription_init_ht(void);