[dpdk-dev] [PATCH v1 4/7] examples/power: add host channel to power manager

[Yao, Lei A]

> -----Original Message-----
> From: dev [mailto:dev-bounces at dpdk.org] On Behalf Of David Hunt
> Sent: Thursday, August 30, 2018 6:54 PM
> To: dev at dpdk.org
> Cc: Mcnamara, John <john.mcnamara at intel.com>; Hunt, David
> <david.hunt at intel.com>
> Subject: [dpdk-dev] [PATCH v1 4/7] examples/power: add host channel to
> power manager
> 
> 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.0
> 
> Signed-off-by: David Hunt <david.hunt at intel.com>
> ---
>  examples/vm_power_manager/channel_manager.c | 108
> +++++++++++++++
>  examples/vm_power_manager/channel_manager.h |  17 ++-
>  examples/vm_power_manager/channel_monitor.c | 146
> +++++++++++++++-----
>  examples/vm_power_manager/main.c            |   2 +
>  4 files changed, 238 insertions(+), 35 deletions(-)
> 
> diff --git a/examples/vm_power_manager/channel_manager.c
> b/examples/vm_power_manager/channel_manager.c
> index 2bb8641d3..bcd106be1 100644
> --- a/examples/vm_power_manager/channel_manager.c
> +++ b/examples/vm_power_manager/channel_manager.c
> @@ -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,35 @@ setup_channel_info(struct virtual_machine_info
> **vm_info_dptr,
>  	return 0;
>  }
> 
> +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 *)0;
> +	chan_info->status = CHANNEL_MGR_CHANNEL_DISCONNECTED;
> +	chan_info->type = CHANNEL_TYPE_JSON;
> +	sprintf(chan_info->channel_path, "%sfifo.0",
> CHANNEL_MGR_SOCKET_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 +533,51 @@ 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;
> +
> +	snprintf(socket_path, sizeof(socket_path), "%sfifo.%u",
> +			CHANNEL_MGR_SOCKET_PATH, 0);
> +
> +	errno = 0;
> +	ret = mkfifo(socket_path, 0666);
> +	if ((errno != EEXIST) && (ret < 0)) {
> +		printf(" %d %d, %d\n", ret, EEXIST, errno);
> +		RTE_LOG(ERR, CHANNEL_MANAGER, "Cannot create fifo '%s'
> error: "
> +				"%s\n", socket_path, strerror(errno));
> +		return 0;
> +	}
> +
> +	errno = 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),
> +			RTE_CACHE_LINE_SIZE);
> +	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), "%sfifo.%u",
> +			CHANNEL_MGR_SOCKET_PATH, 0);
> +	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)
>  {
> diff --git a/examples/vm_power_manager/channel_manager.h
> b/examples/vm_power_manager/channel_manager.h
> index 872ec6140..c157cc22b 100644
> --- a/examples/vm_power_manager/channel_manager.h
> +++ b/examples/vm_power_manager/channel_manager.h
> @@ -37,7 +37,7 @@ struct sockaddr_un _sockaddr_un;
>  #define UNIX_PATH_MAX sizeof(_sockaddr_un.sun_path)
>  #endif
> 
> -#define MAX_VMS 4
> +#define MAX_VMS 64
>  #define MAX_VCPUS 20
> 
> 
> @@ -54,6 +54,11 @@ 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 +71,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 +232,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.
> diff --git a/examples/vm_power_manager/channel_monitor.c
> b/examples/vm_power_manager/channel_monitor.c
> index f180d74e6..0ffa1112a 100644
> --- a/examples/vm_power_manager/channel_monitor.c
> +++ b/examples/vm_power_manager/channel_monitor.c
> @@ -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,46 @@ 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);
> -
> -	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);
> -				}
> +	RTE_LOG(INFO, CHANNEL_MONITOR,
> +			"Looking for pcpu for %s\n", pol->pkt.vm_name);
> +
> +	/*
> +	 * 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 == 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 +199,13 @@ update_policy(struct channel_packet *pkt)
>  	unsigned int updated = 0;
>  	int i;
> 
> +
> +	RTE_LOG(INFO, CHANNEL_MONITOR,
> +			"Updating 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) {
I suggest add warning log here when no VM can match the policy name
which we send through the fifo.0. Otherwise, the user can't aware the 
policy won't be applied.  

> +			/* 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 +233,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 +408,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 +416,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 +425,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 = 1 << pkt->resource_id;
>  		}
>  		if (__builtin_popcountll(core_mask) == 1) {
> 
> @@ -421,12 +486,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 +521,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 +543,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 "
> diff --git a/examples/vm_power_manager/main.c
> b/examples/vm_power_manager/main.c
> index 58c5fa45c..893bf4cdd 100644
> --- a/examples/vm_power_manager/main.c
> +++ b/examples/vm_power_manager/main.c
> @@ -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);
> 
> --
> 2.17.1