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"
36 #include <yaz/otherinfo.h>
37 #include <yaz/diagbib1.h>
38 #include <yazpp/z-query.h>
39 #include <yazpp/record-cache.h>
44 namespace mp = metaproxy_1;
45 namespace yf = metaproxy_1::filter;
47 namespace metaproxy_1 {
50 // key for session.. We'll only share sessions with same InitKey
51 class SessionShared::InitKey {
53 bool operator < (const SessionShared::InitKey &k) const;
54 InitKey(Z_InitRequest *req);
55 InitKey(const InitKey &);
58 char *m_idAuthentication_buf;
59 int m_idAuthentication_size;
60 char *m_otherInfo_buf;
64 // worker thread .. for expiry of sessions
65 class SessionShared::Worker {
67 Worker(SessionShared::Rep *rep);
68 void operator() (void);
70 SessionShared::Rep *m_p;
73 class SessionShared::BackendSet {
75 std::string m_result_set_id;
76 Databases m_databases;
77 Odr_int m_result_set_size;
78 yazpp_1::Yaz_Z_Query m_query;
79 time_t m_time_last_use;
81 yazpp_1::RecordCache m_record_cache;
83 const std::string &result_set_id,
84 const Databases &databases,
85 const yazpp_1::Yaz_Z_Query &query);
87 Package &frontend_package,
88 Package &search_package,
89 const Z_APDU *apdu_req,
90 const BackendInstancePtr bp,
91 Z_Records **z_records);
93 // backend connection instance
94 class SessionShared::BackendInstance {
96 friend class BackendClass;
97 friend class BackendSet;
99 mp::Session m_session;
100 BackendSetList m_sets;
103 int m_result_set_sequence;
104 time_t m_time_last_use;
105 mp::Package * m_close_package;
109 // backends of some class (all with same InitKey)
110 class SessionShared::BackendClass : boost::noncopyable {
112 friend struct Frontend;
113 bool m_named_result_sets;
114 BackendInstanceList m_backend_list;
115 BackendInstancePtr create_backend(const Package &package);
116 void remove_backend(BackendInstancePtr b);
117 BackendInstancePtr get_backend(const Package &package);
118 void use_backend(BackendInstancePtr b);
119 void release_backend(BackendInstancePtr b);
121 yazpp_1::GDU m_init_request;
122 yazpp_1::GDU m_init_response;
123 boost::mutex m_mutex_backend_class;
125 time_t m_backend_set_ttl;
126 time_t m_backend_expiry_ttl;
127 size_t m_backend_set_max;
128 Odr_int m_preferredMessageSize;
129 Odr_int m_maximumRecordSize;
131 BackendClass(const yazpp_1::GDU &init_request,
135 Odr_int preferredRecordSize,
136 Odr_int maximumRecordSize);
139 // frontend result set
140 class SessionShared::FrontendSet {
141 Databases m_databases;
142 yazpp_1::Yaz_Z_Query m_query;
144 const Databases &get_databases();
145 const yazpp_1::Yaz_Z_Query &get_query();
147 const Databases &databases,
148 const yazpp_1::Yaz_Z_Query &query);
152 struct SessionShared::Frontend {
157 Z_Options m_init_options;
158 void search(Package &package, Z_APDU *apdu);
159 void present(Package &package, Z_APDU *apdu);
160 void scan(Package &package, Z_APDU *apdu);
162 void get_set(mp::Package &package,
163 const Z_APDU *apdu_req,
164 const Databases &databases,
165 yazpp_1::Yaz_Z_Query &query,
166 BackendInstancePtr &found_backend,
167 BackendSetPtr &found_set);
168 void override_set(BackendInstancePtr &found_backend,
169 std::string &result_set_id,
170 const Databases &databases,
171 bool out_of_sessions);
174 BackendClassPtr m_backend_class;
175 FrontendSets m_frontend_sets;
178 class SessionShared::Rep {
179 friend class SessionShared;
180 friend struct Frontend;
182 FrontendPtr get_frontend(Package &package);
183 void release_frontend(Package &package);
188 void init(Package &package, const Z_GDU *gdu,
189 FrontendPtr frontend);
191 boost::mutex m_mutex;
192 boost::condition m_cond_session_ready;
193 std::map<mp::Session, FrontendPtr> m_clients;
195 BackendClassMap m_backend_map;
196 boost::mutex m_mutex_backend_map;
197 boost::thread_group m_thrds;
201 bool m_optimize_search;
204 Odr_int m_preferredMessageSize;
205 Odr_int m_maximumRecordSize;
210 yf::SessionShared::FrontendSet::FrontendSet(
211 const Databases &databases,
212 const yazpp_1::Yaz_Z_Query &query)
213 : m_databases(databases), m_query(query)
217 const yf::SessionShared::Databases &
218 yf::SessionShared::FrontendSet::get_databases()
223 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
228 yf::SessionShared::InitKey::InitKey(const InitKey &k)
230 m_odr = odr_createmem(ODR_ENCODE);
232 m_idAuthentication_size = k.m_idAuthentication_size;
233 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
234 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
235 m_idAuthentication_size);
237 m_otherInfo_size = k.m_otherInfo_size;
238 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
239 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
243 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
245 m_odr = odr_createmem(ODR_ENCODE);
247 Z_IdAuthentication *t = req->idAuthentication;
248 z_IdAuthentication(m_odr, &t, 1, 0);
249 m_idAuthentication_buf =
250 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
252 Z_OtherInformation *o = req->otherInfo;
253 z_OtherInformation(m_odr, &o, 1, 0);
254 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
257 yf::SessionShared::InitKey::~InitKey()
262 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
266 c = mp::util::memcmp2(
267 (void*) m_idAuthentication_buf, m_idAuthentication_size,
268 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
274 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
275 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
283 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
285 boost::mutex::scoped_lock lock(m_mutex_backend_class);
290 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
292 BackendInstanceList::iterator it = m_backend_list.begin();
294 while (it != m_backend_list.end())
299 (*it)->m_close_package->response() = odr.create_close(
300 0, Z_Close_lackOfActivity, 0);
301 (*it)->m_close_package->session().close();
302 (*it)->m_close_package->move();
304 it = m_backend_list.erase(it);
313 yf::SessionShared::BackendInstancePtr
314 yf::SessionShared::BackendClass::get_backend(
315 const mp::Package &frontend_package)
318 boost::mutex::scoped_lock lock(m_mutex_backend_class);
320 BackendInstanceList::const_iterator it = m_backend_list.begin();
322 BackendInstancePtr backend1; // null
324 for (; it != m_backend_list.end(); it++)
326 if (!(*it)->m_in_use)
329 || (*it)->m_sequence_this < backend1->m_sequence_this)
335 use_backend(backend1);
339 return create_backend(frontend_package);
342 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
344 backend->m_in_use = true;
345 backend->m_sequence_this = m_sequence_top++;
348 void yf::SessionShared::BackendInstance::timestamp()
351 time(&m_time_last_use);
354 yf::SessionShared::BackendInstance::~BackendInstance()
356 delete m_close_package;
359 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
360 const mp::Package &frontend_package)
362 BackendInstancePtr bp(new BackendInstance);
363 BackendInstancePtr null;
365 bp->m_close_package =
366 new mp::Package(bp->m_session, frontend_package.origin());
367 bp->m_close_package->copy_filter(frontend_package);
369 Package init_package(bp->m_session, frontend_package.origin());
371 init_package.copy_filter(frontend_package);
373 yazpp_1::GDU actual_init_request = m_init_request;
374 Z_GDU *init_pdu = actual_init_request.get();
376 assert(init_pdu->which == Z_GDU_Z3950);
377 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
379 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
380 ODR_MASK_ZERO(req->options);
382 ODR_MASK_SET(req->options, Z_Options_search);
383 ODR_MASK_SET(req->options, Z_Options_present);
384 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
385 ODR_MASK_SET(req->options, Z_Options_scan);
387 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
388 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
389 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
391 if (m_preferredMessageSize)
392 *req->preferredMessageSize = m_preferredMessageSize;
393 if (m_maximumRecordSize)
394 *req->maximumRecordSize = m_maximumRecordSize;
396 init_package.request() = init_pdu;
400 boost::mutex::scoped_lock lock(m_mutex_backend_class);
402 m_named_result_sets = false;
403 Z_GDU *gdu = init_package.response().get();
405 if (gdu && gdu->which == Z_GDU_Z3950
406 && gdu->u.z3950->which == Z_APDU_initResponse)
408 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
409 m_init_response = gdu->u.z3950;
410 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
412 m_named_result_sets = true;
414 if (*gdu->u.z3950->u.initResponse->result
415 && !init_package.session().is_closed())
418 time(&bp->m_time_last_use);
419 bp->m_sequence_this = 0;
420 bp->m_result_set_sequence = 0;
421 m_backend_list.push_back(bp);
427 yazpp_1::GDU empty_gdu;
428 m_init_response = empty_gdu;
431 if (!init_package.session().is_closed())
433 init_package.copy_filter(frontend_package);
434 init_package.session().close();
441 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
445 Odr_int preferredMessageSize,
446 Odr_int maximumRecordSize)
447 : m_named_result_sets(false), m_init_request(init_request),
448 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
449 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
450 m_preferredMessageSize(preferredMessageSize),
451 m_maximumRecordSize(maximumRecordSize)
454 yf::SessionShared::BackendClass::~BackendClass()
457 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
458 FrontendPtr frontend)
460 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
462 frontend->m_is_virtual = true;
463 frontend->m_init_options = *req->options;
466 boost::mutex::scoped_lock lock(m_mutex_backend_map);
467 BackendClassMap::const_iterator it;
468 it = m_backend_map.find(k);
469 if (it == m_backend_map.end())
471 BackendClassPtr b(new BackendClass(gdu->u.z3950,
475 m_preferredMessageSize,
476 m_maximumRecordSize));
477 m_backend_map[k] = b;
478 frontend->m_backend_class = b;
482 frontend->m_backend_class = it->second;
485 BackendClassPtr bc = frontend->m_backend_class;
488 // we only need to get init response from "first" target in
489 // backend class - the assumption being that init response is
491 if (bc->m_backend_list.size() == 0)
493 BackendInstancePtr backend = bc->create_backend(package);
495 bc->release_backend(backend);
498 yazpp_1::GDU init_response;
500 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
502 init_response = bc->m_init_response;
505 if (init_response.get())
507 Z_GDU *response_gdu = init_response.get();
508 mp::util::transfer_referenceId(odr, gdu->u.z3950,
509 response_gdu->u.z3950);
510 Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
511 Z_Options *server_options = init_res->options;
512 Z_Options *client_options = &frontend->m_init_options;
514 for (i = 0; i < 30; i++)
515 if (!ODR_MASK_GET(client_options, i))
516 ODR_MASK_CLEAR(server_options, i);
518 if (!m_preferredMessageSize ||
519 *init_res->preferredMessageSize > *req->preferredMessageSize)
520 *init_res->preferredMessageSize = *req->preferredMessageSize;
522 if (!m_maximumRecordSize ||
523 *init_res->maximumRecordSize > *req->maximumRecordSize)
524 *init_res->maximumRecordSize = *req->maximumRecordSize;
526 package.response() = init_response;
527 if (!*response_gdu->u.z3950->u.initResponse->result)
528 package.session().close();
533 odr.create_initResponse(
534 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
535 "session_shared: target closed connection during init");
536 *apdu->u.initResponse->result = 0;
537 package.response() = apdu;
538 package.session().close();
542 void yf::SessionShared::BackendSet::timestamp()
544 time(&m_time_last_use);
547 yf::SessionShared::BackendSet::BackendSet(
548 const std::string &result_set_id,
549 const Databases &databases,
550 const yazpp_1::Yaz_Z_Query &query) :
551 m_result_set_id(result_set_id),
552 m_databases(databases), m_result_set_size(0), m_query(query)
557 static int get_diagnostic(Z_DefaultDiagFormat *r)
559 return *r->condition;
562 bool yf::SessionShared::BackendSet::search(
563 mp::Package &frontend_package,
564 mp::Package &search_package,
565 const Z_APDU *frontend_apdu,
566 const BackendInstancePtr bp,
567 Z_Records **z_records)
570 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
571 Z_SearchRequest *req = apdu_req->u.searchRequest;
573 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
574 req->query = m_query.get_Z_Query();
576 req->num_databaseNames = m_databases.size();
577 req->databaseNames = (char**)
578 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
579 Databases::const_iterator it = m_databases.begin();
581 for (; it != m_databases.end(); it++)
582 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
584 if (frontend_apdu->which == Z_APDU_searchRequest)
585 req->preferredRecordSyntax =
586 frontend_apdu->u.searchRequest->preferredRecordSyntax;
588 search_package.request() = apdu_req;
590 search_package.move();
592 Z_GDU *gdu = search_package.response().get();
593 if (!search_package.session().is_closed()
594 && gdu && gdu->which == Z_GDU_Z3950
595 && gdu->u.z3950->which == Z_APDU_searchResponse)
597 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
598 *z_records = b_resp->records;
599 m_result_set_size = *b_resp->resultCount;
603 const char *addinfo = "session_shared: "
604 "target closed connection during search";
605 if (frontend_apdu->which == Z_APDU_searchRequest)
606 f_apdu = odr.create_searchResponse(
607 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
608 else if (frontend_apdu->which == Z_APDU_presentRequest)
609 f_apdu = odr.create_presentResponse(
610 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
612 f_apdu = odr.create_close(
613 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
614 frontend_package.response() = f_apdu;
618 void yf::SessionShared::Frontend::override_set(
619 BackendInstancePtr &found_backend,
620 std::string &result_set_id,
621 const Databases &databases,
622 bool out_of_sessions)
624 BackendClassPtr bc = m_backend_class;
625 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
629 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
630 for (; it != bc->m_backend_list.end(); it++)
632 if (!(*it)->m_in_use)
634 BackendSetList::iterator set_it = (*it)->m_sets.begin();
635 for (; set_it != (*it)->m_sets.end(); set_it++)
637 if ((max_sets > 1 || (*set_it)->m_databases == databases)
640 now < (*set_it)->m_time_last_use ||
641 now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
644 result_set_id = (*set_it)->m_result_set_id;
645 found_backend->m_sets.erase(set_it);
651 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
653 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
656 if (bc->m_named_result_sets)
658 result_set_id = boost::io::str(
659 boost::format("%1%") %
660 found_backend->m_result_set_sequence);
661 found_backend->m_result_set_sequence++;
664 result_set_id = "default";
670 void yf::SessionShared::Frontend::get_set(mp::Package &package,
671 const Z_APDU *apdu_req,
672 const Databases &databases,
673 yazpp_1::Yaz_Z_Query &query,
674 BackendInstancePtr &found_backend,
675 BackendSetPtr &found_set)
677 bool session_restarted = false;
680 std::string result_set_id;
681 bool out_of_sessions = false;
682 BackendClassPtr bc = m_backend_class;
684 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
686 if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
687 out_of_sessions = true;
689 if (m_p->m_optimize_search)
691 // look at each backend and see if we have a similar search
692 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
693 for (; it != bc->m_backend_list.end(); it++)
695 if (!(*it)->m_in_use)
697 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
698 for (; set_it != (*it)->m_sets.end(); set_it++)
700 if ((*set_it)->m_databases == databases
701 && query.match(&(*set_it)->m_query))
705 bc->use_backend(found_backend);
706 // found matching set. No need to search again
713 override_set(found_backend, result_set_id, databases, out_of_sessions);
715 bc->use_backend(found_backend);
719 // create a new backend set (and new set) if we're not out of sessions
720 if (!out_of_sessions)
721 found_backend = bc->create_backend(package);
727 const char *addinfo = 0;
730 addinfo = "session_shared: all sessions in use";
732 addinfo = "session_shared: could not create backend";
733 if (apdu_req->which == Z_APDU_searchRequest)
735 f_apdu = odr.create_searchResponse(
736 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
738 else if (apdu_req->which == Z_APDU_presentRequest)
740 f_apdu = odr.create_presentResponse(
741 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
745 f_apdu = odr.create_close(
746 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
748 package.response() = f_apdu;
751 if (bc->m_named_result_sets)
753 result_set_id = boost::io::str(
754 boost::format("%1%") % found_backend->m_result_set_sequence);
755 found_backend->m_result_set_sequence++;
758 result_set_id = "default";
760 found_backend->timestamp();
762 // we must search ...
763 BackendSetPtr new_set(new BackendSet(result_set_id,
765 Z_Records *z_records = 0;
767 Package search_package(found_backend->m_session, package.origin());
768 search_package.copy_filter(package);
770 if (!new_set->search(package, search_package,
771 apdu_req, found_backend, &z_records))
773 bc->remove_backend(found_backend);
774 return; // search error
780 if (z_records->which == Z_Records_NSD)
783 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
785 else if (z_records->which == Z_Records_multipleNSD)
787 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
790 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
791 Z_DiagRec_defaultFormat)
793 condition = get_diagnostic(
794 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
798 if (m_p->m_restart && !session_restarted &&
799 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
801 package.log("session_shared", YLOG_LOG, "restart");
802 bc->remove_backend(found_backend);
803 session_restarted = true;
804 found_backend.reset();
812 if (apdu_req->which == Z_APDU_searchRequest)
814 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
816 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
817 *f_resp->searchStatus = Z_SearchResponse_none;
818 f_resp->records = z_records;
819 package.response() = f_apdu;
821 if (apdu_req->which == Z_APDU_presentRequest)
823 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
825 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
826 f_resp->records = z_records;
827 package.response() = f_apdu;
829 bc->release_backend(found_backend);
830 return; // search error
833 if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
835 package.log("session_shared", YLOG_LOG, "restart");
836 bc->remove_backend(found_backend);
837 session_restarted = true;
838 found_backend.reset();
843 found_set->timestamp();
844 found_backend->m_sets.push_back(found_set);
847 void yf::SessionShared::Frontend::search(mp::Package &package,
850 Z_SearchRequest *req = apdu_req->u.searchRequest;
851 FrontendSets::iterator fset_it =
852 m_frontend_sets.find(req->resultSetName);
853 if (fset_it != m_frontend_sets.end())
855 // result set already exist
856 // if replace indicator is off: we return diagnostic if
857 // result set already exist.
858 if (*req->replaceIndicator == 0)
862 odr.create_searchResponse(
864 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
866 package.response() = apdu;
870 m_frontend_sets.erase(fset_it);
873 yazpp_1::Yaz_Z_Query query;
874 query.set_Z_Query(req->query);
877 for (i = 0; i < req->num_databaseNames; i++)
878 databases.push_back(req->databaseNames[i]);
880 BackendSetPtr found_set; // null
881 BackendInstancePtr found_backend; // null
883 get_set(package, apdu_req, databases, query, found_backend, found_set);
888 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
889 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
890 *f_resp->resultCount = found_set->m_result_set_size;
891 package.response() = f_apdu;
893 FrontendSetPtr fset(new FrontendSet(databases, query));
894 m_frontend_sets[req->resultSetName] = fset;
896 m_backend_class->release_backend(found_backend);
899 void yf::SessionShared::Frontend::present(mp::Package &package,
903 Z_PresentRequest *req = apdu_req->u.presentRequest;
905 FrontendSets::iterator fset_it =
906 m_frontend_sets.find(req->resultSetId);
908 if (fset_it == m_frontend_sets.end())
911 odr.create_presentResponse(
913 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
915 package.response() = apdu;
918 FrontendSetPtr fset = fset_it->second;
920 Databases databases = fset->get_databases();
921 yazpp_1::Yaz_Z_Query query = fset->get_query();
923 BackendClassPtr bc = m_backend_class;
924 BackendSetPtr found_set; // null
925 BackendInstancePtr found_backend;
927 get_set(package, apdu_req, databases, query, found_backend, found_set);
931 Z_NamePlusRecordList *npr_res = 0;
932 if (found_set->m_record_cache.lookup(odr, &npr_res,
933 *req->resultSetStartPoint,
934 *req->numberOfRecordsRequested,
935 req->preferredRecordSyntax,
936 req->recordComposition))
938 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
939 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
941 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
942 " records in cache %p",
943 *req->resultSetStartPoint,
944 *req->numberOfRecordsRequested,
945 &found_set->m_record_cache);
947 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
948 *f_resp->nextResultSetPosition =
949 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
950 // f_resp->presentStatus assumed OK.
951 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
952 f_resp->records->which = Z_Records_DBOSD;
953 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
954 package.response() = f_apdu_res;
955 bc->release_backend(found_backend);
959 found_backend->timestamp();
961 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
962 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
963 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
964 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
965 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
966 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
967 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
968 p_req->recordComposition = req->recordComposition;
970 Package present_package(found_backend->m_session, package.origin());
971 present_package.copy_filter(package);
973 present_package.request() = p_apdu;
975 present_package.move();
977 Z_GDU *gdu = present_package.response().get();
978 if (!present_package.session().is_closed()
979 && gdu && gdu->which == Z_GDU_Z3950
980 && gdu->u.z3950->which == Z_APDU_presentResponse)
982 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
983 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
984 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
986 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
987 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
988 f_resp->presentStatus= b_resp->presentStatus;
989 f_resp->records = b_resp->records;
990 f_resp->otherInfo = b_resp->otherInfo;
991 package.response() = f_apdu_res;
993 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
995 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
996 " records to cache %p",
997 *req->resultSetStartPoint,
998 *f_resp->numberOfRecordsReturned,
999 &found_set->m_record_cache);
1000 found_set->m_record_cache.add(
1002 b_resp->records->u.databaseOrSurDiagnostics,
1003 *req->resultSetStartPoint,
1004 *f_resp->numberOfRecordsReturned);
1006 bc->release_backend(found_backend);
1010 bc->remove_backend(found_backend);
1011 Z_APDU *f_apdu_res =
1012 odr.create_presentResponse(
1013 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1014 "session_shared: target closed connection during present");
1015 package.response() = f_apdu_res;
1019 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1022 BackendClassPtr bc = m_backend_class;
1023 BackendInstancePtr backend = bc->get_backend(frontend_package);
1027 Z_APDU *apdu = odr.create_scanResponse(
1028 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1029 "session_shared: could not create backend");
1030 frontend_package.response() = apdu;
1034 Package scan_package(backend->m_session, frontend_package.origin());
1035 backend->timestamp();
1036 scan_package.copy_filter(frontend_package);
1037 scan_package.request() = apdu_req;
1038 scan_package.move();
1039 frontend_package.response() = scan_package.response();
1040 if (scan_package.session().is_closed())
1042 frontend_package.session().close();
1043 bc->remove_backend(backend);
1046 bc->release_backend(backend);
1050 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1054 void yf::SessionShared::Worker::operator() (void)
1059 void yf::SessionShared::BackendClass::expire_class()
1063 boost::mutex::scoped_lock lock(m_mutex_backend_class);
1064 BackendInstanceList::iterator bit = m_backend_list.begin();
1065 while (bit != m_backend_list.end())
1067 time_t last_use = (*bit)->m_time_last_use;
1069 if ((*bit)->m_in_use)
1073 else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1076 (*bit)->m_close_package->response() = odr.create_close(
1077 0, Z_Close_lackOfActivity, 0);
1078 (*bit)->m_close_package->session().close();
1079 (*bit)->m_close_package->move();
1081 bit = m_backend_list.erase(bit);
1090 void yf::SessionShared::Rep::expire()
1095 boost::xtime_get(&xt,
1096 #if BOOST_VERSION >= 105000
1102 xt.sec += m_session_ttl / 3;
1103 boost::thread::sleep(xt);
1105 BackendClassMap::const_iterator b_it = m_backend_map.begin();
1106 for (; b_it != m_backend_map.end(); b_it++)
1107 b_it->second->expire_class();
1111 yf::SessionShared::Rep::Rep()
1113 m_resultset_ttl = 30;
1114 m_resultset_max = 10;
1116 m_optimize_search = true;
1118 m_session_max = 100;
1119 m_preferredMessageSize = 0;
1120 m_maximumRecordSize = 0;
1123 void yf::SessionShared::Rep::start()
1125 yf::SessionShared::Worker w(this);
1126 m_thrds.add_thread(new boost::thread(w));
1129 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1133 yf::SessionShared::~SessionShared() {
1136 void yf::SessionShared::start() const
1141 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1145 yf::SessionShared::Frontend::~Frontend()
1149 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1151 boost::mutex::scoped_lock lock(m_mutex);
1153 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1157 it = m_clients.find(package.session());
1158 if (it == m_clients.end())
1161 if (!it->second->m_in_use)
1163 it->second->m_in_use = true;
1166 m_cond_session_ready.wait(lock);
1168 FrontendPtr f(new Frontend(this));
1169 m_clients[package.session()] = f;
1174 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1176 boost::mutex::scoped_lock lock(m_mutex);
1177 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1179 it = m_clients.find(package.session());
1180 if (it != m_clients.end())
1182 if (package.session().is_closed())
1184 m_clients.erase(it);
1188 it->second->m_in_use = false;
1190 m_cond_session_ready.notify_all();
1195 void yf::SessionShared::process(mp::Package &package) const
1197 FrontendPtr f = m_p->get_frontend(package);
1199 Z_GDU *gdu = package.request().get();
1201 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1202 Z_APDU_initRequest && !f->m_is_virtual)
1204 m_p->init(package, gdu, f);
1206 else if (!f->m_is_virtual)
1208 else if (gdu && gdu->which == Z_GDU_Z3950)
1210 Z_APDU *apdu = gdu->u.z3950;
1211 if (apdu->which == Z_APDU_initRequest)
1215 package.response() = odr.create_close(
1217 Z_Close_protocolError,
1220 package.session().close();
1222 else if (apdu->which == Z_APDU_close)
1226 package.response() = odr.create_close(
1228 Z_Close_peerAbort, "received close from client");
1229 package.session().close();
1231 else if (apdu->which == Z_APDU_searchRequest)
1233 f->search(package, apdu);
1235 else if (apdu->which == Z_APDU_presentRequest)
1237 f->present(package, apdu);
1239 else if (apdu->which == Z_APDU_scanRequest)
1241 f->scan(package, apdu);
1247 package.response() = odr.create_close(
1248 apdu, Z_Close_protocolError,
1249 "unsupported APDU in filter_session_shared");
1251 package.session().close();
1254 m_p->release_frontend(package);
1257 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1260 for (ptr = ptr->children; ptr; ptr = ptr->next)
1262 if (ptr->type != XML_ELEMENT_NODE)
1264 if (!strcmp((const char *) ptr->name, "resultset"))
1266 const struct _xmlAttr *attr;
1267 for (attr = ptr->properties; attr; attr = attr->next)
1269 if (!strcmp((const char *) attr->name, "ttl"))
1270 m_p->m_resultset_ttl =
1271 mp::xml::get_int(attr->children, 30);
1272 else if (!strcmp((const char *) attr->name, "max"))
1274 m_p->m_resultset_max =
1275 mp::xml::get_int(attr->children, 10);
1277 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1279 m_p->m_optimize_search =
1280 mp::xml::get_bool(attr->children, true);
1282 else if (!strcmp((const char *) attr->name, "restart"))
1284 m_p->m_restart = mp::xml::get_bool(attr->children, true);
1287 throw mp::filter::FilterException(
1288 "Bad attribute " + std::string((const char *)
1292 else if (!strcmp((const char *) ptr->name, "session"))
1294 const struct _xmlAttr *attr;
1295 for (attr = ptr->properties; attr; attr = attr->next)
1297 if (!strcmp((const char *) attr->name, "ttl"))
1298 m_p->m_session_ttl =
1299 mp::xml::get_int(attr->children, 90);
1300 else if (!strcmp((const char *) attr->name, "max"))
1301 m_p->m_session_max =
1302 mp::xml::get_int(attr->children, 100);
1304 throw mp::filter::FilterException(
1305 "Bad attribute " + std::string((const char *)
1309 else if (!strcmp((const char *) ptr->name, "init"))
1311 const struct _xmlAttr *attr;
1312 for (attr = ptr->properties; attr; attr = attr->next)
1314 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1315 m_p->m_maximumRecordSize =
1316 mp::xml::get_int(attr->children, 0);
1317 else if (!strcmp((const char *) attr->name,
1318 "preferred-message-size"))
1319 m_p->m_preferredMessageSize =
1320 mp::xml::get_int(attr->children, 0);
1322 throw mp::filter::FilterException(
1323 "Bad attribute " + std::string((const char *)
1329 throw mp::filter::FilterException("Bad element "
1330 + std::string((const char *)
1336 static mp::filter::Base* filter_creator()
1338 return new mp::filter::SessionShared;
1342 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1352 * c-file-style: "Stroustrup"
1353 * indent-tabs-mode: nil
1355 * vim: shiftwidth=4 tabstop=8 expandtab