tapdisk.c

xen虚拟机源代码安装包

			len = write(fds[WRITE], buf, msglen);
			free(path);
			return 1;
			
		case CTLMSG_NEWDEV:
			msg_dev = (msg_newdev_t *)(buf + sizeof(msg_hdr_t));

			s = get_state(msg->cookie);
			DPRINTF("Retrieving state, cookie %d.....[%s]\n",
				msg->cookie, (s == NULL ? "FAIL":"OK"));
			if (s != NULL) {
				ret = ((map_new_dev(s, msg_dev->devnum) 
					== msg_dev->devnum ? 0: -1));
				connected_disks++;
			}	

			memset(buf, 0x00, MSG_SIZE); 
			msglen = sizeof(msg_hdr_t);
			msg->type = (ret == 0 ? CTLMSG_NEWDEV_RSP 
				              : CTLMSG_NEWDEV_FAIL);
			msg->len = msglen;

			len = write(fds[WRITE], buf, msglen);
			return 1;

		case CTLMSG_CLOSE:
			s = get_state(msg->cookie);
			if (s) unmap_disk(s);
			
			connected_disks--;
			sig_handler(SIGINT);

			return 1;			

		case CTLMSG_PID:
			memset(buf, 0x00, MSG_SIZE);
			msglen = sizeof(msg_hdr_t) + sizeof(msg_pid_t);
			msg->type = CTLMSG_PID_RSP;
			msg->len = msglen;

			msg_pid = (msg_pid_t *)(buf + sizeof(msg_hdr_t));
			process = getpid();
			msg_pid->pid = process;

			len = write(fds[WRITE], buf, msglen);
			return 1;

		default:
			return 0;
		}
	}
	return 0;
}

static inline int write_rsp_to_ring(struct td_state *s, blkif_response_t *rsp)
{
	tapdev_info_t *info = s->ring_info;
	blkif_response_t *rsp_d;
	
	rsp_d = RING_GET_RESPONSE(&info->fe_ring, info->fe_ring.rsp_prod_pvt);
	memcpy(rsp_d, rsp, sizeof(blkif_response_t));
	info->fe_ring.rsp_prod_pvt++;
	
	return 0;
}

static inline void kick_responses(struct td_state *s)
{
	tapdev_info_t *info = s->ring_info;

	if (info->fe_ring.rsp_prod_pvt != info->fe_ring.sring->rsp_prod) 
	{
		RING_PUSH_RESPONSES(&info->fe_ring);
		ioctl(info->fd, BLKTAP_IOCTL_KICK_FE);
	}
}

static void io_done(struct disk_driver *dd, int sid)
{
	struct tap_disk *drv = dd->drv;

	if (!run) return; /*We have received signal to close*/

	if (sid > MAX_IOFD || drv->td_do_callbacks(dd, sid) > 0)
		kick_responses(dd->td_state);

	return;
}

static inline uint64_t
segment_start(blkif_request_t *req, int sidx)
{
	int i;
	uint64_t start = req->sector_number;

	for (i = 0; i < sidx; i++) 
		start += (req->seg[i].last_sect - req->seg[i].first_sect + 1);

	return start;
}

uint64_t sends, responds;
static int send_responses(struct disk_driver *dd, int res, 
		   uint64_t sector, int nr_secs, int idx, void *private)
{
	pending_req_t   *preq;
	blkif_request_t *req;
	int responses_queued = 0;
	struct td_state *s = dd->td_state;
	blkif_t *blkif = s->blkif;
	int sidx = (int)(long)private, secs_done = nr_secs;

	if ( (idx > MAX_REQUESTS-1) )
	{
		DPRINTF("invalid index returned(%u)!\n", idx);
		return 0;
	}
	preq = &blkif->pending_list[idx];
	req  = &preq->req;

	if (res == BLK_NOT_ALLOCATED) {
		res = do_cow_read(dd, req, sidx, sector, nr_secs);
		if (res >= 0) {
			secs_done = res;
			res = 0;
		} else
			secs_done = 0;
	}

	preq->secs_pending -= secs_done;

	if (res == -EBUSY && preq->submitting) 
		return -EBUSY;  /* propagate -EBUSY back to higher layers */
	if (res) 
		preq->status = BLKIF_RSP_ERROR;
	
	if (!preq->submitting && preq->secs_pending == 0) 
	{
		blkif_request_t tmp;
		blkif_response_t *rsp;

		tmp = preq->req;
		rsp = (blkif_response_t *)req;
		
		rsp->id = tmp.id;
		rsp->operation = tmp.operation;
		rsp->status = preq->status;
		
		write_rsp_to_ring(s, rsp);
		responses_queued++;
	}
	return responses_queued;
}

int do_cow_read(struct disk_driver *dd, blkif_request_t *req, 
		int sidx, uint64_t sector, int nr_secs)
{
	char *page;
	int ret, early;
	uint64_t seg_start, seg_end;
	struct td_state  *s = dd->td_state;
	tapdev_info_t *info = s->ring_info;
	struct disk_driver *parent = dd->next;
	
	seg_start = segment_start(req, sidx);
	seg_end   = seg_start + req->seg[sidx].last_sect + 1;
	
	ASSERT(sector >= seg_start && sector + nr_secs <= seg_end);

	page  = (char *)MMAP_VADDR(info->vstart, 
				   (unsigned long)req->id, sidx);
	page += (req->seg[sidx].first_sect << SECTOR_SHIFT);
	page += ((sector - seg_start) << SECTOR_SHIFT);

	if (!parent) {
		memset(page, 0, nr_secs << SECTOR_SHIFT);
		return nr_secs;
	}

	/* reissue request to backing file */
	ret = parent->drv->td_queue_read(parent, sector, nr_secs,
					 page, send_responses, 
					 req->id, (void *)(long)sidx);
	if (ret > 0)
		parent->early += ret;

	return ((ret >= 0) ? 0 : ret);
}

static void get_io_request(struct td_state *s)
{
	RING_IDX          rp, rc, j, i;
	blkif_request_t  *req;
	int idx, nsects, ret;
	uint64_t sector_nr;
	char *page;
	int early = 0; /* count early completions */
	struct disk_driver *dd = s->disks;
	struct tap_disk *drv   = dd->drv;
	blkif_t *blkif = s->blkif;
	tapdev_info_t *info = s->ring_info;
	int page_size = getpagesize();

	if (!run) return; /*We have received signal to close*/

	rp = info->fe_ring.sring->req_prod; 
	xen_rmb();
	for (j = info->fe_ring.req_cons; j != rp; j++)
	{
		int done = 0, start_seg = 0; 

		req = NULL;
		req = RING_GET_REQUEST(&info->fe_ring, j);
		++info->fe_ring.req_cons;
		
		if (req == NULL) continue;

		idx = req->id;

		if (info->busy.req) {
			/* continue where we left off last time */
			ASSERT(info->busy.req == req);
			start_seg = info->busy.seg_idx;
			sector_nr = segment_start(req, start_seg);
			info->busy.seg_idx = 0;
			info->busy.req     = NULL;
		} else {
			ASSERT(blkif->pending_list[idx].secs_pending == 0);
			memcpy(&blkif->pending_list[idx].req, 
			       req, sizeof(*req));
			blkif->pending_list[idx].status = BLKIF_RSP_OKAY;
			blkif->pending_list[idx].submitting = 1;
			sector_nr = req->sector_number;
		}

		if ((dd->flags & TD_RDONLY) && 
		    (req->operation == BLKIF_OP_WRITE)) {
			blkif->pending_list[idx].status = BLKIF_RSP_ERROR;
			goto send_response;
		}

		for (i = start_seg; i < req->nr_segments; i++) {
			nsects = req->seg[i].last_sect - 
				 req->seg[i].first_sect + 1;
	
			if ((req->seg[i].last_sect >= page_size >> 9) ||
			    (nsects <= 0))
				continue;

			page  = (char *)MMAP_VADDR(info->vstart, 
						   (unsigned long)req->id, i);
			page += (req->seg[i].first_sect << SECTOR_SHIFT);

			if (sector_nr >= s->size) {
				DPRINTF("Sector request failed:\n");
				DPRINTF("%s request, idx [%d,%d] size [%llu], "
					"sector [%llu,%llu]\n",
					(req->operation == BLKIF_OP_WRITE ? 
					 "WRITE" : "READ"),
					idx,i,
					(long long unsigned) 
						nsects<<SECTOR_SHIFT,
					(long long unsigned) 
						sector_nr<<SECTOR_SHIFT,
					(long long unsigned) sector_nr);
				continue;
			}

			blkif->pending_list[idx].secs_pending += nsects;

			switch (req->operation) 
			{
			case BLKIF_OP_WRITE:
				ret = drv->td_queue_write(dd, sector_nr,
							  nsects, page, 
							  send_responses,
							  idx, (void *)(long)i);
				if (ret > 0) dd->early += ret;
				else if (ret == -EBUSY) {
					/* put req back on queue */
					--info->fe_ring.req_cons;
					info->busy.req     = req;
					info->busy.seg_idx = i;
					goto out;
				}
				break;
			case BLKIF_OP_READ:
				ret = drv->td_queue_read(dd, sector_nr,
							 nsects, page, 
							 send_responses,
							 idx, (void *)(long)i);
				if (ret > 0) dd->early += ret;
				else if (ret == -EBUSY) {
					/* put req back on queue */
					--info->fe_ring.req_cons;
					info->busy.req     = req;
					info->busy.seg_idx = i;
					goto out;
				}
				break;
			default:
				DPRINTF("Unknown block operation\n");
				break;
			}
			sector_nr += nsects;
		}
	send_response:
		blkif->pending_list[idx].submitting = 0;
		/* force write_rsp_to_ring for synchronous case */
		if (blkif->pending_list[idx].secs_pending == 0)
			dd->early += send_responses(dd, 0, 0, 0, idx, 
						    (void *)(long)0);
	}

 out:
	/*Batch done*/
	td_for_each_disk(s, dd) {
		dd->early += dd->drv->td_submit(dd);
		if (dd->early > 0) {
			io_done(dd, MAX_IOFD + 1);
			dd->early = 0;
		}
	}

	return;
}

int main(int argc, char *argv[])
{
	int len, msglen, ret;
	char *p, *buf;
	fd_set readfds, writefds;	
	fd_list_entry_t *ptr;
	struct td_state *s;
	char openlogbuf[128];

	if (argc != 3) usage();

	daemonize();

	snprintf(openlogbuf, sizeof(openlogbuf), "TAPDISK[%d]", getpid());
	openlog(openlogbuf, LOG_CONS|LOG_ODELAY, LOG_DAEMON);
	/*Setup signal handlers*/
	signal (SIGBUS, sig_handler);
	signal (SIGINT, sig_handler);

	/*Open the control channel*/
	fds[READ]  = open(argv[1],O_RDWR|O_NONBLOCK);
	fds[WRITE] = open(argv[2],O_RDWR|O_NONBLOCK);

	if ( (fds[READ] < 0) || (fds[WRITE] < 0) ) 
	{
		DPRINTF("FD open failed [%d,%d]\n", fds[READ], fds[WRITE]);
		exit(-1);
	}

	buf = calloc(MSG_SIZE, 1);

	if (buf == NULL) 
        {
		DPRINTF("ERROR: allocating memory.\n");
		exit(-1);
	}

	while (run) 
        {
		ret = 0;
		FD_ZERO(&readfds);
		FD_SET(fds[READ], &readfds);
		maxfds = fds[READ];

		/*Set all tap fds*/
		LOCAL_FD_SET(&readfds);

		/*Wait for incoming messages*/
		ret = select(maxfds + 1, &readfds, (fd_set *) 0, 
			     (fd_set *) 0, NULL);

		if (ret > 0) 
		{
			ptr = fd_start;
			while (ptr != NULL) {
				int progress_made = 0;
				struct disk_driver *dd;
				tapdev_info_t *info = ptr->s->ring_info;

				td_for_each_disk(ptr->s, dd) {
					if (dd->io_fd[READ] &&
					    FD_ISSET(dd->io_fd[READ], 
						     &readfds)) {
						io_done(dd, READ);
						progress_made = 1;
					}
				}

				/* completed io from above may have 
				 * queued new requests on chained disks */
				if (progress_made) {
					td_for_each_disk(ptr->s, dd) {
						dd->early += 
							dd->drv->td_submit(dd);
						if (dd->early > 0) {
							io_done(dd, 
								MAX_IOFD + 1);
							dd->early = 0;
						}
					}
				}

				if (FD_ISSET(ptr->tap_fd, &readfds) ||
				    (info->busy.req && progress_made))
					get_io_request(ptr->s);

				ptr = ptr->next;
			}

			if (FD_ISSET(fds[READ], &readfds))
				read_msg(buf);
		}
	}
	free(buf);
	close(fds[READ]);
	close(fds[WRITE]);

	ptr = fd_start;
	while (ptr != NULL) {
		s = ptr->s;
		unmap_disk(s);
		close(ptr->tap_fd);
		ptr = ptr->next;
	}
	closelog();

	return 0;
}