lkml.org 
[lkml]   [2013]   [Oct]   [14]   [last100]   RSS Feed
Views: [wrap][no wrap]   [headers]  [forward] 
 
Messages in this thread
    Patch in this message
    /
    Date
    From
    SubjectRe: [PATCH V2 2/5] trace-cmd: Apply the trace-msg protocol for communication between a server and clients
    On Fri, 13 Sep 2013 11:06:32 +0900
    Yoshihiro YUNOMAE <yoshihiro.yunomae.ez@hitachi.com> wrote:

    > Apply trace-msg protocol for communication between a server and clients.
    >
    > Currently, trace-listen(server) and trace-record -N(client) operate as follows:
    >
    > <server> <client>
    > listen to socket fd
    > connect to socket fd
    > accept the client
    > send "tracecmd"
    > +------------> receive "tracecmd"
    > check "tracecmd"
    > send cpus
    > receive cpus <------------+
    > print "cpus=XXX"
    > send pagesize
    > |
    > receive pagesize <--------+
    > print "pagesize=XXX"
    > send option
    > |
    > receive option <----------+
    > understand option
    > send port_array
    > +------------> receive port_array
    > understand port_array
    > send meta data
    > receive meta data <-------+
    > record meta data
    > (snip)
    > read block
    > --- start sending trace data on child processes ---
    >
    > --- When client finishes sending trace data ---
    > close(socket fd)
    > read size = 0
    > close(socket fd)

    Note, this patch is filled with whitespace errors. Run checkpatch.pl on
    it if you can.

    I applied and fixed up the first patch.

    Also, when I tested this patch I got:

    Running in one terminal:

    # trace-cmd listen -p 12345

    And then in another terminal:

    # trace-cmd record -N localhost:12345 -p function -e all
    /debug/tracing/events/*/filter
    plugin 'function'
    Hit Ctrl^C to stop recording
    trace-cmd: Connection refused
    trace-cmd: Connection refused
    trace-cmd: Connection refused
    recorder error in splice output recorder error in splice output

    recorder error in splice output
    trace-cmd: Connection refused
    recorder error in splice output

    -- Steve



    >
    > All messages are unstructured character strings, so server(client) using the
    > protocol must parse the unstructured messages. Since it is hard to
    > add complex contents in the protocol, structured binary message trace-msg
    > is introduced as the communication protocol.
    >
    > By applying this patch, server and client operate as follows:
    >
    > <server> <client>
    > listen to socket fd
    > connect to socket fd
    > accept the client
    > send "tracecmd"
    > +------------> receive "tracecmd"
    > check "tracecmd"
    > send "V2\0<MAGIC_NUMBER>\00" as the v2 protocol
    > receive "V2" <------------+
    > check "V2"
    > read "<MAGIC_NUMBER>\00"
    > send "V2"
    > +---------------> receive "V2"
    > check "V2"
    > send cpus,pagesize,option(MSG_TINIT)
    > receive MSG_TINIT <-------+
    > print "cpus=XXX"
    > print "pagesize=XXX"
    > understand option
    > send port_array
    > +--MSG_RINIT-> receive MSG_RINIT
    > understand port_array
    > send meta data(MSG_SENDMETA)
    > receive MSG_SENDMETA <----+
    > record meta data
    > (snip)
    > send a message to finish sending meta data
    > | (MSG_FINMETA)
    > receive MSG_FINMETA <-----+
    > read block
    > --- start sending trace data on child processes ---
    >
    > --- When client finishes sending trace data ---
    > send MSG_CLOSE
    > receive MSG_CLOSE <-------+
    > close(socket fd) close(socket fd)
    >
    > By introducing the v2 protocol, after the client checks "tracecmd", the client
    > will send "V2\0<MAGIC_NUMBER>\00\0". This complex message is used when the
    > new client tries to connect to the old server. The new client wants to check
    > whether the reply message from the server is "V2" or not. However, the old
    > server does not respond to the client before receiving cpu numbers, page size,
    > and options. Each message is separated with "\0" in the old server, so the
    > client send "V2" as cpu numbers, "<MAGIC_NUMBER>" as page size, and "0" as
    > no options. On the other hands, the old server will understand the messages
    > as cpus=0, pagesize=<MAGIC_NUMBER>, and options=0, and then the server will
    > send the message "\0" as port numbers. Then, the message which the client
    > receives is not "V2" but "\0", so the client will reconnect to the old server
    > as the v1 protocol.
    >
    > Changes in V2: Regacy porotocol support in order to keep backward compatibility
    >
    > Signed-off-by: Yoshihiro YUNOMAE <yoshihiro.yunomae.ez@hitachi.com>
    > ---
    > Makefile | 2
    > trace-cmd.h | 11 +
    > trace-listen.c | 133 +++++++----
    > trace-msg.c | 683 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    > trace-msg.h | 27 ++
    > trace-output.c | 4
    > trace-record.c | 86 ++++++-
    > 7 files changed, 880 insertions(+), 66 deletions(-)
    > create mode 100644 trace-msg.c
    > create mode 100644 trace-msg.h
    >
    > diff --git a/Makefile b/Makefile
    > index 1964949..054f53d 100644
    > --- a/Makefile
    > +++ b/Makefile
    > @@ -314,7 +314,7 @@ KERNEL_SHARK_OBJS = $(TRACE_VIEW_OBJS) $(TRACE_GRAPH_OBJS) $(TRACE_GUI_OBJS) \
    > PEVENT_LIB_OBJS = event-parse.o trace-seq.o parse-filter.o parse-utils.o
    > TCMD_LIB_OBJS = $(PEVENT_LIB_OBJS) trace-util.o trace-input.o trace-ftrace.o \
    > trace-output.o trace-recorder.o trace-restore.o trace-usage.o \
    > - trace-blk-hack.o kbuffer-parse.o
    > + trace-blk-hack.o kbuffer-parse.o trace-msg.o
    >
    > PLUGIN_OBJS = plugin_hrtimer.o plugin_kmem.o plugin_sched_switch.o \
    > plugin_mac80211.o plugin_jbd2.o plugin_function.o plugin_kvm.o \
    > diff --git a/trace-cmd.h b/trace-cmd.h
    > index cbbc6ed..a2958ac 100644
    > --- a/trace-cmd.h
    > +++ b/trace-cmd.h
    > @@ -248,6 +248,17 @@ void tracecmd_stop_recording(struct tracecmd_recorder *recorder);
    > void tracecmd_stat_cpu(struct trace_seq *s, int cpu);
    > long tracecmd_flush_recording(struct tracecmd_recorder *recorder);
    >
    > +/* for clients */
    > +int tracecmd_msg_send_init_data(int fd);
    > +int tracecmd_msg_metadata_send(int fd, char *buf, int size);
    > +int tracecmd_msg_finish_sending_metadata(int fd);
    > +void tracecmd_msg_send_close_msg();
    > +
    > +/* for server */
    > +int tracecmd_msg_initial_setting(int fd, int *cpus, int *pagesize);
    > +int tracecmd_msg_send_port_array(int fd, int total_cpus, int *ports);
    > +int tracecmd_msg_collect_metadata(int ifd, int ofd);
    > +
    > /* --- Plugin handling --- */
    > extern struct plugin_option trace_ftrace_options[];
    >
    > diff --git a/trace-listen.c b/trace-listen.c
    > index bf187c9..280b1af 100644
    > --- a/trace-listen.c
    > +++ b/trace-listen.c
    > @@ -33,6 +33,7 @@
    > #include <errno.h>
    >
    > #include "trace-local.h"
    > +#include "trace-msg.h"
    >
    > #define MAX_OPTION_SIZE 4096
    >
    > @@ -45,10 +46,10 @@ static FILE *logfp;
    >
    > static int debug;
    >
    > -static int use_tcp;
    > -
    > static int backlog = 5;
    >
    > +static int proto_ver;
    > +
    > #define TEMP_FILE_STR "%s.%s:%s.cpu%d", output_file, host, port, cpu
    > static char *get_temp_file(const char *host, const char *port, int cpu)
    > {
    > @@ -112,10 +113,9 @@ static int process_option(char *option)
    > return 0;
    > }
    >
    > -static int done;
    > static void finish(int sig)
    > {
    > - done = 1;
    > + done = true;
    > }
    >
    > #define LOG_BUF_SIZE 1024
    > @@ -144,7 +144,7 @@ static void __plog(const char *prefix, const char *fmt, va_list ap,
    > fprintf(fp, "%.*s", r, buf);
    > }
    >
    > -static void plog(const char *fmt, ...)
    > +void plog(const char *fmt, ...)
    > {
    > va_list ap;
    >
    > @@ -153,7 +153,7 @@ static void plog(const char *fmt, ...)
    > va_end(ap);
    > }
    >
    > -static void pdie(const char *fmt, ...)
    > +void pdie(const char *fmt, ...)
    > {
    > va_list ap;
    > char *str = "";
    > @@ -324,56 +324,78 @@ static int communicate_with_client(int fd, int *cpus, int *pagesize)
    >
    > *cpus = atoi(buf);
    >
    > - plog("cpus=%d\n", *cpus);
    > - if (*cpus < 0)
    > - return -1;
    > + /* Is the client using the new protocol? */
    > + if (!*cpus) {
    > + if (memcmp(buf, "V2", 2) != 0) {
    > + plog("Cannot handle the protocol %s", buf);
    > + return -1;
    > + }
    >
    > - /* next read the page size */
    > - n = read_string(fd, buf, BUFSIZ);
    > - if (n == BUFSIZ)
    > - /** ERROR **/
    > - return -1;
    > + /* read the rest of dummy data, but not use */
    > + read(fd, buf, sizeof(V2_MAGIC)+1);
    >
    > - *pagesize = atoi(buf);
    > + proto_ver = V2_PROTOCOL;
    >
    > - plog("pagesize=%d\n", *pagesize);
    > - if (*pagesize <= 0)
    > - return -1;
    > + /* Let the client know we use v2 protocol */
    > + write(fd, "V2", 2);
    >
    > - /* Now the number of options */
    > - n = read_string(fd, buf, BUFSIZ);
    > - if (n == BUFSIZ)
    > - /** ERROR **/
    > - return -1;
    > + /* read the CPU count, the page size, and options */
    > + if (tracecmd_msg_initial_setting(fd, cpus, pagesize) < 0)
    > + return -1;
    > + } else {
    > + /* The client is using the v1 protocol */
    >
    > - options = atoi(buf);
    > + plog("cpus=%d\n", *cpus);
    > + if (*cpus < 0)
    > + return -1;
    >
    > - for (i = 0; i < options; i++) {
    > - /* next is the size of the options */
    > + /* next read the page size */
    > n = read_string(fd, buf, BUFSIZ);
    > if (n == BUFSIZ)
    > /** ERROR **/
    > return -1;
    > - size = atoi(buf);
    > - /* prevent a client from killing us */
    > - if (size > MAX_OPTION_SIZE)
    > +
    > + *pagesize = atoi(buf);
    > +
    > + plog("pagesize=%d\n", *pagesize);
    > + if (*pagesize <= 0)
    > return -1;
    > - option = malloc_or_die(size);
    > - do {
    > - t = size;
    > - s = 0;
    > - s = read(fd, option+s, t);
    > - if (s <= 0)
    > - return -1;
    > - t -= s;
    > - s = size - t;
    > - } while (t);
    >
    > - s = process_option(option);
    > - free(option);
    > - /* do we understand this option? */
    > - if (!s)
    > + /* Now the number of options */
    > + n = read_string(fd, buf, BUFSIZ);
    > + if (n == BUFSIZ)
    > + /** ERROR **/
    > return -1;
    > +
    > + options = atoi(buf);
    > +
    > + for (i = 0; i < options; i++) {
    > + /* next is the size of the options */
    > + n = read_string(fd, buf, BUFSIZ);
    > + if (n == BUFSIZ)
    > + /** ERROR **/
    > + return -1;
    > + size = atoi(buf);
    > + /* prevent a client from killing us */
    > + if (size > MAX_OPTION_SIZE)
    > + return -1;
    > + option = malloc_or_die(size);
    > + do {
    > + t = size;
    > + s = 0;
    > + s = read(fd, option+s, t);
    > + if (s <= 0)
    > + return -1;
    > + t -= s;
    > + s = size - t;
    > + } while (t);
    > +
    > + s = process_option(option);
    > + free(option);
    > + /* do we understand this option? */
    > + if (!s)
    > + return -1;
    > + }
    > }
    >
    > if (use_tcp)
    > @@ -442,14 +464,20 @@ static int *create_all_readers(int cpus, const char *node, const char *port,
    > start_port = udp_port + 1;
    > }
    >
    > - /* send the client a comma deliminated set of port numbers */
    > - for (cpu = 0; cpu < cpus; cpu++) {
    > - snprintf(buf, BUFSIZ, "%s%d",
    > - cpu ? "," : "", port_array[cpu]);
    > - write(fd, buf, strlen(buf));
    > + if (proto_ver == V2_PROTOCOL) {
    > + /* send set of port numbers to the client */
    > + if (tracecmd_msg_send_port_array(fd, cpus, port_array) < 0)
    > + goto out_free;
    > + } else {
    > + /* send the client a comma deliminated set of port numbers */
    > + for (cpu = 0; cpu < cpus; cpu++) {
    > + snprintf(buf, BUFSIZ, "%s%d",
    > + cpu ? "," : "", port_array[cpu]);
    > + write(fd, buf, strlen(buf));
    > + }
    > + /* end with null terminator */
    > + write(fd, "\0", 1);
    > }
    > - /* end with null terminator */
    > - write(fd, "\0", 1);
    >
    > return pid_array;
    >
    > @@ -528,7 +556,10 @@ static void process_client(const char *node, const char *port, int fd)
    > return;
    >
    > /* Now we are ready to start reading data from the client */
    > - collect_metadata_from_client(fd, ofd);
    > + if (proto_ver == V2_PROTOCOL)
    > + tracecmd_msg_collect_metadata(fd, ofd);
    > + else
    > + collect_metadata_from_client(fd, ofd);
    >
    > /* wait a little to let our readers finish reading */
    > sleep(1);
    > diff --git a/trace-msg.c b/trace-msg.c
    > new file mode 100644
    > index 0000000..cf82ff6
    > --- /dev/null
    > +++ b/trace-msg.c
    > @@ -0,0 +1,683 @@
    > +/*
    > + * trace-msg.c : define message protocol for communication between clients and
    > + * a server
    > + *
    > + * Copyright (C) 2013 Hitachi, Ltd.
    > + * Created by Yoshihiro YUNOMAE <yoshihiro.yunomae.ez@hitachi.com>
    > + *
    > + * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    > + *
    > + * This program is free software; you can redistribute it and/or modify
    > + * it under the terms of the GNU General Public License as published by
    > + * the Free Software Foundation; version 2 of the License (not later!)
    > + *
    > + * This program is distributed in the hope that it will be useful,
    > + * but WITHOUT ANY WARRANTY; without even the implied warranty of
    > + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    > + * GNU General Public License for more details.
    > + *
    > + * You should have received a copy of the GNU General Public License
    > + * along with this program; if not, see <http://www.gnu.org/licenses>
    > + *
    > + * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    > + */
    > +
    > +#include <errno.h>
    > +#include <poll.h>
    > +#include <fcntl.h>
    > +#include <limits.h>
    > +#include <stddef.h>
    > +#include <stdio.h>
    > +#include <unistd.h>
    > +#include <arpa/inet.h>
    > +#include <sys/types.h>
    > +#include <linux/types.h>
    > +
    > +#include "trace-cmd-local.h"
    > +#include "trace-msg.h"
    > +
    > +typedef __u32 u32;
    > +typedef __be32 be32;
    > +
    > +#define TRACECMD_MSG_MAX_LEN BUFSIZ
    > +
    > + /* size + cmd */
    > +#define TRACECMD_MSG_HDR_LEN ((sizeof(be32)) + (sizeof(be32)))
    > +
    > + /* + size of the metadata */
    > +#define TRACECMD_MSG_META_MIN_LEN \
    > + ((TRACECMD_MSG_HDR_LEN) + (sizeof(be32)))
    > +
    > + /* - header size for error msg */
    > +#define TRACECMD_MSG_META_MAX_LEN \
    > +((TRACECMD_MSG_MAX_LEN) - (TRACECMD_MSG_META_MIN_LEN) - TRACECMD_MSG_HDR_LEN)
    > +
    > + /* size + opt_cmd + size of str */
    > +#define TRACECMD_OPT_MIN_LEN \
    > + ((sizeof(be32)) + (sizeof(be32)) +(sizeof(be32)))
    > +
    > +
    > +#define CPU_MAX 256
    > +
    > +/* for both client and server */
    > +bool use_tcp;
    > +int cpu_count;
    > +
    > +/* for client */
    > +static int psfd;
    > +unsigned int page_size;
    > +int *client_ports;
    > +bool send_metadata;
    > +
    > +/* for server */
    > +static int *port_array;
    > +bool done;
    > +
    > +struct tracecmd_msg_str {
    > + be32 size;
    > + char *buf;
    > +} __attribute__((packed));
    > +
    > +struct tracecmd_msg_opt {
    > + be32 size;
    > + be32 opt_cmd;
    > + struct tracecmd_msg_str str;
    > +};
    > +
    > +struct tracecmd_msg_tinit {
    > + be32 cpus;
    > + be32 page_size;
    > + be32 opt_num;
    > + struct tracecmd_msg_opt *opt;
    > +} __attribute__((packed));
    > +
    > +struct tracecmd_msg_rinit {
    > + be32 cpus;
    > + be32 port_array[CPU_MAX];
    > +} __attribute__((packed));
    > +
    > +struct tracecmd_msg_meta {
    > + struct tracecmd_msg_str str;
    > +};
    > +
    > +struct tracecmd_msg_error {
    > + be32 size;
    > + be32 cmd;
    > + union {
    > + struct tracecmd_msg_tinit tinit;
    > + struct tracecmd_msg_rinit rinit;
    > + struct tracecmd_msg_meta meta;
    > + } data;
    > +} __attribute__((packed));
    > +
    > +enum tracecmd_msg_cmd {
    > + MSG_CLOSE = 1,
    > + MSG_TINIT = 4,
    > + MSG_RINIT = 5,
    > + MSG_SENDMETA = 6,
    > + MSG_FINMETA = 7,
    > +};
    > +
    > +struct tracecmd_msg {
    > + be32 size;
    > + be32 cmd;
    > + union {
    > + struct tracecmd_msg_tinit tinit;
    > + struct tracecmd_msg_rinit rinit;
    > + struct tracecmd_msg_meta meta;
    > + struct tracecmd_msg_error err;
    > + } data;
    > +} __attribute__((packed));
    > +
    > +struct tracecmd_msg *errmsg;
    > +
    > +static ssize_t msg_do_write_check(int fd, struct tracecmd_msg *msg)
    > +{
    > + return __do_write_check(fd, msg, ntohl(msg->size));
    > +}
    > +
    > +static struct tracecmd_msg *tracecmd_msg_alloc(u32 size)
    > +{
    > + size += TRACECMD_MSG_HDR_LEN;
    > + return malloc(size);
    > +}
    > +
    > +static void tracecmd_msg_init(u32 cmd, u32 size, struct tracecmd_msg *msg)
    > +{
    > + size += TRACECMD_MSG_HDR_LEN;
    > + memset(msg, 0, size);
    > + msg->size = htonl(size);
    > + msg->cmd = htonl(cmd);
    > +}
    > +
    > +static void bufcpy(void *dest, u32 offset, const void *buf, u32 buflen)
    > +{
    > + memcpy(dest+offset, buf, buflen);
    > +}
    > +
    > +enum msg_opt_command {
    > + MSGOPT_USETCP = 1,
    > +};
    > +
    > +static struct tracecmd_msg_opt *tracecmd_msg_opt_alloc(u32 len)
    > +{
    > + len += TRACECMD_OPT_MIN_LEN;
    > + return malloc(len);
    > +}
    > +
    > +static void make_option(int opt_cmd, const char *buf,
    > + struct tracecmd_msg_opt *opt)
    > +{
    > + u32 buflen = 0;
    > + u32 size = TRACECMD_OPT_MIN_LEN;
    > +
    > + if (buf) {
    > + buflen = strlen(buf);
    > + size += buflen;
    > + }
    > +
    > + opt->size = htonl(size);
    > + opt->opt_cmd = htonl(opt_cmd);
    > + opt->str.size = htonl(buflen);
    > +
    > + if (buf)
    > + bufcpy(opt, TRACECMD_OPT_MIN_LEN, buf, buflen);
    > +}
    > +
    > +static int add_options_to_tinit(u32 len, struct tracecmd_msg *msg)
    > +{
    > + struct tracecmd_msg_opt *opt;
    > + int offset = offsetof(struct tracecmd_msg, data.tinit.opt);
    > +
    > + if (use_tcp) {
    > + opt = tracecmd_msg_opt_alloc(0);
    > + if (!opt)
    > + return -ENOMEM;
    > +
    > + make_option(MSGOPT_USETCP, NULL, opt);
    > + /* add option */
    > + bufcpy(msg, offset, opt, ntohl(opt->size));
    > + free(opt);
    > + }
    > +
    > + return 0;
    > +}
    > +
    > +static int make_tinit(u32 len, struct tracecmd_msg *msg)
    > +{
    > + int opt_num = 0;
    > + int ret = 0;
    > +
    > + if (use_tcp)
    > + opt_num++;
    > +
    > + if (opt_num) {
    > + ret = add_options_to_tinit(len, msg);
    > + if (ret < 0)
    > + return ret;
    > + }
    > +
    > + msg->data.tinit.cpus = htonl(cpu_count);
    > + msg->data.tinit.page_size = htonl(page_size);
    > + msg->data.tinit.opt_num = htonl(opt_num);
    > +
    > + return 0;
    > +}
    > +
    > +static int make_rinit(struct tracecmd_msg *msg)
    > +{
    > + int i;
    > + u32 offset = TRACECMD_MSG_HDR_LEN;
    > + be32 port;
    > +
    > + msg->data.rinit.cpus = htonl(cpu_count);
    > +
    > + for (i = 0; i < cpu_count; i++) {
    > + /* + rrqports->cpus or rrqports->port_array[i] */
    > + offset += sizeof(be32);
    > + port = htonl(port_array[i]);
    > + bufcpy(msg, offset, &port, sizeof(be32) * cpu_count);
    > + }
    > +
    > + return 0;
    > +}
    > +
    > +static u32 tracecmd_msg_get_body_length(u32 cmd)
    > +{
    > + struct tracecmd_msg *msg;
    > + u32 len = 0;
    > +
    > + switch (cmd) {
    > + case MSG_TINIT:
    > + len = sizeof(msg->data.tinit.cpus)
    > + + sizeof(msg->data.tinit.page_size)
    > + + sizeof(msg->data.tinit.opt_num);
    > +
    > + /*
    > + * If we are using IPV4 and our page size is greater than
    > + * or equal to 64K, we need to punt and use TCP. :-(
    > + */
    > +
    > + /* TODO, test for ipv4 */
    > + if (page_size >= UDP_MAX_PACKET) {
    > + warning("page size too big for UDP using TCP "
    > + "in live read");
    > + use_tcp = true;
    > + }
    > +
    > + if (use_tcp)
    > + len += TRACECMD_OPT_MIN_LEN;
    > +
    > + return len;
    > + case MSG_RINIT:
    > + return sizeof(msg->data.rinit.cpus)
    > + + sizeof(msg->data.rinit.port_array);
    > + case MSG_SENDMETA:
    > + return TRACECMD_MSG_MAX_LEN - TRACECMD_MSG_HDR_LEN;
    > + case MSG_CLOSE:
    > + case MSG_FINMETA:
    > + break;
    > + }
    > +
    > + return 0;
    > +}
    > +
    > +static int tracecmd_msg_make_body(u32 cmd, u32 len, struct tracecmd_msg *msg)
    > +{
    > + switch (cmd) {
    > + case MSG_TINIT:
    > + return make_tinit(len, msg);
    > + case MSG_RINIT:
    > + return make_rinit(msg);
    > + case MSG_CLOSE:
    > + case MSG_SENDMETA: /* meta data is not stored here. */
    > + case MSG_FINMETA:
    > + break;
    > + }
    > +
    > + return 0;
    > +}
    > +
    > +static int tracecmd_msg_create(u32 cmd, struct tracecmd_msg **msg)
    > +{
    > + u32 len = 0;
    > + int ret = 0;
    > +
    > + len = tracecmd_msg_get_body_length(cmd);
    > + if (len > (TRACECMD_MSG_MAX_LEN - TRACECMD_MSG_HDR_LEN)) {
    > + plog("Exceed maximum message size cmd=%d\n", cmd);
    > + return -EINVAL;
    > + }
    > +
    > + *msg = tracecmd_msg_alloc(len);
    > + if (!*msg)
    > + return -ENOMEM;
    > + tracecmd_msg_init(cmd, len, *msg);
    > +
    > + ret = tracecmd_msg_make_body(cmd, len, *msg);
    > + if (ret < 0)
    > + free(*msg);
    > +
    > + return ret;
    > +}
    > +
    > +static int tracecmd_msg_send(int fd, u32 cmd)
    > +{
    > + struct tracecmd_msg *msg = NULL;
    > + int ret = 0;
    > +
    > + if (cmd > MSG_FINMETA) {
    > + plog("Unsupported command: %d\n", cmd);
    > + return -EINVAL;
    > + }
    > +
    > + ret = tracecmd_msg_create(cmd, &msg);
    > + if (ret < 0)
    > + return ret;
    > +
    > + ret = msg_do_write_check(fd, msg);
    > + if (ret < 0) {
    > + free(msg);
    > + return -ECOMM;
    > + }
    > +
    > + return 0;
    > +}
    > +
    > +static int tracecmd_msg_read_extra(int fd, char *buf, u32 size, int *n)
    > +{
    > + int r = 0;
    > +
    > + do {
    > + r = read(fd, buf+*n, size);
    > + if (r < 0) {
    > + if (errno == EINTR)
    > + continue;
    > + return -errno;
    > + } else if (!r)
    > + return -ENOTCONN;
    > + size -= r;
    > + *n += r;
    > + } while (size);
    > +
    > + return 0;
    > +}
    > +
    > +/*
    > + * Read header information of msg first, then read all data
    > + */
    > +static int tracecmd_msg_recv(int fd, char *buf)
    > +{
    > + struct tracecmd_msg *msg;
    > + u32 size = 0;
    > + int n = 0;
    > + int ret;
    > +
    > + ret = tracecmd_msg_read_extra(fd, buf, TRACECMD_MSG_HDR_LEN, &n);
    > + if (ret < 0)
    > + return ret;
    > +
    > + msg = (struct tracecmd_msg *)buf;
    > + size = ntohl(msg->size);
    > + if (size > TRACECMD_MSG_MAX_LEN)
    > + /* too big */
    > + goto error;
    > + else if (size < TRACECMD_MSG_HDR_LEN)
    > + /* too small */
    > + goto error;
    > + else if (size > TRACECMD_MSG_HDR_LEN) {
    > + size -= TRACECMD_MSG_HDR_LEN;
    > + return tracecmd_msg_read_extra(fd, buf, size, &n);
    > + }
    > +
    > + return 0;
    > +error:
    > + plog("Receive an invalid message(size=%d)\n", size);
    > + return -ENOMSG;
    > +}
    > +
    > +static void *tracecmd_msg_buf_access(struct tracecmd_msg *msg, int offset)
    > +{
    > + return (void *)msg + offset;
    > +}
    > +
    > +static int tracecmd_msg_wait_for_msg(int fd, struct tracecmd_msg **msg)
    > +{
    > + char msg_tmp[TRACECMD_MSG_MAX_LEN];
    > + u32 cmd;
    > + int ret;
    > +
    > + ret = tracecmd_msg_recv(fd, msg_tmp);
    > + if (ret < 0)
    > + return ret;
    > +
    > + *msg = (struct tracecmd_msg *)msg_tmp;
    > + cmd = ntohl((*msg)->cmd);
    > + if (cmd == MSG_CLOSE)
    > + return -ECONNABORTED;
    > +
    > + return 0;
    > +}
    > +
    > +static int tracecmd_msg_send_and_wait_for_msg(int fd, u32 cmd, struct tracecmd_msg **msg)
    > +{
    > + int ret;
    > +
    > + ret = tracecmd_msg_send(fd, cmd);
    > + if (ret < 0)
    > + return ret;
    > +
    > + ret = tracecmd_msg_wait_for_msg(fd, msg);
    > + if (ret < 0)
    > + return ret;
    > +
    > + return 0;
    > +}
    > +
    > +int tracecmd_msg_send_init_data(int fd)
    > +{
    > + struct tracecmd_msg *msg;
    > + int i, cpus;
    > + int ret;
    > +
    > + ret = tracecmd_msg_send_and_wait_for_msg(fd, MSG_TINIT, &msg);
    > + if (ret < 0)
    > + return ret;
    > +
    > + cpus = ntohl(msg->data.rinit.cpus);
    > + client_ports = malloc_or_die(sizeof(int) * cpus);
    > + for (i = 0; i < cpus; i++)
    > + client_ports[i] = ntohl(msg->data.rinit.port_array[i]);
    > +
    > + /* Next, send meta data */
    > + send_metadata = true;
    > +
    > + return 0;
    > +}
    > +
    > +static bool process_option(struct tracecmd_msg_opt *opt)
    > +{
    > + /* currently the only option we have is to us TCP */
    > + if (ntohl(opt->opt_cmd) == MSGOPT_USETCP) {
    > + use_tcp = true;
    > + return true;
    > + }
    > + return false;
    > +}
    > +
    > +static void error_operation_for_server(struct tracecmd_msg *msg)
    > +{
    > + u32 cmd;
    > +
    > + cmd = ntohl(msg->cmd);
    > +
    > + warning("Message: cmd=%d size=%d\n", cmd, ntohl(msg->size));
    > +}
    > +
    > +#define MAX_OPTION_SIZE 4096
    > +
    > +int tracecmd_msg_initial_setting(int fd, int *cpus, int *pagesize)
    > +{
    > + struct tracecmd_msg *msg;
    > + struct tracecmd_msg_opt *opt;
    > + char buf[TRACECMD_MSG_MAX_LEN];
    > + int offset = offsetof(struct tracecmd_msg, data.tinit.opt);
    > + int options, i, s;
    > + int ret;
    > + u32 size = 0;
    > + u32 cmd;
    > +
    > + ret = tracecmd_msg_recv(fd, buf);
    > + if (ret < 0)
    > + return ret;
    > +
    > + msg = (struct tracecmd_msg *)buf;
    > + cmd = ntohl(msg->cmd);
    > + if (cmd != MSG_TINIT) {
    > + ret = -EINVAL;
    > + goto error;
    > + }
    > +
    > + *cpus = ntohl(msg->data.tinit.cpus);
    > + plog("cpus=%d\n", *cpus);
    > + if (*cpus < 0) {
    > + ret = -EINVAL;
    > + goto error;
    > + }
    > +
    > + *pagesize = ntohl(msg->data.tinit.page_size);
    > + plog("pagesize=%d\n", *pagesize);
    > + if (*pagesize <= 0) {
    > + ret = -EINVAL;
    > + goto error;
    > + }
    > +
    > + options = ntohl(msg->data.tinit.opt_num);
    > + for (i = 0; i < options; i++) {
    > + offset += size;
    > + opt = tracecmd_msg_buf_access(msg, offset);
    > + size = ntohl(opt->size);
    > + /* prevent a client from killing us */
    > + if (size > MAX_OPTION_SIZE) {
    > + plog("Exceed MAX_OPTION_SIZE\n");
    > + ret = -EINVAL;
    > + goto error;
    > + }
    > + s = process_option(opt);
    > + /* do we understand this option? */
    > + if (!s) {
    > + plog("Cannot understand(%d:%d:%d)\n",
    > + i, ntohl(opt->size), ntohl(opt->opt_cmd));
    > + ret = -EINVAL;
    > + goto error;
    > + }
    > + }
    > +
    > + return 0;
    > +
    > +error:
    > + error_operation_for_server(msg);
    > + return ret;
    > +}
    > +
    > +int tracecmd_msg_send_port_array(int fd, int total_cpus, int *ports)
    > +{
    > + int ret;
    > +
    > + cpu_count = total_cpus;
    > + port_array = ports;
    > +
    > + ret = tracecmd_msg_send(fd, MSG_RINIT);
    > + if (ret < 0)
    > + return ret;
    > +
    > + return 0;
    > +}
    > +
    > +void tracecmd_msg_send_close_msg()
    > +{
    > + tracecmd_msg_send(psfd, MSG_CLOSE);
    > +}
    > +
    > +static void make_meta(const char *buf, int buflen, struct tracecmd_msg *msg)
    > +{
    > + int offset = offsetof(struct tracecmd_msg, data.meta.str.buf);
    > +
    > + msg->data.meta.str.size = htonl(buflen);
    > + bufcpy(msg, offset, buf, buflen);
    > +}
    > +
    > +int tracecmd_msg_metadata_send(int fd, char *buf, int size)
    > +{
    > + struct tracecmd_msg *msg;
    > + int n, len;
    > + int ret;
    > + int count = 0;
    > +
    > + ret = tracecmd_msg_create(MSG_SENDMETA, &msg);
    > + if (ret < 0)
    > + return ret;
    > +
    > + n = size;
    > + do {
    > + if (n > TRACECMD_MSG_META_MAX_LEN) {
    > + make_meta(buf+count, TRACECMD_MSG_META_MAX_LEN, msg);
    > + n -= TRACECMD_MSG_META_MAX_LEN;
    > + count += TRACECMD_MSG_META_MAX_LEN;
    > + } else {
    > + make_meta(buf+count, n, msg);
    > + /*
    > + * TRACECMD_MSG_META_MAX_LEN is stored in msg->size,
    > + * so update the size to the correct value.
    > + */
    > + len = TRACECMD_MSG_META_MIN_LEN + n;
    > + msg->size = htonl(len);
    > + n = 0;
    > + }
    > +
    > + ret = msg_do_write_check(fd, msg);
    > + if (ret < 0)
    > + return ret;
    > + } while (n);
    > +
    > + return 0;
    > +}
    > +
    > +int tracecmd_msg_finish_sending_metadata(int fd)
    > +{
    > + int ret;
    > +
    > + ret = tracecmd_msg_send(fd, MSG_FINMETA);
    > + if (ret < 0)
    > + return ret;
    > +
    > + /* psfd will be used for closing */
    > + psfd = fd;
    > + return 0;
    > +}
    > +
    > +int tracecmd_msg_collect_metadata(int ifd, int ofd)
    > +{
    > + struct tracecmd_msg *msg;
    > + char buf[TRACECMD_MSG_MAX_LEN];
    > + u32 s, t, n, cmd;
    > + int offset = TRACECMD_MSG_META_MIN_LEN;
    > + int ret;
    > +
    > + do {
    > + ret = tracecmd_msg_recv(ifd, buf);
    > + if (ret < 0) {
    > + warning("reading client");
    > + return ret;
    > + }
    > +
    > + msg = (struct tracecmd_msg *)buf;
    > + cmd = ntohl(msg->cmd);
    > + if (cmd == MSG_FINMETA) {
    > + /* Finish receiving meta data */
    > + break;
    > + } else if (cmd != MSG_SENDMETA)
    > + goto error;
    > +
    > + n = ntohl(msg->data.meta.str.size);
    > + t = n;
    > + s = 0;
    > + do {
    > + s = write(ofd, buf+s+offset, t);
    > + if (s < 0) {
    > + if (errno == EINTR)
    > + continue;
    > + warning("writing to file");
    > + return -errno;
    > + }
    > + t -= s;
    > + s = n - t;
    > + } while (t);
    > + } while (cmd == MSG_SENDMETA);
    > +
    > + /* check the finish message of the client */
    > + while(!done) {
    > + ret = tracecmd_msg_recv(ifd, buf);
    > + if (ret < 0) {
    > + warning("reading client");
    > + return ret;
    > + }
    > +
    > + msg = (struct tracecmd_msg *)buf;
    > + cmd = ntohl(msg->cmd);
    > + if (cmd == MSG_CLOSE)
    > + /* Finish this connection */
    > + break;
    > + else {
    > + warning("Not accept the message %d", ntohl(msg->cmd));
    > + ret = -EINVAL;
    > + goto error;
    > + }
    > + }
    > +
    > + return 0;
    > +
    > +error:
    > + error_operation_for_server(msg);
    > + return ret;
    > +}
    > diff --git a/trace-msg.h b/trace-msg.h
    > new file mode 100644
    > index 0000000..b23e72b
    > --- /dev/null
    > +++ b/trace-msg.h
    > @@ -0,0 +1,27 @@
    > +#ifndef _TRACE_MSG_H_
    > +#define _TRACE_MSG_H_
    > +
    > +#include <stdbool.h>
    > +
    > +#define UDP_MAX_PACKET (65536 - 20)
    > +#define V2_MAGIC "677768\0"
    > +
    > +#define V1_PROTOCOL 1
    > +#define V2_PROTOCOL 2
    > +
    > +/* for both client and server */
    > +extern bool use_tcp;
    > +extern int cpu_count;
    > +
    > +/* for client */
    > +extern unsigned int page_size;
    > +extern int *client_ports;
    > +extern bool send_metadata;
    > +
    > +/* for server */
    > +extern bool done;
    > +
    > +void plog(const char *fmt, ...);
    > +void pdie(const char *fmt, ...);
    > +
    > +#endif /* _TRACE_MSG_H_ */
    > diff --git a/trace-output.c b/trace-output.c
    > index bdb478d..6e1298b 100644
    > --- a/trace-output.c
    > +++ b/trace-output.c
    > @@ -36,6 +36,7 @@
    > #include <glob.h>
    >
    > #include "trace-cmd-local.h"
    > +#include "trace-msg.h"
    > #include "version.h"
    >
    > /* We can't depend on the host size for size_t, all must be 64 bit */
    > @@ -80,6 +81,9 @@ struct list_event_system {
    > static stsize_t
    > do_write_check(struct tracecmd_output *handle, void *data, tsize_t size)
    > {
    > + if (send_metadata)
    > + return tracecmd_msg_metadata_send(handle->fd, data, size);
    > +
    > return __do_write_check(handle->fd, data, size);
    > }
    >
    > diff --git a/trace-record.c b/trace-record.c
    > index 0199627..ebfe6c0 100644
    > --- a/trace-record.c
    > +++ b/trace-record.c
    > @@ -45,6 +45,7 @@
    > #include <errno.h>
    >
    > #include "trace-local.h"
    > +#include "trace-msg.h"
    >
    > #define _STR(x) #x
    > #define STR(x) _STR(x)
    > @@ -59,29 +60,21 @@
    > #define STAMP "stamp"
    > #define FUNC_STACK_TRACE "func_stack_trace"
    >
    > -#define UDP_MAX_PACKET (65536 - 20)
    > -
    > static int tracing_on_init_val;
    >
    > static int rt_prio;
    >
    > -static int use_tcp;
    > -
    > -static unsigned int page_size;
    > -
    > static int buffer_size;
    >
    > static const char *output_file = "trace.dat";
    >
    > static int latency;
    > static int sleep_time = 1000;
    > -static int cpu_count;
    > static int recorder_threads;
    > static int *pids;
    > static int buffers;
    >
    > static char *host;
    > -static int *client_ports;
    > static int sfd;
    >
    > /* Max size to let a per cpu file get */
    > @@ -99,6 +92,8 @@ static unsigned recorder_flags;
    > /* Try a few times to get an accurate date */
    > static int date2ts_tries = 5;
    >
    > +static int proto_ver = V2_PROTOCOL;
    > +
    > struct func_list {
    > struct func_list *next;
    > const char *func;
    > @@ -1607,20 +1602,26 @@ static int create_recorder(struct buffer_instance *instance, int cpu, int extrac
    > exit(0);
    > }
    >
    > -static void communicate_with_listener(int fd)
    > +static void check_first_msg_from_server(int fd)
    > {
    > char buf[BUFSIZ];
    > - ssize_t n;
    > - int cpu, i;
    >
    > - n = read(fd, buf, 8);
    > + read(fd, buf, 8);
    >
    > /* Make sure the server is the tracecmd server */
    > if (memcmp(buf, "tracecmd", 8) != 0)
    > die("server not tracecmd server");
    > +}
    >
    > - /* write the number of CPUs we have (in ASCII) */
    > +static void communicate_with_listener_v1(int fd)
    > +{
    > + char buf[BUFSIZ];
    > + ssize_t n;
    > + int cpu, i;
    > +
    > + check_first_msg_from_server(fd);
    >
    > + /* write the number of CPUs we have (in ASCII) */
    > sprintf(buf, "%d", cpu_count);
    >
    > /* include \0 */
    > @@ -1675,6 +1676,46 @@ static void communicate_with_listener(int fd)
    > }
    > }
    >
    > +static void communicate_with_listener_v2(int fd)
    > +{
    > + if (tracecmd_msg_send_init_data(fd) < 0)
    > + die("Cannot communicate with server");
    > +}
    > +
    > +static void check_protocol_version(int fd)
    > +{
    > + char buf[BUFSIZ];
    > +
    > + check_first_msg_from_server(fd);
    > +
    > + /*
    > + * Write the protocol version, the magic number, and the dummy
    > + * option(0) (in ASCII). The client understands whether the client
    > + * uses the v2 protocol or not by checking a reply message from the
    > + * server. If the message is "V2", the server uses v2 protocol. On the
    > + * other hands, if the message is just number strings, the server
    > + * returned port numbers. So, in that time, the client understands the
    > + * server uses the v1 protocol. However, the old server tells the
    > + * client port numbers after reading cpu_count, page_size, and option.
    > + * So, we add the dummy number (the magic number and 0 option) to the
    > + * first client message.
    > + */
    > + write(fd, "V2\0"V2_MAGIC"0", sizeof(V2_MAGIC)+4);
    > +
    > + /* read a reply message */
    > + read(fd, buf, BUFSIZ);
    > +
    > + if (!buf[0]) {
    > + /* the server uses the v1 protocol, so we'll use it */
    > + proto_ver = V1_PROTOCOL;
    > + plog("Use the v1 protocol\n");
    > + } else {
    > + if (memcmp(buf, "V2", 2) != 0)
    > + die("Cannot handle the protocol %s", buf);
    > + /* OK, let's use v2 protocol */
    > + }
    > +}
    > +
    > static void setup_network(void)
    > {
    > struct tracecmd_output *handle;
    > @@ -1703,6 +1744,7 @@ static void setup_network(void)
    > hints.ai_family = AF_UNSPEC;
    > hints.ai_socktype = SOCK_STREAM;
    >
    > +again:
    > s = getaddrinfo(server, port, &hints, &result);
    > if (s != 0)
    > die("getaddrinfo: %s", gai_strerror(s));
    > @@ -1723,16 +1765,32 @@ static void setup_network(void)
    >
    > freeaddrinfo(result);
    >
    > - communicate_with_listener(sfd);
    > + if (proto_ver == V2_PROTOCOL) {
    > + check_protocol_version(sfd);
    > + if (proto_ver == V1_PROTOCOL) {
    > + /* reconnect to the server for using the v1 protocol */
    > + close(sfd);
    > + goto again;
    > + }
    > + communicate_with_listener_v2(sfd);
    > + }
    > +
    > + if (proto_ver == V1_PROTOCOL)
    > + communicate_with_listener_v1(sfd);
    >
    > /* Now create the handle through this socket */
    > handle = tracecmd_create_init_fd_glob(sfd, listed_events);
    >
    > + if (proto_ver == V2_PROTOCOL)
    > + tracecmd_msg_finish_sending_metadata(sfd);
    > +
    > /* OK, we are all set, let'r rip! */
    > }
    >
    > static void finish_network(void)
    > {
    > + if (proto_ver == V2_PROTOCOL)
    > + tracecmd_msg_send_close_msg();
    > close(sfd);
    > free(host);
    > }



    \
     
     \ /
      Last update: 2013-10-15 04:41    [W:6.175 / U:0.700 seconds]
    ©2003-2020 Jasper Spaans|hosted at Digital Ocean and TransIP|Read the blog|Advertise on this site