cavis/libnd4j/include/execution/impl/Threads.cpp

/* ******************************************************************************
 *
 *
 * This program and the accompanying materials are made available under the
 * terms of the Apache License, Version 2.0 which is available at
 * https://www.apache.org/licenses/LICENSE-2.0.
 *
 *  See the NOTICE file distributed with this work for additional
 *  information regarding copyright ownership.
 * 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.
 *
 * SPDX-License-Identifier: Apache-2.0
 ******************************************************************************/

//
// @author raver119@gmail.com
//
#include <execution/Threads.h>
#include <execution/ThreadPool.h>
#include <vector>
#include <thread>
#include <helpers/logger.h>
#include <math/templatemath.h>
#include <helpers/shape.h>


namespace samediff {

    int ThreadsHelper::numberOfThreads(int maxThreads, uint64_t numberOfElements) {
        // let's see how many threads we actually need first
        auto optimalThreads = sd::math::nd4j_max<uint64_t>(1, numberOfElements / 1024);

        // now return the smallest value
        return sd::math::nd4j_min<int>(optimalThreads, maxThreads);
    }

    Span3::Span3(int64_t startX, int64_t stopX, int64_t incX, int64_t startY, int64_t stopY, int64_t incY, int64_t startZ, int64_t stopZ, int64_t incZ) {
        _startX = startX;
        _startY = startY;
        _startZ = startZ;
        _stopX = stopX;
        _stopY = stopY;
        _stopZ = stopZ;
        _incX = incX;
        _incY = incY;
        _incZ = incZ;
    }

    Span3 Span3::build(int loop, uint64_t threadID, uint64_t numThreads, int64_t startX, int64_t stopX, int64_t incX, int64_t startY, int64_t stopY, int64_t incY, int64_t startZ, int64_t stopZ, int64_t incZ) {
        switch (loop) {
            case 1: {
                    auto span = (stopX - startX) / numThreads;
                    auto s = span * threadID;
                    auto e = s + span;
                    if (threadID == numThreads - 1)
                        e = stopX;

                    return Span3(s, e, incX, startY, stopY, incY, startZ, stopZ, incZ);
                }
                break;
            case 2: {
                    auto span = (stopY - startY) / numThreads;
                    auto s = span * threadID;
                    auto e = s + span;
                    if (threadID == numThreads - 1)
                        e = stopY;

                    return Span3(startX, stopX, incX, s, e, incY, startZ, stopZ, incZ);
                }
                break;
            case 3: {
                    auto span = (stopZ - startZ) / numThreads;
                    auto s = span * threadID;
                    auto e = s + span;
                    if (threadID == numThreads - 1)
                        e = stopZ;

                    return Span3(startX, stopX, incX, startY, stopY, incY, s, e, incZ);
                }
                break;
            default:
                throw std::runtime_error("");
        }
        return Span3(startX, stopX, incX, startY, stopY, incY, startZ, stopZ, incZ);
    }

    Span::Span(int64_t startX, int64_t stopX, int64_t incX) {
        _startX = startX;
        _stopX = stopX;
        _incX = incX;
    }

    Span Span::build(uint64_t threadID, uint64_t numThreads, int64_t startX, int64_t stopX, int64_t incX) {
        auto span = (stopX - startX) / numThreads;
        auto s = span * threadID;
        auto e = s + span;
        if (threadID == numThreads - 1)
            e = stopX;

        return Span(s, e, incX);
    }

    Span2::Span2(int64_t startX, int64_t stopX, int64_t incX, int64_t startY, int64_t stopY, int64_t incY) {
        _startX = startX;
        _startY = startY;
        _stopX = stopX;
        _stopY = stopY;
        _incX = incX;
        _incY = incY;
    }


    Span2 Span2::build(int loop, uint64_t threadID, uint64_t numThreads, int64_t startX, int64_t stopX, int64_t incX, int64_t startY, int64_t stopY, int64_t incY) {

        switch (loop) {
            case 1: {
                    auto span = (stopX - startX) / numThreads;
                    auto s = span * threadID;
                    auto e = s + span;
                    if (threadID == numThreads - 1)
                        e = stopX;

                    return Span2(s, e, incX, startY, stopY, incY);
                }
                break;
            case 2: {
                    auto span = (stopY - startY) / numThreads;
                    auto s = span * threadID;
                    auto e = s + span;
                    if (threadID == numThreads - 1)
                        e = stopY;

                    return Span2(startX, stopX, incX, s, e, incY);
                }
                break;
            default:
                throw std::runtime_error("");
        }
    }

    int64_t Span::startX() const {
        return _startX;
    }

    int64_t Span::stopX() const {
        return _stopX;
    }

    int64_t Span::incX() const {
        return _incX;
    }

    int64_t Span2::startX() const {
        return _startX;
    }

    int64_t Span2::startY() const {
        return _startY;
    }

    int64_t Span2::stopX() const {
        return _stopX;
    }

    int64_t Span2::stopY() const {
        return _stopY;
    }

    int64_t Span2::incX() const {
        return _incX;
    }

    int64_t Span2::incY() const {
        return _incY;
    }

    int64_t Span3::startX() const {
        return _startX;
    }

    int64_t Span3::startY() const {
        return _startY;
    }

    int64_t Span3::startZ() const {
        return _startZ;
    }

    int64_t Span3::stopX() const {
        return _stopX;
    }

    int64_t Span3::stopY() const {
        return _stopY;
    }

    int64_t Span3::stopZ() const {
        return _stopZ;
    }

    int64_t Span3::incX() const {
        return _incX;
    }

    int64_t Span3::incY() const {
        return _incY;
    }

    int64_t Span3::incZ() const {
        return _incZ;
    }

    int ThreadsHelper::pickLoop2d(int numThreads, uint64_t itersX, uint64_t itersY) {
        // if one of dimensions is definitely too small - we just pick the other one
        if (itersX < numThreads && itersY >= numThreads)
            return 2;
        if (itersY < numThreads && itersX >= numThreads)
            return 1;

        // next step - we pick the most balanced dimension
        auto remX = itersX % numThreads;
        auto remY = itersY % numThreads;
        auto splitY = itersY / numThreads;

        // if there's no remainder left in some dimension - we're picking that dimension, because it'll be the most balanced work distribution
        if (remX == 0)
            return 1;
        if (remY == 0)
            return 2;

        // if there's no loop without a remainder - we're picking one with smaller remainder
        if (remX < remY)
            return 1;
        if (remY < remX && splitY >= 64) // we don't want too small splits over last dimension, or vectorization will fail
            return 2;
        // if loops are equally sized - give the preference to the first thread
        return 1;
    }


    static int threads_(int maxThreads, uint64_t elements) {

        if (elements == maxThreads) {
            return maxThreads;
        }
        else if (elements > maxThreads) {
            // if we have full load across thread, or at least half of threads can be utilized
            auto rem = elements % maxThreads;
            if (rem == 0 || rem >= maxThreads / 3)
                return maxThreads;
            else
                return threads_(maxThreads - 1, elements);

        }
        else if (elements < maxThreads) {
            return elements;
        }

        return 1;
    }

    int ThreadsHelper::numberOfThreads2d(int maxThreads, uint64_t iters_x, uint64_t iters_y) {
        // in some cases there's nothing to think about, part 1
        if (iters_x < maxThreads && iters_y < maxThreads)
            return sd::math::nd4j_max<int>(iters_x, iters_y);

        auto remX = iters_x % maxThreads;
        auto remY = iters_y % maxThreads;

        // in some cases there's nothing to think about, part 2
        if ((iters_x >= maxThreads && remX == 0 )|| (iters_y >= maxThreads && remY == 0))
            return maxThreads;

        // at this point we suppose that there's no loop perfectly matches number of our threads
        // so let's pick something as equal as possible
        if (iters_x > maxThreads || iters_y > maxThreads)
            return maxThreads;
        else
            return numberOfThreads2d(maxThreads - 1, iters_x, iters_y);
    }

    int ThreadsHelper::numberOfThreads3d(int maxThreads, uint64_t itersX, uint64_t itersY, uint64_t itersZ) {
        // we don't want to run underloaded threads
        if (itersX * itersY * itersZ <= 32)
            return 1;

        auto remX = itersX % maxThreads;
        auto remY = itersY % maxThreads;
        auto remZ = itersZ % maxThreads;

        // if we have perfect balance across one of dimensions - just go for it
        if ((itersX >= maxThreads && remX == 0) || (itersY >= maxThreads && remY == 0) || (itersZ >= maxThreads && remZ == 0))
            return maxThreads;

        int threadsX = 0, threadsY = 0, threadsZ = 0;

        // now we look into possible number of
        threadsX = threads_(maxThreads, itersX);
        threadsY = threads_(maxThreads, itersY);
        threadsZ = threads_(maxThreads, itersZ);

        // we want to split as close to outer loop as possible, so checking it out first
        if (threadsX >= threadsY && threadsX >= threadsZ)
            return threadsX;
        else if (threadsY >= threadsX && threadsY >= threadsZ)
            return threadsY;
        else if (threadsZ >= threadsX && threadsZ >= threadsY)
            return threadsZ;

        return 1;
    }

    int ThreadsHelper::pickLoop3d(int numThreads, uint64_t itersX, uint64_t itersY, uint64_t itersZ) {
        auto remX = itersX % numThreads;
        auto remY = itersY % numThreads;
        auto remZ = itersZ % numThreads;

        auto splitX = itersX / numThreads;
        auto splitY = itersY / numThreads;
        auto splitZ = itersZ / numThreads;

        // if there's no remainder left in some dimension - we're picking that dimension, because it'll be the most balanced work distribution
        if (remX == 0)
            return 1;
        else if (remY == 0)
            return 2;
        else if (remZ == 0) // TODO: we don't want too smal splits over last dimension? or we do?
            return 3;

        if (itersX > numThreads)
            return 1;
        else if (itersY > numThreads)
            return 2;
        else if (itersZ > numThreads)
            return 3;

        return 1;
    }

    int Threads::parallel_tad(FUNC_1D function, int64_t start, int64_t stop, int64_t increment, uint32_t numThreads) {
        if (start > stop)
            throw std::runtime_error("Threads::parallel_for got start > stop");

        auto delta = (stop - start);

        if (numThreads > delta)
            numThreads = delta;

        if (numThreads == 0)
            return 0;

        // shortcut
        if (numThreads == 1) {
            function(0, start, stop, increment);
            return 1;
        }

        auto ticket = ThreadPool::getInstance().tryAcquire(numThreads);
        if (ticket != nullptr) {
            // if we got our threads - we'll run our jobs here
            auto span = delta / numThreads;

            for (uint32_t e = 0; e < numThreads; e++) {
                auto start_ = span * e + start;
                auto stop_  = start_ + span;

                // last thread will process tail
                if (e == numThreads - 1)
                    stop_ = stop;

                // putting the task into the queue for a given thread
                ticket->enqueue(e, numThreads, function, start_, stop_, increment);
            }

            // block and wait till all threads finished the job
            ticket->waitAndRelease();

            // we tell that parallelism request succeeded
            return numThreads;
        } else {
            // if there were no threads available - we'll execute function right within current thread
            function(0, start, stop, increment);

            // we tell that parallelism request declined
            return 1;
        }
    }

    int Threads::parallel_for(FUNC_1D function, int64_t start, int64_t stop, int64_t increment, uint32_t numThreads) {
        if (start > stop)
            throw std::runtime_error("Threads::parallel_for got start > stop");

        auto delta = (stop - start);

        // in some cases we just fire func as is
        if (delta == 0 || numThreads == 1) {
            function(0, start, stop, increment);
            return 1;
        }

        auto numElements = delta / increment;

        // we decide what's optimal number of threads we need here, and execute it in parallel_tad.
        numThreads = ThreadsHelper::numberOfThreads(numThreads, numElements);
        return parallel_tad(function, start, stop, increment, numThreads);
    }

    int Threads::parallel_for(FUNC_2D function, int64_t startX, int64_t stopX, int64_t incX, int64_t startY, int64_t stopY, int64_t incY, uint64_t numThreads, bool debug) {
        if (startX > stopX)
            throw std::runtime_error("Threads::parallel_for got startX > stopX");

        if (startY > stopY)
            throw std::runtime_error("Threads::parallel_for got startY > stopY");

        // number of elements per loop
        auto delta_x = (stopX - startX);
        auto delta_y = (stopY - startY);

        // number of iterations per loop
        auto itersX = delta_x / incX;
        auto itersY = delta_y / incY;

        // total number of iterations
        auto iters_t = itersX * itersY;

        // we are checking the case of number of requested threads was smaller
        numThreads = ThreadsHelper::numberOfThreads2d(numThreads, itersX, itersY);

        // basic shortcut for no-threading cases
        if (numThreads == 1) {
            function(0, startX, stopX, incX, startY, stopY, incY);
            return 1;
        }

        // We have couple of scenarios:
        // either we split workload along 1st loop, or 2nd
        auto splitLoop = ThreadsHelper::pickLoop2d(numThreads, itersX, itersY);

        // for debug mode we execute things inplace, without any threads
        if (debug) {
            for (int e = 0; e < numThreads; e++) {
                auto span = Span2::build(splitLoop, e, numThreads, startX, stopX, incX, startY, stopY, incY);

                function(e, span.startX(), span.stopX(), span.incX(), span.startY(), span.stopY(), span.incY());
            }

            // but we still mimic multithreaded execution
            return numThreads;
        } else {
            auto ticket = ThreadPool::getInstance().tryAcquire(numThreads);
            if (ticket != nullptr) {

                for (int e = 0; e < numThreads; e++) {
                    auto threadId = numThreads - e - 1;
                    auto span = Span2::build(splitLoop, threadId, numThreads, startX, stopX, incX, startY, stopY, incY);

                    ticket->enqueue(e, numThreads, function, span.startX(), span.stopX(), span.incX(), span.startY(), span.stopY(), span.incY());
                }

                // block until all threads finish their job
                ticket->waitAndRelease();

                return numThreads;
            } else {
                // if there were no threads available - we'll execute function right within current thread
                function(0, startX, stopX, incX, startY, stopY, incY);

                // we tell that parallelism request declined
                return 1;
            }
        };
    }


    int Threads::parallel_for(FUNC_3D function, int64_t startX, int64_t stopX, int64_t incX, int64_t startY, int64_t stopY, int64_t incY, int64_t startZ, int64_t stopZ, int64_t incZ, uint64_t numThreads) {
        if (startX > stopX)
            throw std::runtime_error("Threads::parallel_for got startX > stopX");

        if (startY > stopY)
            throw std::runtime_error("Threads::parallel_for got startY > stopY");

        if (startZ > stopZ)
            throw std::runtime_error("Threads::parallel_for got startZ > stopZ");

        auto delta_x = stopX - startX;
        auto delta_y = stopY - startY;
        auto delta_z = stopZ - startZ;

        auto itersX = delta_x / incX;
        auto itersY = delta_y / incY;
        auto itersZ = delta_z / incZ;

        numThreads = ThreadsHelper::numberOfThreads3d(numThreads, itersX, itersY, itersZ);
        if (numThreads == 1) {
            // loop is too small - executing function as is
            function(0, startX, stopX, incX, startY, stopY, incY, startZ, stopZ, incZ);
            return 1;
        }

        auto ticket = ThreadPool::getInstance().tryAcquire(numThreads);
        if (ticket != nullptr) {
            auto splitLoop = ThreadsHelper::pickLoop3d(numThreads, itersX, itersY, itersZ);

            for (int e = 0; e < numThreads; e++) {
                auto thread_id = numThreads - e - 1;
                auto span = Span3::build(splitLoop, thread_id, numThreads, startX, stopX, incX, startY, stopY, incY, startZ, stopZ, incZ);

                ticket->enqueue(e, numThreads, function, span.startX(), span.stopX(), span.incX(), span.startY(), span.stopY(), span.incY(), span.startZ(), span.stopZ(), span.incZ());
            }

            // block until we're done
            ticket->waitAndRelease();

            // we tell that parallelism request succeeded
            return numThreads;
        } else {
            // if there were no threads available - we'll execute function right within current thread
            function(0, startX, stopX, incX, startY, stopY, incY, startZ, stopZ, incZ);

            // we tell that parallelism request declined
            return 1;
        }

    }

    int Threads::parallel_do(FUNC_DO function, uint64_t numThreads) {
        auto ticket = ThreadPool::getInstance().tryAcquire(numThreads - 1);
        if (ticket != nullptr) {

            // submit tasks one by one
            for (uint64_t e = 0; e < numThreads - 1; e++)
                ticket->enqueue(e, numThreads, function);

            function(numThreads - 1, numThreads);

            ticket->waitAndRelease();

            return numThreads;
        } else {
            // if there's no threads available - we'll execute function sequentially one by one
            for (uint64_t e = 0; e < numThreads; e++)
                function(e, numThreads);

            return numThreads;
        }


        return numThreads;
    }

    int64_t Threads::parallel_long(FUNC_RL function, FUNC_AL aggregator, int64_t start, int64_t stop, int64_t increment, uint64_t numThreads) {
        if (start > stop)
            throw std::runtime_error("Threads::parallel_long got start > stop");

        auto delta = (stop - start);
        if (delta == 0 || numThreads == 1)
            return function(0, start, stop, increment);

        auto numElements = delta / increment;

        // we decide what's optimal number of threads we need here, and execute it
        numThreads = ThreadsHelper::numberOfThreads(numThreads, numElements);
        if (numThreads == 1)
            return function(0, start, stop, increment);

        auto ticket = ThreadPool::getInstance().tryAcquire(numThreads - 1);
        if (ticket == nullptr)
            return function(0, start, stop, increment);

        // create temporary array
        int64_t intermediatery[256];
        auto span = (numElements / numThreads) - (numElements % numThreads);

        // execute threads in parallel
        for (uint32_t e = 0; e < numThreads; e++) {
            auto start_ = span * e + start;
            auto stop_ = span * (e + 1) + start;

            if (e == numThreads - 1)
                intermediatery[e] = function(e, start_, stop, increment);
            else
                ticket->enqueue(e, numThreads, &intermediatery[e], function, start_, stop_, increment);
        }

        ticket->waitAndRelease();

        // aggregate results in single thread
        for (uint64_t e = 1; e < numThreads; e++)
            intermediatery[0] = aggregator(intermediatery[0], intermediatery[e]);

        // return accumulated result
        return intermediatery[0];
    }

    double Threads::parallel_double(FUNC_RD function, FUNC_AD aggregator, int64_t start, int64_t stop, int64_t increment, uint64_t numThreads) {
        if (start > stop)
            throw std::runtime_error("Threads::parallel_long got start > stop");

        auto delta = (stop - start);
        if (delta == 0 || numThreads == 1)
            return function(0, start, stop, increment);

        auto numElements = delta / increment;

        // we decide what's optimal number of threads we need here, and execute it
        numThreads = ThreadsHelper::numberOfThreads(numThreads, numElements);
        if (numThreads == 1)
            return function(0, start, stop, increment);

        auto ticket = ThreadPool::getInstance().tryAcquire(numThreads - 1);
        if (ticket == nullptr)
            return function(0, start, stop, increment);

        // create temporary array
        double intermediatery[256];
        auto span = (numElements / numThreads) - (numElements % numThreads);

        // execute threads in parallel
        for (uint32_t e = 0; e < numThreads; e++) {
            auto start_ = span * e + start;
            auto stop_ = span * (e + 1) + start;

            if (e == numThreads - 1)
                intermediatery[e] = function(e, start_, stop, increment);
            else
                ticket->enqueue(e, numThreads, &intermediatery[e], function, start_, stop_, increment);
        }

        ticket->waitAndRelease();

        // aggregate results in single thread
        for (uint64_t e = 1; e < numThreads; e++)
            intermediatery[0] = aggregator(intermediatery[0], intermediatery[e]);

        // return accumulated result
        return intermediatery[0];
    }


    int  Threads::parallel_aligned_increment(FUNC_1D function, int64_t start, int64_t stop, int64_t increment, size_t type_size , uint32_t req_numThreads) {
        if (start > stop)
            throw std::runtime_error("Threads::parallel_for got start > stop");
        auto num_elements = (stop - start);
        //this way we preserve increment starts offset
        //so we will parition considering delta but not total elements
        auto delta = (stop - start) / increment;

        // in some cases we just fire func as is
        if (delta == 0 || req_numThreads == 1) {
            function(0, start, stop, increment);
            return 1;
        }
        int numThreads = 0;

        int adjusted_numThreads = samediff::ThreadsHelper::numberOfThreads(req_numThreads, (num_elements * sizeof(double)) / (200 * type_size));

        if (adjusted_numThreads > delta)
            adjusted_numThreads = delta;
        // shortcut
        if (adjusted_numThreads <= 1) {
            function(0, start, stop, increment);
            return 1;
        }
        //take span as ceil  
        auto spand = std::ceil((double)delta / (double)adjusted_numThreads);
        numThreads = static_cast<int>(std::ceil((double)delta / spand));
        auto span  = static_cast<Nd4jLong>(spand);

        auto ticket = samediff::ThreadPool::getInstance().tryAcquire(numThreads);
        if (ticket != nullptr) {
            //tail_add is additional value of the last part
            //it could be negative or positive
            //we will spread that value across
            auto tail_add = delta - numThreads * span;
            Nd4jLong begin = 0;
            Nd4jLong end = 0;

            //we will try enqueu bigger parts first
            decltype(span) span1, span2;
            int last = 0;
            if (tail_add >= 0) {
                //for span == 1  , tail_add is  0 
                last = tail_add;
                span1 = span + 1;
                span2 = span;
            }
            else {
                last = numThreads + tail_add;// -std::abs(tail_add);
                span1 = span;
                span2 = span - 1;
            }
            for (int i = 0; i < last; i++) {
                end = begin + span1 * increment;
                // putting the task into the queue for a given thread
                ticket->enqueue(i, numThreads, function, begin, end, increment);
                begin = end;
            }
            for (int i = last; i < numThreads - 1; i++) {
                end = begin + span2 * increment;
                // putting the task into the queue for a given thread
                ticket->enqueue(i, numThreads, function, begin, end, increment);
                begin = end;
            }
            //for last one enqueue last offset as stop
            //we need it in case our ((stop-start) % increment ) > 0
            ticket->enqueue(numThreads - 1, numThreads, function, begin, stop, increment);
            // block and wait till all threads finished the job
            ticket->waitAndRelease();
            // we tell that parallelism request succeeded
            return numThreads;
        }
        else {
            // if there were no threads available - we'll execute function right within current thread
            function(0, start, stop, increment);
            // we tell that parallelism request declined
            return 1;
        }
    }


}