From 9417961b171f772caa965d773f866002daa60266 Mon Sep 17 00:00:00 2001
From: Nuno Campos <nuno@boringbits.io>
Date: Fri, 18 Aug 2023 14:20:09 +0100
Subject: [PATCH] Add lock on tee peer cleanup (#9446)

<!-- Thank you for contributing to LangChain!

Replace this entire comment with:
  - Description: a description of the change,
  - Issue: the issue # it fixes (if applicable),
  - Dependencies: any dependencies required for this change,
- Tag maintainer: for a quicker response, tag the relevant maintainer
(see below),
- Twitter handle: we announce bigger features on Twitter. If your PR
gets announced and you'd like a mention, we'll gladly shout you out!

Please make sure your PR is passing linting and testing before
submitting. Run `make format`, `make lint` and `make test` to check this
locally.

See contribution guidelines for more information on how to write/run
tests, lint, etc:

https://github.com/hwchase17/langchain/blob/master/.github/CONTRIBUTING.md

If you're adding a new integration, please include:
1. a test for the integration, preferably unit tests that do not rely on
network access,
2. an example notebook showing its use. These live is docs/extras
directory.

If no one reviews your PR within a few days, please @-mention one of
@baskaryan, @eyurtsev, @hwchase17, @rlancemartin.
 -->
---
 libs/langchain/langchain/utils/aiter.py | 17 +++++++++--------
 libs/langchain/langchain/utils/iter.py  | 17 +++++++++--------
 2 files changed, 18 insertions(+), 16 deletions(-)

diff --git a/libs/langchain/langchain/utils/aiter.py b/libs/langchain/langchain/utils/aiter.py
index a71650cdb4..ca44dee395 100644
--- a/libs/langchain/langchain/utils/aiter.py
+++ b/libs/langchain/langchain/utils/aiter.py
@@ -107,14 +107,15 @@ async def tee_peer(
                             peer_buffer.append(item)
             yield buffer.popleft()
     finally:
-        # this peer is done – remove its buffer
-        for idx, peer_buffer in enumerate(peers):  # pragma: no branch
-            if peer_buffer is buffer:
-                peers.pop(idx)
-                break
-        # if we are the last peer, try and close the iterator
-        if not peers and hasattr(iterator, "aclose"):
-            await iterator.aclose()
+        async with lock:
+            # this peer is done – remove its buffer
+            for idx, peer_buffer in enumerate(peers):  # pragma: no branch
+                if peer_buffer is buffer:
+                    peers.pop(idx)
+                    break
+            # if we are the last peer, try and close the iterator
+            if not peers and hasattr(iterator, "aclose"):
+                await iterator.aclose()
 
 
 class Tee(Generic[T]):
diff --git a/libs/langchain/langchain/utils/iter.py b/libs/langchain/langchain/utils/iter.py
index 498ccfbf73..1b95f180ea 100644
--- a/libs/langchain/langchain/utils/iter.py
+++ b/libs/langchain/langchain/utils/iter.py
@@ -60,14 +60,15 @@ def tee_peer(
                             peer_buffer.append(item)
             yield buffer.popleft()
     finally:
-        # this peer is done – remove its buffer
-        for idx, peer_buffer in enumerate(peers):  # pragma: no branch
-            if peer_buffer is buffer:
-                peers.pop(idx)
-                break
-        # if we are the last peer, try and close the iterator
-        if not peers and hasattr(iterator, "close"):
-            iterator.close()
+        with lock:
+            # this peer is done – remove its buffer
+            for idx, peer_buffer in enumerate(peers):  # pragma: no branch
+                if peer_buffer is buffer:
+                    peers.pop(idx)
+                    break
+            # if we are the last peer, try and close the iterator
+            if not peers and hasattr(iterator, "close"):
+                iterator.close()
 
 
 class Tee(Generic[T]):