pregel.hpp

Go to the documentation of this file.

/*
 *   Copyright 2021 Huawei Technologies Co., Ltd.
 *
 * 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.
 */

#ifndef _H_GRB_INTERFACES_PREGEL
#define _H_GRB_INTERFACES_PREGEL

#include <graphblas.hpp>
#include <graphblas/utils/parser.hpp>

#include <stdexcept> // std::runtime_error


namespace grb {

    namespace interfaces {

        namespace config {

            enum SparsificationStrategy {

                NONE = 0,

                ALWAYS,

                WHEN_REDUCED,

                WHEN_HALVED

            };

            constexpr const SparsificationStrategy out_sparsify = NONE;

        } // end namespace grb::interfaces::config

        struct PregelState {

            bool &active;

            bool &voteToHalt;

            const size_t &num_vertices;

            const size_t &num_edges;

            const size_t &outdegree;

            const size_t &indegree;

            const size_t &round;

            const size_t &vertexID;

        };

        template<
            typename MatrixEntryType
        >
        class Pregel {

            private:

                const size_t n;

                size_t nz;

                grb::Matrix< MatrixEntryType > graph;

                grb::Vector< bool > activeVertices;

                grb::Vector< bool > haltVotes;

                grb::Vector< bool > buffer;

                grb::Vector< size_t > outdegrees;

                grb::Vector< size_t > indegrees;

                grb::Vector< size_t > IDs;

                void initialize() {
                    grb::Semiring<
                        grb::operators::add< size_t >,
                        grb::operators::right_assign_if< bool, size_t, size_t >,
                        grb::identities::zero,
                        grb::identities::logical_true
                    > ring;
                    grb::Vector< size_t > ones( n );
                    if( grb::set( ones, 1 ) != SUCCESS ) {
                        throw std::runtime_error( "Could not set vector ones" );
                    }
                    if( grb::set( outdegrees, 0 ) != SUCCESS ) {
                        throw std::runtime_error( "Could not initialise outdegrees" );
                    }
                    if( grb::mxv< grb::descriptors::dense >(
                            outdegrees, graph, ones, ring
                        ) != SUCCESS
                    ) {
                        throw std::runtime_error( "Could not compute outdegrees" );
                    }
                    if( grb::set( indegrees, 0 ) != SUCCESS ) {
                        throw std::runtime_error( "Could not initialise indegrees" );
                    }
                    if( grb::mxv<
                        grb::descriptors::dense | grb::descriptors::transpose_matrix
                    >(
                        indegrees, graph, ones, ring
                    ) != SUCCESS ) {
                        throw std::runtime_error( "Could not compute indegrees" );
                    }
                    if( grb::set< grb::descriptors::use_index >(
                            IDs, 0
                        ) != SUCCESS
                    ) {
                        throw std::runtime_error( "Could not compute vertex IDs" );
                    }
                }


            protected:

                template< typename IType >
                Pregel(
                    const size_t _n,
                    IType _start, const IType _end,
                    const grb::IOMode _mode
                ) :
                    n( _n ),
                    graph( _n, _n ),
                    activeVertices( _n ),
                    haltVotes( _n ),
                    buffer( _n ),
                    outdegrees( _n ),
                    indegrees( _n ),
                    IDs( _n )
                {
                    if( grb::ncols( graph ) != grb::nrows( graph ) ) {
                        throw std::runtime_error( "Input graph is bipartite" );
                    }
                    if( grb::buildMatrixUnique(
                        graph, _start, _end, _mode
                    ) != SUCCESS ) {
                        throw std::runtime_error( "Could not build graph" );
                    }
                    nz = grb::nnz( graph );
                    initialize();
                }


            public:

                template< typename IType >
                Pregel(
                    const size_t _m, const size_t _n,
                    IType _start, const IType _end,
                    const grb::IOMode _mode
                ) : Pregel( std::max( _m, _n ), _start, _end, _mode ) {}

                template<
                    class Op,
                    template< typename > class Id,
                    class Program,
                    typename IOType,
                    typename GlobalProgramData,
                    typename IncomingMessageType,
                    typename OutgoingMessageType
                >
                grb::RC execute(
                    const Program program,
                    grb::Vector< IOType > &vertex_state,
                    const GlobalProgramData &data,
                    grb::Vector< IncomingMessageType > &in,
                    grb::Vector< OutgoingMessageType > &out,
                    size_t &rounds,
                    grb::Vector< OutgoingMessageType > &out_buffer =
                        grb::Vector< OutgoingMessageType >(0),
                    const size_t max_rounds = 0
                ) {
                    static_assert( grb::is_operator< Op >::value &&
                            grb::is_associative< Op >::value,
                        "The combiner must be an associate operator"
                    );
                    static_assert( std::is_same< typename Op::D1, IncomingMessageType >::value,
                        "The combiner left-hand input domain should match the incoming message "
                        "type." );
                    static_assert( std::is_same< typename Op::D1, IncomingMessageType >::value,
                        "The combiner right-hand input domain should match the incoming message "
                        "type." );
                    static_assert( std::is_same< typename Op::D1, IncomingMessageType >::value,
                        "The combiner output domain should match the incoming message type." );

                    // set default output
                    rounds = 0;

                    // sanity checks
                    if( grb::size(vertex_state) != n ) {
                        return MISMATCH;
                    }
                    if( grb::size(in) != n ) {
                        return MISMATCH;
                    }
                    if( grb::size(out) != n ) {
                        return MISMATCH;
                    }
                    if( grb::capacity(vertex_state) != n ) {
                        return ILLEGAL;
                    }
                    if( grb::capacity(in) != n ) {
                        return ILLEGAL;
                    }
                    if( grb::capacity(out) != n ) {
                        return ILLEGAL;
                    }
                    if( config::out_sparsify && grb::capacity(out_buffer) != n ) {
                        return ILLEGAL;
                    }
                    if( grb::nnz(vertex_state) != n ) {
                        return ILLEGAL;
                    }

                    // define some monoids and semirings
                    grb::Monoid<
                        grb::operators::logical_or< bool >,
                        grb::identities::logical_false
                    > orMonoid;

                    grb::Monoid<
                        grb::operators::logical_and< bool >,
                        grb::identities::logical_true
                    > andMonoid;

                    grb::Semiring<
                        Op,
                        grb::operators::left_assign_if<
                            IncomingMessageType, bool, IncomingMessageType
                        >,
                        Id,
                        grb::identities::logical_true
                    > ring;

                    // set initial round ID
                    size_t step = 0;

                    // activate all vertices
                    grb::RC ret = grb::set( activeVertices, true );

                    // initialise halt votes to all-false
                    if( ret == SUCCESS ) {
                        ret = grb::set( haltVotes, false );
                    }

                    // set default incoming message
                    if( ret == SUCCESS && grb::nnz(in) < n ) {
#ifdef _DEBUG
                        if( grb::nnz(in) > 0 ) {
                            std::cerr << "Overwriting initial incoming messages since it was not a "
                                << "dense vector\n";
                        }
#endif
                        ret = grb::set( in, Id< IncomingMessageType >::value() );
                    }

                    // reset outgoing buffer
                    size_t out_nnz = n;
                    if( ret == SUCCESS ) {
                        ret = grb::set( out, Id< OutgoingMessageType >::value() );
                    }

                    // return if initialisation failed
                    if( ret != SUCCESS ) {
                        assert( ret == FAILED );
                        std::cerr << "Error: initialisation failed, but if workspace holds full "
                            << "capacity, initialisation should never fail. Please submit a bug "
                            << "report.\n";
                        return PANIC;
                    }

                    // while there are active vertices, execute
                    while( ret == SUCCESS ) {

                        assert( max_rounds == 0 || step < max_rounds );
                        // run one step of the program
                        ret = grb::eWiseLambda(
                            [
                                this,
                                &vertex_state,
                                &in,
                                &out,
                                &program,
                                step,
                                &data
                            ]( const size_t i ) {
                                // create Pregel struct
                                PregelState pregel = {
                                    activeVertices[ i ],
                                    haltVotes[ i ],
                                    n,
                                    nz,
                                    outdegrees[ i ],
                                    indegrees[ i ],
                                    step,
                                    IDs[ i ]
                                };
                                // only execute program on active vertices
                                assert( activeVertices[ i ] );
#ifdef _DEBUG
                                std::cout << "Vertex " << i << " remains active in step " << step
                                    << "\n";
#endif
                                program(
                                    vertex_state[ i ],
                                    in[ i ],
                                    out[ i ],
                                    data,
                                    pregel
                                );
#ifdef _DEBUG
                                std::cout << "Vertex " << i << " sends out message " << out[ i ]
                                    << "\n";
#endif
                            }, activeVertices, vertex_state, in, out, outdegrees, haltVotes, indegrees, IDs
                        );

                        // increment counter
                        (void) ++step;

                        // check if everyone voted to halt
                        if( ret == SUCCESS ) {
                            bool halt = true;
                            ret = grb::foldl< grb::descriptors::structural >(
                                halt, haltVotes, activeVertices, andMonoid
                            );
                            assert( ret == SUCCESS );
                            if( ret == SUCCESS && halt ) {
#ifdef _DEBUG
                                std::cout << "\t All active vertices voted to halt; "
                                    << "terminating Pregel program.\n";
#endif
                                break;
                            }
                        }

                        // update active vertices
                        if( ret == SUCCESS ) {
#ifdef _DEBUG
                            std::cout << "\t Number of active vertices was "
                                << grb::nnz( activeVertices ) << ", and ";
#endif
                            ret = grb::clear( buffer );
                            ret = ret ? ret : grb::set( buffer, activeVertices, true );
                            std::swap( buffer, activeVertices );
#ifdef _DEBUG
                            std::cout << " has now become " << grb::nnz( activeVertices ) << "\n";
#endif
                        }

                        // check if there is a next round
                        const size_t curActive = grb::nnz( activeVertices );
                        if( ret == SUCCESS && curActive == 0 ) {
#ifdef _DEBUG
                            std::cout << "\t All vertices are inactive; "
                                << "terminating Pregel program.\n";
#endif
                            break;
                        }

                        // check if we exceed the maximum number of rounds
                        if( max_rounds > 0 && step > max_rounds ) {
#ifdef _DEBUG
                            std::cout << "\t Maximum number of Pregel rounds met "
                                << "without the program returning a valid termination condition. "
                                << "Exiting prematurely with a FAILED error code.\n";
#endif
                            ret = FAILED;
                            break;
                        }

#ifdef _DEBUG
                        std::cout << "\t Starting message exchange\n";
#endif

                        // reset halt votes
                        if( ret == SUCCESS ) {
                            ret = grb::clear( haltVotes );
                            ret = ret ? ret : grb::set< grb::descriptors::structural >(
                                haltVotes, activeVertices, false
                            );
                        }

                        // reset incoming buffer
                        if( ret == SUCCESS ) {
                            ret = grb::clear( in );
                            ret = ret ? ret : grb::set< grb::descriptors::structural >(
                                in, activeVertices, Id< IncomingMessageType >::value()
                            );
                        }

                        // execute communication
                        if( ret == SUCCESS ) {
                            ret = grb::vxm< grb::descriptors::structural >(
                                in, activeVertices, out, graph, ring
                            );
                        }

                        // sparsify and reset outgoing buffer
                        if( config::out_sparsify && ret == SUCCESS ) {
                            if( config::out_sparsify == config::ALWAYS ||
                                (config::out_sparsify == config::WHEN_REDUCED && out_nnz > curActive) ||
                                (config::out_sparsify == config::WHEN_HALVED && curActive <= out_nnz/2)
                            ) {
                                ret = grb::clear( out_buffer );
                                ret = ret ? ret : grb::set< grb::descriptors::structural >(
                                        out_buffer, activeVertices, Id< OutgoingMessageType >::value()
                                    );
                                std::swap( out, out_buffer );
                                out_nnz = curActive;
                            }
                        }

#ifdef _DEBUG
                        std::cout << "\t Resetting outgoing message fields and "
                            << "starting next compute round\n";
#endif

                    }

#ifdef _DEBUG
                    if( grb::spmd<>::pid() == 0 ) {
                        std::cout << "Info: Pregel exits after " << step
                            << " rounds with error code " << ret
                            << " ( " << grb::toString(ret) << " )\n";
                    }
#endif

                    // done
                    rounds = step;
                    return ret;
                }

                size_t numVertices() const noexcept { return n; }

                size_t numEdges() const noexcept { return nz; }

                const grb::Matrix< MatrixEntryType > & getMatrix() const noexcept {
                    return graph;
                }

        };

    } // end namespace ``grb::interfaces''

} // end namespace ``grb''

#endif // end ``_H_GRB_INTERFACES_PREGEL''