cne_graph_worker.h

Go to the documentation of this file.

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright (c) 2020 Marvell International Ltd.
 */
 
#ifndef _CNE_GRAPH_WORKER_H_
#define _CNE_GRAPH_WORKER_H_
 
#include <string.h>
#include <cne_common.h>
#include <cne_cycles.h>
#include <cne_prefetch.h>
#include <cne_branch_prediction.h>
#include <cne_log.h>
 
#include "cne_graph.h"
 
#ifdef __cplusplus
extern "C" {
#endif
 
struct cne_graph {
    uint32_t tail;                 
    uint32_t head;                 
    uint32_t cir_mask;             
    cne_node_t nb_nodes;           
    cne_graph_off_t *cir_start;    
    cne_graph_off_t nodes_start;   
    cne_graph_t id;                
    char name[CNE_GRAPH_NAMESIZE]; 
    uint64_t fence;                
} __cne_cache_aligned;
 
struct cne_node {
    /* Slow path area */
    uint64_t fence;         
    cne_graph_off_t next;   
    cne_node_t id;          
    cne_node_t parent_id;   
    cne_edge_t nb_edges;    
    uint32_t realloc_count; 
    char parent[CNE_NODE_NAMESIZE]; 
    char name[CNE_NODE_NAMESIZE];   
    /* Fast path area */
#define CNE_NODE_CTX_SZ 16
    uint8_t ctx[CNE_NODE_CTX_SZ] __cne_cache_aligned; 
    uint16_t size;                                    
    uint16_t idx;                                     
    cne_graph_off_t off;                              
    uint64_t total_cycles;                            
    uint64_t total_calls;                             
    uint64_t total_objs;                              
    CNE_STD_C11
    union {
        void **objs; 
        uint64_t objs_u64;
    };
    CNE_STD_C11
    union {
        cne_node_process_t process; 
        uint64_t process_u64;
    };
    struct cne_node *nodes[] __cne_cache_min_aligned; 
} __cne_cache_aligned;
 
void __cne_node_stream_alloc(struct cne_graph *graph, struct cne_node *node);
 
void __cne_node_stream_alloc_size(struct cne_graph *graph, struct cne_node *node,
                                  uint16_t req_size);
 
static inline void
cne_graph_walk(struct cne_graph *graph)
{
    const cne_graph_off_t *cir_start = graph->cir_start;
    const cne_node_t mask            = graph->cir_mask;
    uint32_t head                    = graph->head;
    struct cne_node *node;
    uint64_t start;
    uint16_t rc;
    void **objs;
 
    /*
     * Walk on the source node(s) ((cir_start - head) -> cir_start) and then
     * on the pending streams (cir_start -> (cir_start + mask) -> cir_start)
     * in a circular buffer fashion.
     *
     *  +-----+ <= cir_start - head [number of source nodes]
     *  |     |
     *  | ... | <= source nodes
     *  |     |
     *  +-----+ <= cir_start [head = 0] [tail = 0]
     *  |     |
     *  | ... | <= pending streams
     *  |     |
     *  +-----+ <= cir_start + mask
     */
    while (likely(head != graph->tail)) {
        node = CNE_PTR_ADD(graph, cir_start[(int32_t)head++]);
        CNE_ASSERT(node->fence == CNE_GRAPH_FENCE);
        objs = node->objs;
        cne_prefetch0(objs);
 
        if (cne_graph_has_stats_feature()) {
            start = cne_rdtsc();
            rc    = node->process(graph, node, objs, node->idx);
            node->total_cycles += cne_rdtsc() - start;
            node->total_calls++;
            node->total_objs += rc;
        } else
            node->process(graph, node, objs, node->idx);
        node->idx = 0;
        head      = likely((int32_t)head > 0) ? head & mask : head;
    }
    graph->tail = 0;
}
 
/* Fast path helper functions */
 
static __cne_always_inline void
__cne_node_enqueue_tail_update(struct cne_graph *graph, struct cne_node *node)
{
    uint32_t tail;
 
    tail                     = graph->tail;
    graph->cir_start[tail++] = node->off;
    graph->tail              = tail & graph->cir_mask;
}
 
static __cne_always_inline void
__cne_node_enqueue_prologue(struct cne_graph *graph, struct cne_node *node, const uint16_t idx,
                            const uint16_t space)
{
    /* Add to the pending stream list if the node is new */
    if (idx == 0)
        __cne_node_enqueue_tail_update(graph, node);
 
    if (unlikely(node->size < (idx + space)))
        __cne_node_stream_alloc_size(graph, node, node->size + space);
}
 
static __cne_always_inline struct cne_node *
__cne_node_next_node_get(struct cne_node *node, cne_edge_t next)
{
    CNE_ASSERT(next < node->nb_edges);
    CNE_ASSERT(node->fence == CNE_GRAPH_FENCE);
    node = node->nodes[next];
    CNE_ASSERT(node->fence == CNE_GRAPH_FENCE);
 
    return node;
}
 
static inline void
cne_node_add_objects_to_source(struct cne_graph *graph, struct cne_node *node, void **objs,
                               uint16_t nb_objs)
{
    const uint16_t idx = node->idx;
 
    if (unlikely(node->size < (idx + nb_objs)))
        __cne_node_stream_alloc(graph, node);
 
    memcpy(&node->objs[idx], objs, nb_objs * sizeof(void *));
    node->idx = idx + nb_objs;
}
static inline void
cne_node_add_objects_to_input(struct cne_graph *graph, struct cne_node *node, void **objs,
                              uint16_t nb_objs)
{
    const uint16_t idx = node->idx;
 
    __cne_node_enqueue_prologue(graph, node, idx, nb_objs);
 
    memcpy(&node->objs[idx], objs, nb_objs * sizeof(void *));
    node->idx = idx + nb_objs;
}
 
static inline void
cne_node_enqueue(struct cne_graph *graph, struct cne_node *node, cne_edge_t next, void **objs,
                 uint16_t nb_objs)
{
    node               = __cne_node_next_node_get(node, next);
    const uint16_t idx = node->idx;
 
    __cne_node_enqueue_prologue(graph, node, idx, nb_objs);
 
    memcpy(&node->objs[idx], objs, nb_objs * sizeof(void *));
    node->idx = idx + nb_objs;
}
 
static inline void
cne_node_enqueue_x1(struct cne_graph *graph, struct cne_node *node, cne_edge_t next, void *obj)
{
    node         = __cne_node_next_node_get(node, next);
    uint16_t idx = node->idx;
 
    __cne_node_enqueue_prologue(graph, node, idx, 1);
 
    node->objs[idx++] = obj;
    node->idx         = idx;
}
 
static inline void
cne_node_enqueue_x2(struct cne_graph *graph, struct cne_node *node, cne_edge_t next, void *obj0,
                    void *obj1)
{
    node         = __cne_node_next_node_get(node, next);
    uint16_t idx = node->idx;
 
    __cne_node_enqueue_prologue(graph, node, idx, 2);
 
    node->objs[idx++] = obj0;
    node->objs[idx++] = obj1;
    node->idx         = idx;
}
 
static inline void
cne_node_enqueue_x4(struct cne_graph *graph, struct cne_node *node, cne_edge_t next, void *obj0,
                    void *obj1, void *obj2, void *obj3)
{
    node         = __cne_node_next_node_get(node, next);
    uint16_t idx = node->idx;
 
    __cne_node_enqueue_prologue(graph, node, idx, 4);
 
    node->objs[idx++] = obj0;
    node->objs[idx++] = obj1;
    node->objs[idx++] = obj2;
    node->objs[idx++] = obj3;
    node->idx         = idx;
}
 
static inline void
cne_node_enqueue_next(struct cne_graph *graph, struct cne_node *node, cne_edge_t *nexts,
                      void **objs, uint16_t nb_objs)
{
    uint16_t i;
 
    for (i = 0; i < nb_objs; i++)
        cne_node_enqueue_x1(graph, node, nexts[i], objs[i]);
}
 
static inline void **
cne_node_next_stream_get(struct cne_graph *graph, struct cne_node *node, cne_edge_t next,
                         uint16_t nb_objs)
{
    node                = __cne_node_next_node_get(node, next);
    const uint16_t idx  = node->idx;
    uint16_t free_space = node->size - idx;
 
    if (unlikely(free_space < nb_objs))
        __cne_node_stream_alloc_size(graph, node, node->size + nb_objs);
 
    return &node->objs[idx];
}
 
static inline void
cne_node_next_stream_put(struct cne_graph *graph, struct cne_node *node, cne_edge_t next,
                         uint16_t idx)
{
    if (unlikely(!idx))
        return;
 
    node = __cne_node_next_node_get(node, next);
 
    if (node->idx == 0)
        __cne_node_enqueue_tail_update(graph, node);
 
    node->idx += idx;
}
 
static inline void
cne_node_next_stream_move(struct cne_graph *graph, struct cne_node *src, cne_edge_t next)
{
    struct cne_node *dst = __cne_node_next_node_get(src, next);
 
    CNE_DEBUG("Src %-16s', Dst '%-16s' next %d\n", src->name, dst->name, next);
 
    /* Swap the pointers if dst doesn't have valid objs */
    if (likely(dst->idx == 0)) {
        void **dobjs = dst->objs;
        uint16_t dsz = dst->size;
 
        dst->objs = src->objs;
        dst->size = src->size;
        src->objs = dobjs;
        src->size = dsz;
        dst->idx  = src->idx;
 
        __cne_node_enqueue_tail_update(graph, dst);
    } else /* Move the objects from src node to dst node */
        cne_node_enqueue(graph, src, next, src->objs, src->idx);
}
 
#ifdef __cplusplus
}
#endif
 
#endif /* _CNE_GRAPH_WORKER_H_ */