src/main/connection.c

   1 /**
   2  * @file connection.c
   3  * @brief Handle pools of connections (threads, sockets, etc.)
   4  *
   5  * Version:     $Id$
   6  *
   7  *   This program is free software; you can redistribute it and/or modify
   8  *   it under the terms of the GNU General Public License as published by
   9  *   the Free Software Foundation; either version 2 of the License, or
  10  *   (at your option) any later version.
  11  *
  12  *   This program is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with this program; if not, write to the Free Software
  19  *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  20  *
  21  * Copyright 2011  The FreeRADIUS server project
  22  * Copyright 2011  Alan DeKok <aland@deployingradius.com>
  23  */
  24
  25 #include <freeradius-devel/ident.h>
  26 RCSID("$Id$")
  27
  28 #include <freeradius-devel/radiusd.h>
  29
  30 #include <freeradius-devel/connection.h>
  31
  32 #include <freeradius-devel/rad_assert.h>
  33
  34 typedef struct fr_connection_t fr_connection_t;
  35
  36 struct fr_connection_t {
  37         fr_connection_t *prev, *next;
  38
  39         time_t          start;
  40         time_t          last_used;
  41
  42         int             num_uses;
  43         int             used;
  44         int             number;         /* unique ID */
  45         void            *connection;
  46 };
  47
  48 struct fr_connection_pool_t {
  49         int             start;
  50         int             min;
  51         int             max;
  52         int             spare;
  53         int             cleanup_delay;
  54
  55         unsigned int    count;          /* num connections spawned */
  56         int             num;            /* num connections in pool */
  57         int             active;         /* num connections active */
  58
  59         time_t          last_checked;
  60         time_t          last_spawned;
  61         time_t          last_failed;
  62         time_t          last_complained;
  63
  64         int             max_uses;
  65         int             lifetime;
  66         int             idle_timeout;
  67         int             lazy_init;
  68         int             spawning;
  69
  70         fr_connection_t *head, *tail;
  71
  72 #ifdef HAVE_PTHREAD_H
  73         pthread_mutex_t mutex;
  74 #endif
  75
  76         CONF_SECTION    *cs;
  77         void            *ctx;
  78
  79         char            *log_prefix;
  80
  81         fr_connection_create_t  create;
  82         fr_connection_alive_t   alive;
  83         fr_connection_delete_t  delete;
  84 };
  85
  86 #define LOG_PREFIX "rlm_%s (%s)"
  87 #ifndef HAVE_PTHREAD_H
  88 #define pthread_mutex_lock(_x)
  89 #define pthread_mutex_unlock(_x)
  90 #endif
  91
  92 static const CONF_PARSER connection_config[] = {
  93         { "start",    PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, start),
  94           0, "5" },
  95         { "min",      PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, min),
  96           0, "5" },
  97         { "max",      PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, max),
  98           0, "10" },
  99         { "spare",    PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, spare),
 100           0, "3" },
 101         { "uses",     PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, max_uses),
 102           0, "0" },
 103         { "lifetime", PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, lifetime),
 104           0, "0" },
 105         { "cleanup_delay", PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, cleanup_delay),
 106           0, "5" },
 107         { "idle_timeout",  PW_TYPE_INTEGER, offsetof(fr_connection_pool_t, idle_timeout),
 108           0, "60" },
 109         { "lazy",     PW_TYPE_BOOLEAN, offsetof(fr_connection_pool_t, lazy_init),
 110           0, NULL },
 111         { NULL, -1, 0, NULL, NULL }
 112 };
 113
 114 static void fr_connection_unlink(fr_connection_pool_t *fc,
 115                                  fr_connection_t *this)
 116 {
 117
 118         if (this->prev) {
 119                 rad_assert(fc->head != this);
 120                 this->prev->next = this->next;
 121         } else {
 122                 rad_assert(fc->head == this);
 123                 fc->head = this->next;
 124         }
 125         if (this->next) {
 126                 rad_assert(fc->tail != this);
 127                 this->next->prev = this->prev;
 128         } else {
 129                 rad_assert(fc->tail == this);
 130                 fc->tail = this->prev;
 131         }
 132
 133         this->prev = this->next = NULL;
 134 }
 135
 136
 137 static void fr_connection_link(fr_connection_pool_t *fc,
 138                                fr_connection_t *this)
 139 {
 140         rad_assert(fc != NULL);
 141         rad_assert(this != NULL);
 142         rad_assert(fc->head != this);
 143         rad_assert(fc->tail != this);
 144
 145         if (fc->head) fc->head->prev = this;
 146         this->next = fc->head;
 147         this->prev = NULL;
 148         fc->head = this;
 149         if (!fc->tail) {
 150                 rad_assert(this->next == NULL);
 151                 fc->tail = this;
 152         } else {
 153                 rad_assert(this->next != NULL);
 154         }
 155 }
 156
 157
 158 /*
 159  *      Called with the mutex free.
 160  */
 161 static fr_connection_t *fr_connection_spawn(fr_connection_pool_t *fc,
 162                                             time_t now)
 163 {
 164         fr_connection_t *this;
 165         void *conn;
 166
 167         rad_assert(fc != NULL);
 168
 169         pthread_mutex_lock(&fc->mutex);
 170         rad_assert(fc->num <= fc->max);
 171
 172         if ((fc->last_failed == now) || fc->spawning) {
 173                 pthread_mutex_unlock(&fc->mutex);
 174                 return NULL;
 175         }
 176
 177         fc->spawning = TRUE;
 178
 179         /*
 180          *      Unlock the mutex while we try to open a new
 181          *      connection.  If there are issues with the back-end,
 182          *      opening a new connection may take a LONG time.  In
 183          *      that case, we want the other connections to continue
 184          *      to be used.
 185          */
 186         pthread_mutex_unlock(&fc->mutex);
 187
 188         DEBUG("%s: Opening additional connection (%i)",
 189               fc->log_prefix, fc->count);
 190
 191         this = rad_malloc(sizeof(*this));
 192         memset(this, 0, sizeof(*this));
 193
 194         /*
 195          *      This may take a long time, which prevents other
 196          *      threads from releasing connections.  We don't care
 197          *      about other threads opening new connections, as we
 198          *      already have no free connections.
 199          */
 200         conn = fc->create(fc->ctx);
 201         if (!conn) {
 202                 fc->last_failed = now;
 203                 free(this);
 204                 fc->spawning = FALSE; /* atomic, so no lock is needed */
 205                 return NULL;
 206         }
 207
 208         this->start = now;
 209         this->connection = conn;
 210
 211         /*
 212          *      And lock the mutex again while we link the new
 213          *      connection back into the pool.
 214          */
 215         pthread_mutex_lock(&fc->mutex);
 216
 217         this->number = fc->count++;
 218         this->last_used = now;
 219         fr_connection_link(fc, this);
 220         fc->num++;
 221         fc->spawning = FALSE;
 222         fc->last_spawned = time(NULL);
 223
 224         pthread_mutex_unlock(&fc->mutex);
 225
 226         return this;
 227 }
 228
 229 static void fr_connection_close(fr_connection_pool_t *fc,
 230                                 fr_connection_t *this)
 231 {
 232         rad_assert(this->used == FALSE);
 233
 234         fr_connection_unlink(fc, this);
 235         fc->delete(fc->ctx, this->connection);
 236         rad_assert(fc->num > 0);
 237         fc->num--;
 238         free(this);
 239 }
 240
 241
 242
 243 void fr_connection_pool_delete(fr_connection_pool_t *fc)
 244 {
 245         fr_connection_t *this, *next;
 246
 247         DEBUG("%s: Removing connection pool", fc->log_prefix);
 248
 249         pthread_mutex_lock(&fc->mutex);
 250
 251         for (this = fc->head; this != NULL; this = next) {
 252                 next = this->next;
 253                 DEBUG("%s: Closing connection (%i)", fc->log_prefix, this->number);
 254                 fr_connection_close(fc, this);
 255         }
 256
 257         rad_assert(fc->head == NULL);
 258         rad_assert(fc->tail == NULL);
 259         rad_assert(fc->num == 0);
 260
 261         cf_section_parse_free(fc->cs, fc);
 262
 263         free(fc->log_prefix);
 264         free(fc);
 265 }
 266
 267 fr_connection_pool_t *fr_connection_pool_init(CONF_SECTION *parent,
 268                                               void *ctx,
 269                                               fr_connection_create_t c,
 270                                               fr_connection_alive_t a,
 271                                               fr_connection_delete_t d)
 272 {
 273         int i, lp_len;
 274         fr_connection_pool_t *fc;
 275         fr_connection_t *this;
 276         CONF_SECTION *cs;
 277         const char *cs_name1, *cs_name2;
 278         time_t now = time(NULL);
 279
 280         if (!parent || !ctx || !c || !d) return NULL;
 281
 282         cs = cf_section_sub_find(parent, "pool");
 283         if (!cs) {
 284                 cf_log_err(cf_sectiontoitem(parent), "No \"pool\" subsection found");
 285                 return NULL;
 286         }
 287
 288         fc = rad_malloc(sizeof(*fc));
 289         memset(fc, 0, sizeof(*fc));
 290
 291         fc->cs = cs;
 292         fc->ctx = ctx;
 293         fc->create = c;
 294         fc->alive = a;
 295         fc->delete = d;
 296
 297         fc->head = fc->tail = NULL;
 298
 299 #ifdef HAVE_PTHREAD_H
 300         pthread_mutex_init(&fc->mutex, NULL);
 301 #endif
 302
 303         cs_name1 = cf_section_name1(parent);
 304         cs_name2 = cf_section_name2(parent);
 305         if (!cs_name2) {
 306                 cs_name2 = cs_name1;
 307         }
 308
 309         lp_len = (sizeof(LOG_PREFIX) - 4) + strlen(cs_name1) + strlen(cs_name2);
 310         fc->log_prefix = rad_malloc(lp_len);
 311         snprintf(fc->log_prefix, lp_len, LOG_PREFIX, cs_name1, cs_name2);
 312
 313         DEBUG("%s: Initialising connection pool", fc->log_prefix);
 314
 315         if (cf_section_parse(cs, fc, connection_config) < 0) {
 316                 goto error;
 317         }
 318
 319         /*
 320          *      Some simple limits
 321          */
 322         if (fc->max > 1024) fc->max = 1024;
 323         if (fc->start > fc->max) fc->start = fc->max;
 324         if (fc->spare > (fc->max - fc->min)) {
 325                 fc->spare = fc->max - fc->min;
 326         }
 327         if ((fc->lifetime > 0) && (fc->idle_timeout > fc->lifetime)) {
 328                 fc->idle_timeout = 0;
 329         }
 330
 331         /*
 332          *      Create all of the connections, unless the admin says
 333          *      not to.
 334          */
 335         if (!fc->lazy_init) for (i = 0; i < fc->start; i++) {
 336                 this = fr_connection_spawn(fc, now);
 337                 if (!this) {
 338                 error:
 339                         fr_connection_pool_delete(fc);
 340                         return NULL;
 341                 }
 342         }
 343
 344         return fc;
 345 }
 346
 347
 348 /*
 349  *      Called with the mutex lock held.
 350  */
 351 static int fr_connection_manage(fr_connection_pool_t *fc,
 352                                 fr_connection_t *this,
 353                                 time_t now)
 354 {
 355         rad_assert(fc != NULL);
 356         rad_assert(this != NULL);
 357
 358         /*
 359          *      Don't terminated in-use connections
 360          */
 361         if (this->used) return 1;
 362
 363         if ((fc->max_uses > 0) && (this->num_uses >= fc->max_uses)) {
 364                 DEBUG("%s: Closing expired connection (%i): Hit max_uses limit",
 365                         fc->log_prefix, this->number);
 366         do_delete:
 367                 fr_connection_close(fc, this);
 368                 return 0;
 369         }
 370
 371         if ((fc->lifetime > 0) && ((this->start + fc->lifetime) < now)){
 372                 DEBUG("%s: Closing expired connection (%i) ",
 373                         fc->log_prefix, this->number);
 374                 goto do_delete;
 375         }
 376
 377         if ((fc->idle_timeout > 0) && ((this->last_used + fc->idle_timeout) < now)){
 378                 DEBUG("%s: Closing idle connection (%i)",
 379                         fc->log_prefix, this->number);
 380                 goto do_delete;
 381         }
 382
 383         return 1;
 384 }
 385
 386
 387 static int fr_connection_pool_check(fr_connection_pool_t *fc)
 388 {
 389         int spare, spawn;
 390         time_t now = time(NULL);
 391         fr_connection_t *this;
 392
 393         if (fc->last_checked == now) return 1;
 394
 395         pthread_mutex_lock(&fc->mutex);
 396
 397         spare = fc->num - fc->active;
 398
 399         spawn = 0;
 400         if ((fc->num < fc->max) && (spare < fc->spare)) {
 401                 spawn = fc->spare - spare;
 402                 if ((spawn + fc->num) > fc->max) {
 403                         spawn = fc->max - fc->num;
 404                 }
 405                 if (fc->spawning) spawn = 0;
 406
 407                 if (spawn) {
 408                         pthread_mutex_unlock(&fc->mutex);
 409                         fr_connection_spawn(fc, now); /* ignore return code */
 410                 }
 411         }
 412
 413         /*
 414          *      We haven't spawned connections in a while, and there
 415          *      are too many spare ones.  Close the one which has been
 416          *      idle for the longest.
 417          */
 418         if ((now >= (fc->last_spawned + fc->cleanup_delay)) &&
 419             (spare > fc->spare)) {
 420                 fr_connection_t *idle;
 421
 422                 idle = NULL;
 423                 for (this = fc->tail; this != NULL; this = this->prev) {
 424                         if (this->used) continue;
 425
 426                         if (!idle ||
 427                            (this->last_used < idle->last_used)) {
 428                                 idle = this;
 429                         }
 430                 }
 431
 432                 rad_assert(idle != NULL);
 433
 434                 DEBUG("%s: Closing idle connection (%i): Too many free connections",
 435                         fc->log_prefix, idle->number);
 436                 fr_connection_close(fc, idle);
 437         }
 438
 439         /*
 440          *      Pass over all of the connections in the pool, limiting
 441          *      lifetime, idle time, max requests, etc.
 442          */
 443         for (this = fc->head; this != NULL; this = this->next) {
 444                 fr_connection_manage(fc, this, now);
 445         }
 446
 447         fc->last_checked = now;
 448         pthread_mutex_unlock(&fc->mutex);
 449
 450         return 1;
 451 }
 452
 453 int fr_connection_check(fr_connection_pool_t *fc, void *conn)
 454 {
 455         int rcode = 1;
 456         fr_connection_t *this;
 457         time_t now;
 458
 459         if (!fc) return 1;
 460
 461         if (!conn) return fr_connection_pool_check(fc);
 462
 463         now = time(NULL);
 464         pthread_mutex_lock(&fc->mutex);
 465
 466         for (this = fc->head; this != NULL; this = this->next) {
 467                 if (this->connection == conn) {
 468                         rcode = fr_connection_manage(fc, conn, now);
 469                         break;
 470                 }
 471         }
 472
 473         pthread_mutex_unlock(&fc->mutex);
 474
 475         return 1;
 476 }
 477
 478
 479 void *fr_connection_get(fr_connection_pool_t *fc)
 480 {
 481         time_t now;
 482         fr_connection_t *this, *next;
 483
 484         if (!fc) return NULL;
 485
 486         pthread_mutex_lock(&fc->mutex);
 487
 488         now = time(NULL);
 489         for (this = fc->head; this != NULL; this = next) {
 490                 next = this->next;
 491
 492                 if (!fr_connection_manage(fc, this, now)) continue;
 493
 494                 if (!this->used) goto do_return;
 495         }
 496
 497         if (fc->num == fc->max) {
 498                 /*
 499                  *      Rate-limit complaints.
 500                  */
 501                 if (fc->last_complained != now) {
 502                         radlog(L_ERR, "%s: No connections available and at max connection limit",
 503                                fc->log_prefix);
 504                         fc->last_complained = now;
 505                 }
 506                 pthread_mutex_unlock(&fc->mutex);
 507                 return NULL;
 508         }
 509
 510         pthread_mutex_unlock(&fc->mutex);
 511         this = fr_connection_spawn(fc, now);
 512         if (!this) return NULL;
 513         pthread_mutex_lock(&fc->mutex);
 514
 515 do_return:
 516         fc->active++;
 517         this->num_uses++;
 518         this->last_used = now;
 519         this->used = TRUE;
 520
 521         pthread_mutex_unlock(&fc->mutex);
 522
 523         DEBUG("%s: Reserved connection (%i)", fc->log_prefix, this->number);
 524
 525         return this->connection;
 526 }
 527
 528 void fr_connection_release(fr_connection_pool_t *fc, void *conn)
 529 {
 530         fr_connection_t *this;
 531
 532         if (!fc || !conn) return;
 533
 534         pthread_mutex_lock(&fc->mutex);
 535
 536         /*
 537          *      FIXME: This loop could be avoided if we passed a 'void
 538          *      **connection' instead.  We could use "offsetof" in
 539          *      order to find top of the parent structure.
 540          */
 541         for (this = fc->head; this != NULL; this = this->next) {
 542                 if (this->connection == conn) {
 543                         rad_assert(this->used == TRUE);
 544                         this->used = FALSE;
 545
 546                         /*
 547                          *      Put it at the head of the list, so
 548                          *      that it will get re-used quickly.
 549                          */
 550                         if (this != fc->head) {
 551                                 fr_connection_unlink(fc, this);
 552                                 fr_connection_link(fc, this);
 553                         }
 554                         rad_assert(fc->active > 0);
 555                         fc->active--;
 556                         break;
 557                 }
 558         }
 559
 560         pthread_mutex_unlock(&fc->mutex);
 561
 562         DEBUG("%s: Released connection (%i)", fc->log_prefix, this->number);
 563
 564         /*
 565          *      We mirror the "spawn on get" functionality by having
 566          *      "delete on release".  If there are too many spare
 567          *      connections, go manage the pool && clean some up.
 568          */
 569         fr_connection_pool_check(fc);
 570
 571 }
 572
 573 void *fr_connection_reconnect(fr_connection_pool_t *fc, void *conn)
 574 {
 575         void *new_conn;
 576         fr_connection_t *this;
 577         int conn_number;
 578
 579         if (!fc || !conn) return NULL;
 580
 581         pthread_mutex_lock(&fc->mutex);
 582
 583         conn_number = this->number;
 584
 585         /*
 586          *      FIXME: This loop could be avoided if we passed a 'void
 587          *      **connection' instead.  We could use "offsetof" in
 588          *      order to find top of the parent structure.
 589          */
 590         for (this = fc->head; this != NULL; this = this->next) {
 591                 if (this->connection != conn) continue;
 592
 593                 rad_assert(this->used == TRUE);
 594
 595                 DEBUG("%s: Reconnecting (%i)", fc->log_prefix, conn_number);
 596
 597                 new_conn = fc->create(fc->ctx);
 598                 if (!new_conn) {
 599                         time_t now = time(NULL);
 600
 601                         if (fc->last_complained == now) {
 602                                 now = 0;
 603                         } else {
 604                                 fc->last_complained = now;
 605                         }
 606
 607                         fr_connection_close(fc, conn);
 608                         pthread_mutex_unlock(&fc->mutex);
 609
 610                         /*
 611                          *      Can't create a new socket.
 612                          *      Try grabbing a pre-existing one.
 613                          */
 614                         new_conn = fr_connection_get(fc);
 615                         if (new_conn) return new_conn;
 616
 617                         if (!now) return NULL;
 618
 619                         radlog(L_ERR, "%s: Failed to reconnect (%i), and no other connections available",
 620                                fc->log_prefix, conn_number);
 621                         return NULL;
 622                 }
 623
 624                 fc->delete(fc->ctx, conn);
 625                 this->connection = new_conn;
 626                 pthread_mutex_unlock(&fc->mutex);
 627                 return new_conn;
 628         }
 629
 630         pthread_mutex_unlock(&fc->mutex);
 631
 632         /*
 633          *      Caller passed us something that isn't in the pool.
 634          */
 635         return NULL;
 636 }