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, master has been updated
via a1006cd6134d55de466abc073ac89e5a86cca065 (commit)
from 391b9dc071ae7d788fc5e164775b157a6595d040 (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 a1006cd6134d55de466abc073ac89e5a86cca065
Author: Phil Carns <carns(a)mcs.anl.gov>
Date: Mon Jun 4 15:46:18 2012 -0400
wrapper for server to server rpcs
- only used in replicated create operation so far
-----------------------------------------------------------------------
Summary of changes:
code/src/common/traffic-cop.ae | 93 +++++++++++++
code/src/common/traffic-cop.hae | 10 ++
code/src/placement/placement-nearest.ae | 54 ++++++++
code/src/placement/placement.hae | 7 +
code/src/replicated-osd/module.mk.in | 2 +-
.../{replicated-osd-s2s.haer => rosd-s2s.haer} | 0
code/src/replicated-osd/rosd.aer | 138 ++++++++------------
7 files changed, 217 insertions(+), 87 deletions(-)
rename code/src/replicated-osd/{replicated-osd-s2s.haer => rosd-s2s.haer} (100%)
Diff of changes:
diff --git a/code/src/common/traffic-cop.ae b/code/src/common/traffic-cop.ae
index c44dbae..67f4a58 100644
--- a/code/src/common/traffic-cop.ae
+++ b/code/src/common/traffic-cop.ae
@@ -438,6 +438,99 @@ __blocking int traffic_cop_rpc_control(uint64_t *niid, uint128_t target_oid,
return(1);
}
+__blocking int traffic_cop_rpc_control_s2s(
+ uint64_t *niid,
+ uint128_t oid,
+ int replication_factor,
+ int current_position,
+ int *next_position,
+ triton_addr_t *addr,
+ triton_ret_t *ret,
+ int* critical_error)
+{
+ struct triton_hash_link *rpc_link;
+ struct rpc_state* state;
+ triton_node_t peer_node;
+ triton_ret_t tmp_ret;
+ triton_node_t self;
+ int current_position_check;
+
+ *critical_error = 0;
+
+ self = triton_map_self();
+ if(triton_node_equal(self, triton_node_null))
+ {
+ *ret = TRITON_ERR_PLACEMENT;
+ *critical_error = 1;
+ return(0);
+ }
+
+ if(*niid == 0)
+ {
+ *niid = traffic_cop_rpc_register();
+ }
+
+ triton_mutex_lock(&rpc_table_mutex);
+
+ rpc_link = triton_hash_search(rpc_table, niid);
+ assert(rpc_link);
+ state = triton_hash_get_entry(rpc_link, struct rpc_state, hash_link);
+
+ if(state->retry_count == 0)
+ {
+ /* we haven't tried this operation yet, set addr and ret to initial
+ * values
+ */
+ *addr = triton_addr_null;
+ *ret = TRITON_ERR_WRONG_SERVER;
+ }
+ triton_mutex_unlock(&rpc_table_mutex);
+
+ /* see if we are out of retries yet */
+ if(!traffic_cop_rpc_should_retry(*niid, *addr, *ret))
+ {
+ traffic_cop_rpc_unregister(*niid);
+ return(0);
+ }
+
+ /* as the traffic cop if we should redo the
+ * object id -> server id -> server addr
+ * mapping
+ */
+ if(traffic_cop_rpc_should_remap(*niid, *addr, *ret))
+ {
+ int loop = 0;
+ *ret = triton_place_server_position(oid, replication_factor,
+ self, ¤t_position_check, &peer_node, next_position);
+ while(*ret != TRITON_SUCCESS)
+ {
+ triton_timer(2000);
+ *ret = triton_place_server_position(oid, replication_factor,
+ self, ¤t_position_check, &peer_node, next_position);
+ loop++;
+ if(loop == 5)
+ break;
+ }
+ if(*ret != TRITON_SUCCESS)
+ return(0);
+
+ if(current_position_check != current_position || *next_position < 0)
+ {
+ /* The replication sequence is out of whack due to system
+ * changes. Return a "wrong server" error so that the operation
+ * can be retried from the beginning with a new mapping.
+ */
+ *ret = TRITON_ERR_WRONG_SERVER;
+ return(0);
+ }
+ *ret = triton_map_lookup_node(peer_node, addr);
+ if(*ret != TRITON_SUCCESS)
+ return(0);
+ }
+
+ return(1);
+}
+
__blocking int traffic_cop_rpc_control_no_obj(uint64_t *niid, triton_addr_t addr, triton_ret_t *ret)
{
struct triton_hash_link *rpc_link;
diff --git a/code/src/common/traffic-cop.hae b/code/src/common/traffic-cop.hae
index 0ef2839..2daf9cb 100644
--- a/code/src/common/traffic-cop.hae
+++ b/code/src/common/traffic-cop.hae
@@ -59,6 +59,16 @@ __blocking int traffic_cop_rpc_control(uint64_t *niid, uint128_t target_oid,
triton_addr_t *addr, triton_ret_t *ret);
__blocking int traffic_cop_rpc_control_no_obj(uint64_t *niid, triton_addr_t addr, triton_ret_t *ret);
+__blocking int traffic_cop_rpc_control_s2s(
+ uint64_t *niid,
+ uint128_t oid,
+ int replication_factor,
+ int current_position,
+ int *next_position,
+ triton_addr_t *addr,
+ triton_ret_t *ret,
+ int* critical_error);
+
#endif /* __TRAFFIC_COP__ */
/*
diff --git a/code/src/placement/placement-nearest.ae b/code/src/placement/placement-nearest.ae
index 61bfbf2..19f65b7 100644
--- a/code/src/placement/placement-nearest.ae
+++ b/code/src/placement/placement-nearest.ae
@@ -234,6 +234,60 @@ triton_ret_t triton_place_hash(triton_string_t *key,
return TRITON_SUCCESS;
}
+triton_ret_t triton_place_server_position(uint128_t oid,
+ int replication_factor,
+ triton_node_t current_node,
+ int* current_position,
+ triton_node_t* next_node,
+ int* next_position)
+{
+ triton_ret_t tret;
+ triton_node_t * node_array;
+ int i=0;
+
+ node_array = malloc(replication_factor*sizeof(*node_array));
+ if(!node_array)
+ {
+ return(TRITON_ERR_NOMEM);
+ }
+
+ tret = triton_place_lookup(oid, replication_factor, node_array);
+ if(tret != TRITON_SUCCESS)
+ {
+ return(tret);
+ }
+
+ *current_position = -1;
+ for(i=0; i<replication_factor; i++)
+ {
+ if(triton_cmp_uint128(node_array[i], current_node) == 0)
+ {
+ *current_position = i;
+ break;
+ }
+ }
+
+ if(*current_position < 0)
+ {
+ return(TRITON_ERR_WRONG_SERVER);
+ }
+
+ if(*current_position < (replication_factor-1))
+ {
+ *next_position = (*current_position) + 1;
+ *next_node = node_array[*next_position];
+ }
+ else
+ {
+ *next_position = -1;
+ *next_node = triton_node_null;
+ }
+
+ free(node_array);
+
+ return(TRITON_SUCCESS);
+}
+
/*
* Local Variables:
* c-basic-offset: 4
diff --git a/code/src/placement/placement.hae b/code/src/placement/placement.hae
index e5e950c..ecf4db5 100644
--- a/code/src/placement/placement.hae
+++ b/code/src/placement/placement.hae
@@ -39,6 +39,13 @@ triton_ret_t triton_place_replica(uint128_t oid,
const triton_node_t *closest,
int *position);
+triton_ret_t triton_place_server_position(uint128_t oid,
+ int replication_factor,
+ triton_node_t current_node,
+ int* current_position,
+ triton_node_t* next_node,
+ int* next_position);
+
/**
Delegated to implementation
*/
diff --git a/code/src/replicated-osd/module.mk.in b/code/src/replicated-osd/module.mk.in
index 60d840b..9366972 100644
--- a/code/src/replicated-osd/module.mk.in
+++ b/code/src/replicated-osd/module.mk.in
@@ -5,7 +5,7 @@ AESOP_HDR += $(DIR_ROSD)/rosd-trans-nr-cache.hae \
AELIBSRC += $(DIR_ROSD)/rosd-trans-nr-cache.ae \
$(DIR_ROSD)/rosd-test-harness.ae \
- $(DIR_ROSD)/rosd-client.ae
+ $(DIR_ROSD)/rosd-client.ae
ifneq (,$(BUILD_MPI))
LIBSRC += $(DIR_ROSD)/rosd-utils.c
diff --git a/code/src/replicated-osd/replicated-osd-s2s.haer b/code/src/replicated-osd/rosd-s2s.haer
similarity index 100%
rename from code/src/replicated-osd/replicated-osd-s2s.haer
rename to code/src/replicated-osd/rosd-s2s.haer
diff --git a/code/src/replicated-osd/rosd.aer b/code/src/replicated-osd/rosd.aer
index 029ead0..09cc908 100644
--- a/code/src/replicated-osd/rosd.aer
+++ b/code/src/replicated-osd/rosd.aer
@@ -34,7 +34,7 @@
#include "src/replicated-osd/rosd-utils.h"
#include "src/replicated-osd/rosd-trans-nr-cache.hae"
#include "src/replicated-osd/rosd.haer"
-#include "src/replicated-osd/replicated-osd-s2s.haer"
+#include "src/replicated-osd/rosd-s2s.haer"
/* TODO: doxygen format */
@@ -77,6 +77,12 @@ static __blocking triton_ret_t interpret_error(
int(*usage_error_check)(triton_ret_t tret),
int from_client_flag);
static triton_ret_t niid_enabled_updater(const char *key, const char *value);
+static __blocking triton_ret_t client_rosd_s2s_create(
+ uint128_t oid,
+ int replication_factor,
+ int current_position,
+ uint64_t niid,
+ uint32_t flags);
static triton_ret_t rosd_init(void)
{
@@ -397,107 +403,26 @@ static __blocking triton_ret_t rosd_create_do_work(
triton_ret_t local_tret;
triton_ret_t remote_tret = TRITON_SUCCESS;
triton_ret_t tret;
- aer_remote_ctx_t rctx;
- struct rosd_s2s_create_req req;
- uint32_t next_position = my_position + 1;
- int forward_flag = 0;
- int my_new_position;
/* right now only chained replication is supported */
assert(!(flags & ROSD_FLAG_FANOUT));
flags |= ROSD_FLAG_CHAIN;
- /* do we need to forward this operation? */
- /* NOTE: deliberately collecting necessary information for communication
- * before performing local work; this will simplify cleanup if we fail
- * here.
- */
- if(my_position < replication_factor-1)
- {
- forward_flag = 1;
-
- tret = aesop_hints_get("triton.remote.context", sizeof(rctx), &rctx);
- if(tret != TRITON_SUCCESS)
- {
- /* generic hint/comm failure. Assume we won't be able to recover
- * from this and fail ourselves before propagating error.
- */
- trigger_server_fault(tret);
- return(tret);
- }
-
- tret = aer_init_struct_rosd_s2s_create_req(
- &req, &oid, &flags, &replication_factor, &niid, &next_position);
- if(tret != TRITON_SUCCESS)
- {
- /* generic hint/comm failure. Assume we won't be able to recover
- * from this and fail ourselves before propagating error.
- */
- trigger_server_fault(tret);
- return(tret);
- }
- }
-
/* forward request while simultaneously performing local work */
pwait
{
pprivate char oid_str[TRITON_UINT128_STRLEN];
pbranch
{
- uint64_t niid;
-
- if(forward_flag)
+ if(my_position < replication_factor -1)
{
triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid);
triton_debug(rosd_dbg_mask,
"ROSD forwarding create of oid %s to %d'th server.\n",
oid_str, my_position+1);
- remote_tret = TRITON_SUCCESS;
- peer_addr = TRITON_ADDR_NULL;
-
- /* TODO: we need to make a wrapper for this logic, similar
- * to what is done on the client side for rpcs
- */
-
- /* NOTE: we generate a new niid here just to use for
- * tracking RPC retries. The forwarded create request is
- * still using the old niid generated by the client
- */
- niid = traffic_cop_rpc_register();
- do
- {
- if(traffic_cop_rpc_should_remap(niid, peer_addr, remote_tret))
- {
- tret = get_placement_info(
- oid, replication_factor, &my_new_position,
- &peer_addr);
- if(triton_error_equal(tret, TRITON_ERR_NOTFOUND) ||
- my_position != my_new_position)
- {
- /* This position of _this_ server has changed.
- * We need to propagate a "wrong server" error
- * up to the caller to start over.
- */
- remote_tret = TRITON_ERR_WRONG_SERVER;
- break;
- }
- else if(tret != TRITON_SUCCESS)
- {
- /* generic placement failure. Assume we won't be
- * able to recover from this.
- */
- trigger_server_fault(tret);
- remote_tret = tret;
- break;
- }
- }
- remote_tret = remote_rosd_s2s_create(
- rctx, peer_addr, &req, NULL);
- } while(traffic_cop_rpc_should_retry(niid, peer_addr, remote_tret));
- traffic_cop_rpc_unregister(niid);
-
- aer_destroy_struct_rosd_s2s_create_req(&req);
+ remote_tret = client_rosd_s2s_create(oid, replication_factor,
+ my_position, niid, flags);
}
}
pbranch
@@ -509,7 +434,7 @@ static __blocking triton_ret_t rosd_create_do_work(
}
}
- if(forward_flag)
+ if(my_position < replication_factor -1)
tret = interpret_errors(local_tret, remote_tret, my_position,
peer_addr, is_usage_error_create, from_client_flag);
else
@@ -1191,6 +1116,47 @@ static triton_ret_t niid_enabled_updater(const char *key, const char *value)
return TRITON_SUCCESS;
}
+static __blocking triton_ret_t client_rosd_s2s_create(
+ uint128_t oid,
+ int replication_factor,
+ int current_position,
+ uint64_t niid,
+ uint32_t flags)
+{
+ struct rosd_s2s_create_req req;
+ triton_ret_t tret;
+ int retry;
+ uint64_t internal_niid = 0;
+ triton_addr_t addr;
+ int next_position = 0;
+ int critical = 0;
+
+ tret = aer_init_struct_rosd_s2s_create_req(&req, &oid, &flags,
+ &replication_factor, &niid, &next_position);
+ if(tret != TRITON_SUCCESS)
+ return(tret);
+
+ retry = traffic_cop_rpc_control_s2s(&internal_niid, oid, replication_factor,
+ current_position, &next_position, &addr, &tret, &critical);
+ while(retry)
+ {
+ req.next_position = next_position;
+ tret = remote_rosd_s2s_create(AER_DEFAULT_CTX, addr, &req, NULL);
+ retry = traffic_cop_rpc_control_s2s(&internal_niid, oid,
+ replication_factor, current_position, &next_position, &addr, &tret, &critical);
+ }
+ aer_destroy_struct_rosd_s2s_create_req(&req);
+
+ if(critical)
+ {
+ trigger_server_fault(tret);
+ return(0);
+ }
+ return(tret);
+
+}
+
+
/*
* Local variables:
* c-indent-level: 4
hooks/post-receive
--
