[wasm][debugger] Make DevToolsProxy's pending_ops thread-safe

radical · web-flow · commit fdc0d1602087 · 2022-02-22T22:40:02.000Z
Currently, `pending_ops` can get written by different threads at the
same time, and also read in parallel. This causes tests to randomly fail
with:

```
DevToolsProxy::Run: Exception System.ArgumentException: The tasks argument included a null value. (Parameter 'tasks')
    at System.Threading.Tasks.Task.WhenAny(Task[] tasks)
    at Microsoft.WebAssembly.Diagnostics.DevToolsProxy.Run(Uri browserUri, WebSocket ideSocket) in /_/src/mono/wasm/debugger/BrowserDebugProxy/DevToolsProxy.cs:line 269
```

Instead, we use `Channel&lt;T&gt;` to add the ops, and then read those in
the main loop, and add to the *local* `pending_ops` list.
diff --git a/src/mono/wasm/debugger/BrowserDebugProxy/DevToolsProxy.cs b/src/mono/wasm/debugger/BrowserDebugProxy/DevToolsProxy.cs
@@ -8,6 +8,7 @@
 using System.Net.WebSockets;
 using System.Text;
 using System.Threading;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
@@ -24,14 +25,19 @@ internal class DevToolsProxy
         private ClientWebSocket browser;
         private WebSocket ide;
         private int next_cmd_id;
-        private List<Task> pending_ops = new List<Task>();
+        private readonly ChannelWriter<Task> _channelWriter;
+        private readonly ChannelReader<Task> _channelReader;
         private List<DevToolsQueue> queues = new List<DevToolsQueue>();
 
         protected readonly ILogger logger;
 
         public DevToolsProxy(ILoggerFactory loggerFactory)
         {
             logger = loggerFactory.CreateLogger<DevToolsProxy>();
+
+            var channel = Channel.CreateUnbounded<Task>(new UnboundedChannelOptions { SingleReader = true });
+            _channelWriter = channel.Writer;
+            _channelReader = channel.Reader;
         }
 
         protected virtual Task<bool> AcceptEvent(SessionId sessionId, string method, JObject args, CancellationToken token)
@@ -93,7 +99,7 @@ private DevToolsQueue GetQueueForTask(Task task)
             return queues.FirstOrDefault(q => q.CurrentSend == task);
         }
 
-        private void Send(WebSocket to, JObject o, CancellationToken token)
+        private async Task Send(WebSocket to, JObject o, CancellationToken token)
         {
             string sender = browser == to ? "Send-browser" : "Send-ide";
 
@@ -105,7 +111,7 @@ private void Send(WebSocket to, JObject o, CancellationToken token)
 
             Task task = queue.Send(bytes, token);
             if (task != null)
-                pending_ops.Add(task);
+                await _channelWriter.WriteAsync(task, token);
         }
 
         private async Task OnEvent(SessionId sessionId, string method, JObject args, CancellationToken token)
@@ -115,7 +121,7 @@ private async Task OnEvent(SessionId sessionId, string method, JObject args, Can
                 if (!await AcceptEvent(sessionId, method, args, token))
                 {
                     //logger.LogDebug ("proxy browser: {0}::{1}",method, args);
-                    SendEventInternal(sessionId, method, args, token);
+                    await SendEventInternal(sessionId, method, args, token);
                 }
             }
             catch (Exception e)
@@ -131,7 +137,7 @@ private async Task OnCommand(MessageId id, string method, JObject args, Cancella
                 if (!await AcceptCommand(id, method, args, token))
                 {
                     Result res = await SendCommandInternal(id, method, args, token);
-                    SendResponseInternal(id, res, token);
+                    await SendResponseInternal(id, res, token);
                 }
             }
             catch (Exception e)
@@ -152,31 +158,38 @@ private void OnResponse(MessageId id, Result result)
             logger.LogError("Cannot respond to command: {id} with result: {result} - command is not pending", id, result);
         }
 
-        private void ProcessBrowserMessage(string msg, CancellationToken token)
+        private Task ProcessBrowserMessage(string msg, CancellationToken token)
         {
             var res = JObject.Parse(msg);
 
             //if (method != "Debugger.scriptParsed" && method != "Runtime.consoleAPICalled")
             Log("protocol", $"browser: {msg}");
 
             if (res["id"] == null)
-                pending_ops.Add(OnEvent(res.ToObject<SessionId>(), res["method"].Value<string>(), res["params"] as JObject, token));
+            {
+                return OnEvent(res.ToObject<SessionId>(), res["method"].Value<string>(), res["params"] as JObject, token);
+            }
             else
+            {
                 OnResponse(res.ToObject<MessageId>(), Result.FromJson(res));
+                return null;
+            }
         }
 
-        private void ProcessIdeMessage(string msg, CancellationToken token)
+        private Task ProcessIdeMessage(string msg, CancellationToken token)
         {
             Log("protocol", $"ide: {msg}");
             if (!string.IsNullOrEmpty(msg))
             {
                 var res = JObject.Parse(msg);
                 var id = res.ToObject<MessageId>();
-                pending_ops.Add(OnCommand(
+                return OnCommand(
                     id,
                     res["method"].Value<string>(),
-                    res["params"] as JObject, token));
+                    res["params"] as JObject, token);
             }
+
+            return null;
         }
 
         internal async Task<Result> SendCommand(SessionId id, string method, JObject args, CancellationToken token)
@@ -185,7 +198,7 @@ internal async Task<Result> SendCommand(SessionId id, string method, JObject arg
             return await SendCommandInternal(id, method, args, token);
         }
 
-        private Task<Result> SendCommandInternal(SessionId sessionId, string method, JObject args, CancellationToken token)
+        private async Task<Result> SendCommandInternal(SessionId sessionId, string method, JObject args, CancellationToken token)
         {
             int id = Interlocked.Increment(ref next_cmd_id);
 
@@ -203,17 +216,17 @@ private Task<Result> SendCommandInternal(SessionId sessionId, string method, JOb
             //Log ("verbose", $"add cmd id {sessionId}-{id}");
             pending_cmds[msgId] = tcs;
 
-            Send(this.browser, o, token);
-            return tcs.Task;
+            await Send(browser, o, token);
+            return await tcs.Task;
         }
 
-        public void SendEvent(SessionId sessionId, string method, JObject args, CancellationToken token)
+        public Task SendEvent(SessionId sessionId, string method, JObject args, CancellationToken token)
         {
             //Log ("verbose", $"sending event {method}: {args}");
-            SendEventInternal(sessionId, method, args, token);
+            return SendEventInternal(sessionId, method, args, token);
         }
 
-        private void SendEventInternal(SessionId sessionId, string method, JObject args, CancellationToken token)
+        private Task SendEventInternal(SessionId sessionId, string method, JObject args, CancellationToken token)
         {
             var o = JObject.FromObject(new
             {
@@ -223,21 +236,21 @@ private void SendEventInternal(SessionId sessionId, string method, JObject args,
             if (sessionId.sessionId != null)
                 o["sessionId"] = sessionId.sessionId;
 
-            Send(this.ide, o, token);
+            return Send(ide, o, token);
         }
 
         internal void SendResponse(MessageId id, Result result, CancellationToken token)
         {
             SendResponseInternal(id, result, token);
         }
 
-        private void SendResponseInternal(MessageId id, Result result, CancellationToken token)
+        private Task SendResponseInternal(MessageId id, Result result, CancellationToken token)
         {
             JObject o = result.ToJObject(id);
             if (!result.IsOk)
                 logger.LogError($"sending error response for id: {id} -> {result}");
 
-            Send(this.ide, o, token);
+            return Send(this.ide, o, token);
         }
 
         // , HttpContext context)
@@ -257,10 +270,14 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)
                     Log("verbose", $"DevToolsProxy: Client connected on {browserUri}");
                     var x = new CancellationTokenSource();
 
+                    List<Task> pending_ops = new();
+
                     pending_ops.Add(ReadOne(browser, x.Token));
                     pending_ops.Add(ReadOne(ide, x.Token));
                     pending_ops.Add(side_exception.Task);
                     pending_ops.Add(client_initiated_close.Task);
+                    Task<bool> readerTask = _channelReader.WaitToReadAsync(x.Token).AsTask();
+                    pending_ops.Add(readerTask);
 
                     try
                     {
@@ -277,14 +294,26 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)
                                 break;
                             }
 
+                            if (readerTask.IsCompleted)
+                            {
+                                while (_channelReader.TryRead(out Task newTask))
+                                {
+                                    pending_ops.Add(newTask);
+                                }
+
+                                pending_ops[4] = _channelReader.WaitToReadAsync(x.Token).AsTask();
+                            }
+
                             //logger.LogTrace ("pump {0} {1}", task, pending_ops.IndexOf (task));
                             if (completedTask == pending_ops[0])
                             {
                                 string msg = ((Task<string>)completedTask).Result;
                                 if (msg != null)
                                 {
                                     pending_ops[0] = ReadOne(browser, x.Token); //queue next read
-                                    ProcessBrowserMessage(msg, x.Token);
+                                    Task newTask = ProcessBrowserMessage(msg, x.Token);
+                                    if (newTask != null)
+                                        pending_ops.Add(newTask);
                                 }
                             }
                             else if (completedTask == pending_ops[1])
@@ -293,7 +322,9 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)
                                 if (msg != null)
                                 {
                                     pending_ops[1] = ReadOne(ide, x.Token); //queue next read
-                                    ProcessIdeMessage(msg, x.Token);
+                                    Task newTask = ProcessIdeMessage(msg, x.Token);
+                                    if (newTask != null)
+                                        pending_ops.Add(newTask);
                                 }
                             }
                             else if (completedTask == pending_ops[2])
@@ -313,10 +344,13 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)
                                 }
                             }
                         }
+
+                        _channelWriter.Complete();
                     }
                     catch (Exception e)
                     {
                         Log("error", $"DevToolsProxy::Run: Exception {e}");
+                        _channelWriter.Complete(e);
                         //throw;
                     }
                     finally
diff --git a/src/mono/wasm/debugger/BrowserDebugProxy/MonoProxy.cs b/src/mono/wasm/debugger/BrowserDebugProxy/MonoProxy.cs
@@ -139,7 +139,7 @@ protected override async Task<bool> AcceptEvent(SessionId sessionId, string meth
 
                 case "Runtime.executionContextCreated":
                     {
-                        SendEvent(sessionId, method, args, token);
+                        await SendEvent(sessionId, method, args, token);
                         JToken ctx = args?["context"];
                         var aux_data = ctx?["auxData"] as JObject;
                         int id = ctx["id"].Value<int>();
@@ -986,7 +986,7 @@ private async Task<bool> SendCallStack(SessionId sessionId, ExecutionContext con
                 await SendCommand(sessionId, "Debugger.resume", new JObject(), token);
                 return true;
             }
-            SendEvent(sessionId, "Debugger.paused", o, token);
+            await SendEvent(sessionId, "Debugger.paused", o, token);
 
             return true;
 
@@ -1103,7 +1103,7 @@ internal async Task<MethodInfo> LoadSymbolsOnDemand(AssemblyInfo asm, int method
                     foreach (SourceFile source in asm.Sources)
                     {
                         var scriptSource = JObject.FromObject(source.ToScriptSource(context.Id, context.AuxData));
-                        SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);
+                        await SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);
                     }
                     return asm.GetMethodByToken(method_token);
                 }
@@ -1333,7 +1333,7 @@ private async Task OnSourceFileAdded(SessionId sessionId, SourceFile source, Exe
         {
             JObject scriptSource = JObject.FromObject(source.ToScriptSource(context.Id, context.AuxData));
             Log("debug", $"sending {source.Url} {context.Id} {sessionId.sessionId}");
-            SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);
+            await SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);
 
             foreach (var req in context.BreakpointRequests.Values)
             {
@@ -1412,7 +1412,7 @@ private async Task<DebugStore> RuntimeReady(SessionId sessionId, CancellationTok
 
             DebugStore store = await LoadStore(sessionId, token);
             context.ready.SetResult(store);
-            SendEvent(sessionId, "Mono.runtimeReady", new JObject(), token);
+            await SendEvent(sessionId, "Mono.runtimeReady", new JObject(), token);
             context.SdbAgent.ResetStore(store);
             return store;
         }
@@ -1502,7 +1502,7 @@ private async Task SetBreakpoint(SessionId sessionId, DebugStore store, Breakpoi
                 };
 
                 if (sendResolvedEvent)
-                    SendEvent(sessionId, "Debugger.breakpointResolved", JObject.FromObject(resolvedLocation), token);
+                    await SendEvent(sessionId, "Debugger.breakpointResolved", JObject.FromObject(resolvedLocation), token);
             }
 
             req.Locations.AddRange(breakpoints);

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@`
`8`	`8`	`using System.Net.WebSockets;`
`9`	`9`	`using System.Text;`
`10`	`10`	`using System.Threading;`
	`11`	`+using System.Threading.Channels;`
`11`	`12`	`using System.Threading.Tasks;`
`12`	`13`	`using Microsoft.Extensions.Logging;`
`13`	`14`	`using Newtonsoft.Json;`
`@@ -24,14 +25,19 @@ internal class DevToolsProxy`
`24`	`25`	`private ClientWebSocket browser;`
`25`	`26`	`private WebSocket ide;`
`26`	`27`	`private int next_cmd_id;`
`27`		`- private List<Task> pending_ops = new List<Task>();`
	`28`	`+ private readonly ChannelWriter<Task> _channelWriter;`
	`29`	`+ private readonly ChannelReader<Task> _channelReader;`
`28`	`30`	`private List<DevToolsQueue> queues = new List<DevToolsQueue>();`
`29`	`31`
`30`	`32`	`protected readonly ILogger logger;`
`31`	`33`
`32`	`34`	`public DevToolsProxy(ILoggerFactory loggerFactory)`
`33`	`35`	`{`
`34`	`36`	`logger = loggerFactory.CreateLogger<DevToolsProxy>();`
	`37`	`+`
	`38`	`+ var channel = Channel.CreateUnbounded<Task>(new UnboundedChannelOptions { SingleReader = true });`
	`39`	`+ _channelWriter = channel.Writer;`
	`40`	`+ _channelReader = channel.Reader;`
`35`	`41`	`}`
`36`	`42`
`37`	`43`	`protected virtual Task<bool> AcceptEvent(SessionId sessionId, string method, JObject args, CancellationToken token)`
`@@ -93,7 +99,7 @@ private DevToolsQueue GetQueueForTask(Task task)`
`93`	`99`	`return queues.FirstOrDefault(q => q.CurrentSend == task);`
`94`	`100`	`}`
`95`	`101`
`96`		`- private void Send(WebSocket to, JObject o, CancellationToken token)`
	`102`	`+ private async Task Send(WebSocket to, JObject o, CancellationToken token)`
`97`	`103`	`{`
`98`	`104`	`string sender = browser == to ? "Send-browser" : "Send-ide";`
`99`	`105`
`@@ -105,7 +111,7 @@ private void Send(WebSocket to, JObject o, CancellationToken token)`
`105`	`111`
`106`	`112`	`Task task = queue.Send(bytes, token);`
`107`	`113`	`if (task != null)`
`108`		`- pending_ops.Add(task);`
	`114`	`+ await _channelWriter.WriteAsync(task, token);`
`109`	`115`	`}`
`110`	`116`
`111`	`117`	`private async Task OnEvent(SessionId sessionId, string method, JObject args, CancellationToken token)`
`@@ -115,7 +121,7 @@ private async Task OnEvent(SessionId sessionId, string method, JObject args, Can`
`115`	`121`	`if (!await AcceptEvent(sessionId, method, args, token))`
`116`	`122`	`{`
`117`	`123`	`//logger.LogDebug ("proxy browser: {0}::{1}",method, args);`
`118`		`- SendEventInternal(sessionId, method, args, token);`
	`124`	`+ await SendEventInternal(sessionId, method, args, token);`
`119`	`125`	`}`
`120`	`126`	`}`
`121`	`127`	`catch (Exception e)`
`@@ -131,7 +137,7 @@ private async Task OnCommand(MessageId id, string method, JObject args, Cancella`
`131`	`137`	`if (!await AcceptCommand(id, method, args, token))`
`132`	`138`	`{`
`133`	`139`	`Result res = await SendCommandInternal(id, method, args, token);`
`134`		`- SendResponseInternal(id, res, token);`
	`140`	`+ await SendResponseInternal(id, res, token);`
`135`	`141`	`}`
`136`	`142`	`}`
`137`	`143`	`catch (Exception e)`
`@@ -152,31 +158,38 @@ private void OnResponse(MessageId id, Result result)`
`152`	`158`	`logger.LogError("Cannot respond to command: {id} with result: {result} - command is not pending", id, result);`
`153`	`159`	`}`
`154`	`160`
`155`		`- private void ProcessBrowserMessage(string msg, CancellationToken token)`
	`161`	`+ private Task ProcessBrowserMessage(string msg, CancellationToken token)`
`156`	`162`	`{`
`157`	`163`	`var res = JObject.Parse(msg);`
`158`	`164`
`159`	`165`	`//if (method != "Debugger.scriptParsed" && method != "Runtime.consoleAPICalled")`
`160`	`166`	`Log("protocol", $"browser: {msg}");`
`161`	`167`
`162`	`168`	`if (res["id"] == null)`
`163`		`- pending_ops.Add(OnEvent(res.ToObject<SessionId>(), res["method"].Value<string>(), res["params"] as JObject, token));`
	`169`	`+ {`
	`170`	`+ return OnEvent(res.ToObject<SessionId>(), res["method"].Value<string>(), res["params"] as JObject, token);`
	`171`	`+ }`
`164`	`172`	`else`
	`173`	`+ {`
`165`	`174`	`OnResponse(res.ToObject<MessageId>(), Result.FromJson(res));`
	`175`	`+ return null;`
	`176`	`+ }`
`166`	`177`	`}`
`167`	`178`
`168`		`- private void ProcessIdeMessage(string msg, CancellationToken token)`
	`179`	`+ private Task ProcessIdeMessage(string msg, CancellationToken token)`
`169`	`180`	`{`
`170`	`181`	`Log("protocol", $"ide: {msg}");`
`171`	`182`	`if (!string.IsNullOrEmpty(msg))`
`172`	`183`	`{`
`173`	`184`	`var res = JObject.Parse(msg);`
`174`	`185`	`var id = res.ToObject<MessageId>();`
`175`		`- pending_ops.Add(OnCommand(`
	`186`	`+ return OnCommand(`
`176`	`187`	`id,`
`177`	`188`	`res["method"].Value<string>(),`
`178`		`- res["params"] as JObject, token));`
	`189`	`+ res["params"] as JObject, token);`
`179`	`190`	`}`
	`191`	`+`
	`192`	`+ return null;`
`180`	`193`	`}`
`181`	`194`
`182`	`195`	`internal async Task<Result> SendCommand(SessionId id, string method, JObject args, CancellationToken token)`
`@@ -185,7 +198,7 @@ internal async Task<Result> SendCommand(SessionId id, string method, JObject arg`
`185`	`198`	`return await SendCommandInternal(id, method, args, token);`
`186`	`199`	`}`
`187`	`200`
`188`		`- private Task<Result> SendCommandInternal(SessionId sessionId, string method, JObject args, CancellationToken token)`
	`201`	`+ private async Task<Result> SendCommandInternal(SessionId sessionId, string method, JObject args, CancellationToken token)`
`189`	`202`	`{`
`190`	`203`	`int id = Interlocked.Increment(ref next_cmd_id);`
`191`	`204`
`@@ -203,17 +216,17 @@ private Task<Result> SendCommandInternal(SessionId sessionId, string method, JOb`
`203`	`216`	`//Log ("verbose", $"add cmd id {sessionId}-{id}");`
`204`	`217`	`pending_cmds[msgId] = tcs;`
`205`	`218`
`206`		`- Send(this.browser, o, token);`
`207`		`- return tcs.Task;`
	`219`	`+ await Send(browser, o, token);`
	`220`	`+ return await tcs.Task;`
`208`	`221`	`}`
`209`	`222`
`210`		`- public void SendEvent(SessionId sessionId, string method, JObject args, CancellationToken token)`
	`223`	`+ public Task SendEvent(SessionId sessionId, string method, JObject args, CancellationToken token)`
`211`	`224`	`{`
`212`	`225`	`//Log ("verbose", $"sending event {method}: {args}");`
`213`		`- SendEventInternal(sessionId, method, args, token);`
	`226`	`+ return SendEventInternal(sessionId, method, args, token);`
`214`	`227`	`}`
`215`	`228`
`216`		`- private void SendEventInternal(SessionId sessionId, string method, JObject args, CancellationToken token)`
	`229`	`+ private Task SendEventInternal(SessionId sessionId, string method, JObject args, CancellationToken token)`
`217`	`230`	`{`
`218`	`231`	`var o = JObject.FromObject(new`
`219`	`232`	`{`
`@@ -223,21 +236,21 @@ private void SendEventInternal(SessionId sessionId, string method, JObject args,`
`223`	`236`	`if (sessionId.sessionId != null)`
`224`	`237`	`o["sessionId"] = sessionId.sessionId;`
`225`	`238`
`226`		`- Send(this.ide, o, token);`
	`239`	`+ return Send(ide, o, token);`
`227`	`240`	`}`
`228`	`241`
`229`	`242`	`internal void SendResponse(MessageId id, Result result, CancellationToken token)`
`230`	`243`	`{`
`231`	`244`	`SendResponseInternal(id, result, token);`
`232`	`245`	`}`
`233`	`246`
`234`		`- private void SendResponseInternal(MessageId id, Result result, CancellationToken token)`
	`247`	`+ private Task SendResponseInternal(MessageId id, Result result, CancellationToken token)`
`235`	`248`	`{`
`236`	`249`	`JObject o = result.ToJObject(id);`
`237`	`250`	`if (!result.IsOk)`
`238`	`251`	`logger.LogError($"sending error response for id: {id} -> {result}");`
`239`	`252`
`240`		`- Send(this.ide, o, token);`
	`253`	`+ return Send(this.ide, o, token);`
`241`	`254`	`}`
`242`	`255`
`243`	`256`	`// , HttpContext context)`
`@@ -257,10 +270,14 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)`
`257`	`270`	`Log("verbose", $"DevToolsProxy: Client connected on {browserUri}");`
`258`	`271`	`var x = new CancellationTokenSource();`
`259`	`272`
	`273`	`+ List<Task> pending_ops = new();`
	`274`	`+`
`260`	`275`	`pending_ops.Add(ReadOne(browser, x.Token));`
`261`	`276`	`pending_ops.Add(ReadOne(ide, x.Token));`
`262`	`277`	`pending_ops.Add(side_exception.Task);`
`263`	`278`	`pending_ops.Add(client_initiated_close.Task);`
	`279`	`+ Task<bool> readerTask = _channelReader.WaitToReadAsync(x.Token).AsTask();`
	`280`	`+ pending_ops.Add(readerTask);`
`264`	`281`
`265`	`282`	`try`
`266`	`283`	`{`
`@@ -277,14 +294,26 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)`
`277`	`294`	`break;`
`278`	`295`	`}`
`279`	`296`
	`297`	`+ if (readerTask.IsCompleted)`
	`298`	`+ {`
	`299`	`+ while (_channelReader.TryRead(out Task newTask))`
	`300`	`+ {`
	`301`	`+ pending_ops.Add(newTask);`
	`302`	`+ }`
	`303`	`+`
	`304`	`+ pending_ops[4] = _channelReader.WaitToReadAsync(x.Token).AsTask();`
	`305`	`+ }`
	`306`	`+`
`280`	`307`	`//logger.LogTrace ("pump {0} {1}", task, pending_ops.IndexOf (task));`
`281`	`308`	`if (completedTask == pending_ops[0])`
`282`	`309`	`{`
`283`	`310`	`string msg = ((Task<string>)completedTask).Result;`
`284`	`311`	`if (msg != null)`
`285`	`312`	`{`
`286`	`313`	`pending_ops[0] = ReadOne(browser, x.Token); //queue next read`
`287`		`- ProcessBrowserMessage(msg, x.Token);`
	`314`	`+ Task newTask = ProcessBrowserMessage(msg, x.Token);`
	`315`	`+ if (newTask != null)`
	`316`	`+ pending_ops.Add(newTask);`
`288`	`317`	`}`
`289`	`318`	`}`
`290`	`319`	`else if (completedTask == pending_ops[1])`
`@@ -293,7 +322,9 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)`
`293`	`322`	`if (msg != null)`
`294`	`323`	`{`
`295`	`324`	`pending_ops[1] = ReadOne(ide, x.Token); //queue next read`
`296`		`- ProcessIdeMessage(msg, x.Token);`
	`325`	`+ Task newTask = ProcessIdeMessage(msg, x.Token);`
	`326`	`+ if (newTask != null)`
	`327`	`+ pending_ops.Add(newTask);`
`297`	`328`	`}`
`298`	`329`	`}`
`299`	`330`	`else if (completedTask == pending_ops[2])`
`@@ -313,10 +344,13 @@ public async Task Run(Uri browserUri, WebSocket ideSocket)`
`313`	`344`	`}`
`314`	`345`	`}`
`315`	`346`	`}`
	`347`	`+`
	`348`	`+ _channelWriter.Complete();`
`316`	`349`	`}`
`317`	`350`	`catch (Exception e)`
`318`	`351`	`{`
`319`	`352`	`Log("error", $"DevToolsProxy::Run: Exception {e}");`
	`353`	`+ _channelWriter.Complete(e);`
`320`	`354`	`//throw;`
`321`	`355`	`}`
`322`	`356`	`finally`
Original file line number	Diff line number	Diff line change
`@@ -139,7 +139,7 @@ protected override async Task<bool> AcceptEvent(SessionId sessionId, string meth`
`139`	`139`
`140`	`140`	`case "Runtime.executionContextCreated":`
`141`	`141`	`{`
`142`		`- SendEvent(sessionId, method, args, token);`
	`142`	`+ await SendEvent(sessionId, method, args, token);`
`143`	`143`	`JToken ctx = args?["context"];`
`144`	`144`	`var aux_data = ctx?["auxData"] as JObject;`
`145`	`145`	`int id = ctx["id"].Value<int>();`
`@@ -986,7 +986,7 @@ private async Task<bool> SendCallStack(SessionId sessionId, ExecutionContext con`
`986`	`986`	`await SendCommand(sessionId, "Debugger.resume", new JObject(), token);`
`987`	`987`	`return true;`
`988`	`988`	`}`
`989`		`- SendEvent(sessionId, "Debugger.paused", o, token);`
	`989`	`+ await SendEvent(sessionId, "Debugger.paused", o, token);`
`990`	`990`
`991`	`991`	`return true;`
`992`	`992`
`@@ -1103,7 +1103,7 @@ internal async Task<MethodInfo> LoadSymbolsOnDemand(AssemblyInfo asm, int method`
`1103`	`1103`	`foreach (SourceFile source in asm.Sources)`
`1104`	`1104`	`{`
`1105`	`1105`	`var scriptSource = JObject.FromObject(source.ToScriptSource(context.Id, context.AuxData));`
`1106`		`- SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);`
	`1106`	`+ await SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);`
`1107`	`1107`	`}`
`1108`	`1108`	`return asm.GetMethodByToken(method_token);`
`1109`	`1109`	`}`
`@@ -1333,7 +1333,7 @@ private async Task OnSourceFileAdded(SessionId sessionId, SourceFile source, Exe`
`1333`	`1333`	`{`
`1334`	`1334`	`JObject scriptSource = JObject.FromObject(source.ToScriptSource(context.Id, context.AuxData));`
`1335`	`1335`	`Log("debug", $"sending {source.Url} {context.Id} {sessionId.sessionId}");`
`1336`		`- SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);`
	`1336`	`+ await SendEvent(sessionId, "Debugger.scriptParsed", scriptSource, token);`
`1337`	`1337`
`1338`	`1338`	`foreach (var req in context.BreakpointRequests.Values)`
`1339`	`1339`	`{`
`@@ -1412,7 +1412,7 @@ private async Task<DebugStore> RuntimeReady(SessionId sessionId, CancellationTok`
`1412`	`1412`
`1413`	`1413`	`DebugStore store = await LoadStore(sessionId, token);`
`1414`	`1414`	`context.ready.SetResult(store);`
`1415`		`- SendEvent(sessionId, "Mono.runtimeReady", new JObject(), token);`
	`1415`	`+ await SendEvent(sessionId, "Mono.runtimeReady", new JObject(), token);`
`1416`	`1416`	`context.SdbAgent.ResetStore(store);`
`1417`	`1417`	`return store;`
`1418`	`1418`	`}`
`@@ -1502,7 +1502,7 @@ private async Task SetBreakpoint(SessionId sessionId, DebugStore store, Breakpoi`
`1502`	`1502`	`};`
`1503`	`1503`
`1504`	`1504`	`if (sendResolvedEvent)`
`1505`		`- SendEvent(sessionId, "Debugger.breakpointResolved", JObject.FromObject(resolvedLocation), token);`
	`1505`	`+ await SendEvent(sessionId, "Debugger.breakpointResolved", JObject.FromObject(resolvedLocation), token);`
`1506`	`1506`	`}`
`1507`	`1507`
`1508`	`1508`	`req.Locations.AddRange(breakpoints);`