X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=libustcomm%2Fustcomm.c;h=5dfd2a8c946183649dd529f67057c3e3ef74c063;hb=dff06a58d6f9b7b6bf9cbebe455b7403096eb78d;hp=60428df0553082bfaa35130b134008743421a202;hpb=2a79ceebcba229c91c93272705444c61c458bb06;p=ust.git

diff --git a/libustcomm/ustcomm.c b/libustcomm/ustcomm.c
index 60428df..5dfd2a8 100644
--- a/libustcomm/ustcomm.c
+++ b/libustcomm/ustcomm.c
@@ -33,25 +33,10 @@
 #include "ustcomm.h"
 #include "usterr.h"
 #include "share.h"
+#include "multipoll.h"
 
 #define UNIX_PATH_MAX 108
 
-#define MSG_MAX 10000
-
-/* FIXME: ustcomm blocks on message sending, which might be problematic in
- * some cases. Fix the poll() usage so sends are buffered until they don't
- * block.
- */
-
-//static void bt(void)
-//{
-//	void *buffer[100];
-//	int result;
-//
-//	result = backtrace(&buffer, 100);
-//	backtrace_symbols_fd(buffer, result, STDERR_FILENO);
-//}
-
 static int mkdir_p(const char *path, mode_t mode)
 {
 	const char *path_p;
@@ -421,11 +406,11 @@ int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustco
 
 				if(retval == 0) {
 					/* connection finished */
-					close(fds[idx].fd);
-
 					list_for_each_entry(conn, &server->connections, list) {
 						if(conn->fd == fds[idx].fd) {
+							ustcomm_close_app(conn);
 							list_del(&conn->list);
+							free(conn);
 							break;
 						}
 					}
@@ -503,7 +488,7 @@ static int init_named_socket(const char *name, char **path_out)
 			PERROR("unlink of socket file");
 			goto close_sock;
 		}
-		WARN("socket already exists; overwriting");
+		DBG("socket already exists; overwriting");
 	}
 
 	result = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
@@ -636,7 +621,10 @@ int ustcomm_connect_app(pid_t pid, struct ustcomm_connection *conn)
 	return ustcomm_connect_path(path, conn, pid);
 }
 
-/* Close a connection to a traceable app. */
+/* Close a connection to a traceable app. It frees the
+ * resources. It however does not free the
+ * ustcomm_connection itself.
+ */
 
 int ustcomm_close_app(struct ustcomm_connection *conn)
 {
@@ -784,11 +772,15 @@ static void ustcomm_fini_server(struct ustcomm_server *server, int keep_socket_f
 	}
 }
 
+/* Free a traceable application server */
+
 void ustcomm_fini_app(struct ustcomm_app *handle, int keep_socket_file)
 {
 	ustcomm_fini_server(&handle->server, keep_socket_file);
 }
 
+/* Free a ustd server */
+
 void ustcomm_fini_ustd(struct ustcomm_ustd *handle)
 {
 	ustcomm_fini_server(&handle->server, 0);
@@ -878,3 +870,88 @@ char *nth_token(const char *str, int tok_no)
 	return retval;
 }
 
+/* Callback from multipoll.
+ * Receive a new connection on the listening socket.
+ */
+
+static int process_mp_incoming_conn(void *priv, int fd, short events)
+{
+	struct ustcomm_connection *newconn;
+	struct ustcomm_server *server = (struct ustcomm_server *) priv;
+	int newfd;
+	int result;
+
+	result = newfd = accept(server->listen_fd, NULL, NULL);
+	if(result == -1) {
+		PERROR("accept");
+		return -1;
+	}
+
+	newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
+	if(newconn == NULL) {
+		ERR("malloc returned NULL");
+		return -1;
+	}
+
+	ustcomm_init_connection(newconn);
+	newconn->fd = newfd;
+
+	list_add(&newconn->list, &server->connections);
+
+	return 0;
+}
+
+/* Callback from multipoll.
+ * Receive a message on an existing connection.
+ */
+
+static int process_mp_conn_msg(void *priv, int fd, short revents)
+{
+	struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) priv;
+	int result;
+	char *msg;
+	struct ustcomm_source src;
+
+	if(revents) {
+		src.fd = fd;
+
+		result = recv_message_conn(mpinfo->conn, &msg);
+		if(result == -1) {
+			ERR("error in recv_message_conn");
+		}
+
+		else if(result == 0) {
+			/* connection finished */
+			ustcomm_close_app(mpinfo->conn);
+			list_del(&mpinfo->conn->list);
+			free(mpinfo->conn);
+		}
+		else {
+			mpinfo->cb(msg, &src);
+			free(msg);
+		}
+	}
+
+	return 0;
+}
+
+int free_ustcomm_client_poll(void *data)
+{
+	free(data);
+	return 0;
+}
+
+void ustcomm_mp_add_app_clients(struct mpentries *ent, struct ustcomm_app *app, int (*cb)(struct ustcomm_connection *conn, char *msg))
+{
+	struct ustcomm_connection *conn;
+
+	/* add listener socket */
+	multipoll_add(ent, app->server.listen_fd, POLLIN, process_mp_incoming_conn, &app->server, NULL);
+
+	list_for_each_entry(conn, &app->server.connections, list) {
+		struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) malloc(sizeof(struct ustcomm_multipoll_conn_info));
+		mpinfo->conn = conn;
+		mpinfo->cb = cb;
+		multipoll_add(ent, conn->fd, POLLIN, process_mp_conn_msg, mpinfo, free_ustcomm_client_poll);
+	}
+}