Basic setup - proto & java classes copied from Bazel source - set up as 2 modules; one for the proto & one for Java code; working MODULE.bazel files - TODO: CI, Copybara, publish-to-BCR, etc
diff --git a/.bazelversion b/.bazelversion new file mode 100644 index 0000000..b26a34e --- /dev/null +++ b/.bazelversion
@@ -0,0 +1 @@ +7.2.1
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..57c461f --- /dev/null +++ b/.gitignore
@@ -0,0 +1,2 @@ +bazel-* +
diff --git a/java/.bazelrc b/java/.bazelrc new file mode 100644 index 0000000..5e3c7b5 --- /dev/null +++ b/java/.bazelrc
@@ -0,0 +1 @@ +build --java_language_version=21
diff --git a/java/.bazelversion b/java/.bazelversion new file mode 120000 index 0000000..b332604 --- /dev/null +++ b/java/.bazelversion
@@ -0,0 +1 @@ +../.bazelversion \ No newline at end of file
diff --git a/java/BUILD.bazel b/java/BUILD.bazel new file mode 100644 index 0000000..63ee1b0 --- /dev/null +++ b/java/BUILD.bazel
@@ -0,0 +1,7 @@ +package(default_visibility = ["//visibility:public"]) + +java_proto_library( + name = "worker_protocol_java_proto", + deps = ["@bazel_worker_api//:worker_protocol_proto"], +) +
diff --git a/java/MODULE.bazel b/java/MODULE.bazel new file mode 100644 index 0000000..b31ea41 --- /dev/null +++ b/java/MODULE.bazel
@@ -0,0 +1,44 @@ +module( + name = "bazel_worker_java", + version = "0", +) + +bazel_dep(name = "bazel_worker_api", version = "0.0.1") +bazel_dep(name = "rules_jvm_external", version = "6.2") +bazel_dep(name = "protobuf", version = "27.2") + +maven = use_extension("@rules_jvm_external//:extensions.bzl", "maven") +maven.install( + artifacts = [ + "com.google.code.gson:gson:2.10.1", + "com.google.errorprone:error_prone_annotations:2.23.0", + "com.google.guava:guava:33.0.0-jre", + "com.google.protobuf:protobuf-java:4.27.2", + "com.google.protobuf:protobuf-java-util:4.27.2", + ], +) +maven.artifact( + artifact = "junit", + group = "junit", + testonly = True, + version = "4.13.2", +) +maven.artifact( + artifact = "mockito-core", + group = "org.mockito", + testonly = True, + version = "5.4.0", +) +maven.artifact( + artifact = "truth", + group = "com.google.truth", + testonly = True, + version = "1.4.0", +) +use_repo(maven, "maven") + +local_path_override( + module_name = "bazel_worker_api", + path = "../proto", +) +
diff --git a/java/MODULE.bazel.lock b/java/MODULE.bazel.lock new file mode 100644 index 0000000..aa2c11f --- /dev/null +++ b/java/MODULE.bazel.lock
@@ -0,0 +1,336 @@ +{ + "lockFileVersion": 11, + "registryFileHashes": { + "https://bcr.bazel.build/bazel_registry.json": "8a28e4aff06ee60aed2a8c281907fb8bcbf3b753c91fb5a5c57da3215d5b3497", + "https://bcr.bazel.build/modules/abseil-cpp/20210324.2/MODULE.bazel": "7cd0312e064fde87c8d1cd79ba06c876bd23630c83466e9500321be55c96ace2", + "https://bcr.bazel.build/modules/abseil-cpp/20211102.0/MODULE.bazel": "70390338f7a5106231d20620712f7cccb659cd0e9d073d1991c038eb9fc57589", + "https://bcr.bazel.build/modules/abseil-cpp/20230125.1/MODULE.bazel": "89047429cb0207707b2dface14ba7f8df85273d484c2572755be4bab7ce9c3a0", + "https://bcr.bazel.build/modules/abseil-cpp/20230802.0.bcr.1/MODULE.bazel": "1c8cec495288dccd14fdae6e3f95f772c1c91857047a098fad772034264cc8cb", + "https://bcr.bazel.build/modules/abseil-cpp/20230802.0.bcr.1/source.json": "14892cc698e02ffedf4967546e6bedb7245015906888d3465fcf27c90a26da10", + "https://bcr.bazel.build/modules/apple_support/1.5.0/MODULE.bazel": "50341a62efbc483e8a2a6aec30994a58749bd7b885e18dd96aa8c33031e558ef", + "https://bcr.bazel.build/modules/apple_support/1.5.0/source.json": "eb98a7627c0bc486b57f598ad8da50f6625d974c8f723e9ea71bd39f709c9862", + "https://bcr.bazel.build/modules/bazel_features/1.11.0/MODULE.bazel": "f9382337dd5a474c3b7d334c2f83e50b6eaedc284253334cf823044a26de03e8", + "https://bcr.bazel.build/modules/bazel_features/1.13.0/MODULE.bazel": "c14c33c7c3c730612bdbe14ebbb5e61936b6f11322ea95a6e91cd1ba962f94df", + "https://bcr.bazel.build/modules/bazel_features/1.13.0/source.json": "b01f6aaaf93527ff4267421ef416debbd89b3166b70af5c89400c6a95a89c133", + "https://bcr.bazel.build/modules/bazel_skylib/1.0.3/MODULE.bazel": "bcb0fd896384802d1ad283b4e4eb4d718eebd8cb820b0a2c3a347fb971afd9d8", + "https://bcr.bazel.build/modules/bazel_skylib/1.2.0/MODULE.bazel": "44fe84260e454ed94ad326352a698422dbe372b21a1ac9f3eab76eb531223686", + "https://bcr.bazel.build/modules/bazel_skylib/1.2.1/MODULE.bazel": "f35baf9da0efe45fa3da1696ae906eea3d615ad41e2e3def4aeb4e8bc0ef9a7a", + "https://bcr.bazel.build/modules/bazel_skylib/1.3.0/MODULE.bazel": "20228b92868bf5cfc41bda7afc8a8ba2a543201851de39d990ec957b513579c5", + "https://bcr.bazel.build/modules/bazel_skylib/1.4.1/MODULE.bazel": "a0dcb779424be33100dcae821e9e27e4f2901d9dfd5333efe5ac6a8d7ab75e1d", + "https://bcr.bazel.build/modules/bazel_skylib/1.4.2/MODULE.bazel": "3bd40978e7a1fac911d5989e6b09d8f64921865a45822d8b09e815eaa726a651", + "https://bcr.bazel.build/modules/bazel_skylib/1.6.1/MODULE.bazel": "8fdee2dbaace6c252131c00e1de4b165dc65af02ea278476187765e1a617b917", + "https://bcr.bazel.build/modules/bazel_skylib/1.7.1/MODULE.bazel": "3120d80c5861aa616222ec015332e5f8d3171e062e3e804a2a0253e1be26e59b", + "https://bcr.bazel.build/modules/bazel_skylib/1.7.1/source.json": "f121b43eeefc7c29efbd51b83d08631e2347297c95aac9764a701f2a6a2bb953", + "https://bcr.bazel.build/modules/buildozer/7.1.2/MODULE.bazel": "2e8dd40ede9c454042645fd8d8d0cd1527966aa5c919de86661e62953cd73d84", + "https://bcr.bazel.build/modules/buildozer/7.1.2/source.json": "c9028a501d2db85793a6996205c8de120944f50a0d570438fcae0457a5f9d1f8", + "https://bcr.bazel.build/modules/googletest/1.11.0/MODULE.bazel": "3a83f095183f66345ca86aa13c58b59f9f94a2f81999c093d4eeaa2d262d12f4", + "https://bcr.bazel.build/modules/googletest/1.14.0/MODULE.bazel": "cfbcbf3e6eac06ef9d85900f64424708cc08687d1b527f0ef65aa7517af8118f", + "https://bcr.bazel.build/modules/googletest/1.14.0/source.json": "2478949479000fdd7de9a3d0107ba2c85bb5f961c3ecb1aa448f52549ce310b5", + "https://bcr.bazel.build/modules/jsoncpp/1.9.5/MODULE.bazel": "31271aedc59e815656f5736f282bb7509a97c7ecb43e927ac1a37966e0578075", + "https://bcr.bazel.build/modules/jsoncpp/1.9.5/source.json": "4108ee5085dd2885a341c7fab149429db457b3169b86eb081fa245eadf69169d", + "https://bcr.bazel.build/modules/platforms/0.0.10/MODULE.bazel": "8cb8efaf200bdeb2150d93e162c40f388529a25852b332cec879373771e48ed5", + "https://bcr.bazel.build/modules/platforms/0.0.10/source.json": "f22828ff4cf021a6b577f1bf6341cb9dcd7965092a439f64fc1bb3b7a5ae4bd5", + "https://bcr.bazel.build/modules/platforms/0.0.4/MODULE.bazel": "9b328e31ee156f53f3c416a64f8491f7eb731742655a47c9eec4703a71644aee", + "https://bcr.bazel.build/modules/platforms/0.0.5/MODULE.bazel": "5733b54ea419d5eaf7997054bb55f6a1d0b5ff8aedf0176fef9eea44f3acda37", + "https://bcr.bazel.build/modules/platforms/0.0.6/MODULE.bazel": "ad6eeef431dc52aefd2d77ed20a4b353f8ebf0f4ecdd26a807d2da5aa8cd0615", + "https://bcr.bazel.build/modules/platforms/0.0.7/MODULE.bazel": "72fd4a0ede9ee5c021f6a8dd92b503e089f46c227ba2813ff183b71616034814", + "https://bcr.bazel.build/modules/platforms/0.0.8/MODULE.bazel": "9f142c03e348f6d263719f5074b21ef3adf0b139ee4c5133e2aa35664da9eb2d", + "https://bcr.bazel.build/modules/platforms/0.0.9/MODULE.bazel": "4a87a60c927b56ddd67db50c89acaa62f4ce2a1d2149ccb63ffd871d5ce29ebc", + "https://bcr.bazel.build/modules/protobuf/21.7/MODULE.bazel": "a5a29bb89544f9b97edce05642fac225a808b5b7be74038ea3640fae2f8e66a7", + "https://bcr.bazel.build/modules/protobuf/27.2/MODULE.bazel": "32450b50673882e4c8c3d10a83f3bc82161b213ed2f80d17e38bece8f165c295", + "https://bcr.bazel.build/modules/protobuf/27.2/source.json": "37722548d44c5f3e81dc6a3d1dfb49868aee80cc63db90d843fcefbe3fd220af", + "https://bcr.bazel.build/modules/protobuf/3.19.0/MODULE.bazel": "6b5fbb433f760a99a22b18b6850ed5784ef0e9928a72668b66e4d7ccd47db9b0", + "https://bcr.bazel.build/modules/protobuf/3.19.6/MODULE.bazel": "9233edc5e1f2ee276a60de3eaa47ac4132302ef9643238f23128fea53ea12858", + "https://bcr.bazel.build/modules/rules_android/0.1.1/MODULE.bazel": "48809ab0091b07ad0182defb787c4c5328bd3a278938415c00a7b69b50c4d3a8", + "https://bcr.bazel.build/modules/rules_android/0.1.1/source.json": "e6986b41626ee10bdc864937ffb6d6bf275bb5b9c65120e6137d56e6331f089e", + "https://bcr.bazel.build/modules/rules_cc/0.0.1/MODULE.bazel": "cb2aa0747f84c6c3a78dad4e2049c154f08ab9d166b1273835a8174940365647", + "https://bcr.bazel.build/modules/rules_cc/0.0.2/MODULE.bazel": "6915987c90970493ab97393024c156ea8fb9f3bea953b2f3ec05c34f19b5695c", + "https://bcr.bazel.build/modules/rules_cc/0.0.6/MODULE.bazel": "abf360251023dfe3efcef65ab9d56beefa8394d4176dd29529750e1c57eaa33f", + "https://bcr.bazel.build/modules/rules_cc/0.0.8/MODULE.bazel": "964c85c82cfeb6f3855e6a07054fdb159aced38e99a5eecf7bce9d53990afa3e", + "https://bcr.bazel.build/modules/rules_cc/0.0.9/MODULE.bazel": "836e76439f354b89afe6a911a7adf59a6b2518fafb174483ad78a2a2fde7b1c5", + "https://bcr.bazel.build/modules/rules_cc/0.0.9/source.json": "1f1ba6fea244b616de4a554a0f4983c91a9301640c8fe0dd1d410254115c8430", + "https://bcr.bazel.build/modules/rules_java/4.0.0/MODULE.bazel": "5a78a7ae82cd1a33cef56dc578c7d2a46ed0dca12643ee45edbb8417899e6f74", + "https://bcr.bazel.build/modules/rules_java/5.3.5/MODULE.bazel": "a4ec4f2db570171e3e5eb753276ee4b389bae16b96207e9d3230895c99644b86", + "https://bcr.bazel.build/modules/rules_java/7.2.0/MODULE.bazel": "06c0334c9be61e6cef2c8c84a7800cef502063269a5af25ceb100b192453d4ab", + "https://bcr.bazel.build/modules/rules_java/7.4.0/MODULE.bazel": "a592852f8a3dd539e82ee6542013bf2cadfc4c6946be8941e189d224500a8934", + "https://bcr.bazel.build/modules/rules_java/7.6.1/MODULE.bazel": "2f14b7e8a1aa2f67ae92bc69d1ec0fa8d9f827c4e17ff5e5f02e91caa3b2d0fe", + "https://bcr.bazel.build/modules/rules_java/7.6.1/source.json": "8f3f3076554e1558e8e468b2232991c510ecbcbed9e6f8c06ac31c93bcf38362", + "https://bcr.bazel.build/modules/rules_jvm_external/4.4.2/MODULE.bazel": "a56b85e418c83eb1839819f0b515c431010160383306d13ec21959ac412d2fe7", + "https://bcr.bazel.build/modules/rules_jvm_external/5.1/MODULE.bazel": "33f6f999e03183f7d088c9be518a63467dfd0be94a11d0055fe2d210f89aa909", + "https://bcr.bazel.build/modules/rules_jvm_external/5.2/MODULE.bazel": "d9351ba35217ad0de03816ef3ed63f89d411349353077348a45348b096615036", + "https://bcr.bazel.build/modules/rules_jvm_external/6.2/MODULE.bazel": "36a6e52487a855f33cb960724eb56547fa87e2c98a0474c3acad94339d7f8e99", + "https://bcr.bazel.build/modules/rules_jvm_external/6.2/source.json": "7f4c0095f17d1b65f943169e1d4f5f831a7133d205179e7b1b515ffcb39a5aa4", + "https://bcr.bazel.build/modules/rules_kotlin/1.9.5/MODULE.bazel": "043a16a572f610558ec2030db3ff0c9938574e7dd9f58bded1bb07c0192ef025", + "https://bcr.bazel.build/modules/rules_kotlin/1.9.5/source.json": "4ea6e867a7db5ef01187a07636070110db267f94ed038367bec7bbd3aea5e612", + "https://bcr.bazel.build/modules/rules_license/0.0.3/MODULE.bazel": "627e9ab0247f7d1e05736b59dbb1b6871373de5ad31c3011880b4133cafd4bd0", + "https://bcr.bazel.build/modules/rules_license/0.0.7/MODULE.bazel": "088fbeb0b6a419005b89cf93fe62d9517c0a2b8bb56af3244af65ecfe37e7d5d", + "https://bcr.bazel.build/modules/rules_license/0.0.7/source.json": "355cc5737a0f294e560d52b1b7a6492d4fff2caf0bef1a315df5a298fca2d34a", + "https://bcr.bazel.build/modules/rules_pkg/0.7.0/MODULE.bazel": "df99f03fc7934a4737122518bb87e667e62d780b610910f0447665a7e2be62dc", + "https://bcr.bazel.build/modules/rules_pkg/0.7.0/source.json": "c2557066e0c0342223ba592510ad3d812d4963b9024831f7f66fd0584dd8c66c", + "https://bcr.bazel.build/modules/rules_proto/4.0.0/MODULE.bazel": "a7a7b6ce9bee418c1a760b3d84f83a299ad6952f9903c67f19e4edd964894e06", + "https://bcr.bazel.build/modules/rules_proto/5.3.0-21.7/MODULE.bazel": "e8dff86b0971688790ae75528fe1813f71809b5afd57facb44dad9e8eca631b7", + "https://bcr.bazel.build/modules/rules_proto/5.3.0-21.7/source.json": "d57902c052424dfda0e71646cb12668d39c4620ee0544294d9d941e7d12bc3a9", + "https://bcr.bazel.build/modules/rules_python/0.10.2/MODULE.bazel": "cc82bc96f2997baa545ab3ce73f196d040ffb8756fd2d66125a530031cd90e5f", + "https://bcr.bazel.build/modules/rules_python/0.22.1/MODULE.bazel": "26114f0c0b5e93018c0c066d6673f1a2c3737c7e90af95eff30cfee38d0bbac7", + "https://bcr.bazel.build/modules/rules_python/0.23.1/MODULE.bazel": "49ffccf0511cb8414de28321f5fcf2a31312b47c40cc21577144b7447f2bf300", + "https://bcr.bazel.build/modules/rules_python/0.23.1/source.json": "a6d9965700e3bd75df4e19140c0e651851bb720d8b9eb280ecd1ee44b92d7646", + "https://bcr.bazel.build/modules/rules_python/0.4.0/MODULE.bazel": "9208ee05fd48bf09ac60ed269791cf17fb343db56c8226a720fbb1cdf467166c", + "https://bcr.bazel.build/modules/stardoc/0.5.1/MODULE.bazel": "1a05d92974d0c122f5ccf09291442580317cdd859f07a8655f1db9a60374f9f8", + "https://bcr.bazel.build/modules/stardoc/0.5.3/MODULE.bazel": "c7f6948dae6999bf0db32c1858ae345f112cacf98f174c7a8bb707e41b974f1c", + "https://bcr.bazel.build/modules/stardoc/0.7.0/MODULE.bazel": "05e3d6d30c099b6770e97da986c53bd31844d7f13d41412480ea265ac9e8079c", + "https://bcr.bazel.build/modules/stardoc/0.7.0/source.json": "e3c524bf2ef20992539ce2bc4a2243f4853130209ee831689983e28d05769099", + "https://bcr.bazel.build/modules/upb/0.0.0-20220923-a547704/MODULE.bazel": "7298990c00040a0e2f121f6c32544bab27d4452f80d9ce51349b1a28f3005c43", + "https://bcr.bazel.build/modules/zlib/1.2.11/MODULE.bazel": "07b389abc85fdbca459b69e2ec656ae5622873af3f845e1c9d80fe179f3effa0", + "https://bcr.bazel.build/modules/zlib/1.2.12/MODULE.bazel": "3b1a8834ada2a883674be8cbd36ede1b6ec481477ada359cd2d3ddc562340b27", + "https://bcr.bazel.build/modules/zlib/1.3/MODULE.bazel": "6a9c02f19a24dcedb05572b2381446e27c272cd383aed11d41d99da9e3167a72", + "https://bcr.bazel.build/modules/zlib/1.3/source.json": "b6b43d0737af846022636e6e255fd4a96fee0d34f08f3830e6e0bac51465c37c" + }, + "selectedYankedVersions": {}, + "moduleExtensions": { + "@@apple_support~//crosstool:setup.bzl%apple_cc_configure_extension": { + "general": { + "bzlTransitiveDigest": "PjIds3feoYE8SGbbIq2SFTZy3zmxeO2tQevJZNDo7iY=", + "usagesDigest": "aLmqbvowmHkkBPve05yyDNGN7oh7QE9kBADr3QIZTZs=", + "recordedFileInputs": {}, + "recordedDirentsInputs": {}, + "envVariables": {}, + "generatedRepoSpecs": { + "local_config_apple_cc": { + "bzlFile": "@@apple_support~//crosstool:setup.bzl", + "ruleClassName": "_apple_cc_autoconf", + "attributes": {} + }, + "local_config_apple_cc_toolchains": { + "bzlFile": "@@apple_support~//crosstool:setup.bzl", + "ruleClassName": "_apple_cc_autoconf_toolchains", + "attributes": {} + } + }, + "recordedRepoMappingEntries": [ + [ + "apple_support~", + "bazel_tools", + "bazel_tools" + ] + ] + } + }, + "@@platforms//host:extension.bzl%host_platform": { + "general": { + "bzlTransitiveDigest": "xelQcPZH8+tmuOHVjL9vDxMnnQNMlwj0SlvgoqBkm4U=", + "usagesDigest": "V1R2Y2oMxKNfx2WCWpSCaUV1WefW1o8HZGm3v1vHgY4=", + "recordedFileInputs": {}, + "recordedDirentsInputs": {}, + "envVariables": {}, + "generatedRepoSpecs": { + "host_platform": { + "bzlFile": "@@platforms//host:extension.bzl", + "ruleClassName": "host_platform_repo", + "attributes": {} + } + }, + "recordedRepoMappingEntries": [] + } + }, + "@@rules_kotlin~//src/main/starlark/core/repositories:bzlmod_setup.bzl%rules_kotlin_extensions": { + "general": { + "bzlTransitiveDigest": "kZ+9OOxWkWt07Jni1BKXITpFDZRZoYjB0LCALFyjGkc=", + "usagesDigest": "K+i+NLBtqsvNSdO+8wYj1BOGr0DAXxBxW8myZpWViyc=", + "recordedFileInputs": {}, + "recordedDirentsInputs": {}, + "envVariables": {}, + "generatedRepoSpecs": { + "rules_android": { + "bzlFile": "@@bazel_tools//tools/build_defs/repo:http.bzl", + "ruleClassName": "http_archive", + "attributes": { + "sha256": "cd06d15dd8bb59926e4d65f9003bfc20f9da4b2519985c27e190cddc8b7a7806", + "strip_prefix": "rules_android-0.1.1", + "urls": [ + "https://github.com/bazelbuild/rules_android/archive/v0.1.1.zip" + ] + } + }, + "com_github_pinterest_ktlint": { + "bzlFile": "@@bazel_tools//tools/build_defs/repo:http.bzl", + "ruleClassName": "http_file", + "attributes": { + "sha256": "2b3f6f674a944d25bb8d283c3539947bbe86074793012909a55de4b771f74bcc", + "urls": [ + "https://github.com/pinterest/ktlint/releases/download/0.49.1/ktlint" + ], + "executable": true + } + }, + "com_github_jetbrains_kotlin": { + "bzlFile": "@@rules_kotlin~//src/main/starlark/core/repositories:compiler.bzl", + "ruleClassName": "kotlin_compiler_repository", + "attributes": { + "urls": [ + "https://github.com/JetBrains/kotlin/releases/download/v1.9.22/kotlin-compiler-1.9.22.zip" + ], + "sha256": "88b39213506532c816ff56348c07bbeefe0c8d18943bffbad11063cf97cac3e6", + "compiler_version": "1.9.22" + } + }, + "com_github_google_ksp": { + "bzlFile": "@@rules_kotlin~//src/main/starlark/core/repositories:ksp.bzl", + "ruleClassName": "ksp_compiler_plugin_repository", + "attributes": { + "urls": [ + "https://github.com/google/ksp/releases/download/1.9.22-1.0.17/artifacts.zip" + ], + "sha256": "b39b373e09e5edefe700fef628572f71be7d49e6396dec0ea52eb10c16ead39e", + "strip_version": "1.9.22-1.0.17" + } + } + }, + "recordedRepoMappingEntries": [ + [ + "rules_kotlin~", + "bazel_tools", + "bazel_tools" + ] + ] + } + }, + "@@rules_python~//python/extensions:python.bzl%python": { + "general": { + "bzlTransitiveDigest": "XaaZIw4dO4l6naftU5IBdrfCE1mOmelaT/Sq9uyBnhs=", + "usagesDigest": "9rlrm2M/kJEEPWIo3UEIjkAFxHjzsbMIAFR9yrYnKsQ=", + "recordedFileInputs": {}, + "recordedDirentsInputs": {}, + "envVariables": {}, + "generatedRepoSpecs": { + "python_aliases": { + "bzlFile": "@@rules_python~//python/private:toolchains_repo.bzl", + "ruleClassName": "multi_toolchain_aliases", + "attributes": { + "python_versions": { + "3.11": "python_3_11" + } + } + }, + "python_3_11": { + "bzlFile": "@@rules_python~//python/private:toolchains_repo.bzl", + "ruleClassName": "toolchain_aliases", + "attributes": { + "python_version": "3.11.1", + "user_repository_name": "python_3_11" + } + }, + "python_3_11_aarch64-unknown-linux-gnu": { + "bzlFile": "@@rules_python~//python:repositories.bzl", + "ruleClassName": "python_repository", + "attributes": { + "sha256": "debf15783bdcb5530504f533d33fda75a7b905cec5361ae8f33da5ba6599f8b4", + "patches": [], + "platform": "aarch64-unknown-linux-gnu", + "python_version": "3.11.1", + "release_filename": "20230116/cpython-3.11.1+20230116-aarch64-unknown-linux-gnu-install_only.tar.gz", + "urls": [ + "https://github.com/indygreg/python-build-standalone/releases/download/20230116/cpython-3.11.1+20230116-aarch64-unknown-linux-gnu-install_only.tar.gz" + ], + "distutils_content": "", + "strip_prefix": "python", + "ignore_root_user_error": false + } + }, + "python_3_11_aarch64-apple-darwin": { + "bzlFile": "@@rules_python~//python:repositories.bzl", + "ruleClassName": "python_repository", + "attributes": { + "sha256": "4918cdf1cab742a90f85318f88b8122aeaa2d04705803c7b6e78e81a3dd40f80", + "patches": [], + "platform": "aarch64-apple-darwin", + "python_version": "3.11.1", + "release_filename": "20230116/cpython-3.11.1+20230116-aarch64-apple-darwin-install_only.tar.gz", + "urls": [ + "https://github.com/indygreg/python-build-standalone/releases/download/20230116/cpython-3.11.1+20230116-aarch64-apple-darwin-install_only.tar.gz" + ], + "distutils_content": "", + "strip_prefix": "python", + "ignore_root_user_error": false + } + }, + "python_3_11_x86_64-apple-darwin": { + "bzlFile": "@@rules_python~//python:repositories.bzl", + "ruleClassName": "python_repository", + "attributes": { + "sha256": "20a4203d069dc9b710f70b09e7da2ce6f473d6b1110f9535fb6f4c469ed54733", + "patches": [], + "platform": "x86_64-apple-darwin", + "python_version": "3.11.1", + "release_filename": "20230116/cpython-3.11.1+20230116-x86_64-apple-darwin-install_only.tar.gz", + "urls": [ + "https://github.com/indygreg/python-build-standalone/releases/download/20230116/cpython-3.11.1+20230116-x86_64-apple-darwin-install_only.tar.gz" + ], + "distutils_content": "", + "strip_prefix": "python", + "ignore_root_user_error": false + } + }, + "pythons_hub": { + "bzlFile": "@@rules_python~//python/extensions/private:pythons_hub.bzl", + "ruleClassName": "hub_repo", + "attributes": { + "toolchain_prefixes": [ + "_0000_python_3_11_" + ], + "toolchain_python_versions": [ + "3.11" + ], + "toolchain_set_python_version_constraints": [ + "False" + ], + "toolchain_user_repository_names": [ + "python_3_11" + ] + } + }, + "python_3_11_x86_64-pc-windows-msvc": { + "bzlFile": "@@rules_python~//python:repositories.bzl", + "ruleClassName": "python_repository", + "attributes": { + "sha256": "edc08979cb0666a597466176511529c049a6f0bba8adf70df441708f766de5bf", + "patches": [], + "platform": "x86_64-pc-windows-msvc", + "python_version": "3.11.1", + "release_filename": "20230116/cpython-3.11.1+20230116-x86_64-pc-windows-msvc-shared-install_only.tar.gz", + "urls": [ + "https://github.com/indygreg/python-build-standalone/releases/download/20230116/cpython-3.11.1+20230116-x86_64-pc-windows-msvc-shared-install_only.tar.gz" + ], + "distutils_content": "", + "strip_prefix": "python", + "ignore_root_user_error": false + } + }, + "python_3_11_x86_64-unknown-linux-gnu": { + "bzlFile": "@@rules_python~//python:repositories.bzl", + "ruleClassName": "python_repository", + "attributes": { + "sha256": "02a551fefab3750effd0e156c25446547c238688a32fabde2995c941c03a6423", + "patches": [], + "platform": "x86_64-unknown-linux-gnu", + "python_version": "3.11.1", + "release_filename": "20230116/cpython-3.11.1+20230116-x86_64-unknown-linux-gnu-install_only.tar.gz", + "urls": [ + "https://github.com/indygreg/python-build-standalone/releases/download/20230116/cpython-3.11.1+20230116-x86_64-unknown-linux-gnu-install_only.tar.gz" + ], + "distutils_content": "", + "strip_prefix": "python", + "ignore_root_user_error": false + } + } + }, + "recordedRepoMappingEntries": [ + [ + "rules_python~", + "bazel_tools", + "bazel_tools" + ] + ] + } + } + } +}
diff --git a/java/src/main/java/com/google/devtools/build/lib/worker/BUILD.bazel b/java/src/main/java/com/google/devtools/build/lib/worker/BUILD.bazel new file mode 100644 index 0000000..6670f2d --- /dev/null +++ b/java/src/main/java/com/google/devtools/build/lib/worker/BUILD.bazel
@@ -0,0 +1,19 @@ +package(default_visibility = ["//visibility:public"]) + +java_library( + name = "work_request_handlers", + srcs = [ + "JsonWorkerMessageProcessor.java", + "ProtoWorkerMessageProcessor.java", + "WorkRequestHandler.java", + ], + deps = [ + "//:worker_protocol_java_proto", + "@maven//:com_google_code_gson_gson", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + "@maven//:com_google_protobuf_protobuf_java", + "@maven//:com_google_protobuf_protobuf_java_util", + ], +) +
diff --git a/java/src/main/java/com/google/devtools/build/lib/worker/JsonWorkerMessageProcessor.java b/java/src/main/java/com/google/devtools/build/lib/worker/JsonWorkerMessageProcessor.java new file mode 100644 index 0000000..70d746c --- /dev/null +++ b/java/src/main/java/com/google/devtools/build/lib/worker/JsonWorkerMessageProcessor.java
@@ -0,0 +1,185 @@ +// Copyright 2020 The Bazel Authors. All rights reserved. +// +// 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 +// +// http://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.google.devtools.build.lib.worker; + +import com.google.common.collect.ImmutableList; +import com.google.devtools.build.lib.worker.WorkerProtocol.Input; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkRequest; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkResponse; +import com.google.gson.stream.JsonReader; +import com.google.gson.stream.MalformedJsonException; +import com.google.protobuf.ByteString; +import com.google.protobuf.util.JsonFormat; +import com.google.protobuf.util.JsonFormat.Printer; +import java.io.BufferedWriter; +import java.io.EOFException; +import java.io.IOException; +import java.util.List; + +/** Implementation of the Worker Protocol using JSON to communicate with Bazel. */ +public final class JsonWorkerMessageProcessor implements WorkRequestHandler.WorkerMessageProcessor { + /** Reader for reading the WorkResponse. */ + private final JsonReader reader; + /** Printer for printing the WorkRequest. */ + private final Printer jsonPrinter; + /** Writer for writing the WorkRequest to the worker. */ + private final BufferedWriter jsonWriter; + + /** Constructs a {@code WorkRequestHandler} that reads and writes JSON. */ + public JsonWorkerMessageProcessor(JsonReader reader, BufferedWriter jsonWriter) { + this.reader = reader; + reader.setLenient(true); + this.jsonWriter = jsonWriter; + jsonPrinter = + JsonFormat.printer().omittingInsignificantWhitespace().alwaysPrintFieldsWithNoPresence(); + } + + private static ImmutableList<String> readArguments(JsonReader reader) throws IOException { + reader.beginArray(); + ImmutableList.Builder<String> argumentsBuilder = ImmutableList.builder(); + while (reader.hasNext()) { + argumentsBuilder.add(reader.nextString()); + } + reader.endArray(); + return argumentsBuilder.build(); + } + + private static ImmutableList<Input> readInputs(JsonReader reader) throws IOException { + reader.beginArray(); + ImmutableList.Builder<Input> inputsBuilder = ImmutableList.builder(); + while (reader.hasNext()) { + String digest = null; + String path = null; + + reader.beginObject(); + while (reader.hasNext()) { + String name = reader.nextName(); + switch (name) { + case "digest": + if (digest != null) { + throw new IOException("Input cannot have more than one digest"); + } + digest = reader.nextString(); + break; + case "path": + if (path != null) { + throw new IOException("Input cannot have more than one path"); + } + path = reader.nextString(); + break; + default: + // As per https://bazel.build/docs/creating-workers#work-responses, + // unknown fields are ignored. + reader.skipValue(); + break; + } + } + reader.endObject(); + Input.Builder inputBuilder = Input.newBuilder(); + if (digest != null) { + inputBuilder.setDigest(ByteString.copyFromUtf8(digest)); + } + if (path != null) { + inputBuilder.setPath(path); + } + inputsBuilder.add(inputBuilder.build()); + } + reader.endArray(); + return inputsBuilder.build(); + } + + @Override + public WorkRequest readWorkRequest() throws IOException { + List<String> arguments = null; + List<Input> inputs = null; + Integer requestId = null; + Integer verbosity = null; + String sandboxDir = null; + try { + reader.beginObject(); + while (reader.hasNext()) { + String name = reader.nextName(); + switch (name) { + case "arguments": + if (arguments != null) { + throw new IOException("WorkRequest cannot have more than one 'arguments' field"); + } + arguments = readArguments(reader); + break; + case "inputs": + if (inputs != null) { + throw new IOException("WorkRequest cannot have more than one 'inputs' field"); + } + inputs = readInputs(reader); + break; + case "requestId": + if (requestId != null) { + throw new IOException("WorkRequest cannot have more than one requestId"); + } + requestId = reader.nextInt(); + break; + case "verbosity": + if (verbosity != null) { + throw new IOException("Work response cannot have more than one verbosity"); + } + verbosity = reader.nextInt(); + break; + case "sandboxDir": + if (sandboxDir != null) { + throw new IOException("Work response cannot have more than one sandboxDir"); + } + sandboxDir = reader.nextString(); + break; + default: + // As per https://bazel.build/docs/creating-workers#work-responses, + // unknown fields are ignored. + reader.skipValue(); + break; + } + } + reader.endObject(); + } catch (MalformedJsonException | IllegalStateException | EOFException e) { + throw new IOException(e); + } + + WorkRequest.Builder requestBuilder = WorkRequest.newBuilder(); + if (arguments != null) { + requestBuilder.addAllArguments(arguments); + } + if (inputs != null) { + requestBuilder.addAllInputs(inputs); + } + if (requestId != null) { + requestBuilder.setRequestId(requestId); + } + if (verbosity != null) { + requestBuilder.setVerbosity(verbosity); + } + if (sandboxDir != null) { + requestBuilder.setSandboxDir(sandboxDir); + } + return requestBuilder.build(); + } + + @Override + public void writeWorkResponse(WorkResponse response) throws IOException { + jsonPrinter.appendTo(response, jsonWriter); + jsonWriter.flush(); + } + + @Override + public void close() throws IOException { + jsonWriter.close(); + } +}
diff --git a/java/src/main/java/com/google/devtools/build/lib/worker/ProtoWorkerMessageProcessor.java b/java/src/main/java/com/google/devtools/build/lib/worker/ProtoWorkerMessageProcessor.java new file mode 100644 index 0000000..182695d --- /dev/null +++ b/java/src/main/java/com/google/devtools/build/lib/worker/ProtoWorkerMessageProcessor.java
@@ -0,0 +1,54 @@ +// Copyright 2020 The Bazel Authors. All rights reserved. +// +// 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 +// +// http://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.google.devtools.build.lib.worker; + +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkRequest; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkResponse; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +/** Implementation of the Worker Protocol using Proto to communicate with Bazel. */ +public final class ProtoWorkerMessageProcessor + implements WorkRequestHandler.WorkerMessageProcessor { + + /** This worker's stdin. */ + private final InputStream stdin; + + /** This worker's stdout. Only {@link WorkRequest}s should be written here. */ + private final OutputStream stdout; + + /** Constructs a {@link WorkRequestHandler} that reads and writes Protocol Buffers. */ + public ProtoWorkerMessageProcessor(InputStream stdin, OutputStream stdout) { + this.stdin = stdin; + this.stdout = stdout; + } + + @Override + public WorkRequest readWorkRequest() throws IOException { + return WorkRequest.parseDelimitedFrom(stdin); + } + + @Override + public void writeWorkResponse(WorkResponse workResponse) throws IOException { + try { + workResponse.writeDelimitedTo(stdout); + } finally { + stdout.flush(); + } + } + + @Override + public void close() {} +}
diff --git a/java/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java b/java/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java new file mode 100644 index 0000000..dcdd60d --- /dev/null +++ b/java/src/main/java/com/google/devtools/build/lib/worker/WorkRequestHandler.java
@@ -0,0 +1,767 @@ +// Copyright 2020 The Bazel Authors. All rights reserved. +// +// 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 +// +// http://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.google.devtools.build.lib.worker; + +import com.google.common.annotations.VisibleForTesting; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkRequest; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkResponse; +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import com.sun.management.OperatingSystemMXBean; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.PrintStream; +import java.io.PrintWriter; +import java.io.StringWriter; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; + +/** + * A helper class that handles WorkRequests (https://bazel.build/docs/persistent-workers), including + * multiplex workers (https://bazel.build/docs/multiplex-worker). + */ +public class WorkRequestHandler implements AutoCloseable { + /** Contains the logic for reading {@link WorkRequest}s and writing {@link WorkResponse}s. */ + public interface WorkerMessageProcessor { + /** Reads the next incoming request from this worker's stdin. */ + WorkRequest readWorkRequest() throws IOException; + + /** + * Writes the provided {@link WorkResponse} to this worker's stdout. This function is also + * responsible for flushing the stdout. + */ + void writeWorkResponse(WorkResponse workResponse) throws IOException; + + /** Clean up. */ + void close() throws IOException; + } + + /** Holds information necessary to properly handle a request, especially for cancellation. */ + static class RequestInfo { + /** The thread handling the request. */ + final Thread thread; + /** If true, we have received a cancel request for this request. */ + private final AtomicBoolean cancelled = new AtomicBoolean(false); + /** + * The builder for the response to this request. Since only one response must be sent per + * request, this builder must be accessed through takeBuilder(), which zeroes this field and + * returns the builder. + */ + private WorkResponse.Builder responseBuilder = WorkResponse.newBuilder(); + + RequestInfo(Thread thread) { + this.thread = thread; + } + + /** Sets whether this request has been cancelled. */ + void setCancelled() { + cancelled.set(true); + } + + /** Returns true if this request has been cancelled. */ + boolean isCancelled() { + return cancelled.get(); + } + + /** + * Returns the response builder. If called more than once on the same instance, subsequent calls + * will return {@code null}. + */ + synchronized Optional<WorkResponse.Builder> takeBuilder() { + WorkResponse.Builder b = responseBuilder; + responseBuilder = null; + return Optional.ofNullable(b); + } + + /** + * Adds {@code s} as output to when the response eventually gets built. Does nothing if the + * response has already been taken. There is no guarantee that the response hasn't already been + * taken, making this call a no-op. This may be called multiple times. No delimiters are added + * between strings from multiple calls. + */ + synchronized void addOutput(String s) { + if (responseBuilder != null) { + responseBuilder.setOutput(responseBuilder.getOutput() + s); + } + } + } + + /** Requests that are currently being processed. Visible for testing. */ + final ConcurrentMap<Integer, RequestInfo> activeRequests = new ConcurrentHashMap<>(); + + /** The function to be called after each {@link WorkRequest} is read. */ + private final WorkRequestCallback callback; + + /** This worker's stderr. */ + private final PrintStream stderr; + + final WorkerMessageProcessor messageProcessor; + + private final BiConsumer<Integer, Thread> cancelCallback; + /** + * A scheduler that runs garbage collection after a certain amount of CPU time has passed. In our + * experience, explicit GC reclaims much more than implicit GC. This scheduler helps make sure + * very busy workers don't grow ridiculously large. + */ + private final CpuTimeBasedGcScheduler gcScheduler; + /** + * A scheduler that runs garbage collection after a certain amount of time without any activity. + * In our experience, explicit GC reclaims much more than implicit GC. This scheduler helps make + * sure workers don't hang on to excessive memory after they are done working. + */ + private final IdleGcScheduler idleGcScheduler; + + /** + * If set, this worker will stop handling requests and shut itself down. This can happen if + * something throws an {@link Error}. + */ + private final AtomicBoolean shutdownWorker = new AtomicBoolean(false); + + /** + * Creates a {@code WorkRequestHandler} that will call {@code callback} for each WorkRequest + * received. + * + * @param callback Callback method for executing a single WorkRequest in a thread. The first + * argument to {@code callback} is the set of command-line arguments, the second is where all + * error messages and other user-oriented messages should be written to. The callback must + * return an exit code indicating success (zero) or failure (nonzero). + * @param stderr Stream that log messages should be written to, typically the process' stderr. + * @param messageProcessor Object responsible for parsing {@code WorkRequest}s from the server and + * writing {@code WorkResponses} to the server. + */ + @Deprecated + public WorkRequestHandler( + BiFunction<List<String>, PrintWriter, Integer> callback, + PrintStream stderr, + WorkerMessageProcessor messageProcessor) { + this(callback, stderr, messageProcessor, Duration.ZERO, null); + } + + /** + * Creates a {@code WorkRequestHandler} that will call {@code callback} for each WorkRequest + * received. + * + * @param callback Callback method for executing a single WorkRequest in a thread. The first + * argument to {@code callback} is the set of command-line arguments, the second is where all + * error messages and other user-oriented messages should be written to. The callback must + * return an exit code indicating success (zero) or failure (nonzero). + * @param stderr Stream that log messages should be written to, typically the process' stderr. + * @param messageProcessor Object responsible for parsing {@code WorkRequest}s from the server and + * writing {@code WorkResponses} to the server. + * @param cpuUsageBeforeGc The minimum amount of CPU time between explicit garbage collection + * calls. Pass Duration.ZERO to not do explicit garbage collection. + * @deprecated Use WorkRequestHandlerBuilder instead. + */ + @Deprecated() + public WorkRequestHandler( + BiFunction<List<String>, PrintWriter, Integer> callback, + PrintStream stderr, + WorkerMessageProcessor messageProcessor, + Duration cpuUsageBeforeGc) { + this(callback, stderr, messageProcessor, cpuUsageBeforeGc, null); + } + + /** + * Creates a {@code WorkRequestHandler} that will call {@code callback} for each WorkRequest + * received. Only used for the Builder. + * + * @deprecated Use WorkRequestHandlerBuilder instead. + */ + @Deprecated + private WorkRequestHandler( + BiFunction<List<String>, PrintWriter, Integer> callback, + PrintStream stderr, + WorkerMessageProcessor messageProcessor, + Duration cpuUsageBeforeGc, + BiConsumer<Integer, Thread> cancelCallback) { + this( + new WorkRequestCallback((request, pw) -> callback.apply(request.getArgumentsList(), pw)), + stderr, + messageProcessor, + cpuUsageBeforeGc, + cancelCallback, + Duration.ZERO); + } + + /** + * Creates a {@code WorkRequestHandler} that will call {@code callback} for each WorkRequest + * received. Only used for the Builder. + * + * @param callback WorkRequestCallback object with Callback method for executing a single + * WorkRequest in a thread. The first argument to {@code callback} is the WorkRequest, the + * second is where all error messages and other user-oriented messages should be written to. + * The callback must return an exit code indicating success (zero) or failure (nonzero). + */ + private WorkRequestHandler( + WorkRequestCallback callback, + PrintStream stderr, + WorkerMessageProcessor messageProcessor, + Duration cpuUsageBeforeGc, + BiConsumer<Integer, Thread> cancelCallback, + Duration idleTimeBeforeGc) { + this.callback = callback; + this.stderr = stderr; + this.messageProcessor = messageProcessor; + this.gcScheduler = new CpuTimeBasedGcScheduler(cpuUsageBeforeGc); + this.cancelCallback = cancelCallback; + this.idleGcScheduler = new IdleGcScheduler(idleTimeBeforeGc); + } + + /** A wrapper class for the callback BiFunction */ + public static class WorkRequestCallback { + + /** + * Callback method for executing a single WorkRequest in a thread. The first argument to {@code + * callback} is the WorkRequest, the second is where all error messages and other user-oriented + * messages should be written to. The callback must return an exit code indicating success + * (zero) or failure (nonzero). + */ + private final BiFunction<WorkRequest, PrintWriter, Integer> callback; + + public WorkRequestCallback(BiFunction<WorkRequest, PrintWriter, Integer> callback) { + this.callback = callback; + } + + public Integer apply(WorkRequest workRequest, PrintWriter printWriter) + throws InterruptedException { + Integer result = callback.apply(workRequest, printWriter); + if (Thread.interrupted()) { + throw new InterruptedException("Work request interrupted: " + workRequest.getRequestId()); + } + return result; + } + } + + /** Builder class for WorkRequestHandler. Required parameters are passed to the constructor. */ + public static class WorkRequestHandlerBuilder { + private final WorkRequestCallback callback; + private final PrintStream stderr; + private final WorkerMessageProcessor messageProcessor; + private Duration cpuUsageBeforeGc = Duration.ZERO; + private BiConsumer<Integer, Thread> cancelCallback; + private Duration idleTimeBeforeGc = Duration.ZERO; + + /** + * Creates a {@code WorkRequestHandlerBuilder}. + * + * @param callback Callback method for executing a single WorkRequest in a thread. The first + * argument to {@code callback} is the set of command-line arguments, the second is where + * all error messages and other user-oriented messages should be written to. The callback + * must return an exit code indicating success (zero) or failure (nonzero). + * @param stderr Stream that log messages should be written to, typically the process' stderr. + * @param messageProcessor Object responsible for parsing {@code WorkRequest}s from the server + * and writing {@code WorkResponses} to the server. + * @deprecated use WorkRequestHandlerBuilder with WorkRequestCallback instead + */ + @Deprecated + public WorkRequestHandlerBuilder( + BiFunction<List<String>, PrintWriter, Integer> callback, + PrintStream stderr, + WorkerMessageProcessor messageProcessor) { + this( + new WorkRequestCallback((request, pw) -> callback.apply(request.getArgumentsList(), pw)), + stderr, + messageProcessor); + } + + /** + * Creates a {@code WorkRequestHandlerBuilder}. + * + * @param callback WorkRequestCallback object with Callback method for executing a single + * WorkRequest in a thread. The first argument to {@code callback} is the WorkRequest, the + * second is where all error messages and other user-oriented messages should be written to. + * The callback must return an exit code indicating success (zero) or failure (nonzero). + * @param stderr Stream that log messages should be written to, typically the process' stderr. + * @param messageProcessor Object responsible for parsing {@code WorkRequest}s from the server + * and writing {@code WorkResponses} to the server. + */ + public WorkRequestHandlerBuilder( + WorkRequestCallback callback, PrintStream stderr, WorkerMessageProcessor messageProcessor) { + this.callback = callback; + this.stderr = stderr; + this.messageProcessor = messageProcessor; + } + + /** + * Sets the minimum amount of CPU time between explicit garbage collection calls. Pass + * Duration.ZERO to not do explicit garbage collection (the default). + */ + @CanIgnoreReturnValue + public WorkRequestHandlerBuilder setCpuUsageBeforeGc(Duration cpuUsageBeforeGc) { + this.cpuUsageBeforeGc = cpuUsageBeforeGc; + return this; + } + + /** + * Sets a callback will be called when a cancellation message has been received. The callback + * will be call with the request ID and the thread executing the request. + */ + @CanIgnoreReturnValue + public WorkRequestHandlerBuilder setCancelCallback(BiConsumer<Integer, Thread> cancelCallback) { + this.cancelCallback = cancelCallback; + return this; + } + + /** Sets the time without any work that should elapse before forcing a GC. */ + @CanIgnoreReturnValue + public WorkRequestHandlerBuilder setIdleTimeBeforeGc(Duration idleTimeBeforeGc) { + this.idleTimeBeforeGc = idleTimeBeforeGc; + return this; + } + + /** Returns a WorkRequestHandler instance with the values in this Builder. */ + public WorkRequestHandler build() { + return new WorkRequestHandler( + callback, stderr, messageProcessor, cpuUsageBeforeGc, cancelCallback, idleTimeBeforeGc); + } + } + + /** + * Runs an infinite loop of reading {@link WorkRequest} from {@code in}, running the callback, + * then writing the corresponding {@link WorkResponse} to {@code out}. If there is an error + * reading or writing the requests or responses, it writes an error message on {@code err} and + * returns. If {@code in} reaches EOF, it also returns. + * + * <p>This function also wraps the system streams in a {@link WorkerIO} instance that prevents the + * underlying tool from writing to {@link System#out} or reading from {@link System#in}, which + * would corrupt the worker worker protocol. When the while loop exits, the original system + * streams will be swapped back into {@link System}. + */ + public void processRequests() throws IOException { + // Wrap the system streams into a WorkerIO instance to prevent unexpected reads and writes on + // stdin/stdout. + WorkerIO workerIO = WorkerIO.capture(); + + try { + while (!shutdownWorker.get()) { + WorkRequest request = messageProcessor.readWorkRequest(); + idleGcScheduler.markActivity(true); + if (request == null) { + break; + } + if (request.getCancel()) { + respondToCancelRequest(request); + } else { + startResponseThread(workerIO, request); + } + } + } catch (IOException e) { + stderr.println("Error reading next WorkRequest: " + e); + e.printStackTrace(stderr); + } finally { + idleGcScheduler.stop(); + // TODO(b/220878242): Give the outstanding requests a chance to send a "shutdown" response, + // but also try to kill stuck threads. For now, we just interrupt the remaining threads. + // We considered doing System.exit here, but that is hard to test and would deny the callers + // of this method a chance to clean up. Instead, we initiate the cleanup of our resources here + // and the caller can decide whether to wait for an orderly shutdown or now. + for (RequestInfo ri : activeRequests.values()) { + if (ri.thread.isAlive()) { + try { + ri.thread.interrupt(); + } catch (RuntimeException e) { + // If we can't interrupt, we can't do much else. + } + } + } + + try { + // Unwrap the system streams placing the original streams back + workerIO.close(); + } catch (Exception e) { + stderr.println(e.getMessage()); + } + } + } + + /** Starts a thread for the given request. */ + void startResponseThread(WorkerIO workerIO, WorkRequest request) { + Thread currentThread = Thread.currentThread(); + String threadName = + request.getRequestId() > 0 + ? "multiplex-request-" + request.getRequestId() + : "singleplex-request"; + // TODO(larsrc): See if this can be handled with a queue instead, without introducing more + // race conditions. + if (request.getRequestId() == 0) { + while (activeRequests.containsKey(request.getRequestId())) { + // b/194051480: Previous singleplex requests can still be in activeRequests for a bit after + // the response has been sent. We need to wait for them to vanish. + try { + Thread.sleep(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + } + Thread t = + new Thread( + () -> { + RequestInfo requestInfo = activeRequests.get(request.getRequestId()); + if (requestInfo == null) { + // Already cancelled + idleGcScheduler.markActivity(!activeRequests.isEmpty()); + return; + } + try { + respondToRequest(workerIO, request, requestInfo); + } catch (IOException e) { + // IOExceptions here means a problem talking to the server, so we must shut down. + if (!shutdownWorker.compareAndSet(false, true)) { + stderr.println("Error communicating with server, shutting down worker."); + e.printStackTrace(stderr); + currentThread.interrupt(); + } + } finally { + activeRequests.remove(request.getRequestId()); + idleGcScheduler.markActivity(!activeRequests.isEmpty()); + } + }, + threadName); + t.setUncaughtExceptionHandler( + (t1, e) -> { + // Shut down the worker in case of severe issues. We don't handle RuntimeException here, + // as those are not serious enough to merit shutting down the worker. + if (e instanceof Error && shutdownWorker.compareAndSet(false, true)) { + stderr.println("Error thrown by worker thread, shutting down worker."); + e.printStackTrace(stderr); + currentThread.interrupt(); + idleGcScheduler.stop(); + System.exit(1); + } + }); + RequestInfo previous = activeRequests.putIfAbsent(request.getRequestId(), new RequestInfo(t)); + if (previous != null) { + // Kill worker since this shouldn't happen: server didn't follow the worker protocol + throw new IllegalStateException("Request still active: " + request.getRequestId()); + } + t.start(); + } + + /** + * Handles and responds to the given {@link WorkRequest}. + * + * @throws IOException if there is an error talking to the server. Errors from calling the {@link + * #callback} are reported with exit code 1. + */ + @VisibleForTesting + void respondToRequest(WorkerIO workerIO, WorkRequest request, RequestInfo requestInfo) + throws IOException { + int exitCode; + StringWriter sw = new StringWriter(); + try (PrintWriter pw = new PrintWriter(sw)) { + try { + exitCode = callback.apply(request, pw); + } catch (InterruptedException e) { + exitCode = 1; + } catch (RuntimeException e) { + e.printStackTrace(pw); + exitCode = 1; + } + + try { + // Read out the captured string for the final WorkResponse output + String captured = workerIO.readCapturedAsUtf8String().trim(); + if (!captured.isEmpty()) { + pw.write(captured); + } + } catch (IOException e) { + stderr.println(e.getMessage()); + } + } + Optional<WorkResponse.Builder> optBuilder = requestInfo.takeBuilder(); + if (optBuilder.isPresent()) { + WorkResponse.Builder builder = optBuilder.get(); + builder.setRequestId(request.getRequestId()); + if (requestInfo.isCancelled()) { + builder.setWasCancelled(true); + } else { + builder.setOutput(builder.getOutput() + sw).setExitCode(exitCode); + } + WorkResponse response = builder.build(); + synchronized (this) { + messageProcessor.writeWorkResponse(response); + } + } + gcScheduler.maybePerformGc(); + } + + /** + * Marks the given request as cancelled and uses {@link #cancelCallback} to request cancellation. + * + * <p>For simplicity, and to avoid blocking in {@link #cancelCallback}, response to cancellation + * is still handled by {@link #respondToRequest} once the canceled request aborts (or finishes). + */ + void respondToCancelRequest(WorkRequest request) { + // Theoretically, we could have gotten two singleplex requests, and we can't tell those apart. + // However, that's a violation of the protocol, so we don't try to handle it (not least because + // handling it would be quite error-prone). + RequestInfo ri = activeRequests.get(request.getRequestId()); + + if (ri == null) { + return; + } + if (cancelCallback == null) { + ri.setCancelled(); + // This is either an error on the server side or a version mismatch between the server setup + // and the binary. It's better to wait for the regular work to finish instead of breaking the + // build, but we should inform the user about the bad setup. + ri.addOutput( + String.format( + "Cancellation request received for worker request %d, but this worker does not" + + " support cancellation.\n", + request.getRequestId())); + } else { + if (ri.thread.isAlive() && !ri.isCancelled()) { + ri.setCancelled(); + Thread t = + new Thread( + // Response will be sent from request thread once request handler returns. + // We can ignore any exceptions in cancel callback since it's best effort. + () -> cancelCallback.accept(request.getRequestId(), ri.thread)); + t.start(); + } + } + } + + @Override + public void close() throws IOException { + messageProcessor.close(); + } + + /** Schedules GC when the worker has been idle for a while */ + private static class IdleGcScheduler { + private Instant lastActivity = Instant.EPOCH; + private Instant lastGc = Instant.EPOCH; + /** Minimum duration from the end of activity until we perform an idle GC. */ + private final Duration idleTimeBeforeGc; + + private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1); + private ScheduledFuture<?> futureGc = null; + + /** + * Creates a new scheduler. + * + * @param idleTimeBeforeGc The time from the last activity until attempting GC. + */ + public IdleGcScheduler(Duration idleTimeBeforeGc) { + this.idleTimeBeforeGc = idleTimeBeforeGc; + } + + synchronized void start() { + if (!idleTimeBeforeGc.isZero()) { + futureGc = + executor.schedule(this::maybeDoGc, idleTimeBeforeGc.toMillis(), TimeUnit.MILLISECONDS); + } + } + + /** + * Should be called whenever there is some sort of activity starting or ending. Better to call + * too often. + */ + synchronized void markActivity(boolean anythingActive) { + lastActivity = Instant.now(); + if (futureGc != null) { + futureGc.cancel(false); + futureGc = null; + } + if (!anythingActive) { + start(); + } + } + + private void maybeDoGc() { + if (lastGc.isBefore(lastActivity) + && lastActivity.isBefore(Instant.now().minus(idleTimeBeforeGc))) { + System.gc(); + lastGc = Instant.now(); + } else { + start(); + } + } + + synchronized void stop() { + if (futureGc != null) { + futureGc.cancel(false); + futureGc = null; + } + executor.shutdown(); + } + } + + /** + * Class that performs GC occasionally, based on how much CPU time has passed. This strikes a + * compromise between blindly doing GC after e.g. every request, which takes too much CPU, and not + * doing explicit GC at all, which causes poor garbage collection in some cases. + */ + private static class CpuTimeBasedGcScheduler { + /** + * After this much CPU time has elapsed, we may force a GC run. Set to {@link Duration#ZERO} to + * disable. + */ + private final Duration cpuUsageBeforeGc; + + /** The total process CPU time at the last GC run (or from the start of the worker). */ + private final AtomicReference<Duration> cpuTimeAtLastGc; + + /** Used to get the CPU time used by this process. */ + private static final OperatingSystemMXBean bean = + (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean(); + + /** + * Creates a new {@link CpuTimeBasedGcScheduler} that may perform GC after {@code + * cpuUsageBeforeGc} amount of CPU time has been used. + */ + public CpuTimeBasedGcScheduler(Duration cpuUsageBeforeGc) { + this.cpuUsageBeforeGc = cpuUsageBeforeGc; + this.cpuTimeAtLastGc = new AtomicReference<>(getCpuTime()); + } + + private Duration getCpuTime() { + return !cpuUsageBeforeGc.isZero() + ? Duration.ofNanos(bean.getProcessCpuTime()) + : Duration.ZERO; + } + + /** Call occasionally to perform a GC if enough CPU time has been used. */ + private void maybePerformGc() { + if (!cpuUsageBeforeGc.isZero()) { + Duration currentCpuTime = getCpuTime(); + Duration lastCpuTime = cpuTimeAtLastGc.get(); + // Do GC when enough CPU time has been used, but only if nobody else beat us to it. + if (currentCpuTime.minus(lastCpuTime).compareTo(cpuUsageBeforeGc) > 0 + && cpuTimeAtLastGc.compareAndSet(lastCpuTime, currentCpuTime)) { + System.gc(); + // Avoid counting GC CPU time against CPU time before next GC. + cpuTimeAtLastGc.compareAndSet(currentCpuTime, getCpuTime()); + } + } + } + } + + /** + * Class that wraps the standard {@link System#in}, {@link System#out}, and {@link System#err} + * with our own ByteArrayOutputStream that allows {@link WorkRequestHandler} to safely capture + * outputs that can't be directly captured by the PrintStream associated with the work request. + * + * <p>This is most useful when integrating JVM tools that write exceptions and logs directly to + * {@link System#out} and {@link System#err}, which would corrupt the persistent worker protocol. + * We also redirect {@link System#in}, just in case a tool should attempt to read it. + * + * <p>WorkerIO implements {@link AutoCloseable} and will swap the original streams back into + * {@link System} once close has been called. + */ + public static class WorkerIO implements AutoCloseable { + private final InputStream originalInputStream; + private final PrintStream originalOutputStream; + private final PrintStream originalErrorStream; + private final ByteArrayOutputStream capturedStream; + private final AutoCloseable restore; + + /** + * Creates a new {@link WorkerIO} that allows {@link WorkRequestHandler} to capture standard + * output and error streams that can't be directly captured by the PrintStream associated with + * the work request. + */ + @VisibleForTesting + WorkerIO( + InputStream originalInputStream, + PrintStream originalOutputStream, + PrintStream originalErrorStream, + ByteArrayOutputStream capturedStream, + AutoCloseable restore) { + this.originalInputStream = originalInputStream; + this.originalOutputStream = originalOutputStream; + this.originalErrorStream = originalErrorStream; + this.capturedStream = capturedStream; + this.restore = restore; + } + + /** Wraps the standard System streams and WorkerIO instance */ + public static WorkerIO capture() { + // Save the original streams + InputStream originalInputStream = System.in; + PrintStream originalOutputStream = System.out; + PrintStream originalErrorStream = System.err; + + // Replace the original streams with our own instances + ByteArrayOutputStream capturedStream = new ByteArrayOutputStream(); + PrintStream outputBuffer = new PrintStream(capturedStream, true); + ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(new byte[0]); + System.setIn(byteArrayInputStream); + System.setOut(outputBuffer); + System.setErr(outputBuffer); + + return new WorkerIO( + originalInputStream, + originalOutputStream, + originalErrorStream, + capturedStream, + () -> { + System.setIn(originalInputStream); + System.setOut(originalOutputStream); + System.setErr(originalErrorStream); + outputBuffer.close(); + byteArrayInputStream.close(); + }); + } + + /** Returns the original input stream most commonly provided by {@link System#in} */ + @VisibleForTesting + InputStream getOriginalInputStream() { + return originalInputStream; + } + + /** Returns the original output stream most commonly provided by {@link System#out} */ + @VisibleForTesting + PrintStream getOriginalOutputStream() { + return originalOutputStream; + } + + /** Returns the original error stream most commonly provided by {@link System#err} */ + @VisibleForTesting + PrintStream getOriginalErrorStream() { + return originalErrorStream; + } + + /** Returns the captured outputs as a UTF-8 string */ + @VisibleForTesting + String readCapturedAsUtf8String() throws IOException { + capturedStream.flush(); + String captureOutput = capturedStream.toString(StandardCharsets.UTF_8); + capturedStream.reset(); + return captureOutput; + } + + @Override + public void close() throws Exception { + restore.close(); + } + } +}
diff --git a/java/src/test/java/com/google/devtools/build/lib/worker/BUILD.bazel b/java/src/test/java/com/google/devtools/build/lib/worker/BUILD.bazel new file mode 100644 index 0000000..41d81f9 --- /dev/null +++ b/java/src/test/java/com/google/devtools/build/lib/worker/BUILD.bazel
@@ -0,0 +1,12 @@ +java_test( + name = "WorkRequestHandlerTest", + srcs = ["WorkRequestHandlerTest.java"], + deps = [ + "//:worker_protocol_java_proto", + "//src/main/java/com/google/devtools/build/lib/worker:work_request_handlers", + "@maven//:junit_junit", + "@maven//:org_mockito_mockito_core", + "@maven//:com_google_truth_truth", + ], +) +
diff --git a/java/src/test/java/com/google/devtools/build/lib/worker/WorkRequestHandlerTest.java b/java/src/test/java/com/google/devtools/build/lib/worker/WorkRequestHandlerTest.java new file mode 100644 index 0000000..9c1ea29 --- /dev/null +++ b/java/src/test/java/com/google/devtools/build/lib/worker/WorkRequestHandlerTest.java
@@ -0,0 +1,675 @@ +// Copyright 2020 The Bazel Authors. All rights reserved. +// +// 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 +// +// http://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.google.devtools.build.lib.worker; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.devtools.build.lib.worker.WorkRequestHandler.RequestInfo; +import com.google.devtools.build.lib.worker.WorkRequestHandler.WorkRequestCallback; +import com.google.devtools.build.lib.worker.WorkRequestHandler.WorkRequestHandlerBuilder; +import com.google.devtools.build.lib.worker.WorkRequestHandler.WorkerMessageProcessor; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkRequest; +import com.google.devtools.build.lib.worker.WorkerProtocol.WorkResponse; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.io.PrintStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.MockitoAnnotations; + +/** Tests for the WorkRequestHandler */ +@RunWith(JUnit4.class) +public class WorkRequestHandlerTest { + + private final WorkRequestHandler.WorkerIO testWorkerIO = createTestWorkerIO(); + + @Before + public void init() { + MockitoAnnotations.initMocks(this); + } + + @After + public void after() throws Exception { + testWorkerIO.close(); + } + + @Test + public void testNormalWorkRequest() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + WorkRequestHandler handler = + new WorkRequestHandler( + (args, err) -> 1, + new PrintStream(new ByteArrayOutputStream()), + new ProtoWorkerMessageProcessor(new ByteArrayInputStream(new byte[0]), out)); + + List<String> args = Arrays.asList("--sources", "A.java"); + WorkRequest request = WorkRequest.newBuilder().addAllArguments(args).build(); + handler.respondToRequest(testWorkerIO, request, new RequestInfo(null)); + + WorkResponse response = + WorkResponse.parseDelimitedFrom(new ByteArrayInputStream(out.toByteArray())); + assertThat(response.getRequestId()).isEqualTo(0); + assertThat(response.getExitCode()).isEqualTo(1); + assertThat(response.getOutput()).isEmpty(); + } + + @Test + public void testMultiplexWorkRequest() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + WorkRequestHandler handler = + new WorkRequestHandler( + (args, err) -> 0, + new PrintStream(new ByteArrayOutputStream()), + new ProtoWorkerMessageProcessor(new ByteArrayInputStream(new byte[0]), out)); + + List<String> args = Arrays.asList("--sources", "A.java"); + WorkRequest request = WorkRequest.newBuilder().addAllArguments(args).setRequestId(42).build(); + handler.respondToRequest(testWorkerIO, request, new RequestInfo(null)); + + WorkResponse response = + WorkResponse.parseDelimitedFrom(new ByteArrayInputStream(out.toByteArray())); + assertThat(response.getRequestId()).isEqualTo(42); + assertThat(response.getExitCode()).isEqualTo(0); + assertThat(response.getOutput()).isEmpty(); + } + + @Test + public void testMultiplexWorkRequest_stopsThreadsOnShutdown() + throws IOException, InterruptedException { + PipedOutputStream src = new PipedOutputStream(); + PipedInputStream dest = new PipedInputStream(); + + // Work request threads release this when they have started. + Semaphore started = new Semaphore(0); + // Work request threads wait forever on this, so we can see how they react to closed stdin. + Semaphore eternity = new Semaphore(0); + // Released when the work request handler thread has noticed the closed stdin and interrupted + // the work request threads. + Semaphore stopped = new Semaphore(0); + List<Thread> workerThreads = new ArrayList<>(); + StoppableWorkerMessageProcessor messageProcessor = + new StoppableWorkerMessageProcessor( + new ProtoWorkerMessageProcessor( + new PipedInputStream(src), new PipedOutputStream(dest))); + WorkRequestHandler handler = + new WorkRequestHandler( + (args, err) -> { + // Each call to this runs in its own thread. + try { + synchronized (workerThreads) { + workerThreads.add(Thread.currentThread()); + } + started.release(); + eternity.acquire(); // This blocks forever. + } catch (InterruptedException e) { + throw new AssertionError("Unhandled exception", e); + } + return 0; + }, + new PrintStream(new ByteArrayOutputStream()), + messageProcessor); + + List<String> args = Arrays.asList("--sources", "A.java"); + Thread t = + new Thread( + () -> { + try { + handler.processRequests(); + stopped.release(); + } catch (IOException e) { + throw new AssertionError("Unhandled exception", e); + } + }); + t.start(); + WorkRequest request1 = WorkRequest.newBuilder().addAllArguments(args).setRequestId(42).build(); + request1.writeDelimitedTo(src); + WorkRequest request2 = WorkRequest.newBuilder().addAllArguments(args).setRequestId(43).build(); + request2.writeDelimitedTo(src); + src.flush(); + + started.acquire(2); + assertThat(workerThreads).hasSize(2); + // Now both request threads are started, closing the input to the "worker" should shut it down. + src.close(); + stopped.acquire(); + while (workerThreads.get(0).isAlive() || workerThreads.get(1).isAlive()) { + Thread.sleep(1); + } + assertThat(workerThreads.get(0).isAlive()).isFalse(); + assertThat(workerThreads.get(1).isAlive()).isFalse(); + } + + @Test + public void testMultiplexWorkRequest_stopsWorkerOnException() + throws IOException, InterruptedException { + PipedOutputStream src = new PipedOutputStream(); + PipedInputStream dest = new PipedInputStream(); + + // Work request threads release this when they have started. + Semaphore started = new Semaphore(0); + // One work request threads waits forever on this, so the second one can throw an exception + Semaphore eternity = new Semaphore(0); + // Released when the work request handler thread has been stopped after a worker thread died. + Semaphore stopped = new Semaphore(0); + List<Thread> workerThreads = new ArrayList<>(); + StoppableWorkerMessageProcessor messageProcessor = + new StoppableWorkerMessageProcessor( + new ProtoWorkerMessageProcessor( + new PipedInputStream(src), new PipedOutputStream(dest))); + WorkRequestHandler handler = + new WorkRequestHandler( + (args, err) -> { + // Each call to this runs in its own thread. + try { + synchronized (workerThreads) { + workerThreads.add(Thread.currentThread()); + } + started.release(); + if (workerThreads.size() < 2) { + eternity.acquire(); // This blocks forever. + } else { + throw new Error("Intentional death!"); + } + } catch (InterruptedException e) { + throw new AssertionError("Unhandled exception", e); + } + return 0; + }, + new PrintStream(new ByteArrayOutputStream()), + messageProcessor); + + List<String> args = Arrays.asList("--sources", "A.java"); + Thread t = + new Thread( + () -> { + try { + handler.processRequests(); + stopped.release(); + } catch (IOException e) { + throw new AssertionError("Unhandled exception", e); + } + }); + t.start(); + WorkRequest request1 = WorkRequest.newBuilder().addAllArguments(args).setRequestId(42).build(); + request1.writeDelimitedTo(src); + WorkRequest request2 = WorkRequest.newBuilder().addAllArguments(args).setRequestId(43).build(); + request2.writeDelimitedTo(src); + src.flush(); + + started.acquire(2); + assertThat(workerThreads).hasSize(2); + stopped.acquire(); + while (workerThreads.get(0).isAlive() || workerThreads.get(1).isAlive()) { + Thread.sleep(1); + } + assertThat(workerThreads.get(0).isAlive()).isFalse(); + assertThat(workerThreads.get(1).isAlive()).isFalse(); + } + + @Test + public void testOutput() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + WorkRequestHandler handler = + new WorkRequestHandler( + (args, err) -> { + err.println("Failed!"); + return 1; + }, + new PrintStream(new ByteArrayOutputStream()), + new ProtoWorkerMessageProcessor(new ByteArrayInputStream(new byte[0]), out)); + + List<String> args = Arrays.asList("--sources", "A.java"); + WorkRequest request = WorkRequest.newBuilder().addAllArguments(args).build(); + handler.respondToRequest(testWorkerIO, request, new RequestInfo(null)); + + WorkResponse response = + WorkResponse.parseDelimitedFrom(new ByteArrayInputStream(out.toByteArray())); + assertThat(response.getRequestId()).isEqualTo(0); + assertThat(response.getExitCode()).isEqualTo(1); + assertThat(response.getOutput()).contains("Failed!"); + } + + @Test + public void testException() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + WorkRequestHandler handler = + new WorkRequestHandler( + (args, err) -> { + throw new RuntimeException("Exploded!"); + }, + new PrintStream(new ByteArrayOutputStream()), + new ProtoWorkerMessageProcessor(new ByteArrayInputStream(new byte[0]), out)); + + List<String> args = Arrays.asList("--sources", "A.java"); + WorkRequest request = WorkRequest.newBuilder().addAllArguments(args).build(); + handler.respondToRequest(testWorkerIO, request, new RequestInfo(null)); + + WorkResponse response = + WorkResponse.parseDelimitedFrom(new ByteArrayInputStream(out.toByteArray())); + assertThat(response.getRequestId()).isEqualTo(0); + assertThat(response.getExitCode()).isEqualTo(1); + assertThat(response.getOutput()).startsWith("java.lang.RuntimeException: Exploded!"); + } + + @Test + public void testCancelRequest_exactlyOneResponseSent() throws IOException, InterruptedException { + boolean[] handlerCalled = new boolean[] {false}; + boolean[] cancelCalled = new boolean[] {false}; + PipedOutputStream src = new PipedOutputStream(); + PipedInputStream dest = new PipedInputStream(); + Semaphore done = new Semaphore(0); + Semaphore finish = new Semaphore(0); + List<String> failures = new ArrayList<>(); + + StoppableWorkerMessageProcessor messageProcessor = + new StoppableWorkerMessageProcessor( + new ProtoWorkerMessageProcessor( + new PipedInputStream(src), new PipedOutputStream(dest))); + WorkRequestHandler handler = + new WorkRequestHandlerBuilder( + (args, err) -> { + handlerCalled[0] = true; + err.println("Such work! Much progress! Wow!"); + return 1; + }, + new PrintStream(new ByteArrayOutputStream()), + messageProcessor) + .setCancelCallback( + (i, t) -> { + cancelCalled[0] = true; + }) + .build(); + + runRequestHandlerThread(done, handler, finish, failures); + WorkRequest.newBuilder().setRequestId(42).build().writeDelimitedTo(src); + WorkRequest.newBuilder().setRequestId(42).setCancel(true).build().writeDelimitedTo(src); + WorkResponse response = WorkResponse.parseDelimitedFrom(dest); + messageProcessor.stop(); + done.acquire(); + + assertThat(handlerCalled[0] || cancelCalled[0]).isTrue(); + assertThat(response.getRequestId()).isEqualTo(42); + if (response.getWasCancelled()) { + assertThat(response.getOutput()).isEmpty(); + assertThat(response.getExitCode()).isEqualTo(0); + } else { + assertThat(response.getOutput()).isEqualTo("Such work! Much progress! Wow!\n"); + assertThat(response.getExitCode()).isEqualTo(1); + } + + // Checks that nothing more was sent. + assertThat(dest.available()).isEqualTo(0); + finish.release(); + + // Checks that there weren't other unexpected failures. + assertThat(failures).isEmpty(); + } + + @Test + public void testCancelRequest_sendsResponseWhenDone() throws IOException, InterruptedException { + Semaphore waitForCancel = new Semaphore(0); + Semaphore handlerCalled = new Semaphore(0); + Semaphore cancelCalled = new Semaphore(0); + PipedOutputStream src = new PipedOutputStream(); + PipedInputStream dest = new PipedInputStream(); + Semaphore done = new Semaphore(0); + Semaphore requestDone = new Semaphore(0); + Semaphore finish = new Semaphore(0); + List<String> failures = new ArrayList<>(); + + StoppableWorkerMessageProcessor messageProcessor = + new StoppableWorkerMessageProcessor( + new ProtoWorkerMessageProcessor( + new PipedInputStream(src), new PipedOutputStream(dest))); + // We force the regular handling to not finish until after we have read the cancel response, + // to avoid flakiness. + WorkRequestHandler handler = + new WorkRequestHandlerBuilder( + (args, err) -> { + // This handler waits until the main thread has sent a cancel request. + handlerCalled.release(2); + try { + waitForCancel.acquire(); + } catch (InterruptedException e) { + failures.add("Unexpected interrupt waiting for cancel request"); + e.printStackTrace(); + } + requestDone.release(); + return 0; + }, + new PrintStream(new ByteArrayOutputStream()), + messageProcessor) + .setCancelCallback( + (i, t) -> { + cancelCalled.release(); + }) + .build(); + + runRequestHandlerThread(done, handler, finish, failures); + WorkRequest.newBuilder().setRequestId(42).build().writeDelimitedTo(src); + // Make sure the handler is called before sending the cancel request, or we might process + // the cancellation entirely before that. + handlerCalled.acquire(); + WorkRequest.newBuilder().setRequestId(42).setCancel(true).build().writeDelimitedTo(src); + cancelCalled.acquire(); + waitForCancel.release(); + // Give the other request a chance to process, so we can check that no other response is sent + requestDone.acquire(); + messageProcessor.stop(); + done.acquire(); + + WorkResponse response = WorkResponse.parseDelimitedFrom(dest); + assertThat(handlerCalled.availablePermits()).isEqualTo(1); // Released 2, one was acquired + assertThat(cancelCalled.availablePermits()).isEqualTo(0); + assertThat(response.getRequestId()).isEqualTo(42); + assertThat(response.getOutput()).isEmpty(); + assertThat(response.getWasCancelled()).isTrue(); + + // Checks that nothing more was sent. + assertThat(dest.available()).isEqualTo(0); + src.close(); + finish.release(); + + // Checks that there weren't other unexpected failures. + assertThat(failures).isEmpty(); + } + + @Test + public void testCancelRequest_noDoubleCancelResponse() throws IOException, InterruptedException { + Semaphore waitForCancel = new Semaphore(0); + Semaphore cancelCalled = new Semaphore(0); + PipedOutputStream src = new PipedOutputStream(); + PipedInputStream dest = new PipedInputStream(); + Semaphore done = new Semaphore(0); + Semaphore requestsDone = new Semaphore(0); + Semaphore finish = new Semaphore(0); + List<String> failures = new ArrayList<>(); + + // We force the regular handling to not finish until after we have read the cancel response, + // to avoid flakiness. + StoppableWorkerMessageProcessor messageProcessor = + new StoppableWorkerMessageProcessor( + new ProtoWorkerMessageProcessor( + new PipedInputStream(src), new PipedOutputStream(dest))); + WorkRequestHandler handler = + new WorkRequestHandlerBuilder( + (args, err) -> { + try { + waitForCancel.acquire(); + } catch (InterruptedException e) { + failures.add("Unexpected interrupt waiting for cancel request"); + e.printStackTrace(); + } + requestsDone.release(); + return 0; + }, + new PrintStream(new ByteArrayOutputStream()), + messageProcessor) + .setCancelCallback( + (i, t) -> { + cancelCalled.release(); + }) + .build(); + + runRequestHandlerThread(done, handler, finish, failures); + WorkRequest.newBuilder().setRequestId(42).build().writeDelimitedTo(src); + WorkRequest.newBuilder().setRequestId(42).setCancel(true).build().writeDelimitedTo(src); + WorkRequest.newBuilder().setRequestId(42).setCancel(true).build().writeDelimitedTo(src); + cancelCalled.acquire(); + waitForCancel.release(); + requestsDone.acquire(); + messageProcessor.stop(); + done.acquire(); + + WorkResponse response = WorkResponse.parseDelimitedFrom(dest); + assertThat(cancelCalled.availablePermits()).isLessThan(2); + assertThat(response.getRequestId()).isEqualTo(42); + assertThat(response.getOutput()).isEmpty(); + assertThat(response.getWasCancelled()).isTrue(); + + // Checks that nothing more was sent. + assertThat(dest.available()).isEqualTo(0); + src.close(); + finish.release(); + + // Checks that there weren't other unexpected failures. + assertThat(failures).isEmpty(); + } + + @Test + public void testCancelRequest_sendsNoResponseWhenAlreadySent() + throws IOException, InterruptedException { + Semaphore handlerCalled = new Semaphore(0); + PipedOutputStream src = new PipedOutputStream(); + PipedInputStream dest = new PipedInputStream(); + Semaphore done = new Semaphore(0); + Semaphore finish = new Semaphore(0); + List<String> failures = new ArrayList<>(); + + // We force the cancel request to not happen until after we have read the normal response, + // to avoid flakiness. + StoppableWorkerMessageProcessor messageProcessor = + new StoppableWorkerMessageProcessor( + new ProtoWorkerMessageProcessor( + new PipedInputStream(src), new PipedOutputStream(dest))); + WorkRequestHandler handler = + new WorkRequestHandlerBuilder( + (args, err) -> { + handlerCalled.release(); + err.println("Such work! Much progress! Wow!"); + return 2; + }, + new PrintStream(new ByteArrayOutputStream()), + messageProcessor) + .setCancelCallback((i, t) -> {}) + .build(); + + runRequestHandlerThread(done, handler, finish, failures); + WorkRequest.newBuilder().setRequestId(42).build().writeDelimitedTo(src); + WorkResponse response = WorkResponse.parseDelimitedFrom(dest); + WorkRequest.newBuilder().setRequestId(42).setCancel(true).build().writeDelimitedTo(src); + messageProcessor.stop(); + done.acquire(); + + assertThat(response).isNotNull(); + + assertThat(handlerCalled.availablePermits()).isEqualTo(1); + assertThat(response.getRequestId()).isEqualTo(42); + assertThat(response.getWasCancelled()).isFalse(); + assertThat(response.getExitCode()).isEqualTo(2); + assertThat(response.getOutput()).isEqualTo("Such work! Much progress! Wow!\n"); + + // Checks that nothing more was sent. + assertThat(dest.available()).isEqualTo(0); + src.close(); + finish.release(); + + // Checks that there weren't other unexpected failures. + assertThat(failures).isEmpty(); + } + + @Test + public void testWorkRequestHandler_withWorkRequestCallback() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + WorkRequestCallback callback = + new WorkRequestCallback((request, err) -> request.getArgumentsCount()); + WorkRequestHandler handler = + new WorkRequestHandlerBuilder( + callback, + new PrintStream(new ByteArrayOutputStream()), + new ProtoWorkerMessageProcessor(new ByteArrayInputStream(new byte[0]), out)) + .build(); + + List<String> args = Arrays.asList("--sources", "B.java"); + WorkRequest request = WorkRequest.newBuilder().addAllArguments(args).build(); + handler.respondToRequest(testWorkerIO, request, new RequestInfo(null)); + + WorkResponse response = + WorkResponse.parseDelimitedFrom(new ByteArrayInputStream(out.toByteArray())); + assertThat(response.getRequestId()).isEqualTo(0); + assertThat(response.getExitCode()).isEqualTo(2); + assertThat(response.getOutput()).isEmpty(); + } + + private void runRequestHandlerThread( + Semaphore done, WorkRequestHandler handler, Semaphore finish, List<String> failures) { + // This thread just makes sure the WorkRequestHandler does work asynchronously. + new Thread( + () -> { + try { + handler.processRequests(); + while (!handler.activeRequests.isEmpty()) { + Thread.sleep(1); + } + } catch (IOException e) { + failures.add("Unexpected I/O error talking to worker thread"); + e.printStackTrace(); + } catch (InterruptedException e) { + // Getting interrupted while waiting for requests to finish is OK. + } + try { + done.release(); + finish.acquire(); + } catch (InterruptedException e) { + // Getting interrupted at the end is OK. + } + }) + .start(); + } + + @Test + public void testWorkerIO_doesWrapSystemStreams() throws Exception { + // Save the original streams + InputStream originalInputStream = System.in; + PrintStream originalOutputStream = System.out; + PrintStream originalErrorStream = System.err; + + // Swap in the test streams to assert against + ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(new byte[0]); + System.setIn(byteArrayInputStream); + PrintStream outputBuffer = new PrintStream(new ByteArrayOutputStream(), true); + System.setOut(outputBuffer); + System.setErr(outputBuffer); + + try (outputBuffer; + byteArrayInputStream; + WorkRequestHandler.WorkerIO io = WorkRequestHandler.WorkerIO.capture()) { + // Assert that the WorkerIO returns the correct wrapped streams and the new System instance + // has been swapped out with the wrapped one + assertThat(io.getOriginalInputStream()).isSameInstanceAs(byteArrayInputStream); + assertThat(System.in).isNotSameInstanceAs(byteArrayInputStream); + + assertThat(io.getOriginalOutputStream()).isSameInstanceAs(outputBuffer); + assertThat(System.out).isNotSameInstanceAs(outputBuffer); + + assertThat(io.getOriginalErrorStream()).isSameInstanceAs(outputBuffer); + assertThat(System.err).isNotSameInstanceAs(outputBuffer); + } finally { + // Swap back in the original streams + System.setIn(originalInputStream); + System.setOut(originalOutputStream); + System.setErr(originalErrorStream); + } + } + + @Test + public void testWorkerIO_doesCaptureStandardOutAndErrorStreams() throws Exception { + try (WorkRequestHandler.WorkerIO io = WorkRequestHandler.WorkerIO.capture()) { + // Assert that nothing has been captured in the new instance + assertThat(io.readCapturedAsUtf8String()).isEmpty(); + + // Assert that the standard out/error stream redirect to our own streams + System.out.print("This is a standard out message!"); + System.err.print("This is a standard error message!"); + assertThat(io.readCapturedAsUtf8String()) + .isEqualTo("This is a standard out message!This is a standard error message!"); + + // Assert that readCapturedAsUtf8String calls reset on the captured stream after a read + assertThat(io.readCapturedAsUtf8String()).isEmpty(); + + System.out.print("out 1"); + System.err.print("err 1"); + System.out.print("out 2"); + System.err.print("err 2"); + assertThat(io.readCapturedAsUtf8String()).isEqualTo("out 1err 1out 2err 2"); + assertThat(io.readCapturedAsUtf8String()).isEmpty(); + } + } + + private WorkRequestHandler.WorkerIO createTestWorkerIO() { + ByteArrayOutputStream captured = new ByteArrayOutputStream(); + return new WorkRequestHandler.WorkerIO(System.in, System.out, System.err, captured, captured); + } + + /** A wrapper around a WorkerMessageProcessor that can be stopped by calling {@code #stop()}. */ + private static class StoppableWorkerMessageProcessor implements WorkerMessageProcessor { + private final WorkerMessageProcessor delegate; + private final AtomicBoolean stop = new AtomicBoolean(false); + private Thread readerThread; + + public StoppableWorkerMessageProcessor(WorkerMessageProcessor delegate) { + this.delegate = delegate; + } + + @Override + public WorkRequest readWorkRequest() throws IOException { + readerThread = Thread.currentThread(); + if (stop.get()) { + return null; + } else { + try { + return delegate.readWorkRequest(); + } catch (InterruptedIOException e) { + // Being interrupted is only an error if we didn't ask for it. + if (!stop.get()) { + throw e; + } else { + return null; + } + } + } + } + + @Override + public void writeWorkResponse(WorkResponse workResponse) throws IOException { + delegate.writeWorkResponse(workResponse); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + + public void stop() { + stop.set(true); + if (readerThread != null) { + readerThread.interrupt(); + } + } + } +}
diff --git a/proto/.bazelversion b/proto/.bazelversion new file mode 120000 index 0000000..b332604 --- /dev/null +++ b/proto/.bazelversion
@@ -0,0 +1 @@ +../.bazelversion \ No newline at end of file
diff --git a/proto/BUILD.bazel b/proto/BUILD.bazel new file mode 100644 index 0000000..d92dfdd --- /dev/null +++ b/proto/BUILD.bazel
@@ -0,0 +1,7 @@ +package(default_visibility = ["//visibility:public"]) + +proto_library( + name = "worker_protocol_proto", + srcs = ["worker_protocol.proto"], +) +
diff --git a/proto/MODULE.bazel b/proto/MODULE.bazel new file mode 100644 index 0000000..d95d7e8 --- /dev/null +++ b/proto/MODULE.bazel
@@ -0,0 +1,4 @@ +module( + name = "bazel_worker_api", + version = "0", +)
diff --git a/proto/MODULE.bazel.lock b/proto/MODULE.bazel.lock new file mode 100644 index 0000000..f5cd531 --- /dev/null +++ b/proto/MODULE.bazel.lock
@@ -0,0 +1,110 @@ +{ + "lockFileVersion": 11, + "registryFileHashes": { + "https://bcr.bazel.build/bazel_registry.json": "8a28e4aff06ee60aed2a8c281907fb8bcbf3b753c91fb5a5c57da3215d5b3497", + "https://bcr.bazel.build/modules/abseil-cpp/20210324.2/MODULE.bazel": "7cd0312e064fde87c8d1cd79ba06c876bd23630c83466e9500321be55c96ace2", + "https://bcr.bazel.build/modules/abseil-cpp/20211102.0/MODULE.bazel": "70390338f7a5106231d20620712f7cccb659cd0e9d073d1991c038eb9fc57589", + "https://bcr.bazel.build/modules/abseil-cpp/20211102.0/source.json": "7e3a9adf473e9af076ae485ed649d5641ad50ec5c11718103f34de03170d94ad", + "https://bcr.bazel.build/modules/apple_support/1.5.0/MODULE.bazel": "50341a62efbc483e8a2a6aec30994a58749bd7b885e18dd96aa8c33031e558ef", + "https://bcr.bazel.build/modules/apple_support/1.5.0/source.json": "eb98a7627c0bc486b57f598ad8da50f6625d974c8f723e9ea71bd39f709c9862", + "https://bcr.bazel.build/modules/bazel_features/1.11.0/MODULE.bazel": "f9382337dd5a474c3b7d334c2f83e50b6eaedc284253334cf823044a26de03e8", + "https://bcr.bazel.build/modules/bazel_features/1.11.0/source.json": "c9320aa53cd1c441d24bd6b716da087ad7e4ff0d9742a9884587596edfe53015", + "https://bcr.bazel.build/modules/bazel_skylib/1.0.3/MODULE.bazel": "bcb0fd896384802d1ad283b4e4eb4d718eebd8cb820b0a2c3a347fb971afd9d8", + "https://bcr.bazel.build/modules/bazel_skylib/1.2.1/MODULE.bazel": "f35baf9da0efe45fa3da1696ae906eea3d615ad41e2e3def4aeb4e8bc0ef9a7a", + "https://bcr.bazel.build/modules/bazel_skylib/1.3.0/MODULE.bazel": "20228b92868bf5cfc41bda7afc8a8ba2a543201851de39d990ec957b513579c5", + "https://bcr.bazel.build/modules/bazel_skylib/1.6.1/MODULE.bazel": "8fdee2dbaace6c252131c00e1de4b165dc65af02ea278476187765e1a617b917", + "https://bcr.bazel.build/modules/bazel_skylib/1.6.1/source.json": "082ed5f9837901fada8c68c2f3ddc958bb22b6d654f71dd73f3df30d45d4b749", + "https://bcr.bazel.build/modules/buildozer/7.1.2/MODULE.bazel": "2e8dd40ede9c454042645fd8d8d0cd1527966aa5c919de86661e62953cd73d84", + "https://bcr.bazel.build/modules/buildozer/7.1.2/source.json": "c9028a501d2db85793a6996205c8de120944f50a0d570438fcae0457a5f9d1f8", + "https://bcr.bazel.build/modules/googletest/1.11.0/MODULE.bazel": "3a83f095183f66345ca86aa13c58b59f9f94a2f81999c093d4eeaa2d262d12f4", + "https://bcr.bazel.build/modules/googletest/1.11.0/source.json": "c73d9ef4268c91bd0c1cd88f1f9dfa08e814b1dbe89b5f594a9f08ba0244d206", + "https://bcr.bazel.build/modules/platforms/0.0.4/MODULE.bazel": "9b328e31ee156f53f3c416a64f8491f7eb731742655a47c9eec4703a71644aee", + "https://bcr.bazel.build/modules/platforms/0.0.5/MODULE.bazel": "5733b54ea419d5eaf7997054bb55f6a1d0b5ff8aedf0176fef9eea44f3acda37", + "https://bcr.bazel.build/modules/platforms/0.0.6/MODULE.bazel": "ad6eeef431dc52aefd2d77ed20a4b353f8ebf0f4ecdd26a807d2da5aa8cd0615", + "https://bcr.bazel.build/modules/platforms/0.0.7/MODULE.bazel": "72fd4a0ede9ee5c021f6a8dd92b503e089f46c227ba2813ff183b71616034814", + "https://bcr.bazel.build/modules/platforms/0.0.9/MODULE.bazel": "4a87a60c927b56ddd67db50c89acaa62f4ce2a1d2149ccb63ffd871d5ce29ebc", + "https://bcr.bazel.build/modules/platforms/0.0.9/source.json": "cd74d854bf16a9e002fb2ca7b1a421f4403cda29f824a765acd3a8c56f8d43e6", + "https://bcr.bazel.build/modules/protobuf/21.7/MODULE.bazel": "a5a29bb89544f9b97edce05642fac225a808b5b7be74038ea3640fae2f8e66a7", + "https://bcr.bazel.build/modules/protobuf/21.7/source.json": "bbe500720421e582ff2d18b0802464205138c06056f443184de39fbb8187b09b", + "https://bcr.bazel.build/modules/protobuf/3.19.0/MODULE.bazel": "6b5fbb433f760a99a22b18b6850ed5784ef0e9928a72668b66e4d7ccd47db9b0", + "https://bcr.bazel.build/modules/protobuf/3.19.6/MODULE.bazel": "9233edc5e1f2ee276a60de3eaa47ac4132302ef9643238f23128fea53ea12858", + "https://bcr.bazel.build/modules/rules_cc/0.0.1/MODULE.bazel": "cb2aa0747f84c6c3a78dad4e2049c154f08ab9d166b1273835a8174940365647", + "https://bcr.bazel.build/modules/rules_cc/0.0.2/MODULE.bazel": "6915987c90970493ab97393024c156ea8fb9f3bea953b2f3ec05c34f19b5695c", + "https://bcr.bazel.build/modules/rules_cc/0.0.8/MODULE.bazel": "964c85c82cfeb6f3855e6a07054fdb159aced38e99a5eecf7bce9d53990afa3e", + "https://bcr.bazel.build/modules/rules_cc/0.0.9/MODULE.bazel": "836e76439f354b89afe6a911a7adf59a6b2518fafb174483ad78a2a2fde7b1c5", + "https://bcr.bazel.build/modules/rules_cc/0.0.9/source.json": "1f1ba6fea244b616de4a554a0f4983c91a9301640c8fe0dd1d410254115c8430", + "https://bcr.bazel.build/modules/rules_java/4.0.0/MODULE.bazel": "5a78a7ae82cd1a33cef56dc578c7d2a46ed0dca12643ee45edbb8417899e6f74", + "https://bcr.bazel.build/modules/rules_java/7.6.1/MODULE.bazel": "2f14b7e8a1aa2f67ae92bc69d1ec0fa8d9f827c4e17ff5e5f02e91caa3b2d0fe", + "https://bcr.bazel.build/modules/rules_java/7.6.1/source.json": "8f3f3076554e1558e8e468b2232991c510ecbcbed9e6f8c06ac31c93bcf38362", + "https://bcr.bazel.build/modules/rules_jvm_external/4.4.2/MODULE.bazel": "a56b85e418c83eb1839819f0b515c431010160383306d13ec21959ac412d2fe7", + "https://bcr.bazel.build/modules/rules_jvm_external/4.4.2/source.json": "a075731e1b46bc8425098512d038d416e966ab19684a10a34f4741295642fc35", + "https://bcr.bazel.build/modules/rules_license/0.0.3/MODULE.bazel": "627e9ab0247f7d1e05736b59dbb1b6871373de5ad31c3011880b4133cafd4bd0", + "https://bcr.bazel.build/modules/rules_license/0.0.7/MODULE.bazel": "088fbeb0b6a419005b89cf93fe62d9517c0a2b8bb56af3244af65ecfe37e7d5d", + "https://bcr.bazel.build/modules/rules_license/0.0.7/source.json": "355cc5737a0f294e560d52b1b7a6492d4fff2caf0bef1a315df5a298fca2d34a", + "https://bcr.bazel.build/modules/rules_pkg/0.7.0/MODULE.bazel": "df99f03fc7934a4737122518bb87e667e62d780b610910f0447665a7e2be62dc", + "https://bcr.bazel.build/modules/rules_pkg/0.7.0/source.json": "c2557066e0c0342223ba592510ad3d812d4963b9024831f7f66fd0584dd8c66c", + "https://bcr.bazel.build/modules/rules_proto/4.0.0/MODULE.bazel": "a7a7b6ce9bee418c1a760b3d84f83a299ad6952f9903c67f19e4edd964894e06", + "https://bcr.bazel.build/modules/rules_proto/5.3.0-21.7/MODULE.bazel": "e8dff86b0971688790ae75528fe1813f71809b5afd57facb44dad9e8eca631b7", + "https://bcr.bazel.build/modules/rules_proto/5.3.0-21.7/source.json": "d57902c052424dfda0e71646cb12668d39c4620ee0544294d9d941e7d12bc3a9", + "https://bcr.bazel.build/modules/rules_python/0.10.2/MODULE.bazel": "cc82bc96f2997baa545ab3ce73f196d040ffb8756fd2d66125a530031cd90e5f", + "https://bcr.bazel.build/modules/rules_python/0.22.1/MODULE.bazel": "26114f0c0b5e93018c0c066d6673f1a2c3737c7e90af95eff30cfee38d0bbac7", + "https://bcr.bazel.build/modules/rules_python/0.22.1/source.json": "57226905e783bae7c37c2dd662be078728e48fa28ee4324a7eabcafb5a43d014", + "https://bcr.bazel.build/modules/rules_python/0.4.0/MODULE.bazel": "9208ee05fd48bf09ac60ed269791cf17fb343db56c8226a720fbb1cdf467166c", + "https://bcr.bazel.build/modules/stardoc/0.5.1/MODULE.bazel": "1a05d92974d0c122f5ccf09291442580317cdd859f07a8655f1db9a60374f9f8", + "https://bcr.bazel.build/modules/stardoc/0.5.1/source.json": "a96f95e02123320aa015b956f29c00cb818fa891ef823d55148e1a362caacf29", + "https://bcr.bazel.build/modules/upb/0.0.0-20220923-a547704/MODULE.bazel": "7298990c00040a0e2f121f6c32544bab27d4452f80d9ce51349b1a28f3005c43", + "https://bcr.bazel.build/modules/upb/0.0.0-20220923-a547704/source.json": "f1ef7d3f9e0e26d4b23d1c39b5f5de71f584dd7d1b4ef83d9bbba6ec7a6a6459", + "https://bcr.bazel.build/modules/zlib/1.2.11/MODULE.bazel": "07b389abc85fdbca459b69e2ec656ae5622873af3f845e1c9d80fe179f3effa0", + "https://bcr.bazel.build/modules/zlib/1.2.12/MODULE.bazel": "3b1a8834ada2a883674be8cbd36ede1b6ec481477ada359cd2d3ddc562340b27", + "https://bcr.bazel.build/modules/zlib/1.3/MODULE.bazel": "6a9c02f19a24dcedb05572b2381446e27c272cd383aed11d41d99da9e3167a72", + "https://bcr.bazel.build/modules/zlib/1.3/source.json": "b6b43d0737af846022636e6e255fd4a96fee0d34f08f3830e6e0bac51465c37c" + }, + "selectedYankedVersions": {}, + "moduleExtensions": { + "@@apple_support~//crosstool:setup.bzl%apple_cc_configure_extension": { + "general": { + "bzlTransitiveDigest": "PjIds3feoYE8SGbbIq2SFTZy3zmxeO2tQevJZNDo7iY=", + "usagesDigest": "aLmqbvowmHkkBPve05yyDNGN7oh7QE9kBADr3QIZTZs=", + "recordedFileInputs": {}, + "recordedDirentsInputs": {}, + "envVariables": {}, + "generatedRepoSpecs": { + "local_config_apple_cc": { + "bzlFile": "@@apple_support~//crosstool:setup.bzl", + "ruleClassName": "_apple_cc_autoconf", + "attributes": {} + }, + "local_config_apple_cc_toolchains": { + "bzlFile": "@@apple_support~//crosstool:setup.bzl", + "ruleClassName": "_apple_cc_autoconf_toolchains", + "attributes": {} + } + }, + "recordedRepoMappingEntries": [ + [ + "apple_support~", + "bazel_tools", + "bazel_tools" + ] + ] + } + }, + "@@platforms//host:extension.bzl%host_platform": { + "general": { + "bzlTransitiveDigest": "xelQcPZH8+tmuOHVjL9vDxMnnQNMlwj0SlvgoqBkm4U=", + "usagesDigest": "meSzxn3DUCcYEhq4HQwExWkWtU4EjriRBQLsZN+Q0SU=", + "recordedFileInputs": {}, + "recordedDirentsInputs": {}, + "envVariables": {}, + "generatedRepoSpecs": { + "host_platform": { + "bzlFile": "@@platforms//host:extension.bzl", + "ruleClassName": "host_platform_repo", + "attributes": {} + } + }, + "recordedRepoMappingEntries": [] + } + } + } +}
diff --git a/proto/worker_protocol.proto b/proto/worker_protocol.proto new file mode 100644 index 0000000..e90cad4 --- /dev/null +++ b/proto/worker_protocol.proto
@@ -0,0 +1,100 @@ +// Copyright 2024 The Bazel Authors. All rights reserved. +// +// 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 +// +// http://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. + +syntax = "proto3"; + +package bazel_worker; + +option java_package = "com.google.devtools.build.lib.worker"; + +// An input file. +message Input { + // The path in the file system where to read this input artifact from. This is + // either a path relative to the execution root (the worker process is + // launched with the working directory set to the execution root), or an + // absolute path. + string path = 1; + + // A hash-value of the contents. The format of the contents is unspecified and + // the digest should be treated as an opaque token. This can be empty in some + // cases. + bytes digest = 2; +} + +// This represents a single work unit that Blaze sends to the worker. +message WorkRequest { + repeated string arguments = 1; + + // The inputs that the worker is allowed to read during execution of this + // request. + repeated Input inputs = 2; + + // Each WorkRequest must have either a unique + // request_id or request_id = 0. If request_id is 0, this WorkRequest must be + // processed alone (singleplex), otherwise the worker may process multiple + // WorkRequests in parallel (multiplexing). As an exception to the above, if + // the cancel field is true, the request_id must be the same as a previously + // sent WorkRequest. The request_id must be attached unchanged to the + // corresponding WorkResponse. Only one singleplex request may be sent to a + // worker at a time. + int32 request_id = 3; + + // EXPERIMENTAL: When true, this is a cancel request, indicating that a + // previously sent WorkRequest with the same request_id should be cancelled. + // The arguments and inputs fields must be empty and should be ignored. + bool cancel = 4; + + // Values greater than 0 indicate that the worker may output extra debug + // information to stderr (which will go into the worker log). Setting the + // --worker_verbose flag for Bazel makes this flag default to 10. + int32 verbosity = 5; + + // The relative directory inside the workers working directory where the + // inputs and outputs are placed, for sandboxing purposes. For singleplex + // workers, this is unset, as they can use their working directory as sandbox. + // For multiplex workers, this will be set when the + // --experimental_worker_multiplex_sandbox flag is set _and_ the execution + // requirements for the worker includes 'supports-multiplex-sandbox'. + // The paths in `inputs` will not contain this prefix, but the actual files + // will be placed/must be written relative to this directory. The worker + // implementation is responsible for resolving the file paths. + string sandbox_dir = 6; +} + +// The worker sends this message to Blaze when it finished its work on the +// WorkRequest message. +message WorkResponse { + int32 exit_code = 1; + + // Output message for this work unit. + // This is akin to the combined stdout/stderr if the work unit were executed + // as a standalone process. Output pertaining to a work unit should be + // reported here instead of through the stdout/stderr of the worker process. + // Assumed to be UTF-8 encoded. + string output = 2; + + // This field must be set to the same request_id as the WorkRequest it is a + // response to. Since worker processes which support multiplex worker will + // handle multiple WorkRequests in parallel, this ID will be used to + // determined which WorkerProxy does this WorkResponse belong to. + int32 request_id = 3; + + // EXPERIMENTAL When true, indicates that this response was sent due to + // receiving a cancel request. The exit_code and output fields should be empty + // and will be ignored. Exactly one WorkResponse must be sent for each + // non-cancelling WorkRequest received by the worker, but if the worker + // received a cancel request, it doesn't matter if it replies with a regular + // WorkResponse or with one where was_cancelled = true. + bool was_cancelled = 4; +}