src/filter_session_shared.cpp

   1 /* This file is part of Metaproxy.
   2    Copyright (C) 2005-2013 Index Data
   3
   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
   7 version.
   8
   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
  12 for more details.
  13
  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
  17 */
  18
  19 #include "config.hpp"
  20
  21 #include <metaproxy/filter.hpp>
  22 #include <metaproxy/package.hpp>
  23
  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>
  30
  31 #include <metaproxy/util.hpp>
  32 #include "filter_session_shared.hpp"
  33
  34 #include <yaz/log.h>
  35 #include <yaz/zgdu.h>
  36 #include <yaz/otherinfo.h>
  37 #include <yaz/diagbib1.h>
  38 #include <yazpp/z-query.h>
  39 #include <yazpp/record-cache.h>
  40 #include <map>
  41 #include <iostream>
  42 #include <time.h>
  43
  44 namespace mp = metaproxy_1;
  45 namespace yf = metaproxy_1::filter;
  46
  47 namespace metaproxy_1 {
  48
  49     namespace filter {
  50         // key for session.. We'll only share sessions with same InitKey
  51         class SessionShared::InitKey {
  52         public:
  53             bool operator < (const SessionShared::InitKey &k) const;
  54             InitKey(Z_InitRequest *req);
  55             InitKey(const InitKey &);
  56             ~InitKey();
  57         private:
  58             char *m_idAuthentication_buf;
  59             int m_idAuthentication_size;
  60             char *m_otherInfo_buf;
  61             int m_otherInfo_size;
  62             ODR m_odr;
  63         };
  64         // worker thread .. for expiry of sessions
  65         class SessionShared::Worker {
  66         public:
  67             Worker(SessionShared::Rep *rep);
  68             void operator() (void);
  69         private:
  70             SessionShared::Rep *m_p;
  71         };
  72         // backend result set
  73         class SessionShared::BackendSet {
  74         public:
  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;
  80             void timestamp();
  81             yazpp_1::RecordCache m_record_cache;
  82             BackendSet(
  83                 const std::string &result_set_id,
  84                 const Databases &databases,
  85                 const yazpp_1::Yaz_Z_Query &query);
  86             bool search(
  87                 Package &frontend_package,
  88                 Package &search_package,
  89                 const Z_APDU *apdu_req,
  90                 const BackendInstancePtr bp,
  91                 Z_Records **z_records);
  92         };
  93         // backend connection instance
  94         class SessionShared::BackendInstance {
  95             friend class Rep;
  96             friend class BackendClass;
  97             friend class BackendSet;
  98         public:
  99             mp::Session m_session;
 100             BackendSetList m_sets;
 101             bool m_in_use;
 102             int m_sequence_this;
 103             int m_result_set_sequence;
 104             time_t m_time_last_use;
 105             mp::Package * m_close_package;
 106             ~BackendInstance();
 107             void timestamp();
 108         };
 109         // backends of some class (all with same InitKey)
 110         class SessionShared::BackendClass : boost::noncopyable {
 111             friend class Rep;
 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);
 120             void expire_class();
 121             yazpp_1::GDU m_init_request;
 122             yazpp_1::GDU m_init_response;
 123             boost::mutex m_mutex_backend_class;
 124             int m_sequence_top;
 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;
 130         public:
 131             BackendClass(const yazpp_1::GDU &init_request,
 132                          int resultset_ttl,
 133                          int resultset_max,
 134                          int session_ttl,
 135                          Odr_int preferredRecordSize,
 136                          Odr_int maximumRecordSize);
 137             ~BackendClass();
 138         };
 139         // frontend result set
 140         class SessionShared::FrontendSet {
 141             Databases m_databases;
 142             yazpp_1::Yaz_Z_Query m_query;
 143         public:
 144             const Databases &get_databases();
 145             const yazpp_1::Yaz_Z_Query &get_query();
 146             FrontendSet(
 147                 const Databases &databases,
 148                 const yazpp_1::Yaz_Z_Query &query);
 149             FrontendSet();
 150         };
 151         // frontend session
 152         struct SessionShared::Frontend {
 153             Frontend(Rep *rep);
 154             ~Frontend();
 155             bool m_is_virtual;
 156             bool m_in_use;
 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);
 161
 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);
 172
 173             Rep *m_p;
 174             BackendClassPtr m_backend_class;
 175             FrontendSets m_frontend_sets;
 176         };
 177         // representation
 178         class SessionShared::Rep {
 179             friend class SessionShared;
 180             friend struct Frontend;
 181
 182             FrontendPtr get_frontend(Package &package);
 183             void release_frontend(Package &package);
 184             Rep();
 185         public:
 186             void expire();
 187         private:
 188             void init(Package &package, const Z_GDU *gdu,
 189                       FrontendPtr frontend);
 190             void start();
 191             boost::mutex m_mutex;
 192             boost::condition m_cond_session_ready;
 193             std::map<mp::Session, FrontendPtr> m_clients;
 194
 195             BackendClassMap m_backend_map;
 196             boost::mutex m_mutex_backend_map;
 197             boost::thread_group m_thrds;
 198             int m_resultset_ttl;
 199             int m_resultset_max;
 200             int m_session_ttl;
 201             bool m_optimize_search;
 202             bool m_restart;
 203             int m_session_max;
 204             Odr_int m_preferredMessageSize;
 205             Odr_int m_maximumRecordSize;
 206         };
 207     }
 208 }
 209
 210 yf::SessionShared::FrontendSet::FrontendSet(
 211     const Databases &databases,
 212     const yazpp_1::Yaz_Z_Query &query)
 213     : m_databases(databases), m_query(query)
 214 {
 215 }
 216
 217 const yf::SessionShared::Databases &
 218 yf::SessionShared::FrontendSet::get_databases()
 219 {
 220     return m_databases;
 221 }
 222
 223 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
 224 {
 225     return m_query;
 226 }
 227
 228 yf::SessionShared::InitKey::InitKey(const InitKey &k)
 229 {
 230     m_odr = odr_createmem(ODR_ENCODE);
 231
 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);
 236
 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,
 240            m_otherInfo_size);
 241 }
 242
 243 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
 244 {
 245     m_odr = odr_createmem(ODR_ENCODE);
 246
 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);
 251
 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);
 255 }
 256
 257 yf::SessionShared::InitKey::~InitKey()
 258 {
 259     odr_destroy(m_odr);
 260 }
 261
 262 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
 263     const
 264 {
 265     int c;
 266     c = mp::util::memcmp2(
 267         (void*) m_idAuthentication_buf, m_idAuthentication_size,
 268         (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
 269     if (c < 0)
 270         return true;
 271     else if (c > 0)
 272         return false;
 273
 274     c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
 275                           (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
 276     if (c < 0)
 277         return true;
 278     else if (c > 0)
 279         return false;
 280     return false;
 281 }
 282
 283 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
 284 {
 285     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 286     b->m_in_use = false;
 287 }
 288
 289
 290 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
 291 {
 292     BackendInstanceList::iterator it = m_backend_list.begin();
 293
 294     while (it != m_backend_list.end())
 295     {
 296         if (*it == b)
 297         {
 298              mp::odr odr;
 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();
 303
 304             it = m_backend_list.erase(it);
 305         }
 306         else
 307             it++;
 308     }
 309 }
 310
 311
 312
 313 yf::SessionShared::BackendInstancePtr
 314 yf::SessionShared::BackendClass::get_backend(
 315     const mp::Package &frontend_package)
 316 {
 317     {
 318         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 319
 320         BackendInstanceList::const_iterator it = m_backend_list.begin();
 321
 322         BackendInstancePtr backend1; // null
 323
 324         for (; it != m_backend_list.end(); it++)
 325         {
 326             if (!(*it)->m_in_use)
 327             {
 328                 if (!backend1
 329                     || (*it)->m_sequence_this < backend1->m_sequence_this)
 330                     backend1 = *it;
 331             }
 332         }
 333         if (backend1)
 334         {
 335             use_backend(backend1);
 336             return backend1;
 337         }
 338     }
 339     return create_backend(frontend_package);
 340 }
 341
 342 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
 343 {
 344     backend->m_in_use = true;
 345     backend->m_sequence_this = m_sequence_top++;
 346 }
 347
 348 void yf::SessionShared::BackendInstance::timestamp()
 349 {
 350     assert(m_in_use);
 351     time(&m_time_last_use);
 352 }
 353
 354 yf::SessionShared::BackendInstance::~BackendInstance()
 355 {
 356     delete m_close_package;
 357 }
 358
 359 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
 360     const mp::Package &frontend_package)
 361 {
 362     BackendInstancePtr bp(new BackendInstance);
 363     BackendInstancePtr null;
 364
 365     bp->m_close_package =
 366         new mp::Package(bp->m_session, frontend_package.origin());
 367     bp->m_close_package->copy_filter(frontend_package);
 368
 369     Package init_package(bp->m_session, frontend_package.origin());
 370
 371     init_package.copy_filter(frontend_package);
 372
 373     yazpp_1::GDU actual_init_request = m_init_request;
 374     Z_GDU *init_pdu = actual_init_request.get();
 375
 376     assert(init_pdu->which == Z_GDU_Z3950);
 377     assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
 378
 379     Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
 380     ODR_MASK_ZERO(req->options);
 381
 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);
 386
 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);
 390
 391     if (m_preferredMessageSize)
 392         *req->preferredMessageSize = m_preferredMessageSize;
 393     if (m_maximumRecordSize)
 394         *req->maximumRecordSize = m_maximumRecordSize;
 395
 396     init_package.request() = init_pdu;
 397
 398     init_package.move();
 399
 400     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 401
 402     m_named_result_sets = false;
 403     Z_GDU *gdu = init_package.response().get();
 404
 405     if (gdu && gdu->which == Z_GDU_Z3950
 406         && gdu->u.z3950->which == Z_APDU_initResponse)
 407     {
 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))
 411         {
 412             m_named_result_sets = true;
 413         }
 414         if (*gdu->u.z3950->u.initResponse->result
 415             && !init_package.session().is_closed())
 416         {
 417             bp->m_in_use = true;
 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);
 422             return bp;
 423         }
 424     }
 425     else
 426     {
 427         yazpp_1::GDU empty_gdu;
 428         m_init_response = empty_gdu;
 429     }
 430
 431     if (!init_package.session().is_closed())
 432     {
 433         init_package.copy_filter(frontend_package);
 434         init_package.session().close();
 435         init_package.move();
 436     }
 437     return null;
 438 }
 439
 440
 441 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
 442                                               int resultset_ttl,
 443                                               int resultset_max,
 444                                               int session_ttl,
 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)
 452 {}
 453
 454 yf::SessionShared::BackendClass::~BackendClass()
 455 {}
 456
 457 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
 458                                   FrontendPtr frontend)
 459 {
 460     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 461
 462     frontend->m_is_virtual = true;
 463     frontend->m_init_options = *req->options;
 464     InitKey k(req);
 465     {
 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())
 470         {
 471             BackendClassPtr b(new BackendClass(gdu->u.z3950,
 472                                                m_resultset_ttl,
 473                                                m_resultset_max,
 474                                                m_session_ttl,
 475                                                m_preferredMessageSize,
 476                                                m_maximumRecordSize));
 477             m_backend_map[k] = b;
 478             frontend->m_backend_class = b;
 479         }
 480         else
 481         {
 482             frontend->m_backend_class = it->second;
 483         }
 484     }
 485     BackendClassPtr bc = frontend->m_backend_class;
 486     mp::odr odr;
 487
 488     // we only need to get init response from "first" target in
 489     // backend class - the assumption being that init response is
 490     // same for all
 491     if (bc->m_backend_list.size() == 0)
 492     {
 493         BackendInstancePtr backend = bc->create_backend(package);
 494         if (backend)
 495             bc->release_backend(backend);
 496     }
 497
 498     yazpp_1::GDU init_response;
 499     {
 500         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 501
 502         init_response = bc->m_init_response;
 503     }
 504
 505     if (init_response.get())
 506     {
 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;
 513         int i;
 514         for (i = 0; i < 30; i++)
 515             if (!ODR_MASK_GET(client_options, i))
 516                 ODR_MASK_CLEAR(server_options, i);
 517
 518         if (!m_preferredMessageSize ||
 519             *init_res->preferredMessageSize > *req->preferredMessageSize)
 520             *init_res->preferredMessageSize = *req->preferredMessageSize;
 521
 522         if (!m_maximumRecordSize ||
 523             *init_res->maximumRecordSize > *req->maximumRecordSize)
 524             *init_res->maximumRecordSize = *req->maximumRecordSize;
 525
 526         package.response() = init_response;
 527         if (!*response_gdu->u.z3950->u.initResponse->result)
 528             package.session().close();
 529     }
 530     else
 531     {
 532         Z_APDU *apdu =
 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();
 539     }
 540 }
 541
 542 void yf::SessionShared::BackendSet::timestamp()
 543 {
 544     time(&m_time_last_use);
 545 }
 546
 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)
 553 {
 554     timestamp();
 555 }
 556
 557 static int get_diagnostic(Z_DefaultDiagFormat *r)
 558 {
 559     return *r->condition;
 560 }
 561
 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)
 568 {
 569     mp::odr odr;
 570     Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
 571     Z_SearchRequest *req = apdu_req->u.searchRequest;
 572
 573     req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
 574     req->query = m_query.get_Z_Query();
 575
 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();
 580     size_t i = 0;
 581     for (; it != m_databases.end(); it++)
 582         req->databaseNames[i++] = odr_strdup(odr, it->c_str());
 583
 584     if (frontend_apdu->which == Z_APDU_searchRequest)
 585         req->preferredRecordSyntax =
 586             frontend_apdu->u.searchRequest->preferredRecordSyntax;
 587
 588     search_package.request() = apdu_req;
 589
 590     search_package.move();
 591
 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)
 596     {
 597         Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
 598         *z_records = b_resp->records;
 599         m_result_set_size = *b_resp->resultCount;
 600         return true;
 601     }
 602     Z_APDU *f_apdu = 0;
 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);
 611     else
 612         f_apdu = odr.create_close(
 613             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 614     frontend_package.response() = f_apdu;
 615     return false;
 616 }
 617
 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)
 623 {
 624     BackendClassPtr bc = m_backend_class;
 625     BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 626     time_t now;
 627     time(&now);
 628
 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++)
 631     {
 632         if (!(*it)->m_in_use)
 633         {
 634             BackendSetList::iterator set_it = (*it)->m_sets.begin();
 635             for (; set_it != (*it)->m_sets.end(); set_it++)
 636             {
 637                 if ((max_sets > 1 || (*set_it)->m_databases == databases)
 638                     &&
 639                     (out_of_sessions ||
 640                      now < (*set_it)->m_time_last_use ||
 641                      now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
 642                 {
 643                     found_backend = *it;
 644                     result_set_id = (*set_it)->m_result_set_id;
 645                     found_backend->m_sets.erase(set_it);
 646                     return;
 647                 }
 648             }
 649         }
 650     }
 651     for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
 652     {
 653         if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
 654         {
 655             found_backend = *it;
 656             if (bc->m_named_result_sets)
 657             {
 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++;
 662             }
 663             else
 664                 result_set_id = "default";
 665             return;
 666         }
 667     }
 668 }
 669
 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)
 676 {
 677     bool session_restarted = false;
 678
 679 restart:
 680     std::string result_set_id;
 681     bool out_of_sessions = false;
 682     BackendClassPtr bc = m_backend_class;
 683     {
 684         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 685
 686         if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
 687             out_of_sessions = true;
 688
 689         if (m_p->m_optimize_search)
 690         {
 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++)
 694             {
 695                 if (!(*it)->m_in_use)
 696                 {
 697                     BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
 698                     for (; set_it != (*it)->m_sets.end(); set_it++)
 699                     {
 700                         if ((*set_it)->m_databases == databases
 701                             && query.match(&(*set_it)->m_query))
 702                         {
 703                             found_set = *set_it;
 704                             found_backend = *it;
 705                             bc->use_backend(found_backend);
 706                             // found matching set. No need to search again
 707                             return;
 708                         }
 709                     }
 710                 }
 711             }
 712         }
 713         override_set(found_backend, result_set_id, databases, out_of_sessions);
 714         if (found_backend)
 715             bc->use_backend(found_backend);
 716     }
 717     if (!found_backend)
 718     {
 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);
 722
 723         if (!found_backend)
 724         {
 725             Z_APDU *f_apdu = 0;
 726             mp::odr odr;
 727             const char *addinfo = 0;
 728
 729             if (out_of_sessions)
 730                 addinfo = "session_shared: all sessions in use";
 731             else
 732                 addinfo = "session_shared: could not create backend";
 733             if (apdu_req->which == Z_APDU_searchRequest)
 734             {
 735                 f_apdu = odr.create_searchResponse(
 736                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 737             }
 738             else if (apdu_req->which == Z_APDU_presentRequest)
 739             {
 740                 f_apdu = odr.create_presentResponse(
 741                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 742             }
 743             else
 744             {
 745                 f_apdu = odr.create_close(
 746                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 747             }
 748             package.response() = f_apdu;
 749             return;
 750         }
 751         if (bc->m_named_result_sets)
 752         {
 753             result_set_id = boost::io::str(
 754                 boost::format("%1%") % found_backend->m_result_set_sequence);
 755             found_backend->m_result_set_sequence++;
 756         }
 757         else
 758             result_set_id = "default";
 759     }
 760     found_backend->timestamp();
 761
 762     // we must search ...
 763     BackendSetPtr new_set(new BackendSet(result_set_id,
 764                                          databases, query));
 765     Z_Records *z_records = 0;
 766
 767     Package search_package(found_backend->m_session, package.origin());
 768     search_package.copy_filter(package);
 769
 770     if (!new_set->search(package, search_package,
 771                          apdu_req, found_backend, &z_records))
 772     {
 773         bc->remove_backend(found_backend);
 774         return; // search error
 775     }
 776
 777     if (z_records)
 778     {
 779         int condition = 0;
 780         if (z_records->which == Z_Records_NSD)
 781         {
 782             condition =
 783                 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
 784         }
 785         else if (z_records->which == Z_Records_multipleNSD)
 786         {
 787             if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
 788                 &&
 789
 790                 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
 791                 Z_DiagRec_defaultFormat)
 792             {
 793                 condition = get_diagnostic(
 794                     z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
 795
 796             }
 797         }
 798         if (m_p->m_restart && !session_restarted &&
 799             condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
 800         {
 801             package.log("session_shared", YLOG_LOG, "restart");
 802             bc->remove_backend(found_backend);
 803             session_restarted = true;
 804             found_backend.reset();
 805             goto restart;
 806
 807         }
 808
 809         if (condition)
 810         {
 811             mp::odr odr;
 812             if (apdu_req->which == Z_APDU_searchRequest)
 813             {
 814                 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
 815                                                            0, 0);
 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;
 820             }
 821             if (apdu_req->which == Z_APDU_presentRequest)
 822             {
 823                 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
 824                                                             0, 0);
 825                 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
 826                 f_resp->records = z_records;
 827                 package.response() = f_apdu;
 828             }
 829             bc->release_backend(found_backend);
 830             return; // search error
 831         }
 832     }
 833     if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
 834     {
 835         package.log("session_shared", YLOG_LOG, "restart");
 836         bc->remove_backend(found_backend);
 837         session_restarted = true;
 838         found_backend.reset();
 839         goto restart;
 840     }
 841
 842     found_set = new_set;
 843     found_set->timestamp();
 844     found_backend->m_sets.push_back(found_set);
 845 }
 846
 847 void yf::SessionShared::Frontend::search(mp::Package &package,
 848                                          Z_APDU *apdu_req)
 849 {
 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())
 854     {
 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)
 859         {
 860             mp::odr odr;
 861             Z_APDU *apdu =
 862                 odr.create_searchResponse(
 863                     apdu_req,
 864                     YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
 865                     0);
 866             package.response() = apdu;
 867
 868             return;
 869         }
 870         m_frontend_sets.erase(fset_it);
 871     }
 872
 873     yazpp_1::Yaz_Z_Query query;
 874     query.set_Z_Query(req->query);
 875     Databases databases;
 876     int i;
 877     for (i = 0; i < req->num_databaseNames; i++)
 878         databases.push_back(req->databaseNames[i]);
 879
 880     BackendSetPtr found_set; // null
 881     BackendInstancePtr found_backend; // null
 882
 883     get_set(package, apdu_req, databases, query, found_backend, found_set);
 884     if (!found_set)
 885         return;
 886
 887     mp::odr odr;
 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;
 892
 893     FrontendSetPtr fset(new FrontendSet(databases, query));
 894     m_frontend_sets[req->resultSetName] = fset;
 895
 896     m_backend_class->release_backend(found_backend);
 897 }
 898
 899 void yf::SessionShared::Frontend::present(mp::Package &package,
 900                                           Z_APDU *apdu_req)
 901 {
 902     mp::odr odr;
 903     Z_PresentRequest *req = apdu_req->u.presentRequest;
 904
 905     FrontendSets::iterator fset_it =
 906         m_frontend_sets.find(req->resultSetId);
 907
 908     if (fset_it == m_frontend_sets.end())
 909     {
 910         Z_APDU *apdu =
 911             odr.create_presentResponse(
 912                 apdu_req,
 913                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
 914                 req->resultSetId);
 915         package.response() = apdu;
 916         return;
 917     }
 918     FrontendSetPtr fset = fset_it->second;
 919
 920     Databases databases = fset->get_databases();
 921     yazpp_1::Yaz_Z_Query query = fset->get_query();
 922
 923     BackendClassPtr bc = m_backend_class;
 924     BackendSetPtr found_set; // null
 925     BackendInstancePtr found_backend;
 926
 927     get_set(package, apdu_req, databases, query, found_backend, found_set);
 928     if (!found_set)
 929         return;
 930
 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))
 937     {
 938         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
 939         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
 940
 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);
 946
 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);
 956         return;
 957     }
 958
 959     found_backend->timestamp();
 960
 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;
 969
 970     Package present_package(found_backend->m_session, package.origin());
 971     present_package.copy_filter(package);
 972
 973     present_package.request() = p_apdu;
 974
 975     present_package.move();
 976
 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)
 981     {
 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;
 985
 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;
 992
 993         if (b_resp->records && b_resp->records->which ==  Z_Records_DBOSD)
 994         {
 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(
1001                 odr,
1002                 b_resp->records->u.databaseOrSurDiagnostics,
1003                 *req->resultSetStartPoint,
1004                 *f_resp->numberOfRecordsReturned);
1005         }
1006         bc->release_backend(found_backend);
1007     }
1008     else
1009     {
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;
1016     }
1017 }
1018
1019 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1020                                        Z_APDU *apdu_req)
1021 {
1022     BackendClassPtr bc = m_backend_class;
1023     BackendInstancePtr backend = bc->get_backend(frontend_package);
1024     if (!backend)
1025     {
1026         mp::odr odr;
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;
1031     }
1032     else
1033     {
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())
1041         {
1042             frontend_package.session().close();
1043             bc->remove_backend(backend);
1044         }
1045         else
1046             bc->release_backend(backend);
1047     }
1048 }
1049
1050 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1051 {
1052 }
1053
1054 void yf::SessionShared::Worker::operator() (void)
1055 {
1056     m_p->expire();
1057 }
1058
1059 void yf::SessionShared::BackendClass::expire_class()
1060 {
1061     time_t now;
1062     time(&now);
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())
1066     {
1067         time_t last_use = (*bit)->m_time_last_use;
1068
1069         if ((*bit)->m_in_use)
1070         {
1071             bit++;
1072         }
1073         else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1074         {
1075             mp::odr odr;
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();
1080
1081             bit = m_backend_list.erase(bit);
1082         }
1083         else
1084         {
1085             bit++;
1086         }
1087     }
1088 }
1089
1090 void yf::SessionShared::Rep::expire()
1091 {
1092     while (true)
1093     {
1094         boost::xtime xt;
1095         boost::xtime_get(&xt,
1096 #if BOOST_VERSION >= 105000
1097                 boost::TIME_UTC_
1098 #else
1099                 boost::TIME_UTC
1100 #endif
1101                   );
1102         xt.sec += m_session_ttl / 3;
1103         boost::thread::sleep(xt);
1104
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();
1108     }
1109 }
1110
1111 yf::SessionShared::Rep::Rep()
1112 {
1113     m_resultset_ttl = 30;
1114     m_resultset_max = 10;
1115     m_session_ttl = 90;
1116     m_optimize_search = true;
1117     m_restart = false;
1118     m_session_max = 100;
1119     m_preferredMessageSize = 0;
1120     m_maximumRecordSize = 0;
1121 }
1122
1123 void yf::SessionShared::Rep::start()
1124 {
1125     yf::SessionShared::Worker w(this);
1126     m_thrds.add_thread(new boost::thread(w));
1127 }
1128
1129 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1130 {
1131 }
1132
1133 yf::SessionShared::~SessionShared() {
1134 }
1135
1136 void yf::SessionShared::start() const
1137 {
1138     m_p->start();
1139 }
1140
1141 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1142 {
1143 }
1144
1145 yf::SessionShared::Frontend::~Frontend()
1146 {
1147 }
1148
1149 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1150 {
1151     boost::mutex::scoped_lock lock(m_mutex);
1152
1153     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1154
1155     while(true)
1156     {
1157         it = m_clients.find(package.session());
1158         if (it == m_clients.end())
1159             break;
1160
1161         if (!it->second->m_in_use)
1162         {
1163             it->second->m_in_use = true;
1164             return it->second;
1165         }
1166         m_cond_session_ready.wait(lock);
1167     }
1168     FrontendPtr f(new Frontend(this));
1169     m_clients[package.session()] = f;
1170     f->m_in_use = true;
1171     return f;
1172 }
1173
1174 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1175 {
1176     boost::mutex::scoped_lock lock(m_mutex);
1177     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1178
1179     it = m_clients.find(package.session());
1180     if (it != m_clients.end())
1181     {
1182         if (package.session().is_closed())
1183         {
1184             m_clients.erase(it);
1185         }
1186         else
1187         {
1188             it->second->m_in_use = false;
1189         }
1190         m_cond_session_ready.notify_all();
1191     }
1192 }
1193
1194
1195 void yf::SessionShared::process(mp::Package &package) const
1196 {
1197     FrontendPtr f = m_p->get_frontend(package);
1198
1199     Z_GDU *gdu = package.request().get();
1200
1201     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1202         Z_APDU_initRequest && !f->m_is_virtual)
1203     {
1204         m_p->init(package, gdu, f);
1205     }
1206     else if (!f->m_is_virtual)
1207         package.move();
1208     else if (gdu && gdu->which == Z_GDU_Z3950)
1209     {
1210         Z_APDU *apdu = gdu->u.z3950;
1211         if (apdu->which == Z_APDU_initRequest)
1212         {
1213             mp::odr odr;
1214
1215             package.response() = odr.create_close(
1216                 apdu,
1217                 Z_Close_protocolError,
1218                 "double init");
1219
1220             package.session().close();
1221         }
1222         else if (apdu->which == Z_APDU_close)
1223         {
1224             mp::odr odr;
1225
1226             package.response() = odr.create_close(
1227                 apdu,
1228                 Z_Close_peerAbort, "received close from client");
1229             package.session().close();
1230         }
1231         else if (apdu->which == Z_APDU_searchRequest)
1232         {
1233             f->search(package, apdu);
1234         }
1235         else if (apdu->which == Z_APDU_presentRequest)
1236         {
1237             f->present(package, apdu);
1238         }
1239         else if (apdu->which == Z_APDU_scanRequest)
1240         {
1241             f->scan(package, apdu);
1242         }
1243         else
1244         {
1245             mp::odr odr;
1246
1247             package.response() = odr.create_close(
1248                 apdu, Z_Close_protocolError,
1249                 "unsupported APDU in filter_session_shared");
1250
1251             package.session().close();
1252         }
1253     }
1254     m_p->release_frontend(package);
1255 }
1256
1257 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1258                                   const char *path)
1259 {
1260     for (ptr = ptr->children; ptr; ptr = ptr->next)
1261     {
1262         if (ptr->type != XML_ELEMENT_NODE)
1263             continue;
1264         if (!strcmp((const char *) ptr->name, "resultset"))
1265         {
1266             const struct _xmlAttr *attr;
1267             for (attr = ptr->properties; attr; attr = attr->next)
1268             {
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"))
1273                 {
1274                     m_p->m_resultset_max =
1275                         mp::xml::get_int(attr->children, 10);
1276                 }
1277                 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1278                 {
1279                     m_p->m_optimize_search =
1280                         mp::xml::get_bool(attr->children, true);
1281                 }
1282                 else if (!strcmp((const char *) attr->name, "restart"))
1283                 {
1284                     m_p->m_restart = mp::xml::get_bool(attr->children, true);
1285                 }
1286                 else
1287                     throw mp::filter::FilterException(
1288                         "Bad attribute " + std::string((const char *)
1289                                                        attr->name));
1290             }
1291         }
1292         else if (!strcmp((const char *) ptr->name, "session"))
1293         {
1294             const struct _xmlAttr *attr;
1295             for (attr = ptr->properties; attr; attr = attr->next)
1296             {
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);
1303                 else
1304                     throw mp::filter::FilterException(
1305                         "Bad attribute " + std::string((const char *)
1306                                                        attr->name));
1307             }
1308         }
1309         else if (!strcmp((const char *) ptr->name, "init"))
1310         {
1311             const struct _xmlAttr *attr;
1312             for (attr = ptr->properties; attr; attr = attr->next)
1313             {
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);
1321                 else
1322                     throw mp::filter::FilterException(
1323                         "Bad attribute " + std::string((const char *)
1324                                                        attr->name));
1325             }
1326         }
1327         else
1328         {
1329             throw mp::filter::FilterException("Bad element "
1330                                                + std::string((const char *)
1331                                                              ptr->name));
1332         }
1333     }
1334 }
1335
1336 static mp::filter::Base* filter_creator()
1337 {
1338     return new mp::filter::SessionShared;
1339 }
1340
1341 extern "C" {
1342     struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1343         0,
1344         "session_shared",
1345         filter_creator
1346     };
1347 }
1348
1349 /*
1350  * Local variables:
1351  * c-basic-offset: 4
1352  * c-file-style: "Stroustrup"
1353  * indent-tabs-mode: nil
1354  * End:
1355  * vim: shiftwidth=4 tabstop=8 expandtab
1356  */
1357