1 /* This file is part of Metaproxy.
2 Copyright (C) 2005-2013 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>
45 namespace mp = metaproxy_1;
46 namespace yf = metaproxy_1::filter;
48 namespace metaproxy_1 {
51 // key for session.. We'll only share sessions with same InitKey
52 class SessionShared::InitKey {
54 bool operator < (const SessionShared::InitKey &k) const;
55 InitKey(Z_InitRequest *req);
56 InitKey(const InitKey &);
59 char *m_idAuthentication_buf;
60 int m_idAuthentication_size;
61 char *m_otherInfo_buf;
65 // worker thread .. for expiry of sessions
66 class SessionShared::Worker {
68 Worker(SessionShared::Rep *rep);
69 void operator() (void);
71 SessionShared::Rep *m_p;
74 class SessionShared::BackendSet {
76 std::string m_result_set_id;
77 Databases m_databases;
78 Odr_int m_result_set_size;
79 yazpp_1::Yaz_Z_Query m_query;
80 time_t m_time_last_use;
82 yazpp_1::RecordCache m_record_cache;
83 Z_OtherInformation *additionalSearchInfoResponse;
84 NMEM mem_additionalSearchInfoResponse;
86 const std::string &result_set_id,
87 const Databases &databases,
88 const yazpp_1::Yaz_Z_Query &query);
91 Package &frontend_package,
92 Package &search_package,
93 const Z_APDU *apdu_req,
94 const BackendInstancePtr bp,
95 Z_Records **z_records,
96 Z_OtherInformation *additionalSearchInfo);
98 // backend connection instance
99 class SessionShared::BackendInstance {
101 friend class BackendClass;
102 friend class BackendSet;
104 mp::Session m_session;
105 BackendSetList m_sets;
108 int m_result_set_sequence;
109 time_t m_time_last_use;
110 mp::Package * m_close_package;
114 // backends of some class (all with same InitKey)
115 class SessionShared::BackendClass : boost::noncopyable {
117 friend struct Frontend;
118 bool m_named_result_sets;
119 BackendInstanceList m_backend_list;
120 BackendInstancePtr create_backend(const Package &package);
121 void remove_backend(BackendInstancePtr b);
122 BackendInstancePtr get_backend(const Package &package);
123 void use_backend(BackendInstancePtr b);
124 void release_backend(BackendInstancePtr b);
125 bool expire_instances();
126 yazpp_1::GDU m_init_request;
127 yazpp_1::GDU m_init_response;
128 boost::mutex m_mutex_backend_class;
130 time_t m_backend_set_ttl;
131 time_t m_backend_expiry_ttl;
132 size_t m_backend_set_max;
133 Odr_int m_preferredMessageSize;
134 Odr_int m_maximumRecordSize;
136 BackendClass(const yazpp_1::GDU &init_request,
140 Odr_int preferredRecordSize,
141 Odr_int maximumRecordSize);
144 // frontend result set
145 class SessionShared::FrontendSet {
146 Databases m_databases;
147 yazpp_1::Yaz_Z_Query m_query;
149 const Databases &get_databases();
150 const yazpp_1::Yaz_Z_Query &get_query();
152 const Databases &databases,
153 const yazpp_1::Yaz_Z_Query &query);
157 struct SessionShared::Frontend {
162 Z_Options m_init_options;
163 void search(Package &package, Z_APDU *apdu);
164 void present(Package &package, Z_APDU *apdu);
165 void scan(Package &package, Z_APDU *apdu);
167 int result_set_ref(ODR o,
168 const Databases &databases,
169 Z_RPNStructure *s, std::string &rset);
170 void get_set(mp::Package &package,
171 const Z_APDU *apdu_req,
172 const Databases &databases,
173 yazpp_1::Yaz_Z_Query &query,
174 BackendInstancePtr &found_backend,
175 BackendSetPtr &found_set);
176 void override_set(BackendInstancePtr &found_backend,
177 std::string &result_set_id,
178 const Databases &databases,
179 bool out_of_sessions);
182 BackendClassPtr m_backend_class;
183 FrontendSets m_frontend_sets;
186 class SessionShared::Rep {
187 friend class SessionShared;
188 friend struct Frontend;
190 FrontendPtr get_frontend(Package &package);
191 void release_frontend(Package &package);
196 void expire_classes();
198 void init(Package &package, const Z_GDU *gdu,
199 FrontendPtr frontend);
201 boost::mutex m_mutex;
202 boost::condition m_cond_session_ready;
203 std::map<mp::Session, FrontendPtr> m_clients;
205 BackendClassMap m_backend_map;
206 boost::mutex m_mutex_backend_map;
207 boost::thread_group m_thrds;
211 bool m_optimize_search;
214 Odr_int m_preferredMessageSize;
215 Odr_int m_maximumRecordSize;
220 yf::SessionShared::FrontendSet::FrontendSet(
221 const Databases &databases,
222 const yazpp_1::Yaz_Z_Query &query)
223 : m_databases(databases), m_query(query)
227 const yf::SessionShared::Databases &
228 yf::SessionShared::FrontendSet::get_databases()
233 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
238 yf::SessionShared::InitKey::InitKey(const InitKey &k)
240 m_odr = odr_createmem(ODR_ENCODE);
242 m_idAuthentication_size = k.m_idAuthentication_size;
243 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
244 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
245 m_idAuthentication_size);
247 m_otherInfo_size = k.m_otherInfo_size;
248 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
249 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
253 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
255 m_odr = odr_createmem(ODR_ENCODE);
257 Z_IdAuthentication *t = req->idAuthentication;
258 z_IdAuthentication(m_odr, &t, 1, 0);
259 m_idAuthentication_buf =
260 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
262 Z_OtherInformation *o = req->otherInfo;
263 z_OtherInformation(m_odr, &o, 1, 0);
264 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
267 yf::SessionShared::InitKey::~InitKey()
272 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
276 c = mp::util::memcmp2(
277 (void*) m_idAuthentication_buf, m_idAuthentication_size,
278 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
284 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
285 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
293 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
295 boost::mutex::scoped_lock lock(m_mutex_backend_class);
300 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
302 BackendInstanceList::iterator it = m_backend_list.begin();
304 while (it != m_backend_list.end())
309 (*it)->m_close_package->response() = odr.create_close(
310 0, Z_Close_lackOfActivity, 0);
311 (*it)->m_close_package->session().close();
312 (*it)->m_close_package->move();
314 it = m_backend_list.erase(it);
323 yf::SessionShared::BackendInstancePtr
324 yf::SessionShared::BackendClass::get_backend(
325 const mp::Package &frontend_package)
328 boost::mutex::scoped_lock lock(m_mutex_backend_class);
330 BackendInstanceList::const_iterator it = m_backend_list.begin();
332 BackendInstancePtr backend1; // null
334 for (; it != m_backend_list.end(); it++)
336 if (!(*it)->m_in_use)
339 || (*it)->m_sequence_this < backend1->m_sequence_this)
345 use_backend(backend1);
349 return create_backend(frontend_package);
352 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
354 backend->m_in_use = true;
355 backend->m_sequence_this = m_sequence_top++;
358 void yf::SessionShared::BackendInstance::timestamp()
361 time(&m_time_last_use);
364 yf::SessionShared::BackendInstance::~BackendInstance()
366 delete m_close_package;
369 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
370 const mp::Package &frontend_package)
372 BackendInstancePtr bp(new BackendInstance);
373 BackendInstancePtr null;
375 bp->m_close_package =
376 new mp::Package(bp->m_session, frontend_package.origin());
377 bp->m_close_package->copy_filter(frontend_package);
379 Package init_package(bp->m_session, frontend_package.origin());
381 init_package.copy_filter(frontend_package);
383 yazpp_1::GDU actual_init_request = m_init_request;
384 Z_GDU *init_pdu = actual_init_request.get();
386 assert(init_pdu->which == Z_GDU_Z3950);
387 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
389 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
390 ODR_MASK_ZERO(req->options);
392 ODR_MASK_SET(req->options, Z_Options_search);
393 ODR_MASK_SET(req->options, Z_Options_present);
394 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
395 ODR_MASK_SET(req->options, Z_Options_scan);
397 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
398 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
399 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
401 if (m_preferredMessageSize)
402 *req->preferredMessageSize = m_preferredMessageSize;
403 if (m_maximumRecordSize)
404 *req->maximumRecordSize = m_maximumRecordSize;
406 init_package.request() = init_pdu;
410 boost::mutex::scoped_lock lock(m_mutex_backend_class);
412 m_named_result_sets = false;
413 Z_GDU *gdu = init_package.response().get();
415 if (gdu && gdu->which == Z_GDU_Z3950
416 && gdu->u.z3950->which == Z_APDU_initResponse)
418 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
419 m_init_response = gdu->u.z3950;
420 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
422 m_named_result_sets = true;
424 if (*gdu->u.z3950->u.initResponse->result
425 && !init_package.session().is_closed())
428 time(&bp->m_time_last_use);
429 bp->m_sequence_this = 0;
430 bp->m_result_set_sequence = 0;
431 m_backend_list.push_back(bp);
437 yazpp_1::GDU empty_gdu;
438 m_init_response = empty_gdu;
441 if (!init_package.session().is_closed())
443 init_package.copy_filter(frontend_package);
444 init_package.session().close();
451 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
455 Odr_int preferredMessageSize,
456 Odr_int maximumRecordSize)
457 : m_named_result_sets(false), m_init_request(init_request),
458 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
459 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
460 m_preferredMessageSize(preferredMessageSize),
461 m_maximumRecordSize(maximumRecordSize)
464 yf::SessionShared::BackendClass::~BackendClass()
467 void yf::SessionShared::Rep::stat()
470 int no_instances = 0;
471 BackendClassMap::const_iterator it;
473 boost::mutex::scoped_lock lock(m_mutex_backend_map);
474 for (it = m_backend_map.begin(); it != m_backend_map.end(); it++)
476 BackendClassPtr bc = it->second;
478 BackendInstanceList::iterator bit = bc->m_backend_list.begin();
479 for (; bit != bc->m_backend_list.end(); bit++)
483 yaz_log(YLOG_LOG, "backend classes=%d instances=%d", no_classes,
487 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
488 FrontendPtr frontend)
490 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
492 frontend->m_is_virtual = true;
493 frontend->m_init_options = *req->options;
496 boost::mutex::scoped_lock lock(m_mutex_backend_map);
497 BackendClassMap::const_iterator it;
498 it = m_backend_map.find(k);
499 if (it == m_backend_map.end())
501 BackendClassPtr b(new BackendClass(gdu->u.z3950,
505 m_preferredMessageSize,
506 m_maximumRecordSize));
507 m_backend_map[k] = b;
508 frontend->m_backend_class = b;
512 frontend->m_backend_class = it->second;
515 BackendClassPtr bc = frontend->m_backend_class;
518 // we only need to get init response from "first" target in
519 // backend class - the assumption being that init response is
521 if (bc->m_backend_list.size() == 0)
523 BackendInstancePtr backend = bc->create_backend(package);
525 bc->release_backend(backend);
528 yazpp_1::GDU init_response;
530 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
532 init_response = bc->m_init_response;
535 if (init_response.get())
537 Z_GDU *response_gdu = init_response.get();
538 mp::util::transfer_referenceId(odr, gdu->u.z3950,
539 response_gdu->u.z3950);
540 Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
541 Z_Options *server_options = init_res->options;
542 Z_Options *client_options = &frontend->m_init_options;
544 for (i = 0; i < 30; i++)
545 if (!ODR_MASK_GET(client_options, i))
546 ODR_MASK_CLEAR(server_options, i);
548 if (!m_preferredMessageSize ||
549 *init_res->preferredMessageSize > *req->preferredMessageSize)
550 *init_res->preferredMessageSize = *req->preferredMessageSize;
552 if (!m_maximumRecordSize ||
553 *init_res->maximumRecordSize > *req->maximumRecordSize)
554 *init_res->maximumRecordSize = *req->maximumRecordSize;
556 package.response() = init_response;
557 if (!*response_gdu->u.z3950->u.initResponse->result)
558 package.session().close();
563 odr.create_initResponse(
564 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
565 "session_shared: target closed connection during init");
566 *apdu->u.initResponse->result = 0;
567 package.response() = apdu;
568 package.session().close();
572 void yf::SessionShared::BackendSet::timestamp()
574 time(&m_time_last_use);
577 yf::SessionShared::BackendSet::BackendSet(
578 const std::string &result_set_id,
579 const Databases &databases,
580 const yazpp_1::Yaz_Z_Query &query) :
581 m_result_set_id(result_set_id),
582 m_databases(databases), m_result_set_size(0), m_query(query)
585 mem_additionalSearchInfoResponse = nmem_create();
586 additionalSearchInfoResponse = 0;
589 yf::SessionShared::BackendSet::~BackendSet()
591 nmem_destroy(mem_additionalSearchInfoResponse);
594 static int get_diagnostic(Z_DefaultDiagFormat *r)
596 return *r->condition;
599 bool yf::SessionShared::BackendSet::search(
600 mp::Package &frontend_package,
601 mp::Package &search_package,
602 const Z_APDU *frontend_apdu,
603 const BackendInstancePtr bp,
604 Z_Records **z_records,
605 Z_OtherInformation *additionalSearchInfo)
608 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
609 Z_SearchRequest *req = apdu_req->u.searchRequest;
611 req->additionalSearchInfo = additionalSearchInfo;
612 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
613 req->query = m_query.get_Z_Query();
615 req->num_databaseNames = m_databases.size();
616 req->databaseNames = (char**)
617 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
618 Databases::const_iterator it = m_databases.begin();
620 for (; it != m_databases.end(); it++)
621 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
623 if (frontend_apdu->which == Z_APDU_searchRequest)
624 req->preferredRecordSyntax =
625 frontend_apdu->u.searchRequest->preferredRecordSyntax;
627 search_package.request() = apdu_req;
629 search_package.move();
631 Z_GDU *gdu = search_package.response().get();
632 if (!search_package.session().is_closed()
633 && gdu && gdu->which == Z_GDU_Z3950
634 && gdu->u.z3950->which == Z_APDU_searchResponse)
636 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
637 *z_records = b_resp->records;
638 m_result_set_size = *b_resp->resultCount;
640 nmem_reset(mem_additionalSearchInfoResponse);
641 additionalSearchInfoResponse = yaz_clone_z_OtherInformation(
642 b_resp->additionalSearchInfo, mem_additionalSearchInfoResponse);
646 const char *addinfo = "session_shared: "
647 "target closed connection during search";
648 if (frontend_apdu->which == Z_APDU_searchRequest)
649 f_apdu = odr.create_searchResponse(
650 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
651 else if (frontend_apdu->which == Z_APDU_presentRequest)
652 f_apdu = odr.create_presentResponse(
653 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
655 f_apdu = odr.create_close(
656 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
657 frontend_package.response() = f_apdu;
661 void yf::SessionShared::Frontend::override_set(
662 BackendInstancePtr &found_backend,
663 std::string &result_set_id,
664 const Databases &databases,
665 bool out_of_sessions)
667 BackendClassPtr bc = m_backend_class;
668 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
672 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
673 for (; it != bc->m_backend_list.end(); it++)
675 if (!(*it)->m_in_use)
677 BackendSetList::iterator set_it = (*it)->m_sets.begin();
678 for (; set_it != (*it)->m_sets.end(); set_it++)
680 if ((max_sets > 1 || (*set_it)->m_databases == databases)
683 now < (*set_it)->m_time_last_use ||
684 now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
687 result_set_id = (*set_it)->m_result_set_id;
688 found_backend->m_sets.erase(set_it);
694 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
696 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
699 if (bc->m_named_result_sets)
701 result_set_id = boost::io::str(
702 boost::format("%1%") %
703 found_backend->m_result_set_sequence);
704 found_backend->m_result_set_sequence++;
707 result_set_id = "default";
713 void yf::SessionShared::Frontend::get_set(mp::Package &package,
714 const Z_APDU *apdu_req,
715 const Databases &databases,
716 yazpp_1::Yaz_Z_Query &query,
717 BackendInstancePtr &found_backend,
718 BackendSetPtr &found_set)
720 bool session_restarted = false;
721 Z_OtherInformation *additionalSearchInfo = 0;
723 if (apdu_req->which == Z_APDU_searchRequest)
724 additionalSearchInfo = apdu_req->u.searchRequest->additionalSearchInfo;
727 std::string result_set_id;
728 bool out_of_sessions = false;
729 BackendClassPtr bc = m_backend_class;
731 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
733 if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
734 out_of_sessions = true;
736 if (m_p->m_optimize_search && !additionalSearchInfo)
738 // look at each backend and see if we have a similar search
739 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
740 for (; it != bc->m_backend_list.end(); it++)
742 if (!(*it)->m_in_use)
744 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
745 for (; set_it != (*it)->m_sets.end(); set_it++)
747 if ((*set_it)->m_databases == databases
748 && query.match(&(*set_it)->m_query))
752 bc->use_backend(found_backend);
753 // found matching set. No need to search again
760 override_set(found_backend, result_set_id, databases, out_of_sessions);
762 bc->use_backend(found_backend);
766 // create a new backend set (and new set) if we're not out of sessions
767 if (!out_of_sessions)
768 found_backend = bc->create_backend(package);
774 const char *addinfo = 0;
777 addinfo = "session_shared: all sessions in use";
779 addinfo = "session_shared: could not create backend";
780 if (apdu_req->which == Z_APDU_searchRequest)
782 f_apdu = odr.create_searchResponse(
783 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
785 else if (apdu_req->which == Z_APDU_presentRequest)
787 f_apdu = odr.create_presentResponse(
788 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
792 f_apdu = odr.create_close(
793 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
795 package.response() = f_apdu;
798 if (bc->m_named_result_sets)
800 result_set_id = boost::io::str(
801 boost::format("%1%") % found_backend->m_result_set_sequence);
802 found_backend->m_result_set_sequence++;
805 result_set_id = "default";
807 found_backend->timestamp();
809 // we must search ...
810 BackendSetPtr new_set(new BackendSet(result_set_id,
812 Z_Records *z_records = 0;
814 Package search_package(found_backend->m_session, package.origin());
815 search_package.copy_filter(package);
817 if (!new_set->search(package, search_package,
818 apdu_req, found_backend, &z_records,
819 additionalSearchInfo))
821 bc->remove_backend(found_backend);
822 return; // search error
828 if (z_records->which == Z_Records_NSD)
831 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
833 else if (z_records->which == Z_Records_multipleNSD)
835 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
838 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
839 Z_DiagRec_defaultFormat)
841 condition = get_diagnostic(
842 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
846 if (m_p->m_restart && !session_restarted &&
847 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
849 package.log("session_shared", YLOG_LOG, "restart");
850 bc->remove_backend(found_backend);
851 session_restarted = true;
852 found_backend.reset();
860 if (apdu_req->which == Z_APDU_searchRequest)
862 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
864 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
865 *f_resp->searchStatus = Z_SearchResponse_none;
866 f_resp->records = z_records;
867 package.response() = f_apdu;
869 if (apdu_req->which == Z_APDU_presentRequest)
871 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
873 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
874 f_resp->records = z_records;
875 package.response() = f_apdu;
877 bc->release_backend(found_backend);
878 return; // search error
881 if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
883 package.log("session_shared", YLOG_LOG, "restart");
884 bc->remove_backend(found_backend);
885 session_restarted = true;
886 found_backend.reset();
891 found_set->timestamp();
892 found_backend->m_sets.push_back(found_set);
895 int yf::SessionShared::Frontend::result_set_ref(ODR o,
896 const Databases &databases,
903 case Z_RPNStructure_simple:
904 if (s->u.simple->which == Z_Operand_resultSetId)
906 const char *id = s->u.simple->u.resultSetId;
909 FrontendSets::iterator fset_it = m_frontend_sets.find(id);
910 if (fset_it == m_frontend_sets.end())
912 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
914 else if (fset_it->second->get_databases() != databases)
916 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
920 yazpp_1::Yaz_Z_Query query = fset_it->second->get_query();
921 Z_Query *q = yaz_copy_Z_Query(query.get_Z_Query(), o);
922 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
924 s->which = q->u.type_1->RPNStructure->which;
925 s->u.simple = q->u.type_1->RPNStructure->u.simple;
929 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
934 case Z_RPNStructure_complex:
935 ret = result_set_ref(o, databases, s->u.complex->s1, rset);
937 ret = result_set_ref(o, databases, s->u.complex->s2, rset);
943 void yf::SessionShared::Frontend::search(mp::Package &package,
946 Z_SearchRequest *req = apdu_req->u.searchRequest;
947 FrontendSets::iterator fset_it =
948 m_frontend_sets.find(req->resultSetName);
949 if (fset_it != m_frontend_sets.end())
951 // result set already exist
952 // if replace indicator is off: we return diagnostic if
953 // result set already exist.
954 if (*req->replaceIndicator == 0)
958 odr.create_searchResponse(
960 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
962 package.response() = apdu;
965 m_frontend_sets.erase(fset_it);
970 for (i = 0; i < req->num_databaseNames; i++)
971 databases.push_back(req->databaseNames[i]);
974 yazpp_1::Yaz_Z_Query query;
975 query.set_Z_Query(req->query);
977 Z_Query *q = query.get_Z_Query();
978 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
982 int diag = result_set_ref(odr, databases, q->u.type_1->RPNStructure,
987 odr.create_searchResponse(
991 package.response() = apdu;
994 query.set_Z_Query(q);
997 BackendSetPtr found_set; // null
998 BackendInstancePtr found_backend; // null
1000 get_set(package, apdu_req, databases, query, found_backend, found_set);
1005 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
1006 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
1007 *f_resp->resultCount = found_set->m_result_set_size;
1008 f_resp->additionalSearchInfo = found_set->additionalSearchInfoResponse;
1009 package.response() = f_apdu;
1011 FrontendSetPtr fset(new FrontendSet(databases, query));
1012 m_frontend_sets[req->resultSetName] = fset;
1014 m_backend_class->release_backend(found_backend);
1017 void yf::SessionShared::Frontend::present(mp::Package &package,
1021 Z_PresentRequest *req = apdu_req->u.presentRequest;
1023 FrontendSets::iterator fset_it =
1024 m_frontend_sets.find(req->resultSetId);
1026 if (fset_it == m_frontend_sets.end())
1029 odr.create_presentResponse(
1031 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
1033 package.response() = apdu;
1036 FrontendSetPtr fset = fset_it->second;
1038 Databases databases = fset->get_databases();
1039 yazpp_1::Yaz_Z_Query query = fset->get_query();
1041 BackendClassPtr bc = m_backend_class;
1042 BackendSetPtr found_set; // null
1043 BackendInstancePtr found_backend;
1045 get_set(package, apdu_req, databases, query, found_backend, found_set);
1049 Z_NamePlusRecordList *npr_res = 0;
1050 if (found_set->m_record_cache.lookup(odr, &npr_res,
1051 *req->resultSetStartPoint,
1052 *req->numberOfRecordsRequested,
1053 req->preferredRecordSyntax,
1054 req->recordComposition))
1056 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1057 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1059 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1060 " records in cache %p",
1061 *req->resultSetStartPoint,
1062 *req->numberOfRecordsRequested,
1063 &found_set->m_record_cache);
1065 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
1066 *f_resp->nextResultSetPosition =
1067 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
1068 // f_resp->presentStatus assumed OK.
1069 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
1070 f_resp->records->which = Z_Records_DBOSD;
1071 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
1072 package.response() = f_apdu_res;
1073 bc->release_backend(found_backend);
1077 found_backend->timestamp();
1079 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
1080 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
1081 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1082 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
1083 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
1084 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
1085 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1086 p_req->recordComposition = req->recordComposition;
1088 Package present_package(found_backend->m_session, package.origin());
1089 present_package.copy_filter(package);
1091 present_package.request() = p_apdu;
1093 present_package.move();
1095 Z_GDU *gdu = present_package.response().get();
1096 if (!present_package.session().is_closed()
1097 && gdu && gdu->which == Z_GDU_Z3950
1098 && gdu->u.z3950->which == Z_APDU_presentResponse)
1100 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
1101 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1102 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1104 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
1105 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
1106 f_resp->presentStatus= b_resp->presentStatus;
1107 f_resp->records = b_resp->records;
1108 f_resp->otherInfo = b_resp->otherInfo;
1109 package.response() = f_apdu_res;
1111 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
1114 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1115 " records to cache %p",
1116 *req->resultSetStartPoint,
1117 *f_resp->numberOfRecordsReturned,
1118 &found_set->m_record_cache);
1120 found_set->m_record_cache.add(
1122 b_resp->records->u.databaseOrSurDiagnostics,
1123 *req->resultSetStartPoint,
1124 *f_resp->numberOfRecordsReturned);
1126 bc->release_backend(found_backend);
1130 bc->remove_backend(found_backend);
1131 Z_APDU *f_apdu_res =
1132 odr.create_presentResponse(
1133 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1134 "session_shared: target closed connection during present");
1135 package.response() = f_apdu_res;
1139 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1142 BackendClassPtr bc = m_backend_class;
1143 BackendInstancePtr backend = bc->get_backend(frontend_package);
1147 Z_APDU *apdu = odr.create_scanResponse(
1148 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1149 "session_shared: could not create backend");
1150 frontend_package.response() = apdu;
1154 Package scan_package(backend->m_session, frontend_package.origin());
1155 backend->timestamp();
1156 scan_package.copy_filter(frontend_package);
1157 scan_package.request() = apdu_req;
1158 scan_package.move();
1159 frontend_package.response() = scan_package.response();
1160 if (scan_package.session().is_closed())
1162 frontend_package.session().close();
1163 bc->remove_backend(backend);
1166 bc->release_backend(backend);
1170 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1174 void yf::SessionShared::Worker::operator() (void)
1179 bool yf::SessionShared::BackendClass::expire_instances()
1183 boost::mutex::scoped_lock lock(m_mutex_backend_class);
1184 BackendInstanceList::iterator bit = m_backend_list.begin();
1185 while (bit != m_backend_list.end())
1187 time_t last_use = (*bit)->m_time_last_use;
1189 if ((*bit)->m_in_use)
1193 else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1196 (*bit)->m_close_package->response() = odr.create_close(
1197 0, Z_Close_lackOfActivity, 0);
1198 (*bit)->m_close_package->session().close();
1199 (*bit)->m_close_package->move();
1201 bit = m_backend_list.erase(bit);
1208 if (m_backend_list.empty())
1213 void yf::SessionShared::Rep::expire_classes()
1215 boost::mutex::scoped_lock lock(m_mutex_backend_map);
1216 BackendClassMap::iterator b_it = m_backend_map.begin();
1217 while (b_it != m_backend_map.end())
1219 if (b_it->second->expire_instances())
1221 m_backend_map.erase(b_it);
1222 b_it = m_backend_map.begin();
1229 void yf::SessionShared::Rep::expire()
1234 boost::xtime_get(&xt,
1235 #if BOOST_VERSION >= 105000
1241 xt.sec += m_session_ttl / 3;
1242 boost::thread::sleep(xt);
1249 yf::SessionShared::Rep::Rep()
1251 m_resultset_ttl = 30;
1252 m_resultset_max = 10;
1254 m_optimize_search = true;
1256 m_session_max = 100;
1257 m_preferredMessageSize = 0;
1258 m_maximumRecordSize = 0;
1261 void yf::SessionShared::Rep::start()
1263 yf::SessionShared::Worker w(this);
1264 m_thrds.add_thread(new boost::thread(w));
1267 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1271 yf::SessionShared::~SessionShared() {
1274 void yf::SessionShared::start() const
1279 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1283 yf::SessionShared::Frontend::~Frontend()
1287 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1289 boost::mutex::scoped_lock lock(m_mutex);
1291 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1295 it = m_clients.find(package.session());
1296 if (it == m_clients.end())
1299 if (!it->second->m_in_use)
1301 it->second->m_in_use = true;
1304 m_cond_session_ready.wait(lock);
1306 FrontendPtr f(new Frontend(this));
1307 m_clients[package.session()] = f;
1312 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1314 boost::mutex::scoped_lock lock(m_mutex);
1315 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1317 it = m_clients.find(package.session());
1318 if (it != m_clients.end())
1320 if (package.session().is_closed())
1322 m_clients.erase(it);
1326 it->second->m_in_use = false;
1328 m_cond_session_ready.notify_all();
1333 void yf::SessionShared::process(mp::Package &package) const
1335 FrontendPtr f = m_p->get_frontend(package);
1337 Z_GDU *gdu = package.request().get();
1339 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1340 Z_APDU_initRequest && !f->m_is_virtual)
1342 m_p->init(package, gdu, f);
1344 else if (!f->m_is_virtual)
1346 else if (gdu && gdu->which == Z_GDU_Z3950)
1348 Z_APDU *apdu = gdu->u.z3950;
1349 if (apdu->which == Z_APDU_initRequest)
1353 package.response() = odr.create_close(
1355 Z_Close_protocolError,
1358 package.session().close();
1360 else if (apdu->which == Z_APDU_close)
1364 package.response() = odr.create_close(
1366 Z_Close_peerAbort, "received close from client");
1367 package.session().close();
1369 else if (apdu->which == Z_APDU_searchRequest)
1371 f->search(package, apdu);
1373 else if (apdu->which == Z_APDU_presentRequest)
1375 f->present(package, apdu);
1377 else if (apdu->which == Z_APDU_scanRequest)
1379 f->scan(package, apdu);
1385 package.response() = odr.create_close(
1386 apdu, Z_Close_protocolError,
1387 "unsupported APDU in filter_session_shared");
1389 package.session().close();
1392 m_p->release_frontend(package);
1395 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1398 for (ptr = ptr->children; ptr; ptr = ptr->next)
1400 if (ptr->type != XML_ELEMENT_NODE)
1402 if (!strcmp((const char *) ptr->name, "resultset"))
1404 const struct _xmlAttr *attr;
1405 for (attr = ptr->properties; attr; attr = attr->next)
1407 if (!strcmp((const char *) attr->name, "ttl"))
1408 m_p->m_resultset_ttl =
1409 mp::xml::get_int(attr->children, 30);
1410 else if (!strcmp((const char *) attr->name, "max"))
1412 m_p->m_resultset_max =
1413 mp::xml::get_int(attr->children, 10);
1415 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1417 m_p->m_optimize_search =
1418 mp::xml::get_bool(attr->children, true);
1420 else if (!strcmp((const char *) attr->name, "restart"))
1422 m_p->m_restart = mp::xml::get_bool(attr->children, true);
1425 throw mp::filter::FilterException(
1426 "Bad attribute " + std::string((const char *)
1430 else if (!strcmp((const char *) ptr->name, "session"))
1432 const struct _xmlAttr *attr;
1433 for (attr = ptr->properties; attr; attr = attr->next)
1435 if (!strcmp((const char *) attr->name, "ttl"))
1436 m_p->m_session_ttl =
1437 mp::xml::get_int(attr->children, 90);
1438 else if (!strcmp((const char *) attr->name, "max"))
1439 m_p->m_session_max =
1440 mp::xml::get_int(attr->children, 100);
1442 throw mp::filter::FilterException(
1443 "Bad attribute " + std::string((const char *)
1447 else if (!strcmp((const char *) ptr->name, "init"))
1449 const struct _xmlAttr *attr;
1450 for (attr = ptr->properties; attr; attr = attr->next)
1452 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1453 m_p->m_maximumRecordSize =
1454 mp::xml::get_int(attr->children, 0);
1455 else if (!strcmp((const char *) attr->name,
1456 "preferred-message-size"))
1457 m_p->m_preferredMessageSize =
1458 mp::xml::get_int(attr->children, 0);
1460 throw mp::filter::FilterException(
1461 "Bad attribute " + std::string((const char *)
1467 throw mp::filter::FilterException("Bad element "
1468 + std::string((const char *)
1474 static mp::filter::Base* filter_creator()
1476 return new mp::filter::SessionShared;
1480 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1490 * c-file-style: "Stroustrup"
1491 * indent-tabs-mode: nil
1493 * vim: shiftwidth=4 tabstop=8 expandtab