X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/0f5e8daaa6e9f74521068aa75837200bcd182ea6..6a6157abf9619875297ff1d16c1e5c63c526d4c1:/src/gras/Transport/transport_plugin_tcp.c diff --git a/src/gras/Transport/transport_plugin_tcp.c b/src/gras/Transport/transport_plugin_tcp.c index ce5923d26e..0835ece385 100644 --- a/src/gras/Transport/transport_plugin_tcp.c +++ b/src/gras/Transport/transport_plugin_tcp.c @@ -34,6 +34,13 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gras_trp_tcp, gras_trp, *** Specific socket part ***/ +typedef struct { + int port; /* port on this side */ + int peer_port; /* port on the other side */ + char *peer_name; /* hostname of the other side */ + char *peer_proc; /* process on the other side */ +} s_gras_trp_tcp_sock_data_t, *gras_trp_tcp_sock_data_t; + typedef enum { buffering_buf, buffering_iov } buffering_kind; typedef struct { @@ -72,10 +79,18 @@ static int gras_trp_tcp_recv(gras_socket_t sock, char *data, static int _gras_tcp_proto_number(void); -static XBT_INLINE void gras_trp_sock_socket_client(gras_trp_plugin_t - ignored, - gras_socket_t sock) +static XBT_INLINE +void gras_trp_sock_socket_client(gras_trp_plugin_t ignored, + const char *host, + int port, + /*OUT*/gras_socket_t sock) { + gras_trp_tcp_sock_data_t sockdata = xbt_new(s_gras_trp_tcp_sock_data_t,1); + sockdata->port = port; + sockdata->peer_proc = NULL; + sockdata->peer_port = port; + sockdata->peer_name = (char *) strdup(host ? host : "localhost"); + sock->data = sockdata; struct sockaddr_in addr; struct hostent *he; @@ -98,14 +113,14 @@ static XBT_INLINE void gras_trp_sock_socket_client(gras_trp_plugin_t (sock->sd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size)) || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, (char *) &size, sizeof(size))) { - VERB1("setsockopt failed, cannot set buffer size: %s", + XBT_VERB("setsockopt failed, cannot set buffer size: %s", sock_errstr(sock_errno)); } - he = gethostbyname(sock->peer_name); + he = gethostbyname(sockdata->peer_name); if (he == NULL) { THROW2(system_error, 0, "Failed to lookup hostname %s: %s", - sock->peer_name, sock_errstr(sock_errno)); + sockdata->peer_name, sock_errstr(sock_errno)); } haddr = ((struct in_addr *) (he->h_addr_list)[0]); @@ -113,20 +128,20 @@ static XBT_INLINE void gras_trp_sock_socket_client(gras_trp_plugin_t memset(&addr, 0, sizeof(struct sockaddr_in)); memcpy(&addr.sin_addr, haddr, sizeof(struct in_addr)); addr.sin_family = AF_INET; - addr.sin_port = htons(sock->peer_port); + addr.sin_port = htons(sockdata->peer_port); if (connect(sock->sd, (struct sockaddr *) &addr, sizeof(addr)) < 0) { tcp_close(sock->sd); THROW3(system_error, 0, "Failed to connect socket to %s:%d (%s)", - sock->peer_name, sock->peer_port, sock_errstr(sock_errno)); + sockdata->peer_name, sockdata->peer_port, sock_errstr(sock_errno)); } gras_trp_tcp_send(sock, (char *) &myport, sizeof(uint32_t)); - DEBUG1("peerport sent to %d", sock->peer_port); + XBT_DEBUG("peerport sent to %d", sockdata->peer_port); - VERB4("Connect to %s:%d (sd=%d, port %d here)", - sock->peer_name, sock->peer_port, sock->sd, sock->port); + XBT_VERB("Connect to %s:%d (sd=%d, port %d here)", + sockdata->peer_name, sockdata->peer_port, sock->sd, sockdata->port); } /** @@ -134,17 +149,25 @@ static XBT_INLINE void gras_trp_sock_socket_client(gras_trp_plugin_t * * Open a socket used to receive messages. */ -static XBT_INLINE void gras_trp_sock_socket_server(gras_trp_plugin_t - ignored, - gras_socket_t sock) +static XBT_INLINE +void gras_trp_sock_socket_server(gras_trp_plugin_t ignored, + int port, + gras_socket_t sock) { int size = sock->buf_size; int on = 1; struct sockaddr_in server; + gras_trp_tcp_sock_data_t sockdata = xbt_new(s_gras_trp_tcp_sock_data_t,1); + sockdata->port = port; + sockdata->peer_port = -1; + sockdata->peer_name = NULL; + sockdata->peer_proc = NULL; + sock->data=sockdata; + sock->outgoing = 1; /* TCP => duplex mode */ - server.sin_port = htons((u_short) sock->port); + server.sin_port = htons((u_short) sockdata->port); server.sin_addr.s_addr = INADDR_ANY; server.sin_family = AF_INET; if ((sock->sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) @@ -161,26 +184,26 @@ static XBT_INLINE void gras_trp_sock_socket_server(gras_trp_plugin_t (char *) &size, sizeof(size)) || setsockopt(sock->sd, SOL_SOCKET, SO_SNDBUF, (char *) &size, sizeof(size))) { - VERB1("setsockopt failed, cannot set buffer size: %s", + XBT_VERB("setsockopt failed, cannot set buffer size: %s", sock_errstr(sock_errno)); } if (bind(sock->sd, (struct sockaddr *) &server, sizeof(server)) == -1) { tcp_close(sock->sd); THROW2(system_error, 0, - "Cannot bind to port %d: %s", sock->port, + "Cannot bind to port %d: %s", sockdata->port, sock_errstr(sock_errno)); } - DEBUG2("Listen on port %d (sd=%d)", sock->port, sock->sd); + XBT_DEBUG("Listen on port %d (sd=%d)", sockdata->port, sock->sd); if (listen(sock->sd, 5) < 0) { tcp_close(sock->sd); THROW2(system_error, 0, "Cannot listen on port %d: %s", - sock->port, sock_errstr(sock_errno)); + sockdata->port, sock_errstr(sock_errno)); } - VERB2("Openned a server socket on port %d (sd=%d)", sock->port, + XBT_VERB("Openned a server socket on port %d (sd=%d)", sockdata->port, sock->sd); } @@ -200,7 +223,7 @@ static gras_socket_t gras_trp_sock_socket_accept(gras_socket_t sock) uint32_t hisport; - XBT_IN; + XBT_IN(""); gras_trp_socket_new(1, &res); sd = accept(sock->sd, (struct sockaddr *) &peer_in, &peer_in_len); @@ -225,7 +248,7 @@ static gras_socket_t gras_trp_sock_socket_accept(gras_socket_t sock) if (setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size)) || setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *) &size, sizeof(size))) - VERB1("setsockopt failed, cannot set buffer size: %s", + XBT_VERB("setsockopt failed, cannot set buffer size: %s", sock_errstr(tmp_errno)); res->plugin = sock->plugin; @@ -233,15 +256,18 @@ static gras_socket_t gras_trp_sock_socket_accept(gras_socket_t sock) res->outgoing = sock->outgoing; res->accepting = 0; res->sd = sd; - res->port = -1; + gras_trp_tcp_sock_data_t sockdata = xbt_new(s_gras_trp_tcp_sock_data_t,1); + sockdata->port = -1; + res->data=sockdata; + gras_trp_tcp_recv(res, (char *) &hisport, sizeof(hisport)); - res->peer_port = ntohl(hisport); - DEBUG1("peerport %d received", res->peer_port); + sockdata->peer_port = ntohl(hisport); + XBT_DEBUG("peerport %d received", sockdata->peer_port); /* FIXME: Lock to protect inet_ntoa */ if (((struct sockaddr *) &peer_in)->sa_family != AF_INET) { - res->peer_name = (char *) strdup("unknown"); + sockdata->peer_name = (char *) strdup("unknown"); } else { struct in_addr addrAsInAddr; char *tmp; @@ -250,17 +276,17 @@ static gras_socket_t gras_trp_sock_socket_accept(gras_socket_t sock) tmp = inet_ntoa(addrAsInAddr); if (tmp != NULL) { - res->peer_name = (char *) strdup(tmp); + sockdata->peer_name = (char *) strdup(tmp); } else { - res->peer_name = (char *) strdup("unknown"); + sockdata->peer_name = (char *) strdup("unknown"); } } - VERB3("Accepted from %s:%d (sd=%d)", res->peer_name, res->peer_port, sd); + XBT_VERB("Accepted from %s:%d (sd=%d)", sockdata->peer_name, sockdata->peer_port, sd); xbt_dynar_push(((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets, &res); - XBT_OUT; + XBT_OUT(); return res; } @@ -270,7 +296,11 @@ static void gras_trp_sock_socket_close(gras_socket_t sock) if (!sock) return; /* close only once */ - VERB1("close tcp connection %d", sock->sd); + if (((gras_trp_tcp_sock_data_t)sock->data)->peer_name) + free(((gras_trp_tcp_sock_data_t)sock->data)->peer_name); + free(sock->data); + + XBT_VERB("close tcp connection %d", sock->sd); /* ask the listener to close the socket */ gras_msg_listener_close_socket(sock->sd); @@ -295,7 +325,7 @@ static XBT_INLINE void gras_trp_tcp_send(gras_socket_t sock, int status = 0; status = tcp_write(sock->sd, data, (size_t) size); - DEBUG3("write(%d, %p, %ld);", sock->sd, data, size); + XBT_DEBUG("write(%d, %p, %ld);", sock->sd, data, size); if (status < 0) { #ifdef EWOULDBLOCK @@ -338,7 +368,7 @@ gras_trp_tcp_recv_withbuffer(gras_socket_t sock, while (size > got) { int status = 0; - DEBUG5("read(%d, %p, %ld) got %d so far (%s)", + XBT_DEBUG("read(%d, %p, %ld) got %d so far (%s)", sock->sd, data + got, bufsize, got, hexa_str((unsigned char *) data, got, 0)); status = tcp_read(sock->sd, data + got, (size_t) bufsize); @@ -349,7 +379,7 @@ gras_trp_tcp_recv_withbuffer(gras_socket_t sock, sock->sd, data + got, (int) size, gras_socket_peer_name(sock), gras_socket_peer_port(sock), sock_errstr(sock_errno), got); } - DEBUG2("Got %d more bytes (%s)", status, + XBT_DEBUG("Got %d more bytes (%s)", status, hexa_str((unsigned char *) data + got, status, 0)); if (status) { @@ -388,27 +418,27 @@ static void gras_trp_bufiov_flush(gras_socket_t sock) int size; #endif gras_trp_bufdata_t *data = sock->bufdata; - XBT_IN; + XBT_IN(""); - DEBUG0("Flush"); + XBT_DEBUG("Flush"); if (data->out == buffering_buf) { if (XBT_LOG_ISENABLED(gras_trp_tcp, xbt_log_priority_debug)) hexa_print("chunk to send ", (unsigned char *) data->out_buf.data, data->out_buf.size); if ((data->out_buf.size - data->out_buf.pos) != 0) { - DEBUG3("Send the chunk (size=%d) to %s:%d", data->out_buf.size, + XBT_DEBUG("Send the chunk (size=%d) to %s:%d", data->out_buf.size, gras_socket_peer_name(sock), gras_socket_peer_port(sock)); gras_trp_tcp_send(sock, data->out_buf.data, data->out_buf.size); - VERB1("Chunk sent (size=%d)", data->out_buf.size); + XBT_VERB("Chunk sent (size=%d)", data->out_buf.size); data->out_buf.size = 0; } } #ifdef HAVE_READV if (data->out == buffering_iov) { - DEBUG0("Flush out iov"); + XBT_DEBUG("Flush out iov"); vect = sock->bufdata->out_buf_v; if ((size = xbt_dynar_length(vect))) { - DEBUG1("Flush %d chunks out of this socket", size); + XBT_DEBUG("Flush %d chunks out of this socket", size); writev(sock->sd, xbt_dynar_get_ptr(vect, 0), size); xbt_dynar_reset(vect); } @@ -416,10 +446,10 @@ static void gras_trp_bufiov_flush(gras_socket_t sock) } if (data->in == buffering_iov) { - DEBUG0("Flush in iov"); + XBT_DEBUG("Flush in iov"); vect = sock->bufdata->in_buf_v; if ((size = xbt_dynar_length(vect))) { - DEBUG1("Get %d chunks from of this socket", size); + XBT_DEBUG("Get %d chunks from of this socket", size); readv(sock->sd, xbt_dynar_get_ptr(vect, 0), size); xbt_dynar_reset(vect); } @@ -436,13 +466,13 @@ gras_trp_buf_send(gras_socket_t sock, gras_trp_bufdata_t *data = (gras_trp_bufdata_t *) sock->bufdata; int chunk_pos = 0; - XBT_IN; + XBT_IN(""); while (chunk_pos < size) { /* size of the chunk to receive in that shot */ long int thissize = min(size - chunk_pos, data->buffsize - data->out_buf.size); - DEBUG4("Set the chars %d..%ld into the buffer; size=%ld, ctn=(%s)", + XBT_DEBUG("Set the chars %d..%ld into the buffer; size=%ld, ctn=(%s)", (int) data->out_buf.size, ((int) data->out_buf.size) + thissize - 1, size, hexa_str((unsigned char *) chunk, thissize, 0)); @@ -452,7 +482,7 @@ gras_trp_buf_send(gras_socket_t sock, data->out_buf.size += thissize; chunk_pos += thissize; - DEBUG4("New pos = %d; Still to send = %ld of %ld; ctn sofar=(%s)", + XBT_DEBUG("New pos = %d; Still to send = %ld of %ld; ctn sofar=(%s)", data->out_buf.size, size - chunk_pos, size, hexa_str((unsigned char *) chunk, chunk_pos, 0)); @@ -460,7 +490,7 @@ gras_trp_buf_send(gras_socket_t sock, gras_trp_bufiov_flush(sock); } - XBT_OUT; + XBT_OUT(); } static int @@ -470,7 +500,7 @@ gras_trp_buf_recv(gras_socket_t sock, char *chunk, unsigned long int size) gras_trp_bufdata_t *data = sock->bufdata; long int chunk_pos = 0; - XBT_IN; + XBT_IN(""); while (chunk_pos < size) { /* size of the chunk to receive in that shot */ @@ -478,7 +508,7 @@ gras_trp_buf_recv(gras_socket_t sock, char *chunk, unsigned long int size) if (data->in_buf.size == data->in_buf.pos) { /* out of data. Get more */ - DEBUG2("Get more data (size=%d,bufsize=%d)", + XBT_DEBUG("Get more data (size=%d,bufsize=%d)", (int) MIN(size - chunk_pos, data->buffsize), (int) data->buffsize); @@ -498,15 +528,15 @@ gras_trp_buf_recv(gras_socket_t sock, char *chunk, unsigned long int size) data->in_buf.pos += thissize; chunk_pos += thissize; - DEBUG4("New pos = %d; Still to receive = %ld of %ld. Ctn so far=(%s)", + XBT_DEBUG("New pos = %d; Still to receive = %ld of %ld. Ctn so far=(%s)", data->in_buf.pos, size - chunk_pos, size, hexa_str((unsigned char *) chunk, chunk_pos, 0)); } /* indicate on need to the gras_select function that there is more to read on this socket so that it does not actually select */ sock->moredata = (data->in_buf.size > data->in_buf.pos); - DEBUG1("There is %smore data", (sock->moredata ? "" : "no ")); + XBT_DEBUG("There is %smore data", (sock->moredata ? "" : "no ")); - XBT_OUT; + XBT_OUT(); return chunk_pos; } @@ -526,7 +556,7 @@ gras_trp_iov_send(gras_socket_t sock, gras_trp_bufdata_t *data = (gras_trp_bufdata_t *) sock->bufdata; - DEBUG1("Buffer one chunk to be sent later (%s)", + XBT_DEBUG("Buffer one chunk to be sent later (%s)", hexa_str((char *) chunk, size, 0)); elm.iov_len = (size_t) size; @@ -565,7 +595,7 @@ gras_trp_iov_recv(gras_socket_t sock, char *chunk, unsigned long int size) { struct iovec elm; - DEBUG0("Buffer one chunk to be received later"); + XBT_DEBUG("Buffer one chunk to be received later"); elm.iov_base = (void *) chunk; elm.iov_len = (size_t) size; xbt_dynar_push(sock->bufdata->in_buf_v, &elm); @@ -584,8 +614,11 @@ gras_trp_iov_recv(gras_socket_t sock, char *chunk, unsigned long int size) ***/ void gras_trp_buf_socket_client(gras_trp_plugin_t self, + const char *host, + int port, gras_socket_t sock); void gras_trp_buf_socket_server(gras_trp_plugin_t self, + int port, gras_socket_t sock); gras_socket_t gras_trp_buf_socket_accept(gras_socket_t sock); @@ -621,12 +654,43 @@ gras_socket_t gras_trp_buf_init_sock(gras_socket_t sock) return sock; } +/*** + *** Info about who's speaking + ***/ +static int gras_trp_tcp_my_port(gras_socket_t s) { + gras_trp_tcp_sock_data_t sockdata = s->data; + return sockdata->port; +} +static int gras_trp_tcp_peer_port(gras_socket_t s) { + gras_trp_tcp_sock_data_t sockdata = s->data; + return sockdata->peer_port; +} +static const char* gras_trp_tcp_peer_name(gras_socket_t s) { + gras_trp_tcp_sock_data_t sockdata = s->data; + return sockdata->peer_name; +} +static const char* gras_trp_tcp_peer_proc(gras_socket_t s) { + gras_trp_tcp_sock_data_t sockdata = s->data; + return sockdata->peer_proc; +} +static void gras_trp_tcp_peer_proc_set(gras_socket_t s,char *name) { + gras_trp_tcp_sock_data_t sockdata = s->data; + sockdata->peer_proc = xbt_strdup(name); +} + /*** *** Code ***/ void gras_trp_tcp_setup(gras_trp_plugin_t plug) { + plug->my_port = gras_trp_tcp_my_port; + plug->peer_port = gras_trp_tcp_peer_port; + plug->peer_name = gras_trp_tcp_peer_name; + plug->peer_proc = gras_trp_tcp_peer_proc; + plug->peer_proc_set = gras_trp_tcp_peer_proc_set; + + plug->socket_client = gras_trp_buf_socket_client; plug->socket_server = gras_trp_buf_socket_server; plug->socket_accept = gras_trp_buf_socket_accept; @@ -649,10 +713,12 @@ void gras_trp_tcp_setup(gras_trp_plugin_t plug) } void gras_trp_buf_socket_client(gras_trp_plugin_t self, + const char *host, + int port, /* OUT */ gras_socket_t sock) { - gras_trp_sock_socket_client(NULL, sock); + gras_trp_sock_socket_client(NULL, host,port,sock); gras_trp_buf_init_sock(sock); } @@ -662,10 +728,11 @@ void gras_trp_buf_socket_client(gras_trp_plugin_t self, * Open a socket used to receive messages. */ void gras_trp_buf_socket_server(gras_trp_plugin_t self, + int port, /* OUT */ gras_socket_t sock) { - gras_trp_sock_socket_server(NULL, sock); + gras_trp_sock_socket_server(NULL, port, sock); gras_trp_buf_init_sock(sock); } @@ -679,7 +746,7 @@ void gras_trp_buf_socket_close(gras_socket_t sock) gras_trp_bufdata_t *data = sock->bufdata; if (data->in_buf.size != data->in_buf.pos) { - WARN3("Socket closed, but %d bytes were unread (size=%d,pos=%d)", + XBT_WARN("Socket closed, but %d bytes were unread (size=%d,pos=%d)", data->in_buf.size - data->in_buf.pos, data->in_buf.size, data->in_buf.pos); } @@ -687,7 +754,7 @@ void gras_trp_buf_socket_close(gras_socket_t sock) free(data->in_buf.data); if (data->out_buf.size != data->out_buf.pos) { - DEBUG2("Flush the socket before closing (in=%d,out=%d)", + XBT_DEBUG("Flush the socket before closing (in=%d,out=%d)", data->in_buf.size, data->out_buf.size); gras_trp_bufiov_flush(sock); } @@ -697,12 +764,12 @@ void gras_trp_buf_socket_close(gras_socket_t sock) #ifdef HAVE_READV if (data->in_buf_v) { if (xbt_dynar_length(data->in_buf_v)) - WARN0("Socket closed, but some bytes were unread"); + XBT_WARN("Socket closed, but some bytes were unread"); xbt_dynar_free(&data->in_buf_v); } if (data->out_buf_v) { if (xbt_dynar_length(data->out_buf_v)) { - DEBUG0("Flush the socket before closing"); + XBT_DEBUG("Flush the socket before closing"); gras_trp_bufiov_flush(sock); } xbt_dynar_free(&data->out_buf_v);