/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB

   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; either version 2 of the License, or
   (at your option) any later version.

   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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include <my_global.h>
#include <my_sys.h>
#include <m_string.h>
#include <m_ctype.h>
#include "mysql.h"
#include "errmsg.h"
#include <errno.h>
#include <getopt.h>


static pthread_cond_t COND_thread_count;
static pthread_mutex_t LOCK_thread_count;
static uint thread_count,number_of_threads;
static my_bool opt_verbose;

static char *opt_db="test", *opt_user="test", *opt_password=NullS;
static char *lock_priority="";
static char *opt_host=0, *opt_mysql_unix_port=0;
static uint opt_mysql_port;
static uint number_of_insert_threads=10, insert_loop_count=10000;
static uint time_for_check=0;

char *table_name="bench1";

char *table_def="\
CREATE TABLE bench1 (\n\
  URL varchar(200) DEFAULT '' NOT NULL,\n\
  ID int(11) DEFAULT '-1' NOT NULL,\n\
  DIGEST varchar(32) DEFAULT '' NOT NULL,\n\
  TITLE blob,\n\
  EXPIRY timestamp(14),\n\
  MODIFIED_DATE datetime,\n\
  FIRST_ACC_DATE date,\n\
  LAST_ACC_DATE date DEFAULT '0000-00-00' NOT NULL,\n\
  TIMES_INDEXED int(11) DEFAULT '0' NOT NULL,\n\
  ABSTRACT blob,\n\
  FILE_SIZE int(11),\n\
  DELETED char(1) DEFAULT '' NOT NULL,\n\
  TIMES_ACCESSED int(11) DEFAULT '0' NOT NULL,\n\
  LAST_RESPONSE_CODE int(11),\n\
  PRIMARY KEY (URL),\n\
  KEY INDEXEDPAGES_DIGEST (DIGEST,DELETED),\n\
  KEY INDEXEDPAGES_ID (ID,DELETED),\n\
  KEY INDEXEDPAGES_EXPIRY (EXPIRY)\n\
)";

pthread_handler_decl(test_insert,arg);
pthread_handler_decl(test_update,arg);

static void get_options(register int *argc,register char ***argv);
static int init_test_tables();
static int check_table(char *table_name, my_bool verbose);


int main(int argc ,char **argv)
{
  pthread_t tid;
  pthread_attr_t thr_attr;
  uint i;
  int *param, error;
  MY_INIT(argv[0]);
  get_options(&argc,&argv);

  number_of_threads=number_of_insert_threads+1;

  if ((error=pthread_cond_init(&COND_thread_count,NULL)))
  {
    fprintf(stderr,"Got error: %d from pthread_cond_init (errno: %d)",
	    error,errno);
    exit(1);
  }

  pthread_mutex_init(&LOCK_thread_count,NULL);
  if ((error=pthread_attr_init(&thr_attr)))
  {
    fprintf(stderr,"Got error: %d from pthread_attr_init (errno: %d)",
	    error,errno);
    exit(1);
  }

  if ((error=pthread_attr_setdetachstate(&thr_attr,PTHREAD_CREATE_DETACHED)))
  {
    fprintf(stderr,
	    "Got error: %d from pthread_attr_setdetachstate (errno: %d)",
	    error,errno);
    exit(1);
  }
#ifndef pthread_attr_setstacksize		/* void return value */
  if ((error=pthread_attr_setstacksize(&thr_attr,65536L)))
  {
    fprintf(stderr,"Got error: %d from pthread_attr_setstacksize (errno: %d)",
	    error,errno);
    exit(1);
  }
#endif
#ifdef HAVE_THR_SETCONCURRENCY
  VOID(thr_setconcurrency(2));
#endif

  if (opt_verbose)
  {
    printf("Creating test table %s.%s\n",opt_db,table_name);
    fflush(stdout);
  }
  if (init_test_tables())
    exit(1);

  param= (int*) my_malloc(sizeof(int) * number_of_threads,MYF(0));
  for (i=0 ; i < number_of_insert_threads ; i++)
  {
    pthread_mutex_lock(&LOCK_thread_count);
    param[i]=i+1;
    if ((error=pthread_create(&tid,&thr_attr,test_insert,(void*) (param+i))))
    {
      fprintf(stderr,"Got error: %d from pthread_create (errno: %d)\n",
	      error,errno);
      pthread_mutex_unlock(&LOCK_thread_count);
      exit(1);
    }
    thread_count++;
    pthread_mutex_unlock(&LOCK_thread_count);
  }

  pthread_mutex_lock(&LOCK_thread_count);
  param[i]=1;
  if ((error=pthread_create(&tid,&thr_attr,test_update,(void*) (param+i))))
  {
    fprintf(stderr,"Got error: %d from pthread_create (errno: %d)\n",
	    error,errno);
    pthread_mutex_unlock(&LOCK_thread_count);
    exit(1);
  }
  thread_count++; i++;
  pthread_mutex_unlock(&LOCK_thread_count);


  /* Now wait until all threads have finnished */
  pthread_attr_destroy(&thr_attr);
  pthread_mutex_lock(&LOCK_thread_count);
  while (thread_count)
  {
    if ((error=pthread_cond_wait(&COND_thread_count,&LOCK_thread_count)))
      fprintf(stderr,"Got error: %d from pthread_cond_wait\n",error);
  }
  pthread_mutex_unlock(&LOCK_thread_count);

  error=check_table(table_name,opt_verbose);
  for (i=0 ; i < number_of_threads ; i++)
  {
    if (param[i])
    {
      fprintf(stderr, "Thread %d returned error %d\n",i+1,param[i]);
      error=1;
    }
  }
  my_free((char*) param,MYF(0));
  if (!error)
    printf("ok\n");
  return error;
}


static struct option long_options[] =
{
  {"database",	   required_argument,	0, 'D'},
  {"debug",	   required_argument,	0, '#'},
  {"host",	   required_argument,	0, 'h'},
#ifdef __WIN__
  {"pipe",	    no_argument,	0, 'W'},
#endif
  {"port",	    required_argument,	0, 'P'},
  {"socket",	    required_argument,	0, 'S'},
  {"user",	    required_argument,	0, 'u'},
  {"verbose",	    no_argument,	0, 'v'},
  {"version",	    no_argument,	0, 'V'},

  {"insert-threads", required_argument, 0, 'i'},
  {"count",	    required_argument,	0, 'c'},
  {"low-priority",  no_argument,	0, 'l'},
  {"check-time",    no_argument,	0, 'C'},
  {0, 0, 0, 0}
};


static void print_version(void)
{
  printf("%s  Ver 1.0 for %s at %s\n",my_progname,SYSTEM_TYPE,
	 MACHINE_TYPE);
}

static void usage(void)
{
  print_version();
  puts("By Monty, for your professional use");
  puts("This software comes with NO WARRANTY: see the PUBLIC for details.\n");
  puts("Testing of inserting into a table from many threads");
}


static void get_options(register int *argc,register char ***argv)
{
  int c,option_index=0;

  while ((c=getopt_long(*argc,*argv,"D:#:h:WP:S:u:i:Vc:vlC:",
			long_options, &option_index)) != EOF)
  {
    switch(c) {
    case 'D':
      opt_db=optarg;
      break;
    case '#':
      DBUG_PUSH(optarg ? optarg : "d:t:o");
      break;
    case 'h':
      opt_host=optarg;
      break;
    case 'l':
      lock_priority="LOW_PRIORITY";
      break;
    case 'W':
#ifdef __WIN__
      opt_mysql_unix_port=MYSQL_NAMEDPIPE;
#endif
      break;
    case 'P':
      opt_mysql_port= (unsigned int) atoi(optarg);
      break;
    case 'S':
      opt_mysql_unix_port= optarg;
      break;
    case 'u':
      opt_user=optarg;
      break;
    case 'p':
      opt_password=my_strdup(optarg,MYF(0));
      while (*optarg) *optarg++= 'x';	 /* Destroy argument */
      break;
    case 'v':
      opt_verbose=1;
      break;
    case 'd':
    case 'V':
      print_version();
      exit(0);
    case '?':
      usage();
      exit(0);

    case 'i':
      number_of_insert_threads=atoi(optarg);
      break;
    case 'c':
      insert_loop_count=atoi(optarg);
      break;
    case 'C':
      time_for_check=atoi(optarg);
      break;
    }
  }
  (*argc)-=optind;
  (*argv)+=optind;
  if (*argc != 0)
  {
    fprintf(stderr,"Too many arguments (%d)\n",*argc);
    usage();
    exit(-1);
  }
  number_of_threads=number_of_insert_threads+2;
  return;
}

static MYSQL *test_connect()
{
  MYSQL *mysql;

  if (!(mysql=mysql_init((MYSQL*) 0)))
    return 0;
  if (!(mysql_real_connect(mysql,opt_host,opt_user,
			   opt_password, opt_db, opt_mysql_port,
			   opt_mysql_unix_port, 0)))
  {
    fprintf(stderr,"Couldn't connect to engine!\n%s\n\n",mysql_error(mysql));
    my_free((char*) mysql,MYF(0));
    return (0);
  }
  return mysql;
}

static my_bool safe_stmt(MYSQL *mysql, char *query)
{
  if (mysql_query(mysql,query))
  {
    fprintf(stderr, "Got error '%s' (%d) when executing %-.160s\n",
	    mysql_error(mysql),mysql_errno(mysql),query);
    fflush(stderr);
    return 1;
  }
  return 0;
}

static int init_test_tables()
{
  char buff[120];
  MYSQL *mysql;
  int error;

  if (!(mysql=test_connect()))
    return 1;
  sprintf(buff,"drop table if exists %s",table_name);
  error=(safe_stmt(mysql,buff) ||
	 safe_stmt(mysql,table_def));
  mysql_close(mysql);
  return error;
}


static int check_table(char *table_name, my_bool verbose)
{
  char buff[120];
  MYSQL *mysql;
  int error;
  DBUG_ENTER("check_table");

  if (!(mysql=test_connect()))
    return 1;
  sprintf(buff,"check table %s",table_name);
  if (!(error=(safe_stmt(mysql,buff))))
  {
    MYSQL_RES *res;
    MYSQL_ROW row;
    error=1;
    if (!(res=mysql_store_result(mysql)))
    {
      fprintf(stderr,
	      "Got error %s (%d) from mysql_store_result() in check_table",
	      mysql_error(mysql),mysql_errno(mysql));
      goto end;
    }
    while ((row=mysql_fetch_row(res)))
    {
      if (!strcmp(row[1],"status"))
	goto end;
    }
    error=0;
  }

end:
  if (error)
  {
    fprintf(stderr,"ERROR: Table %s is corrupted\n",table_name);
    fflush(stderr);
  }
  else if (verbose)
  {
    printf("Table %s is ok\n",table_name);
    fflush(stdout);
  }
  mysql_close(mysql);
  DBUG_RETURN(error);
}


static char *generate_url(char *to,struct rand_struct *random)
{
  uint name_length=(uint) (rnd(random)*10+50);
  char *pos=strmov(to,"http://www.");
  while (name_length--)
    *pos++= (char) (rnd(random)*27+65);
  *pos=0;
  return to;
}

static char NEAR dig_vec[] =
  "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ";

static char *generate_digest(char *to,struct rand_struct *random)
{
  uint name_length=32;
  char *pos=to;

  while (name_length--)
    *pos++= dig_vec[(int) (rnd(random)*16)];
  *pos=0;
  return to;
}


pthread_handler_decl(test_insert,arg)
{
  int *param=(int*) arg;
  int result=1,thread_nr= *param;
  uint i;
  char buff[255],url[160];
  struct rand_struct random;
  MYSQL *mysql;

  my_thread_init();
  randominit(&random,(ulong) time((time_t*) 0), *param * 1024);

  if (opt_verbose)
  {
    printf("Insert thread %d (%s) started\n",thread_nr,my_thread_name());
    fflush(stdout);
  }

  if (!(mysql=test_connect()))
    goto end;

  for (i=0 ; i < insert_loop_count ; i++)
  {
    sprintf(buff,
	    "INSERT %s INTO %s (URL, EXPIRY, LAST_ACC_DATE, TIMES_INDEXED, DELETED, TIMES_ACCESSED) VALUES ('%s','%s','%s', 0, 'F', 0)",
	    lock_priority,
	    table_name,
	    generate_url(url,&random),
	    "2000-09-13 17:57:37","2000-09-13 17:57:37");
    if (safe_stmt(mysql,buff))
      goto end;
  }
  result=0;

end:
  *param= result;
  if (mysql)
    mysql_close(mysql);
  pthread_mutex_lock(&LOCK_thread_count);
  thread_count--;
  VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
  pthread_mutex_unlock(&LOCK_thread_count);
  if (opt_verbose)
  {
    printf("insert thread %d (%s) ended\n",thread_nr,my_thread_name());
    fflush(stdout);
  }
  my_thread_end();
  pthread_exit(0);
  return 0;
}


pthread_handler_decl(test_update,arg)
{
  int *param=(int*) arg;
  int result=1,not_found=0,thread_nr= *param;
  ulong used_id=0;
  char buff[1024],digest[64];
  struct rand_struct random;
  MYSQL *mysql;
  MYSQL_ROW row;
  MYSQL_RES *res;

  my_thread_init();
  randominit(&random,(ulong) time((time_t*) 0), *param * 1024);

  if (opt_verbose)
  {
    printf("update thread %d (%s) started\n",thread_nr,my_thread_name());
    fflush(stdout);
  }

  if (!(mysql=test_connect()))
    goto end;

  /* Loop until there is no ID's and no threads that updates them */
  for (; ; )
  {
    if (time_for_check && (used_id % time_for_check) == time_for_check-1)
    {
      if (check_table(table_name,0))
	goto end;
    }
    if (used_id & 1)
      sprintf(buff,"SELECT HIGH_PRIORITY url from %s where id=-1",table_name);
    else
      sprintf(buff,"SELECT HIGH_PRIORITY url from %s where digest=''",table_name);
    if (safe_stmt(mysql,buff))
      goto end;
    if (!(res=mysql_store_result(mysql)))
    {
      fprintf(stderr,"Got error %s (%d) from mysql_store_result()",
	      mysql_error(mysql),mysql_errno(mysql));
      goto end;
    }
    if (!mysql_num_rows(res))
    {
      if (not_found++ > 0)
	break;					/* end of test */
      sleep(5);
      continue;
    }
    not_found=0;
    while ((row=mysql_fetch_row(res)))
    {
      sprintf(buff,"UPDATE LOW_PRIORITY %s SET ID=%ld, DIGEST='%s', TITLE = '%s', EXPIRY = '2000-09-01 17:11:02', MODIFIED_DATE = '2000-07-11 18:10:02', FIRST_ACC_DATE = '2000-08-18', LAST_ACC_DATE = '2000-08-18', TIMES_INDEXED = TIMES_INDEXED + 1, ABSTRACT = 'abstract', FILE_SIZE = 2230, DELETED = 'F', TIMES_ACCESSED = TIMES_ACCESSED + 1, LAST_RESPONSE_CODE = 200 WHERE URL = '%s'",
	      table_name,++used_id, generate_digest(digest, &random), "test-title", row[0]);
      if (safe_stmt(mysql,buff))
	goto end;
    }
    mysql_free_result(res);
  }
  result=0;

end:
  *param= result;
  if (mysql)
    mysql_close(mysql);
  pthread_mutex_lock(&LOCK_thread_count);
  thread_count--;
  VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
  pthread_mutex_unlock(&LOCK_thread_count);
  if (opt_verbose)
  {
    printf("update thread %d (%s) ended\n",thread_nr,my_thread_name());
    fflush(stdout);
  }
  my_thread_end();
  pthread_exit(0);
  return 0;
}
