ust: first try at blocking support for consumer
[lttng-ust.git] / libustcomm / ustcomm.c
1 #define _GNU_SOURCE
2 #include <sys/types.h>
3 #include <signal.h>
4 #include <errno.h>
5 #include <sys/socket.h>
6 #include <sys/un.h>
7 #include <unistd.h>
8 #include <poll.h>
9
10 #include <stdio.h>
11 #include <stdlib.h>
12 #include <string.h>
13 #include <execinfo.h>
14
15 #include "ustcomm.h"
16 #include "localerr.h"
17
18 #define UNIX_PATH_MAX 108
19 #define SOCK_DIR "/tmp/socks"
20 #define UST_SIGNAL SIGIO
21
22 #define MSG_MAX 1000
23
24 /* FIXME: ustcomm blocks on message sending, which might be problematic in
25 * some cases. Fix the poll() usage so sends are buffered until they don't
26 * block.
27 */
28
29 //static void bt(void)
30 //{
31 // void *buffer[100];
32 // int result;
33 //
34 // result = backtrace(&buffer, 100);
35 // backtrace_symbols_fd(buffer, result, STDERR_FILENO);
36 //}
37
38 char *strdup_malloc(const char *s)
39 {
40 char *retval;
41
42 if(s == NULL)
43 return NULL;
44
45 retval = (char *) malloc(strlen(s)+1);
46
47 strcpy(retval, s);
48
49 return retval;
50 }
51
52 static void signal_process(pid_t pid)
53 {
54 int result;
55
56 result = kill(pid, UST_SIGNAL);
57 if(result == -1) {
58 PERROR("kill");
59 return;
60 }
61
62 sleep(1);
63 }
64
65 int send_message_fd(int fd, const char *msg, char **reply)
66 {
67 int result;
68
69 result = send(fd, msg, strlen(msg), 0);
70 if(result == -1) {
71 PERROR("send");
72 return -1;
73 }
74 else if(result == 0) {
75 return 0;
76 }
77
78 if(!reply)
79 return 1;
80
81 *reply = (char *) malloc(MSG_MAX+1);
82 result = recv(fd, *reply, MSG_MAX, 0);
83 if(result == -1) {
84 PERROR("recv");
85 return -1;
86 }
87 else if(result == 0) {
88 return 0;
89 }
90
91 (*reply)[result] = '\0';
92
93 return 1;
94 }
95
96 int send_message_path(const char *path, const char *msg, char **reply, int signalpid)
97 {
98 int fd;
99 int result;
100 struct sockaddr_un addr;
101
102 result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
103 if(result == -1) {
104 PERROR("socket");
105 return -1;
106 }
107
108 addr.sun_family = AF_UNIX;
109
110 result = snprintf(addr.sun_path, UNIX_PATH_MAX, "%s", path);
111 if(result >= UNIX_PATH_MAX) {
112 ERR("string overflow allocating socket name");
113 return -1;
114 }
115
116 if(signalpid >= 0)
117 signal_process(signalpid);
118
119 result = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
120 if(result == -1) {
121 PERROR("connect");
122 return -1;
123 }
124
125 return send_message_fd(fd, msg, reply);
126 }
127
128 /* pid: the pid of the trace process that must receive the msg
129 msg: pointer to a null-terminated message to send
130 reply: location where to put the null-terminated string of the reply;
131 it must be free'd after usage
132 */
133
134 int send_message(pid_t pid, const char *msg, char **reply)
135 {
136 int result;
137 char path[UNIX_PATH_MAX];
138
139 result = snprintf(path, UNIX_PATH_MAX, "%s/%d", SOCK_DIR, pid);
140 if(result >= UNIX_PATH_MAX) {
141 fprintf(stderr, "string overflow allocating socket name");
142 return -1;
143 }
144
145 send_message_path(path, msg, reply, pid);
146
147 return 0;
148 }
149
150 /* Called by an app to ask the consumer daemon to connect to it. */
151
152 int ustcomm_request_consumer(pid_t pid, const char *channel)
153 {
154 char path[UNIX_PATH_MAX];
155 int result;
156 char *msg;
157
158 result = snprintf(path, UNIX_PATH_MAX, "%s/ustd", SOCK_DIR);
159 if(result >= UNIX_PATH_MAX) {
160 fprintf(stderr, "string overflow allocating socket name");
161 return -1;
162 }
163
164 asprintf(&msg, "collect %d %s", pid, channel);
165
166 send_message_path(path, msg, NULL, -1);
167 free(msg);
168
169 return 0;
170 }
171
172 /* returns 1 to indicate a message was received
173 * returns 0 to indicate no message was received (cannot happen)
174 * returns -1 to indicate an error
175 */
176
177 static int recv_message_fd(int fd, char **msg, struct ustcomm_source *src)
178 {
179 int result;
180
181 *msg = (char *) malloc(MSG_MAX+1);
182
183 result = recv(fd, *msg, MSG_MAX, 0);
184 if(result == -1) {
185 PERROR("recv");
186 return -1;
187 }
188
189 (*msg)[result] = '\0';
190
191 DBG("ustcomm_app_recv_message: result is %d, message is %s", result, (*msg));
192
193 if(src)
194 src->fd = fd;
195
196 return 1;
197 }
198
199 int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src)
200 {
201 int result;
202
203 result = send_message_fd(src->fd, msg, NULL);
204 if(result) {
205 ERR("error in send_message_fd");
206 return -1;
207 }
208
209 return 0;
210 }
211
212 /* @timeout: max blocking time in milliseconds, -1 means infinity
213 *
214 * returns 1 to indicate a message was received
215 * returns 0 to indicate no message was received
216 * returns -1 to indicate an error
217 */
218
219 int ustcomm_recv_message(struct ustcomm_server *server, char **msg, struct ustcomm_source *src, int timeout)
220 {
221 struct pollfd *fds;
222 struct ustcomm_connection *conn;
223 int result;
224 int retval;
225
226 for(;;) {
227 int idx = 0;
228 int n_fds = 1;
229
230 list_for_each_entry(conn, &server->connections, list) {
231 n_fds++;
232 }
233
234 fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd));
235 if(fds == NULL) {
236 ERR("malloc returned NULL");
237 return -1;
238 }
239
240 /* special idx 0 is for listening socket */
241 fds[idx].fd = server->listen_fd;
242 fds[idx].events = POLLIN;
243 idx++;
244
245 list_for_each_entry(conn, &server->connections, list) {
246 fds[idx].fd = conn->fd;
247 fds[idx].events = POLLIN;
248 idx++;
249 }
250
251 result = poll(fds, n_fds, timeout);
252 if(result == -1) {
253 PERROR("poll");
254 return -1;
255 }
256
257 if(result == 0)
258 return 0;
259
260 if(fds[0].revents) {
261 struct ustcomm_connection *newconn;
262 int newfd;
263
264 result = newfd = accept(server->listen_fd, NULL, NULL);
265 if(result == -1) {
266 PERROR("accept");
267 return -1;
268 }
269
270 newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
271 if(newconn == NULL) {
272 ERR("malloc returned NULL");
273 return -1;
274 }
275
276 newconn->fd = newfd;
277
278 list_add(&newconn->list, &server->connections);
279 }
280
281 for(idx=1; idx<n_fds; idx++) {
282 if(fds[idx].revents) {
283 retval = recv_message_fd(fds[idx].fd, msg, src);
284 if(**msg == 0) {
285 /* connection finished */
286 close(fds[idx].fd);
287
288 list_for_each_entry(conn, &server->connections, list) {
289 if(conn->fd == fds[idx].fd) {
290 list_del(&conn->list);
291 break;
292 }
293 }
294 }
295 else {
296 goto free_fds_return;
297 }
298 }
299 }
300
301 free(fds);
302 }
303
304 free_fds_return:
305 free(fds);
306 return retval;
307 }
308
309 int ustcomm_ustd_recv_message(struct ustcomm_ustd *ustd, char **msg, struct ustcomm_source *src, int timeout)
310 {
311 return ustcomm_recv_message(&ustd->server, msg, src, timeout);
312 }
313
314 int ustcomm_app_recv_message(struct ustcomm_app *app, char **msg, struct ustcomm_source *src, int timeout)
315 {
316 return ustcomm_recv_message(&app->server, msg, src, timeout);
317 }
318
319 /* This removes src from the list of active connections of app.
320 */
321
322 int ustcomm_app_detach_client(struct ustcomm_app *app, struct ustcomm_source *src)
323 {
324 struct ustcomm_server *server = (struct ustcomm_server *)app;
325 struct ustcomm_connection *conn;
326
327 list_for_each_entry(conn, &server->connections, list) {
328 if(conn->fd == src->fd) {
329 list_del(&conn->list);
330 goto found;
331 }
332 }
333
334 return -1;
335 found:
336 return src->fd;
337 }
338
339 static int init_named_socket(char *name, char **path_out)
340 {
341 int result;
342 int fd;
343
344 struct sockaddr_un addr;
345
346 result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
347 if(result == -1) {
348 PERROR("socket");
349 return -1;
350 }
351
352 addr.sun_family = AF_UNIX;
353
354 strncpy(addr.sun_path, name, UNIX_PATH_MAX);
355 addr.sun_path[UNIX_PATH_MAX-1] = '\0';
356
357 result = access(name, F_OK);
358 if(result == 0) {
359 /* file exists */
360 result = unlink(name);
361 if(result == -1) {
362 PERROR("unlink of socket file");
363 goto close_sock;
364 }
365 WARN("socket already exists; overwriting");
366 }
367
368 result = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
369 if(result == -1) {
370 PERROR("bind");
371 goto close_sock;
372 }
373
374 result = listen(fd, 1);
375 if(result == -1) {
376 PERROR("listen");
377 goto close_sock;
378 }
379
380 if(path_out) {
381 *path_out = "";
382 *path_out = strdupa(addr.sun_path);
383 }
384
385 return fd;
386
387 close_sock:
388 close(fd);
389
390 return -1;
391 }
392
393 int ustcomm_init_app(pid_t pid, struct ustcomm_app *handle)
394 {
395 int result;
396 char *name;
397
398 result = asprintf(&name, "%s/%d", SOCK_DIR, (int)pid);
399 if(result >= UNIX_PATH_MAX) {
400 ERR("string overflow allocating socket name");
401 return -1;
402 }
403
404 handle->server.listen_fd = init_named_socket(name, &(handle->server.socketpath));
405 if(handle->server.listen_fd < 0) {
406 ERR("error initializing named socket");
407 goto free_name;
408 }
409 free(name);
410
411 INIT_LIST_HEAD(&handle->server.connections);
412
413 return 0;
414
415 free_name:
416 free(name);
417 return -1;
418 }
419
420 int ustcomm_init_ustd(struct ustcomm_ustd *handle)
421 {
422 int result;
423 char *name;
424
425 result = asprintf(&name, "%s/%s", SOCK_DIR, "ustd");
426 if(result >= UNIX_PATH_MAX) {
427 ERR("string overflow allocating socket name");
428 return -1;
429 }
430
431 handle->server.listen_fd = init_named_socket(name, &handle->server.socketpath);
432 if(handle->server.listen_fd < 0) {
433 ERR("error initializing named socket");
434 goto free_name;
435 }
436 free(name);
437
438 INIT_LIST_HEAD(&handle->server.connections);
439
440 return 0;
441
442 free_name:
443 free(name);
444 return -1;
445 }
446
447 static char *find_tok(char *str)
448 {
449 while(*str == ' ') {
450 str++;
451
452 if(*str == 0)
453 return NULL;
454 }
455
456 return str;
457 }
458
459 static char *find_sep(char *str)
460 {
461 while(*str != ' ') {
462 str++;
463
464 if(*str == 0)
465 break;
466 }
467
468 return str;
469 }
470
471 int nth_token_is(char *str, char *token, int tok_no)
472 {
473 int i;
474 char *start;
475 char *end;
476
477 for(i=0; i<=tok_no; i++) {
478 str = find_tok(str);
479 if(str == NULL)
480 return -1;
481
482 start = str;
483
484 str = find_sep(str);
485 if(str == NULL)
486 return -1;
487
488 end = str;
489 }
490
491 if(end-start != strlen(token))
492 return 0;
493
494 if(strncmp(start, token, end-start))
495 return 0;
496
497 return 1;
498 }
499
500 char *nth_token(char *str, int tok_no)
501 {
502 static char *retval = NULL;
503 int i;
504 char *start;
505 char *end;
506
507 for(i=0; i<=tok_no; i++) {
508 str = find_tok(str);
509 if(str == NULL)
510 return NULL;
511
512 start = str;
513
514 str = find_sep(str);
515 if(str == NULL)
516 return NULL;
517
518 end = str;
519 }
520
521 if(retval) {
522 free(retval);
523 retval = NULL;
524 }
525
526 asprintf(&retval, "%.*s", (int)(end-start), start);
527
528 return retval;
529 }
530
This page took 0.050833 seconds and 4 git commands to generate.