//-----------------------------------------------------------------------------
// Copyright (c) 1994,1995 Southeastern Universities Research Association,
//                         Continuous Electron Beam Accelerator Facility
//
// This software was developed under a United States Government license
// described in the NOTICE file included as part of this distribution.
//
//-----------------------------------------------------------------------------
//
// Description:
//      Implementation of rsvcDatabase
//
// Author:  Jie Chen
//
// Revision History:
//   rsvcDatabase.cc,v
// Revision 1.1  1998/01/22  17:08:09  akers
// Addition of new NameServer
//
//
//
#include <rsvcDatabaseEnv.h>
#include "rsvcDatabase.h"

//=========================================================================
//            All possible comparison functions for B+tree
//=========================================================================
static int rsvcUcharFunc (const DBT *key1, const DBT* key2)
{
  unsigned char s1 = *(unsigned char *)key1->data;
  unsigned char s2 = *(unsigned char *)key2->data;
  return s1 - s2;
}

static int rsvcUshortFunc (const DBT *key1, const DBT* key2)
{
  unsigned short s1;
  unsigned short s2;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));
  
  return s1 - s2;
}

static int rsvcShortFunc (const DBT *key1, const DBT* key2)
{
  short s1;
  short s2;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));
  
  return s1 - s2;
}

static int rsvcUintFunc (const DBT *key1, const DBT* key2)
{
  unsigned int s1;
  unsigned int s2;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));
  
  return s1 - s2;
}

static int rsvcIntFunc (const DBT *key1, const DBT* key2)
{
  int s1;
  int s2;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));
  
  return s1 - s2;
}

static int rsvcFloatFunc (const DBT *key1, const DBT* key2)
{
  float s1;
  float s2;
  float res;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));
  
  res = s1 - s2;
  if (res > 0.00001)
    return 1;
  else if (res < -1*0.00001)
    return -1;
  return 0;
}

static int rsvcDoubleFunc (const DBT *key1, const DBT* key2)
{
  double s1;
  double s2;
  double res;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));
  
  res = s1 - s2;
  if (res > 0.00001)
    return 1;
  else if (res < -1*0.00001)
    return -1;
  return 0;
}

static int rsvcStringFunc (const DBT* key1, const DBT* key2)
{
  return strcmp ((char *)key1->data, (char *)key2->data);
}

static int rsvcTSFunc (const DBT* key1, const DBT* key2)
{
  rsvc_TS_STAMP s1;
  rsvc_TS_STAMP s2;

  memcpy (&s1, key1->data, sizeof (s1));
  memcpy (&s2, key2->data, sizeof (s2));

  double t1 = s1.secPastEpoch + s1.nsec/1000000000.0;
  double t2 = s2.secPastEpoch + s2.nsec/1000000000.0;

  double res = t1 - t2;
  if (res > 0.00001)
    return 1;
  else if (res < -1*0.00001)
    return -1;
  return 0;  
}

rsvcCompFunc
rsvcDatabase::compFuncs_[] = {
  rsvcUcharFunc,
  rsvcShortFunc,
  rsvcUshortFunc,
  rsvcIntFunc,
  rsvcUintFunc,
  rsvcFloatFunc,
  rsvcDoubleFunc,
  rsvcStringFunc,
  rsvcTSFunc
};

#if defined (_RSVC_USE_THREAD) && defined (_REENTRANT)
rsvcDatabase::rsvcDatabase (void)
:rsvcVirtualDbase (), dbp_ (0), dbc_ (0), wsize_ (0), mutex_ ()
#else
rsvcDatabase::rsvcDatabase (void)
:rsvcVirtualDbase (), dbp_ (0), dbc_ (0), wsize_ (0)
#endif
{
#ifdef _TRACE_OBJECTS
  printf ("Create rsvcDatabase Class Object\n");
#endif
}

#if defined (_RSVC_USE_THREAD) && defined (_REENTRANT)
rsvcDatabase::rsvcDatabase (char* name, int flags, int keyType,
			    size_t data_size,
			    size_t cache_size,
			    size_t page_size, int mode)
:rsvcVirtualDbase (name, flags, keyType, data_size, cache_size, page_size, mode),
 dbp_ (0), dbc_ (0), wsize_ (0), mutex_ ()
#else
rsvcDatabase::rsvcDatabase (char* name, int flags, int keyType,
			    size_t data_size,
			    size_t cache_size,
			    size_t page_size, int mode)
:rsvcVirtualDbase (name, flags, keyType, data_size, cache_size, page_size, mode),
 dbp_ (0), dbc_ (0), wsize_ (0)
#endif
{
#ifdef _TRACE_OBJECTS
  printf ("Create rsvcDatabase Class Object\n");
#endif

  create (name, flags, keyType, data_size, cache_size, page_size, mode);
}

rsvcDatabase::~rsvcDatabase (void)
{
#ifdef _TRACE_OBJECTS
  printf ("Delete rsvcDatabase Class Object\n");
#endif

  if (dbc_)
    (void)(*dbc_->c_close)(dbc_); 

  if (dbp_)
    (void)(*dbp_->close)(dbp_, 0);

  dbp_ = 0;

  opened_ = 0;
}

int
rsvcDatabase::create (char* name, int flags, int keyType,
		      size_t data_size, size_t cache_size,
		      size_t page_size, int mode)
{
  // get database environment
  rsvcDatabaseEnv* env = rsvcDatabaseEnv::dbaseEnv ();
  DB_ENV* dbenv = env->dbenv ();

  memset (&b_, 0, sizeof (b_));
  if (cache_size == 0)
    b_.db_cachesize = rsvcServerConfig::dbaseCacheSize ();
  else
    b_.db_cachesize = cache_size;

  if (page_size == 0)
    b_.db_pagesize = rsvcServerConfig::dbasePageSize ();
  else
    b_.db_pagesize = page_size;

  page_size_ = b_.db_pagesize;

  b_.db_lorder = 4321;   // big endien
  b_.flags = DB_DUP;     // allow duplicated key
  // the following are related to b+ tree only
  b_.bt_minkey = page_size/data_size;
  b_.bt_compare = rsvcDatabase::compFuncs_[keyType];
  
  // create database
  name_ = new char[::strlen (name) + 1];
  ::strcpy (name_, name);
  if ((err_ = db_open (name_, DB_BTREE, DB_CREATE | O_RDWR, 
		       mode, dbenv, &b_, &dbp_)) != 0) {
    fprintf (stderr, "Fatal Error: cannot create database %s\n", name_);
    fprintf (stderr, "Err_ is %d\n", err_);
    err_ = err_;
    return RSVC_ERROR;
  }
  else { // force database to flush its meta data onto the disk
    (*dbp_->sync)(dbp_, 0);
    err_ = 0;
    opened_ = 1;
    keyType_ = keyType;
    return RSVC_SUCCESS;
  }
}


int
rsvcDatabase::open (char* name, int flags, int keyType,
		    size_t cache_size, int mode)
{
  if (err_ == 0 && opened_ && dbp_)
    return RSVC_SUCCESS;

  // get database environment
  rsvcDatabaseEnv* env = rsvcDatabaseEnv::dbaseEnv ();
  DB_ENV* dbenv = env->dbenv ();

  memset (&b_, 0, sizeof (b_));
  if (cache_size == 0)
    b_.db_cachesize = rsvcServerConfig::dbaseCacheSize ();
  else
    b_.db_cachesize = cache_size;

  b_.db_lorder = 4321;   // big endien
  b_.flags = DB_DUP;     // allow duplicated key
  b_.bt_compare = rsvcDatabase::compFuncs_[keyType];
  
  // create database
  name_ = new char[::strlen (name) + 1];
  ::strcpy (name_, name);
  if ((err_ = db_open (name_, DB_BTREE, DB_CREATE | O_RDWR, 
		       mode, dbenv, &b_, &dbp_)) != 0) {
    fprintf (stderr, "Fatal Error: cannot create database %s\n", name_);
    fprintf (stderr, "Err_ is %d\n", err_);
    err_ = err_;
    return RSVC_ERROR;
  }
  else { // force database to flush its meta data onto the disk
    err_ = 0;
    opened_ = 1;
    page_size_ = b_.db_pagesize;
    keyType_ = keyType;
    return RSVC_SUCCESS;
  }
}

int
rsvcDatabase::close (void)
{
  if (dbc_)
    (*dbc_->c_close) (dbc_);
  dbc_ = 0;

  if (dbp_)
    (*dbp_->close)(dbp_, 0);
  dbp_ = 0;

  opened_ = 0;
  wsize_ = 0;
  return RSVC_SUCCESS;
}  

int
rsvcDatabase::flush (void)
{
  if (opened_ && dbp_) {
    (*dbp_->sync)(dbp_, 0);
    return RSVC_SUCCESS;
  }
  return RSVC_ERROR;
}

int
rsvcDatabase::opened (void) const
{
  return (opened_ && dbp_);
}

int
rsvcDatabase::fd (void)
{
  if (opened_ && dbp_) {
    int fdp;
    (*dbp_->fd)(dbp_, &fdp);
    return fdp;
  }
  return 0;
}

int
rsvcDatabase::cursorInit (void)
{
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_) 
    return RSVC_ERROR;

  if (dbc_) {
    (*dbc_->c_close)(dbc_);
    dbc_ = 0;
  }
  
  if (err_ = dbp_->cursor (dbp_, 0, &dbc_) != 0) {
    fprintf (stderr, "Cannot initialize the cursor operation\n");
    return RSVC_ERROR;
  }
  return RSVC_SUCCESS;
}

int
rsvcDatabase::cursorSet (rsvcData& data, rsvcData& key)
{ 
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_ || !dbc_) 
    return RSVC_ERROR;

  rsvcDBT tkey;
  DBT     res;   memset (&res, 0, sizeof (res));

  if (rsvcDatabase::convertKey (&tkey, key, keyType_, keybuf_) != RSVC_SUCCESS)
    return RSVC_ERROR;


  err_ = (*dbc_->c_get)(dbc_, tkey.dbt (), &res, DB_SET_RANGE);

  // success
  if (err_ == 0) {
    rsvcDatabase::copyKeyData (key, tkey.dbt(), keyType_);
    return rsvcDatabase::convertData (data, &res);
  }
  else if (err_ != DB_NOTFOUND) {
    // not found
    return RSVC_NOTFOUND;
  }
  else 
    return RSVC_ERROR;
}  


int
rsvcDatabase::cursorNext (rsvcData& data, rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_ || !dbc_) 
    return RSVC_ERROR;

  DBT tkey; memset (&tkey, 0, sizeof (tkey));
  DBT res; memset (&res, 0, sizeof (res));
  
  err_ = (*dbc_->c_get)(dbc_, &tkey, &res, DB_NEXT);
  
  if (err_ == 0) {
    // success
    rsvcDatabase::copyKeyData (key, &tkey, keyType_);
    return rsvcDatabase::convertData (data, &res);
  }
  else if (err_ != DB_NOTFOUND) {
    // not found
    return RSVC_NOTFOUND;
  }
  else 
    return RSVC_ERROR;
}  

int
rsvcDatabase::cursorPrev (rsvcData& data, rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_ || !dbc_) 
    return RSVC_ERROR;

  DBT tkey; memset (&tkey, 0, sizeof (tkey));
  DBT res;  memset (&res, 0, sizeof (res));

  err_ = (*dbc_->c_get)(dbc_, &tkey, &res, DB_PREV);
  
  if (err_ == 0) {
    // success
    rsvcDatabase::copyKeyData (key, &tkey, keyType_);
    return rsvcDatabase::convertData (data, &res);
  }
  else if (err_ != DB_NOTFOUND) {
    // not found
    return RSVC_NOTFOUND;
  }
  else 
    return RSVC_ERROR;
}  

int
rsvcDatabase::cursorUpdate (rsvcData& value)
{
  rsvcDatabaseLocker locker (this);

  if (err_ != 0 || !opened_ || !dbp_ || !dbc_) 
    return RSVC_ERROR;

  rsvcDBT out;

  if (rsvcDatabase::convertData (&out, value) != RSVC_SUCCESS)
    return RSVC_ERROR;

  DBT res, key;
  memset (&res, 0, sizeof (res));
  memset (&key, 0, sizeof (key));

  // get current data
  err_ = (*dbc_->c_get)(dbc_, &key, &res, DB_CURRENT);
  if (err_ != 0)
    return RSVC_ERROR;

  if (out.size () > res.size) {
    fprintf (stderr, "Overflow data buffer with cursor update\n");
    return RSVC_OVERFLOW;
  }

  // clean out old data
  memset (res.data, 0, res.size);
  // copy new data in
  memcpy (res.data, out.data (), out.size ());
  // use database routine to set data
  err_ = (*dbc_->c_put)(dbc_, 0, &res, DB_CURRENT);

  if (err_ == 0) {
    // success
    return RSVC_SUCCESS;
  }
  else 
    return RSVC_ERROR;
}  

int
rsvcDatabase::cursorFirst (rsvcData& data, rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_ || !dbc_) 
    return RSVC_ERROR;

  DBT     tkey;  memset (&tkey, 0, sizeof (tkey));    
  DBT     res;   memset (&res, 0, sizeof (res));

  err_ = (*dbc_->c_get)(dbc_, &tkey, &res, DB_FIRST);

  // success
  if (err_ == 0) {
    rsvcDatabase::copyKeyData (key, &tkey, keyType_);
    return rsvcDatabase::convertData (data, &res);
  }
  else if (err_ != DB_NOTFOUND) {
    // not found
    return RSVC_NOTFOUND;
  }
  else 
    return RSVC_ERROR;
}  

int
rsvcDatabase::cursorLast (rsvcData& data, rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_ || !dbc_) 
    return RSVC_ERROR;

  DBT     tkey;  memset (&tkey, 0, sizeof (tkey));    
  DBT     res;   memset (&res, 0, sizeof (res));

  err_ = (*dbc_->c_get)(dbc_, &tkey, &res, DB_LAST);

  // success
  if (err_ == 0) {
    rsvcDatabase::copyKeyData (key, &tkey, keyType_);
    return rsvcDatabase::convertData (data, &res);
  }
  else if (err_ != DB_NOTFOUND) {
    // not found
    return RSVC_NOTFOUND;
  }
  else 
    return RSVC_ERROR;
}  

int
rsvcDatabase::cursorFinish (void)
{
  rsvcDatabaseLocker locker (this);

  if (err_ !=0 || !opened_ || !dbp_) 
    return RSVC_ERROR;

  if (!dbc_)
    return RSVC_ERROR;
  
  (*dbc_->c_close)(dbc_);
  dbc_ = 0;
  return RSVC_SUCCESS;
}

int
rsvcDatabase::put (rsvcData& value, rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  int status;

  if (err_ != 0 || !opened_ || dbp_ == 0)
    return RSVC_INVALIDOP;

  if (wsize_ >= page_size_) {
    (*dbp_->sync)(dbp_, 0);
    wsize_ = 0;
  }

  rsvcDBT out, tkey;

  if (rsvcDatabase::convertData (&out, &tkey, value, key, keyType_)
      != RSVC_SUCCESS)
    return RSVC_ERROR;

  wsize_ += out.size ();

  if ((status = (*dbp_->put)(dbp_, 0, tkey.dbt (), out.dbt (), 0)) == 0) 
    return RSVC_SUCCESS;
  else
    return RSVC_ERROR;
}

int
rsvcDatabase::get (rsvcData& data, rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  if (err_ != 0 || !opened_ || dbp_ == 0)
    return RSVC_INVALIDOP;

  rsvcDBT tkey;
  DBT res; memset (&res, 0, sizeof (res));

  if (rsvcDatabase::convertKey (&tkey, key, keyType_, keybuf_)!= RSVC_SUCCESS) 
    return RSVC_ERROR;

  int status;

  if ((status = (*dbp_->get)(dbp_, 0, tkey.dbt (), &res, 0)) == 0)  // success
    return rsvcDatabase::convertData (data, &res); 
  else if (status == DB_NOTFOUND) // not found
    return RSVC_NOTFOUND;
  else
    return RSVC_ERROR;
}

int
rsvcDatabase::del (rsvcData& key)
{
  rsvcDatabaseLocker locker (this);

  if (err_ != 0 || !opened_ || dbp_ == 0)
    return RSVC_INVALIDOP;

  rsvcDBT tkey;

  if (rsvcDatabase::convertKey (&tkey, key, keyType_, keybuf_)!= RSVC_SUCCESS) 
    return RSVC_ERROR;

  int status;

  if ((status = (*dbp_->del)(dbp_, 0, tkey.dbt (), 0)) == 0)  // success
    return RSVC_SUCCESS;
  else if (status == DB_NOTFOUND) // not found
    return RSVC_NOTFOUND;
  else
    return RSVC_ERROR;
}


int
rsvcDatabase::convertData (rsvcDBT* out, rsvcDBT* key, 
			   rsvcData & data, rsvcData& tkey,
			   int keyType)
{
  void *keydata = 0;
  int  status = RSVC_SUCCESS;

  // find out whether this key data has key tag in it
  if ((status = tkey.find (rsvcServerConfig::key (), keydata))
      != RSVC_SUCCESS)
    return status;

  char* buffer;
  size_t len;
  // figure out size of the key
  len = rsvcDatabase::keySize (tkey, keydata, keyType);
  if (len == 0) 
    return RSVC_ERROR;

  // allocate memory and copy data to the memory
  buffer = new char[len];
  key->size (len);
  key->data ((void *)buffer);
  key->datacpy (keydata);

  // data is streamed buffer of incoming data
  if ((status = data.streamOut (&buffer, &len)) == RSVC_SUCCESS) {
    out->data ((void *)buffer);
    out->size (len);
  }
  return status;
}


int
rsvcDatabase::convertData (rsvcDBT* out, rsvcData & data)
{
  int  status = RSVC_SUCCESS;

  char* buffer;
  size_t len;
  if ((status = data.streamOut (&buffer, &len)) == RSVC_SUCCESS) {
    out->data ((void *)buffer);
    out->size (len);
  }
  return status;
}

int
rsvcDatabase::convertData (rsvcData& data, const DBT* res)
{
  char* buffer = (char *)res->data;
  int   size   = res->size;

  // note: data out of database in not aligned
  char *nbuf = new char[size];
  memcpy (nbuf, buffer, size);
  if (data.streamIn (nbuf, size) == RSVC_SUCCESS) {
    delete []nbuf;
    return RSVC_SUCCESS;
  }
  else {
    delete []nbuf;
    return RSVC_ERROR;
  }
}
  

int
rsvcDatabase::convertKey (rsvcDBT* key, rsvcData& tkey, int keyType,
			  char* keybuf)
{
  void *keydata = 0;
  int  status = RSVC_SUCCESS;

  if ((status = tkey.find (rsvcServerConfig::key (), keydata)) != RSVC_SUCCESS)
    return status;

  size_t len;
  len = rsvcDatabase::keySize (tkey, keydata, keyType);
  if (len == 0 || len > RSVC_MAX_KEY_LEN) 
    return RSVC_ERROR;

  key->data ((void *)keybuf, 0);
  key->size (len);
  key->datacpy (keydata);

  return RSVC_SUCCESS;
}
  

size_t
rsvcDatabase::keySize (rsvcData& key, void* data, int keyType)
{
  int status = RSVC_SUCCESS;
  size_t ksize;
  rsvcDataTypes keytype;
  
  if ((keytype = key.getType (rsvcServerConfig::key ())) != RSVC_INVALID &&
      keytype == keyType) {
    switch (keytype) {
    case RSVC_BYTE:
      return sizeof ((char)1);
      break;
    case RSVC_INT16:
    case RSVC_UINT16:
      return sizeof ((short)1);
      break;
    case RSVC_INT32:
    case RSVC_UINT32:
      return sizeof ((int)1);
      break;
    case RSVC_FLOAT:
      return sizeof ((float)1.0);
      break;
    case RSVC_DOUBLE:
      return sizeof ((double)1.0);
      break;
    case RSVC_STRING:
      {
 	char* str = (char *)data;
 	return (strlen (str) + 1);
      }
    break;
    case RSVC_TIMESTAMP:
      return sizeof (rsvc_TS_STAMP);
      break;
    default:
      return 0;
    }
  }
  return 0;
}

int
rsvcDatabase::copyKeyData (rsvcData& key, DBT* data, int keyType)
{
  register size_t len;
  register u_char *p1, *p2;

  /*
   * XXX
   * If a size_t doesn't fit in an int, this routine can lose.
   * What we need is a integral type which is guaranteed to be
   * larger than a size_t, and there is no such thing.
   */
  switch (keyType) {
  case RSVC_BYTE:
    {
      if (len = data->size != sizeof ((char)1))
	return RSVC_ERROR;
      char c;
      memcpy (&c, data->data, len);
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_INT16:
    {
      if (len = data->size != sizeof ((short)1))
	return RSVC_ERROR;
      short c;
      memcpy (&c, data->data, len);      
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_UINT16:
    {
      if (len = data->size != sizeof ((unsigned short)1))
	return RSVC_ERROR;
      unsigned short c;
      memcpy (&c, data->data, len);
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_INT32:
    {
      if (len = data->size != sizeof ((int)1))
	return RSVC_ERROR;
      int c;
      memcpy (&c, data->data, len);
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_UINT32:
    {
      if (len = data->size != sizeof ((unsigned int)1))
	return RSVC_ERROR;
      unsigned int c;
      memcpy (&c, data->data, len);
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_FLOAT:
    {
      if (len = data->size != sizeof ((float)1.0))
	return RSVC_ERROR;
      float c;
      memcpy (&c, data->data, len);
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_DOUBLE:
    {
      if (len = data->size != sizeof ((double)1.0))
	return RSVC_ERROR;
      double c;
      memcpy (&c, data->data, len);
      return key.insert (rsvcServerConfig::key (), c);
    }
  break;
  case RSVC_STRING:
    return key.insert (rsvcServerConfig::key (), (char *)data->data);    
  break;
  case RSVC_TIMESTAMP:
    {
      if (len = data->size != sizeof (rsvc_TS_STAMP))
	return RSVC_ERROR;
      rsvc_TS_STAMP ts;
      memcpy (&ts, data->data, len);
      return key.insert (rsvcServerConfig::key (), ts);
    }
  break;
  default:
    break;
  }
  return RSVC_ERROR;
}

#if defined (_RSVC_USE_THREAD) && defined (_REENTRANT)

rsvcDatabaseLocker::rsvcDatabaseLocker (rsvcDatabase* dbase)
:dbase_ (dbase)
{
  dbase_->mutex_.acquire ();
}

rsvcDatabaseLocker::~rsvcDatabaseLocker (void)
{
  dbase_->mutex_.release ();  
}
#else
rsvcDatabaseLocker::rsvcDatabaseLocker (rsvcDatabase* dbase)
:dbase_ (dbase)
{
  // empty
}

rsvcDatabaseLocker::~rsvcDatabaseLocker (void)
{
  // empty
}
#endif


    
    
