multithread

/*
   @mindmaze_header@
*/

/* multithreaded data write
 *
 * example of child process, to be used by pshared-parent.c example.
 *
 * This program writes to a shared text (in shared memory) concurrently with
 * other threads in the same process. When notified, a worker thread tries
 * to write its identification string ("|-thread-X+|") onto a text field of
 * the shared memory. The text after update of several worker threads looks
 * something like:
 *
 *     ...|+thread-Z+||+thread-W+||+thread-X+||+thread-Y+|...
 *
 * This file demonstrates how to:
 *  - map file into memory
 *  - use process shared mutex
 */
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <mmthread.h>

#define NUM_THREAD	6
#define MAX_ID_LEN	16

struct shared_data {
	mm_thr_mutex_t mutex;
	int len;
	char text[1024];
	mm_thr_mutex_t notif_mtx;
	mm_thr_cond_t notif_cond;
	int start;
};

struct thread_data {
	struct shared_data* shdata;
	char id_str[MAX_ID_LEN];
};


/*
 * This function do the update of the shared text. It happens the
 * string |+@id_str+| to the text field in @psh_data.
 */
static
void write_shared_data(struct shared_data* shdata, const char* id_str)
{
	int id_str_len = strlen(id_str);

	// Get the shared lock. Since we are using a normal mutex, we do not
	// have to check the return value
	mm_thr_mutex_lock(&shdata->mutex);

	// Add "|+" in the text
	shdata->text[shdata->len++] = '|';
	shdata->text[shdata->len++] = '+';

	// Append process identifier on text
	memcpy(shdata->text + shdata->len, id_str, id_str_len);
	shdata->len += id_str_len;

	// Add "+|" in the text
	shdata->text[shdata->len++] = '+';
	shdata->text[shdata->len++] = '|';

	mm_thr_mutex_unlock(&shdata->mutex);
}


static
void wait_start_notification(struct shared_data* shdata)
{
	mm_thr_mutex_lock(&shdata->notif_mtx);

	// A while loop is necessary, because a spurious wakeup is always
	// possible
	while (!shdata->start)
		mm_thr_cond_wait(&shdata->notif_cond, &shdata->notif_mtx);

	mm_thr_mutex_unlock(&shdata->notif_mtx);
}


static
void broadcast_start_notification(struct shared_data* shdata)
{
	// We want a worker thread to be be scheduled in a predictable way,
	// so we must own shdata->notif_mtx when calling
	// mm_thr_cond_broadcast()
	mm_thr_mutex_lock(&shdata->notif_mtx);

	shdata->start = 1;
	mm_thr_cond_broadcast(&shdata->notif_cond);

	mm_thr_mutex_unlock(&shdata->notif_mtx);
}


static
void* thread_func(void* data)
{
	struct thread_data* thdata = data;
	struct shared_data* shdata = thdata->shdata;
	const char* id_str = thdata->id_str;

	// Put a wait here to force a litle bit of more contention. This is
	// here only for demonstration purpose... Without it, since the
	// update of text is short and simple, the text would be likely
	// filed in the order of thread creation
	wait_start_notification(shdata);

	write_shared_data(shdata, id_str);

	return NULL;
}


int main(void)
{
	int i;
	mm_thread_t thid[NUM_THREAD];
	struct thread_data thdata[NUM_THREAD];
	struct shared_data shared = {
		.mutex = MM_THR_MUTEX_INITIALIZER,
		.notif_mtx = MM_THR_MUTEX_INITIALIZER,
		.notif_cond = MM_THR_COND_INITIALIZER,
		.start = 0,
	};

	// Create threads and assign each an ID string
	for (i = 0; i < NUM_THREAD; i++) {
		thdata[i].shdata = &shared;
		sprintf(thdata[i].id_str, "thread-%i", i);
		mm_thr_create(&thid[i], thread_func, &thdata[i]);
	}

	// Now that all thread are created, we can signal them to start
	broadcast_start_notification(&shared);

	for (i = 0; i < NUM_THREAD; i++)
		mm_thr_join(thid[i], NULL);

	printf("result string:%s\n", shared.text);
	return EXIT_SUCCESS;
}

parse_args

/*
 * @mindmaze_header@
 */

#include <mmargparse.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <mmsysio.h>

struct config {
	const char* detach_flag;
	unsigned int num_instance;
	const char* ip;
	const char* use_local_storage;
};
static
struct config cfg = {
	.num_instance = 10,
	.ip = "127.0.0.1",
};

#define LOREM_IPSUM "Lorem ipsum dolor sit amet, consectetur adipiscing"   \
	"elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua." \
	"Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi " \
	"ut aliquip ex ea commodo consequat..."

#define DEFAULT_PATH "/default/path"

static
struct mm_arg_opt cmdline_optv[] = {
	{"detach", MM_OPT_NOVAL, "set", {.sptr = &cfg.detach_flag},
	 "detach server process."},
	{"n|num-instance", MM_OPT_NEEDUINT, NULL, {.uiptr = &cfg.num_instance},
	 "Server can accommodate up to @NUM client simultaneously. Here is "
	 "more explanation to test text wrapping. " LOREM_IPSUM},
	{"l|use-local-storage", MM_OPT_OPTSTR, DEFAULT_PATH, {NULL},
	 "Use local storage located at @PATH which must exist. "
	 "If unspecified @PATH is assumed "DEFAULT_PATH "."},
	{.name = "i", .flags = MM_OPT_NEEDSTR, .defval = NULL,
	 {.sptr = &cfg.ip},
	 .desc = "IP address of remote server. @ADDR must have dotted form."},
};


/**
 * parse_option_cb() - validate some option value and parse other
 * @opt:        parser configuration of option recognized
 * @value:      value about to be set for option
 * @data:       callback data
 * @state:      flags indicating the state of option parsing.
 *
 * Return: 0 is parsing must continue, -1 if error has been detect and
 * parsing must stop.
 */
static
int parse_option_cb(const struct mm_arg_opt* opt, union mm_arg_val value,
                    void* data, int state)
{
	struct config* conf = data;
	(void)state;

	switch (mm_arg_opt_get_key(opt)) {
	case 'n':
		if (value.ui < 1) {
			fprintf(stderr,
			        "Server must support at least 1 instance\n");
			return -1;
		}

		// We don't set value here, since variable to set is already
		// configured in option setup ({.strptr = &cfg.ip})
		return 0;

	case 'l':
		if (mm_check_access(value.str, F_OK) != 0) {
			fprintf(stderr,
			        "storage file %s does not exist\n",
			        value.str);
			return -1;
		}

		conf->use_local_storage = value.str;
		return 0;

	default:
		return 0;
	}
}


int main(int argc, char* argv[])
{
	int i, arg_index;
	struct mm_arg_parser parser = {
		.doc = LOREM_IPSUM,
		.args_doc = "[options] cmd argument\n[options] hello",
		.optv = cmdline_optv,
		.num_opt = MM_NELEM(cmdline_optv),
		.cb = parse_option_cb,
		.cb_data = &cfg,
		.execname = argv[0],
	};


	arg_index = mm_arg_parse(&parser, argc, argv);

	fprintf(stdout, "options used:\n\tdetach_flag: %s\n\tinstance: %u\n"
	        "\tserver address: %s\n\tuse local path: %s\n",
	        cfg.detach_flag, cfg.num_instance,
	        cfg.ip, cfg.use_local_storage);

	fprintf(stdout, "Execute ");
	for (i = arg_index; i < argc; i++)
		fprintf(stdout, "%s ", argv[i]);

	fputc('\n', stdout);

	return EXIT_SUCCESS;
}

pshared

pshared-common.h

/*
 * @mindmaze_header@
 */
#ifndef PSHARED_COMMON_H
#define PSHARED_COMMON_H

#include <mmthread.h>

#define SHM_CHILD_FD 3

struct pshared_data {
	mm_thr_mutex_t mutex;
	int len;
	char text[1024];
	mm_thr_mutex_t notif_mtx;
	mm_thr_cond_t notif_cond;
	int start;
};

#endif /* ifndef PSHARED_COMMON_H */

pshared-child.h

/*
 * @mindmaze_header@
 */

/* process shared data: child program
 *
 * example of child process, to be used by pshared-parent.c example.
 *
 * Similar to the multithreaded data write example, this program writes to a
 * shared text (in shared memory) concurrently to other children of the
 * parent process. Each child maps into memory a file descriptor
 * (SHM_CHILD_FD) inherited from parent and initialized there.  The child
 * tries to write its identification string ("|-child-X+|") onto a text
 * field of the shared memory. The text after update of several child looks
 * something like:
 *
 *     ...|+child-Z+||+child-W+||+child-X+||+child-Y+|...
 *
 * Because of the concurrent access, the children use a process shared mutex
 * mapped in the shared memory. They can recover from a child dying while
 * owning the mutex. Put simulate this, the SEGFAULT_IN_CHILD environment
 * variable can be set. If a child process see its identification string
 * ("child-X" for Xth child created), it will provoke a segfault while
 * updating the text.
 *
 * This file demonstrates how to:
 *  - map file into memory
 *  - use process shared mutex
 */

#define MM_LOG_MODULE_NAME "pshared-child"

#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <stdbool.h>
#include <mmthread.h>
#include <mmsysio.h>
#include <mmerrno.h>
#include <mmlib.h>

#include "pshared-common.h"

#define BAD_ADDR (void*)0xDEADBEEF


static
void handle_notif_lock_retval(int lockret, struct pshared_data* psh_data)
{
	// By far the most usual case. We simply got the lock, nothing fancy
	// has happened.
	if (lockret == 0)
		return;

	// Contrary to psh_data->mutex, there is no shared state to recover
	// with psh_data->notif_mtx. Simply mark it consistent
	if (lockret == EOWNERDEAD)
		mm_thr_mutex_consistent(&psh_data->notif_mtx);

	if (lockret == ENOTRECOVERABLE)
		exit(EXIT_FAILURE);
}


/*
 * Wait that parent notifies to start, ie, mark psh_data->start = 1 and
 * broadcast psh_data->notif_cond.
 */
static
void wait_start_notification(struct pshared_data* psh_data)
{
	int lockret;

	lockret = mm_thr_mutex_lock(&psh_data->notif_mtx);
	handle_notif_lock_retval(lockret, psh_data);

	while (!psh_data->start) {
		lockret = mm_thr_cond_wait(&psh_data->notif_cond,
		                           &psh_data->notif_mtx);
		handle_notif_lock_retval(lockret, psh_data);
	}

	mm_thr_mutex_unlock(&psh_data->notif_mtx);
}


/*
 * This function do the update of the shared text. It must be called while
 * holding the lock, ie, called from write_shared_data(). It happens the
 * string |+@id_str+| to the text field in @psh_data.
 *
 * If requested, this function will provoke a segfault in the middle of
 * string appending.
 */
static
void write_shared_text_locked(struct pshared_data* psh_data, const char* id_str,
                              bool provoke_segfault)
{
	int id_str_len = strlen(id_str);

	// Add "|+" in the text
	psh_data->text[psh_data->len++] = '|';
	psh_data->text[psh_data->len++] = '+';

	// Append process identifier on text (psh_data->len not updated yet)
	memcpy(psh_data->text + psh_data->len, id_str, id_str_len);

	// Segfaulting here is a good place for demonstration purpose:
	// psh_data->len will be not consistent with the null-terminated
	// string in psh_data->text
	if (provoke_segfault)
		strcpy(psh_data->text + psh_data->len, BAD_ADDR);

	// Now update psh_data->len
	psh_data->len += id_str_len;

	// Add "+|" in the text
	psh_data->text[psh_data->len++] = '+';
	psh_data->text[psh_data->len++] = '|';
}


/*
 * Function to recover shared state from the situation where the previous
 * owner died while holding the lock.
 */
static
void recover_shared_text_from_owner_dead(struct pshared_data* psh_data)
{
	int len = psh_data->len;
	char* text = psh_data->text;

	// Find index of text immediately after the last occurrence of "+|"
	while (len > 0) {
		if ((len > 2) && (text[len-2] == '+') && (text[len-1] == '|'))
			break;

		len--;
	}

	// Crop string and set to the proper found length
	text[len] = '\0';
	psh_data->len = len;
}



static
void write_shared_data(struct pshared_data* psh_data, const char* id_str,
                       bool provoke_segfault)
{
	int r;

	// Get the shared lock. Since we are using a process shared mutex,
	// we must check return value of the lock operation: If the previous
	// owner has died while owning it, it will be only occasion to know
	// about it and recover from this if we want to continue using it.
	r = mm_thr_mutex_lock(&psh_data->mutex);
	if (r == EOWNERDEAD) {
		// We have the lock, but it is time to recover state since
		// previous owner died while holding the lock
		recover_shared_text_from_owner_dead(psh_data);

		// We have recovered the shared state, so we can mark lock as
		// consistent. After this, we will be back to normal operation
		mm_thr_mutex_consistent(&psh_data->mutex);
	} else if (r == ENOTRECOVERABLE) {
		// There has been an lock owner that has died and the next
		// owner failed (or refused) to mark lock as consistent,
		// thus rendering the lock unusable. This provokes all
		// waiters for the lock to be waken up and ENOTRECOVERABLE
		// is returned. Any new attempt to lock the mutex will return
		// ENOTRECOVERABLE (until it is deinit and init again).

		// So now, we don't have the lock and we can only stop
		return;
	}

	write_shared_text_locked(psh_data, id_str, provoke_segfault);

	mm_thr_mutex_unlock(&psh_data->mutex);
}


int main(int argc, char* argv[])
{
	int mflags;
	bool must_segfault;
	struct pshared_data* psh_data = NULL;
	const char* proc_string;

	// identifier of process is passed in the first argument
	if (argc < 2) {
		fprintf(stderr, "%s is missing argument\n", argv[0]);
		return EXIT_FAILURE;
	}

	proc_string = argv[1];

	// Map shared memory object onto memory.  We know that child is
	// created with shared memrory file descriptor inherited at
	// SHM_CHILD_FD
	mflags = MM_MAP_SHARED|MM_MAP_READ|MM_MAP_WRITE;
	psh_data = mm_mapfile(SHM_CHILD_FD, 0, sizeof(*psh_data), mflags);
	if (!psh_data) {
		mm_print_lasterror("mm_mapfile(%i, ...) failed", SHM_CHILD_FD);
		return EXIT_FAILURE;
	}

	// Close SHM_CHILD_FD because now that it is mapped, we don't need
	// it any longer
	mm_close(SHM_CHILD_FD);

	// Get from environment if this particular instance must simulate a
	// segfault while holding the lock.
	must_segfault = false;
	if (!strcmp(mm_getenv("SEGFAULT_IN_CHILD", ""), proc_string))
		must_segfault = true;

	// Wait until parent notify to start
	wait_start_notification(psh_data);

	// Try to update shared text.
	write_shared_data(psh_data, proc_string, must_segfault);

	mm_unmap(psh_data);
	return EXIT_SUCCESS;
}

pshared-parent.h

/*
 * @mindmaze_header@
 */

/* process shared data: parent program
 *
 * example of parent process, child is implemented in pshared-child.c.
 *
 * This program writes to a shared text (in shared memory) concurrently to
 * other children of the parent process. Each child maps into memory a file
 * descriptor (SHM_CHILD_FD) inherited from parent and initialized there.
 * The child tries to write its identification string ("|-child-X+|") onto a
 * text field of the shared memory. The text after update of several child
 * looks something like:
 *
 *     ...|+child-Z+||+child-W+||+child-X+||+child-Y+|...
 *
 * Because of the concurrent access, the children use a process shared mutex
 * mapped in the shared memory. They can recover from a child dying while
 * owning the mutex. Put simulate this, the SEGFAULT_IN_CHILD environment
 * variable can be set. If a child process see its identification string
 * ("child-X" for Xth child created), it will provoke a segfault while
 * updating the text.
 *
 * Note on how to execute the program:
 * If left untouched, the program assumes that child executable is available
 * in the _current_ directory. Also it assumes that mmlib shared library is
 * accessible at runtime.
 *
 * This file demonstrates how to:
 *  - create an anonymous shared memory object
 *  - map file into memory
 *  - initialize process shared mutex
 *  - create child process with passing file descriptor to them
 */
#define MM_LOG_MODULE_NAME "pshared-parent"

#include <stdlib.h>
#include <stdio.h>
#include <mmthread.h>
#include <mmsysio.h>
#include <mmpredefs.h>
#include <mmerrno.h>
#include <mmlib.h>

#include "pshared-common.h"


#ifdef _WIN32
#  define       BINEXT ".exe"
#else
#  define       BINEXT
#endif

#define NUM_CHILD 6
#define PSHARED_CHILD_BIN "./pshared-child" BINEXT

/*
 * Create, map into memory and initialize the data that will shared with the
 * children. The process shared mutex is initialized here.
 */
static
struct pshared_data* init_shared_mem_data(int* shm_fd)
{
	int fd, mflags;
	struct pshared_data* psh_data = NULL;

	// Create an new anonymous shared memory object. We could have use a
	// normal file (with mm_open()) without changing of the rest of the
	// following if we wanted to keep the result of memory access on the
	// shared memory.
	fd = mm_anon_shm();
	if (fd < 0)
		return NULL;

	// Size it to accommodate the data that will be shared between
	// parent and children.
	if (mm_ftruncate(fd, sizeof(*psh_data)))
		goto failure;

	// Map shared memory object onto memory
	mflags = MM_MAP_SHARED|MM_MAP_READ|MM_MAP_WRITE;
	psh_data = mm_mapfile(fd, 0, sizeof(*psh_data), mflags);
	if (!psh_data)
		goto failure;

	// Reset the while content of structure to 0/NULL fields
	*psh_data = (struct pshared_data) {.start = 0};

	// Initialize synchronization primitives of shared data
	if (mm_thr_mutex_init(&psh_data->mutex, MM_THR_PSHARED)
	    || mm_thr_mutex_init(&psh_data->notif_mtx, MM_THR_PSHARED)
	    || mm_thr_cond_init(&psh_data->notif_cond, MM_THR_PSHARED))
		goto failure;

	*shm_fd = fd;
	return psh_data;

failure:
	mm_close(fd);
	return NULL;
}


/*
 * Starts all children process ensuring that they inherit of the shared
 * memory file descriptor. Pass the string identify a particular process
 * instance as the first argument.
 */
static
int spawn_children(int shm_fd, int num_child, mm_pid_t* children)
{
	int i;
	char process_identifier[32];
	char* argv[] = {PSHARED_CHILD_BIN, process_identifier, NULL};
	struct mm_remap_fd fd_map = {
		.child_fd = SHM_CHILD_FD,
		.parent_fd = shm_fd,
	};

	for (i = 0; i < num_child; i++) {
		// Set the process identifier (it is just a string to
		// identify which child process is running). This string is
		// already set as second element in argv, ie, the first
		// argument
		sprintf(process_identifier, "child-%i", i);

		// Spawn the process
		if (mm_spawn(&children[i], argv[0], 1, &fd_map, 0, argv, NULL))
			return -1;
	}

	return 0;
}


static
int wait_children_termination(int num_child, const mm_pid_t* children)
{
	int i;

	for (i = 0; i < num_child; i++) {
		if (mm_wait_process(children[i], NULL))
			return -1;
	}

	return 0;
}


static
void broadcast_start_notification(struct pshared_data* psh_data)
{
	int lockret;

	// We want a worker thread to be be scheduled in a predictable way,
	// so we must own shdata->notif_mtx when calling
	// mm_thr_cond_broadcast()
	lockret = mm_thr_mutex_lock(&psh_data->notif_mtx);
	if (lockret == ENOTRECOVERABLE)
		return;

	if (lockret == EOWNERDEAD)
		mm_thr_mutex_consistent(&psh_data->notif_mtx);

	psh_data->start = 1;
	mm_thr_cond_broadcast(&psh_data->notif_cond);

	mm_thr_mutex_unlock(&psh_data->notif_mtx);
}


int main(void)
{
	mm_pid_t children[NUM_CHILD];
	int shm_fd = -1;
	struct pshared_data* psh_data = NULL;
	int exitcode = EXIT_FAILURE;

	fprintf(stderr, "SEGFAULT_IN_CHILD=%s\n",
	        mm_getenv("SEGFAULT_IN_CHILD", ""));

	// Create a shared memory object with the right size and map into
	// memory
	psh_data = init_shared_mem_data(&shm_fd);
	if (!psh_data)
		goto exit;

	// Create the children inheriting the shared memory object
	if (spawn_children(shm_fd, MM_NELEM(children), children))
		goto exit;

	// Close shm_fd because now that it is mapped, and transmitted to
	// children, we don't need its file descriptor.
	mm_close(shm_fd);
	shm_fd = -1;

	broadcast_start_notification(psh_data);

	wait_children_termination(MM_NELEM(children), children);
	exitcode = EXIT_SUCCESS;

exit:
	if (exitcode == EXIT_FAILURE)
		mm_print_lasterror("pshared-parent failed");
	else
		printf("result string:%s\n", psh_data->text);

	mm_close(shm_fd);
	mm_unmap(psh_data);
	return exitcode;
}