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])
  33
  34# create active socket
  35sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
  36try:
  37    sock.connect(('localhost', serverPort))
  38except socket.error as e:
  39    sys.exit(1)
  40
  41buf = b''
  42n = 0
  43while n < 1000:
  44    buf += b'+'
  45    n += 1
  46
  47sock.settimeout(1);
  48n = send(sock, buf)
  49n = read(sock, 500)
  50sys.exit(0)
  51