/*
   Copyright (C) 2003-2006 MySQL AB
    All rights reserved. Use is subject to license terms.

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

   This program is also distributed with certain software (including
   but not limited to OpenSSL) that is licensed under separate terms,
   as designated in a particular file or component or in included license
   documentation.  The authors of MySQL hereby grant you an additional
   permission to link the program and your derivative works with the
   separately licensed software that they have included with MySQL.

   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, version 2.0, 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 St, Fifth Floor, Boston, MA 02110-1301  USA
*/

//#define TESTDEBUG 1

#include <ndb_global.h>

#include <kernel_types.h>
#include <Pool.hpp>
#include "AsyncFile.hpp"
#include "NdbOut.hpp"
#include "NdbTick.h"
#include "NdbThread.h"
#include "NdbMain.h"

// Test and benchmark functionality of AsyncFile
// -n Number of files
// -r Number of simultaneous requests
// -s Filesize, number of pages
// -l Number of iterations
// -remove, remove files after close
// -reverse, write files in reverse order, start with the last page

#define MAXFILES 255
#define DEFAULT_NUM_FILES 1
#define MAXREQUESTS 256
#define DEFAULT_NUM_REQUESTS 1
#define MAXFILESIZE 4096
#define DEFAULT_FILESIZE 2048
#define FVERSION 0x01000000
#define PAGESIZE 8192

#define TIMER_START { Uint64 starttick = NdbTick_CurrentMillisecond()
#define TIMER_PRINT(str, ops) Uint64 stoptick = NdbTick_CurrentMillisecond();\
            Uint64 totaltime = (stoptick-starttick);  \
            ndbout << ops << " " << str << \
	    " total time " << (int)totaltime << "ms" << endl;\
            char buf[255];\
            sprintf(buf,  "%d %s/sec\n",(int)((ops*1000)/totaltime), str);\
            ndbout <<buf << endl;}

static int numberOfFiles = DEFAULT_NUM_FILES;
static int numberOfRequests = DEFAULT_NUM_REQUESTS;
static int fileSize = DEFAULT_FILESIZE;
static int removeFiles = 0;
static int writeFilesReverse = 0;
static int numberOfIterations = 1;
Uint32 FileNameArray[4];

Pool<AsyncFile>* files;
AsyncFile* openFiles[MAXFILES];
Pool<Request>* theRequestPool;
MemoryChannelMultipleWriter<Request>* theReportChannel;

char WritePages[MAXFILES][PAGESIZE];
char ReadPages[MAXFILES][PAGESIZE];

int readArguments(int argc, const char** argv);
int openFile(int fileNum);
int openFileWait();
int closeFile(int fileNum);
int closeFileWait();
int writeFile( int fileNum, int pagenum);
int writeFileWait();
int writeSyncFile( int fileNum, int pagenum);
int writeSyncFileWait();
int readFile( int fileNum, int pagenum);
int readFileWait();


NDB_COMMAND(aftest, "aftest", "aftest [-n <Number of files>] [-r <Number of simultaneous requests>] [-s <Filesize, number of pages>] [-l <Number of iterations>] [-remove, remove files after close] [-reverse, write files in reverse order, start with the last page]", "Test the AsyncFile class of Ndb", 8192)
{
  int s, numReq, numOps;

  readArguments(argc, argv);
   
  files = new Pool<AsyncFile>(numberOfFiles, 2);
  theRequestPool = new Pool<Request>;
  theReportChannel = new MemoryChannelMultipleWriter<Request>;

  ndbout << "AsyncFileTest starting" << endl;
  ndbout << "  " << numberOfFiles << " files" << endl;
  ndbout << "  " << numberOfRequests << " requests" << endl;
  ndbout << "  " << fileSize << " * 8k files" << endl << endl;
  ndbout << "  " << numberOfIterations << " iterations" << endl << endl;

  NdbThread_SetConcurrencyLevel(numberOfFiles+2);

  // initialize data to write to files
  for (int i = 0; i < MAXFILES; i++) {
    for (int j = 0; j < PAGESIZE; j++){
      WritePages[i][j] = (64+i+j)%256;
    }
      //      memset(&WritePages[i][0], i+64, PAGESIZE);
  }

  // Set file directory and name
  // /T27/F27/NDBFS/S27Pnn.data
  FileNameArray[0] = 27; // T27
  FileNameArray[1] = 27; // F27
  FileNameArray[2] = 27; // S27 
  FileNameArray[3] = FVERSION; // Version
  
  for (int l = 0; l < numberOfIterations; l++)
    {

  ndbout << "Opening files" << endl;
  // Open files
  for (int f = 0; f < numberOfFiles; f++)
  {
      openFile(f);
 
  }

  // Wait for answer
  openFileWait();

  ndbout << "Files opened!" << endl<< endl;

  // Write to files
  ndbout << "Started writing" << endl;
  TIMER_START;
  s = 0;
  numReq = 0;
  numOps = 0;
  while ( s < fileSize)
  {
    for (int r = 0; r < numberOfRequests; r++)
    {
      for (int f = 0; f < numberOfFiles; f++)
      {
	writeFile(f, s);
	numReq++;
	numOps++;
      }
      
      s++;
    }
    
    while (numReq > 0)
      {	
	writeFileWait();
	numReq--;
      }

  }

  TIMER_PRINT("writes", numOps);


  ndbout << "Started reading" << endl;
  TIMER_START;

  // Read from files
  s = 0;
  numReq = 0;
  numOps = 0;
  while ( s < fileSize)
  {
    for (int r = 0; r < numberOfRequests; r++)
    {
      for (int f = 0; f < numberOfFiles; f++)
      {
	readFile(f, s);
	numReq++;
	numOps++;
      }
      
      s++;
      
    }
    
    while (numReq > 0)
      {	
	readFileWait();
	numReq--;
      }

  }
  TIMER_PRINT("reads", numOps);

  ndbout << "Started writing with sync" << endl;
  TIMER_START;

  // Write to files
  s = 0;
  numReq = 0;
  numOps = 0;
  while ( s < fileSize)
  {
    for (int r = 0; r < numberOfRequests; r++)
    {
      for (int f = 0; f < numberOfFiles; f++)
      {
	writeSyncFile(f, s);
	numReq++;
	numOps++;
      }
      
      s++;
    }
    
    while (numReq > 0)
      {	
	writeSyncFileWait();
	numReq--;
      }

  }

  TIMER_PRINT("writeSync", numOps);
  
  // Close files
  ndbout << "Closing files" << endl;
  for (int f = 0; f < numberOfFiles; f++)
  {
      closeFile(f);
 
  }

  // Wait for answer
  closeFileWait();

  ndbout << "Files closed!" << endl<< endl;
    }

  // Deallocate memory
  delete files;
  delete theReportChannel;
  delete theRequestPool;

  return 0;

}



int forward( AsyncFile * file, Request* request )
{
   file->execute(request);
   ERROR_CHECK 0;
   return 1;
}

int openFile( int fileNum)
{
  AsyncFile* file = (AsyncFile *)files->get();

  FileNameArray[3] = fileNum | FVERSION;
  file->fileName().set( NDBFS_REF, &FileNameArray[0] );
  ndbout << "openFile: " << file->fileName().c_str() << endl;
  
  if( ERROR_STATE ) {
     ERROR_RESET;
     files->put( file );
     ndbout <<  "Failed to set filename" << endl;
     return 1;
  }
  file->reportTo(theReportChannel);
          
  Request* request = theRequestPool->get();
  request->action= Request::open;
  request->error= 0;
  request->par.open.flags = 0x302; //O_RDWR | O_CREAT | O_TRUNC ; // 770
  request->set(NDBFS_REF, 0x23456789, fileNum );
  request->file = file;
  
  if (!forward(file,request)) {
     // Something went wrong
     ndbout << "Could not forward open request" << endl;
     theRequestPool->put(request);
     return 1;
  }
  return 0;
}

int closeFile( int fileNum)
{

  AsyncFile* file = openFiles[fileNum];
          
  Request* request = theRequestPool->get();
  if (removeFiles == 1)
    request->action = Request::closeRemove;
  else
    request->action= Request::close;

  request->error= 0;
  request->set(NDBFS_REF, 0x23456789, fileNum );
  request->file = file;
  
  if (!forward(file,request)) {
     // Something went wrong
     ndbout << "Could not forward close request" << endl;
     theRequestPool->put(request);
     return 1;
  }
  return 0;
}

int writeFile( int fileNum, int pagenum)
{
  AsyncFile* file = openFiles[fileNum];
#ifdef TESTDEBUG
  ndbout << "writeFile" << fileNum <<": "<<pagenum<<", " << file->fileName().c_str()<< endl;
#endif
  Request *request = theRequestPool->get();
  request->action = Request::write;
  request->error = 0;
  request->set(NDBFS_REF, pagenum, fileNum);
  request->file = openFiles[fileNum];

  // Write only one page, choose the correct page for each file using fileNum
  request->par.readWrite.pages[0].buf = &WritePages[fileNum][0];
  request->par.readWrite.pages[0].size = PAGESIZE;
  if (writeFilesReverse == 1)
  {
    // write the last page in the files first
    // This is a normal way for the Blocks in Ndb to write to a file
     request->par.readWrite.pages[0].offset = (fileSize - pagenum - 1) * PAGESIZE;
  }
  else
  {
     request->par.readWrite.pages[0].offset = pagenum * PAGESIZE;
  }
  request->par.readWrite.numberOfPages = 1;
  
  if (!forward(file,request)) {
     // Something went wrong
     ndbout << "Could not forward write request" << endl;
     theRequestPool->put(request);
     return 1;
  }
  return 0;

}

int writeSyncFile( int fileNum, int pagenum)
{
  AsyncFile* file = openFiles[fileNum];
#ifdef TESTDEBUG
  ndbout << "writeFile" << fileNum <<": "<<pagenum<<", " << file->fileName().c_str() << endl;
#endif
  Request *request = theRequestPool->get();
  request->action = Request::writeSync;
  request->error = 0;
  request->set(NDBFS_REF, pagenum, fileNum);
  request->file = openFiles[fileNum];

  // Write only one page, choose the correct page for each file using fileNum
  request->par.readWrite.pages[0].buf = &WritePages[fileNum][0];
  request->par.readWrite.pages[0].size = PAGESIZE;
  request->par.readWrite.pages[0].offset = pagenum * PAGESIZE;
  request->par.readWrite.numberOfPages = 1;
  
  if (!forward(file,request)) {
     // Something went wrong
     ndbout << "Could not forward write request" << endl;
     theRequestPool->put(request);
     return 1;
  }
  return 0;

}

int readFile( int fileNum, int pagenum)
{
  AsyncFile* file = openFiles[fileNum];
#ifdef TESTDEBUG
  ndbout << "readFile" << fileNum <<": "<<pagenum<<", " << file->fileName().c_str() << endl;
#endif
  Request *request = theRequestPool->get();
  request->action = Request::read;
  request->error = 0;
  request->set(NDBFS_REF, pagenum, fileNum);
  request->file = openFiles[fileNum];

  // Read only one page, choose the correct page for each file using fileNum
  request->par.readWrite.pages[0].buf = &ReadPages[fileNum][0];
  request->par.readWrite.pages[0].size = PAGESIZE;
  request->par.readWrite.pages[0].offset = pagenum * PAGESIZE;
  request->par.readWrite.numberOfPages = 1;
  
  if (!forward(file,request)) {
     // Something went wrong
     ndbout << "Could not forward read request" << endl;
     theRequestPool->put(request);
     return 1;
  }
  return 0;

}

int openFileWait()
{
  int openedFiles = 0;
  while (openedFiles < numberOfFiles)
    {
      Request* request = theReportChannel->readChannel();
      if (request) 
	{
	  if (request->action == Request::open)
	    {	      
	      if (request->error ==0)
		{
#ifdef TESTDEBUG
  	          ndbout << "Opened file " << request->file->fileName().c_str() << endl;
#endif
		  openFiles[request->theFilePointer] = request->file;
		}
	      else
		{
		  ndbout << "error while opening file" << endl;
		  exit(1);
		}
	      theRequestPool->put(request);
	      openedFiles++;
	    }
	  else
	    {
	      ndbout << "Unexpected request received" << endl;
	    }
	}
      else
	{
	  ndbout << "Nothing read from theReportChannel" << endl;
	}
    }
  return 0;
}

int closeFileWait()
{
  int closedFiles = 0;
  while (closedFiles < numberOfFiles)
    {
      Request* request = theReportChannel->readChannel();
      if (request) 
	{
	  if (request->action == Request::close || request->action == Request::closeRemove)
	    {	      
	      if (request->error ==0)
		{
#ifdef TESTDEBUG
  	          ndbout << "Closed file " << request->file->fileName().c_str() << endl;
#endif
		  openFiles[request->theFilePointer] = NULL;
		  files->put(request->file);
		}
	      else
		{
		  ndbout << "error while closing file" << endl;
		  exit(1);
		}
	      theRequestPool->put(request);
	      closedFiles++;
	    }
	  else
	    {
	      ndbout << "Unexpected request received" << endl;
	    }
	}
      else
	{
	  ndbout << "Nothing read from theReportChannel" << endl;
	}
    }
  return 0;
}

int writeFileWait()
{
  Request* request = theReportChannel->readChannel();
  if (request) 
    {
      if (request->action == Request::write)
	{	      
	  if (request->error == 0)
	    {
#ifdef TESTDEBUG
	      ndbout << "writeFileWait"<<request->theFilePointer<<", " << request->theUserPointer<<" "<< request->file->fileName().c_str() << endl;
#endif

	    }
	  else
	    {
	      ndbout << "error while writing file, error=" << request->error << endl;
	      exit(1);
	    }
	  theRequestPool->put(request);
	}
      else
	{
	  ndbout << "Unexpected request received" << endl;
	}
    }
  else
    {
      ndbout << "Nothing read from theReportChannel" << endl;
    }
  return 0;
}

int writeSyncFileWait()
{
  Request* request = theReportChannel->readChannel();
  if (request) 
    {
      if (request->action == Request::writeSync)
	{	      
	  if (request->error == 0)
	    {
#ifdef TESTDEBUG
	      ndbout << "writeFileWait"<<request->theFilePointer<<", " << request->theUserPointer<<" "<< request->file->fileName().c_str() << endl;
#endif

	    }
	  else
	    {
	      ndbout << "error while writing file" << endl;
	      exit(1);
	    }
	  theRequestPool->put(request);
	}
      else
	{
	  ndbout << "Unexpected request received" << endl;
	}
    }
  else
    {
      ndbout << "Nothing read from theReportChannel" << endl;
    }
  return 0;
}

int readFileWait()
{
  Request* request = theReportChannel->readChannel();
  if (request) 
    {
      if (request->action == Request::read)
	{	      
	  if (request->error == 0)
	    {
#ifdef TESTDEBUG
	      ndbout << "readFileWait"<<request->theFilePointer<<", " << request->theUserPointer<<" "<< request->file->fileName().c_str() << endl;
#endif
	      if (memcmp(&(ReadPages[request->theFilePointer][0]), &(WritePages[request->theFilePointer][0]), PAGESIZE)!=0) 
		{
		  ndbout <<"Verification error!" << endl;
		  for (int i = 0; i < PAGESIZE; i++ ){
		    ndbout <<" Compare Page " <<  i << " : " << ReadPages[request->theFilePointer][i] <<", " <<WritePages[request->theFilePointer][i] << endl;;
		    if( ReadPages[request->theFilePointer][i] !=WritePages[request->theFilePointer][i])
		
		      exit(1);
		       }
		}
		
	    }
	  else
	    {
	      ndbout << "error while reading file" << endl;
	      exit(1);
	    }
	  theRequestPool->put(request);
	}
      else
	{
	  ndbout << "Unexpected request received" << endl;
	}
    }
  else
    {
      ndbout << "Nothing read from theReportChannel" << endl;
    }
  return 0;
}

int readArguments(int argc, const char** argv)
{

  int i = 1;
  while (argc > 1)
  {
    if (strcmp(argv[i], "-n") == 0)
    {
      numberOfFiles = atoi(argv[i+1]);
      if ((numberOfFiles < 1) || (numberOfFiles > MAXFILES))
      {
	ndbout << "Wrong number of files, default = "<<DEFAULT_NUM_FILES << endl;
	  numberOfFiles = DEFAULT_NUM_FILES;
      }
    }
    else if (strcmp(argv[i], "-r") == 0)
    {
      numberOfRequests = atoi(argv[i+1]);
      if ((numberOfRequests < 1) || (numberOfRequests > MAXREQUESTS))
      {
	ndbout << "Wrong number of requests, default = "<<DEFAULT_NUM_REQUESTS << endl;
	  numberOfRequests = DEFAULT_NUM_REQUESTS;
      }
    }
    else if (strcmp(argv[i], "-s") == 0)
    {
      fileSize = atoi(argv[i+1]);
      if ((fileSize < 1) || (fileSize > MAXFILESIZE))
      {
	ndbout << "Wrong number of 8k pages, default = "<<DEFAULT_FILESIZE << endl;
	  fileSize = DEFAULT_FILESIZE;
      }
    }
    else if (strcmp(argv[i], "-l") == 0)
    {
      numberOfIterations = atoi(argv[i+1]);
      if ((numberOfIterations < 1))
      {
	ndbout << "Wrong number of iterations, default = 1" << endl;
	numberOfIterations = 1;
      }
    }
    else if (strcmp(argv[i], "-remove") == 0)
    {
      removeFiles = 1;
      argc++;
      i--;
    }
    else if (strcmp(argv[i], "-reverse") == 0)
    {
      ndbout << "Writing files reversed" << endl;
      writeFilesReverse = 1;
      argc++;
      i--;
    }

    argc -= 2;
    i = i + 2;
  }
  
  if ((fileSize % numberOfRequests)!= 0)
    {
    numberOfRequests = numberOfRequests - (fileSize % numberOfRequests);
    ndbout <<"numberOfRequest must be modulo of filesize" << endl;
    ndbout << "New numberOfRequest="<<numberOfRequests<<endl;
    }
  return 0;
}


// Needed for linking...

void ErrorReporter::handleError(ErrorCategory type, int messageID,
                                const char* problemData, const char* objRef, NdbShutdownType stype)
{

  ndbout << "ErrorReporter::handleError activated" << endl;
  ndbout << "type= " << type << endl; 
  ndbout << "messageID= " << messageID << endl;
  ndbout << "problemData= " << problemData << endl;
  ndbout << "objRef= " << objRef << endl;

  exit(1);
}

void ErrorReporter::handleAssert(const char* message, const char* file, int line)
{
  ndbout << "ErrorReporter::handleAssert activated" << endl;
  ndbout << "message= " << message << endl;
  ndbout << "file= " << file << endl;
  ndbout << "line= " << line << endl;
  exit(1);
}


GlobalData globalData;


Signal::Signal()
{

}

