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