Ticket #47319 - make connection buffer size adjustable

https://fedorahosted.org/389/ticket/47319
Reviewed by: mreynolds (Thanks!)
Branch: master
Fix Description: Added a new config attribute nsslapd-connection-buffer with
the following values:
0 - no buffering at all - just read a single PDU at a time
1 - default - regular fixed size LDAP_SOCKET_IO_BUFFER_SIZE buffer
2 - adaptable buffer - if the system recv() call returns exactly the
amount of data we requested, assume we could have read more, and increase
the size of the buffer up to BUFSIZ, to try to minimize the number of system
recv calls
2 should be better than 1 in the case where the client is sending a large
amount of data at once - e.g. large add/modify, or cases where many
async requests are coming in over a single connection (e.g. replication)
Platforms tested: RHEL6 x86_64
Flag Day: no
Doc impact: yes - document new config parameter

ldap/servers/slapd/connection.c

@@ -1601,6 +1601,7 @@ struct Conn_private
 	char *c_buffer;			/* pointer to the socket read buffer */
 	size_t c_buffer_bytes; /* number of bytes currently stored in the buffer */
 	size_t c_buffer_offset; /* offset to the location of new data in the buffer */
+	int use_buffer; /* if true, use the buffer - if false, ber_get_next reads directly from socket */
 };
 
 #if defined(USE_OPENLDAP)
@@ -1622,6 +1623,11 @@ openldap_read_function(Sockbuf_IO_Desc *sbiod, void *buf, ber_len_t len)
 
 	conn = (Connection *)sbiod->sbiod_pvt;
 
+	if (CONNECTION_BUFFER_OFF == conn->c_private->use_buffer) {
+		bytes_to_copy = PR_Recv(conn->c_prfd,buf,len,0,PR_INTERVAL_NO_WAIT);
+		goto done;
+	}
+
 	PR_ASSERT(conn->c_private->c_buffer);
 
 	readbuf = conn->c_private->c_buffer;
@@ -1649,6 +1655,7 @@ openldap_read_function(Sockbuf_IO_Desc *sbiod, void *buf, ber_len_t len)
 		SAFEMEMCPY(buf, readbuf + offset, bytes_to_copy);
 		conn->c_private->c_buffer_offset += bytes_to_copy;
 	}
+done:
 	return bytes_to_copy;
 }
 #endif
@@ -1663,11 +1670,12 @@ connection_new_private(Connection *conn)
 			return -1;
 		}
 		conn->c_private = new_private;
+		conn->c_private->use_buffer = config_get_connection_buffer();
 	}
 
 	/* The c_buffer is supposed to be NULL here, cleaned by connection_cleanup, 
 	   double check to avoid memory leak */
-	if (NULL == conn->c_private->c_buffer) {
+	if ((CONNECTION_BUFFER_OFF != conn->c_private->use_buffer) && (NULL == conn->c_private->c_buffer)) {
 		conn->c_private->c_buffer = (char*)slapi_ch_malloc(LDAP_SOCKET_IO_BUFFER_SIZE);
 		if (NULL == conn->c_private->c_buffer) {
 			/* memory allocation failure */
@@ -1682,11 +1690,13 @@ connection_new_private(Connection *conn)
 	 */
 	{
 		char	*c_buffer = conn->c_private->c_buffer;
-		size_t	c_buffer_size = conn->c_private->c_buffer_size;;
+		size_t	c_buffer_size = conn->c_private->c_buffer_size;
+		int use_buffer = conn->c_private->use_buffer;
 
 		memset( conn->c_private, 0, sizeof(Conn_private));
 		conn->c_private->c_buffer = c_buffer;
 		conn->c_private->c_buffer_size = c_buffer_size;
+		conn->c_private->use_buffer = use_buffer;
 	}
 
 	return 0;
@@ -1864,15 +1874,21 @@ get_next_from_buffer( void *buffer, size_t buffer_size, ber_len_t *lenp,
 		/* drop connection */
 		disconnect_server( conn, conn->c_connid, -1, err, syserr );
 		return -1;
-	}
-
-	/* openldap_read_function will advance c_buffer_offset */
+	} else if (CONNECTION_BUFFER_OFF == conn->c_private->use_buffer) {
+		*lenp = bytes_scanned;
+		if ((LBER_OVERFLOW == *tagp || LBER_DEFAULT == *tagp) && 0 == bytes_scanned &&
+		    SLAPD_SYSTEM_WOULD_BLOCK_ERROR(errno)) {
+			return -2; /* tells connection_read_operation we need to try again */
+		}
+	} else {
+		/* openldap_read_function will advance c_buffer_offset */
 #if !defined(USE_OPENLDAP)
-	/* success, or need to wait for more data */
-	/* if openldap could not read a whole pdu, bytes_scanned will be zero -
-	   it does not return partial results */
-	conn->c_private->c_buffer_offset += bytes_scanned;
+		/* success, or need to wait for more data */
+		/* if openldap could not read a whole pdu, bytes_scanned will be zero -
+	   	   it does not return partial results */
+		conn->c_private->c_buffer_offset += bytes_scanned;
 #endif
+	}
 	return 0;
 }
 
@@ -1881,10 +1897,20 @@ static int
 connection_read_ldap_data(Connection *conn, PRInt32 *err)
 {
 	int ret = 0;
-    ret = PR_Recv(conn->c_prfd,conn->c_private->c_buffer,conn->c_private->c_buffer_size,0,PR_INTERVAL_NO_WAIT);
-    if (ret < 0) {
-        *err = PR_GetError();
-    }
+	ret = PR_Recv(conn->c_prfd,conn->c_private->c_buffer,conn->c_private->c_buffer_size,0,PR_INTERVAL_NO_WAIT);
+	if (ret < 0) {
+		*err = PR_GetError();
+	} else if (CONNECTION_BUFFER_ADAPT == conn->c_private->use_buffer) {
+		if ((ret == conn->c_private->c_buffer_size) && (conn->c_private->c_buffer_size < BUFSIZ)) {
+			/* we read exactly what we requested - there could be more that we could have read */
+			/* so increase the buffer size */
+			conn->c_private->c_buffer_size *= 2;
+			if (conn->c_private->c_buffer_size > BUFSIZ) {
+				conn->c_private->c_buffer_size = BUFSIZ;
+			}
+			conn->c_private->c_buffer = slapi_ch_realloc(conn->c_private->c_buffer, conn->c_private->c_buffer_size);
+		}
+	}
 	return ret;
 }
 
@@ -1937,7 +1963,17 @@ int connection_read_operation(Connection *conn, Operation *op, ber_tag_t *tag, i
 		/* We should never get here with data remaining in the buffer */
 		PR_ASSERT( !new_operation || 0 == (conn->c_private->c_buffer_bytes - conn->c_private->c_buffer_offset) );
 		/* We make a non-blocking read call */
-		ret = connection_read_ldap_data(conn,&err);
+		if (CONNECTION_BUFFER_OFF != conn->c_private->use_buffer) {
+			ret = connection_read_ldap_data(conn,&err);
+		} else {
+			ret = get_next_from_buffer( NULL, 0, &len, tag, op->o_ber, conn );
+			if (ret == -1) {
+				return CONN_DONE; /* get_next_from_buffer does the disconnect stuff */
+			} else if (ret == 0) {
+				ret = len;
+			}
+			*remaining_data = 0;
+		}
 		if (ret <= 0) {
 			if (0 == ret) {
 				/* Connection is closed */
@@ -2009,18 +2045,20 @@ int connection_read_operation(Connection *conn, Operation *op, ber_tag_t *tag, i
 			}
 		} else {
 			/* We read some data off the network, do something with it */
-			conn->c_private->c_buffer_bytes = ret;
-			conn->c_private->c_buffer_offset = 0;
-
-			if ( get_next_from_buffer( buffer,
-						conn->c_private->c_buffer_bytes
-						- conn->c_private->c_buffer_offset,
-						&len, tag, op->o_ber, conn ) != 0 ) {
-				return CONN_DONE;
+			if (CONNECTION_BUFFER_OFF != conn->c_private->use_buffer) {
+				conn->c_private->c_buffer_bytes = ret;
+				conn->c_private->c_buffer_offset = 0;
+
+				if ( get_next_from_buffer( buffer,
+				                           conn->c_private->c_buffer_bytes
+				                           - conn->c_private->c_buffer_offset,
+				                           &len, tag, op->o_ber, conn ) != 0 ) {
+					return CONN_DONE;
+				}
 			}
 
 			new_operation = 0;
-			ret = 0;
+			ret = CONN_FOUND_WORK_TO_DO;
 			waits_done = 0;	/* got some data: reset counter */
 		}
 	}

ldap/servers/slapd/libglobs.c

@@ -257,6 +257,7 @@ slapi_onoff_t init_ndn_cache_enabled;
 slapi_onoff_t init_sasl_mapping_fallback;
 slapi_onoff_t init_return_orig_type;
 slapi_onoff_t init_enable_turbo_mode;
+slapi_int_t init_connection_buffer;
 #ifdef MEMPOOL_EXPERIMENTAL
 slapi_onoff_t init_mempool_switch;
 #endif
@@ -1048,7 +1049,11 @@ static struct config_get_and_set {
 	{CONFIG_ENABLE_TURBO_MODE, config_set_enable_turbo_mode,
 	        NULL, 0,
 	        (void**)&global_slapdFrontendConfig.enable_turbo_mode,
-	        CONFIG_ON_OFF, (ConfigGetFunc)config_get_enable_turbo_mode, &init_enable_turbo_mode}
+	        CONFIG_ON_OFF, (ConfigGetFunc)config_get_enable_turbo_mode, &init_enable_turbo_mode},
+	{CONFIG_CONNECTION_BUFFER, config_set_connection_buffer,
+	        NULL, 0,
+	        (void**)&global_slapdFrontendConfig.connection_buffer,
+	        CONFIG_INT, (ConfigGetFunc)config_get_connection_buffer, &init_connection_buffer}
 #ifdef MEMPOOL_EXPERIMENTAL
 	,{CONFIG_MEMPOOL_SWITCH_ATTRIBUTE, config_set_mempool_switch,
 		NULL, 0,
@@ -1489,6 +1494,7 @@ FrontendConfig_init () {
   cfg->unhashed_pw_switch = SLAPD_UNHASHED_PW_ON;
   init_return_orig_type = cfg->return_orig_type = LDAP_OFF;
   init_enable_turbo_mode = cfg->enable_turbo_mode = LDAP_ON;
+  init_connection_buffer = cfg->connection_buffer = CONNECTION_BUFFER_ON;
 
 #ifdef MEMPOOL_EXPERIMENTAL
   init_mempool_switch = cfg->mempool_switch = LDAP_ON;
@@ -6944,6 +6950,43 @@ config_set_enable_turbo_mode( const char *attrname, char *value,
     return retVal;
 }
 
+int
+config_get_connection_buffer(void)
+{
+    int retVal;
+    slapdFrontendConfig_t *slapdFrontendConfig = getFrontendConfig();
+    retVal = (int)slapdFrontendConfig->connection_buffer;
+
+    return retVal;
+}
+
+int
+config_set_connection_buffer( const char *attrname, char *value,
+                            char *errorbuf, int apply )
+{
+    int retVal =  LDAP_SUCCESS;
+    slapdFrontendConfig_t *slapdFrontendConfig = getFrontendConfig();
+
+    if ( config_value_is_null( attrname, value, errorbuf, 0 )) {
+	return LDAP_OPERATIONS_ERROR;
+    }
+
+    if ((strcasecmp(value, "0") != 0) && (strcasecmp(value, "1") != 0) &&
+        (strcasecmp(value, "2") != 0)) {
+        PR_snprintf(errorbuf, SLAPI_DSE_RETURNTEXT_SIZE,
+            "%s: invalid value \"%s\". Valid values are \"0\", "
+            "\"1\", or \"2\".", attrname, value);
+        retVal = LDAP_OPERATIONS_ERROR;
+    }
+
+    if ( !apply ) {
+	return retVal;
+    }
+
+    PR_AtomicSet(&slapdFrontendConfig->connection_buffer, atoi(value));
+    return retVal;
+}
+
 /*
  * This function is intended to be used from the dse code modify callback.  It
  * is "optimized" for that case because it takes a berval** of values, which is

ldap/servers/slapd/proto-slap.h

@@ -572,6 +572,8 @@ int config_get_unhashed_pw_switch();
 int config_get_sasl_maxbufsize();
 int config_get_enable_turbo_mode();
 int config_set_enable_turbo_mode(const char *attrname, char *value, char *errorbuf, int apply);
+int config_get_connection_buffer();
+int config_set_connection_buffer(const char *attrname, char *value, char *errorbuf, int apply);
 
 PLHashNumber hashNocaseString(const void *key);
 PRIntn hashNocaseCompare(const void *v1, const void *v2);

ldap/servers/slapd/slap.h

@@ -2077,6 +2077,7 @@ typedef struct _slapdEntryPoints {
 #define CONFIG_SASL_MAXBUFSIZE "nsslapd-sasl-max-buffer-size"
 #define CONFIG_SEARCH_RETURN_ORIGINAL_TYPE "nsslapd-search-return-original-type-switch"
 #define CONFIG_ENABLE_TURBO_MODE "nsslapd-enable-turbo-mode"
+#define CONFIG_CONNECTION_BUFFER "nsslapd-connection-buffer"
 
 #ifdef MEMPOOL_EXPERIMENTAL
 #define CONFIG_MEMPOOL_SWITCH_ATTRIBUTE "nsslapd-mempool"
@@ -2320,6 +2321,7 @@ typedef struct _slapdFrontendConfig {
   Slapi_Counter *sasl_mapping_fallback;
   slapi_onoff_t unhashed_pw_switch;	/* switch to on/off/nolog unhashed pw */
   slapi_onoff_t enable_turbo_mode;
+  slapi_int_t connection_buffer; /* values are CONNECTION_BUFFER_* below */
 } slapdFrontendConfig_t;
 
 /* possible values for slapdFrontendConfig_t.schemareplace */
@@ -2327,6 +2329,10 @@ typedef struct _slapdFrontendConfig {
 #define CONFIG_SCHEMAREPLACE_STR_ON					"on"
 #define CONFIG_SCHEMAREPLACE_STR_REPLICATION_ONLY	"replication-only"
 
+#define CONNECTION_BUFFER_OFF 0
+#define CONNECTION_BUFFER_ON 1
+#define CONNECTION_BUFFER_ADAPT 2
+
  
 slapdFrontendConfig_t *getFrontendConfig();