hstt/main.py
async def worker_loop(
    args: argparse.Namespace, tasks: Queue, results: Queue, start_event: Event, worker_number: int,
) -> None:
    """Make actual requests in loop."""
    try:
        # trace for tracking time different times inside requests
        async def trace_call(name: str, session: ClientSession, context: SimpleNamespace, params: Any) -> None:
            if name not in context.trace_request_ctx:
                context.trace_request_ctx[name] = time()
        trace = TraceConfig()
        trace.on_connection_create_start.append(partial(trace_call, 'conn_start'))  # noqa
        trace.on_connection_create_end.append(partial(trace_call, 'conn_end'))  # noqa
        trace.on_request_start.append(partial(trace_call, 'req_start'))  # noqa
        trace.on_request_end.append(partial(trace_call, 'req_end'))  # noqa
        trace.on_request_chunk_sent.append(partial(trace_call, 'chunk_sent'))  # noqa
        trace.on_response_chunk_received.append(partial(trace_call, 'chunk_received'))  # noqa

        # wait until all threads will be initialized
        if not start_event.wait(timeout=10):
            return

        if not args.no_reuse:
            # common session for all requests
            session = ClientSession(
                timeout=ClientTimeout(total=args.t),
                skip_auto_headers=('User-Agent',),
                trace_configs=[trace],
            )
        while True:
            # throttle requests
            if worker_number >= enabled_threads:
                sleep(1)
                continue
            if request_delay:
                sleep(request_delay)

            if args.no_reuse:
                # new session for each request
                session = ClientSession(
                    timeout=ClientTimeout(total=args.t),
                    skip_auto_headers=('User-Agent',),
                    trace_configs=[trace],
                )
            else:
                # do not reuse cookies with common session
                session.cookie_jar.clear()
            try:
                url = tasks.get(block=True, timeout=1)
            except Empty:
                # no more tasks, stop worker
                await session.close()
                return
            try:
                context: Dict[str, float] = {}
                headers: Dict[str, str] = {}
                if args.H:
                    headers = dict(map(str.strip, h.split(':', 1)) for h in args.H)  # noqa
                if args.chrome:
                    headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36'  # noqa: E501
                response = await session.request(
                    method=args.m,
                    url=url,
                    data=args.b,
                    headers=headers,
                    allow_redirects=False,
                    ssl=False if args.insecure else None,
                    trace_request_ctx=context,
                )
                body = await response.read()
            except asyncio.TimeoutError:
                result = Result(timestamp=time(), error='timeout')
            except Exception as e:
                result = Result(timestamp=time(), error=e.__class__.__name__)
            else:
                result = Result(
                    timestamp=time(),
                    status=response.status,
                    size=len(body),
                    total_time=context.get('req_end', 0) - context.get('req_start', 0),
                    conn_time=context.get('conn_end', 0) - context.get('conn_start', 0),
                    ttfb_time=context.get('chunk_received', 0) - context.get('chunk_sent', 0),
                )
            results.put(result, block=True)
            if args.no_reuse:
                await session.close()
    except KeyboardInterrupt:
        exit(1)
    except Exception as e:
        print(repr(e))
Make actual requests in loop.
Make actual requests in loop.
Make actual requests in loop .
async def worker_loop(
    args: argparse.Namespace, tasks: Queue, results: Queue, start_event: Event, worker_number: int,
) -> None:
    try:
        async def trace_call(name: str, session: ClientSession, context: SimpleNamespace, params: Any) -> None:
            if name not in context.trace_request_ctx:
                context.trace_request_ctx[name] = time()
        trace = TraceConfig()
        trace.on_connection_create_start.append(partial(trace_call, 'conn_start'))  
        trace.on_connection_create_end.append(partial(trace_call, 'conn_end'))  
        trace.on_request_start.append(partial(trace_call, 'req_start'))  
        trace.on_request_end.append(partial(trace_call, 'req_end'))  
        trace.on_request_chunk_sent.append(partial(trace_call, 'chunk_sent'))  
        trace.on_response_chunk_received.append(partial(trace_call, 'chunk_received'))  
        if not start_event.wait(timeout=10):
            return
        if not args.no_reuse:
            session = ClientSession(
                timeout=ClientTimeout(total=args.t),
                skip_auto_headers=('User-Agent',),
                trace_configs=[trace],
            )
        while True:
            if worker_number >= enabled_threads:
                sleep(1)
                continue
            if request_delay:
                sleep(request_delay)
            if args.no_reuse:
                session = ClientSession(
                    timeout=ClientTimeout(total=args.t),
                    skip_auto_headers=('User-Agent',),
                    trace_configs=[trace],
                )
            else:
                session.cookie_jar.clear()
            try:
                url = tasks.get(block=True, timeout=1)
            except Empty:
                await session.close()
                return
            try:
                context: Dict[str, float] = {}
                headers: Dict[str, str] = {}
                if args.H:
                    headers = dict(map(str.strip, h.split(':', 1)) for h in args.H)  
                if args.chrome:
                    headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36'  : E501
                response = await session.request(
                    method=args.m,
                    url=url,
                    data=args.b,
                    headers=headers,
                    allow_redirects=False,
                    ssl=False if args.insecure else None,
                    trace_request_ctx=context,
                )
                body = await response.read()
            except asyncio.TimeoutError:
                result = Result(timestamp=time(), error='timeout')
            except Exception as e:
                result = Result(timestamp=time(), error=e.__class__.__name__)
            else:
                result = Result(
                    timestamp=time(),
                    status=response.status,
                    size=len(body),
                    total_time=context.get('req_end', 0) - context.get('req_start', 0),
                    conn_time=context.get('conn_end', 0) - context.get('conn_start', 0),
                    ttfb_time=context.get('chunk_received', 0) - context.get('chunk_sent', 0),
                )
            results.put(result, block=True)
            if args.no_reuse:
                await session.close()
    except KeyboardInterrupt:
        exit(1)
    except Exception as e:
        print(repr(e))
async def worker_loop ( args : argparse . Namespace , tasks : Queue , results : Queue , start_event : Event , worker_number : int , ) -> None : try : async def trace_call ( name : str , session : ClientSession , context : SimpleNamespace , params : Any ) -> None : if name not in context . trace_request_ctx : context . trace_request_ctx [ name ] = time ( ) trace = TraceConfig ( ) trace . on_connection_create_start . append ( partial ( trace_call , 'conn_start' ) ) trace . on_connection_create_end . append ( partial ( trace_call , 'conn_end' ) ) trace . on_request_start . append ( partial ( trace_call , 'req_start' ) ) trace . on_request_end . append ( partial ( trace_call , 'req_end' ) ) trace . on_request_chunk_sent . append ( partial ( trace_call , 'chunk_sent' ) ) trace . on_response_chunk_received . append ( partial ( trace_call , 'chunk_received' ) ) if not start_event . wait ( timeout = 10 ) : return if not args . no_reuse : session = ClientSession ( timeout = ClientTimeout ( total = args . t ) , skip_auto_headers = ( 'User-Agent' , ) , trace_configs = [ trace ] , ) while True : if worker_number >= enabled_threads : sleep ( 1 ) continue if request_delay : sleep ( request_delay ) if args . no_reuse : session = ClientSession ( timeout = ClientTimeout ( total = args . t ) , skip_auto_headers = ( 'User-Agent' , ) , trace_configs = [ trace ] , ) else : session . cookie_jar . clear ( ) try : url = tasks . get ( block = True , timeout = 1 ) except Empty : await session . close ( ) return try : context : Dict [ str , float ] = { } headers : Dict [ str , str ] = { } if args . H : headers = dict ( map ( str . strip , h . split ( ':' , 1 ) ) for h in args . H ) if args . chrome : headers [ 'User-Agent' ] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36' response = await session . request ( method = args . m , url = url , data = args . b , headers = headers , allow_redirects = False , ssl = False if args . insecure else None , trace_request_ctx = context , ) body = await response . read ( ) except asyncio . TimeoutError : result = Result ( timestamp = time ( ) , error = 'timeout' ) except Exception as e : result = Result ( timestamp = time ( ) , error = e . __class__ . __name__ ) else : result = Result ( timestamp = time ( ) , status = response . status , size = len ( body ) , total_time = context . get ( 'req_end' , 0 ) - context . get ( 'req_start' , 0 ) , conn_time = context . get ( 'conn_end' , 0 ) - context . get ( 'conn_start' , 0 ) , ttfb_time = context . get ( 'chunk_received' , 0 ) - context . get ( 'chunk_sent' , 0 ) , ) results . put ( result , block = True ) if args . no_reuse : await session . close ( ) except KeyboardInterrupt : exit ( 1 ) except Exception as e : print ( repr ( e ) )
Make actual requests in loop.
Make actual requests in loop .
"""Make actual requests in loop.""" # trace for tracking time different times inside requests # noqa # noqa # noqa # noqa # noqa # noqa # wait until all threads will be initialized # common session for all requests # throttle requests # new session for each request # do not reuse cookies with common session # no more tasks, stop worker # noqa # noqa: E501
args
argparse.Namespace
tasks
Queue
results
Queue
start_event
Event
worker_number
int