X-Git-Url: http://git.indexdata.com/?a=blobdiff_plain;f=src%2Ffilter_session_shared.cpp;h=c243dea6357c976524d3b9ca0712ac926608abf3;hb=4bfec20c393570a81471c644fad6268eed3dba29;hp=6319c3560699e3ffdab2cdb7d92726e22d0549aa;hpb=e9530d6289df9b848e4901d7c73f5a2f27706d66;p=metaproxy-moved-to-github.git diff --git a/src/filter_session_shared.cpp b/src/filter_session_shared.cpp index 6319c35..c243dea 100644 --- a/src/filter_session_shared.cpp +++ b/src/filter_session_shared.cpp @@ -1,5 +1,5 @@ /* This file is part of Metaproxy. - Copyright (C) 2005-2008 Index Data + Copyright (C) 2005-2011 Index Data Metaproxy is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free @@ -18,8 +18,8 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA #include "config.hpp" -#include "filter.hpp" -#include "package.hpp" +#include +#include #include #include @@ -28,7 +28,7 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA #include #include -#include "util.hpp" +#include #include "filter_session_shared.hpp" #include @@ -36,6 +36,7 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA #include #include #include +#include #include #include #include @@ -46,6 +47,7 @@ namespace yf = metaproxy_1::filter; namespace metaproxy_1 { namespace filter { + // key for session.. We'll only share sessions with same InitKey class SessionShared::InitKey { public: bool operator < (const SessionShared::InitKey &k) const; @@ -53,13 +55,13 @@ namespace metaproxy_1 { InitKey(const InitKey &); ~InitKey(); private: - InitKey &operator = (const InitKey &k); char *m_idAuthentication_buf; int m_idAuthentication_size; char *m_otherInfo_buf; int m_otherInfo_size; ODR m_odr; }; + // worker thread .. for expiry of sessions class SessionShared::Worker { public: Worker(SessionShared::Rep *rep); @@ -67,6 +69,7 @@ namespace metaproxy_1 { private: SessionShared::Rep *m_p; }; + // backend result set class SessionShared::BackendSet { public: std::string m_result_set_id; @@ -75,15 +78,19 @@ namespace metaproxy_1 { yazpp_1::Yaz_Z_Query m_query; time_t m_time_last_use; void timestamp(); + yazpp_1::RecordCache m_record_cache; BackendSet( const std::string &result_set_id, const Databases &databases, const yazpp_1::Yaz_Z_Query &query); bool search( Package &frontend_package, + Package &search_package, const Z_APDU *apdu_req, - const BackendInstancePtr bp); + const BackendInstancePtr bp, + Z_Records **z_records); }; + // backend connection instance class SessionShared::BackendInstance { friend class Rep; friend class BackendClass; @@ -98,6 +105,7 @@ namespace metaproxy_1 { mp::Package * m_close_package; ~BackendInstance(); }; + // backends of some class (all with same InitKey) class SessionShared::BackendClass : boost::noncopyable { friend class Rep; friend struct Frontend; @@ -108,7 +116,7 @@ namespace metaproxy_1 { BackendInstancePtr get_backend(const Package &package); void use_backend(BackendInstancePtr b); void release_backend(BackendInstancePtr b); - void expire(); + void expire_class(); yazpp_1::GDU m_init_request; yazpp_1::GDU m_init_response; boost::mutex m_mutex_backend_class; @@ -123,6 +131,7 @@ namespace metaproxy_1 { int session_ttl); ~BackendClass(); }; + // frontend result set class SessionShared::FrontendSet { Databases m_databases; yazpp_1::Yaz_Z_Query m_query; @@ -134,6 +143,7 @@ namespace metaproxy_1 { const yazpp_1::Yaz_Z_Query &query); FrontendSet(); }; + // frontend session struct SessionShared::Frontend { Frontend(Rep *rep); ~Frontend(); @@ -157,6 +167,7 @@ namespace metaproxy_1 { BackendClassPtr m_backend_class; FrontendSets m_frontend_sets; }; + // representation class SessionShared::Rep { friend class SessionShared; friend struct Frontend; @@ -169,6 +180,7 @@ namespace metaproxy_1 { private: void init(Package &package, const Z_GDU *gdu, FrontendPtr frontend); + void start(); boost::mutex m_mutex; boost::condition m_cond_session_ready; std::map m_clients; @@ -179,6 +191,7 @@ namespace metaproxy_1 { int m_resultset_ttl; int m_resultset_max; int m_session_ttl; + bool m_optimize_search; }; } } @@ -270,7 +283,15 @@ void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b) while (it != m_backend_list.end()) { if (*it == b) + { + mp::odr odr; + (*it)->m_close_package->response() = odr.create_close( + 0, Z_Close_lackOfActivity, 0); + (*it)->m_close_package->session().close(); + (*it)->m_close_package->move(); + it = m_backend_list.erase(it); + } else it++; } @@ -359,13 +380,17 @@ yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_ba m_named_result_sets = false; Z_GDU *gdu = init_package.response().get(); - if (!init_package.session().is_closed() - && gdu && gdu->which == Z_GDU_Z3950 - && gdu->u.z3950->which == Z_APDU_initResponse) + if (init_package.session().is_closed()) { + /* already closed. We don't know why */ + return null; + } + else if (gdu && gdu->which == Z_GDU_Z3950 + && gdu->u.z3950->which == Z_APDU_initResponse + && *gdu->u.z3950->u.initResponse->result) + { + /* successful init response */ Z_InitResponse *res = gdu->u.z3950->u.initResponse; - if (!*res->result) - return null; m_init_response = gdu->u.z3950; if (ODR_MASK_GET(res->options, Z_Options_namedResultSets)) { @@ -374,7 +399,10 @@ yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_ba } else { - // did not receive an init response or closed + /* not init or init rejected */ + init_package.copy_filter(frontend_package); + init_package.session().close(); + init_package.move(); return null; } bp->m_in_use = true; @@ -473,15 +501,18 @@ yf::SessionShared::BackendSet::BackendSet( timestamp(); } +static int get_diagnostic(Z_DefaultDiagFormat *r) +{ + return *r->condition; +} + bool yf::SessionShared::BackendSet::search( mp::Package &frontend_package, + mp::Package &search_package, const Z_APDU *frontend_apdu, - const BackendInstancePtr bp) + const BackendInstancePtr bp, + Z_Records **z_records) { - Package search_package(bp->m_session, frontend_package.origin()); - - search_package.copy_filter(frontend_package); - mp::odr odr; Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest); Z_SearchRequest *req = apdu_req->u.searchRequest; @@ -500,41 +531,14 @@ bool yf::SessionShared::BackendSet::search( search_package.request() = apdu_req; search_package.move(); - - Z_Records *z_records_diag = 0; + Z_GDU *gdu = search_package.response().get(); if (!search_package.session().is_closed() && gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which == Z_APDU_searchResponse) { Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse; - if (b_resp->records) - { - if (b_resp->records->which == Z_Records_NSD - || b_resp->records->which == Z_Records_multipleNSD) - z_records_diag = b_resp->records; - } - if (z_records_diag) - { - if (frontend_apdu->which == Z_APDU_searchRequest) - { - Z_APDU *f_apdu = odr.create_searchResponse(frontend_apdu, - 0, 0); - Z_SearchResponse *f_resp = f_apdu->u.searchResponse; - f_resp->records = z_records_diag; - frontend_package.response() = f_apdu; - return false; - } - if (frontend_apdu->which == Z_APDU_presentRequest) - { - Z_APDU *f_apdu = odr.create_presentResponse(frontend_apdu, - 0, 0); - Z_PresentResponse *f_resp = f_apdu->u.presentResponse; - f_resp->records = z_records_diag; - frontend_package.response() = f_apdu; - return false; - } - } + *z_records = b_resp->records; m_result_set_size = *b_resp->resultCount; return true; } @@ -606,30 +610,35 @@ void yf::SessionShared::Frontend::get_set(mp::Package &package, BackendInstancePtr &found_backend, BackendSetPtr &found_set) { + bool session_restarted = false; + +restart: std::string result_set_id; BackendClassPtr bc = m_backend_class; { boost::mutex::scoped_lock lock(bc->m_mutex_backend_class); - - // look at each backend and see if we have a similar search - BackendInstanceList::const_iterator it = bc->m_backend_list.begin(); - for (; it != bc->m_backend_list.end(); it++) + if (m_p->m_optimize_search) { - if (!(*it)->m_in_use) + // look at each backend and see if we have a similar search + BackendInstanceList::const_iterator it = bc->m_backend_list.begin(); + for (; it != bc->m_backend_list.end(); it++) { - BackendSetList::const_iterator set_it = (*it)->m_sets.begin(); - for (; set_it != (*it)->m_sets.end(); set_it++) + if (!(*it)->m_in_use) { - if ((*set_it)->m_databases == databases - && query.match(&(*set_it)->m_query)) + BackendSetList::const_iterator set_it = (*it)->m_sets.begin(); + for (; set_it != (*it)->m_sets.end(); set_it++) { - found_set = *set_it; - found_backend = *it; - bc->use_backend(found_backend); - found_set->timestamp(); - // found matching set. No need to search again - return; + if ((*set_it)->m_databases == databases + && query.match(&(*set_it)->m_query)) + { + found_set = *set_it; + found_backend = *it; + bc->use_backend(found_backend); + found_set->timestamp(); + // found matching set. No need to search again + return; + } } } } @@ -677,11 +686,81 @@ void yf::SessionShared::Frontend::get_set(mp::Package &package, // we must search ... BackendSetPtr new_set(new BackendSet(result_set_id, databases, query)); - if (!new_set->search(package, apdu_req, found_backend)) + Z_Records *z_records = 0; + + Package search_package(found_backend->m_session, package.origin()); + search_package.copy_filter(package); + + if (!new_set->search(package, search_package, + apdu_req, found_backend, &z_records)) { bc->remove_backend(found_backend); return; // search error } + + if (z_records) + { + int condition = 0; + if (z_records->which == Z_Records_NSD) + { + condition = + get_diagnostic(z_records->u.nonSurrogateDiagnostic); + } + else if (z_records->which == Z_Records_multipleNSD) + { + if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1 + && + + z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which == + Z_DiagRec_defaultFormat) + { + condition = get_diagnostic( + z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat); + + } + } + if (!session_restarted && + condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR) + { + bc->remove_backend(found_backend); + session_restarted = true; + found_backend.reset(); + goto restart; + + } + + if (condition) + { + mp::odr odr; + if (apdu_req->which == Z_APDU_searchRequest) + { + Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, + 0, 0); + Z_SearchResponse *f_resp = f_apdu->u.searchResponse; + *f_resp->searchStatus = Z_SearchResponse_none; + f_resp->records = z_records; + package.response() = f_apdu; + } + if (apdu_req->which == Z_APDU_presentRequest) + { + Z_APDU *f_apdu = odr.create_presentResponse(apdu_req, + 0, 0); + Z_PresentResponse *f_resp = f_apdu->u.presentResponse; + f_resp->records = z_records; + package.response() = f_apdu; + } + bc->release_backend(found_backend); + return; // search error + } + } + if (!session_restarted && new_set->m_result_set_size < 0) + { + bc->remove_backend(found_backend); + session_restarted = true; + found_backend.reset(); + goto restart; + } + found_set = new_set; found_set->timestamp(); found_backend->m_sets.push_back(found_set); @@ -771,6 +850,34 @@ void yf::SessionShared::Frontend::present(mp::Package &package, if (!found_set) return; + Z_NamePlusRecordList *npr_res = 0; + if (found_set->m_record_cache.lookup(odr, &npr_res, + *req->resultSetStartPoint, + *req->numberOfRecordsRequested, + req->preferredRecordSyntax, + req->recordComposition)) + { + Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0); + Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse; + + yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF + " records in cache %p", + *req->resultSetStartPoint, + *req->numberOfRecordsRequested, + &found_set->m_record_cache); + + *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested; + *f_resp->nextResultSetPosition = + *req->resultSetStartPoint + *req->numberOfRecordsRequested; + // f_resp->presentStatus assumed OK. + f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records)); + f_resp->records->which = Z_Records_DBOSD; + f_resp->records->u.databaseOrSurDiagnostics = npr_res; + package.response() = f_apdu_res; + bc->release_backend(found_backend); + return; + } + Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest); Z_PresentRequest *p_req = p_apdu->u.presentRequest; p_req->preferredRecordSyntax = req->preferredRecordSyntax; @@ -802,6 +909,20 @@ void yf::SessionShared::Frontend::present(mp::Package &package, f_resp->records = b_resp->records; f_resp->otherInfo = b_resp->otherInfo; package.response() = f_apdu_res; + + if (b_resp->records && b_resp->records->which == Z_Records_DBOSD) + { + yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF + " records to cache %p", + *req->resultSetStartPoint, + *f_resp->numberOfRecordsReturned, + &found_set->m_record_cache); + found_set->m_record_cache.add( + odr, + b_resp->records->u.databaseOrSurDiagnostics, + *req->resultSetStartPoint, + *f_resp->numberOfRecordsReturned); + } bc->release_backend(found_backend); } else @@ -852,7 +973,7 @@ void yf::SessionShared::Worker::operator() (void) m_p->expire(); } -void yf::SessionShared::BackendClass::expire() +void yf::SessionShared::BackendClass::expire_class() { time_t now; time(&now); @@ -895,7 +1016,7 @@ void yf::SessionShared::Rep::expire() BackendClassMap::const_iterator b_it = m_backend_map.begin(); for (; b_it != m_backend_map.end(); b_it++) - b_it->second->expire(); + b_it->second->expire_class(); } } @@ -904,6 +1025,11 @@ yf::SessionShared::Rep::Rep() m_resultset_ttl = 30; m_resultset_max = 10; m_session_ttl = 90; + m_optimize_search = true; +} + +void yf::SessionShared::Rep::start() +{ yf::SessionShared::Worker w(this); m_thrds.add_thread(new boost::thread(w)); } @@ -915,6 +1041,10 @@ yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep) yf::SessionShared::~SessionShared() { } +void yf::SessionShared::start() const +{ + m_p->start(); +} yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep) { @@ -1032,7 +1162,8 @@ void yf::SessionShared::process(mp::Package &package) const m_p->release_frontend(package); } -void yf::SessionShared::configure(const xmlNode *ptr, bool test_only) +void yf::SessionShared::configure(const xmlNode *ptr, bool test_only, + const char *path) { for (ptr = ptr->children; ptr; ptr = ptr->next) { @@ -1051,6 +1182,11 @@ void yf::SessionShared::configure(const xmlNode *ptr, bool test_only) m_p->m_resultset_max = mp::xml::get_int(attr->children, 10); } + else if (!strcmp((const char *) attr->name, "optimizesearch")) + { + m_p->m_optimize_search = + mp::xml::get_bool(attr->children, true); + } else throw mp::filter::FilterException( "Bad attribute " + std::string((const char *) @@ -1096,8 +1232,9 @@ extern "C" { /* * Local variables: * c-basic-offset: 4 + * c-file-style: "Stroustrup" * indent-tabs-mode: nil - * c-file-style: "stroustrup" * End: * vim: shiftwidth=4 tabstop=8 expandtab */ +