gras_error_t gras_trp_sg_chunk_send(gras_socket_t *sd,
char *data,
- size_t size);
+ long int size);
gras_error_t gras_trp_sg_chunk_recv(gras_socket_t *sd,
char *data,
- size_t size);
+ long int size);
/* FIXME
gras_error_t gras_trp_sg_flush(gras_socket_t *sd);
int placeholder; /* nothing plugin specific so far */
} gras_trp_sg_plug_data_t;
-/***
- *** Specific socket part
- ***/
-typedef struct {
- int from_PID; /* process which sent this message */
- int to_PID; /* process to which this message is destinated */
-
- m_host_t to_host; /* Who's on other side */
- m_channel_t to_chan;/* Channel on which the other side is earing */
-} gras_trp_sg_sock_data_t;
-
-
/***
*** Code
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;
}
}
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(),
pr.raw = sock->raw;
TRY(gras_dynar_push(hd->ports,&pr));
- if (sock->raw) {
- if (pd->rawSock)
- WARN1("asked to open two raw server sockets on %s, first one lost",
- MSG_host_get_name(MSG_host_self()));
- pd->rawSock = sock;
- } else {
- if (pd->sock)
- WARN1("asked to open two server sockets on %s, first one lost",
- MSG_host_get_name(MSG_host_self()));
- pd->sock = sock;
- }
-
default:
return errcode;
}
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());
- gras_procdata_t *pd=gras_procdata_get();
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)
+ 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 %d", cpt, gras_dynar_length(hd->ports));
+ if (pr.port == sock->port) {
gras_dynar_cursor_rm(hd->ports, &cpt);
-
- if (sock->raw) {
- pd->rawSock = NULL;
- } else {
- pd->sock = NULL;
- }
return;
}
}
gras_error_t gras_trp_sg_chunk_send(gras_socket_t *sock,
char *data,
- size_t size) {
+ long int size) {
m_task_t task=NULL;
static unsigned int count=0;
char name[256];
task=MSG_task_create(name,0,((double)size)/(1024.0*1024.0),task_data);
+ DEBUG4("send chunk %s from %s to %s on channel %d",
+ name, MSG_host_get_name(MSG_host_self()),
+ MSG_host_get_name(sock_data->to_host), sock_data->to_chan);
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;
+ DEBUG3("recv chunk on %s from %s on channel %d",
+ MSG_host_get_name(MSG_host_self()),
+ MSG_host_get_name(sock_data->to_host), sock_data->to_chan);
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",
+ "Got %d bytes when %ld where expected",
task_data->size, size);
memcpy(data,task_data->data,size);
free(task_data->data);