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..451efd2d90c8ab71ce30503faa1149174d859957 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,82 @@
 
 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,

[msarett, 2016/01/26 21:34:03]

nit:

This is non-critical, but can we use a ceil_div() helper function?  SkGifCodec
actually already has one.  Can we share it in SkCodecPriv.h?

[ebrauer, 2016/01/26 22:50:59]

Would it be Ok to add a Fixit?

[ebrauer, 2016/01/27 15:54:37]

Done.

+                   (areaSize.h + tileSize.h - 1) / tileSize.h);
+}
+
+int num_threads_required(const dng_point& tilesInThread,
+                         const dng_point& tilesInArea) {
+  return ((tilesInArea.v + tilesInThread.v - 1) / tilesInThread.v) *
+         ((tilesInArea.h + tilesInThread.h - 1) / tilesInThread.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 (num_threads_required(tilesInThread, tilesInArea) > maxThreads) {
+      if (tilesInThread.h < tilesInArea.h) {

[msarett, 2016/01/26 21:34:03]

So, when possible, we will always split into tiles horizontally (as much as
possible) before splitting into tiles vertically?

Is there a reason for this?  Or does it not matter?

[ebrauer, 2016/01/26 22:50:58]

I think it's preferable to make them as wide as possible -- i.e. to increase
tiles_in_thread.h as long as possible while keeping tiles_in_thread.v at 1 --
because it gives better locality of reference.

[msarett, 2016/01/27 15:34:42]

Great!  Can you add a comment explaining this?

[ebrauer, 2016/01/27 15:54:37]

Done.

+          ++tilesInThread.h;
+      } else if (tilesInThread.v < tilesInArea.v) {
+          ++tilesInThread.v;
+      } else {
+          ThrowProgramError("num_tiles_per_thread calculation is wrong.");

[msarett, 2016/01/26 21:34:03]

Can we signal an error here without throwing an exception?  (Or maybe not,
because this is used in overriding a function in the dng_sdk?)

If it's code that will never be reached, you might just omit the else case?

Or maybe more Skia-like:
SkASSERT(false);
return {0,0};

[ebrauer, 2016/01/26 22:50:59]

It might be reached on program errors or invalid data. This prevents the
function from producing erroneous results if the caller should ever pass in
max_threads == 0. I think I like the assert.

[msarett, 2016/01/27 15:34:42]

Sorry for being unclear here.

If reaching this code indicates programmer error, then let's use SkASSERT.  This
will assert in Debug mode and do nothing in Release mode.

If reaching this code might indicate invalid input data, we need to indicate
that the decode failed.  AFAICT, the only way to do this is using an exception? 
That we will catch in SkRawCodec and return a failure?

[ebrauer, 2016/01/27 15:54:37]

Discussed this with Anton and we agreed that it is better to be safe here and
throw.

+      }
+  }
+  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 = static_cast<int>(kMaxMPThreads);

[msarett, 2016/01/26 21:34:03]

So this is 32 on 64-bit systems and 8 on 32-bit systems?

Can you explain why we need to get this value from the dng_sdk?

[ebrauer, 2016/01/26 22:50:59]

I think we should usetask.MaxThreads(), because the task that's passed in to
PerformAreaTask() may request to be processed on a smaller number of threads,
though it seems to be either 8 or 32.

[msarett, 2016/01/27 15:34:42]

sgtm.  Thanks for the comment.

[ebrauer, 2016/01/27 15:54:37]

Done.

+        SkTaskGroup taskGroup;
+
+        const dng_point tileSize(task.FindTileSize(area));

[msarett, 2016/01/26 21:34:03]

Can you add a comment explaining how the dng_sdk determines the tile size?

Side note: It would be useful for me to know some typical values of "area" and
"tileSize" (and how those compare to the image size).  I've looked at this
function in dng_sdk and find it a little confusing.

[ebrauer, 2016/01/26 22:50:58]

area is typically something like (4000, 3000) while tileSize is (256, 256). I
have always seen tileSize to be that which is maxTileSize in the dng sdk.

[msarett, 2016/01/27 15:34:42]

Can you add a comment that is something like this?

// tileSize is typically 256x256.

[ebrauer, 2016/01/27 15:54:37]

Done.

+        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 = dng_rect(threadAreaSize) + area.TL();
+
+        const int numThreads = num_threads_required(tilesInThread, tilesInArea);
+
+        task.Start(numThreads, tileSize, &Allocator(), Sniffer());
+        int taskIndex = 0;
+        for (int v = 0; v < tilesInArea.v; v += tilesInThread.v) {
+            for (int h = 0; h < tilesInArea.h; h += tilesInThread.h) {
+                threadArea.l = area.l + h * tileSize.h;
+                threadArea.t = area.t + v * tileSize.v;
+                threadArea.r = Min_int32(threadArea.l + threadAreaSize.h, area.r);
+                threadArea.b = Min_int32(threadArea.t + threadAreaSize.v, area.b);
+
+                taskGroup.add([&task, this, taskIndex, threadArea, tileSize] {
+                    task.ProcessOnThread(taskIndex, threadArea, tileSize, this->Sniffer());
+                });
+                ++taskIndex;
+            }
+        }
+
+        taskGroup.wait();
+        task.Finish(numThreads);
+    }
+
+    uint32 PerformAreaTaskThreads() override {

[msarett, 2016/01/26 21:34:03]

I'm not sure how this is used, but the comments in dng_host.h make it seem like
this should return the actual number of threads (rather than the max).

[ebrauer, 2016/01/26 22:50:58]

The documentation for PerformAreaTaskThreads() isn't completely clear, but my
understanding is that it's supposed to return the maximum number of threads that
PerformAreaTask() is potentially able to use. (This also makes sense given that
PerformAreaTaskThreads() is used in multiple places to check whether
multithreading is possible or not.)

[msarett, 2016/01/27 15:34:42]

Yeah I noticed that they mostly compare it to 1, so we definitely need to return
a value greater than 1.

They also seem to use it to determine the size of the AreaTask.

I don't have insight into how this might matter, but I think the code looks fine
as is.

[ebrauer, 2016/01/27 15:54:37]

Acknowledged.

+        return kMaxMPThreads;
+    }
+
+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) {
@@ -288,7 +366,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 {