diff mbox

[dpdk-dev,v2,4/5] eal: add per rx queue interrupt handling based on VFIO

Message ID 1422951511-28143-5-git-send-email-danny.zhou@intel.com (mailing list archive)
State Superseded, archived
Headers show

Commit Message

Zhou, Danny Feb. 3, 2015, 8:18 a.m. UTC
v2 change:
- Fix compilation issue for a missed header file
- Bug fix: free unreleased resources on the exception path before return
- Consolidate coding style related review comments

This patch does below:
- Create multiple VFIO eventfd for rx queues.
- Handle per rx queue interrupt.
- Eliminate unnecessary suspended DPDK polling thread wakeup mechanism
for rx interrupt by allowing polling thread epoll_wait rx queue
interrupt notification.

Signed-off-by: Danny Zhou <danny.zhou@intel.com>
Tested-by: Yong Liu <yong.liu@intel.com>
---
 lib/librte_eal/common/include/rte_eal.h            |  12 ++
 lib/librte_eal/linuxapp/eal/Makefile               |   1 +
 lib/librte_eal/linuxapp/eal/eal_interrupts.c       | 181 ++++++++++++++++-----
 lib/librte_eal/linuxapp/eal/eal_pci_vfio.c         |  11 +-
 .../linuxapp/eal/include/exec-env/rte_interrupts.h |   4 +
 5 files changed, 167 insertions(+), 42 deletions(-)

Comments

Cunming Liang Feb. 13, 2015, 3:48 a.m. UTC | #1
Hi,

> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Zhou Danny
> Sent: Tuesday, February 03, 2015 4:19 PM
> To: dev@dpdk.org
> Subject: [dpdk-dev] [PATCH v2 4/5] eal: add per rx queue interrupt handling
> based on VFIO
> 
[...]
>  #include "eal_private.h"
>  #include "eal_vfio.h"
> @@ -127,6 +128,7 @@ static pthread_t intr_thread;
>  #ifdef VFIO_PRESENT
> 
>  #define IRQ_SET_BUF_LEN  (sizeof(struct vfio_irq_set) + sizeof(int))
> +#define MSIX_IRQ_SET_BUF_LEN (sizeof(struct vfio_irq_set) + sizeof(int) *
> (VFIO_MAX_QUEUE_ID + 1))
[LCM] Does it better to add comment for '+1' which is the max other interrupts besides rxtx.
> 
>  /* enable legacy (INTx) interrupts */
>  static int
> @@ -221,7 +223,7 @@ vfio_disable_intx(struct rte_intr_handle *intr_handle) {
>  /* enable MSI-X interrupts */
[LCM] typo on comment. 'enable MSI interrupts' ?

> 
> @@ -292,8 +284,8 @@ vfio_disable_msi(struct rte_intr_handle *intr_handle) {
>  /* enable MSI-X interrupts */
>  static int
>  vfio_enable_msix(struct rte_intr_handle *intr_handle) {
> -	int len, ret;
> -	char irq_set_buf[IRQ_SET_BUF_LEN];
> +	int len, ret, max_intr;
> +	char irq_set_buf[MSIX_IRQ_SET_BUF_LEN];
>  	struct vfio_irq_set *irq_set;
>  	int *fd_ptr;
> 
> @@ -301,12 +293,19 @@ vfio_enable_msix(struct rte_intr_handle *intr_handle)
> {
> 
>  	irq_set = (struct vfio_irq_set *) irq_set_buf;
>  	irq_set->argsz = len;
> -	irq_set->count = 1;
> +	if ((!intr_handle->max_intr) ||
> +		(intr_handle->max_intr > VFIO_MAX_QUEUE_ID))
> +		max_intr = VFIO_MAX_QUEUE_ID + 1;
> +	else
> +		max_intr = intr_handle->max_intr;
> +
> +	irq_set->count = max_intr;
>  	irq_set->flags = VFIO_IRQ_SET_DATA_EVENTFD |
> VFIO_IRQ_SET_ACTION_TRIGGER;
>  	irq_set->index = VFIO_PCI_MSIX_IRQ_INDEX;
>  	irq_set->start = 0;
>  	fd_ptr = (int *) &irq_set->data;
> -	*fd_ptr = intr_handle->fd;
> +	memcpy(fd_ptr, intr_handle->queue_fd, sizeof(intr_handle-
> >queue_fd));
> +	fd_ptr[max_intr - 1] = intr_handle->fd;
> 
>  	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
> 
> @@ -316,22 +315,6 @@ vfio_enable_msix(struct rte_intr_handle *intr_handle) {
>  		return -1;
>  	}
> 
> -	/* manually trigger interrupt to enable it */
> -	memset(irq_set, 0, len);
> -	len = sizeof(struct vfio_irq_set);
> -	irq_set->argsz = len;
> -	irq_set->count = 1;
> -	irq_set->flags = VFIO_IRQ_SET_DATA_NONE |
> VFIO_IRQ_SET_ACTION_TRIGGER;
> -	irq_set->index = VFIO_PCI_MSIX_IRQ_INDEX;
> -	irq_set->start = 0;
> -
> -	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
> -
> -	if (ret) {
> -		RTE_LOG(ERR, EAL, "Error triggering MSI-X interrupts for fd %d\n",
> -						intr_handle->fd);
> -		return -1;
> -	}
>  	return 0;
>  }
> 
> @@ -339,7 +322,7 @@ vfio_enable_msix(struct rte_intr_handle *intr_handle) {
>  static int
>  vfio_disable_msix(struct rte_intr_handle *intr_handle) {
>  	struct vfio_irq_set *irq_set;
> -	char irq_set_buf[IRQ_SET_BUF_LEN];
> +	char irq_set_buf[MSIX_IRQ_SET_BUF_LEN];
>  	int len, ret;
> 
>  	len = sizeof(struct vfio_irq_set);
> @@ -824,3 +807,119 @@ rte_eal_intr_init(void)
>  	return -ret;
>  }
> 
> +static void
> +eal_intr_process_rx_interrupts(uint8_t port_id, struct epoll_event *events, int
> nfds)
> +{
> +	int n, bytes_read;
> +	union rte_intr_read_buffer buf;
> +	struct rte_intr_handle intr_handle = rte_eth_devices[port_id].pci_dev-
> >intr_handle;
[LCM] column number large than 80.
> +
> +	for (n = 0; n < nfds; n++) {
> +		/* set the length to be read for different handle type */
> +		switch (intr_handle.type) {
> +		case RTE_INTR_HANDLE_UIO:
> +			bytes_read = sizeof(buf.uio_intr_count);
> +			break;
> +		case RTE_INTR_HANDLE_ALARM:
> +			bytes_read = sizeof(buf.timerfd_num);
> +			break;
> +#ifdef VFIO_PRESENT
> +		case RTE_INTR_HANDLE_VFIO_MSIX:
> +		case RTE_INTR_HANDLE_VFIO_MSI:
> +		case RTE_INTR_HANDLE_VFIO_LEGACY:
> +			bytes_read = sizeof(buf.vfio_intr_count);
> +			break;
> +#endif
> +		default:
> +			bytes_read = 1;
> +			break;
> +		}
> +
> +		/**
> +		* read out to clear the ready-to-be-read flag
> +		* for epoll_wait.
> +		*/
> +		bytes_read = read(events[n].data.fd, &buf, bytes_read);
> +		if (bytes_read < 0)
> +			RTE_LOG(ERR, EAL, "Error reading from file "
> +				"descriptor %d: %s\n", events[n].data.fd,
> +							strerror(errno));
> +		else if (bytes_read == 0)
> +			RTE_LOG(ERR, EAL, "Read nothing from file "
> +				"descriptor %d\n", events[n].data.fd);
> +	}
> +}
> +
> +static void
> +eal_intr_handle_rx_interrupts(uint8_t port_id, int pfd, unsigned totalfds)
> +{
> +	struct epoll_event events[totalfds];
> +	int nfds = 0;
> +
> +	do {
> +		nfds = epoll_wait(pfd, events, totalfds,
> +				EAL_INTR_EPOLL_WAIT_FOREVER);
> +		/* epoll_wait fail */
> +		if (nfds < 0) {
> +			RTE_LOG(ERR, EAL,
> +				"epoll_wait returns with fail\n");
> +			return;
> +		}
> +	} while (nfds == 0);
> +
> +	/* epoll_wait has at least one fd ready to read */
> +	eal_intr_process_rx_interrupts(port_id, events, nfds);
> +}
> +
> +int
> +rte_eal_wait_rx_intr(uint8_t port_id, uint8_t queue_id)
> +{
> +	struct rte_intr_handle intr_handle = rte_eth_devices[port_id].pci_dev-
> >intr_handle;
[LCM] exceed margin.

> +	struct epoll_event ev;
> +	unsigned numfds = 0;
> +
> +	/* create epoll fd */
> +	int pfd = epoll_create(1);
> +	if (pfd < 0) {
> +		RTE_LOG(ERR, EAL, "Cannot create epoll instance\n");
> +		return -1;
> +	}
> +
> +	rte_spinlock_lock(&intr_lock);
> +
> +	ev.events = EPOLLIN | EPOLLPRI;
> +	switch (intr_handle.type) {
> +	case RTE_INTR_HANDLE_UIO:
> +		ev.data.fd = intr_handle.fd;
> +		break;
> +#ifdef VFIO_PRESENT
> +	case RTE_INTR_HANDLE_VFIO_MSIX:
> +	case RTE_INTR_HANDLE_VFIO_MSI:
> +	case RTE_INTR_HANDLE_VFIO_LEGACY:
> +		ev.data.fd = intr_handle.queue_fd[queue_id];
> +		break;
> +#endif
> +	default:
> +		rte_spinlock_unlock(&intr_lock);
> +		close(pfd);
> +		return -1;
> +	}
> +
> +	if (epoll_ctl(pfd, EPOLL_CTL_ADD, ev.data.fd, &ev) < 0) {
> +		RTE_LOG(ERR, EAL, "Error adding fd %d epoll_ctl, %s\n",
> +				intr_handle.queue_fd[queue_id],
> strerror(errno));
> +	} else
> +		numfds++;
> +
> +	rte_spinlock_unlock(&intr_lock);
> +	/* serve the interrupt */
> +	eal_intr_handle_rx_interrupts(port_id, pfd, numfds);
> +
> +	/**
> +	* when we return, we need to rebuild the
> +	* list of fds to monitor.
> +	*/
> +	close(pfd);
> +
> +	return 0;
> +}
> diff --git a/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> b/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> index 20e0977..63d0ae8 100644
> --- a/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> +++ b/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> @@ -283,7 +283,6 @@ pci_vfio_setup_interrupts(struct rte_pci_device *dev, int
> vfio_dev_fd)
> 
>  		dev->intr_handle.fd = fd;
>  		dev->intr_handle.vfio_dev_fd = vfio_dev_fd;
> -
>  		switch (i) {
>  		case VFIO_PCI_MSIX_IRQ_INDEX:
>  			internal_config.vfio_intr_mode = RTE_INTR_MODE_MSIX;
> @@ -302,6 +301,16 @@ pci_vfio_setup_interrupts(struct rte_pci_device *dev,
> int vfio_dev_fd)
>  			return -1;
>  		}
> 
> +		for (i = 0; i < VFIO_MAX_QUEUE_ID; i++) {
[LCM] I think before assigning queue_fd, it shall check VFIO_PCI_MSIX_IRQ_INDEX first.
Otherwise, no matter MSI or INTX, either one will set queue_fd as well.

> +			fd = eventfd(0, 0);
> +			if (fd < 0) {
> +				RTE_LOG(ERR, EAL, "  cannot set up eventfd, "
> +						"error %i (%s)\n", errno,
> strerror(errno));
[LCM] Cleanup the eventfd before return ?


-Steve
Zhou, Danny Feb. 17, 2015, 8:14 a.m. UTC | #2
> -----Original Message-----
> From: Liang, Cunming
> Sent: Friday, February 13, 2015 11:48 AM
> To: Zhou, Danny; dev@dpdk.org
> Subject: RE: [dpdk-dev] [PATCH v2 4/5] eal: add per rx queue interrupt handling based on VFIO
> 
> Hi,
> 
> > -----Original Message-----
> > From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Zhou Danny
> > Sent: Tuesday, February 03, 2015 4:19 PM
> > To: dev@dpdk.org
> > Subject: [dpdk-dev] [PATCH v2 4/5] eal: add per rx queue interrupt handling
> > based on VFIO
> >
> [...]
> >  #include "eal_private.h"
> >  #include "eal_vfio.h"
> > @@ -127,6 +128,7 @@ static pthread_t intr_thread;
> >  #ifdef VFIO_PRESENT
> >
> >  #define IRQ_SET_BUF_LEN  (sizeof(struct vfio_irq_set) + sizeof(int))
> > +#define MSIX_IRQ_SET_BUF_LEN (sizeof(struct vfio_irq_set) + sizeof(int) *
> > (VFIO_MAX_QUEUE_ID + 1))
> [LCM] Does it better to add comment for '+1' which is the max other interrupts besides rxtx.

Will add.

> >
> >  /* enable legacy (INTx) interrupts */
> >  static int
> > @@ -221,7 +223,7 @@ vfio_disable_intx(struct rte_intr_handle *intr_handle) {
> >  /* enable MSI-X interrupts */
> [LCM] typo on comment. 'enable MSI interrupts' ?

Bug in legacy code and will fix it in V3.

> 
> >
> > @@ -292,8 +284,8 @@ vfio_disable_msi(struct rte_intr_handle *intr_handle) {
> >  /* enable MSI-X interrupts */
> >  static int
> >  vfio_enable_msix(struct rte_intr_handle *intr_handle) {
> > -	int len, ret;
> > -	char irq_set_buf[IRQ_SET_BUF_LEN];
> > +	int len, ret, max_intr;
> > +	char irq_set_buf[MSIX_IRQ_SET_BUF_LEN];
> >  	struct vfio_irq_set *irq_set;
> >  	int *fd_ptr;
> >
> > @@ -301,12 +293,19 @@ vfio_enable_msix(struct rte_intr_handle *intr_handle)
> > {
> >
> >  	irq_set = (struct vfio_irq_set *) irq_set_buf;
> >  	irq_set->argsz = len;
> > -	irq_set->count = 1;
> > +	if ((!intr_handle->max_intr) ||
> > +		(intr_handle->max_intr > VFIO_MAX_QUEUE_ID))
> > +		max_intr = VFIO_MAX_QUEUE_ID + 1;
> > +	else
> > +		max_intr = intr_handle->max_intr;
> > +
> > +	irq_set->count = max_intr;
> >  	irq_set->flags = VFIO_IRQ_SET_DATA_EVENTFD |
> > VFIO_IRQ_SET_ACTION_TRIGGER;
> >  	irq_set->index = VFIO_PCI_MSIX_IRQ_INDEX;
> >  	irq_set->start = 0;
> >  	fd_ptr = (int *) &irq_set->data;
> > -	*fd_ptr = intr_handle->fd;
> > +	memcpy(fd_ptr, intr_handle->queue_fd, sizeof(intr_handle-
> > >queue_fd));
> > +	fd_ptr[max_intr - 1] = intr_handle->fd;
> >
> >  	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
> >
> > @@ -316,22 +315,6 @@ vfio_enable_msix(struct rte_intr_handle *intr_handle) {
> >  		return -1;
> >  	}
> >
> > -	/* manually trigger interrupt to enable it */
> > -	memset(irq_set, 0, len);
> > -	len = sizeof(struct vfio_irq_set);
> > -	irq_set->argsz = len;
> > -	irq_set->count = 1;
> > -	irq_set->flags = VFIO_IRQ_SET_DATA_NONE |
> > VFIO_IRQ_SET_ACTION_TRIGGER;
> > -	irq_set->index = VFIO_PCI_MSIX_IRQ_INDEX;
> > -	irq_set->start = 0;
> > -
> > -	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
> > -
> > -	if (ret) {
> > -		RTE_LOG(ERR, EAL, "Error triggering MSI-X interrupts for fd %d\n",
> > -						intr_handle->fd);
> > -		return -1;
> > -	}
> >  	return 0;
> >  }
> >
> > @@ -339,7 +322,7 @@ vfio_enable_msix(struct rte_intr_handle *intr_handle) {
> >  static int
> >  vfio_disable_msix(struct rte_intr_handle *intr_handle) {
> >  	struct vfio_irq_set *irq_set;
> > -	char irq_set_buf[IRQ_SET_BUF_LEN];
> > +	char irq_set_buf[MSIX_IRQ_SET_BUF_LEN];
> >  	int len, ret;
> >
> >  	len = sizeof(struct vfio_irq_set);
> > @@ -824,3 +807,119 @@ rte_eal_intr_init(void)
> >  	return -ret;
> >  }
> >
> > +static void
> > +eal_intr_process_rx_interrupts(uint8_t port_id, struct epoll_event *events, int
> > nfds)
> > +{
> > +	int n, bytes_read;
> > +	union rte_intr_read_buffer buf;
> > +	struct rte_intr_handle intr_handle = rte_eth_devices[port_id].pci_dev-
> > >intr_handle;
> [LCM] column number large than 80.

Will fix this in V3 patechset

> > +
> > +	for (n = 0; n < nfds; n++) {
> > +		/* set the length to be read for different handle type */
> > +		switch (intr_handle.type) {
> > +		case RTE_INTR_HANDLE_UIO:
> > +			bytes_read = sizeof(buf.uio_intr_count);
> > +			break;
> > +		case RTE_INTR_HANDLE_ALARM:
> > +			bytes_read = sizeof(buf.timerfd_num);
> > +			break;
> > +#ifdef VFIO_PRESENT
> > +		case RTE_INTR_HANDLE_VFIO_MSIX:
> > +		case RTE_INTR_HANDLE_VFIO_MSI:
> > +		case RTE_INTR_HANDLE_VFIO_LEGACY:
> > +			bytes_read = sizeof(buf.vfio_intr_count);
> > +			break;
> > +#endif
> > +		default:
> > +			bytes_read = 1;
> > +			break;
> > +		}
> > +
> > +		/**
> > +		* read out to clear the ready-to-be-read flag
> > +		* for epoll_wait.
> > +		*/
> > +		bytes_read = read(events[n].data.fd, &buf, bytes_read);
> > +		if (bytes_read < 0)
> > +			RTE_LOG(ERR, EAL, "Error reading from file "
> > +				"descriptor %d: %s\n", events[n].data.fd,
> > +							strerror(errno));
> > +		else if (bytes_read == 0)
> > +			RTE_LOG(ERR, EAL, "Read nothing from file "
> > +				"descriptor %d\n", events[n].data.fd);
> > +	}
> > +}
> > +
> > +static void
> > +eal_intr_handle_rx_interrupts(uint8_t port_id, int pfd, unsigned totalfds)
> > +{
> > +	struct epoll_event events[totalfds];
> > +	int nfds = 0;
> > +
> > +	do {
> > +		nfds = epoll_wait(pfd, events, totalfds,
> > +				EAL_INTR_EPOLL_WAIT_FOREVER);
> > +		/* epoll_wait fail */
> > +		if (nfds < 0) {
> > +			RTE_LOG(ERR, EAL,
> > +				"epoll_wait returns with fail\n");
> > +			return;
> > +		}
> > +	} while (nfds == 0);
> > +
> > +	/* epoll_wait has at least one fd ready to read */
> > +	eal_intr_process_rx_interrupts(port_id, events, nfds);
> > +}
> > +
> > +int
> > +rte_eal_wait_rx_intr(uint8_t port_id, uint8_t queue_id)
> > +{
> > +	struct rte_intr_handle intr_handle = rte_eth_devices[port_id].pci_dev-
> > >intr_handle;
> [LCM] exceed margin.
> 

Will fix it in V3 patchset.

> > +	struct epoll_event ev;
> > +	unsigned numfds = 0;
> > +
> > +	/* create epoll fd */
> > +	int pfd = epoll_create(1);
> > +	if (pfd < 0) {
> > +		RTE_LOG(ERR, EAL, "Cannot create epoll instance\n");
> > +		return -1;
> > +	}
> > +
> > +	rte_spinlock_lock(&intr_lock);
> > +
> > +	ev.events = EPOLLIN | EPOLLPRI;
> > +	switch (intr_handle.type) {
> > +	case RTE_INTR_HANDLE_UIO:
> > +		ev.data.fd = intr_handle.fd;
> > +		break;
> > +#ifdef VFIO_PRESENT
> > +	case RTE_INTR_HANDLE_VFIO_MSIX:
> > +	case RTE_INTR_HANDLE_VFIO_MSI:
> > +	case RTE_INTR_HANDLE_VFIO_LEGACY:
> > +		ev.data.fd = intr_handle.queue_fd[queue_id];
> > +		break;
> > +#endif
> > +	default:
> > +		rte_spinlock_unlock(&intr_lock);
> > +		close(pfd);
> > +		return -1;
> > +	}
> > +
> > +	if (epoll_ctl(pfd, EPOLL_CTL_ADD, ev.data.fd, &ev) < 0) {
> > +		RTE_LOG(ERR, EAL, "Error adding fd %d epoll_ctl, %s\n",
> > +				intr_handle.queue_fd[queue_id],
> > strerror(errno));
> > +	} else
> > +		numfds++;
> > +
> > +	rte_spinlock_unlock(&intr_lock);
> > +	/* serve the interrupt */
> > +	eal_intr_handle_rx_interrupts(port_id, pfd, numfds);
> > +
> > +	/**
> > +	* when we return, we need to rebuild the
> > +	* list of fds to monitor.
> > +	*/
> > +	close(pfd);
> > +
> > +	return 0;
> > +}
> > diff --git a/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> > b/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> > index 20e0977..63d0ae8 100644
> > --- a/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> > +++ b/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
> > @@ -283,7 +283,6 @@ pci_vfio_setup_interrupts(struct rte_pci_device *dev, int
> > vfio_dev_fd)
> >
> >  		dev->intr_handle.fd = fd;
> >  		dev->intr_handle.vfio_dev_fd = vfio_dev_fd;
> > -
> >  		switch (i) {
> >  		case VFIO_PCI_MSIX_IRQ_INDEX:
> >  			internal_config.vfio_intr_mode = RTE_INTR_MODE_MSIX;
> > @@ -302,6 +301,16 @@ pci_vfio_setup_interrupts(struct rte_pci_device *dev,
> > int vfio_dev_fd)
> >  			return -1;
> >  		}
> >
> > +		for (i = 0; i < VFIO_MAX_QUEUE_ID; i++) {
> [LCM] I think before assigning queue_fd, it shall check VFIO_PCI_MSIX_IRQ_INDEX first.
> Otherwise, no matter MSI or INTX, either one will set queue_fd as well.

Will move this to "case" block.

> 
> > +			fd = eventfd(0, 0);
> > +			if (fd < 0) {
> > +				RTE_LOG(ERR, EAL, "  cannot set up eventfd, "
> > +						"error %i (%s)\n", errno,
> > strerror(errno));
> [LCM] Cleanup the eventfd before return ?

It is not allocated successful if it returns error, so no need to clean it up. 

> 
> 
> -Steve
diff mbox

Patch

diff --git a/lib/librte_eal/common/include/rte_eal.h b/lib/librte_eal/common/include/rte_eal.h
index f4ecd2e..d81331f 100644
--- a/lib/librte_eal/common/include/rte_eal.h
+++ b/lib/librte_eal/common/include/rte_eal.h
@@ -150,6 +150,18 @@  int rte_eal_iopl_init(void);
  *   - On failure, a negative error value.
  */
 int rte_eal_init(int argc, char **argv);
+
+/**
+ * @param port_id
+ *   the port id
+ * @param queue_id
+ *   the queue id
+ * @return
+ *   - On success, return 0
+ *   - On failure, returns -1.
+ */
+int rte_eal_wait_rx_intr(uint8_t port_id, uint8_t queue_id);
+
 /**
  * Usage function typedef used by the application usage function.
  *
diff --git a/lib/librte_eal/linuxapp/eal/Makefile b/lib/librte_eal/linuxapp/eal/Makefile
index 72ecf3a..325957f 100644
--- a/lib/librte_eal/linuxapp/eal/Makefile
+++ b/lib/librte_eal/linuxapp/eal/Makefile
@@ -39,6 +39,7 @@  CFLAGS += -I$(SRCDIR)/include
 CFLAGS += -I$(RTE_SDK)/lib/librte_eal/common
 CFLAGS += -I$(RTE_SDK)/lib/librte_eal/common/include
 CFLAGS += -I$(RTE_SDK)/lib/librte_ring
+CFLAGS += -I$(RTE_SDK)/lib/librte_mbuf
 CFLAGS += -I$(RTE_SDK)/lib/librte_mempool
 CFLAGS += -I$(RTE_SDK)/lib/librte_malloc
 CFLAGS += -I$(RTE_SDK)/lib/librte_ether
diff --git a/lib/librte_eal/linuxapp/eal/eal_interrupts.c b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
index dc2668a..74da06c 100644
--- a/lib/librte_eal/linuxapp/eal/eal_interrupts.c
+++ b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
@@ -64,6 +64,7 @@ 
 #include <rte_malloc.h>
 #include <rte_errno.h>
 #include <rte_spinlock.h>
+#include <rte_ethdev.h>
 
 #include "eal_private.h"
 #include "eal_vfio.h"
@@ -127,6 +128,7 @@  static pthread_t intr_thread;
 #ifdef VFIO_PRESENT
 
 #define IRQ_SET_BUF_LEN  (sizeof(struct vfio_irq_set) + sizeof(int))
+#define MSIX_IRQ_SET_BUF_LEN (sizeof(struct vfio_irq_set) + sizeof(int) * (VFIO_MAX_QUEUE_ID + 1))
 
 /* enable legacy (INTx) interrupts */
 static int
@@ -221,7 +223,7 @@  vfio_disable_intx(struct rte_intr_handle *intr_handle) {
 /* enable MSI-X interrupts */
 static int
 vfio_enable_msi(struct rte_intr_handle *intr_handle) {
-	int len, ret;
+	int len, ret, max_intr;
 	char irq_set_buf[IRQ_SET_BUF_LEN];
 	struct vfio_irq_set *irq_set;
 	int *fd_ptr;
@@ -230,12 +232,19 @@  vfio_enable_msi(struct rte_intr_handle *intr_handle) {
 
 	irq_set = (struct vfio_irq_set *) irq_set_buf;
 	irq_set->argsz = len;
-	irq_set->count = 1;
+	if ((!intr_handle->max_intr) ||
+		(intr_handle->max_intr > VFIO_MAX_QUEUE_ID))
+		max_intr = VFIO_MAX_QUEUE_ID + 1;
+	else
+		max_intr = intr_handle->max_intr;
+
+	irq_set->count = max_intr;
 	irq_set->flags = VFIO_IRQ_SET_DATA_EVENTFD | VFIO_IRQ_SET_ACTION_TRIGGER;
 	irq_set->index = VFIO_PCI_MSI_IRQ_INDEX;
 	irq_set->start = 0;
 	fd_ptr = (int *) &irq_set->data;
-	*fd_ptr = intr_handle->fd;
+	memcpy(fd_ptr, intr_handle->queue_fd, sizeof(intr_handle->queue_fd));
+	fd_ptr[max_intr - 1] = intr_handle->fd;
 
 	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
 
@@ -244,23 +253,6 @@  vfio_enable_msi(struct rte_intr_handle *intr_handle) {
 						intr_handle->fd);
 		return -1;
 	}
-
-	/* manually trigger interrupt to enable it */
-	memset(irq_set, 0, len);
-	len = sizeof(struct vfio_irq_set);
-	irq_set->argsz = len;
-	irq_set->count = 1;
-	irq_set->flags = VFIO_IRQ_SET_DATA_NONE | VFIO_IRQ_SET_ACTION_TRIGGER;
-	irq_set->index = VFIO_PCI_MSI_IRQ_INDEX;
-	irq_set->start = 0;
-
-	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
-
-	if (ret) {
-		RTE_LOG(ERR, EAL, "Error triggering MSI interrupts for fd %d\n",
-						intr_handle->fd);
-		return -1;
-	}
 	return 0;
 }
 
@@ -292,8 +284,8 @@  vfio_disable_msi(struct rte_intr_handle *intr_handle) {
 /* enable MSI-X interrupts */
 static int
 vfio_enable_msix(struct rte_intr_handle *intr_handle) {
-	int len, ret;
-	char irq_set_buf[IRQ_SET_BUF_LEN];
+	int len, ret, max_intr;
+	char irq_set_buf[MSIX_IRQ_SET_BUF_LEN];
 	struct vfio_irq_set *irq_set;
 	int *fd_ptr;
 
@@ -301,12 +293,19 @@  vfio_enable_msix(struct rte_intr_handle *intr_handle) {
 
 	irq_set = (struct vfio_irq_set *) irq_set_buf;
 	irq_set->argsz = len;
-	irq_set->count = 1;
+	if ((!intr_handle->max_intr) ||
+		(intr_handle->max_intr > VFIO_MAX_QUEUE_ID))
+		max_intr = VFIO_MAX_QUEUE_ID + 1;
+	else
+		max_intr = intr_handle->max_intr;
+
+	irq_set->count = max_intr;
 	irq_set->flags = VFIO_IRQ_SET_DATA_EVENTFD | VFIO_IRQ_SET_ACTION_TRIGGER;
 	irq_set->index = VFIO_PCI_MSIX_IRQ_INDEX;
 	irq_set->start = 0;
 	fd_ptr = (int *) &irq_set->data;
-	*fd_ptr = intr_handle->fd;
+	memcpy(fd_ptr, intr_handle->queue_fd, sizeof(intr_handle->queue_fd));
+	fd_ptr[max_intr - 1] = intr_handle->fd;
 
 	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
 
@@ -316,22 +315,6 @@  vfio_enable_msix(struct rte_intr_handle *intr_handle) {
 		return -1;
 	}
 
-	/* manually trigger interrupt to enable it */
-	memset(irq_set, 0, len);
-	len = sizeof(struct vfio_irq_set);
-	irq_set->argsz = len;
-	irq_set->count = 1;
-	irq_set->flags = VFIO_IRQ_SET_DATA_NONE | VFIO_IRQ_SET_ACTION_TRIGGER;
-	irq_set->index = VFIO_PCI_MSIX_IRQ_INDEX;
-	irq_set->start = 0;
-
-	ret = ioctl(intr_handle->vfio_dev_fd, VFIO_DEVICE_SET_IRQS, irq_set);
-
-	if (ret) {
-		RTE_LOG(ERR, EAL, "Error triggering MSI-X interrupts for fd %d\n",
-						intr_handle->fd);
-		return -1;
-	}
 	return 0;
 }
 
@@ -339,7 +322,7 @@  vfio_enable_msix(struct rte_intr_handle *intr_handle) {
 static int
 vfio_disable_msix(struct rte_intr_handle *intr_handle) {
 	struct vfio_irq_set *irq_set;
-	char irq_set_buf[IRQ_SET_BUF_LEN];
+	char irq_set_buf[MSIX_IRQ_SET_BUF_LEN];
 	int len, ret;
 
 	len = sizeof(struct vfio_irq_set);
@@ -824,3 +807,119 @@  rte_eal_intr_init(void)
 	return -ret;
 }
 
+static void
+eal_intr_process_rx_interrupts(uint8_t port_id, struct epoll_event *events, int nfds)
+{
+	int n, bytes_read;
+	union rte_intr_read_buffer buf;
+	struct rte_intr_handle intr_handle = rte_eth_devices[port_id].pci_dev->intr_handle;
+
+	for (n = 0; n < nfds; n++) {
+		/* set the length to be read for different handle type */
+		switch (intr_handle.type) {
+		case RTE_INTR_HANDLE_UIO:
+			bytes_read = sizeof(buf.uio_intr_count);
+			break;
+		case RTE_INTR_HANDLE_ALARM:
+			bytes_read = sizeof(buf.timerfd_num);
+			break;
+#ifdef VFIO_PRESENT
+		case RTE_INTR_HANDLE_VFIO_MSIX:
+		case RTE_INTR_HANDLE_VFIO_MSI:
+		case RTE_INTR_HANDLE_VFIO_LEGACY:
+			bytes_read = sizeof(buf.vfio_intr_count);
+			break;
+#endif
+		default:
+			bytes_read = 1;
+			break;
+		}
+
+		/**
+		* read out to clear the ready-to-be-read flag
+		* for epoll_wait.
+		*/
+		bytes_read = read(events[n].data.fd, &buf, bytes_read);
+		if (bytes_read < 0)
+			RTE_LOG(ERR, EAL, "Error reading from file "
+				"descriptor %d: %s\n", events[n].data.fd,
+							strerror(errno));
+		else if (bytes_read == 0)
+			RTE_LOG(ERR, EAL, "Read nothing from file "
+				"descriptor %d\n", events[n].data.fd);
+	}
+}
+
+static void
+eal_intr_handle_rx_interrupts(uint8_t port_id, int pfd, unsigned totalfds)
+{
+	struct epoll_event events[totalfds];
+	int nfds = 0;
+
+	do {
+		nfds = epoll_wait(pfd, events, totalfds,
+				EAL_INTR_EPOLL_WAIT_FOREVER);
+		/* epoll_wait fail */
+		if (nfds < 0) {
+			RTE_LOG(ERR, EAL,
+				"epoll_wait returns with fail\n");
+			return;
+		}
+	} while (nfds == 0);
+
+	/* epoll_wait has at least one fd ready to read */
+	eal_intr_process_rx_interrupts(port_id, events, nfds);
+}
+
+int
+rte_eal_wait_rx_intr(uint8_t port_id, uint8_t queue_id)
+{
+	struct rte_intr_handle intr_handle = rte_eth_devices[port_id].pci_dev->intr_handle;
+	struct epoll_event ev;
+	unsigned numfds = 0;
+
+	/* create epoll fd */
+	int pfd = epoll_create(1);
+	if (pfd < 0) {
+		RTE_LOG(ERR, EAL, "Cannot create epoll instance\n");
+		return -1;
+	}
+
+	rte_spinlock_lock(&intr_lock);
+
+	ev.events = EPOLLIN | EPOLLPRI;
+	switch (intr_handle.type) {
+	case RTE_INTR_HANDLE_UIO:
+		ev.data.fd = intr_handle.fd;
+		break;
+#ifdef VFIO_PRESENT
+	case RTE_INTR_HANDLE_VFIO_MSIX:
+	case RTE_INTR_HANDLE_VFIO_MSI:
+	case RTE_INTR_HANDLE_VFIO_LEGACY:
+		ev.data.fd = intr_handle.queue_fd[queue_id];
+		break;
+#endif
+	default:
+		rte_spinlock_unlock(&intr_lock);
+		close(pfd);
+		return -1;
+	}
+
+	if (epoll_ctl(pfd, EPOLL_CTL_ADD, ev.data.fd, &ev) < 0) {
+		RTE_LOG(ERR, EAL, "Error adding fd %d epoll_ctl, %s\n",
+				intr_handle.queue_fd[queue_id], strerror(errno));
+	} else
+		numfds++;
+
+	rte_spinlock_unlock(&intr_lock);
+	/* serve the interrupt */
+	eal_intr_handle_rx_interrupts(port_id, pfd, numfds);
+
+	/**
+	* when we return, we need to rebuild the
+	* list of fds to monitor.
+	*/
+	close(pfd);
+
+	return 0;
+}
diff --git a/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c b/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
index 20e0977..63d0ae8 100644
--- a/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
+++ b/lib/librte_eal/linuxapp/eal/eal_pci_vfio.c
@@ -283,7 +283,6 @@  pci_vfio_setup_interrupts(struct rte_pci_device *dev, int vfio_dev_fd)
 
 		dev->intr_handle.fd = fd;
 		dev->intr_handle.vfio_dev_fd = vfio_dev_fd;
-
 		switch (i) {
 		case VFIO_PCI_MSIX_IRQ_INDEX:
 			internal_config.vfio_intr_mode = RTE_INTR_MODE_MSIX;
@@ -302,6 +301,16 @@  pci_vfio_setup_interrupts(struct rte_pci_device *dev, int vfio_dev_fd)
 			return -1;
 		}
 
+		for (i = 0; i < VFIO_MAX_QUEUE_ID; i++) {
+			fd = eventfd(0, 0);
+			if (fd < 0) {
+				RTE_LOG(ERR, EAL, "  cannot set up eventfd, "
+						"error %i (%s)\n", errno, strerror(errno));
+				return -1;
+			}
+			dev->intr_handle.queue_fd[i] = fd;
+		}
+
 		return 0;
 	}
 
diff --git a/lib/librte_eal/linuxapp/eal/include/exec-env/rte_interrupts.h b/lib/librte_eal/linuxapp/eal/include/exec-env/rte_interrupts.h
index 23eafd9..c6982cf 100644
--- a/lib/librte_eal/linuxapp/eal/include/exec-env/rte_interrupts.h
+++ b/lib/librte_eal/linuxapp/eal/include/exec-env/rte_interrupts.h
@@ -38,6 +38,8 @@ 
 #ifndef _RTE_LINUXAPP_INTERRUPTS_H_
 #define _RTE_LINUXAPP_INTERRUPTS_H_
 
+#define VFIO_MAX_QUEUE_ID 32
+
 enum rte_intr_handle_type {
 	RTE_INTR_HANDLE_UNKNOWN = 0,
 	RTE_INTR_HANDLE_UIO,      /**< uio device handle */
@@ -52,6 +54,8 @@  enum rte_intr_handle_type {
 struct rte_intr_handle {
 	int vfio_dev_fd;                 /**< VFIO device file descriptor */
 	int fd;                          /**< file descriptor */
+	int max_intr;                    /**< max interrupt requested */
+	int queue_fd[VFIO_MAX_QUEUE_ID]; /**< rx and tx queue interrupt file descriptor */
 	enum rte_intr_handle_type type;  /**< handle type */
 };