connection.c 16 KB
Newer Older
Ondřej Kuzník's avatar
Ondřej Kuzník committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
/* $OpenLDAP$ */
/* This work is part of OpenLDAP Software <http://www.openldap.org/>.
 *
 * Copyright 1998-2020 The OpenLDAP Foundation.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted only as authorized by the OpenLDAP
 * Public License.
 *
 * A copy of this license is available in the file LICENSE in the
 * top-level directory of the distribution or, alternatively, at
 * <http://www.OpenLDAP.org/license.html>.
 */
/* Portions Copyright (c) 1995 Regents of the University of Michigan.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms are permitted
 * provided that this notice is preserved and that due credit is given
 * to the University of Michigan at Ann Arbor. The name of the University
 * may not be used to endorse or promote products derived from this
 * software without specific prior written permission. This software
 * is provided ``as is'' without express or implied warranty.
 */

#include "portable.h"

#include <stdio.h>
#ifdef HAVE_LIMITS_H
#include <limits.h>
#endif

#include <ac/socket.h>
#include <ac/errno.h>
#include <ac/string.h>
#include <ac/time.h>
#include <ac/unistd.h>

39
#include "lload.h"
Ondřej Kuzník's avatar
Ondřej Kuzník committed
40

41
42
43
#include "lutil.h"
#include "lutil_ldap.h"

Ondřej Kuzník's avatar
Ondřej Kuzník committed
44
45
46
static unsigned long conn_nextid = 0;

static void
47
lload_connection_assign_nextid( LloadConnection *conn )
Ondřej Kuzník's avatar
Ondřej Kuzník committed
48
{
49
    conn->c_connid = __atomic_fetch_add( &conn_nextid, 1, __ATOMIC_RELAXED );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
50
51
}

52
53
54
55
56
/*
 * We start off with the connection muted and c_currentber holding the pdu we
 * received.
 *
 * We run c->c_pdu_cb for each pdu, stopping once we hit an error, have to wait
57
 * on reading or after we process lload_conn_max_pdus_per_cycle pdus so as to
58
59
60
61
62
63
64
65
66
67
 * maintain fairness and not hog the worker thread forever.
 *
 * If we've run out of pdus immediately available from the stream or hit the
 * budget, we unmute the connection.
 *
 * c->c_pdu_cb might return an 'error' and not free the connection. That can
 * happen when changing the state or when client is blocked on writing and
 * already has a pdu pending on the same operation, it's their job to make sure
 * we're woken up again.
 */
68
void *
69
70
handle_pdus( void *ctx, void *arg )
{
71
    LloadConnection *c = arg;
72
    int pdus_handled = 0;
73
    epoch_t epoch;
74

75
76
77
78
    /* A reference was passed on to us */
    assert( IS_ALIVE( c, c_refcnt ) );

    epoch = epoch_join();
79
80
81
82
83
84
    for ( ;; ) {
        BerElement *ber;
        ber_tag_t tag;
        ber_len_t len;

        if ( c->c_pdu_cb( c ) ) {
85
86
            /* Error/reset, get rid ouf our reference and bail */
            goto done;
87
88
        }

89
        if ( ++pdus_handled >= lload_conn_max_pdus_per_cycle ) {
90
91
92
93
            /* Do not read now, re-enable read event instead */
            break;
        }

94
95
        ber = c->c_currentber;
        if ( ber == NULL && (ber = ber_alloc()) == NULL ) {
96
97
98
            Debug( LDAP_DEBUG_ANY, "handle_pdus: "
                    "connid=%lu, ber_alloc failed\n",
                    c->c_connid );
99
100
            CONNECTION_LOCK_DESTROY(c);
            goto done;
101
102
103
        }
        c->c_currentber = ber;

104
        ldap_pvt_thread_mutex_lock( &c->c_io_mutex );
105
        tag = ber_get_next( c->c_sb, &len, ber );
106
        ldap_pvt_thread_mutex_unlock( &c->c_io_mutex );
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
        if ( tag != LDAP_TAG_MESSAGE ) {
            int err = sock_errno();

            if ( err != EWOULDBLOCK && err != EAGAIN ) {
                if ( err || tag == LBER_ERROR ) {
                    char ebuf[128];
                    Debug( LDAP_DEBUG_ANY, "handle_pdus: "
                            "ber_get_next on fd=%d failed errno=%d (%s)\n",
                            c->c_fd, err,
                            sock_errstr( err, ebuf, sizeof(ebuf) ) );
                } else {
                    Debug( LDAP_DEBUG_STATS, "handle_pdus: "
                            "ber_get_next on fd=%d connid=%lu received "
                            "a strange PDU tag=%lx\n",
                            c->c_fd, c->c_connid, tag );
                }

                c->c_currentber = NULL;
                ber_free( ber, 1 );
126
127
                CONNECTION_LOCK_DESTROY(c);
                goto done;
128
129
130
131
132
            }
            break;
        }
    }

133
    event_add( c->c_read_event, c->c_read_timeout );
134
135
136
    Debug( LDAP_DEBUG_CONNS, "handle_pdus: "
            "re-enabled read event on connid=%lu\n",
            c->c_connid );
137
138
139
done:
    RELEASE_REF( c, c_refcnt, c->c_destroy );
    epoch_leave( epoch );
140
141
142
143
144
145
146
147
148
149
150
151
152
    return NULL;
}

/*
 * Initial read on the connection, if we get an LDAP PDU, submit the
 * processing of this and successive ones to the work queue.
 *
 * If we can't submit it to the queue (overload), process this one and return
 * to the event loop immediately after.
 */
void
connection_read_cb( evutil_socket_t s, short what, void *arg )
{
153
    LloadConnection *c = arg;
154
155
156
    BerElement *ber;
    ber_tag_t tag;
    ber_len_t len;
157
    epoch_t epoch;
158
159
160
161

    CONNECTION_LOCK(c);
    if ( !c->c_live ) {
        event_del( c->c_read_event );
162
        CONNECTION_UNLOCK(c);
163
164
165
166
167
        Debug( LDAP_DEBUG_CONNS, "connection_read_cb: "
                "suspended read event on a dead connid=%lu\n",
                c->c_connid );
        return;
    }
168
    CONNECTION_UNLOCK(c);
169

170
171
172
173
    if ( what & EV_TIMEOUT ) {
        Debug( LDAP_DEBUG_CONNS, "connection_read_cb: "
                "connid=%lu, timeout reached, destroying\n",
                c->c_connid );
174
175
176
177
        /* Make sure the connection stays around for us to unlock it */
        epoch = epoch_join();
        CONNECTION_LOCK_DESTROY(c);
        epoch_leave( epoch );
178
179
180
        return;
    }

181
182
183
184
185
    if ( !acquire_ref( &c->c_refcnt ) ) {
        return;
    }
    epoch = epoch_join();

186
187
188
189
190
191
192
193
194
    Debug( LDAP_DEBUG_CONNS, "connection_read_cb: "
            "connection connid=%lu ready to read\n",
            c->c_connid );

    ber = c->c_currentber;
    if ( ber == NULL && (ber = ber_alloc()) == NULL ) {
        Debug( LDAP_DEBUG_ANY, "connection_read_cb: "
                "connid=%lu, ber_alloc failed\n",
                c->c_connid );
195
        goto out;
196
197
198
    }
    c->c_currentber = ber;

199
    ldap_pvt_thread_mutex_lock( &c->c_io_mutex );
200
    tag = ber_get_next( c->c_sb, &len, ber );
201
202
    ldap_pvt_thread_mutex_unlock( &c->c_io_mutex );

203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
    if ( tag != LDAP_TAG_MESSAGE ) {
        int err = sock_errno();

        if ( err != EWOULDBLOCK && err != EAGAIN ) {
            if ( err || tag == LBER_ERROR ) {
                char ebuf[128];
                Debug( LDAP_DEBUG_STATS, "connection_read_cb: "
                        "ber_get_next on fd=%d failed errno=%d (%s)\n",
                        c->c_fd, err,
                        sock_errstr( err, ebuf, sizeof(ebuf) ) );
            } else {
                Debug( LDAP_DEBUG_STATS, "connection_read_cb: "
                        "ber_get_next on fd=%d connid=%lu received "
                        "a strange PDU tag=%lx\n",
                        c->c_fd, c->c_connid, tag );
            }

            c->c_currentber = NULL;
            ber_free( ber, 1 );

            event_del( c->c_read_event );
            Debug( LDAP_DEBUG_CONNS, "connection_read_cb: "
                    "suspended read event on dying connid=%lu\n",
                    c->c_connid );
227
228
            CONNECTION_LOCK_DESTROY(c);
            goto out;
229
        }
230
        event_add( c->c_read_event, c->c_read_timeout );
231
232
233
        Debug( LDAP_DEBUG_CONNS, "connection_read_cb: "
                "re-enabled read event on connid=%lu\n",
                c->c_connid );
234
        goto out;
235
236
    }

237
    event_del( c->c_read_event );
238
    if ( !lload_conn_max_pdus_per_cycle ||
239
240
            ldap_pvt_thread_pool_submit( &connection_pool, handle_pdus, c ) ) {
        /* If we're overloaded or configured as such, process one and resume in
241
242
243
244
         * the next cycle. */
        event_add( c->c_read_event, c->c_read_timeout );
        c->c_pdu_cb( c );
        goto out;
245
246
247
248
249
250
    }

    Debug( LDAP_DEBUG_CONNS, "connection_read_cb: "
            "suspended read event on connid=%lu\n",
            c->c_connid );

251
252
253
254
255
    /*
     * We have scheduled a call to handle_pdus to take care of handling this
     * and further requests, its reference is now owned by that task.
     */
    epoch_leave( epoch );
256
    return;
257
258
259
260

out:
    RELEASE_REF( c, c_refcnt, c->c_destroy );
    epoch_leave( epoch );
261
262
263
264
265
}

void
connection_write_cb( evutil_socket_t s, short what, void *arg )
{
266
    LloadConnection *c = arg;
267
    epoch_t epoch;
268
269

    CONNECTION_LOCK(c);
270
271
272
    Debug( LDAP_DEBUG_CONNS, "connection_write_cb: "
            "considering writing to%s connid=%lu what=%hd\n",
            c->c_live ? " live" : " dead", c->c_connid, what );
273
274
275
276
    if ( !c->c_live ) {
        CONNECTION_UNLOCK(c);
        return;
    }
277
    CONNECTION_UNLOCK(c);
278
279
280
281
282

    if ( what & EV_TIMEOUT ) {
        Debug( LDAP_DEBUG_CONNS, "connection_write_cb: "
                "connid=%lu, timeout reached, destroying\n",
                c->c_connid );
283
284
285
286
        /* Make sure the connection stays around for us to unlock it */
        epoch = epoch_join();
        CONNECTION_LOCK_DESTROY(c);
        epoch_leave( epoch );
287
288
        return;
    }
289
290
291
292

    /* Before we acquire any locks */
    event_del( c->c_write_event );

293
294
295
296
297
298
    if ( !acquire_ref( &c->c_refcnt ) ) {
        return;
    }

    epoch = epoch_join();

299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
    ldap_pvt_thread_mutex_lock( &c->c_io_mutex );
    Debug( LDAP_DEBUG_CONNS, "connection_write_cb: "
            "have something to write to connection connid=%lu\n",
            c->c_connid );

    /* We might have been beaten to flushing the data by another thread */
    if ( c->c_pendingber && ber_flush( c->c_sb, c->c_pendingber, 1 ) ) {
        int err = sock_errno();

        if ( err != EWOULDBLOCK && err != EAGAIN ) {
            char ebuf[128];
            ldap_pvt_thread_mutex_unlock( &c->c_io_mutex );
            Debug( LDAP_DEBUG_ANY, "connection_write_cb: "
                    "ber_flush on fd=%d failed errno=%d (%s)\n",
                    c->c_fd, err, sock_errstr( err, ebuf, sizeof(ebuf) ) );
            CONNECTION_LOCK_DESTROY(c);
315
            goto done;
316
        }
317
        event_add( c->c_write_event, lload_write_timeout );
318
319
320
321
322
    } else {
        c->c_pendingber = NULL;
    }
    ldap_pvt_thread_mutex_unlock( &c->c_io_mutex );

323
324
325
done:
    RELEASE_REF( c, c_refcnt, c->c_destroy );
    epoch_leave( epoch );
326
327
}

Ondřej Kuzník's avatar
Ondřej Kuzník committed
328
void
329
connection_destroy( LloadConnection *c )
Ondřej Kuzník's avatar
Ondřej Kuzník committed
330
331
{
    assert( c );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
332
    Debug( LDAP_DEBUG_CONNS, "connection_destroy: "
Ondřej Kuzník's avatar
Ondřej Kuzník committed
333
            "destroying connection connid=%lu\n",
Ondřej Kuzník's avatar
Ondřej Kuzník committed
334
335
            c->c_connid );

336
337
    assert( c->c_live == 0 );
    assert( c->c_refcnt == 0 );
338
    assert( c->c_state == LLOAD_C_INVALID );
339

Ondřej Kuzník's avatar
Ondřej Kuzník committed
340
341
342
343
    ber_sockbuf_free( c->c_sb );

    if ( c->c_currentber ) {
        ber_free( c->c_currentber, 1 );
344
        c->c_currentber = NULL;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
345
    }
Ondřej Kuzník's avatar
Ondřej Kuzník committed
346
347
    if ( c->c_pendingber ) {
        ber_free( c->c_pendingber, 1 );
348
        c->c_pendingber = NULL;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
349
    }
Ondřej Kuzník's avatar
Ondřej Kuzník committed
350

351
352
353
354
    if ( !BER_BVISNULL( &c->c_sasl_bind_mech ) ) {
        ber_memfree( c->c_sasl_bind_mech.bv_val );
        BER_BVZERO( &c->c_sasl_bind_mech );
    }
355
356
357
358
359
360
361
362
363
#ifdef HAVE_CYRUS_SASL
    if ( c->c_sasl_defaults ) {
        lutil_sasl_freedefs( c->c_sasl_defaults );
        c->c_sasl_defaults = NULL;
    }
    if ( c->c_sasl_authctx ) {
        sasl_dispose( &c->c_sasl_authctx );
    }
#endif /* HAVE_CYRUS_SASL */
364

365
    CONNECTION_UNLOCK(c);
Ondřej Kuzník's avatar
Ondřej Kuzník committed
366
367
368
369
370

    ldap_pvt_thread_mutex_destroy( &c->c_io_mutex );
    ldap_pvt_thread_mutex_destroy( &c->c_mutex );

    ch_free( c );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
371
372

    listeners_reactivate();
Ondřej Kuzník's avatar
Ondřej Kuzník committed
373
374
}

375
/*
Ondřej Kuzník's avatar
Ondřej Kuzník committed
376
377
 * Called holding mutex, will walk cq calling cb on all connections whose
 * c_connid <= cq_last->c_connid that still exist at the time we get to them.
378
379
 */
void
Ondřej Kuzník's avatar
Ondřej Kuzník committed
380
381
382
383
384
385
connections_walk_last(
        ldap_pvt_thread_mutex_t *cq_mutex,
        lload_c_head *cq,
        LloadConnection *cq_last,
        CONNCB cb,
        void *arg )
386
{
387
388
    LloadConnection *c = cq_last;
    uintptr_t last_connid;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
389
390
391
392

    if ( LDAP_CIRCLEQ_EMPTY( cq ) ) {
        return;
    }
393
394
    last_connid = c->c_connid;
    c = LDAP_CIRCLEQ_LOOP_NEXT( cq, c, c_next );
395

396
397
398
399
400
401
    while ( !acquire_ref( &c->c_refcnt ) ) {
        c = LDAP_CIRCLEQ_LOOP_NEXT( cq, c, c_next );
        if ( c->c_connid >= last_connid ) {
            return;
        }
    }
Ondřej Kuzník's avatar
Ondřej Kuzník committed
402
403

    /*
404
     * Notes:
Ondřej Kuzník's avatar
Ondřej Kuzník committed
405
406
     * - we maintain the connections in the cq CIRCLEQ_ in ascending c_connid
     *   order
407
     * - the connection with the highest c_connid is passed in cq_last
Ondřej Kuzník's avatar
Ondřej Kuzník committed
408
409
410
411
412
413
     * - we can only use cq when we hold cq_mutex
     * - connections might be added to or removed from cq while we're busy
     *   processing connections
     * - we need a way to detect we've finished looping around cq for some
     *   definition of looping around
     */
414
415
    do {
        int rc;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
416

417
        ldap_pvt_thread_mutex_unlock( cq_mutex );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
418

419
420
        rc = cb( c, arg );
        RELEASE_REF( c, c_refcnt, c->c_destroy );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
421
422

        ldap_pvt_thread_mutex_lock( cq_mutex );
423
424
        if ( rc || LDAP_CIRCLEQ_EMPTY( cq ) ) {
            break;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
425
426
        }

427
428
429
430
431
432
433
434
        do {
            LloadConnection *old = c;
            c = LDAP_CIRCLEQ_LOOP_NEXT( cq, c, c_next );
            if ( c->c_connid <= old->c_connid || c->c_connid > last_connid ) {
                return;
            }
        } while ( !acquire_ref( &c->c_refcnt ) );
    } while ( c->c_connid <= last_connid );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
}

void
connections_walk(
        ldap_pvt_thread_mutex_t *cq_mutex,
        lload_c_head *cq,
        CONNCB cb,
        void *arg )
{
    LloadConnection *cq_last = LDAP_CIRCLEQ_LAST( cq );
    return connections_walk_last( cq_mutex, cq, cq_last, cb, arg );
}

int
lload_connection_close( LloadConnection *c, void *arg )
{
    int gentle = *(int *)arg;
452
    LloadOperation *op;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
453

454
455
456
457
458
459
    Debug( LDAP_DEBUG_CONNS, "lload_connection_close: "
            "marking connection connid=%lu closing\n",
            c->c_connid );

    /* We were approached from the connection list */
    assert( IS_ALIVE( c, c_refcnt ) );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
460

461
462
    CONNECTION_LOCK(c);
    if ( !gentle || !c->c_ops ) {
Ondřej Kuzník's avatar
Ondřej Kuzník committed
463
464
465
        CONNECTION_DESTROY(c);
        return LDAP_SUCCESS;
    }
466
467
468
469

    /* The first thing we do is make sure we don't get new Operations in */
    c->c_state = LLOAD_C_CLOSING;

470
471
472
    do {
        TAvlnode *node = tavl_end( c->c_ops, TAVL_DIR_LEFT );
        op = node->avl_data;
473

474
475
476
477
        /* Close operations that would need client action to resolve,
         * only SASL binds in progress do that right now */
        if ( op->o_client_msgid || op->o_upstream_msgid ) {
            break;
478
        }
479
480
481
482
483
484
485

        CONNECTION_UNLOCK(c);
        operation_unlink( op );
        CONNECTION_LOCK(c);
    } while ( c->c_ops );

    CONNECTION_UNLOCK(c);
Ondřej Kuzník's avatar
Ondřej Kuzník committed
486
    return LDAP_SUCCESS;
487
488
}

489
490
LloadConnection *
lload_connection_init( ber_socket_t s, const char *peername, int flags )
Ondřej Kuzník's avatar
Ondřej Kuzník committed
491
{
492
    LloadConnection *c;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
493
494
495
496

    assert( peername != NULL );

    if ( s == AC_SOCKET_INVALID ) {
497
        Debug( LDAP_DEBUG_ANY, "lload_connection_init: "
Ondřej Kuzník's avatar
Ondřej Kuzník committed
498
                "init of socket fd=%ld invalid\n",
Ondřej Kuzník's avatar
Ondřej Kuzník committed
499
500
501
502
503
504
                (long)s );
        return NULL;
    }

    assert( s >= 0 );

505
    c = ch_calloc( 1, sizeof(LloadConnection) );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
506

Ondřej Kuzník's avatar
Ondřej Kuzník committed
507
    c->c_fd = s;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
    c->c_sb = ber_sockbuf_alloc();
    ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_SET_FD, &s );

#ifdef LDAP_PF_LOCAL
    if ( flags & CONN_IS_IPC ) {
#ifdef LDAP_DEBUG
        ber_sockbuf_add_io( c->c_sb, &ber_sockbuf_io_debug,
                LBER_SBIOD_LEVEL_PROVIDER, (void *)"ipc_" );
#endif
        ber_sockbuf_add_io( c->c_sb, &ber_sockbuf_io_fd,
                LBER_SBIOD_LEVEL_PROVIDER, (void *)&s );
    } else
#endif /* LDAP_PF_LOCAL */
    {
#ifdef LDAP_DEBUG
        ber_sockbuf_add_io( c->c_sb, &ber_sockbuf_io_debug,
                LBER_SBIOD_LEVEL_PROVIDER, (void *)"tcp_" );
#endif
        ber_sockbuf_add_io( c->c_sb, &ber_sockbuf_io_tcp,
                LBER_SBIOD_LEVEL_PROVIDER, (void *)&s );
    }

#ifdef LDAP_DEBUG
    ber_sockbuf_add_io(
            c->c_sb, &ber_sockbuf_io_debug, INT_MAX, (void *)"lload_" );
#endif

Ondřej Kuzník's avatar
Ondřej Kuzník committed
535
    c->c_next_msgid = 1;
536
    c->c_refcnt = c->c_live = 1;
537
    c->c_destroy = connection_destroy;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
538

539
    LDAP_CIRCLEQ_ENTRY_INIT( c, c_next );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
540

Ondřej Kuzník's avatar
Ondřej Kuzník committed
541
542
543
    ldap_pvt_thread_mutex_init( &c->c_mutex );
    ldap_pvt_thread_mutex_init( &c->c_io_mutex );

544
    lload_connection_assign_nextid( c );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
545

546
    Debug( LDAP_DEBUG_CONNS, "lload_connection_init: "
Ondřej Kuzník's avatar
Ondřej Kuzník committed
547
548
            "connection connid=%lu allocated for socket fd=%d peername=%s\n",
            c->c_connid, s, peername );
Ondřej Kuzník's avatar
Ondřej Kuzník committed
549

550
    c->c_state = LLOAD_C_ACTIVE;
Ondřej Kuzník's avatar
Ondřej Kuzník committed
551

Ondřej Kuzník's avatar
Ondřej Kuzník committed
552
553
    return c;
}