--- parser3/src/main/Attic/pa_db_connection.C 2001/10/24 09:34:26 1.5 +++ parser3/src/main/Attic/pa_db_connection.C 2001/11/12 10:00:31 1.33 @@ -1,194 +1,244 @@ /** @file - Parser: Charset connection implementation. + Parser: sql driver connection implementation. Copyright (c) 2001 ArtLebedev Group (http://www.artlebedev.com) - Author: Alexander Petrosyan (http://design.ru/paf) + Author: Alexander Petrosyan (http://paf.design.ru) - $Id: pa_db_connection.C,v 1.5 2001/10/24 09:34:26 parser Exp $ + $Id: pa_db_connection.C,v 1.33 2001/11/12 10:00:31 paf Exp $ + + developed with LIBDB 2.7.4 */ #include "pa_config_includes.h" -#ifdef HAVE_LIBDB +#ifdef DB2 #include "pa_db_connection.h" +#include "pa_db_table.h" #include "pa_exception.h" +#include "pa_threads.h" +#include "pa_stack.h" +#include "pa_common.h" +#include "pa_vtable.h" -// consts +// defines -const int DATA_STRING_SERIALIZED_VERSION=0x0100; +// consts -// helper types +/// @test increase +const int DB_CHECKPOINT_MINUTES=1; -#ifndef DOXYGEN -struct Data_string_serialized_prolog { - int version; - time_t time_to_die; -}; -#endif +const int EXPIRE_UNUSED_TABLE_SECONDS=60; +const int CHECK_EXPIRED_TABLES_SECONDS=EXPIRE_UNUSED_TABLE_SECONDS*2; -// DB_Connection +// callbacks -void DB_Connection::check(const char *operation, const String *source, int error) { - switch(error) { - case 0: - // no error - 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: - throw Exception(0, 0, - source, - "action failed. db %s error: %s (%d), real filename '%s'", - operation, strerror(error), error, file_spec_cstr); - } +static void db_paniccall(DB_ENV *dbenv, int error) { + throw Exception(0, 0, + 0, + "db_panic: %s (%d)", + strerror(error), error); } -void DB_Connection::key_string_to_dbt(const String& key_string, DBT& key_result) { - memset(&key_result, 0, sizeof(key_result)); - key_result.data=key_string.cstr(String::UL_AS_IS); - key_result.size=key_string.size(); +static void db_errcall(const char *, char *buffer) { + throw Exception(0, 0, + 0, + "db_err: %s", + buffer); } -String& DB_Connection::key_dbt_to_string(const DBT& key_dbt) { - String& result=*new(*fservices_pool) String(*fservices_pool); - if(key_dbt.size) { - char *request_data=(char *)malloc(key_dbt.size); - memcpy(request_data, key_dbt.data, key_dbt.size); - result.APPEND_TAINTED(request_data, key_dbt.size, file_spec_cstr, 0/*line*/); +static void expire_table(const Hash::Key& key, Hash::Val *& value, void *info) { + DB_Table *table=static_cast(value); + time_t older_dies=reinterpret_cast(info); + + if(table->expired(older_dies)) { + table->~DB_Table(); value=0; } - return result; } -void DB_Connection::data_string_to_dbt(const String& data_string, time_t time_to_die, - DBT& data_result) { - memset(&data_result, 0, sizeof(data_result)); +// DB_Connection - data_string.serialize( - sizeof(Data_string_serialized_prolog), - data_result.data, data_result.size); +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; - Data_string_serialized_prolog& prolog= - *static_cast(data_result.data); + // 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; - prolog.version=DATA_STRING_SERIALIZED_VERSION; - prolog.time_to_die=time_to_die; -} + // lockdetector flags + dbenv.lk_detect=DB_LOCK_RANDOM; -String *DB_Connection::data_dbt_to_string(const DBT& data_dbt) { - Data_string_serialized_prolog& prolog= - *static_cast(data_dbt.data); + // init + check("db_appinit", &fdb_home, db_appinit( + db_home_cstr, + db_config, + &dbenv, + flags + )); - if(prolog.version!=DATA_STRING_SERIALIZED_VERSION) + // after some hang noticed this to be null + if(!dbenv.tx_info) 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; + &fdb_home, + "transaction system failed to initialize"); } -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) { +DB_Connection::~DB_Connection() { + // close tables + table_cache.for_each(expire_table, + reinterpret_cast(time(0)/* =now= expire[close] all*/)); + + // destroy connection data + check("db_appexit", &fdb_home, db_appexit(&dbenv)); } -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; -} -/// @test string pieces [get/put preserve lang] -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*/)); +void DB_Connection::check(const char *operation, const String *source, int error) { + switch(error) { + case 0: + // no error + break; + + default: + if(error<0) + throw Exception(0, 0, + source, + "db %s error: %d", + operation, error); + else + throw Exception(0, 0, + source, + "db %s error: %s (%d)", + operation, strerror(error), error); + } } -String *DB_Connection::get(const String& key) { - DBT dbt_key; key_string_to_dbt(key, dbt_key); - // data - DBT dbt_data={0}; // must be zeroed - int error=db->get(db, ftid, &dbt_key, &dbt_data, 0/*flags*/); - if(error==DB_NOTFOUND) - return 0; - else { - check("get", &key, error); - return data_dbt_to_string(dbt_data); - } -} +DB_Table_ptr DB_Connection::get_table_ptr(const String& request_file_name, const String *source) { + // checkpoint + { + 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); + } -void DB_Connection::_delete(const String& key) { - DBT dbt_key; key_string_to_dbt(key, dbt_key); + // first trying to get cached table + DB_Table *result=get_table_from_cache(request_file_name); + if(!result) { // no cached table + // make global_file_name C-string on global pool + const char *request_file_name_cstr=request_file_name.cstr(); + 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); + } - int error=db->del(db, ftid, &dbt_key, 0/*flags*/); - if(error!=DB_NOTFOUND) - check("del", &key, error); + // return auto-it + return DB_Table_ptr(result); } -DB_Cursor DB_Connection::cursor(const String *source) { - return DB_Cursor(*this, source); +// table cache +/// @todo get rid of memory spending Stack [zeros deep inside got accumulated] +DB_Table *DB_Connection::get_table_from_cache(const String& file_name) { + SYNCHRONIZED; + + maybe_expire_table_cache(); + + return static_cast(table_cache.get(file_name)); } -// DB_Cursor +void DB_Connection::put_table_to_cache(const String& file_name, DB_Table& table) { + SYNCHRONIZED; -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*/)); + table_cache.put(file_name, &table); } -DB_Cursor::~DB_Cursor() { - if(cursor) { - check("c_close", fsource, cursor->c_close(cursor)); cursor=0; +void DB_Connection::maybe_expire_table_cache() { + time_t now=time(0); + + if(prev_expiration_pass_time(now-EXPIRE_UNUSED_TABLE_SECONDS)); + + prev_expiration_pass_time=now; } } -bool DB_Cursor::get(String *& key, String *& data, u_int32_t flags) { - DBT dbt_key={0}; // must be zeroed - DBT dbt_data={0}; // must be zeroed - - int error=cursor->c_get(cursor, &dbt_key, &dbt_data, flags); - if(error==DB_NOTFOUND) - return false; - - check("c_get", fsource, error); - - key=&key_dbt_to_string(dbt_key); - data=data_dbt_to_string(dbt_data); - return true; +static void add_table_to_status_table(const Hash::Key& key, Hash::Val *& value, void *info) { + DB_Table *db_table=static_cast(value); + Table& status_table=*static_cast(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