mariadb/unittest/sql/my_apc-t.cc

/*
   Copyright (c) 2012, Monty Program 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; 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/*
  This file does standalone APC system tests.
*/
#include <stdio.h>
#include <my_global.h>
#include <my_pthread.h>
#include <my_sys.h>

#include <tap.h>

#include "../sql/my_apc.h"

#define MY_APC_STANDALONE 1
#include "../sql/my_apc.cc"

volatile bool started= FALSE;
volatile bool service_should_exit= FALSE;
volatile bool requestors_should_exit=FALSE;

/*  Counters for APC calls */
int apcs_served= 0;
int apcs_missed=0;
int apcs_timed_out=0;
mysql_mutex_t apc_counters_mutex;

inline void increment_counter(int *var)
{
  mysql_mutex_lock(&apc_counters_mutex);
  *var= *var+1;
  mysql_mutex_unlock(&apc_counters_mutex);
}

volatile bool have_errors= false;

Apc_target apc_target;
mysql_mutex_t target_mutex;

int int_rand(int size)
{
  return round (((double)rand() / RAND_MAX) * size);
}

/* 
  APC target thread (the one that will serve the APC requests). We will have
  one target.
*/
void *test_apc_service_thread(void *ptr)
{
  my_thread_init();
  mysql_mutex_init(0, &target_mutex, MY_MUTEX_INIT_FAST);
  apc_target.init(&target_mutex);
  apc_target.enable();
  started= TRUE;
  fprintf(stderr, "# test_apc_service_thread started\n");
  while (!service_should_exit)
  {
    //apc_target.disable();
    usleep(10000);
    //apc_target.enable();
    for (int i = 0; i < 10 && !service_should_exit; i++)
    {
      apc_target.process_apc_requests();
      usleep(int_rand(30));
    }
  }
  apc_target.disable();
  apc_target.destroy();
  mysql_mutex_destroy(&target_mutex);
  my_thread_end();
  pthread_exit(0);
}


/*
  One APC request (to write 'value' into *where_to)
*/
class Apc_order : public Apc_target::Apc_call
{
public:
  int value;   // The value 
  int *where_to;  // Where to write it
  Apc_order(int a, int *b) : value(a), where_to(b) {}

  void call_in_target_thread()
  {
    usleep(int_rand(1000));
    *where_to = value;
    increment_counter(&apcs_served);
  }
};


/*
  APC requestor thread. It makes APC requests, and checks if they were actually
  executed.
*/
void *test_apc_requestor_thread(void *ptr)
{
  my_thread_init();
  fprintf(stderr, "# test_apc_requestor_thread started\n");
  while (!requestors_should_exit)
  {
    int dst_value= 0;
    int src_value= int_rand(4*1000*100);
    /* Create an APC to do "dst_value= src_value" assignment */
    Apc_order apc_order(src_value, &dst_value);
    bool timed_out;

    mysql_mutex_lock(&target_mutex);
    bool res= apc_target.make_apc_call(&apc_order, 60, &timed_out);
    if (res)
    {
      if (timed_out)
        increment_counter(&apcs_timed_out);
      else
        increment_counter(&apcs_missed);

      if (dst_value != 0)
      {
        fprintf(stderr, "APC was done even though return value says it wasnt!\n");
        have_errors= true;
      }
    }
    else
    {
      if (dst_value != src_value)
      {
        fprintf(stderr, "APC was not done even though return value says it was!\n");
        have_errors= true;
      }
    }
    //usleep(300);
  }
  fprintf(stderr, "# test_apc_requestor_thread exiting\n");
  my_thread_end();
  return NULL;
}

/* Number of APC requestor threads */
const int N_THREADS=23;


int main(int args, char **argv)
{
  pthread_t service_thr;
  pthread_t request_thr[N_THREADS];
  int i;

  my_thread_global_init();

  mysql_mutex_init(0, &apc_counters_mutex, MY_MUTEX_INIT_FAST);

  plan(1);
  diag("Testing APC delivery and execution");

  pthread_create(&service_thr, NULL, test_apc_service_thread, (void*)NULL);
  while (!started)
    usleep(1000);
  for (i = 0; i < N_THREADS; i++)
    pthread_create(&request_thr[i], NULL, test_apc_requestor_thread, (void*)NULL);
  
  for (i = 0; i < 15; i++)
  {
    usleep(500*1000);
    fprintf(stderr, "# %d APCs served %d missed\n", apcs_served, apcs_missed);
  }
  fprintf(stderr, "# Shutting down requestors\n");
  requestors_should_exit= TRUE;
  for (i = 0; i < N_THREADS; i++)
    pthread_join(request_thr[i], NULL);
  
  fprintf(stderr, "# Shutting down service\n");
  service_should_exit= TRUE;
  pthread_join(service_thr, NULL);

  mysql_mutex_destroy(&apc_counters_mutex);

  fprintf(stderr, "# Done.\n");
  my_thread_end();
  my_thread_global_end();

  ok1(!have_errors);
  return exit_status();
}