CODEWITHSUNDEEP
pythonasynchronousaws-lambdaboto3threadpoolexecutor
Falpangaea
Falpangaea

Reputation: 191

Async, Lambda, Boto3, ThreadPoolExecutor: Connection pool is full, discarding connection

Some Context:

I'm developing a test suite for updating and testing global redirects using CodePipeline and two lambda functions: a wrapper function and a tester function.

The wrapper loops through a list of domains. For each domain, it grabs the associated S3 file, then creates a long list of objects (one object per line in the file). Example object:

redirect = {
     'staging': None,
     'prod': None,
     'country': 'US',
     'language': 'EN',
     'status_code': None,
     'shortlink': None,
     'expected': None
}

The wrapper then takes this list of objects and calls the second lambda (the tester lambda) which sends a series of httpx requests for each object etc. etc.

The most important piece of this is that it is asynchronous - otherwise, the lambda will time out at the 15 minute mark.

The Actual Issue:

The wrapper function can only trigger the tester 10 times. Otherwise, when the result is returned, I get this error:

Connection pool is full, discarding connection: lambda.us-east-1.amazonaws.com

I've bumped up the max_workers for ThreadPoolExecutor and even set max_workers for boto3. I spoke to AWS support and my actual lambda settings are conducive to my tester lambda being triggered as many times as needed. And, yet, I'm still getting the connection pool error.

Relevant Code:

from botocore.client import Config
from boto3.session import Session
from concurrent.futures import ThreadPoolExecutor, as_completed
import boto3

max_pool_connections = 30

config = Config(
    max_pool_connections=max_pool_connections,
    read_timeout = 900
)

def handler(event, context):
    try:
        events = ... # generated via a bunch of nonsense 

        with ThreadPoolExecutor(max_pool_connections) as executor:
            futures = []
            for event in events:
                print(event)
                future = executor.submit(lambda_client.invoke, FunctionName = "site-tester", InvocationType = "RequestResponse", Payload = json.dumps(event))
                futures.append(future)

            for index, future in enumerate(as_completed(futures, timeout=None), start=1):
                ...

This is (obviously) not the full code because the full code is an absolute train wreck of nonsense. I don't have time at the moment to create a full test function but if anyone has any initial thoughts or troubleshooting tips it would be greatly appreciated.

I do want to note, that for the 10 events that do get called and return, everything is working as expected. It's just limited to 10 times and I definitely need it to be more than that.

AGAIN the most important piece of this is that it is asynchronous - otherwise, the lambda will time out at the 15 minute mark.

Concurrency Settings

wrapper function: reserved concurrency set to 30

tester function: reserved concurrency set to 30

Upvotes: 9

Views: 14278

Answers (2)

Sviat Lavrinchuk
Sviat Lavrinchuk

Reputation: 404

As mentioned by JustinTArthur on Github:

This warning is fine. If you dig into the urllib3 connection pool code, it's basically the pool of persistent connections and not the maximum number of concurrent connections you can have. The connections in the pool are re-used if they haven't been closed by the endpoint.

If you'd like to boost the size of this pool, it can be done per-endpoint using the low-level botocore config. boto3 example:

import boto3
import botocore

client_config = botocore.config.Config(
    max_pool_connections=25,
)
boto3.client('lambda', config=client_config)

or

s3_client = boto3.client('s3', config=botocore.client.Config(max_pool_connections=50))

Upvotes: 12

HTF
HTF

Reputation: 7290

So the warning itself comes from the urllib3 library that boto3 is using to make the HTTP requests.

The max_pool_connections config option sets maxsize for the ConnectionPool class.

You can try to increase this option further to see if this will help but don't add more thread workers.

It's not clear how you create the client from your code snipped but it looks like resources are not threads safe and a separate one should be created for each thread / process:

Multithreading and multiprocessing:

Note

Resources are not thread safe. These special classes contain additional meta data that cannot be shared between threads. When using a Resource, it is recommended to instantiate a new Resource for each thread, as is shown in the example above.

Low-level clients are thread safe. When using a low-level client, it is recommended to instantiate your client then pass that client object to each of your threads.

Upvotes: 5

Related Questions