#include "gras_private.h"
#include "transport_private.h"
-#include "Virtu/virtu_sg.h"
+#include "gras/Virtu/virtu_sg.h"
GRAS_LOG_EXTERNAL_CATEGORY(transport);
-GRAS_LOG_NEW_DEFAULT_SUBCATEGORY(trp_sg,transport);
+GRAS_LOG_NEW_DEFAULT_SUBCATEGORY(trp_sg,transport,"SimGrid pseudo-transport");
/***
*** Prototypes
void gras_trp_sg_socket_close(gras_socket_t *sd);
gras_error_t gras_trp_sg_chunk_send(gras_socket_t *sd,
- char *data,
- size_t size);
+ const char *data,
+ long int size);
gras_error_t gras_trp_sg_chunk_recv(gras_socket_t *sd,
char *data,
- size_t size);
-
-/* FIXME
- gras_error_t gras_trp_sg_flush(gras_socket_t *sd);
-*/
+ long int size);
/***
*** Specific plugin part
static gras_error_t find_port(gras_hostdata_t *hd, int port,
gras_sg_portrec_t *hpd) {
int cpt;
- gras_sg_portrec_t pd;
+ gras_sg_portrec_t pr;
gras_assert0(hd,"Please run gras_process_init on each process");
- gras_dynar_foreach(hd->ports, cpt, pd) {
- if (pd.port == port) {
- memcpy(hpd,&pd,sizeof(gras_sg_portrec_t));
+ gras_dynar_foreach(hd->ports, cpt, pr) {
+ if (pr.port == port) {
+ memcpy(hpd,&pr,sizeof(gras_sg_portrec_t));
return no_error;
}
}
gras_error_t
gras_trp_sg_setup(gras_trp_plugin_t *plug) {
- gras_trp_sg_plug_data_t *data=malloc(sizeof(gras_trp_sg_plug_data_t));
+ gras_trp_sg_plug_data_t *data=gras_new(gras_trp_sg_plug_data_t,1);
if (!data)
RAISE_MALLOC;
plug->chunk_send = gras_trp_sg_chunk_send;
plug->chunk_recv = gras_trp_sg_chunk_recv;
+ plug->flush = NULL; /* nothing cached */
return no_error;
}
}
/* create the socket */
- if (!(data = malloc(sizeof(gras_trp_sg_sock_data_t))))
+ if (!(data = gras_new(gras_trp_sg_sock_data_t,1)))
RAISE_MALLOC;
data->from_PID = MSG_process_self_PID();
data->to_chan = pr.tochan;
sock->data = data;
+ sock->incoming = 1;
DEBUG6("%s (PID %d) connects in %s mode to %s:%d (to_PID=%d)",
MSG_process_get_name(MSG_process_self()), MSG_process_self_PID(),
}
/* Create the socket */
- if (!(data = malloc(sizeof(gras_trp_sg_sock_data_t))))
+ if (!(data = gras_new(gras_trp_sg_sock_data_t,1)))
RAISE_MALLOC;
data->from_PID = -1;
INFO6("'%s' (%d) ears on %s:%d%s (%p)",
MSG_process_get_name(MSG_process_self()), MSG_process_self_PID(),
- host,port,sock->raw? " (mode RAW)":"",*sock);
+ host,port,sock->raw? " (mode RAW)":"",sock);
return no_error;
}
void gras_trp_sg_socket_close(gras_socket_t *sock){
gras_hostdata_t *hd=(gras_hostdata_t *)MSG_host_get_data(MSG_host_self());
int cpt;
- gras_sg_portrec_t *pr;
+
+ gras_sg_portrec_t pr;
if (!sock) return;
gras_assert0(hd,"Please run gras_process_init on each process");
- free(sock->data);
+ if (sock->data)
+ gras_free(sock->data);
if (sock->incoming) {
/* server mode socket. Un register it from 'OS' tables */
gras_dynar_foreach(hd->ports, cpt, pr) {
- if (pr->port == sock->port) {
+ DEBUG2("Check pr %d of %lu", cpt, gras_dynar_length(hd->ports));
+ if (pr.port == sock->port) {
gras_dynar_cursor_rm(hd->ports, &cpt);
-
return;
}
}
} sg_task_data_t;
gras_error_t gras_trp_sg_chunk_send(gras_socket_t *sock,
- char *data,
- size_t size) {
+ const char *data,
+ long int size) {
m_task_t task=NULL;
static unsigned int count=0;
char name[256];
sprintf(name,"Chunk[%d]",count++);
- if (!(task_data=malloc(sizeof(sg_task_data_t))))
+ if (!(task_data=gras_new(sg_task_data_t,1)))
RAISE_MALLOC;
- if (!(task_data->data=malloc(size)))
+ if (!(task_data->data=(void*)gras_malloc(size)))
RAISE_MALLOC;
task_data->size = size;
memcpy(task_data->data,data,size);
task=MSG_task_create(name,0,((double)size)/(1024.0*1024.0),task_data);
+ DEBUG5("send chunk %s from %s to %s:%d (size=%ld)",
+ name, MSG_host_get_name(MSG_host_self()),
+ MSG_host_get_name(sock_data->to_host), sock_data->to_chan,size);
if (MSG_task_put(task, sock_data->to_host,sock_data->to_chan) != MSG_OK) {
RAISE0(system_error,"Problem during the MSG_task_put");
}
gras_error_t gras_trp_sg_chunk_recv(gras_socket_t *sock,
char *data,
- size_t size){
+ long int size){
gras_procdata_t *pd=gras_procdata_get();
m_task_t task=NULL;
sg_task_data_t *task_data;
+ gras_trp_sg_sock_data_t *sock_data = sock->data;
+ GRAS_IN;
+ DEBUG4("recv chunk on %s -> %s:%d (size=%ld)",
+ MSG_host_get_name(sock_data->to_host),
+ MSG_host_get_name(MSG_host_self()), sock_data->to_chan, size);
if (MSG_task_get(&task, (sock->raw ? pd->rawChan : pd->chan)) != MSG_OK)
RAISE0(unknown_error,"Error in MSG_task_get()");
task_data = MSG_task_get_data(task);
- gras_assert2(task_data->size == size,
- "Got %d bytes when %d where expected",
- task_data->size, size);
+ if (task_data->size != size)
+ RAISE5(mismatch_error,
+ "Got %d bytes when %ld where expected (in %s->%s:%d)",
+ task_data->size, size,
+ MSG_host_get_name(sock_data->to_host),
+ MSG_host_get_name(MSG_host_self()), sock_data->to_chan);
memcpy(data,task_data->data,size);
- free(task_data->data);
- free(task_data);
+ gras_free(task_data->data);
+ gras_free(task_data);
if (MSG_task_destroy(task) != MSG_OK)
RAISE0(unknown_error,"Error in MSG_task_destroy()");
+ GRAS_OUT;
return no_error;
}
-
-/*FIXME
-
-gras_error_t gras_trp_sg_flush(gras_socket_t *sd){
- RAISE_UNIMPLEMENTED;
-}
-*/