<commit_before>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2013 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 "config.h"

#include "ep_engine.h"
#include "failover-table.h"
#include "tapconnmap.h"
#include "tapthrottle.h"
#include "upr-consumer.h"
#include "upr-response.h"
#include "upr-stream.h"

class Processer : public GlobalTask {
public:
    Processer(EventuallyPersistentEngine* e, connection_t c,
                const Priority &p, double sleeptime = 1, bool shutdown = false)
        : GlobalTask(e, p, sleeptime, shutdown), conn(c) {}

    bool run();

    std::string getDescription();

private:
    connection_t conn;
};

bool Processer::run() {
    UprConsumer* consumer = static_cast<UprConsumer*>(conn.get());
    if (consumer->doDisconnect()) {
        return false;
    }

    switch (consumer->processBufferedItems()) {
        case all_processed:
            snooze(1);
            break;
        case more_to_process:
            snooze(0);
            break;
        case cannot_process:
            snooze(5);
            break;
        default:
            abort();
    }

    return true;
}

std::string Processer::getDescription() {
    std::stringstream ss;
    ss << "Processing buffered items for " << conn->getName();
    return ss.str();
}

UprConsumer::UprConsumer(EventuallyPersistentEngine &engine, const void *cookie,
                         const std::string &name)
    : Consumer(engine, cookie, name), opaqueCounter(0), processTaskId(0),
          itemsToProcess(false) {
    Configuration& config = engine.getConfiguration();
    streams = new passive_stream_t[config.getMaxVbuckets()];
    setSupportAck(false);
    setLogHeader("UPR (Consumer) " + getName() + " -");
    setReserved(true);

    flowControl.enabled = config.isUprEnableFlowControl();
    flowControl.bufferSize = config.getUprConnBufferSize();
    flowControl.maxUnackedBytes = config.getUprMaxUnackedBytes();

    ExTask task = new Processer(&engine, this, Priority::PendingOpsPriority, 1);
    processTaskId = ExecutorPool::get()->schedule(task, NONIO_TASK_IDX);
}

UprConsumer::~UprConsumer() {
    delete[] streams;
}

ENGINE_ERROR_CODE UprConsumer::addStream(uint32_t opaque, uint16_t vbucket,
                                         uint32_t flags) {
    LockHolder lh(streamMutex);
    if (doDisconnect()) {
        return ENGINE_DISCONNECT;
    }

    RCPtr<VBucket> vb = engine_.getVBucket(vbucket);
    if (!vb) {
        LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Add stream failed because this "<commit_msg>MB-11572: CloseAllStreams before deleting them<commit_after>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2013 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 "config.h"

#include "ep_engine.h"
#include "failover-table.h"
#include "tapconnmap.h"
#include "tapthrottle.h"
#include "upr-consumer.h"
#include "upr-response.h"
#include "upr-stream.h"

class Processer : public GlobalTask {
public:
    Processer(EventuallyPersistentEngine* e, connection_t c,
                const Priority &p, double sleeptime = 1, bool shutdown = false)
        : GlobalTask(e, p, sleeptime, shutdown), conn(c) {}

    bool run();

    std::string getDescription();

private:
    connection_t conn;
};

bool Processer::run() {
    UprConsumer* consumer = static_cast<UprConsumer*>(conn.get());
    if (consumer->doDisconnect()) {
        return false;
    }

    switch (consumer->processBufferedItems()) {
        case all_processed:
            snooze(1);
            break;
        case more_to_process:
            snooze(0);
            break;
        case cannot_process:
            snooze(5);
            break;
        default:
            abort();
    }

    return true;
}

std::string Processer::getDescription() {
    std::stringstream ss;
    ss << "Processing buffered items for " << conn->getName();
    return ss.str();
}

UprConsumer::UprConsumer(EventuallyPersistentEngine &engine, const void *cookie,
                         const std::string &name)
    : Consumer(engine, cookie, name), opaqueCounter(0), processTaskId(0),
          itemsToProcess(false) {
    Configuration& config = engine.getConfiguration();
    streams = new passive_stream_t[config.getMaxVbuckets()];
    setSupportAck(false);
    setLogHeader("UPR (Consumer) " + getName() + " -");
    setReserved(true);

    flowControl.enabled = config.isUprEnableFlowControl();
    flowControl.bufferSize = config.getUprConnBufferSize();
    flowControl.maxUnackedBytes = config.getUprMaxUnackedBytes();

    ExTask task = new Processer(&engine, this, Priority::PendingOpsPriority, 1);
    processTaskId = ExecutorPool::get()->schedule(task, NONIO_TASK_IDX);
}

UprConsumer::~UprConsumer() {
    closeAllStreams();
    delete[] streams;
}

ENGINE_ERROR_CODE UprConsumer::addStream(uint32_t opaque, uint16_t vbucket,
                                         uint32_t flags) {
    LockHolder lh(streamMutex);
    if (doDisconnect()) {
        return ENGINE_DISCONNECT;
    }

    RCPtr<VBucket> vb = engine_.getVBucket(vbucket);
    if (!vb) {
        LOG(EXTENSION_LOG_WARNING, "%s (vb %d) Add stream failed because this "
{'fraction_non_alphanumeric': 0.11635617303441743, 'fraction_numerical': 0.008051784022734449, 'mean_word_length': 3.0582959641255605, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2442972', 'n_tokens_mistral': 1960, 'n_tokens_neox': 1887, 'n_words': 499}