<commit_before>/* Copyright (c) 2011-2013 Stanford University
 *
 * Permission to use, copy, modify, and distribute this software for any purpose
 * with or without fee is hereby granted, provided that the above copyright
 * notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR(S) DISCLAIM ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL AUTHORS BE LIABLE FOR ANY
 * SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
 * RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
 * CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

#include <sys/epoll.h>
#include <sys/select.h>
#include <fstream>

#include "ShortMacros.h"
#include "Common.h"
#include "CycleCounter.h"
#include "Cycles.h"
#include "Dispatch.h"
#include "Fence.h"
#include "RawMetrics.h"
#include "NoOp.h"
#include "PerfStats.h"

// Uncomment to print out a human readable name for any poller that takes longer
// than slowPollerCycles to complete. Useful for determining which poller is
// responsible for "Long gap" messages.
// #define DEBUG_SLOW_POLLERS 1

namespace RAMCloud {

/**
 * Default object used to make system calls.
 */
static Syscall defaultSyscall;

/**
 * Used by this class to make all system calls.  In normal production
 * use it points to defaultSyscall; for testing it points to a mock
 * object.
 */
Syscall* Dispatch::sys = &defaultSyscall;

// The following is a debugging assertion used in many methods to make sure
// that either (a) the method was invoked in the dispatch thread or (b) the
// dispatcher was been locked before calling the method.
#define CHECK_LOCK assert((owner->locked.load() != 0) || \
        owner->isDispatchThread())

/**
 * Constructor for Dispatch objects.
 * \param hasDedicatedThread
 *      Pass true if there is a thread which owns this dispatch (this is
 *      true on RAMCloud servers), if false then the dispatch performs
 *      no synchronization and callers must guarantee mutual exclusion
 *      themselves.
 */
Dispatch::Dispatch(bool hasDedicatedThread)
    : currentTime(Cycles::rdtsc())
    , pollers()
    , files()
    , epollFd(-1)
    , epollThread()
    , readyFd(-1)
    , readyEvents(0)
    , fileInvocationSerial(0)
    , timers()
    , earliestTriggerTime(0)
    , ownerId(ThreadId::get())
    , mutex("Dispatch::mutex")
    , lockNeeded(0)
    , locked(0)
    , hasDedicatedThread(hasDedicatedThread)
    , slowPollerCycles(Cycles::fromSeconds(.05))
    , profilerFlag(false)
    , totalElements(0)
    , pollingTimes(NULL)
    , nextInd(0)
{
    exitPipeFds[0] = exitPipeFds[1] = -1;
}

/**
 * Destructor for Dispatch objects.  All existing Pollers, Timers, and Files
 * become useless.
 */
Dispatch::~Dispatch()
{
    if (epollThread) {
        // Writing data to the pipe below signals the epoll thread that it
        // should exit.
        sys->write(exitPipeFds[1], "x", 1);
        epollThread->join();
        epollThread.destroy();
        sys->close(exitPipeFds[0]);
        sys->close(exitPipeFds[1]);
        exitPipeFds[1] = exitPipeFds[0] = -1;
    }
    if (epollFd >= 0) {
        sys->close(epollFd);
        epollFd = -1;
    }
    for (uint32_t i = 0; i < pollers.size(); i++) {
        pollers[i]->owner = NULL;
        pollers[i]->slot = -1;
    }
    pollers.clear();
    for (uint32_t i = 0; i < files.size(); i++) {
        if (files[i] != NULL) {
            files[i]->owner = NULL;
            files[i]->active = false;
            files[i]->events = 0;
            files[i] = NULL;
        }
    }
    readyFd = -1;
    while (timers.size() > 0) {
        Timer* t = timers.back();
        t->stop();
        t->owner = NULL;
    }
    cleanProfiler();
}

/**
 * Check to see if any events need handling.
 *
 * \return
 *      The return value is a count of the number of useful actions
 *      taken during this call (the number of pollers, timer handlers,
 *      and file handlers that did useful work). Zero means there
 *      was no useful work found in this call.
 */
int
Dispatch::poll()
{
    assert(isDispatchThread());
    int result = 0;
    uint64_t previous = currentTime;
    currentTime = Cycles::rdtsc();
    // Dispatch::poll() execution time will be recorded if
    // profilerFlag is set to true.
    if (profilerFlag) {
        pollingTimes[nextInd] = currentTime - previous;
        nextInd = (nextInd + 1) % totalElements;
    }
    if (((currentTime - previous) > slowPollerCycles) && hasDedicatedThread) {
        LOG(NOTICE, "Long gap in dispatcher: %.1f ms",
                Cycles::toSeconds(currentTime - previous)*1e03);
    }
    if (lockNeeded.load() != 0) {
        // Someone wants us locked. Indicate that we are locked,
        // then wait for the lock to be released.
        Fence::leave();
        locked.store(1);
        while (lockNeeded.load() != 0) {
            // Empty loop body.
        }
        locked.store(0);
        Fence::enter();
        uint64_t newCurrent = Cycles::rdtsc();
        if ((newCurrent - currentTime) > slowPollerCycles) {
            LOG(WARNING, "Long lockout in poller: %.1f ms",
                    Cycles::toSeconds(newCurrent - currentTime)*1e03);
        }
        currentTime = newCurrent;
    }
    for (uint32_t i = 0; i < pollers.size(); i++) {
#if DEBUG_SLOW_POLLERS
        uint64_t ticks = 0;<commit_msg>Increased log level to WARNING for Dispatch long gap messages.<commit_after>/* Copyright (c) 2011-2015 Stanford University
 *
 * Permission to use, copy, modify, and distribute this software for any purpose
 * with or without fee is hereby granted, provided that the above copyright
 * notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR(S) DISCLAIM ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL AUTHORS BE LIABLE FOR ANY
 * SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
 * RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
 * CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

#include <sys/epoll.h>
#include <sys/select.h>
#include <fstream>

#include "ShortMacros.h"
#include "Common.h"
#include "CycleCounter.h"
#include "Cycles.h"
#include "Dispatch.h"
#include "Fence.h"
#include "RawMetrics.h"
#include "NoOp.h"
#include "PerfStats.h"

// Uncomment to print out a human readable name for any poller that takes longer
// than slowPollerCycles to complete. Useful for determining which poller is
// responsible for "Long gap" messages.
// #define DEBUG_SLOW_POLLERS 1

namespace RAMCloud {

/**
 * Default object used to make system calls.
 */
static Syscall defaultSyscall;

/**
 * Used by this class to make all system calls.  In normal production
 * use it points to defaultSyscall; for testing it points to a mock
 * object.
 */
Syscall* Dispatch::sys = &defaultSyscall;

// The following is a debugging assertion used in many methods to make sure
// that either (a) the method was invoked in the dispatch thread or (b) the
// dispatcher was been locked before calling the method.
#define CHECK_LOCK assert((owner->locked.load() != 0) || \
        owner->isDispatchThread())

/**
 * Constructor for Dispatch objects.
 * \param hasDedicatedThread
 *      Pass true if there is a thread which owns this dispatch (this is
 *      true on RAMCloud servers), if false then the dispatch performs
 *      no synchronization and callers must guarantee mutual exclusion
 *      themselves.
 */
Dispatch::Dispatch(bool hasDedicatedThread)
    : currentTime(Cycles::rdtsc())
    , pollers()
    , files()
    , epollFd(-1)
    , epollThread()
    , readyFd(-1)
    , readyEvents(0)
    , fileInvocationSerial(0)
    , timers()
    , earliestTriggerTime(0)
    , ownerId(ThreadId::get())
    , mutex("Dispatch::mutex")
    , lockNeeded(0)
    , locked(0)
    , hasDedicatedThread(hasDedicatedThread)
    , slowPollerCycles(Cycles::fromSeconds(.05))
    , profilerFlag(false)
    , totalElements(0)
    , pollingTimes(NULL)
    , nextInd(0)
{
    exitPipeFds[0] = exitPipeFds[1] = -1;
}

/**
 * Destructor for Dispatch objects.  All existing Pollers, Timers, and Files
 * become useless.
 */
Dispatch::~Dispatch()
{
    if (epollThread) {
        // Writing data to the pipe below signals the epoll thread that it
        // should exit.
        sys->write(exitPipeFds[1], "x", 1);
        epollThread->join();
        epollThread.destroy();
        sys->close(exitPipeFds[0]);
        sys->close(exitPipeFds[1]);
        exitPipeFds[1] = exitPipeFds[0] = -1;
    }
    if (epollFd >= 0) {
        sys->close(epollFd);
        epollFd = -1;
    }
    for (uint32_t i = 0; i < pollers.size(); i++) {
        pollers[i]->owner = NULL;
        pollers[i]->slot = -1;
    }
    pollers.clear();
    for (uint32_t i = 0; i < files.size(); i++) {
        if (files[i] != NULL) {
            files[i]->owner = NULL;
            files[i]->active = false;
            files[i]->events = 0;
            files[i] = NULL;
        }
    }
    readyFd = -1;
    while (timers.size() > 0) {
        Timer* t = timers.back();
        t->stop();
        t->owner = NULL;
    }
    cleanProfiler();
}

/**
 * Check to see if any events need handling.
 *
 * \return
 *      The return value is a count of the number of useful actions
 *      taken during this call (the number of pollers, timer handlers,
 *      and file handlers that did useful work). Zero means there
 *      was no useful work found in this call.
 */
int
Dispatch::poll()
{
    assert(isDispatchThread());
    int result = 0;
    uint64_t previous = currentTime;
    currentTime = Cycles::rdtsc();
    // Dispatch::poll() execution time will be recorded if
    // profilerFlag is set to true.
    if (profilerFlag) {
        pollingTimes[nextInd] = currentTime - previous;
        nextInd = (nextInd + 1) % totalElements;
    }
    if (((currentTime - previous) > slowPollerCycles) && hasDedicatedThread) {
        LOG(WARNING, "Long gap in dispatcher: %.1f ms",
                Cycles::toSeconds(currentTime - previous)*1e03);
    }
    if (lockNeeded.load() != 0) {
        // Someone wants us locked. Indicate that we are locked,
        // then wait for the lock to be released.
        Fence::leave();
        locked.store(1);
        while (lockNeeded.load() != 0) {
            // Empty loop body.
        }
        locked.store(0);
        Fence::enter();
        uint64_t newCurrent = Cycles::rdtsc();
        if ((newCurrent - currentTime) > slowPollerCycles) {
            LOG(WARNING, "Long lockout in poller: %.1f ms",
                    Cycles::toSeconds(newCurrent - currentTime)*1e03);
        }
        currentTime = newCurrent;
    }
    for (uint32_t i = 0; i < pollers.size(); i++) {
#if DEBUG_SLOW_POLLERS
        uint64_t ticks = 0;
{'fraction_non_alphanumeric': 0.11492080633549316, 'fraction_numerical': 0.012059035277177825, 'mean_word_length': 2.730446458543135, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 45, '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': '5766568', 'n_tokens_mistral': 3533, 'n_tokens_neox': 3366, 'n_words': 1122}