src/filter_load_balance.cpp

   1 /* This file is part of Metaproxy.
   2    Copyright (C) 2005-2009 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 #include "session.hpp"
  21 #include "package.hpp"
  22 #include "filter.hpp"
  23 #include "filter_load_balance.hpp"
  24 #include "util.hpp"
  25
  26
  27 #include <boost/thread/mutex.hpp>
  28
  29 #include <yaz/zgdu.h>
  30
  31 // remove max macro if already defined (defined later in <limits>)
  32 #ifdef max
  33 #undef max
  34 #endif
  35
  36 //#include <iostream>
  37 #include <list>
  38 #include <map>
  39 #include <limits>
  40
  41 namespace mp = metaproxy_1;
  42 namespace yf = mp::filter;
  43
  44 namespace metaproxy_1
  45 {
  46     namespace filter
  47     {
  48         class LoadBalance::Impl
  49         {
  50         public:
  51             Impl();
  52             ~Impl();
  53             void process(metaproxy_1::Package & package);
  54             void configure(const xmlNode * ptr);
  55         private:
  56             // statistic manipulating functions,
  57             void add_dead(unsigned long session_id);
  58             //void clear_dead(unsigned long session_id);
  59             void add_package(unsigned long session_id);
  60             void remove_package(unsigned long session_id);
  61             void add_session(unsigned long session_id, std::string target);
  62             void remove_session(unsigned long session_id);
  63             std::string find_session_target(unsigned long session_id);
  64
  65             // cost functions
  66             unsigned int cost(std::string target);
  67             unsigned int dead(std::string target);
  68
  69             // local classes
  70             class TargetStat {
  71             public:
  72                 unsigned int sessions;
  73                 unsigned int packages;
  74                 unsigned int deads;
  75                 unsigned int cost() {
  76                     unsigned int c = sessions + packages + deads;
  77                     //std::cout << "stats  c:" << c
  78                     //          << " s:" << sessions
  79                     //          << " p:" << packages
  80                     //          << " d:" << deads
  81                     //          <<"\n";
  82                     return c;
  83                 }
  84             };
  85
  86             // local protected databases
  87             boost::mutex m_mutex;
  88             std::map<std::string, TargetStat> m_target_stat;
  89             std::map<unsigned long, std::string> m_session_target;
  90         };
  91     }
  92 }
  93
  94 // define Pimpl wrapper forwarding to Impl
  95
  96 yf::LoadBalance::LoadBalance() : m_p(new Impl)
  97 {
  98 }
  99
 100 yf::LoadBalance::~LoadBalance()
 101 {  // must have a destructor because of boost::scoped_ptr
 102 }
 103
 104 void yf::LoadBalance::configure(const xmlNode *xmlnode, bool test_only)
 105 {
 106     m_p->configure(xmlnode);
 107 }
 108
 109 void yf::LoadBalance::process(mp::Package &package) const
 110 {
 111     m_p->process(package);
 112 }
 113
 114
 115 yf::LoadBalance::Impl::Impl()
 116 {
 117 }
 118
 119 yf::LoadBalance::Impl::~Impl()
 120 {
 121 }
 122
 123 void yf::LoadBalance::Impl::configure(const xmlNode *xmlnode)
 124 {
 125 }
 126
 127 void yf::LoadBalance::Impl::process(mp::Package &package)
 128 {
 129
 130     bool is_closed_front = false;
 131     bool is_closed_back = false;
 132
 133     // checking for closed front end packages
 134     if (package.session().is_closed())
 135     {
 136         is_closed_front = true;
 137     }
 138
 139     Z_GDU *gdu_req = package.request().get();
 140
 141     // passing anything but z3950 packages
 142     if (gdu_req && gdu_req->which == Z_GDU_Z3950)
 143     {
 144         // target selecting only on Z39.50 init request
 145         if (gdu_req->u.z3950->which == Z_APDU_initRequest)
 146         {
 147             mp::odr odr_en(ODR_ENCODE);
 148             Z_InitRequest *org_init = gdu_req->u.z3950->u.initRequest;
 149
 150             // extracting virtual hosts
 151             std::list<std::string> vhosts;
 152
 153             mp::util::remove_vhost_otherinfo(&(org_init->otherInfo), vhosts);
 154
 155             // choosing one target according to load-balancing algorithm
 156
 157             if (vhosts.size())
 158             {
 159                 std::string target;
 160                 unsigned int cost = std::numeric_limits<unsigned int>::max();
 161                 { //locking scope for local databases
 162                     boost::mutex::scoped_lock scoped_lock(m_mutex);
 163
 164                     // load-balancing algorithm goes here
 165                     //target = *vhosts.begin();
 166                     for (std::list<std::string>::const_iterator ivh
 167                              = vhosts.begin();
 168                          ivh != vhosts.end();
 169                          ivh++)
 170                     {
 171                         if ((*ivh).size() != 0)
 172                         {
 173                             unsigned int vhcost
 174                                 = yf::LoadBalance::Impl::cost(*ivh);
 175                             if (cost > vhcost)
 176                             {
 177                                 cost = vhcost;
 178                                 target = *ivh;
 179                             }
 180                         }
 181                      }
 182
 183                     // updating local database
 184                     add_session(package.session().id(), target);
 185                     yf::LoadBalance::Impl::cost(target);
 186                     add_package(package.session().id());
 187                 }
 188
 189                 // copying new target into init package
 190                 mp::util::set_vhost_otherinfo(&(org_init->otherInfo),
 191                                               odr_en, target, 1);
 192                 package.request() = gdu_req;
 193             }
 194         }
 195         // frontend Z39.50 close request is added to statistics and marked
 196         else if (gdu_req->u.z3950->which == Z_APDU_close)
 197         {
 198             is_closed_front = true;
 199             boost::mutex::scoped_lock scoped_lock(m_mutex);
 200             add_package(package.session().id());
 201         }
 202         // any other Z39.50 package is added to statistics
 203         else
 204         {
 205             boost::mutex::scoped_lock scoped_lock(m_mutex);
 206             add_package(package.session().id());
 207         }
 208     }
 209
 210     // moving all package types
 211     package.move();
 212
 213     // checking for closed back end packages
 214     if (package.session().is_closed())
 215         is_closed_back = true;
 216
 217     Z_GDU *gdu_res = package.response().get();
 218
 219     // passing anything but z3950 packages
 220     if (gdu_res && gdu_res->which == Z_GDU_Z3950)
 221     {
 222         // session closing only on Z39.50 close response
 223         if (gdu_res->u.z3950->which == Z_APDU_close)
 224         {
 225             is_closed_back = true;
 226             boost::mutex::scoped_lock scoped_lock(m_mutex);
 227             remove_package(package.session().id());
 228         }
 229         // any other Z39.50 package is removed from statistics
 230         else
 231         {
 232             boost::mutex::scoped_lock scoped_lock(m_mutex);
 233             remove_package(package.session().id());
 234         }
 235     }
 236
 237     // finally removing sessions and marking deads
 238     if (is_closed_back || is_closed_front)
 239     {
 240         boost::mutex::scoped_lock scoped_lock(m_mutex);
 241
 242         // marking backend dead if backend closed without fronted close
 243         if (is_closed_front == false)
 244             add_dead(package.session().id());
 245
 246         remove_session(package.session().id());
 247
 248         // making sure that package is closed
 249         package.session().close();
 250     }
 251 }
 252
 253 // statistic manipulating functions,
 254 void yf::LoadBalance::Impl::add_dead(unsigned long session_id)
 255 {
 256     std::string target = find_session_target(session_id);
 257
 258     if (target.size() != 0)
 259     {
 260         std::map<std::string, TargetStat>::iterator itarg;
 261         itarg = m_target_stat.find(target);
 262         if (itarg != m_target_stat.end()
 263             && itarg->second.deads < std::numeric_limits<unsigned int>::max())
 264         {
 265             itarg->second.deads += 1;
 266             // std:.cout << "add_dead " << session_id << " " << target
 267             //          << " d:" << itarg->second.deads << "\n";
 268         }
 269     }
 270 }
 271
 272 //void yf::LoadBalance::Impl::clear_dead(unsigned long session_id){
 273 //    std::cout << "clear_dead " << session_id << "\n";
 274 //};
 275
 276 void yf::LoadBalance::Impl::add_package(unsigned long session_id)
 277 {
 278     std::string target = find_session_target(session_id);
 279
 280     if (target.size() != 0)
 281     {
 282         std::map<std::string, TargetStat>::iterator itarg;
 283         itarg = m_target_stat.find(target);
 284         if (itarg != m_target_stat.end()
 285             && itarg->second.packages
 286                < std::numeric_limits<unsigned int>::max())
 287         {
 288             itarg->second.packages += 1;
 289             // std:.cout << "add_package " << session_id << " " << target
 290             //          << " p:" << itarg->second.packages << "\n";
 291         }
 292     }
 293 }
 294
 295 void yf::LoadBalance::Impl::remove_package(unsigned long session_id)
 296 {
 297     std::string target = find_session_target(session_id);
 298
 299     if (target.size() != 0)
 300     {
 301         std::map<std::string, TargetStat>::iterator itarg;
 302         itarg = m_target_stat.find(target);
 303         if (itarg != m_target_stat.end()
 304             && itarg->second.packages > 0)
 305         {
 306             itarg->second.packages -= 1;
 307             // std:.cout << "remove_package " << session_id << " " << target
 308             //          << " p:" << itarg->second.packages << "\n";
 309         }
 310     }
 311 }
 312
 313 void yf::LoadBalance::Impl::add_session(unsigned long session_id,
 314                                         std::string target)
 315 {
 316     // finding and adding session
 317     std::map<unsigned long, std::string>::iterator isess;
 318     isess = m_session_target.find(session_id);
 319     if (isess == m_session_target.end())
 320     {
 321         m_session_target.insert(std::make_pair(session_id, target));
 322     }
 323
 324     // finding and adding target statistics
 325     std::map<std::string, TargetStat>::iterator itarg;
 326     itarg = m_target_stat.find(target);
 327     if (itarg == m_target_stat.end())
 328     {
 329         TargetStat stat;
 330         stat.sessions = 1;
 331         stat.packages = 0;
 332         stat.deads = 0;
 333         m_target_stat.insert(std::make_pair(target, stat));
 334         // std:.cout << "add_session " << session_id << " " << target
 335         //          << " s:1\n";
 336     }
 337     else if (itarg->second.sessions < std::numeric_limits<unsigned int>::max())
 338     {
 339         itarg->second.sessions += 1;
 340         // std:.cout << "add_session " << session_id << " " << target
 341         //          << " s:" << itarg->second.sessions << "\n";
 342     }
 343 }
 344
 345 void yf::LoadBalance::Impl::remove_session(unsigned long session_id)
 346 {
 347     std::string target;
 348
 349     // finding session
 350     std::map<unsigned long, std::string>::iterator isess;
 351     isess = m_session_target.find(session_id);
 352     if (isess == m_session_target.end())
 353         return;
 354     else
 355         target = isess->second;
 356
 357     // finding target statistics
 358     std::map<std::string, TargetStat>::iterator itarg;
 359     itarg = m_target_stat.find(target);
 360     if (itarg == m_target_stat.end())
 361     {
 362         m_session_target.erase(isess);
 363         return;
 364     }
 365
 366     // counting session down
 367     if (itarg->second.sessions > 0)
 368         itarg->second.sessions -= 1;
 369
 370     // std:.cout << "remove_session " << session_id << " " << target
 371     //          << " s:" << itarg->second.sessions << "\n";
 372
 373     // clearing empty sessions and targets
 374     if (itarg->second.sessions == 0 && itarg->second.deads == 0)
 375     {
 376         m_target_stat.erase(itarg);
 377         m_session_target.erase(isess);
 378     }
 379 }
 380
 381 std::string yf::LoadBalance::Impl::find_session_target(unsigned long session_id)
 382 {
 383     std::string target;
 384     std::map<unsigned long, std::string>::iterator isess;
 385     isess = m_session_target.find(session_id);
 386     if (isess != m_session_target.end())
 387         target = isess->second;
 388     return target;
 389 }
 390
 391
 392 // cost functions
 393 unsigned int yf::LoadBalance::Impl::cost(std::string target)
 394 {
 395     unsigned int cost = 0;
 396
 397     if (target.size() != 0)
 398     {
 399         std::map<std::string, TargetStat>::iterator itarg;
 400         itarg = m_target_stat.find(target);
 401         if (itarg != m_target_stat.end())
 402             cost = itarg->second.cost();
 403     }
 404
 405     //std::cout << "cost " << target << " c:" << cost << "\n";
 406     return cost;
 407 }
 408
 409 unsigned int yf::LoadBalance::Impl::dead(std::string target)
 410 {
 411     unsigned int dead = 0;
 412
 413     if (target.size() != 0)
 414     {
 415         std::map<std::string, TargetStat>::iterator itarg;
 416         itarg = m_target_stat.find(target);
 417         if (itarg != m_target_stat.end())
 418             dead = itarg->second.deads;
 419     }
 420
 421     //std::cout << "dead " << target << " d:" << dead << "\n";
 422     return dead;
 423 }
 424
 425
 426 static mp::filter::Base* filter_creator()
 427 {
 428     return new mp::filter::LoadBalance;
 429 }
 430
 431 extern "C" {
 432     struct metaproxy_1_filter_struct metaproxy_1_filter_load_balance = {
 433         0,
 434         "load_balance",
 435         filter_creator
 436     };
 437 }
 438
 439
 440 /*
 441  * Local variables:
 442  * c-basic-offset: 4
 443  * c-file-style: "Stroustrup"
 444  * indent-tabs-mode: nil
 445  * End:
 446  * vim: shiftwidth=4 tabstop=8 expandtab
 447  */
 448