bal_release/src/core/main/bal_worker.c

/******************************************************************************
 *
 *  <:copyright-BRCM:2016:DUAL/GPL:standard
 *  
 *     Copyright (c) 2016 Broadcom
 *     All Rights Reserved
 *  
 *  Unless you and Broadcom execute a separate written software license
 *  agreement governing use of this software, this software is licensed
 *  to you under the terms of the GNU General Public License version 2
 *  (the "GPL"), available at http://www.broadcom.com/licenses/GPLv2.php,
 *  with the following added to such license:
 *  
 *     As a special exception, the copyright holders of this software give
 *     you permission to link this software with independent modules, and
 *     to copy and distribute the resulting executable under terms of your
 *     choice, provided that you also meet, for each linked independent
 *     module, the terms and conditions of the license of that module.
 *     An independent module is a module which is not derived from this
 *     software.  The special exception does not apply to any modifications
 *     of the software.
 *  
 *  Not withstanding the above, under no circumstances may you combine
 *  this software in any way with any other Broadcom software provided
 *  under a license other than the GPL, without Broadcom's express prior
 *  written consent.
 *  
 *  :>
 *
 *****************************************************************************/
 
/**
 * @file bal_worker.c
 * @brief Initialization and message handler functions for the BAL Core
 *        worker thread including all core FSM initialization.
 *
 * @addtogroup ctrlr
 */

/*@{*/

/* Project includes */
#include <bcmos_system.h>
#include <bal_api.h>
#include <bal_msg.h>
#include <bal_osmsg.h>
#include "bal_worker.h"
#include "acc_term_fsm.h"
#include "flow_fsm.h"
#include "group_fsm.h"
#include "sub_term_fsm.h"
#include "tm_sched_fsm.h"
#include "tm_queue_fsm.h"
#include "fsm_common.h"
#include "bal_api_worker.h"
#include "bal_obj_msg_pack_unpack.h"

#include "bal_dpp_acc_term.h"
#include "rsc_mgr.h"

#include <arpa/inet.h>

#ifdef ENABLE_LOG
#include <bcm_dev_log.h>
/*
 * @brief The logging device id for the core
 */
extern dev_log_id log_id_core;
#endif

/* Local function declarations */
static void core_mgmt_msg_handler(bcmos_module_id module_id, bcmos_msg *msg);
static bcmos_errno process_mgmt_msg(void *msg_payload);
static bcmos_errno process_util_msg(void *msg_payload);

static int _bal_mgmt_rx_handler(long data);

static bcmos_errno process_packet_object(void *msg_payload);

/*
 * @brief The BAL core management queue.  These are the queues through which the core 
 * converses with the Public API. The remote endpoint of these queues are the 
 * BAL API backend.
 */
static bcmos_msg_queue core_mgmt_queue;
static bcmos_msg_queue core_api_queue;
static bcmos_msg_queue core_api_ind_queue;
bcmos_msg_queue *p_bal_core_queue;
bcmos_msg_queue *p_bal_core_to_api_queue;
bcmos_msg_queue *p_bal_core_to_api_ind_queue;

/*
 * @brief The rx thread for management message reception
 */
static bcmos_task rx_mgmt_thread;

/*****************************************************************************/
/**
 * @brief Initialize the worker thread
 *
 * A Worker module function that initializes all of the data structures and
 * FSMs associated with the BAL core.  This function called from bal_core_init
 * during BAL core startup. 
 *
 *****************************************************************************/
void core_worker_thread_init(void)
{
    /* 
     * Initialize all of the worker thread hosted FSMs and the resource manager
     */
     
    rsc_mgr_init();

    tm_queue_fsm_init();
		
    tm_sched_fsm_init();

    access_terminal_fsm_init();

    flow_fsm_init();

    sub_term_fsm_init();
    
    group_fsm_init();


    return;
}

/*****************************************************************************/
/**
 * @brief Finish the worker thread
 *
 * A Worker module function that un-initializes all of the data structures and
 * FSMs associated with the BAL core.  This function called from bal_core_init
 * during BAL core cleanup on exit. 
 *
 *****************************************************************************/
void core_worker_thread_finish(void)
{
    /*
     * Un-initialize the worker thread owned data structures. etc.
     */
    bcmos_msg_unregister(BCMBAL_MGMT_MSG, 0, BCMOS_MODULE_ID_WORKER_MGMT);

    bcmos_msg_unregister(BCMBAL_MAC_UTIL_MSG, 0, BCMOS_MODULE_ID_WORKER_MGMT);

    bcmos_task_destroy(&rx_mgmt_thread);

    sub_term_fsm_finish();

    flow_fsm_finish();

    group_fsm_finish();

    tm_sched_fsm_finish();

    tm_queue_fsm_finish();
	
    bcmos_msg_queue_destroy(&core_mgmt_queue);

    if (p_bal_core_to_api_queue == &core_api_queue)
        bcmos_msg_queue_destroy(p_bal_core_to_api_queue);

    if (p_bal_core_to_api_ind_queue == &core_api_ind_queue)
        bcmos_msg_queue_destroy(p_bal_core_to_api_ind_queue);
 
    return;
}


/*****************************************************************************/
/**
 * @brief The BAL core management message queue init
 *
 * A initialization function for the worker module.
 * It creates message queues for communication with API layer and UTILS
 *
 * @param mgmt_queue_info A pointer to the ip:port addresses required for the msg queues.
 *             The ip:port addresses can be NULL in BAL_MONOLITHIC mode. In this case
 *             inter-thread queues will be used for communication.
 *
 * @returns bcmos_errno
 *
 *****************************************************************************/
bcmos_errno core_msg_queue_init(mgmt_queue_addr_ports *mgmt_queue_info)
{
    bcmos_msg_queue_parm msg_q_p = {};
    bcmos_errno ret;

    do
    {
        /* Create core queue. BAL core listens on this queue
         * for public API calls and indications from the UTils
         */
        msg_q_p.name = "mgmt_rx_q";
        if (NULL != mgmt_queue_info->core_mgmt_ip_port)
        {
            /* The queue allows sending from core_mgmt_ip_port and receiving at core_mgmt_ip_port */
            msg_q_p.local_ep_address = mgmt_queue_info->core_mgmt_ip_port;
            msg_q_p.remote_ep_address = mgmt_queue_info->core_mgmt_ip_port;
            msg_q_p.ep_type = BCMOS_MSG_QUEUE_EP_UDP_SOCKET;
        }
        else
        {
            msg_q_p.ep_type = BCMOS_MSG_QUEUE_EP_LOCAL;
        }
        ret = bcmos_msg_queue_create(&core_mgmt_queue, &msg_q_p);
        if (BCM_ERR_OK != ret)
        {
            BCM_LOG(ERROR, log_id_core, "Couldn't create rx mgmt queue\n");
            break;
        }
        p_bal_core_queue = &core_mgmt_queue;
#ifdef BAL_MONOLITHIC
        if (NULL == mgmt_queue_info->core_mgmt_ip_port)
            p_balapi_to_core_queue = p_bal_core_queue;
#endif

        /* Now create a TX queue for sending replies from the core to API.
         * Only do it if using socket transport. In case of inter-thread queues
         * it will be created by API layer
         */
        if (NULL != mgmt_queue_info->balapi_mgmt_ip_port)
        {
            uint16_t portnum;
            char *p_portnum_str;
            char balapi_ind_port_str[256];

            msg_q_p.name = "mgmt_to_api_q";
            msg_q_p.local_ep_address = NULL;
            msg_q_p.remote_ep_address = mgmt_queue_info->balapi_mgmt_ip_port;
            msg_q_p.ep_type = BCMOS_MSG_QUEUE_EP_UDP_SOCKET;
            ret = bcmos_msg_queue_create(&core_api_queue, &msg_q_p);
            if (BCM_ERR_OK != ret)
            {
                BCM_LOG(ERROR, log_id_core, "Couldn't create tx mgmt queue\n");
                break;
            }
            p_bal_core_to_api_queue = &core_api_queue;

            /*
             * make a copy of the user chosen bal api mgmt port
             */
            strcpy(balapi_ind_port_str, mgmt_queue_info->balapi_mgmt_ip_port);

            /* Find the port number */
            p_portnum_str = strchr(balapi_ind_port_str, ':') + 1;

            /* convert to an integer and increment it by one */
            portnum = atoi(p_portnum_str) + 1;

            /* create the new string defining the BAL API indication port */
            sprintf(p_portnum_str,"%d", portnum);

            /* Create core->API indication queue
             */
            msg_q_p.name = "mgmt_ind_q";

            /* Set up the management IP:port access parameters to allow the core to send indications to
             * the BAL public API - this is a TX queue only!. 
             */
            msg_q_p.local_ep_address = NULL;
            msg_q_p.remote_ep_address = balapi_ind_port_str;
            msg_q_p.ep_type = BCMOS_MSG_QUEUE_EP_UDP_SOCKET;

            ret = bcmos_msg_queue_create(&core_api_ind_queue, &msg_q_p);
            if (BCM_ERR_OK != ret)
            {
                BCM_LOG(ERROR, log_id_core, "Couldn't create the core tx indication queue\n");
                break;
            }

            BCM_LOG(DEBUG, log_id_core, "Created the core tx indication queue\n");

            p_bal_core_to_api_ind_queue = &core_api_ind_queue;
        }
    } while (0);

    return ret;
}


/*****************************************************************************/
/**
 * @brief The BAL core management worker module init
 *
 * A initialization function for the worker module.
 * It registers for messages that this module is expected to process.
 *
 * @returns bcmos_errno
 *
 *****************************************************************************/
bcmos_errno _bal_worker_mgmt_module_init(long data)
{

    bcmos_task_parm task_p = {};
    bcmos_errno ret = BCM_ERR_OK;

    do
    {

        /* Create Mgmt RX thread */
        task_p.name = "mgmt_mgmt_rx";
        task_p.priority = TASK_PRIORITY_IPC_RX;
        task_p.handler = _bal_mgmt_rx_handler;
        task_p.data = (long)p_bal_core_queue;

        ret = bcmos_task_create(&rx_mgmt_thread, &task_p);
        if (ret)
        {
            BCM_LOG(ERROR, log_id_core, "Couldn't create Mgmt RX thread\n");
            break;
        }

        /* Register the message types to be handled by the mgmt module
         */
        bcmos_msg_register(BCMBAL_MGMT_MSG, 0, BCMOS_MODULE_ID_WORKER_MGMT, core_mgmt_msg_handler);
        bcmos_msg_register(BCMBAL_MAC_UTIL_MSG, 0, BCMOS_MODULE_ID_WORKER_MGMT, core_util_msg_handler);
    }
    while(0);

    return ret;
}

/*****************************************************************************/
/**
 * @brief Mgmt RX handler
 *
 * This handler is executed in the context of an RX thread.  It's purpose 
 * is to dispatch the message received here to the module that has registered
 * to process this message.  The message is then processed in the context 
 * of the thread to which that module is attached.
 *
 * @param data A pointer to the received message
 *
 * @returns 0 on success, -EINVAL on failure
 *
 *****************************************************************************/
static int _bal_mgmt_rx_handler(long data)
{
    bcmos_msg_queue *rxq = (bcmos_msg_queue *)data;
    bcmos_task *my_task = bcmos_task_current();
    void *payload;
    bcmos_msg *msg;
    bcmos_errno ret = BCM_ERR_OK;

    while (!my_task->destroy_request)
    {
        payload = NULL;
        ret = bcmbal_msg_recv(rxq, BCMOS_WAIT_FOREVER, &payload);
        if (ret)
        {
            /* Unexpected failure */
            BCM_LOG(ERROR, log_id_core, "bcmbal_msg_recv() -> %s\n", bcmos_strerror(ret));
            continue;
        }

        /* Message received */
        BCM_LOG(DEBUG, log_id_core, "bcmbal_msg_recv(%p) -> %s\n", payload, bcmos_strerror(ret));

       /* 
         * Got a message, so now dispatch it.  This will result in one
         * of the modules (registered for the message being processed)
         * executing its message callback handler.
         *
         */
        msg = bcmbal_bcmos_hdr_get(payload);
        ret = bcmos_msg_dispatch(msg, BCMOS_MSG_SEND_AUTO_FREE);
        if (ret)
        {
            BCM_LOG(ERROR, log_id_core, 
                    "Couldn't dispatch message %d:%d\n", 
                    (int)msg->type, (int)msg->instance);
        }
    }

    my_task->destroyed = BCMOS_TRUE;

    return (BCM_ERR_OK == ret) ? 0 : -EINVAL;
}

/*****************************************************************************/
/**
 * @brief The BAL core management message handler
 *
 * A function that handlers messages received from the
 * BAL public API via an RX thread.
 *
 * This function executes in the context of the worker thread.
 *
 * @param module_id  The index of the module that this message handler
 *                   is associated with.
 *
 * @param msg        A pointer to the received message to be processed
 *
 *****************************************************************************/
static void core_mgmt_msg_handler(bcmos_module_id module_id, bcmos_msg *msg)
{

    void *msg_payload;

    /* Parameter checks */
    BUG_ON(NULL == msg);

    /*
     * @to-do 
     * validate the message major and minor version is correct
     */

    /*
     * Point to the message payload for further processing
     */
    msg_payload = bcmbal_payload_ptr_get(bcmbal_bal_hdr_get_by_bcmos_hdr(msg));

    /*
     * These are messages from the BAL Public API or the UTILS
     *
     * - call into appropriate FSM
     *    
     */
    BCM_LOG(DEBUG, log_id_core, "Received a mgmt message (payload at %p)\n", msg_payload);


    if(BCMBAL_MGMT_MSG == bcmbal_type_major_get(msg_payload) && 
       BAL_SUBSYSTEM_PUBLIC_API == bcmbal_sender_get(msg_payload))
    {
        /* Process the message */
        process_mgmt_msg(msg_payload);
    }
    else
    {
        BCM_LOG(FATAL, log_id_core, "message received with wrong major type/subsystem combination (%d/\%d)\n", 
                bcmbal_type_major_get(msg_payload),
                bcmbal_sender_get(msg_payload));

        BUG_ON(BCMOS_TRUE);
    }

}

/*****************************************************************************/
/**
 * @brief The BAL core management message processing function
 *
 * A Worker module function that handles messages received from the
 * BAL public API or UTILs via an RX thread.
 *
 * This function executes in the context of the worker thread
 *
 * @param msg_payload A pointer to the message to be processed
 *
 * @returns bcmos_errno
 *
 *****************************************************************************/
static bcmos_errno process_mgmt_msg(void *msg_payload)
{

    bcmos_errno ret = BCM_ERR_OK;
    bcmbal_obj_id objtype;

    BCM_LOG(DEBUG, log_id_core, "Processing a management message\n");

    objtype = bcmbal_msg_id_obj_get(msg_payload);

    /*
     * Process the message based on the type of BAL object and sender
     * in the message.
     */
    switch(objtype)
    {
        case BCMBAL_OBJ_ID_FLOW:
        {
            ret = process_flow_object(msg_payload);
            break;                
        }
        
        case BCMBAL_OBJ_ID_GROUP:
        {
            ret = process_group_object(msg_payload);
            break;                
        }

        case BCMBAL_OBJ_ID_ACCESS_TERMINAL:
        {
            ret = process_access_terminal_object(msg_payload);
            break;
        }

        case BCMBAL_OBJ_ID_INTERFACE:
        {
            ret = process_interface_object(msg_payload);
            break;
        }

        case BCMBAL_OBJ_ID_SUBSCRIBER_TERMINAL:
        {
            ret = process_subscriber_terminal_object(msg_payload);
            break;
        }

        case BCMBAL_OBJ_ID_PACKET:
        {

            ret = process_packet_object(msg_payload);

            /* We respond to the BAL public API backend with a result. We always 
             * send a complete msg_payload back to the API, but the data portion 
             * of the object is only relevant when a GET or GET-STATS has been requested.
             */
            mgmt_msg_send_balapi_rsp(ret, msg_payload, ((bcmbal_obj *)msg_payload)->type, log_id_core);
    
            break;
 
        }

        case BCMBAL_OBJ_ID_TM_SCHED:
        {
            ret = process_tm_sched_object(msg_payload);
            break;				 
        }
		
        case BCMBAL_OBJ_ID_TM_QUEUE:
        {
            ret = process_tm_queue_object(msg_payload);
            break;				 
        }

        default:
        {
            BCM_LOG(ERROR, log_id_core, 
                    "Unsupported object detected in management message\n");
            ret = BCM_ERR_NOT_SUPPORTED;

            /* We respond to the BAL public API backend with a result. We always 
             * send a complete msg_payload back to the API, but the data portion 
             * of the object is only relevant when a GET or GET-STATS has been requested.
             */
            mgmt_msg_send_balapi_rsp(ret, msg_payload, ((bcmbal_obj *)msg_payload)->type, log_id_core);
    
            break;
        }
    }
    bcmbal_msg_free(msg_payload);

    return ret;
}

/*****************************************************************************/
/**
 * @brief The BAL core util message handler
 *
 * A function that handlers messages received from the
 * BAL utils via an RX thread.
 *
 * This function executes in the context of the worker thread.
 *
 * @param module_id  The index of the module that this message handler
 *                   is associated with.
 *
 * @param msg        A pointer to the received message to be processed
 *
 *****************************************************************************/
void core_util_msg_handler(bcmos_module_id module_id, bcmos_msg *msg)
{

    void *msg_payload;

    /* Parameter checks */
    BUG_ON(NULL == msg);

    /*
     * @to-do 
     * validate the message major and minor version is correct
     */

    /*
     * Point to the message payload for further processing
     */
    msg_payload = bcmbal_payload_ptr_get(bcmbal_bal_hdr_get_by_bcmos_hdr(msg));

    /*
     * These are messages from the BAL Utils
     *
     * - call into appropriate FSM
     *    
     */
    BCM_LOG(DEBUG, log_id_core, "Received a Util message (payload at %p)\n", msg_payload);

    if(BCMBAL_MAC_UTIL_MSG == bcmbal_type_major_get(msg_payload) &&
       BAL_SUBSYSTEM_MAC_UTIL == bcmbal_sender_get(msg_payload))
    {           
        /* Process the message */
        process_util_msg(msg_payload);

    }
    else
    {
        BCM_LOG(FATAL, log_id_core, "message received with wrong major type/subsystem combination (%d/\%d)\n", 
                bcmbal_type_major_get(msg_payload),
                bcmbal_sender_get(msg_payload));

        BUG_ON(BCMOS_TRUE);
    }
}

/*****************************************************************************/
/**
 * @brief The BAL core util message processing function
 *
 * A Worker module function that handles messages received from the
 * BAL utils via an RX thread.
 *
 * This function executes in the context of the worker thread
 *
 * @param msg_payload A pointer to the message to be processed
 *
 * @returns bcmos_errno
 *
 *****************************************************************************/
static bcmos_errno process_util_msg(void *msg_payload)
{

    bcmos_errno ret = BCM_ERR_OK;
    bcmbal_obj_id objtype;

    BCM_LOG(DEBUG, log_id_core, "Processing a util message\n");

    objtype = bcmbal_msg_id_obj_get(msg_payload);

    /*
     * Process the util message based on the type of BAL object
     * in the message.
     */
    switch(objtype)
    {
        case BCMBAL_OBJ_ID_FLOW:
        {
            ret = process_flow_util_msg(msg_payload);
            break;                
        }
        
        case BCMBAL_OBJ_ID_GROUP:
        {
            ret = process_group_util_msg(msg_payload);
            break;                
        }


        case BCMBAL_OBJ_ID_ACCESS_TERMINAL:
        {
            ret = process_access_terminal_util_msg(msg_payload);
            break;
        }

        case BCMBAL_OBJ_ID_INTERFACE:
        {
            ret = process_interface_util_msg(msg_payload);
            break;
        }

        case BCMBAL_OBJ_ID_SUBSCRIBER_TERMINAL:
        {
            ret = process_subscriber_terminal_util_msg(msg_payload);
            break;
        }

        case BCMBAL_OBJ_ID_TM_SCHED:
        {
            ret = process_tm_sched_util_msg(msg_payload);
            break;
        }

        default:
        {
            BCM_LOG(ERROR, log_id_core, 
                    "Unsupported object detected in message received from util\n");
            ret = BCM_ERR_NOT_SUPPORTED;

            break;
        }
    }
    /*
     * Free the message after processing
     */
    bcmbal_msg_free(msg_payload);

    return ret;
}

/*****************************************************************************/
/**
 * @brief The BAL core "packet send" message processing function
 *
 * A Worker module function that handles packet send messages received from the
 * BAL public API via an RX thread.
 *
 * This function executes in the context of the worker thread
 *
 * @param msg_payload A pointer to the message to be processed
 *
 * @returns bcmos_errno
 *
 *****************************************************************************/
static bcmos_errno process_packet_object(void *msg_payload)
{

    bcmos_errno ret = BCM_ERR_OK;
    uint8_t *p_user_pkt;
    uint16_t user_pkt_len;
    uint8_t dst_port_id;
    bcmos_bool b_port_id_is_nni;
    bcmbal_packet_cfg *p_packet_obj;
    uint16_t tunnel_tag_vlan_id = 0;
 
    BCM_LOG(DEBUG, log_id_core, "Processing a \"Packet send\" message\n");

    p_packet_obj = (bcmbal_packet_cfg *)msg_payload;
    
    /* Extract the length of the user packet to be proxied */
    user_pkt_len = p_packet_obj->data.pkt.len;

    /* Derive a pointer to the user packet */
    p_user_pkt = p_packet_obj->data.pkt.val;

    BCM_LOG(DEBUG, log_id_core, "user packet first 12 bytes %02X%02X%02X%02X%02X%02X %02X%02X%02X%02X%02X%02X\n",
            p_user_pkt[0], p_user_pkt[1], p_user_pkt[2], p_user_pkt[3], p_user_pkt[4], p_user_pkt[5], 
            p_user_pkt[6], p_user_pkt[7], p_user_pkt[8], p_user_pkt[9], p_user_pkt[10], p_user_pkt[11]);

    /* Is this packet destined to an NNI or PON port? */
    b_port_id_is_nni = (BCMBAL_DEST_TYPE_NNI == p_packet_obj->key.packet_send_dest.type) ?
        BCMOS_TRUE : BCMOS_FALSE;

    /*
     * Process the message based on the type of BAL object
     * in the message.
     */
    switch(p_packet_obj->hdr.hdr.obj_type)
    {
        case BCMBAL_OBJ_ID_PACKET:
        {
            
            dst_port_id = (BCMOS_TRUE == b_port_id_is_nni) ? 
                          (p_packet_obj->key.packet_send_dest.u.nni.int_id) :
                          (p_packet_obj->key.packet_send_dest.u.sub_term.int_id);
            if(BCMOS_FALSE == b_port_id_is_nni)
            {
                 /*
                 * Packets destined to a PON interface require a tunnel tag
                 *
                 * Get the svc_port_id for the first flow on the subscriber terminal (if there is one)
                 */
                if(BCM_ERR_OK != svc_port_id_for_sub_term_ds_flow_get(p_packet_obj->key.packet_send_dest.u.sub_term.int_id,
                                                                      p_packet_obj->key.packet_send_dest.u.sub_term.sub_term_id,
                                                                      p_packet_obj->key.packet_send_dest.u.sub_term.sub_term_uni,
                                                                      &tunnel_tag_vlan_id))
                {
                    BCM_LOG(ERROR, 
                            log_id_core, 
                            "Packet send could not find any downstream FLOW to send packet for sub_term_id %d (PON %d)\n",
                            p_packet_obj->key.packet_send_dest.u.sub_term.sub_term_id,
                            p_packet_obj->key.packet_send_dest.u.sub_term.int_id);                   
                    ret = BCM_ERR_NOENT;
                    break;
                }    
              
            }
            
            ret = sw_util_pkt_send(dst_port_id, 
                                  (b_port_id_is_nni ? REASON_SEND_TO_NNI : REASON_SEND_TO_PON),
                                   p_user_pkt, 
                                   user_pkt_len,
                                   (int)tunnel_tag_vlan_id);       
        }
        break;                
            
        default:
        {
            BCM_LOG(ERROR, log_id_core, 
                    "Unsupported object detected in \"packet send\" message\n");
            ret = BCM_ERR_NOT_SUPPORTED;
        }
        break;

    }
    /*
     * NOTE: DO NOT free the message after processing here.  It is freed in the calling function
     */
 
    return ret;
}

/*@}*/