0cd5bce53921ebff4fa2f6332e51d7340971bf79
[dpdk.git] / drivers / crypto / scheduler / scheduler_multicore.c
1 /*-
2  *   BSD LICENSE
3  *
4  *   Copyright(c) 2017 Intel Corporation. All rights reserved.
5  *
6  *   Redistribution and use in source and binary forms, with or without
7  *   modification, are permitted provided that the following conditions
8  *   are met:
9  *
10  *     * Redistributions of source code must retain the above copyright
11  *       notice, this list of conditions and the following disclaimer.
12  *     * Redistributions in binary form must reproduce the above copyright
13  *       notice, this list of conditions and the following disclaimer in
14  *       the documentation and/or other materials provided with the
15  *       distribution.
16  *     * Neither the name of Intel Corporation nor the names of its
17  *       contributors may be used to endorse or promote products derived
18  *       from this software without specific prior written permission.
19  *
20  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31  */
32 #include <unistd.h>
33
34 #include <rte_cryptodev.h>
35 #include <rte_malloc.h>
36
37 #include "rte_cryptodev_scheduler_operations.h"
38 #include "scheduler_pmd_private.h"
39
40 #define MC_SCHED_ENQ_RING_NAME_PREFIX   "MCS_ENQR_"
41 #define MC_SCHED_DEQ_RING_NAME_PREFIX   "MCS_DEQR_"
42
43 #define MC_SCHED_BUFFER_SIZE 32
44
45 #define CRYPTO_OP_STATUS_BIT_COMPLETE   0x80
46
47 /** multi-core scheduler context */
48 struct mc_scheduler_ctx {
49         uint32_t num_workers;             /**< Number of workers polling */
50         uint32_t stop_signal;
51
52         struct rte_ring *sched_enq_ring[RTE_CRYPTODEV_SCHEDULER_MAX_NB_WORKER_CORES];
53         struct rte_ring *sched_deq_ring[RTE_CRYPTODEV_SCHEDULER_MAX_NB_WORKER_CORES];
54 };
55
56 struct mc_scheduler_qp_ctx {
57         struct scheduler_slave slaves[RTE_CRYPTODEV_SCHEDULER_MAX_NB_SLAVES];
58         uint32_t nb_slaves;
59
60         uint32_t last_enq_worker_idx;
61         uint32_t last_deq_worker_idx;
62
63         struct mc_scheduler_ctx *mc_private_ctx;
64 };
65
66 static uint16_t
67 schedule_enqueue(void *qp, struct rte_crypto_op **ops, uint16_t nb_ops)
68 {
69         struct mc_scheduler_qp_ctx *mc_qp_ctx =
70                         ((struct scheduler_qp_ctx *)qp)->private_qp_ctx;
71         struct mc_scheduler_ctx *mc_ctx = mc_qp_ctx->mc_private_ctx;
72         uint32_t worker_idx = mc_qp_ctx->last_enq_worker_idx;
73         uint16_t i, processed_ops = 0;
74
75         if (unlikely(nb_ops == 0))
76                 return 0;
77
78         for (i = 0; i <  mc_ctx->num_workers && nb_ops != 0; i++) {
79                 struct rte_ring *enq_ring = mc_ctx->sched_enq_ring[worker_idx];
80                 uint16_t nb_queue_ops = rte_ring_enqueue_burst(enq_ring,
81                         (void *)(&ops[processed_ops]), nb_ops, NULL);
82
83                 nb_ops -= nb_queue_ops;
84                 processed_ops += nb_queue_ops;
85
86                 if (++worker_idx == mc_ctx->num_workers)
87                         worker_idx = 0;
88         }
89         mc_qp_ctx->last_enq_worker_idx = worker_idx;
90
91         return processed_ops;
92 }
93
94 static uint16_t
95 schedule_enqueue_ordering(void *qp, struct rte_crypto_op **ops,
96                 uint16_t nb_ops)
97 {
98         struct rte_ring *order_ring =
99                         ((struct scheduler_qp_ctx *)qp)->order_ring;
100         uint16_t nb_ops_to_enq = get_max_enqueue_order_count(order_ring,
101                         nb_ops);
102         uint16_t nb_ops_enqd = schedule_enqueue(qp, ops,
103                         nb_ops_to_enq);
104
105         scheduler_order_insert(order_ring, ops, nb_ops_enqd);
106
107         return nb_ops_enqd;
108 }
109
110
111 static uint16_t
112 schedule_dequeue(void *qp, struct rte_crypto_op **ops, uint16_t nb_ops)
113 {
114         struct mc_scheduler_qp_ctx *mc_qp_ctx =
115                         ((struct scheduler_qp_ctx *)qp)->private_qp_ctx;
116         struct mc_scheduler_ctx *mc_ctx = mc_qp_ctx->mc_private_ctx;
117         uint32_t worker_idx = mc_qp_ctx->last_deq_worker_idx;
118         uint16_t i, processed_ops = 0;
119
120         for (i = 0; i < mc_ctx->num_workers && nb_ops != 0; i++) {
121                 struct rte_ring *deq_ring = mc_ctx->sched_deq_ring[worker_idx];
122                 uint16_t nb_deq_ops = rte_ring_dequeue_burst(deq_ring,
123                         (void *)(&ops[processed_ops]), nb_ops, NULL);
124
125                 nb_ops -= nb_deq_ops;
126                 processed_ops += nb_deq_ops;
127                 if (++worker_idx == mc_ctx->num_workers)
128                         worker_idx = 0;
129         }
130
131         mc_qp_ctx->last_deq_worker_idx = worker_idx;
132
133         return processed_ops;
134
135 }
136
137 static uint16_t
138 schedule_dequeue_ordering(void *qp, struct rte_crypto_op **ops,
139                 uint16_t nb_ops)
140 {
141         struct rte_ring *order_ring = ((struct scheduler_qp_ctx *)qp)->order_ring;
142         struct rte_crypto_op *op;
143         uint32_t nb_objs = rte_ring_count(order_ring);
144         uint32_t nb_ops_to_deq = 0;
145         uint32_t nb_ops_deqd = 0;
146
147         if (nb_objs > nb_ops)
148                 nb_objs = nb_ops;
149
150         while (nb_ops_to_deq < nb_objs) {
151                 SCHEDULER_GET_RING_OBJ(order_ring, nb_ops_to_deq, op);
152
153                 if (!(op->status & CRYPTO_OP_STATUS_BIT_COMPLETE))
154                         break;
155
156                 op->status &= ~CRYPTO_OP_STATUS_BIT_COMPLETE;
157                 nb_ops_to_deq++;
158         }
159
160         if (nb_ops_to_deq) {
161                 nb_ops_deqd = rte_ring_sc_dequeue_bulk(order_ring,
162                                 (void **)ops, nb_ops_to_deq, NULL);
163         }
164
165         return nb_ops_deqd;
166 }
167
168 static int
169 slave_attach(__rte_unused struct rte_cryptodev *dev,
170                 __rte_unused uint8_t slave_id)
171 {
172         return 0;
173 }
174
175 static int
176 slave_detach(__rte_unused struct rte_cryptodev *dev,
177                 __rte_unused uint8_t slave_id)
178 {
179         return 0;
180 }
181
182 static int
183 mc_scheduler_worker(struct rte_cryptodev *dev)
184 {
185         struct scheduler_ctx *sched_ctx = dev->data->dev_private;
186         struct mc_scheduler_ctx *mc_ctx = sched_ctx->private_ctx;
187         struct rte_ring *enq_ring;
188         struct rte_ring *deq_ring;
189         uint32_t core_id = rte_lcore_id();
190         int i, worker_idx = -1;
191         struct scheduler_slave *slave;
192         struct rte_crypto_op *enq_ops[MC_SCHED_BUFFER_SIZE];
193         struct rte_crypto_op *deq_ops[MC_SCHED_BUFFER_SIZE];
194         uint16_t processed_ops;
195         uint16_t pending_enq_ops = 0;
196         uint16_t pending_enq_ops_idx = 0;
197         uint16_t pending_deq_ops = 0;
198         uint16_t pending_deq_ops_idx = 0;
199         uint16_t inflight_ops = 0;
200         const uint8_t reordering_enabled = sched_ctx->reordering_enabled;
201
202         for (i = 0; i < (int)sched_ctx->nb_wc; i++) {
203                 if (sched_ctx->wc_pool[i] == core_id) {
204                         worker_idx = i;
205                         break;
206                 }
207         }
208         if (worker_idx == -1) {
209                 CS_LOG_ERR("worker on core %u:cannot find worker index!\n", core_id);
210                 return -1;
211         }
212
213         slave = &sched_ctx->slaves[worker_idx];
214         enq_ring = mc_ctx->sched_enq_ring[worker_idx];
215         deq_ring = mc_ctx->sched_deq_ring[worker_idx];
216
217         while (!mc_ctx->stop_signal) {
218                 if (pending_enq_ops) {
219                         processed_ops =
220                                 rte_cryptodev_enqueue_burst(slave->dev_id,
221                                         slave->qp_id, &enq_ops[pending_enq_ops_idx],
222                                         pending_enq_ops);
223                         pending_enq_ops -= processed_ops;
224                         pending_enq_ops_idx += processed_ops;
225                         inflight_ops += processed_ops;
226                 } else {
227                         processed_ops = rte_ring_dequeue_burst(enq_ring, (void *)enq_ops,
228                                                         MC_SCHED_BUFFER_SIZE, NULL);
229                         if (processed_ops) {
230                                 pending_enq_ops_idx = rte_cryptodev_enqueue_burst(
231                                                         slave->dev_id, slave->qp_id,
232                                                         enq_ops, processed_ops);
233                                 pending_enq_ops = processed_ops - pending_enq_ops_idx;
234                                 inflight_ops += pending_enq_ops_idx;
235                         }
236                 }
237
238                 if (pending_deq_ops) {
239                         processed_ops = rte_ring_enqueue_burst(
240                                         deq_ring, (void *)&deq_ops[pending_deq_ops_idx],
241                                                         pending_deq_ops, NULL);
242                         pending_deq_ops -= processed_ops;
243                         pending_deq_ops_idx += processed_ops;
244                 } else if (inflight_ops) {
245                         processed_ops = rte_cryptodev_dequeue_burst(slave->dev_id,
246                                         slave->qp_id, deq_ops, MC_SCHED_BUFFER_SIZE);
247                         if (processed_ops) {
248                                 inflight_ops -= processed_ops;
249                                 if (reordering_enabled) {
250                                         uint16_t j;
251
252                                         for (j = 0; j < processed_ops; j++) {
253                                                 deq_ops[j]->status |=
254                                                         CRYPTO_OP_STATUS_BIT_COMPLETE;
255                                         }
256                                 } else {
257                                         pending_deq_ops_idx = rte_ring_enqueue_burst(
258                                                 deq_ring, (void *)deq_ops, processed_ops,
259                                                 NULL);
260                                         pending_deq_ops = processed_ops -
261                                                                 pending_deq_ops_idx;
262                                 }
263                         }
264                 }
265
266                 rte_pause();
267         }
268
269         return 0;
270 }
271
272 static int
273 scheduler_start(struct rte_cryptodev *dev)
274 {
275         struct scheduler_ctx *sched_ctx = dev->data->dev_private;
276         struct mc_scheduler_ctx *mc_ctx = sched_ctx->private_ctx;
277         uint16_t i;
278
279         mc_ctx->stop_signal = 0;
280
281         for (i = 0; i < sched_ctx->nb_wc; i++)
282                 rte_eal_remote_launch(
283                         (lcore_function_t *)mc_scheduler_worker, dev,
284                                         sched_ctx->wc_pool[i]);
285
286         if (sched_ctx->reordering_enabled) {
287                 dev->enqueue_burst = &schedule_enqueue_ordering;
288                 dev->dequeue_burst = &schedule_dequeue_ordering;
289         } else {
290                 dev->enqueue_burst = &schedule_enqueue;
291                 dev->dequeue_burst = &schedule_dequeue;
292         }
293
294         for (i = 0; i < dev->data->nb_queue_pairs; i++) {
295                 struct scheduler_qp_ctx *qp_ctx = dev->data->queue_pairs[i];
296                 struct mc_scheduler_qp_ctx *mc_qp_ctx =
297                                 qp_ctx->private_qp_ctx;
298                 uint32_t j;
299
300                 memset(mc_qp_ctx->slaves, 0,
301                                 RTE_CRYPTODEV_SCHEDULER_MAX_NB_SLAVES *
302                                 sizeof(struct scheduler_slave));
303                 for (j = 0; j < sched_ctx->nb_slaves; j++) {
304                         mc_qp_ctx->slaves[j].dev_id =
305                                         sched_ctx->slaves[j].dev_id;
306                         mc_qp_ctx->slaves[j].qp_id = i;
307                 }
308
309                 mc_qp_ctx->nb_slaves = sched_ctx->nb_slaves;
310
311                 mc_qp_ctx->last_enq_worker_idx = 0;
312                 mc_qp_ctx->last_deq_worker_idx = 0;
313         }
314
315         return 0;
316 }
317
318 static int
319 scheduler_stop(struct rte_cryptodev *dev)
320 {
321         struct scheduler_ctx *sched_ctx = dev->data->dev_private;
322         struct mc_scheduler_ctx *mc_ctx = sched_ctx->private_ctx;
323         uint16_t i;
324
325         mc_ctx->stop_signal = 1;
326
327         for (i = 0; i < sched_ctx->nb_wc; i++)
328                 rte_eal_wait_lcore(sched_ctx->wc_pool[i]);
329
330         return 0;
331 }
332
333 static int
334 scheduler_config_qp(struct rte_cryptodev *dev, uint16_t qp_id)
335 {
336         struct scheduler_qp_ctx *qp_ctx = dev->data->queue_pairs[qp_id];
337         struct mc_scheduler_qp_ctx *mc_qp_ctx;
338         struct scheduler_ctx *sched_ctx = dev->data->dev_private;
339         struct mc_scheduler_ctx *mc_ctx = sched_ctx->private_ctx;
340
341         mc_qp_ctx = rte_zmalloc_socket(NULL, sizeof(*mc_qp_ctx), 0,
342                         rte_socket_id());
343         if (!mc_qp_ctx) {
344                 CS_LOG_ERR("failed allocate memory for private queue pair");
345                 return -ENOMEM;
346         }
347
348         mc_qp_ctx->mc_private_ctx = mc_ctx;
349         qp_ctx->private_qp_ctx = (void *)mc_qp_ctx;
350
351
352         return 0;
353 }
354
355 static int
356 scheduler_create_private_ctx(struct rte_cryptodev *dev)
357 {
358         struct scheduler_ctx *sched_ctx = dev->data->dev_private;
359         struct mc_scheduler_ctx *mc_ctx;
360         uint16_t i;
361
362         if (sched_ctx->private_ctx)
363                 rte_free(sched_ctx->private_ctx);
364
365         mc_ctx = rte_zmalloc_socket(NULL, sizeof(struct mc_scheduler_ctx), 0,
366                         rte_socket_id());
367         if (!mc_ctx) {
368                 CS_LOG_ERR("failed allocate memory");
369                 return -ENOMEM;
370         }
371
372         mc_ctx->num_workers = sched_ctx->nb_wc;
373         for (i = 0; i < sched_ctx->nb_wc; i++) {
374                 char r_name[16];
375
376                 snprintf(r_name, sizeof(r_name), MC_SCHED_ENQ_RING_NAME_PREFIX "%u", i);
377                 mc_ctx->sched_enq_ring[i] = rte_ring_create(r_name, PER_SLAVE_BUFF_SIZE,
378                                         rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ);
379                 if (!mc_ctx->sched_enq_ring[i]) {
380                         CS_LOG_ERR("Cannot create ring for worker %u", i);
381                         return -1;
382                 }
383                 snprintf(r_name, sizeof(r_name), MC_SCHED_DEQ_RING_NAME_PREFIX "%u", i);
384                 mc_ctx->sched_deq_ring[i] = rte_ring_create(r_name, PER_SLAVE_BUFF_SIZE,
385                                         rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ);
386                 if (!mc_ctx->sched_deq_ring[i]) {
387                         CS_LOG_ERR("Cannot create ring for worker %u", i);
388                         return -1;
389                 }
390         }
391
392         sched_ctx->private_ctx = (void *)mc_ctx;
393
394         return 0;
395 }
396
397 struct rte_cryptodev_scheduler_ops scheduler_mc_ops = {
398         slave_attach,
399         slave_detach,
400         scheduler_start,
401         scheduler_stop,
402         scheduler_config_qp,
403         scheduler_create_private_ctx,
404         NULL,   /* option_set */
405         NULL    /* option_get */
406 };
407
408 struct rte_cryptodev_scheduler mc_scheduler = {
409                 .name = "multicore-scheduler",
410                 .description = "scheduler which will run burst across multiple cpu cores",
411                 .mode = CDEV_SCHED_MODE_MULTICORE,
412                 .ops = &scheduler_mc_ops
413 };
414
415 struct rte_cryptodev_scheduler *multicore_scheduler = &mc_scheduler;