Skip to content

Commit ca10c54

Browse files
committed
feat(ric): Cut response copies on the invocation success path from 4 to 1
A successful response of size S was copied four times before being posted: toByteArray() on the Java heap, then GetByteArrayElements, the std::string construction, and the by-value success() argument in native memory. - toNativeString copies with GetByteArrayRegion straight into a presized std::string, dropping the JVM's temporary copy. - The payload is moved into invocation_response::success instead of copied. - New NativeClient.postInvocationResponseWithLength posts the first N bytes of an array, so AWSLambda passes the response buffer's backing array and size instead of a toByteArray() copy. It has its own JNI name, so the existing postInvocationResponse symbol and behavior are unchanged. - LambdaRuntimeApiClient gets a default reportInvocationSuccess overload taking a length, which falls back to an exact-size copy for other implementations. - The per-thread response buffer is a LambdaByteArrayOutputStream, which exposes its backing array. The remaining copy into std::string is required by aws-lambda-cpp's invocation_response::success(std::string, ...).
1 parent 1412aea commit ca10c54

10 files changed

Lines changed: 386 additions & 19 deletions

File tree

‎aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/AWSLambda.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -327,7 +327,13 @@ private static void startRuntimeLoop(LambdaRequestHandler lambdaRequestHandler,
327327

328328
try {
329329
ByteArrayOutputStream payload = lambdaRequestHandler.call(request);
330-
runtimeClient.reportInvocationSuccess(request.getId(), payload.toByteArray(), request.getInvocationId());
330+
// Post straight from the backing array: toByteArray() would copy the whole response.
331+
ResponseBufferViewer view = ResponseBufferViewer.of(payload);
332+
if (view.isComplete()) {
333+
runtimeClient.reportInvocationSuccess(request.getId(), view.array(), view.length(), request.getInvocationId());
334+
} else {
335+
runtimeClient.reportInvocationSuccess(request.getId(), payload.toByteArray(), request.getInvocationId());
336+
}
331337
// clear interrupted flag in case if it was set by user's code
332338
Thread.interrupted();
333339
} catch (Throwable t) {
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
/*
2+
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package com.amazonaws.services.lambda.runtime.api.client;
7+
8+
import java.io.ByteArrayOutputStream;
9+
import java.io.IOException;
10+
import java.io.OutputStream;
11+
12+
/**
13+
* Gets the backing array of a ByteArrayOutputStream without copying it. ByteArrayOutputStream.writeTo(out) is
14+
* specified to call out.write(buf, 0, count) with its own array, under the stream's lock, so this stream keeps that
15+
* reference instead of writing anywhere. If writeTo does anything else (a subclass that overrides it, for example),
16+
* the view is not complete and the caller falls back to toByteArray().
17+
*/
18+
final class ResponseBufferViewer extends OutputStream {
19+
private byte[] array;
20+
private int length;
21+
private int arrayWrites;
22+
private boolean otherWrites;
23+
24+
/** Runs payload.writeTo on a new view. */
25+
static ResponseBufferViewer of(ByteArrayOutputStream payload) throws IOException {
26+
ResponseBufferViewer view = new ResponseBufferViewer();
27+
payload.writeTo(view);
28+
return view;
29+
}
30+
31+
@Override
32+
public void write(byte[] b, int off, int len) {
33+
if (off != 0) {
34+
otherWrites = true;
35+
return;
36+
}
37+
arrayWrites++;
38+
array = b;
39+
length = len;
40+
}
41+
42+
@Override
43+
public void write(int b) {
44+
otherWrites = true;
45+
}
46+
47+
/** True when writeTo passed a single array from offset 0: array() then holds the whole content. */
48+
boolean isComplete() {
49+
return arrayWrites == 1 && !otherWrites;
50+
}
51+
52+
/** The captured array. Only the first length() bytes are the content. */
53+
byte[] array() {
54+
return array;
55+
}
56+
57+
int length() {
58+
return length;
59+
}
60+
}

‎aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/runtimeapi/LambdaRuntimeApiClient.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import com.amazonaws.services.lambda.runtime.api.client.logging.LambdaContextLogger;
88
import com.amazonaws.services.lambda.runtime.api.client.runtimeapi.dto.InvocationRequest;
99
import java.io.IOException;
10+
import java.util.Arrays;
1011

1112
/**
1213
* Java interface for
@@ -38,6 +39,17 @@ public interface LambdaRuntimeApiClient {
3839
*/
3940
void reportInvocationSuccess(String requestId, byte[] response, String invocationId) throws IOException;
4041

42+
/**
43+
* Report invocation success with the first responseLength bytes of response
44+
* @param requestId request id
45+
* @param response byte array whose first responseLength bytes are the response
46+
* @param responseLength number of bytes of response to send
47+
* @param invocationId invocation id for cross-wiring protection (may be null)
48+
*/
49+
default void reportInvocationSuccess(String requestId, byte[] response, int responseLength, String invocationId) throws IOException {
50+
reportInvocationSuccess(requestId, Arrays.copyOf(response, responseLength), invocationId);
51+
}
52+
4153
/**
4254
* Report invocation error
4355
* @param requestId request id

‎aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/runtimeapi/LambdaRuntimeApiClientImpl.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,12 @@ public void reportInvocationSuccess(String requestId, byte[] response, String in
130130
NativeClient.postInvocationResponse(requestId.getBytes(UTF_8), response, invocationIdBytes);
131131
}
132132

133+
@Override
134+
public void reportInvocationSuccess(String requestId, byte[] response, int responseLength, String invocationId) {
135+
byte[] invocationIdBytes = invocationId != null ? invocationId.getBytes(UTF_8) : null;
136+
NativeClient.postInvocationResponseWithLength(requestId.getBytes(UTF_8), response, responseLength, invocationIdBytes);
137+
}
138+
133139
@Override
134140
public void reportInvocationError(String requestId, LambdaError error, String invocationId) throws IOException {
135141
String endpoint = invocationEndpoint + requestId + "/error";

‎aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/runtimeapi/NativeClient.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,4 +23,7 @@ static void init(String awsLambdaRuntimeApi) {
2323

2424
static native void postInvocationResponse(byte[] requestId, byte[] response, byte[] invocationId);
2525

26+
/** Posts the first responseLength bytes of response, so callers can pass a buffer's backing array without copying it. */
27+
static native void postInvocationResponseWithLength(byte[] requestId, byte[] response, int responseLength, byte[] invocationId);
28+
2629
}

‎aws-lambda-java-runtime-interface-client/src/main/jni/com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient.cpp‎

Lines changed: 29 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -64,15 +64,19 @@ static void throwLambdaRuntimeClientException(JNIEnv *env, std::string message,
6464
env->Throw(lambdaRuntimeException);
6565
}
6666

67-
static std::string toNativeString(JNIEnv *env, jbyteArray jArray) {
68-
int length = env->GetArrayLength(jArray);
69-
jbyte* bytes = env->GetByteArrayElements(jArray, NULL);
70-
std::string nativeString = std::string((char *)bytes, length);
71-
env->ReleaseByteArrayElements(jArray, bytes, JNI_ABORT);
67+
// Copies the first length bytes of the array straight into the string. GetByteArrayElements would usually make a
68+
// temporary copy first. A length outside the array raises ArrayIndexOutOfBoundsException; callers check for it.
69+
static std::string toNativeString(JNIEnv *env, jbyteArray jArray, jsize length) {
70+
std::string nativeString(length > 0 ? length : 0, '\0');
71+
env->GetByteArrayRegion(jArray, 0, length, reinterpret_cast<jbyte*>(&nativeString[0]));
7272
env->DeleteLocalRef(jArray);
7373
return nativeString;
7474
}
7575

76+
static std::string toNativeString(JNIEnv *env, jbyteArray jArray) {
77+
return toNativeString(env, jArray, env->GetArrayLength(jArray));
78+
}
79+
7680
JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient_initializeClient(JNIEnv *env, jobject thisObject, jbyteArray userAgent, jbyteArray awsLambdaRuntimeApi) {
7781
std::string user_agent = toNativeString(env, userAgent);
7882
std::string endpoint = toNativeString(env, awsLambdaRuntimeApi);
@@ -129,12 +133,7 @@ JNIEXPORT jobject JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_
129133
return NULL;
130134
}
131135

132-
JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient_postInvocationResponse
133-
(JNIEnv *env, jobject thisObject, jbyteArray jrequestId, jbyteArray jresponseArray, jbyteArray jinvocationId) {
134-
std::string payload = toNativeString(env, jresponseArray);
135-
if ((env)->ExceptionOccurred()){
136-
return;
137-
}
136+
static void postInvocationResponse(JNIEnv *env, jbyteArray jrequestId, std::string payload, jbyteArray jinvocationId) {
138137
std::string requestId = toNativeString(env, jrequestId);
139138
if ((env)->ExceptionOccurred()){
140139
return;
@@ -148,10 +147,28 @@ JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_run
148147
}
149148
}
150149

151-
auto response = aws::lambda_runtime::invocation_response::success(payload, "application/json");
150+
auto response = aws::lambda_runtime::invocation_response::success(std::move(payload), "application/json");
152151
auto outcome = CLIENT->post_success(requestId, response, invocationId);
153152
if (!outcome.is_success()) {
154153
std::string errorMessage("Failed to post invocation response.");
155154
throwLambdaRuntimeClientException(env, errorMessage, outcome.get_failure());
156155
}
157156
}
157+
158+
JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient_postInvocationResponse
159+
(JNIEnv *env, jobject thisObject, jbyteArray jrequestId, jbyteArray jresponseArray, jbyteArray jinvocationId) {
160+
std::string payload = toNativeString(env, jresponseArray);
161+
if ((env)->ExceptionOccurred()){
162+
return;
163+
}
164+
postInvocationResponse(env, jrequestId, std::move(payload), jinvocationId);
165+
}
166+
167+
JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient_postInvocationResponseWithLength
168+
(JNIEnv *env, jobject thisObject, jbyteArray jrequestId, jbyteArray jresponseArray, jint responseLength, jbyteArray jinvocationId) {
169+
std::string payload = toNativeString(env, jresponseArray, responseLength);
170+
if ((env)->ExceptionOccurred()){
171+
return;
172+
}
173+
postInvocationResponse(env, jrequestId, std::move(payload), jinvocationId);
174+
}

‎aws-lambda-java-runtime-interface-client/src/main/jni/com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@ JNIEXPORT jobject JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_
1919
JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient_postInvocationResponse
2020
(JNIEnv *, jobject, jbyteArray, jbyteArray, jbyteArray);
2121

22+
JNIEXPORT void JNICALL Java_com_amazonaws_services_lambda_runtime_api_client_runtimeapi_NativeClient_postInvocationResponseWithLength
23+
(JNIEnv *, jobject, jbyteArray, jbyteArray, jint, jbyteArray);
24+
2225
#ifdef __cplusplus
2326
}
2427
#endif

‎aws-lambda-java-runtime-interface-client/src/test/java/com/amazonaws/services/lambda/runtime/api/client/AWSLambdaTest.java‎

Lines changed: 69 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,14 @@
88
import static org.junit.Assert.assertEquals;
99
import static org.junit.Assert.assertTrue;
1010
import static org.mockito.ArgumentMatchers.any;
11+
import static org.mockito.ArgumentMatchers.anyInt;
1112
import static org.mockito.ArgumentMatchers.anyString;
1213
import static org.mockito.ArgumentMatchers.eq;
1314
import static org.mockito.Mockito.*;
1415

1516
import java.io.ByteArrayOutputStream;
1617
import java.io.IOError;
18+
import java.io.IOException;
1719
import java.util.ArrayList;
1820
import java.util.HashMap;
1921
import java.util.HashSet;
@@ -242,7 +244,7 @@ void testConcurrentRunWithPlatformThreads() throws Throwable {
242244
AWSLambda.startRuntimeLoops(lambdaRequestHandler, lambdaLogger, concurrencyConfig, runtimeClient);
243245

244246
// Success Reports Must Equal number of tasks that ran successfully.
245-
verify(runtimeClient, times(7)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), any());
247+
verify(runtimeClient, times(7)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), anyInt(), any());
246248
// Hashmap keys should equal the number of threads (runtime loops).
247249
assertEquals(4, SampleHandler.hashMap.size());
248250
// Hashmap total count should equal all tasks that ran * number of iterations per task
@@ -284,7 +286,7 @@ void testConcurrentRunWithPlatformThreadsWithFailures() throws Throwable {
284286
verify(runtimeClient).reportInvocationError(eq(UserFaultID), any(), any());
285287

286288
// Success Reports Must Equal number of tasks that ran successfully.
287-
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), any());
289+
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), anyInt(), any());
288290

289291
// Hashmap keys should equal the minumum between(number of threads (runtime loops) AND number of tasks that ran successfully).
290292
assertEquals(2, SampleHandler.hashMap.size());
@@ -331,7 +333,7 @@ void testConcurrentModeLoopDoesNotExitExceptForLambdaRuntimeClientMaxRetriesExce
331333
verify(runtimeClient).reportInvocationError(eq(IOErrorID), any(), any());
332334

333335
// Success Reports Must Equal number of tasks that ran successfully.
334-
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), any());
336+
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), anyInt(), any());
335337

336338
// Hashmap keys should equal the minumum between(number of threads (runtime loops) AND number of tasks that ran successfully).
337339
assertEquals(1, SampleHandler.hashMap.size());
@@ -520,7 +522,7 @@ void testSequentialWithFatalUserFaultErrorStopsLoop() throws Throwable {
520522
verify(runtimeClient).reportInvocationError(eq(UserFaultID), any(), any());
521523

522524
// Success Reports Must Equal number of tasks that ran successfully. And only 2 Error reports for failImmediatelyRequest and userFaultRequest.
523-
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), any());
525+
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), anyInt(), any());
524526
verify(runtimeClient, times(2)).reportInvocationError(any(), any(), any());
525527

526528
// Hashmap keys should equal one as it is not multithreaded.
@@ -566,7 +568,7 @@ void testSequentialWithVirtualMachineErrorStopsLoop() throws Throwable {
566568
verify(runtimeClient).reportInvocationError(eq(IOErrorID), any(), any());
567569

568570
// Success Reports Must Equal number of tasks that ran successfully. And only 2 Error reports for failImmediatelyRequest and virtualMachineErrorRequest.
569-
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), any());
571+
verify(runtimeClient, times(2)).reportInvocationSuccess(eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), anyInt(), any());
570572
verify(runtimeClient, times(2)).reportInvocationError(any(), any(), any());
571573

572574
// Hashmap keys should equal one as it is not multithreaded.
@@ -671,7 +673,68 @@ void testInvocationIdIsPassedToReportSuccess() throws Throwable {
671673
AWSLambda.startRuntimeLoops(lambdaRequestHandler, lambdaLogger, concurrencyConfig, runtimeClient);
672674

673675
verify(runtimeClient).reportInvocationSuccess(
674-
eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), eq("test-inv-uuid-1234"));
676+
eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), any(), anyInt(), eq("test-inv-uuid-1234"));
677+
}
678+
679+
/** Exposes the backing array so tests can check that it is posted as is. */
680+
private static final class InspectableByteArrayOutputStream extends ByteArrayOutputStream {
681+
byte[] backingArray() {
682+
return buf;
683+
}
684+
}
685+
686+
/** Overrides writeTo the way a subclass could, one byte at a time, so the backing array cannot be captured. */
687+
private static final class BytewiseWriteToByteArrayOutputStream extends ByteArrayOutputStream {
688+
@Override
689+
public synchronized void writeTo(java.io.OutputStream out) throws IOException {
690+
for (byte b : toByteArray()) {
691+
out.write(b);
692+
}
693+
}
694+
}
695+
696+
private void runOneInvocation(LambdaRequestHandler handler) throws Throwable {
697+
when(concurrencyConfig.isMultiConcurrent()).thenReturn(false);
698+
699+
InvocationRequest request = getFakeInvocationRequest(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE);
700+
request.setInvocationId("test-inv-uuid-1234");
701+
702+
// Fatal error to stop the loop after one successful invocation
703+
InvocationRequest fatalRequest = mock(InvocationRequest.class);
704+
when(fatalRequest.getId()).thenThrow(UserFault.makeUserFault(new IOError(new Throwable()), true)).thenReturn("fatal");
705+
706+
when(runtimeClient.nextInvocation())
707+
.thenReturn(request)
708+
.thenReturn(fatalRequest);
709+
710+
AWSLambda.startRuntimeLoops(handler, lambdaLogger, concurrencyConfig, runtimeClient);
711+
}
712+
713+
@Test
714+
@Timeout(value = 1, unit = TimeUnit.MINUTES)
715+
void testResponseBackingArrayIsPostedWithoutCopy() throws Throwable {
716+
InspectableByteArrayOutputStream output = new InspectableByteArrayOutputStream();
717+
output.write("\"success\"".getBytes());
718+
719+
runOneInvocation(request -> output);
720+
721+
// The backing array itself is passed, with the content length, instead of a toByteArray() copy.
722+
verify(runtimeClient).reportInvocationSuccess(
723+
eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), same(output.backingArray()), eq(9), eq("test-inv-uuid-1234"));
724+
verify(runtimeClient, never()).reportInvocationSuccess(anyString(), any(), any());
725+
}
726+
727+
@Test
728+
@Timeout(value = 1, unit = TimeUnit.MINUTES)
729+
void testResponseIsCopiedWhenWriteToDoesNotPassTheBackingArray() throws Throwable {
730+
ByteArrayOutputStream output = new BytewiseWriteToByteArrayOutputStream();
731+
output.write("\"success\"".getBytes());
732+
733+
runOneInvocation(request -> output);
734+
735+
verify(runtimeClient).reportInvocationSuccess(
736+
eq(SampleHandler.ADD_ENTRY_TO_MAP_ID_OP_MODE), eq("\"success\"".getBytes()), eq("test-inv-uuid-1234"));
737+
verify(runtimeClient, never()).reportInvocationSuccess(anyString(), any(), anyInt(), any());
675738
}
676739

677740
@Test

0 commit comments

Comments
 (0)