src/filter_session_shared.cpp

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