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>
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;
84 Z_OtherInformation *additionalSearchInfoRequest;
85 Z_OtherInformation *additionalSearchInfoResponse;
86 NMEM mem_additionalSearchInfo;
88 const std::string &result_set_id,
89 const Databases &databases,
90 const yazpp_1::Yaz_Z_Query &query,
91 Z_OtherInformation *additionalSearchInfoRequest);
94 Package &frontend_package,
95 Package &search_package,
96 const Z_APDU *apdu_req,
97 const BackendInstancePtr bp,
98 Z_Records **z_records);
100 // backend connection instance
101 class SessionShared::BackendInstance {
103 friend class BackendClass;
104 friend class BackendSet;
106 mp::Session m_session;
107 BackendSetList m_sets;
110 int m_result_set_sequence;
111 time_t m_time_last_use;
112 mp::Package * m_close_package;
116 // backends of some class (all with same InitKey)
117 class SessionShared::BackendClass : boost::noncopyable {
119 friend struct Frontend;
120 bool m_named_result_sets;
121 BackendInstanceList m_backend_list;
122 BackendInstancePtr create_backend(const Package &package);
123 void remove_backend(BackendInstancePtr b);
124 BackendInstancePtr get_backend(const Package &package);
125 void use_backend(BackendInstancePtr b);
126 void release_backend(BackendInstancePtr b);
127 bool expire_instances();
128 yazpp_1::GDU m_init_request;
129 yazpp_1::GDU m_init_response;
130 boost::mutex m_mutex_backend_class;
132 time_t m_backend_set_ttl;
133 time_t m_backend_expiry_ttl;
134 size_t m_backend_set_max;
135 Odr_int m_preferredMessageSize;
136 Odr_int m_maximumRecordSize;
138 BackendClass(const yazpp_1::GDU &init_request,
142 Odr_int preferredRecordSize,
143 Odr_int maximumRecordSize);
146 // frontend result set
147 class SessionShared::FrontendSet {
148 Databases m_databases;
149 yazpp_1::Yaz_Z_Query m_query;
151 const Databases &get_databases();
152 const yazpp_1::Yaz_Z_Query &get_query();
154 const Databases &databases,
155 const yazpp_1::Yaz_Z_Query &query);
159 struct SessionShared::Frontend {
164 Z_Options m_init_options;
165 void search(Package &package, Z_APDU *apdu);
166 void present(Package &package, Z_APDU *apdu);
167 void scan(Package &package, Z_APDU *apdu);
169 int result_set_ref(ODR o,
170 const Databases &databases,
171 Z_RPNStructure *s, std::string &rset);
172 void get_set(mp::Package &package,
173 const Z_APDU *apdu_req,
174 const Databases &databases,
175 yazpp_1::Yaz_Z_Query &query,
176 BackendInstancePtr &found_backend,
177 BackendSetPtr &found_set);
178 void override_set(BackendInstancePtr &found_backend,
179 std::string &result_set_id,
180 const Databases &databases,
181 bool out_of_sessions);
184 BackendClassPtr m_backend_class;
185 FrontendSets m_frontend_sets;
188 class SessionShared::Rep {
189 friend class SessionShared;
190 friend struct Frontend;
192 FrontendPtr get_frontend(Package &package);
193 void release_frontend(Package &package);
199 void expire_classes();
201 void init(Package &package, const Z_GDU *gdu,
202 FrontendPtr frontend);
204 boost::mutex m_mutex;
205 boost::condition m_cond_session_ready;
206 boost::condition m_cond_expire_ready;
207 std::map<mp::Session, FrontendPtr> m_clients;
209 BackendClassMap m_backend_map;
210 boost::mutex m_mutex_backend_map;
211 boost::thread_group m_thrds;
215 bool m_optimize_search;
218 Odr_int m_preferredMessageSize;
219 Odr_int m_maximumRecordSize;
225 yf::SessionShared::FrontendSet::FrontendSet(
226 const Databases &databases,
227 const yazpp_1::Yaz_Z_Query &query)
228 : m_databases(databases), m_query(query)
232 const yf::SessionShared::Databases &
233 yf::SessionShared::FrontendSet::get_databases()
238 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
243 yf::SessionShared::InitKey::InitKey(const InitKey &k)
245 m_odr = odr_createmem(ODR_ENCODE);
247 m_idAuthentication_size = k.m_idAuthentication_size;
248 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
249 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
250 m_idAuthentication_size);
252 m_otherInfo_size = k.m_otherInfo_size;
253 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
254 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
258 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
260 m_odr = odr_createmem(ODR_ENCODE);
262 Z_IdAuthentication *t = req->idAuthentication;
263 z_IdAuthentication(m_odr, &t, 1, 0);
264 m_idAuthentication_buf =
265 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
267 Z_OtherInformation *o = req->otherInfo;
268 z_OtherInformation(m_odr, &o, 1, 0);
269 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
272 yf::SessionShared::InitKey::~InitKey()
277 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
281 c = mp::util::memcmp2(
282 (void*) m_idAuthentication_buf, m_idAuthentication_size,
283 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
289 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
290 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
298 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
300 boost::mutex::scoped_lock lock(m_mutex_backend_class);
305 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
307 BackendInstanceList::iterator it = m_backend_list.begin();
309 while (it != m_backend_list.end())
314 (*it)->m_close_package->response() = odr.create_close(
315 0, Z_Close_lackOfActivity, 0);
316 (*it)->m_close_package->session().close();
317 (*it)->m_close_package->move();
319 it = m_backend_list.erase(it);
328 yf::SessionShared::BackendInstancePtr
329 yf::SessionShared::BackendClass::get_backend(
330 const mp::Package &frontend_package)
333 boost::mutex::scoped_lock lock(m_mutex_backend_class);
335 BackendInstanceList::const_iterator it = m_backend_list.begin();
337 BackendInstancePtr backend1; // null
339 for (; it != m_backend_list.end(); it++)
341 if (!(*it)->m_in_use)
344 || (*it)->m_sequence_this < backend1->m_sequence_this)
350 use_backend(backend1);
354 return create_backend(frontend_package);
357 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
359 backend->m_in_use = true;
360 backend->m_sequence_this = m_sequence_top++;
363 void yf::SessionShared::BackendInstance::timestamp()
366 time(&m_time_last_use);
369 yf::SessionShared::BackendInstance::~BackendInstance()
371 delete m_close_package;
374 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
375 const mp::Package &frontend_package)
377 BackendInstancePtr bp(new BackendInstance);
378 BackendInstancePtr null;
380 bp->m_close_package =
381 new mp::Package(bp->m_session, frontend_package.origin());
382 bp->m_close_package->copy_filter(frontend_package);
384 Package init_package(bp->m_session, frontend_package.origin());
386 init_package.copy_filter(frontend_package);
388 yazpp_1::GDU actual_init_request = m_init_request;
389 Z_GDU *init_pdu = actual_init_request.get();
391 assert(init_pdu->which == Z_GDU_Z3950);
392 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
394 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
395 ODR_MASK_ZERO(req->options);
397 ODR_MASK_SET(req->options, Z_Options_search);
398 ODR_MASK_SET(req->options, Z_Options_present);
399 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
400 ODR_MASK_SET(req->options, Z_Options_scan);
402 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
403 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
404 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
406 if (m_preferredMessageSize)
407 *req->preferredMessageSize = m_preferredMessageSize;
408 if (m_maximumRecordSize)
409 *req->maximumRecordSize = m_maximumRecordSize;
411 init_package.request() = init_pdu;
415 boost::mutex::scoped_lock lock(m_mutex_backend_class);
417 m_named_result_sets = false;
418 Z_GDU *gdu = init_package.response().get();
420 if (gdu && gdu->which == Z_GDU_Z3950
421 && gdu->u.z3950->which == Z_APDU_initResponse)
423 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
424 m_init_response = gdu->u.z3950;
425 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
427 m_named_result_sets = true;
429 if (*gdu->u.z3950->u.initResponse->result
430 && !init_package.session().is_closed())
433 time(&bp->m_time_last_use);
434 bp->m_sequence_this = 0;
435 bp->m_result_set_sequence = 0;
436 m_backend_list.push_back(bp);
442 yazpp_1::GDU empty_gdu;
443 m_init_response = empty_gdu;
446 if (!init_package.session().is_closed())
448 init_package.copy_filter(frontend_package);
449 init_package.session().close();
456 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
460 Odr_int preferredMessageSize,
461 Odr_int maximumRecordSize)
462 : m_named_result_sets(false), m_init_request(init_request),
463 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
464 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
465 m_preferredMessageSize(preferredMessageSize),
466 m_maximumRecordSize(maximumRecordSize)
469 yf::SessionShared::BackendClass::~BackendClass()
472 void yf::SessionShared::Rep::stat()
475 int no_instances = 0;
476 BackendClassMap::const_iterator it;
478 boost::mutex::scoped_lock lock(m_mutex_backend_map);
479 for (it = m_backend_map.begin(); it != m_backend_map.end(); it++)
481 BackendClassPtr bc = it->second;
483 BackendInstanceList::iterator bit = bc->m_backend_list.begin();
484 for (; bit != bc->m_backend_list.end(); bit++)
490 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
491 FrontendPtr frontend)
493 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
495 frontend->m_is_virtual = true;
496 frontend->m_init_options = *req->options;
499 boost::mutex::scoped_lock lock(m_mutex_backend_map);
500 BackendClassMap::const_iterator it;
501 it = m_backend_map.find(k);
502 if (it == m_backend_map.end())
504 BackendClassPtr b(new BackendClass(gdu->u.z3950,
508 m_preferredMessageSize,
509 m_maximumRecordSize));
510 m_backend_map[k] = b;
511 frontend->m_backend_class = b;
515 frontend->m_backend_class = it->second;
518 BackendClassPtr bc = frontend->m_backend_class;
521 // we only need to get init response from "first" target in
522 // backend class - the assumption being that init response is
524 if (bc->m_backend_list.size() == 0)
526 BackendInstancePtr backend = bc->create_backend(package);
528 bc->release_backend(backend);
531 yazpp_1::GDU init_response;
533 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
535 init_response = bc->m_init_response;
538 if (init_response.get())
540 Z_GDU *response_gdu = init_response.get();
541 mp::util::transfer_referenceId(odr, gdu->u.z3950,
542 response_gdu->u.z3950);
543 Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
544 Z_Options *server_options = init_res->options;
545 Z_Options *client_options = &frontend->m_init_options;
547 for (i = 0; i < 30; i++)
548 if (!ODR_MASK_GET(client_options, i))
549 ODR_MASK_CLEAR(server_options, i);
551 if (!m_preferredMessageSize ||
552 *init_res->preferredMessageSize > *req->preferredMessageSize)
553 *init_res->preferredMessageSize = *req->preferredMessageSize;
555 if (!m_maximumRecordSize ||
556 *init_res->maximumRecordSize > *req->maximumRecordSize)
557 *init_res->maximumRecordSize = *req->maximumRecordSize;
559 package.response() = init_response;
560 if (!*response_gdu->u.z3950->u.initResponse->result)
561 package.session().close();
566 odr.create_initResponse(
567 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
568 "session_shared: target closed connection during init");
569 *apdu->u.initResponse->result = 0;
570 package.response() = apdu;
571 package.session().close();
575 void yf::SessionShared::BackendSet::timestamp()
577 time(&m_time_last_use);
580 yf::SessionShared::BackendSet::BackendSet(
581 const std::string &result_set_id,
582 const Databases &databases,
583 const yazpp_1::Yaz_Z_Query &query,
584 Z_OtherInformation *additionalSearchInfo) :
585 m_result_set_id(result_set_id),
586 m_databases(databases), m_result_set_size(0), m_query(query)
589 mem_additionalSearchInfo = nmem_create();
590 additionalSearchInfoResponse = 0;
591 additionalSearchInfoRequest =
592 yaz_clone_z_OtherInformation(additionalSearchInfo,
593 mem_additionalSearchInfo);
596 yf::SessionShared::BackendSet::~BackendSet()
598 nmem_destroy(mem_additionalSearchInfo);
601 static int get_diagnostic(Z_DefaultDiagFormat *r)
603 return *r->condition;
606 bool yf::SessionShared::BackendSet::search(
607 mp::Package &frontend_package,
608 mp::Package &search_package,
609 const Z_APDU *frontend_apdu,
610 const BackendInstancePtr bp,
611 Z_Records **z_records)
614 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
615 Z_SearchRequest *req = apdu_req->u.searchRequest;
617 req->additionalSearchInfo = additionalSearchInfoRequest;
618 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
619 req->query = m_query.get_Z_Query();
621 req->num_databaseNames = m_databases.size();
622 req->databaseNames = (char**)
623 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
624 Databases::const_iterator it = m_databases.begin();
626 for (; it != m_databases.end(); it++)
627 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
629 if (frontend_apdu->which == Z_APDU_searchRequest)
630 req->preferredRecordSyntax =
631 frontend_apdu->u.searchRequest->preferredRecordSyntax;
633 search_package.request() = apdu_req;
635 search_package.move();
637 Z_GDU *gdu = search_package.response().get();
638 if (!search_package.session().is_closed()
639 && gdu && gdu->which == Z_GDU_Z3950
640 && gdu->u.z3950->which == Z_APDU_searchResponse)
642 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
643 *z_records = b_resp->records;
644 m_result_set_size = *b_resp->resultCount;
646 additionalSearchInfoResponse = yaz_clone_z_OtherInformation(
647 b_resp->additionalSearchInfo, mem_additionalSearchInfo);
651 const char *addinfo = "session_shared: "
652 "target closed connection during search";
653 if (frontend_apdu->which == Z_APDU_searchRequest)
654 f_apdu = odr.create_searchResponse(
655 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
656 else if (frontend_apdu->which == Z_APDU_presentRequest)
657 f_apdu = odr.create_presentResponse(
658 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
660 f_apdu = odr.create_close(
661 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
662 frontend_package.response() = f_apdu;
666 void yf::SessionShared::Frontend::override_set(
667 BackendInstancePtr &found_backend,
668 std::string &result_set_id,
669 const Databases &databases,
670 bool out_of_sessions)
672 BackendClassPtr bc = m_backend_class;
673 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
677 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
678 for (; it != bc->m_backend_list.end(); it++)
680 if (!(*it)->m_in_use)
682 BackendSetList::iterator set_it = (*it)->m_sets.begin();
683 for (; set_it != (*it)->m_sets.end(); set_it++)
685 if ((max_sets > 1 || (*set_it)->m_databases == databases)
688 now < (*set_it)->m_time_last_use ||
689 now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
692 result_set_id = (*set_it)->m_result_set_id;
693 found_backend->m_sets.erase(set_it);
699 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
701 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
704 if (bc->m_named_result_sets)
706 result_set_id = boost::io::str(
707 boost::format("%1%") %
708 found_backend->m_result_set_sequence);
709 found_backend->m_result_set_sequence++;
712 result_set_id = "default";
718 void yf::SessionShared::Frontend::get_set(mp::Package &package,
719 const Z_APDU *apdu_req,
720 const Databases &databases,
721 yazpp_1::Yaz_Z_Query &query,
722 BackendInstancePtr &found_backend,
723 BackendSetPtr &found_set)
725 bool session_restarted = false;
726 Z_OtherInformation *additionalSearchInfo = 0;
728 if (apdu_req->which == Z_APDU_searchRequest)
729 additionalSearchInfo = apdu_req->u.searchRequest->additionalSearchInfo;
732 std::string result_set_id;
733 bool out_of_sessions = false;
734 BackendClassPtr bc = m_backend_class;
736 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
738 if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
739 out_of_sessions = true;
741 if (m_p->m_optimize_search)
743 // look at each backend and see if we have a similar search
744 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
745 for (; it != bc->m_backend_list.end(); it++)
747 if (!(*it)->m_in_use)
749 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
750 for (; set_it != (*it)->m_sets.end(); set_it++)
752 // for real present request we don't care
753 // if additionalSearchInfo matches: same records
754 if ((*set_it)->m_databases == databases
755 && query.match(&(*set_it)->m_query)
756 && (apdu_req->which != Z_APDU_searchRequest ||
757 yaz_compare_z_OtherInformation(
758 additionalSearchInfo,
759 (*set_it)->additionalSearchInfoRequest)))
763 bc->use_backend(found_backend);
764 // found matching set. No need to search again
771 override_set(found_backend, result_set_id, databases, out_of_sessions);
773 bc->use_backend(found_backend);
777 // create a new backend set (and new set) if we're not out of sessions
778 if (!out_of_sessions)
779 found_backend = bc->create_backend(package);
785 const char *addinfo = 0;
788 addinfo = "session_shared: all sessions in use";
790 addinfo = "session_shared: could not create backend";
791 if (apdu_req->which == Z_APDU_searchRequest)
793 f_apdu = odr.create_searchResponse(
794 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
796 else if (apdu_req->which == Z_APDU_presentRequest)
798 f_apdu = odr.create_presentResponse(
799 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
803 f_apdu = odr.create_close(
804 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
806 package.response() = f_apdu;
809 if (bc->m_named_result_sets)
811 result_set_id = boost::io::str(
812 boost::format("%1%") % found_backend->m_result_set_sequence);
813 found_backend->m_result_set_sequence++;
816 result_set_id = "default";
818 found_backend->timestamp();
820 // we must search ...
821 BackendSetPtr new_set(new BackendSet(result_set_id,
823 additionalSearchInfo));
824 Z_Records *z_records = 0;
826 Package search_package(found_backend->m_session, package.origin());
827 search_package.copy_filter(package);
829 if (!new_set->search(package, search_package,
830 apdu_req, found_backend, &z_records))
832 bc->remove_backend(found_backend);
833 return; // search error
839 if (z_records->which == Z_Records_NSD)
842 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
844 else if (z_records->which == Z_Records_multipleNSD)
846 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
849 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
850 Z_DiagRec_defaultFormat)
852 condition = get_diagnostic(
853 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
857 if (m_p->m_restart && !session_restarted &&
858 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
860 package.log("session_shared", YLOG_LOG, "restart");
861 bc->remove_backend(found_backend);
862 session_restarted = true;
863 found_backend.reset();
871 if (apdu_req->which == Z_APDU_searchRequest)
873 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
875 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
876 *f_resp->searchStatus = Z_SearchResponse_none;
877 f_resp->records = z_records;
878 package.response() = f_apdu;
880 if (apdu_req->which == Z_APDU_presentRequest)
882 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
884 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
885 f_resp->records = z_records;
886 package.response() = f_apdu;
888 bc->release_backend(found_backend);
889 return; // search error
892 if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
894 package.log("session_shared", YLOG_LOG, "restart");
895 bc->remove_backend(found_backend);
896 session_restarted = true;
897 found_backend.reset();
902 found_set->timestamp();
903 found_backend->m_sets.push_back(found_set);
906 int yf::SessionShared::Frontend::result_set_ref(ODR o,
907 const Databases &databases,
914 case Z_RPNStructure_simple:
915 if (s->u.simple->which == Z_Operand_resultSetId)
917 const char *id = s->u.simple->u.resultSetId;
920 FrontendSets::iterator fset_it = m_frontend_sets.find(id);
921 if (fset_it == m_frontend_sets.end())
923 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
925 else if (fset_it->second->get_databases() != databases)
927 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
931 yazpp_1::Yaz_Z_Query query = fset_it->second->get_query();
932 Z_Query *q = yaz_copy_Z_Query(query.get_Z_Query(), o);
933 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
935 s->which = q->u.type_1->RPNStructure->which;
936 s->u.simple = q->u.type_1->RPNStructure->u.simple;
940 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
945 case Z_RPNStructure_complex:
946 ret = result_set_ref(o, databases, s->u.complex->s1, rset);
948 ret = result_set_ref(o, databases, s->u.complex->s2, rset);
954 void yf::SessionShared::Frontend::search(mp::Package &package,
957 Z_SearchRequest *req = apdu_req->u.searchRequest;
958 FrontendSets::iterator fset_it =
959 m_frontend_sets.find(req->resultSetName);
960 if (fset_it != m_frontend_sets.end())
962 // result set already exist
963 // if replace indicator is off: we return diagnostic if
964 // result set already exist.
965 if (*req->replaceIndicator == 0)
969 odr.create_searchResponse(
971 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
973 package.response() = apdu;
976 m_frontend_sets.erase(fset_it);
981 for (i = 0; i < req->num_databaseNames; i++)
982 databases.push_back(req->databaseNames[i]);
985 yazpp_1::Yaz_Z_Query query;
986 query.set_Z_Query(req->query);
988 Z_Query *q = query.get_Z_Query();
989 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
993 int diag = result_set_ref(odr, databases, q->u.type_1->RPNStructure,
998 odr.create_searchResponse(
1002 package.response() = apdu;
1005 query.set_Z_Query(q);
1008 BackendSetPtr found_set; // null
1009 BackendInstancePtr found_backend; // null
1011 get_set(package, apdu_req, databases, query, found_backend, found_set);
1016 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
1017 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
1018 *f_resp->resultCount = found_set->m_result_set_size;
1019 f_resp->additionalSearchInfo = found_set->additionalSearchInfoResponse;
1020 package.response() = f_apdu;
1022 FrontendSetPtr fset(new FrontendSet(databases, query));
1023 m_frontend_sets[req->resultSetName] = fset;
1025 m_backend_class->release_backend(found_backend);
1028 void yf::SessionShared::Frontend::present(mp::Package &package,
1032 Z_PresentRequest *req = apdu_req->u.presentRequest;
1034 FrontendSets::iterator fset_it =
1035 m_frontend_sets.find(req->resultSetId);
1037 if (fset_it == m_frontend_sets.end())
1040 odr.create_presentResponse(
1042 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
1044 package.response() = apdu;
1047 FrontendSetPtr fset = fset_it->second;
1049 Databases databases = fset->get_databases();
1050 yazpp_1::Yaz_Z_Query query = fset->get_query();
1052 BackendClassPtr bc = m_backend_class;
1053 BackendSetPtr found_set; // null
1054 BackendInstancePtr found_backend;
1056 get_set(package, apdu_req, databases, query, found_backend, found_set);
1060 Z_NamePlusRecordList *npr_res = 0;
1061 if (found_set->m_record_cache.lookup(odr, &npr_res,
1062 *req->resultSetStartPoint,
1063 *req->numberOfRecordsRequested,
1064 req->preferredRecordSyntax,
1065 req->recordComposition))
1067 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1068 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1070 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1071 " records in cache %p",
1072 *req->resultSetStartPoint,
1073 *req->numberOfRecordsRequested,
1074 &found_set->m_record_cache);
1076 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
1077 *f_resp->nextResultSetPosition =
1078 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
1079 // f_resp->presentStatus assumed OK.
1080 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
1081 f_resp->records->which = Z_Records_DBOSD;
1082 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
1083 package.response() = f_apdu_res;
1084 bc->release_backend(found_backend);
1088 found_backend->timestamp();
1090 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
1091 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
1092 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1093 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
1094 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
1095 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
1096 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1097 p_req->recordComposition = req->recordComposition;
1099 Package present_package(found_backend->m_session, package.origin());
1100 present_package.copy_filter(package);
1102 present_package.request() = p_apdu;
1104 present_package.move();
1106 Z_GDU *gdu = present_package.response().get();
1107 if (!present_package.session().is_closed()
1108 && gdu && gdu->which == Z_GDU_Z3950
1109 && gdu->u.z3950->which == Z_APDU_presentResponse)
1111 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
1112 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1113 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1115 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
1116 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
1117 f_resp->presentStatus= b_resp->presentStatus;
1118 f_resp->records = b_resp->records;
1119 f_resp->otherInfo = b_resp->otherInfo;
1120 package.response() = f_apdu_res;
1122 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
1125 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1126 " records to cache %p",
1127 *req->resultSetStartPoint,
1128 *f_resp->numberOfRecordsReturned,
1129 &found_set->m_record_cache);
1131 found_set->m_record_cache.add(
1133 b_resp->records->u.databaseOrSurDiagnostics,
1134 *req->resultSetStartPoint, p_req->recordComposition);
1136 bc->release_backend(found_backend);
1140 bc->remove_backend(found_backend);
1141 Z_APDU *f_apdu_res =
1142 odr.create_presentResponse(
1143 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1144 "session_shared: target closed connection during present");
1145 package.response() = f_apdu_res;
1149 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1152 BackendClassPtr bc = m_backend_class;
1153 BackendInstancePtr backend = bc->get_backend(frontend_package);
1157 Z_APDU *apdu = odr.create_scanResponse(
1158 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1159 "session_shared: could not create backend");
1160 frontend_package.response() = apdu;
1164 Package scan_package(backend->m_session, frontend_package.origin());
1165 backend->timestamp();
1166 scan_package.copy_filter(frontend_package);
1167 scan_package.request() = apdu_req;
1168 scan_package.move();
1169 frontend_package.response() = scan_package.response();
1170 if (scan_package.session().is_closed())
1172 frontend_package.session().close();
1173 bc->remove_backend(backend);
1176 bc->release_backend(backend);
1180 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1184 void yf::SessionShared::Worker::operator() (void)
1189 bool yf::SessionShared::BackendClass::expire_instances()
1193 boost::mutex::scoped_lock lock(m_mutex_backend_class);
1194 BackendInstanceList::iterator bit = m_backend_list.begin();
1195 while (bit != m_backend_list.end())
1197 time_t last_use = (*bit)->m_time_last_use;
1199 if ((*bit)->m_in_use)
1203 else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1206 (*bit)->m_close_package->response() = odr.create_close(
1207 0, Z_Close_lackOfActivity, 0);
1208 (*bit)->m_close_package->session().close();
1209 (*bit)->m_close_package->move();
1211 bit = m_backend_list.erase(bit);
1218 if (m_backend_list.empty())
1223 void yf::SessionShared::Rep::expire_classes()
1225 boost::mutex::scoped_lock lock(m_mutex_backend_map);
1226 BackendClassMap::iterator b_it = m_backend_map.begin();
1227 while (b_it != m_backend_map.end())
1229 if (b_it->second->expire_instances())
1231 m_backend_map.erase(b_it);
1232 b_it = m_backend_map.begin();
1239 void yf::SessionShared::Rep::expire()
1244 boost::xtime_get(&xt,
1245 #if BOOST_VERSION >= 105000
1251 xt.sec += m_session_ttl / 3;
1253 boost::mutex::scoped_lock lock(m_mutex);
1254 m_cond_expire_ready.timed_wait(lock, xt);
1263 yf::SessionShared::Rep::Rep()
1265 m_resultset_ttl = 30;
1266 m_resultset_max = 10;
1268 m_optimize_search = true;
1270 m_session_max = 100;
1271 m_preferredMessageSize = 0;
1272 m_maximumRecordSize = 0;
1276 yf::SessionShared::Rep::~Rep()
1279 boost::mutex::scoped_lock lock(m_mutex);
1281 m_cond_expire_ready.notify_all();
1286 void yf::SessionShared::Rep::start()
1288 yf::SessionShared::Worker w(this);
1289 m_thrds.add_thread(new boost::thread(w));
1292 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1296 yf::SessionShared::~SessionShared() {
1299 void yf::SessionShared::start() const
1304 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1308 yf::SessionShared::Frontend::~Frontend()
1312 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1314 boost::mutex::scoped_lock lock(m_mutex);
1316 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1320 it = m_clients.find(package.session());
1321 if (it == m_clients.end())
1324 if (!it->second->m_in_use)
1326 it->second->m_in_use = true;
1329 m_cond_session_ready.wait(lock);
1331 FrontendPtr f(new Frontend(this));
1332 m_clients[package.session()] = f;
1337 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1339 boost::mutex::scoped_lock lock(m_mutex);
1340 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1342 it = m_clients.find(package.session());
1343 if (it != m_clients.end())
1345 if (package.session().is_closed())
1347 m_clients.erase(it);
1351 it->second->m_in_use = false;
1353 m_cond_session_ready.notify_all();
1358 void yf::SessionShared::process(mp::Package &package) const
1360 FrontendPtr f = m_p->get_frontend(package);
1362 Z_GDU *gdu = package.request().get();
1364 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1365 Z_APDU_initRequest && !f->m_is_virtual)
1367 m_p->init(package, gdu, f);
1369 else if (!f->m_is_virtual)
1371 else if (gdu && gdu->which == Z_GDU_Z3950)
1373 Z_APDU *apdu = gdu->u.z3950;
1374 if (apdu->which == Z_APDU_initRequest)
1378 package.response() = odr.create_close(
1380 Z_Close_protocolError,
1383 package.session().close();
1385 else if (apdu->which == Z_APDU_close)
1389 package.response() = odr.create_close(
1391 Z_Close_peerAbort, "received close from client");
1392 package.session().close();
1394 else if (apdu->which == Z_APDU_searchRequest)
1396 f->search(package, apdu);
1398 else if (apdu->which == Z_APDU_presentRequest)
1400 f->present(package, apdu);
1402 else if (apdu->which == Z_APDU_scanRequest)
1404 f->scan(package, apdu);
1410 package.response() = odr.create_close(
1411 apdu, Z_Close_protocolError,
1412 "unsupported APDU in filter_session_shared");
1414 package.session().close();
1417 m_p->release_frontend(package);
1420 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1423 for (ptr = ptr->children; ptr; ptr = ptr->next)
1425 if (ptr->type != XML_ELEMENT_NODE)
1427 if (!strcmp((const char *) ptr->name, "resultset"))
1429 const struct _xmlAttr *attr;
1430 for (attr = ptr->properties; attr; attr = attr->next)
1432 if (!strcmp((const char *) attr->name, "ttl"))
1433 m_p->m_resultset_ttl =
1434 mp::xml::get_int(attr->children, 30);
1435 else if (!strcmp((const char *) attr->name, "max"))
1437 m_p->m_resultset_max =
1438 mp::xml::get_int(attr->children, 10);
1440 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1442 m_p->m_optimize_search =
1443 mp::xml::get_bool(attr->children, true);
1445 else if (!strcmp((const char *) attr->name, "restart"))
1447 m_p->m_restart = mp::xml::get_bool(attr->children, true);
1450 throw mp::filter::FilterException(
1451 "Bad attribute " + std::string((const char *)
1455 else if (!strcmp((const char *) ptr->name, "session"))
1457 const struct _xmlAttr *attr;
1458 for (attr = ptr->properties; attr; attr = attr->next)
1460 if (!strcmp((const char *) attr->name, "ttl"))
1461 m_p->m_session_ttl =
1462 mp::xml::get_int(attr->children, 90);
1463 else if (!strcmp((const char *) attr->name, "max"))
1464 m_p->m_session_max =
1465 mp::xml::get_int(attr->children, 100);
1467 throw mp::filter::FilterException(
1468 "Bad attribute " + std::string((const char *)
1472 else if (!strcmp((const char *) ptr->name, "init"))
1474 const struct _xmlAttr *attr;
1475 for (attr = ptr->properties; attr; attr = attr->next)
1477 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1478 m_p->m_maximumRecordSize =
1479 mp::xml::get_int(attr->children, 0);
1480 else if (!strcmp((const char *) attr->name,
1481 "preferred-message-size"))
1482 m_p->m_preferredMessageSize =
1483 mp::xml::get_int(attr->children, 0);
1485 throw mp::filter::FilterException(
1486 "Bad attribute " + std::string((const char *)
1492 throw mp::filter::FilterException("Bad element "
1493 + std::string((const char *)
1499 static mp::filter::Base* filter_creator()
1501 return new mp::filter::SessionShared;
1505 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1515 * c-file-style: "Stroustrup"
1516 * indent-tabs-mode: nil
1518 * vim: shiftwidth=4 tabstop=8 expandtab