Commit 90ef3021 authored by Xiaodong Liu's avatar Xiaodong Liu Committed by Jim Harris
Browse files

nbd: refactor nbd io workflow 



Apply a state for nbd io instead of using flag
bits like payload_in_progress and resp_in_progress.

Change-Id: I331567f7c4d7a6a886a145390a8b1e4db0f6040c
Signed-off-by: default avatarXiaodong Liu <xiaodong.liu@intel.com>
Reviewed-on: https://review.gerrithub.io/392175


Tested-by: default avatarSPDK Automated Test System <sys_sgsw@intel.com>
Reviewed-by: default avatarChangpeng Liu <changpeng.liu@intel.com>
Reviewed-by: default avatarJim Harris <james.r.harris@intel.com>
parent b5f4a259

lib/nbd/nbd.c

+162 −124
Original line number Diff line number Diff line
@@ -47,21 +47,31 @@ 


#include "spdk_internal/log.h"



enum nbd_io_state_t {

	/* Receiving or ready to receive nbd request header */

	NBD_IO_RECV_REQ = 0,

	/* Receiving write payload */

	NBD_IO_RECV_PAYLOAD,

	/* processing or ready to process of SPDK bdev module */

	NBD_IO_PROCESS,

	/* Transmitting or ready to transmit nbd response header */

	NBD_IO_XMIT_RESP,

	/* Transmitting read payload */

	NBD_IO_XMIT_PAYLOAD,

};



struct nbd_io {

	enum spdk_bdev_io_type	type;

	struct spdk_nbd_disk	*nbd;

	enum nbd_io_state_t	state;



	int			ref;

	void			*payload;



	/* NOTE: for TRIM, this represents number of bytes to trim. */

	uint32_t		payload_size;



	bool			payload_in_progress;



	struct nbd_request	req;

	bool			req_in_progress;



	struct nbd_reply	resp;

	bool			resp_in_progress;



	/*

	 * Tracks current progress on reading/writing a request,
@@ -184,28 +194,6 @@ spdk_nbd_disk_get_bdev_name(struct spdk_nbd_disk *nbd) 
	return spdk_bdev_get_name(nbd->bdev);

}



static bool

is_read(enum spdk_bdev_io_type io_type)

{

	if (io_type == SPDK_BDEV_IO_TYPE_READ) {

		return true;

	} else {

		return false;

	}

}



static bool

is_write(enum spdk_bdev_io_type io_type)

{

	switch (io_type) {

	case SPDK_BDEV_IO_TYPE_WRITE:

	case SPDK_BDEV_IO_TYPE_UNMAP:

		return true;

	default:

		return false;

	}

}



void

nbd_disconnect(struct spdk_nbd_disk *nbd)

{
@@ -308,70 +296,119 @@ static void 
nbd_io_done(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)

{

	struct nbd_io	*io = cb_arg;

	struct spdk_nbd_disk *nbd = io->nbd;



	if (success) {

		io->resp.error = 0;

	} else {

		to_be32(&io->resp.error, EIO);

	}

	io->resp_in_progress = true;



	memcpy(&io->resp.handle, &io->req.handle, sizeof(io->resp.handle));

	io->state = NBD_IO_XMIT_RESP;



	if (bdev_io != NULL) {

		spdk_bdev_free_io(bdev_io);

	}



	io->ref--;

	if (io->ref == 0) {

		_nbd_stop(SPDK_CONTAINEROF(io, struct spdk_nbd_disk, io));

		_nbd_stop(nbd);

	}

}



static void

nbd_submit_bdev_io(struct spdk_bdev *bdev, struct spdk_bdev_desc *desc,

		   struct spdk_io_channel *ch, struct nbd_io *io)

static int

nbd_submit_bdev_io(struct spdk_nbd_disk *nbd, struct nbd_io *io)

{

	int rc;

	struct spdk_bdev_desc *desc = nbd->bdev_desc;

	struct spdk_io_channel *ch = nbd->ch;

	int rc = 0;



	io->ref++;



	switch (io->type) {

	case SPDK_BDEV_IO_TYPE_READ:

	switch (from_be32(&io->req.type)) {

	case NBD_CMD_READ:

		rc = spdk_bdev_read(desc, ch, io->payload, from_be64(&io->req.from),

				    io->payload_size, nbd_io_done, io);

		break;

	case SPDK_BDEV_IO_TYPE_WRITE:

	case NBD_CMD_WRITE:

		rc = spdk_bdev_write(desc, ch, io->payload, from_be64(&io->req.from),

				     io->payload_size, nbd_io_done, io);

		break;

	case SPDK_BDEV_IO_TYPE_UNMAP:

#ifdef NBD_FLAG_SEND_FLUSH

	case NBD_CMD_FLUSH:

		rc = spdk_bdev_flush(desc, ch, 0,

				     spdk_bdev_get_num_blocks(nbd->bdev) * spdk_bdev_get_block_size(nbd->bdev),

				     nbd_io_done, io);

		break;

#endif

#ifdef NBD_FLAG_SEND_TRIM

	case NBD_CMD_TRIM:

		rc = spdk_bdev_unmap(desc, ch, from_be64(&io->req.from),

				     io->payload_size, nbd_io_done, io);

		break;

	case SPDK_BDEV_IO_TYPE_FLUSH:

		rc = spdk_bdev_flush(desc, ch, 0, spdk_bdev_get_num_blocks(bdev) * spdk_bdev_get_block_size(bdev),

				     nbd_io_done, io);

		break;

#endif

	case NBD_CMD_DISC:

		io->ref--;

		return -ECONNRESET;

	default:

		rc = -1;

		break;

	}



	if (rc == -1) {

	if (rc < 0) {

		nbd_io_done(NULL, false, io);

	}



	return 0;

}



static int

process_request(struct spdk_nbd_disk *nbd)

spdk_nbd_io_exec(struct nbd_io *io)

{

	struct nbd_io	*io = &nbd->io;

	struct spdk_nbd_disk *nbd = io->nbd;

	int ret = 0;



	memcpy(&io->resp.handle, &io->req.handle, sizeof(io->resp.handle));

	io->resp.error = 0;

	if (io->state == NBD_IO_PROCESS) {

		ret = nbd_submit_bdev_io(nbd, io);

	}



	return ret;

}



static int

spdk_nbd_io_recv(struct nbd_io *io)

{

	struct spdk_nbd_disk *nbd = io->nbd;

	int ret = 0;



	if (io->state == NBD_IO_RECV_REQ) {

		ret = read_from_socket(nbd->spdk_sp_fd, (char *)&io->req + io->offset,

				       sizeof(io->req) - io->offset);

		if (ret <= 0) {

			return ret;

		}



		io->offset += ret;



		/* request is fully received */

		if (io->offset == sizeof(io->req)) {

			io->offset = 0;



			/* req magic check */

			if (from_be32(&io->req.magic) != NBD_REQUEST_MAGIC) {

				SPDK_ERRLOG("invalid request magic\n");

				return -EINVAL;

			}



			/* io except read/write should ignore payload */

			if (from_be32(&io->req.type) == NBD_CMD_WRITE ||

			    from_be32(&io->req.type) == NBD_CMD_READ) {

				io->payload_size = from_be32(&io->req.len);

	spdk_dma_free(io->payload);

			} else {

				io->payload_size = 0;

			}



			/* io payload allocate */

			if (io->payload_size) {

				io->payload = spdk_dma_malloc(io->payload_size, nbd->buf_align, NULL);

				if (io->payload == NULL) {
@@ -382,113 +419,113 @@ process_request(struct spdk_nbd_disk *nbd) 
				io->payload = NULL;

			}



	if (from_be32(&io->req.magic) != NBD_REQUEST_MAGIC) {

		SPDK_ERRLOG("invalid request magic\n");

		return -EINVAL;

			/* next io step */

			if (from_be32(&io->req.type) == NBD_CMD_WRITE) {

				io->state = NBD_IO_RECV_PAYLOAD;

			} else {

				io->state = NBD_IO_PROCESS;

				ret = spdk_nbd_io_exec(io);

			}



	switch (from_be32(&io->req.type)) {

	case NBD_CMD_READ:

		io->type = SPDK_BDEV_IO_TYPE_READ;

		nbd_submit_bdev_io(nbd->bdev, nbd->bdev_desc, nbd->ch, io);

		break;

	case NBD_CMD_WRITE:

		io->type = SPDK_BDEV_IO_TYPE_WRITE;

		io->payload_in_progress = true;

		break;

	case NBD_CMD_DISC:

		return -ECONNRESET;

#ifdef NBD_FLAG_SEND_FLUSH

	case NBD_CMD_FLUSH:

		io->type = SPDK_BDEV_IO_TYPE_FLUSH;

		nbd_submit_bdev_io(nbd->bdev, nbd->bdev_desc, nbd->ch, io);

		break;

#endif

#ifdef NBD_FLAG_SEND_TRIM

	case NBD_CMD_TRIM:

		io->type = SPDK_BDEV_IO_TYPE_UNMAP;

		nbd_submit_bdev_io(nbd->bdev, nbd->bdev_desc, nbd->ch, io);

		break;

#endif

		}



	return 0;

	}



/**

 * Poll an NBD instance.

 *

 * \return 0 on success or negated errno values on error (e.g. connection closed).

 */

static int

_spdk_nbd_poll(struct spdk_nbd_disk *nbd)

{

	struct nbd_io	*io = &nbd->io;

	int		fd = nbd->spdk_sp_fd;

	int64_t		ret;

	int		rc;



	if (io->req_in_progress) {

		ret = read_from_socket(fd, (char *)&io->req + io->offset, sizeof(io->req) - io->offset);

	if (io->state == NBD_IO_RECV_PAYLOAD) {

		ret = read_from_socket(nbd->spdk_sp_fd, io->payload + io->offset, io->payload_size - io->offset);

		if (ret <= 0) {

			return ret;

		}

		io->offset += ret;

		if (io->offset == sizeof(io->req)) {

			io->req_in_progress = false;

			rc = process_request(nbd);

			if (rc != 0) {

				return rc;

			}

		}

	}



	if (io->payload_in_progress && is_write(io->type)) {

		ret = read_from_socket(fd, io->payload + io->offset, io->payload_size - io->offset);

		if (ret <= 0) {

			return ret;

		}

		io->offset += ret;



		/* request payload is fully received */

		if (io->offset == io->payload_size) {

			io->payload_in_progress = false;

			nbd_submit_bdev_io(nbd->bdev, nbd->bdev_desc, nbd->ch, io);

			io->offset = 0;

			io->state = NBD_IO_PROCESS;

			ret = spdk_nbd_io_exec(io);

		}



	}



	if (io->resp_in_progress) {

		ret = write_to_socket(fd, (char *)&io->resp + io->offset, sizeof(io->resp) - io->offset);

	return ret;

}



static int

spdk_nbd_io_xmit(struct nbd_io *io)

{

	struct spdk_nbd_disk *nbd = io->nbd;

	int ret = 0;



	/* resp error and handler are already set in io_done */



	if (io->state == NBD_IO_XMIT_RESP) {

		ret = write_to_socket(nbd->spdk_sp_fd, (char *)&io->resp + io->offset,

				      sizeof(io->resp) - io->offset);

		if (ret <= 0) {

			return ret;

		}



		io->offset += ret;



		/* response is fully transmitted */

		if (io->offset == sizeof(io->resp)) {

			io->resp_in_progress = false;

			if (is_read(io->type)) {

				io->payload_in_progress = true;

			io->offset = 0;



			/* transmit payload only when NBD_CMD_READ with no resp error */

			if (from_be32(&io->req.type) != NBD_CMD_READ || io->resp.error != 0) {

				io->state = NBD_IO_RECV_REQ;

				if (io->payload) {

					spdk_dma_free(io->payload);

					io->payload = NULL;

				}

			} else {

				io->req_in_progress = true;

				io->state = NBD_IO_XMIT_PAYLOAD;

			}

			io->offset = 0;

		}

	}



	if (io->payload_in_progress && is_read(io->type)) {

		ret = write_to_socket(fd, io->payload + io->offset, io->payload_size - io->offset);

	if (io->state == NBD_IO_XMIT_PAYLOAD) {

		ret = write_to_socket(nbd->spdk_sp_fd, io->payload + io->offset, io->payload_size - io->offset);

		if (ret <= 0) {

			return ret;

		}



		io->offset += ret;



		/* read payload is fully transmitted */

		if (io->offset == io->payload_size) {

			io->payload_in_progress = false;

			io->req_in_progress = true;

			io->offset = 0;

			io->state = NBD_IO_RECV_REQ;

			if (io->payload) {

				spdk_dma_free(io->payload);

				io->payload = NULL;

			}

		}

	}



	return 0;

}



/**

 * Poll an NBD instance.

 *

 * \return 0 on success or negated errno values on error (e.g. connection closed).

 */

static int

_spdk_nbd_poll(struct spdk_nbd_disk *nbd)

{

	struct nbd_io	*io = &nbd->io;

	int rc;



	rc = spdk_nbd_io_recv(io);

	if (rc < 0) {

		return rc;

	}



	rc = spdk_nbd_io_xmit(io);



	return rc;

}



static void

spdk_nbd_poll(void *arg)

{
@@ -544,6 +581,8 @@ spdk_nbd_start(const char *bdev_name, const char *nbd_path) 
	if (nbd == NULL) {

		return NULL;

	}



	nbd->io.nbd = nbd;

	nbd->dev_fd = -1;

	nbd->spdk_sp_fd = -1;

	nbd->kernel_sp_fd = -1;
@@ -640,7 +679,6 @@ spdk_nbd_start(const char *bdev_name, const char *nbd_path) 
	}



	to_be32(&nbd->io.resp.magic, NBD_REPLY_MAGIC);

	nbd->io.req_in_progress = true;



	nbd->nbd_poller = spdk_poller_register(spdk_nbd_poll, nbd, 0);