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