Add transform TSIOBuffer watermark APIs.

doc/admin-guide/plugins/lua.en.rst

@@ -2234,6 +2234,30 @@ Here is an example:
         return 0
     end
 
+`TOP <#ts-lua-plugin>`_
+
+ts.http.resp_transform.get_upstream_watermark_bytes
+---------------------------------------------------
+**syntax:** *ts.http.resp_transform.get_upstream_watermark_bytes()*
+
+**context:** transform handler
+
+**description**: This function can be used to retrive the current watermark bytes for the upstream transform buffer.
+
+
+`TOP <#ts-lua-plugin>`_
+
+ts.http.resp_transform.set_upstream_watermark_bytes
+---------------------------------------------------
+**syntax:** *ts.http.resp_transform.set_upstream_watermark_bytes(NUMBER)*
+
+**context:** transform handler
+
+**description**: This function can be used to set the watermark bytes of the upstream transform buffer. 
+
+Setting the watermark bytes above 32kb may improve the performance of the transform handler.
+
+
 `TOP <#ts-lua-plugin>`_
 
 ts.http.resp_transform.set_downstream_bytes

plugins/lua/ts_lua_common.h

@@ -144,6 +144,7 @@ typedef struct {
 
   ts_lua_http_ctx *hctx;
   int64_t upstream_bytes;
+  int64_t upstream_watermark_bytes;
   int64_t downstream_bytes;
   int64_t total;
 

plugins/lua/ts_lua_http.c

@@ -105,6 +105,8 @@ static void ts_lua_inject_server_state_variables(lua_State *L);
 
 static void ts_lua_inject_http_resp_transform_api(lua_State *L);
 static int ts_lua_http_resp_transform_get_upstream_bytes(lua_State *L);
+static int ts_lua_http_resp_transform_get_upstream_watermark_bytes(lua_State *L);
+static int ts_lua_http_resp_transform_set_upstream_watermark_bytes(lua_State *L);
 static int ts_lua_http_resp_transform_set_downstream_bytes(lua_State *L);
 
 void
@@ -194,6 +196,12 @@ ts_lua_inject_http_resp_transform_api(lua_State *L)
   lua_pushcfunction(L, ts_lua_http_resp_transform_get_upstream_bytes);
   lua_setfield(L, -2, "get_upstream_bytes");
 
+  lua_pushcfunction(L, ts_lua_http_resp_transform_get_upstream_watermark_bytes);
+  lua_setfield(L, -2, "get_upstream_watermark_bytes");
+
+  lua_pushcfunction(L, ts_lua_http_resp_transform_set_upstream_watermark_bytes);
+  lua_setfield(L, -2, "set_upstream_watermark_bytes");
+
   lua_pushcfunction(L, ts_lua_http_resp_transform_set_downstream_bytes);
   lua_setfield(L, -2, "set_downstream_bytes");
 }
@@ -884,6 +892,41 @@ ts_lua_http_resp_transform_get_upstream_bytes(lua_State *L)
   return 1;
 }
 
+static int
+ts_lua_http_resp_transform_get_upstream_watermark_bytes(lua_State *L)
+{
+  ts_lua_http_transform_ctx *transform_ctx;
+
+  transform_ctx = ts_lua_get_http_transform_ctx(L);
+  if (transform_ctx == NULL) {
+    TSError("[ts_lua] missing transform_ctx");
+    return 0;
+  }
+
+  lua_pushnumber(L, transform_ctx->upstream_watermark_bytes);
+
+  return 1;
+}
+
+static int
+ts_lua_http_resp_transform_set_upstream_watermark_bytes(lua_State *L)
+{
+  int64_t n;
+  ts_lua_http_transform_ctx *transform_ctx;
+
+  transform_ctx = ts_lua_get_http_transform_ctx(L);
+  if (transform_ctx == NULL) {
+    TSError("[ts_lua] missing transform_ctx");
+    return 0;
+  }
+
+  n = luaL_checkinteger(L, 1);
+
+  transform_ctx->upstream_watermark_bytes = n;
+
+  return 0;
+}
+
 static int
 ts_lua_http_resp_transform_set_downstream_bytes(lua_State *L)
 {

plugins/lua/ts_lua_transform.c

@@ -66,7 +66,7 @@ ts_lua_transform_handler(TSCont contp, ts_lua_http_transform_ctx *transform_ctx,
   TSVIO input_vio;
   TSIOBufferReader input_reader;
   TSIOBufferBlock blk;
-  int64_t toread, towrite, blk_len, upstream_done, input_avail, l;
+  int64_t toread, towrite, blk_len, upstream_done, input_avail, input_wm_bytes, l;
   const char *start;
   const char *res;
   size_t res_len;
@@ -97,6 +97,13 @@ ts_lua_transform_handler(TSCont contp, ts_lua_http_transform_ctx *transform_ctx,
       TSDebug(TS_LUA_DEBUG_TAG, "[%s] no input VIO and output VIO", __FUNCTION__);
       empty_input = 1;
     }
+  } else { // input VIO exists
+    input_wm_bytes = TSIOBufferWaterMarkGet(TSVIOBufferGet(input_vio));
+    if (transform_ctx->upstream_watermark_bytes >= 0 && transform_ctx->upstream_watermark_bytes != input_wm_bytes) {
+      TSDebug(TS_LUA_DEBUG_TAG, "[%s] Setting input_vio watermark to %" PRId64 " bytes", __FUNCTION__,
+              transform_ctx->upstream_watermark_bytes);
+      TSIOBufferWaterMarkSet(TSVIOBufferGet(input_vio), transform_ctx->upstream_watermark_bytes);
+    }
   }
 
   if (empty_input == 0) {