Commit f68072eb authored by Matthieu Dorier's avatar Matthieu Dorier
Browse files

modified sdskv_get() so it returns the value size in case of SDSKV_ERR_SIZE

parent 830f58ff
...@@ -607,10 +607,14 @@ int sdskv_get(sdskv_provider_handle_t provider, ...@@ -607,10 +607,14 @@ int sdskv_get(sdskv_provider_handle_t provider,
} }
ret = out.ret; ret = out.ret;
*vsize = (hg_size_t)out.value.size;
if(ret == SDSKV_SUCCESS) {
if (out.value.size > 0 && out.ret == SDSKV_SUCCESS) { *vsize = out.vsize;
memcpy(value, out.value.data, out.value.size); if (out.value.size > 0) {
memcpy(value, out.value.data, out.value.size);
}
} else if(ret == SDSKV_ERR_SIZE) {
*vsize = out.vsize;
} }
margo_free_output(handle, &out); margo_free_output(handle, &out);
...@@ -655,7 +659,7 @@ int sdskv_get(sdskv_provider_handle_t provider, ...@@ -655,7 +659,7 @@ int sdskv_get(sdskv_provider_handle_t provider,
} }
ret = out.ret; ret = out.ret;
*vsize = (hg_size_t)out.size; *vsize = out.vsize;
margo_free_output(handle, &out); margo_free_output(handle, &out);
margo_bulk_free(in.handle); margo_bulk_free(in.handle);
......
...@@ -24,14 +24,13 @@ static inline hg_return_t hg_proc_kv_data_t(hg_proc_t proc, void *arg) ...@@ -24,14 +24,13 @@ static inline hg_return_t hg_proc_kv_data_t(hg_proc_t proc, void *arg)
{ {
hg_return_t ret; hg_return_t ret;
kv_data_t *in = (kv_data_t*)arg; kv_data_t *in = (kv_data_t*)arg;
ret = hg_proc_hg_size_t(proc, &in->size); ret = hg_proc_hg_size_t(proc, &in->size);
if(ret != HG_SUCCESS) return ret; if(ret != HG_SUCCESS) return ret;
if (in->size) { if (in->size) {
switch (hg_proc_get_op(proc)) { switch (hg_proc_get_op(proc)) {
case HG_ENCODE: case HG_ENCODE:
ret = hg_proc_raw(proc, in->data, in->size); ret = hg_proc_raw(proc, in->data, in->size);
if(ret != HG_SUCCESS) return ret; if(ret != HG_SUCCESS) return ret;
break; break;
case HG_DECODE: case HG_DECODE:
in->data = (kv_ptr_t)malloc(in->size); in->data = (kv_ptr_t)malloc(in->size);
...@@ -133,7 +132,8 @@ MERCURY_GEN_PROC(get_in_t, ((uint64_t)(db_id))\ ...@@ -133,7 +132,8 @@ MERCURY_GEN_PROC(get_in_t, ((uint64_t)(db_id))\
((hg_size_t)(vsize))) ((hg_size_t)(vsize)))
MERCURY_GEN_PROC(get_out_t, ((int32_t)(ret))\ MERCURY_GEN_PROC(get_out_t, ((int32_t)(ret))\
((kv_data_t)(value))) ((kv_data_t)(value))\
((hg_size_t)(vsize)))
// ------------- LENGTH ------------- // // ------------- LENGTH ------------- //
MERCURY_GEN_PROC(length_in_t, ((uint64_t)(db_id))((kv_data_t)(key))) MERCURY_GEN_PROC(length_in_t, ((uint64_t)(db_id))((kv_data_t)(key)))
...@@ -179,7 +179,7 @@ MERCURY_GEN_PROC(bulk_get_in_t, ((uint64_t)(db_id))\ ...@@ -179,7 +179,7 @@ MERCURY_GEN_PROC(bulk_get_in_t, ((uint64_t)(db_id))\
((kv_data_t)(key))\ ((kv_data_t)(key))\
((hg_size_t)(vsize))\ ((hg_size_t)(vsize))\
((hg_bulk_t)(handle))) ((hg_bulk_t)(handle)))
MERCURY_GEN_PROC(bulk_get_out_t, ((hg_size_t)(size)) ((int32_t)(ret))) MERCURY_GEN_PROC(bulk_get_out_t, ((hg_size_t)(vsize)) ((int32_t)(ret)))
// ------------- PUT MULTI ------------- // // ------------- PUT MULTI ------------- //
MERCURY_GEN_PROC(put_multi_in_t, \ MERCURY_GEN_PROC(put_multi_in_t, \
......
...@@ -871,6 +871,7 @@ static void sdskv_get_ult(hg_handle_t handle) ...@@ -871,6 +871,7 @@ static void sdskv_get_ult(hg_handle_t handle)
out.ret = SDSKV_ERR_MERCURY; out.ret = SDSKV_ERR_MERCURY;
out.value.data = nullptr; out.value.data = nullptr;
out.value.size = 0; out.value.size = 0;
out.vsize = 0;
margo_respond(handle, &out); margo_respond(handle, &out);
margo_destroy(handle); margo_destroy(handle);
return; return;
...@@ -883,6 +884,7 @@ static void sdskv_get_ult(hg_handle_t handle) ...@@ -883,6 +884,7 @@ static void sdskv_get_ult(hg_handle_t handle)
out.ret = SDSKV_ERR_UNKNOWN_DB; out.ret = SDSKV_ERR_UNKNOWN_DB;
out.value.data = nullptr; out.value.data = nullptr;
out.value.size = 0; out.value.size = 0;
out.vsize = 0;
margo_respond(handle, &out); margo_respond(handle, &out);
margo_free_input(handle, &in); margo_free_input(handle, &in);
margo_destroy(handle); margo_destroy(handle);
...@@ -896,15 +898,18 @@ static void sdskv_get_ult(hg_handle_t handle) ...@@ -896,15 +898,18 @@ static void sdskv_get_ult(hg_handle_t handle)
ds_bulk_t vdata; ds_bulk_t vdata;
if(db->get(kdata, vdata)) { if(db->get(kdata, vdata)) {
if(vdata.size() <= in.vsize) { if(vdata.size() <= in.vsize) {
out.vsize = vdata.size();
out.value.size = vdata.size(); out.value.size = vdata.size();
out.value.data = vdata.data(); out.value.data = vdata.data();
out.ret = SDSKV_SUCCESS; out.ret = SDSKV_SUCCESS;
} else { } else {
out.vsize = vdata.size();
out.value.size = 0; out.value.size = 0;
out.value.data = nullptr; out.value.data = nullptr;
out.ret = SDSKV_ERR_SIZE; out.ret = SDSKV_ERR_SIZE;
} }
} else { } else {
out.vsize = 0;
out.value.size = 0; out.value.size = 0;
out.value.data = nullptr; out.value.data = nullptr;
out.ret = SDSKV_ERR_UNKNOWN_KEY; out.ret = SDSKV_ERR_UNKNOWN_KEY;
...@@ -1290,8 +1295,17 @@ static void sdskv_bulk_get_ult(hg_handle_t handle) ...@@ -1290,8 +1295,17 @@ static void sdskv_bulk_get_ult(hg_handle_t handle)
ds_bulk_t vdata; ds_bulk_t vdata;
auto b = db->get(kdata, vdata); auto b = db->get(kdata, vdata);
if(!b || vdata.size() > in.vsize) { if(!b) {
out.size = 0; out.vsize = 0;
out.ret = SDSKV_ERR_UNKNOWN_KEY;
margo_respond(handle, &out);
margo_free_input(handle, &in);
margo_destroy(handle);
return;
}
if(vdata.size() > in.vsize) {
out.vsize = vdata.size();
out.ret = SDSKV_ERR_SIZE; out.ret = SDSKV_ERR_SIZE;
margo_respond(handle, &out); margo_respond(handle, &out);
margo_free_input(handle, &in); margo_free_input(handle, &in);
...@@ -1305,7 +1319,7 @@ static void sdskv_bulk_get_ult(hg_handle_t handle) ...@@ -1305,7 +1319,7 @@ static void sdskv_bulk_get_ult(hg_handle_t handle)
hret = margo_bulk_create(mid, 1, (void**)&buffer, &size, hret = margo_bulk_create(mid, 1, (void**)&buffer, &size,
HG_BULK_READ_ONLY, &bulk_handle); HG_BULK_READ_ONLY, &bulk_handle);
if(hret != HG_SUCCESS) { if(hret != HG_SUCCESS) {
out.size = 0; out.vsize = 0;
out.ret = SDSKV_ERR_MERCURY; out.ret = SDSKV_ERR_MERCURY;
margo_respond(handle, &out); margo_respond(handle, &out);
margo_free_input(handle, &in); margo_free_input(handle, &in);
...@@ -1316,7 +1330,7 @@ static void sdskv_bulk_get_ult(hg_handle_t handle) ...@@ -1316,7 +1330,7 @@ static void sdskv_bulk_get_ult(hg_handle_t handle)
hret = margo_bulk_transfer(mid, HG_BULK_PUSH, info->addr, in.handle, 0, hret = margo_bulk_transfer(mid, HG_BULK_PUSH, info->addr, in.handle, 0,
bulk_handle, 0, vdata.size()); bulk_handle, 0, vdata.size());
if(hret != HG_SUCCESS) { if(hret != HG_SUCCESS) {
out.size = 0; out.vsize = 0;
out.ret = SDSKV_ERR_MERCURY; out.ret = SDSKV_ERR_MERCURY;
margo_respond(handle, &out); margo_respond(handle, &out);
margo_free_input(handle, &in); margo_free_input(handle, &in);
...@@ -1328,7 +1342,7 @@ static void sdskv_bulk_get_ult(hg_handle_t handle) ...@@ -1328,7 +1342,7 @@ static void sdskv_bulk_get_ult(hg_handle_t handle)
margo_bulk_free(bulk_handle); margo_bulk_free(bulk_handle);
} }
out.size = size; out.vsize = size;
out.ret = SDSKV_SUCCESS; out.ret = SDSKV_SUCCESS;
margo_respond(handle, &out); margo_respond(handle, &out);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment