WIP: Chunked image encoding

binding.gyp

@@ -13,6 +13,7 @@
         "src/mapnik_geometry.cpp",
         "src/mapnik_feature.cpp",
         "src/mapnik_image.cpp",
+        "src/mapnik_image_encode_chunked.cpp",
         "src/mapnik_image_view.cpp",
         "src/mapnik_grid.cpp",
         "src/mapnik_grid_view.cpp",

src/callback_streambuf.hpp

@@ -0,0 +1,47 @@
+#ifndef CALLBACK_STREAMBUF_H
+#define CALLBACK_STREAMBUF_H
+
+#include <array>
+#include <iostream>
+
+template<typename Callback, class Char = char>
+class callback_streambuf : public std::basic_streambuf<Char>
+{
+public:
+    using base_type = std::streambuf;
+    using char_type = typename base_type::char_type;
+    using int_type = typename base_type::int_type;
+
+    callback_streambuf(Callback callback, std::size_t buffer_size)
+        : callback_(callback),
+          buffer_(buffer_size)
+    {
+        base_type::setp(buffer_.data(), buffer_.data() + buffer_.size());
+    }
+
+protected:
+    int sync()
+    {
+        bool ok = callback_(base_type::pbase(),
+                            base_type::pptr() - base_type::pbase());
+        base_type::setp(buffer_.data(), buffer_.data() + buffer_.size());
+        return ok ? 0 : -1;
+    }
+
+    int_type overflow(int_type ch)
+    {
+        int ret = sync();
+        if (ch == base_type::traits_type::eof())
+        {
+            return ch;
+        }
+        base_type::sputc(ch);
+        return ret == 0 ? 0 : base_type::traits_type::eof();
+    }
+
+private:
+    Callback callback_;
+    std::vector<char_type> buffer_;
+};
+
+#endif

src/mapnik_image.cpp

@@ -22,12 +22,12 @@
 #include <mapnik/svg/svg_renderer_agg.hpp>
 #include <mapnik/svg/svg_path_attributes.hpp>
 
-#include "mapnik_image.hpp"
+#include "mapnik_image_encode.hpp"
 #include "mapnik_image_view.hpp"
-#include "mapnik_palette.hpp"
 #include "mapnik_color.hpp"
 
 #include "utils.hpp"
+#include "callback_streambuf.hpp"
 
 #include "agg_rasterizer_scanline_aa.h"
 #include "agg_basics.h"
@@ -82,6 +82,7 @@ void Image::Initialize(v8::Local<v8::Object> target) {
     Nan::SetPrototypeMethod(lcons, "setPixel", setPixel);
     Nan::SetPrototypeMethod(lcons, "encodeSync", encodeSync);
     Nan::SetPrototypeMethod(lcons, "encode", encode);
+    Nan::SetPrototypeMethod(lcons, "encodeChunked", encodeChunked);
     Nan::SetPrototypeMethod(lcons, "view", view);
     Nan::SetPrototypeMethod(lcons, "saveSync", saveSync);
     Nan::SetPrototypeMethod(lcons, "save", save);
@@ -3609,17 +3610,6 @@ NAN_METHOD(Image::encodeSync)
     }
 }
 
-typedef struct {
-    uv_work_t request;
-    Image* im;
-    std::string format;
-    palette_ptr palette;
-    bool error;
-    std::string error_name;
-    Nan::Persistent<v8::Function> cb;
-    std::string result;
-} encode_image_baton_t;
-
 /**
  * Encode this image into a buffer of encoded data
  *

src/mapnik_image.hpp

@@ -29,8 +29,11 @@ class Image: public Nan::ObjectWrap {
     static NAN_METHOD(setPixel);
     static NAN_METHOD(encodeSync);
     static NAN_METHOD(encode);
+    static NAN_METHOD(encodeChunked);
     static void EIO_Encode(uv_work_t* req);
     static void EIO_AfterEncode(uv_work_t* req);
+    static void EIO_EncodeChunked(uv_work_t* req);
+    static void EIO_AfterEncodeChunked(uv_work_t* req, int status);
 
     static NAN_METHOD(setGrayScaleToAlpha);
     static NAN_METHOD(width);

src/mapnik_image_encode.hpp

@@ -0,0 +1,18 @@
+#ifndef __NODE_MAPNIK_IMAGE_ENCODE_H__
+#define __NODE_MAPNIK_IMAGE_ENCODE_H__
+
+#include "mapnik_image.hpp"
+#include "mapnik_palette.hpp"
+
+typedef struct {
+    uv_work_t request;
+    Image* im;
+    std::string format;
+    palette_ptr palette;
+    bool error;
+    std::string error_name;
+    Nan::Persistent<v8::Function> cb;
+    std::string result;
+} encode_image_baton_t;
+
+#endif

src/mapnik_image_encode_chunked.cpp

@@ -0,0 +1,225 @@
+// mapnik
+#include <mapnik/image.hpp>             // for image types
+#include <mapnik/image_util.hpp>        // for save_to_stream
+
+#include "mapnik_image_encode.hpp"
+#include "mapnik_color.hpp"
+
+#include "utils.hpp"
+#include "callback_streambuf.hpp"
+
+struct chunked_encode_image_baton_t
+{
+    encode_image_baton_t image_baton;
+
+    uv_async_t async;
+    uv_mutex_t mutex;
+
+    using char_type = char;
+    using buffer_type = std::vector<char_type>;
+    using buffer_list_type = std::vector<buffer_type>;
+    buffer_list_type buffers;
+
+    const std::size_t buffer_size;
+
+    chunked_encode_image_baton_t(std::size_t buffer_size_)
+        : buffer_size(buffer_size_)
+    {
+        // The reinterpret_cast is for backward compatibility
+        // https://github.com/libuv/libuv/commit/db2a9072bce129630214904be5e2eedeaafc9835
+        if (uv_async_init(uv_default_loop(), &async,
+            reinterpret_cast<uv_async_cb>(yield_chunk)))
+        {
+            throw std::runtime_error("Cannot create async handler");
+        }
+
+        if (uv_mutex_init(&mutex))
+        {
+            uv_close(reinterpret_cast<uv_handle_t*>(&async), NULL);
+            throw std::runtime_error("Cannot create mutex");
+        }
+
+        async.data = this;
+    }
+
+    ~chunked_encode_image_baton_t()
+    {
+        uv_mutex_destroy(&mutex);
+    }
+
+    template<class Char, class Size>
+    bool operator()(const Char* buffer, Size size)
+    {
+        uv_mutex_lock(&mutex);
+        buffers.emplace_back(buffer, buffer + size);
+        uv_mutex_unlock(&mutex);
+
+        return uv_async_send(&async) == 0;
+    }
+
+    static void yield_chunk(uv_async_t* handle)
+    {
+        using closure_type = chunked_encode_image_baton_t;
+        closure_type & closure = *reinterpret_cast<closure_type*>(handle->data);
+
+        if (closure.image_baton.error)
+        {
+            uv_close(reinterpret_cast<uv_handle_t*>(handle), async_close_cb);
+            return;
+        }
+
+        buffer_list_type local_buffers;
+
+        uv_mutex_lock(&closure.mutex);
+        closure.buffers.swap(local_buffers);
+        uv_mutex_unlock(&closure.mutex);
+
+        Nan::HandleScope scope;
+        bool done = false;
+
+        for (auto const & buffer : local_buffers)
+        {
+            v8::Local<v8::Value> argv[2] = {
+                Nan::Null(), Nan::CopyBuffer(buffer.data(),
+                buffer.size()).ToLocalChecked() };
+            Nan::MakeCallback(Nan::GetCurrentContext()->Global(),
+                              Nan::New(closure.image_baton.cb), 2, argv);
+            done = buffer.empty();
+        }
+
+        if (done)
+        {
+            uv_close(reinterpret_cast<uv_handle_t*>(handle), async_close_cb);
+        }
+    }
+
+    static void async_close_cb(uv_handle_t* handle)
+    {
+        using closure_type = chunked_encode_image_baton_t;
+        closure_type & closure = *reinterpret_cast<closure_type*>(handle->data);
+
+        if (closure.image_baton.error)
+        {
+            Nan::HandleScope scope;
+            v8::Local<v8::Value> argv[1] = {
+                Nan::Error(closure.image_baton.error_name.c_str()) };
+            Nan::MakeCallback(Nan::GetCurrentContext()->Global(),
+                              Nan::New(closure.image_baton.cb), 1, argv);
+        }
+
+        closure.image_baton.im->_unref();
+        closure.image_baton.cb.Reset();
+        delete &closure;
+    }
+};
+
+void Image::EIO_EncodeChunked(uv_work_t* work)
+{
+    using closure_type = chunked_encode_image_baton_t;
+    closure_type & closure = *reinterpret_cast<closure_type*>(work->data);
+    try
+    {
+        callback_streambuf<closure_type&> streambuf(closure, closure.buffer_size);
+        std::ostream stream(&streambuf);
+
+        if (closure.image_baton.palette)
+        {
+            mapnik::save_to_stream(*closure.image_baton.im->this_,
+                                   stream,
+                                   closure.image_baton.format,
+                                   *closure.image_baton.palette);
+        }
+        else
+        {
+            mapnik::save_to_stream(*closure.image_baton.im->this_,
+                                   stream,
+                                   closure.image_baton.format);
+        }
+
+        stream.flush();
+    }
+    catch (std::exception const& ex)
+    {
+        closure.image_baton.error = true;
+        closure.image_baton.error_name = ex.what();
+    }
+
+    // Signalize end of stream
+    closure(static_cast<closure_type::char_type*>(NULL), 0);
+}
+
+NAN_METHOD(Image::encodeChunked)
+{
+    Image* im = Nan::ObjectWrap::Unwrap<Image>(info.Holder());
+
+    std::string format = "png";
+    palette_ptr palette;
+
+    if (info.Length() != 4)
+    {
+        Nan::ThrowTypeError("Function requires four arguments");
+        return;
+    }
+
+    // accept custom format
+    if (!info[0]->IsString())
+    {
+        Nan::ThrowTypeError("first arg, 'format' must be a string");
+        return;
+    }
+    format = TOSTR(info[0]);
+
+    // options hash
+    if (!info[1]->IsObject())
+    {
+        Nan::ThrowTypeError("second arg must be an options object");
+        return;
+    }
+
+    v8::Local<v8::Object> options = info[1].As<v8::Object>();
+
+    if (options->Has(Nan::New("palette").ToLocalChecked()))
+    {
+        v8::Local<v8::Value> format_opt = options->Get(Nan::New("palette").ToLocalChecked());
+        if (!format_opt->IsObject())
+        {
+            Nan::ThrowTypeError("'palette' must be an object");
+            return;
+        }
+
+        v8::Local<v8::Object> obj = format_opt.As<v8::Object>();
+        if (obj->IsNull() || obj->IsUndefined() || !Nan::New(Palette::constructor)->HasInstance(obj))
+        {
+            Nan::ThrowTypeError("mapnik.Palette expected as second arg");
+            return;
+        }
+
+        palette = Nan::ObjectWrap::Unwrap<Palette>(obj)->palette();
+    }
+
+    int buffer_size;
+    if (!info[2]->IsNumber() || (buffer_size = info[2]->IntegerValue()) < 1)
+    {
+        Nan::ThrowTypeError("third arg must be a positive integer");
+        return;
+    }
+
+    // ensure callback is a function
+    v8::Local<v8::Value> callback = info[info.Length() - 1];
+    if (!callback->IsFunction())
+    {
+        Nan::ThrowTypeError("last argument must be a callback function");
+        return;
+    }
+
+    chunked_encode_image_baton_t *closure = new chunked_encode_image_baton_t(buffer_size);
+    closure->image_baton.request.data = closure;
+    closure->image_baton.im = im;
+    closure->image_baton.format = format;
+    closure->image_baton.palette = palette;
+    closure->image_baton.error = false;
+    closure->image_baton.cb.Reset(callback.As<v8::Function>());
+
+    uv_queue_work(uv_default_loop(), &closure->image_baton.request, EIO_EncodeChunked, NULL);
+    im->Ref();
+}