<commit_before>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2015 Couchbase, Inc
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#include "dcp/consumer.h"

#include "dcp/dcpconnmap.h"
#include "dcp/stream.h"
#include "ep_engine.h"
#include "ep_time.h"
#include "failover-table.h"
#include "replicationthrottle.h"

#include <climits>
#include <phosphor/phosphor.h>

const std::string DcpConsumer::noopCtrlMsg = "enable_noop";
const std::string DcpConsumer::noopIntervalCtrlMsg = "set_noop_interval";
const std::string DcpConsumer::connBufferCtrlMsg = "connection_buffer_size";
const std::string DcpConsumer::priorityCtrlMsg = "set_priority";
const std::string DcpConsumer::extMetadataCtrlMsg = "enable_ext_metadata";
const std::string DcpConsumer::valueCompressionCtrlMsg = "enable_value_compression";
const std::string DcpConsumer::cursorDroppingCtrlMsg = "supports_cursor_dropping";

class Processor : public GlobalTask {
public:
    Processor(EventuallyPersistentEngine* e,
              connection_t c,
              double sleeptime = 1,
              bool completeBeforeShutdown = true)
        : GlobalTask(e, TaskId::Processor, sleeptime, completeBeforeShutdown),
          conn(c),
          description("Processing buffered items for " + conn->getName()) {
    }

    ~Processor() {
        DcpConsumer* consumer = static_cast<DcpConsumer*>(conn.get());
        consumer->taskCancelled();
    }

    bool run() {
        TRACE_EVENT0("ep-engine/task", "Processor");
        DcpConsumer* consumer = static_cast<DcpConsumer*>(conn.get());
        if (consumer->doDisconnect()) {
            return false;
        }

        double sleepFor = 0.0;
        enum process_items_error_t state = consumer->processBufferedItems();
        switch (state) {
            case all_processed:
                sleepFor = INT_MAX;
                break;
            case more_to_process:
                sleepFor = 0.0;
                break;
            case cannot_process:
                sleepFor = 5.0;
                break;
        }

        if (consumer->notifiedProcessor(false)) {
            wakeUp();
            state = more_to_process;
        } else {
            snooze(sleepFor);
            // Check if the processor was notified again,
            // in which case the task should wake immediately.
            if (consumer->notifiedProcessor(false)) {
                wakeUp();
                state = more_to_process;
            }
        }

        consumer->setProcessorTaskState(state);

        return true;
    }<commit_msg>MB-24874: Explain rationale for snooze/wakeUp order<commit_after>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2015 Couchbase, Inc
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#include "dcp/consumer.h"

#include "dcp/dcpconnmap.h"
#include "dcp/stream.h"
#include "ep_engine.h"
#include "ep_time.h"
#include "failover-table.h"
#include "replicationthrottle.h"

#include <climits>
#include <phosphor/phosphor.h>

const std::string DcpConsumer::noopCtrlMsg = "enable_noop";
const std::string DcpConsumer::noopIntervalCtrlMsg = "set_noop_interval";
const std::string DcpConsumer::connBufferCtrlMsg = "connection_buffer_size";
const std::string DcpConsumer::priorityCtrlMsg = "set_priority";
const std::string DcpConsumer::extMetadataCtrlMsg = "enable_ext_metadata";
const std::string DcpConsumer::valueCompressionCtrlMsg = "enable_value_compression";
const std::string DcpConsumer::cursorDroppingCtrlMsg = "supports_cursor_dropping";

class Processor : public GlobalTask {
public:
    Processor(EventuallyPersistentEngine* e,
              connection_t c,
              double sleeptime = 1,
              bool completeBeforeShutdown = true)
        : GlobalTask(e, TaskId::Processor, sleeptime, completeBeforeShutdown),
          conn(c),
          description("Processing buffered items for " + conn->getName()) {
    }

    ~Processor() {
        DcpConsumer* consumer = static_cast<DcpConsumer*>(conn.get());
        consumer->taskCancelled();
    }

    bool run() {
        TRACE_EVENT0("ep-engine/task", "Processor");
        DcpConsumer* consumer = static_cast<DcpConsumer*>(conn.get());
        if (consumer->doDisconnect()) {
            return false;
        }

        double sleepFor = 0.0;
        enum process_items_error_t state = consumer->processBufferedItems();
        switch (state) {
            case all_processed:
                sleepFor = INT_MAX;
                break;
            case more_to_process:
                sleepFor = 0.0;
                break;
            case cannot_process:
                sleepFor = 5.0;
                break;
        }

        // Check if we've been notified of more work to do - if not then sleep;
        // if so then wakeup and re-run the task.
        // Note: The order of the wakeUp / snooze here is *critical* - another
        // thread may concurrently notify us (set processorNotification=true)
        // while we are performing the checks, so we need to ensure we don't
        // loose a wakeup as that would result in this Task sleeping forever
        // (and DCP hanging).
        // To prevent this, we perform an initial check of notifiedProcessor(),
        // which if false we initially sleep, and then check a second time.
        // We could race if the other actor sets processorNotification=true
        // between the second `if(consumer->notifiedProcessor)` and us calling
        // `wakeUp()`; but that's essentially a benign race as it will just
        // result in wakeUp() being called twice which is benign.
        if (consumer->notifiedProcessor(false)) {
            wakeUp();
            state = more_to_process;
        } else {
            snooze(sleepFor);
            // Check if the processor was notified again,
            // in which case the task should wake immediately.
            if (consumer->notifiedProcessor(false)) {
                wakeUp();
                state = more_to_process;
            }
        }

        consumer->setProcessorTaskState(state);

        return true;
    }
{'fraction_non_alphanumeric': 0.09922502075837254, 'fraction_numerical': 0.0056739551619153054, 'mean_word_length': 2.5548450565666503, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5299865', 'n_tokens_mistral': 2046, 'n_tokens_neox': 1952, 'n_words': 595}