Utils/test-socket-signal/peer.c

   1 /*
   2  * inet (BSD) socket peer application
   3  *    starts a socket listener (server part) and connects to another
   4  *    peer (client part)
   5  *
   6  * 2010, Razvan Deaconescu
   7  */
   8
   9 #include <stdio.h>
  10 #include <string.h>
  11 #include <stdlib.h>
  12 #include <time.h>
  13 #include <unistd.h>
  14 #include <sys/types.h>
  15 #include <sys/socket.h>
  16 #include <signal.h>
  17 #include <getopt.h>
  18
  19 #include "sock_util.h"
  20 #include "utils.h"
  21
  22 #define DEFAULT_LEADER_LISTEN_PORT      43210
  23 #define DEFAULT_LEADER_HOSTNAME         "localhost"
  24 #define DEFAULT_FOLLOWER_LISTEN_PORT    54321
  25 #define DEFAULT_FOLLOWER_HOSTNAME       "localhost"
  26
  27 #define DEFAULT_SERVER_BACKLOG          5
  28
  29 #define PACKET_PAYLOAD_SIZE             120
  30 #define PACKET_INDEX_SIZE               sizeof(unsigned long long)
  31 #define PACKET_TIMESPEC_SIZE            sizeof(time_t)
  32 #define PACKET_SIZE                     (PACKET_PAYLOAD_SIZE + PACKET_INDEX_SIZE + PACKET_TIMESPEC_SIZE)
  33
  34 #define TIMER_FREQUENCY_SECS            1
  35 #define CLOCKID                         CLOCK_REALTIME
  36 #define SIG                             SIGRTMIN
  37
  38 static enum {
  39         TYPE_LEADER = 1,
  40         TYPE_FOLLOWER,
  41         TYPE_OLD_LEADER
  42 } client_type;
  43
  44
  45 static char rcv_buf[PACKET_SIZE];
  46 static char snd_buf[PACKET_SIZE];
  47
  48 /* connection socket */
  49 static int connectfd;
  50
  51 static timer_t timerid;
  52
  53 static void init_buffer_random(char *buf, size_t len)
  54 {
  55         size_t i;
  56
  57         srand(time(NULL));
  58
  59         for (i = 0; i < len-2; i++)
  60                 buf[i] = (char) (rand() % 26) + 'a';
  61         buf[i] = '\0';
  62 }
  63
  64 static void init_buffers(void)
  65 {
  66         init_buffer_random(rcv_buf, PACKET_PAYLOAD_SIZE);
  67         init_buffer_random(snd_buf, PACKET_PAYLOAD_SIZE);
  68 }
  69
  70 static void fill_send_buffer(void)
  71 {
  72         static unsigned long long index = 0;
  73         char *ptr;
  74         time_t curr_time_secs;
  75
  76         curr_time_secs = time(NULL);
  77
  78         ptr = snd_buf + PACKET_PAYLOAD_SIZE;
  79         * (unsigned long long *) ptr = index;
  80         ptr += PACKET_INDEX_SIZE;
  81         * (time_t *) ptr = curr_time_secs;
  82
  83         printf("[send] index: %llu curr_time_secs: %lu\n", index, curr_time_secs);
  84         index++;
  85 }
  86
  87 static ssize_t send_buffer(int sockfd)
  88 {
  89         return send(sockfd, snd_buf, PACKET_SIZE, 0);
  90 }
  91
  92 static ssize_t receive_buffer(int sockfd)
  93 {
  94         return recv(sockfd, rcv_buf, PACKET_SIZE, 0);
  95 }
  96
  97 static void timer_handler(int sig, siginfo_t *si, void *uc)
  98 {
  99         ssize_t nbytes;
 100
 101         fill_send_buffer();
 102         nbytes = send_buffer(connectfd);
 103         DIE(nbytes < 0, "send_buffer");
 104 }
 105
 106 static void remove_timer(void)
 107 {
 108         timer_delete(timerid);
 109 }
 110
 111 static void schedule_timer(void)
 112 {
 113         struct sigaction sa;
 114         struct sigevent sev;
 115         sigset_t mask;
 116         struct itimerspec its;
 117         int rc;
 118
 119         memset(&sa, 0, sizeof(sa));
 120         sa.sa_flags = SA_SIGINFO | SA_RESTART;
 121         sa.sa_sigaction = timer_handler;
 122         sigemptyset(&sa.sa_mask);
 123         rc = sigaction(SIG, &sa, NULL);
 124         DIE(rc < 0, "sigaction");
 125
 126         /* Block timer signal temporarily */
 127         sigemptyset(&mask);
 128         sigaddset(&mask, SIG);
 129         rc = sigprocmask(SIG_SETMASK, &mask, NULL);
 130         DIE(rc < 0, "sigprocmask");
 131
 132         /* Create the timer */
 133         sev.sigev_notify = SIGEV_SIGNAL;
 134         sev.sigev_signo = SIG;
 135         sev.sigev_value.sival_ptr = &timerid;
 136         rc = timer_create(CLOCKID, &sev, &timerid);
 137         DIE(rc < 0, "timer_create");
 138
 139         /* Start the timer */
 140         its.it_value.tv_sec = TIMER_FREQUENCY_SECS;
 141         its.it_value.tv_nsec = 0;
 142         its.it_interval.tv_sec = TIMER_FREQUENCY_SECS;
 143         its.it_interval.tv_nsec = 0;
 144         rc = timer_settime(timerid, 0, &its, NULL);
 145         DIE(rc < 0, "timer_settime");
 146
 147         /* Unlock the timer signal, so that timer notification
 148            can be delivered */
 149         rc = sigprocmask(SIG_UNBLOCK, &mask, NULL);
 150         DIE(rc < 0, "sigprocmask");
 151 }
 152
 153 static void print_buffer_meta(void)
 154 {
 155         unsigned long long index;
 156         time_t curr_time_secs;
 157         time_t sender_time_secs;
 158         char *ptr;
 159
 160         curr_time_secs = time(NULL);
 161
 162         ptr = rcv_buf + PACKET_PAYLOAD_SIZE;
 163         index = * (unsigned long long *) ptr;
 164         ptr += PACKET_INDEX_SIZE;
 165         sender_time_secs = * (time_t *) ptr;
 166
 167         printf("[recv] index %llu, ", index);
 168         if (sender_time_secs > curr_time_secs)
 169                 printf("negative latency (weird)\n");
 170         else
 171                 printf("latency %lu seconds (curr_time = %lu, sender_time = %lu)\n",
 172                                 curr_time_secs - sender_time_secs,
 173                                 curr_time_secs,
 174                                 sender_time_secs);
 175 }
 176
 177 static void usage(const char *argv0)
 178 {
 179         fprintf(stderr, "Usage: %s [-l | --leader] [-p | --port <listen-port>] [-b | --buffer-size <buffer-size>] [-f | --frequency <frequency>] <peer-host> <peer-port>\n", argv0);
 180 }
 181
 182 static void parse_args(int argc, char **argv)
 183 {
 184         int c;
 185         int digit_optind = 0;
 186
 187         while (1) {
 188                 int this_option_optind = optind ? optind : 1;
 189                 int option_index = 0;
 190                 static struct option long_options[] = {
 191                         {"leader", 0, NULL, 0},
 192                         {"port", 1, NULL, 0},
 193                         {"buffer-size", 1, NULL, 0},
 194                         {"frequency", 1, NULL, 0},
 195                         {0, 0, 0, 0}
 196                 };
 197
 198                 c = getopt_long(argc, argv, "lp:b:f:",
 199                                 long_options, &option_index);
 200                 if (c == -1)
 201                         break;
 202
 203                 switch (c) {
 204                 case 0:
 205                         printf("option %s", long_options[option_index].name);
 206                         if (optarg)
 207                                 printf(" with arg %s", optarg);
 208                         printf("\n");
 209                         break;
 210
 211                 case 'l':
 212                         printf("option l\n");
 213                         break;
 214
 215                 case 'p':
 216                         printf("option l with value '%s'\n", optarg);
 217                         break;
 218
 219                 case 'b':
 220                         printf("option b with value '%s'\n", optarg);
 221                         break;
 222
 223                 case 'f':
 224                         printf("option f with value '%s'\n", optarg);
 225                         break;
 226
 227                 case '?':
 228                         break;
 229
 230                 default:
 231                         printf("?? getopt returned character code 0%o ??\n", c);
 232                 }
 233         }
 234
 235         if (optind < argc) {
 236                 printf("non-option ARGV-elements: ");
 237                 while (optind < argc)
 238                         printf("%s ", argv[optind++]);
 239                 printf("\n");
 240         }
 241 }
 242
 243 int main(int argc, char **argv)
 244 {
 245         int listenfd;
 246         int sockfd;
 247         struct sockaddr_in addr;
 248         socklen_t addrlen = 0;
 249
 250         parse_args(argc, argv);
 251
 252 #if 0
 253         if (client_type == TYPE_LEADER) {
 254                 listenfd = tcp_listen_connections(DEFAULT_LEADER_LISTEN_PORT,
 255                                 DEFAULT_SERVER_BACKLOG);
 256                 DIE(listenfd < 0, "tcp_listen_connections");
 257
 258                 sockfd = accept(listenfd, (SSA *) &addr, &addrlen);
 259                 DIE(sockfd < 0, "accept");
 260
 261                 connectfd = tcp_connect_to_server(DEFAULT_FOLLOWER_HOSTNAME,
 262                                 DEFAULT_FOLLOWER_LISTEN_PORT);
 263                 DIE(connectfd < 0, "tcp_connect_to_server");
 264         }
 265         else if (client_type == TYPE_FOLLOWER) {
 266                 listenfd = tcp_listen_connections(DEFAULT_FOLLOWER_LISTEN_PORT,
 267                                 DEFAULT_SERVER_BACKLOG);
 268                 DIE(listenfd < 0, "tcp_listen_connections");
 269
 270                 connectfd = tcp_connect_to_server(DEFAULT_LEADER_HOSTNAME,
 271                                 DEFAULT_LEADER_LISTEN_PORT);
 272                 DIE(connectfd < 0, "tcp_connect_to_server");
 273
 274                 sockfd = accept(listenfd, (SSA *) &addr, &addrlen);
 275                 DIE(sockfd < 0, "accept");
 276         }
 277         else if (client_type == TYPE_OLD_LEADER) {
 278                 listenfd = tcp_listen_connections(DEFAULT_LEADER_LISTEN_PORT,
 279                                 DEFAULT_SERVER_BACKLOG);
 280                 DIE(listenfd < 0, "tcp_listen_connections");
 281
 282                 connectfd = tcp_connect_to_server(DEFAULT_FOLLOWER_HOSTNAME,
 283                                 DEFAULT_FOLLOWER_LISTEN_PORT);
 284                 DIE(connectfd < 0, "tcp_connect_to_server");
 285
 286                 sockfd = accept(listenfd, (SSA *) &addr, &addrlen);
 287                 DIE(sockfd < 0, "accept");
 288         }
 289         else {
 290                 usage(argv[0]);
 291                 exit(EXIT_FAILURE);
 292         }
 293
 294         init_buffers();
 295         schedule_timer();
 296
 297         while (1) {
 298                 ssize_t nbytes;
 299
 300                 nbytes = receive_buffer(sockfd);
 301                 DIE(nbytes < 0, "receive_buffer");
 302                 if (nbytes == 0) {
 303                         close(sockfd);
 304                         remove_timer();
 305
 306                         sockfd = accept(listenfd, (SSA *) &addr, &addrlen);
 307                         DIE(sockfd < 0, "accept");
 308
 309                         if (client_type == TYPE_FOLLOWER) {
 310                                 connectfd = tcp_connect_to_server(DEFAULT_LEADER_HOSTNAME, DEFAULT_LEADER_LISTEN_PORT);
 311                                 DIE(connectfd < 0, "tcp_connect_to_server");
 312                         }
 313                         if (client_type == TYPE_LEADER || client_type == TYPE_OLD_LEADER) {
 314                                 connectfd = tcp_connect_to_server(DEFAULT_FOLLOWER_HOSTNAME, DEFAULT_FOLLOWER_LISTEN_PORT);
 315                                 DIE(connectfd < 0, "tcp_connect_to_server");
 316                         }
 317                         schedule_timer();
 318                 }
 319
 320                 print_buffer_meta();
 321         }
 322
 323         close(sockfd);
 324         close(connectfd);
 325
 326 #endif
 327
 328         return 0;
 329 }