Chapter 14 Writing Multithreaded Clients

The Sun™ Java System LDAP SDK for C contains the functionality for multi-threaded clients to access the same data types. This chapter explains how to write multi-threaded LDAP clients. It consists of the following sections:

Specifying Thread Functions

If you are writing a multi-threaded client with different threads access the same LDAP structure, you need to set up the session so that the threads do not interfere with each other. For example, in a single-threaded LDAP client, the LDAP structure contains the error code of the last LDAP operation. In a multi-threaded client, you need to set up the session so that each thread can have its own error code.

The LDAP_OPT_THREAD_FN_PTRS session option lets you set up an ldap_thread_fns structure identifying the functions that are called in multi-threaded environments (for example, functions to lock and unlock critical sections of code and to get and set errors). Because this structure lets you specify these functions, you can use the LDAP SDK for C in different types of threading environments.

Setting Up the ldap_thread_fns Structure

If you are writing a multi-threaded client in which different threads use the same LDAP connection, you need to set up the ldap_thread_fns structure and identify the functions that you want to use. Code Example 14-1 illustrates the ldap_thread_fns structure.

Setting Up the ldap_extra_thread_fns Structure

Table 14-1 ldap_thread_fns Field Descriptions
Field	Description
*ltf_mutex_alloc	Function pointer for allocating a mutex. This function is called by the client when needed if the function pointer is not NULL.
*ltf_mutex_free	Function pointer for freeing a mutex. This function is called by the client when needed if the function pointer is not NULL.
*ltf_mutex_lock	Function pointer for locking critical sections of code. This function is called by the client when needed if the function pointer is not NULL.
*ltf_mutex_unlock	Function pointer for unlocking critical sections of code. This function is called by the client when needed if the function pointer is not NULL.
*ltf_get_errno	Function pointer for getting the value of the errno variable. This function is called by the client when needed if the function pointer is not NULL. In a threaded environment, errno is typically redefined so that it has a value for each thread, rather than a global value for the entire process. This redefinition is done at compile time. Because the libldap library does not know what method your code and threading environment will use to get the value of errno for each thread, it calls this function to return the value of errno.
*ltf_set_errno	Function pointer for setting the value of the errno variable. This function is called by the client when needed if the function pointer is not NULL. In a threaded environment, errno is typically redefined so that it has a value for each thread, rather than a global value for the entire process. This redefinition is done at compile time. Because the libldap library does not know what method your code and threading environment will use to get the value of errno for each thread, it calls this function to set the value of errno.
*ltf_get_lderrno	Function pointer for getting error values from calls to functions in the libldap library. This function is called by the client when needed if the function pointer is not NULL. If this function pointer is not set, the libldap library records these errors in fields in the LDAP structure.
*ltf_set_lderrno	Function pointer for setting error values from calls to functions in the libldap library. This function is called by the client when needed if the function pointer is not NULL. If this function pointer is not set, the libldap library records these errors in fields in the LDAP structure.
*ltf_lderrno_arg	Additional parameter passed to the functions for getting and setting error values from calls to functions in the libldap library. ltf_get_lderrno and ltf_set_lderrno identify these functions.

The LDAP SDK for C supports only the LDAP_TF_TREADID_CALLBACK *ltf_threadid_fn function. (If any of the other extra thread callback functions are set, they will be ignored.) The supported function must return an identifier that is unique to the calling thread, like pthread_self() does. You use this function callback in a multi-threaded application to improve the performance of thread locking.

Setting Session Options

After you set up the ldap_thread_fns structure, you need to associate the structure with the current session. Call the ldap_set_option() function and pass LDAP_OPT_THREAD_FN_PTRS as the value of the option parameter. Pass a pointer to the ldap_thread_fns structure as the value of the optdata parameter. Code Example 14-3 sets up an ldap_thread_fns structure for an LDAP session.

If you also set up the ldap_extra_thread_fns structure, you can associate it with the current session by calling ldap_set_option() function and passing LDAP_OPT_EXTRA_THREAD_FN_PTRS as the value of the option parameter. You must also pass a pointer to the ldap_extra_thread_fns structure as the value of the optdata parameter.

Pthreads Client Application

Code Example 14-4, which uses pthreads (POSIX threads) on the Solaris™ platform, is the source code for a multi-threaded client. The client connects to a specified LDAP server and creates several threads to perform multiple search and update operations simultaneously on the directory.

Code Example 14-4 Using POSIX Threading Under Solaris


#include <stdio.h>
#include <malloc.h>
#include <errno.h>
#include <pthread.h>
#include <ldap.h>
#include <synch.h>

/* Authentication and search information. */
#define NAME "cn=Directory Manager"
#define PASSWORD "rtfm11111"
#define BASE "dc=example,dc=com"
#define SCOPE LDAP_SCOPE_SUBTREE

/* Function declarations */
static void *search_thread();
static void *modify_thread();
static void *add_thread();
static void *delete_thread();
static void set_ld_error();
static int get_ld_error();
static void set_errno();
static int get_errno();
static void tsd_setup();

/* Linked list of LDAPMessage structs for search results. */
typedef struct ldapmsgwrapper {
LDAPMessage *lmw_messagep;
struct ldapmsgwrapper *lmw_next;
} ldapmsgwrapper;

LDAP *ld;
pthread_key_t key;

main( int argc, char **argv )
{
pthread_attr_t attr;
pthread_t search_tid, search_tid2, search_tid3, search_tid4;
pthread_t modify_tid, add_tid, delete_tid;
void *status;
struct ldap_thread_fns tfns;
struct ldap_extra_thread_fns extrafns;
int rc;

/* Check command-line syntax. */
if ( argc != 3 ) {
fprintf( stderr, "usage: %s <host> <port>\n", argv[0] );
exit( 1 );
}

/* Create a key. */
if ( pthread_key_create( &key, free ) != 0 ) {
perror( "pthread_key_create" );
}
tsd_setup();

/* Initialize the LDAP session. */
if ( (ld = ldap_init( argv[1], atoi( argv[2] ) )) == NULL ) {
perror( "ldap_init" );
exit( 1 );
}

/* Set the function pointers for dealing with mutexes
and error information. */
memset( &tfns, '\0', sizeof(struct ldap_thread_fns) );
tfns.ltf_mutex_alloc = (void ()(void)) my_mutex_alloc;
tfns.ltf_mutex_free = (void ()(void )) my_mutex_free;
tfns.ltf_mutex_lock = (int ()(void )) pthread_mutex_lock;
tfns.ltf_mutex_unlock = (int ()(void )) pthread_mutex_unlock;
tfns.ltf_get_errno = get_errno;
tfns.ltf_set_errno = set_errno;
tfns.ltf_get_lderrno = get_ld_error;
tfns.ltf_set_lderrno = set_ld_error;
tfns.ltf_lderrno_arg = NULL;

/* Set up this session to use those function pointers. */
rc = ldap_set_option( ld, LDAP_OPT_THREAD_FN_PTRS, (void *) &tfns );
if ( rc < 0 ) {
fprintf( stderr, "ldap_set_option (LDAP_OPT_THREAD_FN_PTRS): %s\n", ldap_err2string( rc ) );
exit( 1 );
}

/* Set the function pointers for working with semaphores. */
memset( &extrafns, '\0', sizeof(struct ldap_extra_thread_fns) );
extrafns.ltf_mutex_trylock = (int ()(void )) = null;
extrafns.ltf_sema_alloc = (void ()(void)) = null;
extrafns.ltf_sema_free = (void ()(void )) = null;
extrafns.ltf_sema_wait = (int ()(void )) = null;
extrafns.ltf_sema_post = (int ()(void )) = null;
extrafns.ltf_threadid_fn = (void * (*)(void) )pthread_self;
/* Set up this session to use those function pointers. */
rc = ldap_set_option( ld, LDAP_OPT_EXTRA_THREAD_FN_PTRS, (void *) &extrafns );
if ( rc < 0 ) {
fprintf( stderr, "ldap_set_option (LDAP_OPT_EXTRA_THREAD_FN_PTRS): %s\n", ldap_err2string( rc ) );
exit( 1 );
}

/* Attempt to bind to the server. */
rc = ldap_simple_bind_s( ld, NAME, PASSWORD );
if ( rc != LDAP_SUCCESS ) {
fprintf( stderr, "ldap_simple_bind_s: %s\n", ldap_err2string( rc ) );
exit( 1 );
}

/* Initialize the attribute. */
if ( pthread_attr_init( &attr ) != 0 ) {
perror( "pthread_attr_init" );
exit( 1 );
}

/* Specify that the threads are joinable. */
pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_JOINABLE );

/* Create seven threads: one for adding, one for modifying,
one for deleting, and four for searching. */
if ( pthread_create( &search_tid, &attr, search_thread, "1" ) != 0 ) {
perror( "pthread_create search_thread" );
exit( 1 );
}
if ( pthread_create( &modify_tid, &attr, modify_thread, "2" ) != 0 ) {
perror( "pthread_create modify_thread" );
exit( 1 );
}
if ( pthread_create( &search_tid2, &attr, search_thread, "3" ) != 0 ) {
perror( "pthread_create search_thread2" );
exit( 1 );
}
if ( pthread_create( &add_tid, &attr, add_thread, "4" ) != 0 ) {
perror( "pthread_create add_thread" );
exit( 1 );
}
if ( pthread_create( &search_tid3, &attr, search_thread, "5" ) != 0 ) {
perror( "phread_create search_thread3" );
exit( 1 );
}
if ( pthread_create( &delete_tid, &attr, delete_thread, "6" ) != 0 ) {
perror( "pthread_create delete_thread" );
exit( 1 );
}
if ( pthread_create( &search_tid4, &attr, search_thread, "7" ) != 0 ) {
perror( "pthread_create search_thread4" );
exit( 1 );
}

/* Wait until these threads exit. */
pthread_join( modify_tid, &status );
pthread_join( add_tid, &status );
pthread_join( delete_tid, &status );
pthread_join( search_tid, &status );
pthread_join( search_tid2, &status );
pthread_join( search_tid3, &status );
pthread_join( search_tid4, &status );
}

/* Thread for searching the directory.
The results are not printed out. */
static void *
search_thread( char *id )
{
LDAPMessage *res;
LDAPMessage *e;
char *a;
char **v;
char *dn;
BerElement *ber;
int i, rc, parse_rc, msgid, finished;
int num_entries, num_refs;
void *tsd;
struct timeval zerotime;
zerotime.tv_sec = zerotime.tv_usec = 0L;

printf( "Starting search_thread %s.\n", id );
tsd_setup();
/* Continually search the directory. */
for ( ;; ) {
printf( "Thread %s: Searching...\n", id );
finished = 0;
num_entries = 0;
num_refs = 0;
rc = ldap_search_ext( ld, BASE, SCOPE, "(objectclass=*)",
NULL, 0, NULL, NULL, NULL, LDAP_NO_LIMIT, &msgid );
if ( rc != LDAP_SUCCESS ) {
fprintf( stderr, "Thread %s error: ldap_search: %s\n",
id, ldap_err2string( rc ) );
continue;
}

/* Iterate through the results. In this example,
we don't print out all the results. (It's easier
to see the output from the other threads this way.) */
while ( !finished ) {
rc = ldap_result( ld, msgid, LDAP_MSG_ONE, &zerotime, &res );
switch ( rc ) {
case -1:
rc = ldap_get_lderrno( ld, NULL, NULL );
fprintf( stderr, "ldap_result: %s\n", ldap_err2string( rc ) );
finished = 1;
break;
case 0:
break;
/* Keep track of the number of entries found. */
case LDAP_RES_SEARCH_ENTRY:
num_entries++;
break;
/* Keep track of the number of search references. */
case LDAP_RES_SEARCH_REFERENCE:
num_refs++;
break;
case LDAP_RES_SEARCH_RESULT:
finished = 1;
parse_rc = ldap_parse_result( ld, res, &rc, NULL, NULL, NULL, NULL, 1 );
if ( parse_rc != LDAP_SUCCESS ) {
fprintf( stderr, "Thread %s error: can't parse result code.\n", id );
break;
} else {
if ( rc != LDAP_SUCCESS ) {
fprintf( stderr, "Thread %s error: ldap_search: %s\n", id, ldap_err2string( rc ) );
} else {
printf( "Thread %s: Got %d results and %d references.\n", id, num_entries, num_refs );
}
}
break;
default:
break;
}
}
}
}

/* Thread for modifying directory entries.
This thread searches for entries and randomly selects entries from
the search results for modification. */
static void *
modify_thread( char *id )
{
LDAPMessage *res;
LDAPMessage *e;
int i, modentry, num_entries, msgid, rc, parse_rc, finished;
LDAPMod mod;
LDAPMod *mods[2];
char *vals[2];
char *dn;
ldapmsgwrapper list, lmwp, *lastlmwp;
struct timeval zerotime;
zerotime.tv_sec = zerotime.tv_usec = 0L;

printf( "Starting modify_thread %s.\n", id );
tsd_setup();
rc = ldap_search_ext( ld, BASE, SCOPE, "(objectclass=*)",
NULL, 0, NULL, NULL, NULL, LDAP_NO_LIMIT, &msgid );
if ( rc != LDAP_SUCCESS ) {
fprintf( stderr, "Thread %s error: Modify thread: "
"ldap_search_ext: %s\n", id, ldap_err2string( rc ) );
exit( 1 );
}
list = lastlmwp = NULL;
finished = 0;
num_entries = 0;
while ( !finished ) {
rc = ldap_result( ld, msgid, LDAP_MSG_ONE, &zerotime, &res );
switch ( rc ) {
case -1:
rc = ldap_get_lderrno( ld, NULL, NULL );
fprintf( stderr, "ldap_result: %s\n", ldap_err2string( rc ) );
exit( 1 );
break;
case 0:
break;

/* Keep track of the number of entries found. */
case LDAP_RES_SEARCH_ENTRY:
num_entries++;
if (( lmwp = (ldapmsgwrapper *)
malloc( sizeof( ldapmsgwrapper ))) == NULL ) {
fprintf( stderr, "Thread %s: Modify thread: Cannot malloc\n", id );
exit( 1 );
}
lmwp->lmw_messagep = res;
lmwp->lmw_next = NULL;
if ( lastlmwp == NULL ) {
list = lastlmwp = lmwp;
} else {
lastlmwp->lmw_next = lmwp;
}
lastlmwp = lmwp;
break;
case LDAP_RES_SEARCH_REFERENCE:
break;
case LDAP_RES_SEARCH_RESULT:
finished = 1;
parse_rc = ldap_parse_result( ld, res, &rc, NULL, NULL, NULL, NULL, 1 );
if ( parse_rc != LDAP_SUCCESS ) {
fprintf( stderr, "Thread %s error: can't parse result code.\n", id );
exit( 1 );
} else {
if ( rc != LDAP_SUCCESS ) {
fprintf( stderr, "Thread %s error: ldap_search: %s\n", id, ldap_err2string( rc ) );
} else {
printf( "Thread %s: Got %d results.\n", id, num_entries );
}
}
break;
default:
break;
}
}

/* Set up the modifications to be made. */
mods[0] = &mod;
mods[1] = NULL;
vals[0] = "bar";
vals[1] = NULL;

/* Modify randomly selected entries. */
for ( ;; ) {

/* Randomly select the entries. */
modentry = rand() % num_entries;
for ( i = 0, lmwp = list; lmwp != NULL && i < modentry;
i++, lmwp = lmwp->lmw_next ) {
/* Keep iterating. */
}
if ( lmwp == NULL ) {
fprintf( stderr,
"Thread %s: Modify thread could not find entry %d of %d\n",
id, modentry, num_entries );
continue;
}
e = lmwp->lmw_messagep;
printf( "Thread %s: Modify thread picked entry %d of %d\n", id, i, num_entries );

/* Perform the modification. */
dn = ldap_get_dn( ld, e );
mod.mod_op = LDAP_MOD_REPLACE;
mod.mod_type = "description";
mod.mod_values = vals;
printf( "Thread %s: Modifying (%s)\n", id, dn );
rc = ldap_modify_ext_s( ld, dn, mods, NULL, NULL );
if ( rc != LDAP_SUCCESS ) {
rc = ldap_get_lderrno( ld, NULL, NULL );
fprintf( stderr, "ldap_modify_ext_s: %s\n", ldap_err2string( rc ) );
}
free( dn );
}
}

/* Thread for adding directory entries.
This thread randomly generates DNs for entries and attempts to add them to the directory. */
static void *
add_thread( char *id )
{
LDAPMod mod[5];
LDAPMod *mods[6];
char dn[BUFSIZ], name[40];
char cnvals[2], snvals[2], *ocvals[3];
int i, rc;

printf( "Starting add_thread %s.\n", id );
tsd_setup();

/* Set up the entry to be added. */
for ( i = 0; i < 5; i++ ) {
mods[i] = &mod[i];
}
mods[5] = NULL;
mod[0].mod_op = 0;
mod[0].mod_type = "cn";
mod[0].mod_values = cnvals;
cnvals[1] = NULL;
mod[1].mod_op = 0;
mod[1].mod_type = "sn";
mod[1].mod_values = snvals;
snvals[1] = NULL;
mod[2].mod_op = 0;
mod[2].mod_type = "objectclass";
mod[2].mod_values = ocvals;
ocvals[0] = "top";
ocvals[1] = "person";
ocvals[2] = NULL;
mods[3] = NULL;

/* Randomly generate DNs and add entries. */
for ( ;; ) {
sprintf( name, "%d", rand() );
sprintf( dn, "cn=%s, " BASE, name );
cnvals[0] = name;
snvals[0] = name;
printf( "Thread %s: Adding entry (%s)\n", id, dn );
rc = ldap_add_ext_s( ld, dn, mods, NULL, NULL );
if ( rc != LDAP_SUCCESS ) {
rc = ldap_get_lderrno( ld, NULL, NULL );
fprintf( stderr, "ldap_add_ext_s: %s\n", ldap_err2string( rc ) );
}
}
}

/* Thread for deleting directory entries.
This thread randomly selects entries for deletion. */
static void *
delete_thread( char *id )
{
LDAPMessage *res;
char dn[BUFSIZ], name[40];

printf( "Starting delete_thread %s.\n", id );
tsd_setup();

/* Randomly select entries for deletion. */
for ( ;; ) {
sprintf( name, "%d", rand() );
sprintf( dn, "cn=%s, " BASE, name );
printf( "Thread %s: Deleting entry (%s)\n", id, dn );
if ( ldap_delete_ext_s( ld, dn, NULL, NULL ) != LDAP_SUCCESS ) {
ldap_perror( ld, "ldap_delete_ext_s" );
}
}
}

/* Function for allocating a mutex. */
static void *
my_mutex_alloc( void )
{
pthread_mutex_t *mutexp;
if ( (mutexp = malloc( sizeof(pthread_mutex_t) )) != NULL ) {
pthread_mutex_init( mutexp, NULL );
}
return( mutexp );
}

/* Function for freeing a mutex. */
static void
my_mutex_free( void *mutexp )
{
pthread_mutex_destroy( (pthread_mutex_t *) mutexp );
free( mutexp );
}

/* Error structure. */
struct ldap_error {
int le_errno;
char *le_matched;
char *le_errmsg;
};

/* Function to set up thread-specific data. */
static void
tsd_setup()
{
void *tsd;
tsd = pthread_getspecific( key );
if ( tsd != NULL ) {
fprintf( stderr, "tsd non-null!\n" );
pthread_exit( NULL );
}
tsd = (void *) calloc( 1, sizeof(struct ldap_error) );
pthread_setspecific( key, tsd );
}

/* Function for setting an LDAP error. */
static void
set_ld_error( int err, char matched, char errmsg, void *dummy )
{
struct ldap_error *le;
le = pthread_getspecific( key );
le->le_errno = err;
if ( le->le_matched != NULL ) {
ldap_memfree( le->le_matched );
}
le->le_matched = matched;
if ( le->le_errmsg != NULL ) {
ldap_memfree( le->le_errmsg );
}
le->le_errmsg = errmsg;
}

/* Function for getting an LDAP error. */
static int
get_ld_error( char matched, char errmsg, void *dummy )
{
struct ldap_error *le;
le = pthread_getspecific( key );
if ( matched != NULL ) {
*matched = le->le_matched;
}
if ( errmsg != NULL ) {
*errmsg = le->le_errmsg;
}
return( le->le_errno );
}

/* Function for setting errno. */
static void
set_errno( int err )
{
errno = err;
}

/* Function for getting errno. */
static int
get_errno( void )
{
return( errno );
}