| Nominal Animal |
06-08-2012 08:27 AM |
Add synchronization to the worker threads, and use a flag to tell them to not do any work at all: like a starting gate, with a signal flag to tell the participants if the start failed.
All you really need is just a pthread_mutex_t and a flag variable. Have the thread creator first lock the mutex. If all threads can be created successfully, clear the flag, otherwise set it; then release the mutex. The worker threads first lock the same mutex, then check the value of the flag. If the flag is set, they unlock the mutex and abort. Otherwise they just unlock the mutex and start the real work.
It is simpler if you just add the functionality to the actual thread functions, but you can use a simple wrapper function and a related work spec if you need a more generic solution:
Code:
struct spec {
pthread_mutex_t *lock;
int *cancel;
void *payload;
void * (*function)(void *);
};
void *unless_canceled(void *payload)
{
struct spec *const spec = (struct spec *)payload;
/* Make sure the parameters are valid. */
if (!spec || !spec->lock || !spec->function)
return NULL;
/* Grab the lock. */
if (pthread_mutex_lock(spec->lock))
return NULL;
/* Canceled? */
if (spec->cancel && *(spec->cancel)) {
pthread_mutex_unlock(spec->lock);
return NULL;
}
pthread_mutex_unlock(spec->lock);
return spec->function(spec->payload);
}
The way you use the wrapper function is simple:
Code:
pthread_t thread[2];
struct spec spec[2];
int err[2];
pthread_attr_t attrs;
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
int cancel;
/* Create a pthread attribute to limit stack size to 65536 bytes. */
pthread_attr_init(&attrs);
pthread_attr_setstacksize(&attrs, 65536);
/* Grab the mutex to stop the wrapper scripts. */
pthread_mutex_lock(&lock);
/* Point the work specs to the shared states. */
spec[0].lock = &lock;
spec[1].lock = &lock;
spec[0].cancel = &cancel;
spec[1].cancel = &cancel;
/* Set the worker function and payload. */
spec[0].payload = (void *)c1;
spec[0].function = sender;
spec[1].payload = (void *)c2;
spec[1].function = sender;
/* Create the threads. */
err[0] = pthread_create(&thread[0], &attrs, unless_canceled, &spec[0]);
err[1] = pthread_create(&thread[1], &attrs, unless_canceled, &spec[1]);
if (err[0] || err[1]) {
/* One or both failed: cancel the threads. */
cancel = 1;
pthread_mutex_unlock(&lock);
/* Reap aborted threads. Mark err[] so you can check later. */
if (!err[0]) { pthread_join(thread[0], NULL); err[0] = ENOENT; }
if (!err[1]) { pthread_join(thread[1], NULL); err[1] = ENOENT; }
} else {
/* Threads created successfully. Let them proceed. */
cancel = 0;
pthread_mutex_unlock(&lock);
}
/*
* Note: At this point, err[0] and err[1] are both zero
* iff the threads were created successfully.
* Otherwise, both are nonzero errno values, and any
* temporarily created thread(s) have been cleaned up.
*/
/* After no further threads will be created, destroy the attributes. */
pthread_attr_destroy(&attrs);
Note that although I use the term cancel above, I do not refer to thread cancellation, which is a mechanism provided by the pthread library. It is just a name I found logical for the wrapper function operation.
Also note how this implementation is quite clean (other than my messy code). There are no global variables, and everything is transparent to the thread functions themselves. |
