Upstream connection setup

servers/lloadd/Makefile.in

@@ -21,8 +21,8 @@ XSRCS	= version.c
 NT_SRCS = nt_svc.c
 NT_OBJS = nt_svc.o ../../libraries/liblutil/slapdmsg.res
 
-SRCS	= main.c globals.c config.c connection.c client.c daemon.c \
-		ch_malloc.c init.c user.c sl_malloc.c value.c \
+SRCS	= main.c globals.c backend.c config.c connection.c client.c daemon.c \
+		ch_malloc.c init.c user.c sl_malloc.c upstream.c value.c \
 		libevent_support.c \
 		$(@PLAT@_SRCS)
 

servers/lloadd/backend.c

@@ -0,0 +1,149 @@
+/* $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>.
+ */
+
+#include "portable.h"
+
+#include <ac/socket.h>
+#include <ac/errno.h>
+#include <ac/string.h>
+#include <ac/time.h>
+#include <ac/unistd.h>
+
+#include <event2/event.h>
+#include <event2/dns.h>
+
+#include "lutil.h"
+#include "slap.h"
+
+static void
+upstream_name_cb( int result, struct evutil_addrinfo *res, void *arg )
+{
+    Backend *b = arg;
+    Connection *c;
+    ber_socket_t s;
+    int rc;
+
+    if ( result || !res ) {
+        Debug( LDAP_DEBUG_ANY, "upstream_name_cb: "
+                "name resolution failed for backend '%s': %s\n",
+                b->b_bindconf.sb_uri.bv_val, evutil_gai_strerror( result ) );
+        return;
+    }
+
+    s = socket( res->ai_family, SOCK_STREAM, 0 );
+    if ( s == AC_SOCKET_INVALID ) {
+        return;
+    }
+
+    rc = ber_pvt_socket_set_nonblock( s, 1 );
+    if ( rc ) {
+        evutil_closesocket( s );
+        return;
+    }
+
+    if ( res->ai_family == PF_INET ) {
+        struct sockaddr_in *ai = (struct sockaddr_in *)res->ai_addr;
+        ai->sin_port = htons( b->b_port );
+        rc = connect( s, (struct sockaddr *)ai, res->ai_addrlen );
+    } else {
+        struct sockaddr_in6 *ai = (struct sockaddr_in6 *)res->ai_addr;
+        ai->sin6_port = htons( b->b_port );
+        rc = connect( s, (struct sockaddr *)ai, res->ai_addrlen );
+    }
+    if ( rc && errno != EINPROGRESS && errno != EWOULDBLOCK ) {
+        Debug( LDAP_DEBUG_ANY, "upstream_name_cb: "
+                "failed to connect to server '%s'\n",
+                b->b_bindconf.sb_uri.bv_val );
+        evutil_closesocket( s );
+        return;
+    }
+
+    c = upstream_init( s, b );
+    ldap_pvt_thread_mutex_lock( &b->b_lock );
+    b->b_conns = c;
+    ldap_pvt_thread_mutex_unlock( &b->b_lock );
+}
+
+Connection *
+backend_select( Operation *op )
+{
+    Backend *b;
+
+    LDAP_STAILQ_FOREACH ( b, &backend, b_next ) {
+        Connection *c;
+
+        ldap_pvt_thread_mutex_lock( &b->b_lock );
+        c = b->b_conns;
+        ldap_pvt_thread_mutex_lock( &c->c_mutex );
+        if ( c->c_struct_state != SLAP_C_UNINITIALIZED && !c->c_pendingber ) {
+            ldap_pvt_thread_mutex_unlock( &b->b_lock );
+            return b->b_conns;
+        }
+        ldap_pvt_thread_mutex_unlock( &c->c_mutex );
+        ldap_pvt_thread_mutex_unlock( &b->b_lock );
+    }
+
+    return NULL;
+}
+
+void *
+backend_connect( Backend *b )
+{
+    struct evutil_addrinfo hints = {};
+
+#ifdef LDAP_PF_LOCAL
+    if ( b->b_proto == LDAP_PROTO_IPC ) {
+        struct sockaddr_un addr;
+        ber_socket_t s = socket( PF_LOCAL, SOCK_STREAM, 0 );
+        int rc;
+
+        if ( s == AC_SOCKET_INVALID ) {
+            return (void *)-1;
+        }
+
+        rc = ber_pvt_socket_set_nonblock( s, 1 );
+        if ( rc ) {
+            evutil_closesocket( s );
+            return (void *)-1;
+        }
+
+        if ( strlen( b->b_host ) > ( sizeof(addr.sun_path) - 1 ) ) {
+            evutil_closesocket( s );
+            return (void *)-1;
+        }
+        memset( &addr, '\0', sizeof(addr) );
+        addr.sun_family = AF_LOCAL;
+        strcpy( addr.sun_path, b->b_host );
+
+        rc = connect(
+                s, (struct sockaddr *)&addr, sizeof(struct sockaddr_un) );
+        if ( rc && errno != EINPROGRESS && errno != EWOULDBLOCK ) {
+            evutil_closesocket( s );
+            return (void *)-1;
+        }
+
+        b->b_conns = upstream_init( s, b );
+        return NULL;
+    }
+#endif /* LDAP_PF_LOCAL */
+
+    hints.ai_family = AF_UNSPEC;
+    hints.ai_flags = EVUTIL_AI_CANONNAME;
+    hints.ai_socktype = SOCK_STREAM;
+    hints.ai_protocol = IPPROTO_TCP;
+
+    evdns_getaddrinfo( dnsbase, b->b_host, NULL, &hints, upstream_name_cb, b );
+    return NULL;
+}

servers/lloadd/client.c

@@ -38,7 +38,7 @@ client_read_cb( evutil_socket_t s, short what, void *arg )
     client_destroy( c );
 }
 
-static void
+void
 client_write_cb( evutil_socket_t s, short what, void *arg )
 {
     Connection *c = arg;

servers/lloadd/connection.c

@@ -59,14 +59,17 @@ connection_destroy( Connection *c )
             c->c_connid );
 
     assert( c->c_struct_state == SLAP_C_UNINITIALIZED );
-    evutil_closesocket( c->c_fd );
+    ber_sockbuf_free( c->c_sb );
+
+    if ( c->c_currentber ) {
+        ber_free( c->c_currentber, 1 );
+    }
 
     ldap_pvt_thread_mutex_unlock( &c->c_mutex );
 
     ldap_pvt_thread_mutex_destroy( &c->c_io_mutex );
     ldap_pvt_thread_mutex_destroy( &c->c_mutex );
 
-    ber_sockbuf_free( c->c_sb );
     ch_free( c );
 }
 

servers/lloadd/proto-slap.h

@@ -57,6 +57,12 @@ LDAP_SLAPD_F (void) ch_free( void * );
 #define free ch_free
 #endif
 
+/*
+ * client.c
+ */
+LDAP_SLAPD_F (Connection *) client_init( ber_socket_t s, Listener *url, const char *peername, struct event_base *base, int use_tls );
+LDAP_SLAPD_F (void) client_write_cb( evutil_socket_t s, short what, void *arg );
+
 /*
  * config.c
  */
@@ -70,8 +76,6 @@ LDAP_SLAPD_F (void) bindconf_free( slap_bindconf *bc );
  * connection.c
  */
 LDAP_SLAPD_F (Connection *) connection_init( ber_socket_t s, const char *peername, int use_tls );
-LDAP_SLAPD_F (Connection *) client_init( ber_socket_t s, Listener *url, const char *peername, struct event_base *base, int use_tls );
-LDAP_SLAPD_F (Connection *) upstream_init( ber_socket_t s, Backend *b );
 LDAP_SLAPD_F (void) connection_destroy( Connection *c );
 
 /*
@@ -147,6 +151,13 @@ LDAP_SLAPD_F (void *) slap_sl_context( void *ptr );
 /* assumes (x) > (y) returns 1 if true, 0 otherwise */
 #define SLAP_PTRCMP(x, y) ( (x) < (y) ? -1 : (x) > (y) )
 
+/*
+ * upstream.c
+ */
+LDAP_SLAPD_F (void) upstream_write_cb( evutil_socket_t s, short what, void *arg );
+LDAP_SLAPD_F (void) upstream_read_cb( evutil_socket_t s, short what, void *arg );
+LDAP_SLAPD_F (Connection *) upstream_init( ber_socket_t s, Backend *b );
+
 /*
  * user.c
  */

servers/lloadd/upstream.c

@@ -0,0 +1,114 @@
+/* $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>.
+ */
+
+#include "portable.h"
+
+#include <ac/socket.h>
+#include <ac/errno.h>
+#include <ac/string.h>
+#include <ac/time.h>
+#include <ac/unistd.h>
+
+#include "lutil.h"
+#include "slap.h"
+
+static void upstream_destroy( Connection *c );
+
+void
+upstream_read_cb( evutil_socket_t s, short what, void *arg )
+{
+    Connection *c = arg;
+
+    ldap_pvt_thread_mutex_lock( &c->c_mutex );
+    upstream_destroy( c );
+}
+
+void
+upstream_write_cb( evutil_socket_t s, short what, void *arg )
+{
+    Connection *c = arg;
+}
+
+Connection *
+upstream_init( ber_socket_t s, Backend *backend )
+{
+    Connection *c;
+    struct event_base *base = slap_get_base( s );
+    struct event *event;
+    int flags = (backend->b_tls == LLOAD_LDAPS) ? CONN_IS_TLS : 0;
+
+    assert( backend != NULL );
+
+    c = connection_init( s, backend->b_host, flags );
+
+    event = event_new( base, s, EV_READ|EV_PERSIST, upstream_read_cb, c );
+    if ( !event ) {
+        Debug( LDAP_DEBUG_ANY, "Read event could not be allocated\n" );
+        goto fail;
+    }
+    event_add( event, NULL );
+    c->c_read_event = event;
+
+    event = event_new( base, s, EV_WRITE, upstream_write_cb, c );
+    if ( !event ) {
+        Debug( LDAP_DEBUG_ANY, "Write event could not be allocated\n" );
+        goto fail;
+    }
+    /* We only register the write event when we have data pending */
+    c->c_write_event = event;
+
+    c->c_private = backend;
+    ldap_pvt_thread_mutex_unlock( &c->c_mutex );
+
+    return c;
+fail:
+    if ( c->c_write_event ) {
+        event_del( c->c_write_event );
+        event_free( c->c_write_event );
+    }
+    if ( c->c_read_event ) {
+        event_del( c->c_read_event );
+        event_free( c->c_read_event );
+    }
+    connection_destroy( c );
+    return NULL;
+}
+
+static void
+upstream_destroy( Connection *c )
+{
+    Backend *b = c->c_private;
+
+    c->c_struct_state = SLAP_C_UNINITIALIZED;
+    ldap_pvt_thread_mutex_unlock( &c->c_mutex );
+
+    ldap_pvt_thread_mutex_lock( &b->b_lock );
+    if ( !(b->b_conns == c) ) {
+        ldap_pvt_thread_mutex_unlock( &b->b_lock );
+        return;
+    }
+    b->b_conns = NULL;
+    ldap_pvt_thread_mutex_unlock( &b->b_lock );
+
+    ldap_pvt_thread_mutex_lock( &c->c_mutex );
+
+    event_del( c->c_read_event );
+    event_free( c->c_read_event );
+
+    event_del( c->c_write_event );
+    event_free( c->c_write_event );
+
+    connection_destroy( c );
+}