October 2013 - Triton-commits - mailman3.cels.anl.gov

C-Utils Repository branch, master, updated. e11f958e01bea27318a047161add9842f297b0f4
by noreply＠mcs.anl.gov 28 Oct '13

28 Oct '13

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 "C-Utils Repository". The branch, master has been updated via e11f958e01bea27318a047161add9842f297b0f4 (commit) from 95d78c82b943db7ea32cb8e6654c6ab9e653ddd4 (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 e11f958e01bea27318a047161add9842f297b0f4 Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Mon Oct 28 19:14:26 2013 -0500 Remove triton-string from c-utils - Remove triton-string.[ch] - Update triton-hash (string hashing function) - Update triton-hash unit test ----------------------------------------------------------------------- Summary of changes: TODO | 2 +- src/Makefile.subdir | 2 - src/triton-hash.h | 15 ++-- src/triton-string.c | 19 ---- src/triton-string.h | 236 --------------------------------------------------- tests/testhash.c | 40 +++++----- 6 files changed, 30 insertions(+), 284 deletions(-) delete mode 100644 src/triton-string.c delete mode 100644 src/triton-string.h Diff of changes: diff --git a/TODO b/TODO index 72d640a..dc54101 100644 --- a/TODO +++ b/TODO @@ -1,7 +1,7 @@ TODO Items: - cleanup library naming (get rid of triton) - - triton-string -> cu-string.h or something + - triton-xxx -> cu-xxx.h or something - check if easy/possible to rename ev.h when installing to avoid conflict (or should we just use the public libev?) diff --git a/src/Makefile.subdir b/src/Makefile.subdir index dc29495..ec1283b 100644 --- a/src/Makefile.subdir +++ b/src/Makefile.subdir @@ -1,13 +1,11 @@ c_utils_sources += \ src/c-utils-common.c\ src/lookup3.c\ -src/triton-string.c\ src/workqueue.c public_headers += src/c-utils-common.h \ src/triton-base.h \ src/triton-list.h \ - src/triton-string.h \ src/triton-thread.h \ src/triton-hash.h \ src/triton-uint128.h \ diff --git a/src/triton-hash.h b/src/triton-hash.h index 68bd278..0665906 100644 --- a/src/triton-hash.h +++ b/src/triton-hash.h @@ -7,12 +7,13 @@ #ifndef TRITON_HASH_H #define TRITON_HASH_H -#include <errno.h> #include "triton-list.h" -#include "triton-string.h" #include "jenkins-hash.h" +#include <string.h> +#include <errno.h> + #define triton_hash_link triton_list_link /* __typename__ * triton_hash_get_entry(struct triton_hash_link *link, __typename__, __link_field_name__); */ @@ -328,14 +329,16 @@ static inline int triton_hash_strhash(const void *k, int table_size) return pc & (table_size - 1); } +// Hashes const char * +// Note: cannot handle embedded 0 (C string limitation) static inline int triton_hash_string_hash(const void *k, int table_size) { uint32_t pc = 0, pb = 0; - const triton_string_t *s; + const char *s; int result = -1; - s = (const triton_string_t *)k; - assert(s->string); - bj_hashlittle2(s->string, s->size, &pc, &pb); + s = (const char *)k; + assert(s); + bj_hashlittle2(s, strlen(s), &pc, &pb); result = pc & (table_size - 1); return result; } diff --git a/src/triton-string.c b/src/triton-string.c deleted file mode 100644 index 1d2d683..0000000 --- a/src/triton-string.c +++ /dev/null @@ -1,19 +0,0 @@ -/* - * (C) 2009 The University of Chicago - * - * See COPYRIGHT in top-level directory. - */ - - -#include "triton-string.h" - -triton_string_t triton_string_null = { .size = 0, .string = NULL, .const_string = NULL }; - -/* - * Local variables: - * c-indent-level: 4 - * c-basic-offset: 4 - * End: - * - * vim: ft=c ts=8 sts=4 sw=4 expandtab - */ diff --git a/src/triton-string.h b/src/triton-string.h deleted file mode 100644 index 6c1926b..0000000 --- a/src/triton-string.h +++ /dev/null @@ -1,236 +0,0 @@ -/* - * (C) 2009 The University of Chicago - * - * See COPYRIGHT in top-level directory. - */ - -#ifndef __TRITON_STRING_H__ -#define __TRITON_STRING_H__ - -#include <stdio.h> -#include <stdarg.h> -#include <stdlib.h> -#include <string.h> -#include <assert.h> - -#include "jenkins-hash.h" -#include "triton-thread.h" - -/** - * The triton_string_t is a null-terminated string. Used for encoding - * strings. - */ -typedef struct triton_string -{ - uint32_t size; - char * string; - const char *const_string; - - struct triton_string *parent; - uint32_t refcount; - triton_mutex_t reflock; -} triton_string_t; - -static inline void __triton_string_init_internal(triton_string_t *s) -{ - memset(s, 0, sizeof(*s)); - triton_mutex_init(&s->reflock, NULL); - - return; -} - -static inline void triton_string_init_va(triton_string_t *s, const char *format, va_list ap) -{ - va_list aap; - int err; - __triton_string_init_internal(s); - - if(format) - { - va_copy(aap, ap); - s->size = vsnprintf(NULL, 0, format, ap) + 1; - va_end(ap); - s->string = malloc(s->size); - assert(s->string); - err = vsnprintf(s->string, s->size, format, aap); - assert(err >= 0 && err < s->size); - } - else - { - s->size = 0; - s->string = NULL; - } -} - -static inline void triton_string_init(triton_string_t *s, const char *format, ...) __attribute__ ((format (printf, 2, 3))) ; - -static inline void triton_string_init(triton_string_t *s, const char *format, ...) -{ - va_list ap; - va_start(ap, format); - triton_string_init_va(s, format, ap); - va_end(ap); -} - -static inline void triton_string_init_size(triton_string_t *s, const char *value, uint32_t len) -{ - __triton_string_init_internal(s); - - if(len > 0) - { - s->string = malloc(len); - assert(s->string); - s->size = len; - memcpy(s->string, value, len); - } - else - { - s->string = NULL; - s->size = 0; - } -} - -/* does not copy the passed in string */ -static inline void triton_string_init_inplace(triton_string_t *s, char *value) -{ - __triton_string_init_internal(s); - - if(value) - { - s->string = value; - s->size = strlen(value)+1; - } - else - { - s->size = 0; - s->string = NULL; - } -} - -static inline void triton_string_init_inplace_size(triton_string_t *s, char *value, int len) -{ - __triton_string_init_internal(s); - - s->size = len; - if(len > 0) - { - s->string = value; - } - else - { - s->string = NULL; - } -} - -static inline const char* triton_string_get(triton_string_t* str) -{ - while(str->parent) - str = str->parent; - - if(str->size == 0) - return(NULL); - - return(str->const_string ? str->const_string : str->string); -} -#if 0 -#define triton_string_get(__str__) ((__str__)->const_string ? (__str__)->const_string : (__str__)->string) -#endif - -#define triton_string_clear(__val) do { \ - (__val)->parent = NULL; \ - (__val)->refcount = 0; \ - (__val)->size = 0; \ - (__val)->string = NULL; \ - (__val)->const_string = NULL; \ -} while(0) - -#define triton_string_set_null(__val) do { \ - triton_string_clear(__val); \ - triton_mutex_init(&(__val)->reflock, NULL); \ -} while(0) - -static inline void triton_string_decrement(triton_string_t *s) -{ - triton_mutex_lock(&s->reflock); - s->refcount--; - triton_mutex_unlock(&s->reflock); - - if(s->parent) - triton_string_decrement(s->parent); -} - -static inline void triton_string_destroy(triton_string_t *s) -{ - if(s->parent) - triton_string_decrement(s->parent); - - triton_mutex_lock(&s->reflock); - if(s->refcount == 0) - { - if(s->string) - free(s->string); - triton_mutex_unlock(&s->reflock); - triton_mutex_destroy(&s->reflock); - triton_string_clear(s); - } - else - { - triton_mutex_unlock(&s->reflock); - } - - return; -} - -static inline void triton_string_copy(triton_string_t *to, triton_string_t *from) -{ - assert(to != NULL && from != NULL); - triton_string_set_null(to); - triton_mutex_lock(&from->reflock); - to->parent = from; - from->refcount++; - triton_mutex_unlock(&from->reflock); -} - -static inline void triton_string_clone(triton_string_t *to, const triton_string_t *from) -{ - assert(to != NULL && from != NULL); - triton_string_set_null(to); - to->size = from->size; - if(from->const_string) - { - to->string = strdup(from->const_string); - to->size = from->size; - } - else - { - to->size = from->size; - to->string = malloc(from->size); - assert(to->string); - memcpy(to->string, from->string, from->size); - } -} - -extern triton_string_t triton_string_null; -#define TRITON_STRING_NULL (&triton_string_null) - -#define triton_string_init_const(__str__) { .size = sizeof(__str__), .string = NULL, .const_string = __str__ } - -static inline int triton_string_hash(void *k, int table_size) -{ - triton_string_t *s = (triton_string_t *)k; - uint32_t pc, pb; - - bj_hashlittle2(s->string, s->size, &pc, &pb); - return (pc & (table_size - 1)); -} - -#endif - -/* - * Local variables: - * c-indent-level: 4 - * c-basic-offset: 4 - * End: - * - * vim: ft=c ts=8 sts=4 sw=4 expandtab - */ diff --git a/tests/testhash.c b/tests/testhash.c index a7d7bf1..48887c4 100644 --- a/tests/testhash.c +++ b/tests/testhash.c @@ -8,10 +8,11 @@ #include "triton-list.h" #include "triton-hash.h" -#include "triton-string.h" #include "triton-uint128.h" #include "test-tap.h" +#include <string.h> + #define DEFAULT_TEST_SIZE 1000 struct e @@ -22,7 +23,7 @@ struct e struct s { - triton_string_t *v; + char *v; triton_list_link_t link; }; @@ -72,12 +73,12 @@ static int e_compare(const void *key, struct triton_hash_link *link) static int s_compare(const void *key, struct triton_hash_link *link) { - const triton_string_t *value1 = (const triton_string_t *)key; + const char *value1 = (const char *)key; struct s *value2 = triton_hash_get_entry(link, struct s, link); assert(key); assert(link); - return (!strcmp(value2->v->string, value1->string)); + return (!strcmp(value2->v, value1)); } static int n_compare(const void *key, struct triton_hash_link *link) @@ -95,7 +96,7 @@ static void e_destroy(struct e *entry) {} static void s_destroy(struct s *entry) -{} +{ free (entry->v); } int main(int argc, char *argv[]) { @@ -106,7 +107,6 @@ int main(int argc, char *argv[]) struct n *ns; struct triton_hash_link *result; int i, j, testsize, hash; - triton_string_t *string; struct triton_hash_table *ktable; int testcount, failcount; @@ -156,37 +156,37 @@ int main(int argc, char *argv[]) for (i = 0; i < 10; i++) { - string = malloc(sizeof(triton_string_t)); - triton_string_init(string, "string: _%d_", i); - hash = triton_hash_string_hash(string, 23); + char buf[512]; + memset (&buf, 0, sizeof(buf)); + snprintf (&buf[0], sizeof (buf)-1, "string: _%d_", i); + hash = triton_hash_string_hash(&buf[0], 23); /* printf("hash(\"%s\") -> %i\n", string->string, hash); */ - triton_string_destroy(string); - free(string); } table = triton_hash_init(s_compare, triton_hash_string_hash, 23); for(i = 0; i < testsize; ++i) { + char buf[512]; + memset (&buf, 0, sizeof(buf)); + snprintf (&buf[0], sizeof(buf)-1, "[%i]", i); ss = malloc(sizeof(*ss)); - string = malloc(sizeof(triton_string_t)); - triton_string_init(string, "[%i]", i); - ss->v = string; + ss->v = strdup (buf); triton_list_link_clear(&ss->link); - triton_hash_add(table, string, &(ss->link)); + triton_hash_add(table, &buf[0], &(ss->link)); } for(i = 0; i < testsize; ++i) { - string = malloc(sizeof(triton_string_t)); - triton_string_init(string, "[%i]", i); - result = triton_hash_search(table, string); + char buf[512]; + memset (&buf, 0, sizeof(buf)); + snprintf (&buf[0], sizeof(buf)-1, "[%i]", i); + result = triton_hash_search(table, &buf[0]); assert(result != NULL); ss = triton_hash_get_entry(result, struct s, link); - sscanf(ss->v->string, "[%i]", &j); + sscanf(ss->v, "[%i]", &j); if(i != j) break; - free(string); } tap("triton_hash_search (strings)", i==testsize, testcount, failcount); hooks/post-receive -- C-Utils Repository

1 0

branch, master, updated. f856202c7a7414e66dc3754fbd92323f2f24b030
by noreply＠mcs.anl.gov 24 Oct '13

24 Oct '13

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 f856202c7a7414e66dc3754fbd92323f2f24b030 (commit) from 507cd11ae01f2a29fc474fa962dc19890df35daf (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 f856202c7a7414e66dc3754fbd92323f2f24b030 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Thu Oct 24 21:11:05 2013 -0500 spelling ----------------------------------------------------------------------- Summary of changes: code/INSTALL | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) Diff of changes: diff --git a/code/INSTALL b/code/INSTALL index 1e7bd74..3029dab 100644 --- a/code/INSTALL +++ b/code/INSTALL @@ -21,7 +21,7 @@ network transport. git clone git://git.mcs.anl.gov/radix/mercury cd mercury git submodule init - get submodule update + git submodule update mkdir build cd build cmake -DNA_USE_BMI:BOOL=ON \ hooks/post-receive --

1 0

branch, master, updated. 507cd11ae01f2a29fc474fa962dc19890df35daf
by noreply＠mcs.anl.gov 24 Oct '13

24 Oct '13

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 507cd11ae01f2a29fc474fa962dc19890df35daf (commit) from fb8a997b4a426b9e55fa7805e1b2304b6eb5f7b2 (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 507cd11ae01f2a29fc474fa962dc19890df35daf Author: Phil Carns <carns(a)mcs.anl.gov> Date: Thu Oct 24 20:56:22 2013 -0500 update INSTALL steps for mercury ----------------------------------------------------------------------- Summary of changes: code/INSTALL | 6 +++++- 1 files changed, 5 insertions(+), 1 deletions(-) Diff of changes: diff --git a/code/INSTALL b/code/INSTALL index 0caeb71..1e7bd74 100644 --- a/code/INSTALL +++ b/code/INSTALL @@ -16,8 +16,12 @@ network transport. make make install -* Mercury (git clone git://git.mcs.anl.gov/radix/mercury) +* Mercury (note that Mercury _must_ be built with both BMI and Boost support) + git clone git://git.mcs.anl.gov/radix/mercury + cd mercury + git submodule init + get submodule update mkdir build cd build cmake -DNA_USE_BMI:BOOL=ON \ hooks/post-receive --

1 0

branch, master, updated. fb8a997b4a426b9e55fa7805e1b2304b6eb5f7b2
by noreply＠mcs.anl.gov 23 Oct '13

23 Oct '13

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 fb8a997b4a426b9e55fa7805e1b2304b6eb5f7b2 (commit) from 55bb5d3e738194e0a6e48534f942deab1cd2dc41 (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 fb8a997b4a426b9e55fa7805e1b2304b6eb5f7b2 Author: Sumit Narayan <sumit(a)mcs.anl.gov> Date: Wed Oct 23 17:03:29 2013 -0500 - Trac# 262 - A bug in Mercury was causing a segmentation fault when RPC requests larger than 4K were being made. That bug has been fixed, so we can enable large requests in this tests now. ----------------------------------------------------------------------- Summary of changes: code/src/admin-tools/triton-cp.ae | 5 +---- 1 files changed, 1 insertions(+), 4 deletions(-) Diff of changes: diff --git a/code/src/admin-tools/triton-cp.ae b/code/src/admin-tools/triton-cp.ae index 1444c84..789cc19 100644 --- a/code/src/admin-tools/triton-cp.ae +++ b/code/src/admin-tools/triton-cp.ae @@ -17,10 +17,7 @@ #include "src/replicated-osd/rosd.hae" /* TODO: make this configurable */ -/* TODO: if we boost this higher (to 4096) then I get a segfault in bulk - * handling in Mercury. Need to find out why. - */ -#define BUFFER_SZ 128 +#define BUFFER_SZ 4096 enum obj_ref_type { hooks/post-receive --

1 0

branch, master, updated. 55bb5d3e738194e0a6e48534f942deab1cd2dc41
by noreply＠mcs.anl.gov 23 Oct '13

23 Oct '13

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 55bb5d3e738194e0a6e48534f942deab1cd2dc41 (commit) from 646669e95bde92bf3ac91cd52c1de00094a71316 (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 55bb5d3e738194e0a6e48534f942deab1cd2dc41 Author: Sumit Narayan <sumit(a)mcs.anl.gov> Date: Wed Oct 23 15:02:30 2013 -0500 - Changed configure file to pick Mercury's dependencies properly. ----------------------------------------------------------------------- Summary of changes: code/configure.ac | 12 ++---------- 1 files changed, 2 insertions(+), 10 deletions(-) Diff of changes: diff --git a/code/configure.ac b/code/configure.ac index 28267e1..9d65f00 100755 --- a/code/configure.ac +++ b/code/configure.ac @@ -46,14 +46,6 @@ LIBS="$AESOP_LIBS $LIBS" CPPFLAGS="$AESOP_CFLAGS $CPPFLAGS" CFLAGS="$AESOP_CFLAGS $CFLAGS" -USE_BMI=/home/pcarns/working/bmi/install -AC_ARG_WITH(bmi, - [ --with-bmi=<dir> Location of external BMI package], - AC_CHECK_FILE(${withval}/include/bmi.h,[], - [AC_MSG_ERROR([Could not find BMI])]) - USE_BMI=${withval} - ]) - USE_MERCURY= AC_ARG_WITH(mercury, [ --with-mercury=<dir> Location of external mercury package], @@ -65,8 +57,8 @@ AC_ARG_WITH(mercury, [ USE_MERCURY= ]) -MERCURY_LIBS=`PKG_CONFIG_PATH=${withval}/lib/pkgconfig pkg-config mercury --libs` -MERCURY_LIBS="-L${USE_BMI}/lib $MERCURY_LIBS -lbmi" +MERCURY_LIBS=`PKG_CONFIG_PATH=${withval}/lib/pkgconfig pkg-config mercury --libs --static` +MERCURY_LIBS="$MERCURY_LIBS" MERCURY_CFLAGS=`PKG_CONFIG_PATH=${withval}/lib/pkgconfig pkg-config mercury --cflags` AC_SUBST(USE_MERCURY) LIBS="$MERCURY_LIBS $LIBS" hooks/post-receive --

1 0

branch, master, updated. c02e99faccdaf94e1ec90775e3f1a880aa7b380f
by noreply＠mcs.anl.gov 13 Oct '13

13 Oct '13

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 c02e99faccdaf94e1ec90775e3f1a880aa7b380f (commit) from e8ed20214777036155f5836464699373eb3ebfae (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 c02e99faccdaf94e1ec90775e3f1a880aa7b380f Author: cengiz <cengiz(a)snslck.(none)> Date: Sun Oct 13 18:38:34 2013 -0400 Added clearance code to asg paper ----------------------------------------------------------------------- Summary of changes: papers/asg-osd/paper.tex | 4 ++++ 1 files changed, 4 insertions(+), 0 deletions(-) Diff of changes: diff --git a/papers/asg-osd/paper.tex b/papers/asg-osd/paper.tex index ac4da83..3e6da95 100644 --- a/papers/asg-osd/paper.tex +++ b/papers/asg-osd/paper.tex @@ -424,6 +424,10 @@ Kevin Harms, Robert Ross} %Telephone: (800) 555--1212, Fax: (888) 555--1212} %\IEEEauthorblockA{\IEEEauthorrefmark{4}Tyrell Inc., 123 Replicant Street, Los Angeles, California 90210--4321}} +\IEEEoverridecommandlockouts +\IEEEpubid{\makebox[\columnwidth]{\hfill + 978-1-4799-0898-1/13/\$31.00~\copyright2013 + IEEE}\hspace{\columnsep}\makebox[\columnwidth]{ }} % make the title area \maketitle hooks/post-receive --

1 0

branch, master, updated. 646669e95bde92bf3ac91cd52c1de00094a71316
by noreply＠mcs.anl.gov 11 Oct '13

11 Oct '13

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 646669e95bde92bf3ac91cd52c1de00094a71316 (commit) from ab1438bc4251850222edd7a985558422d3d9f742 (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 646669e95bde92bf3ac91cd52c1de00094a71316 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 17:26:59 2013 -0400 ability to set replication factor in triton-cp ----------------------------------------------------------------------- Summary of changes: code/src/admin-tools/triton-cp.ae | 19 ++++++++++++++++--- 1 files changed, 16 insertions(+), 3 deletions(-) Diff of changes: diff --git a/code/src/admin-tools/triton-cp.ae b/code/src/admin-tools/triton-cp.ae index 06f4af1..1444c84 100644 --- a/code/src/admin-tools/triton-cp.ae +++ b/code/src/admin-tools/triton-cp.ae @@ -53,6 +53,8 @@ static __blocking int tc_read(struct obj_ref* ref, char* buffer, int size); static __blocking int tc_write(struct obj_ref* ref, const char* buffer, int size); static __blocking int transfer(struct obj_ref *src, struct obj_ref *dest, char* buffer, int size); +int replication_factor = 1; + __blocking int aesop_main(int argc, char **argv) { triton_ret_t tret; @@ -62,14 +64,25 @@ __blocking int aesop_main(int argc, char **argv) struct obj_ref *dest; char* buffer; - if(argc != 4) + if(argc != 4 && argc != 5) { - fprintf(stderr, "Usage: %s <server> <src> <dest>\n", argv[0]); + fprintf(stderr, "Usage: %s <server> <src> <dest> <replication_factor>\n", argv[0]); fprintf(stderr, " (prefix local files with \"posix:\")\n"); fprintf(stderr, " Example: triton-cp tcp://localhost:3344 posix:/tmp/a.dat 123.123\n"); + fprintf(stderr, " The <replication_factor> argument is optional.\n"); return(-1); } + if(argc == 5) + { + ret = sscanf(argv[4], "%d", &replication_factor); + if(ret != 1) + { + fprintf(stderr, "Error: invalid replication factor.\n"); + return(-1); + } + } + svr_string = argv[1]; tret = triton_debug_init(); @@ -233,7 +246,7 @@ static __blocking int tc_open(struct obj_ref* ref, int create_if_needed) if(create_if_needed) { /* TODO: make replication factor configurable */ - tret = remote_triton_rpc_rosd_create(ref->u.triton.oid, 1, 0); + tret = remote_triton_rpc_rosd_create(ref->u.triton.oid, replication_factor, 0); if(triton_error_equal(tret, TRITON_ERR_EXIST)) { fprintf(stderr, "Error: unimplemented; need to truncate triton object.\n"); hooks/post-receive --

1 0

branch, master, updated. ab1438bc4251850222edd7a985558422d3d9f742
by noreply＠mcs.anl.gov 11 Oct '13

11 Oct '13

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 ab1438bc4251850222edd7a985558422d3d9f742 (commit) from ef5437c7084cf7b8bae095600214905030d63c68 (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 ab1438bc4251850222edd7a985558422d3d9f742 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 16:57:07 2013 -0400 initialize rosd txn nr cache - rosd_write() now works for extremely simple cases ----------------------------------------------------------------------- Summary of changes: code/src/replicated-osd/rosd.ae | 38 ++++++++++++++++++++++++++++++++++++++ code/src/replicated-osd/rosd.hae | 3 +++ code/src/server/triton-server.ae | 16 ++++++++++++++++ 3 files changed, 57 insertions(+), 0 deletions(-) Diff of changes: diff --git a/code/src/replicated-osd/rosd.ae b/code/src/replicated-osd/rosd.ae index 4681068..407b036 100644 --- a/code/src/replicated-osd/rosd.ae +++ b/code/src/replicated-osd/rosd.ae @@ -16,6 +16,44 @@ #include "src/system-state/system-state.hae" #include "src/transactional-osd/transactional-osd.hae" +static triton_mutex_t module_lock = TRITON_MUTEX_INITIALIZER; +static int module_refcount = 0; + +triton_ret_t triton_rosd_svr_init(void) +{ + triton_ret_t tret = TRITON_SUCCESS; + + triton_mutex_lock(&module_lock); + + if(!module_refcount) + { + tret = txn_nr_cache_init(1024); + if(triton_is_error(tret)) + { + triton_mutex_unlock(&module_lock); + return(tret); + } + } + module_refcount++; + triton_mutex_unlock(&module_lock); + + return(tret); +} + +void triton_rosd_svr_finalize(void) +{ + triton_mutex_lock(&module_lock); + module_refcount--; + + if (!module_refcount) + { + txn_nr_cache_finalize(); + } + triton_mutex_unlock(&module_lock); + + return; +} + void triton_rpc_rosd_register(void) { triton_rpc_rosd_create_register(); diff --git a/code/src/replicated-osd/rosd.hae b/code/src/replicated-osd/rosd.hae index 3bdf620..7a1d07a 100644 --- a/code/src/replicated-osd/rosd.hae +++ b/code/src/replicated-osd/rosd.hae @@ -18,6 +18,9 @@ */ #define ROSD_FLAG_COND_WRITE 8 +triton_ret_t triton_rosd_svr_init(void); + +void triton_rosd_svr_finalize(void); void triton_rpc_rosd_register(void); diff --git a/code/src/server/triton-server.ae b/code/src/server/triton-server.ae index fde8707..0a0c451 100644 --- a/code/src/server/triton-server.ae +++ b/code/src/server/triton-server.ae @@ -233,6 +233,19 @@ __blocking int aesop_main(int argc, char **argv) } triton_signal_track(SIGINT); triton_signal_track(SIGTERM); + + /* intialize rosd */ + tret = triton_rosd_svr_init(); + if(triton_is_error(tret)) + { + triton_error_msg_ret(tret, "triton_rosd_svr_init()"); + triton_error_destroy(tret); + tosd_finalize(); + triton_debug_finalize(); + triton_signal_finalize(); + return(-1); + + } /* enable mercury rpc engine */ tret = triton_mercury_engine_init(TRITON_MERCURY_SVR|TRITON_MERCURY_CLIENT, listen_addr); @@ -240,6 +253,7 @@ __blocking int aesop_main(int argc, char **argv) { triton_error_msg_ret(tret, "triton_mercury_engine_init()"); triton_error_destroy(tret); + triton_rosd_svr_finalize(); tosd_finalize(); triton_signal_finalize(); triton_debug_finalize(); @@ -253,6 +267,7 @@ __blocking int aesop_main(int argc, char **argv) triton_error_msg_ret(tret, "triton_core_rpc_init()"); triton_error_destroy(tret); triton_mercury_engine_finalize(); + triton_rosd_svr_finalize(); tosd_finalize(); triton_signal_finalize(); triton_debug_finalize(); @@ -318,6 +333,7 @@ __blocking int aesop_main(int argc, char **argv) triton_info_msg("Server shutting down cleanly\n"); + triton_rosd_svr_finalize(); tosd_finalize(); triton_core_rpc_finalize(); triton_mercury_engine_finalize(); hooks/post-receive --

1 0

branch, master, updated. ef5437c7084cf7b8bae095600214905030d63c68
by noreply＠mcs.anl.gov 11 Oct '13

11 Oct '13

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 ef5437c7084cf7b8bae095600214905030d63c68 (commit) via 746e8dedf086f8fda974416a2a0dd4311940857a (commit) via b63edd74b8f9bd0ad386fa0b3ccd550c5e427f0b (commit) via 5732b17235f56552b1a70fdb34418322eecfa86f (commit) from 6e95e1090eb03dd13a684969215c96fd5a724453 (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 ef5437c7084cf7b8bae095600214905030d63c68 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 15:45:01 2013 -0400 finish porting rosd_write(); untested commit 746e8dedf086f8fda974416a2a0dd4311940857a Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 15:02:41 2013 -0400 incremental progress on rosd_write commit b63edd74b8f9bd0ad386fa0b3ccd550c5e427f0b Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 13:49:13 2013 -0400 incremental progress on rosd_write commit 5732b17235f56552b1a70fdb34418322eecfa86f Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 13:43:01 2013 -0400 incremental progress on rosd_write ----------------------------------------------------------------------- Summary of changes: code/src/replicated-osd/rosd-write.ae | 259 +++++++++++++++++++++++++++++++-- code/src/replicated-osd/rosd.hae | 5 +- 2 files changed, 246 insertions(+), 18 deletions(-) Diff of changes: diff --git a/code/src/replicated-osd/rosd-write.ae b/code/src/replicated-osd/rosd-write.ae index 7baa6bd..84c69a1 100644 --- a/code/src/replicated-osd/rosd-write.ae +++ b/code/src/replicated-osd/rosd-write.ae @@ -7,6 +7,7 @@ #include "src/replicated-osd/rosd.hae" #include "src/replicated-osd/rosd-internal.hae" +#include "src/replicated-osd/rosd-txn-nr-cache.hae" #include "src/common/triton-debug.h" #include "src/common/self.h" #include "src/placement/placement.h" @@ -26,7 +27,7 @@ typedef struct uint128_t oid; uint64_t oid_fork; int64_t offset; - uint64_t size; + int64_t size; uint32_t flags; uint32_t replication_factor; /* cannot be set by clients */ uint32_t expected_position; /* cannot be set by clients */ @@ -44,19 +45,41 @@ static __blocking triton_ret_t __remote_triton_rpc_rosd_write( na_addr_t addr, uint128_t oid, uint64_t fork, - const char* buffer, - uint64_t size, + char* buffer, + int64_t size, int64_t offset, uint32_t flags, - uint32_t replicas, + uint32_t replication_factor, uint32_t expected_position, uint64_t txn_number); +static int is_usage_error_write(triton_ret_t tret); +static __blocking triton_ret_t rosd_write_do_work( + na_addr_t next_addr, + uint128_t oid, + uint64_t oid_fork, + char* buffer, + int64_t size, + int64_t offset, + uint32_t flags, + uint32_t replication_factor, + uint64_t txn_number, + int my_position, + int from_client_flag); +static __blocking triton_ret_t rosd_write_local_storage( + uint128_t oid, + uint64_t oid_fork, + uint64_t txn_number, + char *buffer, + int64_t size, + int64_t offset, + uint32_t flags); + __blocking triton_ret_t remote_triton_rpc_rosd_write( uint128_t oid, uint64_t fork, - const char* buffer, - uint64_t size, + char* buffer, + int64_t size, int64_t offset, uint64_t txn_number, uint32_t flags) @@ -88,6 +111,12 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) na_addr_t next_addr; na_addr_t* addr_array; int my_position; + int64_t obj_offset; + int64_t size; + int64_t out_size; + char* buffer_offsets[1]; + int from_client_flag = 0; + struct txn_nr_cache_entry* entry_p = NULL; triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_write()\n"); @@ -103,6 +132,7 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) triton_mercury_start_output(handle, &out); return(TRITON_SUCCESS); } + from_client_flag = 1; /* check our position assuming a replication factor of 1 to make * sure that we are the master for this object @@ -114,11 +144,86 @@ static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) return(TRITON_SUCCESS); } + if(my_position != 0) + { + out.tret = TRITON_ERR_WRONG_SERVER; + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + + /* retrieve the true replication factor from the object */ + obj_offset = 0; + buffer_offsets[0] = (char*)&in.replication_factor; + size = sizeof(in.replication_factor); + out.tret = tosd_read(in.oid, REP_FACTOR_FORK, buffer_offsets, + &size, 1, &obj_offset, &size, 1, &out_size, 0); + if(triton_is_error(out.tret)) + { + triton_error_msg("tosd_read() failure for replication factor\n"); + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + if(out_size != sizeof(in.replication_factor)) + { + assert(out_size == 0); + in.replication_factor = DEFAULT_REPLICATION; + } + + /* TODO: check flags to make sure this isn't an explicit conditional */ + + /* figure out what transaction number to use */ + out.tret = txn_nr_cache_get(&entry_p, in.oid, &in.txn_number, in.flags); + if(triton_is_error(out.tret)) + { + triton_error_msg("txn_nr_cache_get() failure.\n"); + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + } + + /* map oid to server addresses */ + addr_array = malloc(in.replication_factor*sizeof(*addr_array)); + if(!addr_array) + { + if(entry_p) + txn_nr_cache_put(entry_p); + out.tret = TRITON_ERR_NOMEM; + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + + out.tret = triton_oid_to_addrs(in.oid, in.replication_factor, &my_position, + addr_array); + if(triton_is_error(out.tret)) + { + if(entry_p) + txn_nr_cache_put(entry_p); + free(addr_array); + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + + /* check position in chain */ + if(my_position != in.expected_position) + { + if(entry_p) + txn_nr_cache_put(entry_p); + out.tret = TRITON_ERR_WRONG_SERVER; + free(addr_array); + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + if(my_position < (in.replication_factor-1)) + { + next_addr = addr_array[my_position+1]; } - /* TODO: fill this in. Just returning success for now to tie into test - * program. - */ - out.tret = TRITON_SUCCESS; + + out.tret = rosd_write_do_work(next_addr, in.oid, in.oid_fork, + in.eager_payload, in.eager_payload_size, in.offset, in.flags, + in.replication_factor, in.txn_number, my_position, from_client_flag); + + if(entry_p) + txn_nr_cache_put(entry_p); triton_mercury_start_output(handle, &out); @@ -154,11 +259,11 @@ static __blocking triton_ret_t __remote_triton_rpc_rosd_write( na_addr_t addr, uint128_t oid, uint64_t fork, - const char* buffer, - uint64_t size, + char* buffer, + int64_t size, int64_t offset, uint32_t flags, - uint32_t replicas, + uint32_t replication_factor, uint32_t expected_position, uint64_t txn_number) { @@ -182,7 +287,7 @@ static __blocking triton_ret_t __remote_triton_rpc_rosd_write( in.size = size; in.offset = offset; in.flags = flags; - in.replication_factor = replicas; + in.replication_factor = replication_factor; in.expected_position = expected_position; in.txn_number = txn_number; in.eager_payload = buffer; @@ -240,7 +345,7 @@ static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_wr ret = HG_FAIL; return ret; } - ret = hg_proc_uint64_t(proc, &in_p->size); + ret = hg_proc_int64_t(proc, &in_p->size); if (ret != HG_SUCCESS) { HG_ERROR_DEFAULT("Proc error"); ret = HG_FAIL; @@ -302,7 +407,7 @@ static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_wr ret = HG_FAIL; return ret; } - ret = hg_proc_uint64_t(proc, &in_p->size); + ret = hg_proc_int64_t(proc, &in_p->size); if (ret != HG_SUCCESS) { HG_ERROR_DEFAULT("Proc error"); ret = HG_FAIL; @@ -369,6 +474,128 @@ static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_wr return ret; } +static __blocking triton_ret_t rosd_write_do_work( + na_addr_t next_addr, + uint128_t oid, + uint64_t oid_fork, + char* buffer, + int64_t size, + int64_t offset, + uint32_t flags, + uint32_t replication_factor, + uint64_t txn_number, + int my_position, + int from_client_flag) +{ + triton_ret_t local_tret; + triton_ret_t remote_tret = TRITON_SUCCESS; + triton_ret_t tret; + + /* right now only chained replication is supported */ + assert(!(flags & ROSD_FLAG_FANOUT)); + flags |= ROSD_FLAG_CHAIN; + + /* TODO: implement conditionals */ + pwait + { + pprivate char oid_str[TRITON_UINT128_STRLEN]; + pbranch + { + /* forward on to peers if necessary */ + if(my_position < replication_factor-1) + { + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); + triton_debug(triton_dbg_rosd, "ROSD forwarding write of oid %s to %d'th server.\n", oid_str, my_position+1); + + remote_tret = __remote_triton_rpc_rosd_write( + next_addr, oid, oid_fork, buffer, size, offset, flags, replication_factor, + my_position+1, txn_number); + } + } + pbranch + { + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); + triton_debug(triton_dbg_rosd, "ROSD local write of oid %s on %d'th server, transaction number %llu.\n", oid_str, my_position, llu(txn_number)); + /* perform local operation */ + local_tret = rosd_write_local_storage(oid, oid_fork, txn_number, + buffer, size, offset, flags); + } + } + + if(my_position < replication_factor-1) + tret = interpret_errors(local_tret, remote_tret, my_position, + is_usage_error_write, from_client_flag); + else + tret = interpret_error(local_tret, my_position, is_usage_error_write, + from_client_flag); + + 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 + * error code, 0 if storage fault. + */ +static int is_usage_error_write(triton_ret_t tret) +{ + /* these are the scenarios that a write could fail because of a + * usage (ie, semantic) error rather than because of a system fault. + */ + if(triton_error_equal(tret, TRITON_ERR_NO_SERVERS) || + triton_error_equal(tret, TRITON_ERR_NOENT) || + triton_error_equal(tret, TRITON_ERR_ACCES)) + { + /* if there aren't enough servers to satisfy the replication + * request, or the object doesn't exist, or if + * permission was denied. + */ + return(1); + } + + return(0); +} + +/* rosd_write_local_storage() + * + * performs the local steps needed in a replicated write + */ +static __blocking triton_ret_t rosd_write_local_storage( + uint128_t oid, + uint64_t oid_fork, + uint64_t txn_number, + char *buffer, + int64_t size, + int64_t offset, + uint32_t flags +) +{ + triton_ret_t tret; + uint64_t niid = 0; + int value = 0; + char* buffer_offsets[1]; + int tosd_flags = TOSD_FLAG_AUTO_TXN; + triton_node_t self; + triton_string_t self_str; + + if(flags & ROSD_FLAG_TRUNC_WRITE) + tosd_flags |= TOSD_FLAG_TRUNC_WRITE; + + /* NOTE: a typical write is idempotent; no need for niid information + * here + */ + buffer_offsets[0] = buffer; + tret = tosd_write(oid, oid_fork, txn_number, buffer_offsets, &size, + 1, &offset, &size, 1, tosd_flags, 0, NULL); + if(triton_is_error(tret)) + { + triton_error_msg("tosd_write() failure.\n"); + } + + return(tret); +} + + /* * Local Variables: * c-basic-offset: 4 diff --git a/code/src/replicated-osd/rosd.hae b/code/src/replicated-osd/rosd.hae index 9366321..3bdf620 100644 --- a/code/src/replicated-osd/rosd.hae +++ b/code/src/replicated-osd/rosd.hae @@ -39,11 +39,12 @@ __blocking triton_ret_t remote_triton_rpc_rosd_range( uint32_t *oid_array_count, uint128_t *next_position); +/* TODO: make const correct (needs some downstream fixes to do) */ __blocking triton_ret_t remote_triton_rpc_rosd_write( uint128_t oid, uint64_t fork, - const char* buffer, - uint64_t size, + char* buffer, + int64_t size, int64_t offset, uint64_t txn_number, uint32_t flags); hooks/post-receive --

1 0

branch, master, updated. 6e95e1090eb03dd13a684969215c96fd5a724453
by noreply＠mcs.anl.gov 11 Oct '13

11 Oct '13

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 6e95e1090eb03dd13a684969215c96fd5a724453 (commit) via 50dec515f2fbe18e91d14fb5477460c5e011be6d (commit) from 6ae4fa96eae576fc24d44f4e8c3025838f33b1d2 (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 6e95e1090eb03dd13a684969215c96fd5a724453 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 12:37:25 2013 -0400 split rosd.ae into separate files for each op commit 50dec515f2fbe18e91d14fb5477460c5e011be6d Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 11 11:29:14 2013 -0400 first step to splitting up rosd.ae ----------------------------------------------------------------------- Summary of changes: code/src/replicated-osd/Makefile.subdir | 15 +- code/src/replicated-osd/rosd-create.ae | 345 ++++++++ code/src/replicated-osd/rosd-internal.hae | 49 ++ code/src/replicated-osd/rosd-range.ae | 241 ++++++ code/src/replicated-osd/rosd-remove.ae | 418 ++++++++++ code/src/replicated-osd/rosd-write.ae | 378 +++++++++ code/src/replicated-osd/rosd.ae | 1269 +---------------------------- 7 files changed, 1452 insertions(+), 1263 deletions(-) create mode 100644 code/src/replicated-osd/rosd-create.ae create mode 100644 code/src/replicated-osd/rosd-internal.hae create mode 100644 code/src/replicated-osd/rosd-range.ae create mode 100644 code/src/replicated-osd/rosd-remove.ae create mode 100644 code/src/replicated-osd/rosd-write.ae Diff of changes: diff --git a/code/src/replicated-osd/Makefile.subdir b/code/src/replicated-osd/Makefile.subdir index 2aae65c..db4c94d 100644 --- a/code/src/replicated-osd/Makefile.subdir +++ b/code/src/replicated-osd/Makefile.subdir @@ -1,12 +1,21 @@ src_libtriton_a_SOURCES += \ src/replicated-osd/rosd.ae \ - src/replicated-osd/rosd-txn-nr-cache.ae + src/replicated-osd/rosd-txn-nr-cache.ae \ + src/replicated-osd/rosd-create.ae \ + src/replicated-osd/rosd-remove.ae \ + src/replicated-osd/rosd-range.ae \ + src/replicated-osd/rosd-write.ae AE_SRC += \ src/replicated-osd/rosd.ae \ - src/replicated-osd/rosd-txn-nr-cache.ae + src/replicated-osd/rosd-txn-nr-cache.ae \ + src/replicated-osd/rosd-create.ae \ + src/replicated-osd/rosd-remove.ae \ + src/replicated-osd/rosd-range.ae \ + src/replicated-osd/rosd-write.ae AE_HDR += \ src/replicated-osd/rosd.hae \ - src/replicated-osd/rosd-txn-nr-cache.hae + src/replicated-osd/rosd-txn-nr-cache.hae \ + src/replicated-osd/rosd-internal.hae diff --git a/code/src/replicated-osd/rosd-create.ae b/code/src/replicated-osd/rosd-create.ae new file mode 100644 index 0000000..918018d --- /dev/null +++ b/code/src/replicated-osd/rosd-create.ae @@ -0,0 +1,345 @@ +#include <aesop/aesop.h> +#include <aesop/timer.hae> +#include <triton-list.h> +#include <triton-hash.h> +#include <mercury_macros.h> +#include <mercury_proc.h> + +#include "src/replicated-osd/rosd.hae" +#include "src/replicated-osd/rosd-internal.hae" +#include "src/common/triton-debug.h" +#include "src/common/self.h" +#include "src/placement/placement.h" +#include "src/system-state/system-state.hae" +#include "src/remote/mercury-engine.hae" +#include "src/remote/mercury-encode.h" +#include "src/system-state/system-state.hae" +#include "src/transactional-osd/transactional-osd.hae" + + +/* Mercury RPC structures for rosd_create */ +MERCURY_GEN_PROC(triton_rpc_rosd_create_out_t, ((triton_ret_t)(tret))) +MERCURY_GEN_PROC(triton_rpc_rosd_create_in_t, ((uint128_t)(oid))((uint32_t)(flags))((uint32_t)(expected_position))((uint32_t)(replication_factor))((uint64_t)(niid))) + +static hg_id_t rpc_rosd_create_id; +static int triton_rpc_rosd_create_handler(hg_handle_t handle); +static __blocking triton_ret_t rosd_create_do_work( + uint128_t oid, + uint32_t flags, + uint32_t replication_factor, + int my_position, + na_addr_t next_addr, + uint64_t niid, + int from_client_flag); +static int is_usage_error_create(triton_ret_t tret); +static __blocking triton_ret_t rosd_create_local_storage( + uint128_t oid, + uint32_t replication_factor, + uint64_t niid); + +static __blocking triton_ret_t triton_rpc_rosd_create(hg_handle_t handle) +{ + triton_rpc_rosd_create_out_t out; + triton_rpc_rosd_create_in_t in; + int ret = 0; + char oid_str[TRITON_UINT128_STRLEN]; + int my_position; + na_addr_t *addr_array = NULL; + int got_addrs = 0; + int i; + na_addr_t next_addr; + + triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_create()\n"); + + triton_mercury_get_input(handle, &in, &out); + + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, in.oid); + + triton_debug(triton_dbg_rosd, "triton_rpc_rosd_create() params: oid: %s, flags: %d, replication_factor: %d, niid: %llu\n", + oid_str, in.flags, in.replication_factor, llu(in.niid)); + + out.tret = TRITON_SUCCESS; + + /* map oid to server addresses */ + addr_array = malloc(in.replication_factor*sizeof(*addr_array)); + if(!addr_array) + { + out.tret = TRITON_ERR_NOMEM; + } + if(!triton_is_error(out.tret)) + { + out.tret = triton_oid_to_addrs(in.oid, in.replication_factor, &my_position, + addr_array); + if(!triton_is_error(out.tret)) + got_addrs = 1; + } + + if(!triton_is_error(out.tret)) + { + /* check position in chain */ + if(my_position != in.expected_position) + { + out.tret = TRITON_ERR_WRONG_SERVER; + } + } + if(my_position < (in.replication_factor-1)) + { + next_addr = addr_array[my_position+1]; + } + + out.tret = rosd_create_do_work(in.oid, in.flags, in.replication_factor, + my_position, next_addr, in.niid, 1); + + triton_mercury_start_output(handle, &out); + + for(i=0; i<in.replication_factor && got_addrs; i++) + triton_mercury_addr_free(addr_array[i]); + if(addr_array) + free(addr_array); + + return(TRITON_SUCCESS); +} +TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_create) + +__blocking triton_ret_t __remote_triton_rpc_rosd_create( + na_addr_t addr, + uint128_t oid, + uint32_t replicas, + uint32_t flags, + uint32_t expected_position) +{ + triton_rpc_rosd_create_out_t out; + triton_rpc_rosd_create_in_t in; + hg_request_t request; + int ret; + triton_ret_t tret; + int position; + + in.oid = oid; + in.flags = flags; + in.replication_factor = replicas; + /* TODO: generate real NIIDs */ + in.niid = 0; + in.expected_position = expected_position; + + ret = HG_Forward(addr, rpc_rosd_create_id, &in, + &out, &request); + if(ret != HG_SUCCESS) + { + triton_error_msg("HG_Forward() failure.\n"); + return(TRITON_ERR_UNKNOWN); + } + + tret = triton_mercury_wait(request); + if(triton_is_error(tret)) + { + HG_Request_free(request); + return(tret); + } + + tret = triton_error_dup(out.tret); + + HG_Request_free(request); + + return(tret); +} + +__blocking triton_ret_t remote_triton_rpc_rosd_create( + uint128_t oid, + uint32_t replicas, + uint32_t flags) +{ + triton_ret_t tret; + na_addr_t addr; + int position; + + /* we contact the first server (master) for oid */ + tret = triton_oid_to_addrs(oid, 1, &position, &addr); + if(triton_is_error(tret)) + { + return(tret); + } + + tret = __remote_triton_rpc_rosd_create(addr, oid, replicas, flags, 0); + + triton_mercury_addr_free(addr); + + return(tret); +} + +static __blocking triton_ret_t rosd_create_do_work( + uint128_t oid, + uint32_t flags, + uint32_t replication_factor, + int my_position, + na_addr_t next_addr, + uint64_t niid, + int from_client_flag) +{ + triton_ret_t local_tret; + triton_ret_t remote_tret = TRITON_SUCCESS; + triton_ret_t tret; + + /* right now only chained replication is supported */ + assert(!(flags & ROSD_FLAG_FANOUT)); + flags |= ROSD_FLAG_CHAIN; + + /* forward request while simultaneously performing local work */ + pwait + { + pprivate char oid_str[TRITON_UINT128_STRLEN]; + pbranch + { + if(my_position < replication_factor -1) + { + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); + triton_debug(triton_dbg_rosd, + "ROSD forwarding create of oid %s to %d'th server.\n", + oid_str, my_position+1); + + remote_tret = __remote_triton_rpc_rosd_create(next_addr, oid, + replication_factor, + flags, my_position+1); + } + } + pbranch + { + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); + triton_debug(triton_dbg_rosd, "ROSD local create of oid %s on %d'th server.\n", oid_str, my_position); + /* perform local operation */ + local_tret = rosd_create_local_storage(oid, replication_factor, niid); + } + } + + if(my_position < replication_factor -1) + tret = interpret_errors(local_tret, remote_tret, my_position, + is_usage_error_create, from_client_flag); + else + tret = interpret_error(local_tret, my_position, is_usage_error_create, + from_client_flag); + + 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 + * error code, 0 if storage fault. + */ +static int is_usage_error_create(triton_ret_t tret) +{ + /* these are the scenarios that a create could fail because of a + * usage (ie, semantic) error rather than because of a system fault. + */ + if(triton_error_equal(tret, TRITON_ERR_NO_SERVERS) || + triton_error_equal(tret, TRITON_ERR_EXIST) || + triton_error_equal(tret, TRITON_ERR_ACCES)) + { + /* if there aren't enough servers to satisfy the replication + * request, or if the object in question already exists, or if + * permission was denied. + */ + return(1); + } + + return(0); +} + +/* rosd_create_local_storage() + * + * performs the local steps needed in a replicated create + */ +static __blocking triton_ret_t rosd_create_local_storage( + uint128_t oid, + uint32_t replication_factor, + uint64_t niid +) +{ + triton_ret_t tret; + triton_ret_t tret_tmp; + int value = 0; + int64_t obj_offset; + int64_t size; + char* buffer_offsets[1]; + triton_node_t self; + + self = triton_self_node(); + + tret = tosd_create(oid, niid); + if(triton_error_equal(tret, TRITON_ERR_NIID_DONE)) + { + /* this operation is already done */ + triton_error_destroy(tret); + tret = TRITON_SUCCESS; + } + + if(tret != TRITON_SUCCESS) + { + return triton_error_wrap(tret, self, "tosd_create failed"); + } + + if(replication_factor != DEFAULT_REPLICATION) + { + + /* store replication factor directly in a data fork */ + obj_offset = 0; + buffer_offsets[0] = (char*)&replication_factor; + size = sizeof(replication_factor); + + /* NOTE: no need for tracking NIID for this portion of the operation. + * There is no harm in writing the replication factor twice. We do need + * to look out for conflicts on the transaction number, though. + */ + tret = tosd_write(oid, REP_FACTOR_FORK, 1, buffer_offsets, + &size, 1, &obj_offset, &size, 1, TOSD_FLAG_AUTO_TXN, 0, NULL); + if(tret != TRITON_SUCCESS) + { + /* We created the object but weren't able to write its metadata. + * Give a best effort here to remove the object in order to simplify + * cleanup later. It's ok if this does not succeed. + */ + tret_tmp = tosd_remove(oid, 0); + if(tret_tmp != TRITON_SUCCESS) + { + triton_error_destroy(tret_tmp); + } + /* don't propagate tret_tmp to the caller; we want the caller to see + * the original error code instead, not side-effects from cleanup. + */ + } + } + + return(tret); +} + +void triton_rpc_rosd_create_register(void) +{ + int mflags = triton_mercury_engine_get_flags(); + + assert(mflags); + + if(mflags & TRITON_MERCURY_SVR) + { + MERCURY_HANDLER_REGISTER("triton_rpc_rosd_create", + triton_rpc_rosd_create_handler, + triton_rpc_rosd_create_in_t, + triton_rpc_rosd_create_out_t); + } + + if(mflags & TRITON_MERCURY_CLIENT) + { + rpc_rosd_create_id = MERCURY_REGISTER("triton_rpc_rosd_create", + triton_rpc_rosd_create_in_t, triton_rpc_rosd_create_out_t); + } + + return; +} + + +/* + * Local Variables: + * c-basic-offset: 4 + * End: + * + * vim: ft=c ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/replicated-osd/rosd-internal.hae b/code/src/replicated-osd/rosd-internal.hae new file mode 100644 index 0000000..e0c4c38 --- /dev/null +++ b/code/src/replicated-osd/rosd-internal.hae @@ -0,0 +1,49 @@ +#ifndef __ROSD_INTERNAL_HAE__ +#define __ROSD_INTERNAL_HAE__ + +/* This header provides prototypes for functionality shared among the + * various ROSD operations, but that doesn't need to be exposed to external + * users of the API. + */ + +#include "src/replicated-osd/rosd.hae" + +/* TODO: need a way to configure and/or reserve these values */ +/* default replication level for objects */ +#define DEFAULT_REPLICATION 3 +/* default fork used to store replication factor in object */ +#define REP_FACTOR_FORK UINT64_MAX +#define EAGER_THRESHOLD 4096 + +triton_ret_t triton_oid_to_addrs(uint128_t oid, + uint32_t replication_factor, int* my_position, na_addr_t *addr_array); + +__blocking triton_ret_t interpret_errors( + triton_ret_t local_error_code, + triton_ret_t remote_error_code, + int my_position, + int(*usage_error_check)(triton_ret_t tret), + int from_client_flag); + +__blocking triton_ret_t interpret_error( + triton_ret_t local_error_code, + int my_position, + int(*usage_error_check)(triton_ret_t tret), + int from_client_flag); + +__blocking void trigger_server_fault(triton_ret_t tret); + +void triton_rpc_rosd_create_register(void); +void triton_rpc_rosd_range_register(void); +void triton_rpc_rosd_remove_register(void); +void triton_rpc_rosd_write_register(void); + +#endif /* __ROSD_INTERNAL_HAE */ + +/* + * Local Variables: + * c-basic-offset: 4 + * End: + * + * vim: ft=c ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/replicated-osd/rosd-range.ae b/code/src/replicated-osd/rosd-range.ae new file mode 100644 index 0000000..9b30bd4 --- /dev/null +++ b/code/src/replicated-osd/rosd-range.ae @@ -0,0 +1,241 @@ +#include <aesop/aesop.h> +#include <aesop/timer.hae> +#include <triton-list.h> +#include <triton-hash.h> +#include <mercury_macros.h> +#include <mercury_proc.h> + +#include "src/replicated-osd/rosd.hae" +#include "src/replicated-osd/rosd-internal.hae" +#include "src/common/triton-debug.h" +#include "src/common/self.h" +#include "src/placement/placement.h" +#include "src/system-state/system-state.hae" +#include "src/remote/mercury-engine.hae" +#include "src/remote/mercury-encode.h" +#include "src/system-state/system-state.hae" +#include "src/transactional-osd/transactional-osd.hae" + +/* Mercury RPC structures for rosd_range */ +MERCURY_GEN_PROC(triton_rpc_rosd_range_in_t, + ((uint128_t)(lower))\ + ((uint128_t)(upper))\ + ((uint32_t)(count))) +/* output struct for rosd_range has a manual encoder because it contains + * an array + */ +typedef struct +{ + triton_ret_t tret; + uint128_t next_position; + uint128_t* oid_array; + uint32_t oid_array_count; +} triton_rpc_rosd_range_out_t; +static int hg_proc_triton_rpc_rosd_range_out_t(hg_proc_t proc, triton_rpc_rosd_range_out_t* out_p); + +static hg_id_t rpc_rosd_range_id; +static int triton_rpc_rosd_range_handler(hg_handle_t handle); + +static __blocking triton_ret_t triton_rpc_rosd_range(hg_handle_t handle) +{ + triton_rpc_rosd_range_out_t out; + triton_rpc_rosd_range_in_t in; + int ret = 0; + int out_count; + + triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_range()\n"); + + triton_mercury_get_input(handle, &in, &out); + + out.tret = TRITON_SUCCESS; + triton_uint128_setmax(out.next_position); + out.oid_array_count = 0; + out.oid_array = (uint128_t*)malloc(in.count*sizeof(*out.oid_array)); + if(!out.oid_array) + { + out.tret = TRITON_ERR_NOMEM; + } + + if(!triton_is_error(out.tret)) + { + out.tret = tosd_range(in.lower, in.upper, out.oid_array, in.count, &out_count, &out.next_position); + } + assert(out_count >= 0); + out.oid_array_count = out_count; + + /* TODO: maybe an option to filter out non-master objects? See old + * rosd.aer implementation, where this was the default behavior. + */ + + triton_mercury_start_output(handle, &out); + + if(out.oid_array) + free(out.oid_array); + + return(TRITON_SUCCESS); +} +TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_range) + +__blocking triton_ret_t remote_triton_rpc_rosd_range( + na_addr_t addr, + uint128_t lower, + uint128_t upper, + uint32_t count, + uint128_t *oid_array, + uint32_t *oid_array_count, + uint128_t *next_position) +{ + triton_ret_t tret; + triton_rpc_rosd_range_out_t out; + triton_rpc_rosd_range_in_t in; + hg_request_t request; + int ret; + + in.lower = lower; + in.upper = upper; + in.count = count; + + ret = HG_Forward(addr, rpc_rosd_range_id, &in, + &out, &request); + if(ret != HG_SUCCESS) + { + triton_error_msg("HG_Forward() failure.\n"); + return(TRITON_ERR_UNKNOWN); + } + + tret = triton_mercury_wait(request); + if(triton_is_error(tret)) + { + HG_Request_free(request); + return(tret); + } + + *oid_array_count = out.oid_array_count; + memcpy(oid_array, out.oid_array, out.oid_array_count*sizeof(*oid_array)); + *next_position = out.next_position; + tret = triton_error_dup(out.tret); + + HG_Request_free(request); + + return(tret); +} +void triton_rpc_rosd_range_register(void) +{ + int mflags = triton_mercury_engine_get_flags(); + + assert(mflags); + + if(mflags & TRITON_MERCURY_SVR) + { + MERCURY_HANDLER_REGISTER("triton_rpc_rosd_range", + triton_rpc_rosd_range_handler, + triton_rpc_rosd_range_in_t, + triton_rpc_rosd_range_out_t); + } + + if(mflags & TRITON_MERCURY_CLIENT) + { + rpc_rosd_range_id = MERCURY_REGISTER("triton_rpc_rosd_range", + triton_rpc_rosd_range_in_t, + triton_rpc_rosd_range_out_t); + } + + return; +} + +/* TODO: refactor this */ +static int hg_proc_triton_rpc_rosd_range_out_t(hg_proc_t proc, triton_rpc_rosd_range_out_t* out_p) +{ + int ret = HG_FAIL; + int i; + + switch (hg_proc_get_op(proc)) { + case HG_ENCODE: + ret = hg_proc_triton_ret_t(proc, &out_p->tret); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint128_t(proc, &out_p->next_position); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &out_p->oid_array_count); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + for(i=0; i<out_p->oid_array_count; i++) + { + ret = hg_proc_uint128_t(proc, &out_p->oid_array[i]); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + } + break; + case HG_DECODE: + ret = hg_proc_triton_ret_t(proc, &out_p->tret); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint128_t(proc, &out_p->next_position); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &out_p->oid_array_count); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + out_p->oid_array = (uint128_t*)malloc(out_p->oid_array_count* + sizeof(*out_p->oid_array)); + if(!out_p->oid_array) { + HG_ERROR_DEFAULT("Out of memory"); + ret = HG_FAIL; + return ret; + } + for(i=0; i<out_p->oid_array_count; i++) + { + ret = hg_proc_uint128_t(proc, &out_p->oid_array[i]); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + } + break; + case HG_FREE: + if (!out_p->oid_array) { + HG_ERROR_DEFAULT("Already freed"); + ret = HG_FAIL; + return ret; + } + free(out_p->oid_array); + out_p->oid_array = NULL; + ret = HG_SUCCESS; + break; + default: + break; + } + + return ret; +} + +/* + * Local Variables: + * c-basic-offset: 4 + * End: + * + * vim: ft=c ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/replicated-osd/rosd-remove.ae b/code/src/replicated-osd/rosd-remove.ae new file mode 100644 index 0000000..163e752 --- /dev/null +++ b/code/src/replicated-osd/rosd-remove.ae @@ -0,0 +1,418 @@ +#include <aesop/aesop.h> +#include <aesop/timer.hae> +#include <triton-list.h> +#include <triton-hash.h> +#include <mercury_macros.h> +#include <mercury_proc.h> + +#include "src/replicated-osd/rosd.hae" +#include "src/replicated-osd/rosd-internal.hae" +#include "src/common/triton-debug.h" +#include "src/common/self.h" +#include "src/placement/placement.h" +#include "src/system-state/system-state.hae" +#include "src/remote/mercury-engine.hae" +#include "src/remote/mercury-encode.h" +#include "src/system-state/system-state.hae" +#include "src/transactional-osd/transactional-osd.hae" + +/* + * Mercury RPC structures for rosd_remove + */ +MERCURY_GEN_PROC(triton_rpc_rosd_remove_out_t, + ((triton_ret_t)(tret))) +MERCURY_GEN_PROC(triton_rpc_rosd_remove_in_t, + ((uint128_t)(oid))\ + ((uint32_t)(flags))\ + ((uint32_t)(expected_position))\ + ((uint32_t)(replication_factor))\ + ((uint64_t)(niid))) + + +static int is_usage_error_remove(triton_ret_t tret); +static hg_id_t rpc_rosd_remove_id; +static __blocking triton_ret_t triton_rpc_rosd_remove(hg_handle_t handle); +static int triton_rpc_rosd_remove_handler(hg_handle_t handle); +static __blocking triton_ret_t rosd_remove_local_storage(uint128_t oid, uint64_t niid); +static __blocking triton_ret_t rosd_remove_do_work( + uint128_t oid, + uint32_t flags, + uint32_t replication_factor, + int my_position, + na_addr_t next_addr, + uint64_t niid, + int from_client_flag); +__blocking triton_ret_t __remote_triton_rpc_rosd_remove( + na_addr_t addr, + uint128_t oid, + uint32_t replicas, + uint32_t flags, + uint32_t expected_position); +__blocking triton_ret_t remote_triton_rpc_rosd_remove( + uint128_t oid, + uint32_t flags); + +void triton_rpc_rosd_remove_register(void) +{ + int mflags = triton_mercury_engine_get_flags(); + + assert(mflags); + + if(mflags & TRITON_MERCURY_SVR) + { + MERCURY_HANDLER_REGISTER("triton_rpc_rosd_remove", + triton_rpc_rosd_remove_handler, + triton_rpc_rosd_remove_in_t, + triton_rpc_rosd_remove_out_t); + } + + if(mflags & TRITON_MERCURY_CLIENT) + { + rpc_rosd_remove_id = MERCURY_REGISTER("triton_rpc_rosd_remove", + triton_rpc_rosd_remove_in_t, + triton_rpc_rosd_remove_out_t); + } + + return; +} + +/* + * triton_rpc_rosd_remove + * - main handler for the rosd_remove RPC + */ +static __blocking triton_ret_t triton_rpc_rosd_remove(hg_handle_t handle) +{ + triton_rpc_rosd_remove_out_t out; + triton_rpc_rosd_remove_in_t in; + triton_ret_t trc; + int my_position; + int ret; + int i; + na_addr_t *addr_array; + na_addr_t next_addr; + char oid_str[TRITON_UINT128_STRLEN]; + + triton_ret_t tret; + uint32_t replication_factor; + char* mem_offsets[1]; + int64_t obj_offsets[1]; + int64_t mem_sizes[1]; + int64_t obj_sizes[1]; + int64_t out_size; + + triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_remove()\n"); + + memset(&out, 0, sizeof(triton_rpc_rosd_remove_out_t)); + addr_array = NULL; + trc = TRITON_SUCCESS; + + triton_mercury_get_input(handle, &in, &out); + + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, in.oid); + triton_debug(triton_dbg_rosd, + "triton_rpc_rosd_remove() params: oid: %s, flags: %d, replication_factor: %d, niid: %llu, position:%d\n", + oid_str, + in.flags, + in.replication_factor, + llu(in.niid), + in.expected_position); + + /* + * A replication factor of zero indicates this is the initial client + * request and we need to lookup the replication factor within the object. + */ + if (in.replication_factor == 0) + { + obj_offsets[0] = 0; + mem_offsets[0] = (char*) &replication_factor; + mem_sizes[0] = sizeof(replication_factor); + obj_sizes[0] = sizeof(replication_factor); + + tret = tosd_read(in.oid, + REP_FACTOR_FORK, + mem_offsets, + mem_sizes, + 1, + obj_offsets, + obj_sizes, + 1, + &out_size, + 0); + if (triton_is_error(tret)) + { + if (triton_error_equal(tret, TRITON_ERR_NOENT)) + { + // TODO + // handle NIID logic here if needed + } + out.tret = tret; + trc = tret; + } + else + { + if (out_size == 0) + { + in.replication_factor = DEFAULT_REPLICATION; + } + else + { + in.replication_factor = replication_factor; + } + triton_debug(triton_dbg_rosd, "found replication factor of: %d\n", in.replication_factor); + } + triton_error_destroy(tret); + } + + if (triton_error_equal(trc, TRITON_SUCCESS)) + { + addr_array = malloc(in.replication_factor*sizeof(*addr_array)); + if(!addr_array) + { + out.tret = TRITON_ERR_NOMEM; + } + else if (triton_error_equal(trc, TRITON_SUCCESS)) + { + memset(addr_array, 0, in.replication_factor*sizeof(*addr_array)); + out.tret = triton_oid_to_addrs(in.oid, + in.replication_factor, + &my_position, + addr_array); + if (!triton_is_error(out.tret)) + { + if (my_position == in.expected_position) + { + if (my_position < (in.replication_factor-1)) + { + next_addr = addr_array[my_position+1]; + } + else + { + next_addr = NA_ADDR_NULL; + } + + out.tret = rosd_remove_do_work(in.oid, + in.flags, + in.replication_factor, + my_position, + next_addr, + in.niid, + 1); + } + else + { + out.tret = TRITON_ERR_WRONG_SERVER; + } + } + } + } + + triton_mercury_start_output(handle, &out); + trc = TRITON_SUCCESS; + + /* + * Free resources + */ + for (i=0; ((i < in.replication_factor) && (addr_array) && (addr_array[i])); i++) + { + triton_mercury_addr_free(addr_array[i]); + } + if (addr_array) + { + free(addr_array); + } + + return trc; +} +TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_remove) + +static __blocking triton_ret_t rosd_remove_do_work( + uint128_t oid, + uint32_t flags, + uint32_t replication_factor, + int my_position, + na_addr_t next_addr, + uint64_t niid, + int from_client_flag) +{ + triton_ret_t local_tret; + triton_ret_t remote_tret = TRITON_SUCCESS; + triton_ret_t tret; + + /* right now only chained replication is supported */ + assert(!(flags & ROSD_FLAG_FANOUT)); + flags |= ROSD_FLAG_CHAIN; + + pwait + { + pprivate char oid_str[TRITON_UINT128_STRLEN]; + pbranch + { + /* forward on to peers if necessary */ + if(my_position < replication_factor-1) + { + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); + triton_debug(triton_dbg_rosd, "ROSD forwarding remove of oid %s to %d'th server.\n", oid_str, my_position+1); + + remote_tret = __remote_triton_rpc_rosd_remove( + next_addr, + oid, + replication_factor, + flags, + my_position+1); + } + } + pbranch + { + triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); + triton_debug(triton_dbg_rosd, "ROSD local remove of oid %s on %d'th server.\n", oid_str, my_position); + /* perform local operation */ + local_tret = rosd_remove_local_storage(oid, niid); + } + } + + if(my_position < replication_factor -1) + { + tret = interpret_errors(local_tret, remote_tret, my_position, + is_usage_error_remove, from_client_flag); + } + else + { + tret = interpret_error(local_tret, my_position, is_usage_error_remove, + from_client_flag); + } + + return(tret); +} + +/* rosd_remove_local_storage + * - removes actual data on storage + */ +__blocking triton_ret_t rosd_remove_local_storage( + uint128_t oid, + uint64_t niid) +{ + triton_ret_t tret; + int value = 0; + + tret = tosd_remove(oid, niid); + if(triton_error_equal(tret, TRITON_ERR_NIID_DONE)) + { + /* this operation is already done */ + triton_error_destroy(tret); + tret = TRITON_SUCCESS; + } + + return(tret); +} + +/* + * __remote_triton_rpc_rosd_remove + * - initiate RPC request for rosd_remove and wait for it to complete. + */ +__blocking triton_ret_t __remote_triton_rpc_rosd_remove( + na_addr_t addr, + uint128_t oid, + uint32_t replicas, + uint32_t flags, + uint32_t expected_position) +{ + triton_rpc_rosd_remove_out_t out; + triton_rpc_rosd_remove_in_t in; + hg_request_t request; + int ret; + triton_ret_t tret; + int position; + char oid_str[TRITON_UINT128_STRLEN]; + + in.oid = oid; + in.flags = flags; + in.replication_factor = replicas; + in.niid = 0; /* TODO: generate real NIIDs */ + in.expected_position = expected_position; + + triton_uint128_to_string(oid_str, sizeof(oid_str), in.oid); + triton_debug(triton_dbg_rosd, "__remote_triton_rpc_rosd_remove: post: oid=%s flags=%d rf=%d niid=%lld pos=%d\n", oid_str, in.flags, in.replication_factor, lld(in.niid), in.expected_position); + + ret = HG_Forward(addr, + rpc_rosd_remove_id, + &in, + &out, + &request); + if(ret != HG_SUCCESS) + { + triton_error_msg("HG_Forward() failure.\n"); + return(TRITON_ERR_UNKNOWN); + } + + tret = triton_mercury_wait(request); + if(triton_is_error(tret)) + { + HG_Request_free(request); + return(tret); + } + + tret = triton_error_dup(out.tret); + + HG_Request_free(request); + + return(tret); +} + +/* + * remote_triton_rpc_rosd_remove + * - wrapper for initiating RPC for rosd_remove + */ +__blocking triton_ret_t remote_triton_rpc_rosd_remove( + uint128_t oid, + uint32_t flags) +{ + triton_ret_t tret; + na_addr_t addr; + int position; + + /* we contact the first server (master) for oid */ + tret = triton_oid_to_addrs(oid, 1, &position, &addr); + if(triton_is_error(tret)) + { + return(tret); + } + + tret = __remote_triton_rpc_rosd_remove(addr, oid, 0, flags, 0); + + triton_mercury_addr_free(addr); + + return(tret); +} + +/* + * is_usage_error_remove + * - 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 error code, 0 if storage fault. + */ +static int is_usage_error_remove(triton_ret_t tret) +{ + /* these are the scenarios that a remove could fail because of a + * usage (ie, semantic) error rather than because of a system fault. + */ + if(triton_error_equal(tret, TRITON_ERR_NO_SERVERS) || + triton_error_equal(tret, TRITON_ERR_NOENT) || + triton_error_equal(tret, TRITON_ERR_ACCES)) + { + /* if there aren't enough servers to satisfy the replication + * request, or if the object does not exist, or if + * permission was denied. + */ + return(1); + } + + return(0); +} + +/* + * Local Variables: + * c-basic-offset: 4 + * End: + * + * vim: ft=c ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/replicated-osd/rosd-write.ae b/code/src/replicated-osd/rosd-write.ae new file mode 100644 index 0000000..7baa6bd --- /dev/null +++ b/code/src/replicated-osd/rosd-write.ae @@ -0,0 +1,378 @@ +#include <aesop/aesop.h> +#include <aesop/timer.hae> +#include <triton-list.h> +#include <triton-hash.h> +#include <mercury_macros.h> +#include <mercury_proc.h> + +#include "src/replicated-osd/rosd.hae" +#include "src/replicated-osd/rosd-internal.hae" +#include "src/common/triton-debug.h" +#include "src/common/self.h" +#include "src/placement/placement.h" +#include "src/system-state/system-state.hae" +#include "src/remote/mercury-engine.hae" +#include "src/remote/mercury-encode.h" +#include "src/system-state/system-state.hae" +#include "src/transactional-osd/transactional-osd.hae" + +/* Mercury RPC structures for rosd_write */ +MERCURY_GEN_PROC(triton_rpc_rosd_write_out_t, ((triton_ret_t)(tret))) +/* input struct for rosd_write has a manual encoder because it contains + * an array + */ +typedef struct +{ + uint128_t oid; + uint64_t oid_fork; + int64_t offset; + uint64_t size; + uint32_t flags; + uint32_t replication_factor; /* cannot be set by clients */ + uint32_t expected_position; /* cannot be set by clients */ + uint64_t txn_number; /* cannot be set by clients unless a corresponding conditional flag is set */ + char* eager_payload; + uint32_t eager_payload_size; +} triton_rpc_rosd_write_in_t; +static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_write_in_t* out_p); + + +static hg_id_t rpc_rosd_write_id; +static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle); +static int triton_rpc_rosd_write_handler(hg_handle_t handle); +static __blocking triton_ret_t __remote_triton_rpc_rosd_write( + na_addr_t addr, + uint128_t oid, + uint64_t fork, + const char* buffer, + uint64_t size, + int64_t offset, + uint32_t flags, + uint32_t replicas, + uint32_t expected_position, + uint64_t txn_number); + +__blocking triton_ret_t remote_triton_rpc_rosd_write( + uint128_t oid, + uint64_t fork, + const char* buffer, + uint64_t size, + int64_t offset, + uint64_t txn_number, + uint32_t flags) +{ + triton_ret_t tret; + na_addr_t addr; + int position; + + /* we contact the first server (master) for oid */ + tret = triton_oid_to_addrs(oid, 1, &position, &addr); + if(triton_is_error(tret)) + { + return(tret); + } + + tret = __remote_triton_rpc_rosd_write(addr, oid, fork, buffer, size, offset, flags, 0, 0, txn_number); + + triton_mercury_addr_free(addr); + + return(tret); + +} + +static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) +{ + triton_rpc_rosd_write_out_t out; + triton_rpc_rosd_write_in_t in; + int ret = 0; + na_addr_t next_addr; + na_addr_t* addr_array; + int my_position; + + triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_write()\n"); + + triton_mercury_get_input(handle, &in, &out); + + if(in.replication_factor == 0) + { + /* replication factor is not known yet; this must be a request from + * a client */ + if(in.expected_position != 0) + { + out.tret = TRITON_ERR_INVAL; + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + + /* check our position assuming a replication factor of 1 to make + * sure that we are the master for this object + */ + out.tret = triton_oid_to_addrs(in.oid, 1, &my_position, &next_addr); + if(triton_is_error(out.tret)) + { + triton_mercury_start_output(handle, &out); + return(TRITON_SUCCESS); + } + + } + /* TODO: fill this in. Just returning success for now to tie into test + * program. + */ + out.tret = TRITON_SUCCESS; + + triton_mercury_start_output(handle, &out); + + return(TRITON_SUCCESS); +} +TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_write) + +void triton_rpc_rosd_write_register(void) +{ + int mflags = triton_mercury_engine_get_flags(); + + assert(mflags); + + if(mflags & TRITON_MERCURY_SVR) + { + MERCURY_HANDLER_REGISTER("triton_rpc_rosd_write", + triton_rpc_rosd_write_handler, + triton_rpc_rosd_write_in_t, + triton_rpc_rosd_write_out_t); + } + + if(mflags & TRITON_MERCURY_CLIENT) + { + rpc_rosd_write_id = MERCURY_REGISTER("triton_rpc_rosd_write", + triton_rpc_rosd_write_in_t, + triton_rpc_rosd_write_out_t); + } + + return; +} + +static __blocking triton_ret_t __remote_triton_rpc_rosd_write( + na_addr_t addr, + uint128_t oid, + uint64_t fork, + const char* buffer, + uint64_t size, + int64_t offset, + uint32_t flags, + uint32_t replicas, + uint32_t expected_position, + uint64_t txn_number) +{ + triton_rpc_rosd_write_out_t out; + triton_rpc_rosd_write_in_t in; + hg_request_t request; + int ret; + triton_ret_t tret; + int position; + + if(size > EAGER_THRESHOLD) + { + /* TODO: implement this with RDMA */ + tret = triton_error_wrap(TRITON_ERR_NOSYS, triton_self_node(), "Bulk transfers not implemented for this I/O size"); + return(tret); + } + + in.oid = oid; + in.oid_fork = fork; + in.offset = offset; + in.size = size; + in.offset = offset; + in.flags = flags; + in.replication_factor = replicas; + in.expected_position = expected_position; + in.txn_number = txn_number; + in.eager_payload = buffer; + in.eager_payload_size = size; + + ret = HG_Forward(addr, + rpc_rosd_write_id, + &in, + &out, + &request); + if(ret != HG_SUCCESS) + { + triton_error_msg("HG_Forward() failure.\n"); + return(TRITON_ERR_UNKNOWN); + } + + tret = triton_mercury_wait(request); + if(triton_is_error(tret)) + { + HG_Request_free(request); + return(tret); + } + + tret = triton_error_dup(out.tret); + + HG_Request_free(request); + + return(tret); +} + +/* TODO: refactor this somehow; a big chunk of this function is + * duplicated... + */ +static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_write_in_t* in_p) +{ + int ret = HG_FAIL; + + switch (hg_proc_get_op(proc)) { + case HG_ENCODE: + ret = hg_proc_uint128_t(proc, &in_p->oid); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint64_t(proc, &in_p->oid_fork); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_int64_t(proc, &in_p->offset); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint64_t(proc, &in_p->size); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->flags); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->replication_factor); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->expected_position); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint64_t(proc, &in_p->txn_number); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->eager_payload_size); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_raw(proc, in_p->eager_payload, in_p->eager_payload_size); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + break; + case HG_DECODE: + ret = hg_proc_uint128_t(proc, &in_p->oid); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint64_t(proc, &in_p->oid_fork); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_int64_t(proc, &in_p->offset); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint64_t(proc, &in_p->size); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->flags); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->replication_factor); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->expected_position); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint64_t(proc, &in_p->txn_number); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_uint32_t(proc, &in_p->eager_payload_size); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + in_p->eager_payload = malloc(in_p->eager_payload_size); + if(!in_p->eager_payload) + { + HG_ERROR_DEFAULT("Out of memory"); + ret = HG_FAIL; + return ret; + } + ret = hg_proc_raw(proc, in_p->eager_payload, in_p->eager_payload_size); + if (ret != HG_SUCCESS) { + HG_ERROR_DEFAULT("Proc error"); + ret = HG_FAIL; + return ret; + } + break; + case HG_FREE: + if (!in_p->eager_payload) { + HG_ERROR_DEFAULT("Already freed"); + ret = HG_FAIL; + return ret; + } + free(in_p->eager_payload); + in_p->eager_payload = NULL; + ret = HG_SUCCESS; + break; + default: + break; + } + + return ret; +} + +/* + * Local Variables: + * c-basic-offset: 4 + * End: + * + * vim: ft=c ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/replicated-osd/rosd.ae b/code/src/replicated-osd/rosd.ae index 625b731..4681068 100644 --- a/code/src/replicated-osd/rosd.ae +++ b/code/src/replicated-osd/rosd.ae @@ -1,7 +1,3 @@ - -/* TODO: split up this file, maybe have a separate rosd* file for each rosd - * function? This is going to get too big. - */ #include <aesop/aesop.h> #include <aesop/timer.hae> #include <triton-list.h> @@ -10,6 +6,7 @@ #include <mercury_proc.h> #include "src/replicated-osd/rosd.hae" +#include "src/replicated-osd/rosd-internal.hae" #include "src/common/triton-debug.h" #include "src/common/self.h" #include "src/placement/placement.h" @@ -19,335 +16,12 @@ #include "src/system-state/system-state.hae" #include "src/transactional-osd/transactional-osd.hae" -/* TODO: need a way to configure and/or reserve these values */ -/* default replication level for objects */ -static int default_replication = 3; -/* default fork used to store replication factor in object */ -#define REP_FACTOR_FORK UINT64_MAX -#define EAGER_THRESHOLD 4096 - - -/* Mercury RPC structures for rosd_create */ -MERCURY_GEN_PROC(triton_rpc_rosd_create_out_t, ((triton_ret_t)(tret))) -MERCURY_GEN_PROC(triton_rpc_rosd_create_in_t, ((uint128_t)(oid))((uint32_t)(flags))((uint32_t)(expected_position))((uint32_t)(replication_factor))((uint64_t)(niid))) - -/* - * Mercury RPC structures for rosd_remove - */ -MERCURY_GEN_PROC(triton_rpc_rosd_remove_out_t, - ((triton_ret_t)(tret))) -MERCURY_GEN_PROC(triton_rpc_rosd_remove_in_t, - ((uint128_t)(oid))\ - ((uint32_t)(flags))\ - ((uint32_t)(expected_position))\ - ((uint32_t)(replication_factor))\ - ((uint64_t)(niid))) - -/* Mercury RPC structures for rosd_range */ -MERCURY_GEN_PROC(triton_rpc_rosd_range_in_t, - ((uint128_t)(lower))\ - ((uint128_t)(upper))\ - ((uint32_t)(count))) -/* output struct for rosd_range has a manual encoder because it contains - * an array - */ -typedef struct -{ - triton_ret_t tret; - uint128_t next_position; - uint128_t* oid_array; - uint32_t oid_array_count; -} triton_rpc_rosd_range_out_t; -static int hg_proc_triton_rpc_rosd_range_out_t(hg_proc_t proc, triton_rpc_rosd_range_out_t* out_p); - -/* Mercury RPC structures for rosd_write */ -MERCURY_GEN_PROC(triton_rpc_rosd_write_out_t, ((triton_ret_t)(tret))) -/* input struct for rosd_write has a manual encoder because it contains - * an array - */ -typedef struct -{ - uint128_t oid; - uint64_t oid_fork; - int64_t offset; - uint64_t size; - uint32_t flags; - uint32_t replication_factor; /* cannot be set by clients */ - uint32_t expected_position; /* cannot be set by clients */ - uint64_t txn_number; /* cannot be set by clients unless a corresponding conditional flag is set */ - char* eager_payload; - uint32_t eager_payload_size; -} triton_rpc_rosd_write_in_t; -static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_write_in_t* out_p); - - -static hg_id_t rpc_rosd_range_id; -static int triton_rpc_rosd_range_handler(hg_handle_t handle); -static hg_id_t rpc_rosd_create_id; -static int triton_rpc_rosd_create_handler(hg_handle_t handle); -static triton_ret_t triton_oid_to_addrs(uint128_t oid, - uint32_t replication_factor, int* my_position, na_addr_t *addr_array); -static __blocking triton_ret_t rosd_create_do_work( - uint128_t oid, - uint32_t flags, - uint32_t replication_factor, - int my_position, - na_addr_t next_addr, - uint64_t niid, - int from_client_flag); -static __blocking triton_ret_t interpret_errors( - triton_ret_t local_error_code, - triton_ret_t remote_error_code, - int my_position, - int(*usage_error_check)(triton_ret_t tret), - int from_client_flag); -static __blocking triton_ret_t interpret_error( - triton_ret_t local_error_code, - int my_position, - int(*usage_error_check)(triton_ret_t tret), - int from_client_flag); -static int is_usage_error_create(triton_ret_t tret); -static __blocking void trigger_server_fault(triton_ret_t tret); -static __blocking triton_ret_t rosd_create_local_storage( - uint128_t oid, - uint32_t replication_factor, - uint64_t niid); - -static hg_id_t rpc_rosd_write_id; -static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle); -static int triton_rpc_rosd_write_handler(hg_handle_t handle); - -static int is_usage_error_remove(triton_ret_t tret); -static hg_id_t rpc_rosd_remove_id; -static __blocking triton_ret_t triton_rpc_rosd_remove(hg_handle_t handle); -static int triton_rpc_rosd_remove_handler(hg_handle_t handle); -static __blocking triton_ret_t rosd_remove_local_storage(uint128_t oid, uint64_t niid); -static __blocking triton_ret_t rosd_remove_do_work( - uint128_t oid, - uint32_t flags, - uint32_t replication_factor, - int my_position, - na_addr_t next_addr, - uint64_t niid, - int from_client_flag); -__blocking triton_ret_t __remote_triton_rpc_rosd_remove( - na_addr_t addr, - uint128_t oid, - uint32_t replicas, - uint32_t flags, - uint32_t expected_position); -__blocking triton_ret_t remote_triton_rpc_rosd_remove( - uint128_t oid, - uint32_t flags); -static __blocking triton_ret_t __remote_triton_rpc_rosd_write( - na_addr_t addr, - uint128_t oid, - uint64_t fork, - const char* buffer, - uint64_t size, - int64_t offset, - uint32_t flags, - uint32_t replicas, - uint32_t expected_position, - uint64_t txn_number); - -static __blocking triton_ret_t triton_rpc_rosd_write(hg_handle_t handle) -{ - triton_rpc_rosd_write_out_t out; - triton_rpc_rosd_write_in_t in; - int ret = 0; - na_addr_t next_addr; - na_addr_t* addr_array; - int my_position; - - triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_write()\n"); - - triton_mercury_get_input(handle, &in, &out); - - if(in.replication_factor == 0) - { - /* replication factor is not known yet; this must be a request from - * a client */ - if(in.expected_position != 0) - { - out.tret = TRITON_ERR_INVAL; - triton_mercury_start_output(handle, &out); - return(TRITON_SUCCESS); - } - - /* check our position assuming a replication factor of 1 to make - * sure that we are the master for this object - */ - out.tret = triton_oid_to_addrs(in.oid, 1, &my_position, &next_addr); - if(triton_is_error(out.tret)) - { - triton_mercury_start_output(handle, &out); - return(TRITON_SUCCESS); - } - - } - /* TODO: fill this in. Just returning success for now to tie into test - * program. - */ - out.tret = TRITON_SUCCESS; - - triton_mercury_start_output(handle, &out); - - return(TRITON_SUCCESS); -} -TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_write) - -static __blocking triton_ret_t triton_rpc_rosd_range(hg_handle_t handle) -{ - triton_rpc_rosd_range_out_t out; - triton_rpc_rosd_range_in_t in; - int ret = 0; - int out_count; - - triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_range()\n"); - - triton_mercury_get_input(handle, &in, &out); - - out.tret = TRITON_SUCCESS; - triton_uint128_setmax(out.next_position); - out.oid_array_count = 0; - out.oid_array = (uint128_t*)malloc(in.count*sizeof(*out.oid_array)); - if(!out.oid_array) - { - out.tret = TRITON_ERR_NOMEM; - } - - if(!triton_is_error(out.tret)) - { - out.tret = tosd_range(in.lower, in.upper, out.oid_array, in.count, &out_count, &out.next_position); - } - assert(out_count >= 0); - out.oid_array_count = out_count; - - /* TODO: maybe an option to filter out non-master objects? See old - * rosd.aer implementation, where this was the default behavior. - */ - - triton_mercury_start_output(handle, &out); - - if(out.oid_array) - free(out.oid_array); - - return(TRITON_SUCCESS); -} -TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_range) - -static __blocking triton_ret_t triton_rpc_rosd_create(hg_handle_t handle) -{ - triton_rpc_rosd_create_out_t out; - triton_rpc_rosd_create_in_t in; - int ret = 0; - char oid_str[TRITON_UINT128_STRLEN]; - int my_position; - na_addr_t *addr_array = NULL; - int got_addrs = 0; - int i; - na_addr_t next_addr; - - triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_create()\n"); - - triton_mercury_get_input(handle, &in, &out); - - triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, in.oid); - - triton_debug(triton_dbg_rosd, "triton_rpc_rosd_create() params: oid: %s, flags: %d, replication_factor: %d, niid: %llu\n", - oid_str, in.flags, in.replication_factor, llu(in.niid)); - - out.tret = TRITON_SUCCESS; - - /* map oid to server addresses */ - addr_array = malloc(in.replication_factor*sizeof(*addr_array)); - if(!addr_array) - { - out.tret = TRITON_ERR_NOMEM; - } - if(!triton_is_error(out.tret)) - { - out.tret = triton_oid_to_addrs(in.oid, in.replication_factor, &my_position, - addr_array); - if(!triton_is_error(out.tret)) - got_addrs = 1; - } - - if(!triton_is_error(out.tret)) - { - /* check position in chain */ - if(my_position != in.expected_position) - { - out.tret = TRITON_ERR_WRONG_SERVER; - } - } - if(my_position < (in.replication_factor-1)) - { - next_addr = addr_array[my_position+1]; - } - - out.tret = rosd_create_do_work(in.oid, in.flags, in.replication_factor, - my_position, next_addr, in.niid, 1); - - triton_mercury_start_output(handle, &out); - - for(i=0; i<in.replication_factor && got_addrs; i++) - triton_mercury_addr_free(addr_array[i]); - if(addr_array) - free(addr_array); - - return(TRITON_SUCCESS); -} -TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_create) - void triton_rpc_rosd_register(void) { - int mflags = triton_mercury_engine_get_flags(); - - assert(mflags); - - if(mflags & TRITON_MERCURY_SVR) - { - MERCURY_HANDLER_REGISTER("triton_rpc_rosd_create", - triton_rpc_rosd_create_handler, - triton_rpc_rosd_create_in_t, - triton_rpc_rosd_create_out_t); - - MERCURY_HANDLER_REGISTER("triton_rpc_rosd_remove", - triton_rpc_rosd_remove_handler, - triton_rpc_rosd_remove_in_t, - triton_rpc_rosd_remove_out_t); - - MERCURY_HANDLER_REGISTER("triton_rpc_rosd_range", - triton_rpc_rosd_range_handler, - triton_rpc_rosd_range_in_t, - triton_rpc_rosd_range_out_t); - - MERCURY_HANDLER_REGISTER("triton_rpc_rosd_write", - triton_rpc_rosd_write_handler, - triton_rpc_rosd_write_in_t, - triton_rpc_rosd_write_out_t); - } - - if(mflags & TRITON_MERCURY_CLIENT) - { - rpc_rosd_create_id = MERCURY_REGISTER("triton_rpc_rosd_create", - triton_rpc_rosd_create_in_t, triton_rpc_rosd_create_out_t); - - rpc_rosd_remove_id = MERCURY_REGISTER("triton_rpc_rosd_remove", - triton_rpc_rosd_remove_in_t, - triton_rpc_rosd_remove_out_t); - - rpc_rosd_range_id = MERCURY_REGISTER("triton_rpc_rosd_range", - triton_rpc_rosd_range_in_t, - triton_rpc_rosd_range_out_t); - - rpc_rosd_write_id = MERCURY_REGISTER("triton_rpc_rosd_write", - triton_rpc_rosd_write_in_t, - triton_rpc_rosd_write_out_t); - } - + triton_rpc_rosd_create_register(); + triton_rpc_rosd_range_register(); + triton_rpc_rosd_remove_register(); + triton_rpc_rosd_write_register(); return; } @@ -358,7 +32,7 @@ void triton_rpc_rosd_register(void) * * TODO: In the long run this functionality should reside in the traffic cop. */ -static triton_ret_t triton_oid_to_addrs(uint128_t oid, uint32_t replication_factor, int* my_position, na_addr_t *addr_array) +triton_ret_t triton_oid_to_addrs(uint128_t oid, uint32_t replication_factor, int* my_position, na_addr_t *addr_array) { triton_node_t* node_array; int node_array_size; @@ -434,263 +108,6 @@ static triton_ret_t triton_oid_to_addrs(uint128_t oid, uint32_t replication_fact return(TRITON_SUCCESS); } - -__blocking triton_ret_t __remote_triton_rpc_rosd_create( - na_addr_t addr, - uint128_t oid, - uint32_t replicas, - uint32_t flags, - uint32_t expected_position) -{ - triton_rpc_rosd_create_out_t out; - triton_rpc_rosd_create_in_t in; - hg_request_t request; - int ret; - triton_ret_t tret; - int position; - - in.oid = oid; - in.flags = flags; - in.replication_factor = replicas; - /* TODO: generate real NIIDs */ - in.niid = 0; - in.expected_position = expected_position; - - ret = HG_Forward(addr, rpc_rosd_create_id, &in, - &out, &request); - if(ret != HG_SUCCESS) - { - triton_error_msg("HG_Forward() failure.\n"); - return(TRITON_ERR_UNKNOWN); - } - - tret = triton_mercury_wait(request); - if(triton_is_error(tret)) - { - HG_Request_free(request); - return(tret); - } - - tret = triton_error_dup(out.tret); - - HG_Request_free(request); - - return(tret); -} - -__blocking triton_ret_t remote_triton_rpc_rosd_create( - uint128_t oid, - uint32_t replicas, - uint32_t flags) -{ - triton_ret_t tret; - na_addr_t addr; - int position; - - /* we contact the first server (master) for oid */ - tret = triton_oid_to_addrs(oid, 1, &position, &addr); - if(triton_is_error(tret)) - { - return(tret); - } - - tret = __remote_triton_rpc_rosd_create(addr, oid, replicas, flags, 0); - - triton_mercury_addr_free(addr); - - return(tret); -} - -__blocking triton_ret_t remote_triton_rpc_rosd_write( - uint128_t oid, - uint64_t fork, - const char* buffer, - uint64_t size, - int64_t offset, - uint64_t txn_number, - uint32_t flags) -{ - triton_ret_t tret; - na_addr_t addr; - int position; - - /* we contact the first server (master) for oid */ - tret = triton_oid_to_addrs(oid, 1, &position, &addr); - if(triton_is_error(tret)) - { - return(tret); - } - - tret = __remote_triton_rpc_rosd_write(addr, oid, fork, buffer, size, offset, flags, 0, 0, txn_number); - - triton_mercury_addr_free(addr); - - return(tret); - -} - -static __blocking triton_ret_t __remote_triton_rpc_rosd_write( - na_addr_t addr, - uint128_t oid, - uint64_t fork, - const char* buffer, - uint64_t size, - int64_t offset, - uint32_t flags, - uint32_t replicas, - uint32_t expected_position, - uint64_t txn_number) -{ - triton_rpc_rosd_write_out_t out; - triton_rpc_rosd_write_in_t in; - hg_request_t request; - int ret; - triton_ret_t tret; - int position; - - if(size > EAGER_THRESHOLD) - { - /* TODO: implement this with RDMA */ - tret = triton_error_wrap(TRITON_ERR_NOSYS, triton_self_node(), "Bulk transfers not implemented for this I/O size"); - return(tret); - } - - in.oid = oid; - in.oid_fork = fork; - in.offset = offset; - in.size = size; - in.offset = offset; - in.flags = flags; - in.replication_factor = replicas; - in.expected_position = expected_position; - in.txn_number = txn_number; - in.eager_payload = buffer; - in.eager_payload_size = size; - - ret = HG_Forward(addr, - rpc_rosd_write_id, - &in, - &out, - &request); - if(ret != HG_SUCCESS) - { - triton_error_msg("HG_Forward() failure.\n"); - return(TRITON_ERR_UNKNOWN); - } - - tret = triton_mercury_wait(request); - if(triton_is_error(tret)) - { - HG_Request_free(request); - return(tret); - } - - tret = triton_error_dup(out.tret); - - HG_Request_free(request); - - return(tret); -} - -__blocking triton_ret_t remote_triton_rpc_rosd_range( - na_addr_t addr, - uint128_t lower, - uint128_t upper, - uint32_t count, - uint128_t *oid_array, - uint32_t *oid_array_count, - uint128_t *next_position) -{ - triton_ret_t tret; - triton_rpc_rosd_range_out_t out; - triton_rpc_rosd_range_in_t in; - hg_request_t request; - int ret; - - in.lower = lower; - in.upper = upper; - in.count = count; - - ret = HG_Forward(addr, rpc_rosd_range_id, &in, - &out, &request); - if(ret != HG_SUCCESS) - { - triton_error_msg("HG_Forward() failure.\n"); - return(TRITON_ERR_UNKNOWN); - } - - tret = triton_mercury_wait(request); - if(triton_is_error(tret)) - { - HG_Request_free(request); - return(tret); - } - - *oid_array_count = out.oid_array_count; - memcpy(oid_array, out.oid_array, out.oid_array_count*sizeof(*oid_array)); - *next_position = out.next_position; - tret = triton_error_dup(out.tret); - - HG_Request_free(request); - - return(tret); -} - - -static __blocking triton_ret_t rosd_create_do_work( - uint128_t oid, - uint32_t flags, - uint32_t replication_factor, - int my_position, - na_addr_t next_addr, - uint64_t niid, - int from_client_flag) -{ - triton_ret_t local_tret; - triton_ret_t remote_tret = TRITON_SUCCESS; - triton_ret_t tret; - - /* right now only chained replication is supported */ - assert(!(flags & ROSD_FLAG_FANOUT)); - flags |= ROSD_FLAG_CHAIN; - - /* forward request while simultaneously performing local work */ - pwait - { - pprivate char oid_str[TRITON_UINT128_STRLEN]; - pbranch - { - if(my_position < replication_factor -1) - { - triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); - triton_debug(triton_dbg_rosd, - "ROSD forwarding create of oid %s to %d'th server.\n", - oid_str, my_position+1); - - remote_tret = __remote_triton_rpc_rosd_create(next_addr, oid, - replication_factor, - flags, my_position+1); - } - } - pbranch - { - triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); - triton_debug(triton_dbg_rosd, "ROSD local create of oid %s on %d'th server.\n", oid_str, my_position); - /* perform local operation */ - local_tret = rosd_create_local_storage(oid, replication_factor, niid); - } - } - - if(my_position < replication_factor -1) - tret = interpret_errors(local_tret, remote_tret, my_position, - is_usage_error_create, from_client_flag); - else - tret = interpret_error(local_tret, my_position, is_usage_error_create, - from_client_flag); - - return(tret); -} - /** * Compares the error code from local operations with forwarded operations * to determine what action the server should take (if any) and what error @@ -701,7 +118,7 @@ static __blocking triton_ret_t rosd_create_do_work( * this function is not to clean up, but to determine how to represent * various cases to clients, and when to explicitly fail servers. */ -static __blocking triton_ret_t interpret_errors( +__blocking triton_ret_t interpret_errors( triton_ret_t local_error_code, triton_ret_t remote_error_code, int my_position, @@ -809,7 +226,7 @@ static __blocking triton_ret_t interpret_errors( * no forwarded operation, and determines what the final response from the * server should be and if it needs to take any additional action. */ -static __blocking triton_ret_t interpret_error( +__blocking triton_ret_t interpret_error( triton_ret_t local_error_code, int my_position, int(*usage_error_check)(triton_ret_t tret), @@ -847,30 +264,6 @@ static __blocking triton_ret_t interpret_error( } /** - * 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 - * error code, 0 if storage fault. - */ -static int is_usage_error_create(triton_ret_t tret) -{ - /* these are the scenarios that a create could fail because of a - * usage (ie, semantic) error rather than because of a system fault. - */ - if(triton_error_equal(tret, TRITON_ERR_NO_SERVERS) || - triton_error_equal(tret, TRITON_ERR_EXIST) || - triton_error_equal(tret, TRITON_ERR_ACCES)) - { - /* if there aren't enough servers to satisfy the replication - * request, or if the object in question already exists, or if - * permission was denied. - */ - return(1); - } - - return(0); -} - -/** * Makes a best effort to initiate a failure condition for the specified * server (may be a remote server or the local server itself). This is used * to take a server out of service if we detect a non-recoverable fault. @@ -878,7 +271,7 @@ static int is_usage_error_create(triton_ret_t tret) * \return void */ /* TODO: write this function */ -static __blocking void trigger_server_fault(triton_ret_t tret) +__blocking void trigger_server_fault(triton_ret_t tret) { triton_node_t self; char* self_str; @@ -895,650 +288,6 @@ static __blocking void trigger_server_fault(triton_ret_t tret) return; } -/* rosd_create_local_storage() - * - * performs the local steps needed in a replicated create - */ -static __blocking triton_ret_t rosd_create_local_storage( - uint128_t oid, - uint32_t replication_factor, - uint64_t niid -) -{ - triton_ret_t tret; - triton_ret_t tret_tmp; - int value = 0; - int64_t obj_offset; - int64_t size; - char* buffer_offsets[1]; - triton_node_t self; - - self = triton_self_node(); - - tret = tosd_create(oid, niid); - if(triton_error_equal(tret, TRITON_ERR_NIID_DONE)) - { - /* this operation is already done */ - triton_error_destroy(tret); - tret = TRITON_SUCCESS; - } - - if(tret != TRITON_SUCCESS) - { - return triton_error_wrap(tret, self, "tosd_create failed"); - } - - if(replication_factor != default_replication) - { - - /* store replication factor directly in a data fork */ - obj_offset = 0; - buffer_offsets[0] = (char*)&replication_factor; - size = sizeof(replication_factor); - - /* NOTE: no need for tracking NIID for this portion of the operation. - * There is no harm in writing the replication factor twice. We do need - * to look out for conflicts on the transaction number, though. - */ - tret = tosd_write(oid, REP_FACTOR_FORK, 1, buffer_offsets, - &size, 1, &obj_offset, &size, 1, TOSD_FLAG_AUTO_TXN, 0, NULL); - if(tret != TRITON_SUCCESS) - { - /* We created the object but weren't able to write its metadata. - * Give a best effort here to remove the object in order to simplify - * cleanup later. It's ok if this does not succeed. - */ - tret_tmp = tosd_remove(oid, 0); - if(tret_tmp != TRITON_SUCCESS) - { - triton_error_destroy(tret_tmp); - } - /* don't propagate tret_tmp to the caller; we want the caller to see - * the original error code instead, not side-effects from cleanup. - */ - } - } - - return(tret); -} - -/* - * triton_rpc_rosd_remove - * - main handler for the rosd_remove RPC - */ -static __blocking triton_ret_t triton_rpc_rosd_remove(hg_handle_t handle) -{ - triton_rpc_rosd_remove_out_t out; - triton_rpc_rosd_remove_in_t in; - triton_ret_t trc; - int my_position; - int ret; - int i; - na_addr_t *addr_array; - na_addr_t next_addr; - char oid_str[TRITON_UINT128_STRLEN]; - - triton_ret_t tret; - uint32_t replication_factor; - char* mem_offsets[1]; - int64_t obj_offsets[1]; - int64_t mem_sizes[1]; - int64_t obj_sizes[1]; - int64_t out_size; - - triton_debug(triton_dbg_rosd, "Called triton_rpc_rosd_remove()\n"); - - memset(&out, 0, sizeof(triton_rpc_rosd_remove_out_t)); - addr_array = NULL; - trc = TRITON_SUCCESS; - - triton_mercury_get_input(handle, &in, &out); - - triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, in.oid); - triton_debug(triton_dbg_rosd, - "triton_rpc_rosd_remove() params: oid: %s, flags: %d, replication_factor: %d, niid: %llu, position:%d\n", - oid_str, - in.flags, - in.replication_factor, - llu(in.niid), - in.expected_position); - - /* - * A replication factor of zero indicates this is the initial client - * request and we need to lookup the replication factor within the object. - */ - if (in.replication_factor == 0) - { - obj_offsets[0] = 0; - mem_offsets[0] = (char*) &replication_factor; - mem_sizes[0] = sizeof(replication_factor); - obj_sizes[0] = sizeof(replication_factor); - - tret = tosd_read(in.oid, - REP_FACTOR_FORK, - mem_offsets, - mem_sizes, - 1, - obj_offsets, - obj_sizes, - 1, - &out_size, - 0); - if (triton_is_error(tret)) - { - if (triton_error_equal(tret, TRITON_ERR_NOENT)) - { - // TODO - // handle NIID logic here if needed - } - out.tret = tret; - trc = tret; - } - else - { - if (out_size == 0) - { - in.replication_factor = default_replication; - } - else - { - in.replication_factor = replication_factor; - } - triton_debug(triton_dbg_rosd, "found replication factor of: %d\n", in.replication_factor); - } - triton_error_destroy(tret); - } - - if (triton_error_equal(trc, TRITON_SUCCESS)) - { - addr_array = malloc(in.replication_factor*sizeof(*addr_array)); - if(!addr_array) - { - out.tret = TRITON_ERR_NOMEM; - } - else if (triton_error_equal(trc, TRITON_SUCCESS)) - { - memset(addr_array, 0, in.replication_factor*sizeof(*addr_array)); - out.tret = triton_oid_to_addrs(in.oid, - in.replication_factor, - &my_position, - addr_array); - if (!triton_is_error(out.tret)) - { - if (my_position == in.expected_position) - { - if (my_position < (in.replication_factor-1)) - { - next_addr = addr_array[my_position+1]; - } - else - { - next_addr = NA_ADDR_NULL; - } - - out.tret = rosd_remove_do_work(in.oid, - in.flags, - in.replication_factor, - my_position, - next_addr, - in.niid, - 1); - } - else - { - out.tret = TRITON_ERR_WRONG_SERVER; - } - } - } - } - - triton_mercury_start_output(handle, &out); - trc = TRITON_SUCCESS; - - /* - * Free resources - */ - for (i=0; ((i < in.replication_factor) && (addr_array) && (addr_array[i])); i++) - { - triton_mercury_addr_free(addr_array[i]); - } - if (addr_array) - { - free(addr_array); - } - - return trc; -} -TRITON_DEFINE_RPC_HANDLER(triton_rpc_rosd_remove) - -static __blocking triton_ret_t rosd_remove_do_work( - uint128_t oid, - uint32_t flags, - uint32_t replication_factor, - int my_position, - na_addr_t next_addr, - uint64_t niid, - int from_client_flag) -{ - triton_ret_t local_tret; - triton_ret_t remote_tret = TRITON_SUCCESS; - triton_ret_t tret; - - /* right now only chained replication is supported */ - assert(!(flags & ROSD_FLAG_FANOUT)); - flags |= ROSD_FLAG_CHAIN; - - pwait - { - pprivate char oid_str[TRITON_UINT128_STRLEN]; - pbranch - { - /* forward on to peers if necessary */ - if(my_position < replication_factor-1) - { - triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); - triton_debug(triton_dbg_rosd, "ROSD forwarding remove of oid %s to %d'th server.\n", oid_str, my_position+1); - - remote_tret = __remote_triton_rpc_rosd_remove( - next_addr, - oid, - replication_factor, - flags, - my_position+1); - } - } - pbranch - { - triton_uint128_to_string(oid_str, TRITON_UINT128_STRLEN, oid); - triton_debug(triton_dbg_rosd, "ROSD local remove of oid %s on %d'th server.\n", oid_str, my_position); - /* perform local operation */ - local_tret = rosd_remove_local_storage(oid, niid); - } - } - - if(my_position < replication_factor -1) - { - tret = interpret_errors(local_tret, remote_tret, my_position, - is_usage_error_remove, from_client_flag); - } - else - { - tret = interpret_error(local_tret, my_position, is_usage_error_remove, - from_client_flag); - } - - return(tret); -} - -/* rosd_remove_local_storage - * - removes actual data on storage - */ -__blocking triton_ret_t rosd_remove_local_storage( - uint128_t oid, - uint64_t niid) -{ - triton_ret_t tret; - int value = 0; - - tret = tosd_remove(oid, niid); - if(triton_error_equal(tret, TRITON_ERR_NIID_DONE)) - { - /* this operation is already done */ - triton_error_destroy(tret); - tret = TRITON_SUCCESS; - } - - return(tret); -} - -/* - * __remote_triton_rpc_rosd_remove - * - initiate RPC request for rosd_remove and wait for it to complete. - */ -__blocking triton_ret_t __remote_triton_rpc_rosd_remove( - na_addr_t addr, - uint128_t oid, - uint32_t replicas, - uint32_t flags, - uint32_t expected_position) -{ - triton_rpc_rosd_remove_out_t out; - triton_rpc_rosd_remove_in_t in; - hg_request_t request; - int ret; - triton_ret_t tret; - int position; - char oid_str[TRITON_UINT128_STRLEN]; - - in.oid = oid; - in.flags = flags; - in.replication_factor = replicas; - in.niid = 0; /* TODO: generate real NIIDs */ - in.expected_position = expected_position; - - triton_uint128_to_string(oid_str, sizeof(oid_str), in.oid); - triton_debug(triton_dbg_rosd, "__remote_triton_rpc_rosd_remove: post: oid=%s flags=%d rf=%d niid=%lld pos=%d\n", oid_str, in.flags, in.replication_factor, lld(in.niid), in.expected_position); - - ret = HG_Forward(addr, - rpc_rosd_remove_id, - &in, - &out, - &request); - if(ret != HG_SUCCESS) - { - triton_error_msg("HG_Forward() failure.\n"); - return(TRITON_ERR_UNKNOWN); - } - - tret = triton_mercury_wait(request); - if(triton_is_error(tret)) - { - HG_Request_free(request); - return(tret); - } - - tret = triton_error_dup(out.tret); - - HG_Request_free(request); - - return(tret); -} - -/* - * remote_triton_rpc_rosd_remove - * - wrapper for initiating RPC for rosd_remove - */ -__blocking triton_ret_t remote_triton_rpc_rosd_remove( - uint128_t oid, - uint32_t flags) -{ - triton_ret_t tret; - na_addr_t addr; - int position; - - /* we contact the first server (master) for oid */ - tret = triton_oid_to_addrs(oid, 1, &position, &addr); - if(triton_is_error(tret)) - { - return(tret); - } - - tret = __remote_triton_rpc_rosd_remove(addr, oid, 0, flags, 0); - - triton_mercury_addr_free(addr); - - return(tret); -} - -/* - * is_usage_error_remove - * - 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 error code, 0 if storage fault. - */ -static int is_usage_error_remove(triton_ret_t tret) -{ - /* these are the scenarios that a remove could fail because of a - * usage (ie, semantic) error rather than because of a system fault. - */ - if(triton_error_equal(tret, TRITON_ERR_NO_SERVERS) || - triton_error_equal(tret, TRITON_ERR_NOENT) || - triton_error_equal(tret, TRITON_ERR_ACCES)) - { - /* if there aren't enough servers to satisfy the replication - * request, or if the object does not exist, or if - * permission was denied. - */ - return(1); - } - - return(0); -} - -static int hg_proc_triton_rpc_rosd_range_out_t(hg_proc_t proc, triton_rpc_rosd_range_out_t* out_p) -{ - int ret = HG_FAIL; - int i; - - switch (hg_proc_get_op(proc)) { - case HG_ENCODE: - ret = hg_proc_triton_ret_t(proc, &out_p->tret); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint128_t(proc, &out_p->next_position); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &out_p->oid_array_count); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - for(i=0; i<out_p->oid_array_count; i++) - { - ret = hg_proc_uint128_t(proc, &out_p->oid_array[i]); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - } - break; - case HG_DECODE: - ret = hg_proc_triton_ret_t(proc, &out_p->tret); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint128_t(proc, &out_p->next_position); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &out_p->oid_array_count); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - out_p->oid_array = (uint128_t*)malloc(out_p->oid_array_count* - sizeof(*out_p->oid_array)); - if(!out_p->oid_array) { - HG_ERROR_DEFAULT("Out of memory"); - ret = HG_FAIL; - return ret; - } - for(i=0; i<out_p->oid_array_count; i++) - { - ret = hg_proc_uint128_t(proc, &out_p->oid_array[i]); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - } - break; - case HG_FREE: - if (!out_p->oid_array) { - HG_ERROR_DEFAULT("Already freed"); - ret = HG_FAIL; - return ret; - } - free(out_p->oid_array); - out_p->oid_array = NULL; - ret = HG_SUCCESS; - break; - default: - break; - } - - return ret; -} - -/* TODO: refactor this somehow; a big chunk of this function is - * duplicated... - */ -static int hg_proc_triton_rpc_rosd_write_in_t(hg_proc_t proc, triton_rpc_rosd_write_in_t* in_p) -{ - int ret = HG_FAIL; - - switch (hg_proc_get_op(proc)) { - case HG_ENCODE: - ret = hg_proc_uint128_t(proc, &in_p->oid); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint64_t(proc, &in_p->oid_fork); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_int64_t(proc, &in_p->offset); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint64_t(proc, &in_p->size); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->flags); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->replication_factor); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->expected_position); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint64_t(proc, &in_p->txn_number); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->eager_payload_size); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_raw(proc, in_p->eager_payload, in_p->eager_payload_size); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - break; - case HG_DECODE: - ret = hg_proc_uint128_t(proc, &in_p->oid); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint64_t(proc, &in_p->oid_fork); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_int64_t(proc, &in_p->offset); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint64_t(proc, &in_p->size); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->flags); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->replication_factor); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->expected_position); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint64_t(proc, &in_p->txn_number); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_uint32_t(proc, &in_p->eager_payload_size); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - in_p->eager_payload = malloc(in_p->eager_payload_size); - if(!in_p->eager_payload) - { - HG_ERROR_DEFAULT("Out of memory"); - ret = HG_FAIL; - return ret; - } - ret = hg_proc_raw(proc, in_p->eager_payload, in_p->eager_payload_size); - if (ret != HG_SUCCESS) { - HG_ERROR_DEFAULT("Proc error"); - ret = HG_FAIL; - return ret; - } - break; - case HG_FREE: - if (!in_p->eager_payload) { - HG_ERROR_DEFAULT("Already freed"); - ret = HG_FAIL; - return ret; - } - free(in_p->eager_payload); - in_p->eager_payload = NULL; - ret = HG_SUCCESS; - break; - default: - break; - } - - return ret; -} - /* * Local Variables: * c-basic-offset: 4 hooks/post-receive --

1 0