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