NF core reassignment on termination

onvm/onvm_mgr/onvm_nf.c

@@ -87,6 +87,17 @@ onvm_nf_ready(struct onvm_nf *nf);
 inline static int
 onvm_nf_stop(struct onvm_nf *nf);
 
+/*
+ * Function to move a NF to another core.
+ *
+ * Input  : instance id of the NF that needs to be moved
+ *          new_core value of where the NF should be moved
+ * Output : an error code
+ *
+ */
+inline int
+onvm_nf_relocate_nf(uint16_t nf, uint16_t new_core);
+
 /*
  * Function that initializes an LPM object
  *
@@ -290,17 +301,19 @@ onvm_nf_stop(struct onvm_nf *nf) {
         uint16_t nf_status;
         uint16_t service_id;
         uint16_t nb_pkts, i;
-        int mapIndex;
         struct onvm_nf_msg *msg;
         struct rte_mempool *nf_info_mp;
         struct rte_mbuf *pkts[PACKET_READ_SIZE];
+        uint16_t candidate_nf_id, candidate_core;
+        int mapIndex;
 
         if (nf == NULL)
                 return 1;
 
         nf_id = nf->instance_id;
         service_id = nf->service_id;
         nf_status = nf->status;
+        candidate_core = nf->thread_info.core;
 
         /* Cleanup the allocated tag */
         if (nf->tag) {
@@ -377,6 +390,15 @@ onvm_nf_stop(struct onvm_nf *nf) {
                 }
         }
 
+        /* As this NF stopped we can reevaluate core mappings */
+        if (ONVM_NF_SHUTDOWN_CORE_REASSIGNMENT) {
+                /* As this NF stopped we can reevaluate core mappings */
+                candidate_nf_id = onvm_threading_find_nf_to_reassign_core(candidate_core, cores);
+                if (candidate_nf_id > 0) {
+                        onvm_nf_relocate_nf(candidate_nf_id, candidate_core);
+                }
+        }
+
         return 0;
 }
 
@@ -395,3 +417,23 @@ onvm_nf_init_lpm_region(struct lpm_request *req_lpm) {
                 req_lpm->status = -1;
         }
 }
+
+inline int
+onvm_nf_relocate_nf(uint16_t dest, uint16_t new_core) {
+        uint16_t *msg_data;
+
+        msg_data = rte_malloc("Change core msg data", sizeof(uint16_t), 0);
+        *msg_data = new_core;
+
+        cores[nfs[dest].thread_info.core].nf_count--;
+
+        onvm_nf_send_msg(dest, MSG_CHANGE_CORE, msg_data);
+
+        /* We probably need logic that handles if everything is successful */
+
+        /* TODO Add core number */
+        onvm_stats_gen_event_nf_info("NF Ready", &nfs[dest]);
+
+        cores[new_core].nf_count++;
+        return 0;
+}

onvm/onvm_nflib/onvm_common.h

@@ -57,6 +57,9 @@
 #include "onvm_config_common.h"
 #include "onvm_msg_common.h"
 
+#define ONVM_NF_HANDLE_TX 1                   // should be true if NFs primarily pass packets to each other
+#define ONVM_NF_SHUTDOWN_CORE_REASSIGNMENT 0  // should be true if on NF shutdown onvm_mgr tries to reallocate cores
+
 #define ONVM_MAX_CHAIN_LENGTH 4  // the maximum chain length
 #define MAX_NFS 128              // total number of concurrent NFs allowed (-1 because ID 0 is reserved)
 #define MAX_SERVICES 32          // total number of unique services allowed
@@ -398,7 +401,6 @@ struct lpm_request {
 #define NF_WAITING_FOR_LPM 13     // NF is waiting for a LPM request to be fulfilled
 
 #define NF_NO_ID -1
-#define ONVM_NF_HANDLE_TX 1  // should be true if NFs primarily pass packets to each other
 
 /*
  * Given the rx queue name template above, get the queue name

onvm/onvm_nflib/onvm_msg_common.h

@@ -53,6 +53,7 @@
 #define MSG_SCALE 5
 #define MSG_FROM_NF 6
 #define MSG_REQUEST_LPM_REGION 7
+#define MSG_CHANGE_CORE 8
 
 struct onvm_nf_msg {
         uint8_t msg_type; /* Constant saying what type of message is */

onvm/onvm_nflib/onvm_nflib.c

@@ -656,11 +656,18 @@ onvm_nflib_handle_msg(struct onvm_nf_msg *msg, struct onvm_nf_local_ctx *nf_loca
                         onvm_nflib_scale((struct onvm_nf_scale_info*)msg->msg_data);
                         break;
                 case MSG_FROM_NF:
-                        RTE_LOG(INFO, APP, "Recieved MSG from other NF");
+                        RTE_LOG(INFO, APP, "Received MSG from other NF\n");
                         if (nf_local_ctx->nf->function_table->msg_handler != NULL) {
                                 nf_local_ctx->nf->function_table->msg_handler(msg->msg_data, nf_local_ctx);
                         }
                         break;
+                case MSG_CHANGE_CORE:
+                        RTE_LOG(INFO, APP, "Received relocation message...\n");
+                        RTE_LOG(INFO, APP, "Moving NF to core %d\n", *(uint16_t *)msg->msg_data);
+                        nf_local_ctx->nf->thread_info.core = *(uint16_t *)msg->msg_data;
+                        onvm_threading_core_affinitize(nf_local_ctx->nf->thread_info.core);
+                        rte_free(msg->msg_data);
+                        break;
                 case MSG_NOOP:
                 default:
                         break;

onvm/onvm_nflib/onvm_threading.c

@@ -137,3 +137,40 @@ onvm_threading_core_affinitize(int cpu) {
 
         return rte_thread_set_affinity(&cpus);
 }
+
+int
+onvm_threading_find_nf_to_reassign_core(uint16_t candidate_core, struct core_status *cores) {
+        uint16_t candidate_nf_id, most_used_core, max_nfs_per_core;
+        int i;
+
+        candidate_nf_id = most_used_core = max_nfs_per_core = 0;
+
+        for (i = 0; i < onvm_threading_get_num_cores(); i++) {
+                if (cores[i].nf_count > max_nfs_per_core) {
+                        max_nfs_per_core = cores[i].nf_count;
+                        most_used_core = i;
+                }
+        }
+
+        /* Core reassignment isn't required if the maximum nf_count doesn't exceed 1 NF
+         * or if the stopped NF was running on one of the most used cores */
+        if (max_nfs_per_core == 1 || cores[candidate_core].nf_count >= max_nfs_per_core - 1)
+                return 0;
+
+        /* Chooses one of the NFs running on candidate core 
+         * TODO (@Deep) if we do this based of NF load this would be impressive, we should 
+         * maintain a variable in onvm_nf struct describing NF load and choose the candidate based on that */
+        for (i = 0; i < MAX_NFS; i++) {
+                if (onvm_nf_is_valid(&nfs[i]) && nfs[i].thread_info.core == most_used_core) {
+                        candidate_nf_id = nfs[i].instance_id;
+                        break;
+                }
+        }
+
+        /* Sanity check, should not happen */
+        if (candidate_nf_id == 0) {
+                rte_exit(EXIT_FAILURE, "Cannot locate NF running on core %u", most_used_core);
+        }
+
+        return candidate_nf_id;
+}

onvm/onvm_nflib/onvm_threading.h

@@ -43,6 +43,12 @@
 
 #include "onvm_common.h"
 
+/********************************Global variables*****************************/
+
+extern struct onvm_nf *nfs;
+
+/********************************Interfaces***********************************/
+
 /**
  * Gets the total number of cores.
  */
@@ -84,4 +90,18 @@ onvm_threading_get_core(uint16_t *core_value, uint8_t flags, struct core_status
 int
 onvm_threading_core_affinitize(int core);
 
+/**
+ * Based on current core usage decides if any NF should be moved to passed candidate core.
+ *
+ * @param candidate_core
+ *    An integer core value to possibly move a NF to
+ * @param cores
+ *    A pointer to the core_status map containing core information
+ *
+ * @return
+ *    An instance ID of a candidate NF that should be moved to candidate_core
+ */
+int
+onvm_threading_find_nf_to_reassign_core(uint16_t candidate_core, struct core_status *cores);
+
 #endif  // _ONVM_THREADING_H_