pycurl/tests/test.py

   1 #! /usr/bin/env python
   2 # -*- coding: iso-8859-1 -*-
   3 # vi:ts=4:et
   4 # $Id$
   5
   6 import sys, threading, time
   7 import pycurl
   8
   9 # We should ignore SIGPIPE when using pycurl.NOSIGNAL - see
  10 # the libcurl tutorial for more info.
  11 try:
  12     import signal
  13     from signal import SIGPIPE, SIG_IGN
  14     signal.signal(signal.SIGPIPE, signal.SIG_IGN)
  15 except ImportError:
  16     pass
  17
  18
  19 class Test(threading.Thread):
  20     def __init__(self, url, ofile):
  21         threading.Thread.__init__(self)
  22         self.curl = pycurl.Curl()
  23         self.curl.setopt(pycurl.URL, url)
  24         self.curl.setopt(pycurl.WRITEDATA, ofile)
  25         self.curl.setopt(pycurl.FOLLOWLOCATION, 1)
  26         self.curl.setopt(pycurl.MAXREDIRS, 5)
  27         self.curl.setopt(pycurl.NOSIGNAL, 1)
  28
  29     def run(self):
  30         self.curl.perform()
  31         self.curl.close()
  32         sys.stdout.write(".")
  33         sys.stdout.flush()
  34
  35
  36 # Read list of URIs from file specified on commandline
  37 try:
  38     urls = open(sys.argv[1]).readlines()
  39 except IndexError:
  40     # No file was specified, show usage string
  41     print "Usage: %s <file with uris to fetch>" % sys.argv[0]
  42     raise SystemExit
  43
  44 # Initialize thread array and the file number
  45 threads = []
  46 fileno = 0
  47
  48 # Start one thread per URI in parallel
  49 t1 = time.time()
  50 for url in urls:
  51     f = open(str(fileno), "wb")
  52     t = Test(url, f)
  53     t.start()
  54     threads.append((t, f))
  55     fileno = fileno + 1
  56 # Wait for all threads to finish
  57 for thread, file in threads:
  58     thread.join()
  59     file.close()
  60 t2 = time.time()
  61 print "\n** Multithreading, %d seconds elapsed for %d uris" % (int(t2-t1), len(urls))
  62
  63 # Start one thread per URI in sequence
  64 fileno = 0
  65 t1 = time.time()
  66 for url in urls:
  67     f = open(str(fileno), "wb")
  68     t = Test(url, f)
  69     t.start()
  70     fileno = fileno + 1
  71     t.join()
  72     f.close()
  73 t2 = time.time()
  74 print "\n** Singlethreading, %d seconds elapsed for %d uris" % (int(t2-t1), len(urls))