]> rtime.felk.cvut.cz Git - frescor/fwp.git/blobdiff - wme_test/wclient.c
wme_test: Use send_async() which is implemented by FWP
[frescor/fwp.git] / wme_test / wclient.c
index 690671c5023bbeb6ffa20267d0660f21783432c5..592e5f545d8752363d1802237f07f333f33dccdd 100644 (file)
 #include <sys/ioctl.h>
 #include <net/if.h>
 #include <inttypes.h>
+#include <ncurses.h>
 
 #ifdef WITH_FWP
 #include <frsh.h>
-#include <ncurses.h>
 #include <fwp_res.h>
+
+/* static UL_LOG_CUST(ulogd); */
+/* static ul_log_domain_t ulogd = {UL_LOGL_MSG, "wclient"}; */
+/* UL_LOGREG_SINGLE_DOMAIN_INIT_FUNCTION(init_ulogd_wclient, ulogd); */
+
 #endif
 
 #define MAX_STREAMS  10 
@@ -43,6 +48,7 @@ unsigned opt_def_period_msec = 0;
 int opt_granularity_usec = MIN_GRANULARITY;
 bool opt_wait_for_queue_is_full; /* Don't gather any statistics until any queue is full */
 char *opt_comment = NULL;
+bool opt_gui = false;
 
 bool some_queue_is_full = false;
 uint64_t reset_timestamp; /* [nsec] */
@@ -104,12 +110,12 @@ struct stream {
        frsh_vres_id_t vres, vres_rcv;
        uint16_t resp_port;
        struct receiver receiver;
-       long wc_delay;          /* worst-case delay  */
 #endif
 
        /* Statistics */
        pthread_mutex_t mutex;
        unsigned long long sent, really_sent, received;
+       long wc_delay;          /* worst-case delay  */
 };
 
 static struct cmsg_ipi {
@@ -450,11 +456,9 @@ void* receiver(void* arg)
                }
                pthread_mutex_unlock(&delay_stats_mutex);
 
-#ifdef WITH_FWP
-               if (trans_time_usec > stream->wc_delay) {
-                       stream->wc_delay = trans_time_usec;
+               if (trans_time_usec > streams[msg.stream].wc_delay) {
+                       streams[msg.stream].wc_delay = trans_time_usec;
                }
-#endif
                receivers[ac].received++;
                
                pthread_mutex_lock(&streams[msg.stream].mutex);
@@ -523,11 +527,15 @@ send_packet_native(struct stream* stream, union msg_buff* buff)
 static inline int 
 send_packet_fwp(struct stream* stream, union msg_buff* buff)
 {
-       int ret;
+       int ret = 0;
 
        buff->msg.resp_port = htons(stream->resp_port);
-       ret = frsh_send_sync(stream->endpoint, buff, stream->packet_size);
-
+       ret = frsh_send_async(stream->endpoint, buff, stream->packet_size);
+       if (ret) {
+               char msg[1024];
+               frsh_strerror(ret, msg, sizeof(msg));
+               fprintf(stderr, "frsh_send error: %s\n", msg);
+       }
        return (ret == 0) ? 0 : -1;
 }
 #endif
@@ -563,11 +571,11 @@ void* sender(void* arg)
        struct timespec ts;
        int ret;
 
-#ifndef WITH_FWP
-       char stream_desc[100];
-       stream_to_text(stream_desc, sizeof(stream_desc), stream, 0);
-       printf("%s\n", stream_desc);
-#endif
+       if (!opt_gui) {
+               char stream_desc[100];
+               stream_to_text(stream_desc, sizeof(stream_desc), stream, 0);
+               printf("%s\n", stream_desc);
+       }
        if (stream->bandwidth_bps == 0)
                goto out;
 
@@ -587,6 +595,7 @@ void* sender(void* arg)
 
                ret = send_packet(stream, &buff);
                if (ret < 0) {
+                       stopper();
                        goto out;
                }
 
@@ -642,7 +651,7 @@ static int negotiate_contract_for_stream_fwp(struct stream *stream)
        frsh_contract_set_basic_params(&contract, &budget, &period, FRSH_WT_BOUNDED, FRSH_CT_DUMMY);
        deadline = frsh_usec_to_rel_time(3*stream->period_usec);
        frsh_contract_set_timing_reqs(&contract, false, &deadline, 0, si, 0, si);
-       
+
        ret = frsh_contract_negotiate(&contract, &stream->vres_rcv);
        frsh_contract_destroy(&contract);
        if (ret != 0) {
@@ -858,11 +867,10 @@ char* parse_bandwidths(char *params)
 void wait_for_all_threads_to_finish_fwp(void)
 {
        int i;
-       /* Wait for all threads to finish */
-       /* FIXME: */
-/*     for (i=0; i < 2*nr_streams; i++) { */
-/*             sem_wait(&sem_thread_finished); */
-/*     } */
+       /* wait for all threads to finish */
+       for (i=0; i < 2*nr_streams; i++) {
+               sem_wait(&sem_thread_finished);
+       }
 }
 #else
 void wait_for_all_threads_to_finish_native(void)
@@ -875,14 +883,78 @@ void wait_for_all_threads_to_finish_native(void)
 }
 #endif
 
-#ifdef WITH_FWP
+WINDOW *logwin;
+
+#if 0
+struct log_params {
+       ul_log_domain_t *domain;
+       int level;
+       const char *format;
+       va_list ap;
+};
+
+
+int locked_log(WINDOW *logwin, void *arg)
+{
+       struct log_params *p = arg;
+       if(!(p->level&UL_LOGL_CONT)) {
+               p->level&=UL_LOGL_MASK;
+               if(p->level)
+                       wprintw(logwin,"<%d>", p->level);
+               if(p->domain && p->domain->name)
+                       wprintw(logwin,"%s: ",p->domain->name);
+       }
+       vwprintw(logwin, p->format, p->ap);
+       wnoutrefresh(logwin);
+       return 0;
+}
+
+void
+wclient_log_fnc(ul_log_domain_t *domain, int level,
+               const char *format, va_list ap)
+{
+       struct log_params p = {
+               .domain = domain,
+               .level = level,
+               .format = format,
+       };
+       va_copy(p.ap, ap);
+       va_end(ap);
+
+       use_window(logwin, locked_log, (void*)&p);
+}
+#endif
+
+
 void init_gui()
 {
-       initscr();
-       cbreak();
-       noecho();
+       if (opt_gui) {
+               initscr();
+               cbreak();
+               noecho();
+/*             nonl(); */
+/*             intrflush(stdscr, FALSE); */
+/*             keypad(stdscr, TRUE); */
+
+               logwin = newwin(0, 0, LINES/2, 0);
+               if (logwin) {
+                       scrollok(logwin, TRUE);
+/*             ul_log_redir(wclient_log_fnc, 0); */
+               }
+       }
+}
+
+void end_gui()
+{
+       if (opt_gui) {
+               endwin();
+               if (logwin) {
+/*             ul_log_redir(NULL, 0); */
+               }
+       }
 }
 
+
 #define addfield(title, format, ...)                                   \
        move(y, x);                                                     \
        x+=strlen(title)+1;                                             \
@@ -892,7 +964,7 @@ void init_gui()
                addstr(str);                                            \
        }
 
-void print_status(int seconds)
+void print_status_gui(int seconds)
 {
        int i;
        char str[200], s1[20];
@@ -911,11 +983,11 @@ void print_status(int seconds)
                addfield("Worst-case delay", "%s", usec_to_text(s1, s->wc_delay));
                addfield("Received responses", "%lld", s->received);
        }
-       refresh();
+       wnoutrefresh(stdscr);
+       doupdate();
 }
-#else
-void init_gui() {}
-void print_status(int seconds)
+
+void print_status_nogui(int seconds)
 {
        int ac;
        fprintf(stderr, "\r%3ds", seconds);
@@ -926,7 +998,6 @@ void print_status(int seconds)
        }
        fflush(stderr);
 }
-#endif
 
 int main(int argc, char *argv[])
 {
@@ -936,7 +1007,7 @@ int main(int argc, char *argv[])
        char opt;
 
 
-       while ((opt = getopt(argc, argv, "B:b:C:c:g:I:j:o:qQ:s:T:")) != -1) {
+       while ((opt = getopt(argc, argv, "B:b:C:c:Gg:I:j:o:qQ:s:T:")) != -1) {
                switch (opt) {
                        case 'B':
                                opt_def_bandwidth = atoi(optarg);
@@ -958,6 +1029,9 @@ int main(int argc, char *argv[])
                        case 'c':
                                opt_count_sec = atoi(optarg);
                                break;
+                       case 'G':
+                               opt_gui = true;
+                               break;
                        case 'g':
                                opt_granularity_usec = atoi(optarg);
                                if (opt_granularity_usec < MIN_GRANULARITY) {
@@ -1048,6 +1122,7 @@ int main(int argc, char *argv[])
        reset_statistics();
 
 #ifdef WITH_FWP
+       //ul_log_domain_arg2levels("6");
        rc = frsh_init();
        if (rc != 0) {
                error(1, errno, "FWP initialization failed");
@@ -1092,36 +1167,40 @@ int main(int argc, char *argv[])
        if (some_contract_not_accepted) {
                stopper();
        } else {
-               //init_gui();
+               init_gui();
 
                seconds = 1;
                frames=0;
                while (!exit_flag) {
-#ifdef WITH_FWP
-                       usleep(40000);
-#else
-                       sleep(1);
-#endif
-                       frames++;
-                       if (frames>=25) {
+                       if (opt_gui) {
+                               usleep(40000);
+                               frames++;
+                               if (frames>=25) {
+                                       seconds++;
+                                       frames = 0;
+                               }
+                               print_status_gui(seconds);
+                       } else {
+                               sleep(1);
                                seconds++;
-                               frames = 0;
+                               print_status_nogui(seconds);
                        }
-                       print_status(seconds);
+
                        if (seconds == opt_count_sec)
                                stopper();
                }
        }
 
+       end_gui();
 #ifdef WITH_FWP
-       //endwin();
        for (i=0; i < nr_streams; i++) {
                if (streams[i].vres)
                        frsh_contract_cancel(streams[i].vres);
                if (streams[i].vres_rcv)
                        frsh_contract_cancel(streams[i].vres_rcv);
        }
-#else
+#endif
+
        fprintf(stderr, "\nWaiting for threads to finish\n");
        wait_for_all_threads_to_finish();
 
@@ -1132,7 +1211,6 @@ int main(int argc, char *argv[])
        measure_length = end_timestamp - reset_timestamp;
 
        save_results(argc, argv, measure_length/1000);
-#endif
 
        return 0;
 }