** * An operator that executes the actions defined in the agent. Upon receiving data from the * upstream, it first wraps the data into an { @link InputEvent}. It then invokes the corresponding * action that is interested in the { @link InputEvent}, and collects the output event produced by * the action. * * <p>For events of type { @link OutputEvent}, the data contained in the event is sent downstream. * For all other event types, the process is repeated: the event triggers the corresponding action, * and the resulting output event is collected for further processing. */ public class ActionExecutionOperator <IN, OUT> extends AbstractStreamOperator <OUT>         implements OneInputStreamOperator <IN, OUT>, BoundedOneInput {     private static final long serialVersionUID = 1L ;     private static final Logger LOG = LoggerFactory.getLogger(ActionExecutionOperator.class);     private static final String RECOVERY_MARKER_STATE_NAME = "recoveryMarker" ;     private static final String MESSAGE_SEQUENCE_NUMBER_STATE_NAME = "messageSequenceNumber" ;     private static final String PENDING_INPUT_EVENT_STATE_NAME = "pendingInputEvents" ;     private final AgentPlan agentPlan;     private final Boolean inputIsJava;     private transient StreamRecord<OUT> reusedStreamRecord;     private transient MapState<String, MemoryObjectImpl.MemoryItem> shortTermMemState;     // PythonActionExecutor for Python actions     private transient PythonActionExecutor pythonActionExecutor;     private transient FlinkAgentsMetricGroupImpl metricGroup;     private transient BuiltInMetrics builtInMetrics;     private transient SegmentedQueue keySegmentQueue;     private final transient MailboxExecutor mailboxExecutor;     // We need to check whether the current thread is the mailbox thread using the mailbox     // processor.  m.gz-lzd.com     // TODO: This is a temporary workaround. In the future, we should add an interface in     // MailboxExecutor to check whether a thread is a mailbox thread, rather than using reflection     // to obtain the MailboxProcessor instance and make the determination.     private transient MailboxProcessor mailboxProcessor;     // An action will be split into one or more ActionTask objects. We use a state to store the     // pending ActionTasks that are waiting to be executed.     private transient ListState<ActionTask> actionTasksKState;     // To avoid processing different InputEvents with the same key, we use a state to store pending     // InputEvents that are waiting to be processed.     private transient ListState<Event> pendingInputEventsKState;     // An operator state is used to track the currently processing keys. This is useful when   wap.wofupay.com         // receiving an EndOfInput signal, as we need to wait until all related events are fully     // processed.     private transient ListState<Object> currentProcessingKeysOpState;     private final transient EventLogger eventLogger;     private final transient List<EventListener> eventListeners;     private transient ActionStateStore actionStateStore;     private transient ValueState<Long> sequenceNumberKState;  kue.zw0769.com         private transient ListState<Object> recoveryMarkerOpState;     private transient Map<Long, Map<Object, Long>> checkpointIdToSeqNums;     // This in memory map keep track of the runner context for the async action task that having  m.yqswh.cn     // been finished   fwn.zw0769.com

• Environment
  • Flink Streaming Runtime Environment：ActionExecutionOperator 是一个Flink 流处理操作符，会被 Flink 流处理运行时环境使用，作为流处理作业 DAG 图中的一个节点。
  • RemoteExecutionEnvironment：在远程执行环境中，RemoteExecutionEnvironment 会创建和使用 ActionExecutionOperator 来执行Agent 逻辑。
• RunnerContext 及其实现：
  • RunnerContextImpl：Java 动作执行上下文
  • PythonRunnerContextImpl：Python 动作执行上下文
• 状态管理组件。以下状态存储组件会被 ActionExecutionOperator 使用。
  • Flink 状态后端，用于短期内存、待处理任务等。
  • ActionStateStore：KafkaActionStateStore：可选的外部状态存储，用于动作状态持久化和恢复。
• Python相关组件。当代理包含Python动作时，以下组件会与 ActionExecutionOperator 协同工作。
  • PythonActionExecutor：管理Python环境和执行Python动作
  • PythonActionTask：表示Python动作任务
  • PythonRunnerContextImpl ：Python动作执行上下文。
• 辅助组件
  • EventLogger：事件日志记录器
  • EventListner：事件监听器
  • BultInMetrics：内置指标收集
  • FlinkAgentsMetricGroupImpl：指标组实现
• 事件处理相关组件
  • 各种Event类型（InputEvent、OutputEvent、PythonEvent等）
  • SegmentedQueue：管理按键分段的事件队列和水印处理

Eden Space（伊甸园区）：新对象的 "默认出生地"， 99% 的新对象 都会优先分配到 Eden 区（除非是超大对象直接进入老年代）。占新生代  80% 。

Survivor Space 0/1（幸存者区 0/1，简称 S0/S1，也叫 From 区 / To 区）：

• 新生代 GC 后存活对象的  "临时中转站"，避免存活对象直接进入老年代。 两个区会动态互换角色。各占新生代  10%，合计 20%。
• S0 和 S1  永远有一个是空的（这是 JVM 的设计巧思）。
• 每经历一次 Minor GC，存活对象的  "年龄 + 1"。

S0/S1 的交互 示例：

• ① 初始状态：Eden 有对象，S0 有对象，S1 为空；
• ② Eden 触发 Minor GC → 回收 Eden+S0 的垃圾对象，存活对象复制到 S1，清空 Eden+S0；
• ③ 此时 S1 非空、S0 为空，两者角色互换（S0=To 区，S1=From 区）；
• ④ 下次 Minor GC 重复上述过程，存活对象在 S0/S1 之间  "来回复制"。

Old Generation（老年代）：

• 存储  长期存活、体积较大 的对象，比如单例对象、缓存对象、经历多次 Minor GC 仍存活的对象。默认占堆总容量  2/3。
• 触发  Major GC / Full GC，频率极低，但耗时极长。
• 老年代大小决定 Full GC 频率 —— 老年代越大，Full GC 频率越低，但单次 Full GC 耗时越长。

对象进入老年代的条件：

• 年龄达标：对象在 S0/S1 之间复制次数达到阈值（默认 15 次， -XX:MaxTenuringThreshold调整）；
• 大对象：超过阈值的对象（ -XX:PretenureSizeThreshold）直接进入老年代；
• 空间担保失败：Minor GC 后 Survivor 区放不下存活对象，多余对象直接晋升老年代；

  for ( int i = 4 ; i >= 0 ; i--)        stride[i] = (i == 4 ) ? 1 : size[i + 1 ] * stride[i + 1 ]; } void cudnn_conv_3d ( const float* input, const float* kernel,     float* output, const int input_dims[ 3],     const int kernel_dims[ 3]) {     // -------------------------- 1. 初始化CUDNN句柄 --------------------------    cudnnHandle_t cudnn;wap.gz-lzd.com    CHECK_CUDNN(cudnnCreate(&cudnn));     // -------------------------- 2. 构造输入张量描述符 --------------------------     int inputDims[ 5 ] = { 1 , 1 , input_dims[ 0 ], input_dims[ 1 ], input_dims[ 2 ] }; // N=1, C=1, D/H/W     int input_stride[ 5 ];    compute_stride(inputDims, input_stride);    cudnnTensorDescriptor_t inputDesc;wap.4008986689.cn    CHECK_CUDNN(cudnnCreateTensorDescriptor(&inputDesc));    CHECK_CUDNN(cudnnSetTensorNdDescriptor(inputDesc, CUDNN_DATA_FLOAT, 5 , inputDims, input_stride));     // -------------------------- 3. 构造卷积核描述符 --------------------------     int filterDims[ 5 ] = { 1 , 1 , kernel_dims[ 0 ], kernel_dims[ 1 ], kernel_dims[ 2 ] }; // 输出通道=1, 输入通道=1, 核尺寸    cudnnFilterDescriptor_t filterDesc;   dgz.zw0769.com    CHECK_CUDNN(cudnnCreateFilterDescriptor(&filterDesc));    CHECK_CUDNN(cudnnSetFilterNdDescriptor(filterDesc, CUDNN_DATA_FLOAT, CUDNN_TENSOR_NCHW, 5 , filterDims));     // -------------------------- 4. 计算SAME填充并构造卷积描述符 --------------------------     int conmv_padA[ 3 ];     for ( int i = 0 ; i < 3 ; i++) {        conmv_padA[i] = (kernel_dims[i] - 1 ) / 2 ; // 计算SAME填充    }     int conv_filterStrideA[ 3 ] = { 1 , 1 , 1 };       // 步长=1     int conv_dilationA[ 3 ] = { 1 , 1 , 1 };     jrszb.ttLyf.cn      // 膨胀=1    cudnnConvolutionDescriptor_t convDesc;    CHECK_CUDNN(cudnnCreateConvolutionDescriptor(&convDesc));    CHECK_CUDNN(cudnnSetConvolutionNdDescriptor(convDesc, 3 , conmv_padA, conv_filterStrideA,        conv_dilationA, CUDNN_CROSS_CORRELATION, CUDNN_DATA_FLOAT));     // -------------------------- 5. 计算输出张量尺寸并构造描述符 --------------------------     int outputDims[ 5 ];jrs.hntcsc.com    CHECK_CUDNN(cudnnGetConvolutionNdForwardOutputDim(convDesc, inputDesc, filterDesc, 5 , outputDims));     int output_stride[ 5 ];    compute_stride(outputDims, output_stride);    cudnnTensorDescriptor_t outputDesc;    CHECK_CUDNN(cudnnCreateTensorDescriptor(&outputDesc));    CHECK_CUDNN(cudnnSetTensorNdDescriptor(outputDesc, CUDNN_DATA_FLOAT, 5 , outputDims, output_stride));     // -------------------------- 6. 选取最优卷积算法 --------------------------     size_t in_bytes, out_bytes;    CHECK_CUDNN(cudnnGetTensorSizeInBytes(inputDesc, &in_bytes));    CHECK_CUDNN(cudnnGetTensorSizeInBytes(outputDesc, &out_bytes));     size_t filt_bytes = 1 ;     for ( int i = 0 ; i < 5 ; i++) filt_bytes *= filterDims[i];    filt_bytes *= sizeof ( float );     // 获取最优卷积算法     int returnedAlgoCount = 0 ;jrs.yqswh.cn    cudnnConvolutionFwdAlgoPerf_t perfResults;    CHECK_CUDNN(cudnnGetConvolutionForwardAlgorithm_v7(cudnn, inputDesc, filterDesc, convDesc, outputDesc,         1 , &returnedAlgoCount, &perfResults));     // 计算工作空间大小     size_t workspace_bytes = 0 ;    CHECK_CUDNN(cudnnGetConvolutionForwardWorkspaceSize(cudnn, inputDesc, filterDesc, convDesc, outputDesc,        perfResults.algo, &workspace_bytes));     // -------------------------- 7. 分配设备内存并拷贝数据 --------------------------     float * d_input = nullptr, * d_filter = nullptr, * d_output = nullptr;     void * d_workspace = nullptr;    CHECK_CUDA(cudaMalloc(( void **)&d_input, in_bytes));m.tinghaoshu.cn    CHECK_CUDA(cudaMalloc(( void **)&d_filter, filt_bytes));    CHECK_CUDA(cudaMalloc(( void **)&d_output, out_bytes));     if (workspace_bytes > 0 ) {        CHECK_CUDA(cudaMalloc(( void **)&d_workspace, workspace_bytes));    }     // 主机到设备拷贝    CHECK_CUDA(cudaMemcpy(d_input, input, in_bytes, cudaMemcpyHostToDevice));    CHECK_CUDA(cudaMemcpy(d_filter, kernel, filt_bytes, cudaMemcpyHostToDevice));     // -------------------------- 8. 执行卷积计算 --------------------------m.czLangrun.com     float alpha = 1.0f , beta = 0.0f ;9aztam.xydgdq.cn    CHECK_CUDNN(cudnnConvolutionForward(cudnn, &alpha, inputDesc, d_input, filterDesc, d_filter,        convDesc, perfResults.algo, d_workspace, workspace_bytes,bgv.zw0769.com        &beta, outputDesc, d_output)); wap.xydgdq.cn     // -------------------------- 9. 设备到主机拷贝输出 --------------------------    CHECK_CUDA(cudaMemcpy(output, d_output, out_bytes, cudaMemcpyDeviceToHost));     // -------------------------- 10. 资源释放 --------------------------    CHECK_CUDA(cudaFree(d_input));    CHECK_CUDA(cudaFree(d_filter));yinuo168.cn    CHECK_CUDA(cudaFree(d_output));    CHECK_CUDA(cudaFree(d_workspace));    CHECK_CUDNN(cudnnDestroyTensorDescriptor(outputDesc));    CHECK_CUDNN(cudnnDestroyConvolutionDescriptor(convDesc));    CHECK_CUDNN(cudnnDestroyFilterDescriptor(filterDesc));m.czLangrun.com    CHECK_CUDNN(cudnnDestroyTensorDescriptor(inputDesc));    CHECK_CUDNN(cudnnDestroy(cudnn));