xenbus_xs.c

来自「linux 内核源代码」· C语言 代码 · 共 862 行 · 第 1/2 页

	char *id_str;

	down_read(&xs_state.transaction_mutex);

	id_str = xs_single(XBT_NIL, XS_TRANSACTION_START, "", NULL);
	if (IS_ERR(id_str)) {
		up_read(&xs_state.transaction_mutex);
		return PTR_ERR(id_str);
	}

	t->id = simple_strtoul(id_str, NULL, 0);
	kfree(id_str);
	return 0;
}
EXPORT_SYMBOL_GPL(xenbus_transaction_start);

/* End a transaction.
 * If abandon is true, transaction is discarded instead of committed.
 */
int xenbus_transaction_end(struct xenbus_transaction t, int abort)
{
	char abortstr[2];
	int err;

	if (abort)
		strcpy(abortstr, "F");
	else
		strcpy(abortstr, "T");

	err = xs_error(xs_single(t, XS_TRANSACTION_END, abortstr, NULL));

	up_read(&xs_state.transaction_mutex);

	return err;
}
EXPORT_SYMBOL_GPL(xenbus_transaction_end);

/* Single read and scanf: returns -errno or num scanned. */
int xenbus_scanf(struct xenbus_transaction t,
		 const char *dir, const char *node, const char *fmt, ...)
{
	va_list ap;
	int ret;
	char *val;

	val = xenbus_read(t, dir, node, NULL);
	if (IS_ERR(val))
		return PTR_ERR(val);

	va_start(ap, fmt);
	ret = vsscanf(val, fmt, ap);
	va_end(ap);
	kfree(val);
	/* Distinctive errno. */
	if (ret == 0)
		return -ERANGE;
	return ret;
}
EXPORT_SYMBOL_GPL(xenbus_scanf);

/* Single printf and write: returns -errno or 0. */
int xenbus_printf(struct xenbus_transaction t,
		  const char *dir, const char *node, const char *fmt, ...)
{
	va_list ap;
	int ret;
#define PRINTF_BUFFER_SIZE 4096
	char *printf_buffer;

	printf_buffer = kmalloc(PRINTF_BUFFER_SIZE, GFP_KERNEL);
	if (printf_buffer == NULL)
		return -ENOMEM;

	va_start(ap, fmt);
	ret = vsnprintf(printf_buffer, PRINTF_BUFFER_SIZE, fmt, ap);
	va_end(ap);

	BUG_ON(ret > PRINTF_BUFFER_SIZE-1);
	ret = xenbus_write(t, dir, node, printf_buffer);

	kfree(printf_buffer);

	return ret;
}
EXPORT_SYMBOL_GPL(xenbus_printf);

/* Takes tuples of names, scanf-style args, and void **, NULL terminated. */
int xenbus_gather(struct xenbus_transaction t, const char *dir, ...)
{
	va_list ap;
	const char *name;
	int ret = 0;

	va_start(ap, dir);
	while (ret == 0 && (name = va_arg(ap, char *)) != NULL) {
		const char *fmt = va_arg(ap, char *);
		void *result = va_arg(ap, void *);
		char *p;

		p = xenbus_read(t, dir, name, NULL);
		if (IS_ERR(p)) {
			ret = PTR_ERR(p);
			break;
		}
		if (fmt) {
			if (sscanf(p, fmt, result) == 0)
				ret = -EINVAL;
			kfree(p);
		} else
			*(char **)result = p;
	}
	va_end(ap);
	return ret;
}
EXPORT_SYMBOL_GPL(xenbus_gather);

static int xs_watch(const char *path, const char *token)
{
	struct kvec iov[2];

	iov[0].iov_base = (void *)path;
	iov[0].iov_len = strlen(path) + 1;
	iov[1].iov_base = (void *)token;
	iov[1].iov_len = strlen(token) + 1;

	return xs_error(xs_talkv(XBT_NIL, XS_WATCH, iov,
				 ARRAY_SIZE(iov), NULL));
}

static int xs_unwatch(const char *path, const char *token)
{
	struct kvec iov[2];

	iov[0].iov_base = (char *)path;
	iov[0].iov_len = strlen(path) + 1;
	iov[1].iov_base = (char *)token;
	iov[1].iov_len = strlen(token) + 1;

	return xs_error(xs_talkv(XBT_NIL, XS_UNWATCH, iov,
				 ARRAY_SIZE(iov), NULL));
}

static struct xenbus_watch *find_watch(const char *token)
{
	struct xenbus_watch *i, *cmp;

	cmp = (void *)simple_strtoul(token, NULL, 16);

	list_for_each_entry(i, &watches, list)
		if (i == cmp)
			return i;

	return NULL;
}

/* Register callback to watch this node. */
int register_xenbus_watch(struct xenbus_watch *watch)
{
	/* Pointer in ascii is the token. */
	char token[sizeof(watch) * 2 + 1];
	int err;

	sprintf(token, "%lX", (long)watch);

	down_read(&xs_state.watch_mutex);

	spin_lock(&watches_lock);
	BUG_ON(find_watch(token));
	list_add(&watch->list, &watches);
	spin_unlock(&watches_lock);

	err = xs_watch(watch->node, token);

	/* Ignore errors due to multiple registration. */
	if ((err != 0) && (err != -EEXIST)) {
		spin_lock(&watches_lock);
		list_del(&watch->list);
		spin_unlock(&watches_lock);
	}

	up_read(&xs_state.watch_mutex);

	return err;
}
EXPORT_SYMBOL_GPL(register_xenbus_watch);

void unregister_xenbus_watch(struct xenbus_watch *watch)
{
	struct xs_stored_msg *msg, *tmp;
	char token[sizeof(watch) * 2 + 1];
	int err;

	sprintf(token, "%lX", (long)watch);

	down_read(&xs_state.watch_mutex);

	spin_lock(&watches_lock);
	BUG_ON(!find_watch(token));
	list_del(&watch->list);
	spin_unlock(&watches_lock);

	err = xs_unwatch(watch->node, token);
	if (err)
		printk(KERN_WARNING
		       "XENBUS Failed to release watch %s: %i\n",
		       watch->node, err);

	up_read(&xs_state.watch_mutex);

	/* Make sure there are no callbacks running currently (unless
	   its us) */
	if (current->pid != xenwatch_pid)
		mutex_lock(&xenwatch_mutex);

	/* Cancel pending watch events. */
	spin_lock(&watch_events_lock);
	list_for_each_entry_safe(msg, tmp, &watch_events, list) {
		if (msg->u.watch.handle != watch)
			continue;
		list_del(&msg->list);
		kfree(msg->u.watch.vec);
		kfree(msg);
	}
	spin_unlock(&watch_events_lock);

	if (current->pid != xenwatch_pid)
		mutex_unlock(&xenwatch_mutex);
}
EXPORT_SYMBOL_GPL(unregister_xenbus_watch);

void xs_suspend(void)
{
	down_write(&xs_state.transaction_mutex);
	down_write(&xs_state.watch_mutex);
	mutex_lock(&xs_state.request_mutex);
	mutex_lock(&xs_state.response_mutex);
}

void xs_resume(void)
{
	struct xenbus_watch *watch;
	char token[sizeof(watch) * 2 + 1];

	mutex_unlock(&xs_state.response_mutex);
	mutex_unlock(&xs_state.request_mutex);
	up_write(&xs_state.transaction_mutex);

	/* No need for watches_lock: the watch_mutex is sufficient. */
	list_for_each_entry(watch, &watches, list) {
		sprintf(token, "%lX", (long)watch);
		xs_watch(watch->node, token);
	}

	up_write(&xs_state.watch_mutex);
}

void xs_suspend_cancel(void)
{
	mutex_unlock(&xs_state.response_mutex);
	mutex_unlock(&xs_state.request_mutex);
	up_write(&xs_state.watch_mutex);
	up_write(&xs_state.transaction_mutex);
}

static int xenwatch_thread(void *unused)
{
	struct list_head *ent;
	struct xs_stored_msg *msg;

	for (;;) {
		wait_event_interruptible(watch_events_waitq,
					 !list_empty(&watch_events));

		if (kthread_should_stop())
			break;

		mutex_lock(&xenwatch_mutex);

		spin_lock(&watch_events_lock);
		ent = watch_events.next;
		if (ent != &watch_events)
			list_del(ent);
		spin_unlock(&watch_events_lock);

		if (ent != &watch_events) {
			msg = list_entry(ent, struct xs_stored_msg, list);
			msg->u.watch.handle->callback(
				msg->u.watch.handle,
				(const char **)msg->u.watch.vec,
				msg->u.watch.vec_size);
			kfree(msg->u.watch.vec);
			kfree(msg);
		}

		mutex_unlock(&xenwatch_mutex);
	}

	return 0;
}

static int process_msg(void)
{
	struct xs_stored_msg *msg;
	char *body;
	int err;

	/*
	 * We must disallow save/restore while reading a xenstore message.
	 * A partial read across s/r leaves us out of sync with xenstored.
	 */
	for (;;) {
		err = xb_wait_for_data_to_read();
		if (err)
			return err;
		mutex_lock(&xs_state.response_mutex);
		if (xb_data_to_read())
			break;
		/* We raced with save/restore: pending data 'disappeared'. */
		mutex_unlock(&xs_state.response_mutex);
	}


	msg = kmalloc(sizeof(*msg), GFP_KERNEL);
	if (msg == NULL) {
		err = -ENOMEM;
		goto out;
	}

	err = xb_read(&msg->hdr, sizeof(msg->hdr));
	if (err) {
		kfree(msg);
		goto out;
	}

	body = kmalloc(msg->hdr.len + 1, GFP_KERNEL);
	if (body == NULL) {
		kfree(msg);
		err = -ENOMEM;
		goto out;
	}

	err = xb_read(body, msg->hdr.len);
	if (err) {
		kfree(body);
		kfree(msg);
		goto out;
	}
	body[msg->hdr.len] = '\0';

	if (msg->hdr.type == XS_WATCH_EVENT) {
		msg->u.watch.vec = split(body, msg->hdr.len,
					 &msg->u.watch.vec_size);
		if (IS_ERR(msg->u.watch.vec)) {
			err = PTR_ERR(msg->u.watch.vec);
			kfree(msg);
			goto out;
		}

		spin_lock(&watches_lock);
		msg->u.watch.handle = find_watch(
			msg->u.watch.vec[XS_WATCH_TOKEN]);
		if (msg->u.watch.handle != NULL) {
			spin_lock(&watch_events_lock);
			list_add_tail(&msg->list, &watch_events);
			wake_up(&watch_events_waitq);
			spin_unlock(&watch_events_lock);
		} else {
			kfree(msg->u.watch.vec);
			kfree(msg);
		}
		spin_unlock(&watches_lock);
	} else {
		msg->u.reply.body = body;
		spin_lock(&xs_state.reply_lock);
		list_add_tail(&msg->list, &xs_state.reply_list);
		spin_unlock(&xs_state.reply_lock);
		wake_up(&xs_state.reply_waitq);
	}

 out:
	mutex_unlock(&xs_state.response_mutex);
	return err;
}

static int xenbus_thread(void *unused)
{
	int err;

	for (;;) {
		err = process_msg();
		if (err)
			printk(KERN_WARNING "XENBUS error %d while reading "
			       "message\n", err);
		if (kthread_should_stop())
			break;
	}

	return 0;
}

int xs_init(void)
{
	int err;
	struct task_struct *task;

	INIT_LIST_HEAD(&xs_state.reply_list);
	spin_lock_init(&xs_state.reply_lock);
	init_waitqueue_head(&xs_state.reply_waitq);

	mutex_init(&xs_state.request_mutex);
	mutex_init(&xs_state.response_mutex);
	init_rwsem(&xs_state.transaction_mutex);
	init_rwsem(&xs_state.watch_mutex);

	/* Initialize the shared memory rings to talk to xenstored */
	err = xb_init_comms();
	if (err)
		return err;

	task = kthread_run(xenwatch_thread, NULL, "xenwatch");
	if (IS_ERR(task))
		return PTR_ERR(task);
	xenwatch_pid = task->pid;

	task = kthread_run(xenbus_thread, NULL, "xenbus");
	if (IS_ERR(task))
		return PTR_ERR(task);

	return 0;
}