Merge branch 'seriyps/zlib-inflate-bound'

* seriyps/zlib-inflate-bound:
  Add zlib limited output buffer size functionality

Conflicts:
	erts/preloaded/ebin/zlib.beam

OTP-12548

4 files changed, 158 insertions, 1 deletions

diff --git a/erts/doc/src/zlib.xml b/erts/doc/src/zlib.xml
index da8ccdecdf..673b743e2e 100644
--- a/erts/doc/src/zlib.xml
+++ b/erts/doc/src/zlib.xml
@@ -312,6 +312,53 @@ list_to_binary([B1,B2])</pre>
       </desc>
     </func>
     <func>
+      <name name="inflateChunk" arity="2"/>
+      <fsummary>Decompress data with limited output size</fsummary>
+      <desc>
+        <p>Like <c>inflate/2</c>, but decompress no more data than
+          will fit in the buffer configured via <c>setBufSize/2</c>.
+          Is is useful when decompressing a stream with a high compression
+          ratio such that a small amount of compressed input may expand up to
+          1000 times.
+          It returns <c>{more, Decompressed}</c>, when there is more output
+          available, and <c>inflateChunk/1</c> should be used to read it.
+          It may introduce some output latency (reading
+          input without producing any output).</p>
+        <p>If a preset dictionary is needed at this point (see
+          <c>inflateSetDictionary</c> below), <c>inflateChunk/2</c> throws a
+          <c>{need_dictionary,Adler}</c> exception where <c>Adler</c> is
+          the adler32 checksum of the dictionary chosen by the
+          compressor.</p>
+
+        <pre>
+walk(Compressed, Handler) ->
+    Z = zlib:open(),
+    zlib:inflateInit(Z),
+    % Limit single uncompressed chunk size to 512kb
+    zlib:setBufSize(Z, 512 * 1024),
+    loop(Z, Handler, zlib:inflateChunk(Z, Compressed)),
+    zlib:inflateEnd(Z),
+    zlib:close(Z).
+
+loop(Z, Handler, {more, Uncompressed}) ->
+    Handler(Uncompressed),
+    loop(Z, Handler, zlib:inflateChunk(Z));
+loop(Z, Handler, Uncompressed) ->
+    Handler(Uncompressed).
+        </pre>
+      </desc>
+    </func>
+    <func>
+      <name name="inflateChunk" arity="1"/>
+      <fsummary>Read next uncompressed chunk</fsummary>
+      <desc>
+        <p>Read next chunk of uncompressed data, initialized by
+          <c>inflateChunk/2</c>.</p>
+        <p>This function should be repeatedly called, while it returns
+          <c>{more, Decompressed}</c>.</p>
+      </desc>
+    </func>
+    <func>
       <name name="inflateSetDictionary" arity="2"/>
       <fsummary>Initialize the decompression dictionary</fsummary>
       <desc>
diff --git a/erts/emulator/drivers/common/zlib_drv.c b/erts/emulator/drivers/common/zlib_drv.c
index 3143e4511d..f7b2d91d23 100644
--- a/erts/emulator/drivers/common/zlib_drv.c
+++ b/erts/emulator/drivers/common/zlib_drv.c
@@ -62,8 +62,17 @@
 #define CRC32_COMBINE   23
 #define ADLER32_COMBINE 24
 
+#define INFLATE_CHUNK   25
+
+
 #define DEFAULT_BUFSZ   4000
 
+/* This flag is used in the same places, where zlib return codes
+ * (Z_OK, Z_STREAM_END, Z_NEED_DICT) are. So, we need to set it to
+ * relatively large value to avoid possible value clashes in future.
+ * */
+#define INFLATE_HAS_MORE 100
+
 static int zlib_init(void);
 static ErlDrvData zlib_start(ErlDrvPort port, char* buf);
 static void zlib_stop(ErlDrvData e);
@@ -295,6 +304,58 @@ static int zlib_inflate(ZLibData* d, int flush)
     return res;
 }
 
+static int zlib_inflate_chunk(ZLibData* d)
+{
+    int res = Z_OK;
+
+    if ((d->bin == NULL) && (zlib_output_init(d) < 0)) {
+	errno = ENOMEM;
+	return Z_ERRNO;
+    }
+
+    while ((driver_sizeq(d->port) > 0) && (d->s.avail_out > 0) &&
+           (res != Z_STREAM_END)) {
+	int vlen;
+	SysIOVec* iov = driver_peekq(d->port, &vlen);
+	int len;
+
+	d->s.next_in = iov[0].iov_base;
+	d->s.avail_in = iov[0].iov_len;
+	while((d->s.avail_in > 0) && (d->s.avail_out > 0) && (res != Z_STREAM_END)) {
+	    res = inflate(&d->s, Z_NO_FLUSH);
+	    if (res == Z_NEED_DICT) {
+		/* Essential to eat the header bytes that zlib has looked at */
+		len = iov[0].iov_len - d->s.avail_in;
+		driver_deq(d->port, len);
+		return res;
+	    }
+	    if (res == Z_BUF_ERROR) {
+		/* Was possible more output, but actually not */
+		res = Z_OK;
+	    }
+	    else if (res < 0) {
+		return res;
+	    }
+	}
+	len = iov[0].iov_len - d->s.avail_in;
+	driver_deq(d->port, len);
+    }
+
+    /* We are here because all input was consumed or EOS reached or output
+     * buffer is full */
+    if (d->want_crc) {
+       d->crc = crc32(d->crc, (unsigned char*) d->bin->orig_bytes,
+		      d->binsz - d->s.avail_out);
+    }
+    zlib_output(d);
+    if ((res == Z_OK) && (d->s.avail_in > 0))
+       res = INFLATE_HAS_MORE;
+    else if (res == Z_STREAM_END) {
+       d->inflate_eos_seen = 1;
+    }
+    return res;
+}
+
 static int zlib_deflate(ZLibData* d, int flush)
 {
     int res = Z_OK;
@@ -568,6 +629,18 @@ static ErlDrvSSizeT zlib_ctl(ErlDrvData drv_data, unsigned int command, char *bu
 	    return zlib_return(res, rbuf, rlen);
 	}
 
+    case INFLATE_CHUNK:
+	if (d->state != ST_INFLATE) goto badarg;
+	if (len != 0) goto badarg;
+	res = zlib_inflate_chunk(d);
+	if (res == INFLATE_HAS_MORE) {
+	    return zlib_value2(4, 0, rbuf, rlen);
+	} else if (res == Z_NEED_DICT) {
+	    return zlib_value2(3, d->s.adler, rbuf, rlen);
+	} else {
+	    return zlib_return(res, rbuf, rlen);
+	}
+
     case GET_QSIZE:
 	return zlib_value(driver_sizeq(d->port), rbuf, rlen);
 
diff --git a/erts/preloaded/ebin/zlib.beam b/erts/preloaded/ebin/zlib.beam
index 193cebdc31..8783f189a4 100644
--- a/erts/preloaded/ebin/zlib.beam
+++ b/erts/preloaded/ebin/zlib.beam
diff --git a/erts/preloaded/src/zlib.erl b/erts/preloaded/src/zlib.erl
index df7b2e6198..5ebc67dcaa 100644
--- a/erts/preloaded/src/zlib.erl
+++ b/erts/preloaded/src/zlib.erl
@@ -24,6 +24,7 @@
 	 deflate/2,deflate/3,deflateEnd/1,
 	 inflateInit/1,inflateInit/2,inflateSetDictionary/2,
 	 inflateSync/1,inflateReset/1,inflate/2,inflateEnd/1,
+	 inflateChunk/1, inflateChunk/2,
 	 setBufSize/2,getBufSize/1,
 	 crc32/1,crc32/2,crc32/3,adler32/2,adler32/3,getQSize/1,
 	 crc32_combine/4,adler32_combine/4,
@@ -100,6 +101,7 @@
 -define(INFLATE_RESET,   12).
 -define(INFLATE_END,     13).
 -define(INFLATE,         14).
+-define(INFLATE_CHUNK,   25).
 
 -define(CRC32_0,         15).
 -define(CRC32_1,         16).
@@ -263,6 +265,39 @@ inflate(Z, Data) ->
 	    erlang:error(badarg) 
     end.
 
+-spec inflateChunk(Z, Data) -> Decompressed | {more, Decompressed} when
+      Z :: zstream(),
+      Data :: iodata(),
+      Decompressed :: iolist().
+inflateChunk(Z, Data) ->
+    try port_command(Z, Data) of
+	true ->
+        inflateChunk(Z)
+    catch
+	error:_Err ->
+	    flush(Z),
+	    erlang:error(badarg)
+    end.
+
+-spec inflateChunk(Z) -> Decompressed | {more, Decompressed} when
+      Z :: zstream(),
+      Decompressed :: iolist().
+inflateChunk(Z) ->
+    Status = call(Z, ?INFLATE_CHUNK, []),
+    Data = receive
+	{Z, {data, Bin}} ->
+	    Bin
+    after 0 ->
+	    []
+    end,
+
+    case Status of
+        Good when (Good == ok) orelse (Good == stream_end) ->
+            Data;
+        inflate_has_more ->
+            {more, Data}
+    end.
+
 -spec inflateEnd(Z) -> 'ok' when
       Z :: zstream().
 inflateEnd(Z) ->
@@ -514,7 +549,9 @@ call(Z, Cmd, Arg) ->
 	[2,A,B,C,D] ->
 	    (A bsl 24)+(B bsl 16)+(C bsl 8)+D;
 	[3,A,B,C,D] ->
-	    erlang:error({need_dictionary,(A bsl 24)+(B bsl 16)+(C bsl 8)+D})
+	    erlang:error({need_dictionary,(A bsl 24)+(B bsl 16)+(C bsl 8)+D});
+	[4, _, _, _, _] ->
+	    inflate_has_more
     catch 
 	error:badarg -> %% Rethrow loses port_control from stacktrace.
 	    erlang:error(badarg)