[server] concurrent execution for batched queries (#1301)

samuelAndalon · web-flow · commit 27f99ede8687 · 2022-02-10T15:46:46.000-06:00
This adds the logic to execute Queries in a concurrent instead of a sequential way, this will only apply for batched operations that are only queries, if there is a mutation, the batch will still be executed in a sequential way.
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -97,6 +97,7 @@ subprojects {
             // even though we don't have any Java code, since we are building using Java LTS version,
             // this is required for Gradle to set the correct JVM versions in the module metadata
             targetCompatibility = JavaVersion.VERSION_1_8
+            sourceCompatibility = JavaVersion.VERSION_1_8
         }
 
         // published artifacts
diff --git a/gradle.properties b/gradle.properties
@@ -40,6 +40,7 @@ mockkVersion = 1.12.0
 mustacheVersion = 0.9.10
 rxjavaVersion = 3.1.0
 wireMockVersion = 2.30.1
+kotlinxBenchmarkVersion = 0.4.0
 
 # plugin versions
 detektVersion = 1.18.0
diff --git a/servers/graphql-kotlin-server/build.gradle.kts b/servers/graphql-kotlin-server/build.gradle.kts
@@ -1,12 +1,40 @@
+import kotlinx.benchmark.gradle.JvmBenchmarkTarget
+
 description = "Common code for running a GraphQL server in any HTTP server framework"
 
 val kotlinCoroutinesVersion: String by project
+val kotlinxBenchmarkVersion: String by project
+
+plugins {
+    id("org.jetbrains.kotlinx.benchmark")
+}
 
 dependencies {
     api(project(path = ":graphql-kotlin-schema-generator"))
     testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$kotlinCoroutinesVersion")
 }
 
+// Benchmarks
+
+// Create a separate source set for benchmarks.
+sourceSets.create("benchmarks")
+
+kotlin.sourceSets.getByName("benchmarks") {
+    dependencies {
+        implementation("org.jetbrains.kotlinx:kotlinx-benchmark-runtime:$kotlinxBenchmarkVersion")
+        implementation(sourceSets.main.get().output)
+        implementation(sourceSets.main.get().runtimeClasspath)
+    }
+}
+
+benchmark {
+    targets {
+        register("benchmarks") {
+            this as JvmBenchmarkTarget
+        }
+    }
+}
+
 tasks {
     jacocoTestCoverageVerification {
         violationRules {
diff --git a/servers/graphql-kotlin-server/src/benchmarks/kotlin/GraphQLRequestBenchmark.kt b/servers/graphql-kotlin-server/src/benchmarks/kotlin/GraphQLRequestBenchmark.kt
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2022 Expedia, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.expediagroup.graphql.server
+
+import com.expediagroup.graphql.server.extensions.isMutation
+import com.expediagroup.graphql.server.types.GraphQLRequest
+import org.openjdk.jmh.annotations.Benchmark
+import org.openjdk.jmh.annotations.Setup
+import org.openjdk.jmh.annotations.State
+import org.openjdk.jmh.annotations.Scope
+import org.openjdk.jmh.annotations.Fork
+import org.openjdk.jmh.annotations.Warmup
+import org.openjdk.jmh.annotations.Measurement
+import java.util.concurrent.TimeUnit
+import kotlin.random.Random
+
+@State(Scope.Benchmark)
+@Fork(1)
+@Warmup(iterations = 2)
+@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+open class GraphQLRequestBenchmark {
+    private val requests = mutableListOf<GraphQLRequest>()
+
+    @Setup
+    fun setUp() {
+        val charPool = ('a'..'z') + ('A'..'Z') + ('0'..'9')
+        val range = (1..3072)
+        repeat(50) {
+            val randomStringForQuery = range
+                .map { Random.nextInt(0, charPool.size) }
+                .map(charPool::get)
+                .joinToString("")
+            val query = """$randomStringForQuery query HeroNameAndFriends(${"$"}episode: Episode) {
+              hero(episode: ${"$"}episode) {
+                name
+                friends {
+                  name
+                }
+              }
+            }"""
+            requests.add(GraphQLRequest(query))
+        }
+        val randomStringForMutation = range
+            .map { Random.nextInt(0, charPool.size) }
+            .map(charPool::get)
+            .joinToString("")
+
+        val mutation = """$randomStringForMutation mutation AddNewPet (${"$"}name: String!,${"$"}petType: PetType) {
+              addPet(name:${"$"}name,petType:${"$"}petType) {
+                name
+                petType
+              }
+            }"""
+        requests.add(GraphQLRequest(mutation))
+    }
+
+    @Benchmark
+    fun isMutationBenchmark(): Boolean {
+        return requests.any(GraphQLRequest::isMutation)
+    }
+}
diff --git a/servers/graphql-kotlin-server/src/main/kotlin/com/expediagroup/graphql/server/execution/GraphQLServer.kt b/servers/graphql-kotlin-server/src/main/kotlin/com/expediagroup/graphql/server/execution/GraphQLServer.kt
@@ -16,13 +16,21 @@
 
 package com.expediagroup.graphql.server.execution
 
+import com.expediagroup.graphql.generator.execution.GraphQLContext
+import com.expediagroup.graphql.server.extensions.isMutation
+import com.expediagroup.graphql.server.extensions.toGraphQLError
+import com.expediagroup.graphql.server.extensions.toGraphQLKotlinType
 import com.expediagroup.graphql.server.types.GraphQLBatchRequest
 import com.expediagroup.graphql.server.types.GraphQLBatchResponse
 import com.expediagroup.graphql.server.types.GraphQLRequest
+import com.expediagroup.graphql.server.types.GraphQLResponse
 import com.expediagroup.graphql.server.types.GraphQLServerResponse
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.supervisorScope
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
 
@@ -58,14 +66,44 @@ open class GraphQLServer<Request>(
 
             when (graphQLRequest) {
                 is GraphQLRequest -> requestHandler.executeRequest(graphQLRequest, deprecatedContext, graphQLContext)
-                is GraphQLBatchRequest -> GraphQLBatchResponse(
-                    graphQLRequest.requests.map {
-                        requestHandler.executeRequest(it, deprecatedContext, graphQLContext)
+                is GraphQLBatchRequest -> when {
+                    graphQLRequest.requests.any(GraphQLRequest::isMutation) -> GraphQLBatchResponse(
+                        graphQLRequest.requests.map {
+                            requestHandler.executeRequest(it, deprecatedContext, graphQLContext)
+                        }
+                    )
+                    else -> {
+                        GraphQLBatchResponse(
+                            handleConcurrently(graphQLRequest, deprecatedContext, graphQLContext)
+                        )
                     }
-                )
+                }
             }
         } else {
             null
         }
     }
+
+    /**
+     * Concurrently execute a [batchRequest], a failure in an specific request will not cause the scope
+     * to fail and does not affect the other requests. The total execution time will be the time of the slowest
+     * request
+     */
+    private suspend fun handleConcurrently(
+        batchRequest: GraphQLBatchRequest,
+        context: GraphQLContext?,
+        graphQLContext: Map<*, Any>
+    ): List<GraphQLResponse<*>> = supervisorScope {
+        batchRequest.requests.map { request ->
+            async {
+                try {
+                    requestHandler.executeRequest(request, context, graphQLContext)
+                } catch (e: Exception) {
+                    GraphQLResponse<Any?>(
+                        errors = listOf(e.toGraphQLError().toGraphQLKotlinType())
+                    )
+                }
+            }
+        }.awaitAll()
+    }
 }
diff --git a/servers/graphql-kotlin-server/src/main/kotlin/com/expediagroup/graphql/server/extensions/requestExtensions.kt b/servers/graphql-kotlin-server/src/main/kotlin/com/expediagroup/graphql/server/extensions/requestExtensions.kt
@@ -23,7 +23,11 @@ import org.dataloader.DataLoaderRegistry
 /**
  * Convert the common [GraphQLRequest] to the execution input used by graphql-java
  */
-fun GraphQLRequest.toExecutionInput(graphQLContext: Any? = null, dataLoaderRegistry: DataLoaderRegistry? = null, graphQLContextMap: Map<*, Any>? = null): ExecutionInput =
+fun GraphQLRequest.toExecutionInput(
+    graphQLContext: Any? = null,
+    dataLoaderRegistry: DataLoaderRegistry? = null,
+    graphQLContextMap: Map<*, Any>? = null
+): ExecutionInput =
     ExecutionInput.newExecutionInput()
         .query(this.query)
         .operationName(this.operationName)
@@ -34,3 +38,5 @@ fun GraphQLRequest.toExecutionInput(graphQLContext: Any? = null, dataLoaderRegis
         }
         .dataLoaderRegistry(dataLoaderRegistry ?: DataLoaderRegistry())
         .build()
+
+fun GraphQLRequest.isMutation(): Boolean = query.contains("mutation ")
diff --git a/servers/graphql-kotlin-server/src/test/kotlin/com/expediagroup/graphql/server/execution/GraphQLServerTest.kt b/servers/graphql-kotlin-server/src/test/kotlin/com/expediagroup/graphql/server/execution/GraphQLServerTest.kt
@@ -21,6 +21,7 @@ import com.expediagroup.graphql.server.types.GraphQLBatchRequest
 import com.expediagroup.graphql.server.types.GraphQLRequest
 import io.mockk.coEvery
 import io.mockk.coVerify
+import io.mockk.every
 import io.mockk.mockk
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runBlockingTest
@@ -36,7 +37,13 @@ class GraphQLServerTest {
     @Test
     fun `the request handler and parser are called`() {
         val mockParser = mockk<GraphQLRequestParser<MockHttpRequest>> {
-            coEvery { parseRequest(any()) } returns GraphQLBatchRequest(requests = listOf(mockk()))
+            coEvery { parseRequest(any()) } returns GraphQLBatchRequest(
+                requests = listOf(
+                    mockk {
+                        every { query } returns "query OperationName { parent { child } }"
+                    }
+                )
+            )
         }
         val mockContextFactory = mockk<GraphQLContextFactory<MockContext, MockHttpRequest>> {
             coEvery { generateContext(any()) } returns MockContext()
@@ -58,10 +65,48 @@ class GraphQLServerTest {
         }
     }
 
+    @Test
+    fun `the request handler and parser are called for a batch with a mutation`() {
+        val mockParser = mockk<GraphQLRequestParser<MockHttpRequest>> {
+            coEvery { parseRequest(any()) } returns GraphQLBatchRequest(
+                requests = listOf(
+                    mockk {
+                        every { query } returns "query OperationName { parent { child } }"
+                    },
+                    mockk {
+                        every { query } returns "mutation OperationName { field { to { mutate } } }"
+                    }
+                )
+            )
+        }
+        val mockContextFactory = mockk<GraphQLContextFactory<MockContext, MockHttpRequest>> {
+            coEvery { generateContext(any()) } returns MockContext()
+            coEvery { generateContextMap(any()) } returns mapOf("foo" to 1)
+        }
+        val mockHandler = mockk<GraphQLRequestHandler> {
+            coEvery { executeRequest(any(), any(), any()) } returns mockk()
+        }
+
+        val server = GraphQLServer(mockParser, mockContextFactory, mockHandler)
+
+        runBlockingTest { server.execute(mockk()) }
+
+        coVerify(exactly = 1) {
+            mockParser.parseRequest(any())
+            mockContextFactory.generateContext(any())
+            mockContextFactory.generateContextMap(any())
+        }
+        coVerify(exactly = 2) {
+            mockHandler.executeRequest(any(), any(), any())
+        }
+    }
+
     @Test
     fun `null context is used and passed to the request handler`() {
         val mockParser = mockk<GraphQLRequestParser<MockHttpRequest>> {
-            coEvery { parseRequest(any()) } returns mockk<GraphQLRequest>()
+            coEvery { parseRequest(any()) } returns mockk<GraphQLRequest> {
+                every { query } returns "query OperationName { parent { child } }"
+            }
         }
         val mockContextFactory = mockk<GraphQLContextFactory<MockContext, MockHttpRequest>> {
             coEvery { generateContext(any()) } returns null
diff --git a/servers/graphql-kotlin-server/src/test/kotlin/com/expediagroup/graphql/server/extensions/RequestExtensionsKtTest.kt b/servers/graphql-kotlin-server/src/test/kotlin/com/expediagroup/graphql/server/extensions/RequestExtensionsKtTest.kt
@@ -22,6 +22,7 @@ import org.dataloader.DataLoaderRegistry
 import org.junit.jupiter.api.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
 
 class RequestExtensionsKtTest {
 
@@ -75,4 +76,17 @@ class RequestExtensionsKtTest {
         val executionInput = request.toExecutionInput(graphQLContextMap = context)
         assertEquals(1, executionInput.graphQLContext.get("foo"))
     }
+
+    @Test
+    fun `verify graphQLRequest is a mutation`() {
+        val request = GraphQLRequest(
+            query = """
+                mutation addPet(name: "name", petType: "type") {
+                name
+                petType
+              }
+            """.trimIndent()
+        )
+        assertTrue(request.isMutation())
+    }
 }
diff --git a/servers/graphql-kotlin-spring-server/src/main/kotlin/com/expediagroup/graphql/server/spring/GraphQLSchemaConfiguration.kt b/servers/graphql-kotlin-spring-server/src/main/kotlin/com/expediagroup/graphql/server/spring/GraphQLSchemaConfiguration.kt
@@ -47,7 +47,7 @@ const val DEFAULT_INSTRUMENTATION_ORDER = 0
 
 /**
  * Configuration class that loads both the federated and non-federation
- * configuraiton and creates the GraphQL schema object and request handler.
+ * configuration and creates the GraphQL schema object and request handler.
  *
  * This config can then be used by all Spring specific configuration classes
  * to handle incoming requests from HTTP routes or subscriptions and send them
diff --git a/servers/graphql-kotlin-spring-server/src/test/kotlin/com/expediagroup/graphql/server/spring/subscriptions/SubscriptionWebSocketHandlerIT.kt b/servers/graphql-kotlin-spring-server/src/test/kotlin/com/expediagroup/graphql/server/spring/subscriptions/SubscriptionWebSocketHandlerIT.kt
@@ -66,7 +66,7 @@ class SubscriptionWebSocketHandlerIT(
         val dataOutput = TestPublisher.create<String>()
 
         val response = client.execute(uri) { session ->
-            executeSubsciption(session, startMessage, dataOutput)
+            executeSubscription(session, startMessage, dataOutput)
         }.subscribe()
 
         StepVerifier.create(dataOutput)
@@ -90,7 +90,7 @@ class SubscriptionWebSocketHandlerIT(
         val dataOutput = TestPublisher.create<String>()
 
         val response = client.execute(uri) { session ->
-            executeSubsciption(session, startMessage, dataOutput)
+            executeSubscription(session, startMessage, dataOutput)
         }.subscribe()
 
         StepVerifier.create(dataOutput)
@@ -116,7 +116,7 @@ class SubscriptionWebSocketHandlerIT(
         headers.set("X-Custom-Header", "junit")
 
         val response = client.execute(uri, headers) { session ->
-            executeSubsciption(session, startMessage, dataOutput)
+            executeSubscription(session, startMessage, dataOutput)
         }.subscribe()
 
         StepVerifier.create(dataOutput)
@@ -127,7 +127,7 @@ class SubscriptionWebSocketHandlerIT(
         response.dispose()
     }
 
-    private fun executeSubsciption(
+    private fun executeSubscription(
         session: WebSocketSession,
         startMessage: String,
         dataOutput: TestPublisher<String>
diff --git a/settings.gradle.kts b/settings.gradle.kts
@@ -2,6 +2,7 @@ pluginManagement {
     val detektVersion: String by settings
     val dokkaVersion: String by settings
     val kotlinVersion: String by settings
+    val kotlinxBenchmarkVersion: String by settings
     val ktlintPluginVersion: String by settings
     val mavenPluginDevelopmentVersion: String by settings
     val nexusPublishPluginVersion: String by settings
@@ -18,6 +19,7 @@ pluginManagement {
         id("io.github.gradle-nexus.publish-plugin") version nexusPublishPluginVersion
         id("io.gitlab.arturbosch.detekt") version detektVersion
         id("org.jetbrains.dokka") version dokkaVersion
+        id("org.jetbrains.kotlinx.benchmark") version kotlinxBenchmarkVersion
         id("org.jlleitschuh.gradle.ktlint") version ktlintPluginVersion
         id("org.springframework.boot") version springBootVersion
     }

Original file line number	Diff line number	Diff line change
`@@ -97,6 +97,7 @@ subprojects {`
`97`	`97`	`// even though we don't have any Java code, since we are building using Java LTS version,`
`98`	`98`	`// this is required for Gradle to set the correct JVM versions in the module metadata`
`99`	`99`	`targetCompatibility = JavaVersion.VERSION_1_8`
	`100`	`+ sourceCompatibility = JavaVersion.VERSION_1_8`
`100`	`101`	`}`
`101`	`102`
`102`	`103`	`// published artifacts`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ const val DEFAULT_INSTRUMENTATION_ORDER = 0`
`47`	`47`
`48`	`48`	`/**`
`49`	`49`	`* Configuration class that loads both the federated and non-federation`
`50`		`- * configuraiton and creates the GraphQL schema object and request handler.`
	`50`	`+ * configuration and creates the GraphQL schema object and request handler.`
`51`	`51`	`*`
`52`	`52`	`* This config can then be used by all Spring specific configuration classes`
`53`	`53`	`* to handle incoming requests from HTTP routes or subscriptions and send them`