Shared memory data transfer between Functions Host and Python worker

.ci/perf_tests/dockerfiles/perf_tests.Dockerfile

@@ -1,6 +1,6 @@
 ARG PYTHON_VERSION=3.8
 
-FROM mcr.microsoft.com/azure-functions/python:3.0.14492-python$PYTHON_VERSION
+FROM mcr.microsoft.com/azure-functions/python:3.0.15418-python$PYTHON_VERSION
 
 # Mounting local machines azure-functions-python-worker and azure-functions-python-library onto it
 RUN rm -rf /azure-functions-host/workers/python/${PYTHON_VERSION}/LINUX/X64/azure_functions_worker
@@ -11,7 +11,8 @@ VOLUME ["/azure-functions-host/workers/python/${PYTHON_VERSION}/LINUX/X64/azure_
 ENV AzureWebJobsScriptRoot=/home/site/wwwroot \
     AzureFunctionsJobHost__Logging__Console__IsEnabled=true \
     FUNCTIONS_WORKER_PROCESS_COUNT=1 \
-    AZURE_FUNCTIONS_ENVIRONMENT=Development
+    AZURE_FUNCTIONS_ENVIRONMENT=Development \
+    FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED=1
 
 RUN apt-get --quiet update && \
     apt-get install --quiet -y git && \

.ci/perf_tests/k6scripts/SyncGetBlobAsBytesReturnHttpResponse.js

@@ -0,0 +1,65 @@
+import { check } from "k6";
+import { Rate } from "k6/metrics";
+import http from "k6/http";
+
+var HOSTNAME = __ENV.HOSTNAME || 'localhost';
+var PORT = __ENV.PORT || '80';
+var PROTOCOL = __ENV.PROTOCOL || (PORT === '80' ? 'http' : 'https');
+var INPUT_FILENAME = 'Input_256MB'
+var CONTENT_SIZE = 1024 * 1024 * 256; // 256 MB
+
+// A custom metric to track failure rates
+var failureRate = new Rate("check_failure_rate");
+
+// Options
+export let options = {
+    stages: [
+        // Linearly ramp up from 1 to 20 VUs during first minute
+        { target: 20, duration: "1m" },
+        // Hold at 20 VUs for the next 3 minutes and 45 seconds
+        { target: 20, duration: "3m45s" },
+        // Linearly ramp down from 20 to 0 VUs over the last 15 seconds
+        { target: 0, duration: "15s" }
+        // Total execution time will be ~5 minutes
+    ],
+    thresholds: {
+        // We want the 95th percentile of all HTTP request durations to be less than 40s
+        "http_req_duration": ["p(95)<40000"],
+        // Thresholds based on the custom metric we defined and use to track application failures
+        "check_failure_rate": [
+            // Global failure rate should be less than 1%
+            "rate<0.01",
+            // Abort the test early if it climbs over 5%
+            { threshold: "rate<=0.05", abortOnFail: true },
+        ],
+    },
+};
+
+// Setup function
+// This will create a blob which will later be used as an input binding
+export function setup() {
+    let no_random_input = true;
+    let url = `${PROTOCOL}://${HOSTNAME}:${PORT}/api/SyncPutBlobAsBytesReturnHttpResponse?content_size=${CONTENT_SIZE}&no_random_input=${no_random_input}&outfile=${INPUT_FILENAME}`;
+    let response = http.get(url);
+
+    // check() returns false if any of the specified conditions fail
+    let checkRes = check(response, {
+        "status is 200": (r) => r.status === 200,
+        "content_size matches": (r) => r.json().content_size === CONTENT_SIZE,
+    });
+}
+
+// Main function
+export default function () {
+    let url = `${PROTOCOL}://${HOSTNAME}:${PORT}/api/SyncGetBlobAsBytesReturnHttpResponse?infile=${INPUT_FILENAME}`;
+    let response = http.get(url);
+
+    // check() returns false if any of the specified conditions fail
+    let checkRes = check(response, {
+        "status is 200": (r) => r.status === 200,
+        "content_size matches": (r) => r.json().content_size === CONTENT_SIZE,
+    });
+
+    // We reverse the check() result since we want to count the failures
+    failureRate.add(!checkRes);
+}

.ci/perf_tests/k6scripts/SyncPutBlobAsBytesReturnHttpResponse.js

@@ -0,0 +1,53 @@
+import { check } from "k6";
+import { Rate } from "k6/metrics";
+import http from "k6/http";
+import { randomIntBetween } from "https://jslib.k6.io/k6-utils/1.0.0/index.js";
+
+var HOSTNAME = __ENV.HOSTNAME || 'localhost';
+var PORT = __ENV.PORT || '80';
+var PROTOCOL = __ENV.PROTOCOL || (PORT === '80' ? 'http' : 'https');
+
+// A custom metric to track failure rates
+var failureRate = new Rate("check_failure_rate");
+
+// Options
+export let options = {
+    stages: [
+        // Linearly ramp up from 1 to 50 VUs during first minute
+        { target: 50, duration: "1m" },
+        // Hold at 50 VUs for the next 3 minutes and 45 seconds
+        { target: 50, duration: "3m45s" },
+        // Linearly ramp down from 50 to 0 VUs over the last 15 seconds
+        { target: 0, duration: "15s" }
+        // Total execution time will be ~5 minutes
+    ],
+    thresholds: {
+        // We want the 95th percentile of all HTTP request durations to be less than 40s
+        "http_req_duration": ["p(95)<40000"],
+        // Thresholds based on the custom metric we defined and use to track application failures
+        "check_failure_rate": [
+            // Global failure rate should be less than 1%
+            "rate<0.01",
+            // Abort the test early if it climbs over 5%
+            { threshold: "rate<=0.05", abortOnFail: true },
+        ],
+    },
+};
+
+// Main function
+export default function () {
+    let content_size = 1024 * 1024 * 256; // 256 MB
+    let no_random_input = true;
+    let outfile = randomIntBetween(1,500000);
+    let url = `${PROTOCOL}://${HOSTNAME}:${PORT}/api/SyncPutBlobAsBytesReturnHttpResponse?content_size=${content_size}&no_random_input=${no_random_input}&outfile=${outfile}`;
+    let response = http.get(url);
+
+    // check() returns false if any of the specified conditions fail
+    let checkRes = check(response, {
+        "status is 200": (r) => r.status === 200,
+        "content_size matches": (r) => r.json().content_size === content_size,
+    });
+
+    // We reverse the check() result since we want to count the failures
+    failureRate.add(!checkRes);
+}

.github/workflows/perf-testing-setup.yml

@@ -6,7 +6,7 @@ on:
     branches: [ dev ]
 
 env:
-  PERF_TESTS_LINK: "https://raw.githubusercontent.com/Azure/azure-functions-python-worker/dev/.ci/perf_tests/k6scripts/"
+  TESTS_DIR_PATH: ".ci/perf_tests/k6scripts/"
   PYTHON_VERSION: "3.8"
   PORT: 8000
 
@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test_to_run: [ SyncHttpTriggerHelloWorld, SyncHttpTriggerWithSyncRequests, AsyncHttpTriggerWithAsyncRequest, SyncHttpTriggerCPUIntensive ]
+        test_to_run: [ SyncHttpTriggerHelloWorld, SyncHttpTriggerWithSyncRequests, AsyncHttpTriggerWithAsyncRequest, SyncHttpTriggerCPUIntensive, SyncPutBlobAsBytesReturnHttpResponse, SyncGetBlobAsBytesReturnHttpResponse ]
     steps:
     - uses: actions/checkout@v2
     - name: Set up Python ${{ env.PYTHON_VERSION }}
@@ -44,12 +44,12 @@ jobs:
     - name: Build and Run the Docker image
       run: |
         docker build --build-arg PYTHON_VERSION=${{ env.PYTHON_VERSION }} --file .ci/perf_tests/dockerfiles/perf_tests.Dockerfile --tag perfimage:latest .
-        docker run -d --env FUNCTIONS_WORKER_RUNTIME_VERSION=${{ env.PYTHON_VERSION }} -p ${PORT}:80 -v $GITHUB_WORKSPACE/azure_functions_worker:/azure-functions-host/workers/python/${{ env.PYTHON_VERSION }}/LINUX/X64/azure_functions_worker perfimage:latest
+        docker run -d --shm-size="2g" --env FUNCTIONS_WORKER_RUNTIME_VERSION=${{ env.PYTHON_VERSION }} -p ${PORT}:80 -v $GITHUB_WORKSPACE/azure_functions_worker:/azure-functions-host/workers/python/${{ env.PYTHON_VERSION }}/LINUX/X64/azure_functions_worker perfimage:latest
         sleep 10  # host needs some time to start.
     - name: Validate if the functions are now running
       run: |
         curl --get http://localhost:${PORT}/api/${{ matrix.test_to_run }}
     - name: Run Throughput tests
       run: |
         chmod 755 .ci/perf_tests/run-perftests.sh
-        .ci/perf_tests/run-perftests.sh localhost $PORT $PERF_TESTS_LINK ${{ matrix.test_to_run }}
+        .ci/perf_tests/run-perftests.sh localhost $PORT ${{ env.TESTS_DIR_PATH }} ${{ matrix.test_to_run }}

azure_functions_worker/bindings/__init__.py

@@ -6,7 +6,8 @@
 from .meta import check_output_type_annotation
 from .meta import has_implicit_output
 from .meta import is_trigger_binding
-from .meta import from_incoming_proto, to_outgoing_proto
+from .meta import from_incoming_proto, to_outgoing_proto, \
+    to_outgoing_param_binding
 from .out import Out
 
 
@@ -16,4 +17,5 @@
     'check_input_type_annotation', 'check_output_type_annotation',
     'has_implicit_output',
     'from_incoming_proto', 'to_outgoing_proto', 'TraceContext',
+    'to_outgoing_param_binding'
 )

azure_functions_worker/bindings/datumdef.py

@@ -1,9 +1,10 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
-from typing import Any
+from typing import Any, Optional
 import json
 from .. import protos
+from ..logging import logger
 
 
 class Datum:
@@ -92,6 +93,81 @@ def from_typed_data(cls, td: protos.TypedData):
 
         return cls(val, tt)
 
+    @classmethod
+    def from_rpc_shared_memory(
+            cls,
+            shmem: protos.RpcSharedMemory,
+            shmem_mgr) -> Optional['Datum']:
+        """
+        Reads the specified shared memory region and converts the read data into
+        a datum object of the corresponding type.
+        """
+        if shmem is None:
+            logger.warning('Cannot read from shared memory. '
+                           'RpcSharedMemory is None.')
+            return None
+
+        mem_map_name = shmem.name
+        offset = shmem.offset
+        count = shmem.count
+        data_type = shmem.type
+        ret_val = None
+
+        if data_type == protos.RpcDataType.bytes:
+            val = shmem_mgr.get_bytes(mem_map_name, offset, count)
+            if val is not None:
+                ret_val = cls(val, 'bytes')
+        elif data_type == protos.RpcDataType.string:
+            val = shmem_mgr.get_string(mem_map_name, offset, count)
+            if val is not None:
+                ret_val = cls(val, 'string')
+
+        if ret_val is not None:
+            logger.info(
+                f'Read {count} bytes from memory map {mem_map_name} '
+                f'for data type {data_type}')
+            return ret_val
+        return None
+
+    @classmethod
+    def to_rpc_shared_memory(
+            cls,
+            datum: 'Datum',
+            shmem_mgr) -> Optional[protos.RpcSharedMemory]:
+        """
+        Writes the given value to shared memory and returns the corresponding
+        RpcSharedMemory object which can be sent back to the functions host over
+        RPC.
+        """
+        if datum.type == 'bytes':
+            value = datum.value
+            shared_mem_meta = shmem_mgr.put_bytes(value)
+            data_type = protos.RpcDataType.bytes
+        elif datum.type == 'string':
+            value = datum.value
+            shared_mem_meta = shmem_mgr.put_string(value)
+            data_type = protos.RpcDataType.string
+        else:
+            raise NotImplementedError(
+                f'Unsupported datum type ({datum.type}) for shared memory'
+            )
+
+        if shared_mem_meta is None:
+            logger.warning('Cannot write to shared memory for type: '
+                           f'{datum.type}')
+            return None
+
+        shmem = protos.RpcSharedMemory(
+            name=shared_mem_meta.mem_map_name,
+            offset=0,
+            count=shared_mem_meta.count_bytes,
+            type=data_type)
+
+        logger.info(
+            f'Wrote {shared_mem_meta.count_bytes} bytes to memory map '
+            f'{shared_mem_meta.mem_map_name} for data type {data_type}')
+        return shmem
+
 
 def datum_as_proto(datum: Datum) -> protos.TypedData:
     if datum.type == 'string':

azure_functions_worker/bindings/meta.py

@@ -8,6 +8,10 @@
 from . import datumdef
 from . import generic
 
+PB_TYPE = 'rpc_data'
+PB_TYPE_DATA = 'data'
+PB_TYPE_RPC_SHARED_MEMORY = 'rpc_shared_memory'
+
 
 def get_binding_registry():
     func = sys.modules.get('azure.functions')
@@ -55,13 +59,11 @@ def has_implicit_output(bind_name: str) -> bool:
 
 def from_incoming_proto(
         binding: str,
-        val: protos.TypedData, *,
+        pb: protos.ParameterBinding, *,
         pytype: typing.Optional[type],
-        trigger_metadata: typing.Optional[typing.Dict[str, protos.TypedData]])\
-        -> typing.Any:
-
+        trigger_metadata: typing.Optional[typing.Dict[str, protos.TypedData]],
+        shmem_mgr) -> typing.Any:
     binding = get_binding(binding)
-    datum = datumdef.Datum.from_typed_data(val)
     if trigger_metadata:
         metadata = {
             k: datumdef.Datum.from_typed_data(v)
@@ -70,22 +72,34 @@ def from_incoming_proto(
     else:
         metadata = {}
 
+    pb_type = pb.WhichOneof(PB_TYPE)
+    if pb_type == PB_TYPE_DATA:
+        val = pb.data
+        datum = datumdef.Datum.from_typed_data(val)
+    elif pb_type == PB_TYPE_RPC_SHARED_MEMORY:
+        # Data was sent over shared memory, attempt to read
+        datum = datumdef.Datum.from_rpc_shared_memory(pb.rpc_shared_memory,
+                                                      shmem_mgr)
+    else:
+        raise TypeError(f'Unknown ParameterBindingType: {pb_type}')
+
     try:
         return binding.decode(datum, trigger_metadata=metadata)
     except NotImplementedError:
         # Binding does not support the data.
         dt = val.WhichOneof('data')
-
         raise TypeError(
             f'unable to decode incoming TypedData: '
             f'unsupported combination of TypedData field {dt!r} '
             f'and expected binding type {binding}')
 
 
-def to_outgoing_proto(binding: str, obj: typing.Any, *,
-                      pytype: typing.Optional[type]) -> protos.TypedData:
+def get_datum(binding: str, obj: typing.Any,
+              pytype: typing.Optional[type]) -> datumdef.Datum:
+    """
+    Convert an object to a datum with the specified type.
+    """
     binding = get_binding(binding)
-
     try:
         datum = binding.encode(obj, expected_type=pytype)
     except NotImplementedError:
@@ -94,5 +108,37 @@ def to_outgoing_proto(binding: str, obj: typing.Any, *,
             f'unable to encode outgoing TypedData: '
             f'unsupported type "{binding}" for '
             f'Python type "{type(obj).__name__}"')
+    return datum
+
 
+def to_outgoing_proto(binding: str, obj: typing.Any, *,
+                      pytype: typing.Optional[type]) -> protos.TypedData:
+    datum = get_datum(binding, obj, pytype)
     return datumdef.datum_as_proto(datum)
+
+
+def to_outgoing_param_binding(binding: str, obj: typing.Any, *,
+                              pytype: typing.Optional[type],
+                              out_name: str,
+                              shmem_mgr) \
+        -> protos.ParameterBinding:
+    datum = get_datum(binding, obj, pytype)
+    shared_mem_value = None
+    # If shared memory is enabled and supported for the given datum, try to
+    # transfer to host over shared memory as a default
+    if shmem_mgr.is_enabled() and shmem_mgr.is_supported(datum):
+        shared_mem_value = datumdef.Datum.to_rpc_shared_memory(datum, shmem_mgr)
+    # Check if data was written into shared memory
+    if shared_mem_value is not None:
+        # If it was, then use the rpc_shared_memory field in response message
+        return protos.ParameterBinding(
+            name=out_name,
+            rpc_shared_memory=shared_mem_value)
+    else:
+        # If not, send it as part of the response message over RPC
+        rpc_val = datumdef.datum_as_proto(datum)
+        if rpc_val is None:
+            raise TypeError('Cannot convert datum to rpc_val')
+        return protos.ParameterBinding(
+            name=out_name,
+            data=rpc_val)