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             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_set->timestamp();
 671                             // found matching set. No need to search again
 672                             return;
 673                         }
 674                     }
 675                 }
 676             }
 677         }
 678         override_set(found_backend, result_set_id, databases, out_of_sessions);
 679         if (found_backend)
 680             bc->use_backend(found_backend);
 681     }
 682     if (!found_backend)
 683     {
 684         // create a new backend set (and new set) if we're not out of sessions
 685         if (!out_of_sessions)
 686             found_backend = bc->create_backend(package);
 687
 688         if (!found_backend)
 689         {
 690             Z_APDU *f_apdu = 0;
 691             mp::odr odr;
 692             const char *addinfo = 0;
 693
 694             if (out_of_sessions)
 695                 addinfo = "session_shared: all sessions in use";
 696             if (apdu_req->which == Z_APDU_searchRequest)
 697             {
 698                 f_apdu = odr.create_searchResponse(
 699                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 700             }
 701             else if (apdu_req->which == Z_APDU_presentRequest)
 702             {
 703                 f_apdu = odr.create_presentResponse(
 704                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 705             }
 706             else
 707             {
 708                 f_apdu = odr.create_close(
 709                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 710             }
 711             package.response() = f_apdu;
 712             return;
 713         }
 714         if (bc->m_named_result_sets)
 715         {
 716             result_set_id = boost::io::str(
 717                 boost::format("%1%") % found_backend->m_result_set_sequence);
 718             found_backend->m_result_set_sequence++;
 719         }
 720         else
 721             result_set_id = "default";
 722     }
 723     found_backend->timestamp();
 724
 725     // we must search ...
 726     BackendSetPtr new_set(new BackendSet(result_set_id,
 727                                          databases, query));
 728     Z_Records *z_records = 0;
 729
 730     Package search_package(found_backend->m_session, package.origin());
 731     search_package.copy_filter(package);
 732
 733     if (!new_set->search(package, search_package,
 734                          apdu_req, found_backend, &z_records))
 735     {
 736         bc->remove_backend(found_backend);
 737         return; // search error
 738     }
 739
 740     if (z_records)
 741     {
 742         int condition = 0;
 743         if (z_records->which == Z_Records_NSD)
 744         {
 745             condition =
 746                 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
 747         }
 748         else if (z_records->which == Z_Records_multipleNSD)
 749         {
 750             if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
 751                 &&
 752
 753                 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
 754                 Z_DiagRec_defaultFormat)
 755             {
 756                 condition = get_diagnostic(
 757                     z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
 758
 759             }
 760         }
 761         if (m_p->m_restart && !session_restarted &&
 762             condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
 763         {
 764             package.log("session_shared", YLOG_LOG, "restart");
 765             bc->remove_backend(found_backend);
 766             session_restarted = true;
 767             found_backend.reset();
 768             goto restart;
 769
 770         }
 771
 772         if (condition)
 773         {
 774             mp::odr odr;
 775             if (apdu_req->which == Z_APDU_searchRequest)
 776             {
 777                 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
 778                                                            0, 0);
 779                 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 780                 *f_resp->searchStatus = Z_SearchResponse_none;
 781                 f_resp->records = z_records;
 782                 package.response() = f_apdu;
 783             }
 784             if (apdu_req->which == Z_APDU_presentRequest)
 785             {
 786                 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
 787                                                             0, 0);
 788                 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
 789                 f_resp->records = z_records;
 790                 package.response() = f_apdu;
 791             }
 792             bc->release_backend(found_backend);
 793             return; // search error
 794         }
 795     }
 796     if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
 797     {
 798         package.log("session_shared", YLOG_LOG, "restart");
 799         bc->remove_backend(found_backend);
 800         session_restarted = true;
 801         found_backend.reset();
 802         goto restart;
 803     }
 804
 805     found_set = new_set;
 806     found_set->timestamp();
 807     found_backend->m_sets.push_back(found_set);
 808 }
 809
 810 void yf::SessionShared::Frontend::search(mp::Package &package,
 811                                          Z_APDU *apdu_req)
 812 {
 813     Z_SearchRequest *req = apdu_req->u.searchRequest;
 814     FrontendSets::iterator fset_it =
 815         m_frontend_sets.find(req->resultSetName);
 816     if (fset_it != m_frontend_sets.end())
 817     {
 818         // result set already exist
 819         // if replace indicator is off: we return diagnostic if
 820         // result set already exist.
 821         if (*req->replaceIndicator == 0)
 822         {
 823             mp::odr odr;
 824             Z_APDU *apdu =
 825                 odr.create_searchResponse(
 826                     apdu_req,
 827                     YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
 828                     0);
 829             package.response() = apdu;
 830
 831             return;
 832         }
 833         m_frontend_sets.erase(fset_it);
 834     }
 835
 836     yazpp_1::Yaz_Z_Query query;
 837     query.set_Z_Query(req->query);
 838     Databases databases;
 839     int i;
 840     for (i = 0; i < req->num_databaseNames; i++)
 841         databases.push_back(req->databaseNames[i]);
 842
 843     BackendSetPtr found_set; // null
 844     BackendInstancePtr found_backend; // null
 845
 846     get_set(package, apdu_req, databases, query, found_backend, found_set);
 847     if (!found_set)
 848         return;
 849
 850     mp::odr odr;
 851     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
 852     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 853     *f_resp->resultCount = found_set->m_result_set_size;
 854     package.response() = f_apdu;
 855
 856     FrontendSetPtr fset(new FrontendSet(databases, query));
 857     m_frontend_sets[req->resultSetName] = fset;
 858
 859     m_backend_class->release_backend(found_backend);
 860 }
 861
 862 void yf::SessionShared::Frontend::present(mp::Package &package,
 863                                           Z_APDU *apdu_req)
 864 {
 865     mp::odr odr;
 866     Z_PresentRequest *req = apdu_req->u.presentRequest;
 867
 868     FrontendSets::iterator fset_it =
 869         m_frontend_sets.find(req->resultSetId);
 870
 871     if (fset_it == m_frontend_sets.end())
 872     {
 873         Z_APDU *apdu =
 874             odr.create_presentResponse(
 875                 apdu_req,
 876                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
 877                 req->resultSetId);
 878         package.response() = apdu;
 879         return;
 880     }
 881     FrontendSetPtr fset = fset_it->second;
 882
 883     Databases databases = fset->get_databases();
 884     yazpp_1::Yaz_Z_Query query = fset->get_query();
 885
 886     BackendClassPtr bc = m_backend_class;
 887     BackendSetPtr found_set; // null
 888     BackendInstancePtr found_backend;
 889
 890     get_set(package, apdu_req, databases, query, found_backend, found_set);
 891     if (!found_set)
 892         return;
 893
 894     Z_NamePlusRecordList *npr_res = 0;
 895     if (found_set->m_record_cache.lookup(odr, &npr_res,
 896                                          *req->resultSetStartPoint,
 897                                          *req->numberOfRecordsRequested,
 898                                          req->preferredRecordSyntax,
 899                                          req->recordComposition))
 900     {
 901         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
 902         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
 903
 904         yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
 905                 " records in cache %p",
 906                 *req->resultSetStartPoint,
 907                 *req->numberOfRecordsRequested,
 908                 &found_set->m_record_cache);
 909
 910         *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
 911         *f_resp->nextResultSetPosition =
 912             *req->resultSetStartPoint + *req->numberOfRecordsRequested;
 913         // f_resp->presentStatus assumed OK.
 914         f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
 915         f_resp->records->which = Z_Records_DBOSD;
 916         f_resp->records->u.databaseOrSurDiagnostics = npr_res;
 917         package.response() = f_apdu_res;
 918         bc->release_backend(found_backend);
 919         return;
 920     }
 921
 922     found_backend->timestamp();
 923
 924     Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
 925     Z_PresentRequest *p_req = p_apdu->u.presentRequest;
 926     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
 927     p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
 928     *p_req->resultSetStartPoint = *req->resultSetStartPoint;
 929     *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
 930     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
 931     p_req->recordComposition = req->recordComposition;
 932
 933     Package present_package(found_backend->m_session, package.origin());
 934     present_package.copy_filter(package);
 935
 936     present_package.request() = p_apdu;
 937
 938     present_package.move();
 939
 940     Z_GDU *gdu = present_package.response().get();
 941     if (!present_package.session().is_closed()
 942         && gdu && gdu->which == Z_GDU_Z3950
 943         && gdu->u.z3950->which == Z_APDU_presentResponse)
 944     {
 945         Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
 946         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
 947         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
 948
 949         f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
 950         f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
 951         f_resp->presentStatus= b_resp->presentStatus;
 952         f_resp->records = b_resp->records;
 953         f_resp->otherInfo = b_resp->otherInfo;
 954         package.response() = f_apdu_res;
 955
 956         if (b_resp->records && b_resp->records->which ==  Z_Records_DBOSD)
 957         {
 958             yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
 959                     " records to cache %p",
 960                     *req->resultSetStartPoint,
 961                     *f_resp->numberOfRecordsReturned,
 962                     &found_set->m_record_cache);
 963             found_set->m_record_cache.add(
 964                 odr,
 965                 b_resp->records->u.databaseOrSurDiagnostics,
 966                 *req->resultSetStartPoint,
 967                 *f_resp->numberOfRecordsReturned);
 968         }
 969         bc->release_backend(found_backend);
 970     }
 971     else
 972     {
 973         bc->remove_backend(found_backend);
 974         Z_APDU *f_apdu_res =
 975             odr.create_presentResponse(
 976                 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 977         package.response() = f_apdu_res;
 978     }
 979 }
 980
 981 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
 982                                        Z_APDU *apdu_req)
 983 {
 984     BackendClassPtr bc = m_backend_class;
 985     BackendInstancePtr backend = bc->get_backend(frontend_package);
 986     if (!backend)
 987     {
 988         mp::odr odr;
 989         Z_APDU *apdu = odr.create_scanResponse(
 990             apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
 991         frontend_package.response() = apdu;
 992     }
 993     else
 994     {
 995         Package scan_package(backend->m_session, frontend_package.origin());
 996         backend->timestamp();
 997         scan_package.copy_filter(frontend_package);
 998         scan_package.request() = apdu_req;
 999         scan_package.move();
1000         frontend_package.response() = scan_package.response();
1001         if (scan_package.session().is_closed())
1002         {
1003             frontend_package.session().close();
1004             bc->remove_backend(backend);
1005         }
1006         else
1007             bc->release_backend(backend);
1008     }
1009 }
1010
1011 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1012 {
1013 }
1014
1015 void yf::SessionShared::Worker::operator() (void)
1016 {
1017     m_p->expire();
1018 }
1019
1020 void yf::SessionShared::BackendClass::expire_class()
1021 {
1022     time_t now;
1023     time(&now);
1024     boost::mutex::scoped_lock lock(m_mutex_backend_class);
1025     BackendInstanceList::iterator bit = m_backend_list.begin();
1026     while (bit != m_backend_list.end())
1027     {
1028         time_t last_use = (*bit)->m_time_last_use;
1029
1030         if ((*bit)->m_in_use)
1031         {
1032             bit++;
1033         }
1034         else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1035         {
1036             mp::odr odr;
1037             (*bit)->m_close_package->response() = odr.create_close(
1038                 0, Z_Close_lackOfActivity, 0);
1039             (*bit)->m_close_package->session().close();
1040             (*bit)->m_close_package->move();
1041
1042             bit = m_backend_list.erase(bit);
1043         }
1044         else
1045         {
1046             bit++;
1047         }
1048     }
1049 }
1050
1051 void yf::SessionShared::Rep::expire()
1052 {
1053     while (true)
1054     {
1055         boost::xtime xt;
1056         boost::xtime_get(&xt, boost::TIME_UTC);
1057         xt.sec += m_session_ttl / 3;
1058         boost::thread::sleep(xt);
1059
1060         BackendClassMap::const_iterator b_it = m_backend_map.begin();
1061         for (; b_it != m_backend_map.end(); b_it++)
1062             b_it->second->expire_class();
1063     }
1064 }
1065
1066 yf::SessionShared::Rep::Rep()
1067 {
1068     m_resultset_ttl = 30;
1069     m_resultset_max = 10;
1070     m_session_ttl = 90;
1071     m_optimize_search = true;
1072     m_restart = true;
1073     m_session_max = 100;
1074 }
1075
1076 void yf::SessionShared::Rep::start()
1077 {
1078     yf::SessionShared::Worker w(this);
1079     m_thrds.add_thread(new boost::thread(w));
1080 }
1081
1082 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1083 {
1084 }
1085
1086 yf::SessionShared::~SessionShared() {
1087 }
1088
1089 void yf::SessionShared::start() const
1090 {
1091     m_p->start();
1092 }
1093
1094 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1095 {
1096 }
1097
1098 yf::SessionShared::Frontend::~Frontend()
1099 {
1100 }
1101
1102 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1103 {
1104     boost::mutex::scoped_lock lock(m_mutex);
1105
1106     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1107
1108     while(true)
1109     {
1110         it = m_clients.find(package.session());
1111         if (it == m_clients.end())
1112             break;
1113
1114         if (!it->second->m_in_use)
1115         {
1116             it->second->m_in_use = true;
1117             return it->second;
1118         }
1119         m_cond_session_ready.wait(lock);
1120     }
1121     FrontendPtr f(new Frontend(this));
1122     m_clients[package.session()] = f;
1123     f->m_in_use = true;
1124     return f;
1125 }
1126
1127 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1128 {
1129     boost::mutex::scoped_lock lock(m_mutex);
1130     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1131
1132     it = m_clients.find(package.session());
1133     if (it != m_clients.end())
1134     {
1135         if (package.session().is_closed())
1136         {
1137             m_clients.erase(it);
1138         }
1139         else
1140         {
1141             it->second->m_in_use = false;
1142         }
1143         m_cond_session_ready.notify_all();
1144     }
1145 }
1146
1147
1148 void yf::SessionShared::process(mp::Package &package) const
1149 {
1150     FrontendPtr f = m_p->get_frontend(package);
1151
1152     Z_GDU *gdu = package.request().get();
1153
1154     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1155         Z_APDU_initRequest && !f->m_is_virtual)
1156     {
1157         m_p->init(package, gdu, f);
1158     }
1159     else if (!f->m_is_virtual)
1160         package.move();
1161     else if (gdu && gdu->which == Z_GDU_Z3950)
1162     {
1163         Z_APDU *apdu = gdu->u.z3950;
1164         if (apdu->which == Z_APDU_initRequest)
1165         {
1166             mp::odr odr;
1167
1168             package.response() = odr.create_close(
1169                 apdu,
1170                 Z_Close_protocolError,
1171                 "double init");
1172
1173             package.session().close();
1174         }
1175         else if (apdu->which == Z_APDU_close)
1176         {
1177             mp::odr odr;
1178
1179             package.response() = odr.create_close(
1180                 apdu,
1181                 Z_Close_peerAbort, "received close from client");
1182             package.session().close();
1183         }
1184         else if (apdu->which == Z_APDU_searchRequest)
1185         {
1186             f->search(package, apdu);
1187         }
1188         else if (apdu->which == Z_APDU_presentRequest)
1189         {
1190             f->present(package, apdu);
1191         }
1192         else if (apdu->which == Z_APDU_scanRequest)
1193         {
1194             f->scan(package, apdu);
1195         }
1196         else
1197         {
1198             mp::odr odr;
1199
1200             package.response() = odr.create_close(
1201                 apdu, Z_Close_protocolError,
1202                 "unsupported APDU in filter_session_shared");
1203
1204             package.session().close();
1205         }
1206     }
1207     m_p->release_frontend(package);
1208 }
1209
1210 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1211                                   const char *path)
1212 {
1213     for (ptr = ptr->children; ptr; ptr = ptr->next)
1214     {
1215         if (ptr->type != XML_ELEMENT_NODE)
1216             continue;
1217         if (!strcmp((const char *) ptr->name, "resultset"))
1218         {
1219             const struct _xmlAttr *attr;
1220             for (attr = ptr->properties; attr; attr = attr->next)
1221             {
1222                 if (!strcmp((const char *) attr->name, "ttl"))
1223                     m_p->m_resultset_ttl =
1224                         mp::xml::get_int(attr->children, 30);
1225                 else if (!strcmp((const char *) attr->name, "max"))
1226                 {
1227                     m_p->m_resultset_max =
1228                         mp::xml::get_int(attr->children, 10);
1229                 }
1230                 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1231                 {
1232                     m_p->m_optimize_search =
1233                         mp::xml::get_bool(attr->children, true);
1234                 }
1235                 else if (!strcmp((const char *) attr->name, "restart"))
1236                 {
1237                     m_p->m_restart = mp::xml::get_bool(attr->children, true);
1238                 }
1239                 else
1240                     throw mp::filter::FilterException(
1241                         "Bad attribute " + std::string((const char *)
1242                                                        attr->name));
1243             }
1244         }
1245         else if (!strcmp((const char *) ptr->name, "session"))
1246         {
1247             const struct _xmlAttr *attr;
1248             for (attr = ptr->properties; attr; attr = attr->next)
1249             {
1250                 if (!strcmp((const char *) attr->name, "ttl"))
1251                     m_p->m_session_ttl =
1252                         mp::xml::get_int(attr->children, 90);
1253                 else if (!strcmp((const char *) attr->name, "max"))
1254                     m_p->m_session_max =
1255                         mp::xml::get_int(attr->children, 100);
1256                 else
1257                     throw mp::filter::FilterException(
1258                         "Bad attribute " + std::string((const char *)
1259                                                        attr->name));
1260             }
1261         }
1262         else
1263         {
1264             throw mp::filter::FilterException("Bad element "
1265                                                + std::string((const char *)
1266                                                              ptr->name));
1267         }
1268     }
1269 }
1270
1271 static mp::filter::Base* filter_creator()
1272 {
1273     return new mp::filter::SessionShared;
1274 }
1275
1276 extern "C" {
1277     struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1278         0,
1279         "session_shared",
1280         filter_creator
1281     };
1282 }
1283
1284 /*
1285  * Local variables:
1286  * c-basic-offset: 4
1287  * c-file-style: "Stroustrup"
1288  * indent-tabs-mode: nil
1289  * End:
1290  * vim: shiftwidth=4 tabstop=8 expandtab
1291  */
1292