1 /* This file is part of Metaproxy.
2 Copyright (C) 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"
34 #include <yaz/copy_types.h>
37 #include <yaz/otherinfo.h>
38 #include <yaz/diagbib1.h>
39 #include <yazpp/z-query.h>
40 #include <yazpp/record-cache.h>
46 namespace mp = metaproxy_1;
47 namespace yf = metaproxy_1::filter;
49 namespace metaproxy_1 {
52 // key for session.. We'll only share sessions with same InitKey
53 class SessionShared::InitKey {
55 bool operator < (const SessionShared::InitKey &k) const;
56 InitKey(Z_InitRequest *req);
57 InitKey(const InitKey &);
60 char *m_idAuthentication_buf;
61 int m_idAuthentication_size;
62 char *m_otherInfo_buf;
66 // worker thread .. for expiry of sessions
67 class SessionShared::Worker {
69 Worker(SessionShared::Rep *rep);
70 void operator() (void);
72 SessionShared::Rep *m_p;
75 class SessionShared::BackendSet {
77 std::string m_result_set_id;
78 Databases m_databases;
79 Odr_int m_result_set_size;
80 yazpp_1::Yaz_Z_Query m_query;
81 time_t m_time_last_use;
83 yazpp_1::RecordCache m_record_cache;
85 Z_OtherInformation *additionalSearchInfoRequest;
86 Z_OtherInformation *additionalSearchInfoResponse;
87 NMEM mem_additionalSearchInfo;
89 const std::string &result_set_id,
90 const Databases &databases,
91 const yazpp_1::Yaz_Z_Query &query,
92 Z_OtherInformation *additionalSearchInfoRequest);
95 Package &frontend_package,
96 Package &search_package,
97 const Z_APDU *apdu_req,
98 const BackendInstancePtr bp,
99 Z_Records **z_records);
101 // backend connection instance
102 class SessionShared::BackendInstance {
104 friend class BackendClass;
105 friend class BackendSet;
107 mp::Session m_session;
108 BackendSetList m_sets;
111 int m_result_set_sequence;
112 time_t m_time_last_use;
113 mp::Package * m_close_package;
117 // backends of some class (all with same InitKey)
118 class SessionShared::BackendClass : boost::noncopyable {
120 friend struct Frontend;
121 bool m_named_result_sets;
122 BackendInstanceList m_backend_list;
123 BackendInstancePtr create_backend(const Package &package);
124 void remove_backend(BackendInstancePtr b);
125 BackendInstancePtr get_backend(const Package &package);
126 void use_backend(BackendInstancePtr b);
127 void release_backend(BackendInstancePtr b);
128 bool expire_instances();
129 yazpp_1::GDU m_init_request;
130 yazpp_1::GDU m_init_response;
131 boost::mutex m_mutex_backend_class;
132 boost::condition m_cond_set_ready;
134 time_t m_backend_set_ttl;
135 time_t m_backend_expiry_ttl;
136 size_t m_backend_set_max;
137 Odr_int m_preferredMessageSize;
138 Odr_int m_maximumRecordSize;
143 BackendClass(const yazpp_1::GDU &init_request,
147 Odr_int preferredRecordSize,
148 Odr_int maximumRecordSize);
151 // frontend result set
152 class SessionShared::FrontendSet {
153 Databases m_databases;
154 yazpp_1::Yaz_Z_Query m_query;
156 const Databases &get_databases();
157 const yazpp_1::Yaz_Z_Query &get_query();
159 const Databases &databases,
160 const yazpp_1::Yaz_Z_Query &query);
164 struct SessionShared::Frontend {
169 Z_Options m_init_options;
170 void search(Package &package, Z_APDU *apdu);
171 void present(Package &package, Z_APDU *apdu);
172 void scan(Package &package, Z_APDU *apdu);
174 int result_set_ref(ODR o,
175 const Databases &databases,
176 Z_RPNStructure *s, std::string &rset);
177 void get_set(mp::Package &package,
178 const Z_APDU *apdu_req,
179 const Databases &databases,
180 yazpp_1::Yaz_Z_Query &query,
181 BackendInstancePtr &found_backend,
182 BackendSetPtr &found_set);
183 void override_set(BackendInstancePtr &found_backend,
184 std::string &result_set_id,
185 const Databases &databases,
186 bool out_of_sessions);
189 BackendClassPtr m_backend_class;
190 FrontendSets m_frontend_sets;
193 class SessionShared::Rep {
194 friend class SessionShared;
195 friend struct Frontend;
197 FrontendPtr get_frontend(Package &package);
198 void release_frontend(Package &package);
204 void expire_classes();
206 void init(Package &package, const Z_GDU *gdu,
207 FrontendPtr frontend);
209 boost::mutex m_mutex;
210 boost::condition m_cond_session_ready;
211 boost::condition m_cond_expire_ready;
212 std::map<mp::Session, FrontendPtr> m_clients;
214 BackendClassMap m_backend_map;
215 boost::mutex m_mutex_backend_map;
216 boost::thread_group m_thrds;
220 bool m_optimize_search;
223 Odr_int m_preferredMessageSize;
224 Odr_int m_maximumRecordSize;
230 yf::SessionShared::FrontendSet::FrontendSet(
231 const Databases &databases,
232 const yazpp_1::Yaz_Z_Query &query)
233 : m_databases(databases), m_query(query)
237 const yf::SessionShared::Databases &
238 yf::SessionShared::FrontendSet::get_databases()
243 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
248 yf::SessionShared::InitKey::InitKey(const InitKey &k)
250 m_odr = odr_createmem(ODR_ENCODE);
252 m_idAuthentication_size = k.m_idAuthentication_size;
253 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
254 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
255 m_idAuthentication_size);
257 m_otherInfo_size = k.m_otherInfo_size;
258 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
259 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
263 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
265 m_odr = odr_createmem(ODR_ENCODE);
267 Z_IdAuthentication *t = req->idAuthentication;
268 z_IdAuthentication(m_odr, &t, 1, 0);
269 m_idAuthentication_buf =
270 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
272 Z_OtherInformation *o = req->otherInfo;
273 z_OtherInformation(m_odr, &o, 1, 0);
274 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
277 yf::SessionShared::InitKey::~InitKey()
282 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
286 c = mp::util::memcmp2(
287 (void*) m_idAuthentication_buf, m_idAuthentication_size,
288 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
294 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
295 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
303 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
305 boost::mutex::scoped_lock lock(m_mutex_backend_class);
306 m_cond_set_ready.notify_all();
311 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
314 boost::mutex::scoped_lock lock(m_mutex_backend_class);
315 BackendInstanceList::iterator it = m_backend_list.begin();
318 if (it == m_backend_list.end())
322 it = m_backend_list.erase(it);
329 b->m_close_package->response() = odr.create_close(
330 0, Z_Close_lackOfActivity, 0);
331 b->m_close_package->session().close();
332 b->m_close_package->move();
336 yf::SessionShared::BackendInstancePtr
337 yf::SessionShared::BackendClass::get_backend(
338 const mp::Package &frontend_package)
341 boost::mutex::scoped_lock lock(m_mutex_backend_class);
343 BackendInstanceList::const_iterator it = m_backend_list.begin();
345 BackendInstancePtr backend1; // null
347 for (; it != m_backend_list.end(); it++)
349 if (!(*it)->m_in_use)
352 || (*it)->m_sequence_this < backend1->m_sequence_this)
358 use_backend(backend1);
362 return create_backend(frontend_package);
365 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
367 backend->m_in_use = true;
368 backend->m_sequence_this = m_sequence_top++;
371 void yf::SessionShared::BackendInstance::timestamp()
374 time(&m_time_last_use);
377 yf::SessionShared::BackendInstance::~BackendInstance()
382 m_close_package->response() = odr.create_close(
383 0, Z_Close_lackOfActivity, 0);
384 m_close_package->session().close();
385 m_close_package->move();
387 delete m_close_package;
390 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
391 const mp::Package &frontend_package)
393 BackendInstancePtr null;
395 boost::mutex::scoped_lock lock(m_mutex_backend_class);
396 if (m_no_failed && !m_no_succeeded)
402 BackendInstancePtr bp(new BackendInstance);
403 bp->m_close_package =
404 new mp::Package(bp->m_session, frontend_package.origin());
405 bp->m_close_package->copy_filter(frontend_package);
407 Package init_package(bp->m_session, frontend_package.origin());
409 init_package.copy_filter(frontend_package);
411 yazpp_1::GDU actual_init_request = m_init_request;
412 Z_GDU *init_pdu = actual_init_request.get();
414 assert(init_pdu->which == Z_GDU_Z3950);
415 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
417 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
418 ODR_MASK_ZERO(req->options);
420 ODR_MASK_SET(req->options, Z_Options_search);
421 ODR_MASK_SET(req->options, Z_Options_present);
422 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
423 ODR_MASK_SET(req->options, Z_Options_scan);
425 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
426 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
427 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
429 if (m_preferredMessageSize)
430 *req->preferredMessageSize = m_preferredMessageSize;
431 if (m_maximumRecordSize)
432 *req->maximumRecordSize = m_maximumRecordSize;
434 init_package.request() = init_pdu;
437 boost::mutex::scoped_lock lock(m_mutex_backend_class);
443 boost::mutex::scoped_lock lock(m_mutex_backend_class);
445 m_named_result_sets = false;
446 Z_GDU *gdu = init_package.response().get();
448 if (gdu && gdu->which == Z_GDU_Z3950
449 && gdu->u.z3950->which == Z_APDU_initResponse)
451 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
452 m_init_response = gdu->u.z3950;
453 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
455 m_named_result_sets = true;
457 if (*gdu->u.z3950->u.initResponse->result
458 && !init_package.session().is_closed())
461 time(&bp->m_time_last_use);
462 bp->m_sequence_this = 0;
463 bp->m_result_set_sequence = 0;
464 m_backend_list.push_back(bp);
471 yazpp_1::GDU empty_gdu;
472 m_init_response = empty_gdu;
475 if (!init_package.session().is_closed())
477 init_package.copy_filter(frontend_package);
478 init_package.session().close();
486 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
490 Odr_int preferredMessageSize,
491 Odr_int maximumRecordSize)
492 : m_named_result_sets(false), m_init_request(init_request),
493 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
494 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
495 m_preferredMessageSize(preferredMessageSize),
496 m_maximumRecordSize(maximumRecordSize),
497 m_no_failed(0), m_no_succeeded(0), m_no_init(0)
500 yf::SessionShared::BackendClass::~BackendClass()
503 void yf::SessionShared::Rep::stat()
506 int no_instances = 0;
507 BackendClassMap::const_iterator it;
509 boost::mutex::scoped_lock lock(m_mutex_backend_map);
510 for (it = m_backend_map.begin(); it != m_backend_map.end(); it++)
512 BackendClassPtr bc = it->second;
514 BackendInstanceList::iterator bit = bc->m_backend_list.begin();
515 for (; bit != bc->m_backend_list.end(); bit++)
521 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
522 FrontendPtr frontend)
524 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
526 frontend->m_is_virtual = true;
527 frontend->m_init_options = *req->options;
530 boost::mutex::scoped_lock lock(m_mutex_backend_map);
531 BackendClassMap::const_iterator it;
532 it = m_backend_map.find(k);
533 if (it == m_backend_map.end())
535 BackendClassPtr b(new BackendClass(gdu->u.z3950,
539 m_preferredMessageSize,
540 m_maximumRecordSize));
541 m_backend_map[k] = b;
542 frontend->m_backend_class = b;
546 frontend->m_backend_class = it->second;
549 BackendClassPtr bc = frontend->m_backend_class;
552 // we only need to get init response from "first" target in
553 // backend class - the assumption being that init response is
555 bool create_first_one = false;
557 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
558 if (!bc->m_no_failed && !bc->m_no_succeeded && !bc->m_no_init)
559 create_first_one = true;
562 // first for first one to finish
563 while (!bc->m_no_failed && !bc->m_no_succeeded && bc->m_no_init)
564 bc->m_cond_set_ready.wait(bc->m_mutex_backend_class);
567 if (create_first_one)
569 BackendInstancePtr backend = bc->create_backend(package);
571 bc->release_backend(backend);
573 yazpp_1::GDU init_response;
575 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
577 init_response = bc->m_init_response;
580 if (init_response.get())
582 Z_GDU *response_gdu = init_response.get();
583 mp::util::transfer_referenceId(odr, gdu->u.z3950,
584 response_gdu->u.z3950);
585 Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
586 Z_Options *server_options = init_res->options;
587 Z_Options *client_options = &frontend->m_init_options;
589 for (i = 0; i < 30; i++)
590 if (!ODR_MASK_GET(client_options, i))
591 ODR_MASK_CLEAR(server_options, i);
593 if (!m_preferredMessageSize ||
594 *init_res->preferredMessageSize > *req->preferredMessageSize)
595 *init_res->preferredMessageSize = *req->preferredMessageSize;
597 if (!m_maximumRecordSize ||
598 *init_res->maximumRecordSize > *req->maximumRecordSize)
599 *init_res->maximumRecordSize = *req->maximumRecordSize;
601 package.response() = init_response;
602 if (!*response_gdu->u.z3950->u.initResponse->result)
603 package.session().close();
608 odr.create_initResponse(
609 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
610 "session_shared: target closed connection during init");
611 *apdu->u.initResponse->result = 0;
612 package.response() = apdu;
613 package.session().close();
617 void yf::SessionShared::BackendSet::timestamp()
619 time(&m_time_last_use);
622 yf::SessionShared::BackendSet::BackendSet(
623 const std::string &result_set_id,
624 const Databases &databases,
625 const yazpp_1::Yaz_Z_Query &query,
626 Z_OtherInformation *additionalSearchInfo) :
627 m_result_set_id(result_set_id),
628 m_databases(databases), m_result_set_size(0), m_query(query)
631 mem_additionalSearchInfo = nmem_create();
632 additionalSearchInfoResponse = 0;
633 additionalSearchInfoRequest =
634 yaz_clone_z_OtherInformation(additionalSearchInfo,
635 mem_additionalSearchInfo);
638 yf::SessionShared::BackendSet::~BackendSet()
640 nmem_destroy(mem_additionalSearchInfo);
643 static int get_diagnostic(Z_DefaultDiagFormat *r)
645 return *r->condition;
648 bool yf::SessionShared::BackendSet::search(
649 mp::Package &frontend_package,
650 mp::Package &search_package,
651 const Z_APDU *frontend_apdu,
652 const BackendInstancePtr bp,
653 Z_Records **z_records)
656 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
657 Z_SearchRequest *req = apdu_req->u.searchRequest;
659 req->additionalSearchInfo = additionalSearchInfoRequest;
660 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
661 req->query = m_query.get_Z_Query();
663 req->num_databaseNames = m_databases.size();
664 req->databaseNames = (char**)
665 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
666 Databases::const_iterator it = m_databases.begin();
668 for (; it != m_databases.end(); it++)
669 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
671 if (frontend_apdu->which == Z_APDU_searchRequest)
672 req->preferredRecordSyntax =
673 frontend_apdu->u.searchRequest->preferredRecordSyntax;
675 search_package.request() = apdu_req;
677 search_package.move();
679 Z_GDU *gdu = search_package.response().get();
680 if (!search_package.session().is_closed()
681 && gdu && gdu->which == Z_GDU_Z3950
682 && gdu->u.z3950->which == Z_APDU_searchResponse)
684 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
685 *z_records = b_resp->records;
686 m_result_set_size = *b_resp->resultCount;
688 additionalSearchInfoResponse = yaz_clone_z_OtherInformation(
689 b_resp->additionalSearchInfo, mem_additionalSearchInfo);
693 const char *addinfo = "session_shared: "
694 "target closed connection during search";
695 if (frontend_apdu->which == Z_APDU_searchRequest)
696 f_apdu = odr.create_searchResponse(
697 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
698 else if (frontend_apdu->which == Z_APDU_presentRequest)
699 f_apdu = odr.create_presentResponse(
700 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
702 f_apdu = odr.create_close(
703 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
704 frontend_package.response() = f_apdu;
708 void yf::SessionShared::Frontend::override_set(
709 BackendInstancePtr &found_backend,
710 std::string &result_set_id,
711 const Databases &databases,
712 bool out_of_sessions)
714 BackendClassPtr bc = m_backend_class;
715 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
719 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
720 for (; it != bc->m_backend_list.end(); it++)
722 if (!(*it)->m_in_use)
724 BackendSetList::iterator set_it = (*it)->m_sets.begin();
725 for (; set_it != (*it)->m_sets.end(); set_it++)
727 if ((max_sets > 1 || (*set_it)->m_databases == databases)
730 now < (*set_it)->m_time_last_use ||
731 now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
734 result_set_id = (*set_it)->m_result_set_id;
735 found_backend->m_sets.erase(set_it);
741 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
743 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
746 if (bc->m_named_result_sets)
748 result_set_id = boost::io::str(
749 boost::format("%1%") %
750 found_backend->m_result_set_sequence);
751 found_backend->m_result_set_sequence++;
754 result_set_id = "default";
760 void yf::SessionShared::Frontend::get_set(mp::Package &package,
761 const Z_APDU *apdu_req,
762 const Databases &databases,
763 yazpp_1::Yaz_Z_Query &query,
764 BackendInstancePtr &found_backend,
765 BackendSetPtr &found_set)
767 bool session_restarted = false;
768 Z_OtherInformation *additionalSearchInfo = 0;
770 if (apdu_req->which == Z_APDU_searchRequest)
771 additionalSearchInfo = apdu_req->u.searchRequest->additionalSearchInfo;
774 std::string result_set_id;
775 bool out_of_sessions = false;
776 BackendClassPtr bc = m_backend_class;
778 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
780 if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
781 out_of_sessions = true;
783 if (m_p->m_optimize_search)
785 // look at each backend and see if we have a similar search
786 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
787 while (it != bc->m_backend_list.end())
789 bool restart = false;
790 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
791 for (; set_it != (*it)->m_sets.end(); set_it++)
793 // for real present request we don't care
794 // if additionalSearchInfo matches: same records
795 if ((*set_it)->m_databases == databases
796 && query.match(&(*set_it)->m_query)
797 && (apdu_req->which != Z_APDU_searchRequest ||
798 yaz_compare_z_OtherInformation(
799 additionalSearchInfo,
800 (*set_it)->additionalSearchInfoRequest)))
804 bc->m_cond_set_ready.wait(
805 bc->m_mutex_backend_class);
813 bc->use_backend(found_backend);
814 // found matching set. No need to search again
820 it = bc->m_backend_list.begin();
825 override_set(found_backend, result_set_id, databases, out_of_sessions);
827 bc->use_backend(found_backend);
831 // create a new backend set (and new set) if we're not out of sessions
832 if (!out_of_sessions)
833 found_backend = bc->create_backend(package);
839 const char *addinfo = 0;
842 addinfo = "session_shared: all sessions in use";
844 addinfo = "session_shared: could not create backend";
845 if (apdu_req->which == Z_APDU_searchRequest)
847 f_apdu = odr.create_searchResponse(
848 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
850 else if (apdu_req->which == Z_APDU_presentRequest)
852 f_apdu = odr.create_presentResponse(
853 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
857 f_apdu = odr.create_close(
858 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
860 package.response() = f_apdu;
863 if (bc->m_named_result_sets)
865 result_set_id = boost::io::str(
866 boost::format("%1%") % found_backend->m_result_set_sequence);
867 found_backend->m_result_set_sequence++;
870 result_set_id = "default";
872 found_backend->timestamp();
874 // we must search ...
875 BackendSetPtr new_set(new BackendSet(result_set_id,
877 additionalSearchInfo));
880 found_set->timestamp();
881 found_backend->m_sets.push_back(found_set);
883 Z_Records *z_records = 0;
885 Package search_package(found_backend->m_session, package.origin());
886 search_package.copy_filter(package);
888 if (!new_set->search(package, search_package,
889 apdu_req, found_backend, &z_records))
891 bc->remove_backend(found_backend);
892 return; // search error
898 if (z_records->which == Z_Records_NSD)
901 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
903 else if (z_records->which == Z_Records_multipleNSD)
905 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
908 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
909 Z_DiagRec_defaultFormat)
911 condition = get_diagnostic(
912 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
916 if (m_p->m_restart && !session_restarted &&
917 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
919 package.log("session_shared", YLOG_LOG, "restart");
920 bc->remove_backend(found_backend);
921 session_restarted = true;
922 found_backend.reset();
930 if (apdu_req->which == Z_APDU_searchRequest)
932 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
934 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
935 *f_resp->searchStatus = Z_SearchResponse_none;
936 f_resp->records = z_records;
937 package.response() = f_apdu;
939 if (apdu_req->which == Z_APDU_presentRequest)
941 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
943 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
944 f_resp->records = z_records;
945 package.response() = f_apdu;
947 bc->release_backend(found_backend);
948 return; // search error
951 if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
953 package.log("session_shared", YLOG_LOG, "restart");
954 bc->remove_backend(found_backend);
955 session_restarted = true;
956 found_backend.reset();
962 found_set->timestamp();
963 found_backend->m_sets.push_back(found_set);
967 int yf::SessionShared::Frontend::result_set_ref(ODR o,
968 const Databases &databases,
975 case Z_RPNStructure_simple:
976 if (s->u.simple->which == Z_Operand_resultSetId)
978 const char *id = s->u.simple->u.resultSetId;
981 FrontendSets::iterator fset_it = m_frontend_sets.find(id);
982 if (fset_it == m_frontend_sets.end())
984 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
986 else if (fset_it->second->get_databases() != databases)
988 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
992 yazpp_1::Yaz_Z_Query query = fset_it->second->get_query();
993 Z_Query *q = yaz_copy_Z_Query(query.get_Z_Query(), o);
994 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
996 s->which = q->u.type_1->RPNStructure->which;
997 s->u.simple = q->u.type_1->RPNStructure->u.simple;
1001 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
1006 case Z_RPNStructure_complex:
1007 ret = result_set_ref(o, databases, s->u.complex->s1, rset);
1009 ret = result_set_ref(o, databases, s->u.complex->s2, rset);
1015 void yf::SessionShared::Frontend::search(mp::Package &package,
1018 Z_SearchRequest *req = apdu_req->u.searchRequest;
1019 FrontendSets::iterator fset_it =
1020 m_frontend_sets.find(req->resultSetName);
1021 if (fset_it != m_frontend_sets.end())
1023 // result set already exist
1024 // if replace indicator is off: we return diagnostic if
1025 // result set already exist.
1026 if (*req->replaceIndicator == 0)
1030 odr.create_searchResponse(
1032 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
1034 package.response() = apdu;
1037 m_frontend_sets.erase(fset_it);
1040 Databases databases;
1042 for (i = 0; i < req->num_databaseNames; i++)
1043 databases.push_back(req->databaseNames[i]);
1046 yazpp_1::Yaz_Z_Query query;
1047 query.set_Z_Query(req->query);
1049 Z_Query *q = query.get_Z_Query();
1050 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
1054 int diag = result_set_ref(odr, databases, q->u.type_1->RPNStructure,
1059 odr.create_searchResponse(
1063 package.response() = apdu;
1066 query.set_Z_Query(q);
1069 BackendSetPtr found_set; // null
1070 BackendInstancePtr found_backend; // null
1072 get_set(package, apdu_req, databases, query, found_backend, found_set);
1077 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
1078 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
1079 *f_resp->resultCount = found_set->m_result_set_size;
1080 f_resp->additionalSearchInfo = found_set->additionalSearchInfoResponse;
1081 package.response() = f_apdu;
1083 FrontendSetPtr fset(new FrontendSet(databases, query));
1084 m_frontend_sets[req->resultSetName] = fset;
1086 m_backend_class->release_backend(found_backend);
1089 void yf::SessionShared::Frontend::present(mp::Package &package,
1093 Z_PresentRequest *req = apdu_req->u.presentRequest;
1095 FrontendSets::iterator fset_it =
1096 m_frontend_sets.find(req->resultSetId);
1098 if (fset_it == m_frontend_sets.end())
1101 odr.create_presentResponse(
1103 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
1105 package.response() = apdu;
1108 FrontendSetPtr fset = fset_it->second;
1110 Databases databases = fset->get_databases();
1111 yazpp_1::Yaz_Z_Query query = fset->get_query();
1113 BackendClassPtr bc = m_backend_class;
1114 BackendSetPtr found_set; // null
1115 BackendInstancePtr found_backend;
1117 get_set(package, apdu_req, databases, query, found_backend, found_set);
1121 Z_NamePlusRecordList *npr_res = 0;
1122 // record_cache.lookup types are int's. Avoid non-fitting values
1123 if (*req->resultSetStartPoint > 0
1124 && *req->resultSetStartPoint < INT_MAX
1125 && *req->numberOfRecordsRequested > 0
1126 && *req->numberOfRecordsRequested < INT_MAX
1127 && found_set->m_record_cache.lookup(odr, &npr_res,
1128 *req->resultSetStartPoint,
1129 *req->numberOfRecordsRequested,
1130 req->preferredRecordSyntax,
1131 req->recordComposition))
1133 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1134 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1136 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1137 " records in cache %p",
1138 *req->resultSetStartPoint,
1139 *req->numberOfRecordsRequested,
1140 &found_set->m_record_cache);
1142 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
1143 *f_resp->nextResultSetPosition =
1144 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
1145 // f_resp->presentStatus assumed OK.
1146 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
1147 f_resp->records->which = Z_Records_DBOSD;
1148 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
1149 package.response() = f_apdu_res;
1150 bc->release_backend(found_backend);
1154 found_backend->timestamp();
1156 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
1157 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
1158 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1159 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
1160 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
1161 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
1162 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1163 p_req->recordComposition = req->recordComposition;
1165 Package present_package(found_backend->m_session, package.origin());
1166 present_package.copy_filter(package);
1168 present_package.request() = p_apdu;
1170 present_package.move();
1172 Z_GDU *gdu = present_package.response().get();
1173 if (!present_package.session().is_closed()
1174 && gdu && gdu->which == Z_GDU_Z3950
1175 && gdu->u.z3950->which == Z_APDU_presentResponse)
1177 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
1178 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1179 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1181 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
1182 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
1183 f_resp->presentStatus= b_resp->presentStatus;
1184 f_resp->records = b_resp->records;
1185 f_resp->otherInfo = b_resp->otherInfo;
1186 package.response() = f_apdu_res;
1188 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
1190 Z_NamePlusRecordList *npr =
1191 b_resp->records->u.databaseOrSurDiagnostics;
1192 // record_cache.add types are int's. Avoid non-fitting values
1193 if (*req->resultSetStartPoint > 0
1194 && npr->num_records + *req->resultSetStartPoint < INT_MAX)
1197 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1198 " records to cache %p",
1199 *req->resultSetStartPoint,
1200 *f_resp->numberOfRecordsReturned,
1201 &found_set->m_record_cache);
1203 found_set->m_record_cache.add(
1204 odr, npr, *req->resultSetStartPoint,
1205 p_req->recordComposition);
1208 bc->release_backend(found_backend);
1212 bc->remove_backend(found_backend);
1213 Z_APDU *f_apdu_res =
1214 odr.create_presentResponse(
1215 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1216 "session_shared: target closed connection during present");
1217 package.response() = f_apdu_res;
1221 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1224 BackendClassPtr bc = m_backend_class;
1225 BackendInstancePtr backend = bc->get_backend(frontend_package);
1229 Z_APDU *apdu = odr.create_scanResponse(
1230 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1231 "session_shared: could not create backend");
1232 frontend_package.response() = apdu;
1236 Package scan_package(backend->m_session, frontend_package.origin());
1237 backend->timestamp();
1238 scan_package.copy_filter(frontend_package);
1239 scan_package.request() = apdu_req;
1240 scan_package.move();
1241 frontend_package.response() = scan_package.response();
1242 if (scan_package.session().is_closed())
1244 frontend_package.session().close();
1245 bc->remove_backend(backend);
1248 bc->release_backend(backend);
1252 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1256 void yf::SessionShared::Worker::operator() (void)
1261 bool yf::SessionShared::BackendClass::expire_instances()
1265 boost::mutex::scoped_lock lock(m_mutex_backend_class);
1266 BackendInstanceList::iterator bit = m_backend_list.begin();
1267 while (bit != m_backend_list.end())
1269 time_t last_use = (*bit)->m_time_last_use;
1271 if ((*bit)->m_in_use)
1275 else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1277 bit = m_backend_list.erase(bit);
1284 if (m_backend_list.empty())
1289 void yf::SessionShared::Rep::expire_classes()
1291 boost::mutex::scoped_lock lock(m_mutex_backend_map);
1292 BackendClassMap::iterator b_it = m_backend_map.begin();
1293 while (b_it != m_backend_map.end())
1295 if (b_it->second->expire_instances())
1297 m_backend_map.erase(b_it);
1298 b_it = m_backend_map.begin();
1305 void yf::SessionShared::Rep::expire()
1310 boost::xtime_get(&xt,
1311 #if BOOST_VERSION >= 105000
1317 xt.sec += m_session_ttl;
1319 boost::mutex::scoped_lock lock(m_mutex);
1320 m_cond_expire_ready.timed_wait(lock, xt);
1329 yf::SessionShared::Rep::Rep()
1331 m_resultset_ttl = 30;
1332 m_resultset_max = 10;
1334 m_optimize_search = true;
1336 m_session_max = 100;
1337 m_preferredMessageSize = 0;
1338 m_maximumRecordSize = 0;
1342 yf::SessionShared::Rep::~Rep()
1345 boost::mutex::scoped_lock lock(m_mutex);
1347 m_cond_expire_ready.notify_all();
1352 void yf::SessionShared::Rep::start()
1354 yf::SessionShared::Worker w(this);
1355 m_thrds.add_thread(new boost::thread(w));
1358 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1362 yf::SessionShared::~SessionShared() {
1365 void yf::SessionShared::start() const
1370 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1374 yf::SessionShared::Frontend::~Frontend()
1378 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1380 boost::mutex::scoped_lock lock(m_mutex);
1382 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1386 it = m_clients.find(package.session());
1387 if (it == m_clients.end())
1390 if (!it->second->m_in_use)
1392 it->second->m_in_use = true;
1395 m_cond_session_ready.wait(lock);
1397 FrontendPtr f(new Frontend(this));
1398 m_clients[package.session()] = f;
1403 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1405 boost::mutex::scoped_lock lock(m_mutex);
1406 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1408 it = m_clients.find(package.session());
1409 if (it != m_clients.end())
1411 if (package.session().is_closed())
1413 m_clients.erase(it);
1417 it->second->m_in_use = false;
1419 m_cond_session_ready.notify_all();
1424 void yf::SessionShared::process(mp::Package &package) const
1426 FrontendPtr f = m_p->get_frontend(package);
1428 Z_GDU *gdu = package.request().get();
1430 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1431 Z_APDU_initRequest && !f->m_is_virtual)
1433 m_p->init(package, gdu, f);
1435 else if (!f->m_is_virtual)
1437 else if (gdu && gdu->which == Z_GDU_Z3950)
1439 Z_APDU *apdu = gdu->u.z3950;
1440 if (apdu->which == Z_APDU_initRequest)
1444 package.response() = odr.create_close(
1446 Z_Close_protocolError,
1449 package.session().close();
1451 else if (apdu->which == Z_APDU_close)
1455 package.response() = odr.create_close(
1457 Z_Close_peerAbort, "received close from client");
1458 package.session().close();
1460 else if (apdu->which == Z_APDU_searchRequest)
1462 f->search(package, apdu);
1464 else if (apdu->which == Z_APDU_presentRequest)
1466 f->present(package, apdu);
1468 else if (apdu->which == Z_APDU_scanRequest)
1470 f->scan(package, apdu);
1476 package.response() = odr.create_close(
1477 apdu, Z_Close_protocolError,
1478 "unsupported APDU in filter_session_shared");
1480 package.session().close();
1483 m_p->release_frontend(package);
1486 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1489 for (ptr = ptr->children; ptr; ptr = ptr->next)
1491 if (ptr->type != XML_ELEMENT_NODE)
1493 if (!strcmp((const char *) ptr->name, "resultset"))
1495 const struct _xmlAttr *attr;
1496 for (attr = ptr->properties; attr; attr = attr->next)
1498 if (!strcmp((const char *) attr->name, "ttl"))
1499 m_p->m_resultset_ttl =
1500 mp::xml::get_int(attr->children, 30);
1501 else if (!strcmp((const char *) attr->name, "max"))
1503 m_p->m_resultset_max =
1504 mp::xml::get_int(attr->children, 10);
1506 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1508 m_p->m_optimize_search =
1509 mp::xml::get_bool(attr->children, true);
1511 else if (!strcmp((const char *) attr->name, "restart"))
1513 m_p->m_restart = mp::xml::get_bool(attr->children, true);
1516 throw mp::filter::FilterException(
1517 "Bad attribute " + std::string((const char *)
1521 else if (!strcmp((const char *) ptr->name, "session"))
1523 const struct _xmlAttr *attr;
1524 for (attr = ptr->properties; attr; attr = attr->next)
1526 if (!strcmp((const char *) attr->name, "ttl"))
1527 m_p->m_session_ttl =
1528 mp::xml::get_int(attr->children, 90);
1529 else if (!strcmp((const char *) attr->name, "max"))
1530 m_p->m_session_max =
1531 mp::xml::get_int(attr->children, 100);
1533 throw mp::filter::FilterException(
1534 "Bad attribute " + std::string((const char *)
1538 else if (!strcmp((const char *) ptr->name, "init"))
1540 const struct _xmlAttr *attr;
1541 for (attr = ptr->properties; attr; attr = attr->next)
1543 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1544 m_p->m_maximumRecordSize =
1545 mp::xml::get_int(attr->children, 0);
1546 else if (!strcmp((const char *) attr->name,
1547 "preferred-message-size"))
1548 m_p->m_preferredMessageSize =
1549 mp::xml::get_int(attr->children, 0);
1551 throw mp::filter::FilterException(
1552 "Bad attribute " + std::string((const char *)
1558 throw mp::filter::FilterException("Bad element "
1559 + std::string((const char *)
1565 static mp::filter::Base* filter_creator()
1567 return new mp::filter::SessionShared;
1571 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1581 * c-file-style: "Stroustrup"
1582 * indent-tabs-mode: nil
1584 * vim: shiftwidth=4 tabstop=8 expandtab