将 AWS Lambda C++ 运行时用作 Lambda 层

引言

在 AWS Lambda C++ Runtime 中，实现了一个用于 AWS Lambda 的 C++ Runtime。您可以在本地创建 C++ 的 AWS Lambda 函数，对其进行编译，创建一个包含可执行文件和 C++ Runtime 的 zip 包，然后将整个包上传到 AWS。这个上传包的体积相当大。为了避免每次构建新的 Lambda C++ 可执行文件时都上传整个 Runtime，Amazon 创建了层 (layers) 的概念。

背景

如果您想了解 Lambda 层的一些基本知识，请点击下面的链接。

使用 Lambda 层

您可能考虑使用层的多个原因

• 减小部署包的大小。 ...
• 将核心函数逻辑与依赖项分离。 ...
• 跨多个函数共享依赖项。 ...

示例应用程序 本指南的 GitHub 仓库提供了空白示例应用程序，演示了如何使用层进行依赖项管理。
Node.js – blank-nodejs
Python – blank-python
Java – blank-java
Ruby – blank-ruby

这里我缺少的是

C++
其他编译型语言

在 AWS Lambda C++ Runtime 的 FAQ & 故障排除中，您可以阅读

1. 为什么 zip 文件这么大？那些都是什么文件？通常，zip 文件之所以大，是因为我们必须打包整个 C 标准库。您可以通过执行以下一项或多项操作来减小文件大小：

但是没有提供像层这样的解决方案。

我在网上找不到任何关于如何实现这一点的文章。因此，我将展示我自己的实现方法。

首先

为什么用 C/C++ 实现 Lambda 函数？

这是一个自定义 Alexa 技能，用于打开和关闭灯，比较了 DurationInMS 和 BilledDurationInMS。一次作为 Python Lambda 函数，第二次作为编译的 C++ Lambda 函数。两者都使用它们在层中的 Runtime。一次使用 Python Runtime 层，第二次使用上面的 C++ Runtime。

该 Lambda 函数是自定义 Alexa 技能“Mein Haus”（我的家）的端点，该技能也是调用短语。

使用的语句是

Licht an 和 Licht aus（开灯，关灯）

其中 an 和 aus 是名为 state 的 OnOff 插槽的值。

以下是“Mein Haus”对话的一个示例

    
// the wake word

// invocation phrase

// response of LaunchRequest

// 1. LichtRequest

// response to 1. LichtRequest

// waiting for ask response of LichtRequest

// comes here

// 2. LichtRequest

// response to 2 LichtRequest

// StopRequest

// response of StopRequest

 

这是控制台中此内容的 Python Lambda 定义

使用此源代码

#- coding: utf-8 -*-

# This sample demonstrates handling intents from an Alexa skill using the Alexa Skills Kit SDK for Python.
# Please visit https://alexa.design/cookbook for additional examples on implementing slots, dialog management,
# session persistence, api calls, and more.
# This sample is built using the handler classes approach in skill builder.
import logging
import ask_sdk_core.utils as ask_utils

from ask_sdk_core.skill_builder import SkillBuilder
from ask_sdk_core.dispatch_components import AbstractRequestHandler
from ask_sdk_core.dispatch_components import AbstractExceptionHandler
from ask_sdk_core.handler_input import HandlerInput
from ask_sdk_model import Response
from ask_sdk_model.slu.entityresolution.status_code import StatusCode

import socket
import struct

UDP_HOST = '<host>'
UDP_PORT =  <port>
  
class LaunchRequestHandler(AbstractRequestHandler):
    """Handler for Skill Launch."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("LaunchRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        return (
            handler_input.response_builder
                .speak('sprich!')
                .ask('was soll ich tun?')
                .response
        )

def getSlotId( slot ):
    if slot.resolutions and len(slot.resolutions.resolutions_per_authority)>0:
        for resolution in  slot.resolutions.resolutions_per_authority:
            if resolution.status and resolution.status.code == StatusCode.ER_SUCCESS_MATCH:
                return resolution.values[0].value.id

class LichtRequestHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        return ask_utils.is_intent_name("licht")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        
        slots = handler_input.request_envelope.request.intent.slots
        if 'state' in slots:
            stateId = getSlotId( slots['state'] )

        if stateId == 'ON':
            method = 'TurnOn'
        else:
            method = 'TurnOff'
        
        """
        message = {
            "device" : 'light',
            "topic"  : 'Alexa.PowerController',
            "method" : method,
            "payload" : {}
        }
        """
        message='{"device":"light","topic":"Alexa.PowerController","method":"%s","payload":{}}' % method
        
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
          s.sendto( message.encode(),(UDP_HOST, UDP_PORT))

        return (
            handler_input.response_builder
                .speak('OK! Schalte Licht ' + slots['state'].value)
                .ask('und weiter?')
                .response
        )

class CancelOrStopIntentHandler(AbstractRequestHandler):
    """Single handler for Cancel and Stop Intent."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return (ask_utils.is_intent_name("AMAZON.CancelIntent")(handler_input) or
                ask_utils.is_intent_name("AMAZON.StopIntent")(handler_input))

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        return (
            handler_input.response_builder
                .speak("Goodbye!")
                .response
        )

class FallbackIntentHandler(AbstractRequestHandler):
    
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.FallbackIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        return (
            handler_input.response_builder
                .speak("Hmm, Ich bin mir nicht sicher. Bitte wiederhole deine Absicht.")
                .ask("Bitte wiederhole deine Absicht.")
                .response
        )

class SessionEndedRequestHandler(AbstractRequestHandler):
    """Handler for Session End."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("SessionEndedRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        # Any cleanup logic goes here.

        return handler_input.response_builder.response

class CatchAllExceptionHandler(AbstractExceptionHandler):
    """Generic error handling to capture any syntax or routing errors. If you receive an error
    stating the request handler chain is not found, you have not implemented a handler for
    the intent being invoked or included it in the skill builder below.
    """
    def can_handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> bool
        return True

    def handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> Response
        logger.error(exception, exc_info=True)

        speak_output = "Sorry, Ich habe Schwierigkeiten mit dem was du sagst. Bitte versuche es erneut."

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )

# The SkillBuilder object acts as the entry point for your skill, routing all request and response
# payloads to the handlers above. Make sure any new handlers or interceptors you've
# defined are included below. The order matters - they're processed top to bottom.

sb = SkillBuilder()

sb.add_request_handler(LichtRequestHandler())
sb.add_request_handler(LaunchRequestHandler())
sb.add_request_handler(CancelOrStopIntentHandler())
sb.add_request_handler(FallbackIntentHandler())
sb.add_request_handler(SessionEndedRequestHandler())
sb.add_exception_handler(CatchAllExceptionHandler())

lambda_handler = sb.lambda_handler()

LichtRequestHandler 类的 handle() 方法首先确定 state 插槽的值/ID。然后，通过 UDP 将包含收件人控制数据的消息（json-object）发送到主机（它是我 24/7 运行的树莓派通过 No-IP 的 IP 地址）上的一个特殊端口。在 Raspi 上，有一个服务在该端口上监听。该服务通过 Node-Red 节点实现。其他 Node-Red 节点将执行所有操作：解析接收到的（json）消息，为 Zigbee2MQTT 生成命令。USB-Zigbee-Coordinator 将这些命令发送到 Zigbee 控制的 LED 灯带。

以下是上述对话中 Python 和 C++ 解决方案的测量毫秒数和“计费毫秒数”

您可以看到对话中 4 个请求（1 次 LaunchRequest，2 次 LichtRequest，1 次 StopRequest）的毫秒数，共两个会话 - 黄色和绿色标记。

Python

C++

正如您所见，在“计费毫秒数”方面，有充分的理由偏好基于 C++ 等编译型语言的解决方案。

如何将 C++ Runtime 放入层

前往 AWS Lambda C++ Runtime 并按照说明构建 Runtime 以及演示示例。我已经在我的树莓派上完成了。当演示运行正常（42！）时，我们将修改一些内容。首先，我们将 Runtime 与应用程序分离。我们的目标是创建一个仅包含 Runtime（在lib 文件夹下）的 zip 包，以及另一个包含应用程序的 zip 包。

1. 从 zip 包中提取演示。
2. 提取bootstrap 并替换为修改后的版本。

   #!/bin/bash
   set -euo pipefail
   export AWS_EXECUTION_ENV=lambda-cpp
   exec $LAMBDA_TASK_ROOT/lib/ld-linux-aarch64.so.1 --library-path $LAMBDA_TASK_ROOT/lib $LAMBDA_TASK_ROOT/bin/demo ${_HANDLER}

   修改后的bootstrap 现在看起来是

   #!/bin/bash
   set -euo pipefail
   export AWS_EXECUTION_ENV=lambda-cpp
   exec /opt/lib/ld-linux-aarch64.so.1 --library-path /opt/lib $LAMBDA_TASK_ROOT/bin/lambda_handler ${_HANDLER}

   1. 提取bootstrap
   2. 编辑bootstrap
      1. 将粗体标记的前两个 $LAMBDA_TASK_ROOT 替换为 /opt
      2. 现在，我们为将来使用此 Runtime 的所有应用程序选择一个通用名称。例如，将其命名为 lambda_handler。在 bootstrap 中将 demo 替换为 lambda_handler。
   3. 将 bootstrap 重新插入 zip 包，并将其命名为，例如，lambda_handler.zip。

      我们的 zip 包lambda_handler.zip 现在看起来是

      lib/<runtime module>
      bootstrap

3. 使用此 zip 包安装层。

   您可以通过 CLI 或 Lambda 控制台完成。我将展示通过控制台的方式。

   登录到 Lambda 控制台的“应用程序”页面。

   AWS Lambda

   （接下来，您将在此处看到此页面的德语本地化屏幕截图）

   

   （右侧显示了我在此处演示的 Lambda 函数。）

   在左侧，点击 **Layer**

   

   现在，在右侧点击 **Layer erstellen/create layer**。

   

   填写表单字段

   1. 名称： cpp-runtime
   2. 我们将上传一个 zip 包。
   3. 在打开的文件浏览器中选择 zip 文件。
   4. 选择架构 arm64 （如果您为此架构编译了 Runtime；我的在树莓派上构建）
   5. 选择 Runtime 和 **Amazon Linux 2023**。

创建使用此层的 Lambda 函数

创建一个名为 lambda_handler.zip 的 zip 包

bin/lambda_handler

这里的lambda-handler 是一个编译过的可执行文件（以前的demo）。

可选地，您可以添加一个src/ 文件夹，其中包含源代码。这样 zip 包看起来就像这样

bin/lambda_handler

src/lambda_handler.cpp

此src 文件夹在调用 Lambda 函数时没有影响。它仅供您方便起见，以便源代码靠近可执行文件，以便在执行 Lambda 函数时快速查看发生了什么。请参见下文。

现在，像往常一样创建一个 Lambda 函数，例如 test-cpp-skill。作为 Runtime，选择 **Amazon Linux 2023**，作为架构 **arm64**。请注意，当您为这种 CPU 架构编译时，例如在树莓派上，您必须指定 **arm64**。

点击 **Hochladen von** 并上传 zip 包。您将看到这个

**注意**：您在这里有可选的源代码，但与 Python 不同，您不能修改它然后只需点击 **Deploy**。您必须在本地编译源代码的任何更改，创建 zip 包并再次上传。（当通过 CLI 完成此上传时，请不要忘记刷新此 Lambda 函数的控制台页面。）

现在将层分配给函数。在“Code”选项卡被选中的情况下，向下滚动页面到最后。

黄色标记的是已分配的层。要实现这一点，请点击 **Einen Layer hinzufügen**。

在打开的表单中，选择 **Benutzerdefinierte Ebenen**，然后在打开的文件浏览器中选择您上面生成的 cpp-runtime，然后选择一个版本。完成后，点击 **Hinzufügen**。

结论

可以将 C++ Runtime 放入层，并从 C++ 编译的 Lambda 函数中使用此层。C++ 编译的 Lambda 函数在“billedMS”方面具有巨大优势。

历史

• 2024 年 2 月 21 日：初始版本