pktgen.cc 14.5 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/*
 * Copyright 2021 Max Planck Institute for Software Systems, and
 * National University of Singapore
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to
 * the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 * CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#include <arpa/inet.h>
Antoine Kaufmann's avatar
Antoine Kaufmann committed
26
27
28
29
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <pcap/pcap.h>
#include <unistd.h>
30
31
32
33
34
35
36

#include <cassert>
#include <climits>
#include <csignal>
#include <cstdio>
#include <cstdlib>
#include <cstring>
37
#include <string>
38
39
40
#include <unordered_map>
#include <vector>

41
#include <simbricks/base/cxxatomicfix.h>
42
extern "C" {
43
#include <simbricks/network/if.h>
44
45
46
#include <simbricks/nicif/nicif.h>
};

Antoine Kaufmann's avatar
Antoine Kaufmann committed
47
// #define NETSWITCH_DEBUG
48
49
#define NETSWITCH_STAT

50
struct SimbricksBaseIfParams netParams;
51
static pcap_dumper_t *dumpfile = nullptr;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
52
53
54
#define PKT_LEN 1500                                           // byte
static uint64_t bit_rate = 100 * 1000ULL * 1000ULL * 1000ULL;  // 100 Gbps
static uint64_t target_tick = 1 * 1000ULL * 1000ULL * 1000ULL * 1000ULL;  // 1s
55
56
57
58
59
static uint64_t last_pkt_sent = 0;
static uint64_t pkt_recv_num = 0;
static uint64_t pkt_recv_byte = 0;
static uint64_t pkt_tx_num = 0;
static uint64_t pkt_tx_byte = 0;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
60
static uint64_t period = (1E12 * 8 * PKT_LEN) / bit_rate;  // per packet
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
static uint8_t packet[PKT_LEN];

#ifdef NETSWITCH_STAT
#endif

#ifdef NETSWITCH_STAT
static uint64_t d2n_poll_total = 0;
static uint64_t d2n_poll_suc = 0;
static uint64_t d2n_poll_sync = 0;

static uint64_t s_d2n_poll_total = 0;
static uint64_t s_d2n_poll_suc = 0;
static uint64_t s_d2n_poll_sync = 0;

static int stat_flag = 0;
#endif

/* MAC address type */
struct MAC {
  const uint8_t *data;

  explicit MAC(const uint8_t *data) : data(data) {
  }

  bool operator==(const MAC &other) const {
    for (int i = 0; i < 6; i++) {
      if (data[i] != other.data[i]) {
        return false;
      }
    }
    return true;
  }
};

Antoine Kaufmann's avatar
Antoine Kaufmann committed
95
struct mac_addr {
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
  uint8_t addr[6];
};

namespace std {
template <>
struct hash<MAC> {
  size_t operator()(const MAC &m) const {
    size_t res = 0;
    for (int i = 0; i < 6; i++) {
      res = (res << 4) | (res ^ m.data[i]);
    }
    return res;
  }
};
}  // namespace std

/** Abstract base switch port */
class Port {
 public:
  enum RxPollState {
    kRxPollSuccess = 0,
    kRxPollFail = 1,
    kRxPollSync = 2,
  };
120

121
122
  struct mac_addr my_mac;
  struct mac_addr dest_mac;
123

124
125
126
127
128
129
  virtual ~Port() = default;

  virtual bool Connect(const char *path, int sync) = 0;
  virtual bool IsSync() = 0;
  virtual void Sync(uint64_t cur_ts) = 0;
  virtual uint64_t NextTimestamp() = 0;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
130
131
  virtual enum RxPollState RxPacket(const void *&data, size_t &len,
                                    uint64_t cur_ts) = 0;
132
133
134
135
136
137
138
  virtual void RxDone() = 0;
  virtual bool TxPacket(const void *data, size_t len, uint64_t cur_ts) = 0;
};

/** Normal network switch port (conneting to a NIC) */
class NetPort : public Port {
 protected:
139
140
141
  struct SimbricksNetIf netifObj_;
  struct SimbricksNetIf *netif_;
  volatile union SimbricksProtoNetMsg *rx_;
142
143
144
  int sync_;

 public:
145
146
147
148
  NetPort() : netif_(&netifObj_), rx_(nullptr), sync_(0) {
    memset(&netifObj_, 0, sizeof(netifObj_));
    memset(&my_mac, 0, sizeof(my_mac));
    memset(&dest_mac, 0, sizeof(dest_mac));
149
150
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
151
152
153
154
155
156
  NetPort(const NetPort &other)
      : netifObj_(other.netifObj_),
        netif_(&netifObj_),
        rx_(other.rx_),
        sync_(other.sync_) {
  }
157

Antoine Kaufmann's avatar
Antoine Kaufmann committed
158
  bool Connect(const char *path, int sync) override {
159
    sync_ = sync;
160
    return SimbricksNetIfInit(netif_, &netParams, path, &sync_) == 0;
161
162
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
163
  bool IsSync() override {
164
165
166
    return sync_;
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
167
168
169
  void Sync(uint64_t cur_ts) override {
    while (SimbricksNetIfOutSync(netif_, cur_ts)) {
    }
170
171
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
172
  uint64_t NextTimestamp() override {
173
    return SimbricksNetIfInTimestamp(netif_);
174
175
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
176
177
  enum RxPollState RxPacket(const void *&data, size_t &len,
                            uint64_t cur_ts) override {
178
179
    assert(rx_ == nullptr);

180
    rx_ = SimbricksNetIfInPoll(netif_, cur_ts);
181
182
183
    if (!rx_)
      return kRxPollFail;

184
185
186
187
    uint8_t type = SimbricksNetIfInType(netif_, rx_);
    if (type == SIMBRICKS_PROTO_NET_MSG_PACKET) {
      data = (const void *)rx_->packet.data;
      len = rx_->packet.len;
188
      return kRxPollSuccess;
189
    } else if (type == SIMBRICKS_PROTO_MSG_TYPE_SYNC) {
190
191
192
193
194
195
196
      return kRxPollSync;
    } else {
      fprintf(stderr, "switch_pkt: unsupported type=%u\n", type);
      abort();
    }
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
197
  void RxDone() override {
198
199
    assert(rx_ != nullptr);

200
    SimbricksNetIfInDone(netif_, rx_);
201
202
203
    rx_ = nullptr;
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
204
  bool TxPacket(const void *data, size_t len, uint64_t cur_ts) override {
205
    volatile union SimbricksProtoNetMsg *msg_to =
Antoine Kaufmann's avatar
Antoine Kaufmann committed
206
        SimbricksNetIfOutAlloc(netif_, cur_ts);
207
    if (!msg_to && !sync_) {
208
      return false;
209
210
211
212
213
214
    } else if (!msg_to && sync_) {
      while (!msg_to)
        msg_to = SimbricksNetIfOutAlloc(netif_, cur_ts);
    }
    volatile struct SimbricksProtoNetMsgPacket *rx;
    rx = &msg_to->packet;
215
216
217
218
    rx->len = len;
    rx->port = 0;
    memcpy((void *)rx->data, data, len);

219
    SimbricksNetIfOutSend(netif_, msg_to, SIMBRICKS_PROTO_NET_MSG_PACKET);
220
221
222
223
224
    return true;
  }
};

/** Hosting network switch port (connected to another network) */
225
class NetHostPort : public NetPort {
226
227
228
229
 protected:
  struct SimbricksNicIf nicif_;

 public:
230
231
  NetHostPort() {
    netif_ = &nicif_.net;
232
233
234
    memset(&nicif_, 0, sizeof(nicif_));
  }

235
236
237
  NetHostPort(const NetHostPort &other) : NetPort(other), nicif_(other.nicif_) {
    netif_ = &nicif_.net;
  }
238

Antoine Kaufmann's avatar
Antoine Kaufmann committed
239
  bool Connect(const char *path, int sync) override {
240
241
242
    sync_ = sync;
    std::string shm_path = path;
    shm_path += "-shm";
243
    struct SimbricksBaseIfParams params = netParams;
244
    params.sock_path = path;
245
    if (!sync)
246
      params.sync_mode = kSimbricksBaseIfSyncDisabled;
247
    int ret = SimbricksNicIfInit(&nicif_, shm_path.c_str(), &params, nullptr,
Antoine Kaufmann's avatar
Antoine Kaufmann committed
248
                                 nullptr);
249
    sync_ = SimbricksBaseIfSyncEnabled(&netif_->base);
250
251
252
    return ret == 0;
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
253
  bool IsSync() override {
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
    return sync_;
  }
};

/* Global variables */
static uint64_t cur_ts = 0;
static int exiting = 0;
static const uint8_t bcast[6] = {0xFF};
static const MAC bcast_addr(bcast);
static std::vector<Port *> ports;
static std::unordered_map<MAC, int> mac_table;

static void sigint_handler(int dummy) {
  exiting = 1;
}

static void sigusr1_handler(int dummy) {
  fprintf(stderr, "main_time = %lu\n", cur_ts);
}

#ifdef NETSWITCH_STAT
static void sigusr2_handler(int dummy) {
  stat_flag = 1;
}
#endif

static void forward_pkt(const void *pkt_data, size_t pkt_len, size_t port_id) {
  struct pcap_pkthdr ph;
  Port &dest_port = *ports[port_id];

  // log to pcap file if initialized
  if (dumpfile) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
286
287
288
289
290
291
    memset(&ph, 0, sizeof(ph));
    ph.ts.tv_sec = cur_ts / 1000000000000ULL;
    ph.ts.tv_usec = (cur_ts % 1000000000000ULL) / 1000ULL;
    ph.caplen = pkt_len;
    ph.len = pkt_len;
    pcap_dump((unsigned char *)dumpfile, &ph, (unsigned char *)pkt_data);
292
293
294
295
296
297
298
  }
  // print sending tick: [packet type] source_IP -> dest_IP len:

#ifdef NETSWITCH_DEBUG
  uint16_t eth_proto;
  struct ethhdr *hdr;
  struct iphdr *iph;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
299
  hdr = (struct ethhdr *)pkt_data;
300
301
302
  eth_proto = ntohs(hdr->h_proto);
  iph = (struct iphdr *)(hdr + 1);
  fprintf(stderr, "%20lu: ", cur_ts);
Antoine Kaufmann's avatar
Antoine Kaufmann committed
303
  if (eth_proto == ETH_P_IP) {
304
    fprintf(stderr, "[ IP] ");
Antoine Kaufmann's avatar
Antoine Kaufmann committed
305
306

  } else if (eth_proto == ETH_P_ARP) {
307
    fprintf(stderr, "[ARP] ");
Antoine Kaufmann's avatar
Antoine Kaufmann committed
308
  } else {
309
310
311
    fprintf(stderr, "unkwon eth type\n");
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
312
313
  fprintf(stderr, "%8X -> %8X len: %lu\n ", iph->saddr, iph->daddr,
          iph->tot_len + sizeof(struct ethhdr));
314
315
316
317
318
319
320
321
322
323
324
325
326
327
#endif

  if (!dest_port.TxPacket(pkt_data, pkt_len, cur_ts))
    fprintf(stderr, "forward_pkt: dropping packet on port %zu\n", port_id);
}

static void pollq(Port &port, size_t iport) {
  // poll N2D queue
  // send packet
  const void *pkt_data;
  size_t pkt_len;

#ifdef NETSWITCH_STAT
  d2n_poll_total += 1;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
328
  if (stat_flag) {
329
330
331
332
333
    s_d2n_poll_total += 1;
  }
#endif

#ifdef NETSWITCH_STAT
Antoine Kaufmann's avatar
Antoine Kaufmann committed
334
335
336
337
  d2n_poll_suc += 1;
  if (stat_flag) {
    s_d2n_poll_suc += 1;
  }
338
339
340
341
#endif

  enum Port::RxPollState poll = port.RxPacket(pkt_data, pkt_len, cur_ts);
  if (poll == Port::kRxPollFail) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
342
    return;  // do nothing
343
344
  }

Antoine Kaufmann's avatar
Antoine Kaufmann committed
345
346
  if (poll == Port::kRxPollSuccess) {
    // stat received bytes
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
    pkt_recv_num++;
    pkt_recv_byte += pkt_len;

    // Get MAC addresses
    // MAC dst((const uint8_t *)pkt_data), src((const uint8_t *)pkt_data + 6);
    // // MAC learning
    // if (!(src == bcast_addr)) {
    //   mac_table[src] = iport;
    // }
    // // L2 forwarding
    // auto i = mac_table.find(dst);
    // if (i != mac_table.end()) {
    //   size_t eport = i->second;
    //   forward_pkt(pkt_data, pkt_len, eport);
    // } else {
    //   // Broadcast
    //   for (size_t eport = 0; eport < ports.size(); eport++) {
    //     if (eport != iport) {
    //       // Do not forward to ingress port
    //       forward_pkt(pkt_data, pkt_len, eport);
    //     }
    //   }
    // }
  } else if (poll == Port::kRxPollSync) {
#ifdef NETSWITCH_STAT
    d2n_poll_sync += 1;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
373
    if (stat_flag) {
374
375
376
377
378
379
380
381
382
383
      s_d2n_poll_sync += 1;
    }
#endif
  } else {
    fprintf(stderr, "pktgen: unsupported poll result=%u\n", poll);
    abort();
  }
  port.RxDone();
}

Antoine Kaufmann's avatar
Antoine Kaufmann committed
384
385
386
387
static void sendq(Port &port, size_t iport) {
  // then send
  if (port.IsSync()) {
    while ((last_pkt_sent + period) <= cur_ts) {
388
389
390
391
392
      port.TxPacket(packet, PKT_LEN, last_pkt_sent + period);
      last_pkt_sent += period;
      pkt_tx_num++;
      pkt_tx_byte += PKT_LEN;
    }
Antoine Kaufmann's avatar
Antoine Kaufmann committed
393
  } else {
394
395
396
397
    port.TxPacket(packet, PKT_LEN, last_pkt_sent + period);
  }
  // if not sync: send packet
  // else: send packet periodically until allowed time
Antoine Kaufmann's avatar
Antoine Kaufmann committed
398
399
  // while(allowed_timestamp){ since_last_send + period = to_send_time <=
  // curtick
400
401
402
403
404
405
406
407
408
409
  //  txpacket(todest)
  // }
}

int main(int argc, char *argv[]) {
  int c;
  int bad_option = 0;
  int sync_eth = 1;
  pcap_t *pc = nullptr;
  int my_num = 0;
Hejing Li's avatar
Hejing Li committed
410
  int brate = 10;
411

412
413
  SimbricksNetIfDefaultParams(&netParams);

414
  // Parse command line argument
Hejing Li's avatar
Hejing Li committed
415
  while ((c = getopt(argc, argv, "s:h:uS:E:p:n:b:")) != -1 && !bad_option) {
416
417
418
    switch (c) {
      case 's': {
        NetPort *port = new NetPort;
Hejing Li's avatar
Hejing Li committed
419
        fprintf(stderr, "pktgen connecting to: %s\n", optarg);
420
421
422
423
424
425
426
427
428
429
        if (!port->Connect(optarg, sync_eth)) {
          fprintf(stderr, "connecting to %s failed\n", optarg);
          return EXIT_FAILURE;
        }
        ports.push_back(port);
        break;
      }

      case 'h': {
        NetHostPort *port = new NetHostPort;
Hejing Li's avatar
Hejing Li committed
430
        fprintf(stderr, "pktgen listening on: %s\n", optarg);
431
432
433
434
435
436
437
438
439
440
441
442
443
        if (!port->Connect(optarg, sync_eth)) {
          fprintf(stderr, "listening on %s failed\n", optarg);
          return EXIT_FAILURE;
        }
        ports.push_back(port);
        break;
      }

      case 'u':
        sync_eth = 0;
        break;

      case 'S':
444
        netParams.sync_interval = strtoull(optarg, NULL, 0) * 1000ULL;
445
446
447
        break;

      case 'E':
448
        netParams.link_latency = strtoull(optarg, NULL, 0) * 1000ULL;
449
450
451
452
453
454
        break;

      case 'p':
        pc = pcap_open_dead_with_tstamp_precision(DLT_EN10MB, 65535,
                                                  PCAP_TSTAMP_PRECISION_NANO);
        if (pc == nullptr) {
Antoine Kaufmann's avatar
Antoine Kaufmann committed
455
456
          perror("pcap_open_dead failed");
          return EXIT_FAILURE;
457
458
459
460
461
462
        }

        dumpfile = pcap_dump_open(pc, optarg);
        break;
      case 'n':
        my_num = strtol(optarg, NULL, 0);
Hejing Li's avatar
Hejing Li committed
463
464
        fprintf(stderr, "my_num is: %d\n", my_num);
        assert(my_num < 255);
465
466
        break;

Hejing Li's avatar
Hejing Li committed
467
468
469
      case 'b':
        brate = strtol(optarg, NULL, 0);
        fprintf(stderr, "bit rate set to: %d Gbps\n", brate);
Antoine Kaufmann's avatar
Antoine Kaufmann committed
470
        if (brate == 0) {
Hejing Li's avatar
Hejing Li committed
471
          period = ULLONG_MAX;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
472
473
474
        } else {
          bit_rate = brate * 1000ULL * 1000ULL * 1000ULL;
          period = (1E12 * 8 * PKT_LEN) / bit_rate;  // per packet
Hejing Li's avatar
Hejing Li committed
475
476
477
478
        }
        assert(brate < 200);
        break;

479
480
481
482
483
484
485
486
487
      default:
        fprintf(stderr, "unknown option %c\n", c);
        bad_option = 1;
        break;
    }
  }

  if (ports.empty() || bad_option) {
    fprintf(stderr,
Hejing Li's avatar
Hejing Li committed
488
489
            "Usage: pktgen [-S SYNC-PERIOD] [-E ETH-LATENCY] "
            "-s SOCKET-A [-s SOCKET-B ...] [-n my_num] [-b bitrate(GB)]\n");
490
491
492
493
494
    return EXIT_FAILURE;
  }

  Port *pkt_port = ports.front();
  pkt_port->my_mac.addr[5] = my_num;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
495
  if (my_num % 2) {  // odd num
496
    pkt_port->dest_mac.addr[5] = my_num - 1;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
497
  } else {  // even number
498
499
    pkt_port->dest_mac.addr[5] = my_num + 1;
  }
Antoine Kaufmann's avatar
Antoine Kaufmann committed
500
501
502
503
  struct mac_addr *mac_tmp = (struct mac_addr *)(&packet[0]);
  mac_tmp->addr[5] = pkt_port->dest_mac.addr[5];  // dest mac
  mac_tmp = (struct mac_addr *)(&packet[6]);
  mac_tmp->addr[5] = pkt_port->my_mac.addr[5];  // source mac
504
505

  int kk;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
506
  for (kk = 12; kk < PKT_LEN - 12; kk++) {
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
    packet[kk] = 0xFF;
  }

  signal(SIGINT, sigint_handler);
  signal(SIGTERM, sigint_handler);
  signal(SIGUSR1, sigusr1_handler);

#ifdef NETSWITCH_STAT
  signal(SIGUSR2, sigusr2_handler);
#endif

  printf("start polling\n");
  while (!exiting) {
    // Sync all interfaces
    for (auto port : ports)
      port->Sync(cur_ts);

    // Switch packets
    uint64_t min_ts;
    do {
      min_ts = ULLONG_MAX;
      for (size_t port_i = 0; port_i < ports.size(); port_i++) {
        auto &port = *ports[port_i];
        pollq(port, port_i);
        sendq(port, port_i);
        if (port.IsSync()) {
          uint64_t ts = port.NextTimestamp();
          min_ts = ts < min_ts ? ts : min_ts;
        }
      }
    } while (!exiting && (min_ts <= cur_ts));

    // Update cur_ts
    if (min_ts < ULLONG_MAX) {
      // a bit broken but should probably do
542
      cur_ts = min_ts;
Antoine Kaufmann's avatar
Antoine Kaufmann committed
543
      if (cur_ts >= target_tick) {
544
545
546
547
548
549
550
        printf("run to %lu tics\n", cur_ts);
        exiting = 1;
      }
    }
  }

#ifdef NETSWITCH_STAT
Antoine Kaufmann's avatar
Antoine Kaufmann committed
551
552
553
554
  fprintf(stderr, "sent packet: %20lu  [%20lu Byte]\n", pkt_tx_num,
          pkt_tx_byte);
  fprintf(stderr, "recv packet: %20lu  [%20lu Byte]\n", pkt_recv_num,
          pkt_recv_byte);
555
556
557
558
559

#endif

  return 0;
}