This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 7a8ed9f71aca3d788d1450d42168ea51cd3d5ca4 (commit) via e67d6b5687b12a823500c743905c6aed1537eddb (commit) via 16ae957ced5c51ca1815c1f75cdc5073c6f2fbcc (commit) from 442f5ddc7e78a1d8188bff82c0464c8dfb4eaff7 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 7a8ed9f71aca3d788d1450d42168ea51cd3d5ca4 Author: Phil Carns <[email protected]> Date: Thu Feb 10 12:46:18 2011 -0500 bug fixes to thread-per-txn model commit e67d6b5687b12a823500c743905c6aed1537eddb Author: Phil Carns <[email protected]> Date: Thu Feb 10 11:44:39 2011 -0500 zeroconf hook for thread-per-txn model commit 16ae957ced5c51ca1815c1f75cdc5073c6f2fbcc Author: Phil Carns <[email protected]> Date: Thu Feb 10 11:42:24 2011 -0500 first cut at thread-per-txn model, untested ----------------------------------------------------------------------- Summary of changes: code/src/common/triton-hash.h | 7 + .../prototype/bdb-resource/bdb-resource.c | 225 +++++++++++++++++--- .../prototype/bdb-resource/bdb-resource.hae | 1 + 3 files changed, 198 insertions(+), 35 deletions(-) Diff of changes: diff --git a/code/src/common/triton-hash.h b/code/src/common/triton-hash.h index 25b7ded..f8a8feb 100644 --- a/code/src/common/triton-hash.h +++ b/code/src/common/triton-hash.h @@ -373,6 +373,13 @@ static inline int triton_hash_string_hash(const void *k, int table_size) return result; } +static inline int triton_hash_ptr_hash(const void *k, int table_size) +{ + uint32_t pc = 0, pb = 0; + bj_hashlittle2(k, sizeof(k), &pc, &pb); + return pc & (table_size - 1); +} + #endif /* TRITON_HASH_H */ /* diff --git a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c index 09b23ca..1b9d1fb 100644 --- a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c +++ b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c @@ -31,6 +31,7 @@ static ae_opcache_t bdb_opcache; static ae_ops_t bdb_poll_oplist; static ae_ops_t bdb_thread_oplist; static triton_mutex_t bdb_mutex = TRITON_MUTEX_INITIALIZER; +static struct triton_hash_table* txn_thread_table = NULL;; static int triton_bdb_resource_id; static triton_debug_mask_t bdb_r_mask; @@ -44,6 +45,17 @@ static const char *flush_progress_mode_confkey = BDB_CONFKEY_PREFIX ".flush_prog static int txn_progress_mode = TRITON_BDB_PROG_THREAD_PER_OP; static int flush_progress_mode = TRITON_BDB_PROG_THREAD_PER_OP; +/* maps txns to threads in the thread-per-txn model */ +struct txn_thread_entry +{ + DB_TXN* txn; + pthread_t tid; + ae_ops_t oplist; + triton_mutex_t mutex; + triton_cond_t cond; + triton_list_link_t hash_link; +}; + struct bdb_op { ae_op_id_t op_id; @@ -144,6 +156,67 @@ static int txn_checkpoint_work_fn(struct ae_op* op); static int log_flush_work_fn(struct ae_op* op); static triton_ret_t progress_mode_updater(const char *key, const char *value); __attribute__((constructor)) void triton_bdb_init_register(void); +static triton_ret_t start_detached_thread(pthread_t* tid, void *(*start_routine)(void*), void * arg); +static int txn_compare(const void *key, struct triton_hash_link *hash_link); + +static void* thread_per_txn_fn(void* foo) +{ + struct txn_thread_entry* tmp_entry = foo; + struct ae_op* op = NULL; + struct bdb_op* b_op = NULL; + int done = 0; + + while(1) + { + triton_mutex_lock(&tmp_entry->mutex); + while(ae_ops_empty(&tmp_entry->oplist)) + { + pthread_cond_wait(&tmp_entry->cond, &tmp_entry->mutex); + } + op = ae_ops_dequeue(&tmp_entry->oplist); + b_op = ae_op_entry(op, struct bdb_op, op); + /* if we see a commit or abort, then we need to get rid of this + * entry and destroy the thread once we are done + */ + if(b_op->work_fn == txn_commit_work_fn || b_op->work_fn == txn_abort_work_fn) + { + triton_list_del(&tmp_entry->hash_link); + done = 1; + } + triton_mutex_unlock(&tmp_entry->mutex); + + /* spin on op worker */ + while(b_op->work_fn(op) != 1); + + /* call cleanup function if present */ + if(b_op->cleanup_fn) + b_op->cleanup_fn(op); + + /* trigger completion of the operation */ + ae_opcache_complete_op(bdb_opcache, op, triton_ret_t, b_op->return_code); + /* request poll; not because ae_poll() needs to drive any work, but + * simply because we want it to break out if any callers are blocking on + * it + */ + ae_resource_request_poll(NULL, triton_bdb_resource_id); + + if(done) + { + /* no more work to do on this txn; clean up and exit thread */ + triton_mutex_lock(&bdb_mutex); + triton_mutex_unlock(&bdb_mutex); + + assert(ae_ops_empty(&tmp_entry->oplist)); + triton_mutex_destroy(&tmp_entry->mutex); + triton_cond_destroy(&tmp_entry->cond); + free(tmp_entry); + return(NULL); + } + } + + return(NULL); +} + static void* thread_fn(void* foo) { @@ -174,7 +247,7 @@ static void* thread_fn(void* foo) return(NULL); } -/* equivalent to bdb_launch_op(), except that it executes the operation i +/* equivalent to bdb_launch_op(), except that it executes the operation * immediately rather than asynchronously */ static void bdb_launch_op_here(struct ae_op *op) @@ -194,12 +267,14 @@ static void bdb_launch_op_here(struct ae_op *op) return; } -static triton_ret_t bdb_launch_op(struct ae_op *op, enum bdb_progress_mode mode) +static triton_ret_t bdb_launch_op(struct ae_op *op, enum bdb_progress_mode mode, DB_TXN* txn) { int ret; pthread_t tid; - pthread_attr_t attr; struct bdb_op *b_op; + struct triton_hash_link* tmp_link; + struct txn_thread_entry* tmp_entry; + triton_ret_t tret; b_op = ae_op_entry(op, struct bdb_op, op); @@ -214,6 +289,55 @@ static triton_ret_t bdb_launch_op(struct ae_op *op, enum bdb_progress_mode mode) ae_resource_request_poll(NULL, triton_bdb_resource_id); triton_mutex_unlock(&bdb_mutex); break; + case TRITON_BDB_PROG_THREAD_PER_TXN: + triton_mutex_lock(&bdb_mutex); + tmp_link = triton_hash_search(txn_thread_table, txn); + if(!tmp_link) + { + tmp_entry = malloc(sizeof(*tmp_entry)); + if(!tmp_entry) + { + triton_mutex_unlock(&bdb_mutex); + return(TRITON_ERR_NOMEM); + } + tmp_entry->txn = txn; + ae_ops_init(&tmp_entry->oplist); + triton_mutex_init(&tmp_entry->mutex, NULL); + triton_cond_init(&tmp_entry->cond, NULL); + triton_list_link_clear(&tmp_entry->hash_link); + triton_hash_add(txn_thread_table, txn, &tmp_entry->hash_link); + triton_mutex_unlock(&bdb_mutex); + + triton_mutex_lock(&tmp_entry->mutex); + ae_ops_enqueue(op, &tmp_entry->oplist); + triton_mutex_unlock(&tmp_entry->mutex); + + tret = start_detached_thread(&tmp_entry->tid, + thread_per_txn_fn, tmp_entry); + if(tret != TRITON_SUCCESS) + { + /* TODO: error handling */ + return(tret); + } + } + else + { + triton_mutex_unlock(&bdb_mutex); + /* NOTE: once we have found the entry we don't have to hold + * the mutex any more; the thread won't delete this entry + * until its processed an abort or commit + */ + tmp_entry = triton_list_get_entry(tmp_link, + struct txn_thread_entry, + hash_link); + + /* add op to queue and signal thread */ + triton_mutex_lock(&tmp_entry->mutex); + ae_ops_enqueue(op, &tmp_entry->oplist); + pthread_cond_signal(&tmp_entry->cond); + triton_mutex_unlock(&tmp_entry->mutex); + } + break; case TRITON_BDB_PROG_THREAD_PER_OP: /* put on the queue */ triton_mutex_lock(&bdb_mutex); @@ -221,28 +345,12 @@ static triton_ret_t bdb_launch_op(struct ae_op *op, enum bdb_progress_mode mode) triton_mutex_unlock(&bdb_mutex); /* launch a thread to do the work */ - ret = pthread_attr_init(&attr); - if(ret != 0) + tret = start_detached_thread(&tid, thread_fn, op); + if(tret != TRITON_SUCCESS) { ae_ops_del(op); - return(triton_ret_from_bdb(ret)); + return(tret); } - ret = pthread_attr_setdetachstate(&attr, - PTHREAD_CREATE_DETACHED); - if(ret != 0) - { - ae_ops_del(op); - pthread_attr_destroy(&attr); - return(triton_ret_from_bdb(ret)); - } - ret = pthread_create(&tid, &attr, thread_fn, op); - if(ret != 0) - { - ae_ops_del(op); - pthread_attr_destroy(&attr); - return(triton_ret_from_bdb(ret)); - } - pthread_attr_destroy(&attr); break; default: return(TRITON_ERR_INVAL); @@ -319,7 +427,7 @@ triton_ret_t bdb_init(void) ret = triton_zeroconf_register(txn_progress_mode_confkey, "THREAD_PER_OP", progress_mode_updater, - "Progress mode for BDB transactions (THREAD_PER_OP"); + "Progress mode for BDB transactions (THREAD_PER_OP or THREAD_PER_TXN"); if(ret != TRITON_SUCCESS && ret != TRITON_ERR_EXIST) { return(ret); @@ -343,6 +451,12 @@ triton_ret_t bdb_init(void) return(ret); } + txn_thread_table = triton_hash_init(txn_compare, triton_hash_ptr_hash, 200); + if(!txn_thread_table) + { + return(TRITON_ERR_NOMEM); + } + ret = ae_resource_register(&triton_bdb_resource, &triton_bdb_resource_id); if(ret != TRITON_SUCCESS) { @@ -394,7 +508,7 @@ ae_define_post(triton_ret_t, bdb_txn_begin, *__ae_op_id = b_op->op_id; /* hard code this one; it shouldn't ever block on locks or I/O */ - bdb_launch_op(op, TRITON_BDB_PROG_IN_PLACE); + bdb_launch_op(op, TRITON_BDB_PROG_IN_PLACE, NULL); return(TRITON_SUCCESS); } @@ -419,7 +533,7 @@ ae_define_post(triton_ret_t, bdb_log_flush, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, flush_progress_mode); + bdb_launch_op(op, flush_progress_mode, NULL); return(TRITON_SUCCESS); } @@ -466,7 +580,7 @@ ae_define_post(triton_ret_t, bdb_txn_checkpoint, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, flush_progress_mode); + bdb_launch_op(op, flush_progress_mode, NULL); return(TRITON_SUCCESS); } @@ -533,7 +647,7 @@ ae_define_post(triton_ret_t, bdb_txn_commit, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, tid); return(TRITON_SUCCESS); } @@ -575,7 +689,7 @@ ae_define_post(triton_ret_t, bdb_txn_abort, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, tid); return(TRITON_SUCCESS); } @@ -630,7 +744,7 @@ ae_define_post(triton_ret_t, bdb_get, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, txnid); return(TRITON_SUCCESS); } @@ -680,7 +794,7 @@ ae_define_post(triton_ret_t, bdb_cursor_get, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, DBcursor->txn); return(TRITON_SUCCESS); } @@ -725,7 +839,7 @@ ae_define_post(triton_ret_t, bdb_cursor_del, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, DBcursor->txn); return(TRITON_SUCCESS); } @@ -766,7 +880,7 @@ ae_define_post(triton_ret_t, bdb_cursor_close, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, DBcursor->txn); return(TRITON_SUCCESS); } @@ -814,7 +928,7 @@ ae_define_post(triton_ret_t, bdb_put, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, txnid); return(TRITON_SUCCESS); } @@ -864,7 +978,7 @@ ae_define_post(triton_ret_t, bdb_del, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, txnid); return(TRITON_SUCCESS); } @@ -913,7 +1027,7 @@ ae_define_post(triton_ret_t, bdb_cursor, b_op->cleanup_fn = NULL; *__ae_op_id = b_op->op_id; - bdb_launch_op(op, txn_progress_mode); + bdb_launch_op(op, txn_progress_mode, txnid); return(TRITON_SUCCESS); } @@ -967,6 +1081,8 @@ static triton_ret_t progress_mode_updater(const char *key, const char *value) { if(!strcmp(value, "THREAD_PER_OP")) txn_progress_mode = TRITON_BDB_PROG_THREAD_PER_OP; + else if(!strcmp(value, "THREAD_PER_TXN")) + txn_progress_mode = TRITON_BDB_PROG_THREAD_PER_TXN; else return(TRITON_ERR_INVAL); } @@ -992,6 +1108,45 @@ static triton_ret_t progress_mode_updater(const char *key, const char *value) return TRITON_SUCCESS; } +static int txn_compare(const void *key, struct triton_hash_link *hash_link) +{ + struct txn_thread_entry *entry; + + entry = triton_hash_get_entry(hash_link, struct txn_thread_entry, hash_link); + if(key == entry->txn) + return(1); + else + return(0); +} + +static triton_ret_t start_detached_thread(pthread_t* tid, void *(*start_routine)(void*), void * arg) +{ + int ret; + pthread_attr_t attr; + + ret = pthread_attr_init(&attr); + if(ret != 0) + { + return(triton_ret_from_bdb(ret)); + } + ret = pthread_attr_setdetachstate(&attr, + PTHREAD_CREATE_DETACHED); + if(ret != 0) + { + pthread_attr_destroy(&attr); + return(triton_ret_from_bdb(ret)); + } + ret = pthread_create(tid, &attr, start_routine, arg); + if(ret != 0) + { + pthread_attr_destroy(&attr); + return(triton_ret_from_bdb(ret)); + } + pthread_attr_destroy(&attr); + + return(TRITON_SUCCESS); +} + /* * Local variables: * c-indent-level: 4 diff --git a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae index 4a6c857..c976481 100644 --- a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae +++ b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae @@ -23,6 +23,7 @@ enum bdb_progress_mode TRITON_BDB_PROG_POLL, /**< requires active polling by client */ TRITON_BDB_PROG_THREAD_PER_OP, /**< spawns new thread for each op */ TRITON_BDB_PROG_IN_PLACE, /**< immediate execution */ + TRITON_BDB_PROG_THREAD_PER_TXN,/**< spawns new thread for each txn */ TRITON_BDB_PROG_MAX, /**< invalid mode */ }; hooks/post-receive -- Triton Repository