+/** @brief Opens a client socket to a remote host */
+gras_socket_t gras_socket_client(const char *host, unsigned short port)
+{
+ return gras_socket_client_ext(host, port, 0, 0);
+}
+
+/** @brief Opens a client socket to a remote host specified as '\a host:\a port' */
+gras_socket_t gras_socket_client_from_string(const char *host)
+{
+ xbt_peer_t p = xbt_peer_from_string(host);
+ gras_socket_t res = gras_socket_client_ext(p->name, p->port, 0, 0);
+ xbt_peer_free(p);
+ return res;
+}
+
+void gras_socket_close_voidp(void *sock) {
+ gras_socket_close((gras_socket_t)sock);
+}
+
+/** \brief Close socket */
+void gras_socket_close(gras_socket_t sock)
+{
+ xbt_dynar_t sockets =
+ ((gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id))->sockets;
+ gras_socket_t sock_iter = NULL;
+ unsigned int cursor;
+
+ XBT_IN;
+ VERB1("Close %p", sock);
+ if (sock == _gras_lastly_selected_socket) {
+ xbt_assert0(!gras_opt_trp_nomoredata_on_close || !sock->moredata,
+ "Closing a socket having more data in buffer while the nomoredata_on_close option is activated");
+
+ if (sock->moredata)
+ CRITICAL0
+ ("Closing a socket having more data in buffer. Option nomoredata_on_close disabled, so continuing.");
+ _gras_lastly_selected_socket = NULL;
+ }
+
+ /* FIXME: Issue an event when the socket is closed */
+ DEBUG1("sockets pointer before %p", sockets);
+ if (sock) {
+ /* FIXME: Cannot get the dynar mutex, because it can be already locked */
+// _xbt_dynar_foreach(sockets,cursor,sock_iter) {
+ for (cursor = 0; cursor < xbt_dynar_length(sockets); cursor++) {
+ _xbt_dynar_cursor_get(sockets, cursor, &sock_iter);
+ if (sock == sock_iter) {
+ DEBUG2("remove sock cursor %d dize %lu\n", cursor,
+ xbt_dynar_length(sockets));
+ xbt_dynar_cursor_rm(sockets, &cursor);
+ if (sock->plugin->socket_close)
+ (*sock->plugin->socket_close) (sock);
+
+ /* free the memory */
+ if (sock->peer_name)
+ free(sock->peer_name);
+ free(sock);
+ XBT_OUT;
+ return;
+ }
+ }
+ WARN1("Ignoring request to free an unknown socket (%p). Execution stack:",
+ sock);
+ xbt_backtrace_display_current();
+ }
+ XBT_OUT;
+}
+
+/**
+ * gras_trp_send:
+ *
+ * Send a bunch of bytes from on socket
+ * (stable if we know the storage will keep as is until the next trp_flush)
+ */
+void gras_trp_send(gras_socket_t sd, char *data, long int size, int stable)
+{
+ xbt_assert0(sd->outgoing, "Socket not suited for data send");
+ (*sd->plugin->send) (sd, data, size, stable);
+}
+
+/**
+ * gras_trp_chunk_recv:
+ *
+ * Receive a bunch of bytes from a socket
+ */
+void gras_trp_recv(gras_socket_t sd, char *data, long int size)
+{
+ xbt_assert0(sd->incoming, "Socket not suited for data receive");
+ (sd->plugin->recv) (sd, data, size);
+}
+
+/**
+ * gras_trp_flush:
+ *
+ * Make sure all pending communications are done
+ */
+void gras_trp_flush(gras_socket_t sd)
+{
+ if (sd->plugin->flush)
+ (sd->plugin->flush) (sd);
+}
+
+gras_trp_plugin_t gras_trp_plugin_get_by_name(const char *name)
+{
+ return xbt_dict_get(_gras_trp_plugins, name);
+}
+
+int gras_socket_my_port(gras_socket_t sock)
+{
+ return sock->port;
+}
+
+int gras_socket_peer_port(gras_socket_t sock)
+{
+ return sock->peer_port;
+}
+
+char *gras_socket_peer_name(gras_socket_t sock)
+{
+ return sock->peer_name;
+}
+
+char *gras_socket_peer_proc(gras_socket_t sock)
+{
+ return sock->peer_proc;
+}
+
+void gras_socket_peer_proc_set(gras_socket_t sock, char *peer_proc)
+{
+ sock->peer_proc = peer_proc;
+}
+
+/** \brief Check if the provided socket is a measurement one (or a regular one) */
+int gras_socket_is_meas(gras_socket_t sock)
+{
+ return sock->meas;
+}
+
+/** \brief Send a chunk of (random) data over a measurement socket
+ *
+ * @param peer measurement socket to use for the experiment
+ * @param timeout timeout (in seconds)
+ * @param msg_size size of each chunk sent over the socket (in bytes).
+ * @param msg_amount how many of these packets you want to send.
+ *
+ * Calls to gras_socket_meas_send() and gras_socket_meas_recv() on
+ * each side of the socket should be paired.
+ *
+ * The exchanged data is zeroed to make sure it's initialized, but
+ * there is no way to control what is sent (ie, you cannot use these
+ * functions to exchange data out of band).
+ *
+ * @warning: in SimGrid version 3.1 and previous, the numerical arguments
+ * were the total amount of data to send and the msg_size. This
+ * was changed for the fool wanting to send more than MAXINT
+ * bytes in a fat pipe.
+ */
+void gras_socket_meas_send(gras_socket_t peer,
+ unsigned int timeout,
+ unsigned long int msg_size,
+ unsigned long int msg_amount)
+{
+ char *chunk = NULL;
+ unsigned long int sent_sofar;
+
+ XBT_IN;
+
+ if (gras_if_RL())
+ chunk = xbt_malloc0(msg_size);
+
+ xbt_assert0(peer->meas,
+ "Asked to send measurement data on a regular socket");
+ xbt_assert0(peer->outgoing,
+ "Socket not suited for data send (was created with gras_socket_server(), not gras_socket_client())");
+
+ for (sent_sofar = 0; sent_sofar < msg_amount; sent_sofar++) {
+ CDEBUG5(gras_trp_meas,
+ "Sent %lu msgs of %lu (size of each: %ld) to %s:%d", sent_sofar,
+ msg_amount, msg_size, gras_socket_peer_name(peer),
+ gras_socket_peer_port(peer));
+ (*peer->plugin->raw_send) (peer, chunk, msg_size);
+ }
+ CDEBUG5(gras_trp_meas, "Sent %lu msgs of %lu (size of each: %ld) to %s:%d",
+ sent_sofar, msg_amount, msg_size,
+ gras_socket_peer_name(peer), gras_socket_peer_port(peer));
+
+ if (gras_if_RL())
+ free(chunk);
+
+ XBT_OUT;
+}
+
+/** \brief Receive a chunk of data over a measurement socket
+ *
+ * Calls to gras_socket_meas_send() and gras_socket_meas_recv() on
+ * each side of the socket should be paired.
+ *
+ * @warning: in SimGrid version 3.1 and previous, the numerical arguments
+ * were the total amount of data to send and the msg_size. This
+ * was changed for the fool wanting to send more than MAXINT
+ * bytes in a fat pipe.
+ */
+void gras_socket_meas_recv(gras_socket_t peer,
+ unsigned int timeout,
+ unsigned long int msg_size,
+ unsigned long int msg_amount)
+{
+
+ char *chunk = NULL;
+ unsigned long int got_sofar;
+
+ XBT_IN;
+
+ if (gras_if_RL())
+ chunk = xbt_malloc(msg_size);
+
+ xbt_assert0(peer->meas,
+ "Asked to receive measurement data on a regular socket");
+ xbt_assert0(peer->incoming, "Socket not suited for data receive");
+
+ for (got_sofar = 0; got_sofar < msg_amount; got_sofar++) {
+ CDEBUG5(gras_trp_meas,
+ "Recvd %ld msgs of %lu (size of each: %ld) from %s:%d", got_sofar,
+ msg_amount, msg_size, gras_socket_peer_name(peer),
+ gras_socket_peer_port(peer));
+ (peer->plugin->raw_recv) (peer, chunk, msg_size);
+ }
+ CDEBUG5(gras_trp_meas,
+ "Recvd %ld msgs of %lu (size of each: %ld) from %s:%d", got_sofar,
+ msg_amount, msg_size, gras_socket_peer_name(peer),
+ gras_socket_peer_port(peer));
+
+ if (gras_if_RL())
+ free(chunk);
+ XBT_OUT;
+}
+
+/**
+ * \brief Something similar to the good old accept system call.
+ *
+ * Make sure that there is someone speaking to the provided server socket.
+ * In RL, it does an accept(2) and return the result as last argument.
+ * In SG, as accepts are useless, it returns the provided argument as result.
+ * You should thus test whether (peer != accepted) before closing both of them.
+ *
+ * You should only call this on measurement sockets. It is automatically
+ * done for regular sockets, but you usually want more control about
+ * what's going on with measurement sockets.
+ */
+gras_socket_t gras_socket_meas_accept(gras_socket_t peer)
+{
+ gras_socket_t res;
+
+ xbt_assert0(peer->meas,
+ "No need to accept on non-measurement sockets (it's automatic)");
+
+ if (!peer->accepting) {
+ /* nothing to accept here (must be in SG) */
+ /* BUG: FIXME: this is BAD! it makes tricky to free the accepted socket */
+ return peer;
+ }
+
+ res = (peer->plugin->socket_accept) (peer);
+ res->meas = peer->meas;
+ CDEBUG1(gras_trp_meas, "meas_accepted onto %d", res->sd);
+
+ return res;
+}
+
+
+/*
+ * Creating procdata for this module
+ */
+static void *gras_trp_procdata_new(void)
+{
+ gras_trp_procdata_t res = xbt_new(s_gras_trp_procdata_t, 1);
+
+ res->name = xbt_strdup("gras_trp");
+ res->name_len = 0;
+ res->sockets = xbt_dynar_new_sync(sizeof(gras_socket_t *), NULL);
+ res->myport = 0;
+
+ return (void *) res;
+}
+
+/*
+ * Freeing procdata for this module
+ */
+static void gras_trp_procdata_free(void *data)
+{
+ gras_trp_procdata_t res = (gras_trp_procdata_t) data;
+
+ xbt_dynar_free(&(res->sockets));
+ free(res->name);
+ free(res);
+}
+
+void gras_trp_socketset_dump(const char *name)
+{
+ gras_trp_procdata_t procdata =
+ (gras_trp_procdata_t) gras_libdata_by_id(gras_trp_libdata_id);
+
+ unsigned int it;
+ gras_socket_t s;
+
+ INFO1("** Dump the socket set %s", name);
+ xbt_dynar_foreach(procdata->sockets, it, s) {
+ INFO4(" %p -> %s:%d %s",
+ s, gras_socket_peer_name(s), gras_socket_peer_port(s),
+ s->valid ? "(valid)" : "(peer dead)");
+ }
+ INFO1("** End of socket set %s", name);
+}
+
+/*
+ * Module registration
+ */
+int gras_trp_libdata_id;
+void gras_trp_register()
+{
+ gras_trp_libdata_id =
+ gras_procdata_add("gras_trp", gras_trp_procdata_new,
+ gras_trp_procdata_free);
+}