LXR linux/tools/testing/selftests/bpf/tcp

   1#!/usr/bin/env python3
   2#
   3# SPDX-License-Identifier: GPL-2.0
   4#
   5
   6import sys, os, os.path, getopt
   7import socket, time
   8import subprocess
   9import select
  10
  11def read(sock, n):
  12    buf = b''
  13    while len(buf) < n:
  14        rem = n - len(buf)
  15        try: s = sock.recv(rem)
  16        except (socket.error) as e: return b''
  17        buf += s
  18    return buf
  19
  20def send(sock, s):
  21    total = len(s)
  22    count = 0
  23    while count < total:
  24        try: n = sock.send(s)
  25        except (socket.error) as e: n = 0
  26        if n == 0:
  27            return count;
  28        count += n
  29    return count
  30
  31
  32serverPort = int(sys.argv[1])
  33HostName = socket.gethostname()
  34
  35# create active socket
  36sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
  37try:
  38    sock.connect((HostName, serverPort))
  39except socket.error as e:
  40    sys.exit(1)
  41
  42buf = b''
  43n = 0
  44while n < 1000:
  45    buf += b'+'
  46    n += 1
  47
  48sock.settimeout(1);
  49n = send(sock, buf)
  50n = read(sock, 500)
  51sys.exit(0)
  52