d/numam-dpdk
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

crypto/scheduler: fix multicore reordering

Browse Source 
Operations can be dequeued from the reordering ring only after they
were dequeued from the crypto pmd with rte_cryptodev_dequeue_burst()
function. It is not correct to dequeue them when status just changed
from RTE_CRYPTO_OP_STATUS_NOT_PROCESSED to any other value, as the
operations still can be processed by crypto pmd internally.
Now multicore scheduler workers mark status of all dequeued from
crypto pmd operations with CRYPTO_OP_STATUS_BIT_COMPLETE bit set.
Scheduler will dequeue crypto operations from reordering ring only
when this status bit is set. Prior to put this operation to output
buffer, scheduler clears this bit, so the application gets
unmodified status from crypto pmd.

Fixes: 4c07e0552f ("crypto/scheduler: add multicore scheduling mode")

Signed-off-by: Kirill Rybalchenko <kirill.rybalchenko@intel.com>
Acked-by: Declan Doherty <declan.doherty@intel.com>
This commit is contained in:
Kirill Rybalchenko 2017-07-18 11:35:38 +01:00 committed by Pablo de Lara
parent 26c4ab1a16
commit 89244ea485
1 changed files with 67 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
drivers/crypto/scheduler/scheduler_multicore.c
View File

@ -42,6 +42,8 @@
#define MC_SCHED_BUFFER_SIZE 32
#define CRYPTO_OP_STATUS_BIT_COMPLETE 0x80
/** multi-core scheduler context */
struct mc_scheduler_ctx {
uint32_t num_workers; /**< Number of workers polling */
@ -136,10 +138,31 @@ static uint16_t
schedule_dequeue_ordering(void *qp, struct rte_crypto_op **ops,
uint16_t nb_ops)
{
struct rte_ring *order_ring =
((struct scheduler_qp_ctx *)qp)->order_ring;
struct rte_ring *order_ring = ((struct scheduler_qp_ctx *)qp)->order_ring;
struct rte_crypto_op *op;
uint32_t nb_objs = rte_ring_count(order_ring);
uint32_t nb_ops_to_deq = 0;
uint32_t nb_ops_deqd = 0;
return scheduler_order_drain(order_ring, ops, nb_ops);
if (nb_objs > nb_ops)
nb_objs = nb_ops;
while (nb_ops_to_deq < nb_objs) {
SCHEDULER_GET_RING_OBJ(order_ring, nb_ops_to_deq, op);
if (!(op->status & CRYPTO_OP_STATUS_BIT_COMPLETE))
break;
op->status &= ~CRYPTO_OP_STATUS_BIT_COMPLETE;
nb_ops_to_deq++;
}
if (nb_ops_to_deq) {
nb_ops_deqd = rte_ring_sc_dequeue_bulk(order_ring,
(void **)ops, nb_ops_to_deq, NULL);
}
return nb_ops_deqd;
}
static int
@ -169,9 +192,12 @@ mc_scheduler_worker(struct rte_cryptodev *dev)
struct rte_crypto_op *enq_ops[MC_SCHED_BUFFER_SIZE];
struct rte_crypto_op *deq_ops[MC_SCHED_BUFFER_SIZE];
uint16_t processed_ops;
uint16_t left_op = 0;
uint16_t left_op_idx = 0;
uint16_t pending_enq_ops = 0;
uint16_t pending_enq_ops_idx = 0;
uint16_t pending_deq_ops = 0;
uint16_t pending_deq_ops_idx = 0;
uint16_t inflight_ops = 0;
const uint8_t reordering_enabled = sched_ctx->reordering_enabled;
for (i = 0; i < (int)sched_ctx->nb_wc; i++) {
if (sched_ctx->wc_pool[i] == core_id) {
@ -189,37 +215,51 @@ mc_scheduler_worker(struct rte_cryptodev *dev)
deq_ring = mc_ctx->sched_deq_ring[worker_idx];
while (!mc_ctx->stop_signal) {
if (left_op) {
if (pending_enq_ops) {
processed_ops =
rte_cryptodev_enqueue_burst(slave->dev_id,
slave->qp_id,
&enq_ops[left_op_idx], left_op);
left_op -= processed_ops;
left_op_idx += processed_ops;
} else {
uint16_t nb_deq_ops = rte_ring_dequeue_burst(enq_ring,
(void *)enq_ops, MC_SCHED_BUFFER_SIZE, NULL);
if (nb_deq_ops) {
processed_ops = rte_cryptodev_enqueue_burst(slave->dev_id,
slave->qp_id, enq_ops, nb_deq_ops);
if (unlikely(processed_ops < nb_deq_ops)) {
left_op = nb_deq_ops - processed_ops;
left_op_idx = processed_ops;
}
slave->qp_id, &enq_ops[pending_enq_ops_idx],
pending_enq_ops);
pending_enq_ops -= processed_ops;
pending_enq_ops_idx += processed_ops;
inflight_ops += processed_ops;
} else {
processed_ops = rte_ring_dequeue_burst(enq_ring, (void *)enq_ops,
MC_SCHED_BUFFER_SIZE, NULL);
if (processed_ops) {
pending_enq_ops_idx = rte_cryptodev_enqueue_burst(
slave->dev_id, slave->qp_id,
enq_ops, processed_ops);
pending_enq_ops = processed_ops - pending_enq_ops_idx;
inflight_ops += pending_enq_ops_idx;
}
}
if (inflight_ops > 0) {
if (pending_deq_ops) {
processed_ops = rte_ring_enqueue_burst(
deq_ring, (void *)&deq_ops[pending_deq_ops_idx],
pending_deq_ops, NULL);
pending_deq_ops -= processed_ops;
pending_deq_ops_idx += processed_ops;
} else if (inflight_ops) {
processed_ops = rte_cryptodev_dequeue_burst(slave->dev_id,
slave->qp_id, deq_ops, MC_SCHED_BUFFER_SIZE);
if (processed_ops) {
uint16_t nb_enq_ops = rte_ring_enqueue_burst(deq_ring,
(void *)deq_ops, processed_ops, NULL);
inflight_ops -= nb_enq_ops;
inflight_ops -= processed_ops;
if (reordering_enabled) {
uint16_t j;
for (j = 0; j < processed_ops; j++) {
deq_ops[j]->status |=
CRYPTO_OP_STATUS_BIT_COMPLETE;
}
} else {
pending_deq_ops_idx = rte_ring_enqueue_burst(
deq_ring, (void *)deq_ops, processed_ops,
NULL);
pending_deq_ops = processed_ops -
pending_deq_ops_idx;
}
}
}