@samitouri / QOSAMI-WSL-1 / commits / f05690c3

Fix broken WSLCorePort channel after receive timeout (#14455)

Refactor the socket channel to resolve the broken state after a timeout: * Replace the sequence number with transaction id. And add transaction step in the message header. * The transaction and non-transaction messages are differentiated by the transaction step. * The old sequence number logic stays the same for non-transaction messages. * For transaction messages. The reply side will reply with the same id in the incoming transaction request. * For transaction messages. The receive loop will discard stale messages based on the transaction id.

Feng Wang committed Apr 21, 2026 at 10:14 UTC f05690c3fd254d7f949e04e8ecabd2ddc0132200
20 files changed +517 -162
src/linux/init/GnsEngine.cpp
+8 -6
@@ -25,12 +25,13 @@ constexpr auto c_ipStrings = {"ip", "ip6"};
25 const char* c_loopbackInterfaceName = "lo";
26
27 GnsEngine::GnsEngine(
28 + wsl::shared::SocketChannel& channel,
29 const NotificationRoutine& notificationRoutine,
30 const StatusRoutine& statusRoutine,
31 NetworkManager& manager,
32 std::optional<int> dnsTunnelingFd,
33 const std::string& dnsTunnelingIpAddress) :
33 - notificationRoutine(notificationRoutine), statusRoutine(statusRoutine), manager(manager)
34 + channel(channel), notificationRoutine(notificationRoutine), statusRoutine(statusRoutine), manager(manager)
35 {
36 if (dnsTunnelingFd.has_value())
37 {
@@ -363,11 +364,11 @@ void GnsEngine::ProcessLinkChange(Interface& interface, const wsl::shared::hns::
364 }
365 }
366
366 -std::tuple<bool, int> GnsEngine::ProcessNextMessage()
367 +std::tuple<bool, int> GnsEngine::ProcessNextMessage(wsl::shared::Transaction& transaction)
368 {
369 int return_value = 0;
370
370 - auto payload = notificationRoutine();
371 + auto payload = notificationRoutine(transaction);
372 if (!payload.has_value())
373 {
374 GNS_LOG_ERROR("Received empty message, exiting");
@@ -723,22 +724,23 @@ void GnsEngine::run()
724
725 while (true)
726 {
727 + auto transaction = channel.ReceiveTransaction();
728 try
729 {
730 GNS_LOG_INFO("Processing Next Message");
729 - auto [should_continue, return_value] = ProcessNextMessage();
731 + auto [should_continue, return_value] = ProcessNextMessage(transaction);
732 if (!should_continue)
733 {
734 break;
735 }
736
737 GNS_LOG_INFO("Processing Next Message Successful ({:#x})", return_value);
736 - statusRoutine(return_value, "");
738 + statusRoutine(return_value, "", transaction);
739 }
740 catch (const std::exception& e)
741 {
742 GNS_LOG_ERROR("Error while processing message: {}", e.what());
741 - statusRoutine(-1, e.what());
743 + statusRoutine(-1, e.what(), transaction);
744 }
745 }
746
src/linux/init/GnsEngine.h
+5 -3
@@ -21,10 +21,11 @@ public:
21 std::optional<GUID> AdapterId;
22 };
23
24 - using NotificationRoutine = std::function<std::optional<Message>()>;
25 - using StatusRoutine = std::function<void(int, const std::string&)>;
24 + using NotificationRoutine = std::function<std::optional<Message>(wsl::shared::Transaction&)>;
25 + using StatusRoutine = std::function<void(int, const std::string&, wsl::shared::Transaction&)>;
26
27 GnsEngine(
28 + wsl::shared::SocketChannel& channel,
29 const NotificationRoutine& notificationRoutine,
30 const StatusRoutine& statusRoutine,
31 NetworkManager& manager,
@@ -36,7 +37,7 @@ public:
37 void run();
38
39 private:
39 - std::tuple<bool, int> ProcessNextMessage();
40 + std::tuple<bool, int> ProcessNextMessage(wsl::shared::Transaction& transaction);
41
42 void ProcessNotification(const nlohmann::json& payload, Interface& interface);
43
@@ -69,6 +70,7 @@ private:
70 void ProcessNotificationImpl(
71 Interface& interface, const nlohmann::json& payload, void (GnsEngine::*routine)(Interface&, const T&, wsl::shared::hns::ModifyRequestType));
72
73 + wsl::shared::SocketChannel& channel;
74 const NotificationRoutine& notificationRoutine;
75 const StatusRoutine& statusRoutine;
76 NetworkManager& manager;
src/linux/init/binfmt.cpp
+2 -1
@@ -174,7 +174,8 @@ try
174 // Send the create process message to the interop server.
175 //
176
177 - channel.SendMessage<LX_INIT_CREATE_NT_PROCESS_UTILITY_VM>(Span);
177 + auto transaction = channel.StartTransaction();
178 + transaction.Send<LX_INIT_CREATE_NT_PROCESS_UTILITY_VM>(Span);
179
180 //
181 // Accept connections from the interop server.
src/linux/init/config.cpp
+16 -15
@@ -334,7 +334,7 @@ try
334 CATCH_LOG()
335
336 void ConfigHandleInteropMessage(
337 - wsl::shared::SocketChannel& ResponseChannel,
337 + wsl::shared::Transaction& Transaction,
338 wsl::shared::SocketChannel& InteropChannel,
339 bool Elevated,
340 gsl::span<gsl::byte> Message,
@@ -350,7 +350,7 @@ Routine Description:
350
351 Arguments:
352
353 - ResponseChannel - Supplies channel used to send responses.
353 + Transaction - Supplies transaction used to send responses.
354
355 InteropChannel - Supplies a channel to the host to be used for create
356 process requests.
@@ -381,7 +381,7 @@ try
381
382 case LxInitMessageQueryDrvfsElevated:
383 {
384 - ResponseChannel.SendResultMessage<bool>(Elevated);
384 + Transaction.SendResultMessage<bool>(Elevated);
385 break;
386 }
387
@@ -397,7 +397,7 @@ try
397 auto Value = UtilGetEnvironmentVariable(Query->Buffer);
398 wsl::shared::MessageWriter<LX_INIT_QUERY_ENVIRONMENT_VARIABLE> Response(LxInitMessageQueryEnvironmentVariable);
399 Response.WriteString(Value);
400 - ResponseChannel.SendMessage<LX_INIT_QUERY_ENVIRONMENT_VARIABLE>(Response.Span());
400 + Transaction.Send<LX_INIT_QUERY_ENVIRONMENT_VARIABLE>(Response.Span());
401 }
402
403 break;
@@ -405,7 +405,7 @@ try
405 case LxInitMessageQueryFeatureFlags:
406 {
407 assert(Config.FeatureFlags.has_value());
408 - ResponseChannel.SendResultMessage<int32_t>(Config.FeatureFlags.value());
408 + Transaction.SendResultMessage<int32_t>(Config.FeatureFlags.value());
409 break;
410 }
411
@@ -419,7 +419,7 @@ try
419 }
420
421 bool success = false;
422 - auto sendResponse = wil::scope_exit([&]() { ResponseChannel.SendResultMessage<bool>(success); });
422 + auto sendResponse = wil::scope_exit([&]() { Transaction.SendResultMessage<bool>(success); });
423
424 if (!Config.BootInit || Config.InitPid.value_or(0) != getpid())
425 {
@@ -435,7 +435,7 @@ try
435
436 case LxInitMessageQueryNetworkingMode:
437 assert(Config.NetworkingMode.has_value());
438 - ResponseChannel.SendResultMessage<uint8_t>(static_cast<uint8_t>(Config.NetworkingMode.value()));
438 + Transaction.SendResultMessage<uint8_t>(static_cast<uint8_t>(Config.NetworkingMode.value()));
439 break;
440
441 case LxInitMessageQueryVmId:
@@ -446,7 +446,7 @@ try
446 Response.WriteString(Config.VmId.value());
447 }
448
449 - ResponseChannel.SendMessage<LX_INIT_QUERY_VM_ID>(Response.Span());
449 + Transaction.Send<LX_INIT_QUERY_VM_ID>(Response.Span());
450 break;
451 }
452
@@ -618,7 +618,7 @@ try
618 }
619 CATCH_LOG()
620
621 -int ConfigInitializeInstance(wsl::shared::SocketChannel& Channel, gsl::span<gsl::byte> Buffer, wsl::linux::WslDistributionConfig& Config)
621 +int ConfigInitializeInstance(const std::function<void(const gsl::span<gsl::byte>&)>& SendResponse, gsl::span<gsl::byte> Buffer, wsl::linux::WslDistributionConfig& Config)
622
623 /*++
624
@@ -632,7 +632,7 @@ Routine Description:
632
633 Arguments:
634
635 - MessageFd - Supplies a file descriptor to send the response message.
635 + SendResponse - Supplies a function to send the response message.
636
637 Buffer - Supplies the message buffer.
638
@@ -923,7 +923,7 @@ try
923 Response.WriteString(Response->VersionIndex, Version->c_str());
924 }
925
926 - Channel.SendMessage<LX_INIT_CONFIGURATION_INFORMATION_RESPONSE>(Response.Span());
926 + SendResponse(Response.Span());
927
928 //
929 // Accept the interop connection.
@@ -973,13 +973,14 @@ try
973 continue;
974 }
975
976 - auto [Message, Span] = ClientChannel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
976 + auto transaction = ClientChannel.ReceiveTransaction();
977 + auto [Message, Span] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
978 if (Message == nullptr)
979 {
980 continue;
981 }
982
982 - ConfigHandleInteropMessage(ClientChannel, InteropChannel, Elevated, Span, Message, Config);
983 + ConfigHandleInteropMessage(transaction, InteropChannel, Elevated, Span, Message, Config);
984 }
985 });
986
@@ -2186,7 +2187,7 @@ Return Value:
2187 return Result;
2188 }
2189
2189 -int ConfigRemountDrvFs(gsl::span<gsl::byte> Buffer, wsl::shared::SocketChannel& Channel, const wsl::linux::WslDistributionConfig& Config)
2190 +int ConfigRemountDrvFs(gsl::span<gsl::byte> Buffer, wsl::shared::Transaction& Transaction, const wsl::linux::WslDistributionConfig& Config)
2191
2192 /*++
2193
@@ -2207,7 +2208,7 @@ Return Value:
2208
2209 --*/
2210 {
2210 - Channel.SendResultMessage<int32_t>(ConfigRemountDrvFsImpl(Buffer, Config));
2211 + Transaction.SendResultMessage<int32_t>(ConfigRemountDrvFsImpl(Buffer, Config));
2212
2213 return 0;
2214 }
src/linux/init/config.h
+4 -3
@@ -20,6 +20,7 @@ Abstract:
20 #include <set>
21 #include <string_view>
22 #include <optional>
23 +#include <functional>
24 #include "SocketChannel.h"
25 #include "WslDistributionConfig.h"
26
@@ -399,7 +400,7 @@ std::set<std::pair<unsigned int, std::string>> ConfigGetMountedDrvFsVolumes(void
400 std::vector<std::pair<std::string, std::string>> ConfigGetWslgEnvironmentVariables(const wsl::linux::WslDistributionConfig& Config);
401
402 void ConfigHandleInteropMessage(
402 - wsl::shared::SocketChannel& ResponseChannel,
403 + wsl::shared::Transaction& Transaction,
404 wsl::shared::SocketChannel& InteropChannel,
405 bool Elevated,
406 gsl::span<gsl::byte> Message,
@@ -408,7 +409,7 @@ void ConfigHandleInteropMessage(
409
410 void ConfigInitializeCgroups(wsl::linux::WslDistributionConfig& Config);
411
411 -int ConfigInitializeInstance(wsl::shared::SocketChannel& Channel, gsl::span<gsl::byte> Buffer, wsl::linux::WslDistributionConfig& Config);
412 +int ConfigInitializeInstance(const std::function<void(const gsl::span<gsl::byte>&)>& SendResponse, gsl::span<gsl::byte> Buffer, wsl::linux::WslDistributionConfig& Config);
413
414 void ConfigMountDrvFsVolumes(unsigned int DrvFsVolumes, uid_t OwnerUid, std::optional<bool> Admin, const wsl::linux::WslDistributionConfig& Config);
415
@@ -420,7 +421,7 @@ int ConfigRegisterBinfmtInterpreter(void);
421
422 int ConfigSetMountNamespace(bool Elevated);
423
423 -int ConfigRemountDrvFs(gsl::span<gsl::byte> Buffer, wsl::shared::SocketChannel& Channel, const wsl::linux::WslDistributionConfig& Config);
424 +int ConfigRemountDrvFs(gsl::span<gsl::byte> Buffer, wsl::shared::Transaction& Transaction, const wsl::linux::WslDistributionConfig& Config);
425
426 int ConfigRemountDrvFsImpl(gsl::span<gsl::byte> Buffer, const wsl::linux::WslDistributionConfig& Config);
427
src/linux/init/drvfs.cpp
+3 -2
@@ -209,8 +209,9 @@ Return Value:
209 QueryPortMessage.MessageType = LxInitMessageQueryDrvfsElevated;
210 QueryPortMessage.MessageSize = sizeof(QueryPortMessage);
211
212 - channel.SendMessage(QueryPortMessage);
213 - return channel.ReceiveMessage<RESULT_MESSAGE<bool>>().Result;
212 + auto transaction = channel.StartTransaction();
213 + transaction.Send(QueryPortMessage);
214 + return transaction.Receive<RESULT_MESSAGE<bool>>().Result;
215 }
216
217 int MountFilesystem(const char* FsType, const char* Source, const char* Target, const char* Options, int* ExitCode)
src/linux/init/init.cpp
+71 -38
@@ -116,10 +116,15 @@ int InitConnectToServer(int LxBusFd, bool WaitForServer);
116 int InitCreateProcessUtilityVm(
117 gsl::span<gsl::byte> Message,
118 const LX_INIT_CREATE_PROCESS_UTILITY_VM& Header,
119 - wsl::shared::SocketChannel& MessageFd,
119 + wsl::shared::Transaction& Transaction,
120 const wsl::linux::WslDistributionConfig& Config);
121
122 -int InitCreateSessionLeader(gsl::span<gsl::byte> Buffer, wsl::shared::SocketChannel& Channel, int LxBusFd, wsl::linux::WslDistributionConfig& Config);
122 +int InitCreateSessionLeader(
123 + gsl::span<gsl::byte> Buffer,
124 + wsl::shared::SocketChannel& Channel,
125 + const std::function<void(LX_INIT_CREATE_SESSION_RESPONSE&)>& SendResponse,
126 + int LxBusFd,
127 + wsl::linux::WslDistributionConfig& Config);
128
129 void InitEntry(int Argc, char* Argv[]);
130
@@ -127,7 +132,7 @@ void InitEntryWsl(wsl::linux::WslDistributionConfig& Config);
132
133 void InitEntryUtilityVm(wsl::linux::WslDistributionConfig& Config);
134
130 -void InitTerminateInstance(gsl::span<gsl::byte> Buffer, wsl::shared::SocketChannel& Channel, wsl::linux::WslDistributionConfig& Config);
135 +void InitTerminateInstance(gsl::span<gsl::byte> Buffer, const std::function<void(bool)>& SendResult, wsl::linux::WslDistributionConfig& Config);
136
137 void InitTerminateInstanceInternal(const wsl::linux::WslDistributionConfig& Config);
138
@@ -1111,7 +1116,12 @@ Return Value:
1116 return 0;
1117 }
1118
1114 -int InitCreateSessionLeader(gsl::span<gsl::byte> Buffer, wsl::shared::SocketChannel& Channel, int LxBusFd, wsl::linux::WslDistributionConfig& Config)
1119 +int InitCreateSessionLeader(
1120 + gsl::span<gsl::byte> Buffer,
1121 + wsl::shared::SocketChannel& Channel,
1122 + const std::function<void(LX_INIT_CREATE_SESSION_RESPONSE&)>& SendResponse,
1123 + int LxBusFd,
1124 + wsl::linux::WslDistributionConfig& Config)
1125
1126 /*++
1127
@@ -1228,7 +1238,7 @@ try
1238 Response.Header.MessageType = LxInitMessageCreateSessionResponse;
1239 Response.Header.MessageSize = sizeof(Response);
1240 Response.Port = SocketAddress.svm_port;
1231 - Channel.SendMessage(Response);
1241 + SendResponse(Response);
1242
1243 if (!ListenSocket)
1244 {
@@ -1329,7 +1339,7 @@ Return Value:
1339 int InitCreateProcessUtilityVm(
1340 gsl::span<gsl::byte> Span,
1341 const LX_INIT_CREATE_PROCESS_UTILITY_VM& CreateProcess,
1332 - wsl::shared::SocketChannel& Channel,
1342 + wsl::shared::Transaction& Transaction,
1343 const wsl::linux::WslDistributionConfig& Config)
1344
1345 /*++
@@ -1414,7 +1424,7 @@ Return Value:
1424 // Tell the service which sockets ports to connect to.
1425 //
1426
1417 - Channel.SendResultMessage<uint32_t>(SocketAddress.svm_port);
1427 + Transaction.SendResultMessage<uint32_t>(SocketAddress.svm_port);
1428
1429 //
1430 // Exit if creating the listening socket failed.
@@ -1978,13 +1988,14 @@ Return Value:
1988 continue;
1989 }
1990
1981 - auto [Header, Span] = channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
1991 + auto transaction = channel.ReceiveTransaction();
1992 + auto [Header, Span] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
1993 if (Header != nullptr)
1994 {
1995 try
1996 {
1997 ConfigHandleInteropMessage(
1987 - channel, ControlChannel, WI_IsFlagSet(CreateProcess.Common.Flags, LxInitCreateProcessFlagsElevated), Span, Header, Config);
1998 + transaction, ControlChannel, WI_IsFlagSet(CreateProcess.Common.Flags, LxInitCreateProcessFlagsElevated), Span, Header, Config);
1999 }
2000 CATCH_LOG();
2001 }
@@ -2455,7 +2466,8 @@ Return Value:
2466 }
2467 else if (PollDescriptors[0].revents & POLLIN)
2468 {
2458 - auto [Header, Span] = channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
2469 + auto transaction = channel.ReceiveTransaction();
2470 + auto [Header, Span] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
2471 if (Header == nullptr)
2472 {
2473 break;
@@ -2464,16 +2476,23 @@ Return Value:
2476 switch (Header->MessageType)
2477 {
2478 case LxInitMessageCreateSession:
2467 - if (InitCreateSessionLeader(Span, channel, -1, Config) < 0)
2479 + {
2480 + auto SendResponse = [&](LX_INIT_CREATE_SESSION_RESPONSE& response) { transaction.Send(response); };
2481 + if (InitCreateSessionLeader(Span, channel, SendResponse, -1, Config) < 0)
2482 {
2483 FATAL_ERROR("InitCreateSessionLeader failed");
2484 }
2471 -
2472 - break;
2485 + }
2486 + break;
2487
2488 case LxInitMessageInitialize:
2475 - ConfigInitializeInstance(channel, Span, Config);
2476 - break;
2489 + {
2490 + auto SendResponse = [&](const gsl::span<gsl::byte>& span) {
2491 + transaction.Send<LX_INIT_CONFIGURATION_INFORMATION_RESPONSE>(span);
2492 + };
2493 + ConfigInitializeInstance(SendResponse, Span, Config);
2494 + }
2495 + break;
2496
2497 case LxInitMessageTimezoneInformation:
2498 UpdateTimezone(Span, Config);
@@ -2489,15 +2508,18 @@ Return Value:
2508 //
2509
2510 WaitForBootProcess(Config);
2492 - ConfigRemountDrvFs(Span, channel, Config);
2511 + ConfigRemountDrvFs(Span, transaction, Config);
2512 break;
2513
2514 case LxInitMessageTerminateInstance:
2496 - InitTerminateInstance(Span, channel, Config);
2497 - break;
2515 + {
2516 + auto SendResult = [&](bool result) { transaction.SendResultMessage<bool>(result); };
2517 + InitTerminateInstance(Span, SendResult, Config);
2518 + }
2519 + break;
2520
2521 case LxInitCreateProcess:
2500 - ProcessCreateProcessMessage(channel, Span);
2522 + ProcessCreateProcessMessage(transaction, Span);
2523 break;
2524
2525 default:
@@ -2612,7 +2634,9 @@ Return Value:
2634 switch (Header->MessageType)
2635 {
2636 case LxInitMessageCreateSession:
2615 - if (InitCreateSessionLeader(Message, Channel, LxBusFd.get(), Config) < 0)
2637 + {
2638 + auto SendResponse = [&](LX_INIT_CREATE_SESSION_RESPONSE& response) { Channel.SendMessage(response); };
2639 + if (InitCreateSessionLeader(Message, Channel, SendResponse, LxBusFd.get(), Config) < 0)
2640 {
2641 //
2642 // If this distro has no children, exit on failure.
@@ -2626,24 +2650,32 @@ Return Value:
2650
2651 LOG_ERROR("InitCreateSessionLeader failed");
2652 }
2629 -
2630 - break;
2653 + }
2654 + break;
2655
2656 case LxInitMessageNetworkInformation:
2657 ConfigUpdateNetworkInformation(Message, Config);
2658 break;
2659
2660 case LxInitMessageInitialize:
2637 - ConfigInitializeInstance(Channel, Message, Config);
2638 - break;
2661 + {
2662 + auto SendResponse = [&](const gsl::span<gsl::byte>& span) {
2663 + Channel.SendMessage<LX_INIT_CONFIGURATION_INFORMATION_RESPONSE>(span);
2664 + };
2665 + ConfigInitializeInstance(SendResponse, Message, Config);
2666 + }
2667 + break;
2668
2669 case LxInitMessageTimezoneInformation:
2670 UpdateTimezone(Message, Config);
2671 break;
2672
2673 case LxInitMessageTerminateInstance:
2645 - InitTerminateInstance(Message, Channel, Config);
2646 - break;
2674 + {
2675 + auto SendResult = [&](bool result) { Channel.SendResultMessage<bool>(result); };
2676 + InitTerminateInstance(Message, SendResult, Config);
2677 + }
2678 + break;
2679
2680 default:
2681 FATAL_ERROR("Unexpected message {}", Header->MessageType);
@@ -2653,7 +2685,7 @@ Return Value:
2685 return;
2686 }
2687
2656 -void InitTerminateInstance(gsl::span<gsl::byte> Buffer, wsl::shared::SocketChannel& Channel, wsl::linux::WslDistributionConfig& Config)
2688 +void InitTerminateInstance(gsl::span<gsl::byte> Buffer, const std::function<void(bool)>& SendResult, wsl::linux::WslDistributionConfig& Config)
2689
2690 /*++
2691
@@ -2665,7 +2697,7 @@ Arguments:
2697
2698 Buffer - Supplies the message buffer.
2699
2668 - Channel - Supplies a channel to send the response.
2700 + SendResult - Supplies a function to send the response.
2701
2702 Config - Supplies the distribution config.
2703
@@ -2690,7 +2722,7 @@ try
2722
2723 if (!StopPlan9Server(Message->Force, Config))
2724 {
2693 - Channel.SendResultMessage<bool>(false);
2725 + SendResult(false);
2726 return;
2727 }
2728
@@ -3036,7 +3068,8 @@ Return Value:
3068
3069 for (;;)
3070 {
3039 - auto [Message, Span] = channel.ReceiveMessageOrClosed<LX_INIT_CREATE_PROCESS_UTILITY_VM>();
3071 + auto transaction = channel.ReceiveTransaction();
3072 + auto [Message, Span] = transaction.ReceiveOrClosed<LX_INIT_CREATE_PROCESS_UTILITY_VM>();
3073 if (Message == nullptr)
3074 {
3075 _exit(0);
@@ -3045,7 +3078,7 @@ Return Value:
3078 switch (Message->Header.MessageType)
3079 {
3080 case LxInitMessageCreateProcessUtilityVm:
3048 - if (InitCreateProcessUtilityVm(Span, *Message, channel, Config) < 0)
3081 + if (InitCreateProcessUtilityVm(Span, *Message, transaction, Config) < 0)
3082 {
3083 FATAL_ERROR("InitCreateProcessUtilityVm failed");
3084 }
@@ -3294,7 +3327,7 @@ unsigned int StartGns(int Argc, char** Argv)
3327
3328 if (channel.Socket() == -1)
3329 {
3297 - readNotification = [&]() -> std::optional<GnsEngine::Message> {
3330 + readNotification = [&](wsl::shared::Transaction&) -> std::optional<GnsEngine::Message> {
3331 std::string content{std::istreambuf_iterator<char>(std::cin), std::istreambuf_iterator<char>()};
3332 if (content.empty())
3333 {
@@ -3308,7 +3341,7 @@ unsigned int StartGns(int Argc, char** Argv)
3341 return {{AdapterId.has_value() ? LxGnsMessageNotification : LxGnsMessageInterfaceConfiguration, content, AdapterId}};
3342 };
3343
3311 - returnStatus = [&](int Result, const std::string& Error) {
3344 + returnStatus = [&](int Result, const std::string& Error, wsl::shared::Transaction&) {
3345 GNS_LOG_INFO("Returning LxGnsMessageResult (no output fd) [{} - {}]", Result, Error.c_str());
3346 // exitCode keeps the most recent error in the test path
3347 if (Result != 0)
@@ -3320,9 +3353,9 @@ unsigned int StartGns(int Argc, char** Argv)
3353 }
3354 else
3355 {
3323 - readNotification = [&]() -> std::optional<GnsEngine::Message> {
3356 + readNotification = [&](wsl::shared::Transaction& transaction) -> std::optional<GnsEngine::Message> {
3357 std::vector<gsl::byte> Buffer;
3325 - auto [Message, Span] = channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
3358 + auto [Message, Span] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
3359 if (Message == nullptr)
3360 {
3361 return {};
@@ -3385,7 +3418,7 @@ unsigned int StartGns(int Argc, char** Argv)
3418 }
3419 };
3420
3388 - returnStatus = [&](int Result, const std::string& Error) {
3421 + returnStatus = [&](int Result, const std::string& Error, wsl::shared::Transaction& transaction) {
3422 std::vector<gsl::byte> Buffer(sizeof(LX_GNS_RESULT) + Error.size() + 1);
3423
3424 GNS_LOG_INFO("Returning LxGnsMessageResult [{} - {}]", Result, Error.c_str());
@@ -3397,13 +3430,13 @@ unsigned int StartGns(int Argc, char** Argv)
3430 response.WriteString(Error);
3431 }
3432
3400 - return channel.SendMessage<LX_GNS_RESULT>(response.Span());
3433 + return transaction.Send<LX_GNS_RESULT>(response.Span());
3434 };
3435 }
3436
3437 RoutingTable routingTable(RT_TABLE_MAIN);
3438 NetworkManager manager(routingTable);
3406 - GnsEngine engine(readNotification, returnStatus, manager, DnsFd, DnsTunnelingIp);
3439 + GnsEngine engine(channel, readNotification, returnStatus, manager, DnsFd, DnsTunnelingIp);
3440
3441 engine.run();
3442
src/linux/init/localhost.cpp
+4 -2
@@ -246,7 +246,8 @@ try
246 {
247 auto message = SockToRelayMessage(sock);
248 message.Header.MessageType = LxGnsMessagePortListenerRelayStart;
249 - channel.SendMessage(message);
249 + auto transaction = channel.StartTransaction();
250 + transaction.Send(message);
251
252 return 0;
253 }
@@ -257,7 +258,8 @@ try
258 {
259 auto message = SockToRelayMessage(sock);
260 message.Header.MessageType = LxGnsMessagePortListenerRelayStop;
260 - channel.SendMessage(message);
261 + auto transaction = channel.StartTransaction();
262 + transaction.Send(message);
263
264 return 0;
265 }
src/linux/init/main.cpp
+11 -12
@@ -193,7 +193,7 @@ int MountInit(const char* Target);
193
194 int MountPlan9(const char* Name, const char* Target, bool ReadOnly, std::optional<int> BufferSize = {});
195
196 -int ProcessMessage(wsl::shared::SocketChannel& channel, LX_MESSAGE_TYPE Type, gsl::span<gsl::byte> Buffer, VmConfiguration& Config);
196 +int ProcessMessage(wsl::shared::Transaction& Transaction, LX_MESSAGE_TYPE Type, gsl::span<gsl::byte> Buffer, VmConfiguration& Config);
197
198 wil::unique_fd RegisterSeccompHook();
199
@@ -2808,7 +2808,7 @@ void ProcessImportExportMessage(gsl::span<gsl::byte> Buffer, wsl::shared::Socket
2808 }
2809 }
2810
2811 -int ProcessMountFolderMessage(wsl::shared::SocketChannel& Channel, gsl::span<gsl::byte> Buffer)
2811 +int ProcessMountFolderMessage(wsl::shared::Transaction& Transaction, gsl::span<gsl::byte> Buffer)
2812
2813 /*++
2814
@@ -2844,7 +2844,7 @@ Return Value:
2844 }
2845
2846 int Result = MountPlan9(Name, Target, Message->ReadOnly);
2847 - Channel.SendResultMessage<int32_t>(Result);
2847 + Transaction.SendResultMessage<int32_t>(Result);
2848 return 0;
2849 }
2850
@@ -3163,7 +3163,7 @@ try
3163 }
3164 CATCH_RETURN_ERRNO();
3165
3166 -int ProcessMessage(wsl::shared::SocketChannel& Channel, LX_MESSAGE_TYPE Type, gsl::span<gsl::byte> Buffer, VmConfiguration& Config)
3166 +int ProcessMessage(wsl::shared::Transaction& Transaction, LX_MESSAGE_TYPE Type, gsl::span<gsl::byte> Buffer, VmConfiguration& Config)
3167
3168 /*++
3169
@@ -3173,9 +3173,7 @@ Routine Description:
3173
3174 Arguments:
3175
3176 - MessageFd - Supplies a file descriptor to the socket on which the message was
3177 - received. This is used for operations that require responses, for example a
3178 - VHD eject request.
3176 + Transaction - Supplies the transaction for replying to the message.
3177
3178 Buffer - Supplies the message.
3179
@@ -3259,7 +3257,7 @@ try
3257 return -1;
3258 }
3259
3262 - Channel.SendResultMessage(EjectScsi(EjectMessage->Lun));
3260 + Transaction.SendResultMessage(EjectScsi(EjectMessage->Lun));
3261 return 0;
3262 }
3263
@@ -3495,10 +3493,10 @@ try
3493 return 0;
3494
3495 case LxMiniInitMountFolder:
3498 - return ProcessMountFolderMessage(Channel, Buffer);
3496 + return ProcessMountFolderMessage(Transaction, Buffer);
3497
3498 case LxInitCreateProcess:
3501 - return ProcessCreateProcessMessage(Channel, Buffer);
3499 + return ProcessCreateProcessMessage(Transaction, Buffer);
3500
3501 case LxMiniInitMessageWaitForPmemDevice:
3502 {
@@ -4175,13 +4173,14 @@ int main(int Argc, char* Argv[])
4173 }
4174 else if (PollDescriptors[0].revents & POLLIN)
4175 {
4178 - auto [Message, Range] = channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
4176 + auto transaction = channel.ReceiveTransaction();
4177 + auto [Message, Range] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
4178 if (Message == nullptr)
4179 {
4180 break; // Socket was closed, exit
4181 }
4182
4184 - Result = ProcessMessage(channel, Message->MessageType, Range, Config);
4183 + Result = ProcessMessage(transaction, Message->MessageType, Range, Config);
4184 if (Result < 0)
4185 {
4186 goto ErrorExit;
src/linux/init/plan9.cpp
+3 -2
@@ -157,13 +157,14 @@ try
157 std::vector<gsl::byte> Buffer;
158 for (;;)
159 {
160 - auto [Message, _] = channel.ReceiveMessageOrClosed<LX_INIT_STOP_PLAN9_SERVER>();
160 + auto transaction = channel.ReceiveTransaction();
161 + auto [Message, _] = transaction.ReceiveOrClosed<LX_INIT_STOP_PLAN9_SERVER>();
162 if (Message == nullptr)
163 {
164 _exit(0);
165 }
166
166 - channel.SendResultMessage<bool>(StopPlan9Server(fileSystem, Message->Force));
167 + transaction.SendResultMessage<bool>(StopPlan9Server(fileSystem, Message->Force));
168 }
169 }
170 CATCH_LOG();
src/linux/init/util.cpp
+14 -10
@@ -1110,13 +1110,14 @@ try
1110 wsl::shared::MessageWriter<LX_INIT_QUERY_ENVIRONMENT_VARIABLE> Message(LxInitMessageQueryEnvironmentVariable);
1111 Message.WriteString(Name);
1112
1113 - channel.SendMessage<LX_INIT_QUERY_ENVIRONMENT_VARIABLE>(Message.Span());
1113 + auto transaction = channel.StartTransaction();
1114 + transaction.Send<LX_INIT_QUERY_ENVIRONMENT_VARIABLE>(Message.Span());
1115
1116 //
1117 // Read a response, this will contain the environment variable value if it exists.
1118 //
1119
1119 - Value = channel.ReceiveMessage<LX_INIT_QUERY_ENVIRONMENT_VARIABLE>().Buffer;
1120 + Value = transaction.Receive<LX_INIT_QUERY_ENVIRONMENT_VARIABLE>().Buffer;
1121
1122 //
1123 // Set the environment variable for future queries.
@@ -1195,8 +1196,9 @@ Return Value:
1196 Message.MessageType = LxInitMessageQueryFeatureFlags;
1197 Message.MessageSize = sizeof(Message);
1198
1198 - channel.SendMessage(Message);
1199 - FeatureFlags = channel.ReceiveMessage<RESULT_MESSAGE<int32_t>>().Result;
1199 + auto transaction = channel.StartTransaction();
1200 + transaction.Send(Message);
1201 + FeatureFlags = transaction.Receive<RESULT_MESSAGE<int32_t>>().Result;
1202 }
1203
1204 UtilSetFeatureFlags(FeatureFlags, FeatureFlagEnv == nullptr);
@@ -1264,9 +1266,10 @@ try
1266 Message.MessageType = LxInitMessageQueryNetworkingMode;
1267 Message.MessageSize = sizeof(Message);
1268
1267 - channel.SendMessage(Message);
1269 + auto transaction = channel.StartTransaction();
1270 + transaction.Send(Message);
1271
1269 - const auto& response = channel.ReceiveMessage<RESULT_MESSAGE<uint8_t>>();
1272 + const auto& response = transaction.Receive<RESULT_MESSAGE<uint8_t>>();
1273 auto NetworkingMode = static_cast<LX_MINI_INIT_NETWORKING_MODE>(response.Result);
1274
1275 THROW_ERRNO_IF(EINVAL, NetworkingMode < LxMiniInitNetworkingModeNone || NetworkingMode > LxMiniInitNetworkingModeVirtioProxy);
@@ -1358,9 +1361,10 @@ try
1361 THROW_LAST_ERROR_IF(channel.Socket() < 0);
1362
1363 wsl::shared::MessageWriter<LX_INIT_QUERY_VM_ID> Message(LxInitMessageQueryVmId);
1361 - channel.SendMessage<LX_INIT_QUERY_VM_ID>(Message.Span());
1364 + auto transaction = channel.StartTransaction();
1365 + transaction.Send<LX_INIT_QUERY_VM_ID>(Message.Span());
1366
1363 - return channel.ReceiveMessage<LX_INIT_QUERY_VM_ID>().Buffer;
1367 + return transaction.Receive<LX_INIT_QUERY_VM_ID>().Buffer;
1368 }
1369 catch (...)
1370 {
@@ -3344,7 +3348,7 @@ Return Value:
3348 return 0;
3349 }
3350
3347 -int ProcessCreateProcessMessage(wsl::shared::SocketChannel& channel, gsl::span<gsl::byte> Buffer)
3351 +int ProcessCreateProcessMessage(wsl::shared::Transaction& Transaction, gsl::span<gsl::byte> Buffer)
3352 {
3353 auto* Message = gslhelpers::try_get_struct<CREATE_PROCESS_MESSAGE>(Buffer);
3354 if (!Message)
@@ -3353,7 +3357,7 @@ int ProcessCreateProcessMessage(wsl::shared::SocketChannel& channel, gsl::span<g
3357 return -1;
3358 }
3359
3356 - auto sendResult = [&](unsigned long Result) { channel.SendResultMessage<int32_t>(Result); };
3360 + auto sendResult = [&](unsigned long Result) { Transaction.SendResultMessage<int32_t>(Result); };
3361
3362 sockaddr_vm SocketAddress{};
3363 wil::unique_fd ListenSocket{UtilListenVsockAnyPort(&SocketAddress, 1, false)};
src/linux/init/util.h
+3 -2
@@ -35,7 +35,8 @@ Abstract:
35
36 namespace wsl::shared {
37 class SocketChannel;
38 -}
38 +class Transaction;
39 +} // namespace wsl::shared
40
41 namespace wsl::linux {
42 struct WslDistributionConfig;
@@ -312,4 +313,4 @@ uint16_t UtilWinAfToLinuxAf(uint16_t AddressFamily);
313
314 int WriteToFile(const char* Path, const char* Content, int permissions = 0644);
315
315 -int ProcessCreateProcessMessage(wsl::shared::SocketChannel& channel, gsl::span<gsl::byte> Buffer);
\ No newline at end of file
316 +int ProcessCreateProcessMessage(wsl::shared::Transaction& Transaction, gsl::span<gsl::byte> Buffer);
\ No newline at end of file
src/shared/inc/SocketChannel.h
+317 -34
@@ -14,6 +14,7 @@ Abstract:
14
15 #pragma once
16
17 +#include <atomic>
18 #include <mutex>
19 #include "socketshared.h"
20 #include "lxinitshared.h"
@@ -41,6 +42,44 @@ constexpr timeval* DefaultSocketTimeout = nullptr;
42
43 #endif
44
45 +class SocketChannel;
46 +
47 +class Transaction
48 +{
49 + friend class SocketChannel;
50 +
51 +public:
52 + ~Transaction() = default;
53 +
54 + NON_COPYABLE(Transaction);
55 +
56 + template <typename TMessage>
57 + void Send(gsl::span<gsl::byte> span);
58 +
59 + template <typename TMessage>
60 + void Send(TMessage& message);
61 +
62 + template <typename TResult>
63 + void SendResultMessage(TResult value);
64 +
65 + template <typename TMessage>
66 + std::pair<TMessage*, gsl::span<gsl::byte>> ReceiveOrClosed(TTimeout timeout = DefaultSocketTimeout);
67 +
68 + template <typename TMessage>
69 + TMessage& Receive(gsl::span<gsl::byte>* responseSpan = nullptr, TTimeout timeout = DefaultSocketTimeout);
70 +
71 +private:
72 + Transaction(SocketChannel& channel, uint32_t id) :
73 + m_channel(channel), m_id(id), m_step(static_cast<uint32_t>(TRANSACTION_STEP::REQUEST))
74 + {
75 + }
76 +
77 + SocketChannel& m_channel;
78 + uint32_t m_id;
79 + /** Use uint32_t as step can go beyond FIRST_REPLY */
80 + uint32_t m_step;
81 +};
82 +
83 class SocketChannel
84 {
85
@@ -63,6 +102,9 @@ public:
102 m_exitEvent = std::move(other.m_exitEvent);
103 #endif
104 m_ignore_sequence = other.m_ignore_sequence;
105 + m_sent_non_transaction_messages = other.m_sent_non_transaction_messages;
106 + m_received_non_transaction_messages = other.m_received_non_transaction_messages;
107 + m_transaction_id_seed = other.m_transaction_id_seed.load();
108
109 return *this;
110 }
@@ -82,7 +124,7 @@ public:
124 #endif
125
126 template <typename TMessage>
85 - void SendMessage(gsl::span<gsl::byte> span)
127 + void SendMessage(gsl::span<gsl::byte> span, uint32_t transactionStep = static_cast<uint32_t>(TRANSACTION_STEP::NONE), uint32_t transactionId = 0)
128 {
129 // Ensure that no other thread is using this channel.
130 const std::unique_lock<std::mutex> lock{m_sendMutex, std::try_to_lock};
@@ -103,12 +145,20 @@ public:
145
146 THROW_INVALID_ARG_IF(m_name == nullptr || span.size() < sizeof(TMessage));
147
106 - m_sent_messages++;
107 -
148 auto* header = gslhelpers::try_get_struct<MESSAGE_HEADER>(span);
149 WI_ASSERT(header->MessageSize == span.size());
150
111 - header->SequenceNumber = m_sent_messages;
151 + if (transactionStep == static_cast<unsigned int>(TRANSACTION_STEP::NONE))
152 + {
153 + m_sent_non_transaction_messages++;
154 + header->TransactionId = m_sent_non_transaction_messages;
155 + header->TransactionStep = static_cast<unsigned int>(TRANSACTION_STEP::NONE);
156 + }
157 + else
158 + {
159 + header->TransactionId = transactionId;
160 + header->TransactionStep = transactionStep;
161 + }
162
163 #ifdef WIN32
164
@@ -150,7 +200,7 @@ public:
200 }
201
202 template <typename TMessage>
153 - void SendMessage(TMessage& message)
203 + void SendMessage(TMessage& message, uint32_t transactionStep = static_cast<uint32_t>(TRANSACTION_STEP::NONE), uint32_t transactionId = 0)
204 {
205 // Catch situations where the other SendMessage() method should be used
206 const auto& header = GetMessageHeader(message);
@@ -164,7 +214,7 @@ public:
214 #endif
215 }
216
167 - SendMessage<TMessage>(gslhelpers::struct_as_writeable_bytes(message));
217 + SendMessage<TMessage>(gslhelpers::struct_as_writeable_bytes(message), transactionStep, transactionId);
218 }
219
220 template <typename TResult>
@@ -179,7 +229,10 @@ public:
229 }
230
231 template <typename TMessage>
182 - std::pair<TMessage*, gsl::span<gsl::byte>> ReceiveMessageOrClosed(TTimeout timeout = DefaultSocketTimeout)
232 + std::pair<TMessage*, gsl::span<gsl::byte>> ReceiveMessageOrClosed(
233 + TTimeout timeout = DefaultSocketTimeout,
234 + uint32_t expectedTransactionStep = static_cast<uint32_t>(TRANSACTION_STEP::NONE),
235 + uint32_t expectedTransactionId = 0)
236 {
237 WI_ASSERT(m_name != nullptr);
238
@@ -199,20 +252,180 @@ public:
252 #endif
253 }
254
202 - m_received_messages++;
203 -
204 - auto receivedSpan = ReceiveImpl(TMessage::Type, timeout);
205 - if (receivedSpan.empty())
255 + gsl::span<gsl::byte> receivedSpan{};
256 + for (;;)
257 {
258 + if (expectedTransactionStep == static_cast<uint32_t>(TRANSACTION_STEP::NONE))
259 + {
260 + // Adhere to the old ++ before receive behavior for non-transaction messages.
261 + m_received_non_transaction_messages++;
262 + }
263 +
264 + receivedSpan = ReceiveImpl(TMessage::Type, timeout);
265 + if (receivedSpan.empty())
266 + {
267 +
268 +#ifdef WIN32
269 + if (errno == HCS_E_CONNECTION_TIMEOUT)
270 + {
271 + THROW_HR_MSG(HCS_E_CONNECTION_TIMEOUT, "Timeout: %u, expected type: %hs, channel: %hs", timeout, ToString(TMessage::Type), m_name);
272 + }
273 +#endif
274 +
275 + return {nullptr, {}};
276 + }
277 +
278 + auto* header = gslhelpers::try_get_struct<MESSAGE_HEADER>(receivedSpan);
279 + if (header == nullptr)
280 + {
281 +#ifdef WIN32
282 + THROW_HR_MSG(E_UNEXPECTED, "Message too small for header: %zd, channel: %hs", receivedSpan.size(), m_name);
283 +#else
284 + LOG_ERROR("Message too small for header: {}, channel: {}", receivedSpan.size(), m_name);
285 + THROW_ERRNO(EINVAL);
286 +#endif
287 + }
288 +
289 + if (expectedTransactionStep == static_cast<uint32_t>(TRANSACTION_STEP::NONE))
290 + {
291 + // Handle non-transaction messages with legacy logic.
292 + if (!m_ignore_sequence)
293 + {
294 + if (header->TransactionStep != static_cast<unsigned int>(TRANSACTION_STEP::NONE))
295 + {
296 +#ifdef WIN32
297 + THROW_HR_MSG(
298 + E_UNEXPECTED,
299 + "Unexpected transaction message received on non-transaction channel: %hs, message type: %hs",
300 + m_name,
301 + ToString(header->MessageType));
302 +#else
303 + LOG_ERROR(
304 + "Unexpected transaction message received on non-transaction channel: {}, message type: {}",
305 + m_name,
306 + ToString(header->MessageType));
307 + THROW_ERRNO(EINVAL);
308 +#endif
309 + }
310 + if (header->TransactionId != m_received_non_transaction_messages)
311 + {
312 +#ifdef WIN32
313 + THROW_HR_MSG(
314 + E_UNEXPECTED,
315 + "Unexpected non-transaction message id: %u, expected: %u, channel: %hs",
316 + header->TransactionId,
317 + m_received_non_transaction_messages,
318 + m_name);
319 +#else
320 + LOG_ERROR("Unexpected non-transaction message id: {}, expected: {}, channel: {}", header->TransactionId, m_received_non_transaction_messages, m_name);
321 + THROW_ERRNO(EINVAL);
322 +#endif
323 + }
324 + }
325 + break;
326 + }
327
328 + // Handle transaction messages
329 + if (header->TransactionStep == static_cast<uint32_t>(TRANSACTION_STEP::NONE))
330 + {
331 + // Skip stale non-transaction messages
332 #ifdef WIN32
209 - if (errno == HCS_E_CONNECTION_TIMEOUT)
333 + WSL_LOG(
334 + "DiscardStaleNonTransactionMessage",
335 + TraceLoggingValue(m_name, "Name"),
336 + TraceLoggingValue(ToString(header->MessageType), "MessageType"),
337 + TraceLoggingValue(ToString(TMessage::Type), "ExpectedMessageType"),
338 + TraceLoggingValue(header->TransactionId, "StaleNonTransactionId"),
339 + TraceLoggingValue(m_received_non_transaction_messages, "ExpectedNonTransactionId"));
340 +#else
341 + LOG_WARNING(
342 + "Discard stale non-transaction message on channel: {}. MessageType: {}, ExpectedMessageType: {}, "
343 + "StaleNonTransactionId: {}, ExpectedNonTransactionId: {}",
344 + m_name,
345 + header->MessageType,
346 + TMessage::Type,
347 + header->TransactionId,
348 + m_received_non_transaction_messages);
349 +#endif
350 + continue;
351 + }
352 +
353 + if (expectedTransactionStep == static_cast<uint32_t>(TRANSACTION_STEP::REQUEST))
354 {
211 - THROW_HR_MSG(HCS_E_CONNECTION_TIMEOUT, "Timeout: %d, expected type: %hs, channel: %hs", timeout, ToString(TMessage::Type), m_name);
355 + // Skip until we get the next request. No matter the transaction id.
356 + if (header->TransactionStep != static_cast<unsigned int>(TRANSACTION_STEP::REQUEST))
357 + {
358 +#ifdef WIN32
359 + WSL_LOG(
360 + "DiscardOutOfOrderTransactionMessage",
361 + TraceLoggingValue(m_name, "Name"),
362 + TraceLoggingValue(ToString(header->MessageType), "MessageType"),
363 + TraceLoggingValue(ToString(TMessage::Type), "ExpectedMessageType"),
364 + TraceLoggingValue(header->TransactionStep, "StaleTransactionStep"),
365 + TraceLoggingValue(expectedTransactionStep, "ExpectedTransactionStep"));
366 +#else
367 + LOG_WARNING(
368 + "Discard out of order transaction message on channel: {}. MessageType: {}, ExpectedMessageType: {}, "
369 + "StaleTransactionStep: {}, ExpectedTransactionStep: {}",
370 + m_name,
371 + header->MessageType,
372 + TMessage::Type,
373 + header->TransactionStep,
374 + expectedTransactionStep);
375 +#endif
376 + continue;
377 + }
378 + break;
379 }
380 +
381 + auto diff = static_cast<int32_t>(header->TransactionId - expectedTransactionId);
382 + if (diff < 0)
383 + {
384 + // Skip stale transaction messages
385 +#ifdef WIN32
386 + WSL_LOG(
387 + "DiscardStaleTransactionMessage",
388 + TraceLoggingValue(m_name, "Name"),
389 + TraceLoggingValue(ToString(header->MessageType), "MessageType"),
390 + TraceLoggingValue(ToString(TMessage::Type), "ExpectedMessageType"),
391 + TraceLoggingValue(header->TransactionId, "StaleTransactionId"),
392 + TraceLoggingValue(expectedTransactionId, "ExpectedTransactionId"));
393 +#else
394 + LOG_WARNING(
395 + "Discard stale transaction message on channel: {}. MessageType: {}, ExpectedMessageType: {}, "
396 + "StaleTransactionId: {}, ExpectedTransactionId: {}",
397 + m_name,
398 + header->MessageType,
399 + TMessage::Type,
400 + header->TransactionId,
401 + expectedTransactionId);
402 +#endif
403 + continue;
404 + }
405 +
406 + if (diff > 0)
407 + {
408 + // Message is from the future.
409 +#ifdef WIN32
410 + THROW_HR_MSG(E_UNEXPECTED, "Unexpected transaction message id: %u, expected: %u, channel: %hs", header->TransactionId, expectedTransactionId, m_name);
411 +#else
412 + LOG_ERROR("Unexpected transaction message id: {}, expected: {}, channel: {}", header->TransactionId, expectedTransactionId, m_name);
413 + THROW_ERRNO(EINVAL);
414 +#endif
415 + }
416 +
417 + if (header->TransactionStep != expectedTransactionStep)
418 + {
419 + // Broken transaction.
420 +#ifdef WIN32
421 + THROW_HR_MSG(E_UNEXPECTED, "Unexpected transaction message step: %u, expected: %u, channel: %hs", header->TransactionStep, expectedTransactionStep, m_name);
422 +#else
423 + LOG_ERROR("Unexpected transaction message step: {}, expected: {}, channel: {}", header->TransactionStep, expectedTransactionStep, m_name);
424 + THROW_ERRNO(EINVAL);
425 #endif
426 + }
427
215 - return {nullptr, {}};
428 + break;
429 }
430
431 auto* message = gslhelpers::try_get_struct<TMessage>(receivedSpan);
@@ -228,7 +441,7 @@ public:
441 #endif
442 }
443
231 - ValidateMessageHeader(GetMessageHeader(*message), TMessage::Type, m_received_messages);
444 + ValidateMessageHeader(GetMessageHeader(*message), TMessage::Type);
445
446 #ifdef WIN32
447 WSL_LOG(
@@ -243,9 +456,13 @@ public:
456 }
457
458 template <typename TMessage>
246 - TMessage& ReceiveMessage(gsl::span<gsl::byte>* responseSpan = nullptr, TTimeout timeout = DefaultSocketTimeout)
459 + TMessage& ReceiveMessage(
460 + gsl::span<gsl::byte>* responseSpan = nullptr,
461 + TTimeout timeout = DefaultSocketTimeout,
462 + uint32_t expectedTransactionStep = static_cast<uint32_t>(TRANSACTION_STEP::NONE),
463 + uint32_t expectedTransactionId = 0)
464 {
248 - auto [message, span] = ReceiveMessageOrClosed<TMessage>(timeout);
465 + auto [message, span] = ReceiveMessageOrClosed<TMessage>(timeout, expectedTransactionStep, expectedTransactionId);
466 if (message == nullptr)
467 {
468 #ifdef WIN32
@@ -264,16 +481,28 @@ public:
481 return *message;
482 }
483
267 - template <typename TSentMessage>
268 - TSentMessage::TResponse& Transaction(gsl::span<gsl::byte> message, gsl::span<gsl::byte>* responseSpan = nullptr, TTimeout timeout = DefaultSocketTimeout)
484 + Transaction StartTransaction()
485 + {
486 + uint32_t transactionId = m_transaction_id_seed++;
487 + return wsl::shared::Transaction(*this, transactionId);
488 + }
489 +
490 + Transaction ReceiveTransaction()
491 {
270 - SendMessage<TSentMessage>(message);
492 + // Transaction id should follow the received one on the receive end.
493 + return wsl::shared::Transaction(*this, 0);
494 + }
495
272 - return ReceiveMessage<typename TSentMessage::TResponse>(responseSpan, timeout);
496 + template <typename TSentMessage>
497 + typename TSentMessage::TResponse& Transaction(gsl::span<gsl::byte> message, gsl::span<gsl::byte>* responseSpan = nullptr, TTimeout timeout = DefaultSocketTimeout)
498 + {
499 + auto transaction = StartTransaction();
500 + transaction.Send<TSentMessage>(message);
501 + return transaction.Receive<typename TSentMessage::TResponse>(responseSpan, timeout);
502 }
503
504 template <typename TSentMessage>
276 - TSentMessage::TResponse& Transaction(TSentMessage& message, gsl::span<gsl::byte>* responseSpan = nullptr, TTimeout timeout = DefaultSocketTimeout)
505 + typename TSentMessage::TResponse& Transaction(TSentMessage& message, gsl::span<gsl::byte>* responseSpan = nullptr, TTimeout timeout = DefaultSocketTimeout)
506 {
507 WI_ASSERT(message.Header.MessageSize == sizeof(message));
508
@@ -321,33 +550,33 @@ private:
550
551 #endif
552
324 - void ValidateMessageHeader(const MESSAGE_HEADER& header, LX_MESSAGE_TYPE expected, unsigned int expectedSequence) const
553 + void ValidateMessageHeader(const MESSAGE_HEADER& header, LX_MESSAGE_TYPE expected) const
554 {
326 - if (header.MessageSize < sizeof(header) || (expected != LxMiniInitMessageAny && header.MessageType != expected) ||
327 - (!m_ignore_sequence && header.SequenceNumber != expectedSequence))
555 +
556 + if (header.MessageSize < sizeof(header) || (expected != LxMiniInitMessageAny && header.MessageType != expected))
557 {
558 #ifdef WIN32
559
560 THROW_HR_MSG(
561 E_UNEXPECTED,
333 - "Protocol error: Received message size: %u, type: %u, sequence: %u. Expected type: %u, expected sequence: %u, "
562 + "Protocol error: Received message size: %u, type: %u, id: %u, step: %u. Expected type: %u, "
563 "channel: %hs",
564 header.MessageSize,
565 header.MessageType,
337 - header.SequenceNumber,
566 + header.TransactionId,
567 + header.TransactionStep,
568 expected,
339 - expectedSequence,
569 m_name);
570 #else
571
572 LOG_ERROR(
344 - "Protocol error: Received message size: {}, type: {}, sequence: {}. Expected type: {}, expected sequence: {}, "
573 + "Protocol error: Received message size: {}, type: {}, id: {}, step: {}. Expected type: {}, "
574 "channel: {}",
575 header.MessageSize,
576 header.MessageType,
348 - header.SequenceNumber,
577 + header.TransactionId,
578 + header.TransactionStep,
579 expected,
350 - expectedSequence,
580 m_name);
581
582 THROW_ERRNO(EINVAL);
@@ -392,11 +621,65 @@ private:
621 HANDLE m_exitEvent{};
622
623 #endif
395 - uint32_t m_sent_messages = 0;
396 - uint32_t m_received_messages = 0;
624 + uint32_t m_sent_non_transaction_messages = 0;
625 + uint32_t m_received_non_transaction_messages = 0;
626 + std::atomic<uint32_t> m_transaction_id_seed = 0;
627 bool m_ignore_sequence = false;
628 const char* m_name{};
629 std::mutex m_sendMutex;
630 std::mutex m_receiveMutex;
631 };
402 -} // namespace wsl::shared
\ No newline at end of file
632 +
633 +template <typename TMessage>
634 +void Transaction::Send(gsl::span<gsl::byte> span)
635 +{
636 + m_channel.SendMessage<TMessage>(span, m_step, m_id);
637 + m_step++;
638 +}
639 +
640 +template <typename TMessage>
641 +void Transaction::Send(TMessage& message)
642 +{
643 + Send<TMessage>(gslhelpers::struct_as_writeable_bytes(message));
644 +}
645 +
646 +template <typename TResult>
647 +void Transaction::SendResultMessage(TResult value)
648 +{
649 + RESULT_MESSAGE<TResult> Result{};
650 + Result.Header.MessageSize = sizeof(Result);
651 + Result.Header.MessageType = RESULT_MESSAGE<TResult>::Type;
652 + Result.Result = value;
653 +
654 + Send(Result);
655 +}
656 +
657 +template <typename TMessage>
658 +std::pair<TMessage*, gsl::span<gsl::byte>> Transaction::ReceiveOrClosed(TTimeout timeout)
659 +{
660 + auto result = m_channel.ReceiveMessageOrClosed<TMessage>(timeout, m_step, m_id);
661 + if (m_step == static_cast<uint32_t>(TRANSACTION_STEP::REQUEST) && result.first != nullptr)
662 + {
663 + // Use the request's id for the reply side transaction.
664 + MESSAGE_HEADER& header = m_channel.GetMessageHeader(*result.first);
665 + m_id = header.TransactionId;
666 + }
667 + m_step++;
668 + return result;
669 +}
670 +
671 +template <typename TMessage>
672 +TMessage& Transaction::Receive(gsl::span<gsl::byte>* responseSpan, TTimeout timeout)
673 +{
674 + auto& message = m_channel.ReceiveMessage<TMessage>(responseSpan, timeout, m_step, m_id);
675 + if (m_step == static_cast<uint32_t>(TRANSACTION_STEP::REQUEST))
676 + {
677 + // Use the request's id for the reply side transaction.
678 + MESSAGE_HEADER& header = m_channel.GetMessageHeader(message);
679 + m_id = header.TransactionId;
680 + }
681 + m_step++;
682 + return message;
683 +}
684 +
685 +} // namespace wsl::shared
src/shared/inc/lxinitshared.h
+11 -3
@@ -475,15 +475,23 @@ inline void PrettyPrint(std::stringstream& Out, LX_MESSAGE_TYPE Value)
475 Out << ToString(Value);
476 }
477
478 +enum class TRANSACTION_STEP : unsigned int
479 +{
480 + NONE = 0,
481 + REQUEST = 1,
482 + FIRST_REPLY = 2,
483 +};
484 +
485 struct MESSAGE_HEADER
486 {
487 static inline auto Type = LxMiniInitMessageAny; // Setting this allows using MESSAGE_HEADER to receive any type of message
488
489 LX_MESSAGE_TYPE MessageType;
490 unsigned int MessageSize;
484 - unsigned int SequenceNumber;
491 + unsigned int TransactionId;
492 + unsigned int TransactionStep;
493
486 - PRETTY_PRINT(FIELD(MessageType), FIELD(MessageSize), FIELD(SequenceNumber));
494 + PRETTY_PRINT(FIELD(MessageType), FIELD(MessageSize), FIELD(TransactionId), FIELD(TransactionStep));
495 };
496
497 //
@@ -771,7 +779,7 @@ typedef struct _LX_GNS_SET_PORT_LISTENER
779 PRETTY_PRINT(FIELD(Header), FIELD(HvSocketPort));
780 } LX_GNS_SET_PORT_LISTENER, *PLX_GNS_SET_PORT_LISTENER;
781
774 -static_assert(sizeof(LX_GNS_SET_PORT_LISTENER) == 16);
782 +static_assert(sizeof(LX_GNS_SET_PORT_LISTENER) == 20);
783
784 typedef struct _LX_GNS_PORT_LISTENER_RELAY
785 {
src/shared/inc/socketshared.h
+4 -4
@@ -95,18 +95,18 @@ try
95
96 LOG_HR_MSG(
97 E_UNEXPECTED,
98 - "Socket closed while reading message. Size: %u, type: %i, sequence: %u",
98 + "Socket closed while reading message. Size: %u, type: %i, id: %u",
99 Header->MessageSize,
100 Header->MessageType,
101 - Header->SequenceNumber);
101 + Header->TransactionId);
102
103 #elif defined(__GNUC__)
104
105 LOG_ERROR(
106 - "Socket closed while reading message. Size: {}, type: {}, sequence: {}",
106 + "Socket closed while reading message. Size: {}, type: {}, id: {}",
107 Header->MessageSize,
108 Header->MessageType,
109 - Header->SequenceNumber);
109 + Header->TransactionId);
110
111 #endif
112
src/windows/common/GnsPortTrackerChannel.cpp
+5 -3
@@ -33,7 +33,8 @@ void GnsPortTrackerChannel::Run()
33 {
34 for (;;)
35 {
36 - auto [header, range] = m_channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
36 + auto transaction = m_channel.ReceiveTransaction();
37 + auto [header, range] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
38 if (header == nullptr)
39 {
40 return;
@@ -46,7 +47,8 @@ void GnsPortTrackerChannel::Run()
47 const auto* message = gslhelpers::try_get_struct<LX_GNS_PORT_ALLOCATION_REQUEST>(range);
48 THROW_HR_IF_MSG(E_UNEXPECTED, !message, "Unexpected message size: %i", header->MessageSize);
49
49 - m_channel.SendResultMessage<int32_t>(m_callback(ConvertPortRequestToSockAddr(message), message->Protocol, message->Allocate));
50 + transaction.SendResultMessage<int32_t>(
51 + m_callback(ConvertPortRequestToSockAddr(message), message->Protocol, message->Allocate));
52 }
53 break;
54 case LxGnsMessageIfStateChangeRequest:
@@ -55,7 +57,7 @@ void GnsPortTrackerChannel::Run()
57 THROW_HR_IF_MSG(E_UNEXPECTED, !message, "Unexpected message size: %i", header->MessageSize);
58
59 m_interfaceStateCallback(message->InterfaceName, message->InterfaceUp);
58 - m_channel.SendResultMessage<int32_t>(0);
60 + transaction.SendResultMessage<int32_t>(0);
61 }
62 break;
63 default:
src/windows/service/exe/LxssCreateProcess.h
+3 -3
@@ -85,15 +85,15 @@ public:
85 wsl::shared::MessageWriter<CREATE_PROCESS_MESSAGE> message(LxInitCreateProcess);
86 message.WriteString(message->PathIndex, Path);
87 gsl::copy(as_bytes(gsl::span(ArgumentsData)), message.InsertBuffer(message->CommandLineIndex, ArgumentsData.size()));
88 - channel.SendMessage<CREATE_PROCESS_MESSAGE>(message.Span());
88 + auto transaction = channel.StartTransaction();
89 + transaction.Send<CREATE_PROCESS_MESSAGE>(message.Span());
90
91 auto readResult = [&]() {
91 - const auto& message = channel.ReceiveMessage<RESULT_MESSAGE<int32_t>>(nullptr, Timeout);
92 + const auto& message = transaction.Receive<RESULT_MESSAGE<int32_t>>(nullptr, Timeout);
93 return message.Result;
94 };
95
96 auto processSocket = wsl::windows::common::hvsocket::Connect(RuntimeId, readResult(), terminatingEvent);
96 -
97 const auto execResult = readResult();
98 THROW_HR_IF_MSG(E_FAIL, execResult != 0, "Failed to execute '%hs', error=%d", Path, execResult);
99
src/windows/service/exe/WslCoreInstance.cpp
+8 -5
@@ -344,7 +344,8 @@ void WslCoreInstance::UpdateTimezone()
344 wsl::windows::common::helpers::GenerateTimezoneUpdateMessage(wsl::windows::common::helpers::GetLinuxTimezone(m_userToken.get()));
345
346 auto lock = m_initChannel->Lock();
347 - m_initChannel->GetChannel().SendMessage<LX_INIT_TIMEZONE_INFORMATION>(gsl::make_span(message));
347 + auto transaction = m_initChannel->GetChannel().StartTransaction();
348 + transaction.Send<LX_INIT_TIMEZONE_INFORMATION>(gsl::make_span(message));
349 }
350
351 ULONG64 WslCoreInstance::GetLifetimeManagerId() const
@@ -389,11 +390,12 @@ void WslCoreInstance::Initialize()
390 auto config = wsl::windows::common::helpers::GenerateConfigurationMessage(
391 m_configuration.Name, fixedDrives, m_defaultUid, timezone, {}, m_featureFlags, drvfsMount);
392
392 - m_initChannel->GetChannel().SendMessage<LX_INIT_CONFIGURATION_INFORMATION>(gsl::span(config));
393 + auto transaction = m_initChannel->GetChannel().StartTransaction();
394 + transaction.Send<LX_INIT_CONFIGURATION_INFORMATION>(gsl::span(config));
395
396 // Init replies with information about the distribution.
397 gsl::span<gsl::byte> span;
396 - const auto& response = m_initChannel->GetChannel().ReceiveMessage<LX_INIT_CONFIGURATION_INFORMATION_RESPONSE>(&span);
398 + const auto& response = transaction.Receive<LX_INIT_CONFIGURATION_INFORMATION_RESPONSE>(&span);
399 m_defaultUid = response.DefaultUid;
400 m_plan9Port = response.Plan9Port;
401 m_distributionInfo.PidNamespace = response.PidNamespace;
@@ -473,8 +475,9 @@ bool WslCoreInstance::RequestStop(_In_ bool Force)
475 terminateMessage.Header.MessageSize = sizeof(terminateMessage);
476 terminateMessage.Force = Force;
477
476 - m_initChannel->GetChannel().SendMessage(terminateMessage);
477 - auto [message, span] = m_initChannel->GetChannel().ReceiveMessageOrClosed<RESULT_MESSAGE<bool>>(m_socketTimeout);
478 + auto transaction = m_initChannel->GetChannel().StartTransaction();
479 + transaction.Send(terminateMessage);
480 + auto [message, span] = transaction.ReceiveOrClosed<RESULT_MESSAGE<bool>>(m_socketTimeout);
481 if (message)
482 {
483 shutdown = message->Result;
src/windows/service/exe/WslCoreVm.cpp
+22 -12
@@ -527,7 +527,8 @@ void WslCoreVm::Initialize(const GUID& VmId, const wil::shared_handle& UserToken
527 message.WriteString(message->KernelModulesListOffset, m_vmConfig.KernelModulesList);
528 message->DnsTunnelingIpAddress = m_vmConfig.DnsTunnelingIpAddress.value_or(0);
529
530 - m_miniInitChannel.SendMessage<LX_MINI_INIT_EARLY_CONFIG_MESSAGE>(message.Span());
530 + auto transaction = m_miniInitChannel.StartTransaction();
531 + transaction.Send<LX_MINI_INIT_EARLY_CONFIG_MESSAGE>(message.Span());
532
533 {
534 ExecutionContext context(Context::ConfigureNetworking);
@@ -1098,7 +1099,8 @@ void WslCoreVm::CollectCrashDumps(wil::unique_socket&& listenSocket) const
1099
1100 auto channel = wsl::shared::SocketChannel{std::move(socket.value()), "crash_dump", m_terminatingEvent.get()};
1101
1101 - const auto& message = channel.ReceiveMessage<LX_PROCESS_CRASH>();
1102 + auto transaction = channel.ReceiveTransaction();
1103 + const auto& message = transaction.Receive<LX_PROCESS_CRASH>();
1104 const char* process = reinterpret_cast<const char*>(&message.Buffer);
1105
1106 constexpr auto dumpExtension = ".dmp";
@@ -1146,7 +1148,7 @@ void WslCoreVm::CollectCrashDumps(wil::unique_socket&& listenSocket) const
1148 wil::unique_hfile file{CreateFileW(fullPath.c_str(), GENERIC_WRITE, 0, nullptr, CREATE_NEW, FILE_ATTRIBUTE_TEMPORARY, nullptr)};
1149 THROW_LAST_ERROR_IF(!file);
1150
1149 - channel.SendResultMessage<std::int32_t>(0);
1151 + transaction.SendResultMessage<std::int32_t>(0);
1152
1153 wsl::windows::common::relay::InterruptableRelay(reinterpret_cast<HANDLE>(channel.Socket()), file.get(), nullptr);
1154 }
@@ -1206,7 +1208,8 @@ std::shared_ptr<LxssRunningInstance> WslCoreVm::CreateInstance(
1208 message.WriteString(message->SharedMemoryRootOffset, sharedMemoryRoot);
1209 message.WriteString(message->InstallPathOffset, installPath);
1210 message.WriteString(message->UserProfileOffset, userProfile);
1209 - m_miniInitChannel.SendMessage<LX_MINI_INIT_MESSAGE>(message.Span());
1211 + auto transaction = m_miniInitChannel.StartTransaction();
1212 + transaction.Send<LX_MINI_INIT_MESSAGE>(message.Span());
1213
1214 return CreateInstanceInternal(
1215 InstanceId, Configuration, ReceiveTimeout, DefaultUid, ClientLifetimeId, WI_IsFlagSet(flags, LxMiniInitMessageFlagLaunchSystemDistro), ConnectPort);
@@ -1844,7 +1847,8 @@ void WslCoreVm::InitializeGuest()
1847 }
1848
1849 // Send the message.
1847 - m_miniInitChannel.SendMessage<LX_MINI_INIT_CONFIG_MESSAGE>(message.Span());
1850 + auto transaction = m_miniInitChannel.StartTransaction();
1851 + transaction.Send<LX_MINI_INIT_CONFIG_MESSAGE>(message.Span());
1852
1853 // If port tracker or localhost relay are enabled, establish a connection with the guest and start processing messages.
1854 switch (message->NetworkingConfiguration.PortTrackerType)
@@ -1978,7 +1982,8 @@ WslCoreVm::DiskMountResult WslCoreVm::MountDiskLockHeld(
1982 message.WriteString(message->OptionsOffset, Options);
1983
1984 // Send the message.
1981 - m_miniInitChannel.SendMessage<LX_MINI_INIT_MOUNT_MESSAGE>(message.Span());
1985 + auto transaction = m_miniInitChannel.StartTransaction();
1986 + transaction.Send<LX_MINI_INIT_MOUNT_MESSAGE>(message.Span());
1987
1988 // Accept a connection from mini_init
1989 wsl::shared::SocketChannel channel{AcceptConnection(m_vmConfig.KernelBootTimeout), "MountResult", m_terminatingEvent.get()};
@@ -2106,7 +2111,8 @@ void WslCoreVm::WaitForPmemDeviceInVm(_In_ ULONG PmemId)
2111 {
2112 auto lock = m_lock.lock_exclusive();
2113
2109 - m_miniInitChannel.SendMessage(message);
2114 + auto transaction = m_miniInitChannel.StartTransaction();
2115 + transaction.Send(message);
2116 channel = {
2117 AcceptConnection(m_vmConfig.KernelBootTimeout),
2118 "WaitForPmem",
@@ -2415,7 +2421,8 @@ void WslCoreVm::ResizeDistribution(_In_ ULONG Lun, _In_ HANDLE OutputHandle, _In
2421 message.ScsiLun = Lun;
2422 message.NewSize = NewSize;
2423
2418 - m_miniInitChannel.SendMessage(message);
2424 + auto transaction = m_miniInitChannel.StartTransaction();
2425 + transaction.Send(message);
2426
2427 wsl::shared::SocketChannel channel{AcceptConnection(m_vmConfig.KernelBootTimeout), "ResizeDistribution", m_terminatingEvent.get()};
2428 auto outputChannel = AcceptConnection(m_vmConfig.KernelBootTimeout);
@@ -2492,7 +2499,8 @@ std::pair<int, LX_MINI_MOUNT_STEP> WslCoreVm::UnmountDisk(_In_ const AttachedDis
2499 message.Header.MessageSize = sizeof(message);
2500 message.ScsiLun = State.Lun;
2501
2495 - m_miniInitChannel.SendMessage(message);
2502 + auto transaction = m_miniInitChannel.StartTransaction();
2503 + transaction.Send(message);
2504
2505 // Accept a connection from mini_init.
2506 wsl::shared::SocketChannel channel{AcceptConnection(m_vmConfig.KernelBootTimeout), "MountResult", m_terminatingEvent.get()};
@@ -2507,7 +2515,8 @@ std::pair<int, LX_MINI_MOUNT_STEP> WslCoreVm::UnmountVolume(_In_ const AttachedD
2515 message.WriteString(Name);
2516
2517 // Send the message.
2510 - m_miniInitChannel.SendMessage<LX_MINI_INIT_UNMOUNT_MESSAGE>(message.Span());
2518 + auto transaction = m_miniInitChannel.StartTransaction();
2519 + transaction.Send<LX_MINI_INIT_UNMOUNT_MESSAGE>(message.Span());
2520
2521 // Accept a connection from mini_init.
2522 wsl::shared::SocketChannel channel{AcceptConnection(m_vmConfig.KernelBootTimeout), "MountResult", m_terminatingEvent.get()};
@@ -2573,7 +2582,8 @@ try
2582 {
2583 wsl::windows::common::wslutil::SetThreadDescription(L"VirtioFs - Request");
2584
2576 - auto [message, span] = channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
2585 + auto transaction = channel.ReceiveTransaction();
2586 + auto [message, span] = transaction.ReceiveOrClosed<MESSAGE_HEADER>();
2587 if (message == nullptr)
2588 {
2589 return;
@@ -2587,7 +2597,7 @@ try
2597 response.WriteString(response->TagOffset, tag);
2598 response.WriteString(response->SourceOffset, source);
2599
2590 - channel.SendMessage<LX_INIT_ADD_VIRTIOFS_SHARE_RESPONSE_MESSAGE>(response.Span());
2600 + transaction.Send<LX_INIT_ADD_VIRTIOFS_SHARE_RESPONSE_MESSAGE>(response.Span());
2601 };
2602
2603 if (message->MessageType == LxInitMessageAddVirtioFsDevice)
src/windows/wslrelay/localhost.cpp
+3 -2
@@ -81,7 +81,8 @@ void wsl::windows::wslrelay::localhost::RelayWorker(_In_ wsl::shared::SocketChan
81
82 for (;;)
83 {
84 - auto [Message, Span] = Channel.ReceiveMessageOrClosed<MESSAGE_HEADER>();
84 + auto Transaction = Channel.ReceiveTransaction();
85 + auto [Message, Span] = Transaction.ReceiveOrClosed<MESSAGE_HEADER>();
86 if (Message == nullptr)
87 {
88 break;
@@ -151,7 +152,7 @@ void wsl::windows::wslrelay::localhost::RelayWorker(_In_ wsl::shared::SocketChan
152 }
153 }
154
154 - Channel.SendMessage(Response);
155 + Transaction.Send(Response);
156 break;
157 }
158