Query 11

Query 11

#include <algorithm>
#include <ctime>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <iterator>
#include <map>
#include <numeric>
#include <sstream>
#include <stdint.h>
#include <thrust/device_vector.h>
#include <thrust/host_vector.h>
#include <thrust/sort.h>
#include <thrust/merge.h>
#include <thrust/unique.h>
#include <thrust/generate.h>
#include <thrust/system/cuda/experimental/pinned_allocator.h>
#include <time.h>
#include <random>
#include <limits>

using namespace std;
using namespace thrust::placeholders;

size_t getFreeMem()
{
	size_t available, total;
	cudaMemGetInfo(&available, &total);
	return available;
};


struct check_records {
  const unsigned int *time6;
  const char *bool5;
  bool *res;
  const unsigned int *time6_from;
  const unsigned int *time6_to;
  const char *bool5_val;

  check_records(const unsigned int *_time6, const char *_bool5, bool *_res,
                const unsigned int *_time6_from, const unsigned int *_time6_to,
                const char *_bool5_val)
      : time6(_time6), bool5(_bool5), res(_res), time6_from(_time6_from),
        time6_to(_time6_to), bool5_val(_bool5_val) {}
  template <typename IndexType>
  __host__ __device__ void operator()(const IndexType &i) {

    if (time6[i] >= time6_from[0] && time6[i] <= time6_to[0] &&
        bool5[i] == bool5_val[0]) {
      res[i] = 1;
    } else
      res[i] = 0;
  }
};

int main(int ac, char **av) {
  unsigned int time6_from = 19000101, time6_to = 20300101; 
  uint64_t line_count, segment_size = 200000000;
  char bool5_val = '0';
  //string usage = "Usage : query1 [-time6_from TIME6_FROM] [-time6_to TIME6_TO] [-bool5 BOOL5] [-line_count LINE_COUNT] [-segment_size SEGMENT_SIZE]";
  string usage = "Usage : query1 [-line_count LINE_COUNT] [-segment_size SEGMENT_SIZE]";

  if (ac != 5) {
    cout << usage << endl;
    exit(1);
  };

  for (unsigned int i = 1; i < ac; i++) {
    if (strcmp(av[i], "-line_count") == 0) {
      if (i + 1 < ac) {
        line_count = atoi(av[i + 1]);
        i++;
      } 
	  else {
        cout << usage << endl;
        exit(1);
	  };		
    }
    else if (strcmp(av[i], "-segment_size") == 0) {
      if (i + 1 < ac) {
        segment_size = atoi(av[i + 1]);
        i++;
      } 
	  else {
        cout << usage << endl;
        exit(1);
      };	
    };
  };

  std::clock_t start1 = 0, start2;
  
  string file_name = "col1.bin";
  std::fstream f(file_name, std::ios_base::in | ios::binary);
  thrust::device_vector<int> id1;
  thrust::host_vector<uint64_t, thrust::cuda::experimental::pinned_allocator<uint64_t> > keys(segment_size);  
  file_name = "col6.bin";
  std::fstream f1(file_name, std::ios_base::in | ios::binary);
  string date_string;
  thrust::device_vector<unsigned int> time6;
  thrust::host_vector<unsigned int, thrust::cuda::experimental::pinned_allocator<unsigned int> > date_keys(segment_size);  
  file_name = "col5.bin";
  std::fstream f2(file_name, std::ios_base::in | ios::binary);
  thrust::device_vector<char> bool5;
  thrust::host_vector<char, thrust::cuda::experimental::pinned_allocator<char> > bool_keys(segment_size);  
  unsigned int curr_cnt = 0;
  unsigned long long int total_count = 0, total_line_cnt = 0;  
  thrust::device_vector<bool> res;
  bool merged = 0;
  bool data_entered = 0;
 
  int * m_A; 
  cudaMallocManaged(&m_A, sizeof(int)*line_count); 
  thrust::device_ptr<int> h_values(m_A); 

  int * m_A_tmp; 
  cudaMallocManaged(&m_A_tmp, sizeof(int)*line_count); 
  thrust::device_ptr<int> h_values_tmp(m_A_tmp);   

  while(1) {
	  while(total_line_cnt < line_count) {
		  curr_cnt = 0;	  
		  if (f) {
			f.read((char*)keys.data(), segment_size*8);
			curr_cnt = f.gcount()/8;
			total_line_cnt = total_line_cnt + curr_cnt;
			
		  } 
		  else {
			cout << "Could not open file " << file_name << endl;
			exit(0);
		  };

		  curr_cnt = 0;
		  if (f1) {
			f1.read((char*)date_keys.data(), segment_size*4);
			curr_cnt = f1.gcount()/4;
			
		  }
		  else {
			cout << "Could not open file " << file_name << endl;
			exit(0);
		  };

		  curr_cnt = 0;
		  if (f2) {
			f2.read(bool_keys.data(), segment_size);
			curr_cnt = f2.gcount();		
			
		  }
		  else {
			cout << "Could not open file " << file_name << endl;
			exit(0);
		  };	  
		  
		  id1.resize(curr_cnt);		
		  time6.resize(curr_cnt);		
		  bool5.resize(curr_cnt);
		  thrust::copy(keys.begin(), keys.begin() + curr_cnt, id1.begin());
		  thrust::copy(date_keys.begin(), date_keys.begin() + curr_cnt, time6.begin());
		  thrust::copy(bool_keys.begin(), bool_keys.begin() + curr_cnt, bool5.begin());
		  
		  // Now we have all 3 arrays in a device memory	  
		  thrust::device_vector<bool> res(curr_cnt);
		  
		  if(!data_entered) {
			  string s;
			  cout << "Enter date_from (default:19000101)" <<  endl;  
			  getline(cin,s);	  
			  if(s.length())
				  time6_from = stoi(s);
			  else
				  time6_from = 19000101;
			  cout << "Enter date_to (default:20300101)" <<  endl;  
			  getline(cin,s);	  
			  if(s.length())
				  time6_to = stoi(s);
			  else
				  time6_to = 20300101;
			  cout << "Enter bool (default:0)" <<  endl;  
			  getline(cin,s);	  
			  if(s.length())
				  bool5_val = s[0];  
			  else
				  bool5_val = '0';  
			  data_entered = 1;
		  };	  
		  
		  start1 = 0;  	  
		  
		  // SQL WHERE condition check
		  thrust::device_vector<unsigned int> dev_time6_from(1);
		  thrust::device_vector<unsigned int> dev_time6_to(1);
		  thrust::device_vector<char> dev_bool5(1);
		  dev_time6_from[0] = time6_from;
		  dev_time6_to[0] = time6_to;
		  dev_bool5[0] = bool5_val;
		  start2 = std::clock();

		  thrust::counting_iterator<unsigned int> begin(0);
		  check_records ff(
			  (const unsigned int *)thrust::raw_pointer_cast(time6.data()),
			  (const char *)thrust::raw_pointer_cast(bool5.data()),
			  thrust::raw_pointer_cast(res.data()),
			  (const unsigned int *)thrust::raw_pointer_cast(dev_time6_from.data()),
			  (const unsigned int *)thrust::raw_pointer_cast(dev_time6_to.data()),
			  (const char *)thrust::raw_pointer_cast(dev_bool5.data()));
		  thrust::for_each(begin, begin + res.size(), ff);

		  //time6.resize(0);
		  //time6.shrink_to_fit();
		  //bool5.resize(0);
		  //bool5.shrink_to_fit();			
			
		  // copy_if the results
		  thrust::device_vector<int> id1_cpy(res.size());
		  
		  auto w_count = thrust::copy_if(id1.begin(), id1.end(), res.begin(),
										 id1_cpy.begin(), thrust::identity<bool>()) -
						 id1_cpy.begin();
						 
						 
		  // SQL DISTINCT
		  thrust::sort(id1_cpy.begin(), id1_cpy.begin() + w_count);  
		  auto distinct_cnt =
			  thrust::unique(id1_cpy.begin(), id1_cpy.begin() + w_count) -
			  id1_cpy.begin();
		  cout << "distinct count " << distinct_cnt << " " <<  endl;  
			   
		  if(total_count == 0) {
			thrust::copy(id1_cpy.begin(), id1_cpy.begin() + distinct_cnt, h_values);    		
		  }	
		  else {		
			thrust::merge(h_values, h_values + total_count, id1_cpy.begin(), id1_cpy.begin() + distinct_cnt, h_values_tmp);  		
			thrust::copy(h_values_tmp, h_values_tmp + distinct_cnt + total_count, h_values);		
			merged = 1;
		  };	
		  total_count = total_count + distinct_cnt;
		  start1 = start1 + (std::clock() - start2);
	  };	
  
	  start2 = std::clock();
	  /*res.resize(0);
	  res.shrink_to_fit();				     
	  id1.resize(0);
	  id1.shrink_to_fit();				 
	  time6.resize(0);
	  time6.shrink_to_fit();				 
	  bool5.resize(0);
	  bool5.shrink_to_fit();				 
	  */
	  
	  uint64_t final_distinct_cnt;
	  if (merged)
		final_distinct_cnt =
			thrust::unique(h_values, h_values + total_count) - h_values;
	  else		
		final_distinct_cnt = total_count;
		  
	  cout << "merge distinct count " << final_distinct_cnt << endl;        

	  start1 = start1 + (std::clock() - start2);
	  std::cout << "gpu running time " <<  start1/(double)CLOCKS_PER_SEC   << '\n';	
	  data_entered = 0;
	  total_line_cnt = 0;
	  total_count = 0;
	  f.seekg(ios_base::beg);
	  f1.seekg(ios_base::beg);
	  f2.seekg(ios_base::beg);
  };	  
  cudaFree(m_A);  			
  cudaFree(m_A_tmp);  

  return 0;
}