#include <stdlib.h>
#include "obj.h"
#include "thread.h"
#ifndef THREAD_PIPELINES_MAX
#define THREAD_PIPELINES_MAX 256
#endif
#ifndef THREAD_MSGQ_SIZE
#define THREAD_MSGQ_SIZE 64
#endif
#ifndef THREAD_TIMER_PERIOD_MS
#define THREAD_TIMER_PERIOD_MS 100
#endif
#ifndef PIPELINE_INSTR_QUANTA
#define PIPELINE_INSTR_QUANTA 1000
#endif
struct thread {
uint32_t enabled;
};
static struct thread thread[RTE_MAX_LCORE];
struct pipeline_data {
struct rte_swx_pipeline *p;
uint64_t timer_period;
uint64_t time_next;
};
struct thread_data {
struct rte_swx_pipeline *p[THREAD_PIPELINES_MAX];
uint32_t n_pipelines;
struct pipeline_data pipeline_data[THREAD_PIPELINES_MAX];
uint64_t timer_period;
uint64_t time_next;
uint64_t time_next_min;
static struct thread_data thread_data[RTE_MAX_LCORE];
static void
thread_free(void)
{
uint32_t i;
for (i = 0; i < RTE_MAX_LCORE; i++) {
struct thread *t = &thread[i];
continue;
}
}
int
thread_init(void)
{
uint32_t i;
char name[NAME_MAX];
struct thread *t = &thread[i];
struct thread_data *t_data = &thread_data[i];
snprintf(name, sizeof(name), "THREAD-%04x-MSGQ-REQ", i);
THREAD_MSGQ_SIZE,
cpu_id,
if (msgq_req == NULL) {
thread_free();
return -1;
}
snprintf(name, sizeof(name), "THREAD-%04x-MSGQ-RSP", i);
THREAD_MSGQ_SIZE,
cpu_id,
if (msgq_rsp == NULL) {
thread_free();
return -1;
}
t->msgq_req = msgq_req;
t->msgq_rsp = msgq_rsp;
t->enabled = 1;
t_data->n_pipelines = 0;
t_data->msgq_req = msgq_req;
t_data->msgq_rsp = msgq_rsp;
t_data->timer_period =
t_data->time_next_min = t_data->time_next;
}
return 0;
}
static inline int
thread_is_running(uint32_t thread_id)
{
return (thread_state ==
RUNNING) ? 1 : 0;
}
enum thread_req_type {
THREAD_REQ_PIPELINE_ENABLE = 0,
THREAD_REQ_PIPELINE_DISABLE,
THREAD_REQ_MAX
};
struct thread_msg_req {
enum thread_req_type type;
union {
struct {
struct rte_swx_pipeline *p;
uint32_t timer_period_ms;
} pipeline_enable;
struct {
struct rte_swx_pipeline *p;
} pipeline_disable;
};
};
struct thread_msg_rsp {
int status;
};
static struct thread_msg_req *
thread_msg_alloc(void)
{
size_t size =
RTE_MAX(
sizeof(
struct thread_msg_req),
sizeof(struct thread_msg_rsp));
return calloc(1, size);
}
static void
thread_msg_free(struct thread_msg_rsp *rsp)
{
free(rsp);
}
static struct thread_msg_rsp *
thread_msg_send_recv(uint32_t thread_id,
struct thread_msg_req *req)
{
struct thread *t = &thread[thread_id];
struct rte_ring *msgq_req = t->msgq_req;
struct rte_ring *msgq_rsp = t->msgq_rsp;
struct thread_msg_rsp *rsp;
int status;
do {
} while (status == -ENOBUFS);
do {
} while (status != 0);
return rsp;
}
static int
thread_is_pipeline_enabled(uint32_t thread_id, struct rte_swx_pipeline *p)
{
struct thread *t = &thread[thread_id];
struct thread_data *td = &thread_data[thread_id];
uint32_t i;
if (!t->enabled)
return 0;
for (i = 0; i < td->n_pipelines; i++)
if (td->p[i] == p)
return 1;
return 0 ;
}
int
thread_pipeline_enable(uint32_t thread_id, struct rte_swx_pipeline *p, uint32_t timer_period_ms)
{
struct thread *t;
struct thread_msg_req *req;
struct thread_msg_rsp *rsp;
int status;
if ((thread_id >= RTE_MAX_LCORE) || !p || !timer_period_ms)
return -1;
t = &thread[thread_id];
if (t->enabled == 0)
return -1;
if (!thread_is_running(thread_id)) {
struct thread_data *td = &thread_data[thread_id];
struct pipeline_data *tdp = &td->pipeline_data[td->n_pipelines];
if (td->n_pipelines >= THREAD_PIPELINES_MAX)
return -1;
td->p[td->n_pipelines] = p;
tdp->p = p;
td->n_pipelines++;
return 0;
}
req = thread_msg_alloc();
if (req == NULL)
return -1;
req->type = THREAD_REQ_PIPELINE_ENABLE;
req->pipeline_enable.p = p;
req->pipeline_enable.timer_period_ms = timer_period_ms;
rsp = thread_msg_send_recv(thread_id, req);
status = rsp->status;
thread_msg_free(rsp);
if (status)
return status;
return 0;
}
int
thread_pipeline_disable(uint32_t thread_id, struct rte_swx_pipeline *p)
{
struct thread *t;
struct thread_msg_req *req;
struct thread_msg_rsp *rsp;
int status;
if ((thread_id >= RTE_MAX_LCORE) || !p)
return -1;
t = &thread[thread_id];
if (t->enabled == 0)
return -1;
if (!thread_is_pipeline_enabled(thread_id, p))
return 0;
if (!thread_is_running(thread_id)) {
struct thread_data *td = &thread_data[thread_id];
uint32_t i;
for (i = 0; i < td->n_pipelines; i++) {
struct pipeline_data *tdp = &td->pipeline_data[i];
if (tdp->p != p)
continue;
if (i < td->n_pipelines - 1) {
struct rte_swx_pipeline *pipeline_last =
td->p[td->n_pipelines - 1];
struct pipeline_data *tdp_last =
&td->pipeline_data[td->n_pipelines - 1];
td->p[i] = pipeline_last;
memcpy(tdp, tdp_last, sizeof(*tdp));
}
td->n_pipelines--;
break;
}
return 0;
}
req = thread_msg_alloc();
if (req == NULL)
return -1;
req->type = THREAD_REQ_PIPELINE_DISABLE;
req->pipeline_disable.p = p;
rsp = thread_msg_send_recv(thread_id, req);
status = rsp->status;
thread_msg_free(rsp);
if (status)
return status;
return 0;
}
static inline struct thread_msg_req *
thread_msg_recv(
struct rte_ring *msgq_req)
{
struct thread_msg_req *req;
if (status != 0)
return NULL;
return req;
}
static inline void
thread_msg_send(
struct rte_ring *msgq_rsp,
struct thread_msg_rsp *rsp)
{
int status;
do {
} while (status == -ENOBUFS);
}
static struct thread_msg_rsp *
thread_msg_handle_pipeline_enable(struct thread_data *t,
struct thread_msg_req *req)
{
struct thread_msg_rsp *rsp = (struct thread_msg_rsp *) req;
struct pipeline_data *p = &t->pipeline_data[t->n_pipelines];
if (t->n_pipelines >= THREAD_PIPELINES_MAX) {
rsp->status = -1;
return rsp;
}
t->p[t->n_pipelines] = req->pipeline_enable.p;
p->p = req->pipeline_enable.p;
req->pipeline_enable.timer_period_ms) / 1000;
t->n_pipelines++;
rsp->status = 0;
return rsp;
}
static struct thread_msg_rsp *
thread_msg_handle_pipeline_disable(struct thread_data *t,
struct thread_msg_req *req)
{
struct thread_msg_rsp *rsp = (struct thread_msg_rsp *) req;
uint32_t n_pipelines = t->n_pipelines;
struct rte_swx_pipeline *
pipeline = req->pipeline_disable.
p;
uint32_t i;
for (i = 0; i < n_pipelines; i++) {
struct pipeline_data *p = &t->pipeline_data[i];
continue;
if (i < n_pipelines - 1) {
struct rte_swx_pipeline *pipeline_last =
t->p[n_pipelines - 1];
struct pipeline_data *p_last =
&t->pipeline_data[n_pipelines - 1];
t->p[i] = pipeline_last;
memcpy(p, p_last, sizeof(*p));
}
t->n_pipelines--;
rsp->status = 0;
return rsp;
}
rsp->status = 0;
return rsp;
}
static void
thread_msg_handle(struct thread_data *t)
{
for ( ; ; ) {
struct thread_msg_req *req;
struct thread_msg_rsp *rsp;
req = thread_msg_recv(t->msgq_req);
if (req == NULL)
break;
switch (req->type) {
case THREAD_REQ_PIPELINE_ENABLE:
rsp = thread_msg_handle_pipeline_enable(t, req);
break;
case THREAD_REQ_PIPELINE_DISABLE:
rsp = thread_msg_handle_pipeline_disable(t, req);
break;
default:
rsp = (struct thread_msg_rsp *) req;
rsp->status = -1;
}
thread_msg_send(t->msgq_rsp, rsp);
}
}
int
{
struct thread_data *t;
uint32_t thread_id, i;
t = &thread_data[thread_id];
for (i = 0; ; i++) {
uint32_t j;
for (j = 0; j < t->n_pipelines; j++)
if ((i & 0xF) == 0) {
uint64_t time_next_min = UINT64_MAX;
if (time < t->time_next_min)
continue;
{
uint64_t time_next = t->time_next;
if (time_next <= time) {
thread_msg_handle(t);
time_next = time + t->timer_period;
t->time_next = time_next;
}
if (time_next < time_next_min)
time_next_min = time_next;
}
t->time_next_min = time_next_min;
}
}
return 0;
}
#define __rte_cache_aligned
static uint64_t rte_get_tsc_cycles(void)
uint64_t rte_get_tsc_hz(void)
enum rte_lcore_state_t rte_eal_get_lcore_state(unsigned int worker_id)
unsigned int rte_lcore_to_socket_id(unsigned int lcore_id)
int rte_lcore_is_enabled(unsigned int lcore_id)
#define RTE_LCORE_FOREACH_WORKER(i)
static unsigned rte_lcore_id(void)
void rte_ring_free(struct rte_ring *r)
struct rte_ring * rte_ring_create(const char *name, unsigned int count, int socket_id, unsigned int flags)
static __rte_always_inline int rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
static __rte_always_inline int rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
__rte_experimental void rte_swx_pipeline_run(struct rte_swx_pipeline *p, uint32_t n_instructions)