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;
84 const std::string &result_set_id,
85 const Databases &databases,
86 const yazpp_1::Yaz_Z_Query &query);
88 Package &frontend_package,
89 Package &search_package,
90 const Z_APDU *apdu_req,
91 const BackendInstancePtr bp,
92 Z_Records **z_records);
94 // backend connection instance
95 class SessionShared::BackendInstance {
97 friend class BackendClass;
98 friend class BackendSet;
100 mp::Session m_session;
101 BackendSetList m_sets;
104 int m_result_set_sequence;
105 time_t m_time_last_use;
106 mp::Package * m_close_package;
110 // backends of some class (all with same InitKey)
111 class SessionShared::BackendClass : boost::noncopyable {
113 friend struct Frontend;
114 bool m_named_result_sets;
115 BackendInstanceList m_backend_list;
116 BackendInstancePtr create_backend(const Package &package);
117 void remove_backend(BackendInstancePtr b);
118 BackendInstancePtr get_backend(const Package &package);
119 void use_backend(BackendInstancePtr b);
120 void release_backend(BackendInstancePtr b);
122 yazpp_1::GDU m_init_request;
123 yazpp_1::GDU m_init_response;
124 boost::mutex m_mutex_backend_class;
126 time_t m_backend_set_ttl;
127 time_t m_backend_expiry_ttl;
128 size_t m_backend_set_max;
129 Odr_int m_preferredMessageSize;
130 Odr_int m_maximumRecordSize;
132 BackendClass(const yazpp_1::GDU &init_request,
136 Odr_int preferredRecordSize,
137 Odr_int maximumRecordSize);
140 // frontend result set
141 class SessionShared::FrontendSet {
142 Databases m_databases;
143 yazpp_1::Yaz_Z_Query m_query;
145 const Databases &get_databases();
146 const yazpp_1::Yaz_Z_Query &get_query();
148 const Databases &databases,
149 const yazpp_1::Yaz_Z_Query &query);
153 struct SessionShared::Frontend {
158 Z_Options m_init_options;
159 void search(Package &package, Z_APDU *apdu);
160 void present(Package &package, Z_APDU *apdu);
161 void scan(Package &package, Z_APDU *apdu);
163 int result_set_ref(ODR o,
164 const Databases &databases,
165 Z_RPNStructure *s, std::string &rset);
166 void get_set(mp::Package &package,
167 const Z_APDU *apdu_req,
168 const Databases &databases,
169 yazpp_1::Yaz_Z_Query &query,
170 BackendInstancePtr &found_backend,
171 BackendSetPtr &found_set);
172 void override_set(BackendInstancePtr &found_backend,
173 std::string &result_set_id,
174 const Databases &databases,
175 bool out_of_sessions);
178 BackendClassPtr m_backend_class;
179 FrontendSets m_frontend_sets;
182 class SessionShared::Rep {
183 friend class SessionShared;
184 friend struct Frontend;
186 FrontendPtr get_frontend(Package &package);
187 void release_frontend(Package &package);
192 void init(Package &package, const Z_GDU *gdu,
193 FrontendPtr frontend);
195 boost::mutex m_mutex;
196 boost::condition m_cond_session_ready;
197 std::map<mp::Session, FrontendPtr> m_clients;
199 BackendClassMap m_backend_map;
200 boost::mutex m_mutex_backend_map;
201 boost::thread_group m_thrds;
205 bool m_optimize_search;
208 Odr_int m_preferredMessageSize;
209 Odr_int m_maximumRecordSize;
214 yf::SessionShared::FrontendSet::FrontendSet(
215 const Databases &databases,
216 const yazpp_1::Yaz_Z_Query &query)
217 : m_databases(databases), m_query(query)
221 const yf::SessionShared::Databases &
222 yf::SessionShared::FrontendSet::get_databases()
227 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
232 yf::SessionShared::InitKey::InitKey(const InitKey &k)
234 m_odr = odr_createmem(ODR_ENCODE);
236 m_idAuthentication_size = k.m_idAuthentication_size;
237 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
238 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
239 m_idAuthentication_size);
241 m_otherInfo_size = k.m_otherInfo_size;
242 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
243 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
247 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
249 m_odr = odr_createmem(ODR_ENCODE);
251 Z_IdAuthentication *t = req->idAuthentication;
252 z_IdAuthentication(m_odr, &t, 1, 0);
253 m_idAuthentication_buf =
254 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
256 Z_OtherInformation *o = req->otherInfo;
257 z_OtherInformation(m_odr, &o, 1, 0);
258 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
261 yf::SessionShared::InitKey::~InitKey()
266 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
270 c = mp::util::memcmp2(
271 (void*) m_idAuthentication_buf, m_idAuthentication_size,
272 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
278 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
279 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
287 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
289 boost::mutex::scoped_lock lock(m_mutex_backend_class);
294 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
296 BackendInstanceList::iterator it = m_backend_list.begin();
298 while (it != m_backend_list.end())
303 (*it)->m_close_package->response() = odr.create_close(
304 0, Z_Close_lackOfActivity, 0);
305 (*it)->m_close_package->session().close();
306 (*it)->m_close_package->move();
308 it = m_backend_list.erase(it);
317 yf::SessionShared::BackendInstancePtr
318 yf::SessionShared::BackendClass::get_backend(
319 const mp::Package &frontend_package)
322 boost::mutex::scoped_lock lock(m_mutex_backend_class);
324 BackendInstanceList::const_iterator it = m_backend_list.begin();
326 BackendInstancePtr backend1; // null
328 for (; it != m_backend_list.end(); it++)
330 if (!(*it)->m_in_use)
333 || (*it)->m_sequence_this < backend1->m_sequence_this)
339 use_backend(backend1);
343 return create_backend(frontend_package);
346 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
348 backend->m_in_use = true;
349 backend->m_sequence_this = m_sequence_top++;
352 void yf::SessionShared::BackendInstance::timestamp()
355 time(&m_time_last_use);
358 yf::SessionShared::BackendInstance::~BackendInstance()
360 delete m_close_package;
363 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
364 const mp::Package &frontend_package)
366 BackendInstancePtr bp(new BackendInstance);
367 BackendInstancePtr null;
369 bp->m_close_package =
370 new mp::Package(bp->m_session, frontend_package.origin());
371 bp->m_close_package->copy_filter(frontend_package);
373 Package init_package(bp->m_session, frontend_package.origin());
375 init_package.copy_filter(frontend_package);
377 yazpp_1::GDU actual_init_request = m_init_request;
378 Z_GDU *init_pdu = actual_init_request.get();
380 assert(init_pdu->which == Z_GDU_Z3950);
381 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
383 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
384 ODR_MASK_ZERO(req->options);
386 ODR_MASK_SET(req->options, Z_Options_search);
387 ODR_MASK_SET(req->options, Z_Options_present);
388 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
389 ODR_MASK_SET(req->options, Z_Options_scan);
391 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
392 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
393 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
395 if (m_preferredMessageSize)
396 *req->preferredMessageSize = m_preferredMessageSize;
397 if (m_maximumRecordSize)
398 *req->maximumRecordSize = m_maximumRecordSize;
400 init_package.request() = init_pdu;
404 boost::mutex::scoped_lock lock(m_mutex_backend_class);
406 m_named_result_sets = false;
407 Z_GDU *gdu = init_package.response().get();
409 if (gdu && gdu->which == Z_GDU_Z3950
410 && gdu->u.z3950->which == Z_APDU_initResponse)
412 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
413 m_init_response = gdu->u.z3950;
414 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
416 m_named_result_sets = true;
418 if (*gdu->u.z3950->u.initResponse->result
419 && !init_package.session().is_closed())
422 time(&bp->m_time_last_use);
423 bp->m_sequence_this = 0;
424 bp->m_result_set_sequence = 0;
425 m_backend_list.push_back(bp);
431 yazpp_1::GDU empty_gdu;
432 m_init_response = empty_gdu;
435 if (!init_package.session().is_closed())
437 init_package.copy_filter(frontend_package);
438 init_package.session().close();
445 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
449 Odr_int preferredMessageSize,
450 Odr_int maximumRecordSize)
451 : m_named_result_sets(false), m_init_request(init_request),
452 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
453 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
454 m_preferredMessageSize(preferredMessageSize),
455 m_maximumRecordSize(maximumRecordSize)
458 yf::SessionShared::BackendClass::~BackendClass()
461 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
462 FrontendPtr frontend)
464 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
466 frontend->m_is_virtual = true;
467 frontend->m_init_options = *req->options;
470 boost::mutex::scoped_lock lock(m_mutex_backend_map);
471 BackendClassMap::const_iterator it;
472 it = m_backend_map.find(k);
473 if (it == m_backend_map.end())
475 BackendClassPtr b(new BackendClass(gdu->u.z3950,
479 m_preferredMessageSize,
480 m_maximumRecordSize));
481 m_backend_map[k] = b;
482 frontend->m_backend_class = b;
486 frontend->m_backend_class = it->second;
489 BackendClassPtr bc = frontend->m_backend_class;
492 // we only need to get init response from "first" target in
493 // backend class - the assumption being that init response is
495 if (bc->m_backend_list.size() == 0)
497 BackendInstancePtr backend = bc->create_backend(package);
499 bc->release_backend(backend);
502 yazpp_1::GDU init_response;
504 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
506 init_response = bc->m_init_response;
509 if (init_response.get())
511 Z_GDU *response_gdu = init_response.get();
512 mp::util::transfer_referenceId(odr, gdu->u.z3950,
513 response_gdu->u.z3950);
514 Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
515 Z_Options *server_options = init_res->options;
516 Z_Options *client_options = &frontend->m_init_options;
518 for (i = 0; i < 30; i++)
519 if (!ODR_MASK_GET(client_options, i))
520 ODR_MASK_CLEAR(server_options, i);
522 if (!m_preferredMessageSize ||
523 *init_res->preferredMessageSize > *req->preferredMessageSize)
524 *init_res->preferredMessageSize = *req->preferredMessageSize;
526 if (!m_maximumRecordSize ||
527 *init_res->maximumRecordSize > *req->maximumRecordSize)
528 *init_res->maximumRecordSize = *req->maximumRecordSize;
530 package.response() = init_response;
531 if (!*response_gdu->u.z3950->u.initResponse->result)
532 package.session().close();
537 odr.create_initResponse(
538 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
539 "session_shared: target closed connection during init");
540 *apdu->u.initResponse->result = 0;
541 package.response() = apdu;
542 package.session().close();
546 void yf::SessionShared::BackendSet::timestamp()
548 time(&m_time_last_use);
551 yf::SessionShared::BackendSet::BackendSet(
552 const std::string &result_set_id,
553 const Databases &databases,
554 const yazpp_1::Yaz_Z_Query &query) :
555 m_result_set_id(result_set_id),
556 m_databases(databases), m_result_set_size(0), m_query(query)
561 static int get_diagnostic(Z_DefaultDiagFormat *r)
563 return *r->condition;
566 bool yf::SessionShared::BackendSet::search(
567 mp::Package &frontend_package,
568 mp::Package &search_package,
569 const Z_APDU *frontend_apdu,
570 const BackendInstancePtr bp,
571 Z_Records **z_records)
574 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
575 Z_SearchRequest *req = apdu_req->u.searchRequest;
577 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
578 req->query = m_query.get_Z_Query();
580 req->num_databaseNames = m_databases.size();
581 req->databaseNames = (char**)
582 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
583 Databases::const_iterator it = m_databases.begin();
585 for (; it != m_databases.end(); it++)
586 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
588 if (frontend_apdu->which == Z_APDU_searchRequest)
589 req->preferredRecordSyntax =
590 frontend_apdu->u.searchRequest->preferredRecordSyntax;
592 search_package.request() = apdu_req;
594 search_package.move();
596 Z_GDU *gdu = search_package.response().get();
597 if (!search_package.session().is_closed()
598 && gdu && gdu->which == Z_GDU_Z3950
599 && gdu->u.z3950->which == Z_APDU_searchResponse)
601 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
602 *z_records = b_resp->records;
603 m_result_set_size = *b_resp->resultCount;
607 const char *addinfo = "session_shared: "
608 "target closed connection during search";
609 if (frontend_apdu->which == Z_APDU_searchRequest)
610 f_apdu = odr.create_searchResponse(
611 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
612 else if (frontend_apdu->which == Z_APDU_presentRequest)
613 f_apdu = odr.create_presentResponse(
614 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
616 f_apdu = odr.create_close(
617 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
618 frontend_package.response() = f_apdu;
622 void yf::SessionShared::Frontend::override_set(
623 BackendInstancePtr &found_backend,
624 std::string &result_set_id,
625 const Databases &databases,
626 bool out_of_sessions)
628 BackendClassPtr bc = m_backend_class;
629 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
633 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
634 for (; it != bc->m_backend_list.end(); it++)
636 if (!(*it)->m_in_use)
638 BackendSetList::iterator set_it = (*it)->m_sets.begin();
639 for (; set_it != (*it)->m_sets.end(); set_it++)
641 if ((max_sets > 1 || (*set_it)->m_databases == databases)
644 now < (*set_it)->m_time_last_use ||
645 now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
648 result_set_id = (*set_it)->m_result_set_id;
649 found_backend->m_sets.erase(set_it);
655 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
657 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
660 if (bc->m_named_result_sets)
662 result_set_id = boost::io::str(
663 boost::format("%1%") %
664 found_backend->m_result_set_sequence);
665 found_backend->m_result_set_sequence++;
668 result_set_id = "default";
674 void yf::SessionShared::Frontend::get_set(mp::Package &package,
675 const Z_APDU *apdu_req,
676 const Databases &databases,
677 yazpp_1::Yaz_Z_Query &query,
678 BackendInstancePtr &found_backend,
679 BackendSetPtr &found_set)
681 bool session_restarted = false;
684 std::string result_set_id;
685 bool out_of_sessions = false;
686 BackendClassPtr bc = m_backend_class;
688 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
690 if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
691 out_of_sessions = true;
693 if (m_p->m_optimize_search)
695 // look at each backend and see if we have a similar search
696 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
697 for (; it != bc->m_backend_list.end(); it++)
699 if (!(*it)->m_in_use)
701 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
702 for (; set_it != (*it)->m_sets.end(); set_it++)
704 if ((*set_it)->m_databases == databases
705 && query.match(&(*set_it)->m_query))
709 bc->use_backend(found_backend);
710 // found matching set. No need to search again
717 override_set(found_backend, result_set_id, databases, out_of_sessions);
719 bc->use_backend(found_backend);
723 // create a new backend set (and new set) if we're not out of sessions
724 if (!out_of_sessions)
725 found_backend = bc->create_backend(package);
731 const char *addinfo = 0;
734 addinfo = "session_shared: all sessions in use";
736 addinfo = "session_shared: could not create backend";
737 if (apdu_req->which == Z_APDU_searchRequest)
739 f_apdu = odr.create_searchResponse(
740 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
742 else if (apdu_req->which == Z_APDU_presentRequest)
744 f_apdu = odr.create_presentResponse(
745 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
749 f_apdu = odr.create_close(
750 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
752 package.response() = f_apdu;
755 if (bc->m_named_result_sets)
757 result_set_id = boost::io::str(
758 boost::format("%1%") % found_backend->m_result_set_sequence);
759 found_backend->m_result_set_sequence++;
762 result_set_id = "default";
764 found_backend->timestamp();
766 // we must search ...
767 BackendSetPtr new_set(new BackendSet(result_set_id,
769 Z_Records *z_records = 0;
771 Package search_package(found_backend->m_session, package.origin());
772 search_package.copy_filter(package);
774 if (!new_set->search(package, search_package,
775 apdu_req, found_backend, &z_records))
777 bc->remove_backend(found_backend);
778 return; // search error
784 if (z_records->which == Z_Records_NSD)
787 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
789 else if (z_records->which == Z_Records_multipleNSD)
791 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
794 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
795 Z_DiagRec_defaultFormat)
797 condition = get_diagnostic(
798 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
802 if (m_p->m_restart && !session_restarted &&
803 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
805 package.log("session_shared", YLOG_LOG, "restart");
806 bc->remove_backend(found_backend);
807 session_restarted = true;
808 found_backend.reset();
816 if (apdu_req->which == Z_APDU_searchRequest)
818 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
820 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
821 *f_resp->searchStatus = Z_SearchResponse_none;
822 f_resp->records = z_records;
823 package.response() = f_apdu;
825 if (apdu_req->which == Z_APDU_presentRequest)
827 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
829 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
830 f_resp->records = z_records;
831 package.response() = f_apdu;
833 bc->release_backend(found_backend);
834 return; // search error
837 if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
839 package.log("session_shared", YLOG_LOG, "restart");
840 bc->remove_backend(found_backend);
841 session_restarted = true;
842 found_backend.reset();
847 found_set->timestamp();
848 found_backend->m_sets.push_back(found_set);
851 int yf::SessionShared::Frontend::result_set_ref(ODR o,
852 const Databases &databases,
859 case Z_RPNStructure_simple:
860 if (s->u.simple->which == Z_Operand_resultSetId)
862 const char *id = s->u.simple->u.resultSetId;
865 FrontendSets::iterator fset_it = m_frontend_sets.find(id);
866 if (fset_it == m_frontend_sets.end())
868 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
870 else if (fset_it->second->get_databases() != databases)
872 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
876 yazpp_1::Yaz_Z_Query query = fset_it->second->get_query();
877 Z_Query *q = yaz_copy_Z_Query(query.get_Z_Query(), o);
878 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
880 s->which = q->u.type_1->RPNStructure->which;
881 s->u.simple = q->u.type_1->RPNStructure->u.simple;
885 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
890 case Z_RPNStructure_complex:
891 ret = result_set_ref(o, databases, s->u.complex->s1, rset);
893 ret = result_set_ref(o, databases, s->u.complex->s2, rset);
899 void yf::SessionShared::Frontend::search(mp::Package &package,
902 Z_SearchRequest *req = apdu_req->u.searchRequest;
903 FrontendSets::iterator fset_it =
904 m_frontend_sets.find(req->resultSetName);
905 if (fset_it != m_frontend_sets.end())
907 // result set already exist
908 // if replace indicator is off: we return diagnostic if
909 // result set already exist.
910 if (*req->replaceIndicator == 0)
914 odr.create_searchResponse(
916 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
918 package.response() = apdu;
921 m_frontend_sets.erase(fset_it);
926 for (i = 0; i < req->num_databaseNames; i++)
927 databases.push_back(req->databaseNames[i]);
930 yazpp_1::Yaz_Z_Query query;
931 query.set_Z_Query(req->query);
933 Z_Query *q = query.get_Z_Query();
934 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
938 int diag = result_set_ref(odr, databases, q->u.type_1->RPNStructure,
943 odr.create_searchResponse(
947 package.response() = apdu;
950 query.set_Z_Query(q);
953 BackendSetPtr found_set; // null
954 BackendInstancePtr found_backend; // null
956 get_set(package, apdu_req, databases, query, found_backend, found_set);
961 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
962 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
963 *f_resp->resultCount = found_set->m_result_set_size;
964 package.response() = f_apdu;
966 FrontendSetPtr fset(new FrontendSet(databases, query));
967 m_frontend_sets[req->resultSetName] = fset;
969 m_backend_class->release_backend(found_backend);
972 void yf::SessionShared::Frontend::present(mp::Package &package,
976 Z_PresentRequest *req = apdu_req->u.presentRequest;
978 FrontendSets::iterator fset_it =
979 m_frontend_sets.find(req->resultSetId);
981 if (fset_it == m_frontend_sets.end())
984 odr.create_presentResponse(
986 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
988 package.response() = apdu;
991 FrontendSetPtr fset = fset_it->second;
993 Databases databases = fset->get_databases();
994 yazpp_1::Yaz_Z_Query query = fset->get_query();
996 BackendClassPtr bc = m_backend_class;
997 BackendSetPtr found_set; // null
998 BackendInstancePtr found_backend;
1000 get_set(package, apdu_req, databases, query, found_backend, found_set);
1004 Z_NamePlusRecordList *npr_res = 0;
1005 if (found_set->m_record_cache.lookup(odr, &npr_res,
1006 *req->resultSetStartPoint,
1007 *req->numberOfRecordsRequested,
1008 req->preferredRecordSyntax,
1009 req->recordComposition))
1011 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1012 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1014 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1015 " records in cache %p",
1016 *req->resultSetStartPoint,
1017 *req->numberOfRecordsRequested,
1018 &found_set->m_record_cache);
1020 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
1021 *f_resp->nextResultSetPosition =
1022 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
1023 // f_resp->presentStatus assumed OK.
1024 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
1025 f_resp->records->which = Z_Records_DBOSD;
1026 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
1027 package.response() = f_apdu_res;
1028 bc->release_backend(found_backend);
1032 found_backend->timestamp();
1034 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
1035 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
1036 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1037 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
1038 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
1039 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
1040 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1041 p_req->recordComposition = req->recordComposition;
1043 Package present_package(found_backend->m_session, package.origin());
1044 present_package.copy_filter(package);
1046 present_package.request() = p_apdu;
1048 present_package.move();
1050 Z_GDU *gdu = present_package.response().get();
1051 if (!present_package.session().is_closed()
1052 && gdu && gdu->which == Z_GDU_Z3950
1053 && gdu->u.z3950->which == Z_APDU_presentResponse)
1055 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
1056 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1057 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1059 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
1060 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
1061 f_resp->presentStatus= b_resp->presentStatus;
1062 f_resp->records = b_resp->records;
1063 f_resp->otherInfo = b_resp->otherInfo;
1064 package.response() = f_apdu_res;
1066 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
1068 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1069 " records to cache %p",
1070 *req->resultSetStartPoint,
1071 *f_resp->numberOfRecordsReturned,
1072 &found_set->m_record_cache);
1073 found_set->m_record_cache.add(
1075 b_resp->records->u.databaseOrSurDiagnostics,
1076 *req->resultSetStartPoint,
1077 *f_resp->numberOfRecordsReturned);
1079 bc->release_backend(found_backend);
1083 bc->remove_backend(found_backend);
1084 Z_APDU *f_apdu_res =
1085 odr.create_presentResponse(
1086 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1087 "session_shared: target closed connection during present");
1088 package.response() = f_apdu_res;
1092 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1095 BackendClassPtr bc = m_backend_class;
1096 BackendInstancePtr backend = bc->get_backend(frontend_package);
1100 Z_APDU *apdu = odr.create_scanResponse(
1101 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1102 "session_shared: could not create backend");
1103 frontend_package.response() = apdu;
1107 Package scan_package(backend->m_session, frontend_package.origin());
1108 backend->timestamp();
1109 scan_package.copy_filter(frontend_package);
1110 scan_package.request() = apdu_req;
1111 scan_package.move();
1112 frontend_package.response() = scan_package.response();
1113 if (scan_package.session().is_closed())
1115 frontend_package.session().close();
1116 bc->remove_backend(backend);
1119 bc->release_backend(backend);
1123 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1127 void yf::SessionShared::Worker::operator() (void)
1132 void yf::SessionShared::BackendClass::expire_class()
1136 boost::mutex::scoped_lock lock(m_mutex_backend_class);
1137 BackendInstanceList::iterator bit = m_backend_list.begin();
1138 while (bit != m_backend_list.end())
1140 time_t last_use = (*bit)->m_time_last_use;
1142 if ((*bit)->m_in_use)
1146 else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1149 (*bit)->m_close_package->response() = odr.create_close(
1150 0, Z_Close_lackOfActivity, 0);
1151 (*bit)->m_close_package->session().close();
1152 (*bit)->m_close_package->move();
1154 bit = m_backend_list.erase(bit);
1163 void yf::SessionShared::Rep::expire()
1168 boost::xtime_get(&xt,
1169 #if BOOST_VERSION >= 105000
1175 xt.sec += m_session_ttl / 3;
1176 boost::thread::sleep(xt);
1178 BackendClassMap::const_iterator b_it = m_backend_map.begin();
1179 for (; b_it != m_backend_map.end(); b_it++)
1180 b_it->second->expire_class();
1184 yf::SessionShared::Rep::Rep()
1186 m_resultset_ttl = 30;
1187 m_resultset_max = 10;
1189 m_optimize_search = true;
1191 m_session_max = 100;
1192 m_preferredMessageSize = 0;
1193 m_maximumRecordSize = 0;
1196 void yf::SessionShared::Rep::start()
1198 yf::SessionShared::Worker w(this);
1199 m_thrds.add_thread(new boost::thread(w));
1202 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1206 yf::SessionShared::~SessionShared() {
1209 void yf::SessionShared::start() const
1214 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1218 yf::SessionShared::Frontend::~Frontend()
1222 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1224 boost::mutex::scoped_lock lock(m_mutex);
1226 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1230 it = m_clients.find(package.session());
1231 if (it == m_clients.end())
1234 if (!it->second->m_in_use)
1236 it->second->m_in_use = true;
1239 m_cond_session_ready.wait(lock);
1241 FrontendPtr f(new Frontend(this));
1242 m_clients[package.session()] = f;
1247 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1249 boost::mutex::scoped_lock lock(m_mutex);
1250 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1252 it = m_clients.find(package.session());
1253 if (it != m_clients.end())
1255 if (package.session().is_closed())
1257 m_clients.erase(it);
1261 it->second->m_in_use = false;
1263 m_cond_session_ready.notify_all();
1268 void yf::SessionShared::process(mp::Package &package) const
1270 FrontendPtr f = m_p->get_frontend(package);
1272 Z_GDU *gdu = package.request().get();
1274 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1275 Z_APDU_initRequest && !f->m_is_virtual)
1277 m_p->init(package, gdu, f);
1279 else if (!f->m_is_virtual)
1281 else if (gdu && gdu->which == Z_GDU_Z3950)
1283 Z_APDU *apdu = gdu->u.z3950;
1284 if (apdu->which == Z_APDU_initRequest)
1288 package.response() = odr.create_close(
1290 Z_Close_protocolError,
1293 package.session().close();
1295 else if (apdu->which == Z_APDU_close)
1299 package.response() = odr.create_close(
1301 Z_Close_peerAbort, "received close from client");
1302 package.session().close();
1304 else if (apdu->which == Z_APDU_searchRequest)
1306 f->search(package, apdu);
1308 else if (apdu->which == Z_APDU_presentRequest)
1310 f->present(package, apdu);
1312 else if (apdu->which == Z_APDU_scanRequest)
1314 f->scan(package, apdu);
1320 package.response() = odr.create_close(
1321 apdu, Z_Close_protocolError,
1322 "unsupported APDU in filter_session_shared");
1324 package.session().close();
1327 m_p->release_frontend(package);
1330 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1333 for (ptr = ptr->children; ptr; ptr = ptr->next)
1335 if (ptr->type != XML_ELEMENT_NODE)
1337 if (!strcmp((const char *) ptr->name, "resultset"))
1339 const struct _xmlAttr *attr;
1340 for (attr = ptr->properties; attr; attr = attr->next)
1342 if (!strcmp((const char *) attr->name, "ttl"))
1343 m_p->m_resultset_ttl =
1344 mp::xml::get_int(attr->children, 30);
1345 else if (!strcmp((const char *) attr->name, "max"))
1347 m_p->m_resultset_max =
1348 mp::xml::get_int(attr->children, 10);
1350 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1352 m_p->m_optimize_search =
1353 mp::xml::get_bool(attr->children, true);
1355 else if (!strcmp((const char *) attr->name, "restart"))
1357 m_p->m_restart = mp::xml::get_bool(attr->children, true);
1360 throw mp::filter::FilterException(
1361 "Bad attribute " + std::string((const char *)
1365 else if (!strcmp((const char *) ptr->name, "session"))
1367 const struct _xmlAttr *attr;
1368 for (attr = ptr->properties; attr; attr = attr->next)
1370 if (!strcmp((const char *) attr->name, "ttl"))
1371 m_p->m_session_ttl =
1372 mp::xml::get_int(attr->children, 90);
1373 else if (!strcmp((const char *) attr->name, "max"))
1374 m_p->m_session_max =
1375 mp::xml::get_int(attr->children, 100);
1377 throw mp::filter::FilterException(
1378 "Bad attribute " + std::string((const char *)
1382 else if (!strcmp((const char *) ptr->name, "init"))
1384 const struct _xmlAttr *attr;
1385 for (attr = ptr->properties; attr; attr = attr->next)
1387 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1388 m_p->m_maximumRecordSize =
1389 mp::xml::get_int(attr->children, 0);
1390 else if (!strcmp((const char *) attr->name,
1391 "preferred-message-size"))
1392 m_p->m_preferredMessageSize =
1393 mp::xml::get_int(attr->children, 0);
1395 throw mp::filter::FilterException(
1396 "Bad attribute " + std::string((const char *)
1402 throw mp::filter::FilterException("Bad element "
1403 + std::string((const char *)
1409 static mp::filter::Base* filter_creator()
1411 return new mp::filter::SessionShared;
1415 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1425 * c-file-style: "Stroustrup"
1426 * indent-tabs-mode: nil
1428 * vim: shiftwidth=4 tabstop=8 expandtab