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                 pthread_mutex_unlock(&fc->mutex);
 369                 return 0;
 370         }
 371
 372         if ((fc->lifetime > 0) && ((this->start + fc->lifetime) < now)){
 373                 DEBUG("%s: Closing expired connection (%i) ",
 374                         fc->log_prefix, this->number);
 375                 goto do_delete;
 376         }
 377
 378         if ((fc->idle_timeout > 0) && ((this->last_used + fc->idle_timeout) < now)){
 379                 DEBUG("%s: Closing idle connection (%i)",
 380                         fc->log_prefix, this->number);
 381                 goto do_delete;
 382         }
 383
 384         return 1;
 385 }
 386
 387
 388 static int fr_connection_pool_check(fr_connection_pool_t *fc)
 389 {
 390         int spare, spawn;
 391         time_t now = time(NULL);
 392         fr_connection_t *this;
 393
 394         if (fc->last_checked == now) return 1;
 395
 396         pthread_mutex_lock(&fc->mutex);
 397
 398         spare = fc->num - fc->active;
 399
 400         spawn = 0;
 401         if ((fc->num < fc->max) && (spare < fc->spare)) {
 402                 spawn = fc->spare - spare;
 403                 if ((spawn + fc->num) > fc->max) {
 404                         spawn = fc->max - fc->num;
 405                 }
 406                 if (fc->spawning) spawn = 0;
 407
 408                 if (spawn) {
 409                         pthread_mutex_unlock(&fc->mutex);
 410                         fr_connection_spawn(fc, now); /* ignore return code */
 411                 }
 412         }
 413
 414         /*
 415          *      We haven't spawned connections in a while, and there
 416          *      are too many spare ones.  Close the one which has been
 417          *      idle for the longest.
 418          */
 419         if ((now >= (fc->last_spawned + fc->cleanup_delay)) &&
 420             (spare > fc->spare)) {
 421                 fr_connection_t *idle;
 422
 423                 idle = NULL;
 424                 for (this = fc->tail; this != NULL; this = this->prev) {
 425                         if (this->used) continue;
 426
 427                         if (!idle ||
 428                            (this->last_used < idle->last_used)) {
 429                                 idle = this;
 430                         }
 431                 }
 432
 433                 rad_assert(idle != NULL);
 434
 435                 DEBUG("%s: Closing idle connection (%i): Too many free connections",
 436                         fc->log_prefix, idle->number);
 437                 fr_connection_close(fc, idle);
 438         }
 439
 440         /*
 441          *      Pass over all of the connections in the pool, limiting
 442          *      lifetime, idle time, max requests, etc.
 443          */
 444         for (this = fc->head; this != NULL; this = this->next) {
 445                 fr_connection_manage(fc, this, now);
 446         }
 447
 448         fc->last_checked = now;
 449         pthread_mutex_unlock(&fc->mutex);
 450
 451         return 1;
 452 }
 453
 454 int fr_connection_check(fr_connection_pool_t *fc, void *conn)
 455 {
 456         int rcode = 1;
 457         fr_connection_t *this;
 458         time_t now;
 459
 460         if (!fc) return 1;
 461
 462         if (!conn) return fr_connection_pool_check(fc);
 463
 464         now = time(NULL);
 465         pthread_mutex_lock(&fc->mutex);
 466
 467         for (this = fc->head; this != NULL; this = this->next) {
 468                 if (this->connection == conn) {
 469                         rcode = fr_connection_manage(fc, conn, now);
 470                         break;
 471                 }
 472         }
 473
 474         pthread_mutex_unlock(&fc->mutex);
 475
 476         return 1;
 477 }
 478
 479
 480 void *fr_connection_get(fr_connection_pool_t *fc)
 481 {
 482         time_t now;
 483         fr_connection_t *this, *next;
 484
 485         if (!fc) return NULL;
 486
 487         pthread_mutex_lock(&fc->mutex);
 488
 489         now = time(NULL);
 490         for (this = fc->head; this != NULL; this = next) {
 491                 next = this->next;
 492
 493                 if (!fr_connection_manage(fc, this, now)) continue;
 494
 495                 if (!this->used) goto do_return;
 496         }
 497
 498         if (fc->num == fc->max) {
 499                 /*
 500                  *      Rate-limit complaints.
 501                  */
 502                 if (fc->last_complained != now) {
 503                         radlog(L_ERR, "%s: No connections available and at max connection limit",
 504                                fc->log_prefix);
 505                         fc->last_complained = now;
 506                 }
 507                 pthread_mutex_unlock(&fc->mutex);
 508                 return NULL;
 509         }
 510
 511         pthread_mutex_unlock(&fc->mutex);
 512         this = fr_connection_spawn(fc, now);
 513         if (!this) return NULL;
 514         pthread_mutex_lock(&fc->mutex);
 515
 516 do_return:
 517         fc->active++;
 518         this->num_uses++;
 519         this->last_used = now;
 520         this->used = TRUE;
 521
 522         pthread_mutex_unlock(&fc->mutex);
 523
 524         DEBUG("%s: Reserved connection (%i)", fc->log_prefix, this->number);
 525
 526         return this->connection;
 527 }
 528
 529 void fr_connection_release(fr_connection_pool_t *fc, void *conn)
 530 {
 531         fr_connection_t *this;
 532
 533         if (!fc || !conn) return;
 534
 535         pthread_mutex_lock(&fc->mutex);
 536
 537         /*
 538          *      FIXME: This loop could be avoided if we passed a 'void
 539          *      **connection' instead.  We could use "offsetof" in
 540          *      order to find top of the parent structure.
 541          */
 542         for (this = fc->head; this != NULL; this = this->next) {
 543                 if (this->connection == conn) {
 544                         rad_assert(this->used == TRUE);
 545                         this->used = FALSE;
 546
 547                         /*
 548                          *      Put it at the head of the list, so
 549                          *      that it will get re-used quickly.
 550                          */
 551                         if (this != fc->head) {
 552                                 fr_connection_unlink(fc, this);
 553                                 fr_connection_link(fc, this);
 554                         }
 555                         rad_assert(fc->active > 0);
 556                         fc->active--;
 557                         break;
 558                 }
 559         }
 560
 561         pthread_mutex_unlock(&fc->mutex);
 562
 563         DEBUG("%s: Released connection (%i)", fc->log_prefix, this->number);
 564
 565         /*
 566          *      We mirror the "spawn on get" functionality by having
 567          *      "delete on release".  If there are too many spare
 568          *      connections, go manage the pool && clean some up.
 569          */
 570         fr_connection_pool_check(fc);
 571
 572 }
 573
 574 void *fr_connection_reconnect(fr_connection_pool_t *fc, void *conn)
 575 {
 576         void *new_conn;
 577         fr_connection_t *this;
 578         int conn_number;
 579
 580         if (!fc || !conn) return NULL;
 581
 582         pthread_mutex_lock(&fc->mutex);
 583
 584         conn_number = this->number;
 585
 586         /*
 587          *      FIXME: This loop could be avoided if we passed a 'void
 588          *      **connection' instead.  We could use "offsetof" in
 589          *      order to find top of the parent structure.
 590          */
 591         for (this = fc->head; this != NULL; this = this->next) {
 592                 if (this->connection != conn) continue;
 593
 594                 rad_assert(this->used == TRUE);
 595
 596                 DEBUG("%s: Reconnecting (%i)", fc->log_prefix, conn_number);
 597
 598                 new_conn = fc->create(fc->ctx);
 599                 if (!new_conn) {
 600                         time_t now = time(NULL);
 601
 602                         if (fc->last_complained == now) {
 603                                 now = 0;
 604                         } else {
 605                                 fc->last_complained = now;
 606                         }
 607
 608                         fr_connection_close(fc, conn);
 609                         pthread_mutex_unlock(&fc->mutex);
 610
 611                         /*
 612                          *      Can't create a new socket.
 613                          *      Try grabbing a pre-existing one.
 614                          */
 615                         new_conn = fr_connection_get(fc);
 616                         if (new_conn) return new_conn;
 617
 618                         if (!now) return NULL;
 619
 620                         radlog(L_ERR, "%s: Failed to reconnect (%i), and no other connections available",
 621                                fc->log_prefix, conn_number);
 622                         return NULL;
 623                 }
 624
 625                 fc->delete(fc->ctx, conn);
 626                 this->connection = new_conn;
 627                 pthread_mutex_unlock(&fc->mutex);
 628                 return new_conn;
 629         }
 630
 631         pthread_mutex_unlock(&fc->mutex);
 632
 633         /*
 634          *      Caller passed us something that isn't in the pool.
 635          */
 636         return NULL;
 637 }