konevsa/haproxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MEDIUM: spoe: Refactor SPOE appctx creation

Browse Source 
A .init callback function is defined for the spoe_applet applet. This
function finishes the spoe_appctx initialization. It also finishes the
appctx startup by calling appctx_finalize_startup() and its handles the
stream customization.
This commit is contained in:
Christopher Faulet 2022-05-12 15:28:51 +02:00
parent 9223851a80
commit 69ebc3093a
1 changed files with 68 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
src/flt_spoe.c
View File

@ -1214,6 +1214,52 @@ spoe_process_appctx(struct task * task, void *context, unsigned int state)
return task;
}
static int
spoe_init_appctx(struct appctx *appctx)
{
struct spoe_appctx *spoe_appctx = SPOE_APPCTX(appctx);
struct spoe_agent *agent = spoe_appctx->agent;
struct task *task;
struct stream *s;
if ((task = task_new_here()) == NULL)
goto out_free_task;
task->process = spoe_process_appctx;
task->context = appctx;
if (appctx_finalize_startup(appctx, &agent->spoe_conf->agent_fe, &BUF_NULL) == -1)
goto out_error;
spoe_appctx->owner = appctx;
spoe_appctx->task = task;
LIST_INIT(&spoe_appctx->buffer_wait.list);
spoe_appctx->buffer_wait.target = appctx;
spoe_appctx->buffer_wait.wakeup_cb = (int (*)(void *))spoe_wakeup_appctx;
s = appctx_strm(appctx);
stream_set_backend(s, agent->b.be);
/* applet is waiting for data */
cs_cant_get(s->csf);
s->do_log = NULL;
s->res.flags |= CF_READ_DONTWAIT;
HA_SPIN_LOCK(SPOE_APPLET_LOCK, &agent->rt[tid].lock);
LIST_APPEND(&agent->rt[tid].applets, &spoe_appctx->list);
HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &agent->rt[tid].lock);
_HA_ATOMIC_INC(&agent->counters.applets);
appctx->st0 = SPOE_APPCTX_ST_CONNECT;
task_wakeup(spoe_appctx->task, TASK_WOKEN_INIT);
return 0;
out_free_task:
task_destroy(task);
out_error:
return -1;
}
/* Callback function that releases a SPOE applet. This happens when the
* connection with the agent is closed. */
static void
@ -1980,6 +2026,7 @@ struct applet spoe_applet = {
.obj_type = OBJ_TYPE_APPLET,
.name = "<SPOE>", /* used for logging */
.fct = spoe_handle_appctx,
.init = spoe_init_appctx,
.release = spoe_release_appctx,
};
@ -1988,74 +2035,39 @@ struct applet spoe_applet = {
static struct appctx *
spoe_create_appctx(struct spoe_config *conf)
{
struct spoe_appctx *spoe_appctx;
struct appctx *appctx;
struct session *sess;
struct conn_stream *cs;
struct stream *strm;
if ((appctx = appctx_new(&spoe_applet, NULL)) == NULL)
spoe_appctx = pool_zalloc(pool_head_spoe_appctx);
if (spoe_appctx == NULL)
goto out_error;
appctx->svcctx = pool_zalloc(pool_head_spoe_appctx);
if (SPOE_APPCTX(appctx) == NULL)
goto out_free_appctx;
spoe_appctx->agent = conf->agent;
spoe_appctx->version = 0;
spoe_appctx->max_frame_size = conf->agent->max_frame_size;
spoe_appctx->flags = 0;
spoe_appctx->status_code = SPOE_FRM_ERR_NONE;
spoe_appctx->buffer = BUF_NULL;
spoe_appctx->cur_fpa = 0;
LIST_INIT(&spoe_appctx->list);
LIST_INIT(&spoe_appctx->waiting_queue);
appctx->st0 = SPOE_APPCTX_ST_CONNECT;
if ((SPOE_APPCTX(appctx)->task = task_new_here()) == NULL)
if ((appctx = appctx_new(&spoe_applet, NULL)) == NULL)
goto out_free_spoe_appctx;
SPOE_APPCTX(appctx)->owner = appctx;
SPOE_APPCTX(appctx)->task->process = spoe_process_appctx;
SPOE_APPCTX(appctx)->task->context = appctx;
SPOE_APPCTX(appctx)->agent = conf->agent;
SPOE_APPCTX(appctx)->version = 0;
SPOE_APPCTX(appctx)->max_frame_size = conf->agent->max_frame_size;
SPOE_APPCTX(appctx)->flags = 0;
SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NONE;
SPOE_APPCTX(appctx)->buffer = BUF_NULL;
SPOE_APPCTX(appctx)->cur_fpa = 0;
appctx->svcctx = spoe_appctx;
if (appctx_init(appctx) == -1)
goto out_free_appctx;
LIST_INIT(&SPOE_APPCTX(appctx)->buffer_wait.list);
SPOE_APPCTX(appctx)->buffer_wait.target = appctx;
SPOE_APPCTX(appctx)->buffer_wait.wakeup_cb = (int (*)(void *))spoe_wakeup_appctx;
LIST_INIT(&SPOE_APPCTX(appctx)->list);
LIST_INIT(&SPOE_APPCTX(appctx)->waiting_queue);
sess = session_new(&conf->agent_fe, NULL, &appctx->obj_type);
if (!sess)
goto out_free_spoe;
appctx->sess = sess;
cs = cs_new_from_endp(appctx->endp, sess, &BUF_NULL);
if (!cs)
goto out_free_spoe;
strm = DISGUISE(cs_strm(cs));
stream_set_backend(strm, conf->agent->b.be);
/* applet is waiting for data */
cs_cant_get(strm->csf);
appctx_wakeup(appctx);
strm->do_log = NULL;
strm->res.flags |= CF_READ_DONTWAIT;
HA_SPIN_LOCK(SPOE_APPLET_LOCK, &conf->agent->rt[tid].lock);
LIST_APPEND(&conf->agent->rt[tid].applets, &SPOE_APPCTX(appctx)->list);
HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &conf->agent->rt[tid].lock);
_HA_ATOMIC_INC(&conf->agent->counters.applets);
task_wakeup(SPOE_APPCTX(appctx)->task, TASK_WOKEN_INIT);
return appctx;
/* Error unrolling */
out_free_spoe:
task_destroy(SPOE_APPCTX(appctx)->task);
out_free_spoe_appctx:
pool_free(pool_head_spoe_appctx, SPOE_APPCTX(appctx));
out_free_appctx:
appctx_free(appctx);
appctx_free_on_early_error(appctx);
out_free_spoe_appctx:
pool_free(pool_head_spoe_appctx, spoe_appctx);
out_error:
return NULL;
}