From: Wenkai Lin <linwenkai6@hisilicon.com> For big data acceleration algorithms, the API layer needs to be adapted to the new UADK heterogeneous hybrid acceleration framework, thereby ensuring that big data acceleration functionality can be adapted to and support the new hybrid acceleration framework. Signed-off-by: Wenkai Lin <linwenkai6@hisilicon.com> Signed-off-by: Zhushuai Yin <yinzhushuai@huawei.com> --- drv/hisi_dae.c | 50 ++++-- drv/hisi_dae.h | 8 +- drv/hisi_dae_common.c | 101 ++++++----- drv/hisi_dae_join_gather.c | 36 ++-- include/drv/wd_agg_drv.h | 6 +- include/drv/wd_join_gather_drv.h | 12 +- wd_agg.c | 239 +++++++++++++++++--------- wd_join_gather.c | 286 +++++++++++++++++++------------ 8 files changed, 459 insertions(+), 279 deletions(-) diff --git a/drv/hisi_dae.c b/drv/hisi_dae.c index d86b9f6..d495541 100644 --- a/drv/hisi_dae.c +++ b/drv/hisi_dae.c @@ -126,6 +126,8 @@ static void fill_hashagg_task_type(struct wd_agg_msg *msg, struct dae_sqe *sqe, else sqe->task_type_ext = DAE_HASHAGG_OUTPUT; break; + default: + break; } } @@ -200,6 +202,8 @@ static void fill_hashagg_table_data(struct dae_sqe *sqe, struct dae_addr_list *a hw_table = &addr_list->src_table; table_data = &agg_ctx->rehash_table; break; + default: + break; } sqe->table_row_size = agg_ctx->row_size; @@ -282,7 +286,6 @@ static void fill_hashagg_key_data(struct dae_sqe *sqe, struct dae_ext_sqe *ext_s } } } - static void fill_hashagg_merge_key_data(struct dae_sqe *sqe, struct dae_ext_sqe *ext_sqe, struct dae_addr_list *addr_list, struct wd_agg_msg *msg) { @@ -350,6 +353,8 @@ static void fill_hashagg_input_data(struct dae_sqe *sqe, struct dae_ext_sqe *ext agg_col_num = cols_data->output_num; fill_hashagg_data_info(sqe, ext_sqe, cols_data->input_data, cols_data->input_num); break; + default: + break; } for (i = 0; i < agg_col_num; i++) { @@ -426,7 +431,7 @@ static int check_hashagg_param(struct wd_agg_msg *msg) return WD_SUCCESS; } -static int hashagg_send(struct wd_alg_driver *drv, handle_t ctx, void *hashagg_msg) +static int hashagg_send(handle_t ctx, void *hashagg_msg) { handle_t h_qp = (handle_t)wd_ctx_get_priv(ctx); struct hisi_qp *qp = (struct hisi_qp *)h_qp; @@ -448,6 +453,10 @@ static int hashagg_send(struct wd_alg_driver *drv, handle_t ctx, void *hashagg_m fill_hashagg_task_type(msg, &sqe, qp->q_info.hw_type); sqe.data_row_num = msg->row_count; + if (qp->q_info.hw_type >= HISI_QM_API_VER5_BASE && msg->pos == WD_AGG_STREAM_INPUT) { + sqe.init_row_num = msg->in_row_count; + sqe.break_point_en = sqe.init_row_num ? 1 : 0; + } idx = get_free_ext_addr(ext_addr); if (idx < 0) @@ -504,7 +513,7 @@ static void fill_sum_overflow_cols(struct dae_sqe *sqe, struct wd_agg_msg *msg, } static void fill_hashagg_msg_task_done(struct dae_sqe *sqe, struct wd_agg_msg *msg, - struct wd_agg_msg *temp_msg, struct hashagg_ctx *agg_ctx) + struct wd_agg_msg *temp_msg, __u16 hw_type) { if (sqe->task_type_ext == DAE_HASHAGG_OUTPUT) { msg->out_row_count = sqe->out_raw_num; @@ -515,11 +524,15 @@ static void fill_hashagg_msg_task_done(struct dae_sqe *sqe, struct wd_agg_msg *m msg->out_row_count = temp_msg->row_count; } else { msg->in_row_count = temp_msg->row_count; + if (hw_type >= HISI_QM_API_VER5_BASE) { + if (!sqe->output_end) + msg->in_row_count = sqe->data_row_offset; + } } } static void fill_hashagg_msg_task_err(struct dae_sqe *sqe, struct wd_agg_msg *msg, - struct wd_agg_msg *temp_msg, struct hashagg_ctx *agg_ctx) + struct wd_agg_msg *temp_msg, __u16 hw_type) { switch (sqe->err_type) { case DAE_TASK_BD_ERROR_MIN ... DAE_TASK_BD_ERROR_MAX: @@ -541,6 +554,11 @@ static void fill_hashagg_msg_task_err(struct dae_sqe *sqe, struct wd_agg_msg *ms case DAE_HASHAGG_RESULT_OVERFLOW: msg->in_row_count = temp_msg->row_count; msg->result = WD_AGG_SUM_OVERFLOW; + if (hw_type >= HISI_QM_API_VER5_BASE && + sqe->task_type_ext == DAE_HASH_AGGREGATE) { + if (!sqe->output_end) + msg->in_row_count = sqe->data_row_offset; + } break; case DAE_TASK_BUS_ERROR: WD_ERR("failed to do hashagg task, bus error! etype %u!\n", sqe->err_type); @@ -557,13 +575,14 @@ static void fill_hashagg_msg_task_err(struct dae_sqe *sqe, struct wd_agg_msg *ms break; } - if (sqe->task_type_ext == DAE_HASHAGG_OUTPUT) { + if (sqe->task_type_ext == DAE_HASHAGG_OUTPUT || + sqe->task_type_ext == DAE_HASHAGG_MERGE) { msg->out_row_count = sqe->out_raw_num; msg->output_done = sqe->output_end; } } -static int hashagg_recv(struct wd_alg_driver *drv, handle_t ctx, void *hashagg_msg) +static int hashagg_recv(handle_t ctx, void *hashagg_msg) { handle_t h_qp = (handle_t)wd_ctx_get_priv(ctx); struct hisi_qp *qp = (struct hisi_qp *)h_qp; @@ -606,9 +625,9 @@ static int hashagg_recv(struct wd_alg_driver *drv, handle_t ctx, void *hashagg_m msg->in_row_count = 0; if (likely(sqe.done_flag == DAE_HW_TASK_DONE)) { - fill_hashagg_msg_task_done(&sqe, msg, temp_msg, agg_ctx); + fill_hashagg_msg_task_done(&sqe, msg, temp_msg, qp->q_info.hw_type); } else if (sqe.done_flag == DAE_HW_TASK_ERR) { - fill_hashagg_msg_task_err(&sqe, msg, temp_msg, agg_ctx); + fill_hashagg_msg_task_err(&sqe, msg, temp_msg, qp->q_info.hw_type); } else { msg->result = WD_AGG_PARSE_ERROR; WD_ERR("failed to do hashagg task, hardware does not process the task!\n"); @@ -1214,17 +1233,11 @@ static void hashagg_sess_priv_uninit(struct wd_alg_driver *drv, void *priv) free(agg_ctx); } -static int hashagg_sess_priv_init(struct wd_alg_driver *drv, - struct wd_agg_sess_setup *setup, void **priv) +static int hashagg_sess_priv_init(struct wd_agg_sess_setup *setup, void **priv) { struct hashagg_ctx *agg_ctx; int ret; - if (!drv || !drv->priv) { - WD_ERR("invalid: dae drv is NULL!\n"); - return -WD_EINVAL; - } - if (!setup || !priv) { WD_ERR("invalid: dae sess priv is NULL!\n"); return -WD_EINVAL; @@ -1265,8 +1278,7 @@ static int agg_get_row_size(struct wd_alg_driver *drv, void *param) return agg_ctx->row_size; } -static int agg_hash_table_init(struct wd_alg_driver *drv, - struct wd_dae_hash_table *hash_table, void *priv) +static int agg_hash_table_init(struct wd_dae_hash_table *hash_table, void *priv) { struct hashagg_ctx *agg_ctx = priv; @@ -1297,6 +1309,8 @@ static struct wd_alg_driver hashagg_driver = { .alg_name = "hashagg", .calc_type = UADK_ALG_HW, .priority = 100, + .priv_size = sizeof(struct hisi_dae_ctx), + .ops_size = sizeof(struct wd_agg_ops), .queue_num = DAE_CTX_Q_NUM_DEF, .op_type_num = 1, .fallback = 0, @@ -1306,6 +1320,8 @@ static struct wd_alg_driver hashagg_driver = { .recv = hashagg_recv, .get_usage = dae_get_usage, .get_extend_ops = dae_get_extend_ops, + .alloc_ctx = wd_hw_alloc_ctx, + .free_ctx = wd_hw_free_ctx, }; #ifdef WD_STATIC_DRV diff --git a/drv/hisi_dae.h b/drv/hisi_dae.h index f82b13c..3ac50eb 100644 --- a/drv/hisi_dae.h +++ b/drv/hisi_dae.h @@ -14,6 +14,7 @@ #include "config.h" #include "wd_alg.h" #include "wd_dae.h" +#include "wd_drv.h" #ifdef __cplusplus extern "C" { @@ -207,11 +208,12 @@ struct hash_table_data { }; struct hisi_dae_ctx { - struct wd_ctx_config_internal config; + struct wd_ctx_internal **ctxs; + __u32 ctx_num; }; -void dae_exit(struct wd_alg_driver *drv); -int dae_init(struct wd_alg_driver *drv, void *conf); +void dae_exit(void *priv); +int dae_init(void *conf, void *priv); int dae_hash_table_init(struct hash_table_data *hw_table, struct hash_table_data *rehash_table, struct wd_dae_hash_table *hash_table, diff --git a/drv/hisi_dae_common.c b/drv/hisi_dae_common.c index c077d1d..24f6481 100644 --- a/drv/hisi_dae_common.c +++ b/drv/hisi_dae_common.c @@ -308,36 +308,60 @@ update_table: return ret; } -int dae_init(struct wd_alg_driver *drv, void *conf) +int dae_init(void *conf, void *priv) { struct wd_ctx_config_internal *config = conf; + struct hisi_dae_ctx *dae_ctx = priv; struct hisi_qm_priv qm_priv; - struct hisi_dae_ctx *priv; - handle_t h_qp = 0; - handle_t h_ctx; - __u32 i, j; + __u32 i, j, count = 0; + bool *is_match; + handle_t h_qp; int ret; if (!config || !config->ctx_num) { - WD_ERR("invalid: dae init config is null or ctx num is 0!\n"); + WD_ERR("invalid: input config or ctx num is null!\n"); + return -WD_EINVAL; + } + + is_match = malloc(config->ctx_num * sizeof(bool)); + if (!is_match) + return -WD_ENOMEM; + + for (i = 0; i < config->ctx_num; i++) { + if (config->ctxs[i].ctx && config->ctxs[i].drv && + !strcmp(config->ctxs[i].drv->drv_name, "hisi_zip")) { + is_match[i] = true; + count++; + } else { + is_match[i] = false; + } + } + if (!count) { + WD_ERR("invalid: valid driver number is zero!\n"); + free(is_match); return -WD_EINVAL; } - priv = malloc(sizeof(struct hisi_dae_ctx)); - if (!priv) + dae_ctx->ctxs = calloc(count, sizeof(struct wd_ctx_internal *)); + if (!dae_ctx->ctxs) { + free(is_match); return -WD_ENOMEM; + } + dae_ctx->ctx_num = count; qm_priv.op_type = DAE_SQC_ALG_TYPE; qm_priv.sqe_size = sizeof(struct dae_sqe); + count = 0; /* Allocate qp for each context */ for (i = 0; i < config->ctx_num; i++) { - h_ctx = config->ctxs[i].ctx; + if (!is_match[i]) + continue; qm_priv.qp_mode = config->ctxs[i].ctx_mode; /* Setting the epoll en to 0 for ASYNC ctx */ qm_priv.epoll_en = (qm_priv.qp_mode == CTX_MODE_SYNC) ? config->epoll_en : 0; qm_priv.idx = i; - h_qp = hisi_qm_alloc_qp(&qm_priv, h_ctx); + h_qp = hisi_qm_alloc_qp(&qm_priv, config->ctxs[i].ctx); if (!h_qp) { ret = -WD_ENOMEM; goto out; @@ -346,58 +370,57 @@ int dae_init(struct wd_alg_driver *drv, void *conf) ret = dae_init_qp_priv(h_qp); if (ret) goto free_h_qp; + dae_ctx->ctxs[count++] = &config->ctxs[i]; } - memcpy(&priv->config, config, sizeof(struct wd_ctx_config_internal)); - drv->priv = priv; + free(is_match); return WD_SUCCESS; free_h_qp: hisi_qm_free_qp(h_qp); out: - for (j = 0; j < i; j++) { - h_qp = (handle_t)wd_ctx_get_priv(config->ctxs[j].ctx); + for (j = 0; j < count; j++) { + h_qp = (handle_t)wd_ctx_get_priv(dae_ctx->ctxs[j]->ctx); if (h_qp) { dae_uninit_qp_priv(h_qp); hisi_qm_free_qp(h_qp); } } - free(priv); + free(dae_ctx->ctxs); + free(is_match); return ret; } -void dae_exit(struct wd_alg_driver *drv) +void dae_exit(void *priv) { - struct wd_ctx_config_internal *config; - struct hisi_dae_ctx *priv; + struct hisi_dae_ctx *dae_ctx = priv; handle_t h_qp; __u32 i; - if (!drv || !drv->priv) + if (!priv) { + WD_ERR("invalid: input parameter is NULL!\n"); return; + } - priv = (struct hisi_dae_ctx *)drv->priv; - config = &priv->config; - for (i = 0; i < config->ctx_num; i++) { - h_qp = (handle_t)wd_ctx_get_priv(config->ctxs[i].ctx); + for (i = 0; i < dae_ctx->ctx_num; i++) { + h_qp = (handle_t)wd_ctx_get_priv(dae_ctx->ctxs[i]->ctx); if (h_qp) { dae_uninit_qp_priv(h_qp); hisi_qm_free_qp(h_qp); } } - - free(priv); - drv->priv = NULL; + if (dae_ctx->ctxs) { + free(dae_ctx->ctxs); + dae_ctx->ctxs = NULL; + } } int dae_get_usage(void *param) { struct hisi_dev_usage *dae_usage = (struct hisi_dev_usage *)param; struct wd_alg_driver *drv = dae_usage->drv; - struct wd_ctx_config_internal *config; - struct hisi_dae_ctx *priv; + struct hisi_dae_ctx *dae_ctx; char *ctx_dev_name; - handle_t ctx = 0; handle_t qp = 0; __u32 i; @@ -406,24 +429,18 @@ int dae_get_usage(void *param) return -WD_EINVAL; } - priv = (struct hisi_dae_ctx *)drv->priv; - if (!priv) + dae_ctx = (struct hisi_dae_ctx *)drv->drv_data; + if (!dae_ctx) return -WD_EACCES; - config = &priv->config; - for (i = 0; i < config->ctx_num; i++) { - ctx_dev_name = wd_ctx_get_dev_name(config->ctxs[i].ctx); + for (i = 0; i < dae_ctx->ctx_num; i++) { + ctx_dev_name = wd_ctx_get_dev_name(dae_ctx->ctxs[i]->ctx); if (!strcmp(dae_usage->dev_name, ctx_dev_name)) { - ctx = config->ctxs[i].ctx; - break; + qp = (handle_t)wd_ctx_get_priv(dae_ctx->ctxs[i]->ctx); + if (qp) + return hisi_qm_get_usage(qp, 0); } } - if (ctx) - qp = (handle_t)wd_ctx_get_priv(ctx); - - if (qp) - return hisi_qm_get_usage(qp, DAE_SQC_ALG_TYPE); - return -WD_EACCES; } diff --git a/drv/hisi_dae_join_gather.c b/drv/hisi_dae_join_gather.c index 5a96d76..98c998f 100644 --- a/drv/hisi_dae_join_gather.c +++ b/drv/hisi_dae_join_gather.c @@ -455,7 +455,7 @@ static int check_join_gather_param(struct wd_join_gather_msg *msg) return WD_SUCCESS; } -static int join_gather_send(struct wd_alg_driver *drv, handle_t ctx, void *send_msg) +static int join_gather_send(handle_t ctx, void *send_msg) { handle_t h_qp = (handle_t)wd_ctx_get_priv(ctx); struct hisi_qp *qp = (struct hisi_qp *)h_qp; @@ -540,7 +540,7 @@ static void fill_join_gather_task_err(struct dae_sqe *sqe, struct wd_join_gather } } -static int join_gather_recv(struct wd_alg_driver *drv, handle_t hctx, void *recv_msg) +static int join_gather_recv(handle_t hctx, void *recv_msg) { handle_t h_qp = (handle_t)wd_ctx_get_priv(hctx); struct hisi_qp *qp = (struct hisi_qp *)h_qp; @@ -861,6 +861,17 @@ static void gather_get_batch_rowsize(struct join_gather_col_data *cols_data, gather_data[i].data_info); batch_row_size[n] = row_count_size + GATHER_ROW_BATCH_EMPTY_SIZE; + /* Align batch_row_size for better gather performance: + * 1-63B -> 32B aligned + * 64-127B -> 64B aligned + * >=128B -> 128B aligned + */ + if (batch_row_size[n] < ROW_SIZE64) + batch_row_size[n] = ALIGN(batch_row_size[n], ROW_SIZE32); + else if (batch_row_size[n] < ROW_SIZE128) + batch_row_size[n] = ALIGN(batch_row_size[n], ROW_SIZE64); + else + batch_row_size[n] = ALIGN(batch_row_size[n], ROW_SIZE128); } } @@ -892,7 +903,7 @@ static int join_gather_fill_ctx(struct join_gather_ctx *ctx, return WD_SUCCESS; } -static void join_gather_sess_priv_uninit(struct wd_alg_driver *drv, void *priv) +static void join_gather_sess_priv_uninit(void *priv) { struct join_gather_ctx *ctx = priv; @@ -904,17 +915,11 @@ static void join_gather_sess_priv_uninit(struct wd_alg_driver *drv, void *priv) free(ctx); } -static int join_gather_sess_priv_init(struct wd_alg_driver *drv, - struct wd_join_gather_sess_setup *setup, void **priv) +static int join_gather_sess_priv_init(struct wd_join_gather_sess_setup *setup, void **priv) { struct join_gather_ctx *ctx; int ret; - if (!drv || !drv->priv) { - WD_ERR("invalid: dae drv is NULL!\n"); - return -WD_EINVAL; - } - if (!setup || !priv) { WD_ERR("invalid: dae sess priv is NULL!\n"); return -WD_EINVAL; @@ -941,7 +946,7 @@ free_ctx: return ret; } -static int join_get_table_row_size(struct wd_alg_driver *drv, void *param) +static int join_get_table_row_size(void *param) { struct join_gather_ctx *ctx = param; @@ -951,7 +956,7 @@ static int join_get_table_row_size(struct wd_alg_driver *drv, void *param) return ctx->hash_table_row_size; } -static int gather_get_batch_row_size(struct wd_alg_driver *drv, void *param, +static int gather_get_batch_row_size(void *param, __u32 *row_size, __u32 size) { struct join_gather_ctx *ctx = param; @@ -967,8 +972,7 @@ static int gather_get_batch_row_size(struct wd_alg_driver *drv, void *param, return 0; } -static int join_hash_table_init(struct wd_alg_driver *drv, - struct wd_dae_hash_table *table, void *priv) +static int join_hash_table_init(struct wd_dae_hash_table *table, void *priv) { struct join_gather_ctx *ctx = priv; @@ -1002,6 +1006,8 @@ static int join_gather_get_extend_ops(void *ops) .alg_name = (dae_alg_name),\ .calc_type = UADK_ALG_HW,\ .priority = 100,\ + .priv_size = sizeof(struct hisi_dae_ctx), \ + .ops_size = sizeof(struct wd_join_gather_ops), \ .queue_num = DAE_CTX_Q_NUM_DEF,\ .op_type_num = 1,\ .fallback = 0,\ @@ -1011,6 +1017,8 @@ static int join_gather_get_extend_ops(void *ops) .recv = join_gather_recv,\ .get_extend_ops = join_gather_get_extend_ops,\ .get_usage = dae_get_usage,\ + .alloc_ctx = wd_hw_alloc_ctx,\ + .free_ctx = wd_hw_free_ctx, \ } static struct wd_alg_driver join_gather_driver[] = { diff --git a/include/drv/wd_agg_drv.h b/include/drv/wd_agg_drv.h index b26b25d..400fc09 100644 --- a/include/drv/wd_agg_drv.h +++ b/include/drv/wd_agg_drv.h @@ -43,11 +43,9 @@ struct wd_agg_msg { struct wd_agg_ops { int (*get_row_size)(struct wd_alg_driver *drv, void *priv); - int (*sess_init)(struct wd_alg_driver *drv, - struct wd_agg_sess_setup *setup, void **priv); + int (*sess_init)(struct wd_agg_sess_setup *setup, void **priv); void (*sess_uninit)(struct wd_alg_driver *drv, void *priv); - int (*hash_table_init)(struct wd_alg_driver *drv, - struct wd_dae_hash_table *hash_table, void *priv); + int (*hash_table_init)(struct wd_dae_hash_table *hash_table, void *priv); }; struct wd_agg_msg *wd_agg_get_msg(__u32 idx, __u32 tag); diff --git a/include/drv/wd_join_gather_drv.h b/include/drv/wd_join_gather_drv.h index dbf4ee7..d7f136d 100644 --- a/include/drv/wd_join_gather_drv.h +++ b/include/drv/wd_join_gather_drv.h @@ -33,14 +33,12 @@ struct wd_join_gather_msg { }; struct wd_join_gather_ops { - int (*get_table_row_size)(struct wd_alg_driver *drv, void *priv); - int (*get_batch_row_size)(struct wd_alg_driver *drv, void *priv, + int (*get_table_row_size)(void *priv); + int (*get_batch_row_size)(void *priv, __u32 *batch_row_size, __u32 size); - int (*sess_init)(struct wd_alg_driver *drv, - struct wd_join_gather_sess_setup *setup, void **priv); - void (*sess_uninit)(struct wd_alg_driver *drv, void *priv); - int (*hash_table_init)(struct wd_alg_driver *drv, - struct wd_dae_hash_table *hash_table, void *priv); + int (*sess_init)(struct wd_join_gather_sess_setup *setup, void **priv); + void (*sess_uninit)(void *priv); + int (*hash_table_init)(struct wd_dae_hash_table *hash_table, void *priv); }; struct wd_join_gather_msg *wd_join_gather_get_msg(__u32 idx, __u32 tag); diff --git a/wd_agg.c b/wd_agg.c index bb834cb..95cb869 100644 --- a/wd_agg.c +++ b/wd_agg.c @@ -32,8 +32,6 @@ struct wd_agg_setting { struct wd_ctx_config_internal config; struct wd_sched sched; struct wd_async_msg_pool pool; - struct wd_alg_driver *driver; - void *priv; void *dlhandle; void *dlh_list; } wd_agg_setting; @@ -78,7 +76,6 @@ static void wd_agg_close_driver(void) wd_dlclose_drv(wd_agg_setting.dlh_list); wd_agg_setting.dlh_list = NULL; #else - wd_release_drv(wd_agg_setting.driver); hisi_dae_remove(); #endif } @@ -353,27 +350,41 @@ out_key: return -WD_ENOMEM; } -static int wd_agg_init_sess_priv(struct wd_agg_sess *sess, struct wd_agg_sess_setup *setup) +static int init_single_driver_session(struct wd_alg_driver *drv, + struct wd_agg_sess *sess, + struct wd_agg_sess_setup *setup) { + struct wd_agg_ops *eops; int ret; - if (sess->ops.sess_init) { - if (!sess->ops.sess_uninit) { + if (!drv->get_extend_ops || !drv->extend_ops) + return WD_SUCCESS; + + ret = drv->get_extend_ops(drv->extend_ops); + if (ret) { + WD_ERR("failed to set session extend ops!\n"); + return -WD_EINVAL; + } + + eops = drv->extend_ops; + + if (eops->sess_init) { + if (!eops->sess_uninit) { WD_ERR("failed to get session uninit ops!\n"); return -WD_EINVAL; } - ret = sess->ops.sess_init(wd_agg_setting.driver, setup, &sess->priv); + ret = eops->sess_init(setup, &sess->priv); if (ret) { WD_ERR("failed to init session priv!\n"); return ret; } } - if (sess->ops.get_row_size) { - ret = sess->ops.get_row_size(wd_agg_setting.driver, sess->priv); + if (eops->get_row_size && sess->priv) { + ret = eops->get_row_size(drv, sess->priv); if (ret <= 0) { - if (sess->ops.sess_uninit) - sess->ops.sess_uninit(wd_agg_setting.driver, sess->priv); + if (eops->sess_uninit) + eops->sess_uninit(drv, sess->priv); WD_ERR("failed to get hash table row size: %d!\n", ret); return -WD_EINVAL; } @@ -383,10 +394,61 @@ static int wd_agg_init_sess_priv(struct wd_agg_sess *sess, struct wd_agg_sess_se return WD_SUCCESS; } +static void cleanup_partial_init(struct wd_agg_sess *sess) +{ + struct wd_ctx_config_internal *config = &wd_agg_setting.config; + struct wd_alg_driver *drv; + struct wd_agg_ops *eops; + + if (!sess->priv) + return; + + drv = config->ctxs[0].drv; + if (drv->extend_ops) { + eops = drv->extend_ops; + if (eops->sess_uninit) + eops->sess_uninit(drv, sess->priv); + } +} + +static int wd_agg_init_sess_priv(struct wd_agg_sess *sess, struct wd_agg_sess_setup *setup) +{ + struct wd_ctx_config_internal *config = &wd_agg_setting.config; + struct wd_alg_driver *first_drv = NULL; + __u32 i; + + /* + * The algorithm always works in stream mode, so it supports only + * the switchover between ctxs of the same driver type. + */ + for (i = 0; i < config->ctx_num; i++) { + if (!first_drv) { + first_drv = config->ctxs[i].drv; + } else if (config->ctxs[i].drv != first_drv) { + WD_ERR("incompatible driver: %s vs %s, only single driver type allowed!\n", + config->ctxs[i].drv->drv_name, first_drv->drv_name); + return -WD_EINVAL; + } + } + + return init_single_driver_session(first_drv, sess, setup); +} + +static int wd_agg_uninit_sess_priv(struct wd_agg_sess *sess) +{ + if (!sess || !sess->priv) + return WD_SUCCESS; + + cleanup_partial_init(sess); + sess->priv = NULL; + return WD_SUCCESS; +} + handle_t wd_agg_alloc_sess(struct wd_agg_sess_setup *setup) { - __u32 out_agg_cols_num = 0; + struct wd_sched_params params; struct wd_agg_sess *sess; + __u32 out_agg_cols_num = 0; int ret; ret = wd_agg_check_sess_params(setup, &out_agg_cols_num); @@ -402,7 +464,7 @@ handle_t wd_agg_alloc_sess(struct wd_agg_sess_setup *setup) sess->agg_conf.out_cols_num = out_agg_cols_num; sess->alg_name = wd_agg_alg_name; - ret = wd_drv_alg_support(sess->alg_name, wd_agg_setting.driver); + ret = wd_drv_alg_support(sess->alg_name, &wd_agg_setting.config); if (!ret) { WD_ERR("failed to support agg algorithm: %s!\n", sess->alg_name); goto free_sess; @@ -415,14 +477,12 @@ handle_t wd_agg_alloc_sess(struct wd_agg_sess_setup *setup) WD_ERR("failed to init agg session schedule key!\n"); goto free_sess; } + params.alg_name = sess->alg_name; + params.ctxs = wd_agg_setting.config.ctxs; + wd_agg_setting.sched.set_param( + wd_agg_setting.sched.h_sched_ctx, + sess->sched_key, ¶ms); - if (wd_agg_setting.driver->get_extend_ops) { - ret = wd_agg_setting.driver->get_extend_ops(&sess->ops); - if (ret) { - WD_ERR("failed to get agg extend ops!\n"); - goto free_key; - } - } ret = wd_agg_init_sess_priv(sess, setup); if (ret) @@ -437,10 +497,16 @@ handle_t wd_agg_alloc_sess(struct wd_agg_sess_setup *setup) return (handle_t)sess; uninit_priv: - if (sess->ops.sess_uninit) - sess->ops.sess_uninit(wd_agg_setting.driver, sess->priv); + wd_agg_uninit_sess_priv(sess); free_key: - free(sess->sched_key); + if (sess->sched_key) { + if (wd_agg_setting.sched.sched_uninit) + wd_agg_setting.sched.sched_uninit( + wd_agg_setting.sched.h_sched_ctx, + (handle_t)sess->sched_key); + else + free(sess->sched_key); + } free_sess: free(sess); return (handle_t)0; @@ -459,10 +525,15 @@ void wd_agg_free_sess(handle_t h_sess) free(sess->agg_conf.cols_info); free(sess->key_conf.data_size); - if (sess->ops.sess_uninit) - sess->ops.sess_uninit(wd_agg_setting.driver, sess->priv); - if (sess->sched_key) - free(sess->sched_key); + wd_agg_uninit_sess_priv(sess); + if (sess->sched_key) { + if (wd_agg_setting.sched.sched_uninit) + wd_agg_setting.sched.sched_uninit( + wd_agg_setting.sched.h_sched_ctx, + (handle_t)sess->sched_key); + else + free(sess->sched_key); + } free(sess); } @@ -510,8 +581,10 @@ static int wd_agg_check_sess_state(struct wd_agg_sess *sess, enum wd_agg_sess_st int wd_agg_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info) { struct wd_agg_sess *sess = (struct wd_agg_sess *)h_sess; + struct wd_ctx_config_internal *config = &wd_agg_setting.config; struct wd_dae_hash_table *hash_table, *rehash_table; enum wd_agg_sess_state expected; + struct wd_agg_ops *eops; int ret; if (!sess || !info) { @@ -548,12 +621,12 @@ int wd_agg_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info) hash_table = &sess->hash_table; rehash_table = &sess->rehash_table; - memcpy(rehash_table, hash_table, sizeof(struct wd_dae_hash_table)); memcpy(hash_table, info, sizeof(struct wd_dae_hash_table)); - if (sess->ops.hash_table_init) { - ret = sess->ops.hash_table_init(wd_agg_setting.driver, hash_table, sess->priv); + eops = config->ctxs[0].drv->extend_ops; + if (eops && eops->hash_table_init) { + ret = eops->hash_table_init(hash_table, sess->priv); if (ret) { memcpy(hash_table, rehash_table, sizeof(struct wd_dae_hash_table)); memset(rehash_table, 0, sizeof(struct wd_dae_hash_table)); @@ -562,12 +635,13 @@ int wd_agg_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info) } return WD_SUCCESS; - out: __atomic_store_n(&sess->state, expected, __ATOMIC_RELEASE); return ret; } +static bool wd_agg_atfork_registered; + static void wd_agg_clear_status(void) { wd_alg_clear_init(&wd_agg_setting.status); @@ -596,14 +670,8 @@ static int wd_agg_alg_init(struct wd_ctx_config *config, struct wd_sched *sched) if (ret < 0) goto out_clear_sched; - ret = wd_alg_init_driver(&wd_agg_setting.config, wd_agg_setting.driver); - if (ret) - goto out_clear_pool; - return WD_SUCCESS; -out_clear_pool: - wd_uninit_async_request_pool(&wd_agg_setting.pool); out_clear_sched: wd_clear_sched(&wd_agg_setting.sched); out_clear_ctx_config: @@ -611,23 +679,12 @@ out_clear_ctx_config: return ret; } -static int wd_agg_alg_uninit(void) +static void wd_agg_alg_uninit(void) { - enum wd_status status; - - wd_alg_get_init(&wd_agg_setting.status, &status); - if (status == WD_UNINIT) - return -WD_EINVAL; - /* Uninit async request pool */ wd_uninit_async_request_pool(&wd_agg_setting.pool); - /* Unset config, sched, driver */ wd_clear_sched(&wd_agg_setting.sched); - - wd_alg_uninit_driver(&wd_agg_setting.config, wd_agg_setting.driver); - - return WD_SUCCESS; } int wd_agg_init(char *alg, __u32 sched_type, int task_type, struct wd_ctx_params *ctx_params) @@ -635,16 +692,20 @@ int wd_agg_init(char *alg, __u32 sched_type, int task_type, struct wd_ctx_params struct wd_ctx_params agg_ctx_params = {0}; struct wd_ctx_nums agg_ctx_num = {0}; int ret = -WD_EINVAL; + int try_cnt = 0; int state; bool flag; - pthread_atfork(NULL, NULL, wd_agg_clear_status); + if (!wd_agg_atfork_registered) { + if (pthread_atfork(NULL, NULL, wd_agg_clear_status) == 0) + wd_agg_atfork_registered = true; + } state = wd_alg_try_init(&wd_agg_setting.status); if (state) return state; - if (!alg || sched_type >= SCHED_POLICY_BUTT || + if (!alg || sched_type >= SCHED_POLICY_BUTT || task_type == TASK_MIX || task_type < 0 || task_type >= TASK_MAX_TYPE) { WD_ERR("invalid: agg init input param is wrong!\n"); goto out_uninit; @@ -661,38 +722,30 @@ int wd_agg_init(char *alg, __u32 sched_type, int task_type, struct wd_ctx_params goto out_uninit; while (ret != 0) { - memset(&wd_agg_setting.config, 0, sizeof(struct wd_ctx_config_internal)); - - /* Get alg driver and dev name */ - wd_agg_setting.driver = wd_alg_drv_bind(task_type, alg); - if (!wd_agg_setting.driver) { - WD_ERR("failed to bind %s driver.\n", alg); - goto out_dlopen; + if (try_cnt++ >= WD_INIT2_MAX_RETRY) { + WD_ERR("failed to init2 after %d retries.\n", + WD_INIT2_MAX_RETRY); + goto out_driver; } - + memset(&wd_agg_setting.config, 0, sizeof(struct wd_ctx_config_internal)); agg_ctx_params.ctx_set_num = &agg_ctx_num; - ret = wd_ctx_param_init(&agg_ctx_params, ctx_params, wd_agg_setting.driver, + ret = wd_ctx_param_init(&agg_ctx_params, ctx_params, alg, WD_AGG_TYPE, 1); if (ret) { - if (ret == -WD_EAGAIN) { - wd_disable_drv(wd_agg_setting.driver); - wd_alg_drv_unbind(wd_agg_setting.driver); + if (ret == -WD_EAGAIN) continue; - } goto out_driver; } (void)strcpy(wd_agg_init_attrs.alg, alg); wd_agg_init_attrs.sched_type = sched_type; - wd_agg_init_attrs.driver = wd_agg_setting.driver; + wd_agg_init_attrs.task_type = task_type; wd_agg_init_attrs.ctx_params = &agg_ctx_params; wd_agg_init_attrs.alg_init = wd_agg_alg_init; wd_agg_init_attrs.alg_poll_ctx = wd_agg_poll_ctx; ret = wd_alg_attrs_init(&wd_agg_init_attrs); if (ret) { if (ret == -WD_ENODEV) { - wd_disable_drv(wd_agg_setting.driver); - wd_alg_drv_unbind(wd_agg_setting.driver); wd_ctx_param_uninit(&agg_ctx_params); continue; } @@ -700,17 +753,31 @@ int wd_agg_init(char *alg, __u32 sched_type, int task_type, struct wd_ctx_params goto out_params_uninit; } } + ret = wd_ctx_bind_drivers(&wd_agg_setting.config, + wd_agg_init_attrs.ctx_config_internal, + WD_TYPE_V2); + if (ret) { + WD_ERR("failed to bind driver for hashagg!\n"); + goto out_common_uninit; + } + + ret = wd_alg_init_driver(&wd_agg_setting.config); + if (ret) + goto out_unbind_drivers; wd_alg_set_init(&wd_agg_setting.status); wd_ctx_param_uninit(&agg_ctx_params); return WD_SUCCESS; +out_unbind_drivers: + wd_ctx_unbind_drivers(&wd_agg_setting.config); +out_common_uninit: + wd_agg_alg_uninit(); + wd_alg_attrs_uninit(&wd_agg_init_attrs); out_params_uninit: wd_ctx_param_uninit(&agg_ctx_params); out_driver: - wd_alg_drv_unbind(wd_agg_setting.driver); -out_dlopen: wd_agg_close_driver(); out_uninit: wd_alg_clear_init(&wd_agg_setting.status); @@ -719,14 +786,19 @@ out_uninit: void wd_agg_uninit(void) { - int ret; + enum wd_status status; - ret = wd_agg_alg_uninit(); - if (ret) + wd_alg_get_init(&wd_agg_setting.status, &status); + if (status != WD_INIT) return; + wd_alg_uninit_driver(&wd_agg_setting.config); + wd_ctx_unbind_drivers(&wd_agg_setting.config); + wd_agg_setting.config.drv_array = NULL; + wd_agg_setting.config.drv_count = 0; + wd_agg_alg_uninit(); + wd_alg_attrs_uninit(&wd_agg_init_attrs); - wd_alg_drv_unbind(wd_agg_setting.driver); wd_agg_close_driver(); wd_alg_clear_init(&wd_agg_setting.status); } @@ -740,7 +812,6 @@ static void fill_request_msg_input(struct wd_agg_msg *msg, struct wd_agg_req *re msg->agg_cols_num = sess->agg_conf.cols_num; memcpy(&msg->hash_table, &sess->hash_table, sizeof(struct wd_dae_hash_table)); msg->row_count = req->in_row_count; - msg->priv = sess->priv; if (!is_rehash) { msg->pos = WD_AGG_STREAM_INPUT; msg->agg_cols_info = sess->agg_conf.cols_info; @@ -759,7 +830,6 @@ static void fill_request_msg_output(struct wd_agg_msg *msg, struct wd_agg_req *r msg->key_cols_num = sess->key_conf.cols_num; msg->agg_cols_num = sess->agg_conf.cols_num; - msg->priv = sess->priv; if (!is_rehash) { msg->pos = WD_AGG_STREAM_OUTPUT; msg->is_count_all = sess->agg_conf.is_count_all; @@ -1097,13 +1167,13 @@ static int wd_agg_sync_job(struct wd_agg_sess *sess, struct wd_agg_req *req, wd_dfx_msg_cnt(config, WD_CTX_CNT_NUM, idx); ctx = config->ctxs + idx; + msg->priv = sess->priv; - msg_handle.send = wd_agg_setting.driver->send; - msg_handle.recv = wd_agg_setting.driver->recv; + msg_handle.send = ctx->drv->send; + msg_handle.recv = ctx->drv->recv; pthread_spin_lock(&ctx->lock); - ret = wd_handle_msg_sync(wd_agg_setting.driver, &msg_handle, ctx->ctx, - msg, NULL, config->epoll_en); + ret = wd_handle_msg_sync(&msg_handle, ctx->ctx, msg, NULL, config->epoll_en); pthread_spin_unlock(&ctx->lock); return ret; @@ -1194,17 +1264,16 @@ static int wd_agg_async_job(struct wd_agg_sess *sess, struct wd_agg_req *req, bo ctx = config->ctxs + idx; msg_id = wd_get_msg_from_pool(&wd_agg_setting.pool, idx, (void **)&msg); - if (unlikely(msg_id < 0)) { - WD_ERR("failed to get agg msg from pool!\n"); - return msg_id; - } + if (unlikely(msg_id < 0)) + return -WD_EBUSY; if (is_input) fill_request_msg_input(msg, req, sess, false); else fill_request_msg_output(msg, req, sess, false); + msg->priv = sess->priv; msg->tag = msg_id; - ret = wd_alg_driver_send(wd_agg_setting.driver, ctx->ctx, msg); + ret = ctx->drv->send(ctx->ctx, msg); if (unlikely(ret < 0)) { if (ret != -WD_EBUSY) WD_ERR("wd agg async send err!\n"); @@ -1544,7 +1613,7 @@ static int wd_agg_poll_ctx(__u32 idx, __u32 expt, __u32 *count) ctx = config->ctxs + idx; do { - ret = wd_alg_driver_recv(wd_agg_setting.driver, ctx->ctx, &resp_msg); + ret = ctx->drv->recv(ctx->ctx, &resp_msg); if (ret == -WD_EAGAIN) { return ret; } else if (unlikely(ret < 0)) { diff --git a/wd_join_gather.c b/wd_join_gather.c index 37e0022..8857c8f 100644 --- a/wd_join_gather.c +++ b/wd_join_gather.c @@ -32,8 +32,6 @@ struct wd_join_gather_setting { struct wd_ctx_config_internal config; struct wd_sched sched; struct wd_async_msg_pool pool; - struct wd_alg_driver *driver; - void *priv; void *dlhandle; void *dlh_list; }; @@ -56,7 +54,6 @@ struct wd_join_gather_sess { enum multi_batch_index_type index_type; enum wd_join_sess_state state; enum wd_join_gather_alg alg; - struct wd_join_gather_ops ops; struct wd_join_cols_conf join_conf; struct wd_gather_tables_conf gather_conf; struct wd_dae_hash_table hash_table; @@ -79,7 +76,6 @@ static void wd_join_gather_close_driver(void) wd_dlclose_drv(wd_join_gather_setting.dlh_list); wd_join_gather_setting.dlh_list = NULL; #else - wd_release_drv(wd_join_gather_setting.driver); hisi_dae_join_gather_remove(); #endif } @@ -404,70 +400,140 @@ free_join: return -WD_ENOMEM; } +static void cleanup_partial_init(struct wd_join_gather_sess *sess) +{ + struct wd_ctx_config_internal *config = &wd_join_gather_setting.config; + struct wd_alg_driver *drv; + struct wd_join_gather_ops *eops; + + if (!sess->priv) + return; + + drv = config->ctxs[0].drv; + if (drv->extend_ops) { + eops = drv->extend_ops; + if (eops->sess_uninit) + eops->sess_uninit(sess->priv); + } +} + static void wd_join_gather_uninit_sess(struct wd_join_gather_sess *sess) { - if (sess->gather_conf.batch_row_size) - free(sess->gather_conf.batch_row_size); + if (!sess) + return; - if (sess->ops.sess_uninit) - sess->ops.sess_uninit(wd_join_gather_setting.driver, sess->priv); + cleanup_partial_init(sess); + free(sess->gather_conf.batch_row_size); + sess->gather_conf.batch_row_size = NULL; } -static int wd_join_gather_init_sess(struct wd_join_gather_sess *sess, - struct wd_join_gather_sess_setup *setup) +static int handle_algorithm_specific_init(struct wd_join_gather_ops *eops, + struct wd_join_gather_sess *sess, + struct wd_join_gather_sess_setup *setup) { - struct wd_alg_driver *drv = wd_join_gather_setting.driver; __u32 array_size; int ret; - if (sess->ops.sess_init) { - if (!sess->ops.sess_uninit) { - WD_ERR("failed to get session uninit ops!\n"); - return -WD_EINVAL; - } - ret = sess->ops.sess_init(drv, setup, &sess->priv); - if (ret) { - WD_ERR("failed to init session priv!\n"); - return ret; - } - } + if (!sess->priv) + return WD_SUCCESS; - if (sess->ops.get_table_row_size && setup->alg != WD_GATHER) { - ret = sess->ops.get_table_row_size(drv, sess->priv); + if (eops->get_table_row_size && setup->alg != WD_GATHER) { + ret = eops->get_table_row_size(sess->priv); if (ret <= 0) { WD_ERR("failed to get hash table row size: %d!\n", ret); - goto uninit; + return -WD_EINVAL; } sess->hash_table.table_row_size = ret; } - if (sess->ops.get_batch_row_size && setup->alg != WD_JOIN) { + if (eops->get_batch_row_size && setup->alg != WD_JOIN) { array_size = setup->gather_table_num * sizeof(__u32); - sess->gather_conf.batch_row_size = malloc(array_size); - if (!sess->gather_conf.batch_row_size) - goto uninit; + if (!sess->gather_conf.batch_row_size) { + sess->gather_conf.batch_row_size = malloc(array_size); + if (!sess->gather_conf.batch_row_size) + return -WD_ENOMEM; + ret = eops->get_batch_row_size(sess->priv, + sess->gather_conf.batch_row_size, + array_size); + if (ret) { + free(sess->gather_conf.batch_row_size); + sess->gather_conf.batch_row_size = NULL; + WD_ERR("failed to get batch table row size!\n"); + return -WD_EINVAL; + } + } + } + + return WD_SUCCESS; +} + +static int init_single_driver_session(struct wd_alg_driver *drv, + struct wd_join_gather_sess *sess, + struct wd_join_gather_sess_setup *setup) +{ + struct wd_join_gather_ops *eops; + int ret; + + if (!drv->get_extend_ops || !drv->extend_ops) + return WD_SUCCESS; + + ret = drv->get_extend_ops(drv->extend_ops); + if (ret) { + WD_ERR("failed to set session extend ops!\n"); + return -WD_EINVAL; + } - ret = sess->ops.get_batch_row_size(drv, sess->priv, - sess->gather_conf.batch_row_size, - array_size); + eops = drv->extend_ops; + + if (eops->sess_init) { + if (!eops->sess_uninit) { + WD_ERR("failed to get session uninit ops!\n"); + return -WD_EINVAL; + } + ret = eops->sess_init(setup, &sess->priv); if (ret) { - WD_ERR("failed to get batch table row size!\n"); - goto free_batch; + WD_ERR("failed to init session priv!\n"); + return ret; } } + ret = handle_algorithm_specific_init(eops, sess, setup); + if (ret != WD_SUCCESS) { + if (eops->sess_uninit) + eops->sess_uninit(sess->priv); + return ret; + } + return WD_SUCCESS; +} -free_batch: - free(sess->gather_conf.batch_row_size); -uninit: - if (sess->ops.sess_uninit) - sess->ops.sess_uninit(drv, sess->priv); - return -WD_EINVAL; +static int wd_join_gather_init_sess(struct wd_join_gather_sess *sess, + struct wd_join_gather_sess_setup *setup) +{ + struct wd_ctx_config_internal *config = &wd_join_gather_setting.config; + struct wd_alg_driver *first_drv = NULL; + __u32 i; + + /* + * The algorithm always works in stream mode, so it supports only + * the switchover between ctxs of the same driver type. + */ + for (i = 0; i < config->ctx_num; i++) { + if (!first_drv) { + first_drv = config->ctxs[i].drv; + } else if (config->ctxs[i].drv != first_drv) { + WD_ERR("incompatible driver: %s vs %s, only single driver type allowed!\n", + config->ctxs[i].drv->drv_name, first_drv->drv_name); + return -WD_EINVAL; + } + } + + return init_single_driver_session(first_drv, sess, setup); } handle_t wd_join_gather_alloc_sess(struct wd_join_gather_sess_setup *setup) { + struct wd_sched_params params; struct wd_join_gather_sess *sess; int ret; @@ -486,7 +552,7 @@ handle_t wd_join_gather_alloc_sess(struct wd_join_gather_sess_setup *setup) sess->index_type = setup->index_type; sess->join_conf.key_output_enable = setup->join_table.key_output_enable; - ret = wd_drv_alg_support(wd_join_gather_alg[sess->alg], wd_join_gather_setting.driver); + ret = wd_drv_alg_support(wd_join_gather_alg[sess->alg], &wd_join_gather_setting.config); if (!ret) { WD_ERR("failed to check driver alg: %s!\n", wd_join_gather_alg[sess->alg]); goto free_sess; @@ -499,14 +565,11 @@ handle_t wd_join_gather_alloc_sess(struct wd_join_gather_sess_setup *setup) WD_ERR("failed to init join_gather session schedule key!\n"); goto free_sess; } - - if (wd_join_gather_setting.driver->get_extend_ops) { - ret = wd_join_gather_setting.driver->get_extend_ops(&sess->ops); - if (ret) { - WD_ERR("failed to get join gather extend ops!\n"); - goto free_key; - } - } + params.alg_name = wd_join_gather_alg[sess->alg]; + params.ctxs = wd_join_gather_setting.config.ctxs; + wd_join_gather_setting.sched.set_param( + wd_join_gather_setting.sched.h_sched_ctx, + sess->sched_key, ¶ms); ret = wd_join_gather_init_sess(sess, setup); if (ret) @@ -523,7 +586,14 @@ handle_t wd_join_gather_alloc_sess(struct wd_join_gather_sess_setup *setup) uninit_sess: wd_join_gather_uninit_sess(sess); free_key: - free(sess->sched_key); + if (sess->sched_key) { + if (wd_join_gather_setting.sched.sched_uninit) + wd_join_gather_setting.sched.sched_uninit( + wd_join_gather_setting.sched.h_sched_ctx, + (handle_t)sess->sched_key); + else + free(sess->sched_key); + } free_sess: free(sess); return (handle_t)0; @@ -542,8 +612,14 @@ void wd_join_gather_free_sess(handle_t h_sess) wd_join_gather_uninit_sess(sess); - if (sess->sched_key) - free(sess->sched_key); + if (sess->sched_key) { + if (wd_join_gather_setting.sched.sched_uninit) + wd_join_gather_setting.sched.sched_uninit( + wd_join_gather_setting.sched.h_sched_ctx, + (handle_t)sess->sched_key); + else + free(sess->sched_key); + } free(sess); } @@ -616,6 +692,8 @@ int wd_join_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info) { struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess; enum wd_join_sess_state expected; + struct wd_ctx_config_internal *config = &wd_join_gather_setting.config; + struct wd_join_gather_ops *eops; int ret; if (!sess || !info) { @@ -655,9 +733,9 @@ int wd_join_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info) if (!info->ext_table_row_num || !info->ext_table) WD_INFO("info: extern hash table is NULL!\n"); - if (sess->ops.hash_table_init) { - ret = sess->ops.hash_table_init(wd_join_gather_setting.driver, - info, sess->priv); + eops = config->ctxs[0].drv->extend_ops; + if (eops && eops->hash_table_init) { + ret = eops->hash_table_init(info, sess->priv); if (ret) goto out; } @@ -671,6 +749,8 @@ out: return ret; } +static bool wd_join_gather_atfork_registered = false; + static void wd_join_gather_clear_status(void) { wd_alg_clear_init(&wd_join_gather_setting.status); @@ -699,14 +779,8 @@ static int wd_join_gather_alg_init(struct wd_ctx_config *config, struct wd_sched if (ret < 0) goto out_clear_sched; - ret = wd_alg_init_driver(&wd_join_gather_setting.config, wd_join_gather_setting.driver); - if (ret) - goto out_clear_pool; - return WD_SUCCESS; -out_clear_pool: - wd_uninit_async_request_pool(&wd_join_gather_setting.pool); out_clear_sched: wd_clear_sched(&wd_join_gather_setting.sched); out_clear_ctx_config: @@ -714,23 +788,13 @@ out_clear_ctx_config: return ret; } -static int wd_join_gather_alg_uninit(void) +static void wd_join_gather_alg_uninit(void) { - enum wd_status status; - - wd_alg_get_init(&wd_join_gather_setting.status, &status); - if (status == WD_UNINIT) - return -WD_EINVAL; - /* Uninit async request pool */ wd_uninit_async_request_pool(&wd_join_gather_setting.pool); /* Unset config, sched, driver */ wd_clear_sched(&wd_join_gather_setting.sched); - - wd_alg_uninit_driver(&wd_join_gather_setting.config, wd_join_gather_setting.driver); - - return WD_SUCCESS; } int wd_join_gather_init(char *alg, __u32 sched_type, int task_type, @@ -742,13 +806,16 @@ int wd_join_gather_init(char *alg, __u32 sched_type, int task_type, int state; bool flag; - pthread_atfork(NULL, NULL, wd_join_gather_clear_status); + if (!wd_join_gather_atfork_registered) { + if (pthread_atfork(NULL, NULL, wd_join_gather_clear_status) == 0) + wd_join_gather_atfork_registered = true; + } state = wd_alg_try_init(&wd_join_gather_setting.status); if (state) return state; - if (!alg || sched_type >= SCHED_POLICY_BUTT || + if (!alg || sched_type >= SCHED_POLICY_BUTT || task_type == TASK_MIX || task_type < 0 || task_type >= TASK_MAX_TYPE) { WD_ERR("invalid: join_gathe init input param is wrong!\n"); goto out_uninit; @@ -766,38 +833,24 @@ int wd_join_gather_init(char *alg, __u32 sched_type, int task_type, while (ret != 0) { memset(&wd_join_gather_setting.config, 0, sizeof(struct wd_ctx_config_internal)); - - /* Get alg driver and dev name */ - wd_join_gather_setting.driver = wd_alg_drv_bind(task_type, alg); - if (!wd_join_gather_setting.driver) { - WD_ERR("failed to bind %s driver.\n", alg); - goto out_dlopen; - } - join_gather_ctx_params.ctx_set_num = &join_gather_ctx_num; - ret = wd_ctx_param_init(&join_gather_ctx_params, ctx_params, - wd_join_gather_setting.driver, + ret = wd_ctx_param_init(&join_gather_ctx_params, ctx_params, alg, WD_JOIN_GATHER_TYPE, 1); if (ret) { - if (ret == -WD_EAGAIN) { - wd_disable_drv(wd_join_gather_setting.driver); - wd_alg_drv_unbind(wd_join_gather_setting.driver); + if (ret == -WD_EAGAIN) continue; - } goto out_driver; } (void)strcpy(wd_join_gather_init_attrs.alg, alg); wd_join_gather_init_attrs.sched_type = sched_type; - wd_join_gather_init_attrs.driver = wd_join_gather_setting.driver; + wd_join_gather_init_attrs.task_type = task_type; wd_join_gather_init_attrs.ctx_params = &join_gather_ctx_params; wd_join_gather_init_attrs.alg_init = wd_join_gather_alg_init; wd_join_gather_init_attrs.alg_poll_ctx = wd_join_gather_poll_ctx; ret = wd_alg_attrs_init(&wd_join_gather_init_attrs); if (ret) { if (ret == -WD_ENODEV) { - wd_disable_drv(wd_join_gather_setting.driver); - wd_alg_drv_unbind(wd_join_gather_setting.driver); wd_ctx_param_uninit(&join_gather_ctx_params); continue; } @@ -806,16 +859,31 @@ int wd_join_gather_init(char *alg, __u32 sched_type, int task_type, } } + ret = wd_ctx_bind_drivers(&wd_join_gather_setting.config, + wd_join_gather_init_attrs.ctx_config_internal, + WD_TYPE_V2); + if (ret) { + WD_ERR("failed to bind driver for hashjoin!\n"); + goto out_common_uninit; + } + + ret = wd_alg_init_driver(&wd_join_gather_setting.config); + if (ret) + goto out_unbind_drivers; + wd_alg_set_init(&wd_join_gather_setting.status); wd_ctx_param_uninit(&join_gather_ctx_params); return WD_SUCCESS; +out_unbind_drivers: + wd_ctx_unbind_drivers(&wd_join_gather_setting.config); +out_common_uninit: + wd_join_gather_alg_uninit(); + wd_alg_attrs_uninit(&wd_join_gather_init_attrs); out_params_uninit: wd_ctx_param_uninit(&join_gather_ctx_params); out_driver: - wd_alg_drv_unbind(wd_join_gather_setting.driver); -out_dlopen: wd_join_gather_close_driver(); out_uninit: wd_alg_clear_init(&wd_join_gather_setting.status); @@ -824,14 +892,19 @@ out_uninit: void wd_join_gather_uninit(void) { - int ret; + enum wd_status status; - ret = wd_join_gather_alg_uninit(); - if (ret) + wd_alg_get_init(&wd_join_gather_setting.status, &status); + if (status != WD_INIT) return; + wd_alg_uninit_driver(&wd_join_gather_setting.config); + wd_ctx_unbind_drivers(&wd_join_gather_setting.config); + wd_join_gather_setting.config.drv_array = NULL; + wd_join_gather_setting.config.drv_count = 0; + wd_join_gather_alg_uninit(); + wd_alg_attrs_uninit(&wd_join_gather_init_attrs); - wd_alg_drv_unbind(wd_join_gather_setting.driver); wd_join_gather_close_driver(); wd_alg_clear_init(&wd_join_gather_setting.status); } @@ -871,7 +944,6 @@ static void fill_join_gather_msg(struct wd_join_gather_msg *msg, struct wd_join_ struct wd_join_gather_sess *sess) { memcpy(&msg->req, req, sizeof(struct wd_join_gather_req)); - msg->priv = sess->priv; msg->op_type = req->op_type; switch (req->op_type) { @@ -1211,11 +1283,12 @@ static int wd_join_gather_sync_job(struct wd_join_gather_sess *sess, wd_dfx_msg_cnt(config, WD_CTX_CNT_NUM, idx); ctx = config->ctxs + idx; - msg_handle.send = setting->driver->send; - msg_handle.recv = setting->driver->recv; + msg->priv = sess->priv; + msg_handle.send = ctx->drv->send; + msg_handle.recv = ctx->drv->recv; pthread_spin_lock(&ctx->lock); - ret = wd_handle_msg_sync(setting->driver, &msg_handle, ctx->ctx, + ret = wd_handle_msg_sync(&msg_handle, ctx->ctx, msg, NULL, config->epoll_en); pthread_spin_unlock(&ctx->lock); @@ -1304,14 +1377,13 @@ static int wd_join_gather_async_job(struct wd_join_gather_sess *sess, ctx = config->ctxs + idx; msg_id = wd_get_msg_from_pool(&setting->pool, idx, (void **)&msg); - if (msg_id < 0) { - WD_ERR("failed to get join gather msg from pool!\n"); - return msg_id; - } + if (msg_id < 0) + return -WD_EBUSY; fill_join_gather_msg(msg, req, sess); + msg->priv = sess->priv; msg->tag = msg_id; - ret = wd_alg_driver_send(setting->driver, ctx->ctx, msg); + ret = ctx->drv->send(ctx->ctx, msg); if (ret < 0) { if (ret != -WD_EBUSY) WD_ERR("wd join gather async send err!\n"); @@ -1781,7 +1853,7 @@ static int wd_join_gather_poll_ctx(__u32 idx, __u32 expt, __u32 *count) ctx = config->ctxs + idx; do { - ret = wd_alg_driver_recv(wd_join_gather_setting.driver, ctx->ctx, &resp_msg); + ret = ctx->drv->recv(ctx->ctx, &resp_msg); if (ret == -WD_EAGAIN) { return ret; } else if (ret < 0) { -- 2.43.0