nxt_router.c (165:a1156473a8f1) nxt_router.c (166:347dae08632b)
1
2/*
3 * Copyright (C) Igor Sysoev
4 * Copyright (C) Valentin V. Bartenev
5 * Copyright (C) NGINX, Inc.
6 */
7
8#include <nxt_router.h>

--- 159 unchanged lines hidden (view full) ---

168 nxt_mp_release(sw->mem_pool, sw);
169}
170
171
172void
173nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
174{
175 nxt_start_worker_t *sw;
1
2/*
3 * Copyright (C) Igor Sysoev
4 * Copyright (C) Valentin V. Bartenev
5 * Copyright (C) NGINX, Inc.
6 */
7
8#include <nxt_router.h>

--- 159 unchanged lines hidden (view full) ---

168 nxt_mp_release(sw->mem_pool, sw);
169}
170
171
172void
173nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
174{
175 nxt_start_worker_t *sw;
176 nxt_event_engine_t *engine;
176
177 nxt_port_new_port_handler(task, msg);
178
179 if (msg->new_port == NULL || msg->new_port->type != NXT_PROCESS_WORKER) {
180 return;
181 }
182
183 sw = nxt_router_sw_find_remove(task, nxt_router, msg->port_msg.stream);

--- 4 unchanged lines hidden (view full) ---

188 sw->app->workers++;
189
190 nxt_assert(sw->app->pending_workers != 0);
191
192 sw->app->pending_workers--;
193
194 nxt_router_app_release_port(task, msg->new_port, sw->app);
195
177
178 nxt_port_new_port_handler(task, msg);
179
180 if (msg->new_port == NULL || msg->new_port->type != NXT_PROCESS_WORKER) {
181 return;
182 }
183
184 sw = nxt_router_sw_find_remove(task, nxt_router, msg->port_msg.stream);

--- 4 unchanged lines hidden (view full) ---

189 sw->app->workers++;
190
191 nxt_assert(sw->app->pending_workers != 0);
192
193 sw->app->pending_workers--;
194
195 nxt_router_app_release_port(task, msg->new_port, sw->app);
196
197 engine = sw->work.data;
198
196 sw->work.handler = nxt_router_sw_release;
199 sw->work.handler = nxt_router_sw_release;
200 sw->work.task = &engine->task;
197
198 nxt_debug(task, "post sw #%uxD release to %p", sw->stream,
199 sw->work.data);
200
201
202 nxt_debug(task, "post sw #%uxD release to %p", sw->stream,
203 sw->work.data);
204
201 nxt_event_engine_post(sw->work.data, &sw->work);
205 nxt_event_engine_post(engine, &sw->work);
202 }
203}
204
205
206void
207nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
208{
209 size_t dump_size;

--- 1562 unchanged lines hidden (view full) ---

1772static void
1773nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
1774{
1775 nxt_buf_t *b;
1776 nxt_app_t *app;
1777 nxt_port_t *port;
1778 nxt_runtime_t *rt;
1779 nxt_start_worker_t *sw;
206 }
207}
208
209
210void
211nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
212{
213 size_t dump_size;

--- 1562 unchanged lines hidden (view full) ---

1776static void
1777nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
1778{
1779 nxt_buf_t *b;
1780 nxt_app_t *app;
1781 nxt_port_t *port;
1782 nxt_runtime_t *rt;
1783 nxt_start_worker_t *sw;
1784 nxt_event_engine_t *engine;
1780
1781 sw = obj;
1782 app = sw->app;
1783
1784 if (app->workers + app->pending_workers >= app->max_workers) {
1785
1786 sw = obj;
1787 app = sw->app;
1788
1789 if (app->workers + app->pending_workers >= app->max_workers) {
1790 engine = sw->work.data;
1791
1785 sw->work.handler = nxt_router_sw_release;
1792 sw->work.handler = nxt_router_sw_release;
1793 sw->work.task = &engine->task;
1786
1787 nxt_debug(task, "%uD/%uD running/penging workers, post sw #%uxD "
1788 "release to %p", sw->stream, sw->work.data);
1789
1794
1795 nxt_debug(task, "%uD/%uD running/penging workers, post sw #%uxD "
1796 "release to %p", sw->stream, sw->work.data);
1797
1790 nxt_event_engine_post(sw->work.data, &sw->work);
1798 nxt_event_engine_post(engine, &sw->work);
1791
1792 return;
1793 }
1794
1795 app->pending_workers++;
1796
1797 nxt_debug(task, "send sw #%uD", sw->stream);
1798

--- 72 unchanged lines hidden (view full) ---

1871
1872 if (task->thread->engine != port->engine) {
1873 work = &port->work;
1874
1875 nxt_debug(task, "post release port to engine %p", port->engine);
1876
1877 work->next = NULL;
1878 work->handler = nxt_router_app_release_port;
1799
1800 return;
1801 }
1802
1803 app->pending_workers++;
1804
1805 nxt_debug(task, "send sw #%uD", sw->stream);
1806

--- 72 unchanged lines hidden (view full) ---

1879
1880 if (task->thread->engine != port->engine) {
1881 work = &port->work;
1882
1883 nxt_debug(task, "post release port to engine %p", port->engine);
1884
1885 work->next = NULL;
1886 work->handler = nxt_router_app_release_port;
1879 work->task = port->socket.task;
1887 work->task = &port->engine->task;
1880 work->obj = port;
1881 work->data = app;
1882
1883 nxt_event_engine_post(port->engine, work);
1884
1885 return;
1886 }
1887

--- 130 unchanged lines hidden (view full) ---

2018
2019 nxt_memzero(sw, sizeof(nxt_start_worker_t));
2020
2021 sw->stream = nxt_random(&task->thread->random);
2022 sw->app = app;
2023 sw->rc = rc;
2024 sw->mem_pool = c->mem_pool;
2025
1888 work->obj = port;
1889 work->data = app;
1890
1891 nxt_event_engine_post(port->engine, work);
1892
1893 return;
1894 }
1895

--- 130 unchanged lines hidden (view full) ---

2026
2027 nxt_memzero(sw, sizeof(nxt_start_worker_t));
2028
2029 sw->stream = nxt_random(&task->thread->random);
2030 sw->app = app;
2031 sw->rc = rc;
2032 sw->mem_pool = c->mem_pool;
2033
2034 rt = task->thread->runtime;
2035 master_port = rt->port_by_type[NXT_PROCESS_MASTER];
2036
2026 sw->work.handler = nxt_router_send_sw_request;
2037 sw->work.handler = nxt_router_send_sw_request;
2027 sw->work.task = task;
2038 sw->work.task = &master_port->engine->task;
2028 sw->work.obj = sw;
2029 sw->work.data = task->thread->engine;
2030
2039 sw->work.obj = sw;
2040 sw->work.data = task->thread->engine;
2041
2031 rt = task->thread->runtime;
2032
2033 master_port = rt->port_by_type[NXT_PROCESS_MASTER];
2034
2035 nxt_debug(task, "post send sw %uxD to master engine %p", sw->stream,
2036 master_port->engine);
2037
2038 nxt_event_engine_post(master_port->engine, &sw->work);
2039
2040 return NXT_AGAIN;
2041}
2042

--- 457 unchanged lines hidden ---
2042 nxt_debug(task, "post send sw %uxD to master engine %p", sw->stream,
2043 master_port->engine);
2044
2045 nxt_event_engine_post(master_port->engine, &sw->work);
2046
2047 return NXT_AGAIN;
2048}
2049

--- 457 unchanged lines hidden ---