<commit_before>/*
 * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

#include <yajr/rpc/rpc.hpp>
#include <yajr/internal/comms.hpp>

#include <opflex/logging/internal/logging.hpp>
#include <boost/scoped_ptr.hpp>

#include <uv.h>

namespace { void prepareAgainCB(uv_timer_t *) { LOG(DEBUG); } }

int ::yajr::initLoop(uv_loop_t * loop) {

    if (!(loop->data = new (std::nothrow)
                ::yajr::comms::internal::Peer::LoopData(loop))) {
        LOG(WARNING) <<
            ": out of memory, cannot instantiate uv_loop LoopData";
        return UV_ENOMEM;
    }

    return 0;

}

void ::yajr::finiLoop(uv_loop_t * loop) {
    LOG(INFO);

    static_cast< ::yajr::comms::internal::Peer::LoopData *>(loop->data)->destroy();

}


namespace yajr { namespace comms {

using namespace yajr::comms::internal;

void internal::Peer::LoopData::onPrepareLoop() {

    if (destroying_ && !refCount_) {

        CountHandle countHandle = { this, 0 };

        uv_walk(prepare_.loop, walkAndCountHandlesCb, &countHandle);

        if (countHandle.counter) {
            LOG(INFO) << "Still waiting on " << countHandle.counter << " handles";

            return;
        }

        LOG(INFO) << this << " Stopping and closing loop watcher";
        uv_prepare_stop(&prepare_);
        uv_close((uv_handle_t*)&prepare_, &fini);

        assert(prepare_.data == this);

        return;
    }

    uint64_t now = uv_now(prepare_.loop);

    peers[TO_RESOLVE]
        .clear_and_dispose(RetryPeer());

    peers[TO_LISTEN]
        .clear_and_dispose(RetryPeer());

    if (now - lastRun_ < 750) {
        goto prepared;
    }

    if (peers[RETRY_TO_CONNECT].begin() !=
        peers[RETRY_TO_CONNECT].end()) {

        LOG(INFO) << "retrying first RETRY_TO_CONNECT peer";

        /* retry just the first active peer in the queue */
        peers[RETRY_TO_CONNECT]
            .erase_and_dispose(peers[RETRY_TO_CONNECT].begin(), RetryPeer());

        LOG(DEBUG)
            << " Stopping prepareAgain_ @"
            << reinterpret_cast<void *>(&prepareAgain_)
        ;

        uv_timer_stop(&prepareAgain_);
        /* if there are more, we will uv_timer_start() down below */
    }

    if (now - lastRun_ > 15000) {

        LOG(INFO) << "retrying all RETRY_TO_LISTEN peers";

        /* retry all listeners */
        peers[RETRY_TO_LISTEN]
            .clear_and_dispose(RetryPeer());
    }

    lastRun_ = now;

prepared:
    uv_walk(prepare_.loop, walkAndDumpHandlesCb<DEBUG>, this);

    if (peers[RETRY_TO_CONNECT].begin() !=
        peers[RETRY_TO_CONNECT].end()) {
        /* We need to make sure we unblock */

        if (!uv_is_active((uv_handle_t *)&prepareAgain_)) {
            LOG(DEBUG)
                << " Starting prepareAgain_ @"
                << reinterpret_cast<void *>(&prepareAgain_)
            ;
        } // else we are just pushing it out in time :)
        uv_timer_start(&prepareAgain_, prepareAgainCB, 1250, 0);
    }
}

void internal::Peer::LoopData::onPrepareLoop(uv_prepare_t * h) {

    static_cast< ::yajr::comms::internal::Peer::LoopData *>(h->data)
        ->onPrepareLoop();

}

void internal::Peer::LoopData::fini(uv_handle_t * h) {
    LOG(INFO);

    delete static_cast< ::yajr::comms::internal::Peer::LoopData *>(h->data);
}

void internal::Peer::LoopData::destroy(bool now) {
    LOG(INFO);

    assert(prepare_.data == this);

    assert(!destroying_);
    if (destroying_) {
        LOG(WARNING)
            << this
            << " Double destroy() detected"
        ;

        return;
    }

    destroying_ = 1;
    down();

    for (size_t i=0; i < Peer::LoopData::TOTAL_STATES; ++i) {
        peers[Peer::LoopData::PeerState(i)]
            .clear_and_dispose(PeerDisposer(now));

    }
}



/*
                    ____
                   / ___|___  _ __ ___  _ __ ___   ___  _ __
                  | |   / _ \| '_ ` _ \| '_ ` _ \ / _ \| '_ \
                  | |__| (_) | | | | | | | | | | | (_) | | | |
                   \____\___/|_| |_| |_|_| |_| |_|\___/|_| |_|

                     ____      _ _ _                _
                    / ___|__ _| | | |__   __ _  ___| | _____
                   | |   / _` | | | '_ \ / _` |/ __| |/ / __|
                   | |__| (_| | | | |_) | (_| | (__|   <\__ \
                    \____\__,_|_|_|_.__/ \__,_|\___|_|\_\___/

                                                              (Common Callbacks)
*/

namespace internal {

void on_close(uv_handle_t * h) {

    CommunicationPeer * peer = Peer::get<CommunicationPeer>(h);

    LOG(DEBUG) << peer << " down() for an on_close()";
    peer->down();

    return;
}

void on_write(uv_write_t *req, int status) {

    LOG(DEBUG);

    if (status == UV_ECANCELED || status == UV_ECONNRESET) {
        LOG(INFO) << "[" << uv_err_name(status) << "] " << uv_strerror(status);
        return;  /* Handle has been closed. */
    }

    /* see here if more output can be dequeued */

    CommunicationPeer * peer = Peer::get(req);

    peer->onWrite();

    return;
}

} /* yajr::comms::internal namespace */

}} /* yajr::comms and opflex namespaces */

#ifdef BOOST_NO_EXCEPTIONS

namespace boost
{

void throw_exception(std::exception const & e) {
    assert(0);

    LOG(FATAL) << e.what();

    exit(127);
}

}

#endif

<commit_msg>fixlet for retries not communicating<commit_after>/*
 * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

#include <yajr/rpc/rpc.hpp>
#include <yajr/internal/comms.hpp>

#include <opflex/logging/internal/logging.hpp>
#include <boost/scoped_ptr.hpp>

#include <uv.h>

namespace { void prepareAgainCB(uv_timer_t *) { LOG(DEBUG); } }

int ::yajr::initLoop(uv_loop_t * loop) {

    if (!(loop->data = new (std::nothrow)
                ::yajr::comms::internal::Peer::LoopData(loop))) {
        LOG(WARNING) <<
            ": out of memory, cannot instantiate uv_loop LoopData";
        return UV_ENOMEM;
    }

    return 0;

}

void ::yajr::finiLoop(uv_loop_t * loop) {
    LOG(INFO);

    static_cast< ::yajr::comms::internal::Peer::LoopData *>(loop->data)->destroy();

}


namespace yajr { namespace comms {

using namespace yajr::comms::internal;

void internal::Peer::LoopData::onPrepareLoop() {

    if (destroying_ && !refCount_) {

        CountHandle countHandle = { this, 0 };

        uv_walk(prepare_.loop, walkAndCountHandlesCb, &countHandle);

        if (countHandle.counter) {
            LOG(INFO) << "Still waiting on " << countHandle.counter << " handles";

            return;
        }

        LOG(INFO) << this << " Stopping and closing loop watcher";
        uv_prepare_stop(&prepare_);
        uv_close((uv_handle_t*)&prepare_, &fini);

        assert(prepare_.data == this);

        return;
    }

    uint64_t now = uv_now(prepare_.loop);

    peers[TO_RESOLVE]
        .clear_and_dispose(RetryPeer());

    peers[TO_LISTEN]
        .clear_and_dispose(RetryPeer());

    if (now - lastRun_ < 750) {
        goto prepared;
    }

    if (peers[RETRY_TO_CONNECT].begin() !=
        peers[RETRY_TO_CONNECT].end()) {

        LOG(INFO) << "retrying first RETRY_TO_CONNECT peer";

        /* retry just the first active peer in the queue */
        peers[RETRY_TO_CONNECT]
            .erase_and_dispose(peers[RETRY_TO_CONNECT].begin(), RetryPeer());

        LOG(DEBUG)
            << " Stopping prepareAgain_ @"
            << reinterpret_cast<void *>(&prepareAgain_)
        ;

        uv_timer_stop(&prepareAgain_);
        /* if there are more, we will uv_timer_start() down below */
    }

    if (now - lastRun_ > 15000) {

        LOG(INFO) << "retrying all RETRY_TO_LISTEN peers";

        /* retry all listeners */
        peers[RETRY_TO_LISTEN]
            .clear_and_dispose(RetryPeer());
    }

    lastRun_ = now;

prepared:
    uv_walk(prepare_.loop, walkAndDumpHandlesCb<DEBUG>, this);

    if (peers[RETRY_TO_CONNECT].begin() !=
        peers[RETRY_TO_CONNECT].end()) {
        /* We need to make sure we unblock */

        if (!uv_is_active((uv_handle_t *)&prepareAgain_)) {
            LOG(DEBUG)
                << " Starting prepareAgain_ @"
                << reinterpret_cast<void *>(&prepareAgain_)
            ;
        } // else we are just pushing it out in time :)
        uv_timer_start(&prepareAgain_, prepareAgainCB, 1250, 0);
    }
}

void internal::Peer::LoopData::onPrepareLoop(uv_prepare_t * h) {

    static_cast< ::yajr::comms::internal::Peer::LoopData *>(h->data)
        ->onPrepareLoop();

}

void internal::Peer::LoopData::fini(uv_handle_t * h) {
    LOG(INFO);

    delete static_cast< ::yajr::comms::internal::Peer::LoopData *>(h->data);
}

void internal::Peer::LoopData::destroy(bool now) {
    LOG(INFO);

    assert(prepare_.data == this);

    assert(!destroying_);
    if (destroying_) {
        LOG(WARNING)
            << this
            << " Double destroy() detected"
        ;

        return;
    }

    destroying_ = 1;
    down();

    for (size_t i=0; i < Peer::LoopData::TOTAL_STATES; ++i) {
        peers[Peer::LoopData::PeerState(i)]
            .clear_and_dispose(PeerDisposer(now));

    }
}



/*
                    ____
                   / ___|___  _ __ ___  _ __ ___   ___  _ __
                  | |   / _ \| '_ ` _ \| '_ ` _ \ / _ \| '_ \
                  | |__| (_) | | | | | | | | | | | (_) | | | |
                   \____\___/|_| |_| |_|_| |_| |_|\___/|_| |_|

                     ____      _ _ _                _
                    / ___|__ _| | | |__   __ _  ___| | _____
                   | |   / _` | | | '_ \ / _` |/ __| |/ / __|
                   | |__| (_| | | | |_) | (_| | (__|   <\__ \
                    \____\__,_|_|_|_.__/ \__,_|\___|_|\_\___/

                                                              (Common Callbacks)
*/

namespace internal {

void on_close(uv_handle_t * h) {

    CommunicationPeer * peer = Peer::get<CommunicationPeer>(h);

    LOG(DEBUG) << peer << " down() for an on_close()";
    peer->down();

    peer->choked_ = 1;

    return;
}

void on_write(uv_write_t *req, int status) {

    LOG(DEBUG);

    if (status == UV_ECANCELED || status == UV_ECONNRESET) {
        LOG(INFO) << "[" << uv_err_name(status) << "] " << uv_strerror(status);
        return;  /* Handle has been closed. */
    }

    /* see here if more output can be dequeued */

    CommunicationPeer * peer = Peer::get(req);

    peer->onWrite();

    return;
}

} /* yajr::comms::internal namespace */

}} /* yajr::comms and opflex namespaces */

#ifdef BOOST_NO_EXCEPTIONS

namespace boost
{

void throw_exception(std::exception const & e) {
    assert(0);

    LOG(FATAL) << e.what();

    exit(127);
}

}

#endif


{'fraction_non_alphanumeric': 0.14919463691172502, 'fraction_numerical': 0.005669036263835148, 'mean_word_length': 2.19551466359977, 'pattern_counts': {'":': 2, '<': 117, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4522916', 'n_tokens_mistral': 3864, 'n_tokens_neox': 3643, 'n_words': 668}