src/filter_session_shared.cpp

   1 /* This file is part of Metaproxy.
   2    Copyright (C) 2005-2012 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         public:
 129             BackendClass(const yazpp_1::GDU &init_request,
 130                          int resultset_ttl,
 131                          int resultset_max,
 132                          int session_ttl);
 133             ~BackendClass();
 134         };
 135         // frontend result set
 136         class SessionShared::FrontendSet {
 137             Databases m_databases;
 138             yazpp_1::Yaz_Z_Query m_query;
 139         public:
 140             const Databases &get_databases();
 141             const yazpp_1::Yaz_Z_Query &get_query();
 142             FrontendSet(
 143                 const Databases &databases,
 144                 const yazpp_1::Yaz_Z_Query &query);
 145             FrontendSet();
 146         };
 147         // frontend session
 148         struct SessionShared::Frontend {
 149             Frontend(Rep *rep);
 150             ~Frontend();
 151             bool m_is_virtual;
 152             bool m_in_use;
 153             Z_Options m_init_options;
 154             void search(Package &package, Z_APDU *apdu);
 155             void present(Package &package, Z_APDU *apdu);
 156             void scan(Package &package, Z_APDU *apdu);
 157
 158             void get_set(mp::Package &package,
 159                          const Z_APDU *apdu_req,
 160                          const Databases &databases,
 161                          yazpp_1::Yaz_Z_Query &query,
 162                          BackendInstancePtr &found_backend,
 163                          BackendSetPtr &found_set);
 164             void override_set(BackendInstancePtr &found_backend,
 165                               std::string &result_set_id,
 166                               const Databases &databases,
 167                               bool out_of_sessions);
 168
 169             Rep *m_p;
 170             BackendClassPtr m_backend_class;
 171             FrontendSets m_frontend_sets;
 172         };
 173         // representation
 174         class SessionShared::Rep {
 175             friend class SessionShared;
 176             friend struct Frontend;
 177
 178             FrontendPtr get_frontend(Package &package);
 179             void release_frontend(Package &package);
 180             Rep();
 181         public:
 182             void expire();
 183         private:
 184             void init(Package &package, const Z_GDU *gdu,
 185                       FrontendPtr frontend);
 186             void start();
 187             boost::mutex m_mutex;
 188             boost::condition m_cond_session_ready;
 189             std::map<mp::Session, FrontendPtr> m_clients;
 190
 191             BackendClassMap m_backend_map;
 192             boost::mutex m_mutex_backend_map;
 193             boost::thread_group m_thrds;
 194             int m_resultset_ttl;
 195             int m_resultset_max;
 196             int m_session_ttl;
 197             bool m_optimize_search;
 198             bool m_restart;
 199             int m_session_max;
 200         };
 201     }
 202 }
 203
 204 yf::SessionShared::FrontendSet::FrontendSet(
 205     const Databases &databases,
 206     const yazpp_1::Yaz_Z_Query &query)
 207     : m_databases(databases), m_query(query)
 208 {
 209 }
 210
 211 const yf::SessionShared::Databases &
 212 yf::SessionShared::FrontendSet::get_databases()
 213 {
 214     return m_databases;
 215 }
 216
 217 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
 218 {
 219     return m_query;
 220 }
 221
 222 yf::SessionShared::InitKey::InitKey(const InitKey &k)
 223 {
 224     m_odr = odr_createmem(ODR_ENCODE);
 225
 226     m_idAuthentication_size =  k.m_idAuthentication_size;
 227     m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
 228     memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
 229            m_idAuthentication_size);
 230
 231     m_otherInfo_size =  k.m_otherInfo_size;
 232     m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
 233     memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
 234            m_otherInfo_size);
 235 }
 236
 237 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
 238 {
 239     m_odr = odr_createmem(ODR_ENCODE);
 240
 241     Z_IdAuthentication *t = req->idAuthentication;
 242     z_IdAuthentication(m_odr, &t, 1, 0);
 243     m_idAuthentication_buf =
 244         odr_getbuf(m_odr, &m_idAuthentication_size, 0);
 245
 246     Z_OtherInformation *o = req->otherInfo;
 247     z_OtherInformation(m_odr, &o, 1, 0);
 248     m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
 249 }
 250
 251 yf::SessionShared::InitKey::~InitKey()
 252 {
 253     odr_destroy(m_odr);
 254 }
 255
 256 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
 257     const
 258 {
 259     int c;
 260     c = mp::util::memcmp2(
 261         (void*) m_idAuthentication_buf, m_idAuthentication_size,
 262         (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
 263     if (c < 0)
 264         return true;
 265     else if (c > 0)
 266         return false;
 267
 268     c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
 269                           (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
 270     if (c < 0)
 271         return true;
 272     else if (c > 0)
 273         return false;
 274     return false;
 275 }
 276
 277 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
 278 {
 279     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 280     b->m_in_use = false;
 281 }
 282
 283
 284 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
 285 {
 286     BackendInstanceList::iterator it = m_backend_list.begin();
 287
 288     while (it != m_backend_list.end())
 289     {
 290         if (*it == b)
 291         {
 292              mp::odr odr;
 293             (*it)->m_close_package->response() = odr.create_close(
 294                 0, Z_Close_lackOfActivity, 0);
 295             (*it)->m_close_package->session().close();
 296             (*it)->m_close_package->move();
 297
 298             it = m_backend_list.erase(it);
 299         }
 300         else
 301             it++;
 302     }
 303 }
 304
 305
 306
 307 yf::SessionShared::BackendInstancePtr
 308 yf::SessionShared::BackendClass::get_backend(
 309     const mp::Package &frontend_package)
 310 {
 311     {
 312         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 313
 314         BackendInstanceList::const_iterator it = m_backend_list.begin();
 315
 316         BackendInstancePtr backend1; // null
 317
 318         for (; it != m_backend_list.end(); it++)
 319         {
 320             if (!(*it)->m_in_use)
 321             {
 322                 if (!backend1
 323                     || (*it)->m_sequence_this < backend1->m_sequence_this)
 324                     backend1 = *it;
 325             }
 326         }
 327         if (backend1)
 328         {
 329             use_backend(backend1);
 330             return backend1;
 331         }
 332     }
 333     return create_backend(frontend_package);
 334 }
 335
 336 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
 337 {
 338     backend->m_in_use = true;
 339     backend->m_sequence_this = m_sequence_top++;
 340 }
 341
 342 void yf::SessionShared::BackendInstance::timestamp()
 343 {
 344     assert(m_in_use);
 345     time(&m_time_last_use);
 346 }
 347
 348 yf::SessionShared::BackendInstance::~BackendInstance()
 349 {
 350     delete m_close_package;
 351 }
 352
 353 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
 354     const mp::Package &frontend_package)
 355 {
 356     BackendInstancePtr bp(new BackendInstance);
 357     BackendInstancePtr null;
 358
 359     bp->m_close_package =
 360         new mp::Package(bp->m_session, frontend_package.origin());
 361     bp->m_close_package->copy_filter(frontend_package);
 362
 363     Package init_package(bp->m_session, frontend_package.origin());
 364
 365     init_package.copy_filter(frontend_package);
 366
 367     yazpp_1::GDU actual_init_request = m_init_request;
 368     Z_GDU *init_pdu = actual_init_request.get();
 369
 370     assert(init_pdu->which == Z_GDU_Z3950);
 371     assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
 372
 373     Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
 374     ODR_MASK_ZERO(req->options);
 375
 376     ODR_MASK_SET(req->options, Z_Options_search);
 377     ODR_MASK_SET(req->options, Z_Options_present);
 378     ODR_MASK_SET(req->options, Z_Options_namedResultSets);
 379     ODR_MASK_SET(req->options, Z_Options_scan);
 380
 381     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
 382     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
 383     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
 384
 385     init_package.request() = init_pdu;
 386
 387     init_package.move();
 388
 389     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 390
 391     m_named_result_sets = false;
 392     Z_GDU *gdu = init_package.response().get();
 393     if (init_package.session().is_closed())
 394     {
 395         /* already closed. We don't know why */
 396         return null;
 397     }
 398     else if (gdu && gdu->which == Z_GDU_Z3950
 399              && gdu->u.z3950->which == Z_APDU_initResponse
 400              && *gdu->u.z3950->u.initResponse->result)
 401     {
 402         /* successful init response */
 403         Z_InitResponse *res = gdu->u.z3950->u.initResponse;
 404         m_init_response = gdu->u.z3950;
 405         if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
 406         {
 407             m_named_result_sets = true;
 408         }
 409     }
 410     else
 411     {
 412         /* not init or init rejected */
 413         init_package.copy_filter(frontend_package);
 414         init_package.session().close();
 415         init_package.move();
 416         return null;
 417     }
 418     bp->m_in_use = true;
 419     time(&bp->m_time_last_use);
 420     bp->m_sequence_this = 0;
 421     bp->m_result_set_sequence = 0;
 422     m_backend_list.push_back(bp);
 423
 424     return bp;
 425 }
 426
 427
 428 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
 429                                               int resultset_ttl,
 430                                               int resultset_max,
 431                                               int session_ttl)
 432     : m_named_result_sets(false), m_init_request(init_request),
 433       m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
 434       m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max)
 435 {}
 436
 437 yf::SessionShared::BackendClass::~BackendClass()
 438 {}
 439
 440 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
 441                                   FrontendPtr frontend)
 442 {
 443     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 444
 445     frontend->m_is_virtual = true;
 446     frontend->m_init_options = *req->options;
 447     InitKey k(req);
 448     {
 449         boost::mutex::scoped_lock lock(m_mutex_backend_map);
 450         BackendClassMap::const_iterator it;
 451         it = m_backend_map.find(k);
 452         if (it == m_backend_map.end())
 453         {
 454             BackendClassPtr b(new BackendClass(gdu->u.z3950,
 455                                                m_resultset_ttl,
 456                                                m_resultset_max,
 457                                                m_session_ttl));
 458             m_backend_map[k] = b;
 459             frontend->m_backend_class = b;
 460         }
 461         else
 462         {
 463             frontend->m_backend_class = it->second;
 464         }
 465     }
 466     BackendClassPtr bc = frontend->m_backend_class;
 467     BackendInstancePtr backend;
 468     mp::odr odr;
 469
 470     // we only need to get init response from "first" target in
 471     // backend class - the assumption being that init response is
 472     // same for all
 473     if (bc->m_init_response.get() == 0)
 474     {
 475         backend = bc->get_backend(package);
 476     }
 477     {
 478         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 479         if (bc->m_init_response.get() == 0)
 480         {
 481             Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
 482             *apdu->u.initResponse->result = 0;
 483             package.response() = apdu;
 484             package.session().close();
 485         }
 486         else
 487         {
 488             yazpp_1::GDU init_response = bc->m_init_response;
 489             Z_GDU *response_gdu = init_response.get();
 490             mp::util::transfer_referenceId(odr, gdu->u.z3950,
 491                                            response_gdu->u.z3950);
 492
 493             Z_Options *server_options =
 494                 response_gdu->u.z3950->u.initResponse->options;
 495             Z_Options *client_options = &frontend->m_init_options;
 496
 497             int i;
 498             for (i = 0; i < 30; i++)
 499                 if (!ODR_MASK_GET(client_options, i))
 500                     ODR_MASK_CLEAR(server_options, i);
 501             package.response() = init_response;
 502         }
 503     }
 504     if (backend)
 505         bc->release_backend(backend);
 506 }
 507
 508 void yf::SessionShared::BackendSet::timestamp()
 509 {
 510     time(&m_time_last_use);
 511 }
 512
 513 yf::SessionShared::BackendSet::BackendSet(
 514     const std::string &result_set_id,
 515     const Databases &databases,
 516     const yazpp_1::Yaz_Z_Query &query) :
 517     m_result_set_id(result_set_id),
 518     m_databases(databases), m_result_set_size(0), m_query(query)
 519 {
 520     timestamp();
 521 }
 522
 523 static int get_diagnostic(Z_DefaultDiagFormat *r)
 524 {
 525     return *r->condition;
 526 }
 527
 528 bool yf::SessionShared::BackendSet::search(
 529     mp::Package &frontend_package,
 530     mp::Package &search_package,
 531     const Z_APDU *frontend_apdu,
 532     const BackendInstancePtr bp,
 533     Z_Records **z_records)
 534 {
 535     mp::odr odr;
 536     Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
 537     Z_SearchRequest *req = apdu_req->u.searchRequest;
 538
 539     req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
 540     req->query = m_query.get_Z_Query();
 541
 542     req->num_databaseNames = m_databases.size();
 543     req->databaseNames = (char**)
 544         odr_malloc(odr, req->num_databaseNames * sizeof(char *));
 545     Databases::const_iterator it = m_databases.begin();
 546     size_t i = 0;
 547     for (; it != m_databases.end(); it++)
 548         req->databaseNames[i++] = odr_strdup(odr, it->c_str());
 549
 550     if (frontend_apdu->which == Z_APDU_searchRequest)
 551         req->preferredRecordSyntax =
 552             frontend_apdu->u.searchRequest->preferredRecordSyntax;
 553
 554     search_package.request() = apdu_req;
 555
 556     search_package.move();
 557
 558     Z_GDU *gdu = search_package.response().get();
 559     if (!search_package.session().is_closed()
 560         && gdu && gdu->which == Z_GDU_Z3950
 561         && gdu->u.z3950->which == Z_APDU_searchResponse)
 562     {
 563         Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
 564         *z_records = b_resp->records;
 565         m_result_set_size = *b_resp->resultCount;
 566         return true;
 567     }
 568     Z_APDU *f_apdu = 0;
 569     if (frontend_apdu->which == Z_APDU_searchRequest)
 570         f_apdu = odr.create_searchResponse(
 571             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 572     else if (frontend_apdu->which == Z_APDU_presentRequest)
 573         f_apdu = odr.create_presentResponse(
 574             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 575     else
 576         f_apdu = odr.create_close(
 577             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 578     frontend_package.response() = f_apdu;
 579     return false;
 580 }
 581
 582 void yf::SessionShared::Frontend::override_set(
 583     BackendInstancePtr &found_backend,
 584     std::string &result_set_id,
 585     const Databases &databases,
 586     bool out_of_sessions)
 587 {
 588     BackendClassPtr bc = m_backend_class;
 589     BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 590     time_t now;
 591     time(&now);
 592
 593     size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
 594     for (; it != bc->m_backend_list.end(); it++)
 595     {
 596         if (!(*it)->m_in_use)
 597         {
 598             BackendSetList::iterator set_it = (*it)->m_sets.begin();
 599             for (; set_it != (*it)->m_sets.end(); set_it++)
 600             {
 601                 if ((max_sets > 1 || (*set_it)->m_databases == databases)
 602                     &&
 603                     (out_of_sessions ||
 604                      now < (*set_it)->m_time_last_use ||
 605                      now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
 606                 {
 607                     found_backend = *it;
 608                     result_set_id = (*set_it)->m_result_set_id;
 609                     found_backend->m_sets.erase(set_it);
 610                     return;
 611                 }
 612             }
 613         }
 614     }
 615     for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
 616     {
 617         if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
 618         {
 619             found_backend = *it;
 620             if (bc->m_named_result_sets)
 621             {
 622                 result_set_id = boost::io::str(
 623                     boost::format("%1%") %
 624                     found_backend->m_result_set_sequence);
 625                 found_backend->m_result_set_sequence++;
 626             }
 627             else
 628                 result_set_id = "default";
 629             return;
 630         }
 631     }
 632 }
 633
 634 void yf::SessionShared::Frontend::get_set(mp::Package &package,
 635                                           const Z_APDU *apdu_req,
 636                                           const Databases &databases,
 637                                           yazpp_1::Yaz_Z_Query &query,
 638                                           BackendInstancePtr &found_backend,
 639                                           BackendSetPtr &found_set)
 640 {
 641     bool session_restarted = false;
 642
 643 restart:
 644     std::string result_set_id;
 645     bool out_of_sessions = false;
 646     BackendClassPtr bc = m_backend_class;
 647     {
 648         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 649
 650         if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
 651             out_of_sessions = true;
 652
 653         if (m_p->m_optimize_search)
 654         {
 655             // look at each backend and see if we have a similar search
 656             BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 657             for (; it != bc->m_backend_list.end(); it++)
 658             {
 659                 if (!(*it)->m_in_use)
 660                 {
 661                     BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
 662                     for (; set_it != (*it)->m_sets.end(); set_it++)
 663                     {
 664                         if ((*set_it)->m_databases == databases
 665                             && query.match(&(*set_it)->m_query))
 666                         {
 667                             found_set = *set_it;
 668                             found_backend = *it;
 669                             bc->use_backend(found_backend);
 670                             // found matching set. No need to search again
 671                             return;
 672                         }
 673                     }
 674                 }
 675             }
 676         }
 677         override_set(found_backend, result_set_id, databases, out_of_sessions);
 678         if (found_backend)
 679             bc->use_backend(found_backend);
 680     }
 681     if (!found_backend)
 682     {
 683         // create a new backend set (and new set) if we're not out of sessions
 684         if (!out_of_sessions)
 685             found_backend = bc->create_backend(package);
 686
 687         if (!found_backend)
 688         {
 689             Z_APDU *f_apdu = 0;
 690             mp::odr odr;
 691             const char *addinfo = 0;
 692
 693             if (out_of_sessions)
 694                 addinfo = "session_shared: all sessions in use";
 695             if (apdu_req->which == Z_APDU_searchRequest)
 696             {
 697                 f_apdu = odr.create_searchResponse(
 698                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 699             }
 700             else if (apdu_req->which == Z_APDU_presentRequest)
 701             {
 702                 f_apdu = odr.create_presentResponse(
 703                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 704             }
 705             else
 706             {
 707                 f_apdu = odr.create_close(
 708                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 709             }
 710             package.response() = f_apdu;
 711             return;
 712         }
 713         if (bc->m_named_result_sets)
 714         {
 715             result_set_id = boost::io::str(
 716                 boost::format("%1%") % found_backend->m_result_set_sequence);
 717             found_backend->m_result_set_sequence++;
 718         }
 719         else
 720             result_set_id = "default";
 721     }
 722     found_backend->timestamp();
 723
 724     // we must search ...
 725     BackendSetPtr new_set(new BackendSet(result_set_id,
 726                                          databases, query));
 727     Z_Records *z_records = 0;
 728
 729     Package search_package(found_backend->m_session, package.origin());
 730     search_package.copy_filter(package);
 731
 732     if (!new_set->search(package, search_package,
 733                          apdu_req, found_backend, &z_records))
 734     {
 735         bc->remove_backend(found_backend);
 736         return; // search error
 737     }
 738
 739     if (z_records)
 740     {
 741         int condition = 0;
 742         if (z_records->which == Z_Records_NSD)
 743         {
 744             condition =
 745                 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
 746         }
 747         else if (z_records->which == Z_Records_multipleNSD)
 748         {
 749             if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
 750                 &&
 751
 752                 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
 753                 Z_DiagRec_defaultFormat)
 754             {
 755                 condition = get_diagnostic(
 756                     z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
 757
 758             }
 759         }
 760         if (m_p->m_restart && !session_restarted &&
 761             condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
 762         {
 763             package.log("session_shared", YLOG_LOG, "restart");
 764             bc->remove_backend(found_backend);
 765             session_restarted = true;
 766             found_backend.reset();
 767             goto restart;
 768
 769         }
 770
 771         if (condition)
 772         {
 773             mp::odr odr;
 774             if (apdu_req->which == Z_APDU_searchRequest)
 775             {
 776                 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
 777                                                            0, 0);
 778                 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 779                 *f_resp->searchStatus = Z_SearchResponse_none;
 780                 f_resp->records = z_records;
 781                 package.response() = f_apdu;
 782             }
 783             if (apdu_req->which == Z_APDU_presentRequest)
 784             {
 785                 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
 786                                                             0, 0);
 787                 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
 788                 f_resp->records = z_records;
 789                 package.response() = f_apdu;
 790             }
 791             bc->release_backend(found_backend);
 792             return; // search error
 793         }
 794     }
 795     if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
 796     {
 797         package.log("session_shared", YLOG_LOG, "restart");
 798         bc->remove_backend(found_backend);
 799         session_restarted = true;
 800         found_backend.reset();
 801         goto restart;
 802     }
 803
 804     found_set = new_set;
 805     found_set->timestamp();
 806     found_backend->m_sets.push_back(found_set);
 807 }
 808
 809 void yf::SessionShared::Frontend::search(mp::Package &package,
 810                                          Z_APDU *apdu_req)
 811 {
 812     Z_SearchRequest *req = apdu_req->u.searchRequest;
 813     FrontendSets::iterator fset_it =
 814         m_frontend_sets.find(req->resultSetName);
 815     if (fset_it != m_frontend_sets.end())
 816     {
 817         // result set already exist
 818         // if replace indicator is off: we return diagnostic if
 819         // result set already exist.
 820         if (*req->replaceIndicator == 0)
 821         {
 822             mp::odr odr;
 823             Z_APDU *apdu =
 824                 odr.create_searchResponse(
 825                     apdu_req,
 826                     YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
 827                     0);
 828             package.response() = apdu;
 829
 830             return;
 831         }
 832         m_frontend_sets.erase(fset_it);
 833     }
 834
 835     yazpp_1::Yaz_Z_Query query;
 836     query.set_Z_Query(req->query);
 837     Databases databases;
 838     int i;
 839     for (i = 0; i < req->num_databaseNames; i++)
 840         databases.push_back(req->databaseNames[i]);
 841
 842     BackendSetPtr found_set; // null
 843     BackendInstancePtr found_backend; // null
 844
 845     get_set(package, apdu_req, databases, query, found_backend, found_set);
 846     if (!found_set)
 847         return;
 848
 849     mp::odr odr;
 850     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
 851     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 852     *f_resp->resultCount = found_set->m_result_set_size;
 853     package.response() = f_apdu;
 854
 855     FrontendSetPtr fset(new FrontendSet(databases, query));
 856     m_frontend_sets[req->resultSetName] = fset;
 857
 858     m_backend_class->release_backend(found_backend);
 859 }
 860
 861 void yf::SessionShared::Frontend::present(mp::Package &package,
 862                                           Z_APDU *apdu_req)
 863 {
 864     mp::odr odr;
 865     Z_PresentRequest *req = apdu_req->u.presentRequest;
 866
 867     FrontendSets::iterator fset_it =
 868         m_frontend_sets.find(req->resultSetId);
 869
 870     if (fset_it == m_frontend_sets.end())
 871     {
 872         Z_APDU *apdu =
 873             odr.create_presentResponse(
 874                 apdu_req,
 875                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
 876                 req->resultSetId);
 877         package.response() = apdu;
 878         return;
 879     }
 880     FrontendSetPtr fset = fset_it->second;
 881
 882     Databases databases = fset->get_databases();
 883     yazpp_1::Yaz_Z_Query query = fset->get_query();
 884
 885     BackendClassPtr bc = m_backend_class;
 886     BackendSetPtr found_set; // null
 887     BackendInstancePtr found_backend;
 888
 889     get_set(package, apdu_req, databases, query, found_backend, found_set);
 890     if (!found_set)
 891         return;
 892
 893     Z_NamePlusRecordList *npr_res = 0;
 894     if (found_set->m_record_cache.lookup(odr, &npr_res,
 895                                          *req->resultSetStartPoint,
 896                                          *req->numberOfRecordsRequested,
 897                                          req->preferredRecordSyntax,
 898                                          req->recordComposition))
 899     {
 900         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
 901         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
 902
 903         yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
 904                 " records in cache %p",
 905                 *req->resultSetStartPoint,
 906                 *req->numberOfRecordsRequested,
 907                 &found_set->m_record_cache);
 908
 909         *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
 910         *f_resp->nextResultSetPosition =
 911             *req->resultSetStartPoint + *req->numberOfRecordsRequested;
 912         // f_resp->presentStatus assumed OK.
 913         f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
 914         f_resp->records->which = Z_Records_DBOSD;
 915         f_resp->records->u.databaseOrSurDiagnostics = npr_res;
 916         package.response() = f_apdu_res;
 917         bc->release_backend(found_backend);
 918         return;
 919     }
 920
 921     found_backend->timestamp();
 922
 923     Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
 924     Z_PresentRequest *p_req = p_apdu->u.presentRequest;
 925     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
 926     p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
 927     *p_req->resultSetStartPoint = *req->resultSetStartPoint;
 928     *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
 929     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
 930     p_req->recordComposition = req->recordComposition;
 931
 932     Package present_package(found_backend->m_session, package.origin());
 933     present_package.copy_filter(package);
 934
 935     present_package.request() = p_apdu;
 936
 937     present_package.move();
 938
 939     Z_GDU *gdu = present_package.response().get();
 940     if (!present_package.session().is_closed()
 941         && gdu && gdu->which == Z_GDU_Z3950
 942         && gdu->u.z3950->which == Z_APDU_presentResponse)
 943     {
 944         Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
 945         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
 946         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
 947
 948         f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
 949         f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
 950         f_resp->presentStatus= b_resp->presentStatus;
 951         f_resp->records = b_resp->records;
 952         f_resp->otherInfo = b_resp->otherInfo;
 953         package.response() = f_apdu_res;
 954
 955         if (b_resp->records && b_resp->records->which ==  Z_Records_DBOSD)
 956         {
 957             yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
 958                     " records to cache %p",
 959                     *req->resultSetStartPoint,
 960                     *f_resp->numberOfRecordsReturned,
 961                     &found_set->m_record_cache);
 962             found_set->m_record_cache.add(
 963                 odr,
 964                 b_resp->records->u.databaseOrSurDiagnostics,
 965                 *req->resultSetStartPoint,
 966                 *f_resp->numberOfRecordsReturned);
 967         }
 968         bc->release_backend(found_backend);
 969     }
 970     else
 971     {
 972         bc->remove_backend(found_backend);
 973         Z_APDU *f_apdu_res =
 974             odr.create_presentResponse(
 975                 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 976         package.response() = f_apdu_res;
 977     }
 978 }
 979
 980 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
 981                                        Z_APDU *apdu_req)
 982 {
 983     BackendClassPtr bc = m_backend_class;
 984     BackendInstancePtr backend = bc->get_backend(frontend_package);
 985     if (!backend)
 986     {
 987         mp::odr odr;
 988         Z_APDU *apdu = odr.create_scanResponse(
 989             apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 990         frontend_package.response() = apdu;
 991     }
 992     else
 993     {
 994         Package scan_package(backend->m_session, frontend_package.origin());
 995         backend->timestamp();
 996         scan_package.copy_filter(frontend_package);
 997         scan_package.request() = apdu_req;
 998         scan_package.move();
 999         frontend_package.response() = scan_package.response();
1000         if (scan_package.session().is_closed())
1001         {
1002             frontend_package.session().close();
1003             bc->remove_backend(backend);
1004         }
1005         else
1006             bc->release_backend(backend);
1007     }
1008 }
1009
1010 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1011 {
1012 }
1013
1014 void yf::SessionShared::Worker::operator() (void)
1015 {
1016     m_p->expire();
1017 }
1018
1019 void yf::SessionShared::BackendClass::expire_class()
1020 {
1021     time_t now;
1022     time(&now);
1023     boost::mutex::scoped_lock lock(m_mutex_backend_class);
1024     BackendInstanceList::iterator bit = m_backend_list.begin();
1025     while (bit != m_backend_list.end())
1026     {
1027         time_t last_use = (*bit)->m_time_last_use;
1028
1029         if ((*bit)->m_in_use)
1030         {
1031             bit++;
1032         }
1033         else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1034         {
1035             mp::odr odr;
1036             (*bit)->m_close_package->response() = odr.create_close(
1037                 0, Z_Close_lackOfActivity, 0);
1038             (*bit)->m_close_package->session().close();
1039             (*bit)->m_close_package->move();
1040
1041             bit = m_backend_list.erase(bit);
1042         }
1043         else
1044         {
1045             bit++;
1046         }
1047     }
1048 }
1049
1050 void yf::SessionShared::Rep::expire()
1051 {
1052     while (true)
1053     {
1054         boost::xtime xt;
1055         boost::xtime_get(&xt, boost::TIME_UTC);
1056         xt.sec += m_session_ttl / 3;
1057         boost::thread::sleep(xt);
1058
1059         BackendClassMap::const_iterator b_it = m_backend_map.begin();
1060         for (; b_it != m_backend_map.end(); b_it++)
1061             b_it->second->expire_class();
1062     }
1063 }
1064
1065 yf::SessionShared::Rep::Rep()
1066 {
1067     m_resultset_ttl = 30;
1068     m_resultset_max = 10;
1069     m_session_ttl = 90;
1070     m_optimize_search = true;
1071     m_restart = false;
1072     m_session_max = 100;
1073 }
1074
1075 void yf::SessionShared::Rep::start()
1076 {
1077     yf::SessionShared::Worker w(this);
1078     m_thrds.add_thread(new boost::thread(w));
1079 }
1080
1081 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1082 {
1083 }
1084
1085 yf::SessionShared::~SessionShared() {
1086 }
1087
1088 void yf::SessionShared::start() const
1089 {
1090     m_p->start();
1091 }
1092
1093 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1094 {
1095 }
1096
1097 yf::SessionShared::Frontend::~Frontend()
1098 {
1099 }
1100
1101 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1102 {
1103     boost::mutex::scoped_lock lock(m_mutex);
1104
1105     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1106
1107     while(true)
1108     {
1109         it = m_clients.find(package.session());
1110         if (it == m_clients.end())
1111             break;
1112
1113         if (!it->second->m_in_use)
1114         {
1115             it->second->m_in_use = true;
1116             return it->second;
1117         }
1118         m_cond_session_ready.wait(lock);
1119     }
1120     FrontendPtr f(new Frontend(this));
1121     m_clients[package.session()] = f;
1122     f->m_in_use = true;
1123     return f;
1124 }
1125
1126 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1127 {
1128     boost::mutex::scoped_lock lock(m_mutex);
1129     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1130
1131     it = m_clients.find(package.session());
1132     if (it != m_clients.end())
1133     {
1134         if (package.session().is_closed())
1135         {
1136             m_clients.erase(it);
1137         }
1138         else
1139         {
1140             it->second->m_in_use = false;
1141         }
1142         m_cond_session_ready.notify_all();
1143     }
1144 }
1145
1146
1147 void yf::SessionShared::process(mp::Package &package) const
1148 {
1149     FrontendPtr f = m_p->get_frontend(package);
1150
1151     Z_GDU *gdu = package.request().get();
1152
1153     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1154         Z_APDU_initRequest && !f->m_is_virtual)
1155     {
1156         m_p->init(package, gdu, f);
1157     }
1158     else if (!f->m_is_virtual)
1159         package.move();
1160     else if (gdu && gdu->which == Z_GDU_Z3950)
1161     {
1162         Z_APDU *apdu = gdu->u.z3950;
1163         if (apdu->which == Z_APDU_initRequest)
1164         {
1165             mp::odr odr;
1166
1167             package.response() = odr.create_close(
1168                 apdu,
1169                 Z_Close_protocolError,
1170                 "double init");
1171
1172             package.session().close();
1173         }
1174         else if (apdu->which == Z_APDU_close)
1175         {
1176             mp::odr odr;
1177
1178             package.response() = odr.create_close(
1179                 apdu,
1180                 Z_Close_peerAbort, "received close from client");
1181             package.session().close();
1182         }
1183         else if (apdu->which == Z_APDU_searchRequest)
1184         {
1185             f->search(package, apdu);
1186         }
1187         else if (apdu->which == Z_APDU_presentRequest)
1188         {
1189             f->present(package, apdu);
1190         }
1191         else if (apdu->which == Z_APDU_scanRequest)
1192         {
1193             f->scan(package, apdu);
1194         }
1195         else
1196         {
1197             mp::odr odr;
1198
1199             package.response() = odr.create_close(
1200                 apdu, Z_Close_protocolError,
1201                 "unsupported APDU in filter_session_shared");
1202
1203             package.session().close();
1204         }
1205     }
1206     m_p->release_frontend(package);
1207 }
1208
1209 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1210                                   const char *path)
1211 {
1212     for (ptr = ptr->children; ptr; ptr = ptr->next)
1213     {
1214         if (ptr->type != XML_ELEMENT_NODE)
1215             continue;
1216         if (!strcmp((const char *) ptr->name, "resultset"))
1217         {
1218             const struct _xmlAttr *attr;
1219             for (attr = ptr->properties; attr; attr = attr->next)
1220             {
1221                 if (!strcmp((const char *) attr->name, "ttl"))
1222                     m_p->m_resultset_ttl =
1223                         mp::xml::get_int(attr->children, 30);
1224                 else if (!strcmp((const char *) attr->name, "max"))
1225                 {
1226                     m_p->m_resultset_max =
1227                         mp::xml::get_int(attr->children, 10);
1228                 }
1229                 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1230                 {
1231                     m_p->m_optimize_search =
1232                         mp::xml::get_bool(attr->children, true);
1233                 }
1234                 else if (!strcmp((const char *) attr->name, "restart"))
1235                 {
1236                     m_p->m_restart = mp::xml::get_bool(attr->children, true);
1237                 }
1238                 else
1239                     throw mp::filter::FilterException(
1240                         "Bad attribute " + std::string((const char *)
1241                                                        attr->name));
1242             }
1243         }
1244         else if (!strcmp((const char *) ptr->name, "session"))
1245         {
1246             const struct _xmlAttr *attr;
1247             for (attr = ptr->properties; attr; attr = attr->next)
1248             {
1249                 if (!strcmp((const char *) attr->name, "ttl"))
1250                     m_p->m_session_ttl =
1251                         mp::xml::get_int(attr->children, 90);
1252                 else if (!strcmp((const char *) attr->name, "max"))
1253                     m_p->m_session_max =
1254                         mp::xml::get_int(attr->children, 100);
1255                 else
1256                     throw mp::filter::FilterException(
1257                         "Bad attribute " + std::string((const char *)
1258                                                        attr->name));
1259             }
1260         }
1261         else
1262         {
1263             throw mp::filter::FilterException("Bad element "
1264                                                + std::string((const char *)
1265                                                              ptr->name));
1266         }
1267     }
1268 }
1269
1270 static mp::filter::Base* filter_creator()
1271 {
1272     return new mp::filter::SessionShared;
1273 }
1274
1275 extern "C" {
1276     struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1277         0,
1278         "session_shared",
1279         filter_creator
1280     };
1281 }
1282
1283 /*
1284  * Local variables:
1285  * c-basic-offset: 4
1286  * c-file-style: "Stroustrup"
1287  * indent-tabs-mode: nil
1288  * End:
1289  * vim: shiftwidth=4 tabstop=8 expandtab
1290  */
1291