/*
   Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.

   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
*/

#include <ndb_global.h>
#include "DictCache.hpp"
#include "NdbDictionaryImpl.hpp"
#include <NdbTick.h>
#include <NdbCondition.h>
#include <NdbSleep.h>

static NdbTableImpl * f_invalid_table = 0;
static NdbTableImpl * f_altered_table = 0;

static int ndb_dict_cache_count = 0;

Ndb_local_table_info *
Ndb_local_table_info::create(NdbTableImpl *table_impl, Uint32 sz)
{
  assert(! is_ndb_blob_table(table_impl));
  Uint32 tot_size= sizeof(Ndb_local_table_info) - sizeof(Uint64)
    + ((sz+7) & ~7); // round to Uint64
  void *data= malloc(tot_size);
  if (data == 0)
    return 0;
  memset(data, 0, tot_size);
  new (data) Ndb_local_table_info(table_impl);
  return (Ndb_local_table_info *) data;
}

void Ndb_local_table_info::destroy(Ndb_local_table_info *info)
{
  free((void *)info);
}

Ndb_local_table_info::Ndb_local_table_info(NdbTableImpl *table_impl)
{
  assert(! is_ndb_blob_table(table_impl));
  m_table_impl= table_impl;
  m_tuple_id_range.reset();
}

Ndb_local_table_info::~Ndb_local_table_info()
{
}

LocalDictCache::LocalDictCache(){
  m_tableHash.createHashTable();
}

LocalDictCache::~LocalDictCache(){
  m_tableHash.releaseHashTable();
}

Ndb_local_table_info * 
LocalDictCache::get(const char * name){
  ASSERT_NOT_MYSQLD;
  assert(! is_ndb_blob_table(name));
  const Uint32 len = (Uint32)strlen(name);
  return m_tableHash.getData(name, len);
}

void 
LocalDictCache::put(const char * name, Ndb_local_table_info * tab_info){
  ASSERT_NOT_MYSQLD;
  assert(! is_ndb_blob_table(name));
  const Uint32 id = tab_info->m_table_impl->m_id;
  m_tableHash.insertKey(name, (Uint32)strlen(name), id, tab_info);
}

void
LocalDictCache::drop(const char * name){
  ASSERT_NOT_MYSQLD;
  assert(! is_ndb_blob_table(name));
  Ndb_local_table_info *info= m_tableHash.deleteKey(name, (Uint32)strlen(name));
  DBUG_ASSERT(info != 0);
  Ndb_local_table_info::destroy(info);
}

/*****************************************************************
 * Global cache
 */
GlobalDictCache::GlobalDictCache(){
  DBUG_ENTER("GlobalDictCache::GlobalDictCache");
  m_tableHash.createHashTable();
  m_waitForTableCondition = NdbCondition_Create();
  if (f_invalid_table == NULL)
    f_invalid_table = new NdbTableImpl();
  if (f_altered_table == NULL)
    f_altered_table = new NdbTableImpl();
  ndb_dict_cache_count++;
  DBUG_VOID_RETURN;
}

GlobalDictCache::~GlobalDictCache(){
  DBUG_ENTER("GlobalDictCache::~GlobalDictCache");
  if (--ndb_dict_cache_count == 0)
  {
    if (f_invalid_table)
    {
      delete f_invalid_table;
      f_invalid_table = 0;
    }
    if (f_altered_table)
    {
      delete f_altered_table;
      f_altered_table = 0;
    }
  }
  NdbElement_t<Vector<TableVersion> > * curr = m_tableHash.getNext(0);
  while(curr != 0){
    Vector<TableVersion> * vers = curr->theData;
    const unsigned sz = vers->size();
    for(unsigned i = 0; i<sz ; i++){
      TableVersion tv= (*vers)[i];
      DBUG_PRINT("  ", ("vers[%d]: ver: %d, refCount: %d, status: %d",
                        i, tv.m_version, tv.m_refCount, tv.m_status));
      if(tv.m_impl != 0)
      {
        DBUG_PRINT("  ", ("m_impl: internalname: %s",
                          tv.m_impl->m_internalName.c_str()));
	delete (* vers)[i].m_impl;
      }
    }
    delete curr->theData;
    curr->theData= NULL;
    curr = m_tableHash.getNext(curr);
  }
  m_tableHash.releaseHashTable();
  NdbCondition_Destroy(m_waitForTableCondition);
  DBUG_VOID_RETURN;
}

void GlobalDictCache::printCache()
{
  DBUG_ENTER("GlobalDictCache::printCache");
  NdbElement_t<Vector<TableVersion> > * curr = m_tableHash.getNext(0);
  while(curr != 0){
    DBUG_PRINT("curr", ("len: %d, hash: %d, lk: %d, str: %s",
                        curr->len, curr->hash, curr->localkey1,
                        (char*) curr->str));
    if (curr->theData){
      Vector<TableVersion> * vers = curr->theData;
      const unsigned sz = vers->size();
      for(unsigned i = 0; i<sz ; i++){
        TableVersion tv= (*vers)[i];
        DBUG_PRINT("  ", ("impl: %p  vers[%d]: ver: %d, refCount: %d, status: %d",
                          tv.m_impl, i, tv.m_version, tv.m_refCount, tv.m_status));
        if(tv.m_impl != 0)
        {
          DBUG_PRINT("  ", ("m_impl: internalname: %s",
                            tv.m_impl->m_internalName.c_str()));
        }
      }
    }
    else
    {
      DBUG_PRINT("  ", ("NULL"));
    }
    curr = m_tableHash.getNext(curr);
  }
  DBUG_VOID_RETURN;
}

NdbTableImpl *
GlobalDictCache::get(const char * name, int *error)
{
  DBUG_ENTER("GlobalDictCache::get");
  DBUG_PRINT("enter", ("name: %s", name));
  assert(! is_ndb_blob_table(name));

  const Uint32 len = (Uint32)strlen(name);
  Vector<TableVersion> * versions = 0;
  versions = m_tableHash.getData(name, len);
  if(versions == 0){
    versions = new Vector<TableVersion>(2);
    if (versions == NULL)
    {
      *error = -1;
      DBUG_RETURN(0);
    }
    m_tableHash.insertKey(name, len, 0, versions);
  }

  int waitTime = 100;

  bool retreive = false;
  while(versions->size() > 0 && !retreive){
    TableVersion * ver = & versions->back();
    switch(ver->m_status){
    case OK:
      if (ver->m_impl->m_status == NdbDictionary::Object::Invalid)
      {
        ver->m_status = DROPPED;
        retreive = true; // Break loop
        if (ver->m_refCount == 0)
        {
          delete ver->m_impl;
          versions->erase(versions->size() - 1);
        }
        break;
      }
      ver->m_refCount++;
      DBUG_PRINT("info", ("Table OK tab: %p  version=%x.%x refCount=%u",
                          ver->m_impl,
                          ver->m_impl->m_version & 0xFFFFFF,
                          ver->m_impl->m_version >> 24,
                          ver->m_refCount));
      DBUG_RETURN(ver->m_impl);
    case DROPPED:
      retreive = true; // Break loop
      break;
    case RETREIVING:
      DBUG_PRINT("info", ("Wait for retrieving thread"));
      NdbCondition_WaitTimeout(m_waitForTableCondition, m_mutex, waitTime);
      continue;
    }
  }
  
  /**
   * Create new...
   */
  TableVersion tmp;
  tmp.m_version = 0;
  tmp.m_impl = 0;
  tmp.m_status = RETREIVING;
  tmp.m_refCount = 1; // The one retreiving it
  if (versions->push_back(tmp))
  {
    *error = -1;
     DBUG_RETURN(0);
  }
  DBUG_PRINT("info", ("No table found"));
  DBUG_RETURN(0);
}

NdbTableImpl *
GlobalDictCache::put(const char * name, NdbTableImpl * tab)
{
  DBUG_ENTER("GlobalDictCache::put");
  DBUG_PRINT("enter", ("tab: %p  name: %s, internal_name: %s version: %x.%x",
                       tab, name,
                       tab ? tab->m_internalName.c_str() : "tab NULL",
                       tab ? tab->m_version & 0xFFFFFF : 0,
                       tab ? tab->m_version >> 24 : 0));
  assert(! is_ndb_blob_table(name));

  const Uint32 len = (Uint32)strlen(name);
  Vector<TableVersion> * vers = m_tableHash.getData(name, len);
  if(vers == 0){
    // Should always tried to retreive it first 
    // and thus there should be a record
    abort(); 
  }

  const Uint32 sz = vers->size();
  if(sz == 0){
    // Should always tried to retreive it first 
    // and thus there should be a record
    abort(); 
  }
  
  TableVersion & ver = vers->back();
  if(ver.m_status != RETREIVING || 
     !(ver.m_impl == 0 || 
       ver.m_impl == f_invalid_table || ver.m_impl == f_altered_table) || 
     ver.m_version != 0 || 
     ver.m_refCount == 0){
    abort();
  }
  
  if(tab == 0)
  {
    DBUG_PRINT("info", ("No table found in db"));
    vers->erase(sz - 1);
  } 
  else if (ver.m_impl == 0) {
    DBUG_PRINT("info", ("Table OK"));
    ver.m_impl = tab;
    ver.m_version = tab->m_version;
    ver.m_status = OK;
  } 
  else if (ver.m_impl == f_invalid_table) 
  {
    DBUG_PRINT("info", ("Table DROPPED invalid"));
    ver.m_impl = tab;
    ver.m_version = tab->m_version;
    ver.m_status = DROPPED;
    ver.m_impl->m_status = NdbDictionary::Object::Invalid;    
  }
  else if(ver.m_impl == f_altered_table)
  {
    DBUG_PRINT("info", ("Table DROPPED altered"));
    ver.m_impl = tab;
    ver.m_version = tab->m_version;
    ver.m_status = DROPPED;
    ver.m_impl->m_status = NdbDictionary::Object::Altered;    
  }
  else
  {
    abort();
  }
  NdbCondition_Broadcast(m_waitForTableCondition);
  DBUG_RETURN(tab);
} 

unsigned
GlobalDictCache::get_size()
{
  NdbElement_t<Vector<TableVersion> > * curr = m_tableHash.getNext(0);
  int sz = 0;
  while(curr != 0){
    sz += curr->theData->size();
    curr = m_tableHash.getNext(curr);
  }
  if (sz)
  {
    printCache();
  }
  return sz;
}

void
GlobalDictCache::invalidate_all()
{
  DBUG_ENTER("GlobalDictCache::invalidate_all");
  NdbElement_t<Vector<TableVersion> > * curr = m_tableHash.getNext(0);
  while(curr != 0){
    Vector<TableVersion> * vers = curr->theData;
    if (vers->size())
    {
      TableVersion * ver = & vers->back();
      if (ver->m_status != RETREIVING)
      {
        ver->m_impl->m_status = NdbDictionary::Object::Invalid;
        ver->m_status = DROPPED;
        if (ver->m_refCount == 0)
        {
          delete ver->m_impl;
          vers->erase(vers->size() - 1);
        }
      }
    }
    curr = m_tableHash.getNext(curr);
  }
  DBUG_VOID_RETURN;
}

void
GlobalDictCache::invalidateDb(const char * name, size_t len)
{
  DBUG_ENTER("GlobalDictCache::invalidateDb");
  NdbElement_t<Vector<TableVersion> > * curr = m_tableHash.getNext(0);
  while(curr != 0)
  {
    Vector<TableVersion> * vers = curr->theData;
    if (vers->size())
    {
      TableVersion * ver = & vers->back();
      if (ver->m_status != RETREIVING)
      {
        if (ver->m_impl->matchDb(name, len))
        {
          ver->m_impl->m_status = NdbDictionary::Object::Invalid;
          ver->m_status = DROPPED;
          if (ver->m_refCount == 0)
          {
            delete ver->m_impl;
            vers->erase(vers->size() - 1);
          }
        }
      }
    }
    curr = m_tableHash.getNext(curr);
  }
  DBUG_VOID_RETURN;
}

void
GlobalDictCache::release(const NdbTableImpl * tab, int invalidate)
{
  DBUG_ENTER("GlobalDictCache::release");
  DBUG_PRINT("enter", ("tab: %p  internal_name: %s",
                       tab, tab->m_internalName.c_str()));
  assert(! is_ndb_blob_table(tab));

  unsigned i;
  const Uint32 len = (Uint32)strlen(tab->m_internalName.c_str());
  Vector<TableVersion> * vers = 
    m_tableHash.getData(tab->m_internalName.c_str(), len);
  if(vers == 0){
    // Should always tried to retreive it first 
    // and thus there should be a record
    abort(); 
  }

  const Uint32 sz = vers->size();
  if(sz == 0){
    // Should always tried to retreive it first 
    // and thus there should be a record
    abort(); 
  }
  
  for(i = 0; i < sz; i++){
    TableVersion & ver = (* vers)[i];
    if(ver.m_impl == tab){
      if(ver.m_refCount == 0 || ver.m_status == RETREIVING || 
	 ver.m_version != tab->m_version){
	DBUG_PRINT("info", ("Releasing with refCount=%d status=%d impl=%p",
                            ver.m_refCount, ver.m_status, ver.m_impl));
	break;
      }
      
      ver.m_refCount--;
      if (ver.m_impl->m_status == NdbDictionary::Object::Invalid || invalidate)
      {
        ver.m_impl->m_status = NdbDictionary::Object::Invalid;
        ver.m_status = DROPPED;
      }
      if (ver.m_refCount == 0 && ver.m_status == DROPPED)
      {
        DBUG_PRINT("info", ("refCount is zero, deleting m_impl"));
        delete ver.m_impl;
        vers->erase(i);
      }
      DBUG_VOID_RETURN;
    }
  }
  
  for(i = 0; i<sz; i++){
    TableVersion & ver = (* vers)[i];
    ndbout_c("%d: version: %d refCount: %d status: %d impl: %p",
	     i, ver.m_version, ver.m_refCount,
	     ver.m_status, ver.m_impl);
  }
  
  abort();
}

void
GlobalDictCache::alter_table_rep(const char * name, 
				 Uint32 tableId, 
				 Uint32 tableVersion,
				 bool altered)
{
  DBUG_ENTER("GlobalDictCache::alter_table_rep");
  const Uint32 len = (Uint32)strlen(name);
  Vector<TableVersion> * vers = 
    m_tableHash.getData(name, len);
  
  if(vers == 0)
  {
    DBUG_VOID_RETURN;
  }

  assert(! is_ndb_blob_table(name));
  const Uint32 sz = vers->size();
  if(sz == 0)
  {
    DBUG_VOID_RETURN;
  }
  
  for(Uint32 i = 0; i < sz; i++)
  {
    TableVersion & ver = (* vers)[i];
    if(ver.m_version == tableVersion && ver.m_impl && 
       (Uint32) ver.m_impl->m_id == tableId)
    {
      ver.m_status = DROPPED;
      ver.m_impl->m_status = altered ? 
	NdbDictionary::Object::Altered : NdbDictionary::Object::Invalid;
      if (ver.m_refCount == 0)
      {
        delete ver.m_impl;
        vers->erase(i);
      }
      DBUG_VOID_RETURN;
    }

    if(i == sz - 1 && ver.m_status == RETREIVING)
    {
      ver.m_impl = altered ? f_altered_table : f_invalid_table;
      DBUG_VOID_RETURN;
    } 
  }
  DBUG_VOID_RETURN;
}

int
GlobalDictCache::chg_ref_count(const NdbTableImpl * impl, int value)
{
  DBUG_ENTER("GlobalDictCache::chg_ref_count");
  const char * name = impl->m_internalName.c_str();
  assert(! is_ndb_blob_table(name));

  const Uint32 tableId = impl->m_id;
  const Uint32 tableVersion = impl->m_version;

  const Uint32 len = (Uint32)strlen(name);
  Vector<TableVersion> * vers = 
    m_tableHash.getData(name, len);
  
  if(vers == 0)
  {
    DBUG_RETURN(-1);
  }

  const Uint32 sz = vers->size();
  if(sz == 0)
  {
    DBUG_RETURN(-1);
  }
  
  for(Uint32 i = 0; i < sz; i++)
  {
    TableVersion & ver = (* vers)[i];
    if(ver.m_impl == impl)
    {
      if (value == +1)
      {
        DBUG_PRINT("info", ("%s id=%u ver=0x%x: inc old ref count %u",
                            name, tableId, tableVersion, ver.m_refCount));
        ver.m_refCount++;
      }
      else if (value == -1)
      {
        DBUG_PRINT("info", ("%s id=%u ver=0x%x: dec old ref count %u",
                            name, tableId, tableVersion, ver.m_refCount));
        if (ver.m_refCount == 0)
          abort();
        ver.m_refCount--;
        if (ver.m_refCount == 0)
        {
          delete ver.m_impl;
          vers->erase(i);
        }
      }
      else
        abort();
      DBUG_RETURN(0);
    }
  }
  DBUG_RETURN(0);
}

template class Vector<GlobalDictCache::TableVersion>;