-
Notifications
You must be signed in to change notification settings - Fork 0
/
example.py
59 lines (42 loc) · 1.25 KB
/
example.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import asyncio
import time
import requests
from railgun.railgun import Railgun
async def random_task_async(num, times):
_ = num * times
print(_)
await asyncio.sleep(times)
return _
def random_task(num, times):
_ = num * times
time.sleep(times)
return _
def call_web(url='http://www.google.com'):
return requests.get(url=url).status_code
async def call_web_async(url='http://www.google.com'):
status_code = requests.get(url=url).status_code
print(status_code)
await asyncio.sleep(0.1)
return status_code
# tasks = [random_task(5,10), random_task(1,1), random_task(2,5), random_task_async(6,3)]
tasks = [call_web()] * 100
print('before railgun')
rail_gun = Railgun(semaphores_count=500)
# time.sleep(3)
print('After rail gun')
# test = rail_gun.run(tasks)
print('After rail gun run')
# print(test)
# print(f'len of returns {len(test)}')
loop = asyncio.get_event_loop()
async_tasks_1 = []
async_tasks_2 = []
start = time.time()
# for x in range(0, 299):
# async_tasks_1.append(call_web_async())
# async_tasks_2.append(call_web_async())
print(f'after for loop: {time.time() - start}')
async_test_3 = rail_gun.repeat(random_task, [12, 0.01], 200)
print(async_test_3)
print(len(async_test_3))
start = time.time()