pycurl/tests/test_multi_vs_thread.py

   1 #! /usr/bin/env python
   2 # -*- coding: iso-8859-1 -*-
   3 # vi:ts=4:et
   4 # $Id: test_multi_vs_thread.py,v 1.15 2004/12/26 17:31:53 mfx Exp $
   5
   6 import os, sys, time
   7 from threading import Thread, RLock
   8 try:
   9     from cStringIO import StringIO
  10 except ImportError:
  11     from StringIO import StringIO
  12 import pycurl
  13
  14 # We should ignore SIGPIPE when using pycurl.NOSIGNAL - see
  15 # the libcurl tutorial for more info.
  16 try:
  17     import signal
  18     from signal import SIGPIPE, SIG_IGN
  19     signal.signal(signal.SIGPIPE, signal.SIG_IGN)
  20 except ImportError:
  21     pass
  22
  23 # The conclusion is: the multi interface is fastest!
  24
  25 NUM_PAGES = 30
  26 NUM_THREADS = 10
  27 assert NUM_PAGES % NUM_THREADS == 0
  28
  29 ##URL = "http://pycurl.sourceforge.net/tests/testgetvars.php?%d"
  30 URL = "http://pycurl.sourceforge.net/tests/teststaticpage.html?%d"
  31
  32
  33 #
  34 # util
  35 #
  36
  37 class Curl:
  38     def __init__(self, url):
  39         self.url = url
  40         self.body = StringIO()
  41         self.http_code = -1
  42         # pycurl API calls
  43         self._curl = pycurl.Curl()
  44         self._curl.setopt(pycurl.URL, self.url)
  45         self._curl.setopt(pycurl.WRITEFUNCTION, self.body.write)
  46         self._curl.setopt(pycurl.NOSIGNAL, 1)
  47
  48     def perform(self):
  49         self._curl.perform()
  50
  51     def close(self):
  52         self.http_code = self._curl.getinfo(pycurl.HTTP_CODE)
  53         self._curl.close()
  54
  55
  56 def print_result(items):
  57     return  # DO NOTHING
  58     #
  59     for c in items:
  60         data = c.body.getvalue()
  61         if 0:
  62             print "**********", c.url, "**********"
  63             print data
  64         elif 1:
  65             print "%-60s   %3d   %6d" % (c.url, c.http_code, len(data))
  66
  67
  68 ###
  69 ### 1) multi
  70 ###
  71
  72 def test_multi():
  73     clock1 = time.time()
  74
  75     # init
  76     handles = []
  77     m = pycurl.CurlMulti()
  78     for i in range(NUM_PAGES):
  79         c = Curl(URL %i)
  80         m.add_handle(c._curl)
  81         handles.append(c)
  82
  83     clock2 = time.time()
  84
  85     # stir state machine into action
  86     while 1:
  87         ret, num_handles = m.perform()
  88         if ret != pycurl.E_CALL_MULTI_PERFORM:
  89             break
  90
  91     # get data
  92     while num_handles:
  93         m.select()
  94         while 1:
  95             ret, num_handles = m.perform()
  96             if ret != pycurl.E_CALL_MULTI_PERFORM:
  97                 break
  98
  99     clock3 = time.time()
 100
 101     # close handles
 102     for c in handles:
 103         c.close()
 104     m.close()
 105
 106     clock4 = time.time()
 107     print "multi  interface:        %d pages: perform %5.2f secs, total %5.2f secs" % (NUM_PAGES, clock3 - clock2, clock4 - clock1)
 108
 109     # print result
 110     print_result(handles)
 111
 112
 113
 114 ###
 115 ### 2) thread
 116 ###
 117
 118 class Test(Thread):
 119     def __init__(self, lock=None):
 120         Thread.__init__(self)
 121         self.lock = lock
 122         self.items = []
 123
 124     def run(self):
 125         if self.lock:
 126             self.lock.acquire()
 127             self.lock.release()
 128         for c in self.items:
 129             c.perform()
 130
 131
 132 def test_threads(lock=None):
 133     clock1 = time.time()
 134
 135     # create and start threads, but block them
 136     if lock:
 137         lock.acquire()
 138
 139     # init (FIXME - this is ugly)
 140     threads = []
 141     handles = []
 142     t = None
 143     for i in range(NUM_PAGES):
 144         if i % (NUM_PAGES / NUM_THREADS) == 0:
 145             t = Test(lock)
 146             if lock:
 147                 t.start()
 148             threads.append(t)
 149         c = Curl(URL % i)
 150         t.items.append(c)
 151         handles.append(c)
 152     assert len(handles) == NUM_PAGES
 153     assert len(threads) == NUM_THREADS
 154
 155     clock2 = time.time()
 156
 157     #
 158     if lock:
 159         # release lock to let the blocked threads run
 160         lock.release()
 161     else:
 162         # start threads
 163         for t in threads:
 164             t.start()
 165     # wait for threads to finish
 166     for t in threads:
 167         t.join()
 168
 169     clock3 = time.time()
 170
 171     # close handles
 172     for c in handles:
 173         c.close()
 174
 175     clock4 = time.time()
 176     if lock:
 177         print "thread interface [lock]: %d pages: perform %5.2f secs, total %5.2f secs" % (NUM_PAGES, clock3 - clock2, clock4 - clock1)
 178     else:
 179         print "thread interface:        %d pages: perform %5.2f secs, total %5.2f secs" % (NUM_PAGES, clock3 - clock2, clock4 - clock1)
 180
 181     # print result
 182     print_result(handles)
 183
 184
 185
 186 ###
 187 ### 3) thread - threads grab curl objects on demand from a shared pool
 188 ###
 189
 190 class TestPool(Thread):
 191     def __init__(self, lock, pool):
 192         Thread.__init__(self)
 193         self.lock = lock
 194         self.pool = pool
 195
 196     def run(self):
 197         while 1:
 198             self.lock.acquire()
 199             c = None
 200             if self.pool:
 201                 c = self.pool.pop()
 202             self.lock.release()
 203             if c is None:
 204                 break
 205             c.perform()
 206
 207
 208 def test_thread_pool(lock):
 209     clock1 = time.time()
 210
 211     # init
 212     handles = []
 213     for i in range(NUM_PAGES):
 214         c = Curl(URL %i)
 215         handles.append(c)
 216
 217     # create and start threads, but block them
 218     lock.acquire()
 219     threads = []
 220     pool = handles[:]   # shallow copy of the list, shared for pop()
 221     for i in range(NUM_THREADS):
 222         t = TestPool(lock, pool)
 223         t.start()
 224         threads.append(t)
 225     assert len(pool) == NUM_PAGES
 226     assert len(threads) == NUM_THREADS
 227
 228     clock2 = time.time()
 229
 230     # release lock to let the blocked threads run
 231     lock.release()
 232
 233     # wait for threads to finish
 234     for t in threads:
 235         t.join()
 236
 237     clock3 = time.time()
 238
 239     # close handles
 240     for c in handles:
 241         c.close()
 242
 243     clock4 = time.time()
 244     print "thread interface [pool]: %d pages: perform %5.2f secs, total %5.2f secs" % (NUM_PAGES, clock3 - clock2, clock4 - clock1)
 245
 246     # print result
 247     print_result(handles)
 248
 249
 250
 251 lock = RLock()
 252 if 1:
 253     test_multi()
 254     test_threads()
 255     test_threads(lock)
 256     test_thread_pool(lock)
 257 else:
 258     test_thread_pool(lock)
 259     test_threads(lock)
 260     test_threads()
 261     test_multi()
 262