[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index] [Xen-changelog] xenstored now supports multiple concurrent transactions per
# HG changeset patch # User kaf24@xxxxxxxxxxxxxxxxxxxx # Node ID 015f8ae8127649f0c69904fd063ca45d304d4e0c # Parent 1ac39c7a043541cfa94655f0e9ab98d4503c29a2 xenstored now supports multiple concurrent transactions per connection, plus interleaving of transactional and non-transactional accesses. A transaction identifier is added to the xsd_sockmsg header structure (0 means 'not in context of a transaction'). The user and kernel xs interfaces accept a pointer to a transaction handle where appropriate -- currently this is directly cast to an integer identifier in the client library / kernel driver, but will allow for keeping extra dynamic client-side state in future if we need to. The transaction mutex has now gone. It's replaced with a read-write mutex, but this is only acquired for exclusive access during suspend/resume, to ensure there are no in-progress transactions. Signed-off-by: Keir Fraser <keir@xxxxxxxxxxxxx> diff -r 1ac39c7a0435 -r 015f8ae81276 linux-2.6-xen-sparse/drivers/xen/xenbus/xenbus_dev.c --- a/linux-2.6-xen-sparse/drivers/xen/xenbus/xenbus_dev.c Mon Oct 10 13:46:53 2005 +++ b/linux-2.6-xen-sparse/drivers/xen/xenbus/xenbus_dev.c Mon Oct 10 14:38:01 2005 @@ -45,8 +45,14 @@ #include <asm-xen/xen_proc.h> #include <asm/hypervisor.h> +struct xenbus_dev_transaction { + struct list_head list; + struct xenbus_transaction *handle; +}; + struct xenbus_dev_data { - int in_transaction; + /* In-progress transaction. */ + struct list_head transactions; /* Partial request. */ unsigned int len; @@ -103,6 +109,7 @@ size_t len, loff_t *ppos) { struct xenbus_dev_data *u = filp->private_data; + struct xenbus_dev_transaction *trans; void *reply; int err = 0; @@ -129,13 +136,24 @@ case XS_RM: case XS_SET_PERMS: reply = xenbus_dev_request_and_reply(&u->u.msg); - if (IS_ERR(reply)) + if (IS_ERR(reply)) { err = PTR_ERR(reply); - else { - if (u->u.msg.type == XS_TRANSACTION_START) - u->in_transaction = 1; - if (u->u.msg.type == XS_TRANSACTION_END) - u->in_transaction = 0; + } else { + if (u->u.msg.type == XS_TRANSACTION_START) { + trans = kmalloc(sizeof(*trans), GFP_KERNEL); + trans->handle = (struct xenbus_transaction *) + simple_strtoul(reply, NULL, 0); + list_add(&trans->list, &u->transactions); + } else if (u->u.msg.type == XS_TRANSACTION_END) { + list_for_each_entry(trans, &u->transactions, + list) + if ((unsigned long)trans->handle == + (unsigned long)u->u.msg.tx_id) + break; + BUG_ON(&trans->list == &u->transactions); + list_del(&trans->list); + kfree(trans); + } queue_reply(u, (char *)&u->u.msg, sizeof(u->u.msg)); queue_reply(u, (char *)reply, u->u.msg.len); kfree(reply); @@ -169,6 +187,7 @@ return -ENOMEM; memset(u, 0, sizeof(*u)); + INIT_LIST_HEAD(&u->transactions); init_waitqueue_head(&u->read_waitq); filp->private_data = u; @@ -179,9 +198,13 @@ static int xenbus_dev_release(struct inode *inode, struct file *filp) { struct xenbus_dev_data *u = filp->private_data; - - if (u->in_transaction) - xenbus_transaction_end((struct xenbus_transaction *)1, 1); + struct xenbus_dev_transaction *trans, *tmp; + + list_for_each_entry_safe(trans, tmp, &u->transactions, list) { + xenbus_transaction_end(trans->handle, 1); + list_del(&trans->list); + kfree(trans); + } kfree(u); diff -r 1ac39c7a0435 -r 015f8ae81276 linux-2.6-xen-sparse/drivers/xen/xenbus/xenbus_xs.c --- a/linux-2.6-xen-sparse/drivers/xen/xenbus/xenbus_xs.c Mon Oct 10 13:46:53 2005 +++ b/linux-2.6-xen-sparse/drivers/xen/xenbus/xenbus_xs.c Mon Oct 10 14:38:01 2005 @@ -71,38 +71,14 @@ /* One request at a time. */ struct semaphore request_mutex; - /* One transaction at a time. */ - struct semaphore transaction_mutex; - int transaction_pid; + /* Protect transactions against save/restore. */ + struct rw_semaphore suspend_mutex; }; static struct xs_handle xs_state; static LIST_HEAD(watches); static DEFINE_SPINLOCK(watches_lock); - -/* Can wait on !xs_resuming for suspend/resume cycle to complete. */ -static int xs_resuming; -static DECLARE_WAIT_QUEUE_HEAD(xs_resuming_waitq); - -static void request_mutex_acquire(void) -{ - /* - * We can't distinguish non-transactional from transactional - * requests right now. So temporarily acquire the transaction mutex - * if this task is outside transaction context. - */ - if (xs_state.transaction_pid != current->pid) - down(&xs_state.transaction_mutex); - down(&xs_state.request_mutex); -} - -static void request_mutex_release(void) -{ - up(&xs_state.request_mutex); - if (xs_state.transaction_pid != current->pid) - up(&xs_state.transaction_mutex); -} static int get_error(const char *errorstring) { @@ -152,17 +128,17 @@ /* Emergency write. */ void xenbus_debug_write(const char *str, unsigned int count) { - struct xsd_sockmsg msg; + struct xsd_sockmsg msg = { 0 }; msg.type = XS_DEBUG; msg.len = sizeof("print") + count + 1; - request_mutex_acquire(); + down(&xs_state.request_mutex); xb_write(&msg, sizeof(msg)); xb_write("print", sizeof("print")); xb_write(str, count); xb_write("", 1); - request_mutex_release(); + up(&xs_state.request_mutex); } void *xenbus_dev_request_and_reply(struct xsd_sockmsg *msg) @@ -171,12 +147,10 @@ struct xsd_sockmsg req_msg = *msg; int err; - if (req_msg.type == XS_TRANSACTION_START) { - down(&xs_state.transaction_mutex); - xs_state.transaction_pid = current->pid; - } - - request_mutex_acquire(); + if (req_msg.type == XS_TRANSACTION_START) + down_read(&xs_state.suspend_mutex); + + down(&xs_state.request_mutex); err = xb_write(msg, sizeof(*msg) + msg->len); if (err) { @@ -186,20 +160,19 @@ ret = read_reply(&msg->type, &msg->len); } - request_mutex_release(); + up(&xs_state.request_mutex); if ((msg->type == XS_TRANSACTION_END) || ((req_msg.type == XS_TRANSACTION_START) && - (msg->type == XS_ERROR))) { - xs_state.transaction_pid = -1; - up(&xs_state.transaction_mutex); - } + (msg->type == XS_ERROR))) + up_read(&xs_state.suspend_mutex); return ret; } /* Send message to xs, get kmalloc'ed reply. ERR_PTR() on error. */ -static void *xs_talkv(enum xsd_sockmsg_type type, +static void *xs_talkv(struct xenbus_transaction *t, + enum xsd_sockmsg_type type, const struct kvec *iovec, unsigned int num_vecs, unsigned int *len) @@ -209,12 +182,13 @@ unsigned int i; int err; + msg.tx_id = (u32)(unsigned long)t; msg.type = type; msg.len = 0; for (i = 0; i < num_vecs; i++) msg.len += iovec[i].iov_len; - request_mutex_acquire(); + down(&xs_state.request_mutex); err = xb_write(&msg, sizeof(msg)); if (err) { @@ -225,14 +199,14 @@ for (i = 0; i < num_vecs; i++) { err = xb_write(iovec[i].iov_base, iovec[i].iov_len);; if (err) { - request_mutex_release(); + up(&xs_state.request_mutex); return ERR_PTR(err); } } ret = read_reply(&msg.type, len); - request_mutex_release(); + up(&xs_state.request_mutex); if (IS_ERR(ret)) return ret; @@ -248,14 +222,16 @@ } /* Simplified version of xs_talkv: single message. */ -static void *xs_single(enum xsd_sockmsg_type type, - const char *string, unsigned int *len) +static void *xs_single(struct xenbus_transaction *t, + enum xsd_sockmsg_type type, + const char *string, + unsigned int *len) { struct kvec iovec; iovec.iov_base = (void *)string; iovec.iov_len = strlen(string) + 1; - return xs_talkv(type, &iovec, 1, len); + return xs_talkv(t, type, &iovec, 1, len); } /* Many commands only need an ack, don't care what it says. */ @@ -322,7 +298,7 @@ char *strings; unsigned int len; - strings = xs_single(XS_DIRECTORY, join(dir, node), &len); + strings = xs_single(t, XS_DIRECTORY, join(dir, node), &len); if (IS_ERR(strings)) return (char **)strings; @@ -352,7 +328,7 @@ void *xenbus_read(struct xenbus_transaction *t, const char *dir, const char *node, unsigned int *len) { - return xs_single(XS_READ, join(dir, node), len); + return xs_single(t, XS_READ, join(dir, node), len); } EXPORT_SYMBOL(xenbus_read); @@ -372,7 +348,7 @@ iovec[1].iov_base = (void *)string; iovec[1].iov_len = strlen(string); - return xs_error(xs_talkv(XS_WRITE, iovec, ARRAY_SIZE(iovec), NULL)); + return xs_error(xs_talkv(t, XS_WRITE, iovec, ARRAY_SIZE(iovec), NULL)); } EXPORT_SYMBOL(xenbus_write); @@ -380,14 +356,14 @@ int xenbus_mkdir(struct xenbus_transaction *t, const char *dir, const char *node) { - return xs_error(xs_single(XS_MKDIR, join(dir, node), NULL)); + return xs_error(xs_single(t, XS_MKDIR, join(dir, node), NULL)); } EXPORT_SYMBOL(xenbus_mkdir); /* Destroy a file or directory (directories must be empty). */ int xenbus_rm(struct xenbus_transaction *t, const char *dir, const char *node) { - return xs_error(xs_single(XS_RM, join(dir, node), NULL)); + return xs_error(xs_single(t, XS_RM, join(dir, node), NULL)); } EXPORT_SYMBOL(xenbus_rm); @@ -396,18 +372,21 @@ */ struct xenbus_transaction *xenbus_transaction_start(void) { - int err; - - down(&xs_state.transaction_mutex); - xs_state.transaction_pid = current->pid; - - err = xs_error(xs_single(XS_TRANSACTION_START, "", NULL)); - if (err) { - xs_state.transaction_pid = -1; - up(&xs_state.transaction_mutex); - } - - return err ? ERR_PTR(err) : (struct xenbus_transaction *)1; + char *id_str; + unsigned long id; + + down_read(&xs_state.suspend_mutex); + + id_str = xs_single(NULL, XS_TRANSACTION_START, "", NULL); + if (IS_ERR(id_str)) { + up_read(&xs_state.suspend_mutex); + return (struct xenbus_transaction *)id_str; + } + + id = simple_strtoul(id_str, NULL, 0); + kfree(id_str); + + return (struct xenbus_transaction *)id; } EXPORT_SYMBOL(xenbus_transaction_start); @@ -419,17 +398,14 @@ char abortstr[2]; int err; - BUG_ON(t == NULL); - if (abort) strcpy(abortstr, "F"); else strcpy(abortstr, "T"); - err = xs_error(xs_single(XS_TRANSACTION_END, abortstr, NULL)); - - xs_state.transaction_pid = -1; - up(&xs_state.transaction_mutex); + err = xs_error(xs_single(t, XS_TRANSACTION_END, abortstr, NULL)); + + up_read(&xs_state.suspend_mutex); return err; } @@ -567,7 +543,8 @@ iov[1].iov_base = (void *)token; iov[1].iov_len = strlen(token) + 1; - return xs_error(xs_talkv(XS_WATCH, iov, ARRAY_SIZE(iov), NULL)); + return xs_error(xs_talkv(NULL, XS_WATCH, iov, + ARRAY_SIZE(iov), NULL)); } static int xs_unwatch(const char *path, const char *token) @@ -579,7 +556,8 @@ iov[1].iov_base = (char *)token; iov[1].iov_len = strlen(token) + 1; - return xs_error(xs_talkv(XS_UNWATCH, iov, ARRAY_SIZE(iov), NULL)); + return xs_error(xs_talkv(NULL, XS_UNWATCH, iov, + ARRAY_SIZE(iov), NULL)); } static struct xenbus_watch *find_watch(const char *token) @@ -603,6 +581,8 @@ int err; sprintf(token, "%lX", (long)watch); + + down_read(&xs_state.suspend_mutex); spin_lock(&watches_lock); BUG_ON(find_watch(token)); @@ -617,6 +597,8 @@ spin_unlock(&watches_lock); } + up_read(&xs_state.suspend_mutex); + return err; } EXPORT_SYMBOL(register_xenbus_watch); @@ -627,14 +609,13 @@ int err; sprintf(token, "%lX", (long)watch); + + down_read(&xs_state.suspend_mutex); spin_lock(&watches_lock); BUG_ON(!find_watch(token)); list_del(&watch->list); spin_unlock(&watches_lock); - - /* Ensure xs_resume() is not in progress (see comments there). */ - wait_event(xs_resuming_waitq, !xs_resuming); err = xs_unwatch(watch->node, token); if (err) @@ -642,6 +623,8 @@ "XENBUS Failed to release watch %s: %i\n", watch->node, err); + up_read(&xs_state.suspend_mutex); + /* Make sure watch is not in use. */ flush_scheduled_work(); } @@ -649,58 +632,24 @@ void xs_suspend(void) { - down(&xs_state.transaction_mutex); + down_write(&xs_state.suspend_mutex); down(&xs_state.request_mutex); } void xs_resume(void) { - struct list_head *ent, *prev_ent = &watches; struct xenbus_watch *watch; char token[sizeof(watch) * 2 + 1]; - /* Protect against concurrent unregistration and freeing of watches. */ - BUG_ON(xs_resuming); - xs_resuming = 1; - up(&xs_state.request_mutex); - up(&xs_state.transaction_mutex); - - /* - * Iterate over the watch list re-registering each node. We must - * be careful about concurrent registrations and unregistrations. - * We search for the node immediately following the previously - * re-registered node. If we get no match then either we are done - * (previous node is last in list) or the node was unregistered, in - * which case we restart from the beginning of the list. - * register_xenbus_watch() + unregister_xenbus_watch() is safe because - * it will only ever move a watch node earlier in the list, so it - * cannot cause us to skip nodes. - */ - for (;;) { - spin_lock(&watches_lock); - list_for_each(ent, &watches) - if (ent->prev == prev_ent) - break; - spin_unlock(&watches_lock); - - /* No match because prev_ent is at the end of the list? */ - if ((ent == &watches) && (watches.prev == prev_ent)) - break; /* We're done! */ - - if ((prev_ent = ent) != &watches) { - /* - * Safe even with watch_lock not held. We are saved by - * (xs_resumed==1) check in unregister_xenbus_watch. - */ - watch = list_entry(ent, struct xenbus_watch, list); - sprintf(token, "%lX", (long)watch); - xs_watch(watch->node, token); - } - } - - xs_resuming = 0; - wake_up(&xs_resuming_waitq); + + /* No need for watches_lock: the suspend_mutex is sufficient. */ + list_for_each_entry(watch, &watches, list) { + sprintf(token, "%lX", (long)watch); + xs_watch(watch->node, token); + } + + up_write(&xs_state.suspend_mutex); } static void xenbus_fire_watch(void *arg) @@ -801,8 +750,7 @@ init_waitqueue_head(&xs_state.reply_waitq); init_MUTEX(&xs_state.request_mutex); - init_MUTEX(&xs_state.transaction_mutex); - xs_state.transaction_pid = -1; + init_rwsem(&xs_state.suspend_mutex); /* Initialize the shared memory rings to talk to xenstored */ err = xb_init_comms(); diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/utils.h --- a/tools/xenstore/utils.h Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/utils.h Mon Oct 10 14:38:01 2005 @@ -55,4 +55,34 @@ #define dprintf(_fmt, _args...) ((void)0) #endif +/* + * Mux errno values onto returned pointers. + */ + +static inline void *ERR_PTR(long error) +{ + return (void *)error; +} + +static inline long PTR_ERR(const void *ptr) +{ + return (long)ptr; +} + +static inline long IS_ERR(const void *ptr) +{ + return ((unsigned long)ptr > (unsigned long)-1000L); +} + + #endif /* _UTILS_H */ + +/* + * Local variables: + * c-file-style: "linux" + * indent-tabs-mode: t + * c-indent-level: 8 + * c-basic-offset: 8 + * tab-width: 8 + * End: + */ diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/xenstore_client.c --- a/tools/xenstore/xenstore_client.c Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/xenstore_client.c Mon Oct 10 14:38:01 2005 @@ -34,14 +34,10 @@ struct xs_handle *xsh; struct xs_transaction_handle *xth; bool success; - int ret = 0; + int ret = 0, socket = 0; #if defined(CLIENT_read) || defined(CLIENT_list) int prefix = 0; #endif - - xsh = xs_domain_open(); - if (xsh == NULL) - err(1, "xs_domain_open"); while (1) { int c, index = 0; @@ -50,10 +46,11 @@ #if defined(CLIENT_read) || defined(CLIENT_list) {"prefix", 0, 0, 'p'}, #endif + {"socket", 0, 0, 's'}, {0, 0, 0, 0} }; - c = getopt_long(argc, argv, "h" + c = getopt_long(argc, argv, "hs" #if defined(CLIENT_read) || defined(CLIENT_list) "p" #endif @@ -65,6 +62,9 @@ case 'h': usage(argv[0]); /* NOTREACHED */ + case 's': + socket = 1; + break; #if defined(CLIENT_read) || defined(CLIENT_list) case 'p': prefix = 1; @@ -83,6 +83,10 @@ /* NOTREACHED */ } #endif + + xsh = socket ? xs_daemon_open() : xs_domain_open(); + if (xsh == NULL) + err(1, socket ? "xs_daemon_open" : "xs_domain_open"); again: xth = xs_transaction_start(xsh); diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/xenstored_core.c --- a/tools/xenstore/xenstored_core.c Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/xenstored_core.c Mon Oct 10 14:38:01 2005 @@ -238,46 +238,47 @@ static bool write_messages(struct connection *conn) { int ret; - struct buffered_data *out, *tmp; - - list_for_each_entry_safe(out, tmp, &conn->out_list, list) { - if (out->inhdr) { - if (verbose) - xprintf("Writing msg %s (%s) out to %p\n", - sockmsg_string(out->hdr.msg.type), - out->buffer, conn); - ret = conn->write(conn, out->hdr.raw + out->used, - sizeof(out->hdr) - out->used); - if (ret < 0) - return false; - - out->used += ret; - if (out->used < sizeof(out->hdr)) - return true; - - out->inhdr = false; - out->used = 0; - - /* Second write might block if non-zero. */ - if (out->hdr.msg.len && !conn->domain) - return true; - } - - ret = conn->write(conn, out->buffer + out->used, - out->hdr.msg.len - out->used); - + struct buffered_data *out; + + out = list_top(&conn->out_list, struct buffered_data, list); + if (out == NULL) + return true; + + if (out->inhdr) { + if (verbose) + xprintf("Writing msg %s (%s) out to %p\n", + sockmsg_string(out->hdr.msg.type), + out->buffer, conn); + ret = conn->write(conn, out->hdr.raw + out->used, + sizeof(out->hdr) - out->used); if (ret < 0) return false; out->used += ret; - if (out->used != out->hdr.msg.len) + if (out->used < sizeof(out->hdr)) return true; - trace_io(conn, "OUT", out); - - list_del(&out->list); - talloc_free(out); - } + out->inhdr = false; + out->used = 0; + + /* Second write might block if non-zero. */ + if (out->hdr.msg.len && !conn->domain) + return true; + } + + ret = conn->write(conn, out->buffer + out->used, + out->hdr.msg.len - out->used); + if (ret < 0) + return false; + + out->used += ret; + if (out->used != out->hdr.msg.len) + return true; + + trace_io(conn, "OUT", out); + + list_del(&out->list); + talloc_free(out); return true; } @@ -1042,6 +1043,17 @@ */ static void process_message(struct connection *conn, struct buffered_data *in) { + struct transaction *trans; + + trans = transaction_lookup(conn, in->hdr.msg.tx_id); + if (IS_ERR(trans)) { + send_error(conn, -PTR_ERR(trans)); + return; + } + + assert(conn->transaction == NULL); + conn->transaction = trans; + switch (in->hdr.msg.type) { case XS_DIRECTORY: send_directory(conn, onearg(in)); @@ -1116,11 +1128,13 @@ do_get_domain_path(conn, onearg(in)); break; - case XS_WATCH_EVENT: default: eprintf("Client unknown operation %i", in->hdr.msg.type); send_error(conn, ENOSYS); - } + break; + } + + conn->transaction = NULL; } static int out_of_mem(void *data) @@ -1239,15 +1253,14 @@ if (!new) return NULL; + memset(new, 0, sizeof(*new)); new->fd = -1; - new->id = 0; - new->domain = NULL; - new->transaction = NULL; new->write = write; new->read = read; new->can_write = true; INIT_LIST_HEAD(&new->out_list); INIT_LIST_HEAD(&new->watches); + INIT_LIST_HEAD(&new->transaction_list); talloc_set_fail_handler(out_of_mem, &talloc_fail); if (setjmp(talloc_fail)) { @@ -1410,6 +1423,7 @@ static struct option options[] = { + { "no-domain-init", 0, NULL, 'D' }, { "pid-file", 1, NULL, 'F' }, { "no-fork", 0, NULL, 'N' }, { "output-pid", 0, NULL, 'P' }, @@ -1424,11 +1438,15 @@ fd_set inset, outset; bool dofork = true; bool outputpid = false; + bool no_domain_init = false; const char *pidfile = NULL; - while ((opt = getopt_long(argc, argv, "F:NPT:V", options, + while ((opt = getopt_long(argc, argv, "DF:NPT:V", options, NULL)) != -1) { switch (opt) { + case 'D': + no_domain_init = true; + break; case 'F': pidfile = optarg; break; @@ -1501,7 +1519,8 @@ setup_structure(); /* Listen to hypervisor. */ - event_fd = domain_init(); + if (!no_domain_init) + event_fd = domain_init(); /* Restore existing connections. */ restore_existing_connections(); diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/xenstored_core.h --- a/tools/xenstore/xenstored_core.h Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/xenstored_core.h Mon Oct 10 14:38:01 2005 @@ -71,8 +71,12 @@ /* Buffered output data */ struct list_head out_list; - /* My transaction, if any. */ + /* Transaction context for current request (NULL if none). */ struct transaction *transaction; + + /* List of in-progress transactions. */ + struct list_head transaction_list; + u32 next_transaction_id; /* The domain I'm associated with, if any. */ struct domain *domain; diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/xenstored_transaction.c --- a/tools/xenstore/xenstored_transaction.c Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/xenstored_transaction.c Mon Oct 10 14:38:01 2005 @@ -37,7 +37,7 @@ struct changed_node { - /* The list within this transaction. */ + /* List of all changed nodes in the context of this transaction. */ struct list_head list; /* The name of the node. */ @@ -49,14 +49,14 @@ struct transaction { - /* Global list of transactions. */ + /* List of all transactions active on this connection. */ struct list_head list; + + /* Connection-local identifier for this transaction. */ + u32 id; /* Generation when transaction started. */ unsigned int generation; - - /* My owner (conn->transaction == me). */ - struct connection *conn; /* TDB to work on, and filename */ TDB_CONTEXT *tdb; @@ -65,7 +65,7 @@ /* List of changed nodes. */ struct list_head changes; }; -static LIST_HEAD(transactions); + static unsigned int generation; /* Return tdb context to use for this connection. */ @@ -100,7 +100,6 @@ { struct transaction *trans = _transaction; - list_del(&trans->list); trace_destroy(trans, "transaction"); if (trans->tdb) tdb_close(trans->tdb); @@ -108,10 +107,26 @@ return 0; } +struct transaction *transaction_lookup(struct connection *conn, u32 id) +{ + struct transaction *trans; + + if (id == 0) + return NULL; + + list_for_each_entry(trans, &conn->transaction_list, list) + if (trans->id == id) + return trans; + + return ERR_PTR(-ENOENT); +} + void do_transaction_start(struct connection *conn, struct buffered_data *in) { - struct transaction *trans; - + struct transaction *trans, *exists; + char id_str[20]; + + /* We don't support nested transactions. */ if (conn->transaction) { send_error(conn, EBUSY); return; @@ -120,7 +135,6 @@ /* Attach transaction to input for autofree until it's complete */ trans = talloc(in, struct transaction); INIT_LIST_HEAD(&trans->changes); - trans->conn = conn; trans->generation = generation; trans->tdb_name = talloc_asprintf(trans, "%s.%p", xs_daemon_tdb(), trans); @@ -132,11 +146,19 @@ /* Make it close if we go away. */ talloc_steal(trans, trans->tdb); + /* Pick an unused transaction identifier. */ + do { + trans->id = conn->next_transaction_id; + exists = transaction_lookup(conn, conn->next_transaction_id++); + } while (!IS_ERR(exists)); + /* Now we own it. */ - conn->transaction = talloc_steal(conn, trans); - list_add_tail(&trans->list, &transactions); + list_add_tail(&trans->list, &conn->transaction_list); + talloc_steal(conn, trans); talloc_set_destructor(trans, destroy_transaction); - send_ack(conn, XS_TRANSACTION_START); + + sprintf(id_str, "%u", trans->id); + send_reply(conn, XS_TRANSACTION_START, id_str, strlen(id_str)+1); } void do_transaction_end(struct connection *conn, const char *arg) @@ -149,13 +171,13 @@ return; } - if (!conn->transaction) { + if ((trans = conn->transaction) == NULL) { send_error(conn, ENOENT); return; } - trans = conn->transaction; conn->transaction = NULL; + list_del(&trans->list); /* Attach transaction to arg for auto-cleanup */ talloc_steal(arg, trans); @@ -181,3 +203,12 @@ send_ack(conn, XS_TRANSACTION_END); } +/* + * Local variables: + * c-file-style: "linux" + * indent-tabs-mode: t + * c-indent-level: 8 + * c-basic-offset: 8 + * tab-width: 8 + * End: + */ diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/xenstored_transaction.h --- a/tools/xenstore/xenstored_transaction.h Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/xenstored_transaction.h Mon Oct 10 14:38:01 2005 @@ -25,10 +25,11 @@ void do_transaction_start(struct connection *conn, struct buffered_data *node); void do_transaction_end(struct connection *conn, const char *arg); -bool transaction_block(struct connection *conn); +struct transaction *transaction_lookup(struct connection *conn, u32 id); /* This node was changed: can fail and longjmp. */ -void add_change_node(struct transaction *trans, const char *node, bool recurse); +void add_change_node(struct transaction *trans, const char *node, + bool recurse); /* Return tdb context to use for this connection. */ TDB_CONTEXT *tdb_transaction_context(struct transaction *trans); diff -r 1ac39c7a0435 -r 015f8ae81276 tools/xenstore/xs.c --- a/tools/xenstore/xs.c Mon Oct 10 13:46:53 2005 +++ b/tools/xenstore/xs.c Mon Oct 10 14:38:01 2005 @@ -75,36 +75,9 @@ /* One request at a time. */ pthread_mutex_t request_mutex; - - /* One transaction at a time. */ - pthread_mutex_t transaction_mutex; - pthread_t transaction_pthread; }; -struct xs_transaction_handle { - int id; -}; - static void *read_thread(void *arg); - -static void request_mutex_acquire(struct xs_handle *h) -{ - /* - * We can't distinguish non-transactional from transactional - * requests right now. So temporarily acquire the transaction mutex - * if this task is outside transaction context. - */ - if (h->transaction_pthread != pthread_self()) - pthread_mutex_lock(&h->transaction_mutex); - pthread_mutex_lock(&h->request_mutex); -} - -static void request_mutex_release(struct xs_handle *h) -{ - pthread_mutex_unlock(&h->request_mutex); - if (h->transaction_pthread != pthread_self()) - pthread_mutex_unlock(&h->transaction_mutex); -} int xs_fileno(struct xs_handle *h) { @@ -186,8 +159,6 @@ pthread_cond_init(&h->reply_condvar, NULL); pthread_mutex_init(&h->request_mutex, NULL); - pthread_mutex_init(&h->transaction_mutex, NULL); - h->transaction_pthread = -1; if (pthread_create(&h->read_thr, NULL, read_thread, h) != 0) goto error; @@ -223,7 +194,6 @@ { struct xs_stored_msg *msg, *tmsg; - pthread_mutex_lock(&h->transaction_mutex); pthread_mutex_lock(&h->request_mutex); pthread_mutex_lock(&h->reply_mutex); pthread_mutex_lock(&h->watch_mutex); @@ -242,7 +212,6 @@ free(msg); } - pthread_mutex_unlock(&h->transaction_mutex); pthread_mutex_unlock(&h->request_mutex); pthread_mutex_unlock(&h->reply_mutex); pthread_mutex_unlock(&h->watch_mutex); @@ -321,8 +290,10 @@ } /* Send message to xs, get malloc'ed reply. NULL and set errno on error. */ -static void *xs_talkv(struct xs_handle *h, enum xsd_sockmsg_type type, - const struct iovec *iovec, unsigned int num_vecs, +static void *xs_talkv(struct xs_handle *h, struct xs_transaction_handle *t, + enum xsd_sockmsg_type type, + const struct iovec *iovec, + unsigned int num_vecs, unsigned int *len) { struct xsd_sockmsg msg; @@ -331,6 +302,7 @@ unsigned int i; struct sigaction ignorepipe, oldact; + msg.tx_id = (u32)(unsigned long)t; msg.type = type; msg.len = 0; for (i = 0; i < num_vecs; i++) @@ -341,7 +313,7 @@ ignorepipe.sa_flags = 0; sigaction(SIGPIPE, &ignorepipe, &oldact); - request_mutex_acquire(h); + pthread_mutex_lock(&h->request_mutex); if (!xs_write_all(h->fd, &msg, sizeof(msg))) goto fail; @@ -354,7 +326,7 @@ if (!ret) goto fail; - request_mutex_release(h); + pthread_mutex_unlock(&h->request_mutex); sigaction(SIGPIPE, &oldact, NULL); if (msg.type == XS_ERROR) { @@ -375,7 +347,7 @@ fail: /* We're in a bad state, so close fd. */ saved_errno = errno; - request_mutex_release(h); + pthread_mutex_unlock(&h->request_mutex); sigaction(SIGPIPE, &oldact, NULL); close_fd: close(h->fd); @@ -393,14 +365,16 @@ } /* Simplified version of xs_talkv: single message. */ -static void *xs_single(struct xs_handle *h, enum xsd_sockmsg_type type, - const char *string, unsigned int *len) +static void *xs_single(struct xs_handle *h, struct xs_transaction_handle *t, + enum xsd_sockmsg_type type, + const char *string, + unsigned int *len) { struct iovec iovec; iovec.iov_base = (void *)string; iovec.iov_len = strlen(string) + 1; - return xs_talkv(h, type, &iovec, 1, len); + return xs_talkv(h, t, type, &iovec, 1, len); } static bool xs_bool(char *reply) @@ -417,7 +391,7 @@ char *strings, *p, **ret; unsigned int len; - strings = xs_single(h, XS_DIRECTORY, path, &len); + strings = xs_single(h, t, XS_DIRECTORY, path, &len); if (!strings) return NULL; @@ -446,7 +420,7 @@ void *xs_read(struct xs_handle *h, struct xs_transaction_handle *t, const char *path, unsigned int *len) { - return xs_single(h, XS_READ, path, len); + return xs_single(h, t, XS_READ, path, len); } /* Write the value of a single file. @@ -462,7 +436,8 @@ iovec[1].iov_base = (void *)data; iovec[1].iov_len = len; - return xs_bool(xs_talkv(h, XS_WRITE, iovec, ARRAY_SIZE(iovec), NULL)); + return xs_bool(xs_talkv(h, t, XS_WRITE, iovec, + ARRAY_SIZE(iovec), NULL)); } /* Create a new directory. @@ -471,7 +446,7 @@ bool xs_mkdir(struct xs_handle *h, struct xs_transaction_handle *t, const char *path) { - return xs_bool(xs_single(h, XS_MKDIR, path, NULL)); + return xs_bool(xs_single(h, t, XS_MKDIR, path, NULL)); } /* Destroy a file or directory (directories must be empty). @@ -480,7 +455,7 @@ bool xs_rm(struct xs_handle *h, struct xs_transaction_handle *t, const char *path) { - return xs_bool(xs_single(h, XS_RM, path, NULL)); + return xs_bool(xs_single(h, t, XS_RM, path, NULL)); } /* Get permissions of node (first element is owner). @@ -494,7 +469,7 @@ unsigned int len; struct xs_permissions *ret; - strings = xs_single(h, XS_GET_PERMS, path, &len); + strings = xs_single(h, t, XS_GET_PERMS, path, &len); if (!strings) return NULL; @@ -544,7 +519,7 @@ goto unwind; } - if (!xs_bool(xs_talkv(h, XS_SET_PERMS, iov, 1+num_perms, NULL))) + if (!xs_bool(xs_talkv(h, t, XS_SET_PERMS, iov, 1+num_perms, NULL))) goto unwind; for (i = 0; i < num_perms; i++) free(iov[i+1].iov_base); @@ -571,7 +546,8 @@ iov[1].iov_base = (void *)token; iov[1].iov_len = strlen(token) + 1; - return xs_bool(xs_talkv(h, XS_WATCH, iov, ARRAY_SIZE(iov), NULL)); + return xs_bool(xs_talkv(h, NULL, XS_WATCH, iov, + ARRAY_SIZE(iov), NULL)); } /* Find out what node change was on (will block if nothing pending). @@ -637,7 +613,8 @@ iov[1].iov_base = (char *)token; iov[1].iov_len = strlen(token) + 1; - return xs_bool(xs_talkv(h, XS_UNWATCH, iov, ARRAY_SIZE(iov), NULL)); + return xs_bool(xs_talkv(h, NULL, XS_UNWATCH, iov, + ARRAY_SIZE(iov), NULL)); } /* Start a transaction: changes by others will not be seen during this @@ -647,18 +624,17 @@ */ struct xs_transaction_handle *xs_transaction_start(struct xs_handle *h) { - bool rc; - - pthread_mutex_lock(&h->transaction_mutex); - h->transaction_pthread = pthread_self(); - - rc = xs_bool(xs_single(h, XS_TRANSACTION_START, "", NULL)); - if (!rc) { - h->transaction_pthread = -1; - pthread_mutex_unlock(&h->transaction_mutex); - } - - return (struct xs_transaction_handle *)rc; + char *id_str; + unsigned long id; + + id_str = xs_single(h, NULL, XS_TRANSACTION_START, "", NULL); + if (id_str == NULL) + return NULL; + + id = strtoul(id_str, NULL, 0); + free(id_str); + + return (struct xs_transaction_handle *)id; } /* End a transaction. @@ -670,22 +646,13 @@ bool abort) { char abortstr[2]; - bool rc; - - if (t == NULL) - return -EINVAL; if (abort) strcpy(abortstr, "F"); else strcpy(abortstr, "T"); - rc = xs_bool(xs_single(h, XS_TRANSACTION_END, abortstr, NULL)); - - h->transaction_pthread = -1; - pthread_mutex_unlock(&h->transaction_mutex); - - return rc; + return xs_bool(xs_single(h, t, XS_TRANSACTION_END, abortstr, NULL)); } /* Introduce a new domain. @@ -713,7 +680,8 @@ iov[3].iov_base = (char *)path; iov[3].iov_len = strlen(path) + 1; - return xs_bool(xs_talkv(h, XS_INTRODUCE, iov, ARRAY_SIZE(iov), NULL)); + return xs_bool(xs_talkv(h, NULL, XS_INTRODUCE, iov, + ARRAY_SIZE(iov), NULL)); } bool xs_release_domain(struct xs_handle *h, domid_t domid) @@ -722,7 +690,7 @@ sprintf(domid_str, "%u", domid); - return xs_bool(xs_single(h, XS_RELEASE, domid_str, NULL)); + return xs_bool(xs_single(h, NULL, XS_RELEASE, domid_str, NULL)); } char *xs_get_domain_path(struct xs_handle *h, domid_t domid) @@ -731,7 +699,7 @@ sprintf(domid_str, "%u", domid); - return xs_single(h, XS_GET_DOMAIN_PATH, domid_str, NULL); + return xs_single(h, NULL, XS_GET_DOMAIN_PATH, domid_str, NULL); } /* Only useful for DEBUG versions */ @@ -745,7 +713,8 @@ iov[1].iov_base = data; iov[1].iov_len = len; - return xs_talkv(h, XS_DEBUG, iov, ARRAY_SIZE(iov), NULL); + return xs_talkv(h, NULL, XS_DEBUG, iov, + ARRAY_SIZE(iov), NULL); } static void *read_thread(void *arg) _______________________________________________ Xen-changelog mailing list Xen-changelog@xxxxxxxxxxxxxxxxxxx http://lists.xensource.com/xen-changelog
|
![]() |
Lists.xenproject.org is hosted with RackSpace, monitoring our |