Method compressionThreadMain

runtime/RuntimeLogger.cc:347–621

* Main compression thread that handles scanning through the StagingBuffers, * compressing log entries, and outputting a compressed log file. */

* compressing log entries, and outputting a compressed log file.
*/
void
RuntimeLogger::compressionThreadMain() {
    // Index of the last StagingBuffer checked for uncompressed log messages
    size_t lastStagingBufferChecked = 0;

    // Marks when the thread wakes up. This value should be used to calculate
    // the number of cyclesActive right before blocking/sleeping and then updated
    // to the latest rdtsc() when the thread re-awakens.
    uint64_t cyclesAwakeStart = PerfUtils::Cycles::rdtsc();
    cycleAtThreadStart = cyclesAwakeStart;

    // Manages the state associated with compressing log messages
    Log::Encoder encoder(compressingBuffer, NanoLogConfig::OUTPUT_BUFFER_SIZE);

    // Indicates whether a compression operation failed or not due
    // to insufficient space in the outputBuffer
    bool outputBufferFull = false;

    // Indicates that in scanning the StagingBuffers, we have passed the
    // zero-th index, but have not yet encoded that in he compressed output
    bool wrapAround = false;

    // Keeps a shadow mapping of the log identifiers to static information
    // to allow the logging threads to register in parallel with compression
    // lookup
    std::vector<StaticLogInfo> shadowStaticInfo;

    // Each iteration of this loop scans for uncompressed log messages in the
    // thread buffers, compresses as much as possible, and outputs it to a file.
    while (!compressionThreadShouldExit) {
        coreId = sched_getcpu();

        // Indicates how many bytes we have consumed from the StagingBuffers
        // in a single iteration of the while above. A value of 0 means we
        // were unable to consume anymore data any of the stagingBuffers
        // (either due to empty stagingBuffers or a full output encoder)
        uint64_t bytesConsumedThisIteration = 0;

        uint64_t start = PerfUtils::Cycles::rdtsc();
        // Step 1: Find buffers with entries and compress them
        {
            std::unique_lock<std::mutex> lock(bufferMutex);
            size_t i = lastStagingBufferChecked;

            // Output new dictionary entries, if necessary
            if (nextInvocationIndexToBePersisted < invocationSites.size())
            {
                std::unique_lock<std::mutex> lock (registrationMutex);
                encoder.encodeNewDictionaryEntries(
                                               nextInvocationIndexToBePersisted,
                                               invocationSites);

                // update our shadow copy
                for (uint64_t i = shadowStaticInfo.size();
                                    i < nextInvocationIndexToBePersisted; ++i)
                {
                    shadowStaticInfo.push_back(invocationSites.at(i));
                }