1 /* This file is part of Metaproxy.
2 Copyright (C) 2005-2012 Index Data
4 Metaproxy is free software; you can redistribute it and/or modify it under
5 the terms of the GNU General Public License as published by the Free
6 Software Foundation; either version 2, or (at your option) any later
9 Metaproxy is distributed in the hope that it will be useful, but WITHOUT ANY
10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 You should have received a copy of the GNU General Public License
15 along with this program; if not, write to the Free Software
16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
21 #include <metaproxy/filter.hpp>
22 #include <metaproxy/package.hpp>
24 #include <boost/thread/mutex.hpp>
25 #include <boost/thread/condition.hpp>
26 #include <boost/thread/thread.hpp>
27 #include <boost/thread/xtime.hpp>
28 #include <boost/shared_ptr.hpp>
29 #include <boost/format.hpp>
31 #include <metaproxy/util.hpp>
32 #include "filter_session_shared.hpp"
36 #include <yaz/otherinfo.h>
37 #include <yaz/diagbib1.h>
38 #include <yazpp/z-query.h>
39 #include <yazpp/record-cache.h>
44 namespace mp = metaproxy_1;
45 namespace yf = metaproxy_1::filter;
47 namespace metaproxy_1 {
50 // key for session.. We'll only share sessions with same InitKey
51 class SessionShared::InitKey {
53 bool operator < (const SessionShared::InitKey &k) const;
54 InitKey(Z_InitRequest *req);
55 InitKey(const InitKey &);
58 char *m_idAuthentication_buf;
59 int m_idAuthentication_size;
60 char *m_otherInfo_buf;
64 // worker thread .. for expiry of sessions
65 class SessionShared::Worker {
67 Worker(SessionShared::Rep *rep);
68 void operator() (void);
70 SessionShared::Rep *m_p;
73 class SessionShared::BackendSet {
75 std::string m_result_set_id;
76 Databases m_databases;
77 int m_result_set_size;
78 yazpp_1::Yaz_Z_Query m_query;
79 time_t m_time_last_use;
81 yazpp_1::RecordCache m_record_cache;
83 const std::string &result_set_id,
84 const Databases &databases,
85 const yazpp_1::Yaz_Z_Query &query);
87 Package &frontend_package,
88 Package &search_package,
89 const Z_APDU *apdu_req,
90 const BackendInstancePtr bp,
91 Z_Records **z_records);
93 // backend connection instance
94 class SessionShared::BackendInstance {
96 friend class BackendClass;
97 friend class BackendSet;
99 mp::Session m_session;
100 BackendSetList m_sets;
103 int m_result_set_sequence;
104 time_t m_time_last_use;
105 mp::Package * m_close_package;
109 // backends of some class (all with same InitKey)
110 class SessionShared::BackendClass : boost::noncopyable {
112 friend struct Frontend;
113 bool m_named_result_sets;
114 BackendInstanceList m_backend_list;
115 BackendInstancePtr create_backend(const Package &package);
116 void remove_backend(BackendInstancePtr b);
117 BackendInstancePtr get_backend(const Package &package);
118 void use_backend(BackendInstancePtr b);
119 void release_backend(BackendInstancePtr b);
121 yazpp_1::GDU m_init_request;
122 yazpp_1::GDU m_init_response;
123 boost::mutex m_mutex_backend_class;
125 time_t m_backend_set_ttl;
126 time_t m_backend_expiry_ttl;
127 size_t m_backend_set_max;
129 BackendClass(const yazpp_1::GDU &init_request,
135 // frontend result set
136 class SessionShared::FrontendSet {
137 Databases m_databases;
138 yazpp_1::Yaz_Z_Query m_query;
140 const Databases &get_databases();
141 const yazpp_1::Yaz_Z_Query &get_query();
143 const Databases &databases,
144 const yazpp_1::Yaz_Z_Query &query);
148 struct SessionShared::Frontend {
153 Z_Options m_init_options;
154 void search(Package &package, Z_APDU *apdu);
155 void present(Package &package, Z_APDU *apdu);
156 void scan(Package &package, Z_APDU *apdu);
158 void get_set(mp::Package &package,
159 const Z_APDU *apdu_req,
160 const Databases &databases,
161 yazpp_1::Yaz_Z_Query &query,
162 BackendInstancePtr &found_backend,
163 BackendSetPtr &found_set);
164 void override_set(BackendInstancePtr &found_backend,
165 std::string &result_set_id,
166 const Databases &databases,
167 bool out_of_sessions);
170 BackendClassPtr m_backend_class;
171 FrontendSets m_frontend_sets;
174 class SessionShared::Rep {
175 friend class SessionShared;
176 friend struct Frontend;
178 FrontendPtr get_frontend(Package &package);
179 void release_frontend(Package &package);
184 void init(Package &package, const Z_GDU *gdu,
185 FrontendPtr frontend);
187 boost::mutex m_mutex;
188 boost::condition m_cond_session_ready;
189 std::map<mp::Session, FrontendPtr> m_clients;
191 BackendClassMap m_backend_map;
192 boost::mutex m_mutex_backend_map;
193 boost::thread_group m_thrds;
197 bool m_optimize_search;
204 yf::SessionShared::FrontendSet::FrontendSet(
205 const Databases &databases,
206 const yazpp_1::Yaz_Z_Query &query)
207 : m_databases(databases), m_query(query)
211 const yf::SessionShared::Databases &
212 yf::SessionShared::FrontendSet::get_databases()
217 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
222 yf::SessionShared::InitKey::InitKey(const InitKey &k)
224 m_odr = odr_createmem(ODR_ENCODE);
226 m_idAuthentication_size = k.m_idAuthentication_size;
227 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
228 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
229 m_idAuthentication_size);
231 m_otherInfo_size = k.m_otherInfo_size;
232 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
233 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
237 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
239 m_odr = odr_createmem(ODR_ENCODE);
241 Z_IdAuthentication *t = req->idAuthentication;
242 z_IdAuthentication(m_odr, &t, 1, 0);
243 m_idAuthentication_buf =
244 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
246 Z_OtherInformation *o = req->otherInfo;
247 z_OtherInformation(m_odr, &o, 1, 0);
248 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
251 yf::SessionShared::InitKey::~InitKey()
256 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
260 c = mp::util::memcmp2(
261 (void*) m_idAuthentication_buf, m_idAuthentication_size,
262 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
268 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
269 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
277 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
279 boost::mutex::scoped_lock lock(m_mutex_backend_class);
284 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
286 BackendInstanceList::iterator it = m_backend_list.begin();
288 while (it != m_backend_list.end())
293 (*it)->m_close_package->response() = odr.create_close(
294 0, Z_Close_lackOfActivity, 0);
295 (*it)->m_close_package->session().close();
296 (*it)->m_close_package->move();
298 it = m_backend_list.erase(it);
307 yf::SessionShared::BackendInstancePtr
308 yf::SessionShared::BackendClass::get_backend(
309 const mp::Package &frontend_package)
312 boost::mutex::scoped_lock lock(m_mutex_backend_class);
314 BackendInstanceList::const_iterator it = m_backend_list.begin();
316 BackendInstancePtr backend1; // null
318 for (; it != m_backend_list.end(); it++)
320 if (!(*it)->m_in_use)
323 || (*it)->m_sequence_this < backend1->m_sequence_this)
329 use_backend(backend1);
333 return create_backend(frontend_package);
336 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
338 backend->m_in_use = true;
339 backend->m_sequence_this = m_sequence_top++;
342 void yf::SessionShared::BackendInstance::timestamp()
345 time(&m_time_last_use);
348 yf::SessionShared::BackendInstance::~BackendInstance()
350 delete m_close_package;
353 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
354 const mp::Package &frontend_package)
356 BackendInstancePtr bp(new BackendInstance);
357 BackendInstancePtr null;
359 bp->m_close_package =
360 new mp::Package(bp->m_session, frontend_package.origin());
361 bp->m_close_package->copy_filter(frontend_package);
363 Package init_package(bp->m_session, frontend_package.origin());
365 init_package.copy_filter(frontend_package);
367 yazpp_1::GDU actual_init_request = m_init_request;
368 Z_GDU *init_pdu = actual_init_request.get();
370 assert(init_pdu->which == Z_GDU_Z3950);
371 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
373 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
374 ODR_MASK_ZERO(req->options);
376 ODR_MASK_SET(req->options, Z_Options_search);
377 ODR_MASK_SET(req->options, Z_Options_present);
378 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
379 ODR_MASK_SET(req->options, Z_Options_scan);
381 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
382 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
383 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
385 init_package.request() = init_pdu;
389 boost::mutex::scoped_lock lock(m_mutex_backend_class);
391 m_named_result_sets = false;
392 Z_GDU *gdu = init_package.response().get();
393 if (init_package.session().is_closed())
395 /* already closed. We don't know why */
398 else if (gdu && gdu->which == Z_GDU_Z3950
399 && gdu->u.z3950->which == Z_APDU_initResponse
400 && *gdu->u.z3950->u.initResponse->result)
402 /* successful init response */
403 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
404 m_init_response = gdu->u.z3950;
405 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
407 m_named_result_sets = true;
412 /* not init or init rejected */
413 init_package.copy_filter(frontend_package);
414 init_package.session().close();
419 time(&bp->m_time_last_use);
420 bp->m_sequence_this = 0;
421 bp->m_result_set_sequence = 0;
422 m_backend_list.push_back(bp);
428 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
432 : m_named_result_sets(false), m_init_request(init_request),
433 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
434 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max)
437 yf::SessionShared::BackendClass::~BackendClass()
440 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
441 FrontendPtr frontend)
443 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
445 frontend->m_is_virtual = true;
446 frontend->m_init_options = *req->options;
449 boost::mutex::scoped_lock lock(m_mutex_backend_map);
450 BackendClassMap::const_iterator it;
451 it = m_backend_map.find(k);
452 if (it == m_backend_map.end())
454 BackendClassPtr b(new BackendClass(gdu->u.z3950,
458 m_backend_map[k] = b;
459 frontend->m_backend_class = b;
463 frontend->m_backend_class = it->second;
466 BackendClassPtr bc = frontend->m_backend_class;
467 BackendInstancePtr backend;
470 // we only need to get init response from "first" target in
471 // backend class - the assumption being that init response is
473 if (bc->m_init_response.get() == 0)
475 backend = bc->get_backend(package);
478 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
479 if (bc->m_init_response.get() == 0)
481 Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
482 *apdu->u.initResponse->result = 0;
483 package.response() = apdu;
484 package.session().close();
488 yazpp_1::GDU init_response = bc->m_init_response;
489 Z_GDU *response_gdu = init_response.get();
490 mp::util::transfer_referenceId(odr, gdu->u.z3950,
491 response_gdu->u.z3950);
493 Z_Options *server_options =
494 response_gdu->u.z3950->u.initResponse->options;
495 Z_Options *client_options = &frontend->m_init_options;
498 for (i = 0; i < 30; i++)
499 if (!ODR_MASK_GET(client_options, i))
500 ODR_MASK_CLEAR(server_options, i);
501 package.response() = init_response;
505 bc->release_backend(backend);
508 void yf::SessionShared::BackendSet::timestamp()
510 time(&m_time_last_use);
513 yf::SessionShared::BackendSet::BackendSet(
514 const std::string &result_set_id,
515 const Databases &databases,
516 const yazpp_1::Yaz_Z_Query &query) :
517 m_result_set_id(result_set_id),
518 m_databases(databases), m_result_set_size(0), m_query(query)
523 static int get_diagnostic(Z_DefaultDiagFormat *r)
525 return *r->condition;
528 bool yf::SessionShared::BackendSet::search(
529 mp::Package &frontend_package,
530 mp::Package &search_package,
531 const Z_APDU *frontend_apdu,
532 const BackendInstancePtr bp,
533 Z_Records **z_records)
536 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
537 Z_SearchRequest *req = apdu_req->u.searchRequest;
539 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
540 req->query = m_query.get_Z_Query();
542 req->num_databaseNames = m_databases.size();
543 req->databaseNames = (char**)
544 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
545 Databases::const_iterator it = m_databases.begin();
547 for (; it != m_databases.end(); it++)
548 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
550 if (frontend_apdu->which == Z_APDU_searchRequest)
551 req->preferredRecordSyntax =
552 frontend_apdu->u.searchRequest->preferredRecordSyntax;
554 search_package.request() = apdu_req;
556 search_package.move();
558 Z_GDU *gdu = search_package.response().get();
559 if (!search_package.session().is_closed()
560 && gdu && gdu->which == Z_GDU_Z3950
561 && gdu->u.z3950->which == Z_APDU_searchResponse)
563 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
564 *z_records = b_resp->records;
565 m_result_set_size = *b_resp->resultCount;
569 if (frontend_apdu->which == Z_APDU_searchRequest)
570 f_apdu = odr.create_searchResponse(
571 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
572 else if (frontend_apdu->which == Z_APDU_presentRequest)
573 f_apdu = odr.create_presentResponse(
574 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
576 f_apdu = odr.create_close(
577 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
578 frontend_package.response() = f_apdu;
582 void yf::SessionShared::Frontend::override_set(
583 BackendInstancePtr &found_backend,
584 std::string &result_set_id,
585 const Databases &databases,
586 bool out_of_sessions)
588 BackendClassPtr bc = m_backend_class;
589 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
593 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
594 for (; it != bc->m_backend_list.end(); it++)
596 if (!(*it)->m_in_use)
598 BackendSetList::iterator set_it = (*it)->m_sets.begin();
599 for (; set_it != (*it)->m_sets.end(); set_it++)
601 if ((max_sets > 1 || (*set_it)->m_databases == databases)
604 now < (*set_it)->m_time_last_use ||
605 now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
608 result_set_id = (*set_it)->m_result_set_id;
609 found_backend->m_sets.erase(set_it);
615 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
617 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
620 if (bc->m_named_result_sets)
622 result_set_id = boost::io::str(
623 boost::format("%1%") %
624 found_backend->m_result_set_sequence);
625 found_backend->m_result_set_sequence++;
628 result_set_id = "default";
634 void yf::SessionShared::Frontend::get_set(mp::Package &package,
635 const Z_APDU *apdu_req,
636 const Databases &databases,
637 yazpp_1::Yaz_Z_Query &query,
638 BackendInstancePtr &found_backend,
639 BackendSetPtr &found_set)
641 bool session_restarted = false;
644 std::string result_set_id;
645 bool out_of_sessions = false;
646 BackendClassPtr bc = m_backend_class;
648 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
650 if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
651 out_of_sessions = true;
653 if (m_p->m_optimize_search)
655 // look at each backend and see if we have a similar search
656 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
657 for (; it != bc->m_backend_list.end(); it++)
659 if (!(*it)->m_in_use)
661 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
662 for (; set_it != (*it)->m_sets.end(); set_it++)
664 if ((*set_it)->m_databases == databases
665 && query.match(&(*set_it)->m_query))
669 bc->use_backend(found_backend);
670 found_set->timestamp();
671 // found matching set. No need to search again
678 override_set(found_backend, result_set_id, databases, out_of_sessions);
680 bc->use_backend(found_backend);
684 // create a new backend set (and new set) if we're not out of sessions
685 if (!out_of_sessions)
686 found_backend = bc->create_backend(package);
692 const char *addinfo = 0;
695 addinfo = "session_shared: all sessions in use";
696 if (apdu_req->which == Z_APDU_searchRequest)
698 f_apdu = odr.create_searchResponse(
699 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
701 else if (apdu_req->which == Z_APDU_presentRequest)
703 f_apdu = odr.create_presentResponse(
704 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
708 f_apdu = odr.create_close(
709 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
711 package.response() = f_apdu;
714 if (bc->m_named_result_sets)
716 result_set_id = boost::io::str(
717 boost::format("%1%") % found_backend->m_result_set_sequence);
718 found_backend->m_result_set_sequence++;
721 result_set_id = "default";
723 found_backend->timestamp();
725 // we must search ...
726 BackendSetPtr new_set(new BackendSet(result_set_id,
728 Z_Records *z_records = 0;
730 Package search_package(found_backend->m_session, package.origin());
731 search_package.copy_filter(package);
733 if (!new_set->search(package, search_package,
734 apdu_req, found_backend, &z_records))
736 bc->remove_backend(found_backend);
737 return; // search error
743 if (z_records->which == Z_Records_NSD)
746 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
748 else if (z_records->which == Z_Records_multipleNSD)
750 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
753 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
754 Z_DiagRec_defaultFormat)
756 condition = get_diagnostic(
757 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
761 if (m_p->m_restart && !session_restarted &&
762 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
764 package.log("session_shared", YLOG_LOG, "restart");
765 bc->remove_backend(found_backend);
766 session_restarted = true;
767 found_backend.reset();
775 if (apdu_req->which == Z_APDU_searchRequest)
777 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
779 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
780 *f_resp->searchStatus = Z_SearchResponse_none;
781 f_resp->records = z_records;
782 package.response() = f_apdu;
784 if (apdu_req->which == Z_APDU_presentRequest)
786 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
788 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
789 f_resp->records = z_records;
790 package.response() = f_apdu;
792 bc->release_backend(found_backend);
793 return; // search error
796 if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
798 package.log("session_shared", YLOG_LOG, "restart");
799 bc->remove_backend(found_backend);
800 session_restarted = true;
801 found_backend.reset();
806 found_set->timestamp();
807 found_backend->m_sets.push_back(found_set);
810 void yf::SessionShared::Frontend::search(mp::Package &package,
813 Z_SearchRequest *req = apdu_req->u.searchRequest;
814 FrontendSets::iterator fset_it =
815 m_frontend_sets.find(req->resultSetName);
816 if (fset_it != m_frontend_sets.end())
818 // result set already exist
819 // if replace indicator is off: we return diagnostic if
820 // result set already exist.
821 if (*req->replaceIndicator == 0)
825 odr.create_searchResponse(
827 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
829 package.response() = apdu;
833 m_frontend_sets.erase(fset_it);
836 yazpp_1::Yaz_Z_Query query;
837 query.set_Z_Query(req->query);
840 for (i = 0; i < req->num_databaseNames; i++)
841 databases.push_back(req->databaseNames[i]);
843 BackendSetPtr found_set; // null
844 BackendInstancePtr found_backend; // null
846 get_set(package, apdu_req, databases, query, found_backend, found_set);
851 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
852 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
853 *f_resp->resultCount = found_set->m_result_set_size;
854 package.response() = f_apdu;
856 FrontendSetPtr fset(new FrontendSet(databases, query));
857 m_frontend_sets[req->resultSetName] = fset;
859 m_backend_class->release_backend(found_backend);
862 void yf::SessionShared::Frontend::present(mp::Package &package,
866 Z_PresentRequest *req = apdu_req->u.presentRequest;
868 FrontendSets::iterator fset_it =
869 m_frontend_sets.find(req->resultSetId);
871 if (fset_it == m_frontend_sets.end())
874 odr.create_presentResponse(
876 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
878 package.response() = apdu;
881 FrontendSetPtr fset = fset_it->second;
883 Databases databases = fset->get_databases();
884 yazpp_1::Yaz_Z_Query query = fset->get_query();
886 BackendClassPtr bc = m_backend_class;
887 BackendSetPtr found_set; // null
888 BackendInstancePtr found_backend;
890 get_set(package, apdu_req, databases, query, found_backend, found_set);
894 Z_NamePlusRecordList *npr_res = 0;
895 if (found_set->m_record_cache.lookup(odr, &npr_res,
896 *req->resultSetStartPoint,
897 *req->numberOfRecordsRequested,
898 req->preferredRecordSyntax,
899 req->recordComposition))
901 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
902 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
904 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
905 " records in cache %p",
906 *req->resultSetStartPoint,
907 *req->numberOfRecordsRequested,
908 &found_set->m_record_cache);
910 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
911 *f_resp->nextResultSetPosition =
912 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
913 // f_resp->presentStatus assumed OK.
914 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
915 f_resp->records->which = Z_Records_DBOSD;
916 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
917 package.response() = f_apdu_res;
918 bc->release_backend(found_backend);
922 found_backend->timestamp();
924 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
925 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
926 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
927 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
928 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
929 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
930 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
931 p_req->recordComposition = req->recordComposition;
933 Package present_package(found_backend->m_session, package.origin());
934 present_package.copy_filter(package);
936 present_package.request() = p_apdu;
938 present_package.move();
940 Z_GDU *gdu = present_package.response().get();
941 if (!present_package.session().is_closed()
942 && gdu && gdu->which == Z_GDU_Z3950
943 && gdu->u.z3950->which == Z_APDU_presentResponse)
945 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
946 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
947 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
949 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
950 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
951 f_resp->presentStatus= b_resp->presentStatus;
952 f_resp->records = b_resp->records;
953 f_resp->otherInfo = b_resp->otherInfo;
954 package.response() = f_apdu_res;
956 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
958 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
959 " records to cache %p",
960 *req->resultSetStartPoint,
961 *f_resp->numberOfRecordsReturned,
962 &found_set->m_record_cache);
963 found_set->m_record_cache.add(
965 b_resp->records->u.databaseOrSurDiagnostics,
966 *req->resultSetStartPoint,
967 *f_resp->numberOfRecordsReturned);
969 bc->release_backend(found_backend);
973 bc->remove_backend(found_backend);
975 odr.create_presentResponse(
976 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
977 package.response() = f_apdu_res;
981 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
984 BackendClassPtr bc = m_backend_class;
985 BackendInstancePtr backend = bc->get_backend(frontend_package);
989 Z_APDU *apdu = odr.create_scanResponse(
990 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
991 frontend_package.response() = apdu;
995 Package scan_package(backend->m_session, frontend_package.origin());
996 backend->timestamp();
997 scan_package.copy_filter(frontend_package);
998 scan_package.request() = apdu_req;
1000 frontend_package.response() = scan_package.response();
1001 if (scan_package.session().is_closed())
1003 frontend_package.session().close();
1004 bc->remove_backend(backend);
1007 bc->release_backend(backend);
1011 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1015 void yf::SessionShared::Worker::operator() (void)
1020 void yf::SessionShared::BackendClass::expire_class()
1024 boost::mutex::scoped_lock lock(m_mutex_backend_class);
1025 BackendInstanceList::iterator bit = m_backend_list.begin();
1026 while (bit != m_backend_list.end())
1028 time_t last_use = (*bit)->m_time_last_use;
1030 if ((*bit)->m_in_use)
1034 else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1037 (*bit)->m_close_package->response() = odr.create_close(
1038 0, Z_Close_lackOfActivity, 0);
1039 (*bit)->m_close_package->session().close();
1040 (*bit)->m_close_package->move();
1042 bit = m_backend_list.erase(bit);
1051 void yf::SessionShared::Rep::expire()
1056 boost::xtime_get(&xt, boost::TIME_UTC);
1057 xt.sec += m_session_ttl / 3;
1058 boost::thread::sleep(xt);
1060 BackendClassMap::const_iterator b_it = m_backend_map.begin();
1061 for (; b_it != m_backend_map.end(); b_it++)
1062 b_it->second->expire_class();
1066 yf::SessionShared::Rep::Rep()
1068 m_resultset_ttl = 30;
1069 m_resultset_max = 10;
1071 m_optimize_search = true;
1073 m_session_max = 100;
1076 void yf::SessionShared::Rep::start()
1078 yf::SessionShared::Worker w(this);
1079 m_thrds.add_thread(new boost::thread(w));
1082 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1086 yf::SessionShared::~SessionShared() {
1089 void yf::SessionShared::start() const
1094 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1098 yf::SessionShared::Frontend::~Frontend()
1102 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1104 boost::mutex::scoped_lock lock(m_mutex);
1106 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1110 it = m_clients.find(package.session());
1111 if (it == m_clients.end())
1114 if (!it->second->m_in_use)
1116 it->second->m_in_use = true;
1119 m_cond_session_ready.wait(lock);
1121 FrontendPtr f(new Frontend(this));
1122 m_clients[package.session()] = f;
1127 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1129 boost::mutex::scoped_lock lock(m_mutex);
1130 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1132 it = m_clients.find(package.session());
1133 if (it != m_clients.end())
1135 if (package.session().is_closed())
1137 m_clients.erase(it);
1141 it->second->m_in_use = false;
1143 m_cond_session_ready.notify_all();
1148 void yf::SessionShared::process(mp::Package &package) const
1150 FrontendPtr f = m_p->get_frontend(package);
1152 Z_GDU *gdu = package.request().get();
1154 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1155 Z_APDU_initRequest && !f->m_is_virtual)
1157 m_p->init(package, gdu, f);
1159 else if (!f->m_is_virtual)
1161 else if (gdu && gdu->which == Z_GDU_Z3950)
1163 Z_APDU *apdu = gdu->u.z3950;
1164 if (apdu->which == Z_APDU_initRequest)
1168 package.response() = odr.create_close(
1170 Z_Close_protocolError,
1173 package.session().close();
1175 else if (apdu->which == Z_APDU_close)
1179 package.response() = odr.create_close(
1181 Z_Close_peerAbort, "received close from client");
1182 package.session().close();
1184 else if (apdu->which == Z_APDU_searchRequest)
1186 f->search(package, apdu);
1188 else if (apdu->which == Z_APDU_presentRequest)
1190 f->present(package, apdu);
1192 else if (apdu->which == Z_APDU_scanRequest)
1194 f->scan(package, apdu);
1200 package.response() = odr.create_close(
1201 apdu, Z_Close_protocolError,
1202 "unsupported APDU in filter_session_shared");
1204 package.session().close();
1207 m_p->release_frontend(package);
1210 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1213 for (ptr = ptr->children; ptr; ptr = ptr->next)
1215 if (ptr->type != XML_ELEMENT_NODE)
1217 if (!strcmp((const char *) ptr->name, "resultset"))
1219 const struct _xmlAttr *attr;
1220 for (attr = ptr->properties; attr; attr = attr->next)
1222 if (!strcmp((const char *) attr->name, "ttl"))
1223 m_p->m_resultset_ttl =
1224 mp::xml::get_int(attr->children, 30);
1225 else if (!strcmp((const char *) attr->name, "max"))
1227 m_p->m_resultset_max =
1228 mp::xml::get_int(attr->children, 10);
1230 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1232 m_p->m_optimize_search =
1233 mp::xml::get_bool(attr->children, true);
1235 else if (!strcmp((const char *) attr->name, "restart"))
1237 m_p->m_restart = mp::xml::get_bool(attr->children, true);
1240 throw mp::filter::FilterException(
1241 "Bad attribute " + std::string((const char *)
1245 else if (!strcmp((const char *) ptr->name, "session"))
1247 const struct _xmlAttr *attr;
1248 for (attr = ptr->properties; attr; attr = attr->next)
1250 if (!strcmp((const char *) attr->name, "ttl"))
1251 m_p->m_session_ttl =
1252 mp::xml::get_int(attr->children, 90);
1253 else if (!strcmp((const char *) attr->name, "max"))
1254 m_p->m_session_max =
1255 mp::xml::get_int(attr->children, 100);
1257 throw mp::filter::FilterException(
1258 "Bad attribute " + std::string((const char *)
1264 throw mp::filter::FilterException("Bad element "
1265 + std::string((const char *)
1271 static mp::filter::Base* filter_creator()
1273 return new mp::filter::SessionShared;
1277 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1287 * c-file-style: "Stroustrup"
1288 * indent-tabs-mode: nil
1290 * vim: shiftwidth=4 tabstop=8 expandtab