Use Persistent wrapper to prevent GC of input Buffer #152

Avoids memcpy of input and output Buffer objects

Improves Buffer and Stream performance by ~3%

docs/install.md

@@ -47,7 +47,13 @@ libvips must be installed before `npm install` is run.
 This can be achieved via homebrew:
 
 ```sh
-brew install homebrew/science/vips --with-webp --with-graphicsmagick
+brew install homebrew/science/vips
+```
+
+For GIF input and WebP output suppport use:
+
+```sh
+brew install homebrew/science/vips --with-graphicsmagick --with-webp
 ```
 
 A missing or incorrectly configured _Xcode Command Line Tools_ installation

src/metadata.cc

@@ -64,20 +64,15 @@ struct MetadataBaton {
     iccLength(0) {}
 };
 
-/*
-  Delete input char[] buffer and notify V8 of memory deallocation
-  Used as the callback function for the "postclose" signal
-*/
-static void DeleteBuffer(VipsObject *object, char *buffer) {
-  if (buffer != NULL) {
-    delete[] buffer;
-  }
-}
-
 class MetadataWorker : public AsyncWorker {
 
  public:
-  MetadataWorker(Callback *callback, MetadataBaton *baton) : AsyncWorker(callback), baton(baton) {}
+  MetadataWorker(Callback *callback, MetadataBaton *baton, const Local<Object> &bufferIn) :
+    AsyncWorker(callback), baton(baton) {
+      if (baton->bufferInLength > 0) {
+        SaveToPersistent("bufferIn", bufferIn);
+      }
+    }
   ~MetadataWorker() {}
 
   void Execute() {
@@ -91,17 +86,12 @@ class MetadataWorker : public AsyncWorker {
       imageType = DetermineImageType(baton->bufferIn, baton->bufferInLength);
       if (imageType != ImageType::UNKNOWN) {
         image = InitImage(baton->bufferIn, baton->bufferInLength, VIPS_ACCESS_RANDOM);
-        if (image != NULL) {
-          // Listen for "postclose" signal to delete input buffer
-          g_signal_connect(image, "postclose", G_CALLBACK(DeleteBuffer), baton->bufferIn);
-        } else {
+        if (image == NULL) {
           (baton->err).append("Input buffer has corrupt header");
           imageType = ImageType::UNKNOWN;
-          DeleteBuffer(NULL, baton->bufferIn);
         }
       } else {
         (baton->err).append("Input buffer contains unsupported image format");
-        DeleteBuffer(NULL, baton->bufferIn);
       }
     } else {
       // From file
@@ -192,6 +182,11 @@ class MetadataWorker : public AsyncWorker {
       }
       argv[1] = info;
     }
+
+    // Dispose of Persistent wrapper around input Buffer so it can be garbage collected
+    if (baton->bufferInLength > 0) {
+      GetFromPersistent("bufferIn");
+    }
     delete baton;
 
     // Return to JavaScript
@@ -215,17 +210,16 @@ NAN_METHOD(metadata) {
   // Input filename
   baton->fileIn = *Utf8String(Get(options, New("fileIn").ToLocalChecked()).ToLocalChecked());
   // Input Buffer object
+  Local<Object> bufferIn;
   if (node::Buffer::HasInstance(Get(options, New("bufferIn").ToLocalChecked()).ToLocalChecked())) {
-    Local<Object> buffer = Get(options, New("bufferIn").ToLocalChecked()).ToLocalChecked().As<Object>();
-    // Take a copy of the input Buffer to avoid problems with V8 heap compaction
-    baton->bufferInLength = node::Buffer::Length(buffer);
-    baton->bufferIn = new char[baton->bufferInLength];
-    memcpy(baton->bufferIn, node::Buffer::Data(buffer), baton->bufferInLength);
+    bufferIn = Get(options, New("bufferIn").ToLocalChecked()).ToLocalChecked().As<Object>();
+    baton->bufferInLength = node::Buffer::Length(bufferIn);
+    baton->bufferIn = node::Buffer::Data(bufferIn);
   }
 
   // Join queue for worker thread
   Callback *callback = new Callback(info[1].As<Function>());
-  AsyncQueueWorker(new MetadataWorker(callback, baton));
+  AsyncQueueWorker(new MetadataWorker(callback, baton, bufferIn));
 
   // Increment queued task counter
   g_atomic_int_inc(&counterQueue);

src/pipeline.cc

@@ -32,7 +32,7 @@ using Nan::Get;
 using Nan::Set;
 using Nan::To;
 using Nan::New;
-using Nan::CopyBuffer;
+using Nan::NewBuffer;
 using Nan::Null;
 using Nan::Equals;
 
@@ -167,21 +167,15 @@ struct PipelineBaton {
     }
 };
 
-/*
-  Delete input char[] buffer and notify V8 of memory deallocation
-  Used as the callback function for the "postclose" signal
-*/
-static void DeleteBuffer(VipsObject *object, char *buffer) {
-  if (buffer != NULL) {
-    delete[] buffer;
-  }
-}
-
 class PipelineWorker : public AsyncWorker {
 
  public:
-  PipelineWorker(Callback *callback, PipelineBaton *baton, Callback *queueListener) :
-    AsyncWorker(callback), baton(baton), queueListener(queueListener) {}
+  PipelineWorker(Callback *callback, PipelineBaton *baton, Callback *queueListener, const Local<Object> &bufferIn) :
+    AsyncWorker(callback), baton(baton), queueListener(queueListener) {
+      if (baton->bufferInLength > 0) {
+        SaveToPersistent("bufferIn", bufferIn);
+      }
+    }
   ~PipelineWorker() {}
 
   /*
@@ -208,18 +202,13 @@ class PipelineWorker : public AsyncWorker {
       inputImageType = DetermineImageType(baton->bufferIn, baton->bufferInLength);
       if (inputImageType != ImageType::UNKNOWN) {
         image = InitImage(baton->bufferIn, baton->bufferInLength, baton->accessMethod);
-        if (image != NULL) {
-          // Listen for "postclose" signal to delete input buffer
-          g_signal_connect(image, "postclose", G_CALLBACK(DeleteBuffer), baton->bufferIn);
-        } else {
+        if (image == NULL) {
           // Could not read header data
           (baton->err).append("Input buffer has corrupt header");
           inputImageType = ImageType::UNKNOWN;
-          DeleteBuffer(NULL, baton->bufferIn);
         }
       } else {
         (baton->err).append("Input buffer contains unsupported image format");
-        DeleteBuffer(NULL, baton->bufferIn);
       }
     } else {
       // From file
@@ -969,10 +958,8 @@ class PipelineWorker : public AsyncWorker {
       Set(info, New("height").ToLocalChecked(), New<Uint32>(static_cast<uint32_t>(height)));
 
       if (baton->bufferOutLength > 0) {
-        // Copy data to new Buffer
-        argv[1] = CopyBuffer(static_cast<char*>(baton->bufferOut), baton->bufferOutLength).ToLocalChecked();
-        // bufferOut was allocated via g_malloc
-        g_free(baton->bufferOut);
+        // Pass ownership of output data to Buffer instance
+        argv[1] = NewBuffer(static_cast<char*>(baton->bufferOut), baton->bufferOutLength).ToLocalChecked();
         // Add buffer size to info
         Set(info, New("size").ToLocalChecked(), New<Uint32>(static_cast<uint32_t>(baton->bufferOutLength)));
         argv[2] = info;
@@ -984,8 +971,12 @@ class PipelineWorker : public AsyncWorker {
         argv[1] = info;
       }
     }
+
+    // Dispose of Persistent wrapper around input Buffer so it can be garbage collected
+    if (baton->bufferInLength > 0) {
+      GetFromPersistent("bufferIn");
+    }
     delete baton;
-    // to here
 
     // Decrement processing task counter
     g_atomic_int_dec_and_test(&counterProcess);
@@ -1132,12 +1123,11 @@ NAN_METHOD(pipeline) {
     To<bool>(Get(options, New("sequentialRead").ToLocalChecked()).ToLocalChecked()).FromJust() ?
     VIPS_ACCESS_SEQUENTIAL : VIPS_ACCESS_RANDOM;
   // Input Buffer object
+  Local<Object> bufferIn;
   if (node::Buffer::HasInstance(Get(options, New("bufferIn").ToLocalChecked()).ToLocalChecked())) {
-    Local<Object> buffer = Get(options, New("bufferIn").ToLocalChecked()).ToLocalChecked().As<Object>();
-    // Take a copy of the input Buffer to avoid problems with V8 heap compaction
-    baton->bufferInLength = node::Buffer::Length(buffer);
-    baton->bufferIn = new char[baton->bufferInLength];
-    memcpy(baton->bufferIn, node::Buffer::Data(buffer), baton->bufferInLength);
+    bufferIn = Get(options, New("bufferIn").ToLocalChecked()).ToLocalChecked().As<Object>();
+    baton->bufferInLength = node::Buffer::Length(bufferIn);
+    baton->bufferIn = node::Buffer::Data(bufferIn);
   }
   // ICC profile to use when input CMYK image has no embedded profile
   baton->iccProfilePath = *Utf8String(Get(options, New("iccProfilePath").ToLocalChecked()).ToLocalChecked());
@@ -1212,11 +1202,10 @@ NAN_METHOD(pipeline) {
 
   // Join queue for worker thread
   Callback *callback = new Callback(info[1].As<Function>());
-  AsyncQueueWorker(new PipelineWorker(callback, baton, queueListener));
+  AsyncQueueWorker(new PipelineWorker(callback, baton, queueListener, bufferIn));
 
   // Increment queued task counter
   g_atomic_int_inc(&counterQueue);
   Local<Value> queueLength[1] = { New<Uint32>(counterQueue) };
   queueListener->Call(1, queueLength);
 }
-

test/bench/package.json

@@ -8,12 +8,12 @@
     "test": "node perf && node random && node parallel"
   },
   "devDependencies": {
-    "async": "^1.4.2",
+    "async": "^1.5.0",
     "benchmark": "^1.0.0",
-    "gm": "^1.20.0",
+    "gm": "^1.21.0",
     "imagemagick": "^0.1.3",
     "imagemagick-native": "^1.8.0",
-    "jimp": "^0.2.8",
+    "jimp": "^0.2.19",
     "lwip": "^0.0.8",
     "semver": "^5.0.3"
   },

test/leak/sharp.supp

@@ -278,11 +278,18 @@
   ...
   fun:vips__magick_read_header
 }
+{
+  cond_magick_is_palette_image
+  Memcheck:Cond
+  fun:IsPaletteImage
+  ...
+  fun:get_bands
+}
 
 # glib g_file_read_link
 # https://github.com/GNOME/glib/commit/49a5d0f6f2aed99cd78f25655f137f4448e47d92
 {
-  leak_magick
+  leak_g_file_read_link
   Memcheck:Leak
   match-leak-kinds: definite,indirect,possible
   ...