From 4e5092db3a3cf64a70467bef494948dbcae2dcf5 Mon Sep 17 00:00:00 2001 From: Philip Davis Date: Wed, 5 Mar 2025 14:00:23 -0500 Subject: [PATCH 1/2] [feat] Add dspaces_get_mod, imrpove module abstractions --- bindings/python/dspaces.py | 18 ++ bindings/python/dspaces_wrapper.c | 78 ++++++++ bindings/python/dspaces_wrapper.h | 3 + include/dspaces-common.h | 16 ++ include/dspaces-modules.h | 9 + include/dspaces.h | 23 +++ include/ss_data.h | 135 +++++++++++++- modules/azure_mod.py | 285 +++++++++++++++++++----------- modules/ds_reg.py | 17 +- modules/s3nc_mod.py | 11 +- modules/url_mod.py | 9 +- src/CMakeLists.txt | 2 +- src/dspaces-client.c | 135 ++++++++++++-- src/dspaces-modules.c | 111 +++++++++++- src/dspaces-server.c | 191 +++++++++++++++++++- 15 files changed, 907 insertions(+), 136 deletions(-) diff --git a/bindings/python/dspaces.py b/bindings/python/dspaces.py index 2aacd9f3..22ef57d9 100644 --- a/bindings/python/dspaces.py +++ b/bindings/python/dspaces.py @@ -90,6 +90,24 @@ def Get(self, name, version, lb, ub, timeout, dtype = None): passed_type = None if dtype == None else np.dtype(dtype) return wrapper_dspaces_get(self.client, (self.nspace + name).encode('ascii'), version, lb, ub, passed_type, timeout) + def GetModule(self, module, params = {}): + if not isinstance(module, str): + raise TypeError("module should be a module name") + if not isinstance(params, dict): + raise TypeError("params should be a dictionary.") + serialized_params = {key: json.dumps(val) for key,val in params.items()} + result, err = wrapper_dspaces_get_module(self.client, module.encode('ascii'), serialized_params) + if err < 0: + if err == -3: + raise DSModuleError(err) + elif err == -2 or err == -1 or err == -4 or err == -5: + raise DSRemoteFaultError(err) + elif err == -6: + raise DSConnectionError(err) + else: + raise Exception("unknown failure") + return(result) + def Exec(self, name, version, lb=None, ub=None, fn=None): arg = DSObject(name=name, version=version, lb=lb, ub=ub) return(self.VecExec([arg], fn)) diff --git a/bindings/python/dspaces_wrapper.c b/bindings/python/dspaces_wrapper.c index 250117d7..fd40f15d 100644 --- a/bindings/python/dspaces_wrapper.c +++ b/bindings/python/dspaces_wrapper.c @@ -261,6 +261,7 @@ PyObject *wrapper_dspaces_get(PyObject *clientppy, const char *name, dspaces_get_req(*clientp, &in_req, &out_req, timeout); Py_END_ALLOW_THREADS // clang-format on + data = out_req.buf; free(in_req.var_name); @@ -287,6 +288,82 @@ PyObject *wrapper_dspaces_get(PyObject *clientppy, const char *name, return (arr); } +PyObject *wrapper_dspaces_get_module(PyObject *clientppy, const char *module, + PyObject *params) +{ + dspaces_client_t *clientp = PyLong_AsVoidPtr(clientppy); + // char ***param_str; + struct dspaces_mod_param *dsp_params; + int num_params; + PyObject *key, *val, *key_bstr, *val_bstr, *keys, *ret, *arr; + struct dspaces_req out; + PyArray_Descr *descr; + void *data; + int ndim; + npy_intp *dims; + int i, err; + + keys = PyDict_Keys(params); + num_params = PyList_Size(keys); + + if(num_params > 0) { + dsp_params = malloc(sizeof(*dsp_params) * num_params); + } + for(i = 0; i < num_params; i++) { + dsp_params[i].type = DSP_JSON; + key = PyList_GetItem(keys, i); + key_bstr = PyUnicode_AsASCIIString(key); + dsp_params[i].key = strdup(PyBytes_AsString(key_bstr)); + Py_DECREF(key_bstr); + val = PyDict_GetItem(params, key); + val_bstr = PyUnicode_AsASCIIString(val); + dsp_params[i].val.s = strdup(PyBytes_AsString(val_bstr)); + Py_DECREF(val_bstr); + } + Py_DECREF(keys); + + // clang-format off + Py_BEGIN_ALLOW_THREADS + err = dspaces_get_module(*clientp, module, dsp_params, num_params, &out); + Py_END_ALLOW_THREADS + // clang-format on + + data = out.buf; + for(i = 0; i < num_params; i++) { + free(dsp_params[i].key); + free(dsp_params[i].val.s); + } + if(num_params > 0) { + free(dsp_params); + } + + ret = PyTuple_New(2); + + PyTuple_SET_ITEM(ret, 1, PyLong_FromLong(err)); + + if(!data) { + Py_INCREF(Py_None); + PyTuple_SET_ITEM(ret, 0, Py_None); + return (ret); + } + descr = PyArray_DescrNewFromType(out.tag); + ndim = out.ndim; + dims = malloc(sizeof(*dims) * ndim); + for(i = 0; i < ndim; i++) { + dims[(ndim - i) - 1] = (out.ub[i] - out.lb[i]) + 1; + } + arr = PyArray_NewFromDescr(&PyArray_Type, descr, ndim, dims, NULL, data, 0, + NULL); + if(!arr) { + PyErr_SetString(PyExc_RuntimeError, "failed to create numpy array"); + return (NULL); + } + PyTuple_SET_ITEM(ret, 0, arr); + free(dims); + + return (ret); +} + PyObject *wrapper_dspaces_pexec(PyObject *clientppy, PyObject *req_list, PyObject *fn, const char *fn_name) { @@ -714,6 +791,7 @@ PyObject *wrapper_dspaces_register(PyObject *clientppy, const char *type, PyTuple_SET_ITEM(ret, 0, PyUnicode_DecodeASCII(nspace, strlen(nspace), NULL)); } else { + Py_INCREF(Py_None); PyTuple_SET_ITEM(ret, 0, Py_None); } diff --git a/bindings/python/dspaces_wrapper.h b/bindings/python/dspaces_wrapper.h index 3b240898..4e2d1ebc 100644 --- a/bindings/python/dspaces_wrapper.h +++ b/bindings/python/dspaces_wrapper.h @@ -26,6 +26,9 @@ PyObject *wrapper_dspaces_get(PyObject *clientppy, const char *name, int version, PyObject *lbt, PyObject *ubt, PyObject *dtype, int timeout); +PyObject *wrapper_dspaces_get_module(PyObject *clientppy, const char *module, + PyObject *params); + PyObject *wrapper_dspaces_pexec(PyObject *clientppy, PyObject *req_list, PyObject *fn, const char *fn_name); diff --git a/include/dspaces-common.h b/include/dspaces-common.h index f1018219..aa629c82 100644 --- a/include/dspaces-common.h +++ b/include/dspaces-common.h @@ -11,6 +11,10 @@ #include #include +#include + +#include "dspaces-logging.h" + #if defined(__cplusplus) extern "C" { #endif @@ -61,6 +65,8 @@ static inline void ignore_result(int unused_result) { (void)unused_result; } #define DSP_INT16 -15 #define DSP_INT32 -16 #define DSP_INT64 -17 +#define DSP_STR -18 +#define DSP_JSON -19 static size_t type_to_size_map[] = {0, sizeof(float), @@ -90,6 +96,16 @@ static int type_to_size(int type_id) return (type_to_size_map[-type_id]); } +#define HG_TRY(op, ret, jmp, estr, ...) \ + do { \ + hg_return_t hret = op; \ + if(hret != HG_SUCCESS) { \ + DEBUG_OUT(estr, ##__VA_ARGS__); \ + err = ret; \ + goto jmp; \ + } \ + } while(0); + #if defined(__cplusplus) } #endif diff --git a/include/dspaces-modules.h b/include/dspaces-modules.h index fc77f2ca..c405c6e8 100644 --- a/include/dspaces-modules.h +++ b/include/dspaces-modules.h @@ -25,7 +25,9 @@ struct dspaces_module { typedef enum dspaces_module_arg_type { DSPACES_ARG_REAL, DSPACES_ARG_INT, + DSPACES_ARG_BOOL, DSPACES_ARG_STR, + DSPACES_ARG_JSON, DSPACES_ARG_NONE } dspaces_mod_arg_type_t; @@ -68,6 +70,10 @@ struct dspaces_module_ret { int dspaces_init_mods(struct list_head *mods); +int build_module_args_from_dict(size_t dict_len, int8_t *types, char **keys, + void **vals, + struct dspaces_module_args **argsp); + int build_module_args_from_odsc(obj_descriptor *odsc, struct dspaces_module_args **argsp); @@ -91,4 +97,7 @@ struct dspaces_module_ret *dspaces_module_exec(struct dspaces_module *mod, int dspaces_module_names(struct list_head *mods, char ***names); +int odsc_from_ret(struct dspaces_module_ret *ret, obj_descriptor **odsc_out, + char name[OD_MAX_NAME_LEN], int version); + #endif // __DSPACES_MODULES_H__ diff --git a/include/dspaces.h b/include/dspaces.h index ac63997c..b8635aec 100644 --- a/include/dspaces.h +++ b/include/dspaces.h @@ -314,6 +314,29 @@ struct dspaces_req { int dspaces_get_req(dspaces_client_t client, struct dspaces_req *in_req, struct dspaces_req *out_req, int timeout); +struct dspaces_mod_param { + int type; + char *key; + union { + float f; + double d; + int8_t b : 1; + int8_t i8; + int16_t i16; + int32_t i32; + int64_t i64; + uint8_t u8; + uint16_t u16; + uint32_t u32; + uint64_t u64; + char *s; + } val; +}; + +int dspaces_get_module(dspaces_client_t client, const char *module, + struct dspaces_mod_param *params, int num_param, + struct dspaces_req *out); + int dspaces_pexec(dspaces_client_t client, const char *var_name, unsigned int ver, int ndim, uint64_t *lb, uint64_t *ub, const char *fn, unsigned int fnsz, const char *fn_name, diff --git a/include/ss_data.h b/include/ss_data.h index ececc6a7..c8366481 100644 --- a/include/ss_data.h +++ b/include/ss_data.h @@ -20,6 +20,8 @@ #include #include +#include "dspaces-common.h" + #include #define MAX_VERSIONS 10 @@ -35,15 +37,18 @@ typedef struct { enum storage_type { row_major, column_major }; +#define OD_MAX_NAME_LEN 150 +#define OD_MAX_OWNER_LEN 128 + typedef struct { - char name[150]; + char name[OD_MAX_NAME_LEN]; enum storage_type st; uint32_t flags; int tag; int type; - char owner[128]; + char owner[OD_MAX_OWNER_LEN]; unsigned int version; /* Global bounding box descriptor. */ @@ -376,6 +381,128 @@ static inline hg_return_t hg_proc_name_list_t(hg_proc_t proc, void *data) return (ret); } +typedef struct dsp_dict { + hg_size_t len; + hg_int8_t *types; + hg_string_t *keys; + void **vals; +} dsp_dict_t; + +static inline hg_return_t dict_encode_val(hg_proc_t proc, hg_int8_t type, + void *val) +{ + switch(type) { + case DSP_BOOL: + case DSP_CHAR: + case DSP_BYTE: + case DSP_UINT8: + case DSP_INT8: + return (hg_proc_raw(proc, val, 1)); + case DSP_UINT16: + case DSP_INT16: + return (hg_proc_raw(proc, val, 2)); + case DSP_FLOAT: + case DSP_INT: + case DSP_UINT: + case DSP_UINT32: + case DSP_INT32: + return (hg_proc_raw(proc, val, 4)); + case DSP_LONG: + case DSP_DOUBLE: + case DSP_ULONG: + case DSP_UINT64: + case DSP_INT64: + return (hg_proc_raw(proc, val, 8)); + case DSP_STR: + case DSP_JSON: + return (hg_proc_hg_string_t(proc, &val)); + default: + return (HG_INVALID_PARAM); + } +} + +static inline hg_return_t dict_decode_val(hg_proc_t proc, hg_int8_t type, + void **val) +{ + hg_return_t ret = HG_SUCCESS; + + switch(type) { + case DSP_BOOL: + case DSP_CHAR: + case DSP_BYTE: + case DSP_UINT8: + case DSP_INT8: + *val = malloc(1); + return (hg_proc_raw(proc, *val, 1)); + case DSP_UINT16: + case DSP_INT16: + *val = malloc(2); + return (hg_proc_raw(proc, *val, 2)); + case DSP_FLOAT: + case DSP_INT: + case DSP_UINT: + case DSP_UINT32: + case DSP_INT32: + *val = malloc(4); + return (hg_proc_raw(proc, *val, 4)); + case DSP_LONG: + case DSP_DOUBLE: + case DSP_ULONG: + case DSP_UINT64: + case DSP_INT64: + *val = malloc(8); + return (hg_proc_raw(proc, *val, 8)); + case DSP_STR: + case DSP_JSON: + ret = hg_proc_hg_string_t(proc, val); + return (ret); + default: + return (HG_INVALID_PARAM); + } +} + +static inline hg_return_t hg_proc_dsp_dict_t(hg_proc_t proc, void *arg) +{ + hg_return_t ret = HG_SUCCESS; + dsp_dict_t *in = (dsp_dict_t *)arg; + int i; + + ret = hg_proc_hg_size_t(proc, &in->len); + if(ret != HG_SUCCESS) + return ret; + + if(hg_proc_get_op(proc) == HG_DECODE) { + in->types = malloc(sizeof(*in->types) * in->len); + in->keys = malloc(sizeof(*in->keys) * in->len); + in->vals = malloc(sizeof(*in->vals) * in->len); + } + for(i = 0; i < in->len; i++) { + ret = hg_proc_hg_int8_t(proc, &in->types[i]); + if(ret != HG_SUCCESS) { + return (ret); + } + ret = hg_proc_hg_string_t(proc, &in->keys[i]); + if(ret != HG_SUCCESS) { + return (ret); + } + if(hg_proc_get_op(proc) == HG_ENCODE) { + ret = dict_encode_val(proc, in->types[i], in->vals[i]); + } else if(hg_proc_get_op(proc) == HG_DECODE) { + ret = dict_decode_val(proc, in->types[i], &in->vals[i]); + } else { + free(in->vals[i]); + } + if(ret != HG_SUCCESS) + return (ret); + } + if(hg_proc_get_op(proc) == HG_FREE) { + free(in->keys); + free(in->vals); + } + + return (ret); +} + MERCURY_GEN_PROC(bulk_gdim_t, ((odsc_hdr_with_gdim)(odsc))((hg_bulk_t)(handle))) MERCURY_GEN_PROC(bulk_in_t, ((odsc_hdr)(odsc))((hg_bulk_t)(handle))((uint8_t)(flags))) @@ -400,6 +527,8 @@ MERCURY_GEN_PROC(get_var_objs_in_t, ((hg_string_t)(var_name))((int32_t)(src))) MERCURY_GEN_PROC(reg_in_t, ((hg_string_t)(type))((hg_string_t)(name))( (hg_string_t)(reg_data))((int32_t)(src))((uint64_t)(id))) +MERCURY_GEN_PROC(get_mod_in_t, ((hg_string_t)(name))((dsp_dict_t)(params))) +MERCURY_GEN_PROC(get_mod_out_t, ((odsc_list_t)(odscs))((int32_t)(ret))) char *obj_desc_sprint(obj_descriptor *); // @@ -488,4 +617,4 @@ struct lock_data *create_lock(struct list_head *list, char *name); char **addr_str_buf_to_list(char *buf, int num_addrs); -#endif /* __SS_DATA_H_ */ +#endif /* __SS_DATA_H_ */ \ No newline at end of file diff --git a/modules/azure_mod.py b/modules/azure_mod.py index ae60c726..bfc317d9 100644 --- a/modules/azure_mod.py +++ b/modules/azure_mod.py @@ -1,71 +1,74 @@ -import planetary_computer -from netCDF4 import Dataset -import fsspec -import pystac_client -import numpy as np -import os -from urllib.parse import urlparse -from urllib.request import urlretrieve -from bitstring import Bits, pack -from datetime import date, timedelta +import planetary_computer +from netCDF4 import Dataset +import pystac_client +import numpy as np +import os +from urllib.parse import urlparse +from urllib.request import urlretrieve +from bitstring import Bits, pack +from datetime import date, timedelta +from dateutil import parser import sys base_date = date(1950, 1, 1) present_date = date(2015, 1, 1) last_date = date(2100, 12, 31) -cache_base='.azrcache' +cache_base = '.azrcache' -try: - catalog = pystac_client.Client.open( - "https://planetarycomputer.microsoft.com/api/stac/v1", - modifier=planetary_computer.sign_inplace, - ) +try: + catalog = pystac_client.Client.open("https://planetarycomputer.microsoft.com/api/stac/v1", + modifier = planetary_computer.sign_inplace, ) collection = catalog.get_collection("nasa-nex-gddp-cmip6") - variable_list=collection.summaries.get_list("cmip6:variable") - model_list=collection.summaries.get_list("cmip6:model")[:10] - scenario_list=collection.summaries.get_list("cmip6:scenario") - have_pc = True + variable_list = collection.summaries.get_list("cmip6:variable") + model_list = collection.summaries.get_list("cmip6:model")[ : 10] + scenario_list = collection.summaries.get_list("cmip6:scenario") + have_pc = True except pystac_client.exceptions.APIError: - print("don't have planetary computer api access") + print("don't have planetary computer api access") have_pc = False + def _get_gddp_time_ranges(version): - bits = Bits(uint=version, length=32) + bits = Bits(uint = version, length = 32) start_day, span_days = bits.unpack('uint:16, uint:16') start_date = base_date + timedelta(days = start_day) - end_date = start_date + timedelta(days = span_days) + end_date = start_date + timedelta(days = span_days) if start_date > last_date: - raise ValueError(f'WARNING: start_date of {start_date} is too far in the future') + raise ValueError(f'WARNING: start_date of {start_date} is too far in the future') if end_date > last_date: - print(f"WARNING: truncating end_date of {end_date} to {last_date}") - end_date = last_date + print(f'WARNING: truncating end_date of {end_date} to {last_date}') + end_date = last_date return start_date, end_date +def _validate_gddp_params(gparams): + model = gparams.get('model', 'NO MODEL') + scenario = gparams.get('scenario', 'NO SCENARIO') + variable = gparams.get('variable', 'NO VARIABLE') + if model not in model_list: + raise ValueError(f'model {model} not available.') + if scenario not in scenario_list: + raise ValueError(f'scenario {scenario} not available.') + if variable not in variable_list: + raise ValueError(f'variable {variable} not available.') + +def _get_default_params(): + return {'model' : 'CESM2', 'scenario' : 'ssp585', 'variable' : None } + def _get_gddp_params(name): - model = 'CESM2' - scenario = 'ssp585' - variable = None - var_name = name.split('\\')[-1] - quality = 0 + gddp_params = _get_default_params() + var_name = name.split('\\')[- 1] name_parts = var_name.split(',') - for part in name_parts: + for part in name_parts: if part[0] == 'm': - model = part[2:] - if have_pc and model not in model_list: - raise ValueError(f"model {model} not available.") + gddp_params['model'] = part[2 : ] if part[0] == 's': - scenario = part[2:] - if have_pc and scenario not in scenario_list: - raise ValueError(f"scenario {scenario} not available.") + gddp_params['scenario'] = part[2 : ] if part[0] == 'v': - variable = part[2:] - if have_pc and variable not in variable_list: - raise ValueError(f"variable {variable} not available.") - if part[0] == 'q': - quality = int(part[2:]) - if variable == None: - raise ValueError('No variable name specified') - return model, scenario, variable, quality + gddp_params['variable'] = part[2 : ] + if 'variable' not in gddp_params: + raise ValueError('No variable name specified') + _validate_gddp_params(gddp_params) + return gddp_params def _get_dataset(url): path = urlparse(url).path @@ -73,84 +76,150 @@ def _get_dataset(url): if not os.path.exists(cache_entry): cache_dir = os.path.dirname(cache_entry) if not os.path.exists(cache_dir): - os.makedirs(os.path.dirname(cache_entry)) - urlretrieve(url, filename=cache_entry) - return(Dataset(cache_entry)) + os.makedirs(os.path.dirname(cache_entry)) + urlretrieve(url, filename = cache_entry) + return Dataset(cache_entry) def _get_azure_url(url, var_name): - ds = _get_dataset(url) - return(ds[var_name]) + ds = _get_dataset(url) + return ds[var_name] -def _get_cmip6_data(model, scenario, variable, start_date, end_date, lb, ub): - result = None - result_days = (end_date - start_date).days + 1 - if have_pc: - search = catalog.search( - collections=["nasa-nex-gddp-cmip6"], - datetime=f'{start_date}/{end_date}', - query = { - "cmip6:model": { - "eq": model - }, - "cmip6:scenario": { - "in": ['historical', scenario] - }, - }, - sortby=[{'field':'cmip6:year','direction':'asc'}] - ) - items = search.item_collection() - +def _get_urls(start_date, end_date, model, scenario, variable): + url_list = [] + search = catalog.search(collections = ["nasa-nex-gddp-cmip6"], + datetime = f'{start_date}/{end_date}', + query = {"cmip6:model" : {"eq" : model }, + "cmip6:scenario" : {"in" : ['historical', scenario] }, }, + sortby = [{'field' : 'cmip6:year', 'direction' : 'asc' }]) + items = search.item_collection() for item in items: - if have_pc: - year = item.properties['cmip6:year'] - url = item.assets[variable].href - ds = _get_dataset(url) - else: - # TODO - in case indexing is offline, we still want to hit the cache - pass + year = item.properties['cmip6:year'] + url = item.assets[variable].href + url_list.append((year, url)) + return url_list + +def _get_cmip6_data(start_date, end_date, lb, ub, model, scenario, variable): + result = None + result_days = (end_date - start_date).days + 1 + + url_list = _get_urls(start_date, end_date, model, scenario, variable) + + for year, url in url_list: + ds = _get_dataset(url) data = ds[variable] - if result is None: - if lb[0] >= data[0].shape[0] or lb[1] >= data[0].shape[1]: - return None - ub = (min(ub[0]+1, data[0].shape[0]), min(ub[1]+1, data[0].shape[1])) + if result is None: + if lb[0] >= data[0].shape[0] or lb[1] >= data[0].shape[1]: + return None + ub = (min(ub[0] + 1, data[0].shape[0]), min(ub[1] + 1, data[0].shape[1])) shape = (result_days, ub[0] - lb[0], ub[1] - lb[1]) result = np.ndarray(shape, dtype = data.dtype) - item_start = max(start_date, date(year, 1,1)) - item_end = min(date(year,12,31), end_date) - start_gidx = (item_start - start_date).days - end_gidx = (item_end - start_date).days + 1 - start_iidx = (item_start - date(year, 1 , 1)).days - end_iidx = (item_end - date(year, 1, 1)).days + 1 - result[start_gidx:end_gidx,:,:] = data[start_iidx:end_iidx,lb[0]:ub[0],lb[1]:ub[1]] - return(result) - -def validate(url, **kwargs): + item_start = max(start_date, date(year, 1, 1)) + item_end = min(date(year, 12, 31), end_date) + start_gidx =(item_start - start_date).days + end_gidx =(item_end - start_date).days + 1 + start_iidx =(item_start - date(year, 1, 1)).days + end_iidx =(item_end - date(year, 1, 1)).days + 1 + result[start_gidx:end_gidx, :, : ] = data[start_iidx:end_iidx, lb[0] :ub[0], lb[1] :ub[1]] + return result + +def validate(start_date, end_date, variable, **kwargs): pass -def reg_query(name, version, lb, ub, params, url, var_name): - array = _get_azure_url(url, var_name) +def _get_reg_time_bounds(params, start_date, end_date): + query_time = params.get('query_time') + if query_time: + query_start = parser.parse(query_time) + query_end = parser.parse(query_time) + else: + query_start = parser.parse(params.get('query_start')) + query_end = parser.parse(params.get('query_end')) + start_date = parser.parse(start_date) + end_date = parser.parse(end_date) + query_start = max(query_start, start_date) + query_end = min(query_end, end_date) + result_start = (query_start - start_date).days + result_end =(query_end - start_date).days + + return result_start, result_end + +def _get_reg_params(variable, model, scenario): + gddp_params = _get_default_params() + gddp_params['variable'] = variable + if model: + gddp_params['model'] = model + if scenario: + gddp_params['scenario'] = scenario + return gddp_params + +def _get_reg_data(gddp_params, start_date, end_date): + url_list = _get_urls(start_date, end_date, + gddp_params['model'], + gddp_params['scenario'], + gddp_params['variable']) + if len(url_list) != 1: + raise ValueError('registry queries should involve exactly one file.') + _, url = url_list[0] + ds = _get_dataset(url) + return ds[gddp_params['variable']] + +def reg_query(name, params, variable, start_date, end_date, model = None, scenario = None): + result_start, result_end = _get_reg_time_bounds(params, start_date, end_date) + + gddp_params = _get_reg_params(variable, model, scenario) + array = _get_reg_data(gddp_params, start_date, end_date) + + lb = params.get('lb') + ub = params.get('ub') if lb: - index = [ slice(lb[x], ub[x]+1) for x in range(len(lb)) ] + result = np.ndarray(((result_end - result_start) + 1, + (ub[0] - lb[0]) + 1, (ub[1] - lb[1] + 1)), + dtype = array.dtype) + index = [slice(result_start, result_end + 1), + slice(lb[0], ub[0] + 1), + slice(lb[1], ub[1] + 1)] else: - index = [ slice(0, x) for x in array.shape] - return(array[index]) + result = np.ndarray(((result_end - result_start) + 1, + array.shape[0], array.shape[1]), + dtype = array.dtype) + index = [slice(result_start, result_end + 1), + slice(0, array.shape[0]), + slice(0, array.shape[1])] + result[ :, :, : ] = array[index] + return result +def pquery(variable, start_date, end_date, lb, ub, model = 'CESM2', scenario = 'ssp585', **kwargs): + result = _get_cmip6_data(model, scenario, variable, start_date, end_date, lb, ub) + return result def query(name, version, lb, ub): start_date, end_date = _get_gddp_time_ranges(version) - model, scenario, variable, quality = _get_gddp_params(name) - result = _get_cmip6_data(model, scenario, variable, start_date, end_date, lb, ub) - return(result) + gddp_params = _get_gddp_params(name) + result = _get_cmip6_data(start_date, end_date, lb, ub, **gddp_params) + return result +""" if __name__ == '__main__': - s = date(2013, 5, 2) - e = date(2013, 5, 2) - start = (s - base_date).days - span = (e - s).days - lb = (0,0) - ub = (599,1399) - version = pack('uint:16, uint:16', start, span).uint - res = query(name='cmip6-planetary\\m:ACCESS-ESM1-5,v:tas', version=1, lb=lb, ub=ub) + params = {'lb' : (100, 200), + 'ub' : (150, 230), + 'query_start' : '2013-05-02', + 'query_end' : '2013-06-03'} + #params = {'query_time' : '2013-05-02' } + start_date = '2013-01-01' + end_date = '2013-12-31' + model = 'ACCESS-ESM1-5' + scenario = 'historical' + variable = 'tas' + res = reg_query('foo', params, variable, start_date, end_date, model, scenario) print(res.shape) +""" - +if __name__ == '__main__': + s = date(2013, 5, 2) + e = date(2013, 5, 2) + start =(s - base_date).days + span = (e - s).days + lb =(100, 200) + ub =(150, 230) + version = pack('uint:16, uint:16', start, span).uint + res = query(name = 'cmip6-planetary\\m:ACCESS-ESM1-5,v:tas', version = 1, lb = lb, ub = ub) + print(res) diff --git a/modules/ds_reg.py b/modules/ds_reg.py index 1db5c3a3..ca2a4989 100644 --- a/modules/ds_reg.py +++ b/modules/ds_reg.py @@ -36,7 +36,7 @@ def Add(self, reg_id, reg_type, name, data): cursor.close() return(reg_id) - def Query(self, reg_id, version, lb, ub, params): + def Query(self, reg_id, params): db_conn = sqlite3.connect(self.db_name) cursor = db_conn.cursor() res = cursor.execute("SELECT type.name, registry.name, registry.data FROM registry INNER JOIN type ON registry.reg_type == type.id WHERE registry.id == ?", (reg_id,)) @@ -44,7 +44,7 @@ def Query(self, reg_id, version, lb, ub, params): db_conn.close() data = json.loads(ser_data) module = self.modules[reg_type] - return(module.reg_query(name, version, lb, ub, params, **data)) + return(module.reg_query(name, params, **data)) def HighestIDBetween(self, min_id, max_id): db_conn = sqlite3.connect(self.db_name) @@ -81,10 +81,19 @@ def bootstrap_id(rank): def query(name, version, lb, ub): id_part = name.split('\\')[-1] id, params = _get_query_params(name) - return(reg.Query(id, version, lb, ub, params)) + params['version'] = version + params['lb'] = lb + params['ub'] = ub + result = reg.Query(id, params) + return(result) + +def pquery(id, **kwargs): + return(reg.Query(id, kwargs)) if __name__ == '__main__': - register('s3nc_mod', 'abc', json.dumps({'bucket':'noaa-goes17','path':'ABI-L1b-RadM/2020/215/15/OR_ABI-L1b-RadM1-M6C02_G17_s20202151508255_e20202151508312_c20202151508338.nc'}), 45) + id = register('s3nc_mod', 'abc', json.dumps({'bucket':'noaa-goes17','path':'ABI-L1b-RadM/2020/215/15/OR_ABI-L1b-RadM1-M6C02_G17_s20202151508255_e20202151508312_c20202151508338.nc'}), 45) res = query('abcdef\id:45,var_name:Rad', 2, (1,1), (4,2)) + params = {'id':id, 'var_name': 'Rad', 'version':2, 'lb':(1,1), 'ub':(4,2)} + #res = pquery(**params) print(res) diff --git a/modules/s3nc_mod.py b/modules/s3nc_mod.py index 203f8c95..325abc00 100644 --- a/modules/s3nc_mod.py +++ b/modules/s3nc_mod.py @@ -88,10 +88,12 @@ def validate(bucket, path): if not path.endswith('.nc'): raise ValueError("S3 module can only access netCDF files.") -def reg_query(name, version, lb, ub, params, bucket, path): +def reg_query(name, params, bucket, path): if 'var_name' not in params: raise ValueError var_name = params['var_name'] + lb = params.get('lb') + ub = params.get('ub') cache_entry = f'{cache_base}/{bucket}/{path}' if not os.path.exists(cache_entry): s3.get(f'{bucket}/{path}', cache_entry) @@ -121,6 +123,9 @@ def query(name, version, lb, ub): return(array[index]) if __name__ == '__main__': - var_name = 'goes17\\RadM/M1/C2/Rad' + var_name = 'Rad' version = 505081608 - print(query(var_name, version, (1,1), (4,2))) + bucket = 'noaa-goes17' + path = 'ABI-L1b-RadM/2020/215/15/OR_ABI-L1b-RadM1-M6C04_G17_s20202151508255_e20202151508312_c20202151508354.nc' + params = {'version':version, 'var_name': var_name, 'lb':(1,1), 'ub':(4,2)} + print(reg_query('foo', params, bucket, path)) diff --git a/modules/url_mod.py b/modules/url_mod.py index 0fee9595..afcda762 100644 --- a/modules/url_mod.py +++ b/modules/url_mod.py @@ -29,8 +29,12 @@ def _get_url_array(url, name, var_name): data = Dataset(cache_entry) return(data[var_name]) -def reg_query(name, version, lb, ub, params, url): +def reg_query(name, params, url): + if 'var_name' not in params: + raise ValueError var_name = params['var_name'] + lb = params.get('lb') + ub = params.get('ub') array = _get_url_array(url, name, var_name) if lb: index = [ slice(lb[x], ub[x]+1) for x in range(len(lb)) ] @@ -50,5 +54,6 @@ def query(name, version, lb, ub): if __name__ == '__main__': url = 'https://www.unidata.ucar.edu/software/netcdf/examples/sresa1b_ncar_ccsm3-example.nc' - a = reg_query('foo', 0, (10, 10), (20,30), {'var_name':'area'}, url) + params = {'lb':(10,10), 'ub':(20,20), 'var_name':'area'} + a = reg_query('foo', params, url) print(a) \ No newline at end of file diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index c8ef0a1d..b13e3df0 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -5,7 +5,7 @@ if(HAVE_DRC) endif() # list of source files -set(dspaces-src util.c bbox.c ss_data.c dspaces-client.c dspaces-ops.c) +set(dspaces-src util.c bbox.c ss_data.c dspaces-client.c dspaces-ops.c dspaces-logging.c) # load package helper for generating cmake CONFIG packages include (CMakePackageConfigHelpers) diff --git a/src/dspaces-client.c b/src/dspaces-client.c index 0473253e..db2dcc78 100644 --- a/src/dspaces-client.c +++ b/src/dspaces-client.c @@ -4,6 +4,7 @@ * * See COPYRIGHT in top-level directory. */ +#include "dspaces-logging.h" #include "dspaces-ops.h" #include "dspaces.h" #include "dspacesp.h" @@ -40,18 +41,6 @@ #define APEX_TIMER_STOP(num) (void)0; #endif -#define DEBUG_OUT(dstr, ...) \ - do { \ - if(client->f_debug) { \ - ABT_unit_id tid; \ - ABT_thread_self_id(&tid); \ - fprintf(stderr, \ - "Rank %i: TID: %" PRIu64 " %s, line %i (%s): " dstr, \ - client->rank, tid, __FILE__, __LINE__, __func__, \ - ##__VA_ARGS__); \ - } \ - } while(0); - #define SUB_HASH_SIZE 16 #define MB (1024 * 1024) #define BULK_TRANSFER_MAX (128 * MB) @@ -109,6 +98,7 @@ struct dspaces_client { hg_id_t get_var_objs_id; hg_id_t reg_id; hg_id_t get_mods_id; + hg_id_t get_mod_id; struct dc_gspace *dcg; char **server_address; char **node_names; @@ -462,6 +452,7 @@ static int dspaces_init_internal(int rank, dspaces_client_t *c) } client->rank = rank; + dspaces_init_logging(client->rank); // now do dcg_alloc and store gid client->dcg = dcg_alloc(client); @@ -595,6 +586,8 @@ static int dspaces_init_margo(dspaces_client_t client, margo_registered_name(client->mid, "reg_rpc", &client->reg_id, &flag); margo_registered_name(client->mid, "get_mods_rpc", &client->get_mods_id, &flag); + margo_registered_name(client->mid, "get_mod_rpc", &client->get_mod_id, + &flag); } else { client->put_id = MARGO_REGISTER(client->mid, "put_rpc", bulk_gdim_t, bulk_out_t, NULL); @@ -660,6 +653,8 @@ static int dspaces_init_margo(dspaces_client_t client, MARGO_REGISTER(client->mid, "reg_rpc", reg_in_t, uint64_t, NULL); client->get_mods_id = MARGO_REGISTER(client->mid, "get_mods_rpc", void, name_list_t, NULL); + client->get_mod_id = MARGO_REGISTER(client->mid, "get_mod_rpc", + get_mod_in_t, get_mod_out_t, NULL); } return (dspaces_SUCCESS); @@ -1342,6 +1337,8 @@ static int get_data(dspaces_client_t client, int num_odscs, in[i].odsc.size = sizeof(obj_descriptor); in[i].odsc.raw_odsc = (char *)(&odsc_tab[i]); + DEBUG_OUT("retrieving %s\n", obj_desc_sprint(&odsc_tab[i])); + rdma_size[i] = (req_obj.size) * bbox_volume(&odsc_tab[i].bb); DEBUG_OUT("For odsc %i, element size is %zi, and there are %" PRIu64 @@ -1881,6 +1878,120 @@ int dspaces_get(dspaces_client_t client, const char *var_name, unsigned int ver, return (ret); } +int compare_params(const void *v1, const void *v2) +{ + struct dspaces_mod_param *p1 = (struct dspaces_mod_param *)v1; + struct dspaces_mod_param *p2 = (struct dspaces_mod_param *)v2; + + return (strcmp(p1->key, p2->key)); +} + +int dspaces_get_module(dspaces_client_t client, const char *module, + struct dspaces_mod_param *params, int num_param, + struct dspaces_req *req_out) +{ + get_mod_in_t in; + get_mod_out_t out; + hg_addr_t server_addr; + hg_handle_t handle; + int num_odscs; + obj_descriptor *odsc_tab; + obj_descriptor odsc = {0}; + int num_elem, elem_size; + void *data; + int i, err; + + memset(req_out, 0, sizeof(*req_out)); + + qsort(params, num_param, sizeof(*params), compare_params); + + in.name = strdup(module); + in.params.len = num_param; + in.params.types = malloc(sizeof(*in.params.types) * num_param); + in.params.keys = malloc(sizeof(*in.params.keys) * num_param); + in.params.vals = malloc(sizeof(*in.params.vals) * num_param); + for(i = 0; i < num_param; i++) { + in.params.types[i] = params[i].type; + in.params.keys[i] = params[i].key; + if(in.params.types[i] == DSP_STR || in.params.types[i] == DSP_JSON) { + in.params.vals[i] = params[i].val.s; + } else { + in.params.vals[i] = ¶ms[i].val; + } + } + + get_server_address(client, &server_addr); + HG_TRY(margo_create(client->mid, server_addr, client->get_mod_id, &handle), + DS_MOD_ECLIENT, err_out, "margo_create() failed.\n"); + HG_TRY(margo_forward(handle, &in), DS_MOD_ECLIENT, err_destroy, + "margo_forward() failed.\n"); + HG_TRY(margo_get_output(handle, &out), DS_MOD_ECLIENT, err_destroy, + "margo_get_output() failed\n"); + + num_odscs = (out.odscs.odsc_list.size) / sizeof(obj_descriptor); + if(num_odscs == 0) { + err = DS_MOD_EFAULT; + goto err_destroy; + } + odsc_tab = malloc(out.odscs.odsc_list.size); + memcpy(odsc_tab, out.odscs.odsc_list.raw_odsc, out.odscs.odsc_list.size); + margo_free_output(handle, &out); + margo_addr_free(client->mid, server_addr); + margo_destroy(handle); + + free(in.name); + free(in.params.types); + free(in.params.keys); + free(in.params.vals); + + DEBUG_OUT("Finished module query - need to fetch %d objects\n", num_odscs); + memcpy(&odsc, odsc_tab, sizeof(odsc)); + + // send request to get the obj_desc + if(num_odscs != 0) { + elem_size = odsc_tab[0].size; + } else { + DEBUG_OUT("not setting element size because there are no result " + "descriptors.\n"); + data = NULL; + return (DS_MOD_EFAULT); + } + + num_elem = 1; + odsc.size = elem_size; + DEBUG_OUT("element size is %zi\n", odsc.size); + for(i = 0; i < odsc.bb.num_dims; i++) { + num_elem *= (odsc.bb.ub.c[i] - odsc.bb.lb.c[i]) + 1; + } + DEBUG_OUT("data buffer size is %d\n", num_elem * elem_size); + + odsc.tag = odsc_tab[0].tag; + for(i = 1; i < num_odscs; i++) { + if(odsc_tab[i].tag != odsc_tab[0].tag) { + fprintf(stderr, "WARNING: multiple distinct tag values returned in " + "query result. Returning first one.\n"); + break; + } + } + + data = malloc(num_elem * elem_size); + get_data(client, num_odscs, odsc, odsc_tab, data); + if(req_out) { + fill_req(client, &odsc, data, req_out); + } + + return (0); +err_destroy: + margo_addr_free(client->mid, server_addr); + margo_destroy(handle); +err_out: + free(in.name); + free(in.params.types); + free(in.params.keys); + free(in.params.vals); + return (err); +} + int dspaces_get_meta(dspaces_client_t client, const char *name, int mode, int current, int *version, void **data, unsigned int *len) { diff --git a/src/dspaces-modules.c b/src/dspaces-modules.c index a9e6e3ca..e2695172 100644 --- a/src/dspaces-modules.c +++ b/src/dspaces-modules.c @@ -140,6 +140,62 @@ int build_module_args_from_reg(reg_in_t *reg, return (nargs); } +int build_module_args_from_dict(size_t dict_len, int8_t *types, char **keys, + void **vals, struct dspaces_module_args **argsp) +{ + struct dspaces_module_args *args; + int i; + + args = malloc(sizeof(*args) * dict_len); + + for(i = 0; i < dict_len; i++) { + args[i].name = strdup(keys[i]); + switch(types[i]) { + case DSP_FLOAT: + args[i].type = DSPACES_ARG_REAL; + args[i].len = -1; + args[i].rval = *(float *)(vals[i]); + break; + case DSP_INT: + args[i].type = DSPACES_ARG_INT; + args[i].len = -1; + args[i].ival = *(int *)(vals[i]); + break; + case DSP_LONG: + args[i].type = DSPACES_ARG_INT; + args[i].len = -1; + args[i].ival = *(long *)(vals[i]); + break; + case DSP_DOUBLE: + args[i].type = DSPACES_ARG_REAL; + args[i].len = -1; + args[i].ival = *(double *)(vals[i]); + break; + case DSP_BOOL: + args[i].type = DSPACES_ARG_BOOL; + args[i].len = -1; + args[i].ival = *(uint8_t *)(vals[i]); + break; + case DSP_STR: + args[i].type = DSPACES_ARG_STR; + args[i].len = 1 + strlen(vals[i]); + args[i].strval = vals[i]; + break; + case DSP_JSON: + args[i].type = DSPACES_ARG_JSON; + args[i].len = 1 + strlen(vals[i]); + args[i].strval = vals[i]; + break; + default: + args[i].type = DSPACES_ARG_NONE; + args[i].len = -1; + break; + } + } + *argsp = args; + return (dict_len); +} + int build_module_args_from_odsc(obj_descriptor *odsc, struct dspaces_module_args **argsp) { @@ -290,7 +346,8 @@ static struct dspaces_module_ret *ds_module_ret_err(int err) #ifdef DSPACES_HAVE_PYTHON PyObject *py_obj_from_arg(struct dspaces_module_args *arg) { - PyObject *pArg; + PyObject *pArg, *jArg, *jObj; + static PyObject *json_mod = NULL; int i; switch(arg->type) { @@ -312,8 +369,32 @@ PyObject *py_obj_from_arg(struct dspaces_module_args *arg) PyTuple_SetItem(pArg, i, PyLong_FromLong(arg->iarray[i])); } return (pArg); + case DSPACES_ARG_BOOL: + if(arg->len == -1) { + return (PyBool_FromLong(arg->ival)); + } + pArg = PyTuple_New(arg->len); + for(i = 0; i < arg->len; i++) { + PyTuple_SetItem(pArg, i, PyBool_FromLong(arg->iarray[i])); + } + return (pArg); case DSPACES_ARG_STR: return (PyUnicode_DecodeFSDefault(arg->strval)); + case DSPACES_ARG_JSON: + if(json_mod == NULL && + (json_mod = PyImport_ImportModule("json")) == NULL) { + fprintf(stderr, "ERROR: could not load json python module.\n"); + PyErr_Print(); + return (Py_None); + } + jArg = PyBytes_FromStringAndSize(arg->strval, arg->len - 1); + jObj = PyObject_CallMethodObjArgs( + json_mod, PyUnicode_FromString("loads"), jArg, NULL); + if(!jObj) { + PyErr_Print(); + } + Py_XDECREF(jArg); + return (jObj); case DSPACES_ARG_NONE: return (Py_None); default: @@ -487,3 +568,31 @@ int dspaces_module_names(struct list_head *mods, char ***names) return (num_mods); } + +int odsc_from_ret(struct dspaces_module_ret *ret, obj_descriptor **odsc_out, + char name[OD_MAX_NAME_LEN], int version) +{ + obj_descriptor *odsc; + + if(odsc_out == NULL) { + fprintf(stderr, "ERROR: %s: bad oddsc_out pointer.\n", __func__); + return (-1); + } + if(ret->type != DSPACES_MOD_RET_ARRAY) { + fprintf(stderr, + "WARNING: %s: failed to make obj_data from non-array.\n", + __func__); + return (-1); + } + odsc = calloc(1, sizeof(*odsc)); + odsc->size = ret->elem_size; + odsc->bb.num_dims = ret->ndim; + obj_data_resize(odsc, ret->dim); + odsc->tag = ret->tag; + memcpy(odsc->name, name, OD_MAX_NAME_LEN); + odsc->version = version; + + *odsc_out = odsc; + + return (0); +} diff --git a/src/dspaces-server.c b/src/dspaces-server.c index 610826a3..db7f646e 100644 --- a/src/dspaces-server.c +++ b/src/dspaces-server.c @@ -89,6 +89,7 @@ struct dspaces_provider { hg_id_t get_var_objs_id; hg_id_t reg_id; hg_id_t get_mods_id; + hg_id_t get_mod_id; struct list_head mods; struct ds_gspace *dsg; char **server_address; @@ -156,6 +157,7 @@ DECLARE_MARGO_RPC_HANDLER(get_vars_rpc); DECLARE_MARGO_RPC_HANDLER(get_var_objs_rpc); DECLARE_MARGO_RPC_HANDLER(reg_rpc); DECLARE_MARGO_RPC_HANDLER(get_mods_rpc); +DECLARE_MARGO_RPC_HANDLER(get_mod_rpc); static int init_sspace(dspaces_provider_t server, struct bbox *default_domain, struct ds_gspace *dsg_l) @@ -977,6 +979,10 @@ int dspaces_server_init(const char *listen_addr_str, MPI_Comm comm, &flag); DS_HG_REGISTER(hg, server->get_mods_id, void, name_list_t, get_mods_rpc); + margo_registered_name(server->mid, "get_mod_rpc", &server->get_mod_id, + &flag); + DS_HG_REGISTER(hg, server->get_mod_id, get_mod_in_t, get_mod_out_t, + get_mod_rpc); } else { server->put_id = MARGO_REGISTER(server->mid, "put_rpc", bulk_gdim_t, bulk_out_t, put_rpc); @@ -1079,6 +1085,11 @@ int dspaces_server_init(const char *listen_addr_str, MPI_Comm comm, name_list_t, get_mods_rpc); margo_register_data(server->mid, server->get_mods_id, (void *)server, NULL); + server->get_mod_id = + MARGO_REGISTER(server->mid, "get_mod_rpc", get_mod_in_t, + get_mod_out_t, get_mod_rpc); + margo_register_data(server->mid, server->get_mod_id, (void *)server, + NULL); } int err = dsg_alloc(server, conf_file, comm); if(err) { @@ -1962,6 +1973,183 @@ static void query_rpc(hg_handle_t handle) } DEFINE_MARGO_RPC_HANDLER(query_rpc) +static int param_hash(int type, void *val, int mod) +{ + char *sval; + int hval = 1; + int i; + + switch(type) { + case DSP_BOOL: + case DSP_CHAR: + case DSP_BYTE: + case DSP_UINT8: + case DSP_INT8: + return (*(uint8_t *)val); + case DSP_UINT16: + case DSP_INT16: + return (*(uint16_t *)val); + case DSP_FLOAT: + case DSP_INT: + case DSP_UINT: + case DSP_UINT32: + case DSP_INT32: + return (*(uint32_t *)val); + case DSP_LONG: + case DSP_DOUBLE: + case DSP_ULONG: + case DSP_UINT64: + case DSP_INT64: + return ((*(uint64_t *)val) % mod); + case DSP_STR: + case DSP_JSON: + sval = val; + for(i = 0; sval[i]; i++) { + hval = (hval * sval[i]) % mod; + } + return (hval); + } +} + +static unsigned int params_hash(dsp_dict_t *params) +{ + long hval = 1; + static const int mod = 1000000007; + int i; + + for(i = 0; i < params->len; i++) { + hval = + (hval * param_hash(params->types[i], params->vals[i], mod)) % mod; + } + + return (hval); +} + +static void write_param_name(dsp_dict_t *params, char name[OD_MAX_NAME_LEN]) +{ + char *pstr_start = name; + static const int mod = 1000000007; + int hval; + int i; + + for(i = 0; i < params->len; i++) { + hval = param_hash(params->types[i], params->vals[i], mod); + sprintf(pstr_start, "%x.4", hval); + pstr_start += 8; + if(pstr_start - name >= 136) { + pstr_start = name; + } + } +} + +static void get_version_name(dsp_dict_t *params, unsigned int *version, + char name[OD_MAX_NAME_LEN]) +{ + int ver_set = 0; + int name_set = 0; + int i; + + for(i = 0; i < params->len; i++) { + if(params->types[i] == DSP_STR && + strcmp(params->keys[i], "name") == 0) { + memcpy(name, params->vals[i], 149); + name_set = 1; + } + if(params->types[i] == DSP_UINT && + strcmp(params->keys[i], "version") == 0) { + *version = *(unsigned int *)(params->vals[i]); + ver_set = 1; + } + } + if(!ver_set) { + *version = params_hash(params); + } + if(!name_set) { + write_param_name(params, name); + } +} + +static void get_mod_rpc(hg_handle_t handle) +{ + margo_instance_id mid = margo_hg_handle_get_instance(handle); + const struct hg_info *info = margo_get_info(handle); + dspaces_provider_t server = + (dspaces_provider_t)margo_registered_data(mid, info->id); + get_mod_in_t in; + get_mod_out_t out = {0}; + struct dspaces_module *mod; + struct dspaces_module_args *args; + int nargs; + struct dspaces_module_ret *res = NULL; + hg_return_t hret; + char name[OD_MAX_NAME_LEN] = {0}; + unsigned int version; + obj_descriptor *odsc; + struct obj_data *od; + int8_t *types; + int i, err; + + HG_TRY(margo_get_input(handle, &in), 0, err_destroy, + "margo_get_input() failed.\n"); + + mod = dspaces_mod_by_name(&server->mods, in.name); + if(!mod) { + fprintf(stderr, "ERROR: %s: module '%s' requested but not available.\n", + __func__, in.name); + out.ret = DS_MOD_ENOMOD; + goto err_respond; + } + + // To keep modules from being depedent on hg types. + // vals are void * ; we can probably depend on hg_string_t + // being a typedef of char *, and there's no hg_string_t + // conversion interface anyway... + types = malloc(sizeof(*types) * in.params.len); + for(i = 0; i < in.params.len; i++) { + types[i] = in.params.types[i]; + } + + nargs = build_module_args_from_dict(in.params.len, types, in.params.keys, + in.params.vals, &args); + free(types); + res = + dspaces_module_exec(mod, "pquery", args, nargs, DSPACES_MOD_RET_ARRAY); + if(res && res->type == DSPACES_MOD_RET_ERR) { + DEBUG_OUT("Module failure. Returning EFAULT to user.\n") + out.ret = DS_MOD_EFAULT; + free(res); + goto err_respond; + } else if(res) { + get_version_name(&in.params, &version, name); + } + + odsc_from_ret(res, &odsc, name, version); + odsc_take_ownership(server, odsc); + od = obj_data_alloc_no_data(odsc, res->data); + ABT_mutex_lock(server->ls_mutex); + ls_add_obj(server->dsg->ls, od); + ABT_mutex_unlock(server->ls_mutex); + obj_update_dht(server, od, DS_OBJ_NEW); + + out.odscs.odsc_list.raw_odsc = (char *)odsc; + out.odscs.odsc_list.size = sizeof(*odsc); + + margo_respond(handle, &out); + margo_free_input(handle, &in); + margo_destroy(handle); + + free(odsc); + + return; + +err_respond: + margo_respond(handle, &out); + margo_free_input(handle, &in); +err_destroy: + margo_destroy(handle); +} +DEFINE_MARGO_RPC_HANDLER(get_mod_rpc) + static int peek_meta_remotes(dspaces_provider_t server, peek_meta_in_t *in) { margo_request *reqs; @@ -2806,8 +2994,7 @@ static void pexec_rpc(hg_handle_t handle) gstate = PyGILState_Ensure(); array = build_ndarray_from_od(arg_obj); - if((pklmod == NULL) && - (pklmod = PyImport_ImportModuleNoBlock("dill")) == NULL) { + if((pklmod == NULL) && (pklmod = PyImport_ImportModule("dill")) == NULL) { margo_respond(handle, &out); margo_free_input(handle, &in); margo_destroy(handle); From 46f11641b8e6f03369d3b987f304c274798bde45 Mon Sep 17 00:00:00 2001 From: Philip Davis Date: Tue, 22 Apr 2025 01:34:29 -0400 Subject: [PATCH 2/2] Add add_module RPC and client bindings --- bindings/python/dspaces.py | 3 + bindings/python/dspaces_wrapper.c | 26 ++++- bindings/python/dspaces_wrapper.h | 2 + docs/running.rst | 2 +- include/dspaces-conf.h | 2 + include/dspaces-modules.h | 4 + include/dspaces.h | 2 + include/ss_data.h | 1 + src/dspaces-client.c | 43 +++++++ src/dspaces-conf.c | 184 +++++++++++++++++++++--------- src/dspaces-modules.c | 102 ++++++++++++----- src/dspaces-server.c | 40 +++++++ 12 files changed, 327 insertions(+), 84 deletions(-) diff --git a/bindings/python/dspaces.py b/bindings/python/dspaces.py index 22ef57d9..22bcf55f 100644 --- a/bindings/python/dspaces.py +++ b/bindings/python/dspaces.py @@ -180,6 +180,9 @@ def GetVarObjs(self, var_name): def GetModules(self): return wrapper_dspaces_get_modules(self.client) + def AddModule(self, name, namespace, url): + wrapper_dspaces_add_module(self.client, name.encode('ascii'), namespace.encode('ascii'), url.encode('ascii')) + def _get_expr(obj, client): if isinstance(obj, DSExpr): return(obj) diff --git a/bindings/python/dspaces_wrapper.c b/bindings/python/dspaces_wrapper.c index fd40f15d..e9c78df2 100644 --- a/bindings/python/dspaces_wrapper.c +++ b/bindings/python/dspaces_wrapper.c @@ -222,7 +222,7 @@ void wrapper_dspaces_put(PyObject *clientppy, PyObject *obj, const char *name, Py_END_ALLOW_THREADS // clang-format on - return; + return; } PyObject *wrapper_dspaces_get(PyObject *clientppy, const char *name, @@ -430,9 +430,9 @@ PyObject *wrapper_dspaces_pexec(PyObject *clientppy, PyObject *req_list, dspaces_mpexec(*clientp, num_reqs, reqs, PyBytes_AsString(fn), PyBytes_Size(fn) + 1, fn_name, &data, &data_size); Py_END_ALLOW_THREADS - // clang-format on + // clang-format on - if(data_size > 0) + if(data_size > 0) { result = PyBytes_FromStringAndSize(data, data_size); } @@ -799,3 +799,23 @@ PyObject *wrapper_dspaces_register(PyObject *clientppy, const char *type, return (ret); } + +PyObject *wrapper_dspaces_add_module(PyObject *clientppy, const char *name, const char *namespace, const char *url) +{ + dspaces_client_t *clientp = PyLong_AsVoidPtr(clientppy); + int8_t result; + + // clang-format off + Py_BEGIN_ALLOW_THREADS + result = dspaces_add_module(*clientp, name, namespace, url); + Py_END_ALLOW_THREADS + // clang-format on + + if(result != 0) { + PyErr_SetString(PyExc_RuntimeError, "dspaces_add_module() failed"); + return(NULL); + } + + Py_INCREF(Py_None); + return (Py_None); +} diff --git a/bindings/python/dspaces_wrapper.h b/bindings/python/dspaces_wrapper.h index 4e2d1ebc..eb4efe07 100644 --- a/bindings/python/dspaces_wrapper.h +++ b/bindings/python/dspaces_wrapper.h @@ -65,3 +65,5 @@ PyObject *wrapper_dspaces_ops_calc(PyObject *clientppy, PyObject *exprppy); PyObject *wrapper_dspaces_register(PyObject *clientppy, const char *type, const char *name, const char *data); + + PyObject *wrapper_dspaces_add_module(PyObject *clientppy, const char *name, const char *namespace, const char *url); \ No newline at end of file diff --git a/docs/running.rst b/docs/running.rst index c80fbbd8..b25b814c 100644 --- a/docs/running.rst +++ b/docs/running.rst @@ -41,7 +41,7 @@ Bootstrapping communication --------------------------- The server produces a bootstrap file during its init phase, ``conf.ds``. This file must be read by the clients (or rank zero of the clients if ``dspaces_init_mpi()`` is being used. This file provides the clients with enough information to make initial contact with the server and -perform wire-up. In order to find this file, the server and client application must be run in the same working directory, or at last a symlink of ``ds.conf`` should be present. +perform wire-up. In order to find this file, the server and client application must be run in the same working directory, or at last a symlink of ``conf.ds`` should be present. Environment variables --------------------- diff --git a/include/dspaces-conf.h b/include/dspaces-conf.h index 7dcbf94f..9b8c9f5f 100644 --- a/include/dspaces-conf.h +++ b/include/dspaces-conf.h @@ -26,4 +26,6 @@ int parse_conf_toml(const char *fname, struct ds_conf *conf); void print_conf(struct ds_conf *conf); +char *dspaces_download_module(const char *name, const char *url, char *file); + #endif // __DSPACES_CONF_H \ No newline at end of file diff --git a/include/dspaces-modules.h b/include/dspaces-modules.h index c405c6e8..f868f052 100644 --- a/include/dspaces-modules.h +++ b/include/dspaces-modules.h @@ -70,6 +70,10 @@ struct dspaces_module_ret { int dspaces_init_mods(struct list_head *mods); +int dspaces_server_add_module(struct list_head *mods, const char *name, + const char *namespace, const char *url, + enum dspaces_mod_type type); + int build_module_args_from_dict(size_t dict_len, int8_t *types, char **keys, void **vals, struct dspaces_module_args **argsp); diff --git a/include/dspaces.h b/include/dspaces.h index b8635aec..9ae62f8e 100644 --- a/include/dspaces.h +++ b/include/dspaces.h @@ -490,6 +490,8 @@ int dspaces_get_modules(dspaces_client_t client, char ***mod_names); long dspaces_register_simple(dspaces_client_t client, const char *type, const char *name, const char *data, char **nspace); + int dspaces_add_module(dspaces_client_t client, const char *name, const char *namespace, const char *url); + #if defined(__cplusplus) } #endif diff --git a/include/ss_data.h b/include/ss_data.h index c8366481..dd01ba69 100644 --- a/include/ss_data.h +++ b/include/ss_data.h @@ -529,6 +529,7 @@ MERCURY_GEN_PROC(reg_in_t, (hg_string_t)(reg_data))((int32_t)(src))((uint64_t)(id))) MERCURY_GEN_PROC(get_mod_in_t, ((hg_string_t)(name))((dsp_dict_t)(params))) MERCURY_GEN_PROC(get_mod_out_t, ((odsc_list_t)(odscs))((int32_t)(ret))) +MERCURY_GEN_PROC(add_mod_in_t, ((hg_string_t)(name))((int8_t)(type))((hg_string_t)(url))((hg_string_t)(namespace))) char *obj_desc_sprint(obj_descriptor *); // diff --git a/src/dspaces-client.c b/src/dspaces-client.c index db2dcc78..c482e488 100644 --- a/src/dspaces-client.c +++ b/src/dspaces-client.c @@ -99,6 +99,7 @@ struct dspaces_client { hg_id_t reg_id; hg_id_t get_mods_id; hg_id_t get_mod_id; + hg_id_t add_mod_id; struct dc_gspace *dcg; char **server_address; char **node_names; @@ -588,6 +589,7 @@ static int dspaces_init_margo(dspaces_client_t client, &flag); margo_registered_name(client->mid, "get_mod_rpc", &client->get_mod_id, &flag); + margo_registered_name(client->mid, "add_mod_rpc", &client->add_mod_id, &flag); } else { client->put_id = MARGO_REGISTER(client->mid, "put_rpc", bulk_gdim_t, bulk_out_t, NULL); @@ -655,6 +657,8 @@ static int dspaces_init_margo(dspaces_client_t client, name_list_t, NULL); client->get_mod_id = MARGO_REGISTER(client->mid, "get_mod_rpc", get_mod_in_t, get_mod_out_t, NULL); + client->add_mod_id = MARGO_REGISTER(client->mid, "add_mod_rpc", add_mod_in_t, int8_t, NULL); + } return (dspaces_SUCCESS); @@ -3205,3 +3209,42 @@ int dspaces_get_modules(dspaces_client_t client, char ***mod_names) return (ret); } + +int dspaces_add_module(dspaces_client_t client, const char *name, const char *namespace, const char *url) +{ + hg_addr_t server_addr; + hg_handle_t h; + add_mod_in_t in; + int8_t result; + hg_return_t hret; + int err; + + in.name = strdup(name); + in.namespace = strdup(namespace); + in.url = strdup(url); + // rpc handler ignores in.type - assumes python module + in.type = 0; + + get_server_address(client, &server_addr); + HG_TRY(margo_create(client->mid, server_addr, client->add_mod_id, &h), + DS_MOD_ECLIENT, err_out, "margo_create() failed.\n"); + HG_TRY(margo_forward(h, &in), DS_MOD_ECLIENT, err_destroy, + "margo_forward() failed.\n"); + HG_TRY(margo_get_output(h, &result), DS_MOD_ECLIENT, err_destroy, + "margo_get_output() failed\n"); + if(result == 0) { + err = 0; + } else { + DEBUG_OUT("WARNING: (%s): server responded with %d\n", __func__, result); + err = DS_MOD_EFAULT; + } + +err_destroy: + margo_addr_free(client->mid, server_addr); + margo_destroy(h); +err_out: + free(in.name); + free(in.namespace); + free(in.url); + return (err); +} \ No newline at end of file diff --git a/src/dspaces-conf.c b/src/dspaces-conf.c index 0fc68ac5..679193fc 100644 --- a/src/dspaces-conf.c +++ b/src/dspaces-conf.c @@ -237,75 +237,151 @@ static void parse_storage_table(toml_table_t *storage, struct ds_conf *conf) } } -static void parse_modules_table(toml_table_t *modules, struct ds_conf *conf) +char *dspaces_download_module(const char *name, const char *url, char *file) { - struct dspaces_module *mod; - toml_table_t *module; - char *server, *file, *url, *type, *ext; + char *arg_part; char *fname; + FILE *mod_file; + +#ifdef DSPACES_HAVE_CURL + static CURL *curl = NULL; + if(!curl) { + curl = curl_easy_init(); + } + if(!curl) { + fprintf(stderr, "ERROR: could not initialize curl.\n"); + return(NULL); + } + if(!file) { + file = strdup(strrchr(url, '/')) + 1; + arg_part = strchr(file, '?'); + if(arg_part) { + arg_part[0] = '\0'; + } + } + fname = malloc(strlen(xstr(DSPACES_MOD_DIR)) + strlen(file) + 2); + sprintf(fname, "%s/%s", xstr(DSPACES_MOD_DIR), file); + DEBUG_OUT("downloading module '%s' to '%s'\n", name, fname); + mod_file = fopen(fname, "wb"); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, mod_file); + curl_easy_setopt(curl, CURLOPT_URL, url); + if(curl_easy_perform(curl) != CURLE_OK) { + fprintf(stderr, + "ERROR: could not download module '%s' from '%s'.\n", + name, url); + fclose(mod_file); + free(fname); + return(NULL); + } + fclose(mod_file); + DEBUG_OUT("downloaded completed successfully.\n"); + free(fname); +#else + fprintf(stderr, + "WARNING: could not download module '%s': compiled without " + "curl support.\n", + name); + return(NULL); +#endif // DSPACES_HAVE_CURL + return(file); +} + +static struct dspaces_module *parse_module(char *name, toml_table_t *module) +{ + struct dspaces_module *mod; + char *file, *url, *ext; char *arg_part; + char *fname; FILE *mod_file; + + mod = calloc(1, sizeof(*mod)); + if(!mod) { + fprintf(stderr, "ERROR: could not allocate memory for module '%s'.\n", name); + return NULL; + } + mod->name = strdup(name); + get_toml_str(module, "namespace", &mod->namespace); + if(!mod->namespace) { + fprintf(stderr, + "WARNING: No namespace for '%s'. Query matching currently " + "requires a namespace.\n", + mod->name); + } + url = NULL; + file = NULL; + get_toml_str(module, "url", &url); + get_toml_str(module, "file", &file); + if(url) { + if(file) { + fprintf(stderr, + "WARNING: both 'url' and 'file' specified for module '%s'. " + "Using 'url'.\n", + mod->name); + } + file = dspaces_download_module(name, url, file); + if(!file) { + fprintf(stderr, + "ERROR: could not download module '%s' from '%s'.\n", + mod->name, url); + free(url); + free(mod); + return NULL; + } + free(url); + } + if(file) { + ext = strrchr(file, '.'); + if(strcmp(ext, ".py") == 0) { + mod->type = DSPACES_MOD_PY; + ext[0] = '\0'; + mod->file = file; + } else { + fprintf(stderr, + "WARNING: could not determine type of module '%s', " + "with extension '%s'. Skipping.\n", + mod->name, ext); + free(file); + return NULL; + } + } + return mod; +} + +static void parse_modules_table(toml_table_t *modules, struct ds_conf *conf) +{ + struct dspaces_module *mod; + toml_table_t *module; + char *name; int nmod; #ifdef DSPACES_HAVE_CURL CURL *curl = curl_easy_init(); + if(!curl) { + fprintf(stderr, "ERROR: could not initialize curl.\n"); + } #endif int i; nmod = toml_table_ntab(modules); + DEBUG_OUT("found %d modules\n", nmod); for(i = 0; i < nmod; i++) { - mod = calloc(1, sizeof(*mod)); - mod->name = strdup(toml_key_in(modules, i)); - module = toml_table_in(modules, mod->name); - get_toml_str(module, "namespace", &mod->namespace); - if(!mod->namespace) { - fprintf(stderr, - "WARNING: No namespace for '%s'. Query matching currently " - "requires a namespace.\n", - mod->name); + name = strdup(toml_key_in(modules, i)); + if(!name) { + fprintf(stderr, "ERROR: could not get module name.\n"); + continue; } - url = NULL; - file = NULL; - get_toml_str(module, "url", &url); - get_toml_str(module, "file", &file); - if(url) { -#ifdef DSPACES_HAVE_CURL - if(!file) { - file = strdup(strrchr(url, '/')) + 1; - arg_part = strchr(file, '?'); - if(arg_part) { - arg_part[0] = '\0'; - } - } - fname = malloc(strlen(xstr(DSPACES_MOD_DIR)) + strlen(file) + 2); - sprintf(fname, "%s/%s", xstr(DSPACES_MOD_DIR), file); - mod_file = fopen(fname, "wb"); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, mod_file); - curl_easy_setopt(curl, CURLOPT_URL, url); - curl_easy_perform(curl); - fclose(mod_file); - free(fname); -#else - fprintf(stderr, - "WARNING: could not download module '%s': compiled without " - "curl support.\n", - mod->name); -#endif // DSPACES_HAVE_CURL + DEBUG_OUT("configuring module '%s'\n", name); + module = toml_table_in(modules, name); + if(!module) { + fprintf(stderr, "ERROR: could not get module '%s'.\n", name); + continue; } - if(file) { - ext = strrchr(file, '.'); - if(strcmp(ext, ".py") == 0) { - mod->type = DSPACES_MOD_PY; - ext[0] = '\0'; - mod->file = file; - } else { - fprintf(stderr, - "WARNING: could not determine type of module '%s', " - "with extension '%s'. Skipping.\n", - mod->name, ext); - free(file); - continue; - } + mod = parse_module(name, module); + if(!mod) { + fprintf(stderr, "ERROR: could not parse module '%s'.\n", mod->name); + continue; } + free(name); + DEBUG_OUT("adding module '%s' to module list\n", mod->name); list_add(&mod->entry, conf->mods); } #ifdef DSPACES_HAVE_CURL diff --git a/src/dspaces-modules.c b/src/dspaces-modules.c index e2695172..4e17f8bf 100644 --- a/src/dspaces-modules.c +++ b/src/dspaces-modules.c @@ -1,5 +1,6 @@ #include "dspaces-modules.h" #include "dspaces-common.h" +#include "dspaces-conf.h" #include "dspaces-logging.h" #include "ss_data.h" @@ -21,6 +22,7 @@ static int dspaces_init_py_mod(struct dspaces_module *mod) return (-1); } + DEBUG_OUT("loading module '%s' as %s\n", mod->name, mod->file); pName = PyUnicode_DecodeFSDefault(mod->file); mod->pModule = PyImport_Import(pName); if(!mod->pModule) { @@ -63,15 +65,9 @@ static void add_builtin_mods(struct list_head *mods) } } -int dspaces_init_mods(struct list_head *mods) +static int dspaces_init_mod(struct dspaces_module *mod) { - struct dspaces_module *mod; - - add_builtin_mods(mods); - - list_for_each_entry(mod, mods, struct dspaces_module, entry) - { - switch(mod->type) { + switch(mod->type) { case DSPACES_MOD_PY: #ifdef DSPACES_HAVE_PYTHON dspaces_init_py_mod(mod); @@ -81,18 +77,88 @@ int dspaces_init_mods(struct list_head *mods) "module. DataSpaces was compiled without Python support.\n", mod->name); #endif // DSPACES_HAVE_PYTHON - break; + return(0); default: fprintf(stderr, "WARNING: unknown type %i for module '%s'. Corruption?\n", mod->type, mod->name); - } + return(-1); // TODO: unlink module on init failure? } +} + +int dspaces_init_mods(struct list_head *mods) +{ + struct dspaces_module *mod; + + add_builtin_mods(mods); + + list_for_each_entry(mod, mods, struct dspaces_module, entry) + { + dspaces_init_mod(mod); + } return (0); } +static struct dspaces_module *find_mod(struct list_head *mods, + const char *mod_name) +{ + struct dspaces_module *mod; + int i; + + list_for_each_entry(mod, mods, struct dspaces_module, entry) + { + if(strcmp(mod_name, mod->name) == 0) { + return (mod); + } + } + + return (NULL); +} + +int dspaces_server_add_module(struct list_head *mods, const char *name, + const char *namespace, const char *url, + enum dspaces_mod_type type) +{ + struct dspaces_module *mod; + char *ext; + int ret; + + if(find_mod(mods, name)) { + fprintf(stderr, "WARNING: module '%s' already exists.\n", name); + return (-1); + } + + mod = malloc(sizeof(*mod)); + if(!mod) { + return (-1); + } + + mod->name = strdup(name); + mod->namespace = strdup(namespace); + mod->file = dspaces_download_module(name, url, mod->file); + ext = strrchr(mod->file, '.'); + if(strcmp(ext, ".py") == 0) { + mod->type = DSPACES_MOD_PY; + ext[0] = '\0'; + } else { + mod->type = type; + } + list_add(&mod->entry, mods); + +#ifdef DSPACES_HAVE_PYTHON + PyGILState_STATE gstate; + gstate = PyGILState_Ensure(); +#endif // DSPACES_HAVE_PYTHON + ret = dspaces_init_mod(mod); +#ifdef DSPACES_HAVE_PYTHON + PyGILState_Release(gstate); +#endif // DSPACES_HAVE_PYTHON + + return(ret); +} + int build_module_arg_from_rank(long rank, struct dspaces_module_args *arg) { arg->name = strdup("rank"); @@ -285,22 +351,6 @@ void free_arg_list(struct dspaces_module_args *args, int len) } } -static struct dspaces_module *find_mod(struct list_head *mods, - const char *mod_name) -{ - struct dspaces_module *mod; - int i; - - list_for_each_entry(mod, mods, struct dspaces_module, entry) - { - if(strcmp(mod_name, mod->name) == 0) { - return (mod); - } - } - - return (NULL); -} - struct dspaces_module *dspaces_mod_by_od(struct list_head *mods, obj_descriptor *odsc) { diff --git a/src/dspaces-server.c b/src/dspaces-server.c index db7f646e..a6c8ef10 100644 --- a/src/dspaces-server.c +++ b/src/dspaces-server.c @@ -90,6 +90,7 @@ struct dspaces_provider { hg_id_t reg_id; hg_id_t get_mods_id; hg_id_t get_mod_id; + hg_id_t add_mod_id; struct list_head mods; struct ds_gspace *dsg; char **server_address; @@ -112,6 +113,7 @@ struct dspaces_provider { ABT_mutex dht_mutex; ABT_mutex sspace_mutex; ABT_mutex kill_mutex; + ABT_mutex mod_mutex; ABT_xstream drain_xstream; ABT_pool drain_pool; @@ -158,6 +160,7 @@ DECLARE_MARGO_RPC_HANDLER(get_var_objs_rpc); DECLARE_MARGO_RPC_HANDLER(reg_rpc); DECLARE_MARGO_RPC_HANDLER(get_mods_rpc); DECLARE_MARGO_RPC_HANDLER(get_mod_rpc); +DECLARE_MARGO_RPC_HANDLER(add_mod_rpc); static int init_sspace(dspaces_provider_t server, struct bbox *default_domain, struct ds_gspace *dsg_l) @@ -896,6 +899,7 @@ int dspaces_server_init(const char *listen_addr_str, MPI_Comm comm, ABT_mutex_create(&server->dht_mutex); ABT_mutex_create(&server->sspace_mutex); ABT_mutex_create(&server->kill_mutex); + ABT_mutex_create(&server->mod_mutex); hg = margo_get_class(server->mid); @@ -983,6 +987,10 @@ int dspaces_server_init(const char *listen_addr_str, MPI_Comm comm, &flag); DS_HG_REGISTER(hg, server->get_mod_id, get_mod_in_t, get_mod_out_t, get_mod_rpc); + margo_registered_name(server->mid, "add_mod_rpc", &server->add_mod_id, + &flag); + DS_HG_REGISTER(hg, server->add_mod_id, add_mod_in_t, int8_t, + add_mod_rpc); } else { server->put_id = MARGO_REGISTER(server->mid, "put_rpc", bulk_gdim_t, bulk_out_t, put_rpc); @@ -1090,6 +1098,11 @@ int dspaces_server_init(const char *listen_addr_str, MPI_Comm comm, get_mod_out_t, get_mod_rpc); margo_register_data(server->mid, server->get_mod_id, (void *)server, NULL); + server->add_mod_id = + MARGO_REGISTER(server->mid, "add_mod_rpc", add_mod_in_t, int8_t, + add_mod_rpc); + margo_register_data(server->mid, server->add_mod_id, (void *)server, + NULL); } int err = dsg_alloc(server, conf_file, comm); if(err) { @@ -3632,6 +3645,33 @@ static void get_mods_rpc(hg_handle_t handle) } DEFINE_MARGO_RPC_HANDLER(get_mods_rpc); +static void add_mod_rpc(hg_handle_t handle) +{ + margo_instance_id mid = margo_hg_handle_get_instance(handle); + const struct hg_info *info = margo_get_info(handle); + dspaces_provider_t server = + (dspaces_provider_t)margo_registered_data(mid, info->id); + add_mod_in_t in; + int8_t err; + + margo_get_input(handle, &in); + DEBUG_OUT("Received request to add module '%s'.\n", in.name); + + ABT_mutex_lock(server->mod_mutex); + // TOOD: don't ignore in.type + err = dspaces_server_add_module(&server->mods, in.name, in.namespace, in.url, DSPACES_MOD_PY); + ABT_mutex_unlock(server->mod_mutex); + if(err != 0) { + fprintf(stderr, "ERROR: (%s): failed to add module '%s' with %d\n", + __func__, in.name, err); + } + + margo_respond(handle, &err); + margo_free_input(handle, &in); + margo_destroy(handle); +} +DEFINE_MARGO_RPC_HANDLER(add_mod_rpc); + void dspaces_server_fini(dspaces_provider_t server) { int err = 0;