nsc.c

Go to the documentation of this file.

#include "nsc.h"

#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <openssl/sha.h>
#include <signal.h>

#include "CRC64.h"

/* ############ Simple Context MGT ############ */
char * Context_host_from_env()
{
        char *raw_addr = getenv( "NODE_SHARED_CACHE" );

        if( !raw_addr )
        {
                raw_addr = "127.0.0.1:9874";
        }

        return raw_addr;
}

char * Context_passwd_from_env()
{
        char *pswd = getenv( "NODE_SHARED_CACHE_PWD" );

        if( !pswd )
        {
                pswd = "";
        }

        return pswd;
}


int Context_init_data( struct Context *ctx, char *phost_port, char *password )
{
        if( phost_port )
        {
        
                char * host_port = strdup( phost_port );
                
                char *sep = strstr( host_port , ":");

                if( !sep )
                {
                        printf("Error address must be in the 'ip:port' format\n");
                        return 1;
                }

                *sep = '\0';

                ctx->addr = strdup(host_port);
                ctx->s_port = strdup(sep + 1);
                        
                free( host_port );

                char *endptr;

                ctx->port = strtol(ctx->s_port, &endptr, 10);


                if( endptr == ctx->s_port )
                {
                        printf("Error parsing port number\n");
                        return 1;
                }
                
        }
        else
        {
                ctx->addr = NULL;
                ctx->s_port = 0;
        }

        if( password )
        {
                ctx->passwd = strdup(password);
        }
        else
        {
                ctx->passwd = strdup("");
        }

        return 0;
}

int Context_init( struct Context *ctx )
{
        return Context_init_data( ctx, Context_host_from_env(), Context_passwd_from_env() );
}

void Context_release( struct Context *ctx )
{
        free( ctx->addr );
        ctx->addr = NULL;

        free( ctx->s_port );
        ctx->s_port  = NULL;

        free( ctx->passwd );
        ctx->passwd  = NULL;

        ctx->port = -1;
}

/*
        NODE SHARED CACHE
*/


/* Internals */
static inline int _nodeSharedCache_send( int socket, char *key, uint32_t len_key,  char *json, uint32_t len_json, uint32_t action );

static inline int _nodeSharedCache_send_up( struct nodeSharedCache *nsc , char *key, uint32_t len_key,  char *json, uint32_t len_json, uint32_t action );
int nodeSharedCache_send_up( struct nodeSharedCache *nsc , char *key,  char *json, uint32_t action );

static inline int _nodeSharedCache_send_down( struct nodeSharedCache *nsc , char *key, uint32_t len_key,  char *json, uint32_t len_json, uint32_t action );
int nodeSharedCache_send_down( struct nodeSharedCache *nsc , char *key,  char *json, uint32_t action );

void socket_flag_timeouts( int socket, int *socket_flagged_read, int *socket_flagged_write)
{

   struct timeval tv;
   tv.tv_sec = 5;
   tv.tv_usec = 0;

        
   /* Try to set some time out flags */
   if( setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv,sizeof(struct timeval) )  != -1 )
   {
           if( socket_flagged_read )
                *socket_flagged_read = 1;
   }
   
   if( setsockopt(socket, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv,sizeof(struct timeval) )  != -1 )
   {
           if( socket_flagged_write )
                *socket_flagged_write = 1;
   }
}


int nodeSharedCache_incoming_set( struct nodeSharedCache *nsc , char * key_buff, char *read_buff, uint32_t data_len )
{
        read_buff[data_len] = '\0';
        
        //printf("==> SET { %s : %s }\n", key_buff, read_buff );
        
        jsonCache_set_json_nocommit( &nsc->cache, key_buff, read_buff );

        read_buff[0] = '\0';

        return 0;
}



int safe_write( int fd, void *src, size_t size )
{
        errno = 0;
        size_t left_to_write = size;

        do
        {
                struct sigaction new, old;

        new.sa_handler = SIG_IGN;
        sigemptyset (&new.sa_mask);
        new.sa_flags = 0;
                
                sigaction( SIGPIPE, &new, &old );
                
                int ret = write( fd, src, size );

                sigaction( SIGPIPE, &old, NULL );

                if( ret <= 0 )
                {
                        if( errno == EAGAIN )
                                continue;

                        else if( errno != 0 )
                        {
                                //perror("write");
                                return -1;
                        }

                        return 0;
                }
                else
                {
                        left_to_write -= ret;
                }


        }while( left_to_write );


        return size;
}

static inline int _nodeSharedCache_send( int socket, char *key, uint32_t len_key,  char *json, uint32_t len_json, uint32_t action )
{
        uint32_t sizes[4];
        sizes[0] = 0x47; /* Sync byte (just in case) */
        sizes[1] = action;
        sizes[2] = len_key;
        sizes[3] = len_json;

        /* Send sizes */
        if( safe_write( socket, sizes, sizeof(uint32_t) * 4 ) < 0 )
        {
                return 1;
        }

        /* Send key */
        if( safe_write( socket, key, sizes[2]) < 0 )
        {
                return 1;
        }

        /* Send data */
        if( safe_write( socket, json, sizes[3]) < 0 )
        {
                return 1;
        }

        return 0;
}

static inline int _nodeSharedCache_send_up( struct nodeSharedCache *nsc , char *key, uint32_t len_key,  char *json, uint32_t len_json, uint32_t action )
{
        int ret = 0;
        
        if( !nsc->connected )
        {
                /* In this case we are the root so we have nothing to send up */
                return 0;
        }
        
        pthread_mutex_lock( &nsc->socket_w_lock );
        
        ret = _nodeSharedCache_send( nsc->socket, key, len_key,  json, len_json, action );
        
        pthread_mutex_unlock( &nsc->socket_w_lock );
        
        return ret;
}


int nodeSharedCache_send_up( struct nodeSharedCache *nsc , char *key,  char *json, uint32_t action )
{
        int len_key = strlen( key );
        int len_json = strlen( json );

        if( !len_key  || !len_json )
                return 1;

        return _nodeSharedCache_send_up( nsc, key, len_key,  json, len_json, action );
}


static inline int _nodeSharedCache_send_down( struct nodeSharedCache *nsc , char *key, uint32_t len_key,  char *json, uint32_t len_json, uint32_t action )
{
        int ret = 0;
        int i;
        
        for( i = 0 ; i < nsc->connected_clients ; i++ )
        {
                //printf("Sending to %d\n", i );
                pthread_mutex_lock( &nsc->client_socket_w_lock[i] );
                
                ret = _nodeSharedCache_send( nsc->clients[i], key, len_key,  json, len_json, action );
                
                if( ret )
                {
                        pthread_mutex_unlock( &nsc->client_socket_w_lock[i] );
                        break;
                }
                
                pthread_mutex_unlock( &nsc->client_socket_w_lock[i] );
        }
        
        return ret;
}


int nodeSharedCache_send_down( struct nodeSharedCache *nsc , char *key,  char *json, uint32_t action )
{       
        int len_key = strlen( key );
        int len_json = strlen( json );

        if( !len_key  || !len_json )
                return 1;

        return _nodeSharedCache_send_down( nsc, key, len_key,  json, len_json, action );
}



static inline int nodeSharedCacheRead_Header( int socket, uint32_t *action, uint32_t *key_len, uint32_t *data_len )
{
        uint32_t header[4];
        
        errno = 0;
        int ret = read( socket, header, sizeof(uint32_t) * 4 );
        
        if( ret !=  sizeof(uint32_t) * 4 )
        {
                if( ret == 0 || errno == EBADF )
                        return 1;

                printf("BAD protocol : Could not read header\n");
                //perror("read");
                return 1;
        }
        
        if( header[0] != 0x47 )
        {
                printf("BAD protocol : sync byte is erroneous\n");
                return 1;
        }
        
        *action = header[1];
        *key_len = header[2];
        *data_len = header[3];

        return 0;
}

static inline int nodeSharedCacheRead_Key( int socket, char *key_buff, uint32_t key_len )
{
        int ret = 0;
        errno = 0;
        ret = read( socket, key_buff, key_len );

        if( ret !=  key_len )
        {
                if( ret == 0 || errno == EBADF )
                        return 1;

                printf("Could not read KEY\n");
                //perror("read");
                return 1;
        }

        key_buff[key_len] = '\0';
        
        return 0;
}

static inline int nodeSharedCacheRead_Payload( int socket, char *read_buff, uint32_t data_len, uint32_t max_buffer_size )
{
        if( ( max_buffer_size - 1 ) <= data_len )
        {
                printf("ERROR Message too large BAD PROTOCOL");
                return 1;
        }
        else
        {
                errno = 0;
                int ret = read( socket,  read_buff, data_len );

                if( ret !=  data_len )
                {
                        if( ret == 0 || errno == EBADF )
                                return 1;

                        printf("COULD not read payload");
                        //perror("read");
                        return 1;
                }

        }

        return 0;
}




void * nodeSharedCache_read_loop( void *pnsc )
{
        struct nodeSharedCache *nsc = (struct nodeSharedCache *) pnsc;
        
        char *key_buff = malloc( nsc->max_key_len + 1 );
        
        if( !key_buff )
        {
                perror("malloc");
                abort();
        }

        char *read_buff = malloc( nsc->max_buffer_size );

        if( !read_buff )
        {
                perror("malloc");
                abort();
        }

        
        while( 1 )
        {       
                uint32_t action = 0;
                uint32_t key_len = 0;
                uint32_t data_len = 0;
                
                if( nodeSharedCacheRead_Header( nsc->socket, &action, &key_len, &data_len ) )
                {
                        break;
                }
                
                //printf("INCOMING NSC UP %d\n", action );
                
                if( nodeSharedCacheRead_Key( nsc->socket, key_buff, key_len ) )
                {
                        break;
                }
                
                if(  nodeSharedCacheRead_Payload( nsc->socket, read_buff, data_len, nsc->max_buffer_size ) )
                {
                        break;
                }
                
                                        
                if( action == NSC_CLOSE )
                {
                        break;
                }

                switch( action )
                {
                        case NSC_BCAST:
                                printf("BCAST %s\n", key_buff);
                                /* Send DOWN the tree and set LOCAL */
                                if(  _nodeSharedCache_send_down( nsc , key_buff, key_len,  read_buff, data_len, NSC_BCAST ) )
                                {
                                        break;
                                }
                        case NSC_REDUCE:
                                printf("REDUCE %s\n", key_buff);
                                /* As reduce are just supposed to go up just set local */
                        case NSC_SET :
                                printf("=> SET %s\n", key_buff);
                                /* Set local from up the tree */
                                nodeSharedCache_incoming_set( nsc , key_buff, read_buff, data_len );
                        break;
                        case NSC_DELETE:
                                printf("DEL %s\n", key_buff);
                                /* Delete local from up the tree */
                                jsonCache_delete_nocommit( &nsc->cache, key_buff );                             
                        break;
                }
                
                sched_yield();
        }

        free( read_buff );
        free( key_buff );

        close( nsc->socket );

        //printf("Closed connection\n");
        
        return NULL;
}


uint64_t nodeSharedCache_compute_auth_digest( char *passwd, char *challenge )
{
        MALP_Trace_crc64_init();
        
        int pwdlen = strlen(passwd);
        
        uint64_t crc1, crc2; 
        
        crc1 = MALP_Trace_crc64(passwd, pwdlen );
        crc2 = MALP_Trace_crc64(challenge, 512 );
        
        unsigned char CRCS[64];

        sprintf(CRCS, "%016llx%016llx",crc1, crc2 );

        return MALP_Trace_crc64(CRCS, 32 );
}



int nodeSharedCache_do_auth( struct nodeSharedCache *nsc )
{
        char challenge[512];

        if( read( nsc->socket, challenge, 512 ) != 512 )
        {
                //perror("read");
                return 1;
        }

        uint64_t hex_digest =  nodeSharedCache_compute_auth_digest( nsc->ctx.passwd, challenge );

        if( safe_write( nsc->socket, &hex_digest, sizeof(uint64_t) ) != sizeof(uint64_t) )
        {
                //perror("write");
                return 1;
        }

        uint32_t ret;

        if( read( nsc->socket, &ret, sizeof(uint32_t) ) != sizeof(uint32_t) )
        {
                //perror("read");
                return 1;
        }

        return ret;
}


int _nodeSharedCache_init( struct nodeSharedCache *nsc)
{
        if( !nsc )
                return 1;

        nsc->been_waited = 0;
        
        nsc->client_socket_w_lock = NULL;
        nsc->client_threads = NULL;
        nsc->clients = NULL;
        nsc->connected = 0;
        
        /* ############ Setup Datastructures ############ */
        jsonCache_init( &nsc->cache, nodeSharedCache_set_handler, nodeSharedCache_delete_handler, nsc );
        
        /* If we do not connect to a server just return */
        if( !nsc->ctx.addr )
        {
                /* Set a default config for future clients */
                nsc->max_key_len = 1024;
                nsc->max_buffer_size = 1024*1024;
                return 0;       
        }
        /* ############ Connect to NODE.js ############ */

        errno = 0;

        struct addrinfo *results;

        int ret = getaddrinfo( nsc->ctx.addr, nsc->ctx.s_port, NULL, &results);

        if( ret != 0 )
        {
                printf("Error resolving name : %s\n", gai_strerror(ret) );
                return 1;
        }

        struct addrinfo *current = results;

        int connected = 0;

        while( current )
        {
                errno = 0;
                nsc->socket =socket( current->ai_family, current->ai_socktype, current->ai_protocol );

                if( nsc->socket < 0 )
                {
                        perror( "socket" );
                }
                else
                {
                        errno = 0;
                        if (connect(nsc->socket, current->ai_addr, current->ai_addrlen) != -1)
                        {
                                //printf("Connected to %s:%d\n", nsc->ctx.addr, nsc->ctx.port );
                                connected = 1;
                                break;
                        }
                        else
                        {
                                perror("connect");
                        }
                
                }

                close( nsc->socket );

                current = current->ai_next;
        }

        freeaddrinfo(results);

        if( connected == 0 )
        {
                printf("Failed to connect to %s:%d\n", nsc->ctx.addr, nsc->ctx.port);
                return 1;
        }

        /* SAY HELLO and LOAD REMOTE CONFIG */
        char *greetings = "Hello NodeSharedCache";
        int greetings_len = strlen( greetings );

        if( safe_write( nsc->socket, greetings, greetings_len ) != greetings_len )
        {
                printf("Failed to register to node.js server\n");
                close( nsc->socket );
                return 1;
        }

        /* Authenticate */

        if( nodeSharedCache_do_auth( nsc ) )
        {
                printf("Server refused password\n");
                close( nsc->socket );
                return 1;
        }

        /* Retrieve remote config */
        uint32_t remote_config[3] = { 0 , 0 , 0};

        if( read( nsc->socket, remote_config, 3 * sizeof( uint32_t ) )  != (3 * sizeof( uint32_t )) )
        {
                perror("Failed to read remote configuration...\n");
                close( nsc->socket );
                return 1;
        }

        if( remote_config[0] != 777 )
        {
                printf("Error bad server protocol\n");
                close( nsc->socket );
                return 1;
        }

        nsc->max_key_len = remote_config[1];
        nsc->max_buffer_size = remote_config[2];

        pthread_mutex_init( &nsc->socket_w_lock, NULL);


        /* ############ Start Reading/writing Threads ############ */
        pthread_create( &nsc->th, NULL, nodeSharedCache_read_loop, (void *)nsc);
        
        //pthread_detach( nsc->th );

        nsc->connected = 1;

        return 0;
}

int nodeSharedCache_init( struct nodeSharedCache *nsc )
{
        if( !nsc )
                return 1;

        /* Fill in ctx with default values */
        if( Context_init( &nsc->ctx ) )
                return 1;

        nsc->expected_clients = 0;
        nsc->connected_clients = 0;

        /* Now connect to the server/proxy */
        if(  _nodeSharedCache_init( nsc) )
                return 1;

        return 0;
}


int nodeSharedCache_proxy_greetings( int socket )
{
        int ret = 0;
        char *greetings = "Hello NodeSharedCache";
        int greetings_len = strlen( greetings );

        char buff[50];
                                                           
        if( (ret = recv( socket, buff, greetings_len, 0 ) ) == -1 )
        {
           return -1;
        }

        /* Check greetings format */
        if( ret != greetings_len )
        {
           printf("Bad protocol in proxy (BAD LENGTH got %d)\n", ret);
           return -1;
        }

        if( memcmp( buff, greetings, greetings_len ) )
        {
                printf("Bad protocol in proxy (BAD greetings)\n");
                return -1;
        }
        
        return 0;
}

int nodeSharedCache_proxy_auth( struct nodeSharedCache *nsc, int socket )
{
        char challenge[512];
        int i;

        /* Prepare a challenge */
        for( i = 0 ; i < 512 ; i++ )
                challenge[i] = rand();

        uint64_t auth_digest = nodeSharedCache_compute_auth_digest( nsc->ctx.passwd, challenge );
        
        /* Send the challenge */
        if( send( socket, challenge, 512, 0 ) == -1 )
        {
                return -1;
        }
        
        uint64_t buff;
        /* Read the digest */
        if( recv( socket, &buff, sizeof(uint64_t), 0 )  == -1 )
        {
                return -1;
        }       
        
        uint32_t bad_password = 0;
        
        if(  memcmp(&auth_digest, &buff, sizeof(uint64_t)) )
        {
                printf("Bad password\n");
                bad_password = 1;
        }
        
        /* Inform the client of our decision */
        if( send( socket, (char *)&bad_password, sizeof(uint32_t), 0 )  == -1 )
        {
                return -1;
        }
        

        
        return !bad_password;
}

int nodeSharedCache_proxy_send_config( struct nodeSharedCache *nsc, int socket )
{
        /* Retrieve remote config */
        uint32_t remote_config[3] = { 777 , nsc->max_key_len , nsc->max_buffer_size};

        if( send( socket, (char *)remote_config, 3*sizeof(uint32_t), 0 ) == -1 )
        {
                printf("Failed to send nodeSharedCache config\n");
                return -1;
        }
        
        return 0;
}


struct nscRoutingThreadCtx
{
        struct nodeSharedCache *nsc;
        int id;
};


void * nodeSharedCache_proxy_routing_thread( void *arg )
{
        struct nscRoutingThreadCtx *ctx = (struct nscRoutingThreadCtx *)arg;
        
        int my_socket = ctx->nsc->clients[ ctx->id ];

        char *key_buff = malloc( ctx->nsc->max_key_len + 1 );
        
        if( !key_buff )
        {
                perror("malloc");
                abort();
        }

        char *read_buff = malloc( ctx->nsc->max_buffer_size );

        if( !read_buff )
        {
                perror("malloc");
                abort();
        }

        
        while( 1 )
        {
                uint32_t action = 0;
                uint32_t key_len = 0;
                uint32_t data_len = 0;
                
                if( nodeSharedCacheRead_Header( my_socket, &action, &key_len, &data_len ) )
                {
                        break;
                }


                if( nodeSharedCacheRead_Key( my_socket, key_buff, key_len ) )
                {
                        break;
                }
                
                if(  nodeSharedCacheRead_Payload( my_socket, read_buff, data_len, ctx->nsc->max_buffer_size ) )
                {
                        break;
                }
                
                                
                //printf("[DOWN %d] (%s : %s) over %d\n", action, key_buff, read_buff,  ctx->id );
                /* Here we want to speedup the closing process */
                if( action == NSC_CLOSE )
                {
                        _nodeSharedCache_send( my_socket, "EOF", 4,  "{}", 3, NSC_CLOSE );
                        break;
                }

                switch( action )
                {
                        case NSC_REDUCE:
                                /* Send UP the tree */
                                if(  _nodeSharedCache_send_up( ctx->nsc , key_buff, key_len,  read_buff, data_len, NSC_REDUCE ) )
                                {
                                        break;
                                }
                        case NSC_BCAST:
                                /* As bcast are just supposed to go down just set local */
                        case NSC_SET :
                                /* Set local from down the tree */
                                nodeSharedCache_incoming_set( ctx->nsc , key_buff, read_buff, data_len );
                        break;
                        case NSC_DELETE:
                                /* Delete local from down the tree */
                                jsonCache_delete_nocommit( &ctx->nsc->cache, key_buff );                                
                        break;
                }
                
                sched_yield();
        }

        free( read_buff );
        free( key_buff );
        free( ctx );
        
        close( my_socket );
        
        //printf("Proxy client disconnected\n");

        return NULL;
}


int nodeSharedCache_proxy_accept_clients( struct nodeSharedCache *nsc )
{
        
        if( nsc->expected_clients != 0 )
        {
                   /* We now accept clients */
                   while( nsc->connected_clients  < nsc->expected_clients )
                   {
                           socklen_t len = sizeof( struct sockaddr );
                           struct sockaddr remote_infos;
                           
                           int client_socket =  accept(nsc->listening_socket, &remote_infos, &len);
                           
                           if( client_socket == -1 )
                           {
                                   perror("accept");
                                   continue;
                           }
                           else
                           {
                                        /* Here a client is connected */

                                        /* Receive protocol greetings */
                                        if( nodeSharedCache_proxy_greetings( client_socket ) == -1 )
                                        {
                                                close(client_socket);
                                                continue;
                                        }

                                        /* Here the client seems to be a proxy
                                        * Lets do the authentification */
                                        int did_auth = 0;
                                        
                                        if( (did_auth = nodeSharedCache_proxy_auth( nsc, client_socket )) == -1 )
                                        {
                                                close(client_socket);
                                                continue;
                                        }
                                        
                                        if( did_auth == 0 )
                                        {
                                                printf("Client failed authentification\n");
                                                close(client_socket);
                                                continue;
                                        }
                                        
                                        /* We now send the config */
                                        if( nodeSharedCache_proxy_send_config( nsc, client_socket ) == -1 )
                                        {
                                                close(client_socket);
                                                continue;
                                        }
                                        
                                        /* Client is sucessfully authentified
                                        * start routing loop */
                                        nsc->clients[nsc->connected_clients] = client_socket;
                                        nsc->connected_clients ++;
                           }
                           
                   }

                        /* Here we close the listening socket
                         * as every clients are connected  */
                        close(nsc->listening_socket);
                   

                        /* We now create the threads for clients */
                        nsc->client_threads = malloc( nsc->expected_clients * sizeof( pthread_t ) );
                        
                        if( !nsc->client_threads )
                        {
                                perror( "malloc" );
                                return -1;
                        }

                        int i;
                        for( i = 0 ; i < nsc->expected_clients ; i++ )
                        {
                                struct nscRoutingThreadCtx *ctx = malloc( sizeof( struct nscRoutingThreadCtx ) );
                                
                                if( !ctx )
                                {
                                        perror( "malloc" );
                                        return -1;
                                }
                                
                                ctx->nsc = nsc;
                                ctx->id = i;
                                
                                pthread_create(  &nsc->client_threads[i], NULL, nodeSharedCache_proxy_routing_thread, (void *) ctx );
                                
                        }
   
        }
        
        return 0;
}





int nodeSharedCache_proxy_init( struct nodeSharedCache *nsc, char *destination_server, int expected_clients )
{
        if( Context_init_data( &nsc->ctx, destination_server, Context_passwd_from_env() ) )
                return -1;
        
        /* Fill in the suplementary expected clients attribute */
        nsc->expected_clients = expected_clients;
        nsc->connected_clients = 0;
        
        /* Connect to server/relay */
        if(  _nodeSharedCache_init( nsc) )
                return -1;

        nsc->client_socket_w_lock = malloc( nsc->expected_clients * sizeof( pthread_mutex_t ) + sizeof(int) );
        
        if( !nsc->client_socket_w_lock )
        {
                perror("malloc");
                return -1;
        }
        
        int i;
        for( i = 0 ; i < nsc->expected_clients ; i++ )
                pthread_mutex_init( &nsc->client_socket_w_lock[i], NULL);
        

                
        int proxy_port = 0;


        /* Start the proxy server to receive #expected_clients connections*/
        if( expected_clients != 0 )
        {
                errno = 0;
                
                struct addrinfo hints, *results;

                memset( &hints, 0, sizeof( struct addrinfo ) );
                hints.ai_family = AF_UNSPEC;
                hints.ai_socktype = SOCK_STREAM;
                hints.ai_flags = AI_PASSIVE;

                /* Note that we bind ourselves to port 0 in order to get a free port */
                int ret = getaddrinfo( NULL, "0", &hints, &results);

                if( ret != 0 )
                {
                        printf("Could not find a binding method : %s\n", gai_strerror(ret) );
                        return -1;
                }

                struct addrinfo *current = results;

                nsc->listening_socket = 0;
                
                nsc->clients = malloc( expected_clients * sizeof( int ) );
                
                if( !nsc->clients )
                {
                        perror("malloc");
                        return -1;
                }

                while( current )
                {
                        errno = 0;
                        nsc->listening_socket =socket( current->ai_family, current->ai_socktype, current->ai_protocol );

                        if( nsc->listening_socket < 0 )
                        {
                                perror( "socket" );
                                continue;
                        }

                        errno = 0;
                        
                        /* Bind the socket to the port */
                        if (bind(nsc->listening_socket, current->ai_addr, current->ai_addrlen) != -1)
                        {
                                        /* Start listening on that port */
                                   if (listen(nsc->listening_socket,  expected_clients) != -1)
                                   {
                                                        /* As we are bound to "0" we want to get the actual port */
                                                        struct sockaddr_in socket_infos;
                                                        socklen_t infolen = sizeof(struct sockaddr);
                                                        
                                                        if(getsockname(nsc->listening_socket, (struct sockaddr*)&socket_infos, &infolen) == -1 )
                                                        {
                                                                perror( "getsockname" );
                                                                return -1;
                                                        }
                                                        else
                                                        {
                                                                if( infolen == sizeof(struct sockaddr_in) )
                                                                        proxy_port = ntohs(socket_infos.sin_port);
                                                        }                          
                                           
                                                   //printf("Proxy listening on port %d\n", proxy_port);

                                                   /* Proxy server is now up and listening for clients */
                                                   break;
                   
                                   }
                                   else
                                   {
                                           perror("listen");
                                           close(nsc->listening_socket);
                                           return -1;
                                           
                                   }
                        }
                        else
                        {
                           perror("bind");
                           close(nsc->listening_socket);
                           return -1;
                        }

                        current = current->ai_next;
                }

                freeaddrinfo(results);

                if( nsc->listening_socket == 0 )
                {
                        printf("Failed to set a proxy server and establish connections\n");
                        return -1;
                }
        }
        
        return proxy_port;
}


void nodeSharedCache_wait( struct nodeSharedCache *nsc)
{       
        if( ! nsc->been_waited )
        {
                if( nsc->expected_clients == 0 )
                        nodeSharedCache_send_up( nsc , "EOF",  "{}", NSC_CLOSE );
        
                int i;
        
                for( i = 0 ; i < nsc->expected_clients ; i++ )
                {
                        pthread_join( nsc->client_threads[i], NULL );
                }

                if( nsc->expected_clients != 0 )
                        nodeSharedCache_send_up( nsc , "EOF",  "{}", NSC_CLOSE );
                                
                if( nsc->connected )
                {
                        pthread_join( nsc->th, NULL );
                }               


                nsc->been_waited = 1;
        }
}

void nodeSharedCache_release( struct nodeSharedCache *nsc)
{       
        nodeSharedCache_wait( nsc );

        jsonCache_release( &nsc->cache );

        Context_release( &nsc->ctx );
        pthread_mutex_destroy( &nsc->socket_w_lock );

        int i;
        for( i = 0 ; i < nsc->expected_clients ; i++ )
                pthread_mutex_destroy( &nsc->client_socket_w_lock[i] );
        
        free( nsc->client_socket_w_lock );
        free( nsc->client_threads );
        free( nsc->clients );
}



void nodeSharedCache_set_handler( char *key, json_t *elem, void *ctx )
{
        struct nodeSharedCache *nsc = (struct nodeSharedCache *)ctx;

        char *json = json_dump( elem, JSON_COMPACT | JSON_NO_LOCK );

        if( json )
        {
                nodeSharedCache_send_up( nsc, key, json, NSC_SET );
        }

        free( json );
        

}

void nodeSharedCache_delete_handler( char *key, void *ctx )
{
        struct nodeSharedCache *nsc = (struct nodeSharedCache *)ctx;

        nodeSharedCache_send_up( nsc, key, "", NSC_DELETE );
}


void nodeSharedCache_set_json( struct nodeSharedCache *nsc, char *key, char *json )
{
        /* Here we directly send to avoid a sumplementary DUMPING */
        nodeSharedCache_send_up( nsc, key, json, NSC_SET );
        
        jsonCache_set_json_nocommit( &nsc->cache, key, json );
}

void nodeSharedCache_set( struct nodeSharedCache *nsc, char *key, json_t *json )
{
        jsonCache_set( &nsc->cache, key, json );
}


void nodeSharedCache_reduce_json( struct nodeSharedCache *nsc, char *key, char *json )
{
        /* Here we directly send to avoid a sumplementary DUMPING */
        nodeSharedCache_send_up( nsc, key, json, NSC_REDUCE );
        
        jsonCache_set_json_nocommit( &nsc->cache, key, json );
}

void nodeSharedCache_reduce( struct nodeSharedCache *nsc, char *key, json_t *elem )
{               
        /* Here we explicitly do the parsing as we have no way to change the flag
         * to NSC_REDUCE in the set_handler */
        char *json = json_dump( elem, JSON_COMPACT | JSON_NO_LOCK );

        //printf("%d REDUCING KEY %s\n =============== \n %s \n ===================\n", getpid(), key,  json);

        if( json )
        {
                nodeSharedCache_send_up( nsc, key, json, NSC_REDUCE );
        }

        free( json );

        jsonCache_set_nocommit( &nsc->cache, key, elem );
}


json_t * nodeSharedCache_get( struct nodeSharedCache *nsc, char *key )
{
        return jsonCache_get( &nsc->cache, key );
}



char * nodeSharedCache_get_json( struct nodeSharedCache *nsc, char *key )
{
        json_t *ret = jsonCache_get( &nsc->cache, key );

        if( !ret )
                return strdup("{}");

        char *json = json_dump( ret, JSON_COMPACT );

        json_decref( ret );

        return json;
}


void nodeSharedCache_delete( struct nodeSharedCache *nsc, char *key )
{
        jsonCache_delete( &nsc->cache, key );
}

void nodeSharedCache_commit( struct nodeSharedCache *nsc, char *key )
{
        jsonCache_commit( &nsc->cache, key );
}