src/filter_session_shared.cpp

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