forked from dotnet/fsharp
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathReactor.fs
More file actions
executable file
·163 lines (143 loc) · 8.85 KB
/
Copy pathReactor.fs
File metadata and controls
executable file
·163 lines (143 loc) · 8.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
// Copyright (c) Microsoft Corporation. All Rights Reserved. Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
namespace Microsoft.FSharp.Compiler.SourceCodeServices
open System
open System.Diagnostics
open System.Globalization
open System.Threading
open Microsoft.FSharp.Control
open Microsoft.FSharp.Compiler
open Microsoft.FSharp.Compiler.Lib
open Microsoft.FSharp.Compiler.ErrorLogger
open Microsoft.FSharp.Compiler.AbstractIL.Internal.Library
/// Represents the capability to schedule work in the compiler service operations queue for the compilation thread
type internal IReactorOperations =
abstract EnqueueAndAwaitOpAsync : string * (CompilationThreadToken -> Cancellable<'T>) -> Async<'T>
abstract EnqueueOp: string * (CompilationThreadToken -> unit) -> unit
[<NoEquality; NoComparison>]
type internal ReactorCommands =
/// Kick off a build.
| SetBackgroundOp of (CompilationThreadToken -> bool) option
/// Do some work not synchronized in the mailbox.
| Op of string * CancellationToken * (CompilationThreadToken -> unit) * (unit -> unit)
/// Finish the background building
| WaitForBackgroundOpCompletion of AsyncReplyChannel<unit>
/// Finish all the queued ops
| CompleteAllQueuedOps of AsyncReplyChannel<unit>
[<AutoSerializable(false);Sealed>]
/// There is one global Reactor for the entire language service, no matter how many projects or files
/// are open.
type Reactor() =
static let pauseBeforeBackgroundWorkDefault = GetEnvInteger "FCS_PauseBeforeBackgroundWorkMilliseconds" 1000
static let theReactor = Reactor()
let mutable pauseBeforeBackgroundWork = pauseBeforeBackgroundWorkDefault
// We need to store the culture for the VS thread that is executing now,
// so that when the reactor picks up a thread from the threadpool we can set the culture
let culture = new CultureInfo(CultureInfo.CurrentUICulture.Name)
/// Mailbox dispatch function.
let builder =
MailboxProcessor<_>.Start <| fun inbox ->
// Async workflow which receives messages and dispatches to worker functions.
let rec loop (bgOpOpt, onComplete, bg) =
async { Trace.TraceInformation("Reactor: receiving..., remaining {0}, mem {1}, gc2 {2}", inbox.CurrentQueueLength, GC.GetTotalMemory(false)/1000000L, GC.CollectionCount(2))
// Explanation: The reactor thread acts as the compilation thread in hosted scenarios
let ctok = AssumeCompilationThreadWithoutEvidence()
// Messages always have priority over the background op.
let! msg =
async { match bgOpOpt, onComplete with
| None, None ->
let! msg = inbox.Receive()
return Some msg
| _, Some _ ->
return! inbox.TryReceive(0)
| Some _, _ ->
let timeout = (if bg then 0 else pauseBeforeBackgroundWork)
return! inbox.TryReceive(timeout) }
#if FX_RESHAPED_GLOBALIZATION
CultureInfo.CurrentUICulture <- culture
#else
Thread.CurrentThread.CurrentUICulture <- culture
#endif
match msg with
| Some (SetBackgroundOp bgOpOpt) ->
Trace.TraceInformation("Reactor: --> set background op, remaining {0}, mem {1}, gc2 {2}", inbox.CurrentQueueLength, GC.GetTotalMemory(false)/1000000L, GC.CollectionCount(2))
return! loop (bgOpOpt, onComplete, false)
| Some (Op (desc, ct, op, ccont)) ->
if ct.IsCancellationRequested then ccont() else
Trace.TraceInformation("Reactor: --> {0}, remaining {1}, mem {2}, gc2 {3}", desc, inbox.CurrentQueueLength, GC.GetTotalMemory(false)/1000000L, GC.CollectionCount(2))
let time = System.DateTime.Now
op ctok
let span = System.DateTime.Now - time
//if span.TotalMilliseconds > 100.0 then
Trace.TraceInformation("Reactor: <-- {0}, remaining {1}, took {2}ms", desc, inbox.CurrentQueueLength, span.TotalMilliseconds)
return! loop (bgOpOpt, onComplete, false)
| Some (WaitForBackgroundOpCompletion channel) ->
Trace.TraceInformation("Reactor: --> wait for background (debug only), remaining {0}, mem {1}, gc2 {2}", inbox.CurrentQueueLength, GC.GetTotalMemory(false)/1000000L, GC.CollectionCount(2))
match bgOpOpt with
| None -> ()
| Some bgOp ->
while bgOp ctok do
()
channel.Reply(())
return! loop (None, onComplete, false)
| Some (CompleteAllQueuedOps channel) ->
Trace.TraceInformation("Reactor: --> stop background work and complete all queued ops, remaining {0}, mem {1}, gc2 {2}", inbox.CurrentQueueLength, GC.GetTotalMemory(false)/1000000L, GC.CollectionCount(2))
return! loop (None, Some channel, false)
| None ->
match bgOpOpt, onComplete with
| _, Some onComplete -> onComplete.Reply()
| Some bgOp, None ->
Trace.TraceInformation("Reactor: --> background step, remaining {0}, mem {1}, gc2 {2}", inbox.CurrentQueueLength, GC.GetTotalMemory(false)/1000000L, GC.CollectionCount(2))
let time = System.DateTime.Now
let res = bgOp ctok
let span = System.DateTime.Now - time
//if span.TotalMilliseconds > 100.0 then
Trace.TraceInformation("Reactor: <-- background step, remaining {0}, took {1}ms", inbox.CurrentQueueLength, span.TotalMilliseconds)
return! loop ((if res then Some bgOp else None), onComplete, true)
| None, None -> failwith "unreachable, should have used inbox.Receive"
}
async {
while true do
try
do! loop (None, None, false)
with e ->
Debug.Assert(false,String.Format("unexpected failure in reactor loop {0}, restarting", e))
}
// [Foreground Mailbox Accessors] -----------------------------------------------------------
member r.SetBackgroundOp(build) =
Trace.TraceInformation("Reactor: enqueue start background, length {0}", builder.CurrentQueueLength)
builder.Post(SetBackgroundOp build)
member r.EnqueueOp(desc, op) =
Trace.TraceInformation("Reactor: enqueue {0}, length {1}", desc, builder.CurrentQueueLength)
builder.Post(Op(desc, CancellationToken.None, op, (fun () -> ())))
member r.EnqueueOpPrim(desc, ct, op, ccont) =
Trace.TraceInformation("Reactor: enqueue {0}, length {1}", desc, builder.CurrentQueueLength)
builder.Post(Op(desc, ct, op, ccont))
member r.CurrentQueueLength =
builder.CurrentQueueLength
// This is for testing only
member r.WaitForBackgroundOpCompletion() =
Trace.TraceInformation("Reactor: enqueue wait for background, length {0}", builder.CurrentQueueLength)
builder.PostAndReply WaitForBackgroundOpCompletion
// This is for testing only
member r.CompleteAllQueuedOps() =
Trace.TraceInformation("Reactor: enqueue wait for all ops, length {0}", builder.CurrentQueueLength)
builder.PostAndReply CompleteAllQueuedOps
member r.EnqueueAndAwaitOpAsync (desc, f) =
async {
let! ct = Async.CancellationToken
let resultCell = AsyncUtil.AsyncResultCell<_>()
r.EnqueueOpPrim(desc, ct,
op=(fun ctok ->
let result =
try
match Cancellable.run ct (f ctok) with
| ValueOrCancelled.Value r -> AsyncUtil.AsyncOk r
| ValueOrCancelled.Cancelled e -> AsyncUtil.AsyncCanceled e
with e -> e |> AsyncUtil.AsyncException
resultCell.RegisterResult(result)),
ccont=(fun () -> resultCell.RegisterResult (AsyncUtil.AsyncCanceled(OperationCanceledException())) )
)
return! resultCell.AsyncResult
}
member __.PauseBeforeBackgroundWork with get() = pauseBeforeBackgroundWork and set v = pauseBeforeBackgroundWork <- v
static member Singleton = theReactor