Add notion of transactions + loging to NGI

master
JT Acquaviva 2019-08-25 22:33:51 +02:00
parent 79e601775e
commit fed97e48bc
3 changed files with 148 additions and 0 deletions

View File

@ -1,4 +1,5 @@
#include <ngi.h>
#include <ngi_log.h>
#include <assert.h>
#include <float.h>
#include <math.h>
@ -22,6 +23,99 @@ int server_reduce(FILE * file, ngi_op_t op, ngi_type_t type, void * out_buff, si
return 0;
}
/*
* this function should be called by IME client on the compute node
*/
int ngi_client_side_reduce(FILE * file, ngi_op_t op, ngi_type_t type, void * out_buff, size_t off, size_t size){
ngi_dbg("Starting offset %lu size %lu Byte\n",off, size);
assert(size <= NGI_EXT_SZ);
fseek(file, off, SEEK_SET);
assert(type == NGI_TYPE_FLOAT);
assert(op == NGI_OP_MAX);
int res = ngi_server_side_reduce(file, op, type, out_buff, off, size);
assert(res == 0);
exit(0);
return res;
}
/*
* this function should be called by IME server on the storage appliance
*/
int ngi_server_side_reduce(FILE * file, ngi_op_t op, ngi_type_t type, void * out_buff, size_t off, size_t ext_sz){
ngi_dbg("Starting offset %lu size %lu Byte\n",off, ext_sz);
assert(type == NGI_TYPE_FLOAT);
assert(op == NGI_OP_MAX);
assert(ext_sz <= NGI_EXT_SZ);
float * data = malloc(ext_sz);
fseek(file, off, 0);
fread(data, ext_sz, 1, file);
float mx = -INFINITY;
for(size_t i=0; i < ext_sz/sizeof(float); i++){
mx = data[i] > mx ? data[i]: mx;
}
ngi_dbg("Result: %f \n", mx);
*(float*) out_buff = mx;
free(data);
return 0;
}
/*
* this function split the reduction in a set of client server transaction
*/
int ngi_reduce_stub(FILE * file, ngi_op_t op, ngi_type_t type, void * out_buff, size_t off, size_t size){
fseek(file, off, SEEK_SET);
assert(type == NGI_TYPE_FLOAT);
assert(op == NGI_OP_MAX);
u_int32_t nb_tx = size / NGI_EXT_SZ;
// handle remainder in extend computation
if (size % NGI_EXT_SZ != 0)
nb_tx += 1;
float *tmp_red = (float*) malloc(sizeof(float*) * nb_tx);
/* will be executed in parallel withint the client */
ngi_dbg("Reduction from %lu to %lu Byte in %u transactions\n", off, size, nb_tx);
for(u_int32_t i = 0; i < nb_tx ; i++){
size_t ext_off = off + NGI_EXT_SZ * i;
int res = ngi_client_side_reduce(file, op, type, &tmp_red[i], ext_off, NGI_EXT_SZ);
assert(res == 0);
}
float mx = -INFINITY;
for(u_int32_t i = 0; i < nb_tx ; i++){
mx = tmp_red[i] > mx ? tmp_red[i]: mx;
}
*(float*) out_buff = mx;
free(tmp_red);
return 0;
}
int ngi_reduce(FILE * file, ngi_op_t op, ngi_type_t type, void * out_buff, size_t off, size_t size){
fseek(file, off, SEEK_SET);
assert(type == NGI_TYPE_FLOAT);
assert(op == NGI_OP_MAX);
float * data = malloc(size);
fread(data, size, 1, file);
float mx = -INFINITY;
for(size_t i=0; i < size/sizeof(float); i++){
mx = data[i] > mx ? data[i]: mx;
}
*(float*) out_buff = mx;
free(data);
return 0;
}
int server_reduce_any_func(FILE * file, void (*func)(int8_t * data, void * out_buff, size_t off, size_t size), void * out_buff, size_t max_out_data_size, size_t off, size_t size){
fseek(file, off, SEEK_SET);

View File

@ -3,6 +3,12 @@
#include <stdlib.h>
#include <stdio.h>
//#include "ngi_log.h"
/*
* Define the size in B of file extent exchanged between server and clint
*/
#define NGI_EXT_SZ 1048576
typedef enum NGI_OP{
NGI_OP_MAX
@ -12,6 +18,12 @@ typedef enum NGI_TYPE{
NGI_TYPE_FLOAT
} ngi_type_t;
/*
*
*/
int ngi_server_side_reduce(FILE * , ngi_op_t , ngi_type_t , void * , size_t , size_t );
int ngi_client_side_reduce(FILE * , ngi_op_t , ngi_type_t , void * , size_t , size_t );
int ngi_reduce_stub(FILE * , ngi_op_t , ngi_type_t , void * , size_t , size_t );
/*
* This function invoces a well known reduction operation on data of a well known type
* Return 0 upon success, any other code, if e.g., short read

View File

@ -0,0 +1,42 @@
/******************************************************************************
*
* NGI log utilities
*
******************************************************************************/
#ifndef COMMON_NGI_LOG_H
#define COMMON_NGI_LOG_H
#include <stdlib.h>
#include <stdio.h>
enum ngi_log_level
{
NGI_LOG_DBG,
NGI_LOG_WARN,
NGI_LOG_ERROR,
NGI_LOG_MAX
};
enum ngi_log_level NGI_Log_Level = NGI_LOG_DBG;
/*
#define ngi_log(lvl, fmt, ...) \
do { if (lvl >= NGI_Log_Level) fprintf(stderr, "%s:%d:%s(): %s" fmt,\
__func__, __VA_ARGS__); } while (0)
*/
#define ngi_log(lvl, fmt, ...) \
do { if (lvl >= NGI_Log_Level ) fprintf(stderr, "== ngi_log == %s: " fmt, __func__, ## __VA_ARGS__);}\
while (0)
#define ngi_dbg(msg, ...) ngi_log(NGI_LOG_DBG, msg, ##__VA_ARGS__)
#define ngi_warn(msg, ...) ngi_log(NGI_LOG_WARN, msg, ##__VA_ARGS__)
#define ngi_error(msg, ...) ngi_log(NGI_LOG_ERROR, msg, ##__VA_ARGS__)
/* getter and setters */
#define ngi_log_level_set(ngi_log_level) NGI_Log_Level = ngi_log_level;
#define ngi_log_level_get(ngi_log_level) ngi_log_level;
#endif