Multithreaded dng host implementation

src/codec/SkRawCodec.cpp

 /*
  * Copyright 2016 Google Inc.
  *
  * Use of this source code is governed by a BSD-style license that can be
  * found in the LICENSE file.
  */
 
 #include "SkCodec.h"
 #include "SkCodecPriv.h"
 #include "SkColorPriv.h"
 #include "SkData.h"
 #if !defined(GOOGLE3)
 #include "SkJpegCodec.h"
 #endif
 #include "SkRawCodec.h"
 #include "SkRefCnt.h"
 #include "SkStream.h"
 #include "SkStreamPriv.h"
 #include "SkSwizzler.h"
+#include "SkTaskGroup.h"
 #include "SkTemplates.h"
 #include "SkTypes.h"
 
+#include "dng_area_task.h"
 #include "dng_color_space.h"
 #include "dng_exceptions.h"
 #include "dng_host.h"
 #include "dng_info.h"
 #include "dng_memory.h"
 #include "dng_render.h"
 #include "dng_stream.h"
 
 #include "src/piex.h"
 
 #include <cmath>  // for std::round,floor,ceil
 #include <limits>
 
 namespace {
 
+// Caluclates the number of tiles of tile_size that fit into the area in vertical and horizontal
+// directions.
+dng_point num_tiles_in_area(const dng_point &areaSize,
+                            const dng_point_real64 &tileSize) {
+  return dng_point((areaSize.v + tileSize.v - 1) / tileSize.v,
+                   (areaSize.h + tileSize.h - 1) / tileSize.h);
+}
+
+// Calculate the number of tiles to process per thread, taking into account the maximum number of
+// threads.
+dng_point num_tiles_per_thread(const int maxThreads,
+                               const dng_point &tilesInArea) {
+  dng_point tilesInThread = {1, 1};
+  while (((tilesInArea.v + tilesInThread.v - 1) / tilesInThread.v) *
+             ((tilesInArea.h + tilesInThread.h - 1) / tilesInThread.h) >
+         maxThreads) {
+    if (tilesInArea.h > tilesInThread.h) {
+        ++tilesInThread.h;
+    } else {
+        ++tilesInThread.v;
+    }
+  }
+  return tilesInThread;
+}
+
+class SkDngHost : public dng_host {
+public:
+    using dng_host::dng_host;
+
+    void PerformAreaTask(dng_area_task& task, const dng_rect& area) override {
+        const int maxThreads = Min_int32(sk_num_cores(), static_cast<int>(task.MaxThreads()));
+        SkTaskGroup threadPool;

[mtklein, 2016/01/25 19:50:59]

This isn't really an independent thread pool; it represents a group of tasks
that are multiplexed onto a global threadpool.  Creating it does not create a
thread pool, nor does destroying it destroy any threads.

You should be fine to break your problem up into as many small parts as
necessary.  It'll usually perform better if you add more tasks to the task group
than there are cpu threads, to allow for uneven task runtime.

[ebrauer, 2016/01/26 09:53:08]

Changed the name. The dng_task has a maximum we must not exceed.

+
+        const dng_point tileSize(task.FindTileSize(area));
+        const dng_point tilesInArea = num_tiles_in_area(area.Size(), tileSize);
+
+        const dng_point tilesInThread = num_tiles_per_thread(maxThreads, tilesInArea);
+        const dng_point threadAreaSize = {tilesInThread.v * tileSize.v,
+                                          tilesInThread.h * tileSize.h};
+        dng_rect threadArea(area.t, area.l, threadAreaSize.v + area.t, threadAreaSize.h + area.l);
+
+        task.Start(maxThreads, tileSize, &Allocator(), Sniffer());
+        int taskIndex = 0;
+        for (int v = 0; v < tilesInArea.v; v += tilesInThread.v) {
+            threadArea.l = area.l;
+            threadArea.r = threadAreaSize.h + threadArea.l;
+            for (int h = 0; h < tilesInArea.h; h += tilesInThread.h) {
+                std::function<void()> func = std::bind(&dng_area_task::ProcessOnThread, &task,

[mtklein, 2016/01/25 19:50:59]

Generally I'd rather see a lambda for this than std::bind.
You can usually pass them right into SkTaskGroup.add() without giving them a
name.

[ebrauer, 2016/01/26 09:53:08]

Done.

+                                                       taskIndex, threadArea, tileSize, Sniffer());
+                threadPool.add(func);

[mtklein, 2016/01/25 21:02:01]

Something like,

threadPool.add([&task, this, taskIndex, threadArea, tileSize] {
    task.ProcessOnThread(taskIndex, threadArea, tileSize, this->Sniffer());
});

[ebrauer, 2016/01/26 09:53:07]

Done.

+
+                threadArea = threadArea + dng_point(0, threadAreaSize.h);
+                threadArea.r = Min_int32(threadArea.r, area.r);
+            }
+            threadArea = threadArea + dng_point(threadAreaSize.v, 0);
+            threadArea.b = Min_int32(threadArea.b, area.b);
+            ++taskIndex;
+        }
+
+        threadPool.wait();
+        task.Finish(maxThreads);
+    }
+
+private:
+    typedef dng_host INHERITED;
+};
+
 // T must be unsigned type.
 template <class T>
 bool safe_add_to_size_t(T arg1, T arg2, size_t* result) {
     SkASSERT(arg1 >= 0);
     SkASSERT(arg2 >= 0);
     if (arg1 >= 0 && arg2 <= std::numeric_limits<T>::max() - arg1) {
         T sum = arg1 + arg2;
         if (sum <= std::numeric_limits<size_t>::max()) {
             *result = static_cast<size_t>(sum);
             return true;
@@ skipping 179 matching lines @@
      *   returned size could be: 2000 x 1500
      */
     dng_image* render(int width, int height) {
         if (!fHost || !fInfo || !fNegative || !fDngStream) {
             if (!this->readDng()) {
                 return nullptr;
             }
         }
 
         // render() takes ownership of fHost, fInfo, fNegative and fDngStream when available.
-        SkAutoTDelete<dng_host> host(fHost.release());
+        SkAutoTDelete<SkDngHost> host(fHost.release());

[adaubert, 2016/01/26 09:26:37]

This change is not necessary.

[ebrauer, 2016/01/26 09:53:08]

Done.

         SkAutoTDelete<dng_info> info(fInfo.release());
         SkAutoTDelete<dng_negative> negative(fNegative.release());
         SkAutoTDelete<dng_stream> dngStream(fDngStream.release());
 
         // DNG SDK preserves the aspect ratio, so it only needs to know the longer dimension.
         const int preferredSize = SkTMax(width, height);
         try {
             host->SetPreferredSize(preferredSize);
             host->ValidateSizes();
 
@@ skipping 33 matching lines @@
         return fIsScalable;
     }
 
     bool isXtransImage() const {
         return fIsXtransImage;
     }
 
 private:
     bool readDng() {
         // Due to the limit of DNG SDK, we need to reset host and info.
-        fHost.reset(new dng_host(&fAllocator));
+        fHost.reset(new SkDngHost(&fAllocator));
         fInfo.reset(new dng_info);
         fDngStream.reset(new SkDngStream(fStream));
         try {
             fHost->ValidateSizes();
             fInfo->Parse(*fHost, *fDngStream);
             fInfo->PostParse(*fHost);
             if (!fInfo->IsValidDNG()) {
                 return false;
             }
 
@@ skipping 18 matching lines @@
             fNegative.reset(nullptr);
             return false;
         }
     }
 
     SkDngImage(SkRawStream* stream)
         : fStream(stream) {}
 
     SkDngMemoryAllocator fAllocator;
     SkAutoTDelete<SkRawStream> fStream;
-    SkAutoTDelete<dng_host> fHost;
+    SkAutoTDelete<SkDngHost> fHost;

[adaubert, 2016/01/26 09:26:38]

This change is not necessary.

[ebrauer, 2016/01/26 09:53:08]

Done.

     SkAutoTDelete<dng_info> fInfo;
     SkAutoTDelete<dng_negative> fNegative;
     SkAutoTDelete<dng_stream> fDngStream;
 
     SkImageInfo fImageInfo;
     bool fIsScalable;
     bool fIsXtransImage;
 };
 
 /*
@@ skipping 125 matching lines @@
     SkISize sizeFloor = this->onGetScaledDimensions(1.f / std::floor(fullShortEdge / shortEdge));
     SkISize sizeCeil = this->onGetScaledDimensions(1.f / std::ceil(fullShortEdge / shortEdge));
     return sizeFloor == dim || sizeCeil == dim;
 }
 
 SkRawCodec::~SkRawCodec() {}
 
 SkRawCodec::SkRawCodec(SkDngImage* dngImage)
     : INHERITED(dngImage->getImageInfo(), nullptr)
     , fDngImage(dngImage) {}