Chapter 12-12
Multi-Threaded Application Support
Introduction
The HCL C API for Domino and Notes supports mult-threaded applications within the same process. Each thread must call the NotesInitThread and NotesTermThread routines.
NotesInitThread performs per-thread initialization for a new thread. A call to this routine is required when a new thread that uses the C API starts, so that internal per-thread storage is initialized and the process thread count is incremented. The thread data includes all local and global Domino database and note handles. Threads that do not call NotesInitThread and reference global Domino handles will corrupt each other's handle values during execution.
NotesTermThread performs per-thread termination for a thread. A call to this routine is required when a thread terminates, so that internal per-thread storage is deallocated and the process thread count is decremented. Threads that fail to call the NotesTermThread routine prior to thread termination will leak any allocated resources.
The main program thread performs initialization and termination for the process by calling NotesInitExtended and NotesTerm, so it does not need to call NotesInitThread and NotesTermThread. Although any thread can call NotesTerm (allowing for flexible process termination), it must always be called by the last thread to ensure appropriate resource deallocation.
Sharing Domino Handles
NotesInitThread provides the capability for separate threads to use global Domino handles when calling the C API. This means that each thread has its own copy of all declared Domino handle global variables, so that modifying the handle values in one thread does not corrupt the handle values of another thread. With this safety comes the limitation that created handle values (for example, as returned by NSFDbOpen or NSFNoteCreate) are only valid in the scope of the thread and cannot be used by a different thread. Even though the handle variables are global to the program, a thread that calls NotesInitThread is still required to create the relevant handles.
A global database handle that is opened by one thread can still be shared by a different thread by using NSFDbReopen to acquire a thread-specific copy. Even though you need to reference an additional database handle, overall thread performance will improve if you use NSFDbReopen instead of NSFDbOpen, because the physical database file access is eliminated.
Multi-Threaded Server Add-In Sample
The sample program THREADS demonstrates the multi-threaded application support provided by the C API, implementing the basics for calling the C API from multiple threads. The program also uses Message Queues for communication from the main thread to the worker threads.
The source for this sample is in the SAMPLES\SERVER\THREADS directory. THREADS is a Domino server add-in program that launches three threads, one for each of the following operations:
- Every 30 seconds send a message to worker thread 1 that will open the sample database, write a note, and close the database.
- Every minute send a message to worker thread 2 that will open the sample database, write a note, and close the database.
- Every other minute send a message to worker thread 3 that will open the sample database, read and display the notes, and close the database.
Before calling the main program thread entry point AddInMain, Domino calls NotesInitExtended, which initializes the program and the main thread. The three threads share the sample global database and note handle when accessing the sample database. Each thread created must call NotesInitThread and NotesTermThread.
The THREADS sample program also contains basic semaphoring and file locking logic. The main program thread uses semaphores to determine when the three threads have ended. The worker threads use file locking to serialize requests to open the local sample database file. Note that file locking would be unnecessary if the database were located on a remote server.
The main thread communicates with the worker threads through Message Queues. A Message Queue is created and opened in the main thread for each worker thread and messages are sent to the threads to perform their operations. The main thread controls the timing of all messages and sends a Quit message after a specific number of messages have been sent to terminate each thread. When each worker thread has ended the main thread then terminates.
| threads.c - Global Domino Handle Declarations and Main Thread. |
The following code fragment shows the global data declarations and main program thread for the THREADS sample program. The program thread uses the following globals -- a database and note handle, a semaphore, a file lock count, a generic message queue name, 3 specific message queue names, and an array of 3 message queue handles. The main program thread opens up the sample database, starts the three threads, and sends messages to the threads to perform their operations based on time intervals.
... <missing global declarations>
/ Global variables /
DBHANDLE db_handle; / open handle to sample db /
NOTEHANDLE note_handle; / note handle /
int semaphore; / thread semaphore count /
int filelock; / nsf file lock count /
char MsgQueueName[3][128]; / generic Message queue name /
char MsgQueue1[] = TASK_QUEUE_PREFIX "MSG_Q1"; / Message queue name /
char MsgQueue2[] = TASK_QUEUE_PREFIX "MSG_Q2"; / Message queue name /
char MsgQueue3[] = TASK_QUEUE_PREFIX "MSG_Q3"; / Message queue name /
MQHANDLE hQueue[3]; / Handles to message queue /
... <missing global declarations>
STATUS LNPUBLIC AddInMain (HMODULE hModule, int argc, char argv[])
{
... <missing local declarations>
/
Initialize the addin task -
Set the task name and status string of this add-in task. The task
name and status string appear on the status line at the Domino
server in response to the 'show tasks' command. Get the handle to
this default status line descriptor and delete it. Then create a new
status line and set the status to 'Initializing'.
/
... <missing code>
/ Open sample database and assign global handle /
if (error = NSFDbOpen (DATABASE_NAME, &db_handle))
{
AddInLogMessageText("Error opening database.", ERR(error));
return(ERR(error));
}
/ Initialize semaphore and lock counts and spawn threads for each of
the addin operations
/
semaphore = 0;
filelock = 0;
#ifdef WIN32
_beginthread (ThirtySecOps, 0, NULL); / 30 Second op thread /
_beginthread (OneMinOps, 0, NULL); / 1 Minute op thread /
_beginthread (TwoMinOps, 0, NULL); / 2 Minute op thread */
else
_beginthread (ThirtySecOps, NULL, 16384, NULL); /* 30 Second op thread */<br>
_beginthread (OneMinOps, NULL, 16384, NULL); /* 1 Minute op thread */<br>
_beginthread (TwoMinOps, NULL, 16384, NULL); /* 2 Minute op thread */<br>
endif
/ Create and Open Message Queues for worker threads /
strcpy(MsgQueueName[0], MsgQueue1);
strcpy(MsgQueueName[1], MsgQueue2);
strcpy(MsgQueueName[2], MsgQueue3);
for (i=0; i<3; i++)
{
error = MQCreate (MsgQueueName[i], 0, 0); / No quota on messages /
if (NOERROR != error)
return (ERR(error));
error = MQOpen (MsgQueueName[i], 0, &hQueue[i]);
if (NOERROR != error)
return (ERR(error));
}
/ Initialize semaphore and lock counts and spawn threads for each of
the addin operations
/
semaphore = 0;
filelock = 0;
#ifdef WIN32
_beginthread (ThirtySecOps, 0, NULL); / 30 Second op thread /
_beginthread (OneMinOps, 0, NULL); / 1 Minute op thread /
_beginthread (TwoMinOps, 0, NULL); / 2 Minute op thread /
#else
_beginthread (ThirtySecOps, NULL, 16384, NULL); / 30 Second op thread /
_beginthread (OneMinOps, NULL, 16384, NULL); / 1 Minute op thread /
_beginthread (TwoMinOps, NULL, 16384, NULL); / 2 Minute op thread /
#endif
/ get current time and initialize thread operation times /
TIME(&cur_time);
op_time_1 = cur_time;
op_time_2 = cur_time;
op_time_3 = cur_time;
/ The main add-in loop assigns jobs to each thread by sending messages through
the thread message queues. Each thread is then told to terminate after a
specific number of job assignments have been sent.
/
while ((!AddInIdleDelay(5000L)) && (threadsdone < 3))
{
OSPreemptOccasionally();
if (semaphore == 0)
goto Done;
/ get current time /
TIME(&cur_time);
/ send message every 30 seconds to worker thread 1 /
if (cur_time == op_time_1 + THIRTYSECONDS)
{
/ if assigned jobs == 6 tell thread to quit /
if (++counter1 == 6)
{
strcpy(MsgBuffer[0], "QUIT");
MsgLen = strlen(MsgBuffer[0]);
/ put the QUIT message in the first queue /
error = MQPut (hQueue[0], 1, MsgBuffer[0], MsgLen, 0);
if (error)
goto Done;
threadsdone++;
}
else / tell thread to run job /
{
strcpy(MsgBuffer[0], "RUN");
MsgLen = strlen(MsgBuffer[0]);
/ put the RUN message in the first queue /
error = MQPut (hQueue[0], 1, MsgBuffer[0], MsgLen, 0);
if (error)
goto Done;
}
op_time_1 += THIRTYSECONDS; / bump up time another 30 seconds /
}
/ send message every 60 seconds to worker thread 2 /
else if (cur_time == op_time_2 + SIXTYSECONDS)
{
/ if assigned jobs == 3 tell thread to quit /
if (++counter2 == 3)
{
strcpy(MsgBuffer[1], "QUIT");
MsgLen = strlen(MsgBuffer[1]);
/ put the QUIT message in the second queue /
error = MQPut (hQueue[1], 1, MsgBuffer[1], MsgLen, 0);
if (error)
goto Done;
threadsdone++;
}
else / tell thread to run job /
{
strcpy(MsgBuffer[1], "RUN");
MsgLen = strlen(MsgBuffer[1]);
/ put the RUN message in the second queue /
error = MQPut (hQueue[1], 1, MsgBuffer[1], MsgLen, 0);
if (error)
goto Done;
}
op_time_2 += SIXTYSECONDS; / bump up time another 60 seconds /
}
/ send message every two minutes to worker thread 3 /
else if (cur_time == op_time_3 + TWOMINUTES)
{
/ if assigned jobs == 2 tell thread to quit /
if (++counter3 == 2)
{
strcpy(MsgBuffer[2], "QUIT");
MsgLen = strlen(MsgBuffer[2]);
/ put the QUIT message in the third queue /
error = MQPut (hQueue[2], 1, MsgBuffer[2], MsgLen, 0);
if (error)
goto Done;
threadsdone++;
}
else / tell thread to run job /
{
strcpy(MsgBuffer[2], "RUN");
MsgLen = strlen(MsgBuffer[2]);
/ put the RUN message in the third queue /
error = MQPut (hQueue[2], 1, MsgBuffer[2], MsgLen, 0);
if (error)
goto Done;
}
op_time_3 += TWOMINUTES; / bump up time another 2 minutes /
}
}
/ We get here when the server notifies us that it is time to terminate.
This can occur when
1) The user has entered "quit" at the server console.
2) The user has entered "tell <ThisTask> quit" at the server console.
3) All three operation threads have ended.
/
/ Wait for threads to terminate or for user to exit /
AddInSetStatusText("Terminating Add-In Threads");
while ( semaphore )
SLEEP(5000L);
Done:
/ close the message queues /
for (i=0; i<3; i++)
{
error = MQClose (hQueue[i], 0);
if (error)
return (ERR(error));
}
NSFDbClose (db_handle);
AddInSetStatusText("Terminating Add-in");
AddInLogMessageText("THREADS: Termination complete.", NOERROR);
/ End of add-in task. We must "return" here rather than "exit". /
return (NOERROR);
}
| threads.c - "Writer" Thread and Service Routines. |
The following code fragment shows the two threads functions called by AddInMain and the service routine called by the threads to write a note to the sample database. Each thread increments the semaphore and calls NotesInitThread prior to entering its service loop, and decrements the semaphore and calls NotesTermThread prior to ending the function. In the service loop, the thread function:
- Waits for a message from the main thread to perform an operation and periodically checks to see if the add-in has been terminated by the user.
- If the RUNOP message is received it sets the database file lock (when available).
- Performs the database operation by calling the service routine WriteOpNote.
- If the QUITOP message is received the thread terminates.
The WriteOpNote routine reopens the global database handle to a local handle and uses the thread's s database and note handles with the corresponding C API routines to write the note. This routine unlocks the database file at completion and returns to the calling routine.
/**********
FUNCTION: ThirtySecOps
PURPOSE: Add-in thread routine that performs "30 Second" operations
by writing a note to the sample database.
**********/
#ifdef WIN32
void ThirtySecOps(void dummy)
else
void _Optlink ThirtySecOps(void *dummy)
endif
{
STATUS error;
int timer;
int counter = 0;
int op = 0;
char MsgBuffer [MAX_MESSAGE + 1]; / Buffer for messages /
WORD MsgLen; / Size of message /
/ First increment semaphore /
semaphore++;
/ Initialize Domino thread - required by threads calling the C API. /
error = NotesInitThread ();
if (error)
{
AddInLogMessageText("Error initializing 30 Second operation thread.", NOERROR);
semaphore--;
_endthread();
}
/ while there isn't an error loop for messages from main thread /
while (NOERROR == error)
{
/ if the main thread terminates we're done /
if (AddInShouldTerminate())
goto Done;
/ attempt to get a message from the queue and wait for 5 seconds /
error = MQGet (hQueue[0], MsgBuffer, MAX_MESSAGE,
MQ_WAIT_FOR_MSG, (5 * 1000), &MsgLen);
MsgBuffer[MsgLen] = '\0';
/ if we received a message... /
if (NOERROR == error)
{
/ get the message from the buffer /
op = ProcessMessage (MsgBuffer, MsgLen);
/ perform operation depending on message received /
switch(op)
{
/ run the job assignment /
case RUNOP:
AddInSetStatusText("Performing 30 Second operation");
/ Do not write note until the database is no longer opened
* by one of the other threads.
/
while ( filelock )
SLEEP (50L);
filelock++; / lock database /
if (error = WriteOpNote(THIRTYSECOP)) / and write note /
{
AddInLogMessageText("Error writing 30 Second
operation Note to database.", ERR(error));
goto Done;
}
AddInLogMessageText("THREADS: 30 Second operation
complete.", NOERROR);
AddInSetStatusText("30 Second operation Idle");
break;
/ quit the job /
case QUITOP:
goto Done;
break;
default:
error = op;
break;
}
}
/ if we have a timeout loop again /
else if (ERR_MQ_TIMEOUT == error)
{
error = NOERROR;
}
} / End of main task loop. /
/ We get here when the server notifies us that it is time to terminate.
Clean up anything we have been doing.
/
Done:
AddInLogMessageText("THREADS: Ending 30 Second operation thread.", NOERROR);
AddInSetStatusText("Exiting 30 Second operation");
/ Terminate Notes thread, decrement semaphore count, and end OS thread /
NotesTermThread ();
semaphore--;
_endthread();
}
/**********
FUNCTION: OneMinOps
PURPOSE: Add-in thread routine that performs "1 Minute" operations
by writing a note to the sample database.
**********/
#ifdef WIN32
void OneMinOps(void dummy)
else
void _Optlink OneMinOps(void *dummy)
endif
{
STATUS error;
int timer;
int counter = 0;
int op = 0;
char MsgBuffer [MAX_MESSAGE + 1]; / Buffer for messages /
WORD MsgLen; / Size of message /
/ First increment semaphore /
semaphore++;
/ Initialize Notes thread - required by threads calling the C API. /
error = NotesInitThread ();
if (error)
{
AddInLogMessageText("Error initializing 1 Minute operation thread.", NOERROR);
semaphore--;
_endthread();
}
/ while there isn't an error loop for messages from main thread /
while (NOERROR == error)
{
/ if the main thread terminates we're done /
if (AddInShouldTerminate())
goto Done;
/ attempt to get a message from the queue and wait for 5 seconds /
error = MQGet (hQueue[1], MsgBuffer, MAX_MESSAGE,
MQ_WAIT_FOR_MSG, (5 * 1000), &MsgLen);
MsgBuffer[MsgLen] = '\0';
/ if we received a message... /
if (NOERROR == error)
{
/ get the message from the buffer /
op = ProcessMessage (MsgBuffer, MsgLen);
/ perform operation depending on message received /
switch(op)
{
/ run the job assignment /
case RUNOP:
AddInSetStatusText("Performing 1 Minute operation");
/ Do not write note until the database is no longer opened by one of the other threads.
/
while ( filelock )
SLEEP (50L);
filelock++; / lock database /
if (error = WriteOpNote(ONEMINOP)) / and write note /
{
AddInLogMessageText("Error writing 1 Minute operation note to database.", ERR(error));
goto Done;
}
AddInLogMessageText("THREADS: 1 Minute operation complete.", NOERROR);
AddInSetStatusText("1 Minute operation idle");
break;
/ quit the job /
case QUITOP:
goto Done;
break;
default:
error = op;
break;
}
}
else if (ERR_MQ_TIMEOUT == error)
{
error = NOERROR;
}
} / End of main task loop. /
/ We get here when the server notifies us that it is time to terminate.
Clean up anything we have been doing.
/
Done:
AddInLogMessageText("THREADS: Ending 1 Minute operation thread.", NOERROR);
AddInSetStatusText("Exiting 1 Minute operation");
/ Terminate Notes thread, decrement semaphore count, and end OS thread /
NotesTermThread ();
semaphore--;
_endthread();
}
/**********
FUNCTION: WriteOpNote
PURPOSE: Local function called by the operation thread routines to
write a note to the sample database. Checks that file
locks are in place before opening local database.
**********/
STATUS WriteOpNote (char op_thread)
{
/ Local data declarations. /
DBHANDLE write_handle; / local database handle /
STATUS error = NOERROR; / return code from API calls /
TIMEDATE timedate; / contents of a time/date field /
char thread_text[64]; / thread specific item text /
/ First ensure that the file has been locked. /
if (filelock < 1)
{
filelock = 0;
return (ERR(ERR_LOCK_FAILED));
}
/ Then ensure that the database file has been locked by only one thread. /
if (filelock > 1)
{
error = ERR_LOCK;
goto Done;
}
/ Reopen the database from the global handle. /
if (error = NSFDbReopen (db_handle, &write_handle))
goto Done;
/ Create a new data note. /
if (error = NSFNoteCreate (write_handle, ¬e_handle))
{
NSFDbClose (write_handle);
goto Done;
}
/ Write the form name to the note. /
if (error = NSFItemSetText (note_handle, "Form", "AddInForm", MAXWORD))
{
NSFNoteClose (note_handle);
NSFDbClose (write_handle);
goto Done;
}
/ Write a text field to the note. /
... <missing code>
/ Write the current time into the note. /
... <missing code>
/ Add the note to the database. /
if (error = NSFNoteUpdate (note_handle, 0))
{
NSFNoteClose (note_handle);
NSFDbClose (write_handle);
goto Done;
}
/ Deallocate the new note from memory. /
if (error = NSFNoteClose (note_handle))
{
NSFDbClose (write_handle);
goto Done;
}
/ Close the database. /
if (error = NSFDbClose (write_handle))
goto Done;
/ End of function. Unlock database file and return status. /
Done:
filelock--;
return (error);
}
| threads.c - "Reader" Thread and Service Routines. |
The following code fragment shows the thread functions called by AddInMain and the service routine called by the thread to read and display the notes in the sample database. The thread increments the semaphore and calls NotesInitThread before entering its service loop, and decrements the semaphore and calls NotesTermThread before ending the function. Within the service loop, the thread function:
- Waits for a message from the main thread to perform an operation and periodically checks to see if the add-in has been terminated by the user.
- If the RUNOP message is received it sets the file lock (when available) and reopens the global database handle to a local handle.
- Performs the database operation by calling the NSFSearch service routine ReadOpNote.Closes and unlocks the database.
- If the QUITOP message is received the thread terminates.
The ReadOpNote routine uses the thread's database and note handle values with the corresponding C API routines to open and read the notes from the database.
/**********
FUNCTION: TwoMinOps
PURPOSE: Addin thread routine that performs "2 Minute" operations..
by reading and displaying all the notes in the sample database.
**********/
#ifdef WIN32
void TwoMinOps(void dummy)
else
void _Optlink TwoMinOps(void *dummy)
endif
{
DBHANDLE read_handle; / local database handle /
STATUS error;
int counter = 0;
int op = 0;
char MsgBuffer [MAX_MESSAGE + 1]; / Buffer for messages /
WORD MsgLen; / Size of message /
/ First increment semaphore /
semaphore++;
/ Initialize Notes thread - required by threads calling the C API. /
error = NotesInitThread ();
if (error)
{
AddInLogMessageText("Error initializing 2 Minute operation thread.", NOERROR);
semaphore--;
_endthread();
}
/ while there isn't an error loop for messages from main thread /
while (NOERROR == error)
{
/ if the main thread terminates we're done /
if (AddInShouldTerminate())
goto Done;
/ attempt to get a message from the queue and wait for 5 seconds /
error = MQGet (hQueue[2], MsgBuffer, MAX_MESSAGE,
MQ_WAIT_FOR_MSG, (5 * 1000), &MsgLen);
MsgBuffer[MsgLen] = '\0';
/ if we received a message... /
if (NOERROR == error)
{
/ get the message from the buffer /
op = ProcessMessage (MsgBuffer, MsgLen);
/ perform operation depending on message received /
switch(op)
{
/ run the job assignment /
case RUNOP:
AddInSetStatusText("Performing 2 Minute operation");
AddInLogMessageText("THREADS: Begin 2 Minute Database Summary:",NOERROR);
/ Do not read note until the database is no longer opened
by one of the other threads.
/
while ( filelock )
SLEEP (50L); / sleep length based on typical lock time /
/ Lock and reopen the database from the global handle. /
filelock++;
if (error = NSFDbReopen (db_handle, &read_handle))
{
AddInLogMessageText("Error opening database.",
ERR(error));
filelock--;
goto Done;
}
/ Call NSFSearch to find all data notes in the database. /
if (error = NSFSearch (
read_handle, / database handle /
NULLHANDLE, / selection formula(all)/
NULL, / title of view in selection formula /
0, / search flags /
NOTE_CLASS_DOCUMENT, / note class to find /
NULL, / starting date (unused) /
ReadOpNote,/action routine for notes found/
&read_handle, / argument to action routine /
NULL)) / returned ending date (unused) /
{
NSFDbClose (read_handle);
AddInLogMessageText("Error reading notes from
database.", ERR(error));
filelock--;
goto Done;
}
/ Close and unlock the database. /
if (error = NSFDbClose (read_handle))
{
AddInLogMessageText("Error closing database.",
ERR(error));
filelock--;
goto Done;
}
filelock--;
AddInLogMessageText("THREADS: 2 Minute operation
complete.", NOERROR);
AddInSetStatusText("2 Minute operation Idle");
break;
/ quit the job /
case QUITOP:
goto Done;
break;
default:
error = op;
break;
}
}
else if (ERR_MQ_TIMEOUT == error)
{
error = NOERROR;
}
} / End of main task loop. /
/ We get here when the server notifies us that it is time to terminate.
Clean up anything we have been doing.
/
Done:
AddInLogMessageText("THREADS: Ending 2 Minute operation thread.",NOERROR);
AddInSetStatusText("Exiting 2 Minute operation");
/ Terminate Notes thread, decrement semaphore count, and end OS thread /
NotesTermThread ();
semaphore--;
_endthread();
}
/**********
FUNCTION: ReadOpNote
PURPOSE: Local function called by the operation thread routines to
read and print out all the notes from the sample database.
Assumes that file locks are in place for local database.
This routine is called by NSFSearch for each note that
matches the selection criteria (in this case, all the notes).
INPUTS:
The first argument to this function is the optional argument
that we supplied when we called NSFSearch.
The second argument is supplied by NSFSearch. It is
a structure of information about the note that was found.
The third argument is also supplied by NSFSearch and is
the summary buffer for this note.
**********/
STATUS LNPUBLIC ReadOpNote ( VOID read_handle,
SEARCH_MATCH search_info,
ITEM_TABLE summary_info)
{
/ Local data declarations. /
STATUS error;
char field_text[64]; / contents of a AddIn_Text field /
WORD field_len; / length of field /
char msg_text[80]; / add-in message string /
SEARCH_MATCH SearchMatch; / local copy of search match /
memcpy ((char)(&SearchMatch), (char )search_info, sizeof(SEARCH_MATCH));
/ Skip this note if it does not really match the search criteria (it is
* now deleted or modified). This is not necessary for full searches,
* but is shown here in case a starting date was used in the search.
/
if (!(SearchMatch.SERetFlags & SE_FMATCH))
return (NOERROR);
/ Open the note. /
if (error = NSFNoteOpen (
(DBHANDLE )read_handle, / database handle /
SearchMatch.ID.NoteID, / note ID /
0, / open flags /
¬e_handle)) / note handle (return) /
return (ERR(error));
/ Get the note text item. /
... <missing code>
/ Build message text: note ID + field text. /
... <missing code>
/ Display the note information message. /
... <missing code>
/ Close the note. /
if (error = NSFNoteClose (note_handle))
return (ERR(error));
/ End of subroutine. /
return (NOERROR);
}