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, shared-lib has been updated via df1cc8d73f9e4446004ffb25b8debfb115dafb87 (commit) via 2c09c00fc8daa7e78d0f601dfe7d7e8809adb89f (commit) via 03b9c0f3895c25b8157bb5ef467d25b48be89c4b (commit) from 40fd462d6268eb9e645ec61b051bec81f19d5c05 (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 df1cc8d73f9e4446004ffb25b8debfb115dafb87 Author: John Jenkins <[email protected]> Date: Wed Mar 4 09:01:23 2015 -0600 tweaks to conditional test commit 2c09c00fc8daa7e78d0f601dfe7d7e8809adb89f Author: John Jenkins <[email protected]> Date: Thu Feb 26 13:26:43 2015 -0600 initial re-support for conditionals without txn cache, limited in what it can do (auto_update_id will hard-error), but at least seems to work for cond-all concurrent updates still need to be checked commit 03b9c0f3895c25b8157bb5ef467d25b48be89c4b Author: John Jenkins <[email protected]> Date: Thu Feb 26 09:22:01 2015 -0600 invalid argument return code ----------------------------------------------------------------------- Summary of changes: code/include/asg.h | 1 + code/src/asg/asg-internal.ae | 78 ++++++++++++----- code/src/replicated-osd/rosd-internal.hae | 4 + code/src/replicated-osd/rosd-read.ae | 63 ++++++++++++-- code/src/replicated-osd/rosd-write.ae | 87 +++++++++++++++---- code/src/replicated-osd/rosd.ae | 18 ++++ code/src/replicated-osd/rosd.hae | 21 +++++- code/tests/asg/test-asg-cond.c | 134 +++++++++++++++++------------ code/tests/asg/test-asg-simple.c | 2 +- 9 files changed, 299 insertions(+), 109 deletions(-) Diff of changes: diff --git a/code/include/asg.h b/code/include/asg.h index f73f303..59fdc94 100644 --- a/code/include/asg.h +++ b/code/include/asg.h @@ -166,6 +166,7 @@ typedef enum ASG_ERR_NOENT = 3, /**< entity does not exist */ ASG_ERR_BUFFER_SMALL = 4, /**< buffer too small to hold data */ ASG_ERR_RECORDLEN_INVAL = 5, /**< invalid record length */ + ASG_ERR_INVAL = 6, /**< invalid arguments */ ASG_ERR_OTHER = 54321 /**< all other failure modes */ } asg_ret_t; diff --git a/code/src/asg/asg-internal.ae b/code/src/asg/asg-internal.ae index c11cbdf..415cddf 100644 --- a/code/src/asg/asg-internal.ae +++ b/code/src/asg/asg-internal.ae @@ -34,6 +34,21 @@ void asg_callback (void *user_ptr, int ret) return; } +static uint32_t convert_flags_to_rosd(asg_flags_t flags) { + switch (flags) { + case ASG_COND_NONE: /* ASG_COND_UNCONDITIONAL currently the same */ + return ROSD_COND_UNCONDITIONAL; + case ASG_COND_UNTIL: /* currently unsupported */ + return -1; + case ASG_COND_ALL: + return ROSD_COND_ALL; + case ASG_AUTO_UPDATE_ID: /* currently unsupported */ + return -1; + default: /* unknown flags */ + return -1; + } +} + /** * should initialize all the server name and their address * first step, use options as the server address we want to connect @@ -112,21 +127,12 @@ __blocking int asg_i_write ( asg_size_t * transferred) { triton_ret_t tret; + int aret; + + uint32_t rosd_flags; const char * replication_factor_env; uint32_t replication_factor; - //build oid and get server info from it; - - /** - * last two args are: uint64_t txn_number, uint32_t flags; - * remote_triton_rpc_rosd_write does not provide the ability to check - * whether current condition satisfy the 'version_condition', we need to - * implement it before calling write function. - - printf("oid is %s\n", triton_uint128_to_string(oid)); - printf("fork_id: %ld\n", (uint64_t)fork_id); - printf("data: %s, has %d lengh with new version: %ld with flag:%d\n", data, recordlen, (uint64_t)new_version, flags); - */ /* hack: use environment variable to control the replication factor */ replication_factor = 0; @@ -138,21 +144,31 @@ __blocking int asg_i_write ( } } - tret = remote_triton_rpc_rosd_write(container, object, fork, - start_record, recordcount, recordlen, data, flags, version_condition, - new_version, replication_factor); + rosd_flags = convert_flags_to_rosd(flags); + if (rosd_flags == -1) { + return ASG_ERR_INVAL; + } - if (triton_is_error(tret)) + tret = remote_triton_rpc_rosd_write(container, object, fork, + start_record, recordcount, recordlen, data, + rosd_flags, version_condition, new_version, + replication_factor); + + 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); - return(ASG_ERR_OTHER); + aret = ASG_ERR_OTHER; + } + else { + *transferred = recordcount * recordlen; + aret = ASG_SUCCESS; } - triton_error_destroy(tret); - - *transferred = recordcount * recordlen; - return(ASG_SUCCESS); + triton_error_destroy(tret); + return aret; } __blocking int asg_i_read_sequence ( @@ -172,9 +188,15 @@ __blocking int asg_i_read_sequence ( triton_ret_t tret; int ret = ASG_SUCCESS; struct rosd_record_info rinfo; + uint32_t rosd_flags; 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; + } + tret = remote_triton_rpc_rosd_read_sequence( (uint64_t)container, (uint64_t)object, @@ -182,12 +204,14 @@ __blocking int asg_i_read_sequence ( (uint64_t)record_start, (uint64_t)record_count, (uint64_t)record_expected_len, - (uint32_t)flags, + rosd_flags, (uint64_t)update_id_condition, buf, &rinfo); - if (triton_is_error(tret)) { + if (triton_error_equal(tret, TRITON_ERR_CONDITIONAL)) + ret = ASG_ERR_UPDATE_ID; + else if (triton_is_error(tret)) { /* TODO: do we really want _print fns in the library? */ triton_error_print(tret, "remote_triton_rpc_rosd_read_sequence"); ret = ASG_ERR_OTHER; @@ -222,13 +246,19 @@ __blocking int asg_i_read_one ( uint64_t record_len; uint64_t record_update_id; int ret; + uint32_t rosd_flags; + + rosd_flags = convert_flags_to_rosd(flags); + if (rosd_flags == -1 || rosd_flags == ROSD_COND_AUTO) { + return ASG_ERR_INVAL; + } tret = remote_triton_rpc_rosd_read_one( container, object, fork, record, - flags, + rosd_flags, update_id_condition, buf, buf_size, diff --git a/code/src/replicated-osd/rosd-internal.hae b/code/src/replicated-osd/rosd-internal.hae index 76ebdd3..0302e78 100644 --- a/code/src/replicated-osd/rosd-internal.hae +++ b/code/src/replicated-osd/rosd-internal.hae @@ -7,6 +7,7 @@ */ #include "src/replicated-osd/rosd.hae" +#include "hoss.hae" /* default replication level for objects (set via zeroconf) */ extern uint32_t rosd_default_replication_factor; @@ -14,6 +15,9 @@ extern uint32_t rosd_default_replication_factor; /* default fork used to store replication factor in object */ #define REP_FACTOR_FORK UINT64_MAX +/* helper function to convert rosd conditionals into hoss conditionals */ +hoss_flags_t rosd_convert_flags_to_hoss(uint32_t flags); + triton_ret_t triton_oid_to_addrs( uint64_t oid, uint32_t replication_factor, diff --git a/code/src/replicated-osd/rosd-read.ae b/code/src/replicated-osd/rosd-read.ae index e174d3c..de5e2f0 100644 --- a/code/src/replicated-osd/rosd-read.ae +++ b/code/src/replicated-osd/rosd-read.ae @@ -180,7 +180,7 @@ __blocking triton_ret_t remote_triton_rpc_rosd_read_one( uint64_t update_id_condition, void * buf, size_t buf_size, - uint64_t record_len, + uint64_t * record_len, uint64_t * record_update_id) { triton_rpc_rosd_read_one_in_t in; @@ -209,6 +209,11 @@ __blocking triton_ret_t remote_triton_rpc_rosd_read_one( tret = triton_error_dup(out.tret); + if (!triton_is_error(tret)) { + *record_len = out.record_len; + *record_update_id = out.record_update_id; + } + triton_mercury_addr_free(addr); HG_Free_output(handle, &out); HG_Destroy(handle); @@ -240,6 +245,8 @@ static __blocking triton_ret_t rosd_read_one_local_storage( hoss_size_t nrecs; hoss_update_t update_info; struct hoss_record_info rec_info; + hoss_flags_t hoss_flags; + int do_commit; soid.nids = 3; soid.ids = soid_eids; @@ -247,6 +254,13 @@ static __blocking triton_ret_t rosd_read_one_local_storage( soid_eids[1] = object; soid_eids[2] = fork; + 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"); + return TRITON_ERR_INVAL; + } + /* TODO: handle canceled error code (or whatever hoss will report to * indicate that a txn failed and should be retried) */ @@ -254,26 +268,42 @@ static __blocking triton_ret_t rosd_read_one_local_storage( if(ret != 0) return(TRITON_ERR_IO); - ret = hoss_read(&soid, record, 1, HOSS_NONE, update_id_condition, buf, - buf_sz, &rec_info, 1, grp, &nbytes, &nrecs, &update_info, + ret = hoss_read(&soid, record, 1, hoss_flags, update_id_condition, buf, + buf_sz, &rec_info, 1, grp, record_len, &nrecs, &update_info, &hoss_rc); + do_commit = 0; if (ret != 0) tret = TRITON_ERR_IO; /* check for short buffer */ + else if (hoss_rc == HOSS_ECONDFAIL) { + /* TODO: have to "commit" here, else hoss/idb segfaults */ + tret = TRITON_ERR_CONDITIONAL; + do_commit = 1; + } else if (hoss_rc == ENOSPC) { + /* TODO: this also requires a "commit"? */ tret = TRITON_ERR_RECV_TOO_SMALL; + do_commit = 1; } /* check for other io errors */ else if (hoss_rc != 0) tret = TRITON_ERR_IO; /* check for nonexistent record */ - else if (nrecs == 0) + else if (nrecs == 0) { + /* TODO: this also requires a "commit"? */ + do_commit = 1; tret = TRITON_ERR_NOENT; - else + } + else { tret = TRITON_SUCCESS; + *new_update_id = rec_info.update_id; + do_commit = 1; + } - hoss_end(grp, 1); + ret = hoss_end(grp, do_commit); + if (ret != 0) + return TRITON_ERR_IO; return tret; } @@ -297,6 +327,7 @@ static __blocking triton_ret_t rosd_read_sequence_local_storage( int ret, hoss_rc; hoss_size_t n_recs; hoss_update_t update_info; + hoss_flags_t hoss_flags; /* NOTE: we provide two hoss_record_info structs here. In normal * operation we only expect 1 of them to be filled in. The second one @@ -306,6 +337,13 @@ static __blocking triton_ret_t rosd_read_sequence_local_storage( */ struct hoss_record_info rec_array[2]; + 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"); + return TRITON_ERR_INVAL; + } + /* TODO: handle canceled error code (or whatever hoss will report to * indicate that a txn failed and should be retried) */ @@ -321,15 +359,21 @@ static __blocking triton_ret_t rosd_read_sequence_local_storage( soid_eids[1] = object; soid_eids[2] = fork; - ret = hoss_read(&soid, record_start, record_count, HOSS_NONE, update_id_condition, buffer, record_count*record_expected_len, rec_array, 2, grp, &nbytes, &n_recs, &update_info, &hoss_rc); + ret = hoss_read(&soid, record_start, record_count, hoss_flags, update_id_condition, buffer, record_count*record_expected_len, rec_array, 2, grp, &nbytes, &n_recs, &update_info, &hoss_rc); /* check for fundamental I/O error */ - if(ret != 0 || hoss_rc != 0) + if(ret != 0) { - triton_error_msg("hoss_read() failure, return code %d\n", ret); + triton_error_msg("hoss_read() failure, return %d, hoss rc %d\n", ret, + hoss_rc); hoss_end(grp, 0); return(TRITON_ERR_IO); } + else if (hoss_rc == HOSS_ECONDFAIL) { + /* TODO: currently have to "commit" - segfaults otherwise */ + hoss_end(grp, 1); + return TRITON_ERR_CONDITIONAL; + } /* check for unexpected record size */ if(n_recs == 2 || (n_recs ==1 && rec_array[0].reclen != record_expected_len)) { @@ -352,6 +396,7 @@ static __blocking triton_ret_t rosd_read_sequence_local_storage( { assert(n_recs == 1); *records_read = rec_array[0].nrecs; + *new_update_id = rec_array[0].update_id; } else { diff --git a/code/src/replicated-osd/rosd-write.ae b/code/src/replicated-osd/rosd-write.ae index fb3c0f7..6985897 100644 --- a/code/src/replicated-osd/rosd-write.ae +++ b/code/src/replicated-osd/rosd-write.ae @@ -70,7 +70,8 @@ static __blocking triton_ret_t rosd_write_do_work( int my_position, uint64_t start_record, uint64_t recordcount, - triton_rpc_rosd_write_in_t* in); + triton_rpc_rosd_write_in_t* in, + uint64_t *new_update_id); static __blocking triton_ret_t rosd_write_local_storage( uint64_t container, @@ -124,7 +125,8 @@ static __blocking triton_ret_t rosd_handle_write_segment( triton_rpc_rosd_write_in_t* in, uint64_t start_record, int64_t remote_offset, - uint64_t recordcount) + uint64_t recordcount, + uint64_t *new_update_id) { char *buffer; triton_ret_t tret; @@ -172,7 +174,8 @@ static __blocking triton_ret_t rosd_handle_write_segment( my_position, start_record, recordcount, - in); + in, + new_update_id); buffer_mgmt_free(token); @@ -184,7 +187,8 @@ static __blocking triton_ret_t rosd_write_run_pipeline( na_addr_t src_addr, na_addr_t next_addr, int my_position, - triton_rpc_rosd_write_in_t* in + triton_rpc_rosd_write_in_t* in, + uint64_t *new_update_id ) { int64_t current_remote_offset = 0; @@ -192,6 +196,9 @@ static __blocking triton_ret_t rosd_write_run_pipeline( triton_ret_t out_tret = TRITON_SUCCESS; triton_mutex_t pipeline_mutex; uint64_t records_per_buffer; + int is_update_id_set; + *new_update_id = 0; + is_update_id_set = 0; triton_mutex_init(&pipeline_mutex, NULL); @@ -211,6 +218,7 @@ static __blocking triton_ret_t rosd_write_run_pipeline( pprivate uint64_t this_start_record; pprivate int64_t this_remote_offset; pprivate uint64_t this_recordcount; + pprivate uint64_t this_update_id; /* One pbranch for pipeline depth. Each pbranch can be though of * like a worker in a thread pool model. @@ -260,13 +268,23 @@ static __blocking triton_ret_t rosd_write_run_pipeline( in, this_start_record, this_remote_offset, - this_recordcount); + this_recordcount, + &this_update_id); triton_mutex_lock(&pipeline_mutex); + /* TODO: revisit update ID checking - everything should + * currently work its way to UINT64_MAX if there are any + * discrepencies */ if(triton_is_error(b_tret)) { out_tret = b_tret; } + 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) + *new_update_id = this_update_id; + triton_mutex_unlock(&pipeline_mutex); } } @@ -515,7 +533,6 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) } } #else - in.update_id_condition = 0; #endif if(my_position < (in.replication_factor-1)) @@ -526,7 +543,7 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) triton_debug(triton_dbg_rosd, "rosd_write() triggered, running pipeline.\n"); out.tret = rosd_write_run_pipeline(handle, src_addr, next_addr, my_position, - &in); + &in, &out.new_update_id); #if 0 if(entry_p) @@ -596,6 +613,7 @@ static __blocking triton_ret_t __remote_triton_rpc_rosd_write( triton_mercury_run_rpc_with_bulk(addr, rpc_rosd_write_id, handle, &in, &out, in.bulk_handle, 1, &data, &bulk_size, 0, HG_BULK_READ_ONLY); + *new_update_id = out.new_update_id; tret = triton_error_dup(out.tret); HG_Free_output(handle, &out); @@ -610,12 +628,13 @@ static __blocking triton_ret_t rosd_write_do_work( int my_position, uint64_t start_record, uint64_t recordcount, - triton_rpc_rosd_write_in_t* in) + triton_rpc_rosd_write_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; + uint64_t new_update_id_local, new_update_id_remote; /* right now only chained replication is supported */ assert(!(in->flags & ROSD_FLAG_FANOUT)); @@ -633,7 +652,7 @@ static __blocking triton_ret_t rosd_write_do_work( remote_tret = __remote_triton_rpc_rosd_write( next_addr, in->container, in->object, in->fork, start_record, recordcount, in->recordlen, buffer, - in->flags, in->update_id_condition, &new_update_id, + in->flags, in->update_id_condition, &new_update_id_remote, my_position+1, in->replication_factor); } } @@ -657,13 +676,14 @@ static __blocking triton_ret_t rosd_write_do_work( recordcount, in->recordlen, buffer, - 0, + in->flags, in->update_id_condition, - &new_update_id); + &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); @@ -671,6 +691,12 @@ static __blocking triton_ret_t rosd_write_do_work( 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); } @@ -686,7 +712,8 @@ static int is_usage_error_write(triton_ret_t tret) */ if(triton_error_equal(tret, TRITON_ERR_NO_SERVERS) || triton_error_equal(tret, TRITON_ERR_NOENT) || - triton_error_equal(tret, TRITON_ERR_ACCES)) + triton_error_equal(tret, TRITON_ERR_ACCES) || + triton_error_equal(tret, TRITON_ERR_CONDITIONAL)) { /* if there aren't enough servers to satisfy the replication * request, or the object doesn't exist, or if @@ -714,8 +741,9 @@ static __blocking triton_ret_t rosd_write_local_storage( struct hoss_oid soid; hoss_eid_t soid_eids[3]; hoss_size_t nbytes; - hoss_update_t update_info; + hoss_update_t update_check, update_new; int ret, hoss_rc; + hoss_flags_t hoss_flags; /* TODO: handle canceled error code (or whatever hoss will report to * indicate that a txn failed and should be retried) @@ -723,6 +751,7 @@ static __blocking triton_ret_t rosd_write_local_storage( 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"); return(TRITON_ERR_IO); } @@ -732,8 +761,24 @@ static __blocking triton_ret_t rosd_write_local_storage( 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"); + return TRITON_ERR_INVAL; + } + + /* TODO: what to do when no update id is given */ + if (hoss_flags == HOSS_NONE) + update_new = 0; + else + update_new = update_id_condition; + /* TODO: the update id arguments probably aren't right here... */ - ret = hoss_write(&soid, start_record, recordcount, recordlen, HOSS_NONE, 0, update_id_condition, buffer, grp, &nbytes, &hoss_rc); + ret = hoss_write(&soid, start_record, recordcount, recordlen, hoss_flags, + update_new, update_new, buffer, grp, &nbytes, + &hoss_rc); if(ret != 0) { triton_error_msg("hoss_write() failure, return code %d\n", ret); @@ -742,19 +787,25 @@ static __blocking triton_ret_t rosd_write_local_storage( } ret = hoss_end(grp, 1); - if(ret != 0) + if (ret == HOSS_ECONDFAIL) + return TRITON_ERR_CONDITIONAL; + else if(ret != 0) { + triton_error_msg("hoss_end() failure (write), return code %d\n", ret); return(TRITON_ERR_IO); } - /* NOTE: output arguments to hoss_write (rc and nbytes) are not filled * in until completion of hoss_end() */ - if(hoss_rc != 0 || nbytes != recordcount*recordlen) + else if(hoss_rc != 0 || nbytes != recordcount*recordlen) { triton_error_msg("hoss_write() failure or short write.\n"); return(TRITON_ERR_IO); } + + /* success - set update condition */ + *new_update_id = update_new; + return(TRITON_SUCCESS); } diff --git a/code/src/replicated-osd/rosd.ae b/code/src/replicated-osd/rosd.ae index d430b52..e2e6851 100644 --- a/code/src/replicated-osd/rosd.ae +++ b/code/src/replicated-osd/rosd.ae @@ -177,6 +177,24 @@ void triton_rpc_rosd_register(void) return; } +/* TODO: don't have a hoss-equivalent for UNTIL or AUTO yet, AFAIK */ +/* TODO: not sure if HOSS_NONE usage is correct */ +hoss_flags_t rosd_convert_flags_to_hoss(uint32_t flags) { + switch (ROSD_COND_MASK(flags)) { + case ROSD_COND_UNCONDITIONAL: + return HOSS_NONE; + case ROSD_COND_UNTIL: + return -1; + case ROSD_COND_ALL: + return HOSS_COND_ALL; + case ROSD_COND_AUTO: + return -1; + default: + /* unknown flags */ + return -1; + } +} + triton_ret_t triton_oid_to_addrs( uint64_t oid, uint32_t replication_factor, diff --git a/code/src/replicated-osd/rosd.hae b/code/src/replicated-osd/rosd.hae index 58f2290..6d47f12 100644 --- a/code/src/replicated-osd/rosd.hae +++ b/code/src/replicated-osd/rosd.hae @@ -18,6 +18,7 @@ #define ROSD_FLAG_TRUNC_WRITE 4 /* to make writes conditional (provided version number must be strictly * greater than current version number for the region in the object) + * NOTE: deprecated (currently unused in non-ancient revisions of triton */ #define ROSD_FLAG_COND_WRITE 8 /* set by master to tell replicas that this is expected to be the first @@ -26,6 +27,24 @@ */ #define ROSD_FLAG_FIRST_WRITE 16 +#define ROSD_FLAG_MAX 32 + +#define ROSD_FLAG_MASK(x) (x & (ROSD_FLAG_MAX-1)) +#define ROSD_COND_MASK(x) (x & (~(ROSD_FLAG_MAX-1))) + +/* ROSD conditional operation flags (basically mirrors the ASG equivalents) */ +typedef enum { + /* no condition - version numbers unchanged (but still returned) */ + ROSD_COND_UNCONDITIONAL = ROSD_FLAG_MAX, + /* read: read in order until finding a record with >= version number + * write/punch: update in order until finding a record with >= version + * number */ + ROSD_COND_UNTIL = (ROSD_FLAG_MAX << 1), + /* read/write: commit if *all* records in range have version #s strictly + * smaller than the provided */ + ROSD_COND_ALL = (ROSD_FLAG_MAX << 2), + ROSD_COND_AUTO = (ROSD_FLAG_MAX << 3) +} rosd_cond_flags_t; /* special values */ /* TODO: how to sync these with ASG equivalents cleanly? */ @@ -104,7 +123,7 @@ __blocking triton_ret_t remote_triton_rpc_rosd_read_one( uint64_t update_id_condition, void * buf, size_t buf_size, - uint64_t record_len, + uint64_t * record_len, uint64_t * record_update_id); __blocking triton_ret_t rosd_pipeline_setup_one_buffer( diff --git a/code/tests/asg/test-asg-cond.c b/code/tests/asg/test-asg-cond.c index 0e4c3a9..cd04b07 100644 --- a/code/tests/asg/test-asg-cond.c +++ b/code/tests/asg/test-asg-cond.c @@ -10,7 +10,7 @@ // expects at least a non-empty string fmt #define ERR(...) \ do { \ - fprintf(stderr, "%s:%d:", __FILE__, __LINE__); \ + fprintf(stderr, "Error: %s:%d: ", __FILE__, __LINE__); \ fprintf(stderr, __VA_ARGS__); \ return -1; \ } while (0) @@ -25,7 +25,7 @@ int main(int argc, char **argv) asg_fork_id_t fid; asg_record_id_t rid; asg_size_t out_size; - char fill = 'c'; + int fill; asg_record_info_t rinfo; if(argc != 2) @@ -43,77 +43,84 @@ int main(int argc, char **argv) ret = asg_initialize(&ait, argv[1]); if(ret != ASG_SUCCESS) - ERR("Error: asg_initialize(..., \"%s\")\n", argv[1]); + ERR("asg_initialize(..., \"%s\")\n", argv[1]); - out_ver = -1; - out_size = -1; + out_ver = 1111; + out_size = 1111; /* shorthand */ #define LOC \ ait, ASG_LOCATION_AUTO, cid, oid, fid, rid + fill = 5; /* write data (effectively a create) */ ret = asg_write( LOC, 1, /* rec count */ - 1, /* rec len */ + sizeof(fill), /* rec len */ ASG_COND_ALL, /* flags */ 1, /* update id */ &out_ver, /* assigned version */ &fill, /* data */ &out_size); /* resulting size */ if(ret != ASG_SUCCESS) - ERR("Error: asg_write()\n"); - else if (out_size != 1) - ERR("Error: asg_write() (short write)\n"); + ERR("asg_write() (return %d)\n", ret); + else if (out_size != sizeof(fill)) + ERR("asg_write() (short write)\n"); /* the update ID should be exactly what we gave it */ else if (out_ver != 1) - ERR("Error: gave ver 1 to asg_write, got %lu\n", out_ver); + ERR("gave ver 1 to asg_write, got %lu\n", out_ver); - out_ver = -1; - out_size = -1; + out_ver = 1111; + out_size = 1111; /* overwrite with a version ID = last - should fail */ ret = asg_write( LOC, 1, - 1, + sizeof(fill), ASG_COND_ALL, 1, &out_ver, &fill, &out_size); if (ret != ASG_ERR_UPDATE_ID) - ERR("asg_write(): expected update id error\n"); + ERR("asg_write(): expected update id error, got %d\n", ret); + fill = 255; memset(&rinfo, 0xff, sizeof(rinfo)); /* read with a version ID - should fail (ver is >=) */ ret = asg_read_sequence( LOC, 1, /* rec count */ - 1, /* expected record length */ + sizeof(fill), /* expected record length */ ASG_COND_ALL, /* flags */ 1, /* update id */ &fill, &rinfo); if(ret != ASG_ERR_UPDATE_ID) - ERR("asg_read_sequence(): expected update id error\n"); + ERR("asg_read_sequence(): expected update id error, got %d\n", ret); /* succeed (ver is >) */ ret = asg_read_sequence( LOC, 1, /* rec count */ - 1, /* expected record length */ + sizeof(fill), /* expected record length */ ASG_COND_ALL, /* flags */ 2, /* update id */ &fill, &rinfo); if(ret != ASG_SUCCESS) - ERR("Error: asg_read_sequence()\n"); + ERR("asg_read_sequence() (return %d)\n", ret); else if (rinfo.record_id != 1 || - rinfo.seq_len != 1 || rinfo.record_len != 1) - ERR("Error: asg_read_sequence()\n"); + rinfo.seq_len != 1 || rinfo.record_len != sizeof(fill)) + ERR("asg_read_sequence() (unexpected record info)\n"); else if (rinfo.record_update_id != 1) - ERR("Error: expected ver 1 from asg_read_sequence\n"); + ERR("expected ver 1 from asg_read_sequence, got %lu\n", + rinfo.record_update_id); + else if (fill != 5) + ERR("expected data \"5\" from asg_read_sequence, got %d\n", + (int)fill); + fill = 255; memset(&rinfo, 0xff, sizeof(rinfo)); /* read with a version ID - should fail (ver is >=) */ ret = asg_read_one( @@ -121,60 +128,66 @@ int main(int argc, char **argv) ASG_COND_ALL, 1, &fill, - 1, + sizeof(fill), &rinfo); if (ret != ASG_ERR_UPDATE_ID) - ERR("asg_read_one(): expected update id error\n"); + ERR("asg_read_one(): expected update id error, got %d\n", ret); /* should succeed (ver is >) */ ret = asg_read_one( LOC, ASG_COND_ALL, - 1, - &fill, 2, + &fill, + sizeof(fill), &rinfo); if(ret != ASG_SUCCESS) - ERR("Error: asg_read_one()\n"); + ERR("asg_read_one() (return %d)\n", ret); else if (rinfo.record_id != 1 || - rinfo.seq_len != 1 || rinfo.record_len != 1) - ERR("Error: asg_read_one()\n"); + rinfo.seq_len != 1 || rinfo.record_len != sizeof(fill)) + ERR("asg_read_one() (unexpected record info)\n"); else if (rinfo.record_update_id != 1) - ERR("Error: expected ver 1 from asg_read_one\n"); + ERR("expected ver 1 from asg_read_one, got %lu\n", + rinfo.record_update_id); + else if (fill != 5) + ERR("expected data \"5\" from asg_read_one, got %d\n", + (int)fill); - out_size = -1; - out_ver = -1; + fill = 6; + out_size = 1111; + out_ver = 1111; /* now try to increment */ ret = asg_write( LOC, 1, - 1, + sizeof(fill), ASG_COND_ALL, 2, &out_ver, &fill, &out_size); if(ret != ASG_SUCCESS) - ERR("Error: asg_write()\n"); - else if (out_size != 1) - ERR("Error: asg_write() (short write)\n"); + ERR("asg_write() (return %d)\n", ret); + else if (out_size != sizeof(fill)) + ERR("asg_write() (short write)\n"); /* the update ID should be exactly what we gave it */ else if (out_ver != 2) - ERR("Error: gave ver 2 to asg_write, got %lu\n", out_ver); + ERR("gave ver 2 to asg_write, got %lu\n", out_ver); - out_ver = -1; out_size = -1; + out_ver = 1111; out_size = 1111; /* overwrite with a version ID <= last - should fail */ ret = asg_write( LOC, 1, - 1, + sizeof(fill), ASG_COND_ALL, 2, &out_ver, &fill, &out_size); if (ret != ASG_ERR_UPDATE_ID) - ERR("asg_write(): expected update id error\n"); + ERR("asg_write(): expected update id error, got %d\n", ret); + fill = 255; memset(&rinfo, 0xff, sizeof(rinfo)); /* read with a version ID - should fail (ver is >=) */ ret = asg_read_sequence( @@ -186,54 +199,63 @@ int main(int argc, char **argv) &fill, &rinfo); if(ret != ASG_ERR_UPDATE_ID) - ERR("asg_read_sequence(): expected update id error\n"); + ERR("asg_read_sequence(): expected update id error, got %d\n", ret); /* succeed (ver is >) */ ret = asg_read_sequence( LOC, 1, /* rec count */ - 1, /* expected record length */ + sizeof(fill), /* expected record length */ ASG_COND_ALL, /* flags */ 3, /* update id */ &fill, &rinfo); if(ret != ASG_SUCCESS) - ERR("Error: asg_read_sequence()\n"); + ERR("asg_read_sequence() (return %d)\n", ret); else if (rinfo.record_id != 1 || - rinfo.seq_len != 1 || rinfo.record_len != 1) - ERR("Error: asg_read_one()\n"); + rinfo.seq_len != 1 || rinfo.record_len != sizeof(fill)) + ERR("asg_read_one() (unexpected record info)\n"); else if (rinfo.record_update_id != 2) - ERR("Error: expected ver 2 from asg_read_sequence\n"); + ERR("expected ver 2 from asg_read_sequence, got %lu\n", + rinfo.record_update_id); + else if (fill != 6) + ERR("expected data \"6\" from asg_read_sequence, got %d\n", + (int)fill); + fill = 255; memset(&rinfo, 0xff, sizeof(rinfo)); /* read with a version ID - should fail (ver is >=) */ ret = asg_read_one( LOC, ASG_COND_ALL, - 1, - &fill, 2, + &fill, + 1, &rinfo); if (ret != ASG_ERR_UPDATE_ID) - ERR("asg_read_one(): expected update id error\n"); + ERR("asg_read_one(): expected update id error, got %d\n", ret); /* should succeed (ver is >) */ ret = asg_read_one( LOC, ASG_COND_ALL, - 1, - &fill, 3, + &fill, + sizeof(fill), &rinfo); if(ret != ASG_SUCCESS) - ERR("Error: asg_read_one()\n"); + ERR("asg_read_one() (return %d)\n", ret); else if (rinfo.record_id != 1 || - rinfo.seq_len != 1 || rinfo.record_len != 1) - ERR("Error: asg_read_one()\n"); - else if (rinfo.record_update_id != 1) - ERR("Error: expected ver 1 from asg_read_one\n"); + rinfo.seq_len != 1 || rinfo.record_len != sizeof(fill)) + ERR("asg_read_one() (unexpected record info)\n"); + else if (rinfo.record_update_id != 2) + ERR("expected ver 1 from asg_read_one, got %lu\n", + rinfo.record_update_id); + else if (fill != 6) + ERR("expected data \"6\" from asg_read_one, got %d\n", + (int)fill); ret = asg_finalize(ait); if(ret != ASG_SUCCESS) - ERR("Error: asg_finalize()\n"); + ERR("asg_finalize() (return %d)\n", ret); return(0); } diff --git a/code/tests/asg/test-asg-simple.c b/code/tests/asg/test-asg-simple.c index 4001e81..56ab632 100644 --- a/code/tests/asg/test-asg-simple.c +++ b/code/tests/asg/test-asg-simple.c @@ -79,7 +79,7 @@ int main(int argc, char **argv) rid, /* start rec */ buffer_sz, /* rec count */ 1, /* rec len */ - ASG_AUTO_UPDATE_ID, /* flags */ + ASG_COND_NONE, /* flags */ 0, /* update id */ &out_ver, /* assigned version */ buffer, /* data */ hooks/post-receive --