<commit_before>/*
 Copyright 2015 Nervana Systems 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 <assert.h>

#include <vector>
#include <cstdio>
#include <iostream>
#include <chrono>
#include <utility>
#include <algorithm>

#include "matrix.hpp"
#include "pyLoader.hpp"
#include "batch_loader_cpio_cache.hpp"
#include "sequential_batch_iterator.hpp"
#include "shuffled_batch_iterator.hpp"
#include "minibatch_iterator.hpp"

using namespace std;

pyDecodeThreadPool::pyDecodeThreadPool(int count,
                                       const shared_ptr<buffer_pool_in>& in,
                                       const shared_ptr<buffer_pool_out>& out,
                                       const shared_ptr<pyBackendWrapper>& pbe)
: ThreadPool(count), _in(in), _out(out), _pbe(pbe)
{
    _batchSize = _pbe->_batchSize;
    _itemsPerThread = (_batchSize - 1) / _count + 1;

    assert(_itemsPerThread * count >= _batchSize);
    assert(_itemsPerThread * (count - 1) < _batchSize);
}


void pyDecodeThreadPool::add_provider(std::shared_ptr<nervana::provider_interface> prov)
{
    _providers.push_back(prov);
    _startSignaled.push_back(0);

    _startInds.push_back(0);
    _endInds.push_back(0);
}

pyDecodeThreadPool::~pyDecodeThreadPool()
{
    if (_manager != 0) {
        _manager->join();
        delete _manager;
    }
    // The other thread objects are freed in the destructor
    // of the parent class.
}

void pyDecodeThreadPool::start()
{
    for (int i = 0; i < _count; i++) {
        _threads.push_back(new thread(&pyDecodeThreadPool::run, this, i));
    }
    _manager = new thread(&pyDecodeThreadPool::manage, this);
}

void pyDecodeThreadPool::stop()
{
    ThreadPool::stop();
    while (stopped() == false) {
        std::this_thread::yield();
        _in->advanceWritePos();
        _in->signalNonEmpty();
    }

    _stopManager = true;
    while (_managerStopped == false) {
        std::this_thread::yield();
        _in->advanceWritePos();
        _in->signalNonEmpty();
        _endSignaled++;
        _ended.notify_one();
    }
}

void pyDecodeThreadPool::run(int id)
{
    // Initialize worker threads by computing memory offsets for the
    // data this thread should work on
    assert(id < _count);
    _startInds[id] = id * _itemsPerThread;
    int itemCount = _itemsPerThread;
    if (id == _count - 1) {
        itemCount = _batchSize - id * _itemsPerThread;
    }

    _endInds[id] = _startInds[id] + itemCount;

    while (_done == false) {
        work(id);
    }

    _stopped[id] = true;
}

void pyDecodeThreadPool::work(int id)
{
    // Thread function.
    {
        unique_lock<mutex> lock(_mutex);
        while (_startSignaled[id] == 0) {
            _started.wait(lock);
            if (_done == true) {
                return;
            }
        }
        _startSignaled[id]--;
        assert(_startSignaled[id] == 0);
    }

    // No locking required because threads write into non-overlapping regions.
    buffer_out_array& outBuf = _out->getForWrite();

    for (int i = _startInds[id]; i < _endInds[id]; i++) {
        _providers[id]->provide(i, *_inputBuf, outBuf);
    }

    {
        lock_guard<mutex> lock(_mutex);
        _endSignaled++;
        assert(_endSignaled <= _count);
    }
    _ended.notify_one();
}

void pyDecodeThreadPool::produce()
{
    // Produce a minibatch.
    {
        unique_lock<mutex> lock(_out->getMutex());
        while (_out->full() == true) {
            _out->waitForNonFull(lock);
        }
        {
            lock_guard<mutex> lock(_mutex);
            for (unsigned int i = 0; i < _startSignaled.size(); i++) {
                _startSignaled[i] = 1;
            }
        }
        _started.notify_all();
        {
            unique_lock<mutex> lock(_mutex);
            while (_endSignaled < _count) {
                _ended.wait(lock);
            }
            _endSignaled = 0;
        }
        // At this point, we have decoded data for the whole minibatch.
        buffer_out_array& outBuf = _out->getForWrite();

        // Copy to device.
        _pbe->call_backend_transfer(outBuf, _bufferIndex);

        _bufferIndex = (_bufferIndex == 0) ? 1 : 0;
        _out->advanceWritePos();
    }
    _out->signalNonEmpty();
}

void pyDecodeThreadPool::consume()
{
    // Consume an input buffer.
    {
        unique_lock<mutex> lock(_in->getMutex());
        while (_in->empty() == true) {
            _in->waitForNonEmpty(lock);
            if (_stopManager == true) {
                return;
            }
        }
        _inputBuf = &_in->getForRead();
        produce();
        _in->advanceReadPos();
    }
    _in->signalNonFull();
}

void pyDecodeThreadPool::manage()<commit_msg>update comments<commit_after>/*
 Copyright 2015 Nervana Systems 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 <assert.h>

#include <vector>
#include <cstdio>
#include <iostream>
#include <chrono>
#include <utility>
#include <algorithm>

#include "matrix.hpp"
#include "pyLoader.hpp"
#include "batch_loader_cpio_cache.hpp"
#include "sequential_batch_iterator.hpp"
#include "shuffled_batch_iterator.hpp"
#include "minibatch_iterator.hpp"

using namespace std;

pyDecodeThreadPool::pyDecodeThreadPool(int count,
                                       const shared_ptr<buffer_pool_in>& in,
                                       const shared_ptr<buffer_pool_out>& out,
                                       const shared_ptr<pyBackendWrapper>& pbe)
: ThreadPool(count), _in(in), _out(out), _pbe(pbe)
{
    _batchSize = _pbe->_batchSize;
    _itemsPerThread = (_batchSize - 1) / _count + 1;

    assert(_itemsPerThread * count >= _batchSize);
    assert(_itemsPerThread * (count - 1) < _batchSize);
}


void pyDecodeThreadPool::add_provider(std::shared_ptr<nervana::provider_interface> prov)
{
    _providers.push_back(prov);
    _startSignaled.push_back(0);

    _startInds.push_back(0);
    _endInds.push_back(0);
}

pyDecodeThreadPool::~pyDecodeThreadPool()
{
    if (_manager != 0) {
        _manager->join();
        delete _manager;
    }
    // The other thread objects are freed in the destructor
    // of the parent class.
}

void pyDecodeThreadPool::start()
{
    for (int i = 0; i < _count; i++) {
        _threads.push_back(new thread(&pyDecodeThreadPool::run, this, i));
    }
    _manager = new thread(&pyDecodeThreadPool::manage, this);
}

void pyDecodeThreadPool::stop()
{
    ThreadPool::stop();
    while (stopped() == false) {
        std::this_thread::yield();
        _in->advanceWritePos();
        _in->signalNonEmpty();
    }

    _stopManager = true;
    while (_managerStopped == false) {
        std::this_thread::yield();
        _in->advanceWritePos();
        _in->signalNonEmpty();
        _endSignaled++;
        _ended.notify_one();
    }
}

void pyDecodeThreadPool::run(int id)
{
    // Initialize worker threads by computing memory offsets for the
    // data this thread should work on
    assert(id < _count);
    _startInds[id] = id * _itemsPerThread;
    int itemCount = _itemsPerThread;
    if (id == _count - 1) {
        itemCount = _batchSize - id * _itemsPerThread;
    }

    _endInds[id] = _startInds[id] + itemCount;

    while (_done == false) {
        work(id);
    }

    _stopped[id] = true;
}

void pyDecodeThreadPool::work(int id)
{
    // Thread function.
    {
        unique_lock<mutex> lock(_mutex);
        while (_startSignaled[id] == 0) {
            _started.wait(lock);
            if (_done == true) {
                return;
            }
        }
        _startSignaled[id]--;
        assert(_startSignaled[id] == 0);
    }

    // No locking required because threads write into non-overlapping regions.
    buffer_out_array& outBuf = _out->getForWrite();

    for (int i = _startInds[id]; i < _endInds[id]; i++) {
        _providers[id]->provide(i, *_inputBuf, outBuf);
    }

    {
        lock_guard<mutex> lock(_mutex);
        _endSignaled++;
        assert(_endSignaled <= _count);
    }
    _ended.notify_one();
}

void pyDecodeThreadPool::produce()
{
    // lock on output buffers and copy to device
    {
        unique_lock<mutex> lock(_out->getMutex());
        while (_out->full() == true) {
            _out->waitForNonFull(lock);
        }
        {
            lock_guard<mutex> lock(_mutex);
            for (unsigned int i = 0; i < _startSignaled.size(); i++) {
                _startSignaled[i] = 1;
            }
        }
        _started.notify_all();
        {
            unique_lock<mutex> lock(_mutex);
            while (_endSignaled < _count) {
                _ended.wait(lock);
            }
            _endSignaled = 0;
        }
        // At this point, we have decoded data for the whole minibatch.
        buffer_out_array& outBuf = _out->getForWrite();

        // Copy to device.
        _pbe->call_backend_transfer(outBuf, _bufferIndex);

        _bufferIndex = (_bufferIndex == 0) ? 1 : 0;
        _out->advanceWritePos();
    }
    _out->signalNonEmpty();
}

void pyDecodeThreadPool::consume()
{
    // lock on input buffers and call produce
    {
        unique_lock<mutex> lock(_in->getMutex());
        while (_in->empty() == true) {
            _in->waitForNonEmpty(lock);
            if (_stopManager == true) {
                return;
            }
        }
        _inputBuf = &_in->getForRead();
        produce();
        _in->advanceReadPos();
    }
    _in->signalNonFull();
}

void pyDecodeThreadPool::manage()
{'fraction_non_alphanumeric': 0.12007288769540615, 'fraction_numerical': 0.0047952431188261245, 'mean_word_length': 2.4393139841688654, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 81, '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': '3064835', 'n_tokens_mistral': 3266, 'n_tokens_neox': 3171, 'n_words': 758}