<commit_before>@doc doc"""
YarnClientProtocol: Hadoop RPC client for application client to Yarn resource manager protocol.
""" ->
typealias YarnClientProtocol HadoopRpcProtocol{ApplicationClientProtocolServiceBlockingStub}

for fn in (:getClusterMetrics, :getClusterNodes, :getNewApplication, :submitApplication, :forceKillApplication, :getApplicationReport, :getApplicationAttempts)
    @eval begin
       (hadoop.yarn.$fn)(p::YarnClientProtocol, inp) = (hadoop.yarn.$fn)(p.stub, p.controller, inp)
    end
end

@doc doc"""
YarnAMRMProtocol: Hadoop RPC client for Yarn application master to resource manager protocol.
""" ->
typealias YarnAMRMProtocol HadoopRpcProtocol{ApplicationMasterProtocolServiceBlockingStub}

for fn in (:registerApplicationMaster, :finishApplicationMaster, :allocate)
    @eval begin
        (hadoop.yarn.$fn)(p::YarnAMRMProtocol, inp) = (hadoop.yarn.$fn)(p.stub, p.controller, inp)
    end
end

@doc doc"""
YarnAMNMProtocol: Hadoop RPC client for Yarn application master to node manager protocol.
""" ->
typealias YarnAMNMProtocol HadoopRpcProtocol{ContainerManagementProtocolServiceBlockingStub}

for fn in (:startContainers, :stopContainers, :getContainerStatuses)
    @eval begin
        (hadoop.yarn.$fn)(p::YarnAMNMProtocol, inp) = (hadoop.yarn.$fn)(p.stub, p.controller, inp)
    end
end 



@doc doc"""
# YarnException
Thrown by Yarn APIs.
""" ->
type YarnException <: Exception
    message::AbstractString
end

function show(io::IO, serex::SerializedExceptionProto)
    print(io, "Exception: ")
    isfilled(serex, :class_name) && print(io, serex.class_name, ": ")
    isfilled(serex, :message) && print(io, serex.message)
    isfilled(serex, :trace) && print(io, '\n', serex.trace, '\n')

    if isfilled(serex, :cause)
        println(io, "Caused by:")
        show(io, serex.cause)
    end
    nothing
end

function YarnException(cex::ContainerExceptionMapProto)
    cont_id = cex.ccontainer_id
    app_id = cont_id.app_id
    atmpt_id = cont_id.app_attempt_id
    iob = IOBuffer()
    println(iob, "Error in container $(cont_id.id) of app $(app_id.id) attempt $(atmpt_id.attemptId):")
    show(iob, cex.exception)
    YarnException(takebuf_string(iob))
end



@doc doc"""
YarnNode represents a node manager in the yarn cluster and its
communication address, resource state and run state.
""" ->
type YarnNode
    host::AbstractString
    port::Int32
    rack::AbstractString
    ncontainers::Int32
    mem::Int32
    cores::Int32
    memused::Int32
    coresused::Int32
    state::Int32
    isrunning::Bool
    report::NodeReportProto

    function YarnNode(node::NodeReportProto)
        host = node.nodeId.host
        port = node.nodeId.port
        rack = node.rackName
        ncontainers = node.numContainers

        state = node.node_state
        isrunning = (state == NodeStateProto.NS_RUNNING)

        if isrunning
            mem = node.capability.memory
            cores = node.capability.virtual_cores

            memused = node.used.memory
            coresused = node.used.virtual_cores
        else
            mem = cores = memused = coresused = 0
        end

        new(host, port, rack, ncontainers, mem, cores, memused, coresused, state, isrunning, node)
    end
end

@doc doc"""
NODE_STATES: enum value to state map. Used for converting state for display.
""" ->
const NODE_STATES = [:new, :running, :unhealthy, :decommissioned, :lost, :rebooted]

function show(io::IO, node::YarnNode)
    print(io, "YarnNode: $(node.rack)/$(node.host):$(node.port) $(NODE_STATES[node.state])")
    println(io, node.isrunning ? ", Used mem: $(node.memused)/$(node.mem), cores: $(node.coresused)/$(node.cores)" : "")
    nothing
end



@doc doc"""
YarnNodes holds node information as visible to the application master.
It also caches connection to node masters. Connection are reused if they are required before a set keepalivesecs time.
""" ->
type YarnNodes
    count::Int32
    status::Dict{NodeIdProto,YarnNode}
    conn::Dict{NodeIdProto,Tuple}
    ugi::UserGroupInformation
    keepalivesecs::UInt64

    function YarnNodes(ugi::UserGroupInformation)
        new(0, Dict{NodeIdProto,YarnNode}(), Dict{NodeIdProto,Tuple}(), ugi, YARN_NM_CONN_KEEPALIVE_SECS)
    end
end

function show(io::IO, nodes::YarnNodes)
    println(io, "YarnNodes: $(nodes.count) (connected to $(length(nodes.conn)))")
    for n in values(nodes.status)
        show(io, n)
    end
    nothing
end

function update(nodes::YarnNodes, arp::AllocateResponseProto)
    isfilled(arp, :num_cluster_nodes) && (nodes.count = arp.num_cluster_nodes)
    if isfilled(arp, :updated_nodes)
        for nrep in arp.updated_nodes
            nodes.status[nrep.nodeId] = YarnNode(nrep)
        end
    end
    nothing
end

function update(nodes::YarnNodes, gcnrp::GetClusterNodesResponseProto)
    isfilled(gcnrp, :nodeReports) || return

    nlist = gcnrp.nodeReports
    nodes.count = length(nlist)
    for nrep in nlist
        nodes.status[nrep.nodeId] = YarnNode(nrep)
    end 
    nothing
end

function get_connection(nodes::YarnNodes, nodeid::NodeIdProto)
    (nodeid in keys(nodes.status)) || throw(YarnException("Unknown Yarn node: $(nodeid.host):$(nodeid.port)"))
    node = nodes.status[nodeid]
    node.isrunning || throw(YarnException("Yarn node $(nodeid.host):$(nodeid.port) is not running"))

    t = time()
    if nodeid in keys(nodes.conn)
        (conn,lastusetime) = nodes.conn[nodeid]
        (t < (lastusetime + nodes.keepalivesecs)) && (return conn)
        try
            disconnect(conn.channel)
        finally
            delete!(nodes.conn, nodeid)
        end
    end
        
    conn = YarnAMNMProtocol(nodeid.host, nodeid.port, nodes.ugi)
    nodes.conn[nodeid] = (conn, t)
    conn
end

function release_connection(nodes::YarnNodes, nodeid::NodeIdProto, conn::YarnAMNMProtocol, reuse::Bool)
    if reuse
        nodes.conn[nodeid] = (conn, time())
    else
        try
            disconnect(conn.channel)
        finally
            delete!(nodes.conn, nodeid)
        end
    end
    nothing
end


@doc doc"""
RequestPipeline holds entities while they are requested for from yarn resource manager.
Application master thread extracts pending items and requests them from RM, whereupon they are moved on to the requested state.
""" ->
type RequestPipeline{T}
    pending::Vector{T}
    requested::Vector{T}

    function RequestPipeline()
        new(T[], T[])
    end
end

pending{T}(pipe::RequestPipeline{T}, item::T) = push!(pipe.pending, item)
function torequest{T}(pipe::RequestPipeline{T})<commit_msg>need to lock node connection<commit_after>@doc doc"""
YarnClientProtocol: Hadoop RPC client for application client to Yarn resource manager protocol.
""" ->
typealias YarnClientProtocol HadoopRpcProtocol{ApplicationClientProtocolServiceBlockingStub}

for fn in (:getClusterMetrics, :getClusterNodes, :getNewApplication, :submitApplication, :forceKillApplication, :getApplicationReport, :getApplicationAttempts)
    @eval begin
       (hadoop.yarn.$fn)(p::YarnClientProtocol, inp) = (hadoop.yarn.$fn)(p.stub, p.controller, inp)
    end
end

@doc doc"""
YarnAMRMProtocol: Hadoop RPC client for Yarn application master to resource manager protocol.
""" ->
typealias YarnAMRMProtocol HadoopRpcProtocol{ApplicationMasterProtocolServiceBlockingStub}

for fn in (:registerApplicationMaster, :finishApplicationMaster, :allocate)
    @eval begin
        (hadoop.yarn.$fn)(p::YarnAMRMProtocol, inp) = (hadoop.yarn.$fn)(p.stub, p.controller, inp)
    end
end

@doc doc"""
YarnAMNMProtocol: Hadoop RPC client for Yarn application master to node manager protocol.
""" ->
typealias YarnAMNMProtocol HadoopRpcProtocol{ContainerManagementProtocolServiceBlockingStub}

for fn in (:startContainers, :stopContainers, :getContainerStatuses)
    @eval begin
        (hadoop.yarn.$fn)(p::YarnAMNMProtocol, inp) = (hadoop.yarn.$fn)(p.stub, p.controller, inp)
    end
end 



@doc doc"""
# YarnException
Thrown by Yarn APIs.
""" ->
type YarnException <: Exception
    message::AbstractString
end

function show(io::IO, serex::SerializedExceptionProto)
    print(io, "Exception: ")
    isfilled(serex, :class_name) && print(io, serex.class_name, ": ")
    isfilled(serex, :message) && print(io, serex.message)
    isfilled(serex, :trace) && print(io, '\n', serex.trace, '\n')

    if isfilled(serex, :cause)
        println(io, "Caused by:")
        show(io, serex.cause)
    end
    nothing
end

function YarnException(cex::ContainerExceptionMapProto)
    cont_id = cex.ccontainer_id
    app_id = cont_id.app_id
    atmpt_id = cont_id.app_attempt_id
    iob = IOBuffer()
    println(iob, "Error in container $(cont_id.id) of app $(app_id.id) attempt $(atmpt_id.attemptId):")
    show(iob, cex.exception)
    YarnException(takebuf_string(iob))
end



@doc doc"""
YarnNode represents a node manager in the yarn cluster and its
communication address, resource state and run state.
""" ->
type YarnNode
    host::AbstractString
    port::Int32
    rack::AbstractString
    ncontainers::Int32
    mem::Int32
    cores::Int32
    memused::Int32
    coresused::Int32
    state::Int32
    isrunning::Bool
    report::NodeReportProto

    function YarnNode(node::NodeReportProto)
        host = node.nodeId.host
        port = node.nodeId.port
        rack = node.rackName
        ncontainers = node.numContainers

        state = node.node_state
        isrunning = (state == NodeStateProto.NS_RUNNING)

        if isrunning
            mem = node.capability.memory
            cores = node.capability.virtual_cores

            memused = node.used.memory
            coresused = node.used.virtual_cores
        else
            mem = cores = memused = coresused = 0
        end

        new(host, port, rack, ncontainers, mem, cores, memused, coresused, state, isrunning, node)
    end
end

@doc doc"""
NODE_STATES: enum value to state map. Used for converting state for display.
""" ->
const NODE_STATES = [:new, :running, :unhealthy, :decommissioned, :lost, :rebooted]

function show(io::IO, node::YarnNode)
    print(io, "YarnNode: $(node.rack)/$(node.host):$(node.port) $(NODE_STATES[node.state])")
    println(io, node.isrunning ? ", Used mem: $(node.memused)/$(node.mem), cores: $(node.coresused)/$(node.cores)" : "")
    nothing
end



@doc doc"""
YarnNodes holds node information as visible to the application master.
It also caches connection to node masters. Connection are reused if they are required before a set keepalivesecs time.
""" ->
type YarnNodes
    count::Int32
    status::Dict{NodeIdProto,YarnNode}
    conn::Dict{NodeIdProto,Tuple}
    ugi::UserGroupInformation
    keepalivesecs::UInt64
    lck::ReentrantLock

    function YarnNodes(ugi::UserGroupInformation)
        new(0, Dict{NodeIdProto,YarnNode}(), Dict{NodeIdProto,Tuple}(), ugi, YARN_NM_CONN_KEEPALIVE_SECS, ReentrantLock())
    end
end

function show(io::IO, nodes::YarnNodes)
    println(io, "YarnNodes: $(nodes.count) (connected to $(length(nodes.conn)))")
    for n in values(nodes.status)
        show(io, n)
    end
    nothing
end

function update(nodes::YarnNodes, arp::AllocateResponseProto)
    isfilled(arp, :num_cluster_nodes) && (nodes.count = arp.num_cluster_nodes)
    if isfilled(arp, :updated_nodes)
        for nrep in arp.updated_nodes
            nodes.status[nrep.nodeId] = YarnNode(nrep)
        end
    end
    nothing
end

function update(nodes::YarnNodes, gcnrp::GetClusterNodesResponseProto)
    isfilled(gcnrp, :nodeReports) || return

    nlist = gcnrp.nodeReports
    nodes.count = length(nlist)
    for nrep in nlist
        nodes.status[nrep.nodeId] = YarnNode(nrep)
    end 
    nothing
end

function _new_or_existing_conn(nodes::YarnNodes, nodeid::NodeIdProto)
    lock(nodes.lck)
    try
        t = time()
        (conn,lastusetime,lck) = (nodeid in keys(nodes.conn)) ? nodes.conn[nodeid] : (YarnAMNMProtocol(nodeid.host, nodeid.port, nodes.ugi), t, ReentrantLock())
        if t > (lastusetime + nodes.keepalivesecs)
            try
                disconnect(conn.channel)
            finally
                conn = YarnAMNMProtocol(nodeid.host, nodeid.port, nodes.ugi)
                lastusetime = t
            end
        end
        nodes.conn[nodeid] = (conn, lastusetime, lck)
        return (conn, lastusetime, lck)
    finally
        unlock(nodes.lck)
    end
end

function get_connection(nodes::YarnNodes, nodeid::NodeIdProto)
    (nodeid in keys(nodes.status)) || throw(YarnException("Unknown Yarn node: $(nodeid.host):$(nodeid.port)"))
    node = nodes.status[nodeid]
    node.isrunning || throw(YarnException("Yarn node $(nodeid.host):$(nodeid.port) is not running"))

    conn, lastusetime, lck = _new_or_existing_conn(nodes, nodeid)
    # lock the connection to mark in use
    lock(lck)
    conn
end

function release_connection(nodes::YarnNodes, nodeid::NodeIdProto, conn::YarnAMNMProtocol, reuse::Bool)
    lock(nodes.lck)
    try
        # disconnect if we don't want to reuse the connection
        if !reuse
            try
                disconnect(conn.channel)
            end
        end

        if nodeid in keys(nodes.conn)
            # release the lock if it's an existing connection
            (conn_old, lastusetime, lck) = nodes.conn[nodeid]
            (conn === conn_old) && unlock(lck)
            if reuse
                nodes.conn[nodeid] = (conn, time(), lck)
            else
                delete!(nodes.conn, nodeid)
            end
        else
            # add connection to cache for reuse
            if reuse
                nodes.conn[nodeid] = (conn, time(), ReentrantLock())
            end
        end
    finally
        unlock(nodes.lck)
    end
    nothing
end


@doc doc"""
RequestPipeline holds entities while they are requested for from yarn resource manager.
Application master thread extracts pending items and requests them from RM, whereupon they are moved on to the requested state.
""" ->
type RequestPipeline{T}
    pending::Vector{T}
    requested::Vector{T}

    function RequestPipeline()
        new(T[], T[])
    end
end

pending{T}(pipe::RequestPipeline{T}, item::T) = push!(pipe.pending, item)
function torequest{T}(pipe::RequestPipeline{T})
{'fraction_non_alphanumeric': 0.11292372881355932, 'fraction_numerical': 0.002824858757062147, 'mean_word_length': 3.468602082675923, 'pattern_counts': {'":': 2, '<': 6, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '878929', 'n_tokens_mistral': 4608, 'n_tokens_neox': 4444, 'n_words': 1094}