src/filter_session_shared.cpp

   1 /* This file is part of Metaproxy.
   2    Copyright (C) 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/copy_types.h>
  35 #include <yaz/log.h>
  36 #include <yaz/zgdu.h>
  37 #include <yaz/otherinfo.h>
  38 #include <yaz/diagbib1.h>
  39 #include <yazpp/z-query.h>
  40 #include <yazpp/record-cache.h>
  41 #include <map>
  42 #include <iostream>
  43 #include <time.h>
  44 #include <limits.h>
  45
  46 namespace mp = metaproxy_1;
  47 namespace yf = metaproxy_1::filter;
  48
  49 namespace metaproxy_1 {
  50
  51     namespace filter {
  52         // key for session.. We'll only share sessions with same InitKey
  53         class SessionShared::InitKey {
  54         public:
  55             bool operator < (const SessionShared::InitKey &k) const;
  56             InitKey(Z_InitRequest *req);
  57             InitKey(const InitKey &);
  58             ~InitKey();
  59         private:
  60             char *m_idAuthentication_buf;
  61             int m_idAuthentication_size;
  62             char *m_otherInfo_buf;
  63             int m_otherInfo_size;
  64             ODR m_odr;
  65         };
  66         // worker thread .. for expiry of sessions
  67         class SessionShared::Worker {
  68         public:
  69             Worker(SessionShared::Rep *rep);
  70             void operator() (void);
  71         private:
  72             SessionShared::Rep *m_p;
  73         };
  74         // backend result set
  75         class SessionShared::BackendSet {
  76         public:
  77             std::string m_result_set_id;
  78             Databases m_databases;
  79             Odr_int m_result_set_size;
  80             yazpp_1::Yaz_Z_Query m_query;
  81             time_t m_time_last_use;
  82             void timestamp();
  83             yazpp_1::RecordCache m_record_cache;
  84
  85             Z_OtherInformation *additionalSearchInfoRequest;
  86             Z_OtherInformation *additionalSearchInfoResponse;
  87             NMEM mem_additionalSearchInfo;
  88             BackendSet(
  89                 const std::string &result_set_id,
  90                 const Databases &databases,
  91                 const yazpp_1::Yaz_Z_Query &query,
  92                 Z_OtherInformation *additionalSearchInfoRequest);
  93             ~BackendSet();
  94             bool search(
  95                 Package &frontend_package,
  96                 Package &search_package,
  97                 const Z_APDU *apdu_req,
  98                 const BackendInstancePtr bp,
  99                 Z_Records **z_records);
 100         };
 101         // backend connection instance
 102         class SessionShared::BackendInstance {
 103             friend class Rep;
 104             friend class BackendClass;
 105             friend class BackendSet;
 106         public:
 107             mp::Session m_session;
 108             BackendSetList m_sets;
 109             bool m_in_use;
 110             int m_sequence_this;
 111             int m_result_set_sequence;
 112             time_t m_time_last_use;
 113             mp::Package * m_close_package;
 114             ~BackendInstance();
 115             void timestamp();
 116         };
 117         // backends of some class (all with same InitKey)
 118         class SessionShared::BackendClass : boost::noncopyable {
 119             friend class Rep;
 120             friend struct Frontend;
 121             bool m_named_result_sets;
 122             BackendInstanceList m_backend_list;
 123             BackendInstancePtr create_backend(const Package &package);
 124             void remove_backend(BackendInstancePtr b);
 125             BackendInstancePtr get_backend(const Package &package);
 126             void use_backend(BackendInstancePtr b);
 127             void release_backend(BackendInstancePtr b);
 128             bool expire_instances();
 129             yazpp_1::GDU m_init_request;
 130             yazpp_1::GDU m_init_response;
 131             boost::mutex m_mutex_backend_class;
 132             boost::condition m_cond_set_ready;
 133             int m_sequence_top;
 134             time_t m_backend_set_ttl;
 135             time_t m_backend_expiry_ttl;
 136             size_t m_backend_set_max;
 137             Odr_int m_preferredMessageSize;
 138             Odr_int m_maximumRecordSize;
 139             int m_no_failed;
 140             int m_no_succeeded;
 141             int m_no_init;
 142         public:
 143             BackendClass(const yazpp_1::GDU &init_request,
 144                          int resultset_ttl,
 145                          int resultset_max,
 146                          int session_ttl,
 147                          Odr_int preferredRecordSize,
 148                          Odr_int maximumRecordSize);
 149             ~BackendClass();
 150         };
 151         // frontend result set
 152         class SessionShared::FrontendSet {
 153             Databases m_databases;
 154             yazpp_1::Yaz_Z_Query m_query;
 155         public:
 156             const Databases &get_databases();
 157             const yazpp_1::Yaz_Z_Query &get_query();
 158             FrontendSet(
 159                 const Databases &databases,
 160                 const yazpp_1::Yaz_Z_Query &query);
 161             FrontendSet();
 162         };
 163         // frontend session
 164         struct SessionShared::Frontend {
 165             Frontend(Rep *rep);
 166             ~Frontend();
 167             bool m_is_virtual;
 168             bool m_in_use;
 169             Z_Options m_init_options;
 170             void search(Package &package, Z_APDU *apdu);
 171             void present(Package &package, Z_APDU *apdu);
 172             void scan(Package &package, Z_APDU *apdu);
 173
 174             int result_set_ref(ODR o,
 175                                const Databases &databases,
 176                                Z_RPNStructure *s, std::string &rset);
 177             void get_set(mp::Package &package,
 178                          const Z_APDU *apdu_req,
 179                          const Databases &databases,
 180                          yazpp_1::Yaz_Z_Query &query,
 181                          BackendInstancePtr &found_backend,
 182                          BackendSetPtr &found_set);
 183             void override_set(BackendInstancePtr &found_backend,
 184                               std::string &result_set_id,
 185                               const Databases &databases,
 186                               bool out_of_sessions);
 187
 188             Rep *m_p;
 189             BackendClassPtr m_backend_class;
 190             FrontendSets m_frontend_sets;
 191         };
 192         // representation
 193         class SessionShared::Rep {
 194             friend class SessionShared;
 195             friend struct Frontend;
 196
 197             FrontendPtr get_frontend(Package &package);
 198             void release_frontend(Package &package);
 199             Rep();
 200         public:
 201             ~Rep();
 202             void expire();
 203         private:
 204             void expire_classes();
 205             void stat();
 206             void init(Package &package, const Z_GDU *gdu,
 207                       FrontendPtr frontend);
 208             void start();
 209             boost::mutex m_mutex;
 210             boost::condition m_cond_session_ready;
 211             boost::condition m_cond_expire_ready;
 212             std::map<mp::Session, FrontendPtr> m_clients;
 213
 214             BackendClassMap m_backend_map;
 215             boost::mutex m_mutex_backend_map;
 216             boost::thread_group m_thrds;
 217             int m_resultset_ttl;
 218             int m_resultset_max;
 219             int m_session_ttl;
 220             bool m_optimize_search;
 221             bool m_restart;
 222             int m_session_max;
 223             Odr_int m_preferredMessageSize;
 224             Odr_int m_maximumRecordSize;
 225             bool close_down;
 226         };
 227     }
 228 }
 229
 230 yf::SessionShared::FrontendSet::FrontendSet(
 231     const Databases &databases,
 232     const yazpp_1::Yaz_Z_Query &query)
 233     : m_databases(databases), m_query(query)
 234 {
 235 }
 236
 237 const yf::SessionShared::Databases &
 238 yf::SessionShared::FrontendSet::get_databases()
 239 {
 240     return m_databases;
 241 }
 242
 243 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
 244 {
 245     return m_query;
 246 }
 247
 248 yf::SessionShared::InitKey::InitKey(const InitKey &k)
 249 {
 250     m_odr = odr_createmem(ODR_ENCODE);
 251
 252     m_idAuthentication_size =  k.m_idAuthentication_size;
 253     m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
 254     memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
 255            m_idAuthentication_size);
 256
 257     m_otherInfo_size =  k.m_otherInfo_size;
 258     m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
 259     memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
 260            m_otherInfo_size);
 261 }
 262
 263 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
 264 {
 265     m_odr = odr_createmem(ODR_ENCODE);
 266
 267     Z_IdAuthentication *t = req->idAuthentication;
 268     z_IdAuthentication(m_odr, &t, 1, 0);
 269     m_idAuthentication_buf =
 270         odr_getbuf(m_odr, &m_idAuthentication_size, 0);
 271
 272     Z_OtherInformation *o = req->otherInfo;
 273     z_OtherInformation(m_odr, &o, 1, 0);
 274     m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
 275 }
 276
 277 yf::SessionShared::InitKey::~InitKey()
 278 {
 279     odr_destroy(m_odr);
 280 }
 281
 282 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
 283     const
 284 {
 285     int c;
 286     c = mp::util::memcmp2(
 287         (void*) m_idAuthentication_buf, m_idAuthentication_size,
 288         (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
 289     if (c < 0)
 290         return true;
 291     else if (c > 0)
 292         return false;
 293
 294     c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
 295                           (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
 296     if (c < 0)
 297         return true;
 298     else if (c > 0)
 299         return false;
 300     return false;
 301 }
 302
 303 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
 304 {
 305     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 306     m_cond_set_ready.notify_all();
 307     b->m_in_use = false;
 308 }
 309
 310
 311 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
 312 {
 313     {
 314         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 315         BackendInstanceList::iterator it = m_backend_list.begin();
 316         for (;;)
 317         {
 318             if (it == m_backend_list.end())
 319                 return;
 320             if (*it == b)
 321             {
 322                 it = m_backend_list.erase(it);
 323                 break;
 324             }
 325             it++;
 326         }
 327     }
 328     mp::odr odr;
 329     b->m_close_package->response() = odr.create_close(
 330         0, Z_Close_lackOfActivity, 0);
 331     b->m_close_package->session().close();
 332     b->m_close_package->move();
 333 }
 334
 335
 336 yf::SessionShared::BackendInstancePtr
 337 yf::SessionShared::BackendClass::get_backend(
 338     const mp::Package &frontend_package)
 339 {
 340     {
 341         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 342
 343         BackendInstanceList::const_iterator it = m_backend_list.begin();
 344
 345         BackendInstancePtr backend1; // null
 346
 347         for (; it != m_backend_list.end(); it++)
 348         {
 349             if (!(*it)->m_in_use)
 350             {
 351                 if (!backend1
 352                     || (*it)->m_sequence_this < backend1->m_sequence_this)
 353                     backend1 = *it;
 354             }
 355         }
 356         if (backend1)
 357         {
 358             use_backend(backend1);
 359             return backend1;
 360         }
 361     }
 362     return create_backend(frontend_package);
 363 }
 364
 365 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
 366 {
 367     backend->m_in_use = true;
 368     backend->m_sequence_this = m_sequence_top++;
 369 }
 370
 371 void yf::SessionShared::BackendInstance::timestamp()
 372 {
 373     assert(m_in_use);
 374     time(&m_time_last_use);
 375 }
 376
 377 yf::SessionShared::BackendInstance::~BackendInstance()
 378 {
 379     if (m_close_package)
 380     {
 381         mp::odr odr;
 382         m_close_package->response() = odr.create_close(
 383             0, Z_Close_lackOfActivity, 0);
 384         m_close_package->session().close();
 385         m_close_package->move();
 386     }
 387     delete m_close_package;
 388 }
 389
 390 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
 391     const mp::Package &frontend_package)
 392 {
 393     BackendInstancePtr null;
 394     {
 395         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 396         if (m_no_failed && !m_no_succeeded)
 397         {
 398             m_no_failed++;
 399             return null;
 400         }
 401     }
 402     BackendInstancePtr bp(new BackendInstance);
 403     bp->m_close_package =
 404         new mp::Package(bp->m_session, frontend_package.origin());
 405     bp->m_close_package->copy_filter(frontend_package);
 406
 407     Package init_package(bp->m_session, frontend_package.origin());
 408
 409     init_package.copy_filter(frontend_package);
 410
 411     yazpp_1::GDU actual_init_request = m_init_request;
 412     Z_GDU *init_pdu = actual_init_request.get();
 413
 414     assert(init_pdu->which == Z_GDU_Z3950);
 415     assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
 416
 417     Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
 418     ODR_MASK_ZERO(req->options);
 419
 420     ODR_MASK_SET(req->options, Z_Options_search);
 421     ODR_MASK_SET(req->options, Z_Options_present);
 422     ODR_MASK_SET(req->options, Z_Options_namedResultSets);
 423     ODR_MASK_SET(req->options, Z_Options_scan);
 424
 425     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
 426     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
 427     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
 428
 429     if (m_preferredMessageSize)
 430         *req->preferredMessageSize = m_preferredMessageSize;
 431     if (m_maximumRecordSize)
 432         *req->maximumRecordSize = m_maximumRecordSize;
 433
 434     init_package.request() = init_pdu;
 435
 436     {
 437         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 438         m_no_init++;
 439     }
 440
 441     init_package.move();
 442
 443     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 444
 445     m_named_result_sets = false;
 446     Z_GDU *gdu = init_package.response().get();
 447
 448     if (gdu && gdu->which == Z_GDU_Z3950
 449         && gdu->u.z3950->which == Z_APDU_initResponse)
 450     {
 451         Z_InitResponse *res = gdu->u.z3950->u.initResponse;
 452         m_init_response = gdu->u.z3950;
 453         if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
 454         {
 455             m_named_result_sets = true;
 456         }
 457         if (*gdu->u.z3950->u.initResponse->result
 458             && !init_package.session().is_closed())
 459         {
 460             bp->m_in_use = true;
 461             time(&bp->m_time_last_use);
 462             bp->m_sequence_this = 0;
 463             bp->m_result_set_sequence = 0;
 464             m_backend_list.push_back(bp);
 465             m_no_succeeded++;
 466             return bp;
 467         }
 468     }
 469     else
 470     {
 471         yazpp_1::GDU empty_gdu;
 472         m_init_response = empty_gdu;
 473     }
 474
 475     if (!init_package.session().is_closed())
 476     {
 477         init_package.copy_filter(frontend_package);
 478         init_package.session().close();
 479         init_package.move();
 480     }
 481     m_no_failed++;
 482     return null;
 483 }
 484
 485
 486 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
 487                                               int resultset_ttl,
 488                                               int resultset_max,
 489                                               int session_ttl,
 490                                               Odr_int preferredMessageSize,
 491                                               Odr_int maximumRecordSize)
 492     : m_named_result_sets(false), m_init_request(init_request),
 493       m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
 494       m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
 495       m_preferredMessageSize(preferredMessageSize),
 496       m_maximumRecordSize(maximumRecordSize),
 497       m_no_failed(0), m_no_succeeded(0), m_no_init(0)
 498 {}
 499
 500 yf::SessionShared::BackendClass::~BackendClass()
 501 {}
 502
 503 void yf::SessionShared::Rep::stat()
 504 {
 505     int no_classes = 0;
 506     int no_instances = 0;
 507     BackendClassMap::const_iterator it;
 508     {
 509         boost::mutex::scoped_lock lock(m_mutex_backend_map);
 510         for (it = m_backend_map.begin(); it != m_backend_map.end(); it++)
 511         {
 512             BackendClassPtr bc = it->second;
 513             no_classes++;
 514             BackendInstanceList::iterator bit = bc->m_backend_list.begin();
 515             for (; bit != bc->m_backend_list.end(); bit++)
 516                 no_instances++;
 517         }
 518     }
 519 }
 520
 521 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
 522                                   FrontendPtr frontend)
 523 {
 524     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 525
 526     frontend->m_is_virtual = true;
 527     frontend->m_init_options = *req->options;
 528     InitKey k(req);
 529     {
 530         boost::mutex::scoped_lock lock(m_mutex_backend_map);
 531         BackendClassMap::const_iterator it;
 532         it = m_backend_map.find(k);
 533         if (it == m_backend_map.end())
 534         {
 535             BackendClassPtr b(new BackendClass(gdu->u.z3950,
 536                                                m_resultset_ttl,
 537                                                m_resultset_max,
 538                                                m_session_ttl,
 539                                                m_preferredMessageSize,
 540                                                m_maximumRecordSize));
 541             m_backend_map[k] = b;
 542             frontend->m_backend_class = b;
 543         }
 544         else
 545         {
 546             frontend->m_backend_class = it->second;
 547         }
 548     }
 549     BackendClassPtr bc = frontend->m_backend_class;
 550     mp::odr odr;
 551
 552     // we only need to get init response from "first" target in
 553     // backend class - the assumption being that init response is
 554     // same for all
 555     bool create_first_one = false;
 556     {
 557         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 558         if (!bc->m_no_failed && !bc->m_no_succeeded && !bc->m_no_init)
 559             create_first_one = true;
 560         else
 561         {
 562             // first for first one to finish
 563             while (!bc->m_no_failed && !bc->m_no_succeeded && bc->m_no_init)
 564                 bc->m_cond_set_ready.wait(bc->m_mutex_backend_class);
 565         }
 566     }
 567     if (create_first_one)
 568     {
 569         BackendInstancePtr backend = bc->create_backend(package);
 570         if (backend)
 571             bc->release_backend(backend);
 572     }
 573     yazpp_1::GDU init_response;
 574     {
 575         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 576
 577         init_response = bc->m_init_response;
 578     }
 579
 580     if (init_response.get())
 581     {
 582         Z_GDU *response_gdu = init_response.get();
 583         mp::util::transfer_referenceId(odr, gdu->u.z3950,
 584                                        response_gdu->u.z3950);
 585         Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
 586         Z_Options *server_options = init_res->options;
 587         Z_Options *client_options = &frontend->m_init_options;
 588         int i;
 589         for (i = 0; i < 30; i++)
 590             if (!ODR_MASK_GET(client_options, i))
 591                 ODR_MASK_CLEAR(server_options, i);
 592
 593         if (!m_preferredMessageSize ||
 594             *init_res->preferredMessageSize > *req->preferredMessageSize)
 595             *init_res->preferredMessageSize = *req->preferredMessageSize;
 596
 597         if (!m_maximumRecordSize ||
 598             *init_res->maximumRecordSize > *req->maximumRecordSize)
 599             *init_res->maximumRecordSize = *req->maximumRecordSize;
 600
 601         package.response() = init_response;
 602         if (!*response_gdu->u.z3950->u.initResponse->result)
 603             package.session().close();
 604     }
 605     else
 606     {
 607         Z_APDU *apdu =
 608             odr.create_initResponse(
 609                 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
 610                 "session_shared: target closed connection during init");
 611         *apdu->u.initResponse->result = 0;
 612         package.response() = apdu;
 613         package.session().close();
 614     }
 615 }
 616
 617 void yf::SessionShared::BackendSet::timestamp()
 618 {
 619     time(&m_time_last_use);
 620 }
 621
 622 yf::SessionShared::BackendSet::BackendSet(
 623     const std::string &result_set_id,
 624     const Databases &databases,
 625     const yazpp_1::Yaz_Z_Query &query,
 626     Z_OtherInformation *additionalSearchInfo) :
 627     m_result_set_id(result_set_id),
 628     m_databases(databases), m_result_set_size(0), m_query(query)
 629 {
 630     timestamp();
 631     mem_additionalSearchInfo = nmem_create();
 632     additionalSearchInfoResponse = 0;
 633     additionalSearchInfoRequest =
 634         yaz_clone_z_OtherInformation(additionalSearchInfo,
 635                                      mem_additionalSearchInfo);
 636 }
 637
 638 yf::SessionShared::BackendSet::~BackendSet()
 639 {
 640     nmem_destroy(mem_additionalSearchInfo);
 641 }
 642
 643 static int get_diagnostic(Z_DefaultDiagFormat *r)
 644 {
 645     return *r->condition;
 646 }
 647
 648 bool yf::SessionShared::BackendSet::search(
 649     mp::Package &frontend_package,
 650     mp::Package &search_package,
 651     const Z_APDU *frontend_apdu,
 652     const BackendInstancePtr bp,
 653     Z_Records **z_records)
 654 {
 655     mp::odr odr;
 656     Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
 657     Z_SearchRequest *req = apdu_req->u.searchRequest;
 658
 659     req->additionalSearchInfo = additionalSearchInfoRequest;
 660     req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
 661     req->query = m_query.get_Z_Query();
 662
 663     req->num_databaseNames = m_databases.size();
 664     req->databaseNames = (char**)
 665         odr_malloc(odr, req->num_databaseNames * sizeof(char *));
 666     Databases::const_iterator it = m_databases.begin();
 667     size_t i = 0;
 668     for (; it != m_databases.end(); it++)
 669         req->databaseNames[i++] = odr_strdup(odr, it->c_str());
 670
 671     if (frontend_apdu->which == Z_APDU_searchRequest)
 672         req->preferredRecordSyntax =
 673             frontend_apdu->u.searchRequest->preferredRecordSyntax;
 674
 675     search_package.request() = apdu_req;
 676
 677     search_package.move();
 678
 679     Z_GDU *gdu = search_package.response().get();
 680     if (!search_package.session().is_closed()
 681         && gdu && gdu->which == Z_GDU_Z3950
 682         && gdu->u.z3950->which == Z_APDU_searchResponse)
 683     {
 684         Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
 685         *z_records = b_resp->records;
 686         m_result_set_size = *b_resp->resultCount;
 687
 688         additionalSearchInfoResponse = yaz_clone_z_OtherInformation(
 689             b_resp->additionalSearchInfo, mem_additionalSearchInfo);
 690         return true;
 691     }
 692     Z_APDU *f_apdu = 0;
 693     const char *addinfo = "session_shared: "
 694         "target closed connection during search";
 695     if (frontend_apdu->which == Z_APDU_searchRequest)
 696         f_apdu = odr.create_searchResponse(
 697             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 698     else if (frontend_apdu->which == Z_APDU_presentRequest)
 699         f_apdu = odr.create_presentResponse(
 700             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 701     else
 702         f_apdu = odr.create_close(
 703             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 704     frontend_package.response() = f_apdu;
 705     return false;
 706 }
 707
 708 void yf::SessionShared::Frontend::override_set(
 709     BackendInstancePtr &found_backend,
 710     std::string &result_set_id,
 711     const Databases &databases,
 712     bool out_of_sessions)
 713 {
 714     BackendClassPtr bc = m_backend_class;
 715     BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 716     time_t now;
 717     time(&now);
 718
 719     size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
 720     for (; it != bc->m_backend_list.end(); it++)
 721     {
 722         if (!(*it)->m_in_use)
 723         {
 724             BackendSetList::iterator set_it = (*it)->m_sets.begin();
 725             for (; set_it != (*it)->m_sets.end(); set_it++)
 726             {
 727                 if ((max_sets > 1 || (*set_it)->m_databases == databases)
 728                     &&
 729                     (out_of_sessions ||
 730                      now < (*set_it)->m_time_last_use ||
 731                      now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
 732                 {
 733                     found_backend = *it;
 734                     result_set_id = (*set_it)->m_result_set_id;
 735                     found_backend->m_sets.erase(set_it);
 736                     return;
 737                 }
 738             }
 739         }
 740     }
 741     for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
 742     {
 743         if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
 744         {
 745             found_backend = *it;
 746             if (bc->m_named_result_sets)
 747             {
 748                 result_set_id = boost::io::str(
 749                     boost::format("%1%") %
 750                     found_backend->m_result_set_sequence);
 751                 found_backend->m_result_set_sequence++;
 752             }
 753             else
 754                 result_set_id = "default";
 755             return;
 756         }
 757     }
 758 }
 759
 760 void yf::SessionShared::Frontend::get_set(mp::Package &package,
 761                                           const Z_APDU *apdu_req,
 762                                           const Databases &databases,
 763                                           yazpp_1::Yaz_Z_Query &query,
 764                                           BackendInstancePtr &found_backend,
 765                                           BackendSetPtr &found_set)
 766 {
 767     bool session_restarted = false;
 768     Z_OtherInformation *additionalSearchInfo = 0;
 769
 770     if (apdu_req->which == Z_APDU_searchRequest)
 771         additionalSearchInfo = apdu_req->u.searchRequest->additionalSearchInfo;
 772
 773 restart:
 774     std::string result_set_id;
 775     bool out_of_sessions = false;
 776     BackendClassPtr bc = m_backend_class;
 777     {
 778         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 779
 780         if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
 781             out_of_sessions = true;
 782
 783         if (m_p->m_optimize_search)
 784         {
 785             // look at each backend and see if we have a similar search
 786             BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 787             while (it != bc->m_backend_list.end())
 788             {
 789                 bool restart = false;
 790                 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
 791                 for (; set_it != (*it)->m_sets.end(); set_it++)
 792                 {
 793                     // for real present request we don't care
 794                     // if additionalSearchInfo matches: same records
 795                     if ((*set_it)->m_databases == databases
 796                         && query.match(&(*set_it)->m_query)
 797                         && (apdu_req->which != Z_APDU_searchRequest ||
 798                             yaz_compare_z_OtherInformation(
 799                                 additionalSearchInfo,
 800                                 (*set_it)->additionalSearchInfoRequest)))
 801                     {
 802                         if ((*it)->m_in_use)
 803                         {
 804                             bc->m_cond_set_ready.wait(
 805                                 bc->m_mutex_backend_class);
 806                             restart = true;
 807                             break;
 808                         }
 809                         else
 810                         {
 811                             found_set = *set_it;
 812                             found_backend = *it;
 813                             bc->use_backend(found_backend);
 814                             // found matching set. No need to search again
 815                             return;
 816                         }
 817                     }
 818                 }
 819                 if (restart)
 820                     it = bc->m_backend_list.begin();
 821                 else
 822                     it++;
 823             }
 824         }
 825         override_set(found_backend, result_set_id, databases, out_of_sessions);
 826         if (found_backend)
 827             bc->use_backend(found_backend);
 828     }
 829     if (!found_backend)
 830     {
 831         // create a new backend set (and new set) if we're not out of sessions
 832         if (!out_of_sessions)
 833             found_backend = bc->create_backend(package);
 834
 835         if (!found_backend)
 836         {
 837             Z_APDU *f_apdu = 0;
 838             mp::odr odr;
 839             const char *addinfo = 0;
 840
 841             if (out_of_sessions)
 842                 addinfo = "session_shared: all sessions in use";
 843             else
 844                 addinfo = "session_shared: could not create backend";
 845             if (apdu_req->which == Z_APDU_searchRequest)
 846             {
 847                 f_apdu = odr.create_searchResponse(
 848                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 849             }
 850             else if (apdu_req->which == Z_APDU_presentRequest)
 851             {
 852                 f_apdu = odr.create_presentResponse(
 853                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 854             }
 855             else
 856             {
 857                 f_apdu = odr.create_close(
 858                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 859             }
 860             package.response() = f_apdu;
 861             return;
 862         }
 863         if (bc->m_named_result_sets)
 864         {
 865             result_set_id = boost::io::str(
 866                 boost::format("%1%") % found_backend->m_result_set_sequence);
 867             found_backend->m_result_set_sequence++;
 868         }
 869         else
 870             result_set_id = "default";
 871     }
 872     found_backend->timestamp();
 873
 874     // we must search ...
 875     BackendSetPtr new_set(new BackendSet(result_set_id,
 876                                          databases, query,
 877                                          additionalSearchInfo));
 878
 879     found_set = new_set;
 880     found_set->timestamp();
 881     found_backend->m_sets.push_back(found_set);
 882
 883     Z_Records *z_records = 0;
 884
 885     Package search_package(found_backend->m_session, package.origin());
 886     search_package.copy_filter(package);
 887
 888     if (!new_set->search(package, search_package,
 889                          apdu_req, found_backend, &z_records))
 890     {
 891         bc->remove_backend(found_backend);
 892         return; // search error
 893     }
 894
 895     if (z_records)
 896     {
 897         int condition = 0;
 898         if (z_records->which == Z_Records_NSD)
 899         {
 900             condition =
 901                 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
 902         }
 903         else if (z_records->which == Z_Records_multipleNSD)
 904         {
 905             if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
 906                 &&
 907
 908                 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
 909                 Z_DiagRec_defaultFormat)
 910             {
 911                 condition = get_diagnostic(
 912                     z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
 913
 914             }
 915         }
 916         if (m_p->m_restart && !session_restarted &&
 917             condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
 918         {
 919             package.log("session_shared", YLOG_LOG, "restart");
 920             bc->remove_backend(found_backend);
 921             session_restarted = true;
 922             found_backend.reset();
 923             goto restart;
 924
 925         }
 926
 927         if (condition)
 928         {
 929             mp::odr odr;
 930             if (apdu_req->which == Z_APDU_searchRequest)
 931             {
 932                 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
 933                                                            0, 0);
 934                 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 935                 *f_resp->searchStatus = Z_SearchResponse_none;
 936                 f_resp->records = z_records;
 937                 package.response() = f_apdu;
 938             }
 939             if (apdu_req->which == Z_APDU_presentRequest)
 940             {
 941                 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
 942                                                             0, 0);
 943                 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
 944                 f_resp->records = z_records;
 945                 package.response() = f_apdu;
 946             }
 947             bc->release_backend(found_backend);
 948             return; // search error
 949         }
 950     }
 951     if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
 952     {
 953         package.log("session_shared", YLOG_LOG, "restart");
 954         bc->remove_backend(found_backend);
 955         session_restarted = true;
 956         found_backend.reset();
 957         goto restart;
 958     }
 959
 960 #if 0
 961     found_set = new_set;
 962     found_set->timestamp();
 963     found_backend->m_sets.push_back(found_set);
 964 #endif
 965 }
 966
 967 int yf::SessionShared::Frontend::result_set_ref(ODR o,
 968                                                 const Databases &databases,
 969                                                 Z_RPNStructure *s,
 970                                                 std::string &rset)
 971 {
 972     int ret = 0;
 973     switch (s->which)
 974     {
 975     case Z_RPNStructure_simple:
 976         if (s->u.simple->which == Z_Operand_resultSetId)
 977         {
 978             const char *id = s->u.simple->u.resultSetId;
 979             rset = id;
 980
 981             FrontendSets::iterator fset_it = m_frontend_sets.find(id);
 982             if (fset_it == m_frontend_sets.end())
 983             {
 984                 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
 985             }
 986             else if (fset_it->second->get_databases() != databases)
 987             {
 988                 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
 989             }
 990             else
 991             {
 992                 yazpp_1::Yaz_Z_Query query = fset_it->second->get_query();
 993                 Z_Query *q = yaz_copy_Z_Query(query.get_Z_Query(), o);
 994                 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
 995                 {
 996                     s->which = q->u.type_1->RPNStructure->which;
 997                     s->u.simple = q->u.type_1->RPNStructure->u.simple;
 998                 }
 999                 else
1000                 {
1001                     ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
1002                 }
1003             }
1004         }
1005         break;
1006     case Z_RPNStructure_complex:
1007         ret = result_set_ref(o, databases, s->u.complex->s1, rset);
1008         if (!ret)
1009             ret = result_set_ref(o, databases, s->u.complex->s2, rset);
1010         break;
1011     }
1012     return ret;
1013 }
1014
1015 void yf::SessionShared::Frontend::search(mp::Package &package,
1016                                          Z_APDU *apdu_req)
1017 {
1018     Z_SearchRequest *req = apdu_req->u.searchRequest;
1019     FrontendSets::iterator fset_it =
1020         m_frontend_sets.find(req->resultSetName);
1021     if (fset_it != m_frontend_sets.end())
1022     {
1023         // result set already exist
1024         // if replace indicator is off: we return diagnostic if
1025         // result set already exist.
1026         if (*req->replaceIndicator == 0)
1027         {
1028             mp::odr odr;
1029             Z_APDU *apdu =
1030                 odr.create_searchResponse(
1031                     apdu_req,
1032                     YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
1033                     0);
1034             package.response() = apdu;
1035             return;
1036         }
1037         m_frontend_sets.erase(fset_it);
1038     }
1039
1040     Databases databases;
1041     int i;
1042     for (i = 0; i < req->num_databaseNames; i++)
1043         databases.push_back(req->databaseNames[i]);
1044
1045
1046     yazpp_1::Yaz_Z_Query query;
1047     query.set_Z_Query(req->query);
1048
1049     Z_Query *q = query.get_Z_Query();
1050     if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
1051     {
1052         mp::odr odr;
1053         std::string rset;
1054         int diag = result_set_ref(odr, databases, q->u.type_1->RPNStructure,
1055                                   rset);
1056         if (diag)
1057         {
1058             Z_APDU *apdu =
1059                 odr.create_searchResponse(
1060                     apdu_req,
1061                     diag,
1062                     rset.c_str());
1063             package.response() = apdu;
1064             return;
1065         }
1066         query.set_Z_Query(q);
1067     }
1068
1069     BackendSetPtr found_set; // null
1070     BackendInstancePtr found_backend; // null
1071
1072     get_set(package, apdu_req, databases, query, found_backend, found_set);
1073     if (!found_set)
1074         return;
1075
1076     mp::odr odr;
1077     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
1078     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
1079     *f_resp->resultCount = found_set->m_result_set_size;
1080     f_resp->additionalSearchInfo = found_set->additionalSearchInfoResponse;
1081     package.response() = f_apdu;
1082
1083     FrontendSetPtr fset(new FrontendSet(databases, query));
1084     m_frontend_sets[req->resultSetName] = fset;
1085
1086     m_backend_class->release_backend(found_backend);
1087 }
1088
1089 void yf::SessionShared::Frontend::present(mp::Package &package,
1090                                           Z_APDU *apdu_req)
1091 {
1092     mp::odr odr;
1093     Z_PresentRequest *req = apdu_req->u.presentRequest;
1094
1095     FrontendSets::iterator fset_it =
1096         m_frontend_sets.find(req->resultSetId);
1097
1098     if (fset_it == m_frontend_sets.end())
1099     {
1100         Z_APDU *apdu =
1101             odr.create_presentResponse(
1102                 apdu_req,
1103                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
1104                 req->resultSetId);
1105         package.response() = apdu;
1106         return;
1107     }
1108     FrontendSetPtr fset = fset_it->second;
1109
1110     Databases databases = fset->get_databases();
1111     yazpp_1::Yaz_Z_Query query = fset->get_query();
1112
1113     BackendClassPtr bc = m_backend_class;
1114     BackendSetPtr found_set; // null
1115     BackendInstancePtr found_backend;
1116
1117     get_set(package, apdu_req, databases, query, found_backend, found_set);
1118     if (!found_set)
1119         return;
1120
1121     Z_NamePlusRecordList *npr_res = 0;
1122     // record_cache.lookup types are int's. Avoid non-fitting values
1123     if (*req->resultSetStartPoint > 0
1124         && *req->resultSetStartPoint < INT_MAX
1125         && *req->numberOfRecordsRequested > 0
1126         && *req->numberOfRecordsRequested < INT_MAX
1127         && found_set->m_record_cache.lookup(odr, &npr_res,
1128                                             *req->resultSetStartPoint,
1129                                             *req->numberOfRecordsRequested,
1130                                             req->preferredRecordSyntax,
1131                                             req->recordComposition))
1132     {
1133         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1134         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1135
1136         yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1137                 " records in cache %p",
1138                 *req->resultSetStartPoint,
1139                 *req->numberOfRecordsRequested,
1140                 &found_set->m_record_cache);
1141
1142         *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
1143         *f_resp->nextResultSetPosition =
1144             *req->resultSetStartPoint + *req->numberOfRecordsRequested;
1145         // f_resp->presentStatus assumed OK.
1146         f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
1147         f_resp->records->which = Z_Records_DBOSD;
1148         f_resp->records->u.databaseOrSurDiagnostics = npr_res;
1149         package.response() = f_apdu_res;
1150         bc->release_backend(found_backend);
1151         return;
1152     }
1153
1154     found_backend->timestamp();
1155
1156     Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
1157     Z_PresentRequest *p_req = p_apdu->u.presentRequest;
1158     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1159     p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
1160     *p_req->resultSetStartPoint = *req->resultSetStartPoint;
1161     *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
1162     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1163     p_req->recordComposition = req->recordComposition;
1164
1165     Package present_package(found_backend->m_session, package.origin());
1166     present_package.copy_filter(package);
1167
1168     present_package.request() = p_apdu;
1169
1170     present_package.move();
1171
1172     Z_GDU *gdu = present_package.response().get();
1173     if (!present_package.session().is_closed()
1174         && gdu && gdu->which == Z_GDU_Z3950
1175         && gdu->u.z3950->which == Z_APDU_presentResponse)
1176     {
1177         Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
1178         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1179         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1180
1181         f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
1182         f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
1183         f_resp->presentStatus= b_resp->presentStatus;
1184         f_resp->records = b_resp->records;
1185         f_resp->otherInfo = b_resp->otherInfo;
1186         package.response() = f_apdu_res;
1187
1188         if (b_resp->records && b_resp->records->which ==  Z_Records_DBOSD)
1189         {
1190             Z_NamePlusRecordList *npr =
1191                 b_resp->records->u.databaseOrSurDiagnostics;
1192             // record_cache.add types are int's. Avoid non-fitting values
1193             if (*req->resultSetStartPoint > 0
1194                 && npr->num_records + *req->resultSetStartPoint < INT_MAX)
1195             {
1196 #if 0
1197                 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1198                         " records to cache %p",
1199                         *req->resultSetStartPoint,
1200                         *f_resp->numberOfRecordsReturned,
1201                         &found_set->m_record_cache);
1202 #endif
1203                 found_set->m_record_cache.add(
1204                     odr, npr, *req->resultSetStartPoint,
1205                     p_req->recordComposition);
1206             }
1207         }
1208         bc->release_backend(found_backend);
1209     }
1210     else
1211     {
1212         bc->remove_backend(found_backend);
1213         Z_APDU *f_apdu_res =
1214             odr.create_presentResponse(
1215                 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1216                 "session_shared: target closed connection during present");
1217         package.response() = f_apdu_res;
1218     }
1219 }
1220
1221 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1222                                        Z_APDU *apdu_req)
1223 {
1224     BackendClassPtr bc = m_backend_class;
1225     BackendInstancePtr backend = bc->get_backend(frontend_package);
1226     if (!backend)
1227     {
1228         mp::odr odr;
1229         Z_APDU *apdu = odr.create_scanResponse(
1230             apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1231             "session_shared: could not create backend");
1232         frontend_package.response() = apdu;
1233     }
1234     else
1235     {
1236         Package scan_package(backend->m_session, frontend_package.origin());
1237         backend->timestamp();
1238         scan_package.copy_filter(frontend_package);
1239         scan_package.request() = apdu_req;
1240         scan_package.move();
1241         frontend_package.response() = scan_package.response();
1242         if (scan_package.session().is_closed())
1243         {
1244             frontend_package.session().close();
1245             bc->remove_backend(backend);
1246         }
1247         else
1248             bc->release_backend(backend);
1249     }
1250 }
1251
1252 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1253 {
1254 }
1255
1256 void yf::SessionShared::Worker::operator() (void)
1257 {
1258     m_p->expire();
1259 }
1260
1261 bool yf::SessionShared::BackendClass::expire_instances()
1262 {
1263     time_t now;
1264     time(&now);
1265     boost::mutex::scoped_lock lock(m_mutex_backend_class);
1266     BackendInstanceList::iterator bit = m_backend_list.begin();
1267     while (bit != m_backend_list.end())
1268     {
1269         time_t last_use = (*bit)->m_time_last_use;
1270
1271         if ((*bit)->m_in_use)
1272         {
1273             bit++;
1274         }
1275         else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1276         {
1277             bit = m_backend_list.erase(bit);
1278         }
1279         else
1280         {
1281             bit++;
1282         }
1283     }
1284     if (m_backend_list.empty())
1285         return true;
1286     return false;
1287 }
1288
1289 void yf::SessionShared::Rep::expire_classes()
1290 {
1291     boost::mutex::scoped_lock lock(m_mutex_backend_map);
1292     BackendClassMap::iterator b_it = m_backend_map.begin();
1293     while (b_it != m_backend_map.end())
1294     {
1295         if (b_it->second->expire_instances())
1296         {
1297             m_backend_map.erase(b_it);
1298             b_it = m_backend_map.begin();
1299         }
1300         else
1301             b_it++;
1302     }
1303 }
1304
1305 void yf::SessionShared::Rep::expire()
1306 {
1307     while (true)
1308     {
1309         boost::xtime xt;
1310         boost::xtime_get(&xt,
1311 #if BOOST_VERSION >= 105000
1312                 boost::TIME_UTC_
1313 #else
1314                 boost::TIME_UTC
1315 #endif
1316                   );
1317         xt.sec += m_session_ttl;
1318         {
1319             boost::mutex::scoped_lock lock(m_mutex);
1320             m_cond_expire_ready.timed_wait(lock, xt);
1321             if (close_down)
1322                 break;
1323         }
1324         stat();
1325         expire_classes();
1326     }
1327 }
1328
1329 yf::SessionShared::Rep::Rep()
1330 {
1331     m_resultset_ttl = 30;
1332     m_resultset_max = 10;
1333     m_session_ttl = 90;
1334     m_optimize_search = true;
1335     m_restart = false;
1336     m_session_max = 100;
1337     m_preferredMessageSize = 0;
1338     m_maximumRecordSize = 0;
1339     close_down = false;
1340 }
1341
1342 yf::SessionShared::Rep::~Rep()
1343 {
1344     {
1345         boost::mutex::scoped_lock lock(m_mutex);
1346         close_down = true;
1347         m_cond_expire_ready.notify_all();
1348     }
1349     m_thrds.join_all();
1350 }
1351
1352 void yf::SessionShared::Rep::start()
1353 {
1354     yf::SessionShared::Worker w(this);
1355     m_thrds.add_thread(new boost::thread(w));
1356 }
1357
1358 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1359 {
1360 }
1361
1362 yf::SessionShared::~SessionShared() {
1363 }
1364
1365 void yf::SessionShared::start() const
1366 {
1367     m_p->start();
1368 }
1369
1370 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1371 {
1372 }
1373
1374 yf::SessionShared::Frontend::~Frontend()
1375 {
1376 }
1377
1378 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1379 {
1380     boost::mutex::scoped_lock lock(m_mutex);
1381
1382     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1383
1384     while(true)
1385     {
1386         it = m_clients.find(package.session());
1387         if (it == m_clients.end())
1388             break;
1389
1390         if (!it->second->m_in_use)
1391         {
1392             it->second->m_in_use = true;
1393             return it->second;
1394         }
1395         m_cond_session_ready.wait(lock);
1396     }
1397     FrontendPtr f(new Frontend(this));
1398     m_clients[package.session()] = f;
1399     f->m_in_use = true;
1400     return f;
1401 }
1402
1403 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1404 {
1405     boost::mutex::scoped_lock lock(m_mutex);
1406     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1407
1408     it = m_clients.find(package.session());
1409     if (it != m_clients.end())
1410     {
1411         if (package.session().is_closed())
1412         {
1413             m_clients.erase(it);
1414         }
1415         else
1416         {
1417             it->second->m_in_use = false;
1418         }
1419         m_cond_session_ready.notify_all();
1420     }
1421 }
1422
1423
1424 void yf::SessionShared::process(mp::Package &package) const
1425 {
1426     FrontendPtr f = m_p->get_frontend(package);
1427
1428     Z_GDU *gdu = package.request().get();
1429
1430     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1431         Z_APDU_initRequest && !f->m_is_virtual)
1432     {
1433         m_p->init(package, gdu, f);
1434     }
1435     else if (!f->m_is_virtual)
1436         package.move();
1437     else if (gdu && gdu->which == Z_GDU_Z3950)
1438     {
1439         Z_APDU *apdu = gdu->u.z3950;
1440         if (apdu->which == Z_APDU_initRequest)
1441         {
1442             mp::odr odr;
1443
1444             package.response() = odr.create_close(
1445                 apdu,
1446                 Z_Close_protocolError,
1447                 "double init");
1448
1449             package.session().close();
1450         }
1451         else if (apdu->which == Z_APDU_close)
1452         {
1453             mp::odr odr;
1454
1455             package.response() = odr.create_close(
1456                 apdu,
1457                 Z_Close_peerAbort, "received close from client");
1458             package.session().close();
1459         }
1460         else if (apdu->which == Z_APDU_searchRequest)
1461         {
1462             f->search(package, apdu);
1463         }
1464         else if (apdu->which == Z_APDU_presentRequest)
1465         {
1466             f->present(package, apdu);
1467         }
1468         else if (apdu->which == Z_APDU_scanRequest)
1469         {
1470             f->scan(package, apdu);
1471         }
1472         else
1473         {
1474             mp::odr odr;
1475
1476             package.response() = odr.create_close(
1477                 apdu, Z_Close_protocolError,
1478                 "unsupported APDU in filter_session_shared");
1479
1480             package.session().close();
1481         }
1482     }
1483     m_p->release_frontend(package);
1484 }
1485
1486 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1487                                   const char *path)
1488 {
1489     for (ptr = ptr->children; ptr; ptr = ptr->next)
1490     {
1491         if (ptr->type != XML_ELEMENT_NODE)
1492             continue;
1493         if (!strcmp((const char *) ptr->name, "resultset"))
1494         {
1495             const struct _xmlAttr *attr;
1496             for (attr = ptr->properties; attr; attr = attr->next)
1497             {
1498                 if (!strcmp((const char *) attr->name, "ttl"))
1499                     m_p->m_resultset_ttl =
1500                         mp::xml::get_int(attr->children, 30);
1501                 else if (!strcmp((const char *) attr->name, "max"))
1502                 {
1503                     m_p->m_resultset_max =
1504                         mp::xml::get_int(attr->children, 10);
1505                 }
1506                 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1507                 {
1508                     m_p->m_optimize_search =
1509                         mp::xml::get_bool(attr->children, true);
1510                 }
1511                 else if (!strcmp((const char *) attr->name, "restart"))
1512                 {
1513                     m_p->m_restart = mp::xml::get_bool(attr->children, true);
1514                 }
1515                 else
1516                     throw mp::filter::FilterException(
1517                         "Bad attribute " + std::string((const char *)
1518                                                        attr->name));
1519             }
1520         }
1521         else if (!strcmp((const char *) ptr->name, "session"))
1522         {
1523             const struct _xmlAttr *attr;
1524             for (attr = ptr->properties; attr; attr = attr->next)
1525             {
1526                 if (!strcmp((const char *) attr->name, "ttl"))
1527                     m_p->m_session_ttl =
1528                         mp::xml::get_int(attr->children, 90);
1529                 else if (!strcmp((const char *) attr->name, "max"))
1530                     m_p->m_session_max =
1531                         mp::xml::get_int(attr->children, 100);
1532                 else
1533                     throw mp::filter::FilterException(
1534                         "Bad attribute " + std::string((const char *)
1535                                                        attr->name));
1536             }
1537         }
1538         else if (!strcmp((const char *) ptr->name, "init"))
1539         {
1540             const struct _xmlAttr *attr;
1541             for (attr = ptr->properties; attr; attr = attr->next)
1542             {
1543                 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1544                     m_p->m_maximumRecordSize =
1545                         mp::xml::get_int(attr->children, 0);
1546                 else if (!strcmp((const char *) attr->name,
1547                                  "preferred-message-size"))
1548                     m_p->m_preferredMessageSize =
1549                         mp::xml::get_int(attr->children, 0);
1550                 else
1551                     throw mp::filter::FilterException(
1552                         "Bad attribute " + std::string((const char *)
1553                                                        attr->name));
1554             }
1555         }
1556         else
1557         {
1558             throw mp::filter::FilterException("Bad element "
1559                                                + std::string((const char *)
1560                                                              ptr->name));
1561         }
1562     }
1563 }
1564
1565 static mp::filter::Base* filter_creator()
1566 {
1567     return new mp::filter::SessionShared;
1568 }
1569
1570 extern "C" {
1571     struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1572         0,
1573         "session_shared",
1574         filter_creator
1575     };
1576 }
1577
1578 /*
1579  * Local variables:
1580  * c-basic-offset: 4
1581  * c-file-style: "Stroustrup"
1582  * indent-tabs-mode: nil
1583  * End:
1584  * vim: shiftwidth=4 tabstop=8 expandtab
1585  */
1586