Skip to content

Commit 5795c5d

Browse files
mikhail-dclmikhail-dcl
committed
Distribute bots across multiple processes to reduce messages loss through a single socket
Signed-off-by: Mikhail Agapov <mikhail.agapov@decentraland.org>
1 parent b06d956 commit 5795c5d

7 files changed

Lines changed: 229 additions & 20 deletions

File tree

‎src/DCLPulseTestClient/BotBehaviorSettings.cs‎

Lines changed: 31 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,20 +17,44 @@ public class BotBehaviorSettings
1717
public float JumpVelocity => MathF.Sqrt(2f * Gravity * JumpHeight);
1818

1919
public static BotBehaviorSettings Load()
20+
{
21+
JsonElement? section = LoadSection("BotBehavior");
22+
23+
if (section is null)
24+
return new BotBehaviorSettings();
25+
26+
return JsonSerializer.Deserialize(section.Value.GetRawText(), BotBehaviorJsonContext.Default.BotBehaviorSettings)
27+
?? new BotBehaviorSettings();
28+
}
29+
30+
public static int LoadBotsPerProcess()
31+
{
32+
JsonElement? root = LoadRoot();
33+
34+
if (root is null || !root.Value.TryGetProperty("BotsPerProcess", out JsonElement value))
35+
return 10;
36+
37+
return value.GetInt32();
38+
}
39+
40+
private static JsonElement? LoadRoot()
2041
{
2142
string path = Path.Combine(AppContext.BaseDirectory, "appsettings.json");
2243

2344
if (!File.Exists(path))
24-
return new BotBehaviorSettings();
45+
return null;
2546

26-
string json = File.ReadAllText(path);
27-
var doc = JsonDocument.Parse(json);
47+
return JsonDocument.Parse(File.ReadAllText(path)).RootElement;
48+
}
2849

29-
if (!doc.RootElement.TryGetProperty("BotBehavior", out JsonElement section))
30-
return new BotBehaviorSettings();
50+
private static JsonElement? LoadSection(string name)
51+
{
52+
JsonElement? root = LoadRoot();
3153

32-
return JsonSerializer.Deserialize(section.GetRawText(), BotBehaviorJsonContext.Default.BotBehaviorSettings)
33-
?? new BotBehaviorSettings();
54+
if (root is null || !root.Value.TryGetProperty(name, out JsonElement section))
55+
return null;
56+
57+
return section;
3458
}
3559
}
3660

‎src/DCLPulseTestClient/BotSession.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,5 +22,6 @@ public class BotSession
2222
public int JumpCount { get; set; }
2323
public uint LastFrameTick { get; set; }
2424
public Dictionary<uint, uint> KnownSeqBySubject { get; } = new ();
25+
public HashSet<uint> PendingResyncs { get; } = new ();
2526
public Dictionary<uint, Web3Address> PeerAddresses { get; } = new ();
2627
}

‎src/DCLPulseTestClient/ClientOptions.cs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@ public class ClientOptions
1212
public float PositionZ { get; init; } = 5f;
1313
public float SpawnRadius { get; init; } = 10f;
1414
public float DispersionRadius { get; init; } = 20f;
15+
public int BotOffset { get; init; }
16+
public int TotalBotCount { get; init; }
1517

1618
public static ClientOptions FromArgs(string[] args)
1719
{
@@ -30,6 +32,8 @@ string Arg(string name, string fallback) =>
3032
PositionZ = float.Parse(Arg("pos-z", "5")),
3133
SpawnRadius = float.Parse(Arg("spawn-radius", "10")),
3234
DispersionRadius = float.Parse(Arg("dispersion-radius", "20")),
35+
BotOffset = int.Parse(Arg("bot-offset", "0")),
36+
TotalBotCount = int.Parse(Arg("total-bot-count", "0")),
3337
};
3438
}
3539
}

‎src/DCLPulseTestClient/ProcessOrchestrator.cs‎

Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
using System.Diagnostics;
2+
3+
namespace PulseTestClient;
4+
5+
public static class ProcessOrchestrator
6+
{
7+
public static async Task<int> RunAsync(ClientOptions options, int botsPerProcess, CancellationToken ct)
8+
{
9+
int totalBots = options.BotCount;
10+
int processCount = (totalBots + botsPerProcess - 1) / botsPerProcess;
11+
12+
Console.WriteLine($"Spawning {totalBots} bots across {processCount} processes ({botsPerProcess} per process)..");
13+
14+
// Account creation must be sequential across all bots
15+
for (var i = 0; i < totalBots; i++)
16+
{
17+
var accountName = $"{options.AccountPrefix}-{i}";
18+
Console.WriteLine($"[{accountName}] Ensuring account exists..");
19+
await MetaForge.RunCommandAsync($"account create {accountName} --skip-update-check --skip-auto-login", ct);
20+
}
21+
22+
var processes = new List<Process>();
23+
24+
for (var p = 0; p < processCount; p++)
25+
{
26+
int offset = p * botsPerProcess;
27+
int count = Math.Min(botsPerProcess, totalBots - offset);
28+
29+
string childArgs = BuildChildArgs(options, offset, count, totalBots);
30+
31+
var process = new Process
32+
{
33+
StartInfo = new ProcessStartInfo
34+
{
35+
FileName = Environment.ProcessPath!,
36+
Arguments = childArgs,
37+
UseShellExecute = false,
38+
RedirectStandardOutput = true,
39+
RedirectStandardError = true,
40+
},
41+
};
42+
43+
process.OutputDataReceived += (_, e) =>
44+
{
45+
if (e.Data != null) Console.WriteLine(e.Data);
46+
};
47+
48+
process.ErrorDataReceived += (_, e) =>
49+
{
50+
if (e.Data != null) Console.Error.WriteLine(e.Data);
51+
};
52+
53+
process.Start();
54+
process.BeginOutputReadLine();
55+
process.BeginErrorReadLine();
56+
processes.Add(process);
57+
58+
Console.WriteLine($"[orchestrator] Process {p} started (PID {process.Id}): bots {offset}..{offset + count - 1}");
59+
}
60+
61+
Console.WriteLine($"[orchestrator] All {processCount} processes running. Press q+Enter or Ctrl+C to stop.");
62+
63+
// Watch for quit
64+
_ = Task.Run(() =>
65+
{
66+
while (!ct.IsCancellationRequested)
67+
{
68+
string? line = Console.ReadLine();
69+
70+
if (line is "q" or "Q" or "quit")
71+
{
72+
Console.WriteLine("[orchestrator] Quit requested, stopping child processes..");
73+
SignalStop();
74+
}
75+
}
76+
});
77+
78+
// When parent is cancelled, signal children
79+
ct.Register(SignalStop);
80+
81+
// Wait for all children
82+
Task[] tasks = processes.Select(p => p.WaitForExitAsync(CancellationToken.None)).ToArray();
83+
await Task.WhenAll(tasks);
84+
85+
// Clean up the stop file after all children have exited
86+
CleanupStopFile();
87+
88+
int failed = processes.Count(p => p.ExitCode != 0);
89+
90+
if (failed > 0)
91+
Console.WriteLine($"[orchestrator] {failed}/{processCount} processes exited with errors.");
92+
else
93+
Console.WriteLine($"[orchestrator] All {processCount} processes exited cleanly.");
94+
95+
return failed > 0 ? 1 : 0;
96+
}
97+
98+
private static string BuildChildArgs(ClientOptions options, int offset, int count, int totalBots)
99+
{
100+
var parts = new List<string>
101+
{
102+
$"--account={options.AccountPrefix}",
103+
$"--bot-count={count}",
104+
$"--bot-offset={offset}",
105+
$"--total-bot-count={totalBots}",
106+
$"--ip={options.ServerIp}",
107+
$"--port={options.ServerPort}",
108+
$"--pos-x={options.PositionX}",
109+
$"--pos-y={options.PositionY}",
110+
$"--pos-z={options.PositionZ}",
111+
$"--spawn-radius={options.SpawnRadius}",
112+
$"--dispersion-radius={options.DispersionRadius}",
113+
$"--rotate-speed={options.RotateSpeed}",
114+
};
115+
116+
return string.Join(' ', parts);
117+
}
118+
119+
private static void SignalStop()
120+
{
121+
string stopFile = Path.Combine(Path.GetTempPath(), "dcl-pulse-test-client.stop");
122+
123+
try { File.WriteAllText(stopFile, ""); }
124+
catch
125+
{ /* best effort */
126+
}
127+
}
128+
129+
private static void CleanupStopFile()
130+
{
131+
string stopFile = Path.Combine(Path.GetTempPath(), "dcl-pulse-test-client.stop");
132+
133+
try { File.Delete(stopFile); }
134+
catch
135+
{ /* best effort */
136+
}
137+
}
138+
}

‎src/DCLPulseTestClient/Program.cs‎

Lines changed: 45 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,27 @@
99

1010
var options = ClientOptions.FromArgs(args);
1111
var behaviorSettings = BotBehaviorSettings.Load();
12+
int botsPerProcess = BotBehaviorSettings.LoadBotsPerProcess();
13+
14+
// If bot count exceeds per-process limit and we're not already a child worker, orchestrate
15+
bool isWorker = options.BotOffset > 0 || options.TotalBotCount > 0;
16+
17+
if (!isWorker && options.BotCount > botsPerProcess)
18+
{
19+
using var orchestratorCts = new CancellationTokenSource();
20+
21+
Console.CancelKeyPress += (_, e) =>
22+
{
23+
e.Cancel = true;
24+
orchestratorCts.Cancel();
25+
};
26+
27+
return await ProcessOrchestrator.RunAsync(options, botsPerProcess, orchestratorCts.Token);
28+
}
29+
30+
// --- Worker mode: run bots in this process ---
31+
32+
int totalBotCount = options.TotalBotCount > 0 ? options.TotalBotCount : options.BotCount;
1233

1334
IAuthenticator authenticator = new MetaForgeAuthenticator();
1435
using var profileGateway = new CatalystProfileGateway();
@@ -28,18 +49,28 @@
2849
var sharedTransport = new ENetTransport(new ENetTransportOptions { PeerLimit = options.BotCount });
2950
sharedTransport.Initialize();
3051

31-
// Account creation must be sequential — MetaForge's account store isn't safe for concurrent writes
52+
// When running as a worker child, accounts are pre-created by the orchestrator
3253
var accountNames = new string[options.BotCount];
3354

34-
for (var i = 0; i < options.BotCount; i++)
55+
if (!isWorker)
3556
{
36-
accountNames[i] = options.BotCount == 1 ? options.AccountPrefix : $"{options.AccountPrefix}-{i}";
37-
Console.WriteLine($"[{accountNames[i]}] Ensuring account exists..");
38-
await MetaForge.RunCommandAsync($"account create {accountNames[i]} --skip-update-check --skip-auto-login", lifeCycleCts.Token);
57+
for (var i = 0; i < options.BotCount; i++)
58+
{
59+
accountNames[i] = options.BotCount == 1 ? options.AccountPrefix : $"{options.AccountPrefix}-{i}";
60+
Console.WriteLine($"[{accountNames[i]}] Ensuring account exists..");
61+
await MetaForge.RunCommandAsync($"account create {accountNames[i]} --skip-update-check --skip-auto-login", lifeCycleCts.Token);
62+
}
63+
}
64+
else
65+
{
66+
for (var i = 0; i < options.BotCount; i++)
67+
accountNames[i] = $"{options.AccountPrefix}-{options.BotOffset + i}";
3968
}
4069

4170
// Auth, profile fetch, and connect can run in parallel
42-
Task<BotSession>[] sessionTasks = Enumerable.Range(0, options.BotCount).Select(i => CreateBotSessionAsync(i, accountNames[i])).ToArray();
71+
Task<BotSession>[] sessionTasks = Enumerable.Range(0, options.BotCount)
72+
.Select(i => CreateBotSessionAsync(i, options.BotOffset + i, totalBotCount, accountNames[i]))
73+
.ToArray();
4374
var sessions = (await Task.WhenAll(sessionTasks)).ToList();
4475

4576
Console.WriteLine(options.BotCount == 1
@@ -62,11 +93,11 @@
6293

6394
await Task.Delay(200);
6495
sharedTransport.Dispose();
65-
return;
96+
return 0;
6697

6798
// --- Bot session factory ---
6899

69-
async Task<BotSession> CreateBotSessionAsync(int index, string accountName)
100+
async Task<BotSession> CreateBotSessionAsync(int localIndex, int globalIndex, int total, string accountName)
70101
{
71102
Console.WriteLine($"[{accountName}] Signing auth chain..");
72103
LoginResult login = await authenticator.LoginAsync(accountName, lifeCycleCts.Token);
@@ -78,8 +109,8 @@ async Task<BotSession> CreateBotSessionAsync(int index, string accountName)
78109
var botTransport = new BotTransport(sharedTransport, pipe);
79110
var service = new PulseMultiplayerService(botTransport, pipe);
80111

81-
float angle = options.BotCount > 1 ? 2f * MathF.PI * index / options.BotCount : 0f;
82-
float botSpawnOffset = options.BotCount > 1 ? options.SpawnRadius : 0f;
112+
float angle = total > 1 ? 2f * MathF.PI * globalIndex / total : 0f;
113+
float botSpawnOffset = total > 1 ? options.SpawnRadius : 0f;
83114

84115
var spawnOrigin = new Vector3(options.PositionX, options.PositionY, options.PositionZ);
85116
var position = new Vector3(
@@ -127,15 +158,17 @@ async Task<BotSession> CreateBotSessionAsync(int index, string accountName)
127158
void WatchForStopFile(CancellationTokenSource cts)
128159
{
129160
string stopFile = Path.Combine(Path.GetTempPath(), "dcl-pulse-test-client.stop");
130-
File.Delete(stopFile);
161+
162+
// Only the top-level process (not a worker child) cleans up the stop file on startup
163+
if (!isWorker)
164+
File.Delete(stopFile);
131165

132166
_ = Task.Run(async () =>
133167
{
134168
while (!cts.Token.IsCancellationRequested)
135169
{
136170
if (File.Exists(stopFile))
137171
{
138-
File.Delete(stopFile);
139172
Console.WriteLine("Stop file detected, shutting down..");
140173
await cts.CancelAsync();
141174
break;

‎src/DCLPulseTestClient/ServerEventHandler.cs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,16 @@ private static async Task OnDeltaState(BotSession bot, CancellationToken ct)
2424
{
2525
uint subjectId = delta.SubjectId;
2626

27+
// Already waiting for a full state — drop silently
28+
if (bot.PendingResyncs.Contains(subjectId))
29+
continue;
30+
2731
if (bot.KnownSeqBySubject.TryGetValue(subjectId, out uint lastSeq) && delta.BaselineSeq != lastSeq)
2832
{
2933
Console.WriteLine($"[{bot.AccountName}] Seq gap for subject {subjectId}: expected {lastSeq}, got {delta.BaselineSeq}. Requesting resync.");
3034

35+
bot.PendingResyncs.Add(subjectId);
36+
3137
bot.Pipe.Send(new MessagePipe.OutgoingMessage(new ClientMessage
3238
{
3339
Resync = new ResyncRequest { SubjectId = subjectId, KnownSeq = lastSeq },
@@ -46,6 +52,7 @@ private static async Task OnFullState(BotSession bot, CancellationToken ct)
4652
ServerMessage.MessageOneofCase.PlayerStateFull, ct))
4753
{
4854
bot.KnownSeqBySubject[full.SubjectId] = full.Sequence;
55+
bot.PendingResyncs.Remove(full.SubjectId);
4956
Console.WriteLine($"[{bot.AccountName}] Full state for subject {full.SubjectId}, seq={full.Sequence}");
5057
}
5158
}
@@ -68,6 +75,7 @@ private static async Task OnPeerLeft(BotSession bot, CancellationToken ct)
6875
{
6976
bot.PeerAddresses.TryGetValue(left.SubjectId, out Web3Address address);
7077
bot.KnownSeqBySubject.Remove(left.SubjectId);
78+
bot.PendingResyncs.Remove(left.SubjectId);
7179
bot.PeerAddresses.Remove(left.SubjectId);
7280
Console.WriteLine($"[{bot.AccountName}] Player left: {address}");
7381
}

‎src/DCLPulseTestClient/appsettings.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,5 +5,6 @@
55
"JumpMaxInterval": 10,
66
"JumpHeight": 2.5,
77
"Gravity": 15
8-
}
8+
},
9+
"BotsPerProcess": 5
910
}

0 commit comments

Comments
 (0)