[2.7] smaller lock in produce item (#4174)

### Description

Do not hold the lock around produce_item. It is not needed and this operation can be slow. We do not want/need to hold up everying during this time.


### Types of changes
<!--- Put an `x` in all the boxes that apply, and remove the not
applicable items -->
- [x] Non-breaking change (fix or new feature that would not break
existing functionality).
- [ ] Breaking change (fix or new feature that would cause existing
functionality to change).
- [ ] New tests added to cover the changes.
- [ ] Quick tests passed locally by running `./runtest.sh`.
- [ ] In-line docstrings updated.
- [ ] Documentation updated.

Author: pcnudde

nvflare/fuel/f3/streaming/cacheable.py

@@ -90,20 +90,35 @@ def clear_cache(self):
     def _get_item(self, index: int, requester: str) -> bytes:
         with self.lock:
             if not self.cache:
-                # the cache has been cleared
+                cache_available = False
                 data = None
             else:
+                cache_available = True
                 data, _ = self.cache[index]
 
-            if data is None:
-                data = self.produce_item(index)
-                if self.cache:
-                    self.cache[index] = (data, 0)
-                    self.logger.debug(f"created and cached item {index} for {requester}: {len(data)} bytes")
-            else:
-                self.logger.debug(f"got item {index} from cache for {requester}")
+        if not cache_available:
+            return self.produce_item(index)
+
+        if data is not None:
+            self.logger.debug(f"got item {index} from cache for {requester}")
             return data
 
+        # Produce outside the lock so concurrent receivers aren't blocked.
+        # If two receivers produce the same item simultaneously, the first
+        # to re-acquire the lock stores its result; the second uses it.
+        data = self.produce_item(index)
+
+        with self.lock:
+            if self.cache:
+                existing, count = self.cache[index]
+                if existing is None:
+                    self.cache[index] = (data, count)
+                    self.logger.debug(f"created and cached item {index} for {requester}: {len(data)} bytes")
+                else:
+                    data = existing
+                    self.logger.debug(f"got item {index} from cache for {requester} (produced concurrently)")
+        return data
+
     def _adjust_cache(self, start: int, count: int):
         with self.lock:
             if not self.cache: