mariadb-columnstore-engine/utils/querytele/serverdriver.cpp

/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

#include <iostream>
#include <string>
#include <ctime>
using namespace std;

#include <boost/shared_ptr.hpp>
using namespace boost;

#include "thrift/TProcessor.h"
#include "thrift/protocol/TProtocol.h"
#include "thrift/protocol/TBinaryProtocol.h"
#include "thrift/transport/TServerTransport.h"
#include "thrift/transport/TServerSocket.h"
#include "thrift/transport/TTransport.h"
#include "thrift/transport/TBufferTransports.h"
#include "thrift/server/TSimpleServer.h"
namespace at = apache::thrift;
namespace atp = at::protocol;
namespace att = at::transport;
namespace ats = at::server;

#include "QueryTeleService.h"
using namespace querytele;

namespace
{
class QueryTeleServiceHandler : public QueryTeleServiceIf
{
 public:
  void postQuery(const QueryTele&);
  void postStep(const StepTele&);
  void postImport(const ImportTele&);

 protected:
 private:
};

const string st2str(enum StepType::type t)
{
  switch (t)
  {
    case StepType::T_HJS: return "HJS";

    case StepType::T_DSS: return "DSS";

    case StepType::T_CES: return "CES";

    case StepType::T_SQS: return "SQS";

    case StepType::T_TAS: return "TAS";

    case StepType::T_TNS: return "TNS";

    case StepType::T_BPS: return "BPS";

    case StepType::T_TCS: return "TCS";

    case StepType::T_HVS: return "HVS";

    case StepType::T_WFS: return "WFS";

    case StepType::T_SAS: return "SAS";

    case StepType::T_TUN: return "TUN";

    default: return "INV";
  }

  return "INV";
}

void QueryTeleServiceHandler::postQuery(const QueryTele& qt)
{
  cout << "postQuery: " << endl;
  cout << "  uuid: " << qt.query_uuid << endl;

  if (qt.msg_type == QTType::QT_SUMMARY)
    cout << "  mt: SUMMARY" << endl;
  else if (qt.msg_type == QTType::QT_START)
    cout << "  mt: START" << endl;
  else
    cout << "  mt: PROGRESS" << endl;

  cout << "  qry: " << qt.query << endl;
  cout << "  mmpct: " << qt.max_mem_pct << endl;
  cout << "  cache: " << qt.cache_io << endl;
  cout << "  nmsgs: " << qt.msg_rcv_cnt << endl;
  cout << "  rows: " << qt.rows << endl;
  cout << "  qt: " << qt.query_type << endl;
  int64_t tt = qt.start_time;
  cout << "  st: (" << tt << ") ";
  tt /= 1000;
  cout << ctime(&tt);
  tt = qt.end_time;
  cout << "  et: (" << tt << ") ";
  tt /= 1000;
  cout << ctime(&tt);
  cout << "  sn: " << qt.system_name << endl;
  cout << "  mn: " << qt.module_name << endl;
  cout << "  lq: " << qt.local_query << endl;
  cout << "  dn: " << qt.schema_name << endl;
  cout << endl;
}

void QueryTeleServiceHandler::postStep(const StepTele& qt)
{
  cout << "postStep: " << endl;
  cout << "  quuid: " << qt.query_uuid << endl;
  cout << "  uuid: " << qt.step_uuid << endl;

  if (qt.msg_type == STType::ST_SUMMARY)
    cout << "  mt: SUMMARY" << endl;
  else if (qt.msg_type == STType::ST_START)
    cout << "  mt: START" << endl;
  else
    cout << "  mt: PROGRESS" << endl;

  cout << "  st: " << st2str(qt.step_type) << endl;
  cout << "  cache: " << qt.cache_io << endl;
  cout << "  nmsgs: " << qt.msg_rcv_cnt << endl;
  cout << "  rows: " << qt.rows << endl;

  if (qt.total_units_of_work > 0)
    cout << "  pct: " << qt.units_of_work_completed * 100 / qt.total_units_of_work << endl;
  else
    cout << "  pct: n/a" << endl;

  int64_t tt = qt.start_time;
  cout << "  st: (" << tt << ") ";
  tt /= 1000;
  cout << ctime(&tt);
  tt = qt.end_time;
  cout << "  et: (" << tt << ") ";
  tt /= 1000;
  cout << ctime(&tt);
  cout << endl;
}

void QueryTeleServiceHandler::postImport(const ImportTele& qt)
{
  cout << "importStep: " << endl;
  cout << "  juuid: " << qt.job_uuid << endl;
  cout << "  iuuid: " << qt.import_uuid << endl;

  if (qt.msg_type == ITType::IT_SUMMARY)
    cout << "  mt: SUMMARY" << endl;
  else if (qt.msg_type == ITType::IT_START)
    cout << "  mt: START" << endl;
  else if (qt.msg_type == ITType::IT_TERM)
    cout << "  mt: TERM" << endl;
  else
    cout << "  mt: PROGRESS" << endl;

  if (qt.table_list.empty())
    cout << "  tn: "
         << "(empty)" << endl;
  else
    cout << "  tn: " << qt.table_list[0] << endl;

  if (qt.rows_so_far.empty())
    cout << "  rows: "
         << "(empty)" << endl;
  else
    cout << "  rows: " << qt.rows_so_far[0] << endl;

  int64_t tt = qt.start_time;
  cout << "  st: (" << tt << ") ";
  tt /= 1000;
  cout << ctime(&tt);
  tt = qt.end_time;
  cout << "  et: (" << tt << ") ";
  tt /= 1000;
  cout << ctime(&tt);
  cout << "  sn: " << qt.system_name << endl;
  cout << "  mn: " << qt.module_name << endl;
  cout << "  dn: " << qt.schema_name << endl;
  cout << endl;
}

}  // namespace

int main(int argc, char** argv)
{
  shared_ptr<atp::TProtocolFactory> protocolFactory(new atp::TBinaryProtocolFactory());
  shared_ptr<QueryTeleServiceHandler> handler(new QueryTeleServiceHandler());
  shared_ptr<at::TProcessor> processor(new QueryTeleServiceProcessor(handler));
  shared_ptr<att::TServerTransport> serverTransport(new att::TServerSocket(9990));
  shared_ptr<att::TTransportFactory> transportFactory(new att::TBufferedTransportFactory());

  ats::TSimpleServer server(processor, serverTransport, transportFactory, protocolFactory);

  cout << "Starting the server..." << endl;
  server.serve();
  cout << "done." << endl;
  return 0;
}