Large wait time after concurrent.futures closes

Viewed 126

I have a list of URLs which I am trying to get JSON data from. I am going through a list of 10,000s items so speed is key for me. This took forever in serial processing so I opted for using sessions. I have not used it before, but I have been able to produce something that is fast enough.

ar_list=['https//:www.foo.com',...,'https//:www.foo2.com']
adapter = HTTPAdapter(pool_connections=workers_num, pool_maxsize=workers_num)
data_list = []
with sessions.FuturesSession(max_workers=workers_num) as session:
    session.mount('http://', adapter)
    session.mount('https://', adapter)  
    futures = []
    for idx, ar_url in enumerate(ar_list):
        resp = session.get(ar_url,headers=headers)
        futures.append(resp)
    end_time = datetime.datetime.now()
    time_delta=end_time-start_time

This works nicely and runs a query within about 70us (with 10,000 queries). I unfortunately have a problem getting the data. futures.append(resp) gives me a Future object, and I need to run .result().json() on this to actually get usable information out. However when I add these, I get extremely slow operation speeds at about 0.4s per query (much slower than 70us!)

I have tried running the FutureSession quickly (as the code shows above), and then performing actions on the futures list later using:

  1. for loop
  2. list comprehension: [item.result().json() for item in as_completed(futures)]
  3. lambda list operation: list(map(lambda i: func(futures, i), range(0, len(futures))))
  4. finally trying another session to perform the operation:
data=[]
with concurrent.futures.ThreadPoolExecutor() as e:
    fut = [e.submit(func, i) for i in futures]
    for idx,r in enumerate(concurrent.futures.as_completed(fut)):
        print(str(idx))
        data.append(r.result())

Now this ThreadPoolExecutor works really well also, performing the task around 70us or so. However, I am measuring at the start and end of with and notice that there is a large delay between these two with structures. It looks like this all together:


with sessions.FuturesSession(max_workers=workers_num) as session:
    #some fast operations here

"""
A VERY LARGE DELAY (about 60 seconds)
"""
with concurrent.futures.ThreadPoolExecutor() as e:
    #some fast operations here

Is there some cleanup of all of the sessions I create between the two with conditions that take a long time to resolve? I can see that there must be a better way of doing this than two sessions processes to just get some json data formatted correctly, but I can't seem to find the faster way.

1 Answers

After a lot of research reagrding this, I have discovered the big wait time that I experience is the as_completed call. As @Simon says, the large wait time is simply me waiting for the responses. If I remove as_completed it becomes as fast as ever, but of course the reponses are garbage or None.

I am looking into avoiding the use of as_completed with a callback to avoid the need to wait for as_completed to occur on the futures, and just get them from a completed list. Hopefully I return soon with a "work around"

Related