Multithreaded dng host implementation

src/codec/SkRawCodec.cpp

Index: src/codec/SkRawCodec.cpp
diff --git a/src/codec/SkRawCodec.cpp b/src/codec/SkRawCodec.cpp
index 05f18ac635918d448149fe112374ebd57636789c..5e167dbf7b59d62b7a0fc482c440348b06ea7eda 100644
--- a/src/codec/SkRawCodec.cpp
+++ b/src/codec/SkRawCodec.cpp
@@ -17,9 +17,11 @@
 #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"
@@ -35,6 +37,73 @@
 
 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) {
@@ -234,7 +303,7 @@ public:
         }
 
         // 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());
@@ -288,7 +357,7 @@ public:
 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 {
@@ -327,7 +396,7 @@ private:
 
     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;