Update gras_simix. Modified the implementation of ports.

Now the pmm example works.

git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@3510 48e7efb5-ca39-0410-a469-dd3cf9ba447f

diff --git a/src/gras_simix/Msg/gras_simix_sg_msg.c b/src/gras_simix/Msg/gras_simix_sg_msg.c
index 89e544b..ee09ec1 100644 (file)
--- a/src/gras_simix/Msg/gras_simix_sg_msg.c
+++ b/src/gras_simix/Msg/gras_simix_sg_msg.c
@@ -38,6 +38,7 @@ void gras_msg_send_ext(gras_socket_t   sock,
                                This is used to report the load onto the simulator. It also counts the size of pointed stuff */
 
        sock_data = (gras_trp_sg_sock_data_t *)sock->data;
+
        hd = (gras_hostdata_t *)SIMIX_host_get_data(SIMIX_host_self());
 
   xbt_assert1(!gras_socket_is_meas(sock), 
@@ -45,7 +46,7 @@ void gras_msg_send_ext(gras_socket_t   sock,
        
        /* got the mutex my port */
        DEBUG1("Sock port %d",sock->port);
-       SIMIX_mutex_lock(hd->mutex_port[sock->port]);
+       SIMIX_mutex_lock(sock_data->mutex);
 
        /*initialize gras message */
        msg = xbt_new(s_gras_msg_t,1);
@@ -83,23 +84,23 @@ void gras_msg_send_ext(gras_socket_t   sock,
        SIMIX_cond_signal(trp_remote_proc->cond);
 
        /* wait for the receiver */
-       SIMIX_cond_wait(hd->cond_port[sock->port], hd->mutex_port[sock->port]);
+       SIMIX_cond_wait(sock_data->cond, sock_data->mutex);
 
        /* creates simix action and waits its ends, waits in the sender host condition*/
        act = SIMIX_action_communicate(sock_data->to_host, SIMIX_host_self(),msgtype->name, msg->payl_size, -1);
-       SIMIX_register_action_to_condition(act,hd->cond_port[sock->port]);
-       SIMIX_register_condition_to_action(act,hd->cond_port[sock->port]);
+       SIMIX_register_action_to_condition(act,sock_data->cond);
+       SIMIX_register_condition_to_action(act,sock_data->cond);
 
   VERB5("Sending to %s(%s) a message type '%s' kind '%s' ID %lu",
        SIMIX_host_get_name(sock_data->to_host),SIMIX_process_get_name(sock_data->to_process),
        msg->type->name,e_gras_msg_kind_names[msg->kind],       msg->ID);
        
-       SIMIX_cond_wait(hd->cond_port[sock->port], hd->mutex_port[sock->port]);
+       SIMIX_cond_wait(sock_data->cond, sock_data->mutex);
        /* error treatmeant */
 
        /* cleanup structures */
        SIMIX_action_destroy(act);
-       SIMIX_mutex_unlock(hd->mutex_port[sock->port]);
+       SIMIX_mutex_unlock(sock_data->mutex);
 
        VERB0("Message sent");
 
@@ -112,6 +113,7 @@ gras_msg_recv(gras_socket_t    sock,
              gras_msg_t       msg) {
 
        gras_trp_sg_sock_data_t *sock_data; 
+       gras_trp_sg_sock_data_t *remote_sock_data; 
        gras_hostdata_t *remote_hd;
   gras_msg_t msg_got;
        gras_msg_procdata_t msg_procdata = (gras_msg_procdata_t)gras_libdata_by_name("gras_msg");
@@ -122,6 +124,7 @@ gras_msg_recv(gras_socket_t    sock,
   xbt_assert0(msg,"msg is an out parameter of gras_msg_recv...");
 
        sock_data = (gras_trp_sg_sock_data_t *)sock->data;
+       remote_sock_data = ((gras_trp_sg_sock_data_t *)sock->data)->to_socket->data;
        DEBUG3("Remote host %s, Remote Port: %d Local port %d", SIMIX_host_get_name(sock_data->to_host), sock->peer_port, sock->port);
        remote_hd = (gras_hostdata_t *)SIMIX_host_get_data(sock_data->to_host);
 
@@ -131,17 +134,17 @@ gras_msg_recv(gras_socket_t    sock,
        DEBUG1("Size msg_to_receive buffer: %d", xbt_fifo_size(msg_procdata->msg_to_receive_queue));
   msg_got = xbt_fifo_shift(msg_procdata->msg_to_receive_queue);
 
-       SIMIX_mutex_lock(remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_mutex_lock(remote_sock_data->mutex);
 /* ok, I'm here, you can continuate the communication */
-       SIMIX_cond_signal(remote_hd->cond_port[sock->peer_port]);
+       SIMIX_cond_signal(remote_sock_data->cond);
 
 /* wait for communication end */
-       SIMIX_cond_wait(remote_hd->cond_port[sock->peer_port],remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_cond_wait(remote_sock_data->cond,remote_sock_data->mutex);
 
        msg_got->expe= msg->expe;
   memcpy(msg,msg_got,sizeof(s_gras_msg_t));
        xbt_free(msg_got);
-       SIMIX_mutex_unlock(remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_mutex_unlock(remote_sock_data->mutex);
 
        VERB3("Received a message type '%s' kind '%s' ID %lu",// from %s",
        msg->type->name,

diff --git a/src/gras_simix/Transport/gras_simix_sg_transport.c b/src/gras_simix/Transport/gras_simix_sg_transport.c
index 31c89c7..297fdd2 100644 (file)
--- a/src/gras_simix/Transport/gras_simix_sg_transport.c
+++ b/src/gras_simix/Transport/gras_simix_sg_transport.c
@@ -65,8 +65,11 @@ gras_socket_t gras_trp_select(double timeout) {
 
                if (sock_iter->meas || !sock_iter->outgoing)
                        continue;
+                       /*
                if ((sock_iter->peer_port == active_socket->port) && 
                                (((gras_trp_sg_sock_data_t*)sock_iter->data)->to_host == SIMIX_process_get_host(((gras_trp_sg_sock_data_t*)active_socket->data)->from_process))) {
+                               */
+               if ( (((gras_trp_sg_sock_data_t*)sock_iter->data)->to_socket == active_socket) && (((gras_trp_sg_sock_data_t*)sock_iter->data)->to_host == SIMIX_process_get_host(((gras_trp_sg_sock_data_t*)active_socket->data)->from_process)) ) {
                        SIMIX_mutex_unlock(pd->mutex);
                        return sock_iter;
                }
@@ -87,8 +90,8 @@ gras_socket_t gras_trp_select(double timeout) {
 
        res->port = -1;
 
-       /* unitialize the ports */
-       res->peer_port = active_socket->port;
+       /* initialize the ports */
+       //res->peer_port = active_socket->port;
        res->port = active_socket->peer_port;
 
        /* create sockdata */
@@ -96,6 +99,13 @@ gras_socket_t gras_trp_select(double timeout) {
        sockdata->from_process = SIMIX_process_self();
        sockdata->to_process   = ((gras_trp_sg_sock_data_t*)(active_socket->data))->from_process;
        
+       res->peer_port = ((gras_trp_procdata_t)gras_libdata_by_name_from_remote("gras_trp",sockdata->to_process))->myport;
+       sockdata->to_socket = active_socket;
+       /*update the peer to_socket  variable */
+       ((gras_trp_sg_sock_data_t*)active_socket->data)->to_socket = res;
+       sockdata->cond = SIMIX_cond_init();
+       sockdata->mutex = SIMIX_mutex_init();
+
        sockdata->to_host  = SIMIX_process_get_host(((gras_trp_sg_sock_data_t*)(active_socket->data))->from_process);
 
        res->data = sockdata;

diff --git a/src/gras_simix/Transport/gras_simix_transport_interface.h b/src/gras_simix/Transport/gras_simix_transport_interface.h
index bd24870..ad212b4 100644 (file)
--- a/src/gras_simix/Transport/gras_simix_transport_interface.h
+++ b/src/gras_simix/Transport/gras_simix_transport_interface.h
@@ -108,8 +108,7 @@ typedef struct {
        xbt_fifo_t active_socket;
        long int pid;
        long int ppid;
-  //int chan;    /* Formated messages channel */
-  //int measChan; /* Unformated echange channel for performance measurement*/
+
    
 } s_gras_trp_procdata_t,*gras_trp_procdata_t;
 

diff --git a/src/gras_simix/Transport/gras_simix_transport_plugin_sg.c b/src/gras_simix/Transport/gras_simix_transport_plugin_sg.c
index 0debd25..7803cf2 100644 (file)
--- a/src/gras_simix/Transport/gras_simix_transport_plugin_sg.c
+++ b/src/gras_simix/Transport/gras_simix_transport_plugin_sg.c
@@ -97,10 +97,8 @@ void gras_trp_sg_socket_client(gras_trp_plugin_t self,
 
   smx_host_t peer;
   gras_hostdata_t *hd;
-  gras_hostdata_t *local_hd;
   gras_trp_sg_sock_data_t *data;
   gras_sg_portrec_t pr;
-       int i;
 
   /* make sure this socket will reach someone */
   if (!(peer=SIMIX_host_get_by_name(sock->peer_name))) 
@@ -139,16 +137,10 @@ void gras_trp_sg_socket_client(gras_trp_plugin_t self,
        data->to_process         = pr.process;
   data->to_host      = peer;
 
-       /* searches for a free port on this host */
-       local_hd = (gras_hostdata_t *)SIMIX_host_get_data(SIMIX_host_self());
-       for (i=1;i<65536;i++) {
-               if (local_hd->cond_port[i] == NULL)
-                       break;
-       }
-       if (i == 65536) THROW0(system_error,0,"No port free");
-       sock->port = i;
-       local_hd->cond_port[i] = SIMIX_cond_init();
-       local_hd->mutex_port[i] = SIMIX_mutex_init();
+       /* initialize mutex and condition of the socket */
+       data->mutex = SIMIX_mutex_init();
+       data->cond = SIMIX_cond_init();
+       data->to_socket = pr.socket; 
 
   sock->data = data;
   sock->incoming = 1;
@@ -163,7 +155,6 @@ void gras_trp_sg_socket_server(gras_trp_plugin_t self,
                               gras_socket_t sock){
 
   gras_hostdata_t *hd=(gras_hostdata_t *)SIMIX_host_get_data(SIMIX_host_self());
-  //gras_trp_procdata_t pd=(gras_trp_procdata_t)gras_libdata_by_id(gras_trp_libdata_id);
   gras_sg_portrec_t pr;
   gras_trp_sg_sock_data_t *data;
   volatile int found;
@@ -193,11 +184,9 @@ void gras_trp_sg_socket_server(gras_trp_plugin_t self,
 
   pr.port   = sock->port;
   pr.meas    = sock->meas;
+       pr.socket = sock;
        pr.process = SIMIX_process_self();
   xbt_dynar_push(hd->ports,&pr);
-
-       hd->cond_port[sock->port] = SIMIX_cond_init();
-       hd->mutex_port[sock->port] = SIMIX_mutex_init();
   
   /* Create the socket */
   data = xbt_new(gras_trp_sg_sock_data_t,1);
@@ -205,6 +194,9 @@ void gras_trp_sg_socket_server(gras_trp_plugin_t self,
   data->to_process       = NULL;
        data->to_host      = SIMIX_host_self();
   
+       data->cond = SIMIX_cond_init();
+       data->mutex = SIMIX_mutex_init();
+
   sock->data = data;
 
   VERB6("'%s' (%ld) ears on %s:%d%s (%p)",
@@ -214,10 +206,6 @@ void gras_trp_sg_socket_server(gras_trp_plugin_t self,
 }
 
 void gras_trp_sg_socket_close(gras_socket_t sock){
-       xbt_dynar_t sockets = ((gras_trp_procdata_t) gras_libdata_by_name("gras_trp"))->sockets;
-       gras_socket_t sock_iter;
-       int cursor;
-       int found;
   gras_hostdata_t *hd=(gras_hostdata_t *)SIMIX_host_get_data(SIMIX_host_self());
   int cpt;
   gras_sg_portrec_t pr; 
@@ -228,25 +216,11 @@ void gras_trp_sg_socket_close(gras_socket_t sock){
 
   xbt_assert0(hd,"Please run gras_process_init on each process");
 
-  if (sock->data)
-    free(sock->data);
-
-       /* search for a socket in the list that is using the mutex and condition. It can happen because we create 2 sockets to communicate (incomming and outgoing) */
-       found = 0;
-       xbt_dynar_foreach(sockets,cursor,sock_iter) {
-               if (sock_iter->port == sock->port) {
-                       found = 1;
-                       break;
-               }
+  if (sock->data) {
+       SIMIX_cond_destroy(((gras_trp_sg_sock_data_t*)sock->data)->cond);
+               SIMIX_mutex_destroy(((gras_trp_sg_sock_data_t*)sock->data)->mutex);
+               free(sock->data);
        }
-       /* if not found, it is the last socket opened in this port and we can free the mutex and condition */
-       if (!found) {
-               SIMIX_cond_destroy(hd->cond_port[sock->port]);
-               hd->cond_port[sock->port] = NULL;
-               SIMIX_mutex_destroy(hd->mutex_port[sock->port]);
-               hd->mutex_port[sock->port] = NULL;
-       }
-
 
   if (sock->incoming && !sock->outgoing && sock->port >= 0) {
     /* server mode socket. Unregister it from 'OS' tables */
@@ -285,6 +259,7 @@ void gras_trp_sg_chunk_send_raw(gras_socket_t sock,
 
        smx_action_t act; /* simix action */
        gras_trp_sg_sock_data_t *sock_data; 
+       gras_trp_sg_sock_data_t *remote_sock_data; 
        gras_hostdata_t *hd;
        gras_hostdata_t *remote_hd;
        gras_trp_procdata_t trp_remote_proc;
@@ -292,6 +267,7 @@ void gras_trp_sg_chunk_send_raw(gras_socket_t sock,
        gras_msg_t msg; /* message to send */
 
        sock_data = (gras_trp_sg_sock_data_t *)sock->data;
+       remote_sock_data = ((gras_trp_sg_sock_data_t *)sock->data)->to_socket->data;
        hd = (gras_hostdata_t *)SIMIX_host_get_data(SIMIX_host_self());
        remote_hd = (gras_hostdata_t *)SIMIX_host_get_data(sock_data->to_host);
 
@@ -319,9 +295,11 @@ void gras_trp_sg_chunk_send_raw(gras_socket_t sock,
        trp_remote_proc = (gras_trp_procdata_t)gras_libdata_by_name_from_remote("gras_trp",sock_data->to_process);
        SIMIX_cond_signal(trp_remote_proc->cond);
 
-       SIMIX_mutex_lock(remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_mutex_lock(remote_sock_data->mutex);
+       //SIMIX_mutex_lock(remote_hd->mutex_port[sock->peer_port]);
        /* wait for the receiver */
-       SIMIX_cond_wait(remote_hd->cond_port[sock->peer_port], remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_cond_wait(remote_sock_data->cond,remote_sock_data->mutex);
+       //SIMIX_cond_wait(remote_hd->cond_port[sock->peer_port], remote_hd->mutex_port[sock->peer_port]);
 
        /* creates simix action and waits its ends, waits in the sender host condition*/
   DEBUG5("send chunk %s from %s to  %s:%d (size=%ld)",
@@ -329,18 +307,26 @@ void gras_trp_sg_chunk_send_raw(gras_socket_t sock,
         SIMIX_host_get_name(sock_data->to_host), sock->peer_port,size);
 
        act = SIMIX_action_communicate(sock_data->to_host, SIMIX_host_self(),name, size, -1);
+       /*
        SIMIX_register_action_to_condition(act,remote_hd->cond_port[sock->peer_port]);
        SIMIX_register_condition_to_action(act,remote_hd->cond_port[sock->peer_port]);
+       */
+       SIMIX_register_action_to_condition(act,remote_sock_data->cond);
+       SIMIX_register_condition_to_action(act,remote_sock_data->cond);
 
        SIMIX_host_get_name(sock_data->to_host),SIMIX_process_get_name(sock_data->to_process),
        
-       SIMIX_cond_wait(remote_hd->cond_port[sock->peer_port], remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_cond_wait(remote_sock_data->cond,remote_sock_data->mutex);
+       //SIMIX_cond_wait(remote_hd->cond_port[sock->peer_port], remote_hd->mutex_port[sock->peer_port]);
        /* error treatmeant */
 
        /* cleanup structures */
        SIMIX_action_destroy(act);
-       SIMIX_mutex_unlock(remote_hd->mutex_port[sock->peer_port]);
-       SIMIX_cond_signal(remote_hd->cond_port[sock->peer_port]);
+
+       SIMIX_mutex_unlock(remote_sock_data->mutex);
+       //SIMIX_mutex_unlock(remote_hd->mutex_port[sock->peer_port]);
+       SIMIX_cond_signal(remote_sock_data->cond);
+       //SIMIX_cond_signal(remote_hd->cond_port[sock->peer_port]);
                                /*
   m_task_t task=NULL;
   char name[256];
@@ -400,12 +386,15 @@ int gras_trp_sg_chunk_recv(gras_socket_t sock,
 
   msg_got = xbt_fifo_shift(msg_procdata->msg_to_receive_queue);
 
-       SIMIX_mutex_lock(local_hd->mutex_port[sock->port]);
+       SIMIX_mutex_lock(sock_data->mutex);
+       //SIMIX_mutex_lock(local_hd->mutex_port[sock->port]);
 /* ok, I'm here, you can continue the communication */
-       SIMIX_cond_signal(local_hd->cond_port[sock->port]);
+       SIMIX_cond_signal(sock_data->cond);
+       //SIMIX_cond_signal(local_hd->cond_port[sock->port]);
 
 /* wait for communication end */
-       SIMIX_cond_wait(local_hd->cond_port[sock->port],local_hd->mutex_port[sock->port]);
+       SIMIX_cond_wait(sock_data->cond,sock_data->mutex);
+       //SIMIX_cond_wait(local_hd->cond_port[sock->port],local_hd->mutex_port[sock->port]);
 
 
   if (msg_got->payl_size != size)
@@ -420,9 +409,13 @@ int gras_trp_sg_chunk_recv(gras_socket_t sock,
        if (msg_got->payl)
                xbt_free(msg_got->payl);        
        xbt_free(msg_got);
+       SIMIX_cond_wait(sock_data->cond,sock_data->mutex);
+       SIMIX_mutex_unlock(sock_data->mutex);
+       //SIMIX_mutex_unlock(local_hd->mutex_port[sock->port]);
+       /*
        SIMIX_cond_wait(local_hd->cond_port[sock->port],local_hd->mutex_port[sock->port]);
        SIMIX_mutex_unlock(local_hd->mutex_port[sock->port]);
-
+       */
                                        /*
   gras_trp_procdata_t pd=(gras_trp_procdata_t)gras_libdata_by_id(gras_trp_libdata_id);
 

diff --git a/src/gras_simix/Virtu/gras_simix_sg_process.c b/src/gras_simix/Virtu/gras_simix_sg_process.c
index 4a0189b..6a88042 100644 (file)
--- a/src/gras_simix/Virtu/gras_simix_sg_process.c
+++ b/src/gras_simix/Virtu/gras_simix_sg_process.c
@@ -19,7 +19,6 @@ static long int PID = 1;
 
 void
 gras_process_init() {
-       int i;
   gras_hostdata_t *hd=(gras_hostdata_t *)SIMIX_host_get_data(SIMIX_host_self());
   gras_procdata_t *pd=xbt_new0(gras_procdata_t,1);
   gras_trp_procdata_t trp_pd;
@@ -34,11 +33,6 @@ gras_process_init() {
     hd=xbt_new(gras_hostdata_t,1);
     hd->refcount = 1;
     hd->ports = xbt_dynar_new(sizeof(gras_sg_portrec_t),NULL);
-
-               for (i=0;i<65536;i++) {
-                       hd->cond_port[i] =NULL;
-                       hd->mutex_port[i] =NULL;
-               }
                SIMIX_host_set_data(SIMIX_host_self(),(void*)hd);
   } else {
     hd->refcount++;
@@ -63,7 +57,6 @@ gras_process_init() {
 
 void
 gras_process_exit() {
-       int i;
        xbt_dynar_t sockets = ((gras_trp_procdata_t) gras_libdata_by_name("gras_trp"))->sockets;
   gras_socket_t sock_iter;
   int cursor;
@@ -97,14 +90,6 @@ gras_process_exit() {
        }
   if ( ! --(hd->refcount)) {
     xbt_dynar_free(&hd->ports);
-               for (i=0; i<65536; i++) {
-                       if(hd->mutex_port[i] != NULL) {
-                               SIMIX_mutex_destroy(hd->mutex_port[i]);
-                               SIMIX_cond_destroy(hd->cond_port[i]);
-                               hd->mutex_port[i] = NULL;
-                               hd->cond_port[i] = NULL;
-                       }
-               }
     free(hd);
   }
   gras_procdata_exit();

diff --git a/src/gras_simix/Virtu/gras_simix_virtu_sg.h b/src/gras_simix/Virtu/gras_simix_virtu_sg.h
index d2fb67d..692a88a 100644 (file)
--- a/src/gras_simix/Virtu/gras_simix_virtu_sg.h
+++ b/src/gras_simix/Virtu/gras_simix_virtu_sg.h
@@ -14,20 +14,19 @@
 #include "xbt/dynar.h"
 //#include "msg/msg.h" /* SimGrid header */
 #include "simix/simix.h" /* SimGrid header */
+#include "gras_simix/Transport/gras_simix_transport_private.h"
 
 typedef struct {
   int port;  /* list of ports used by a server socket */
        int meas;   /* (boolean) the channel is for measurements or for messages */
        smx_process_t process;
+       gras_socket_t socket;
 } gras_sg_portrec_t;
 
 /* Data for each host */
 typedef struct {
   int refcount;
 
-       smx_cond_t cond_port[65536];
-       smx_mutex_t mutex_port[65536];
-
   xbt_dynar_t ports;
 
 } gras_hostdata_t;
@@ -41,6 +40,9 @@ typedef struct {
 
   smx_host_t to_host;   /* Who's on other side */
 
+       smx_cond_t cond;
+       smx_mutex_t mutex;
+       gras_socket_t to_socket;
 } gras_trp_sg_sock_data_t;