[Bug] api client multiprocessing usage incompatible with AWS lambda

[kazzmir]

Is this a new bug in the Pinecone Python client?

• I believe this is a new bug in the Pinecone Python Client
• I have searched the existing issues, and I could not find an existing issue for this bug

Current Behavior

Using the python api client ultimately results in an invocation of _multiprocessing.SemLock, which is not implemented on AWS lambda using a python3.11 runtime. Related stackoverflow: https://stackoverflow.com/questions/34005930/multiprocessing-semlock-is-not-implemented-when-running-on-aws-lambda

Expected Behavior

Using the python API client should work without issue on AWS lambda.

index = pinecone.Index("xyz")
index.insert(...)

Steps To Reproduce

1. create a docker container with pinecone-client[grpc]==2.2.4 installed, and a python script that uses the pinecone python client
2. upload container to AWS ECR
3. create AWS lambda using that container and invoke the lambda

Relevant log output

Here is roughly my stack trace (extracted from cloudwatch)

File "/var/lang/lib/python3.11/site-packages/langchain/schema/vectorstore.py", line 122, in add_documents
--
return self.add_texts(texts, metadatas, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/langchain/vectorstores/pinecone.py", line 138, in add_texts
async_res = [
^
File "/var/lang/lib/python3.11/site-packages/langchain/vectorstores/pinecone.py", line 139, in <listcomp>
self._index.upsert(
File "/var/lang/lib/python3.11/site-packages/pinecone/core/utils/error_handling.py", line 17, in inner_func
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/index.py", line 150, in upsert
return self._upsert_batch(vectors, namespace, _check_type, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/index.py", line 237, in _upsert_batch
return self._vector_api.upsert(
^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/core/client/api_client.py", line 776, in __call__
return self.callable(self, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/core/client/api/vector_operations_api.py", line 956, in __upsert
return self.call_with_http_info(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/core/client/api_client.py", line 838, in call_with_http_info
return self.api_client.call_api(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/core/client/api_client.py", line 421, in call_api
return self.pool.apply_async(self.__call_api, (resource_path,
^^^^^^^^^
File "/var/lang/lib/python3.11/site-packages/pinecone/core/client/api_client.py", line 107, in pool
self._pool = ThreadPool(self.pool_threads)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/multiprocessing/pool.py", line 930, in __init__
Pool.__init__(self, processes, initializer, initargs)
File "/var/lang/lib/python3.11/multiprocessing/pool.py", line 196, in __init__
self._change_notifier = self._ctx.SimpleQueue()
^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/multiprocessing/context.py", line 113, in SimpleQueue
return SimpleQueue(ctx=self.get_context())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/multiprocessing/queues.py", line 341, in __init__
self._rlock = ctx.Lock()
^^^^^^^^^^
File "/var/lang/lib/python3.11/multiprocessing/context.py", line 68, in Lock
return Lock(ctx=self.get_context())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/lang/lib/python3.11/multiprocessing/synchronize.py", line 169, in __init__
SemLock.__init__(self, SEMAPHORE, 1, 1, ctx=ctx)
File "/var/lang/lib/python3.11/multiprocessing/synchronize.py", line 57, in __init__
sl = self._semlock = _multiprocessing.SemLock(
^^^^^^^^^^^^^^^^^^^^^^^^^
OSError: [Errno 38] Function not implemented

### Environment

```markdown
- OS: linux
- Python: 3.11
- pinecone: 2.2.4

Additional Context

https://stackoverflow.com/questions/34005930/multiprocessing-semlock-is-not-implemented-when-running-on-aws-lambda

[kazzmir]

I haven't dug into the implementation of pinecone for the client that much, but maybe its possible to use a concurrent.futures.ThreadPoolExecutor instead of a multiprocessing.ThreadPool? Or maybe it can be an option for situations like this.

[kazzmir]

A little more context on this: multiprocessing is used in case the async_req argument is passed as true, as in index.upsert(..., async_req=True), which is how langchain uses pinecone.

https://github.com/langchain-ai/langchain/blob/a2d30428237695f076060dec881bae0258123775/libs/community/langchain_community/vectorstores/pinecone.py#L142

self._index.upsert(
                    vectors=batch,
                    namespace=namespace,
                    async_req=True,
                    **kwargs,
                )

[kazzmir]

I'm using this workaround for now. A class that wraps a ThreadPoolExecutor and sets the _pool field of Index to an instance of it.

class PineconePool(object):
    def __init__(self):
        self.pool = ThreadPoolExecutor()
        
    def apply_async(self, f, args):
        future = self.pool.submit(f, *args)
        future.get = future.result
        return future

index = pinecone.Index(PINECONE_INDEX)
index._pool = PineconePool()

[kazzmir]

Using these versions of the langchain libraries

langchain==0.1.17
langchain-pinecone==0.1.0
langchain-community==0.0.37
langchain-core==0.1.52

langchain no longer hard codes async_req=True, which allows my own code to pass in async_req=False now, so I can avoid the PineconePool workaround.

my code:

from langchain_pinecone import Pinecone
index = pinecone_client.Index(...)
store = Pinecone(index, ...)
store.add_documents(docs, async_req=False)

[daverigby]

Thanks for the update @kazzmir - glad it's working for you with the newer langchain version. Closing this issue now.