d/numam-dpdk
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

examples/power: add host channel to power manager

Browse Source 
This patch adds a fifo channel to the vm_power_manager app through which
we can send commands and polices. Intended for sending JSON strings.
The fifo is at /tmp/powermonitor/fifo

Signed-off-by: David Hunt <david.hunt@intel.com>
Acked-by: Anatoly Burakov <anatoly.burakov@intel.com>
This commit is contained in:
David Hunt 2018-10-17 14:05:29 +01:00 committed by Thomas Monjalon
parent 39701c19dd
commit 3618326f6c
4 changed files with 235 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
examples/vm_power_manager/channel_manager.c
View File

@ -13,6 +13,7 @@
#include <sys/queue.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/select.h>
@ -284,6 +285,38 @@ open_non_blocking_channel(struct channel_info *info)
return 0;
}
static int
open_host_channel(struct channel_info *info)
{
int flags;
info->fd = open(info->channel_path, O_RDWR | O_RSYNC);
if (info->fd == -1) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Error(%s) opening fifo for '%s'\n",
strerror(errno),
info->channel_path);
return -1;
}
/* Get current flags */
flags = fcntl(info->fd, F_GETFL, 0);
if (flags < 0) {
RTE_LOG(WARNING, CHANNEL_MANAGER, "Error(%s) fcntl get flags socket for"
"'%s'\n", strerror(errno), info->channel_path);
return 1;
}
/* Set to Non Blocking */
flags |= O_NONBLOCK;
if (fcntl(info->fd, F_SETFL, flags) < 0) {
RTE_LOG(WARNING, CHANNEL_MANAGER,
"Error(%s) setting non-blocking "
"socket for '%s'\n",
strerror(errno), info->channel_path);
return -1;
}
return 0;
}
static int
setup_channel_info(struct virtual_machine_info **vm_info_dptr,
struct channel_info **chan_info_dptr, unsigned channel_num)
@ -294,6 +327,7 @@ setup_channel_info(struct virtual_machine_info **vm_info_dptr,
chan_info->channel_num = channel_num;
chan_info->priv_info = (void *)vm_info;
chan_info->status = CHANNEL_MGR_CHANNEL_DISCONNECTED;
chan_info->type = CHANNEL_TYPE_BINARY;
if (open_non_blocking_channel(chan_info) < 0) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Could not open channel: "
"'%s' for VM '%s'\n",
@ -316,6 +350,42 @@ setup_channel_info(struct virtual_machine_info **vm_info_dptr,
return 0;
}
static void
fifo_path(char *dst, unsigned int len)
{
snprintf(dst, len, "%sfifo", CHANNEL_MGR_SOCKET_PATH);
}
static int
setup_host_channel_info(struct channel_info **chan_info_dptr,
unsigned int channel_num)
{
struct channel_info *chan_info = *chan_info_dptr;
chan_info->channel_num = channel_num;
chan_info->priv_info = (void *)NULL;
chan_info->status = CHANNEL_MGR_CHANNEL_DISCONNECTED;
chan_info->type = CHANNEL_TYPE_JSON;
fifo_path(chan_info->channel_path, sizeof(chan_info->channel_path));
if (open_host_channel(chan_info) < 0) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Could not open host channel: "
"'%s'\n",
chan_info->channel_path);
return -1;
}
if (add_channel_to_monitor(&chan_info) < 0) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Could add channel: "
"'%s' to epoll ctl\n",
chan_info->channel_path);
return -1;
}
chan_info->status = CHANNEL_MGR_CHANNEL_CONNECTED;
return 0;
}
int
add_all_channels(const char *vm_name)
{
@ -470,6 +540,45 @@ add_channels(const char *vm_name, unsigned *channel_list,
return num_channels_enabled;
}
int
add_host_channel(void)
{
struct channel_info *chan_info;
char socket_path[PATH_MAX];
int num_channels_enabled = 0;
int ret;
fifo_path(socket_path, sizeof(socket_path));
ret = mkfifo(socket_path, 0660);
if ((errno != EEXIST) && (ret < 0)) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Cannot create fifo '%s' error: "
"%s\n", socket_path, strerror(errno));
return 0;
}
if (access(socket_path, F_OK) < 0) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Channel path '%s' error: "
"%s\n", socket_path, strerror(errno));
return 0;
}
chan_info = rte_malloc(NULL, sizeof(*chan_info), 0);
if (chan_info == NULL) {
RTE_LOG(ERR, CHANNEL_MANAGER, "Error allocating memory for "
"channel '%s'\n", socket_path);
return 0;
}
snprintf(chan_info->channel_path,
sizeof(chan_info->channel_path), "%s", socket_path);
if (setup_host_channel_info(&chan_info, 0) < 0) {
rte_free(chan_info);
return 0;
}
num_channels_enabled++;
return num_channels_enabled;
}
int
remove_channel(struct channel_info **chan_info_dptr)
{

17
examples/vm_power_manager/channel_manager.h
View File

@ -54,6 +54,13 @@ enum channel_status { CHANNEL_MGR_CHANNEL_DISCONNECTED = 0,
CHANNEL_MGR_CHANNEL_DISABLED,
CHANNEL_MGR_CHANNEL_PROCESSING};
/* Communication Channel Type */
enum channel_type {
CHANNEL_TYPE_BINARY = 0,
CHANNEL_TYPE_INI,
CHANNEL_TYPE_JSON
};
/* VM libvirt(qemu/KVM) connection status */
enum vm_status { CHANNEL_MGR_VM_INACTIVE = 0, CHANNEL_MGR_VM_ACTIVE};
@ -66,6 +73,7 @@ struct channel_info {
volatile uint32_t status; /**< Connection status(enum channel_status) */
int fd; /**< AF_UNIX socket fd */
unsigned channel_num; /**< CHANNEL_MGR_SOCKET_PATH/<vm_name>.channel_num */
enum channel_type type; /**< Binary, ini, json, etc. */
void *priv_info; /**< Pointer to private info, do not modify */
};
@ -226,6 +234,15 @@ int add_all_channels(const char *vm_name);
int add_channels(const char *vm_name, unsigned *channel_list,
unsigned num_channels);
/**
* Set up a fifo by which host applications can send command an policies
* through a fifo to the vm_power_manager
*
* @return
* - 0 for success
*/
int add_host_channel(void);
/**
* Remove a channel definition from the channel manager. This must only be
* called from the channel monitor thread.

140
examples/vm_power_manager/channel_monitor.c
View File

@ -85,6 +85,33 @@ core_share_status(int pNo)
}
}
static int
pcpu_monitor(struct policy *pol, struct core_info *ci, int pcpu, int count)
{
int ret = 0;
if (pol->pkt.policy_to_use == BRANCH_RATIO) {
ci->cd[pcpu].oob_enabled = 1;
ret = add_core_to_monitor(pcpu);
if (ret == 0)
RTE_LOG(INFO, CHANNEL_MONITOR,
"Monitoring pcpu %d OOB for %s\n",
pcpu, pol->pkt.vm_name);
else
RTE_LOG(ERR, CHANNEL_MONITOR,
"Error monitoring pcpu %d OOB for %s\n",
pcpu, pol->pkt.vm_name);
} else {
pol->core_share[count].pcpu = pcpu;
RTE_LOG(INFO, CHANNEL_MONITOR,
"Monitoring pcpu %d for %s\n",
pcpu, pol->pkt.vm_name);
}
return ret;
}
static void
get_pcpu_to_control(struct policy *pol)
{
@ -94,34 +121,42 @@ get_pcpu_to_control(struct policy *pol)
int pcpu, count;
uint64_t mask_u64b;
struct core_info *ci;
int ret;
ci = get_core_info();
RTE_LOG(INFO, CHANNEL_MONITOR, "Looking for pcpu for %s\n",
pol->pkt.vm_name);
get_info_vm(pol->pkt.vm_name, &info);
RTE_LOG(INFO, CHANNEL_MONITOR,
"Looking for pcpu for %s\n", pol->pkt.vm_name);
for (count = 0; count < pol->pkt.num_vcpu; count++) {
mask_u64b = info.pcpu_mask[pol->pkt.vcpu_to_control[count]];
for (pcpu = 0; mask_u64b; mask_u64b &= ~(1ULL << pcpu++)) {
if ((mask_u64b >> pcpu) & 1) {
if (pol->pkt.policy_to_use == BRANCH_RATIO) {
ci->cd[pcpu].oob_enabled = 1;
ret = add_core_to_monitor(pcpu);
if (ret == 0)
printf("Monitoring pcpu %d via Branch Ratio\n",
pcpu);
else
printf("Failed to start OOB Monitoring pcpu %d\n",
pcpu);
} else {
pol->core_share[count].pcpu = pcpu;
printf("Monitoring pcpu %d\n", pcpu);
}
/*
* So now that we're handling virtual and physical cores, we need to
* differenciate between them when adding them to the branch monitor.
* Virtual cores need to be converted to physical cores.
*/
if (pol->pkt.core_type == CORE_TYPE_VIRTUAL) {
/*
* If the cores in the policy are virtual, we need to map them
* to physical core. We look up the vm info and use that for
* the mapping.
*/
get_info_vm(pol->pkt.vm_name, &info);
for (count = 0; count < pol->pkt.num_vcpu; count++) {
mask_u64b =
info.pcpu_mask[pol->pkt.vcpu_to_control[count]];
for (pcpu = 0; mask_u64b;
mask_u64b &= ~(1ULL << pcpu++)) {
if ((mask_u64b >> pcpu) & 1)
pcpu_monitor(pol, ci, pcpu, count);
}
}
} else {
/*
* If the cores in the policy are physical, we just use
* those core id's directly.
*/
for (count = 0; count < pol->pkt.num_vcpu; count++) {
pcpu = pol->pkt.vcpu_to_control[count];
pcpu_monitor(pol, ci, pcpu, count);
}
}
}
@ -160,8 +195,13 @@ update_policy(struct channel_packet *pkt)
unsigned int updated = 0;
int i;
RTE_LOG(INFO, CHANNEL_MONITOR,
"Applying policy for %s\n", pkt->vm_name);
for (i = 0; i < MAX_VMS; i++) {
if (strcmp(policies[i].pkt.vm_name, pkt->vm_name) == 0) {
/* Copy the contents of *pkt into the policy.pkt */
policies[i].pkt = *pkt;
get_pcpu_to_control(&policies[i]);
if (get_pfid(&policies[i]) == -1) {
@ -189,6 +229,24 @@ update_policy(struct channel_packet *pkt)
return 0;
}
static int
remove_policy(struct channel_packet *pkt __rte_unused)
{
int i;
/*
* Disabling the policy is simply a case of setting
* enabled to 0
*/
for (i = 0; i < MAX_VMS; i++) {
if (strcmp(policies[i].pkt.vm_name, pkt->vm_name) == 0) {
policies[i].enabled = 0;
return 0;
}
}
return -1;
}
static uint64_t
get_pkt_diff(struct policy *pol)
{
@ -346,7 +404,6 @@ apply_policy(struct policy *pol)
apply_workload_profile(pol);
}
static int
process_request(struct channel_packet *pkt, struct channel_info *chan_info)
{
@ -355,6 +412,8 @@ process_request(struct channel_packet *pkt, struct channel_info *chan_info)
if (chan_info == NULL)
return -1;
RTE_LOG(INFO, CHANNEL_MONITOR, "Processing Request %s\n", pkt->vm_name);
if (rte_atomic32_cmpset(&(chan_info->status), CHANNEL_MGR_CHANNEL_CONNECTED,
CHANNEL_MGR_CHANNEL_PROCESSING) == 0)
return -1;
@ -362,10 +421,12 @@ process_request(struct channel_packet *pkt, struct channel_info *chan_info)
if (pkt->command == CPU_POWER) {
core_mask = get_pcpus_mask(chan_info, pkt->resource_id);
if (core_mask == 0) {
RTE_LOG(ERR, CHANNEL_MONITOR, "Error get physical CPU mask for "
"channel '%s' using vCPU(%u)\n", chan_info->channel_path,
(unsigned)pkt->unit);
return -1;
/*
* Core mask will be 0 in the case where
* hypervisor is not available so we're working in
* the host, so use the core as the mask.
*/
core_mask = 1ULL << pkt->resource_id;
}
if (__builtin_popcountll(core_mask) == 1) {
@ -421,12 +482,20 @@ process_request(struct channel_packet *pkt, struct channel_info *chan_info)
}
if (pkt->command == PKT_POLICY) {
RTE_LOG(INFO, CHANNEL_MONITOR, "\nProcessing Policy request from Guest\n");
RTE_LOG(INFO, CHANNEL_MONITOR,
"\nProcessing Policy request\n");
update_policy(pkt);
policy_is_set = 1;
}
/* Return is not checked as channel status may have been set to DISABLED
if (pkt->command == PKT_POLICY_REMOVE) {
RTE_LOG(INFO, CHANNEL_MONITOR,
"Removing policy %s\n", pkt->vm_name);
remove_policy(pkt);
}
/*
* Return is not checked as channel status may have been set to DISABLED
* from management thread
*/
rte_atomic32_cmpset(&(chan_info->status), CHANNEL_MGR_CHANNEL_PROCESSING,
@ -448,13 +517,16 @@ add_channel_to_monitor(struct channel_info **chan_info)
"to epoll\n", info->channel_path);
return -1;
}
RTE_LOG(ERR, CHANNEL_MONITOR, "Added channel '%s' "
"to monitor\n", info->channel_path);
return 0;
}
int
remove_channel_from_monitor(struct channel_info *chan_info)
{
if (epoll_ctl(global_event_fd, EPOLL_CTL_DEL, chan_info->fd, NULL) < 0) {
if (epoll_ctl(global_event_fd, EPOLL_CTL_DEL,
chan_info->fd, NULL) < 0) {
RTE_LOG(ERR, CHANNEL_MONITOR, "Unable to remove channel '%s' "
"from epoll\n", chan_info->channel_path);
return -1;
@ -467,11 +539,13 @@ channel_monitor_init(void)
{
global_event_fd = epoll_create1(0);
if (global_event_fd == 0) {
RTE_LOG(ERR, CHANNEL_MONITOR, "Error creating epoll context with "
"error %s\n", strerror(errno));
RTE_LOG(ERR, CHANNEL_MONITOR,
"Error creating epoll context with error %s\n",
strerror(errno));
return -1;
}
global_events_list = rte_malloc("epoll_events", sizeof(*global_events_list)
global_events_list = rte_malloc("epoll_events",
sizeof(*global_events_list)
* MAX_EVENTS, RTE_CACHE_LINE_SIZE);
if (global_events_list == NULL) {
RTE_LOG(ERR, CHANNEL_MONITOR, "Unable to rte_malloc for "

2
examples/vm_power_manager/main.c
View File

@ -421,6 +421,8 @@ main(int argc, char **argv)
return -1;
}
add_host_channel();
printf("Running core monitor on lcore id %d\n", lcore_id);
rte_eal_remote_launch(run_core_monitor, NULL, lcore_id);