tunbench.py

   1 import os
   2 import sys
   3 import threading
   4 import time
   5 import cProfile
   6 import pstats
   7
   8 from nepi.util import tunchannel
   9
  10 remote = open("/dev/zero","r+b")
  11 tun = open("/dev/zero","r+b")
  12
  13 def rwrite(remote, packet, remote_fd = remote.fileno(), os_write=os.write, len=len):
  14     global bytes
  15     bytes += len(packet)
  16     return os_write(remote_fd, packet)
  17
  18 def rread(remote, maxlen, remote_fd = remote.fileno(), os_read=os.read):
  19     global bytes
  20     rv = os_read(remote_fd, maxlen)
  21     bytes += len(rv)
  22     return rv
  23
  24 def test(cipher, passphrase):
  25    TERMINATE = []
  26    def stopme():
  27        time.sleep(100)
  28        TERMINATE.append(None)
  29    t = threading.Thread(target=stopme)
  30    t.start()
  31    tunchannel.tun_fwd(tun, remote, True, True, passphrase, True, TERMINATE, None, tunkqueue=500,
  32         rwrite = rwrite, rread = rread, cipher=cipher)
  33
  34 # Swallow exceptions on decryption
  35 def decrypt(packet, crypter, super=tunchannel.decrypt):
  36     try:
  37         return super(packet, crypter)
  38     except:
  39         return packet
  40 tunchannel.decrypt = decrypt
  41
  42 for cipher in (None, 'AES', 'Blowfish', 'DES', 'DES3'):
  43     if cipher is None:
  44         passphrase = None
  45     else:
  46         passphrase = 'Abracadabra'
  47     bytes = 0
  48     cProfile.runctx('test(%r,%r)' % (cipher, passphrase),globals(),locals(),'tunchannel.%s.profile' % (cipher,))
  49
  50     print "Profile (%s):" % ( cipher, )
  51     pstats.Stats('tunchannel.%s.profile' % cipher).strip_dirs().sort_stats('time').print_stats()
  52
  53     print "Bandwidth (%s): %.4fMb/s" % ( cipher, bytes / 200.0 * 8 / 2**20, )
  54
  55