/******************************************************************************
 *
 *  <: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 flow_fsm.c
 * @brief Code to support the BAL flow FSM
 *
 * @addtogroup flow
 *
 */

/*@{*/

#include <bcmos_system.h>
#include <flow_fsm.h>
#include <tm_sched_fsm.h>
#include <tm_queue_fsm.h>


#include <bal_msg.h>
#include <bal_osmsg.h>
#include "bal_worker.h"
#include "bal_mac_util.h"
#include "bal_switch_util.h"
#include "rsc_mgr.h"

#include <bal_objs.h>
#include <fsm_common.h>
#include <bal_switch_flow.h>

#ifdef ENABLE_LOG
#include <bcm_dev_log.h>

 /*
 * @brief The logging device id for flow
 */
static dev_log_id log_id_flow;
#endif

/* local function declarations */
static bcmos_errno flow_fsm_admin_up_start(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event);

static bcmos_errno flow_fsm_admin_up_error(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event);

static bcmos_errno flow_fsm_admin_dn_start(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event);

static bcmos_errno flow_fsm_admin_dn_ok(flow_inst *p_flow_inst, 
                                        void *msg, 
                                        flow_fsm_event *p_event);

static bcmos_errno flow_fsm_admin_dn_error(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event);

static bcmos_errno flow_fsm_ignore_util_msg(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event);

static bcmos_errno flow_fsm_removing_process_util_msg(flow_inst *p_flow_inst, 
                                                     void *msg, 
                                                     flow_fsm_event *p_event);

static bcmos_errno flow_fsm_removing_process_util_auto_msg(flow_inst *p_flow_inst, 
                                                          void *msg, 
                                                          flow_fsm_event *p_event);

static bcmos_errno flow_fsm_null_process_util_auto_msg(flow_inst *p_flow_inst, 
                                                      void *msg, 
                                                      flow_fsm_event *p_event);

static bcmos_errno flow_fsm_process_util_msg(flow_inst *p_flow_inst, 
                                                        void *msg, 
                                                        flow_fsm_event *p_event);

static bcmos_errno flow_fsm_process_util_auto_msg(flow_inst *p_flow_inst, 
                                                             void *msg, 
                                                             flow_fsm_event *p_event);

static bcmos_errno flow_fsm_clear_start(flow_inst *p_flow_inst, 
                                                             void *msg, 
                                                             flow_fsm_event *p_event);


static bcmos_errno flow_fsm_state_err(flow_inst *p_flow_inst, 
                                      void *msg, 
                                      flow_fsm_event *p_event);

static bcmos_errno flow_fsm_exec(flow_inst *p_flow_inst, flow_fsm_event *p_event);

static flow_inst *flow_inst_get(bcmbal_flow_key *key, flow_flag search_flag, bcmos_bool *is_new_flow);


#ifdef FREE_FLOW_BY_KEY_SUPPORTED
static bcmos_errno flow_free_by_key(bcmbal_flow_key *key);
#endif
static bcmos_errno flow_free_by_entry(flow_inst *p_entry);

static bcmos_errno flow_tm_get(bcmbal_flow_cfg *p_flow_info, tm_sched_inst **p_tm_sched_inst);
static bcmos_errno flow_queue_validate(bcmbal_flow_cfg *p_flow_cfg, tm_queue_inst **p_tm_queue_inst);


/*
 * @brief The Global flow fsm context data structure
 */
static flow_fsm_ctx g_flow_fsm_flow_list_ctx;

/*
 * Macros for flow ctx access
 */
#define FLOW_FSM_FLOW_LIST_CTX      (g_flow_fsm_flow_list_ctx)
#define FLOW_FSM_FLOW_LIST_CTX_PTR  (&g_flow_fsm_flow_list_ctx)

/*
 *  @brief The definition of a flow FSM state processing function
 */
typedef bcmos_errno (* flow_fsm_state_processor)(flow_inst *, void *, flow_fsm_event *);

/*
 *  @brief The Flow FSM state processing array
 */
static flow_fsm_state_processor flow_states[FLOW_FSM_STATE__NUM_OF][FLOW_FSM_EVENT_TYPE__NUM_OF] =
{
 
    [FLOW_FSM_STATE_NULL] = 
    {        
        /*
         * Next state: CONFIGURING
         */ 
        [FLOW_FSM_EVENT_TYPE_ADMIN_UP]     = flow_fsm_admin_up_start,   

        /* 
         * Next state: NULL 
         */   
        [FLOW_FSM_EVENT_TYPE_ADMIN_DN]     = flow_fsm_admin_dn_ok,

        /* 
         * Next state: NULL 
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_MSG]      = flow_fsm_ignore_util_msg,

        /* 
         * Next state: NULL 
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_AUTO_MSG] = flow_fsm_null_process_util_auto_msg,
    },
    [FLOW_FSM_STATE_CONFIGURING] =                                        
    {   
        /*
         * Next state: CONFIGURING
         */ 
        [FLOW_FSM_EVENT_TYPE_ADMIN_UP]     = flow_fsm_admin_up_start,   

        /* 
         * Next state: CONFIGURING 
         */   
        [FLOW_FSM_EVENT_TYPE_ADMIN_DN]     = flow_fsm_admin_dn_start,

        /* 
         * Next state: CONFIGURING | CONFIGURED 
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_MSG]      = flow_fsm_process_util_msg,

        /* 
         * Next state: REMOVING
         */
        [FLOW_FSM_EVENT_TYPE_REMOVE]     = flow_fsm_clear_start,

        /* 
         * Next state: CONFIGURING 
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_AUTO_MSG] = flow_fsm_process_util_auto_msg,

    },

    [FLOW_FSM_STATE_CONFIGURED] =                                         
    {     
        /*
         * Next state: CONFIGURED
         */ 
        [FLOW_FSM_EVENT_TYPE_ADMIN_UP]     = flow_fsm_admin_up_start,   

        /* 
         * Next state: CONFIGURING
         */   
        [FLOW_FSM_EVENT_TYPE_ADMIN_DN]     = flow_fsm_admin_dn_start,

        /* 
         * Next state: REMOVING
         */
        [FLOW_FSM_EVENT_TYPE_REMOVE]     = flow_fsm_clear_start,

        /* 
         * Next state: CONFIGURING
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_MSG]      = flow_fsm_process_util_msg,

        /* 
         * Next state: CONFIGURED
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_AUTO_MSG] = flow_fsm_process_util_auto_msg,

    },

    [FLOW_FSM_STATE_REMOVING] =                                      
    {        
        /*
         * Next state: REMOVING 
         */ 
        [FLOW_FSM_EVENT_TYPE_ADMIN_UP]     = flow_fsm_admin_up_error,   

        /* 
         * Next state: REMOVING 
         */   
        [FLOW_FSM_EVENT_TYPE_ADMIN_DN]     = flow_fsm_admin_dn_error,

         /* 
         * Next state: REMOVING | NULL
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_MSG]      = flow_fsm_removing_process_util_msg,

        /* 
         * Next state: REMOVING 
         */
        [FLOW_FSM_EVENT_TYPE_UTIL_AUTO_MSG] = flow_fsm_removing_process_util_auto_msg,
    },

};

static char *state_name_str[] = 
{
    "FLOW_FSM_STATE_NULL",
    "FLOW_FSM_STATE_CONFIGURING",
    "FLOW_FSM_STATE_CONFIGURED",
    "FLOW_FSM_STATE_REMOVING",
};

/* Ensure that the name array size matches the associated enum */
BAL_STATIC_ASSERT (FLOW_FSM_STATE__LAST == (sizeof (state_name_str) / sizeof (char *)), flow_fsm_state);

static char *flow_state_name_get(flow_fsm_state state)
{
    if(state < FLOW_FSM_STATE__LAST)
    {
        return state_name_str[state];
    }
    else
    {
        return "FLOW_UNKNOWN";
    }
}

static char *event_name_str[] = 
{
    "FLOW_FSM_ADMIN_UP_EVENT",
    "FLOW_FSM_ADMIN_DN_EVENT",
    "FLOW_FSM_REMOVE_EVENT",
    "FLOW_FSM_UTIL_MSG_EVENT",
    "FLOW_FSM_UTIL_AUTO_MSG_EVENT",
};

/* Ensure that the name array size matches the associated enum */
BAL_STATIC_ASSERT (FLOW_FSM_EVENT_TYPE__LAST == (sizeof (event_name_str) / sizeof (char *)), flow_fsm_event_type);

static char *flow_event_name_get(flow_fsm_event_type event)
{
    if(event < FLOW_FSM_EVENT_TYPE__LAST)
    {
        return event_name_str[event];
    }
    else
    {
        return "FLOW_EVT_UNKNOWN";
    }
}

/*****************************************************************************/
/**
 * @brief A function to initialize the current_flow_info object of the 
 *        supplied entry.
 *
 * @param p_entry A pointer to the entry to be initialized
 *
 *
 * @returns void
 *****************************************************************************/
static void flow_inst_entry_obj_init(flow_inst *p_entry)
{
    /* The actual key content is irrelevant for free flows */
    bcmbal_flow_key key = { .flow_id = 0, .flow_type = BCMBAL_FLOW_TYPE_DOWNSTREAM };

    BCMBAL_CFG_INIT(&p_entry->current_flow_info,
                    flow,
                    key);
        
    BCMBAL_CFG_PROP_SET(&p_entry->current_flow_info,
                    flow,
                    admin_state,
                    BCMBAL_STATE_DOWN);

    BCMBAL_CFG_PROP_SET(&p_entry->current_flow_info,
                        flow,
                        oper_status,
                        BCMBAL_STATUS_DOWN);
            
    BCMBAL_OBJ_IN_PROGRESS_SET(&(p_entry->current_flow_info), BCMOS_FALSE);

}

/*****************************************************************************/
/**
 * @brief A function to initialize the Flow FSM infrastructure.
 *  
 *        NOTE: This is called once on startup and NOT for each FSM instance.
 *
 * @returns bcmos_errno
 *****************************************************************************/
bcmos_errno flow_fsm_init(void)
{
    int ii;
    flow_inst *new_entry;
    bcmos_errno ret = BCM_ERR_OK;

#ifdef ENABLE_LOG
    log_id_flow = bcm_dev_log_id_register("FLOW", DEV_LOG_LEVEL_INFO, DEV_LOG_ID_TYPE_BOTH);
    BUG_ON(log_id_flow == DEV_LOG_INVALID_ID);
#endif

    /* Initialize all of the flow queues */
    TAILQ_INIT(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list);
    TAILQ_INIT(&FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list);

    /* Populate the free list with it's initial set of flows
     */
    for(ii=0; ii<FLOW_ALLOCATION_BLOCK_SIZE; ii++)
    {
        
        new_entry = bcmos_calloc(sizeof(flow_inst));

        if (NULL == new_entry)
        {
            BCM_LOG(FATAL, log_id_flow,  "Failed to initialize the flow free list - FATAL\n");
            ret = BCM_ERR_NOMEM;
            break;
        }

       flow_free_by_entry(new_entry);
    }

    return ret;
}

/*****************************************************************************/
/**
 * @brief A function to un-initialize the Flow FSM infrastructure.
 *  
 *        NOTE: This is called once on shutdown and NOT for each FSM instance.
 *
 * @returns bcmos_errno
 *****************************************************************************/
bcmos_errno flow_fsm_finish(void)
{

    flow_inst *current_entry, *p_temp_entry;

    /* Free all the entries on the active list */
    TAILQ_FOREACH_SAFE(current_entry, 
                       &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, 
                       flow_inst_next,
                       p_temp_entry)
    {
        /* Remove it from the active list */
        TAILQ_REMOVE(&FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, current_entry, flow_inst_next);

        bcmos_free(current_entry);

    }

    /* Free all the entries on the free list */
    TAILQ_FOREACH_SAFE(current_entry, 
                       &FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list, 
                       flow_inst_next,
                       p_temp_entry)
    {
        /* Remove it from the active list */
        TAILQ_REMOVE(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list, current_entry, flow_inst_next);

        bcmos_free(current_entry);
    }
     
    return BCM_ERR_OK;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing executive function
 *
 * @param p_flow_inst  Pointer to a flow instance
 * @param p_event      Pointer to a flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_exec(flow_inst *p_flow_inst, flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;
    flow_fsm_state pre_state; 
    flow_fsm_state_processor flow_state_processor;     

    /* Parameter checks */
    BUG_ON(NULL == p_flow_inst);
    BUG_ON(NULL == p_event);

    /* Record the present state for debug printing
     */
    pre_state = p_flow_inst->fsm_state;

    /*
     * Get the state processing function 
     */
    flow_state_processor = flow_states[p_flow_inst->fsm_state][p_event->event_type];

    /*
     * If there's a state processing function for this event and state, execute it.
     * Otherwise, process a generic error.
     */
    if (flow_state_processor)
    {
        ret = flow_state_processor(p_flow_inst, p_event->msg, p_event);
    } else        
    {
        flow_fsm_state_err(p_flow_inst, p_event->msg, p_event);
    }
    
    if(BCM_ERR_OK != ret)
    {
        BCM_LOG(ERROR, log_id_flow, "*** Error detected during state processing\n");
        p_flow_inst->fsm_state = pre_state;
    }

    BCM_LOG(DEBUG, log_id_flow,  "*** Event %s, State: %s --> %s\n\n",
            flow_event_name_get(p_event->event_type),
            flow_state_name_get(pre_state), 
            flow_state_name_get(p_flow_inst->fsm_state)); 

    return ret;
}


/*****************************************************************************/
/**
 * @brief The Flow FSM state processing for a flow admin-up command received 
 *        from the BAL Public API when the specified flow instance is in the 
 *        admin-down state (i.e. when the flow instance FSM is in the NULL state).
 *
 * @param p_flow_inst      Pointer to an flow instance
 * @param msg              Pointer to a BAL message received from the BAL Public API
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_admin_up_start(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event)
{

    bcmos_errno ret = BCM_ERR_OK;
    bcmos_bool is_ds_flow_to_host, is_ds_n_to_1;


    BCM_LOG(INFO, log_id_flow,  "Got admin UP request from BAL API - bringing up FLOW\n");

    do
    {
        /* change Flow state to CONFIGURING */
        p_flow_inst->fsm_state = FLOW_FSM_STATE_CONFIGURING;

        /*– Core calls Switch Utils to add applicable CMDs */
        if(BCM_ERR_OK != (ret = sw_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_ADD)))
        {
            BCM_LOG(ERROR, log_id_flow,  "error %s detected by switch util while adding flow\n", bcmos_strerror(ret));
            break;
        }

        /*– Core calls Mac Utils add applicable CMDs */
        if(BCM_ERR_OK != (ret = mac_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_ADD)))
        {
            BCM_LOG(ERROR, log_id_flow,  "error %s detected by mac util\n", bcmos_strerror(ret));

            /* Remove the (just added) flow from the switch otherwise the switch utils 
             * will remember it and complain when this flow is added later.  There's not
             * much we can do about it if removing this flow fails.
             */
            if(BCM_ERR_OK != sw_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_CLEAR))
            {
                BCM_LOG(ERROR, log_id_flow,  
                        "error detected by switch util while removing flow\n");
            }

            break;
        }

        /* The hardware has properly accepted the object info, so the request object becomes
         * the current state.
         */
        bcmbal_flow_object_overlay_w_src_priority(&p_flow_inst->current_flow_info,
                                                  &p_flow_inst->api_req_flow_info);
 
        is_ds_flow_to_host = (BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_inst->api_req_flow_info.key.flow_type &&
                             (BCMOS_TRUE == BCMBAL_CFG_PROP_IS_SET(&p_flow_inst->api_req_flow_info, flow, action) && 
                             (p_flow_inst->api_req_flow_info.data.action.cmds_bitmask & 
                              BCMBAL_ACTION_CMD_ID_TRAP_TO_HOST)));
                              
        is_ds_n_to_1 = ((BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_inst->api_req_flow_info.key.flow_type) && 
                        (BCMBAL_CFG_PROP_IS_SET(&p_flow_inst->api_req_flow_info, flow, group_id)  ));


        /* Add the svc_port_id record to the sub_term record for upstream flows, 
         * or for downstream flows that are not destined to the host
         */
        if(BCMBAL_FLOW_TYPE_UPSTREAM   == p_flow_inst->api_req_flow_info.key.flow_type  ||
           (BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_inst->api_req_flow_info.key.flow_type &&
            !(is_ds_flow_to_host) && !(is_ds_n_to_1)))
    
        {
            bcmbal_sub_term_svc_port_id_list_entry_add(p_flow_inst->p_sub_term_inst,
                                                       p_flow_inst->api_req_flow_info.data.svc_port_id);
        }
 
        /* Add the agg_port_id to the sub_term record (only for upstream flows) */
        if(BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type)
        {
            bcmbal_sub_term_agg_port_id_list_entry_add(p_flow_inst->p_sub_term_inst,
                                                       p_flow_inst->api_req_flow_info.data.agg_port_id);
        }
 
        BCMBAL_OBJ_IN_PROGRESS_SET(&(p_flow_inst->current_flow_info), BCMOS_TRUE);

    }while(0);

    /* If there were errors during processing, then report the error to the API and free the flow */
    if(BCM_ERR_OK != ret)
    {
        mgmt_msg_send_balapi_ind(ret,
                                 msg, 
                                 log_id_flow);

        flow_free_by_entry(p_flow_inst);
    }

    return ret;

}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing for a flow admin-up command received 
 *        from the BAL Public API when the specified flow FSM is already 
 *        in the REMOVING state.
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from the BAL Public API
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_admin_up_error(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_STATE;

    BCM_LOG(DEBUG, log_id_flow,  
            "Received an admin UP request from BAL API"
            " - returning ERROR to the API - no further function\n");

    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing for a flow admin-down command 
 *        received from the BAL Public API when the specified flow is
 *        admin-up (i.e when the specified flow instance FSM is in the
 *        CONFIGURED state).
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from the BAL Public API
 * @param p_event          Pointer to a flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_admin_dn_start(flow_inst *p_flow_inst, 
                                           void *msg,
                                           flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;

    BCM_LOG(INFO, log_id_flow,
            "Got admin DOWN request from BAL API - bringing down FLOW\n");

    /* change Flow state to CONFIGURING */
    p_flow_inst->fsm_state = FLOW_FSM_STATE_CONFIGURING;

    do
    {
        /*– Core calls Switch Utils to remove applicable CMDs */
        if(BCM_ERR_OK != (ret = sw_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_REMOVE)))
        {
            BCM_LOG(ERROR, log_id_flow,  "error %s detected by switch util\n", bcmos_strerror(ret));
            break;
        }

        /*– Core calls Mac Utils remove applicable CMDs */
        if(BCM_ERR_OK != (ret = mac_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_REMOVE)))
        {
            BCM_LOG(ERROR, log_id_flow,  "error %s detected by mac util\n", bcmos_strerror(ret));
            break;
        }

        /* The hardware has properly accepted the object info but we do
         * not overwrite the current flow data as there is nothing in the request
         * that is relevant besides the admin_state
         */
        BCMBAL_OBJ_IN_PROGRESS_SET(&(p_flow_inst->current_flow_info), BCMOS_TRUE);

    }while(0);
    /* Report any error found to the API immediately */
    if(BCM_ERR_OK != ret)
    {
        mgmt_msg_send_balapi_ind(ret,
                                 msg, 
                                 log_id_flow);
    }

    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing for a flow clear command 
 *        received from the BAL Public API. 
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from the BAL Public API
 * @param p_event          Pointer to a flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_clear_start(flow_inst *p_flow_inst, 
                                           void *msg,
                                           flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;
    tm_sched_inst *p_tm_sched_inst;
    uint32_t alloc_ref_count;
    bcmos_bool b_flow_is_destined_to_host;
	tm_queue_inst *p_tm_queue_inst = NULL;
    
    BCM_LOG(INFO, log_id_flow,
    "Got CLEAR request from BAL API - removing FLOW\n");
    
    /* change Flow state to REMOVING */
    p_flow_inst->fsm_state = FLOW_FSM_STATE_REMOVING;

	
	b_flow_is_destined_to_host = ((BCMBAL_CFG_PROP_IS_SET(&(p_flow_inst->api_req_flow_info), flow, action) && 
								   (p_flow_inst->api_req_flow_info.data.action.cmds_bitmask & 
									BCMBAL_ACTION_CMD_ID_TRAP_TO_HOST)) ? BCMOS_TRUE : BCMOS_FALSE);
    do
    {
        /*Core calls Switch Utils to clear applicable CMDs */
        if(BCM_ERR_OK != (ret = sw_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_CLEAR)))
        {
            BCM_LOG(ERROR, log_id_flow,  "error %s detected by switch util\n", bcmos_strerror(ret));
            break;
        }
        
        /*Core calls Mac Utils clear applicable CMDs */
        if(BCM_ERR_OK != (ret = mac_util_flow_set(p_flow_inst, BAL_UTIL_OPER_FLOW_CLEAR)))
        {
            /* if entry does not exist for a clear, that is fine, since anyway that would have been the end goal */
            if (BCM_ERR_NOENT != ret)
            {
                BCM_LOG(ERROR, log_id_flow,  "error %s detected by mac util\n", bcmos_strerror(ret));
                break;
            }
        }
        
        if(BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type)
        {
            /*handle the alloc id and alloc - tm sched*/
            if(BCM_ERR_OK != rsc_mgr_alloc_id_free(p_flow_inst->api_req_flow_info.data.access_int_id,
                                                   p_flow_inst->api_req_flow_info.data.agg_port_id, p_flow_inst))
            {
                BCM_LOG(ERROR, log_id_flow, 
                    " error encountered during release of flow resources (agg_port_id: %d, intf_id:%d\n",
                    p_flow_inst->api_req_flow_info.data.access_int_id,
                    p_flow_inst->api_req_flow_info.data.agg_port_id);
            }
            /*check if tm sched should be removed - ref count = 1 and tm sched is auto created*/
            if(BCM_ERR_OK != (ret = flow_tm_get(&(p_flow_inst->api_req_flow_info), &p_tm_sched_inst)))
            {
            
                BCM_LOG(ERROR, log_id_flow, 
                    " could not find tm sched for agg_port_id: %d, intf_id:%d\n",
                    p_flow_inst->api_req_flow_info.data.access_int_id,
                    p_flow_inst->api_req_flow_info.data.agg_port_id);
                break;
            }
            ret = rsc_mgr_alloc_id_get_ref_count(p_flow_inst->api_req_flow_info.data.access_int_id,p_flow_inst->api_req_flow_info.data.agg_port_id, &alloc_ref_count);
            
            if(BCM_ERR_OK == ret
                && BCMBAL_TM_CREATION_MODE_AUTO == p_tm_sched_inst->req_tm_sched_info.data.creation_mode
                && 1 == alloc_ref_count)
            {
                ret = bcmbal_tm_sched_fsm_active_destroy(p_tm_sched_inst);
            }
        }

        /*if the flow is not a cpu flow (to host), should handle it sched/queue setting*/
        if(!b_flow_is_destined_to_host)
        {
            /*remove the flow from the the tm queue list*/			 
            /*find tm queue instance*/
            ret = flow_queue_validate(&p_flow_inst->api_req_flow_info, &p_tm_queue_inst); 
        
            if (ret != BCM_ERR_OK)
            {
                ret =  BCM_ERR_NOENT;
                break;
            }
        
            ret = bcmbal_tm_queue_use_set(p_tm_queue_inst, BCMOS_FALSE);
            if (ret != BCM_ERR_OK)
            {
                ret = BCM_ERR_INTERNAL;
                break;
            }
       	}
        
        /* The hardware has properly accepted the object info, so the request object becomes
        * the current state, except for the oper_status.
        */
        bcmbal_flow_object_overlay_w_src_priority(&p_flow_inst->current_flow_info,
                                                  &p_flow_inst->api_req_flow_info);
        
        BCMBAL_OBJ_IN_PROGRESS_SET(&(p_flow_inst->current_flow_info), BCMOS_TRUE);
        
    }while(0);
    
    /* Report any error found to the API immediately */
    if(BCM_ERR_OK != ret)
    {
        mgmt_msg_send_balapi_ind(ret,
                                 msg, 
                                 log_id_flow);
    }
    
    return ret;

}


/*****************************************************************************/
/**
 * @brief The Flow FSM state processing for a flow admin-down command 
 *        from the BAL Public API when the specified flow is already 
 *        admin-down.
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from the BAL Public API
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_admin_dn_ok(flow_inst *p_flow_inst, 
                                        void *msg, 
                                        flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;

    BCM_LOG(DEBUG, log_id_flow,  
            "Received an admin DOWN request from BAL API"
            " - returning OK to the API - no further function\n");

    mgmt_msg_send_balapi_ind(ret,
                             msg, 
                             log_id_flow);
    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing for a flow admin-down command 
 *        received from the BAL Public API when the specified flow FSM 
 *        is in the CONFIGURING state.
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from the BAL Public API
 * @param p_event          Pointer to a flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_admin_dn_error(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_STATE;

    BCM_LOG(DEBUG, log_id_flow,  
            "Received an admin DOWN request from BAL API"
            " - returning ERROR to the API - no further function\n");

    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing function to ignore a received message.
 *
 * @param p_flow_inst      Pointer to an flow instance
 * @param msg              Pointer to a BAL message received from the BAL utils
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_ignore_util_msg(flow_inst *p_flow_inst, 
                                           void *msg, 
                                           flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;

    BCM_LOG(DEBUG, log_id_flow,  "Ignoring message from BAL utils\n");
    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing function to process an  AUTO IND 
 *        message from one of the BAL apps.
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from one of
 *                         the BAL apps.
 * @param p_event          Pointer to a flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_null_process_util_auto_msg(flow_inst *p_flow_inst, 
                                                      void *msg, 
                                                      flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;

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

    BCM_LOG(DEBUG, log_id_flow,  "Received an AUTO IND in the NULL state\n");

    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing function to process a message from 
 *        one of the BAL apps received when the specified flow instance FSM 
 *        is in the CONFIGURING state.
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from one of
 *                         the BAL apps.
 * @param p_event          Pointer to an access terminal event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_process_util_msg(flow_inst *p_flow_inst, 
                                                        void *msg, 
                                                        flow_fsm_event *p_event)
{
    bcmos_errno ret;
    bal_util_msg_ind *ind_msg;

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

    ind_msg = (bal_util_msg_ind *)msg;

    /*
     * NOTE: AUTO_IND messages are not processed in this function,
     * so there is no need to consider them in this logic.
     */

    BCM_LOG(DEBUG, log_id_flow, 
            " Received an IND message from BAL UTIL (%s) during %s state\n",
            subsystem_str[bcmbal_sender_get(msg)],flow_state_name_get(p_flow_inst->fsm_state));

    BCM_LOG(DEBUG, log_id_flow, 
            "%s, thread %s, module %d\n", __FUNCTION__, bcmos_task_current()->name, bcmos_module_current());

    /* Handle indication */
    ret = ind_msg->status;

    /* Reflect the execution status in the object being returned in the indication
     */
    if(BCM_ERR_OK == ret)
    {
        p_flow_inst->current_flow_info.data.oper_status =
            p_flow_inst->api_req_flow_info.data.oper_status;
		
        /*
         * The flow has been successfully configured
         */
        p_flow_inst->fsm_state = FLOW_FSM_STATE_CONFIGURED;
    }
    else
    {
        /* Error */
        BCM_LOG(ERROR, log_id_flow,  "Flow %d: Failed in state %s. Error %s\n",
                p_flow_inst->api_req_flow_info.key.flow_id,
                flow_state_name_get(p_flow_inst->fsm_state),
                bcmos_strerror(ret));
    }

    BCMBAL_OBJ_IN_PROGRESS_SET(&(p_flow_inst->current_flow_info), BCMOS_FALSE);
    p_flow_inst->current_flow_info.hdr.hdr.status = ret;

    /*
     * Send the indication back to the BAL public API here
     */
    mgmt_msg_send_balapi_ind(ret,
                             &p_flow_inst->current_flow_info.hdr, 
                             log_id_flow);

    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing function to process a message from 
 *        one of the BAL apps received 
 *
 * @param p_flow_inst      Pointer to a flow instance
 * @param msg              Pointer to a BAL message received from one of
 *                         the BAL apps.
 * @param p_event          Pointer to an access terminal event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_process_util_auto_msg(flow_inst *p_flow_inst, 
                                                            void *msg, 
                                                            flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;
    bal_util_msg_auto_ind *ind_msg;
    
    
    /* Parameter checks */
    BUG_ON(NULL == p_flow_inst);
    BUG_ON(NULL == msg);
    BUG_ON(NULL == p_event);
    
    BCM_LOG(DEBUG, log_id_flow,  
            "Received an AUTO IND message in the %s state\n",
            flow_state_name_get(p_flow_inst->fsm_state));
    
    ind_msg = (bal_util_msg_auto_ind *)msg;
    
    /* Handle indication */
    ret = ind_msg->status;
    
    if(BCM_ERR_OK == ret)
    {
	    /* data reflects the new oper_status in the object being indicated */
        memcpy(&p_flow_inst->current_flow_info.data.oper_status, ind_msg->data, sizeof(bcmbal_status));
        
        /* 
        * Send the indication back to the BAL public API here
        */
        mgmt_msg_send_balapi_ind(ret,
            &p_flow_inst->current_flow_info.hdr, 
            log_id_flow);

		
        p_flow_inst->fsm_state = FLOW_FSM_STATE_CONFIGURED;
		
    }
    
    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing function to process a
 *        message from one of the BAL apps received when the specified 
 *        flow instance FSM is in the REMOVING state.
 *
 * @param p_flow_inst      Pointer to an flow instance
 * @param msg              Pointer to a BAL message received from one of
 *                         the BAL apps.
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_removing_process_util_msg(flow_inst *p_flow_inst, 
                                                     void *msg, 
                                                     flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;
    bal_util_msg_ind *ind_msg;
    bcmos_bool is_ds_flow_to_host;
    /* Parameter checks */
    BUG_ON(NULL == p_flow_inst);
    BUG_ON(NULL == msg);
    BUG_ON(NULL == p_event);

    ind_msg = (bal_util_msg_ind *)msg;

    /*
     * NOTE: AUTO_IND messages are not processed in this function,
     * so there is no need to consider them in this logic.
     */

    BCM_LOG(DEBUG, log_id_flow, 
            " Received an IND message from BAL UTIL (%s) during REMOVING state\n",
            subsystem_str[bcmbal_sender_get(msg)]);

    do{

        /* Handle indication */
        ret = ind_msg->status;
    
        /* Reflect the execution status in the object being returned in the indication
         */
        if(BCM_ERR_OK == ret)
        {
            if(BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type ||
               BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_inst->api_req_flow_info.key.flow_type)
            {
                is_ds_flow_to_host = (BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_inst->api_req_flow_info.key.flow_type &&
                                     (BCMOS_TRUE == BCMBAL_CFG_PROP_IS_SET(&p_flow_inst->api_req_flow_info, flow, action) && 
                                      (p_flow_inst->api_req_flow_info.data.action.cmds_bitmask & 
                                       BCMBAL_ACTION_CMD_ID_TRAP_TO_HOST)));
    
                if (bcm_topo_pon_get_pon_family(p_flow_inst->api_req_flow_info.data.access_int_id) == BCM_TOPO_PON_FAMILY_GPON)
                {
    
                    /* Don't attempt to release a GEM for a downstream flow that is TRAP_TO_HOST, because there
                     * is no subscriber terminal involved in this flow (there was no GEM allocated for this flow).
                     */
                    if(BCMOS_FALSE == is_ds_flow_to_host)
                    {
                        if(BCM_ERR_OK != rsc_mgr_gem_free(p_flow_inst->api_req_flow_info.data.access_int_id,
                                                          p_flow_inst->api_req_flow_info.data.svc_port_id, p_flow_inst))
                        {
                            BCM_LOG(ERROR, log_id_flow, 
                                    " error encountered during release of flow resources (svc_port_id: %d, intf_id:%d\n",
                                    p_flow_inst->api_req_flow_info.data.svc_port_id,
                                    p_flow_inst->api_req_flow_info.data.access_int_id);
                        }   
                    }               
    
    
                }
         
                /* Remove the svc_port_id record from the sub_term record for upstream flows, 
                 * or for downstream flows that are not destined to the host
                 */
                if((BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type) ||
                   ((BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_inst->api_req_flow_info.key.flow_type) &&
                    !(is_ds_flow_to_host)))
                {
                    bcmbal_sub_term_svc_port_id_list_entry_remove(p_flow_inst->p_sub_term_inst,
                                                                  p_flow_inst->current_flow_info.data.svc_port_id);
                }
    
			    /* Remove the agg_port_id from the sub_term record (only for upstream flows) */
    			if(BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type)
                {
    
                    bcmbal_sub_term_agg_port_id_list_entry_remove(p_flow_inst->p_sub_term_inst,
                                                                  p_flow_inst->current_flow_info.data.agg_port_id);
                }
            }
            
            p_flow_inst->current_flow_info.hdr.hdr.status = ret;
    
            /* This is the proper state and status for the indication about to be sent */
            p_flow_inst->current_flow_info.data.admin_state = BCMBAL_STATE_DOWN;
            p_flow_inst->current_flow_info.data.oper_status = BCMBAL_STATUS_DOWN;
            
            BCMBAL_OBJ_IN_PROGRESS_SET(&(p_flow_inst->current_flow_info), BCMOS_FALSE);
    
            /*
         * Send the success indication back to the BAL public API here
             */
            mgmt_msg_send_balapi_ind(ret,
                                     &p_flow_inst->current_flow_info.hdr, 
                                     log_id_flow);
    
            /* Return the flow to the free pool regardless of the errors encountered above */
            flow_free_by_entry(p_flow_inst);
        }
        else
        {
        /*
         * Send the failure indication back to the BAL public API here
         */
            mgmt_msg_send_balapi_ind(ret,
                                     &p_flow_inst->current_flow_info.hdr, 
                                     log_id_flow);
        }
    }while(0);
    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM state processing function to process a
 *        AUTO IND message from one of the BAL apps received when the specified 
 *        flow instance FSM is in the REMOVING state.
 *
 * @param p_flow_inst      Pointer to an flow instance
 * @param msg              Pointer to a BAL message received from one of
 *                         the BAL apps.
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
  *****************************************************************************/
static bcmos_errno flow_fsm_removing_process_util_auto_msg(flow_inst *p_flow_inst, 
                                                          void *msg, 
                                                          flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_OK;

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

    BCM_LOG(DEBUG, log_id_flow,  
            "Received an AUTO IND in the removing state"
            " - no further function\n");

    return ret;
}

/*****************************************************************************/
/**
 * @brief The Flow FSM function which is executed when an error 
 *        is encountered during FSM processing.
 *
 * @param p_flow_inst      Pointer to an flow instance
 * @param msg              Pointer to a BAL message received from one of
 *                         the BAL apps.
 * @param p_event          Pointer to an flow event structure
 * 
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_fsm_state_err(flow_inst *p_flow_inst, 
                                      void *msg, 
                                      flow_fsm_event *p_event)
{
    bcmos_errno ret = BCM_ERR_INVALID_OP;

    BCM_LOG(DEBUG, log_id_flow,  
            "Error encountered processing FLOW FSM"
            " - BAD EVENT ()\n");

    return ret;
}
static bcmos_errno flow_queue_validate(bcmbal_flow_cfg *p_flow_cfg, tm_queue_inst **p_tm_queue_inst)
{
    bcmos_errno ret = BCM_ERR_OK;
    bcmbal_interface_key intf_key;
    bcmbal_tm_sched_key tm_key;
    tm_sched_inst *p_tm_sched;
    bcmbal_tm_queue_key queue_key;
    bcmbal_tm_queue_ref queue_ref;
    bcmos_bool is_auto_set = BCMOS_TRUE;
    
    do
    {
        if (BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_cfg->key.flow_type)
        {
            intf_key.intf_type = BCMBAL_INTF_TYPE_NNI;
            tm_key.dir = BCMBAL_TM_SCHED_DIR_US;
            intf_key.intf_id = p_flow_cfg->data.network_int_id;
        }       
        else /*BCMBAL_FLOW_TYPE_DOWNSTREM or BCMBAL_FLOW_TYPE_MULTICAST */
        {
            intf_key.intf_type = BCMBAL_INTF_TYPE_PON;
            tm_key.dir = BCMBAL_TM_SCHED_DIR_DS;
            intf_key.intf_id = p_flow_cfg->data.access_int_id;
        }
        
        if (BCMBAL_CFG_PROP_IS_SET(p_flow_cfg, flow, queue))
        {
            queue_ref = p_flow_cfg->data.queue;
            is_auto_set = BCMOS_FALSE;
        }
        else
        {        
            /*look for the auto created tm sched and queue*/
            ret = bcmbal_interface_tm_get(intf_key, &tm_key.id);
            if (BCM_ERR_OK != ret)
            {
                BCM_LOG(ERROR, log_id_flow, 
                        "could not get interface instance"
                        " to set flow queue reference intf_key.intf_type = %d,"
                        " intf_key.intf_id = %d\n",
				    intf_key.intf_type, intf_key.intf_id);
                break;
            }
            p_tm_sched = tm_sched_inst_get(tm_key, TM_SCHED_FLAG_ACTIVE);
            if (NULL == p_tm_sched)
            {
                BCM_LOG(ERROR, log_id_flow, 
                        "could not get tm sched instance to "
                        "set flow queue reference intf_key.intf_type = %d,"
                        " intf_key.intf_id = %d tm_key.dir = %s\n"
				    ,intf_key.intf_type, intf_key.intf_id, TM_SCHED_DIR_TO_STR(tm_key.dir));
                ret = BCM_ERR_NOENT;
                break;
            }

            if(BCMBAL_TM_CREATION_MODE_AUTO != p_tm_sched->current_tm_sched_info.data.creation_mode)
            {
                BCM_LOG(ERROR, log_id_flow, 
                        "can not set flow queue reference if the "
                        "interface tm sched is not auto created. "
                        "intf_key.intf_type = %d, intf_key.intf_id = %d tm_key.dir = %s\n",
				    intf_key.intf_type, intf_key.intf_id, TM_SCHED_DIR_TO_STR(tm_key.dir));
                ret = BCM_ERR_PARM;
                break;
            }
            
            /*if the tm sched exist and it is auto created, queue 0 should be there as well*/
            queue_ref.sched_id = tm_key.id;
            queue_ref.queue_id = 0;
            BCM_LOG(INFO, log_id_flow, 
                    "flow will be assign to queue: node id=%d "
                    "node dir=%s queue id = %d\n",
                    tm_key.id, TM_SCHED_DIR_TO_STR(tm_key.dir), queue_ref.queue_id);     
            
            BCMBAL_CFG_PROP_SET(p_flow_cfg, flow, queue, queue_ref);
        }
        
        queue_key.id = queue_ref.queue_id;
        queue_key.sched_id = queue_ref.sched_id;
        queue_key.sched_dir = tm_key.dir;
        
        /*validate a given tm queue exist and match flow type*/
        *p_tm_queue_inst = tm_queue_inst_get(queue_key, TM_QUEUE_FLAG_ACTIVE);
        if (NULL == *p_tm_queue_inst)
        {
            BCM_LOG(ERROR, log_id_flow,
                    "could not find the queue to assign the flow to "
                    ":tm sched dir = %s tm sched id = %d queue id = %d\n",
			    TM_SCHED_DIR_TO_STR(queue_key.sched_dir), queue_key.sched_id,queue_key.id);
            ret = BCM_ERR_NOENT;
            break;
        }
        if (BCMOS_FALSE == is_auto_set)
        {
            /*should validate queue is related to the flow intf/sub_term/sub_term_uni*/
            tm_key.dir = queue_key.sched_dir;
            tm_key.id =  queue_key.sched_id;
            
            p_tm_sched = tm_sched_inst_get(tm_key, TM_SCHED_FLAG_ACTIVE);
            if (NULL == p_tm_sched)
            {
                BCM_LOG(ERROR, log_id_flow, 
                        "could not get tm sched instance to set flow queue "
                        "reference intf_key.intf_type = %d, intf_key.intf_id = %d tm_key.dir  = %s\n",
    			    intf_key.intf_type, intf_key.intf_id, TM_SCHED_DIR_TO_STR(tm_key.dir));
                ret = BCM_ERR_NOENT;
                break;
            }
            
            if(BCMBAL_CFG_PROP_IS_SET(&p_tm_sched->req_tm_sched_info,tm_sched,owner))
            {
            switch(p_tm_sched->req_tm_sched_info.data.owner.type)
            {
                case BCMBAL_TM_SCHED_OWNER_TYPE_INTERFACE:
                {
                }
                break;
                case BCMBAL_TM_SCHED_OWNER_TYPE_SUB_TERM:
                {
                    if (p_tm_sched->req_tm_sched_info.data.owner.u.sub_term.intf_id != p_flow_cfg->data.access_int_id
                        || p_tm_sched->req_tm_sched_info.data.owner.u.sub_term.sub_term_id!= p_flow_cfg->data.sub_term_id)
                    {                
                        BCM_LOG(ERROR, log_id_flow, 
                                "queue referenced by flow, attached to tm sched instance that is owned by "
                                "sub term intf_id = %d sub_term_id = %d while flow is related to sub term"
                                " intf_id = %d sub_term_id = %d therefor cannot be set as flow queue\n",
                            p_tm_sched->req_tm_sched_info.data.owner.u.sub_term.intf_id,
                            p_tm_sched->req_tm_sched_info.data.owner.u.sub_term.sub_term_id,
                            p_flow_cfg->data.access_int_id,
                            p_flow_cfg->data.sub_term_uni_idx);
                        ret = BCM_ERR_PARM;
                        break;
                    }
                }
                break;
                case BCMBAL_TM_SCHED_OWNER_TYPE_UNI:
                {            
                    if (p_tm_sched->req_tm_sched_info.data.owner.u.uni.intf_id != p_flow_cfg->data.access_int_id
                        || p_tm_sched->req_tm_sched_info.data.owner.u.uni.sub_term_id!= p_flow_cfg->data.sub_term_id
                        || p_tm_sched->req_tm_sched_info.data.owner.u.uni.idx != p_flow_cfg->data.sub_term_uni_idx)
                    {
                        BCM_LOG(ERROR, log_id_flow, 
                                "queue referenced by flow , attached to tm sched "
                                "instance that is owned by sub term uni intf_id = %d "
                                "sub_term_id = %d uni = %d while flow is related to uni sub "
                                "term intf_id = %d sub_term_id = %d uni = %d therefor cannot be "
                                "set as flow queue\n",
                            p_tm_sched->req_tm_sched_info.data.owner.u.uni.intf_id,
                            p_tm_sched->req_tm_sched_info.data.owner.u.uni.sub_term_id,
                            p_tm_sched->req_tm_sched_info.data.owner.u.uni.idx,
                            p_flow_cfg->data.access_int_id,
                            p_flow_cfg->data.sub_term_id,
                            p_flow_cfg->data.sub_term_uni_idx);
                        ret = BCM_ERR_PARM;
                    }
                }
                break;
                default:
                    BCM_LOG(ERROR, log_id_flow, 
                            "tm sched instance is owned by %d therefor "
                            "cannot be set as flow queue\n",
			            p_tm_sched->req_tm_sched_info.data.owner.type);
                    ret = BCM_ERR_PARM;
                break;        
            }
        }
        else
        {
            BCM_LOG(ERROR, log_id_flow, 
                    "tm sched instance is not set with an owner therefor "
                    "cannot be set as flow queue intf_key.intf_type = %d, "
                    "intf_key.intf_id = %d tm_key.dir = %s\n",
			    intf_key.intf_type, intf_key.intf_id, TM_SCHED_DIR_TO_STR(tm_key.dir));
            ret = BCM_ERR_PARM;
            break;
        }    
        }
    }while (0);
    return ret;

}
            

/*****************************************************************************/
/**
 * @brief A function called by the core worker thread to process an 
 *        flow object message (SET, GET, CLEAR, STATS) received 
 *        from the BAL Public API.
 *
 * @param msg_payload      Pointer to a BAL message received from the 
 *                         BAL Public API.
 *                         
 * @returns bcmos_errno 
 *****************************************************************************/
bcmos_errno process_flow_object(void *msg_payload)
{
    bcmos_errno ret = BCM_ERR_OK, rsp_ret = BCM_ERR_OK;
    bcmbal_flow_cfg *p_flow_cfg = (bcmbal_flow_cfg *)msg_payload;
    flow_inst *p_flow_inst = NULL;
    flow_fsm_event fsm_event;
    bcmbal_flow_key *p_flow_key;
    bcmbal_state admin_state_req;
    bcmos_bool b_flow_is_destined_to_host;
    bcmbal_obj_msg_type oper_type;
    bcmbal_subscriber_terminal_key sub_term_key;
    sub_term_inst *p_sub_term_inst = NULL;
    bcmos_bool is_multicast = BCMOS_FALSE;
    bcmos_bool is_unicast = BCMOS_FALSE;
    
    bcmos_bool is_us_n_to_1 = BCMOS_FALSE;
    bcmos_bool is_ds_n_to_1 = BCMOS_FALSE;
    
    tm_queue_inst *p_tm_queue_inst = NULL;    
	
    BUG_ON(NULL == msg_payload);

    BCM_LOG(DEBUG, log_id_flow,  "Processing a flow object\n");

    p_flow_key = &p_flow_cfg->key;

    oper_type = p_flow_cfg->hdr.hdr.type;

    is_multicast = (BCMBAL_FLOW_TYPE_MULTICAST == p_flow_key->flow_type);

    is_unicast   = ((BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_key->flow_type) || 
                    (BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_key->flow_type));                

    is_us_n_to_1 = ((BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_key->flow_type) && 
                    (BCMBAL_CFG_PROP_IS_SET(p_flow_cfg, flow, group_id)));

    is_ds_n_to_1 = ((BCMBAL_FLOW_TYPE_DOWNSTREAM == p_flow_key->flow_type) && 
                    (BCMBAL_CFG_PROP_IS_SET(p_flow_cfg, flow, group_id)  ));


    /*
     * A message pointer may be passed inside the event structure.
     */
    fsm_event.msg = msg_payload;

    /* SET or GET or CLEAR...? */
    switch (oper_type)
    {
        case (BCMBAL_OBJ_MSG_TYPE_SET):
        {
            bcmos_bool b_generate_event = BCMOS_FALSE;
            bcmos_bool found_new_flow = BCMOS_FALSE;

            BCM_LOG(DEBUG, log_id_flow, 
                    "Processing a flow SET REQ mgmt message\n");
            
            do
            {
                if(BCMBAL_STATUS_UP != acc_term_status_get())
                {
                    BCM_LOG(ERROR, log_id_flow,  
                            "ERROR - Access-terminal is not UP.  No further processing\n");
                    ret = BCM_ERR_STATE;
                    break;
                }

                b_flow_is_destined_to_host = ((BCMBAL_CFG_PROP_IS_SET(p_flow_cfg, flow, action) && 
                                               (p_flow_cfg->data.action.cmds_bitmask & 
                                                BCMBAL_ACTION_CMD_ID_TRAP_TO_HOST)) ? BCMOS_TRUE : BCMOS_FALSE);

                admin_state_req = p_flow_cfg->data.admin_state;

                BCM_LOG(INFO, log_id_flow,  
                        "flow %d:%s request - admin state requested is %s\n",
                        p_flow_key->flow_id,
                        CORE_FSM_FLOW_TYPE_GET_STR(p_flow_key->flow_type),
                        (BCMBAL_STATE_UP == admin_state_req) ? "UP" : "DOWN");

                /*
                 * Find or create the specified flow instance
                 */
                p_flow_inst = flow_inst_get(p_flow_key, FLOW_FLAG_ANY, &found_new_flow);
                if(NULL == p_flow_inst)
                {
                    /* This is a fatal error condition
                     */
                    BCM_LOG(ERROR, log_id_flow,  
                            "ERROR - Flow not found.  No further processing\n");
                    ret = BCM_ERR_NOMEM;
                    break;
                }

                /*
                 * Fill in the local flow info data structure
                 */
                p_flow_inst->api_req_flow_info = *p_flow_cfg;

                /* For flows that have already been configured, merge the
                 * requested flow data with the current flow data, and this results in the new request.
                 */
                if((BCMOS_FALSE == found_new_flow) && 
                   (p_flow_inst->api_req_flow_info.data.oper_status != p_flow_inst->current_flow_info.data.oper_status))
                {
                    bcmbal_flow_object_overlay_w_dst_priority(&p_flow_inst->api_req_flow_info,
                                                              &p_flow_inst->current_flow_info);                     
                } 

                BCM_LOG(INFO, log_id_flow,  
                        "flow access_int_id: %d, sub_term_id: %d\n",
                        p_flow_inst->api_req_flow_info.data.access_int_id, 
                        p_flow_inst->api_req_flow_info.data.sub_term_id);

                /* Next, find sub term instance, if is flow UP */
                if (BCMBAL_STATE_UP == admin_state_req)
                {
                    sub_term_key.intf_id = p_flow_inst->api_req_flow_info.data.access_int_id;
                    sub_term_key.sub_term_id = p_flow_inst->api_req_flow_info.data.sub_term_id;

                    p_sub_term_inst = sub_term_inst_get(&sub_term_key, SUB_TERM_FLAG_ACTIVE);

                    if (!p_sub_term_inst && !is_multicast && !is_ds_n_to_1 && !b_flow_is_destined_to_host)
                    {
                        BCM_LOG(ERROR, log_id_flow, 
                            "No active subscriber terminal with id=%u, and for flow type %s \n", 
                            sub_term_key.sub_term_id,
                            CORE_FSM_FLOW_TYPE_GET_STR(p_flow_key->flow_type));

                        ret = BCM_ERR_NOENT;
                        break;
                    }

                    /*if the flow is not a cpu flow (to host), then we should validate/use the sched/queue setting*/
                    if(!b_flow_is_destined_to_host)
                    {
                        /*find tm queue instance*/
                        ret = flow_queue_validate(&(p_flow_inst->api_req_flow_info), &p_tm_queue_inst); 
                        if (ret != BCM_ERR_OK)
                        {
                            ret = BCM_ERR_NOENT;
                            break;
                        }
                        ret = bcmbal_tm_queue_use_set(p_tm_queue_inst, BCMOS_TRUE);
                        if (ret != BCM_ERR_OK)
                        {
                            ret = BCM_ERR_PARM;
                            break;
                        }
                    }
                }

                p_flow_inst->p_sub_term_inst = p_sub_term_inst;
                                                
                /*
                 * Process the request
                 */
                if(((BCMBAL_STATE_UP == admin_state_req) || found_new_flow)
                   && (BCMBAL_STATE_UP != p_flow_inst->current_flow_info.data.admin_state))
                {
                    bcmbal_service_port_id            svc_port_id = 0;
                    flow_inst                         *p_peer_flow;
                    uint8_t                           svc_port_id_range;

                    do
                    {
                        if (bcm_topo_pon_get_pon_family(p_flow_inst->api_req_flow_info.data.access_int_id) == BCM_TOPO_PON_FAMILY_GPON)
                        {
                            /* There's no need to do anything with the MAC in downstream flows that 
                             * that are destined to the host CPU (i.e. NNI->CPU)
                             */
                            if((BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type) ||
                               !(b_flow_is_destined_to_host))
                            {

                                /*
                                 * GEM resolution does not rely on the pbits in a packet, however, 
                                 * we currently only support allocate a svc_port_id range of 1.
                                 */
                                svc_port_id_range = 1;
                        
                                /* Is there a peer flow configured already? (only upstream and downstream flows 
                                 * can have a peer flow) 
                                 * If not, look in the active list for a match.  NOTE: It might not be there.
                                 */
                                if( is_unicast && NULL == p_flow_inst->p_peer_flow_inst)
                                {
                                    /*
                                     * See if we can find a peer flow (a peer flow is a flow that has
                                     * the same flow ID as this flow does, but has the opposite direction 
                                     * in the key).
                                     */
                            
                                    /* Look for an active flow, but ignore the direction (BEWARE: we might
                                     * find ourself!!)
                                     */
                                    p_peer_flow  = flow_inst_get(p_flow_key, 
                                                                 FLOW_FLAG_ACTIVE | FLOW_FLAG_IGNORE_DIR, 
                                                                 NULL);
                            
                                    /* If the flow that we found isn't us, then link it to our flow
                                     */
                                    if(p_peer_flow->api_req_flow_info.key.flow_type != 
                                       p_flow_inst->api_req_flow_info.key.flow_type)
                                    {
                                        p_flow_inst->p_peer_flow_inst = p_peer_flow;
                                    }
                                }
                        
                                /* If a peer flow exists, copy the GEM ID from the peer flow into this flow
                                 */
                                if(NULL != p_flow_inst->p_peer_flow_inst)
                                {
                                    svc_port_id = 
                                        p_flow_inst->p_peer_flow_inst->current_flow_info.data.svc_port_id;

                                    BCM_LOG(DEBUG, log_id_flow,  
                                            "Using GEM ID from peer flow (%d) on access_id %d\n", 
                                            svc_port_id,
                                            p_flow_inst->api_req_flow_info.data.access_int_id);
                            
                                    /* even though we have gem Id, get it allocated by rsrc mgr so that it can 
                                     * manage ref counts.
                                     */
                                    ret = rsc_mgr_gem_alloc_unicast(
                                           p_flow_inst->api_req_flow_info.data.access_int_id,
                                           &svc_port_id,
                                           svc_port_id_range,
                                           p_flow_inst); /* A multicast flow cannot have a peer flow. So this means this is definitely not a multicast. */
                                    if (BCM_ERR_OK != ret)
                                    {
                                        /* 
                                         * An error has occurred trying to get mandatory data
                                         */
                                
                                        BCM_LOG(ERROR, log_id_flow,  "Failed to get base GEM from resource manager\n");
                                
                                        /*
                                         * @todo If the flow instance (that we got) is not active, then return it to the 
                                         * free pool.
                                         */
                                
                                        break;      
                                    }
                                }
                                else /* A peer flow does not exist */
                                {
                                    /* needs single GEM for flows which is not multicast and not downstream N:1 service.
                                       In another words, all upstream flows and downstream flows which is not N:1 service needs one GEM */
                                    if(!is_multicast && !is_ds_n_to_1)
                                    {
                                        if(BCMOS_TRUE == BCMBAL_CFG_PROP_IS_SET(&p_flow_inst->api_req_flow_info,
                                                                                flow,
                                                                                svc_port_id))
                                        {
                                            svc_port_id = p_flow_inst->api_req_flow_info.data.svc_port_id;


                                            BCM_LOG(DEBUG, log_id_flow,  
                                                    "Using the base GEM ID supplied by the user (%d) on access_id %d\n", 
                                                    svc_port_id,
                                                    p_flow_inst->api_req_flow_info.data.access_int_id);
                                         }
                                        else
                                        {
                                            BCM_LOG(DEBUG, log_id_flow,  
                                                    "Getting a new base GEM ID on access_id %d from resource manager\n",
                                                    p_flow_inst->api_req_flow_info.data.access_int_id);
                                            svc_port_id = 0; /* 0 is a magic number telling the resource manager that it should provide
                                                              * the initial allocation of the base GEM */
                                        }
                            
                                        ret = rsc_mgr_gem_alloc_unicast(p_flow_inst->api_req_flow_info.data.access_int_id,
                                                                &svc_port_id,
                                                                svc_port_id_range,
                                                                p_flow_inst);

                                        if (BCM_ERR_OK != ret)
                                        {
                                            /* 
                                             * An error has occurred trying to get mandatory data
                                             */
                                    
                                            BCM_LOG(ERROR, log_id_flow,  "Failed to get base GEM from resource manager\n");
                                    
                                            /*
                                             * @todo If the flow instance (that we got) is not active, then return it to the 
                                             * free pool.
                                             */
                                    
                                            break;      
                                        }
                                    }
                                    /* make sure the all members in the group assoficated with the flow has a GEM */
                                    if(BCMBAL_CFG_PROP_IS_SET(&p_flow_inst->api_req_flow_info, flow, group_id))
                                    {
                                        bcmbal_group_key group_key;
                                        bcmbal_group_owner group_owner;
                                        
                                        group_key.group_id = p_flow_inst->api_req_flow_info.data.group_id;
                                        if(is_ds_n_to_1 || is_us_n_to_1 )
                                        {
                                            group_owner = BCMBAL_GROUP_OWNER_UNICAST;
                                        }
                                        else
                                        {
                                            group_owner = BCMBAL_GROUP_OWNER_MULTICAST;
                                        }
                                        
                                        ret = group_owner_set(group_key, group_owner);
                                        
                                        if(BCM_ERR_OK != ret)
                                        {
                                            BCM_LOG(ERROR, log_id_flow,  
                                                           "error %s while updating group owner\n", bcmos_strerror(ret));
                                            break;
                                        }                                        
                                    } 
                                }
                                /* 
                                 * Set the GEM ID into the object being processed
                                 */
                                if(svc_port_id)
                                {
                                    BCMBAL_CFG_PROP_SET(&p_flow_inst->api_req_flow_info, flow, svc_port_id,svc_port_id);
                                    BCM_LOG(DEBUG, log_id_flow,  "GEM %d being used\n", svc_port_id);
                                }
                                /*
                                 * alloc ID is only required on an UPSTREAM FLOW
                                 */
                                if(BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type)
                                {
									tm_sched_inst * dummy_p_tm_sched_inst;
									ret = flow_tm_get(&p_flow_inst->api_req_flow_info, &dummy_p_tm_sched_inst);
									
									if (BCM_ERR_OK != ret)
									{
										/* An error has occurred trying to get mandatory data  */
										BCM_LOG(ERROR, log_id_flow,  "Failed to get required tm sched for agg port id\n");
										break;		
									}
									ret = rsc_mgr_alloc_id_alloc( p_flow_inst->api_req_flow_info.data.access_int_id, 
									&p_flow_inst->api_req_flow_info.data.agg_port_id, 1, p_flow_inst);
									if (BCM_ERR_OK != ret)
									{
										BCM_LOG(ERROR, log_id_flow,  "Failed to get ALLOC ID from resource manager\n");
										break;		
									}

                                    BCM_LOG(DEBUG, log_id_flow,  "ALLOC ID %d being used\n", p_flow_inst->api_req_flow_info.data.agg_port_id);
                                }

                                /*
                                 * Perform the validation check(s) that the utils require
                                 */
                                if(BCM_ERR_OK != (ret = mac_util_flow_info_validate(&p_flow_inst->api_req_flow_info)))
                                {
                                    BCM_LOG(ERROR, log_id_flow, "Failed mac validation\n");
                                    break;
                                }
                            }
                        }
                        else if (bcm_topo_pon_get_pon_family(p_flow_inst->api_req_flow_info.data.access_int_id) == BCM_TOPO_PON_FAMILY_EPON)
                        {
                            /* There's no need to do anything with the MAC in downstream flows that 
                             * that are destined to the host CPU (i.e. NNI->CPU)
                             */
                            if((BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type) ||
                               !(b_flow_is_destined_to_host))
                            {

                                sub_term_key.intf_id = p_flow_inst->api_req_flow_info.data.access_int_id;
                                sub_term_key.sub_term_id = p_flow_inst->api_req_flow_info.data.sub_term_id;

                                p_sub_term_inst = sub_term_inst_get(&sub_term_key, SUB_TERM_FLAG_ACTIVE);

                                if (!p_sub_term_inst)
                                {
                                    BCM_LOG(ERROR, log_id_flow, 
                                            "Failed to get subscriber terminal with id=%u\n", 
                                            sub_term_key.sub_term_id);
                                    ret = BCM_ERR_NOENT;
                                    break;
                                }
                                /*
                                 * set svc_port to the subtunnel id.
                                 */
                                BCMBAL_CFG_PROP_SET(&p_flow_inst->api_req_flow_info, flow, 
                                                    svc_port_id,p_sub_term_inst->current_sub_term_info.data.svc_port_id);

                                /*
                                 * Perform the validation check(s) that the utils require
                                 */
                                if(BCM_ERR_OK != (ret = mac_util_flow_info_validate(&p_flow_inst->api_req_flow_info)))
                                {
                                    BCM_LOG(ERROR, log_id_flow, "Failed mac validation\n");
                                    break;
                                }
                            }
                        }

                        /* No need to do anything in the switch for upstream flows that 
                         * that are destined to the host CPU (i.e. PON->CPU), so no switch
                         * validation is necessary.
                         */
                        if(!((BCMBAL_FLOW_TYPE_UPSTREAM == p_flow_inst->api_req_flow_info.key.flow_type) &&
                             b_flow_is_destined_to_host))
                        {
                            if(BCM_ERR_OK != (ret = sw_util_flow_info_validate(&p_flow_inst->api_req_flow_info)))
                            {
                                BCM_LOG(ERROR, log_id_flow, "Failed switch validation\n");
                                break;
                            }
                        }

                        p_flow_inst->current_flow_info.data.admin_state = BCMBAL_STATE_UP;
                    
                        /* Set the expected state of the oper_status upon success */
                        p_flow_inst->api_req_flow_info.data.oper_status = BCMBAL_STATUS_UP;
                    
                        fsm_event.event_type = FLOW_FSM_EVENT_TYPE_ADMIN_UP;
                        b_generate_event = BCMOS_TRUE;

                    } while(0);

                    if(BCM_ERR_OK != ret)
                    {
                        flow_free_by_entry(p_flow_inst);
                        break;
                    }
                }
                /*
                 * NOTE: This is not a complete implementation of the admin down processing.
                 *
                 * @todo - complete admin down processing 
                 */
                else if(((BCMBAL_STATE_DOWN == admin_state_req) || found_new_flow)
                        && (BCMBAL_STATE_DOWN != p_flow_inst->current_flow_info.data.admin_state))
                {
                    p_flow_inst->current_flow_info.data.admin_state = BCMBAL_STATE_DOWN;

                    /* Set the expected state of the oper_status upon success */
                    p_flow_inst->api_req_flow_info.data.oper_status = BCMBAL_STATUS_DOWN;

                    fsm_event.event_type = FLOW_FSM_EVENT_TYPE_ADMIN_DN;
                    b_generate_event = BCMOS_TRUE;
                }
                else
                {
                    /* @todo implement a MODIFY here */

                    break;   /* no state change detected - do nothing for now */
                }

            }while(0);

            /* 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.
             */
            rsp_ret = mgmt_msg_send_balapi_rsp(ret, msg_payload, oper_type, log_id_flow);

            if(BCM_ERR_OK != rsp_ret || BCM_ERR_OK != ret)
            {
                /* the mgmt_msg_send_balapi_rsp function above logs any errors that occur there */
                ret = (BCM_ERR_OK != rsp_ret) ? rsp_ret : ret;
                break;
            }

             /* If there was an event generated, call the state machine exec */
            if(BCMOS_TRUE == b_generate_event)
            {
                /*
                 * Run the flow FSM to process this event
                 */
				ret = flow_fsm_exec(p_flow_inst, &fsm_event);
            }
            break;
        }

        case (BCMBAL_OBJ_MSG_TYPE_GET):
        {

            BCM_LOG(DEBUG, log_id_flow,  "Processing a flow GET REQ mgmt message\n");

            do
            {
                if(BCMBAL_STATUS_UP != acc_term_status_get())
                {
                    BCM_LOG(ERROR, log_id_flow,  
                            "ERROR - Access-terminal is not UP.  No further processing\n");
                    ret = BCM_ERR_STATE;
                }
                else
                {
                    /*
                     * Find the specified flow instance
                     */
                    p_flow_inst = flow_inst_get(p_flow_key, FLOW_FLAG_ACTIVE, NULL);
                }

                if(NULL == p_flow_inst)
                {
                    if(BCM_ERR_STATE != ret)
                    {
                        /* This is not a fatal error condition
                         */
                        BCM_LOG(ERROR, log_id_flow,  "ERROR - Specified flow (%d:%s) not found\n",
                                p_flow_key->flow_id,
                                CORE_FSM_FLOW_TYPE_GET_STR(p_flow_key->flow_type));
                        ret =  BCM_ERR_NOENT;
                    }
 
                    break;
                }

                /* 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.
                 */
                p_flow_inst->current_flow_info.hdr.hdr.comm_hdr = ((bcmbal_obj *)msg_payload)->comm_hdr;
                *((bcmbal_flow_cfg *)msg_payload) = p_flow_inst->current_flow_info;

            } while (0);

            mgmt_msg_send_balapi_rsp(ret, msg_payload, oper_type, log_id_flow);

        }
        break;            

        case (BCMBAL_OBJ_MSG_TYPE_CLEAR):
        {
            BCM_LOG(DEBUG, log_id_flow,  "Processing a flow CLEAR REQ mgmt message\n");
 
            do
            {
                if(BCMBAL_STATUS_UP != acc_term_status_get())
                {
                    BCM_LOG(ERROR, log_id_flow,  
                            "ERROR - Access-terminal is not UP.  No further processing\n");
                    ret = BCM_ERR_STATE;
                    break;
                }

                /*
                 * Find the specified flow instance
                 */
                p_flow_inst = flow_inst_get(p_flow_key, FLOW_FLAG_ACTIVE, NULL);

                if(NULL == p_flow_inst)
                {
                    /* This is a fatal error condition
                     */
                    BCM_LOG(ERROR, log_id_flow,  "ERROR - Specified flow (%d:%s) not found\n",
                            p_flow_key->flow_id,
                            CORE_FSM_FLOW_TYPE_GET_STR(p_flow_key->flow_type));
                    ret = BCM_ERR_NOENT;
                    break;
                }
            } while(0);

            /* 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.
             */
            rsp_ret = mgmt_msg_send_balapi_rsp(ret, msg_payload, oper_type, log_id_flow);

            if(BCM_ERR_OK != rsp_ret || BCM_ERR_OK != ret)
            {
                /* the mgmt_msg_send_balapi_rsp function above logs any errors that occur there */
                ret = (BCM_ERR_OK != rsp_ret) ? rsp_ret : ret;
                break;
            }

            /* Merge the requested flow data with the current flow data, 
             * and this is the new request.
             */
            bcmbal_flow_object_overlay_w_dst_priority(&p_flow_inst->api_req_flow_info,
                                                      &p_flow_inst->current_flow_info);                     
            /*
             * Run the flow FSM to process this event
             */
            if(BCM_ERR_OK == ret)
            {
                fsm_event.event_type = FLOW_FSM_EVENT_TYPE_REMOVE;
            
                ret = flow_fsm_exec(p_flow_inst, &fsm_event);
            }

            break;
        }

        default:
        {
            BCM_LOG(ERROR, log_id_flow,  "Unsupported operation on flow object (%d)\n",
                    oper_type );
            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, oper_type, log_id_flow);

            break;
        }
    }

    BCM_LOG(DEBUG, log_id_flow, "%s returns : %s\n", __FUNCTION__, bcmos_strerror(ret));

    return ret;
}


/*****************************************************************************/
/**
 * @brief A function to process a flow object event received
 *        from one of the BAL apps.
 *
 * @param msg_payload  A pointer to the util message
 *                         
 * @returns bcmos_errno 
 *****************************************************************************/
bcmos_errno process_flow_util_msg(void *msg_payload)
{
    bcmos_errno ret = BCM_ERR_OK;
    flow_inst *p_flow_inst;
    flow_fsm_event fsm_event;
    bcmbal_msg_type type;
    bcmbal_flow_key key;

    type = bcmbal_type_minor_get(msg_payload);

    BCM_LOG(DEBUG, log_id_flow,  "processing a flow %s util message from %s\n",
            bcmbal_msg_t_str[type],
            subsystem_str[bcmbal_sender_get(msg_payload)]);

    /* recover the key from the message */
    key = ((bal_util_msg_ind *)msg_payload)->obj_key.flow_key;

    do
    {
        BCM_LOG(DEBUG, log_id_flow,  "Got flow key id from util message (%d)\n", key.flow_id);

        /*
         * Get the flow instance that's being referenced
         */
        if(NULL == (p_flow_inst = flow_inst_get(&key, FLOW_FLAG_ACTIVE, NULL)))
        {  
            BCM_LOG(ERROR, log_id_flow,  "invalid flow (%d, %s) found while processing a util message from %s\n",
                    key.flow_id,
                    CORE_FSM_FLOW_TYPE_GET_STR(key.flow_type),
                    subsystem_str[bcmbal_sender_get(msg_payload)]);

            ret = BCM_ERR_INTERNAL;

            break;
        }

        /*
         * Record the msg for further processing access
         */
        fsm_event.msg = msg_payload;

        if (BAL_MSG_TYPE_IND == type)
        {
            fsm_event.event_type = FLOW_FSM_EVENT_TYPE_UTIL_MSG;
        }
        else if (BAL_MSG_TYPE_AUTO_IND == type)
        {
            fsm_event.event_type = FLOW_FSM_EVENT_TYPE_UTIL_AUTO_MSG;
        }
        else
        {
            ret = BCM_ERR_NOT_SUPPORTED;
            BCM_LOG(ERROR, log_id_flow,  
                    "Unknown message type received from the UTIL"
                    " (not one of IND:AUTO_IND) (type:%d)\n",
                    type);
            break;
        }

        /*
         * Run the Flow FSM to process this event
         */
        if(BCM_ERR_OK == ret)
        {
            ret = flow_fsm_exec(p_flow_inst, &fsm_event);
        }
    }
    while(0);

    return ret;
}

/*
 * Helper functions
 */

/*****************************************************************************/
/**
 * @brief A function to retrieve a flow instance of the specified
 *        class.
 *
 * @param key            A pointer to the key of the flow being 
 *                       referenced
 * @param search_flag    A flag specifying the type of flow
 *                       instance to be retrieved
 *
 * @param is_new_flow    A returned value signifying whether a found flow was on the free list
 *                         
 * @returns flow_inst_t* A pointer to the found flow instance,
 *                       or NULL if one is not found
 *
 *****************************************************************************/
static flow_inst *flow_inst_get(bcmbal_flow_key *key, flow_flag search_flag, bcmos_bool *is_new_flow)
{
    flow_inst *current_entry = NULL;

    if(NULL != is_new_flow)
    {
        *is_new_flow = BCMOS_FALSE;
    }

    /*
     * First, check the active list if the caller has chosen to do so
     */
    if(FLOW_FLAG_ACTIVE & search_flag)
    {
        TAILQ_FOREACH(current_entry, 
                      &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, 
                      flow_inst_next)
        {
 
            if((current_entry->api_req_flow_info.key.flow_id == key->flow_id) 
               && 
               ((FLOW_FLAG_IGNORE_DIR & search_flag) ? 
                1:(current_entry->api_req_flow_info.key.flow_type == key->flow_type)))
            {              
                BCM_LOG(DEBUG, log_id_flow,  "Found active flow\n");
                /* The flow instance pointer is in current_entry */
                break;
            }
        }
    } 
        
    /*
     * Next, check the free list if the caller has chosen to do so
     */
    if((FLOW_FLAG_FREE & search_flag) && (NULL == current_entry))
    {
        /* Now check the free list */
        if(!TAILQ_EMPTY(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list))
        {
            /* Just grab the first entry */
            current_entry = TAILQ_FIRST(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list);

            /* Remove it from the free list */
            TAILQ_REMOVE(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list, current_entry, flow_inst_next);

            /* And add it to the active list */
            TAILQ_INSERT_TAIL(&FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, current_entry, flow_inst_next);

            /*
             * Initialize the fsm state and some of the fields
             */
            current_entry->fsm_state = FLOW_FSM_STATE_NULL;
            current_entry->p_peer_flow_inst = NULL;

            if(NULL != is_new_flow)
            {
                *is_new_flow = BCMOS_TRUE;
            }

            BCM_LOG(DEBUG, log_id_flow,  "Using new flow\n");

        }
    }

    if((FLOW_FLAG_ANY & search_flag) && (NULL == current_entry))
    {
        /*A flow was not found on either list */
        BCM_LOG(DEBUG, log_id_flow,  "************** ERROR: no flow found\n");
    }
    
    return current_entry;
}


/*****************************************************************************/
/**
 * @brief A function to retrieve a flow instance by access_if_id and svc_port_id
 *
 * @param access_if_id   access interface id of the flow being searched
 * @param type           flow type (direction)
 * @param svc_port_id    svc_port_id of the flow being searched
 *
 * @returns flow_inst_t* A pointer to the found flow instance,
 *                       or NULL if one is not found
 *****************************************************************************/
flow_inst *flow_get_by_svc_id(uint16_t access_if_id, bcmbal_flow_type type, bcmbal_service_port_id svc_port_id)
{
    flow_inst *current_entry = NULL;

    /*
     * First, check the active list if the caller has chosen to do so
     */
    TAILQ_FOREACH(current_entry,
        &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list,
        flow_inst_next)
    {

        if((current_entry->api_req_flow_info.data.access_int_id == access_if_id)
           &&
           (current_entry->api_req_flow_info.key.flow_type == type)
           &&
           (current_entry->api_req_flow_info.data.svc_port_id == svc_port_id))
        {
            /* The flow instance pointer is in current_entry */
            break;
        }
    }

    return current_entry;
}

/*****************************************************************************/
/**
 * @brief A function to retrieve the current flow info for the specified 
 *        flow instance.
 *
 * @param key    A flow key 
 *
 * @returns bcmbal_flow_cfg* A pointer to the current flow info for the
 *                           specified flow, or NULL if the flow is not found
 *****************************************************************************/
bcmbal_flow_cfg *flow_get_current_info_by_key(bcmbal_flow_key key)
{
    flow_inst *current_entry = NULL;

    /*
     * Check the active list 
     */
    TAILQ_FOREACH(current_entry,
        &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list,
        flow_inst_next)
    {

        if((current_entry->current_flow_info.key.flow_id == key.flow_id)
           &&
           (current_entry->current_flow_info.key.flow_type == key.flow_type))
        {
            /* The flow instance pointer is in current_entry */
            break;
        }
    }
    
    if(current_entry)
    {
        return &(current_entry->current_flow_info);
    }
    else
    { 
        return NULL;
    }
}

/*****************************************************************************/
/**
 * @brief A function to retrieve a flow instance by access_if_id and agg_port_id
 *
 * @param access_if_id   access interface id of the flow being searched
 * @param agg_port_id    svc_port_id of the flow being searched
 *
 * @returns flow_inst_t* A pointer to the found flow instance,
 *                       or NULL if one is not found
 *****************************************************************************/
flow_inst *flow_get_by_agg_id(uint16_t access_if_id, bcmbal_aggregation_port_id agg_port_id)
{
    flow_inst *current_entry = NULL;

    /*
     * First, check the active list if the caller has chosen to do so
     */
    TAILQ_FOREACH(current_entry,
        &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list,
        flow_inst_next)
    {

        if((current_entry->current_flow_info.data.access_int_id == access_if_id)
           &&
           (current_entry->current_flow_info.key.flow_type == BCMBAL_FLOW_TYPE_UPSTREAM)
           &&
           (current_entry->current_flow_info.data.agg_port_id == agg_port_id))
        {
            /* The flow instance pointer is in current_entry */
            break;
        }
    }

    return current_entry;
}


/*****************************************************************************/
/**
 * @brief A function to retrieve the first flow instance associated with
 *        a subscriber terminal
 *
 * @param access_if_id   access interface id of the flow being searched
 * @param type           flow type (direction)
 * @param sub_term_id    the subscriber terminal associated with the flow being
 *                       searched 
 * @param sub_term_uni   uni port on ONU
 *
 * @returns flow_inst_t* A pointer to the found flow instance,
 *                       or NULL if one is not found
 *****************************************************************************/
static flow_inst *flow_get_first_by_sub_term(uint16_t access_if_id, 
                                               bcmbal_flow_type type, 
                                               uint16_t sub_term_id,
                                               uint16_t sub_term_uni)
{
    flow_inst *current_entry = NULL;
    flow_inst *matched_entry = NULL;

    /*
     * First, check the active list if the caller has chosen to do so
     */
    TAILQ_FOREACH(current_entry,
        &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list,
        flow_inst_next)
    {

        if((current_entry->current_flow_info.data.access_int_id == access_if_id)
           &&
           (current_entry->current_flow_info.key.flow_type == type)
           &&
           (current_entry->current_flow_info.data.sub_term_id == sub_term_id)
           &&
           (current_entry->current_flow_info.data.sub_term_uni_idx == sub_term_uni))
        {
            /* The flow instance pointer is in current_entry */
            matched_entry = current_entry;
            break;
        }
        else if((current_entry->current_flow_info.data.access_int_id == access_if_id)
                &&
                (current_entry->current_flow_info.key.flow_type == type)
                &&
                (current_entry->current_flow_info.data.sub_term_id == sub_term_id))
        {
            if(NULL == matched_entry)
                matched_entry = current_entry; /* store the first matched entry irrespective of uni port match */
        }
    }

    return matched_entry;
}


bcmos_errno svc_port_id_for_sub_term_ds_flow_get(uint16_t access_if_id, 
                                                 uint16_t sub_term_id,
                                                 uint16_t sub_term_uni,
                                                 uint16_t *svc_port_id)
{
    flow_inst *p_flow_inst;
    bcmos_errno ret = BCM_ERR_OK;

    if(NULL == (p_flow_inst = flow_get_first_by_sub_term(access_if_id, 
                                                         BCMBAL_FLOW_TYPE_DOWNSTREAM, 
                                                         sub_term_id,
                                                         sub_term_uni)))
    {
        ret = BCM_ERR_NOENT;
    }
    else
    {
        *svc_port_id = p_flow_inst->current_flow_info.data.svc_port_id;
    }

    return ret;

}

#ifdef FREE_FLOW_BY_KEY_SUPPORTED
/*****************************************************************************/
/**
 * @brief A function to free a flow instance specified by a supplied key.
 *
 *
 * @param key A pointer to the key of the subscriber terminal instance to be freed
 *               
 *                        
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_free_by_key(bcmbal_flow_key *key)
{
    bcmos_errno ret = BCM_ERR_OK;
    flow_inst *current_entry;
    flow_inst *p_temp_entry;

    BUG_ON(NULL == key);

    /*
     * First, check the active list (an active flow can be in the adding or removing state)
     */
    TAILQ_FOREACH_SAFE(current_entry, 
                       &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, 
                       flow_inst_next,
                       p_temp_entry)
    {
        if((current_entry->api_req_flow_info.key.flow_id == key->flow_id) &&
           (current_entry->api_req_flow_info.key.flow_type == key->flow_type))
        {

            /* Remove it from the active list */
            TAILQ_REMOVE(&FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, current_entry, flow_inst_next);

            /* And add it to the free list */
            current_entry->fsm_state = FLOW_FSM_STATE_NULL;
            current_entry->p_sub_term_fsm = NULL;
            TAILQ_INSERT_TAIL(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list, current_entry, flow_inst_next);
            break;
        }
    }

    return ret;
}
#endif

/*****************************************************************************/
/**
 * @brief A function to free a flow instance specified by a the supplied 
 *        entry pointer.
 *
 * @param p_entry A pointer to the entry to be freed
 *               
 *                         
 * @returns bcmos_errno 
 *****************************************************************************/
static bcmos_errno flow_free_by_entry(flow_inst *p_entry)
{
    bcmos_errno ret = BCM_ERR_OK;
    flow_inst *current_entry;
    flow_inst *p_temp_entry;

    /*
     * First, check the active list (an active flow can be in the adding or removing state)
     */
    TAILQ_FOREACH_SAFE(current_entry, 
                       &FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, 
                       flow_inst_next,
                       p_temp_entry)
    {
        if(current_entry == p_entry)
        {
           /* Remove it from the active list */
            TAILQ_REMOVE(&FLOW_FSM_FLOW_LIST_CTX_PTR->active_flow_list, current_entry, flow_inst_next);
            break;
        }
    }
        
    /* And add it to the free list */
    p_entry->fsm_state = FLOW_FSM_STATE_NULL;
            
    /* And initialize the current object in the flow instance */
    flow_inst_entry_obj_init(p_entry);

    TAILQ_INSERT_TAIL(&FLOW_FSM_FLOW_LIST_CTX_PTR->free_flow_list, p_entry, flow_inst_next);

    return ret;
}

static bcmos_errno flow_tm_auto_create(bcmbal_flow_cfg *p_flow_info, tm_sched_inst **p_tm_sched_inst)
{
  	bcmos_errno ret = BCM_ERR_OK;
  	bcmbal_tm_sched_cfg tm_sched_default_cfg;
    bcmbal_tm_sched_owner owner;	
    bcmbal_tm_sched_key tm_sched_key;
    do
    {
        tm_sched_key.dir = BCMBAL_TM_SCHED_DIR_US;
        BCMBAL_CFG_INIT(&tm_sched_default_cfg, tm_sched, tm_sched_key);
        owner.type = BCMBAL_TM_SCHED_OWNER_TYPE_AGG_PORT;
        owner.u.agg_port.intf_id = p_flow_info->data.access_int_id;
        owner.u.agg_port.sub_term_id = p_flow_info->data.sub_term_id;
        BCMBAL_CFG_PROP_SET(&tm_sched_default_cfg, tm_sched, owner, owner);
        BCMBAL_CFG_PROP_SET(&tm_sched_default_cfg, tm_sched, creation_mode, BCMBAL_TM_CREATION_MODE_AUTO);
        if(BCMBAL_CFG_PROP_IS_SET(p_flow_info,flow,sla))
        {    
            BCMBAL_ATTRIBUTE_PROP_SET(&(tm_sched_default_cfg.data.rate),tm_shaping, sbr, p_flow_info->data.sla.min_rate);
            BCMBAL_ATTRIBUTE_PROP_SET(&(tm_sched_default_cfg.data.rate),tm_shaping, pbr, p_flow_info->data.sla.max_rate);
        }
        if (BCM_ERR_OK != (ret = bcmbal_tm_sched_auto_create(tm_sched_default_cfg, p_tm_sched_inst)))
        {
            BCM_LOG(ERROR, log_id_flow, "Could not create the auto tm sched\n");
            break;
        }
    }while(0);
    return ret;
}

static bcmos_errno flow_tm_get(bcmbal_flow_cfg *p_flow_info, tm_sched_inst **p_tm_sched_inst)
{
    bcmos_errno ret = BCM_ERR_OK;
    do
    {
        if(BCMOS_TRUE == BCMBAL_CFG_PROP_IS_SET(p_flow_info,
                                                flow,
                                                agg_port_id))
        {
            BCM_LOG(DEBUG, log_id_flow,  
                    "Using ALLOC ID supplied by the user (%d)\n",
                    p_flow_info->data.agg_port_id );
		
            *p_tm_sched_inst = tm_sched_find_agg_port_node(p_flow_info->data.access_int_id, p_flow_info->data.agg_port_id);
            if(NULL == *p_tm_sched_inst)
            {
                BCM_LOG(ERROR, log_id_flow,  
                        "agg port %d (intf id %d sub term id %d ) has no tm sched\n",
                        p_flow_info->data.agg_port_id, p_flow_info->data.access_int_id, p_flow_info->data.sub_term_id);
                ret = BCM_ERR_PARM;
                break;
            }
            if(p_flow_info->data.sub_term_id != (*p_tm_sched_inst)->req_tm_sched_info.data.owner.u.agg_port.sub_term_id)
            {
                BCM_LOG(ERROR, log_id_flow,  
                        "agg port %d (intf id %d) is already assign to sub term %d (not %d) \n",
                        p_flow_info->data.agg_port_id, p_flow_info->data.access_int_id, 
                        (*p_tm_sched_inst)->req_tm_sched_info.data.owner.u.agg_port.sub_term_id, 
                        p_flow_info->data.sub_term_id);
                ret = BCM_ERR_PARM;
	            break;
            }
    	}
        else
    	{
            BCM_LOG(DEBUG, log_id_flow,  "Getting a new ALLOC ID from resource manager\n");
            /*create a new agg port tm and allocate a new agg port id */  
            ret = flow_tm_auto_create(p_flow_info, p_tm_sched_inst);
            if(BCM_ERR_OK != ret)
            {
                BCM_LOG(ERROR, log_id_flow,  
                        "could not create the auto tm sched for agg port %d (intf id %d)\n",
                        p_flow_info->data.agg_port_id, p_flow_info->data.access_int_id);				
     			break;
            }
            BCMBAL_CFG_PROP_SET(p_flow_info, flow, agg_port_id, 
                                (*p_tm_sched_inst)->req_tm_sched_info.data.owner.u.agg_port.agg_port_id);
        }
    }while(0);
    return ret;
}
/*@}*/