<commit_before>package com.rimmer.metrics

import com.rimmer.metrics.generated.type.*
import org.joda.time.DateTime
import java.util.*

interface MetricsWriter {
    /**
     * Starts a new profiler event within the current call.
     * @return An event index to send to endQuery.
     */
    fun startEvent(eventType: EventType, type: String): Int

    /** Indicates that the provided event id has finished. */
    fun endEvent(id: Int)
}

class Metrics: MetricsWriter {
    inner class Call(val path: String, val startDate: DateTime, val startTime: Long, val parameters: Map<String, String>) {
        val events = ArrayList<Event>()
        var endTime = startTime
        var failed = false
        var error = false
        var failReason = ""
        var failTrace = ""
    }

    inner class Path {
        var lastSend = System.nanoTime()
        val calls = ArrayList<Call>()
    }

    var sender: Sender? = null
    private val paths = HashMap<String, Path>()
    private val currentCall = ThreadLocal<Call>()

    /** Indicates the start of a new action. All timed actions performed from this thread are added to the action. */
    fun start(path: String, parameters: Map<String, String>) {
        val startTime = System.nanoTime()
        val startDate = DateTime.now()
        currentCall.set(Call(path, startDate, startTime, parameters))
    }

    /** Indicates that the current action has completed. It is added to the metrics. */
    fun finish() {
        val time = System.nanoTime()
        val call = currentCall.get()
        call.endTime = time

        val path = synchronized(this) {
            var path = paths[call.path]
            if(path === null) {
                path = Path()
                paths.put(call.path, path)
            }
            path
        }!!

        synchronized(path) {
            path.calls.add(call)
            if(time - path.lastSend > 60000000000L) {
                sendStats(path)
                path.calls.clear()
                path.lastSend = time
            }
        }
    }

    /** Discards the current action without adding to the metrics, for example if the call was forwarded. */
    fun discard() { currentCall.set(null) }

    /**
     * Indicates that the current call did not complete correctly.
     * Incorrect calls are not used in the metrics.
     * @param wasError If set, the failure is reported in the metrics as an internal server error.
     */
    fun fail(wasError: Boolean, reason: String) {
        currentCall.get()?.run {
            failed = true
            error = wasError
            failReason = reason
            finish()
        }
    }

    override fun startEvent(eventType: EventType, type: String): Int {
        val startTime = System.nanoTime()
        val startDate = DateTime.now()
        return currentCall.get()?.run {
            events.add(Event(eventType, type, startDate, startTime, startTime))
            events.size - 1
        } ?: 0
    }

    override fun endEvent(id: Int) {
        currentCall.get()?.run {
            val event = events[id]
            events[id] = event.copy(endTime = System.nanoTime())
        }
    }

    private fun sendStats(path: Path) {
        // The sender may throw exceptions, which we don't want to leak.
        try {
            sender?.run {
                // Filter out any failed requests. If it was an internal failure we log it.
                val calls = ArrayList<Call>(path.calls.size)
                path.calls.filterTo(calls) {
                    if (it.error) {
                        sendError(ErrorPacket(it.path, it.startDate, it.failReason, it.failTrace))
                    }
                    !it.failed
                }

                // If all calls failed, we have nothing more to do.
                if (calls.isEmpty()) return

                val date = calls[0].startDate
                val count = calls.size

                // We need to sort the calls on their duration to find the median and percentiles.
                // In the same loop we calculate the average.
                var totalTime = 0L
                calls.sortBy {
                    val elapsed = it.endTime - it.startTime
                    totalTime += elapsed
                    elapsed
                }

                if(calls.size == 1) {
                    totalTime = calls[0].endTime - calls[0].startTime
                }

                val median = calls[count / 2]
                val min = calls[0]
                val max = calls[count - 1]

                // Send the timing intervals for calculating statistics.
                sendStatistic(StatPacket(min.path, "", date, totalTime, calls.map { Interval(it.startTime, it.endTime) }.toCollection(ArrayList())))

                // Send a full profile for the median and maximum values.
                sendProfile(ProfilePacket(median.path, "", median.startDate, median.startTime, median.endTime, median.events))
                sendProfile(ProfilePacket(max.path, "", max.startDate, max.startTime, max.endTime, max.events))
            }
        } catch(e: Throwable) {
            println("Could not send metrics:")
            e.printStackTrace()
        }
    }
}<commit_msg>Support multiple interleaved calls per thread<commit_after>package com.rimmer.metrics

import com.rimmer.metrics.generated.type.*
import org.joda.time.DateTime
import java.util.*

interface MetricsWriter {
    /**
     * Starts a new profiler event within the current call.
     * @return An event index to send to endQuery.
     */
    fun startEvent(call: Int, eventType: EventType, type: String): Int

    /** Indicates that the provided event id has finished. */
    fun endEvent(call: Int, id: Int)
}

class Metrics: MetricsWriter {
    inner class Call(val path: String, val startDate: DateTime, val startTime: Long, val parameters: Map<String, String>) {
        val events = ArrayList<Event>()
        var endTime = startTime
        var failed = false
        var error = false
        var failReason = ""
        var failTrace = ""
    }

    inner class Path {
        var lastSend = System.nanoTime()
        val calls = ArrayList<Call>()
    }

    var sender: Sender? = null
    private val paths = HashMap<String, Path>()
    private val calls = ArrayList<Call?>()
    private var nextCall = 0

    /** Indicates the start of a new action. All timed actions performed from this thread are added to the action. */
    fun start(path: String, parameters: Map<String, String>): Int {
        val startTime = System.nanoTime()
        val startDate = DateTime.now()
        val call = Call(path, startDate, startTime, parameters)

        val i = nextCall
        if(i >= calls.size) {
            calls.add(call)
            nextCall++
        } else {
            calls[i] = call
            val next = calls.indexOfFirst { it == null }
            nextCall = if(next == -1) calls.size else next
        }
        return i
    }

    /** Indicates that the current action has completed. It is added to the metrics. */
    fun finish(callId: Int) {
        val time = System.nanoTime()
        val call = getCall(callId) ?: return
        call.endTime = time

        var path = paths[call.path]
        if(path === null) {
            path = Path()
            paths.put(call.path, path)
        }

        path.calls.add(call)
        if(time - path.lastSend > 60000000000L) {
            sendStats(path)
            path.calls.clear()
            path.lastSend = time
        }

        removeCall(callId)
    }

    /** Discards the current action without adding to the metrics, for example if the call was forwarded. */
    fun discard(call: Int) {
        removeCall(call)
    }

    /**
     * Indicates that the current call did not complete correctly.
     * Incorrect calls are not used in the metrics.
     * @param wasError If set, the failure is reported in the metrics as an internal server error.
     */
    fun fail(call: Int, wasError: Boolean, reason: String) {
        getCall(call)?.run {
            failed = true
            error = wasError
            failReason = reason
            finish(call)
        }
    }

    override fun startEvent(call: Int, eventType: EventType, type: String): Int {
        val startTime = System.nanoTime()
        val startDate = DateTime.now()
        return getCall(call)?.run {
            events.add(Event(eventType, type, startDate, startTime, startTime))
            events.size - 1
        } ?: 0
    }

    override fun endEvent(call: Int, id: Int) {
        getCall(call)?.run {
            val event = events[id]
            events[id] = event.copy(endTime = System.nanoTime())
        }
    }

    private fun sendStats(path: Path) {
        // The sender may throw exceptions, which we don't want to leak.
        try {
            sender?.run {
                // Filter out any failed requests. If it was an internal failure we log it.
                val calls = ArrayList<Call>(path.calls.size)
                path.calls.filterTo(calls) {
                    if (it.error) {
                        sendError(ErrorPacket(it.path, it.startDate, it.failReason, it.failTrace))
                    }
                    !it.failed
                }

                // If all calls failed, we have nothing more to do.
                if (calls.isEmpty()) return

                val date = calls[0].startDate
                val count = calls.size

                // We need to sort the calls on their duration to find the median and percentiles.
                // In the same loop we calculate the average.
                var totalTime = 0L
                calls.sortBy {
                    val elapsed = it.endTime - it.startTime
                    totalTime += elapsed
                    elapsed
                }

                if(calls.size == 1) {
                    totalTime = calls[0].endTime - calls[0].startTime
                }

                val median = calls[count / 2]
                val min = calls[0]
                val max = calls[count - 1]

                // Send the timing intervals for calculating statistics.
                sendStatistic(StatPacket(min.path, "", date, totalTime, calls.map { Interval(it.startTime, it.endTime) }.toCollection(ArrayList())))

                // Send a full profile for the median and maximum values.
                sendProfile(ProfilePacket(median.path, "", median.startDate, median.startTime, median.endTime, median.events))
                sendProfile(ProfilePacket(max.path, "", max.startDate, max.startTime, max.endTime, max.events))
            }
        } catch(e: Throwable) {
            println("Could not send metrics:")
            e.printStackTrace()
        }
    }

    private fun getCall(id: Int): Call? {
        if(calls.size <= id || calls[id] == null) {
            println("Received unknown call id $id")
            return null
        } else {
            return calls[id]
        }
    }

    private fun removeCall(id: Int) {
        calls[id] = null
        nextCall = id
    }
}
{'fraction_non_alphanumeric': 0.09521233432512849, 'fraction_numerical': 0.003967180596880353, 'mean_word_length': 1.7066861883845779, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '1780777', 'n_tokens_mistral': 3020, 'n_tokens_neox': 2880, 'n_words': 933}