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 "". The branch, trac-323-sos has been updated via d929b66f24c75ce386a9cd780466db72a04ece1e (commit) from 0622aa37a508132b868e2a3b8f3fc7e96ac9114d (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 d929b66f24c75ce386a9cd780466db72a04ece1e Author: John Jenkins <[email protected]> Date: Thu Mar 19 14:15:02 2015 -0500 initial record 'list IO' interface (untested) input: contiguous buffer, list of record IDs / record sizes ----------------------------------------------------------------------- Summary of changes: code/src/asg/asg-internal.ae | 108 +++++- code/src/asg/asg-internal.hae | 30 +- code/src/asg/asg-triton-ext.c | 9 +- code/src/asg/asg.c | 5 +- code/src/replicated-osd/rosd-read.ae | 15 + code/src/replicated-osd/rosd-write.ae | 762 ++++++++++++++++++++++++++++++++- code/src/replicated-osd/rosd.hae | 27 ++ 7 files changed, 908 insertions(+), 48 deletions(-) Diff of changes: diff --git a/code/src/asg/asg-internal.ae b/code/src/asg/asg-internal.ae index 2e5fb12..29c1556 100644 --- a/code/src/asg/asg-internal.ae +++ b/code/src/asg/asg-internal.ae @@ -123,7 +123,7 @@ __blocking int asg_i_write ( asg_flags_t flags, asg_update_id_t version_condition, asg_update_id_t * new_version, - const void * data, + void * data, asg_size_t * transferred) { triton_ret_t tret; @@ -154,6 +154,7 @@ __blocking int asg_i_write ( rosd_flags, version_condition, new_version, replication_factor); + /* TODO: handle short writes? */ if (triton_error_equal(tret, TRITON_ERR_CONDITIONAL)) aret = ASG_ERR_UPDATE_ID; else if (triton_is_error(tret)) @@ -182,7 +183,7 @@ __blocking int asg_i_read_sequence ( asg_size_t record_expected_len, asg_flags_t flags, asg_update_id_t update_id_condition, - const void * buf, + void * buf, asg_record_info_t * record_info) { triton_ret_t tret; @@ -749,7 +750,56 @@ __blocking int asg_i_read_set( asg_size_t * num_record_infos, asg_record_info_t * record_info) { - return ASG_ERR_OTHER; + triton_ret_t tret; + int aret; + uint32_t rosd_flags; + struct rosd_record_info *rinfo; + asg_size_t i; + + if (instance != 1) + return ASG_ERR_OTHER; + + rosd_flags = convert_flags_to_rosd(flags); + if (rosd_flags == -1 || rosd_flags == ROSD_COND_AUTO) + return ASG_ERR_INVAL; + + rinfo = malloc(sizeof(struct rosd_record_info) * num_records); + if (rinfo == NULL) + return ASG_ERR_OTHER; + + /* TODO: use safer casts */ + tret = remote_triton_rpc_rosd_read_set( + (uint64_t) container, + (uint64_t) object, + (uint64_t) fork, + (uint64_t) num_records, + (const uint64_t*) record_ids, + rosd_flags, + (uint64_t) update_id_condition, + buf, + buf_size, + (uint64_t*) num_record_infos, + rinfo); + + /* TODO: handle short reads? */ + if (triton_error_equal(tret, TRITON_ERR_CONDITIONAL)) + aret = ASG_ERR_UPDATE_ID; + else if (triton_is_error(tret)) { + triton_error_print(tret, "remote_triton_rpc_rosd_read_set"); + aret = ASG_ERR_OTHER; + } + else { + for (i = 0; i < *num_record_infos; i++) { + record_info[i].record_id = rinfo[i].record_id; + record_info[i].seq_len = rinfo[i].num_records; + record_info[i].record_len = rinfo[i].record_len; + record_info[i].record_update_id = rinfo[i].record_update_id; + } + aret = ASG_SUCCESS; + } + + triton_error_destroy(tret); + return aret; } __blocking int asg_i_write_set( @@ -759,15 +809,59 @@ __blocking int asg_i_write_set( asg_object_id_t object, asg_fork_id_t fork, asg_size_t num_records, - const asg_record_id_t * record_ids, - const asg_size_t * record_sizes, + asg_record_id_t * record_ids, + asg_size_t * record_sizes, asg_flags_t flags, asg_update_id_t update_id_condition, asg_update_id_t * new_update_id, - const void * buf, + void * buf, asg_size_t * records_written) { - return ASG_ERR_OTHER; + triton_ret_t tret; + int aret; + + uint32_t rosd_flags; + + const char * replication_factor_env; + uint32_t replication_factor; + + if (instance != 1) + return ASG_ERR_OTHER; + + /* hack: use environment variable to control the replication factor */ + replication_factor = 0; + replication_factor_env = getenv("TRITON_ASG_RF"); + if (replication_factor_env){ + if (0 == sscanf(replication_factor_env, "%d", &replication_factor)){ + /* simply ignore rather than returning something of interest */ + replication_factor = 0; + } + } + + rosd_flags = convert_flags_to_rosd(flags); + if (rosd_flags == -1) { + return ASG_ERR_INVAL; + } + + tret = remote_triton_rpc_rosd_write_set(container, object, fork, + num_records, record_ids, record_sizes, rosd_flags, + update_id_condition, new_update_id, buf, records_written, + replication_factor); + + /* TODO: how to handle short writes? */ + if (triton_error_equal(tret, TRITON_ERR_CONDITIONAL)) + aret = ASG_ERR_UPDATE_ID; + else if (triton_is_error(tret)) { + triton_error_print(tret, "remote_triton_rpc_rosd_write "); + triton_error_destroy(tret); + aret = ASG_ERR_OTHER; + } + else { + *records_written = num_records; + aret = ASG_SUCCESS; + } + + return aret; } /* diff --git a/code/src/asg/asg-internal.hae b/code/src/asg/asg-internal.hae index aa449d1..8c8fccd 100644 --- a/code/src/asg/asg-internal.hae +++ b/code/src/asg/asg-internal.hae @@ -43,7 +43,7 @@ __blocking int asg_i_write ( asg_flags_t flags, asg_update_id_t version_condition, asg_update_id_t * new_version, - const void * data, + void * data, asg_size_t * transferred); __blocking int asg_i_read_sequence ( @@ -57,7 +57,7 @@ __blocking int asg_i_read_sequence ( asg_size_t record_expected_len, asg_flags_t flags, asg_update_id_t update_id_condition, - const void * buf, + void * buf, asg_record_info_t * record_info); __blocking int asg_i_read_one ( @@ -173,19 +173,19 @@ __blocking int asg_i_read_set( asg_record_info_t * record_info); __blocking int asg_i_write_set( - asg_instance_t instance, - asg_location_t location, - asg_container_id_t container, - asg_object_id_t object, - asg_fork_id_t fork, - asg_size_t num_records, - const asg_record_id_t * record_ids, - const asg_size_t * record_sizes, - asg_flags_t flags, - asg_update_id_t update_id_condition, - asg_update_id_t * new_update_id, - const void * buf, - asg_size_t * records_written); + asg_instance_t instance, + asg_location_t location, + asg_container_id_t container, + asg_object_id_t object, + asg_fork_id_t fork, + asg_size_t num_records, + asg_record_id_t * record_ids, + asg_size_t * record_sizes, + asg_flags_t flags, + asg_update_id_t update_id_condition, + asg_update_id_t * new_update_id, + void * buf, + asg_size_t * records_written); #endif diff --git a/code/src/asg/asg-triton-ext.c b/code/src/asg/asg-triton-ext.c index 1d72065..1b39671 100644 --- a/code/src/asg/asg-triton-ext.c +++ b/code/src/asg/asg-triton-ext.c @@ -79,6 +79,8 @@ int asg_read_set( return aret; } +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wcast-qual" /* write a set of records with possibly non-consecutive IDs */ int asg_write_set( asg_instance_t instance, @@ -99,8 +101,10 @@ int asg_write_set( asg_size_t i; ASG_HANDLE_AE_DEFAULT(asg_i_write_set, aret, instance, location, - container, object, fork, num_records, record_ids, record_sizes, - flags, update_id_condition, new_update_id, buf, records_written) + container, object, fork, num_records, + (asg_record_id_t*) record_ids, (asg_size_t*) record_sizes, + flags, update_id_condition, new_update_id, (void*) buf, + records_written) total_xfer = 0; if (aret == ASG_SUCCESS) { @@ -111,6 +115,7 @@ int asg_write_set( END_PROF(aret, "xfer:%lu", total_xfer) return aret; } +#pragma GCC diagnostic pop /* * Local variables: diff --git a/code/src/asg/asg.c b/code/src/asg/asg.c index b389828..ba8af25 100644 --- a/code/src/asg/asg.c +++ b/code/src/asg/asg.c @@ -112,6 +112,8 @@ int asg_read ( } #endif +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wcast-qual" int asg_write ( asg_instance_t instance, asg_location_t location, @@ -129,10 +131,11 @@ int asg_write ( { ASG_HANDLE_AE_DEFAULT(asg_i_write, aret, instance, location, container, object, fork, start_record, recordcount, recordlen, flags, - version_condition, new_version, data, transferred) + version_condition, new_version, (void*) data, transferred) END_PROF(aret, "xfer:%lu", *transferred) return aret; } +#pragma GCC diagnostic pop int asg_punch ( asg_instance_t instance, diff --git a/code/src/replicated-osd/rosd-read.ae b/code/src/replicated-osd/rosd-read.ae index de5e2f0..6f141a7 100644 --- a/code/src/replicated-osd/rosd-read.ae +++ b/code/src/replicated-osd/rosd-read.ae @@ -221,6 +221,21 @@ __blocking triton_ret_t remote_triton_rpc_rosd_read_one( return tret; } +__blocking triton_ret_t remote_triton_rpc_rosd_read_set( + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + const uint64_t * record_ids, + uint32_t flags, + uint64_t update_id_condition, + void * buf, + size_t buf_size, + uint64_t * num_record_infos, + struct rosd_record_info * record_info) { + return TRITON_ERR_NOT_IMPLEMENTED; +} + /* TODO: merge this with other functions */ static __blocking triton_ret_t rosd_read_one_local_storage( diff --git a/code/src/replicated-osd/rosd-write.ae b/code/src/replicated-osd/rosd-write.ae index 6985897..92da682 100644 --- a/code/src/replicated-osd/rosd-write.ae +++ b/code/src/replicated-osd/rosd-write.ae @@ -25,11 +25,13 @@ extern int64_t rosd_write_xfer_pipeline_depth; extern struct hoss_ctx* g_hoss_ctx; extern int64_t rosd_write_buffer_size; -/* Mercury RPC structures for rsos_write */ +extern struct buffer_mgmt_instance* rosd_write_buffers; + + +/* structures and declarations for rosd_write */ MERCURY_GEN_PROC(triton_rpc_rosd_write_out_t, ((triton_ret_t)(tret))\ ((uint64_t)(new_update_id))) - MERCURY_GEN_PROC(triton_rpc_rosd_write_in_t, ((uint64_t)(container))\ ((uint64_t)(object))\ @@ -42,8 +44,6 @@ MERCURY_GEN_PROC(triton_rpc_rosd_write_in_t, ((uint32_t)(expected_position))\ ((uint32_t)(replication_factor))\ ((hg_bulk_t)(bulk_handle))) - -extern struct buffer_mgmt_instance* rosd_write_buffers; static hg_id_t rpc_rosd_write_id; static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle); static hg_return_t triton_rpc_rosd_write_handler(hg_handle_t handle); @@ -61,9 +61,6 @@ static __blocking triton_ret_t __remote_triton_rpc_rosd_write( uint64_t *new_update_id, uint32_t expected_position, uint32_t replication_factor); -static int is_usage_error_write(triton_ret_t tret); -static __blocking triton_ret_t resolve_replication_factor( - triton_rpc_rosd_write_in_t *in); static __blocking triton_ret_t rosd_write_do_work( na_addr_t next_addr, char* buffer, @@ -72,7 +69,6 @@ static __blocking triton_ret_t rosd_write_do_work( uint64_t recordcount, triton_rpc_rosd_write_in_t* in, uint64_t *new_update_id); - static __blocking triton_ret_t rosd_write_local_storage( uint64_t container, uint64_t object, @@ -85,6 +81,67 @@ static __blocking triton_ret_t rosd_write_local_storage( uint64_t update_id_condition, uint64_t *new_update_id); +/* structures and declarations for rosd_write_set */ +MERCURY_GEN_PROC(triton_rpc_rosd_write_set_out_t, + ((triton_ret_t)(tret)) \ + ((uint64_t)(new_update_id))) +MERCURY_GEN_PROC(triton_rpc_rosd_write_set_in_t, + ((uint64_t)(container)) \ + ((uint64_t)(object)) \ + ((uint64_t)(fork)) \ + ((uint64_t)(num_records)) \ + ((uint64_t)(update_id_condition)) \ + ((uint32_t)(flags)) \ + ((uint32_t)(expected_position)) \ + ((uint32_t)(replication_factor)) \ + ((hg_bulk_t)(bulk_handle))) +static hg_id_t rpc_rosd_write_set_id; +static __blocking triton_ret_t triton_rpc_rosd_write_set(hg_handle_t handle); +static hg_return_t triton_rpc_rosd_write_set_handler(hg_handle_t handle); +static __blocking triton_ret_t __remote_triton_rpc_rosd_write_set( + na_addr_t addr, + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + uint32_t flags, + uint64_t update_id_condition, + uint64_t * new_update_id, + void * buf, + uint64_t * records_written, + uint32_t expected_position, + uint32_t replication_factor); +static __blocking triton_ret_t rosd_write_set_do_work( + na_addr_t next_addr, + void * buffer, + int my_position, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + triton_rpc_rosd_write_set_in_t * in, + uint64_t *new_update_id); +static __blocking triton_ret_t rosd_write_set_local_storage( + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + void * buffer, + uint32_t flags, + uint64_t update_id_condition, + uint64_t *new_update_id); + +static int is_usage_error_write(triton_ret_t tret); +static __blocking triton_ret_t resolve_replication_factor( + uint64_t container, + uint64_t object, + uint32_t expected_position, + uint32_t * replication_factor, + uint32_t * flags); + __blocking triton_ret_t remote_triton_rpc_rosd_write( uint64_t container, uint64_t object, @@ -117,6 +174,37 @@ __blocking triton_ret_t remote_triton_rpc_rosd_write( } +__blocking triton_ret_t remote_triton_rpc_rosd_write_set( + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + uint32_t flags, + uint64_t update_id_condition, + uint64_t * new_update_id, + void * buf, + uint64_t * records_written, + uint32_t replication_factor) +{ + triton_ret_t tret; + na_addr_t addr; + int position; + + tret = triton_oid_to_addrs(object, 1, &position, &addr); + if (triton_is_error(tret)) + return tret; + + tret = __remote_triton_rpc_rosd_write_set(addr, container, object, fork, + num_records, record_ids, record_sizes, flags, update_id_condition, + new_update_id, buf, records_written, 0, replication_factor); + + triton_mercury_addr_free(addr); + + return tret; +} + static __blocking triton_ret_t rosd_handle_write_segment( hg_handle_t handle, na_addr_t src_addr, @@ -181,7 +269,84 @@ static __blocking triton_ret_t rosd_handle_write_segment( return(tret); } - + +static __blocking triton_ret_t rosd_handle_write_set_segment( + hg_handle_t handle, + na_addr_t src_addr, + na_addr_t next_addr, + int my_position, + triton_rpc_rosd_write_set_in_t* in, + uint64_t num_records, + uint64_t *record_ids, + uint64_t *record_sizes, + int64_t remote_offset, + uint64_t *new_update_id) +{ + void *buffer = NULL; + triton_ret_t tret; + int64_t outsize; + struct buffer_mgmt_token *token = NULL; + hg_bulk_t bulk_handle = HG_BULK_NULL; + int ret; + struct hg_info *info = NULL; + size_t bulk_size; + uint64_t i; + uint64_t total_rec_size = 0; + +#define FIN() \ + do { \ + if (token) buffer_mgmt_free(token); \ + if (bulk_handle != HG_BULK_NULL) HG_Bulk_free(bulk_handle); \ + } while(0) +#define END(_tret) \ + do { \ + FIN(); \ + return _tret; \ + } while(0) + + for (i = 0; i < num_records; i++) { + total_rec_size += record_sizes[i]; + } + + /* grab a buffer */ + tret = buffer_mgmt_alloc(rosd_write_buffers, total_rec_size, &outsize, + (char**)&buffer, &token); + if(tret != TRITON_SUCCESS) + END(tret); + assert(outsize == total_rec_size); + bulk_size = outsize; + + info = HG_Get_info(handle); + + /* set up and perform the bulk xfer */ + ret = HG_Bulk_create(info->hg_bulk_class, 1, &buffer, + &bulk_size, HG_BULK_WRITE_ONLY, &bulk_handle); + if (ret != HG_SUCCESS) { + triton_error_msg("HG_Bulk_create() failure\n"); + END(TRITON_ERR_NOMEM); + } + + ret = hg_rsc_bulk_transfer(info->bulk_context, HG_BULK_PULL, src_addr, + in->bulk_handle, remote_offset, bulk_handle, 0, bulk_size); + /* TODO: properly handle cancellations */ + assert(ret != AE_ERR_CANCELLED); + if (ret != HG_SUCCESS) { + triton_error_msg("hg_rsc_bulk_transfer() failure.\n"); + END(TRITON_ERR_UNKNOWN); + } + + tret = rosd_write_set_do_work(next_addr, buffer, my_position, num_records, + record_ids, record_sizes, in, new_update_id); + + FIN(); + return tret; +#undef FIN +#undef END +} + + + + static __blocking triton_ret_t rosd_write_run_pipeline( hg_handle_t handle, na_addr_t src_addr, @@ -282,8 +447,10 @@ static __blocking triton_ret_t rosd_write_run_pipeline( else if (is_update_id_set && *new_update_id != this_update_id) *new_update_id = ROSD_UPDATE_ID_MIXED; - else if (!is_update_id_set) + else if (!is_update_id_set) { *new_update_id = this_update_id; + is_update_id_set = 1; + } triton_mutex_unlock(&pipeline_mutex); } @@ -295,11 +462,179 @@ static __blocking triton_ret_t rosd_write_run_pipeline( return(out_tret); } +/* TODO: do better pipelining */ +static __blocking triton_ret_t rosd_write_set_run_pipeline( + hg_handle_t handle, + na_addr_t src_addr, + na_addr_t next_addr, + int my_position, + triton_rpc_rosd_write_set_in_t * in, + uint64_t * new_update_id) +{ + triton_ret_t out_tret; + triton_mutex_t pipeline_mutex; + int ret; + uint64_t current_record_pos = 0; + uint64_t current_bulk_offset = 0; + uint64_t current_remote_offset = 0; + int is_update_id_set = 0; + + /* HG specific vars */ + struct hg_info *info = NULL; + hg_bulk_t bulk_handle = HG_BULK_NULL; + + /* record bulk metadata (record sizes, ids) */ + uint64_t *rec_ptr; + uint64_t *rec_sizes; + uint64_t *rec_ids; + uint64_t rec_metadata_size; + size_t bulk_size; + +#define FIN() \ + do { \ + if (rec_ptr) free(rec_ptr); \ + if (bulk_handle != HG_BULK_NULL) HG_Bulk_free(bulk_handle); \ + triton_mutex_destroy(&pipeline_mutex); \ + } while (0) +#define END(_tret) \ + do { \ + FIN(); \ + return _tret; \ + } while (0) + + out_tret = TRITON_SUCCESS; + + triton_mutex_init(&pipeline_mutex, NULL); + + rec_metadata_size = 2 * in->num_records * sizeof(*rec_ptr); + + rec_ptr = malloc(rec_metadata_size); + if (rec_ptr) { + rec_ids = rec_ptr; + rec_sizes = rec_ptr + in->num_records; + } + else { + END(TRITON_ERR_NOMEM); + } + + bulk_size = rec_metadata_size; + ret = HG_Bulk_create(info->hg_bulk_class, 1, (void**)&rec_ptr, + &bulk_size, HG_BULK_WRITE_ONLY, &bulk_handle); + if (ret != HG_SUCCESS) { + triton_error_msg("HG_Bulk_create() failure.\n"); + END(TRITON_ERR_NOMEM); + } + + /* get the bulk metadata */ + hg_rsc_bulk_transfer(info->bulk_context, HG_BULK_PULL, src_addr, + in->bulk_handle, 0, bulk_handle, 0, rec_metadata_size); + /* TODO: properly handle cancellations */ + assert(ret != AE_ERR_CANCELLED); + if (ret != HG_SUCCESS) { + triton_error_msg("hg_rsc_bulk_transfer() failure.\n"); + END(TRITON_ERR_UNKNOWN); + } + + /* need to start off the remote offset as after the record metadata */ + current_remote_offset = rec_metadata_size; + + /* ready to start the pipeline */ + pwait + { + pprivate triton_ret_t b_tret; + pprivate int i, j; + pprivate int64_t pb_remote_offset; + pprivate uint64_t pb_record_count; + pprivate uint64_t pb_record_start_pos; + pprivate uint64_t pb_record_curr_pos; + pprivate uint64_t pb_total_size; + pprivate uint64_t pb_update_id; + for (i = 0; i < rosd_write_xfer_pipeline_depth; i++) { + pbranch + { + for (;;) { + /* calculate the next group of record indices + * [pb_record_start_pos, pb_record_curr_pos) */ + triton_mutex_lock(&pipeline_mutex); + if (current_record_pos >= in->num_records) { + triton_mutex_unlock(&pipeline_mutex); + pbreak; + } + pb_total_size = 0; + pb_record_count = 0; + pb_record_start_pos = current_record_pos; + pb_record_curr_pos = current_record_pos; + pb_remote_offset = current_remote_offset; + while (pb_record_curr_pos < in->num_records && + (pb_total_size + rec_sizes[pb_record_curr_pos] + <= rosd_write_buffer_size)) { + pb_total_size += rec_sizes[pb_record_curr_pos]; + pb_record_curr_pos++; + } + pb_record_count = pb_record_curr_pos - pb_record_start_pos; + if (pb_record_count == 0) { + triton_error_msg( + "record %lu's length of %lu is too large\n", + pb_record_curr_pos, + rec_sizes[pb_record_curr_pos]); + out_tret = TRITON_ERR_INVAL; + triton_mutex_unlock(&pipeline_mutex); + pbreak; + } + current_record_pos = pb_record_curr_pos; + current_remote_offset += pb_total_size; + + triton_mutex_unlock(&pipeline_mutex); + + /* we now have a lock on record indices + * [pb_record_start_pos, pb_record_curr_pos) */ + b_tret = rosd_handle_write_set_segment( + handle, + src_addr, + next_addr, + my_position, + in, + pb_record_count, + rec_ids + pb_record_start_pos, + rec_sizes + pb_record_start_pos, + pb_remote_offset, + &pb_update_id); + + triton_mutex_lock(&pipeline_mutex); + + if (triton_is_error(b_tret)) + out_tret = b_tret; + else if (is_update_id_set && + *new_update_id != pb_update_id) + *new_update_id = ROSD_UPDATE_ID_MIXED; + else if (!is_update_id_set) { + *new_update_id = pb_update_id; + is_update_id_set = 1; + } + + triton_mutex_unlock(&pipeline_mutex); + + } + } + } + } + + FIN(); + return out_tret; +#undef FIN +#undef END +} + /* This function validates the requested replication factor, storing it on * disk within the object if necessary + * replication_factor and flags are inout args */ static __blocking triton_ret_t resolve_replication_factor( - triton_rpc_rosd_write_in_t *in) + uint64_t container, + uint64_t object, + uint32_t expected_position, + uint32_t * replication_factor, + uint32_t * flags) { struct hoss_grp *grp; struct hoss_oid soid; @@ -313,8 +648,8 @@ static __blocking triton_ret_t resolve_replication_factor( /* go ahead and make the soid */ soid.nids = 3; soid.ids = soid_eids; - soid_eids[0] = in->container; - soid_eids[1] = in->object; + soid_eids[0] = container; + soid_eids[1] = object; soid_eids[2] = REP_FACTOR_FORK; /* TODO: check for "CANCELED" return code from hoss once supported. In @@ -328,7 +663,7 @@ static __blocking triton_ret_t resolve_replication_factor( return(TRITON_ERR_IO); } - if(in->expected_position == 0) + if(expected_position == 0) { /* This request is intended for the master. Validate replication * factor here against what is stored on disk (if present) @@ -340,9 +675,9 @@ static __blocking triton_ret_t resolve_replication_factor( { triton_debug(triton_dbg_rosd, "rosd_write() found on-disk replication factor of %d\n", ondisk_rep_factor); - if(in->replication_factor == 0) - in->replication_factor = ondisk_rep_factor; - else if(in->replication_factor != ondisk_rep_factor) + if(*replication_factor == 0) + *replication_factor = ondisk_rep_factor; + else if(*replication_factor != ondisk_rep_factor) { /* caller requested a replication factor that doesn't match * current replication for this object. @@ -354,22 +689,22 @@ static __blocking triton_ret_t resolve_replication_factor( } else { - in->flags |= ROSD_FLAG_FIRST_WRITE; + *flags |= ROSD_FLAG_FIRST_WRITE; } } - if(in->flags & ROSD_FLAG_FIRST_WRITE) + if(*flags & ROSD_FLAG_FIRST_WRITE) { /* store replication factor on disk */ /* this may happen on both master and replicas */ - if(in->replication_factor == 0) + if(*replication_factor == 0) { - in->replication_factor = rosd_default_replication_factor; + *replication_factor = rosd_default_replication_factor; ondisk_rep_factor = rosd_default_replication_factor; } else { - ondisk_rep_factor = in->replication_factor; + ondisk_rep_factor = *replication_factor; } triton_debug(triton_dbg_rosd, "rosd_write() writing on-disk replication factor of %d\n", ondisk_rep_factor); @@ -488,7 +823,8 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) out.tret = TRITON_ERR_CANCELED; while(triton_error_equal(out.tret, TRITON_ERR_CANCELED)) { - out.tret = resolve_replication_factor(&in); + out.tret = resolve_replication_factor(in.container, in.object, + in.expected_position, &in.replication_factor, &in.flags); } if(in.replication_factor > test_rep_factor) @@ -561,6 +897,118 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) } TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_write) +static __blocking triton_ret_t triton_rpc_rosd_write_set(hg_handle_t handle) +{ + triton_rpc_rosd_write_set_out_t out; + triton_rpc_rosd_write_set_in_t in; + int ret = 0; + na_addr_t next_addr; + na_addr_t* addr_array; + int my_position; + na_addr_t src_addr; + struct hg_info *info = NULL; + int test_rep_factor; + +#define FIN() \ + do {\ + triton_mercury_respond(handle, &out); \ + HG_Destroy(handle); \ + if (addr_array) free(addr_array); \ + } while (0) + +#define END(_server_tret) \ + do { \ + FIN(); \ + return _server_tret; \ + } while (0) + + triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_write_set()\n"); + + addr_array = NULL; + info = HG_Get_info(handle); + src_addr = info->addr; + + triton_mercury_get_input(handle, &in, &out); + + triton_debug(triton_dbg_rosd, + "rosd_write_set() with incoming replication factor of %d " + "and incoming expected position of %d\n", + in.replication_factor, in.expected_position); + + /* downstream replicas must have rep. factor defined for them */ + if (in.expected_position != 0 && in.replication_factor == 0) { + out.tret = TRITON_ERR_INVAL; + triton_error_msg("ERROR: non-primary server " + "expected nonzero replication factor for \n"); + END(TRITON_ERR_INVAL); + } + /* no writing to "null" container/object/forks */ + else if (in.container == 0 || in.object == 0 || in.fork == 0) { + out.tret = TRITON_ERR_INVAL; + END(TRITON_SUCCESS); + } + /* if we're the primary, use rep factor of 1 just for testing purposes */ + else if (in.expected_position == 0) + test_rep_factor = 1; + else + test_rep_factor = in.replication_factor; + + addr_array = malloc(test_rep_factor*sizeof(*addr_array)); + if(!addr_array) { + out.tret = TRITON_ERR_NOMEM; + END(TRITON_SUCCESS); + } + + out.tret = triton_oid_to_addrs(in.object, test_rep_factor, &my_position, + addr_array); + if (triton_is_error(out.tret)) + END(TRITON_SUCCESS); + else if (my_position != in.expected_position) { + out.tret = TRITON_ERR_WRONG_SERVER; + END(TRITON_SUCCESS); + } + + /* check that the replication factor is valid and store on disk if needed */ + /* TODO: fix error codes; need to differentiate hoss canceled (which + * means we should retry) from aesop canceled (which means the operation + * has been explicitly canceled by another thread, likely due to error + * cleanup) */ + /* TODO: don't loop infinitely */ + out.tret = TRITON_ERR_CANCELED; + while (triton_error_equal(out.tret, TRITON_ERR_CANCELED)) + out.tret = resolve_replication_factor(in.container, in.object, + in.expected_position, &in.replication_factor, &in.flags); + + if (in.replication_factor > test_rep_factor) { + /* earlier rep factor assumption wrong, look up mapping again */ + free(addr_array); + addr_array = malloc(in.replication_factor * sizeof(*addr_array)); + if (!addr_array) { + out.tret = TRITON_ERR_NOMEM; + END(TRITON_SUCCESS); + } + + out.tret = triton_oid_to_addrs(in.object, in.replication_factor, + &my_position, addr_array); + if (triton_is_error(out.tret)) + END(TRITON_SUCCESS); + } + + if (my_position < in.replication_factor - 1) + next_addr = addr_array[my_position + 1]; + else + next_addr = NA_ADDR_NULL; + + out.tret = rosd_write_set_run_pipeline(handle, src_addr, next_addr, + my_position, &in, &out.new_update_id); + + FIN(); + return TRITON_SUCCESS; +#undef FIN +#undef END +} +TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_write_set); + void triton_rpc_rosd_write_register(void) { hg_class_t* hg_class = NULL; @@ -572,6 +1020,11 @@ void triton_rpc_rosd_write_register(void) triton_rpc_rosd_write_in_t, triton_rpc_rosd_write_out_t, triton_rpc_rosd_write_handler); + rpc_rosd_write_set_id = + MERCURY_REGISTER(hg_class, "triton_rpc_rosd_write_set", + triton_rpc_rosd_write_set_in_t, + triton_rpc_rosd_write_set_out_t, + triton_rpc_rosd_write_handler); return; } @@ -622,6 +1075,73 @@ static __blocking triton_ret_t __remote_triton_rpc_rosd_write( return(tret); } +static __blocking triton_ret_t __remote_triton_rpc_rosd_write_set( + na_addr_t addr, + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + uint32_t flags, + uint64_t update_id_condition, + uint64_t * new_update_id, + void * buf, + uint64_t * records_written, + uint32_t expected_position, + uint32_t replication_factor) +{ + triton_rpc_rosd_write_set_out_t out; + triton_rpc_rosd_write_set_in_t in; + int ret; + triton_ret_t tret; + hg_handle_t handle = HG_HANDLE_NULL; + size_t bulk_size; + uint64_t i; + void * buffers[3]; + size_t buffer_sizes[3]; + + bulk_size = 0; + for (i = 0; i < num_records; i++) { + bulk_size += record_sizes[i]; + } + + in.container = container; + in.object = object; + in.fork = fork; + in.num_records = num_records; + in.flags = flags; + in.update_id_condition = update_id_condition; + in.expected_position = expected_position; + in.replication_factor = replication_factor; + in.bulk_handle = HG_BULK_NULL; + + /* set up the buffers for bulk transfer */ + buffers[0] = (void*) record_ids; + buffers[1] = (void*) record_sizes; + buffers[2] = (void*) buf; + + buffer_sizes[0] = num_records * sizeof(*record_ids); + buffer_sizes[1] = num_records * sizeof(*record_ids); + buffer_sizes[2] = bulk_size; + + triton_mercury_run_rpc_with_bulk(addr, rpc_rosd_write_set_id, handle, + &in, &out, in.bulk_handle, 3, buffers, buffer_sizes, 0, + HG_BULK_READ_ONLY); + + tret = triton_error_dup(out.tret); + + if (!triton_is_error(tret) || + triton_error_equal(tret, TRITON_ERR_CONDITIONAL)) + *new_update_id = out.new_update_id; + + HG_Free_output(handle, &out); + HG_Destroy(handle); + + return tret; +} + + static __blocking triton_ret_t rosd_write_do_work( na_addr_t next_addr, char* buffer, @@ -700,6 +1220,86 @@ static __blocking triton_ret_t rosd_write_do_work( return(tret); } +static __blocking triton_ret_t rosd_write_set_do_work( + na_addr_t next_addr, + void * buffer, + int my_position, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + triton_rpc_rosd_write_set_in_t * in, + uint64_t *new_update_id) +{ + + triton_ret_t local_tret; + triton_ret_t remote_tret = TRITON_SUCCESS; + triton_ret_t tret; + uint64_t new_update_id_local, new_update_id_remote; + uint64_t records_written_total = 0; + + /* right now only chained replication is supported */ + assert(!(in->flags & ROSD_FLAG_FANOUT)); + in->flags |= ROSD_FLAG_CHAIN; + + pwait + { + pprivate uint64_t records_written; + pbranch + { + /* forward on to peers if necessary */ + if(my_position < in->replication_factor-1) { + triton_debug(triton_dbg_rosd, + "ROSD forwarding write of oid %llu to %d'th server.\n", + llu(in->object), my_position+1); + + records_written = 0; + remote_tret = __remote_triton_rpc_rosd_write_set( + next_addr, in->container, in->object, in->fork, + num_records, record_ids, record_sizes, in->flags, + in->update_id_condition, new_update_id, buffer, + &records_written, my_position + 1, + in->replication_factor); + } + } + pbranch + { + triton_debug(triton_dbg_rosd, + "ROSD local write of oid %llu on %d'th server, " + "update_id %llu.\n", llu(in->object), my_position, + llu(in->update_id_condition)); + /* perform local operation */ + /* TODO: fix error codes; need to differentiate hoss canceled + * (which means we should retry) from aesop canceled (which means + * the operation has been explicitly canceled by another thread, + * likely due to error cleanup) + */ + local_tret = TRITON_ERR_CANCELED; + while(local_tret == TRITON_ERR_CANCELED) { + local_tret = rosd_write_set_local_storage( + in->container, in->object, in->fork, num_records, + record_ids, record_sizes, buffer, in->flags, + in->update_id_condition, &new_update_id_local); + } + } + } + + + if(my_position < in->replication_factor-1) + tret = interpret_errors(local_tret, remote_tret, my_position, + is_usage_error_write, !in->expected_position); + else + tret = interpret_error(local_tret, my_position, is_usage_error_write, + !in->expected_position); + + /* TODO: proper error checking for this */ + if (tret == TRITON_SUCCESS && my_position < in->replication_factor-1) { + assert(new_update_id_local == new_update_id_remote); + } + *new_update_id = new_update_id_local; + + return(tret); +} + /** * Checks the triton_ret_t to determine if the error code represents a * usage error rather than an a storage fault. Returns 1 if usage @@ -809,6 +1409,122 @@ static __blocking triton_ret_t rosd_write_local_storage( return(TRITON_SUCCESS); } +static __blocking triton_ret_t rosd_write_set_local_storage( + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + void * buffer, + uint32_t flags, + uint64_t update_id_condition, + uint64_t *new_update_id) +{ + struct hoss_grp *grp = NULL; + struct hoss_oid soid; + hoss_eid_t soid_eids[3]; + hoss_size_t *nbytes; + hoss_update_t update_in; + int ret, hoss_rc; + hoss_flags_t hoss_flags; + uint64_t i; + triton_ret_t tret; + + struct { + hoss_size_t nbytes; + int rc; + } * outputs; + +#define END(_tret)\ + do { \ + if (outputs) free(outputs); \ + if (grp) hoss_end(grp, 0); \ + return _tret; \ + } while (0) + + outputs = malloc(num_records * sizeof(*outputs)); + if (outputs == NULL) { + END(TRITON_ERR_NOMEM); + } + + /* TODO: handle canceled error code (or whatever hoss will report to + * indicate that a txn failed and should be retried) + */ + ret = hoss_begin(&grp, NULL, HOSS_MSYNC|HOSS_ORDERED|HOSS_DEFERRED, + g_hoss_ctx); + if(ret != 0) { + triton_error_msg("hoss_begin() failure (write)\n"); + END(TRITON_ERR_IO); + } + + soid.nids = 3; + soid.ids = soid_eids; + soid_eids[0] = container; + soid_eids[1] = object; + soid_eids[2] = fork; + + /* convert and sanity check the flags */ + hoss_flags = rosd_convert_flags_to_hoss(flags); + if (hoss_flags == -1) { + triton_error_msg( + "invalid ROSD flags passed in (should be caught earlier)\n"); + END(TRITON_ERR_INVAL); + } + + /* TODO: what to do when no update id is given */ + if (hoss_flags == HOSS_NONE) + update_in = 0; + else + update_in = update_id_condition; + + for (i = 0; i < num_records; i++) { + /* TODO: the update id arguments probably aren't right here... */ + ret = hoss_write(&soid, record_ids[i], 1, record_sizes[i], + hoss_flags, update_in, update_in, buffer, + grp, &outputs[i].nbytes, &outputs[i].rc); + if(ret != 0) { + triton_error_msg("hoss_write() failure, return code %d\n", ret); + END(TRITON_ERR_IO); + } + } + + ret = hoss_end(grp, 1); + grp = NULL; + + if (ret == HOSS_ECONDFAIL) { + END(TRITON_ERR_CONDITIONAL); + } + else if(ret != 0) { + triton_error_msg("hoss_end() failure (write), return code %d\n", ret); + END(TRITON_ERR_IO); + } + /* NOTE: output arguments to hoss_write (rc and nbytes) are not filled + * in until completion of hoss_end() + */ + else { + for (i = 0 ; i < num_records; i++) { + if (outputs[i].rc != 0 || outputs[i].nbytes != record_sizes[i]) { + triton_error_msg("hoss_write() failure, record %lu\n", + record_ids[i]); + END(TRITON_ERR_IO); + } + else if (outputs[i].nbytes != record_sizes[i]) { + triton_error_msg("hoss_write() failure (short write), " + "record %lu\n", record_ids[i]); + END(TRITON_ERR_IO); + } + } + } + + /* success */ + *new_update_id = update_in; + + return TRITON_SUCCESS; +#undef END +} + + /* * Local Variables: diff --git a/code/src/replicated-osd/rosd.hae b/code/src/replicated-osd/rosd.hae index 6d47f12..1a23317 100644 --- a/code/src/replicated-osd/rosd.hae +++ b/code/src/replicated-osd/rosd.hae @@ -102,6 +102,20 @@ __blocking triton_ret_t remote_triton_rpc_rosd_write( uint64_t *new_update_id, uint32_t replication_factor); +__blocking triton_ret_t remote_triton_rpc_rosd_write_set( + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + uint64_t * record_ids, + uint64_t * record_sizes, + uint32_t flags, + uint64_t update_id_condition, + uint64_t * new_update_id, + void * buf, + uint64_t * records_written, + uint32_t replication_factor); + __blocking triton_ret_t remote_triton_rpc_rosd_read_sequence( uint64_t container, uint64_t object, @@ -126,6 +140,19 @@ __blocking triton_ret_t remote_triton_rpc_rosd_read_one( uint64_t * record_len, uint64_t * record_update_id); +__blocking triton_ret_t remote_triton_rpc_rosd_read_set( + uint64_t container, + uint64_t object, + uint64_t fork, + uint64_t num_records, + const uint64_t * record_ids, + uint32_t flags, + uint64_t update_id_condition, + void * buf, + size_t buf_size, + uint64_t * num_record_infos, + struct rosd_record_info * record_info); + __blocking triton_ret_t rosd_pipeline_setup_one_buffer( struct buffer_mgmt_instance* buffer_instance, aesop_sem_t *sem, hooks/post-receive --