/*
 * $Header: /home/projects/pgsql/cvsroot/pgsql/contrib/pgbench/pgbench.c,v 1.8 2001/02/10 02:31:25 tgl Exp $
 *
 * pgbench: a simple TPC-B like benchmark program for PostgreSQL
 * written by Tatsuo Ishii
 *
 * Copyright (c) 2000  Tatsuo Ishii
 *
 * Permission to use, copy, modify, and distribute this software and
 * its documentation for any purpose and without fee is hereby
 * granted, provided that the above copyright notice appear in all
 * copies and that both that copyright notice and this permission
 * notice appear in supporting documentation, and that the name of the
 * author not be used in advertising or publicity pertaining to
 * distribution of the software without specific, written prior
 * permission. The author makes no representations about the
 * suitability of this software for any purpose.  It is provided "as
 * is" without express or implied warranty.
 VERSION 3.2 */

#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#ifdef WIN32
#include "win32.h"
#else
#include <sys/time.h>
#include <unistd.h>

#ifdef HAVE_GETOPT_H
#include <getopt.h>
#endif

#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif

/* for getrlimit */
#include <sys/resource.h>

#endif   /* WIN32 */

/********************************************************************
 * some configurable parameters */

exec sql include sqlca;

#ifdef ORACLE
#define EXIT_THREAD(a)   return(NULL)
#else
#define EXIT_THREAD(a)   pthread_exit(NULL) //return(NULL)
#endif

#define MAXCLIENTS 1024      /* max number of clients allowed */
#define a_result char
int nclients = 1;/* default number of simulated clients */
int ncli_connected ;/* Connected clients */
int nb_sql_failed ;/* nb SQL failed */
int num_cli = 0;
int nxacts = 10;/* default number of transactions per Clients */

/*
 * scaling factor. for example, tps = 10 will make 1000000 tuples of
 * accounts table.
 */
typedef struct
{
int finish;
char name[32];
}      st_connection;

int      tps = 1;
int      gi_debug = 0;    /* debug flag */
char     *gdbName;
st_connection      gtab_cnx[MAXCLIENTS];
#ifdef NO_USED
int      gi_break = 0;
#endif

#ifdef ORACLE
  sql_context tab_ctx[MAXCLIENTS];
#endif


/*
 * end of configurable parameters
 *********************************************************************/

#define nbranches  1
#define ntellers  10
#define naccounts  100000

typedef struct
{
  int      etat;      /* etat No. */
  int      cnt;      /* xacts count */
  int      aid;      /* account id for this transaction */
  int      bid;      /* branch id for this transaction */
  int      tid;      /* teller id for this transaction */
  int      delta;
  int      abalance;
}      Client_State;
char command[128];

/*===========================*/
#ifndef ORACLE
static void
doBegin(pnum_thread)
int pnum_thread;
/*===========================*/
{
exec sql begin work;
if ( gi_debug > 9 )
  {
    fprintf (stderr, "... Begin work");
    if ( pnum_thread >= 0 )
      fprintf (stderr, " for client %d\n", pnum_thread);
    else
      fprintf (stderr, "\n");
  }
}
#endif

/*===========================*/
static void
  doCommit( pnum_thread )
  int pnum_thread;
/*===========================*/
{
  exec sql commit ;

if ( gi_debug> 9  )
  {
    fprintf (stderr, "... Commited");
    if ( pnum_thread >= 0 )
      fprintf (stderr, " for client %d\n", pnum_thread);
    else
      fprintf (stderr, "\n");
  }
}

/*===========================*/
static int
doDisconnect( cnx_name )
char *cnx_name;
/*===========================*/
{
exec sql begin declare section;
  char pg_cnx_name[32];
exec sql end declare section;

  strcpy(pg_cnx_name,cnx_name);
#ifndef ORACLE
  if (gi_debug> 9 )
    fprintf(stderr, "-> exec sql disconnect %s\n",cnx_name);
  exec sql disconnect :pg_cnx_name;
#endif
}

/*===========================*/
static int
#ifdef ORACLE
  doConnect( cnx_name , ctx )
    char *cnx_name;
    sql_context ctx;
#else
  doConnect( cnx_name )
    char *cnx_name;
#endif
/*===========================*/
{
exec sql begin declare section;
  char connection[32];
  char pg_cnx_name[32];
exec sql end declare section;

  strcpy(connection,gdbName);
  strcpy(pg_cnx_name,cnx_name);
  if (gi_debug> 9 )
    fprintf(stderr, "-> exec sql connect to %s as %s\n",gdbName,cnx_name);

#ifdef ORACLE
/*-->    exec sql context use :ctx; <--*/
   strcpy (connection,"scott/tiger");
   EXEC SQL CONNECT :connection;
#else
  exec sql connect to :connection as :pg_cnx_name;
#endif
  if ( sqlca.sqlcode < 0 )
  {
    if (gi_debug> 9 )
      fprintf(stderr, "*** pgbch:Connection to '%s' failed.\n", gdbName); 
    return ( sqlca.sqlcode );
  }
  return(0); /* connection successful */
}

#ifdef NO_USED
/*===========================*/
static void printerr(sqlcb)
struct sqlca sqlcb;
/*===========================*/
{
  if ( sqlcb.sqlcode < 0 )
  {
    nb_sql_failed++;
#ifdef NO_USED
    gi_break=2;
#endif
    fprintf(stderr, "*** Exec SQL failed. ***\n");
  }
}
#endif


/*===========================*/
#ifndef ORACLE
static void printwarn(void)
/*===========================*/
{
   if (sqlca.sqlwarn[0]) printf("sqlca.sqlwarn: %c",sqlca.sqlwarn[0]);
   else return;

   if (sqlca.sqlwarn[1]) putchar('1');
   if (sqlca.sqlwarn[2]) putchar('2');

   putchar('\n');
}
#endif

/*===========================*/
static void
usage()
/*===========================*/
{
  fprintf(stderr, "usage: pgbench [-h hostname][-p port][-c nclients][-t ntransactions][-s scaling_factor][-n][-v][-S][-d][dbname]\n");
  fprintf(stderr, "(initialize mode): pgbench -i [-h hostname][-p port][-s scaling_factor][-d][dbname]\n");
}

/*===========================*/
/* random number generator */
static int
getrand(int min, int max)
/*===========================*/
{
  return (min + (int) (max * 1.0 * rand() / (RAND_MAX + 1.0)));
}


/*===========================*/
/* process a transaction */
static void *
doOne( void * arg )
/*===========================*/
{
  Client_State  st;
  char    sql[256];
  int iret=0;
  int i_am_done=0;

  int numero_thread=(int) arg;
exec sql begin declare section;
  char text[16];
  char text1[16];
  char text2[16];
  char text3[16];
exec sql end declare section;

  if (gi_debug> 9 )
    fprintf(stderr, "### client %d connecting to %s\n", numero_thread, gdbName);
    fflush(stderr);
  sprintf(gtab_cnx[numero_thread].name,"CNX_%d", numero_thread);

#ifdef ORACLE
/*-->   exec sql context allocate :tab_ctx[numero_thread]; <--*/
  if ( gi_debug > 9 )
     fprintf (stderr,"allocate ctx %d\n", numero_thread);
  iret=doConnect (gtab_cnx[numero_thread].name, tab_ctx[numero_thread]);
#else
  iret=doConnect (gtab_cnx[numero_thread].name);
#endif
  if ( iret )
  {
    ncli_connected--;
    fprintf(stderr, "*** FATAL CLIENT %d CAN NOT CONNECT TO DB=%s\n",
                      numero_thread, gdbName);
    fflush(stderr);
    gtab_cnx[numero_thread].finish=0;
#ifdef NO_USED
    gi_break=1;
#endif
    EXIT_THREAD(arg);
  }

#ifdef NO_USED
sleep(1);
#endif

  if ( gi_debug > 0 )
    fprintf(stderr, "    CLIENT %d CONNECTED TO DB=%s\n",
                      numero_thread, gdbName);
#ifdef ORACLE
/*-->   exec sql context use :tab_ctx[numero_thread]; <--*/
#endif
  for(;;)
  {
  switch (st.etat)
  {
    case 0:      /* about to start */
      strcpy(sql, "begin");
      st.aid = getrand(1, naccounts * tps);
      st.bid = getrand(1, nbranches * tps);
      st.tid = getrand(1, ntellers * tps);
      st.delta = getrand(1, 1000);
#ifndef ORACLE
      doBegin(numero_thread);
#endif
      break;
    case 1:
      sprintf(sql, "update accounts set abalance = abalance + %d where aid = %d\n", st.delta, st.aid);
      sprintf(text,"%d",st.delta);
      sprintf(text1,"%d",st.aid);
      exec sql update accounts set abalance = abalance + :text where aid = :text1; 
      break;

    case 2:
      sprintf(sql, "select abalance from accounts where aid = %d", st.aid);
      sprintf(text,"%d", st.aid);
      exec sql select abalance from accounts where aid = :text;
      break;
    case 3:
      sprintf(sql, "update tellers set tbalance = tbalance + %d where tid = %d\n",
          st.delta, st.tid);
      sprintf (text,"%d",st.delta);
      sprintf (text1,"%d",st.tid);
      exec sql update tellers set tbalance = tbalance + :text where tid = :text1;
      break;
    case 4:
      sprintf(sql, "update branches set bbalance = bbalance + %d where bid = %d", st.delta, st.bid);
      sprintf (text,"%d",st.delta);
      sprintf (text1,"%d",st.bid);
      exec sql update branches set bbalance = bbalance + :text where bid = :text1;
      break;
    case 5:
      sprintf(sql, "insert into history(tid,bid,aid,delta,mtime) values(%d,%d,%d,%d,'now')",
          st.tid, st.bid, st.aid, st.delta);
      sprintf (text,"%d",st.tid);
      sprintf (text1,"%d",st.bid);
      sprintf (text2,"%d",st.aid);
      sprintf (text3,"%d",st.delta);
      exec sql insert into history(tid,bid,aid,delta,mtime) 
               values(:text, :text1, :text2, :text3,'now');
      break;

    case 6:
      strcpy(sql, "end");
      doCommit(numero_thread);

      if (++st.cnt >= nxacts) 
      {       
        /* I've done */ 
        gtab_cnx[numero_thread].finish=0;

        if (gi_debug > 0)
        {
          fprintf(stderr, "           CLIENT %d ENDED\n", numero_thread);
          fflush(stderr);
        }       
        EXIT_THREAD(arg);
      }       
      break;
  }
#ifdef DO_IT
+ifdef NO_USED
sleep(1);
#endif

  if (gi_debug > 9)
    fprintf(stderr, "client %d sending %s\n", numero_thread, sql);
  /* increment etat counter */
  st.etat++;
  if (st.etat > 6)
    st.etat = 0;
  }

} /* End of doOne */

/*===========================*/
static void
#ifdef ORACLE
  init(ctx)
  sql_context ctx;
#else
  init()
#endif
/*===========================*/
{
exec sql begin declare section;
  int      i;
  char text[16];
  char text1[16];
exec sql end declare section;

char cnx_name[32];
int iret = -1;

  strcpy(cnx_name, "cnx_init");

#ifdef ORACLE
/*-->   exec sql context use :ctx; <--*/
  iret = doConnect (cnx_name, ctx);
#else
  iret = doConnect (cnx_name);
#endif
  if ( iret )
  {
    fprintf(stderr, "*** FATAL init can not connect to DB=%s\n", gdbName);
    fflush(stderr);
    return;
  }
  fprintf(stderr, "-> Begin creating ...\n");
  exec sql drop table branches;
  exec sql create table branches (
    "bid" int primary key,
    "bbalance" int,
    "filler" char(88));
  fprintf(stderr, "tables branches created ...\n");

  exec sql drop table tellers;
  exec sql create table tellers(
    tid int primary key,
    bid int,
    tbalance int,
    filler char(84));
  fprintf(stderr, "tables tellers created...\n");

  exec sql drop table accounts;
  exec sql create table accounts(
    aid int primary key,
    bid int,
    abalance int,
    filler char(84));
  fprintf(stderr, "tables accounts created ...\n");

  exec sql drop table history;
  exec sql create table history(
    tid int,
    bid int,
    aid int,
    delta int,
    mtime timestamp,
    filler char(22));
  fprintf(stderr, "tables history created  ...\n");

  fprintf(stderr, "-> Create Tables done  ...\n");

  for (i = 0; i < nbranches * tps; i++)
  {
    sprintf(command, "insert into branches (bid,bbalance) values(%d,0)", i+1);
    sprintf( text, "%d", i+1);
    exec sql insert into branches(bid,bbalance) values(:text,0);
  }

  for (i = 0; i < ntellers * tps; i++)
  {
    sprintf(command, "insert into tellers(tid,bid,tbalance) values (%d,%d,0)" ,i + 1, i / ntellers + 1);
    sprintf(text,"%d",i+1);
    sprintf(text1,"%d",i-1/ntellers+1);
    i+1;
    exec sql insert into tellers(tid,bid,tbalance) values(:text,:text1,0);
  }

  for (i = 0; i < naccounts * tps; i++)
  {
     int                     j = i + 1;
     sprintf (text,"%d", j);
     sprintf (text1,"%d", j/ naccounts);
     exec sql insert into accounts(aid,bid,abalance) values(:text,:text1,0);
     if (j % 10000 == 0)
     {
      /*
       * every 10000 tuples, we commit the copy command. this should
       * avoid generating too much WAL logs
       */
       fprintf(stderr, "%d tuples done.\n", j);
     }
  }
  doCommit(-1);
  doDisconnect ( cnx_name );

#ifdef ORACLE
/*-->   exec sql context free :tab_ctx[0]; <--*/
#endif

  fprintf(stderr, "done init !");
} /* Fin init */

/*===========================*/
/* print out results */
static void
printResults(
       int ttype, Client_State * process,
       struct timeval * tv1, struct timeval * tv2,
       struct timeval * tv3)
/*===========================*/
{
  double    t1,
        t2;
  int      i;
  int      normal_xacts = 0;

  normal_xacts = nclients*nxacts ;
  t1 = (tv3->tv_sec - tv1->tv_sec) * 1000000.0 + (tv3->tv_usec - tv1->tv_usec);
  t1 = nxacts*ncli_connected * 1000000.0 / t1;

  t2 = (tv3->tv_sec - tv2->tv_sec) * 1000000.0 + (tv3->tv_usec - tv2->tv_usec);
  t2 = nxacts*ncli_connected * 1000000.0 / t2;

  printf("****************************************************************\n");
  printf("transaction type: %s\n", ttype == 0 ? "TPC-B (sort of)" : "SELECT only");
  printf("scaling factor: %d\n", tps);
  printf("number of clients: %d\n", nclients);
  printf("number of clients connected: %d\n", ncli_connected);
  printf("number of transactions per client: %d\n", nxacts);
  printf("number of transactions actually processed: %d/%d\n", nxacts*ncli_connected - nb_sql_failed, normal_xacts);
  printf("tps = %f(including connections establishing)\n", t1);
  printf("tps = %f(excluding connections establishing)\n", t2);
  printf("****************************************************************\n");
} /* Fin print_resu */

/*===========================*/
int
main(int argc, char **argv)
/*===========================*/
{
  extern char *optarg;
  extern int  optind,
        opterr,
        optopt;
  int      c;
  char     *pghost = NULL;
  char     *pgport = NULL;
  int      is_init_mode = 0;    /* initialize mode? */
  int      is_no_vacuum = 0;    /* no vacuum at all before */
  int      is_drop = 0;    /* no drop before */
  int      is_full_vacuum = 0;    /* do full vacuum before testing? */
  int      ttype = 0;    /* transaction type. 0: TPC-B, 1: SELECT
                 * only */

  static Client_State process[MAXCLIENTS];  /* clients status */

  struct timeval tv1;      /* start up time */
  struct timeval tv2;      /* after establishing all connections to
                 * the backend */
  struct timeval tv3;      /* end time */

  int      i;
  int      i_fois=0;
  int      i_encours=0;
  int      iret;
  pthread_t  thread[MAXCLIENTS];
  pthread_attr_t thread_attr;


#ifdef NO_USED
exec sql whenever sqlerror do printerr(sqlca);
#endif
exec sql whenever sqlerror do printwarn();
#ifndef ORACLE
exec sql whenever not found sqlprint;

exec sql whenever sqlwarning do printwarn();
#endif
 
#ifdef ORACLE
/*-->    exec sql enable threads; <--*/
#endif

  while ((c = getopt(argc, argv, "ih:nNvp:d:c:C:t:s:S")) != EOF)
  {
    switch (c)
    {
      case 'i':
        is_init_mode++;
        break;
      case 'h':
        pghost = optarg;
        break;
      case 'N':
        is_drop++;
        break;
      case 'n':
        is_no_vacuum++;
        break;
      case 'v':
        is_full_vacuum++;
        break;
      case 'p':
        pgport = optarg;
        break;
      case 'd':
        gi_debug = atoi(optarg);
        break;
      case 'S':
        ttype = 1;
        break;

      case 'C':
        num_cli = atoi(optarg);
        break;

      case 'c':
        nclients = atoi(optarg);
        ncli_connected = nclients; 
        if (nclients <= 0 || nclients > MAXCLIENTS)
        {
          fprintf(stderr, "wrong number of clients: %d\n", nclients);
          exit(1);
        }
        break;
      case 's':
        tps = atoi(optarg);
        if (tps <= 0)
        {
          fprintf(stderr, "wrong scaling factor: %d\n", tps);
          exit(1);
        }
        break;
      case 't':
        nxacts = atoi(optarg);
        if (nxacts <= 0)
        {
          fprintf(stderr, "wrong number of transactions: %d\n", nxacts);
          exit(1);
        }
        break;
      default:
        usage();
        exit(1);
        break;
    } /* switch */
  } /* Fin du traitement */

  if (argc > optind)
    gdbName = argv[optind];
  else
  {
    gdbName = getenv("USER");
    if (gdbName == NULL)
      gdbName = "";
  }

  if (is_init_mode)
  {
#ifdef ORACLE
/*-->     exec sql context allocate :tab_ctx[0]; <--*/
    init(tab_ctx[0]);
#else
    init();
#endif
    exit(0);
  }

  if (gi_debug > 9)
  {
    printf("pghost: %s pgport: %s nclients: %d nxacts: %d dbName: %s\n",
         pghost, pgport, nclients, nxacts, gdbName);
  }
  /*
   * get the scaling factor that should be same as count(*) from
   * branches...
   */
  
  if (is_drop) 
  {
#ifdef ORACLE
  
/*-->    exec sql context allocate :tab_ctx[0]; <--*/
/*-->    exec sql context use :tab_ctx[0]; <--*/
#endif
    fprintf(stderr, "starting drop...");
    exec sql drop table branches;
    exec sql drop table tellers;
    exec sql drop table accounts;
    exec sql drop table history;
    exec sql commit;
  } /* Fin du traitement si pas d option -n */

   for (i = 0; i < nclients ; i++)
     gtab_cnx[i].finish=1;

  /* set random seed */
  gettimeofday(&tv1, 0);
  srand((uint) tv1.tv_usec);

  /*====== CREATE THE THREADS ============================*/
  /* get start up time */
   gettimeofday(&tv1, 0);

   if ( pthread_attr_init(&thread_attr) )
     perror ("pthread_attr_init: ");
        
   if (pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_JOINABLE))
     perror("pthread_attr_setdetachstate :");

  /* time after connections set up */
  gettimeofday(&tv2, 0);

   for (i = 0; i < nclients ; i++)
     pthread_create (&thread[i], &thread_attr, doOne, (void *) i);

  /*====== WAITING THE END OF THREADS ===================*/
   for (i = 0; i < nclients ; i++)
   {
     if (pthread_join (thread[i], NULL))
       perror("pthread_join failed by : ");
     else
       if ( gi_debug > 0)
         printf("- thread %d stopped\n", i);

   }
  /* time after the end of transaction */
   gettimeofday(&tv3, 0);
       
#ifdef ORACLE
  /*====== ORACLE: FREE CONTEXT ==========================*/
   for (i = 0; i < nclients ; i++)
   {
     if ( pthread_detach(thread[i]) )
     {
     if ( gi_debug > 0)
       printf(" -> detached thread %d ...", i );
     }
     else
     {
     if ( gi_debug > 0)
       printf(" -> detached thread %d OK", i );
     }
/*-->      exec sql context use :tab_ctx[i]; <--*/
     exec sql commit work release;
/*-->      exec sql context free :tab_ctx[i]; <--*/
     if ( gi_debug > 0)
       fprintf (stderr," free %d \n", i );
   }
#endif /* ORACLE */

  /*====== PRINTING OF RESULTS ==========================*/
   printResults(ttype, process, &tv1, &tv2, &tv3);
       
}