<commit_before>﻿namespace MBrace.Runtime

open System.Diagnostics

open MBrace.Core
open MBrace.Core.Internals
open MBrace.ThreadPool
open MBrace.Runtime.Utils

/// MBrace runtime client handle abstract class.
[<AbstractClass; NoEquality; NoComparison; AutoSerializable(false)>]
type MBraceClient (runtime : IRuntimeManager, defaultFaultPolicy : FaultPolicy) =
    do ignore <| RuntimeManagerRegistry.TryRegister runtime

    let syncRoot = new obj()
    let imem = ThreadPoolRuntime.Create(resources = runtime.ResourceRegistry, memoryEmulation = MemoryEmulation.Shared)
    let storeClient = new CloudStoreClient(runtime.ResourceRegistry)
    let mutable defaultFaultPolicy = defaultFaultPolicy

    let taskManagerClient = new CloudProcessManagerClient(runtime)
    let getWorkers () = async {
        let! workers = runtime.WorkerManager.GetAvailableWorkers()
        return workers |> Array.map (fun w -> WorkerRef.Create(runtime, w.Id))
    }

    let workers = CacheAtom.Create(getWorkers(), intervalMilliseconds = 1000)

    let mutable systemLogPoller : ILogPoller<SystemLogEntry> option = None
    let getSystemLogPoller() =
        match systemLogPoller with
        | Some lp -> lp
        | None ->
            lock syncRoot (fun () ->
                match systemLogPoller with
                | Some lp -> lp
                | None ->
                    let lp = runtime.RuntimeSystemLogManager.CreateLogPoller() |> Async.RunSync
                    systemLogPoller <- Some lp
                    lp)

    /// <summary>
    ///     Creates a fresh cloud cancellation token source for use in the MBrace cluster.
    /// </summary>
    /// <param name="parents">Parent cancellation token sources. New cancellation token will be canceled if any of the parents is canceled.</param>
    member c.CreateCancellationTokenSource ([<O;D(null:obj)>] ?parents : seq<ICloudCancellationToken>) : ICloudCancellationTokenSource =
        async {
            let parents = parents |> Option.map Seq.toArray
            let! dcts = CloudCancellationToken.Create(runtime.CancellationEntryFactory, ?parents = parents, elevate = true)
            return dcts :> ICloudCancellationTokenSource
        } |> Async.RunSync

    /// <summary>
    ///     Asynchronously submits supplied cloud computation for execution in the current MBrace runtime.
    ///     Returns an instance of CloudProcess, which can be queried for information on the progress of the computation.
    /// </summary>
    /// <param name="workflow">Cloud computation to be executed.</param>
    /// <param name="cancellationToken">Cancellation token for computation.</param>
    /// <param name="faultPolicy">Fault retry policy for the process. Defaults to the client FaultPolicy property.</param>
    /// <param name="target">Target worker to initialize computation.</param>
    /// <param name="additionalResources">Additional per-cloud process MBrace resources that can be appended to the computation state.</param>
    /// <param name="taskName">User-specified process name.</param>
    member c.CreateProcessAsync(workflow : Cloud<'T>, [<O;D(null:obj)>] ?cancellationToken : ICloudCancellationToken, 
                                    [<O;D(null:obj)>] ?faultPolicy : FaultPolicy, [<O;D(null:obj)>] ?target : IWorkerRef, 
                                    [<O;D(null:obj)>] ?additionalResources : ResourceRegistry, [<O;D(null:obj)>] ?taskName : string) : Async<CloudProcess<'T>> = async {
        let faultPolicy = defaultArg faultPolicy defaultFaultPolicy
        let dependencies = runtime.AssemblyManager.ComputeDependencies((workflow, faultPolicy))
        let assemblyIds = dependencies |> Array.map (fun d -> d.Id)
        do! runtime.AssemblyManager.UploadAssemblies(dependencies)
        return! Combinators.runStartAsCloudProcess runtime None assemblyIds taskName faultPolicy cancellationToken additionalResources target workflow
    }

    /// <summary>
    ///     Submits supplied cloud computation for execution in the current MBrace runtime.
    ///     Returns an instance of CloudProcess, which can be queried for information on the progress of the computation.
    /// </summary>
    /// <param name="workflow">Cloud computation to be executed.</param>
    /// <param name="cancellationToken">Cancellation token for computation.</param>
    /// <param name="faultPolicy">Fault retry policy for the process. Defaults to the client FaultPolicy property.</param>
    /// <param name="target">Target worker to initialize computation.</param>
    /// <param name="additionalResources">Additional per-cloud process MBrace resources that can be appended to the computation state.</param>
    /// <param name="taskName">User-specified process name.</param>
    member __.CreateProcess(workflow : Cloud<'T>, [<O;D(null:obj)>] ?cancellationToken : ICloudCancellationToken, [<O;D(null:obj)>] ?faultPolicy : FaultPolicy, 
                                [<O;D(null:obj)>] ?target : IWorkerRef, [<O;D(null:obj)>] ?additionalResources : ResourceRegistry, [<O;D(null:obj)>] ?taskName : string) : CloudProcess<'T> =
        __.CreateProcessAsync(workflow, ?cancellationToken = cancellationToken, ?faultPolicy = faultPolicy, 
                                    ?target = target, ?additionalResources = additionalResources, ?taskName = taskName) |> Async.RunSync
<commit_msg>add warning when sending computation to cluster with no workers<commit_after>﻿namespace MBrace.Runtime

open System.Diagnostics

open MBrace.Core
open MBrace.Core.Internals
open MBrace.ThreadPool
open MBrace.Runtime.Utils

/// MBrace runtime client handle abstract class.
[<AbstractClass; NoEquality; NoComparison; AutoSerializable(false)>]
type MBraceClient (runtime : IRuntimeManager, defaultFaultPolicy : FaultPolicy) =
    do ignore <| RuntimeManagerRegistry.TryRegister runtime

    let syncRoot = new obj()
    let imem = ThreadPoolRuntime.Create(resources = runtime.ResourceRegistry, memoryEmulation = MemoryEmulation.Shared)
    let storeClient = new CloudStoreClient(runtime.ResourceRegistry)
    let mutable defaultFaultPolicy = defaultFaultPolicy

    let taskManagerClient = new CloudProcessManagerClient(runtime)
    let getWorkers () = async {
        let! workers = runtime.WorkerManager.GetAvailableWorkers()
        return workers |> Array.map (fun w -> WorkerRef.Create(runtime, w.Id))
    }

    let workers = CacheAtom.Create(getWorkers(), intervalMilliseconds = 1000)

    let mutable systemLogPoller : ILogPoller<SystemLogEntry> option = None
    let getSystemLogPoller() =
        match systemLogPoller with
        | Some lp -> lp
        | None ->
            lock syncRoot (fun () ->
                match systemLogPoller with
                | Some lp -> lp
                | None ->
                    let lp = runtime.RuntimeSystemLogManager.CreateLogPoller() |> Async.RunSync
                    systemLogPoller <- Some lp
                    lp)

    /// <summary>
    ///     Creates a fresh cloud cancellation token source for use in the MBrace cluster.
    /// </summary>
    /// <param name="parents">Parent cancellation token sources. New cancellation token will be canceled if any of the parents is canceled.</param>
    member c.CreateCancellationTokenSource ([<O;D(null:obj)>] ?parents : seq<ICloudCancellationToken>) : ICloudCancellationTokenSource =
        async {
            let parents = parents |> Option.map Seq.toArray
            let! dcts = CloudCancellationToken.Create(runtime.CancellationEntryFactory, ?parents = parents, elevate = true)
            return dcts :> ICloudCancellationTokenSource
        } |> Async.RunSync

    /// <summary>
    ///     Asynchronously submits supplied cloud computation for execution in the current MBrace runtime.
    ///     Returns an instance of CloudProcess, which can be queried for information on the progress of the computation.
    /// </summary>
    /// <param name="workflow">Cloud computation to be executed.</param>
    /// <param name="cancellationToken">Cancellation token for computation.</param>
    /// <param name="faultPolicy">Fault retry policy for the process. Defaults to the client FaultPolicy property.</param>
    /// <param name="target">Target worker to initialize computation.</param>
    /// <param name="additionalResources">Additional per-cloud process MBrace resources that can be appended to the computation state.</param>
    /// <param name="taskName">User-specified process name.</param>
    member c.CreateProcessAsync(workflow : Cloud<'T>, [<O;D(null:obj)>] ?cancellationToken : ICloudCancellationToken, 
                                    [<O;D(null:obj)>] ?faultPolicy : FaultPolicy, [<O;D(null:obj)>] ?target : IWorkerRef, 
                                    [<O;D(null:obj)>] ?additionalResources : ResourceRegistry, [<O;D(null:obj)>] ?taskName : string) : Async<CloudProcess<'T>> = async {
        let faultPolicy = defaultArg faultPolicy defaultFaultPolicy
        let dependencies = runtime.AssemblyManager.ComputeDependencies((workflow, faultPolicy))
        let assemblyIds = dependencies |> Array.map (fun d -> d.Id)
        do! runtime.AssemblyManager.UploadAssemblies(dependencies)
        if workers.Value.Length = 0 then
            runtime.SystemLogger.Logf LogLevel.Warning "No worker instances currently associated with cluster. Computation may never complete."

        return! Combinators.runStartAsCloudProcess runtime None assemblyIds taskName faultPolicy cancellationToken additionalResources target workflow
    }

    /// <summary>
    ///     Submits supplied cloud computation for execution in the current MBrace runtime.
    ///     Returns an instance of CloudProcess, which can be queried for information on the progress of the computation.
    /// </summary>
    /// <param name="workflow">Cloud computation to be executed.</param>
    /// <param name="cancellationToken">Cancellation token for computation.</param>
    /// <param name="faultPolicy">Fault retry policy for the process. Defaults to the client FaultPolicy property.</param>
    /// <param name="target">Target worker to initialize computation.</param>
    /// <param name="additionalResources">Additional per-cloud process MBrace resources that can be appended to the computation state.</param>
    /// <param name="taskName">User-specified process name.</param>
    member __.CreateProcess(workflow : Cloud<'T>, [<O;D(null:obj)>] ?cancellationToken : ICloudCancellationToken, [<O;D(null:obj)>] ?faultPolicy : FaultPolicy, 
                                [<O;D(null:obj)>] ?target : IWorkerRef, [<O;D(null:obj)>] ?additionalResources : ResourceRegistry, [<O;D(null:obj)>] ?taskName : string) : CloudProcess<'T> =
        __.CreateProcessAsync(workflow, ?cancellationToken = cancellationToken, ?faultPolicy = faultPolicy, 
                                    ?target = target, ?additionalResources = additionalResources, ?taskName = taskName) |> Async.RunSync

{'fraction_non_alphanumeric': 0.10375114364135407, 'fraction_numerical': 0.0008234217749313815, 'mean_word_length': 3.263260530421217, 'pattern_counts': {'":': 0, '<': 109, '<?xml version=': 0, '>': 133, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5000727', 'n_tokens_mistral': 2851, 'n_tokens_neox': 2786, 'n_words': 822}