threadsafety.py 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566
  1. # Copyright 2009-2014 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 HTML 4.01//EN">
  25. <html>
  26. <head>
  27. <title></title>
  28. </head>
  29. <body>
  30. hello, world
  31. </body>
  32. </html>
  33. '''
  34. SAMPLE = "hello, world"
  35. NUM_THREADS = 100
  36. NUM_TRIES = 100
  37. class TidyingThread(threading.Thread):
  38. def run(self):
  39. for x in xrange(NUM_TRIES):
  40. output, errors = tidy_document(SAMPLE, keep_doc=True)
  41. if output != DOC:
  42. error_queue.put(output)
  43. def run_test():
  44. threads = []
  45. for i in xrange(NUM_THREADS):
  46. t = TidyingThread()
  47. threads.append(t)
  48. t.start()
  49. for t in threads:
  50. t.join()
  51. if __name__ == '__main__':
  52. run_test()
  53. if not error_queue.empty():
  54. print("About %s errors out of %s" % (error_queue.qsize(), NUM_THREADS * NUM_TRIES))
  55. print(error_queue.get())