<commit_before>﻿/*
 * MIT License
 *
 * Copyright (c) 2016-2019 xiongziliang <<EMAIL>>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
#ifndef UTIL_RINGBUFFER_H_
#define UTIL_RINGBUFFER_H_

#include <atomic>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <condition_variable>
#include <functional>
#include "Poller/EventPoller.h"

using namespace std;

//自适应环形缓存大小的取值范围
#define RING_MIN_SIZE 32
#define RING_MAX_SIZE 512
#define LOCK_GUARD(mtx) lock_guard<decltype(mtx)> lck(mtx)

namespace toolkit {

template<typename T>
class RingDelegate {
public:
	typedef std::shared_ptr<RingDelegate> Ptr;
	RingDelegate(){}
	virtual ~RingDelegate(){}
	virtual void onWrite(const T &in,bool isKey = true) = 0;
};

template<typename T>
class _RingStorage;

template<typename T>
class _RingStorageInternal;

template<typename T>
class _RingReaderDispatcher;

/**
 * 环形缓存读取器
 * 该对象的事件触发都会在绑定的poller线程中执行
 * 所以把锁去掉了
 * 对该对象的一切操作都应该在poller线程中执行
 * @tparam T
 */
template<typename T>
class _RingReader {
public:
    typedef std::shared_ptr<_RingReader> Ptr;
    friend class _RingReaderDispatcher<T>;

    _RingReader(const std::shared_ptr<_RingStorage<T> > &storage,bool useBuffer) {
        _storage = storage;
        _useBuffer = useBuffer;
        resetPos(true);
    }

    ~_RingReader() {}

    void setReadCB(const function<void(const T &)> &cb) {
        if (!cb) {
            _readCB = [](const T &) {};
        } else {
            _readCB = cb;
        }
        resetPos(true);
    }

    void setDetachCB(const function<void()> &cb) {
        if (!cb) {
            _detachCB = []() {};
        } else {
            _detachCB = cb;
        }
    }

    //重新定位至最新的数据位置
    void resetPos(bool key = true) {
        _storageInternal = _storage->getStorageInternal();
        _curpos = _storageInternal->getPos(key);
        _newReset = true;
    }
private:
    void onRead(const T &data,int targetPos) {
        if(!_newReset || !_useBuffer){
            _readCB(data);
        }else{
            const T *pkt  = nullptr;
            while((pkt = read(targetPos))){
                _readCB(*pkt);
            }
            _newReset = false;
        }
    }

    void onDetach() const {
        _detachCB();
    }

    const T* read(int targetPos){
        while (_curpos != targetPos){
            const T *data = (*_storageInternal)[_curpos]; //返回包
            _curpos = _storageInternal->next(_curpos); //更新位置
            if(!data){
                continue;
            }
            return data;
        }
        return nullptr;
    }
private:
    function<void(const T &)> _readCB = [](const T &) {};
    function<void(void)> _detachCB = []() {};

    shared_ptr<_RingStorageInternal<T> > _storageInternal;
    shared_ptr<_RingStorage<T> > _storage;
    int _curpos;
    bool _useBuffer;
    bool _newReset = false;
};
<commit_msg>优化GOP秒开<commit_after>﻿/*
 * MIT License
 *
 * Copyright (c) 2016-2019 xiongziliang <<EMAIL>>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
#ifndef UTIL_RINGBUFFER_H_
#define UTIL_RINGBUFFER_H_

#include <atomic>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <condition_variable>
#include <functional>
#include "Poller/EventPoller.h"

using namespace std;

//自适应环形缓存大小的取值范围
#define RING_MIN_SIZE 32
#define RING_MAX_SIZE 512
#define LOCK_GUARD(mtx) lock_guard<decltype(mtx)> lck(mtx)

namespace toolkit {

template<typename T>
class RingDelegate {
public:
	typedef std::shared_ptr<RingDelegate> Ptr;
	RingDelegate(){}
	virtual ~RingDelegate(){}
	virtual void onWrite(const T &in,bool isKey = true) = 0;
};

template<typename T>
class _RingStorage;

template<typename T>
class _RingStorageInternal;

template<typename T>
class _RingReaderDispatcher;

/**
 * 环形缓存读取器
 * 该对象的事件触发都会在绑定的poller线程中执行
 * 所以把锁去掉了
 * 对该对象的一切操作都应该在poller线程中执行
 * @tparam T
 */
template<typename T>
class _RingReader {
public:
    typedef std::shared_ptr<_RingReader> Ptr;
    friend class _RingReaderDispatcher<T>;

    _RingReader(const std::shared_ptr<_RingStorage<T> > &storage,bool useBuffer) {
        _storage = storage;
        _useBuffer = useBuffer;
    }

    ~_RingReader() {}

    void setReadCB(const function<void(const T &)> &cb) {
        if (!cb) {
            _readCB = [](const T &) {};
        } else {
            _readCB = cb;
        }
        resetPos(true);
    }

    void setDetachCB(const function<void()> &cb) {
        if (!cb) {
            _detachCB = []() {};
        } else {
            _detachCB = cb;
        }
    }

    //重新定位至最新的数据位置
    void resetPos(bool key = true) {
        _storageInternal = _storage->getStorageInternal();
        _curpos = _storageInternal->getPos(key);
        if(key){
            flushPacket(_storageInternal->getPos(false));
        }
    }
private:
    void onRead(const T &data,int targetPos) {
        if(!_useBuffer){
            _readCB(data);
        }else{
            flushPacket(targetPos);
        }
    }

    inline void flushPacket(int targetPos){
        const T *pkt  = nullptr;
        while((pkt = read(targetPos))){
            _readCB(*pkt);
        }
    }

    void onDetach() const {
        _detachCB();
    }

    const T* read(int targetPos){
        while (_curpos != targetPos){
            const T *data = (*_storageInternal)[_curpos]; //返回包
            _curpos = _storageInternal->next(_curpos); //更新位置
            if(!data){
                continue;
            }
            return data;
        }
        return nullptr;
    }
private:
    function<void(const T &)> _readCB = [](const T &) {};
    function<void(void)> _detachCB = []() {};

    shared_ptr<_RingStorageInternal<T> > _storageInternal;
    shared_ptr<_RingStorage<T> > _storage;
    int _curpos;
    bool _useBuffer;
};

{'fraction_non_alphanumeric': 0.10203820023073965, 'fraction_numerical': 0.003589283425201897, 'mean_word_length': 2.9543841865179927, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 64, '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': '5592263', 'n_tokens_mistral': 2549, 'n_tokens_neox': 2482, 'n_words': 718}