threadsafety.py 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566
  1. # Copyright 2009 Jason Stitt
  2. #
  3. # Permission is hereby granted, free of charge, to any person obtaining a copy
  4. # of this software and associated documentation files (the "Software"), to deal
  5. # in the Software without restriction, including without limitation the rights
  6. # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  7. # copies of the Software, and to permit persons to whom the Software is
  8. # furnished to do so, subject to the following conditions:
  9. #
  10. # The above copyright notice and this permission notice shall be included in
  11. # all copies or substantial portions of the Software.
  12. #
  13. # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  14. # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  15. # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  16. # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  17. # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  18. # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  19. # THE SOFTWARE.
  20. import threading
  21. from Queue import Queue
  22. from tidylib import tidy_document
  23. error_queue = Queue()
  24. DOC = '''<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN"
  25. "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd">
  26. <html xmlns="http://www.w3.org/1999/xhtml">
  27. <head>
  28. <title></title>
  29. </head>
  30. <body>
  31. hello, world
  32. </body>
  33. </html>
  34. '''
  35. SAMPLE = "hello, world"
  36. NUM_THREADS = 100
  37. NUM_TRIES = 100
  38. class TidyingThread(threading.Thread):
  39. def run(self):
  40. for x in xrange(NUM_TRIES):
  41. output, errors = tidy_document(SAMPLE, keep_doc=True)
  42. if output != DOC:
  43. error_queue.put(output)
  44. def run_test():
  45. threads = []
  46. for i in xrange(NUM_THREADS):
  47. t = TidyingThread()
  48. threads.append(t)
  49. t.start()
  50. for t in threads:
  51. t.join()
  52. if __name__ == '__main__':
  53. run_test()
  54. if not error_queue.empty():
  55. print "About %s errors out of %s" % (error_queue.qsize(), NUM_THREADS * NUM_TRIES)
  56. print error_queue.get()