compressor.cc

#include "gbbs/encodings/byte_pd_amortized.h"
#include "gbbs/gbbs.h"
#include "gbbs/io.h"
#include "gbbs/parse_command_line.h"
#include "pbbslib/utilities.h"
#include "pbbslib/random.h"

#include <iostream>
#include <fstream>
#include <stdlib.h>
#include <cmath>

// Provides utilities for converting between different compressed
// representations.

namespace gbbs {

  template <class Graph>
  void write_graph_bytepd_amortized_directed(Graph& GA, std::ofstream& out) {
    namespace encoding = bytepd_amortized;
    using W = typename Graph::weight_type;
    size_t n = GA.n;

    // out-edges
    // 1. Calculate total size
    {
      auto degrees = sequence<uintE>(n);
      auto byte_offsets = sequence<uintT>(n+1);
      par_for(0, n, [&] (size_t i) {
        size_t total_bytes = 0;
        uintE last_ngh = 0;
        size_t deg = 0;
        uchar tmp[16];
        auto f = [&] (uintE u, uintE v, W w) {
          long bytes = 0;
          if ((deg % PARALLEL_DEGREE) == 0) {
            bytes = encoding::compressFirstEdge(tmp, bytes, u, v);
            bytes = encoding::compressWeight<W>(tmp, bytes, w);
          } else {
            bytes = encoding::compressEdge(tmp, bytes, v - last_ngh);
            bytes = encoding::compressWeight<W>(tmp, bytes, w);
          }
          last_ngh = v;
          total_bytes += bytes;
          deg++;
          return false;
        };
        GA.get_vertex(i).out_neighbors().map(f, false);

        if (deg > 0) {
          size_t n_chunks = 1+(deg-1)/PARALLEL_DEGREE;
          // To account for the byte offsets
          total_bytes += (n_chunks-1)*sizeof(uintE);
          // To account for the per-block counters
          total_bytes += (n_chunks)*sizeof(uintE);
          // To account for the virtual degree
          total_bytes += sizeof(uintE);
        }

        degrees[i] = deg;
        byte_offsets[i] = total_bytes;
      });
      byte_offsets[n] = 0;
      size_t total_space = pbbslib::scan_add_inplace(byte_offsets);
      std::cout << "total in-space = " << total_space << std::endl;

      // 2. Create compressed format in-memory
      auto edges = sequence<uchar>(total_space);
      par_for(0, n, [&] (size_t i) {
        uintE deg = degrees[i];
        if (deg > 0) {
          auto it = GA.get_vertex(i).out_neighbors().get_iter();
          size_t nbytes = encoding::sequentialCompressEdgeSet<W>(edges.begin() + byte_offsets[i], 0, deg, (uintE)i, it);
          if (nbytes != (byte_offsets[i+1] - byte_offsets[i])) {
            std::cout << "nbytes = " << nbytes << ". Should be: " << (byte_offsets[i+1] - byte_offsets[i]) << " deg = " << deg << " i = " << i << std::endl;
            exit(0);
          }
          assert(nbytes == (byte_offsets[i+1] - byte_offsets[i]));
        }
      });
      std::cout << "Compressed" << std::endl;

      long* sizes = pbbslib::new_array_no_init<long>(3);
      sizes[0] = GA.n;
      sizes[1] = GA.m;
      sizes[2] = total_space;
      out.write((char*)sizes,sizeof(long)*3); //write n, m and space used
      out.write((char*)byte_offsets.begin(),sizeof(uintT)*(n+1)); //write offsets
      out.write((char*)degrees.begin(),sizeof(uintE)*n);
      out.write((char*)edges.begin(),total_space); //write edges
    }

    {
      // in-edges
      // 1. Calculate total size
      auto degrees = sequence<uintE>(n);
      auto byte_offsets = sequence<uintT>(n+1);
      par_for(0, n, [&] (size_t i) {
        size_t total_bytes = 0;
        uintE last_ngh = 0;
        size_t deg = 0;
        uchar tmp[16];
        auto f = [&] (uintE u, uintE v, W w) {
          long bytes = 0;
          if ((deg % PARALLEL_DEGREE) == 0) {
            bytes = encoding::compressFirstEdge(tmp, bytes, u, v);
            bytes = encoding::compressWeight<W>(tmp, bytes, w);
          } else {
            bytes = encoding::compressEdge(tmp, bytes, v - last_ngh);
            bytes = encoding::compressWeight<W>(tmp, bytes, w);
          }
          last_ngh = v;
          total_bytes += bytes;
          deg++;
          return false;
        };
        GA.get_vertex(i).in_neighbors().map(f, false);

        if (deg > 0) {
          size_t n_chunks = 1+(deg-1)/PARALLEL_DEGREE;
          // To account for the byte offsets
          total_bytes += (n_chunks-1)*sizeof(uintE);
          // To account for the per-block counters
          total_bytes += (n_chunks)*sizeof(uintE);
          // To account for the virtual degree
          total_bytes += sizeof(uintE);
        }

        degrees[i] = deg;
        byte_offsets[i] = total_bytes;
      });
      byte_offsets[n] = 0;
      size_t total_space = pbbslib::scan_add_inplace(byte_offsets);
      std::cout << "total in-space = " << total_space << std::endl;

      // 2. Create compressed format in-memory
      auto edges = sequence<uchar>(total_space);
      par_for(0, n, [&] (size_t i) {
        uintE deg = degrees[i];
        if (deg > 0) {
          auto it = GA.get_vertex(i).in_neighbors().get_iter();
          size_t nbytes = encoding::sequentialCompressEdgeSet<W>(edges.begin() + byte_offsets[i], 0, deg, (uintE)i, it);
          if (nbytes != (byte_offsets[i+1] - byte_offsets[i])) {
            std::cout << "nbytes = " << nbytes << ". Should be: " << (byte_offsets[i+1] - byte_offsets[i]) << " deg = " << deg << " i = " << i << std::endl;
            exit(0);
          }
          assert(nbytes == (byte_offsets[i+1] - byte_offsets[i]));
        }
      });
      std::cout << "Compressed" << std::endl;
      long inTotalSpace[1];
      inTotalSpace[0] = total_space;
      out.write((char*)inTotalSpace, sizeof(long)); // in-edges total space
      out.write((char*)byte_offsets.begin(),sizeof(uintT)*(n+1)); //write offsets
      out.write((char*)degrees.begin(),sizeof(uintE)*n);
      out.write((char*)edges.begin(),total_space); //write edges
    }
  }

  template <class Graph>
  void write_graph_bytepd_amortized_format(Graph& GA, std::ofstream& out, bool symmetric) {
    namespace encoding = bytepd_amortized;
    using W = typename Graph::weight_type;
    if (!symmetric) {
      write_graph_bytepd_amortized_directed(GA, out);
      return;
    }
    size_t n = GA.n;

//    auto xors = sequence<size_t>(n);
//    parallel_for(size_t i=0; i<n; i++) {
//      size_t xr = 0;
//      auto map_f = wrap_f<W>([&] (uintE src, uintE ngh) {
//        xr ^= (src ^ ngh);
//      });
//      GA.V[i].mapOutNgh(i, map_f, false);
//      xors[i] = xr;
//    }
//    std::cout << "input graph: output red = " << pbbslib::reduce_xor(xors) << std::endl;
//
//  auto hash_or_lt = [&] (const uintE& src, const uintE& ngh) {
//    uint32_t src_h = pbbslib::hash32(src);
//    uint32_t ngh_h = pbbslib::hash32(ngh);
//    return (src_h < ngh_h) || ((src_h == ngh_h) && src < ngh);
//  };


//    auto self_arr = sequence<size_t>(n);
//    parallel_for(size_t i=0; i<n; i++) {
//      uintE our_deg = pbbslib::log2_up(GA.V[i].getOutDegree());
//      bool selfl = false;
//      size_t pri = 0;
//      auto map_f = wrap_f<W>([&] (uintE src, uintE ngh) {
//        uintE ngh_deg = pbbslib::log2_up(GA.V[ngh].getOutDegree());
//        if (src == ngh) {
//          selfl = true;
//        }
//        if ((ngh_deg > our_deg) || ((ngh_deg == our_deg) && hash_or_lt(src, ngh))) {
//          pri++;
//        }
//      });
//      GA.V[i].mapOutNgh(i, map_f, false);
//      self_arr[i] = selfl;
//      xors[i] = pri;
//    }
//    std::cout << "input graph: priorities = " << pbbslib::reduce_add(xors) << std::endl;
//    std::cout << "input graph: self-loops = " << pbbslib::reduce_add(self_arr) << std::endl;

//    parallel_for(size_t i=0; i<n; i++) {
//      uintE our_deg = pbbslib::log2_up(GA.V[i].getOutDegree());
//      bool selfl = false;
//      size_t pri = 0;
//      auto it = GA.V[i].getOutIter(i);
//      size_t d = 0;
//      size_t degree = GA.V[i].getOutDegree();
//      uintE src = i;
//      if (degree > 0) {
//        uintE ngh = get<0>(it.cur());
//        if (src == ngh) { selfl = true; }
//        uintE ngh_deg = pbbslib::log2_up(GA.V[ngh].getOutDegree());
//        if ((ngh_deg > our_deg) || ((ngh_deg == our_deg) && hash_or_lt(src, ngh))) {
//          pri++;
//        }
//        for (size_t i=1; i<degree; i++) {
//          ngh = get<0>(it.next());
//          ngh_deg = pbbslib::log2_up(GA.V[ngh].getOutDegree());
//          if (src == ngh) { selfl = true; }
//          if ((ngh_deg > our_deg) || ((ngh_deg == our_deg) && hash_or_lt(src, ngh))) {
//            pri++;
//          }
//        }
//      }
//      self_arr[i] = selfl;
//      xors[i] = pri;
//    }
//    std::cout << "input graph: priorities = " << pbbslib::reduce_add(xors) << std::endl;
//    std::cout << "input graph: self-loops = " << pbbslib::reduce_add(self_arr) << std::endl;


    // 1. Calculate total size
    auto degrees = sequence<uintE>(n);
    auto byte_offsets = sequence<uintT>(n+1);
    par_for(0, n, [&] (size_t i) {
      size_t total_bytes = 0;
      uintE last_ngh = 0;
      size_t deg = 0;
      uchar tmp[16];
      auto f = [&] (uintE u, uintE v, W w) {
//        if (u == v) {
//          return;
//        }
        long bytes = 0;
        if ((deg % PARALLEL_DEGREE) == 0) {
          bytes = encoding::compressFirstEdge(tmp, bytes, u, v);
          bytes = encoding::compressWeight<W>(tmp, bytes, w);
        } else {
          bytes = encoding::compressEdge(tmp, bytes, v - last_ngh);
          bytes = encoding::compressWeight<W>(tmp, bytes, w);
        }
        last_ngh = v;
        total_bytes += bytes;
        deg++;
      };
      GA.get_vertex(i).out_neighbors().map(f, false);

      if (deg > 0) {
        size_t n_chunks = 1+(deg-1)/PARALLEL_DEGREE;
        // To account for the byte offsets
        total_bytes += (n_chunks-1)*sizeof(uintE);
        // To account for the per-block counters
        total_bytes += (n_chunks)*sizeof(uintE);
        // To account for the virtual degree
        total_bytes += sizeof(uintE);
      }

      degrees[i] = deg;
      byte_offsets[i] = total_bytes;
    });
    byte_offsets[n] = 0;
    size_t total_space = pbbslib::scan_add_inplace(byte_offsets);
    std::cout << "total space = " << total_space << std::endl;
    auto deg_f = [&] (size_t i) { return degrees[i]; };
    auto deg_im = pbbslib::make_sequence<size_t>(n, deg_f);
    std::cout << "sum degs = " << pbbslib::reduce_add(deg_im) << std::endl;

    // 2. Create compressed format in-memory
    auto edges = sequence<uchar>(total_space);
    par_for(0, n, [&] (size_t i) {
      uintE deg = degrees[i];
      if (deg > 0) {
        auto it = GA.get_vertex(i).out_neighbors().get_iter();
        size_t nbytes = encoding::sequentialCompressEdgeSet<W>(edges.begin() + byte_offsets[i], 0, deg, (uintE)i, it);

//        uchar* edgeArray = edges.begin() + byte_offsets[i];
//        size_t degree = deg;
//
//        size_t current_offset = 0;
//        size_t num_blocks = 1+(degree-1)/PARALLEL_DEGREE;
//        uintE* vertex_ctr = (uintE*)edgeArray;
//        *vertex_ctr = degree;
//
//        uintE* block_offsets = (uintE*)(edgeArray + sizeof(uintE));
//        current_offset += sizeof(uintE) + (num_blocks-1)*sizeof(uintE); // virtual deg + block_offs
//
//        size_t proc = 0;
//        size_t cur_block = 0;
//        uintE* prev_block_deg = nullptr;
//        uintE last_ngh;
//        auto map_f = [&] (uintE u, uintE v, W w) {
//          if ((proc % PARALLEL_DEGREE) == 0) {
//            if (cur_block > 0) {
//              assert(*prev_block_deg == 0);
//              *prev_block_deg = PARALLEL_DEGREE; // full block; write prev block's degree
//              block_offsets[cur_block-1] = current_offset; // write start of this block
//            }
//            prev_block_deg = (uintE*)(edgeArray + current_offset);
//            *prev_block_deg = 0;
//            current_offset += sizeof(uintE);
//            current_offset = compressFirstEdge(edgeArray, current_offset, u, v);
//            current_offset = compressWeight<W>(edgeArray, current_offset, w);
//            cur_block++;
//          } else {
//            current_offset = compressEdge(edgeArray, current_offset, v - last_ngh);
//            current_offset = compressWeight<W>(edgeArray, current_offset, w);
//          }
//          last_ngh = v;
//          proc++;
//        };
//
//        GA.V[i].mapOutNgh(i, map_f, false);
//
//        assert(prev_block_deg != nullptr);
//        assert(*prev_block_deg == 0);
//        *prev_block_deg = deg % PARALLEL_DEGREE;
//
//        assert(current_offset == (byte_offsets[i+1] - byte_offsets[i]));

        if (nbytes != (byte_offsets[i+1] - byte_offsets[i])) {
          std::cout << "nbytes = " << nbytes << ". Should be: " << (byte_offsets[i+1] - byte_offsets[i]) << " deg = " << deg << " i = " << i << std::endl;
          exit(0);
        }
        assert(nbytes == (byte_offsets[i+1] - byte_offsets[i]));
      }
    });
    std::cout << "Compressed" << std::endl;
//    exit(0);

//    parallel_for(size_t i=0; i<n; i++) {
//      size_t xr = 0;
//      auto map_f = [&] (uintE src, uintE ngh, const W& wgh, size_t off) {
//        xr ^= (src ^ ngh);
//        return true;
//      };
//      auto edge_start= edges.begin() + byte_offsets[i];
//      size_t deg = degrees[i];
//      if (deg > 0) {
//        bytepd_amortized::decode<W>(map_f, edge_start, i, deg, false);
//      }
//      xors[i] = xr;
//    }
//    std::cout << "output graph: output red = " << pbbslib::reduce_xor(xors) << std::endl;

//    parallel_for(size_t i=0; i<n; i++) {
//      assert(degrees[i] == GA.V[i].getOutDegree());
//      uintE our_deg = pbbslib::log2_up(degrees[i]);
//      bool selfl = false;
//      size_t pri = 0;
//      auto map_f = [&] (uintE src, uintE ngh, const W& wgh, size_t off) {
//        uintE ngh_deg = pbbslib::log2_up(degrees[ngh]);
//        if (src == ngh) {
//          selfl = true;
//        }
//        if ((ngh_deg > our_deg) || ((ngh_deg == our_deg) && hash_or_lt(src, ngh))) {
//          pri++;
//        }
//        return true;
//      };
//      auto edge_start= edges.begin() + byte_offsets[i];
//      size_t deg = degrees[i];
//      if (deg > 0) {
//        bytepd_amortized::decode<W>(map_f, edge_start, i, deg, false);
//      }
//      self_arr[i] = selfl;
//      xors[i] = pri;
//    }
//    std::cout << "output graph: priorities = " << pbbslib::reduce_add(xors) << std::endl;
//    std::cout << "output graph: self-loops = " << pbbslib::reduce_add(self_arr) << std::endl;


//    exit(0);

    long* sizes = pbbslib::new_array_no_init<long>(3);
    sizes[0] = GA.n;
    sizes[1] = GA.m;
    sizes[2] = total_space;
    out.write((char*)sizes,sizeof(long)*3); //write n, m and space used
    out.write((char*)byte_offsets.begin(),sizeof(uintT)*(n+1)); //write offsets
    out.write((char*)degrees.begin(),sizeof(uintE)*n);
    out.write((char*)edges.begin(),total_space); //write edges
    out.close();
  }

template <class Graph>
double converter(Graph& GA, commandLine P) {
  auto outfile = P.getOptionValue("-o", "");
  bool symmetric = P.getOptionValue("-s");
  std::cout << "Outfile: " << outfile << std::endl;
  if (outfile == "") {
    std::cout << "Please specify an output file" << std::endl;
    exit(0);
  }
  std::ofstream out(outfile.c_str(), std::ofstream::out | std::ios::binary);
  auto encoding = P.getOptionValue("-enc", "bytepd-amortized");

  if (encoding == "bytepd-amortized") {
    write_graph_bytepd_amortized_format(GA, out, symmetric);
  } else {
    std::cout << "Unknown encoding: " << encoding << std::endl;
    exit(0);
  }
  std::cout << "Finished converting." << std::endl;
  exit(0);
  return 0;
}
}  // namespace gbbs


generate_main(gbbs::converter, false);