[BACK]Return to pa_db_connection.C CVS log [TXT] [DIR] Up to [parser3project] / parser3 / src / main

Diff for /parser3/src/main/Attic/pa_db_connection.C between versions 1.10 and 1.33

version 1.10, 2001/10/24 14:11:25 version 1.33, 2001/11/12 10:00:31
Line 1 Line 1
 /** @file  /** @file
         Parser: Charset connection implementation.          Parser: sql driver connection implementation.
   
         Copyright (c) 2001 ArtLebedev Group (http://www.artlebedev.com)          Copyright (c) 2001 ArtLebedev Group (http://www.artlebedev.com)
         Author: Alexander Petrosyan <paf@design.ru> (http://design.ru/paf)          Author: Alexander Petrosyan <paf@design.ru> (http://paf.design.ru)
   
         $Id$          $Id$
   
           developed with LIBDB 2.7.4
 */  */
   
 #include "pa_config_includes.h"  #include "pa_config_includes.h"
 #ifdef HAVE_LIBDB  #ifdef DB2
   
 #include "pa_db_connection.h"  #include "pa_db_connection.h"
   #include "pa_db_table.h"
 #include "pa_exception.h"  #include "pa_exception.h"
   #include "pa_threads.h"
   #include "pa_stack.h"
   #include "pa_common.h"
   #include "pa_vtable.h"
   
 // defines  // defines
   
 #define READ_ADDITIONAL_FLAGS DB_RMW  
   
 // consts  // consts
   
 const int DATA_STRING_SERIALIZED_VERSION=0x0100;  /// @test increase
   const int DB_CHECKPOINT_MINUTES=1; 
   
 // helper types  const int EXPIRE_UNUSED_TABLE_SECONDS=60;
   const int CHECK_EXPIRED_TABLES_SECONDS=EXPIRE_UNUSED_TABLE_SECONDS*2;
   
 #ifndef DOXYGEN  // callbacks
 struct Data_string_serialized_prolog {  
         int version;  static void db_paniccall(DB_ENV *dbenv, int error) {
         time_t time_to_die;          throw Exception(0, 0, 
 };                  0, 
 #endif                  "db_panic: %s (%d)", 
                   strerror(error), error);
   }
   
   static void db_errcall(const char *, char *buffer) {
           throw Exception(0, 0, 
                   0, 
                   "db_err: %s", 
                   buffer);
   }
   
   static void expire_table(const Hash::Key& key, Hash::Val *& value, void *info) {
           DB_Table *table=static_cast<DB_Table *>(value);
           time_t older_dies=reinterpret_cast<time_t>(info);
   
           if(table->expired(older_dies)) {
                   table->~DB_Table();  value=0;
           }
   }
   
 // DB_Connection  // DB_Connection
   
   DB_Connection::DB_Connection(Pool& apool, const String& adb_home) : Pooled(apool),
           time_used(0),
           fdb_home(adb_home), used(0),
           table_cache(apool),
           prev_expiration_pass_time(0) {
           //_asm  int 3;
           char DB_DATA_DIR__VALUE[MAX_STRING];
           char DB_LOG_DIR__VALUE[MAX_STRING];
           char DB_TMP_DIR__VALUE[MAX_STRING];
   
           const char *db_home_cstr=fdb_home.cstr(String::UL_FILE_SPEC);
   
           snprintf(DB_DATA_DIR__VALUE, MAX_STRING, "DB_DATA_DIR %s", db_home_cstr);
           snprintf(DB_LOG_DIR__VALUE, MAX_STRING, "DB_LOG_DIR %s", db_home_cstr);
           snprintf(DB_TMP_DIR__VALUE, MAX_STRING, "DB_TMP_DIR %s", db_home_cstr);
   
           char *db_config[] = {
                   DB_DATA_DIR__VALUE,
                   DB_LOG_DIR__VALUE,
                   DB_TMP_DIR__VALUE, 
                   0
           };
   
           u_int32_t flags=
                   (parser_multithreaded?DB_THREAD:0)
                   | DB_CREATE 
                   | DB_INIT_MPOOL | DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_TXN;
   
           // prepare dbenv structure
           memset(&dbenv, 0, sizeof(dbenv));
           
           // error handlers
   #if DB_VERSION_MINOR >= 7
           dbenv.db_paniccall=db_paniccall;
   #endif
           dbenv.db_errcall=db_errcall;
   
           // lockdetector flags
           dbenv.lk_detect=DB_LOCK_RANDOM;
   
           // init
           check("db_appinit", &fdb_home, db_appinit(
                   db_home_cstr,
                   db_config, 
                   &dbenv, 
                   flags
                   ));
   
           // after some hang noticed this to be null
           if(!dbenv.tx_info) 
                   throw Exception(0, 0,
                           &fdb_home,
                           "transaction system failed to initialize");
   }
   
   DB_Connection::~DB_Connection() {
           // close tables
           table_cache.for_each(expire_table, 
                   reinterpret_cast<void *>(time(0)/* =now= expire[close] all*/));
           
           // destroy connection data
           check("db_appexit", &fdb_home, db_appexit(&dbenv));
   }
   
   
 void DB_Connection::check(const char *operation, const String *source, int error) {  void DB_Connection::check(const char *operation, const String *source, int error) {
         switch(error) {          switch(error) {
         case 0:           case 0: 
                 // no error                  // no error
                 break;                   break; 
                   
         case DB_KEYEXIST:  
                 // DB_KEYEXIST is a "normal" return, so should not be  
                 // thrown as an error  
                 break;   
   
         case DB_RUNRECOVERY:  
                 // mark as unsafe, so not to cache it  
                 needs_recovery=true;  
                 throw Exception(0, 0,   
                         source,  
                         "action failed, RUN RECOVERY UTILITY. db %s error, real filename '%s'",   
                         operation, file_spec_cstr);  
                   
         default:          default:
                 throw Exception(0, 0,                   if(error<0)
                         source,                           throw Exception(0, 0, 
                         "action failed. db %s error: %s (%d), real filename '%s'",                                   source, 
                         operation, strerror(error), error, file_spec_cstr);                                  "db %s error: %d", 
                                   operation, error);
                   else
                           throw Exception(0, 0, 
                                   source, 
                                   "db %s error: %s (%d)", 
                                   operation, strerror(error), error);
         }          }
 }  }
   
 void DB_Connection::key_string_to_dbt(const String& key_string, DBT& key_result) {  DB_Table_ptr DB_Connection::get_table_ptr(const String& request_file_name, const String *source) {
         memset(&key_result, 0, sizeof(key_result));          // checkpoint
         key_result.data=key_string.cstr(String::UL_AS_IS);          { 
         key_result.size=key_string.size();                  int error=txn_checkpoint(dbenv.tx_info, 0/*kbyte*/, DB_CHECKPOINT_MINUTES/*min*/);
 }                  /*
                   DB_INCOMPLETE if there were pages that needed to be written 
                   but that memp_sync was unable to write immediately. 
                   In this case, the txn_checkpoint call should be retried. 
   
                   we'll just ignore that
                   */
                   if(error!=DB_INCOMPLETE)
                           check("checkpoint", source, error);
           }
   
 String& DB_Connection::key_dbt_to_string(const DBT& key_dbt) {          // first trying to get cached table
         String& result=*new(*fservices_pool) String(*fservices_pool);          DB_Table *result=get_table_from_cache(request_file_name);
         if(key_dbt.size) {          if(!result) { // no cached table
                 char *request_data=(char *)malloc(key_dbt.size);                  // make global_file_name C-string on global pool
                 memcpy(request_data, key_dbt.data, key_dbt.size);                  const char *request_file_name_cstr=request_file_name.cstr();
                 result.APPEND_TAINTED(request_data, key_dbt.size, file_spec_cstr, 0/*line*/);                  char *global_file_name_cstr=(char *)malloc(strlen(request_file_name_cstr)+1);
                   strcpy(global_file_name_cstr, request_file_name_cstr);
                   // make global_file_name string on global pool
                   String& global_file_name=*new(this->pool()) String(this->pool(), global_file_name_cstr);
                   
                   // allocate in global pool 
                   // associate with services[request]
                   // NOTE: never freed up!
                   result=new(this->pool()) DB_Table(this->pool(), global_file_name, *this);
                   // cache it
                   put_table_to_cache(global_file_name, *result);
         }          }
         return result;  
 }  
   
 void DB_Connection::data_string_to_dbt(const String& data_string, time_t time_to_die,           // return auto-it
                                                                            DBT& data_result) {          return DB_Table_ptr(result);
         memset(&data_result, 0, sizeof(data_result));  }
   
         data_string.serialize(  // table cache
                 sizeof(Data_string_serialized_prolog),   /// @todo get rid of memory spending Stack [zeros deep inside got accumulated]
                 data_result.data, data_result.size);  DB_Table *DB_Connection::get_table_from_cache(const String& file_name) { 
           SYNCHRONIZED;
   
         Data_string_serialized_prolog& prolog=          maybe_expire_table_cache();
                 *static_cast<Data_string_serialized_prolog *>(data_result.data);  
   
         prolog.version=DATA_STRING_SERIALIZED_VERSION;          return static_cast<DB_Table *>(table_cache.get(file_name));
         prolog.time_to_die=time_to_die;  
 }  }
   
 String *DB_Connection::data_dbt_to_string(const DBT& data_dbt) {  void DB_Connection::put_table_to_cache(const String& file_name, DB_Table& table) { 
         Data_string_serialized_prolog& prolog=          SYNCHRONIZED;
                 *static_cast<Data_string_serialized_prolog *>(data_dbt.data);  
   
         if(prolog.version!=DATA_STRING_SERIALIZED_VERSION)          table_cache.put(file_name, &table);
                 throw Exception(0, 0,  
                         &ffile_spec,  
                         "data string version 0x%04X not equal to 0x%04X, recreate file",  
                                 prolog.version, DATA_STRING_SERIALIZED_VERSION);  
   
         if(prolog.time_to_die/*specified*/ && prolog.time_to_die <= time(0)/*expired*/)  
                 return 0;  
   
         String& result=*new(*fservices_pool) String(*fservices_pool);  
         result.deserialize(  
                 sizeof(Data_string_serialized_prolog),   
                 data_dbt.data, data_dbt.size, file_spec_cstr);  
         return &result;  
 }  
   
 DB_Connection::DB_Connection(Pool& pool, const String& afile_spec, DB_ENV& adbenv) : Pooled(pool),  
         fdbenv(adbenv),  
         ffile_spec(afile_spec), file_spec_cstr(afile_spec.cstr(String::UL_FILE_SPEC)),  
         fservices_pool(0), db(0), ftid(0), needs_recovery(false),   
         time_used(0) {  
 }  
   
 void DB_Connection::connect() {   
         // open  
         DB_INFO dbinfo;  
         memset(&dbinfo, 0, sizeof(dbinfo));  
         check("open/create", &ffile_spec, db_open(  
                 file_spec_cstr,   
                 PA_DB_ACCESS_METHOD,   
                 DB_CREATE /* used in single thread, no need for |DB_THREAD*/,  
                 0666,   
                 &fdbenv, &dbinfo, &db));  
 }  
 /// @todo this one of reasons of not having ^try for now  
 void DB_Connection::disconnect() {   
         check("close", &ffile_spec, db->close(db, 0/*flags*/));  db=0;   
 }  
   
 void DB_Connection::put(const String& key, const String& data, time_t time_to_die) {  
         DBT dbt_key;  key_string_to_dbt(key, dbt_key);  
         DBT dbt_data;  data_string_to_dbt(data, time_to_die, dbt_data);  
         check("put", &key, db->put(db, ftid, &dbt_key, &dbt_data, 0/*flags*/));  
 }  
   
 String *DB_Connection::get(const String& key) {  
         DBT dbt_key;  key_string_to_dbt(key, dbt_key);  
         DBT dbt_data={0}; // must be zeroed  
         int error=db->get(db, ftid, &dbt_key, &dbt_data, READ_ADDITIONAL_FLAGS/*flags*/);  
         if(error==DB_NOTFOUND)  
                 return 0;  
         else {  
                 check("get", &key, error);  
                 String *result=data_dbt_to_string(dbt_data);  
                 if(!result) // save efforts by deleting expired keys  
                         check("del expired", &key, db->del(db, ftid, &dbt_key, 0/*flags*/));  
                 return result;  
         }                 
 }  
   
 void DB_Connection::remove(const String& key) {  
         DBT dbt_key;  key_string_to_dbt(key, dbt_key);  
   
         int error=db->del(db, ftid, &dbt_key, 0/*flags*/);  
         if(error!=DB_NOTFOUND)  
                 check("del", &key, error);  
 }  
   
 // DB_Cursor  
   
 DB_Cursor::DB_Cursor(  
                                          DB_Connection& aconnection,   
                                          const String *asource) : fsource(asource), fconnection(aconnection), cursor(0) {  
         check("cursor", fsource, fconnection.db->cursor(fconnection.db,  
                 fconnection.ftid, &cursor, 0/*flags*/));  
 }  
   
 DB_Cursor::~DB_Cursor() {  
         if(cursor) {  
                 check("c_close", fsource, cursor->c_close(cursor));  cursor=0;  
         }  
 }  }
   
 bool DB_Cursor::get(String *& key, String *& data, u_int32_t flags) {  void DB_Connection::maybe_expire_table_cache() {
         DBT dbt_key={0}; // must be zeroed          time_t now=time(0);
         DBT dbt_data={0}; // must be zeroed  
                   if(prev_expiration_pass_time<now-CHECK_EXPIRED_TABLES_SECONDS) {
         int error=cursor->c_get(cursor, &dbt_key, &dbt_data, flags | READ_ADDITIONAL_FLAGS);                  table_cache.for_each(expire_table, 
         if(error==DB_NOTFOUND)                          reinterpret_cast<void *>(now-EXPIRE_UNUSED_TABLE_SECONDS));
                 return false;  
                   prev_expiration_pass_time=now;
         check("c_get", fsource, error);  
   
         if(data=data_dbt_to_string(dbt_data)) // not expired  
                 key=&key_dbt_to_string(dbt_key);  
         else {  
                 // save efforts by deleting expired keys  
                 remove(0/*flags*/);  
                 key=0;  
         }          }
         return true;  
 }  }
   
 void DB_Cursor::remove(u_int32_t flags) {  static void add_table_to_status_table(const Hash::Key& key, Hash::Val *& value, void *info) {
         check("c_del", fsource,  cursor->c_del(cursor, flags));          DB_Table *db_table=static_cast<DB_Table *>(value);
           Table& status_table=*static_cast<Table *>(info);
           Pool& pool=status_table.pool();
   
           Array& row=*new(pool) Array(pool);
   
           // name
           row+=&db_table->file_name();
           // time
           time_t time_stamp=db_table->get_time_used();
           const char *unsafe_time_cstr=ctime(&time_stamp);
           int time_buf_size=strlen(unsafe_time_cstr);
           char *safe_time_buf=(char *)pool.malloc(time_buf_size);
           memcpy(safe_time_buf, unsafe_time_cstr, time_buf_size);
           row+=new(pool) String(pool, safe_time_buf, time_buf_size);
           // users
           char *users_buf=(char *)pool.malloc(MAX_NUMBER);
           row+=new(pool) String(pool, 
                   users_buf, snprintf(users_buf, MAX_NUMBER, "%d", db_table->get_users_count()));
   
           status_table+=&row;
   }
   Value& DB_Connection::get_status(Pool& pool, const String *source) {
           Array& columns=*new(pool) Array(pool);
           columns+=new(pool) String(pool, "name");
           columns+=new(pool) String(pool, "time");
           columns+=new(pool) String(pool, "users");
           Table& table=*new(pool) Table(pool, 0, &columns, table_cache.size());
   
           table_cache.for_each(add_table_to_status_table, &table);
   
           return *new(pool) VTable(pool, &table);
 }  }
   
 #endif  #endif
Removed from v.1.10  
changed lines
  Added in v.1.33

E-mail: