client.c revision 1.3
1 /* $NetBSD: client.c,v 1.3 2025/09/05 21:16:24 christos Exp $ */ 2 3 /* $OpenLDAP$ */ 4 /* This work is part of OpenLDAP Software <http://www.openldap.org/>. 5 * 6 * Copyright 1998-2024 The OpenLDAP Foundation. 7 * All rights reserved. 8 * 9 * Redistribution and use in source and binary forms, with or without 10 * modification, are permitted only as authorized by the OpenLDAP 11 * Public License. 12 * 13 * A copy of this license is available in the file LICENSE in the 14 * top-level directory of the distribution or, alternatively, at 15 * <http://www.OpenLDAP.org/license.html>. 16 */ 17 18 #include <sys/cdefs.h> 19 __RCSID("$NetBSD: client.c,v 1.3 2025/09/05 21:16:24 christos Exp $"); 20 21 #include "portable.h" 22 23 #include <ac/socket.h> 24 #include <ac/errno.h> 25 #include <ac/string.h> 26 #include <ac/time.h> 27 #include <ac/unistd.h> 28 29 #include "lutil.h" 30 #include "lload.h" 31 32 long lload_client_max_pending = 0; 33 34 lload_c_head clients = LDAP_CIRCLEQ_HEAD_INITIALIZER( clients ); 35 36 ldap_pvt_thread_mutex_t clients_mutex; 37 38 static void client_unlink( LloadConnection *upstream ); 39 40 int 41 request_abandon( LloadConnection *c, LloadOperation *op ) 42 { 43 LloadOperation *request, needle = { .o_client_connid = c->c_connid }; 44 int rc = LDAP_SUCCESS; 45 46 op->o_res = LLOAD_OP_COMPLETED; 47 48 if ( ber_decode_int( &op->o_request, &needle.o_client_msgid ) ) { 49 Debug( LDAP_DEBUG_STATS, "request_abandon: " 50 "connid=%lu msgid=%d invalid integer sent in abandon request\n", 51 c->c_connid, op->o_client_msgid ); 52 53 OPERATION_UNLINK(op); 54 CONNECTION_LOCK_DESTROY(c); 55 return -1; 56 } 57 58 CONNECTION_LOCK(c); 59 request = ldap_tavl_find( c->c_ops, &needle, operation_client_cmp ); 60 if ( !request ) { 61 Debug( LDAP_DEBUG_STATS, "request_abandon: " 62 "connid=%lu msgid=%d requests abandon of an operation " 63 "msgid=%d not being processed anymore\n", 64 c->c_connid, op->o_client_msgid, needle.o_client_msgid ); 65 CONNECTION_UNLOCK(c); 66 goto done; 67 } else if ( request->o_tag == LDAP_REQ_BIND ) { 68 /* RFC 4511 states we must not allow Abandon on Binds */ 69 Debug( LDAP_DEBUG_STATS, "request_abandon: " 70 "connid=%lu msgid=%d requests abandon of a bind operation " 71 "msgid=%d\n", 72 c->c_connid, op->o_client_msgid, needle.o_client_msgid ); 73 CONNECTION_UNLOCK(c); 74 goto done; 75 } 76 Debug( LDAP_DEBUG_STATS, "request_abandon: " 77 "connid=%lu msgid=%d abandoning %s msgid=%d\n", 78 c->c_connid, op->o_client_msgid, 79 lload_msgtype2str( request->o_tag ), needle.o_client_msgid ); 80 81 if ( c->c_state == LLOAD_C_BINDING ) { 82 assert(0); 83 } 84 85 CONNECTION_UNLOCK(c); 86 operation_abandon( request ); 87 88 done: 89 OPERATION_UNLINK(op); 90 return rc; 91 } 92 93 int 94 request_process( LloadConnection *client, LloadOperation *op ) 95 { 96 BerElement *output; 97 LloadConnection *upstream = NULL; 98 LloadBackend *b = NULL; 99 ber_int_t msgid; 100 int res = LDAP_UNAVAILABLE, rc = LDAP_SUCCESS; 101 char *message = "no connections available"; 102 enum op_restriction client_restricted; 103 104 if ( lload_control_actions && !BER_BVISNULL( &op->o_ctrls ) ) { 105 BerElementBuffer copy_berbuf; 106 BerElement *copy = (BerElement *)©_berbuf; 107 struct berval control; 108 109 ber_init2( copy, &op->o_ctrls, 0 ); 110 111 while ( ber_skip_element( copy, &control ) == LBER_SEQUENCE ) { 112 struct restriction_entry *entry, needle = {}; 113 BerElementBuffer control_berbuf; 114 BerElement *control_ber = (BerElement *)&control_berbuf; 115 116 ber_init2( control_ber, &control, 0 ); 117 118 if ( ber_skip_element( control_ber, &needle.oid ) == LBER_ERROR ) { 119 res = LDAP_PROTOCOL_ERROR; 120 message = "invalid control"; 121 122 operation_send_reject( op, res, message, 1 ); 123 goto fail; 124 } 125 126 entry = ldap_tavl_find( 127 lload_control_actions, &needle, lload_restriction_cmp ); 128 if ( entry && op->o_restricted < entry->action ) { 129 op->o_restricted = entry->action; 130 } 131 } 132 } 133 if ( op->o_restricted < LLOAD_OP_RESTRICTED_WRITE && 134 lload_write_coherence && 135 op->o_tag != LDAP_REQ_SEARCH && 136 op->o_tag != LDAP_REQ_COMPARE ) { 137 op->o_restricted = LLOAD_OP_RESTRICTED_WRITE; 138 } 139 140 if ( op->o_restricted == LLOAD_OP_RESTRICTED_REJECT ) { 141 res = LDAP_UNWILLING_TO_PERFORM; 142 message = "extended operation or control disallowed"; 143 144 operation_send_reject( op, res, message, 1 ); 145 goto fail; 146 } 147 148 CONNECTION_LOCK(client); 149 client_restricted = client->c_restricted; 150 if ( client_restricted ) { 151 if ( client_restricted == LLOAD_OP_RESTRICTED_WRITE && 152 client->c_restricted_inflight == 0 && 153 client->c_restricted_at >= 0 && 154 client->c_restricted_at + lload_write_coherence < 155 op->o_start.tv_sec ) { 156 Debug( LDAP_DEBUG_TRACE, "request_process: " 157 "connid=%lu write coherence to backend '%s' expired\n", 158 client->c_connid, client->c_backend->b_name.bv_val ); 159 client->c_backend = NULL; 160 client_restricted = client->c_restricted = LLOAD_OP_NOT_RESTRICTED; 161 } 162 switch ( client_restricted ) { 163 case LLOAD_OP_NOT_RESTRICTED: 164 break; 165 case LLOAD_OP_RESTRICTED_WRITE: 166 case LLOAD_OP_RESTRICTED_BACKEND: 167 b = client->c_backend; 168 assert( b ); 169 break; 170 case LLOAD_OP_RESTRICTED_UPSTREAM: 171 case LLOAD_OP_RESTRICTED_ISOLATE: 172 upstream = client->c_linked_upstream; 173 assert( upstream ); 174 break; 175 default: 176 assert(0); 177 break; 178 } 179 } 180 if ( op->o_restricted < client_restricted ) { 181 op->o_restricted = client_restricted; 182 } 183 CONNECTION_UNLOCK(client); 184 185 if ( upstream ) { 186 b = upstream->c_backend; 187 checked_lock( &b->b_mutex ); 188 if ( !try_upstream( b, NULL, op, upstream, &res, &message ) ) { 189 upstream = NULL; 190 } 191 checked_unlock( &b->b_mutex ); 192 } else if ( b ) { 193 backend_select( b, op, &upstream, &res, &message ); 194 } else { 195 upstream_select( op, &upstream, &res, &message ); 196 } 197 198 if ( !upstream ) { 199 Debug( LDAP_DEBUG_STATS, "request_process: " 200 "connid=%lu, msgid=%d no available connection found\n", 201 op->o_client_connid, op->o_client_msgid ); 202 203 operation_send_reject( op, res, message, 1 ); 204 goto fail; 205 } 206 CONNECTION_ASSERT_LOCKED(upstream); 207 assert_locked( &upstream->c_io_mutex ); 208 op->o_upstream = upstream; 209 op->o_upstream_connid = upstream->c_connid; 210 op->o_res = LLOAD_OP_FAILED; 211 212 /* Was it unlinked in the meantime? No need to send a response since the 213 * client is dead */ 214 if ( !IS_ALIVE( op, o_refcnt ) ) { 215 LloadBackend *b = upstream->c_backend; 216 217 upstream->c_n_ops_executing--; 218 checked_unlock( &upstream->c_io_mutex ); 219 CONNECTION_UNLOCK(upstream); 220 221 checked_lock( &b->b_mutex ); 222 b->b_n_ops_executing--; 223 checked_unlock( &b->b_mutex ); 224 225 assert( !IS_ALIVE( client, c_live ) ); 226 checked_lock( &op->o_link_mutex ); 227 if ( op->o_upstream ) { 228 op->o_upstream = NULL; 229 } 230 checked_unlock( &op->o_link_mutex ); 231 return -1; 232 } 233 234 output = upstream->c_pendingber; 235 if ( output == NULL && (output = ber_alloc()) == NULL ) { 236 LloadBackend *b = upstream->c_backend; 237 238 upstream->c_n_ops_executing--; 239 CONNECTION_UNLOCK(upstream); 240 checked_unlock( &upstream->c_io_mutex ); 241 242 checked_lock( &b->b_mutex ); 243 b->b_n_ops_executing--; 244 operation_update_backend_counters( op, b ); 245 checked_unlock( &b->b_mutex ); 246 247 Debug( LDAP_DEBUG_ANY, "request_process: " 248 "ber_alloc failed\n" ); 249 250 rc = -1; 251 goto fail; 252 } 253 upstream->c_pendingber = output; 254 255 if ( client_restricted < LLOAD_OP_RESTRICTED_UPSTREAM && 256 op->o_restricted >= LLOAD_OP_RESTRICTED_UPSTREAM ) { 257 rc = ldap_tavl_insert( 258 &upstream->c_linked, client, lload_upstream_entry_cmp, 259 ldap_avl_dup_error ); 260 assert( rc == LDAP_SUCCESS ); 261 } 262 263 op->o_upstream_msgid = msgid = upstream->c_next_msgid++; 264 rc = ldap_tavl_insert( 265 &upstream->c_ops, op, operation_upstream_cmp, ldap_avl_dup_error ); 266 267 CONNECTION_UNLOCK(upstream); 268 269 Debug( LDAP_DEBUG_TRACE, "request_process: " 270 "client connid=%lu added %s msgid=%d to upstream connid=%lu as " 271 "msgid=%d\n", 272 op->o_client_connid, lload_msgtype2str( op->o_tag ), 273 op->o_client_msgid, op->o_upstream_connid, op->o_upstream_msgid ); 274 assert( rc == LDAP_SUCCESS ); 275 276 lload_stats.counters[LLOAD_STATS_OPS_OTHER].lc_ops_forwarded++; 277 278 if ( op->o_restricted > client_restricted || 279 client_restricted == LLOAD_OP_RESTRICTED_WRITE ) { 280 CONNECTION_LOCK(client); 281 if ( op->o_restricted > client_restricted ) { 282 client->c_restricted = op->o_restricted; 283 } 284 if ( op->o_restricted == LLOAD_OP_RESTRICTED_WRITE ) { 285 client->c_restricted_inflight++; 286 } 287 if ( op->o_restricted >= LLOAD_OP_RESTRICTED_UPSTREAM ) { 288 if ( client_restricted < LLOAD_OP_RESTRICTED_UPSTREAM ) { 289 client->c_linked_upstream = upstream; 290 } 291 assert( client->c_linked_upstream == upstream ); 292 client->c_backend = NULL; 293 } else if ( op->o_restricted >= LLOAD_OP_RESTRICTED_WRITE ) { 294 if ( client_restricted < LLOAD_OP_RESTRICTED_WRITE ) { 295 client->c_backend = upstream->c_backend; 296 } 297 assert( client->c_backend == upstream->c_backend ); 298 } 299 CONNECTION_UNLOCK(client); 300 } 301 302 if ( (lload_features & LLOAD_FEATURE_PROXYAUTHZ) && 303 client->c_type != LLOAD_C_PRIVILEGED ) { 304 CONNECTION_LOCK(client); 305 Debug( LDAP_DEBUG_TRACE, "request_process: " 306 "proxying identity %s to upstream\n", 307 client->c_auth.bv_val ); 308 ber_printf( output, "t{titOt{{sbO}" /* "}}" */, LDAP_TAG_MESSAGE, 309 LDAP_TAG_MSGID, msgid, 310 op->o_tag, &op->o_request, 311 LDAP_TAG_CONTROLS, 312 LDAP_CONTROL_PROXY_AUTHZ, 1, &client->c_auth ); 313 CONNECTION_UNLOCK(client); 314 315 if ( !BER_BVISNULL( &op->o_ctrls ) ) { 316 ber_write( output, op->o_ctrls.bv_val, op->o_ctrls.bv_len, 0 ); 317 } 318 319 ber_printf( output, /* "{{" */ "}}" ); 320 } else { 321 ber_printf( output, "t{titOtO}", LDAP_TAG_MESSAGE, 322 LDAP_TAG_MSGID, msgid, 323 op->o_tag, &op->o_request, 324 LDAP_TAG_CONTROLS, BER_BV_OPTIONAL( &op->o_ctrls ) ); 325 } 326 checked_unlock( &upstream->c_io_mutex ); 327 328 connection_write_cb( -1, 0, upstream ); 329 return rc; 330 331 fail: 332 if ( upstream ) { 333 CONNECTION_LOCK_DESTROY(upstream); 334 335 /* We have not committed any restrictions in the end */ 336 op->o_restricted = LLOAD_OP_NOT_RESTRICTED; 337 operation_send_reject( op, LDAP_OTHER, "internal error", 0 ); 338 } 339 340 OPERATION_UNLINK(op); 341 if ( rc ) { 342 CONNECTION_LOCK_DESTROY(client); 343 } 344 return rc; 345 } 346 347 int 348 handle_one_request( LloadConnection *c ) 349 { 350 BerElement *ber; 351 LloadOperation *op = NULL; 352 RequestHandler handler = NULL; 353 int over_limit = 0; 354 enum sc_state state; 355 enum sc_io_state io_state; 356 357 ber = c->c_currentber; 358 c->c_currentber = NULL; 359 360 CONNECTION_LOCK(c); 361 op = operation_init( c, ber ); 362 if ( !op ) { 363 Debug( LDAP_DEBUG_ANY, "handle_one_request: " 364 "connid=%lu, operation_init failed\n", 365 c->c_connid ); 366 CONNECTION_DESTROY(c); 367 ber_free( ber, 1 ); 368 return -1; 369 } 370 if ( lload_client_max_pending && 371 c->c_n_ops_executing >= lload_client_max_pending ) { 372 over_limit = 1; 373 } 374 375 /* 376 * Remember the current state so we don't have to lock again, 377 * we're only screening whether we can keep going, e.g. noone can change 378 * state to LLOAD_C_BINDING from under us (would imply a new operation was 379 * received but that's us), but the opposite is possible - a Bind response 380 * could be received and processed in the meantime. 381 */ 382 state = c->c_state; 383 CONNECTION_UNLOCK(c); 384 385 switch ( op->o_tag ) { 386 case LDAP_REQ_UNBIND: 387 /* There is never a response for this operation */ 388 op->o_res = LLOAD_OP_COMPLETED; 389 OPERATION_UNLINK(op); 390 391 Debug( LDAP_DEBUG_STATS, "handle_one_request: " 392 "received unbind, closing client connid=%lu\n", 393 c->c_connid ); 394 CONNECTION_LOCK_DESTROY(c); 395 return -1; 396 case LDAP_REQ_BIND: 397 handler = request_bind; 398 break; 399 case LDAP_REQ_ABANDON: 400 /* We can't send a response to abandon requests even if a bind is 401 * currently in progress */ 402 return request_abandon( c, op ); 403 case LDAP_REQ_EXTENDED: 404 default: 405 if ( state == LLOAD_C_BINDING ) { 406 operation_send_reject( 407 op, LDAP_PROTOCOL_ERROR, "bind in progress", 0 ); 408 return LDAP_SUCCESS; 409 } 410 if ( over_limit ) { 411 operation_send_reject( op, LDAP_BUSY, 412 "pending operation limit reached on this connection", 413 0 ); 414 return LDAP_SUCCESS; 415 } 416 417 checked_lock( &c->c_io_mutex ); 418 io_state = c->c_io_state; 419 checked_unlock( &c->c_io_mutex ); 420 if ( io_state & LLOAD_C_READ_PAUSE ) { 421 operation_send_reject( op, LDAP_BUSY, 422 "writing side backlogged, please keep reading", 0 ); 423 return LDAP_SUCCESS; 424 } 425 426 if ( op->o_tag == LDAP_REQ_EXTENDED ) { 427 handler = request_extended; 428 } else { 429 handler = request_process; 430 } 431 break; 432 } 433 434 if ( state == LLOAD_C_CLOSING ) { 435 operation_send_reject( 436 op, LDAP_UNAVAILABLE, "connection is shutting down", 0 ); 437 return LDAP_SUCCESS; 438 } 439 440 return handler( c, op ); 441 } 442 443 #ifdef HAVE_TLS 444 /* 445 * The connection has a token assigned to it when the callback is set up. 446 */ 447 void 448 client_tls_handshake_cb( evutil_socket_t s, short what, void *arg ) 449 { 450 LloadConnection *c = arg; 451 epoch_t epoch; 452 int rc = 0; 453 454 if ( what & EV_TIMEOUT ) { 455 Debug( LDAP_DEBUG_CONNS, "client_tls_handshake_cb: " 456 "connid=%lu, timeout reached, destroying\n", 457 c->c_connid ); 458 goto fail; 459 } 460 461 /* 462 * In case of StartTLS, make sure we flush the response first. 463 * Also before we try to read anything from the connection, it isn't 464 * permitted to Abandon a StartTLS exop per RFC4511 anyway. 465 */ 466 checked_lock( &c->c_io_mutex ); 467 if ( c->c_pendingber ) { 468 checked_unlock( &c->c_io_mutex ); 469 connection_write_cb( s, what, arg ); 470 471 if ( !IS_ALIVE( c, c_live ) ) { 472 goto fail; 473 } 474 475 /* Do we still have data pending? If so, connection_write_cb would 476 * already have arranged the write callback to trigger again */ 477 checked_lock( &c->c_io_mutex ); 478 if ( c->c_pendingber ) { 479 checked_unlock( &c->c_io_mutex ); 480 return; 481 } 482 } 483 484 rc = ldap_pvt_tls_accept( c->c_sb, LLOAD_TLS_CTX ); 485 checked_unlock( &c->c_io_mutex ); 486 if ( rc < 0 ) { 487 goto fail; 488 } 489 490 if ( rc == 0 ) { 491 struct event_base *base = event_get_base( c->c_read_event ); 492 493 /* 494 * We're finished, replace the callbacks 495 * 496 * This is deadlock-safe, since both share the same base - the one 497 * that's just running us. 498 */ 499 CONNECTION_LOCK(c); 500 event_del( c->c_read_event ); 501 event_del( c->c_write_event ); 502 503 c->c_read_timeout = NULL; 504 event_assign( c->c_read_event, base, c->c_fd, EV_READ|EV_PERSIST, 505 connection_read_cb, c ); 506 if ( IS_ALIVE( c, c_live ) ) { 507 event_add( c->c_read_event, c->c_read_timeout ); 508 } 509 510 event_assign( c->c_write_event, base, c->c_fd, EV_WRITE, 511 connection_write_cb, c ); 512 Debug( LDAP_DEBUG_CONNS, "client_tls_handshake_cb: " 513 "connid=%lu finished\n", 514 c->c_connid ); 515 516 c->c_is_tls = LLOAD_TLS_ESTABLISHED; 517 CONNECTION_UNLOCK(c); 518 return; 519 } else if ( ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_NEEDS_WRITE, NULL ) ) { 520 if ( IS_ALIVE( c, c_live ) ) { 521 CONNECTION_LOCK(c); 522 event_add( c->c_write_event, lload_write_timeout ); 523 CONNECTION_UNLOCK(c); 524 } 525 Debug( LDAP_DEBUG_CONNS, "client_tls_handshake_cb: " 526 "connid=%lu need write rc=%d\n", 527 c->c_connid, rc ); 528 } 529 return; 530 531 fail: 532 Debug( LDAP_DEBUG_CONNS, "client_tls_handshake_cb: " 533 "connid=%lu failed rc=%d\n", 534 c->c_connid, rc ); 535 536 assert( c->c_ops == NULL ); 537 epoch = epoch_join(); 538 CONNECTION_LOCK_DESTROY(c); 539 epoch_leave( epoch ); 540 } 541 #endif /* HAVE_TLS */ 542 543 LloadConnection * 544 client_init( 545 ber_socket_t s, 546 const char *peername, 547 struct event_base *base, 548 int flags ) 549 { 550 LloadConnection *c; 551 struct event *event; 552 event_callback_fn read_cb = connection_read_cb, 553 write_cb = connection_write_cb; 554 555 if ( (c = lload_connection_init( s, peername, flags) ) == NULL ) { 556 return NULL; 557 } 558 559 { 560 ber_len_t max = sockbuf_max_incoming_client; 561 ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_SET_MAX_INCOMING, &max ); 562 } 563 564 c->c_state = LLOAD_C_READY; 565 566 if ( flags & CONN_IS_TLS ) { 567 #ifdef HAVE_TLS 568 int rc; 569 570 c->c_is_tls = LLOAD_LDAPS; 571 572 rc = ldap_pvt_tls_accept( c->c_sb, LLOAD_TLS_CTX ); 573 if ( rc < 0 ) { 574 Debug( LDAP_DEBUG_CONNS, "client_init: " 575 "connid=%lu failed initial TLS accept rc=%d\n", 576 c->c_connid, rc ); 577 CONNECTION_LOCK(c); 578 goto fail; 579 } 580 581 if ( rc ) { 582 c->c_read_timeout = lload_timeout_net; 583 read_cb = write_cb = client_tls_handshake_cb; 584 } 585 #else /* ! HAVE_TLS */ 586 assert(0); 587 #endif /* ! HAVE_TLS */ 588 } 589 590 event = event_new( base, s, EV_READ|EV_PERSIST, read_cb, c ); 591 if ( !event ) { 592 Debug( LDAP_DEBUG_ANY, "client_init: " 593 "Read event could not be allocated\n" ); 594 CONNECTION_LOCK(c); 595 goto fail; 596 } 597 c->c_read_event = event; 598 599 event = event_new( base, s, EV_WRITE, write_cb, c ); 600 if ( !event ) { 601 Debug( LDAP_DEBUG_ANY, "client_init: " 602 "Write event could not be allocated\n" ); 603 CONNECTION_LOCK(c); 604 goto fail; 605 } 606 c->c_write_event = event; 607 608 CONNECTION_LOCK(c); 609 #ifdef BALANCER_MODULE 610 if ( lload_monitor_client_subsys ) { 611 acquire_ref( &c->c_refcnt ); 612 CONNECTION_UNLOCK(c); 613 if ( lload_monitor_conn_entry_create( 614 c, lload_monitor_client_subsys ) ) { 615 CONNECTION_LOCK(c); 616 RELEASE_REF( c, c_refcnt, c->c_destroy ); 617 goto fail; 618 } 619 CONNECTION_LOCK(c); 620 RELEASE_REF( c, c_refcnt, c->c_destroy ); 621 } 622 #endif /* BALANCER_MODULE */ 623 624 c->c_destroy = client_destroy; 625 c->c_unlink = client_unlink; 626 c->c_pdu_cb = handle_one_request; 627 628 /* We only register the write event when we have data pending */ 629 event_add( c->c_read_event, c->c_read_timeout ); 630 631 checked_lock( &clients_mutex ); 632 LDAP_CIRCLEQ_INSERT_TAIL( &clients, c, c_next ); 633 checked_unlock( &clients_mutex ); 634 CONNECTION_UNLOCK(c); 635 636 return c; 637 fail: 638 if ( !IS_ALIVE( c, c_live ) ) { 639 /* 640 * Released while we were unlocked, it's scheduled for destruction 641 * already 642 */ 643 return NULL; 644 } 645 646 if ( c->c_write_event ) { 647 event_free( c->c_write_event ); 648 c->c_write_event = NULL; 649 } 650 if ( c->c_read_event ) { 651 event_free( c->c_read_event ); 652 c->c_read_event = NULL; 653 } 654 655 c->c_state = LLOAD_C_INVALID; 656 c->c_live--; 657 c->c_refcnt--; 658 connection_destroy( c ); 659 return NULL; 660 } 661 662 void 663 client_reset( LloadConnection *c ) 664 { 665 TAvlnode *root; 666 long freed = 0, executing; 667 LloadConnection *linked_upstream = NULL; 668 enum op_restriction restricted = c->c_restricted; 669 670 CONNECTION_ASSERT_LOCKED(c); 671 root = c->c_ops; 672 c->c_ops = NULL; 673 executing = c->c_n_ops_executing; 674 c->c_n_ops_executing = 0; 675 676 if ( !BER_BVISNULL( &c->c_auth ) ) { 677 ch_free( c->c_auth.bv_val ); 678 BER_BVZERO( &c->c_auth ); 679 } 680 if ( !BER_BVISNULL( &c->c_sasl_bind_mech ) ) { 681 ch_free( c->c_sasl_bind_mech.bv_val ); 682 BER_BVZERO( &c->c_sasl_bind_mech ); 683 } 684 685 if ( restricted && restricted < LLOAD_OP_RESTRICTED_ISOLATE ) { 686 if ( c->c_backend ) { 687 assert( c->c_restricted <= LLOAD_OP_RESTRICTED_BACKEND ); 688 assert( c->c_restricted_inflight == 0 ); 689 c->c_backend = NULL; 690 c->c_restricted_at = 0; 691 } else { 692 assert( c->c_restricted == LLOAD_OP_RESTRICTED_UPSTREAM ); 693 assert( c->c_linked_upstream != NULL ); 694 linked_upstream = c->c_linked_upstream; 695 c->c_linked_upstream = NULL; 696 } 697 } 698 CONNECTION_UNLOCK(c); 699 700 if ( root ) { 701 freed = ldap_tavl_free( root, (AVL_FREE)operation_abandon ); 702 Debug( LDAP_DEBUG_TRACE, "client_reset: " 703 "dropped %ld operations\n", 704 freed ); 705 } 706 assert( freed == executing ); 707 708 if ( linked_upstream && restricted == LLOAD_OP_RESTRICTED_UPSTREAM ) { 709 LloadConnection *removed = ldap_tavl_delete( 710 &linked_upstream->c_linked, c, lload_upstream_entry_cmp ); 711 assert( removed == c ); 712 } 713 714 CONNECTION_LOCK(c); 715 CONNECTION_ASSERT_LOCKED(c); 716 } 717 718 void 719 client_unlink( LloadConnection *c ) 720 { 721 enum sc_state state; 722 struct event *read_event, *write_event; 723 724 Debug( LDAP_DEBUG_CONNS, "client_unlink: " 725 "removing client connid=%lu\n", 726 c->c_connid ); 727 728 CONNECTION_ASSERT_LOCKED(c); 729 assert( c->c_state != LLOAD_C_INVALID ); 730 assert( c->c_state != LLOAD_C_DYING ); 731 732 state = c->c_state; 733 c->c_state = LLOAD_C_DYING; 734 735 if ( c->c_restricted == LLOAD_OP_RESTRICTED_ISOLATE ) { 736 /* Allow upstream connection to be severed in client_reset() */ 737 c->c_restricted = LLOAD_OP_RESTRICTED_UPSTREAM; 738 } 739 740 read_event = c->c_read_event; 741 write_event = c->c_write_event; 742 CONNECTION_UNLOCK(c); 743 744 if ( read_event ) { 745 event_del( read_event ); 746 } 747 748 if ( write_event ) { 749 event_del( write_event ); 750 } 751 752 if ( state != LLOAD_C_DYING ) { 753 checked_lock( &clients_mutex ); 754 LDAP_CIRCLEQ_REMOVE( &clients, c, c_next ); 755 checked_unlock( &clients_mutex ); 756 } 757 758 CONNECTION_LOCK(c); 759 client_reset( c ); 760 CONNECTION_ASSERT_LOCKED(c); 761 } 762 763 void 764 client_destroy( LloadConnection *c ) 765 { 766 Debug( LDAP_DEBUG_CONNS, "client_destroy: " 767 "destroying client connid=%lu\n", 768 c->c_connid ); 769 770 CONNECTION_LOCK(c); 771 assert( c->c_state == LLOAD_C_DYING ); 772 773 #ifdef BALANCER_MODULE 774 /* 775 * Can't do this in client_unlink as that could be run from cn=monitor 776 * modify callback. 777 */ 778 if ( !BER_BVISNULL( &c->c_monitor_dn ) ) { 779 lload_monitor_conn_unlink( c ); 780 } 781 #endif /* BALANCER_MODULE */ 782 783 c->c_state = LLOAD_C_INVALID; 784 785 assert( c->c_ops == NULL ); 786 787 if ( c->c_read_event ) { 788 event_free( c->c_read_event ); 789 c->c_read_event = NULL; 790 } 791 792 if ( c->c_write_event ) { 793 event_free( c->c_write_event ); 794 c->c_write_event = NULL; 795 } 796 797 assert( c->c_refcnt == 0 ); 798 connection_destroy( c ); 799 } 800 801 void 802 clients_destroy( int gentle ) 803 { 804 epoch_t epoch = epoch_join(); 805 checked_lock( &clients_mutex ); 806 connections_walk( 807 &clients_mutex, &clients, lload_connection_close, &gentle ); 808 checked_unlock( &clients_mutex ); 809 epoch_leave( epoch ); 810 } 811
Indexes created Thu Aug 27 00:25:27 UTC 2026