Implement client pending operation limits

doc/man/man5/lloadd.conf.5

@@ -329,6 +329,10 @@ read or this limit is reached when the I/O thread can pick it up again.
 Very high values have a potential to cause some connections to be
 starved in a very high-bandwidth environment. The default is 1000.
 .TP
+.B client_max_pending <integer>
+Will cause the load balancer to limit the number unfinished operations for each
+client connection. The default is 0, unlimited.
+.TP
 .B iotimeout <integer>
 Specify the number of milliseconds to wait before forcibly closing
 a connection with an outstanding write. This allows faster recovery from

servers/lloadd/client.c

@@ -24,6 +24,8 @@
 #include "lutil.h"
 #include "lload.h"
 
+long lload_client_max_pending = 0;
+
 lload_c_head clients = LDAP_CIRCLEQ_HEAD_INITIALIZER( clients );
 
 ldap_pvt_thread_mutex_t clients_mutex;
@@ -188,6 +190,7 @@ handle_one_request( LloadConnection *c )
     BerElement *ber;
     LloadOperation *op = NULL;
     RequestHandler handler = NULL;
+    int over_limit = 0;
 
     ber = c->c_currentber;
     c->c_currentber = NULL;
@@ -202,6 +205,10 @@ handle_one_request( LloadConnection *c )
         ber_free( ber, 1 );
         return -1;
     }
+    if ( lload_client_max_pending &&
+            c->c_n_ops_executing >= lload_client_max_pending ) {
+        over_limit = 1;
+    }
     CONNECTION_UNLOCK(c);
 
     switch ( op->o_tag ) {
@@ -223,15 +230,23 @@ handle_one_request( LloadConnection *c )
              * currently in progress */
             return request_abandon( c, op );
         case LDAP_REQ_EXTENDED:
-            handler = request_extended;
-            break;
         default:
             if ( c->c_state == LLOAD_C_BINDING ) {
                 operation_send_reject(
                         op, LDAP_PROTOCOL_ERROR, "bind in progress", 0 );
                 return LDAP_SUCCESS;
             }
-            handler = request_process;
+            if ( over_limit ) {
+                operation_send_reject( op, LDAP_BUSY,
+                        "pending operation limit reached on this connection",
+                        0 );
+                return LDAP_SUCCESS;
+            }
+            if ( op->o_tag == LDAP_REQ_EXTENDED ) {
+                handler = request_extended;
+            } else {
+                handler = request_process;
+            }
             break;
     }
 

servers/lloadd/config.c

@@ -177,6 +177,7 @@ enum {
     CFG_MAX_PENDING_OPS,
     CFG_MAX_PENDING_CONNS,
     CFG_STARTTLS,
+    CFG_CLIENT_PENDING,
 
     CFG_LAST
 };
@@ -616,6 +617,17 @@ static ConfigTable config_back_cf_table[] = {
             "SINGLE-VALUE )",
         NULL, NULL
     },
+    { "client_max_pending", NULL, 2, 2, 0,
+        ARG_MAGIC|ARG_UINT|CFG_CLIENT_PENDING,
+        &config_generic,
+        "( OLcfgBkAt:13.35 "
+            "NAME 'olcBkLloadClientMaxPending' "
+            "DESC 'Maximum pending operations per client connection' "
+            "EQUALITY integerMatch "
+            "SYNTAX OMsInteger "
+            "SINGLE-VALUE )",
+        NULL, NULL
+    },
 
     /* cn=config only options */
 #ifdef BALANCER_MODULE
@@ -803,6 +815,9 @@ config_generic( ConfigArgs *c )
                 c->value_uint = 1000 * lload_write_timeout->tv_sec +
                         lload_write_timeout->tv_usec / 1000;
                 break;
+            case CFG_CLIENT_PENDING:
+                c->value_uint = lload_client_max_pending;
+                break;
             default:
                 rc = 1;
                 break;
@@ -953,6 +968,9 @@ config_generic( ConfigArgs *c )
         case CFG_MAXBUF_UPSTREAM:
             sockbuf_max_incoming_upstream = c->value_uint;
             break;
+        case CFG_CLIENT_PENDING:
+            lload_client_max_pending = c->value_uint;
+            break;
         default:
             Debug( LDAP_DEBUG_ANY, "%s: unknown CFG_TYPE %d\n",
                     c->log, c->type );

servers/lloadd/proto-lload.h

@@ -64,6 +64,7 @@ LDAP_SLAPD_F (LloadConnection *) client_init( ber_socket_t s, LloadListener *url
 LDAP_SLAPD_F (void) client_reset( LloadConnection *c );
 LDAP_SLAPD_F (void) client_destroy( LloadConnection *c );
 LDAP_SLAPD_F (void) clients_destroy( int gentle );
+LDAP_SLAPD_V (long) lload_client_max_pending;
 
 /*
  * config.c