Merge pull request #1434 from girder/tile-frames-paralellism

Compute tile frames in parallel.

CHANGELOG.md

@@ -8,6 +8,7 @@
 - Better filter DICOM adjacent files to ensure they share series instance IDs ([#1424](../../pull/1424))
 - Optimizing small getRegion calls and some tiff tile fetches ([#1427](../../pull/1427))
 - Started adding python types to the core library ([#1432](../../pull/1432), [#1433](../../pull/1433))
+- Use parallelism in computing tile frames ([#1434](../../pull/1434))
 
 ### Changed
 - Cleanup some places where get was needlessly used ([#1428](../../pull/1428))

girder/girder_large_image/rest/large_image_resource.py

@@ -26,10 +26,10 @@
 import time
 
 import cherrypy
-import psutil
 from girder_jobs.constants import JobStatus
 from girder_jobs.models.job import Job
 
+import large_image
 from girder import logger
 from girder.api import access
 from girder.api.describe import Description, autoDescribeRoute, describeRoute
@@ -136,7 +136,8 @@ def createThumbnailsJob(job):
         job, log='Started creating large image thumbnails\n',
         status=JobStatus.RUNNING)
     concurrency = int(job['kwargs'].get('concurrent', 0))
-    concurrency = psutil.cpu_count(logical=True) if concurrency < 1 else concurrency
+    concurrency = large_image.tilesource.utilities.cpu_count(
+        logical=True) if concurrency < 1 else concurrency
     status = {
         'checked': 0,
         'created': 0,

girder_annotation/girder_large_image_annotation/models/annotationelement.py

@@ -18,13 +18,13 @@
 import datetime
 import io
 import math
-import multiprocessing
 import pickle
 import time
 
 import pymongo
 from girder_large_image.models.image_item import ImageItem
 
+import large_image
 from girder import logger
 from girder.constants import AccessType, SortDir
 from girder.models.file import File
@@ -601,7 +601,7 @@ def updateElements(self, annotation):
         if not len(elements):
             return
         now = datetime.datetime.now(datetime.timezone.utc)
-        threads = multiprocessing.cpu_count()
+        threads = large_image.tilesource.utilities.cpu_count()
         chunkSize = int(max(100000 // threads, 10000))
         with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as pool:
             for chunk in range(0, len(elements), chunkSize):

large_image/tilesource/base.py

@@ -24,16 +24,12 @@
 from ..constants import (TILE_FORMAT_IMAGE, TILE_FORMAT_NUMPY, TILE_FORMAT_PIL,
                          SourcePriority, TileInputUnits, TileOutputMimeTypes,
                          TileOutputPILFormat)
+from . import utilities
 from .jupyter import IPyLeafletMixin
 from .tiledict import LazyTileDict
-from .utilities import (ImageBytes, JSONDict,  # noqa: F401
-                        _addRegionTileToTiled, _addSubimageToImage,
-                        _calculateWidthHeight, _encodeImage,
-                        _encodeImageBinary, _imageToNumpy, _imageToPIL,
-                        _letterboxImage, _makeSameChannelDepth,
-                        _vipsAddAlphaBand, _vipsCast, _vipsParameters,
-                        dictToEtree, etreeToDict, getPaletteColors,
-                        histogramThreshold, nearPowerOfTwo)
+from .utilities import (ImageBytes, JSONDict, _imageToNumpy,  # noqa: F401
+                        _imageToPIL, dictToEtree, etreeToDict,
+                        getPaletteColors, histogramThreshold, nearPowerOfTwo)
 
 
 class TileSource(IPyLeafletMixin):
@@ -644,7 +640,7 @@ def _tileIteratorInfo(self, **kwargs) -> Optional[Dict[str, Any]]:  # noqa
                 maxWidth = regionWidth / cast(float, mag['scale'])
                 maxHeight = regionHeight / cast(float, mag['scale'])
                 magRequestedScale = cast(float, mag['scale'])
-        outWidth, outHeight, calcScale = _calculateWidthHeight(
+        outWidth, outHeight, calcScale = utilities._calculateWidthHeight(
             maxWidth, maxHeight, regionWidth, regionHeight)
         requestedScale = calcScale if magRequestedScale is None else magRequestedScale
         if (regionWidth < 0 or regionHeight < 0 or outWidth == 0 or
@@ -1692,7 +1688,7 @@ def _outputTile(
         if getattr(tile, 'fp', None) and self._pilFormatMatches(tile):
             tile.fp.seek(0)  # type: ignore
             return tile.fp.read()  # type: ignore
-        result = _encodeImageBinary(
+        result = utilities._encodeImageBinary(
             tile, self.encoding, self.jpegQuality, self.jpegSubsampling, self.tiffCompression)
         return result
 
@@ -2203,7 +2199,7 @@ def getRegion(self, format: Union[str, Tuple[str]] = (TILE_FORMAT_IMAGE, ), **kw
         iterInfo = self._tileIteratorInfo(**kwargs)
         if iterInfo is None:
             pilimage = PIL.Image.new('RGB', (0, 0))
-            return _encodeImage(pilimage, format=format, **kwargs)
+            return utilities._encodeImage(pilimage, format=format, **kwargs)
         regionWidth = iterInfo['region']['width']
         regionHeight = iterInfo['region']['height']
         top = iterInfo['region']['top']
@@ -2218,10 +2214,10 @@ def getRegion(self, format: Union[str, Tuple[str]] = (TILE_FORMAT_IMAGE, ), **kw
             subimage, _ = _imageToNumpy(tile['tile'])
             x0, y0 = tile['x'] - left, tile['y'] - top
             if tiled:
-                tiledimage = _addRegionTileToTiled(
+                tiledimage = utilities._addRegionTileToTiled(
                     tiledimage, subimage, x0, y0, regionWidth, regionHeight, tile, **kwargs)
             else:
-                image = _addSubimageToImage(
+                image = utilities._addSubimageToImage(
                     image, subimage, x0, y0, regionWidth, regionHeight)
             # Somehow discarding the tile here speeds things up.
             del tile
@@ -2244,8 +2240,9 @@ def getRegion(self, format: Union[str, Tuple[str]] = (TILE_FORMAT_IMAGE, ), **kw
         maxWidth = kwargs.get('output', {}).get('maxWidth')
         maxHeight = kwargs.get('output', {}).get('maxHeight')
         if kwargs.get('fill') and maxWidth and maxHeight:
-            image = _letterboxImage(_imageToPIL(image, mode), maxWidth, maxHeight, kwargs['fill'])
-        return _encodeImage(image, format=format, **kwargs)
+            image = utilities._letterboxImage(
+                _imageToPIL(image, mode), maxWidth, maxHeight, kwargs['fill'])
+        return utilities._encodeImage(image, format=format, **kwargs)
 
     def _encodeTiledImage(
             self, image: Dict[str, Any], outWidth: int, outHeight: int,
@@ -2277,9 +2274,9 @@ def _encodeTiledImage(
         vimg = cast(pyvips.Image, image['strips'][0])
         for y in sorted(image['strips'].keys())[1:]:
             if image['strips'][y].bands + 1 == vimg.bands:
-                image['strips'][y] = _vipsAddAlphaBand(image['strips'][y], vimg)
+                image['strips'][y] = utilities._vipsAddAlphaBand(image['strips'][y], vimg)
             elif vimg.bands + 1 == image['strips'][y].bands:
-                vimg = _vipsAddAlphaBand(vimg, image['strips'][y])
+                vimg = utilities._vipsAddAlphaBand(vimg, image['strips'][y])
             vimg = vimg.insert(image['strips'][y], 0, y, expand=True)
 
         if outWidth != image['width'] or outHeight != image['height']:
@@ -2314,8 +2311,9 @@ def _encodeTiledImageFromVips(
         """
         import pyvips
 
-        convertParams = _vipsParameters(defaultCompression='lzw', **kwargs)
-        vimg = _vipsCast(cast(pyvips.Image, vimg), convertParams['compression'] in {'webp', 'jpeg'})
+        convertParams = utilities._vipsParameters(defaultCompression='lzw', **kwargs)
+        vimg = utilities._vipsCast(
+            cast(pyvips.Image, vimg), convertParams['compression'] in {'webp', 'jpeg'})
         maxWidth = kwargs.get('output', {}).get('maxWidth')
         maxHeight = kwargs.get('output', {}).get('maxHeight')
         if (kwargs.get('fill') and str(kwargs.get('fill')).lower() != 'none' and
@@ -2355,7 +2353,8 @@ def _encodeTiledImageFromVips(
     def tileFrames(
             self, format: Union[str, Tuple[str]] = (TILE_FORMAT_IMAGE, ),
             frameList: Optional[List[int]] = None,
-            framesAcross: Optional[int] = None, **kwargs) -> Tuple[
+            framesAcross: Optional[int] = None,
+            max_workers: Optional[int] = -4, **kwargs) -> Tuple[
                 Union[np.ndarray, PIL.Image.Image, ImageBytes, bytes, pathlib.Path], str]:
         """
         Given the parameters for getRegion, plus a list of frames and the
@@ -2373,9 +2372,14 @@ def tileFrames(
         :param kwargs: optional arguments.  Some options are region, output,
             encoding, jpegQuality, jpegSubsampling, tiffCompression, fill.  See
             tileIterator.
+        :param max_workers: maximum workers for parallelism.  If negative, use
+            the minimum of the absolute value of this number or
+            multiprocessing.cpu_count().
         :returns: regionData, formatOrRegionMime: the image data and either the
             mime type, if the format is TILE_FORMAT_IMAGE, or the format.
         """
+        import concurrent.futures
+
         lastlog = time.time()
         kwargs = kwargs.copy()
         kwargs.pop('tile_position', None)
@@ -2397,7 +2401,7 @@ def tileFrames(
         iterInfo = self._tileIteratorInfo(frame=frameList[0], **kwargs)
         if iterInfo is None:
             pilimage = PIL.Image.new('RGB', (0, 0))
-            return _encodeImage(pilimage, format=format, **kwargs)
+            return utilities._encodeImage(pilimage, format=format, **kwargs)
         frameWidth = iterInfo['output']['width']
         frameHeight = iterInfo['output']['height']
         maxWidth = kwargs.get('output', {}).get('maxWidth')
@@ -2409,29 +2413,36 @@ def tileFrames(
         tile = next(self._tileIterator(iterInfo))
         image = None
         tiledimage = None
-        for idx, frame in enumerate(frameList):
-            subimage, _ = self.getRegion(format=TILE_FORMAT_NUMPY, frame=frame, **kwargs)
-            offsetX = (idx % framesAcross) * frameWidth
-            offsetY = (idx // framesAcross) * frameHeight
-            if time.time() - lastlog > 10:
-                self.logger.info(
-                    'Tiling frame %d (%d/%d), offset %dx%d',
-                    frame, idx, len(frameList), offsetX, offsetY)
-                lastlog = time.time()
-            else:
-                self.logger.debug(
-                    'Tiling frame %d (%d/%d), offset %dx%d',
-                    frame, idx, len(frameList), offsetX, offsetY)
-            if tiled:
-                tiledimage = _addRegionTileToTiled(
-                    tiledimage, subimage, offsetX, offsetY, outWidth, outHeight, tile, **kwargs)
+        if max_workers is not None and max_workers < 0:
+            max_workers = max(-max_workers, utilities.cpu_count(False))
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
+            futures = []
+            for idx, frame in enumerate(frameList):
+                futures.append((idx, frame, pool.submit(
+                    self.getRegion, format=TILE_FORMAT_NUMPY, frame=frame, **kwargs)))
+            for idx, frame, future in futures:
+                subimage, _ = future.result()
+                offsetX = (idx % framesAcross) * frameWidth
+                offsetY = (idx // framesAcross) * frameHeight
+                if time.time() - lastlog > 10:
+                    self.logger.info(
+                        'Tiling frame %d (%d/%d), offset %dx%d',
+                        frame, idx, len(frameList), offsetX, offsetY)
+                    lastlog = time.time()
+                else:
+                    self.logger.debug(
+                        'Tiling frame %d (%d/%d), offset %dx%d',
+                        frame, idx, len(frameList), offsetX, offsetY)
+                if tiled:
+                    tiledimage = utilities._addRegionTileToTiled(
+                        tiledimage, subimage, offsetX, offsetY, outWidth, outHeight, tile, **kwargs)
             else:
-                image = _addSubimageToImage(
+                image = utilities._addSubimageToImage(
                     image, subimage, offsetX, offsetY, outWidth, outHeight)
         if tiled:
             return self._encodeTiledImage(
                 cast(Dict[str, Any], tiledimage), outWidth, outHeight, iterInfo, **kwargs)
-        return _encodeImage(image, format=format, **kwargs)
+        return utilities._encodeImage(image, format=format, **kwargs)
 
     def getRegionAtAnotherScale(
             self, sourceRegion: Dict[str, Any],
@@ -2833,14 +2844,14 @@ def getAssociatedImage(
         width = kwargs.get('width')
         height = kwargs.get('height')
         if width or height:
-            width, height, calcScale = _calculateWidthHeight(
+            width, height, calcScale = utilities._calculateWidthHeight(
                 width, height, imageWidth, imageHeight)
             image = image.resize(
                 (width, height),
                 getattr(PIL.Image, 'Resampling', PIL.Image).BICUBIC
                 if width > imageWidth else
                 getattr(PIL.Image, 'Resampling', PIL.Image).LANCZOS)
-        return _encodeImage(image, **kwargs)
+        return utilities._encodeImage(image, **kwargs)
 
     def getPixel(self, includeTileRecord: bool = False, **kwargs) -> JSONDict:
         """

large_image/tilesource/utilities.py

@@ -1,5 +1,6 @@
 import io
 import math
+import os
 import types
 import xml.etree.ElementTree
 from collections import defaultdict
@@ -1163,6 +1164,32 @@ def histogramThreshold(histogram, threshold, fromMax=False):
     return result
 
 
+def cpu_count(logical: bool = True) -> int:
+    """
+    Get the usable CPU count.  If psutil is available, it is used, since it can
+    determine the number of physical CPUS versus logical CPUs.  This returns
+    the smaller of that value from psutil and the number of cpus allowed by the
+    os scheduler, which means that for physical requests (logical=False), the
+    returned value may be more the the number of physical cpus that are usable.
+
+    :param logical: True to get the logical usable CPUs (which include
+        hyperthreading).  False for the physical usable CPUs.
+    :returns: the number of usable CPUs.
+    """
+    count = os.cpu_count()
+    try:
+        count = min(count, len(os.sched_getaffinity(0)))
+    except AttributeError:
+        pass
+    try:
+        import psutil
+
+        count = min(count, psutil.cpu_count(logical))
+    except ImportError:
+        pass
+    return max(1, count)
+
+
 def addPILFormatsToOutputOptions():
     """
     Check PIL for available formats that be saved and add them to the lists of

sources/openjpeg/large_image_source_openjpeg/__init__.py

@@ -17,7 +17,6 @@
 import builtins
 import io
 import math
-import multiprocessing
 import os
 import queue
 import struct
@@ -29,6 +28,7 @@
 import glymur
 import PIL.Image
 
+import large_image
 from large_image.cache_util import LruCacheMetaclass, methodcache
 from large_image.constants import TILE_FORMAT_NUMPY, SourcePriority
 from large_image.exceptions import TileSourceError, TileSourceFileNotFoundError
@@ -104,7 +104,7 @@ def __init__(self, path, **kwargs):
             if not os.path.isfile(self._largeImagePath):
                 raise TileSourceFileNotFoundError(self._largeImagePath) from None
             raise
-        glymur.set_option('lib.num_threads', multiprocessing.cpu_count())
+        glymur.set_option('lib.num_threads', large_image.tilesource.utilities.cpu_count())
         self._openjpegHandles = queue.LifoQueue()
         for _ in range(self._maxOpenHandles - 1):
             self._openjpegHandles.put(None)

utilities/converter/large_image_converter/__init__.py

@@ -342,7 +342,7 @@ def _concurrency_to_value(_concurrency=None, **kwargs):
     _concurrency = int(_concurrency) if str(_concurrency).isdigit() else 0
     if _concurrency > 0:
         return _concurrency
-    return max(1, psutil.cpu_count(logical=True) + _concurrency)
+    return max(1, large_image.tilesource.utilities.cpu_count(logical=True) + _concurrency)
 
 
 def _get_thread_pool(memoryLimit=None, **kwargs):