add zstd_stream_init, zstd_stream_add

builtin-functions/kphp-full/_functions.txt

@@ -110,6 +110,10 @@ final class DeflateContext {
     private function __construct() ::: DeflateContext;
 }
 
+final class ZstdCtx {
+    private function __construct() ::: ZstdCtx;
+}
+
 /** @var mixed $_SERVER */
 global $_SERVER;
 /** @var mixed $_GET */
@@ -1453,6 +1457,8 @@ function is_kphp_job_workers_enabled() ::: bool;
 function get_job_workers_number() ::: int;
 
 // zstd api
+function zstd_stream_init(array $options = []) ::: ?ZstdCtx;
+function zstd_stream_add(ZstdCtx $context, string $data, bool $end) ::: string | false;
 function zstd_compress(string $data, int $level = 3) ::: string | false;
 function zstd_uncompress(string $data) ::: string | false;
 function zstd_compress_dict(string $data, string $dict) ::: string | false;

runtime/zstd.cpp

@@ -151,3 +151,62 @@ Optional<string> f$zstd_compress_dict(const string &data, const string &dict) no
 Optional<string> f$zstd_uncompress_dict(const string &data, const string &dict) noexcept {
   return zstd_uncompress_impl(data, dict);
 }
+
+class_instance<C$ZstdCtx> f$zstd_stream_init(const array<mixed> &options) noexcept {
+  ZSTD_CCtxPtr zctx{ZSTD_createCCtx_advanced(make_custom_alloc())};
+  if (!zctx) {
+    php_warning("zstd_compress: can not create context");
+    return {};
+  }
+
+  int lbound = ZSTD_minCLevel();
+  int ubound = ZSTD_maxCLevel();
+  int level = ZSTD_CLEVEL_DEFAULT;
+  for (const auto &option : options) {
+    if (!option.is_string_key()) {
+      php_warning("zstd_stream_init() : unsupported option");
+      return {};
+    }
+    if (option.get_string_key() == string("level")) {
+      mixed value = option.get_value();
+      if (value.is_int() && value.as_int() >= lbound && value.as_int() <= ubound) {
+        level = value.as_int();
+      } else {
+        php_warning("zstd_stream_init() : option %s should be number between %d..%d", option.get_string_key().c_str(), lbound, ubound);
+        return {};
+      }
+    }
+  }
+
+  size_t result = ZSTD_CCtx_setParameter(zctx.get(), ZSTD_c_compressionLevel, static_cast<int>(level));
+  if (ZSTD_isError(result)) {
+    php_warning("zstd_stream_init: can not init context: %s", ZSTD_getErrorName(result));
+    return {};
+  }
+
+  class_instance<C$ZstdCtx> ctx;
+  ctx.alloc();
+  ctx.get()->ctx = zctx.get();
+
+  return ctx;
+}
+
+Optional<string> f$zstd_stream_add(const class_instance<C$ZstdCtx> &ctx, const string &data, bool end) noexcept {
+  php_assert(ZSTD_CStreamOutSize() <= StringLibContext::STATIC_BUFFER_LENGTH);
+  ZSTD_outBuffer out{StringLibContext::get().static_buf.data(), StringLibContext::STATIC_BUFFER_LENGTH, 0};
+  ZSTD_inBuffer in{data.c_str(), data.size(), 0};
+
+  size_t result;
+  ZSTD_EndDirective mode = end ? ZSTD_e_end : ZSTD_e_flush;
+  string encoded_string;
+  do {
+    result = ZSTD_compressStream2(ctx->ctx, &out, &in, mode);
+    if (ZSTD_isError(result)) {
+      php_warning("zstd_compress: got zstd stream compression error: %s", ZSTD_getErrorName(result));
+      return false;
+    }
+    encoded_string.append(static_cast<char *>(out.dst), out.pos);
+    out.pos = 0;
+  } while (result);
+  return encoded_string;
+}

runtime/zstd.h

@@ -4,7 +4,10 @@
 
 #pragma once
 
+#include <zstd.h>
+
 #include "runtime-common/core/runtime-core.h"
+#include "runtime/dummy-visitor-methods.h"
 
 constexpr int DEFAULT_COMPRESS_LEVEL = 3;
 
@@ -15,3 +18,19 @@ Optional<string> f$zstd_uncompress(const string &data) noexcept;
 Optional<string> f$zstd_compress_dict(const string &data, const string &dict) noexcept;
 
 Optional<string> f$zstd_uncompress_dict(const string &data, const string &dict) noexcept;
+
+struct C$ZstdCtx : public refcountable_php_classes<C$ZstdCtx>, private DummyVisitorMethods {
+  C$ZstdCtx() = default;
+  using DummyVisitorMethods::accept;
+
+  ~C$ZstdCtx() {
+    dl::CriticalSectionGuard guard;
+    // TODO
+  }
+
+  ZSTD_CStream *ctx;
+};
+
+class_instance<C$ZstdCtx> f$zstd_stream_init(const array<mixed> &options = {}) noexcept;
+
+Optional<string> f$zstd_stream_add(const class_instance<C$ZstdCtx> &ctx, const string &data, bool end) noexcept;

tests/cpp/runtime/zstd-test.cpp

@@ -2,7 +2,9 @@
 
 #include <zstd.h>
 
+#include "runtime-common/core/runtime-core.h"
 #include "runtime-common/stdlib/string/string-context.h"
+#include "runtime/zstd.h"
 
 TEST(zstd_test, test_bounds) {
   ASSERT_LE(ZSTD_CStreamOutSize(), StringLibContext::STATIC_BUFFER_LENGTH);
@@ -11,3 +13,20 @@ TEST(zstd_test, test_bounds) {
   ASSERT_LE(ZSTD_DStreamOutSize(), StringLibContext::STATIC_BUFFER_LENGTH);
   ASSERT_LE(ZSTD_DStreamInSize(), StringLibContext::STATIC_BUFFER_LENGTH);
 }
+
+TEST(zstd_test, test_compress_decompress) {
+  const char *encoded_string = f$zstd_uncompress(f$zstd_compress(string("sample string"), 3).val()).val().c_str();
+  ASSERT_STREQ("sample string", encoded_string);
+}
+
+TEST(zstd_test, test_stream_compression) {
+  class_instance<C$ZstdCtx> ctx = f$zstd_stream_init();
+  string first_chunk = string("consequat omittam consequat suavitate vocibus contentiones dolores causae appetere eius dicit vis penatibus postulant egestas sanctus mollis verear faucibus salutatus");
+  string second_chunk = string("tacimates imperdiet augue equidem oporteat mentitum sapien aliquam primis assueverit pretium maiorum constituto novum taciti expetenda mauris an euripidis quam");
+  string first_chunk_compressed = f$zstd_stream_add(ctx, first_chunk, false).val();
+  string second_chunk_compressed = f$zstd_stream_add(ctx, second_chunk, true).val();
+
+  string whole = first_chunk_compressed.copy_and_make_not_shared().append(second_chunk_compressed);
+  ASSERT_STREQ(f$zstd_uncompress(first_chunk_compressed).val().c_str(), first_chunk.c_str());
+  ASSERT_STREQ(f$zstd_uncompress(whole).val().c_str(), first_chunk.append(second_chunk).c_str());
+}

tests/python/tests/http_server/php/index.php

@@ -178,6 +178,18 @@ public function work(string $output) {
 
         break;
     }
+} else if ($_SERVER["PHP_SELF"] === "/test_streaming_zstd") {
+    header('Content-Encoding: zstd');
+
+    $ctx = zstd_stream_init();
+
+    $chunk_1 = "justo mea orcidignissim corrumpit iuvaret suscipit ocurreret donec turpis eget eros sonet habitasse vix reprimique litora porro nostra nec vidisse pretium facilis referrentur in fusce sententiae appareat quidam repudiandae doctus periculis hac quem suspendisse facilis habemus nibh nisi ex dignissim sociis cursus habitant habitant taciti auctor graeco no facilis persius erat venenatis mandamus graeci falli ipsum elit tation dico nam donec luctus alia vim mucius urna ceteros dicta ad eius postea sea atomorum tale eros comprehensam detracto scripserit non natoque ligula velit eleifend iaculis donec ridiculus ignota quaestio equidem appareat consequat eros percipit mei molestie appetere patrioque dicta mnesarchum<br />";
+    $chunk_1 = (string)zstd_stream_add($ctx, $chunk_1, false);
+    print $chunk_1;
+
+    $chunk_2 = "libris error cum mollis consectetur usu adversarium vulputate te tractatos gloriatur hinc massa an ponderum delenit persius partiendo elementum per nullam omnesque dolorum sodales lorem commodo magnis maecenas nobis consequat curabitur vidisse potenti eloquentiam risus pretium corrumpit postulant dictumst conclusionemque affert ceteros atqui mutat tantas splendide docendi dolor euismod placerat rhoncus sonet aliquid movet nonumes eius qui ei velit nisl mauris vocent doming maximus ignota definiebas has duo posuere repudiandae nobis ex prompta dicit conubia dolore sed neglegentur graeci ludus suscipiantur cetero suspendisse phasellus molestiae vitae labores ferri equidem autem corrumpit brute finibus tellus risus petentium partiendo epicuri aliquet hendrerit";
+    $chunk_2 = (string)zstd_stream_add($ctx, $chunk_2, true);
+    print $chunk_2;
 } else if ($_SERVER["PHP_SELF"] === "/test_ignore_user_abort") {
     register_shutdown_function('shutdown_function');
     /** @var I */

tests/python/tests/http_server/test_streaming_zstd.py

@@ -0,0 +1,18 @@
+import zstandard
+
+from python.lib.testcase import KphpServerAutoTestCase
+
+class TestStreamingZstd(KphpServerAutoTestCase):
+    def test_streaming_zstd(self):
+        cmp = zstandard.ZstdDecompressor()
+
+        url = "/test_streaming_zstd"
+        response = self.kphp_server.http_get(url, headers={
+            "Host": "localhost",
+            "Accept-Encoding": "zstd"
+        })
+        self.assertEqual(response.headers["Content-Encoding"], "zstd")
+        stream_reader = cmp.stream_reader(response.content)
+        uncompressed_body = stream_reader.read().decode('utf-8')
+        stream_reader.close()
+        self.assertEqual(uncompressed_body, """justo mea orcidignissim corrumpit iuvaret suscipit ocurreret donec turpis eget eros sonet habitasse vix reprimique litora porro nostra nec vidisse pretium facilis referrentur in fusce sententiae appareat quidam repudiandae doctus periculis hac quem suspendisse facilis habemus nibh nisi ex dignissim sociis cursus habitant habitant taciti auctor graeco no facilis persius erat venenatis mandamus graeci falli ipsum elit tation dico nam donec luctus alia vim mucius urna ceteros dicta ad eius postea sea atomorum tale eros comprehensam detracto scripserit non natoque ligula velit eleifend iaculis donec ridiculus ignota quaestio equidem appareat consequat eros percipit mei molestie appetere patrioque dicta mnesarchum<br />libris error cum mollis consectetur usu adversarium vulputate te tractatos gloriatur hinc massa an ponderum delenit persius partiendo elementum per nullam omnesque dolorum sodales lorem commodo magnis maecenas nobis consequat curabitur vidisse potenti eloquentiam risus pretium corrumpit postulant dictumst conclusionemque affert ceteros atqui mutat tantas splendide docendi dolor euismod placerat rhoncus sonet aliquid movet nonumes eius qui ei velit nisl mauris vocent doming maximus ignota definiebas has duo posuere repudiandae nobis ex prompta dicit conubia dolore sed neglegentur graeci ludus suscipiantur cetero suspendisse phasellus molestiae vitae labores ferri equidem autem corrumpit brute finibus tellus risus petentium partiendo epicuri aliquet hendrerit""")