Snippet content copied to clipboard.
Are you sure to delete this snippet? No, don't delete
  1. from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
  2. import concurrent.futures
  3. import random
  4. import statistics
  5. import string
  6. import sys
  7. import threading
  8. import time
  9. ​
  10. import requests
  11. ​
  12. HOST = 'http://localhost:9000'
  13. TIMEOUT_s = 1
  14. NUM_REQUESTS = 1000
  15. NUM_CONCURRENT = 5
  16. ​
  17. ​
  18. TIMES = []
  19. ​
  20. ​
  21. def send_request(idx, url, data, headers):
  22. try:
  23. time_before = time.perf_counter()
  24. response = requests.post(
  25. url,
  26. json=data,
  27. headers=headers,
  28. timeout=TIMEOUT_s
  29. )
  30. time_after = time.perf_counter()
  31. if response.status_code != 200:
  32. print(f'Request {idx} failed.')
  33. raise Exception('{response.status_code} NOT 200')
  34. time_duration = time_after - time_before
  35. TIMES.append(time_duration)
  36. except (requests.exceptions.RequestException, concurrent.futures.TimeoutError) as exc:
  37. print(f'Request {idx} failed.')
  38. raise exc
  39. ​
  40. ​
  41. def load_test(url, data_list):
  42. headers = get_headers()
  43. ​
  44. semaphore = threading.BoundedSemaphore(NUM_CONCURRENT)
  45. ​
  46. # Use ThreadPoolExecutor to send requests concurrently
  47. with ThreadPoolExecutor(max_workers=NUM_CONCURRENT) as executor:
  48. # Loop to submit and wait for the requests
  49. for idx, data in enumerate(data_list , 1):
  50. # Acquire the semaphore before submitting the request
  51. success = semaphore.acquire(timeout=TIMEOUT_s)
  52. if not success:
  53. # timeout reached
  54. break
  55. ​
  56. def send_one_request():
  57. send_request(idx, url, data, headers)
  58. semaphore.release()
  59. executor.submit(send_one_request)
  60. ​
  61. print('Done.')
  62. ​
  63. ​
  64. print(f'{len(TIMES)} requests succeeded.')
  65. try:
  66. time_mean = statistics.mean(TIMES)
  67. print(f'Mean: {time_mean:.4f}')
  68. except statistics.StatisticsError:
  69. time_mean = None
  70. print(f'Mean: {time_mean}')
  71. try:
  72. time_max = max(TIMES)
  73. print(f'Max: {time_max:.4f}')
  74. except ValueError:
  75. time_max = 0
  76. print(f'Max: {time_max}')
  77. ​
  78. try:
  79. time_max_index = TIMES.index(time_max)
  80. except ValueError:
  81. time_max_index = None
  82. print(f'Max index: {time_max_index}')

Edit this Snippet