flood/trunk/flood_farm.c

/* Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * Originally developed by Aaron Bannert and Justin Erenkrantz, eBuilt.
 */

#include <apr_errno.h>
#include <apr_thread_proc.h>
#include <apr_strings.h>

#if APR_HAVE_STRINGS_H
#include <strings.h>    /* strncasecmp */
#endif
#if APR_HAVE_STRING_H
#include <string.h>    /* strncasecmp */
#endif
#if APR_HAVE_STDLIB_H
#include <stdlib.h>     /* strtol */
#endif

#include "config.h"
#include "flood_farmer.h"

#include "flood_farm.h"

extern apr_file_t *local_stdout;
extern apr_file_t *local_stderr;

struct farm_t {
    const char *name;
    int n_farmers;
#if APR_HAS_THREADS
    apr_thread_t **farmers; /* pointer to array of threads */
#else
    apr_proc_t **farmers;   /* pointer to array of processes */
#endif
};
typedef struct farm_t farm_t;

struct farmer_worker_info_t {
    const char *farmer_name;
    config_t *config;
};
typedef struct farmer_worker_info_t farmer_worker_info_t;

#if APR_HAS_THREADS
/**
 * Worker function that is assigned to a thread. Each worker is
 * called a farmer in our system.
 */
static void * APR_THREAD_FUNC farmer_worker(apr_thread_t *thd, void *data)
{
    apr_status_t stat;
    apr_pool_t *pool;
    farmer_worker_info_t *info;

    info = (farmer_worker_info_t *)data;
    pool = apr_thread_pool_get(thd);

    /* should we create a subpool here? */
#ifdef FARM_DEBUG
    apr_file_printf(local_stdout, "Starting farmer_worker thread '%s'.\n",
                    info->farmer_name);
#endif

    if ((stat = run_farmer(info->config, info->farmer_name,
                           pool)) != APR_SUCCESS) {
        char buf[256];
        apr_strerror(stat, (char*) &buf, 256);
        apr_file_printf(local_stderr, "Error running farmer '%s': %s.\n",
                        info->farmer_name, (char*) &buf);
        /* just die for now, later try to return status */
    }

#if 0 /* this gets uncommented after apr_thread_exit() fixes are commited */
    apr_thread_exit(thd, APR_SUCCESS);
#endif
    return NULL;
}
#else
static void *farmer_worker(void *data)
{
    apr_status_t stat;
    apr_pool_t *pool;
    farmer_worker_info_t *info;

    info = (farmer_worker_info_t *)data;
    apr_pool_create(&pool, NULL);

    /* should we create a subpool here? */
#ifdef FARM_DEBUG
    apr_file_printf(local_stdout, "Starting farmer_worker child '%s'.\n",
                    info->farmer_name);
#endif

    if ((stat = run_farmer(info->config, info->farmer_name,
                           pool)) != APR_SUCCESS) {
        char buf[256];
        apr_strerror(stat, (char*) &buf, 256);
        apr_file_printf(local_stderr, "Error running farmer '%s': %s.\n",
                        info->farmer_name, (char*) &buf);
        /* just die for now, later try to return status */
    }

    return NULL;
}
#endif

apr_status_t run_farm(config_t *config, const char *farm_name, apr_pool_t *pool)
{
#if APR_HAS_THREADS
    apr_status_t child_stat;
#endif
    apr_status_t stat;
    int usefarmer_count, i, j;
    long farmer_start_count = 1;
    apr_time_t farmer_start_delay;
    char *xml_farm, **usefarmer_names;
    struct apr_xml_elem *e, *root_elem, *farm_elem;
    struct apr_xml_attr *use_elem;
    farm_t *farm;
    farmer_worker_info_t *infovec;

    farmer_start_delay = 0;

    xml_farm = apr_pstrdup(pool, XML_FARM);

    /* get the root config node */
    if ((stat = retrieve_root_xml_elem(&root_elem, config)) != APR_SUCCESS) {
        return stat;
    }

    /* get farmer node from config */
    if ((stat = retrieve_xml_elem_with_childmatch(
             &farm_elem, root_elem,
             xml_farm, XML_FARM_NAME, farm_name)) != APR_SUCCESS)
        return stat;

    /* count the number of "usefarmer" children */
    usefarmer_count = 0;
    for (e = farm_elem->first_child; e; e = e->next) {
        if (strncasecmp(e->name, XML_FARM_USEFARMER, FLOOD_STRLEN_MAX) == 0) {
            int found_count = 0;
            for (use_elem = e->attr; use_elem; use_elem = use_elem->next) {
                if (use_elem->name &&
                    strncasecmp(use_elem->name, XML_FARM_USEFARMER_COUNT, 
                                FLOOD_STRLEN_MAX) == 0) {
                    found_count = 1;
                    usefarmer_count += strtol(use_elem->value, NULL, 10);
                } 
            }
            if (!found_count) {
                usefarmer_count++;
            }
        }
    }

    if (!usefarmer_count)
        usefarmer_count = 1;

    /* create each of the children and put their names in an array */
    usefarmer_names = apr_palloc(pool, sizeof(char*) * (usefarmer_count + 1));
    /* set the sentinel (no need for pcalloc()) */
    usefarmer_names[usefarmer_count] = NULL; 
    i = 0;
    for (e = farm_elem->first_child; e; e = e->next) {
        int handled_usefarmers = 0;

        if (strncasecmp(e->name, XML_FARM_USEFARMER, FLOOD_STRLEN_MAX) == 0) {
            for (use_elem = e->attr; use_elem; use_elem = use_elem->next) {
                if (use_elem->name && strncasecmp(use_elem->name, 
                            XML_FARM_USEFARMER_COUNT, FLOOD_STRLEN_MAX) == 0) {
                    for (j = strtol(use_elem->value, NULL, 10); j > 0; j--) {
                        usefarmer_names[i++] = apr_pstrdup(pool, 
                                                   e->first_cdata.first->text);
                    }
                    handled_usefarmers = 1;
                }
                else if (use_elem->name &&
                         strncasecmp(use_elem->name, XML_FARM_USEFARMER_DELAY, 
                                     FLOOD_STRLEN_MAX) == 0) {
                    char *endptr;
                    farmer_start_delay = strtoll(use_elem->value, &endptr, 10);
                    if (*endptr != '\0')
                    {
                        apr_file_printf(local_stderr,
                                        "Attribute %s has invalid value %s.\n",
                                        XML_FARM_USEFARMER_DELAY, 
                                        use_elem->value);
                        return APR_EGENERAL;
                    }
                    farmer_start_delay *= APR_USEC_PER_SEC;
                }
                else if (use_elem->name && 
                         strncasecmp(use_elem->name, XML_FARM_USEFARMER_START, 
                                     FLOOD_STRLEN_MAX) == 0) {
                    char *endptr;
                    farmer_start_count = strtol(use_elem->value, &endptr, 10);
                    if (*endptr != '\0')
                    {
                        apr_file_printf(local_stderr,
                                        "Attribute %s has invalid value %s.\n",
                                        XML_FARM_USEFARMER_START, 
                                        use_elem->value);
                        return APR_EGENERAL;
                    }
                }
            }
            if (!handled_usefarmers)
                usefarmer_names[i++] = apr_pstrdup(pool, 
                                                   e->first_cdata.first->text);
        }
    }

    /* create the farm object */
    farm = apr_pcalloc(pool, sizeof(farm_t));
    farm->name = apr_pstrdup(pool, farm_name);
    farm->n_farmers = usefarmer_count;
#if APR_HAS_THREADS
    farm->farmers = apr_pcalloc(pool, 
                                sizeof(apr_thread_t*) * (usefarmer_count + 1));
    apr_thread_mutex_create(&config->mutex, APR_THREAD_MUTEX_DEFAULT, pool);
#else
    farm->farmers = apr_pcalloc(pool, 
                                sizeof(apr_proc_t*) * (usefarmer_count + 1));

    for (i = 0; i < usefarmer_count + 1; i++)
        farm->farmers[i] = apr_pcalloc(pool, sizeof(apr_proc_t*));

#endif

    infovec = apr_pcalloc(pool, sizeof(farmer_worker_info_t) * usefarmer_count);

    /* for each of my farmers, start them */
    for (i = 0; i < usefarmer_count; i++) {
        infovec[i].farmer_name = usefarmer_names[i];
        infovec[i].config = config;
#if APR_HAS_THREADS
        if ((stat = apr_thread_create(&farm->farmers[i],
                                      NULL,
                                      farmer_worker,
                                      (void*) &infovec[i],
                                      pool)) != APR_SUCCESS) {
            /* error, perhaps shutdown other threads then exit? */
            return stat;
        }
#else
        if (apr_proc_fork(farm->farmers[i], pool) == APR_INCHILD)
        {
            farmer_worker(&infovec[i]);
            exit(0);
        }
#endif
        if (farmer_start_delay && (i+1) % farmer_start_count == 0)
            apr_sleep(farmer_start_delay);
    }

    for (i = 0; i < usefarmer_count; i++) {
#if APR_HAS_THREADS
        stat = apr_thread_join(&child_stat, farm->farmers[i]);
        if (stat != APR_SUCCESS && stat != APR_INCOMPLETE) {
#else
        if ((stat = apr_proc_wait(farm->farmers[i], NULL, NULL, APR_WAIT)) != APR_CHILD_DONE) {
#endif

            apr_file_printf(local_stderr, "Error joining farmer thread '%d' ('%s').\n",
                            i, usefarmer_names[i]);
            return stat;
        } else {
#ifdef FARM_DEBUG
            apr_file_printf(local_stdout, "Farmer '%d' ('%s') completed successfully.\n",
                            i, usefarmer_names[i]);
#endif
        }
    }

    return APR_SUCCESS;
}

1	/* Licensed to the Apache Software Foundation (ASF) under one or more
2	* contributor license agreements. See the NOTICE file distributed with
3	* this work for additional information regarding copyright ownership.
4	* The ASF licenses this file to You under the Apache License, Version 2.0
5	* (the "License"); you may not use this file except in compliance with
6	* the License. You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*
16	* Originally developed by Aaron Bannert and Justin Erenkrantz, eBuilt.
17	*/
18
19	#include <apr_errno.h>
20	#include <apr_thread_proc.h>
21	#include <apr_strings.h>
22
23	#if APR_HAVE_STRINGS_H
24	#include <strings.h> /* strncasecmp */
25	#endif
26	#if APR_HAVE_STRING_H
27	#include <string.h> /* strncasecmp */
28	#endif
29	#if APR_HAVE_STDLIB_H
30	#include <stdlib.h> /* strtol */
31	#endif
32
33	#include "config.h"
34	#include "flood_farmer.h"
35
36	#include "flood_farm.h"
37
38	extern apr_file_t *local_stdout;
39	extern apr_file_t *local_stderr;
40
41	struct farm_t {
42	const char *name;
43	int n_farmers;
44	#if APR_HAS_THREADS
45	apr_thread_t *farmers; / pointer to array of threads */
46	#else
47	apr_proc_t *farmers; / pointer to array of processes */
48	#endif
49	};
50	typedef struct farm_t farm_t;
51
52	struct farmer_worker_info_t {
53	const char *farmer_name;
54	config_t *config;
55	};
56	typedef struct farmer_worker_info_t farmer_worker_info_t;
57
58	#if APR_HAS_THREADS
59	/**
60	* Worker function that is assigned to a thread. Each worker is
61	* called a farmer in our system.
62	*/
63	static void * APR_THREAD_FUNC farmer_worker(apr_thread_t thd, void data)
64	{
65	apr_status_t stat;
66	apr_pool_t *pool;
67	farmer_worker_info_t *info;
68
69	info = (farmer_worker_info_t *)data;
70	pool = apr_thread_pool_get(thd);
71
72	/* should we create a subpool here? */
73	#ifdef FARM_DEBUG
74	apr_file_printf(local_stdout, "Starting farmer_worker thread '%s'.\n",
75	info->farmer_name);
76	#endif
77
78	if ((stat = run_farmer(info->config, info->farmer_name,
79	pool)) != APR_SUCCESS) {
80	char buf[256];
81	apr_strerror(stat, (char*) &buf, 256);
82	apr_file_printf(local_stderr, "Error running farmer '%s': %s.\n",
83	info->farmer_name, (char*) &buf);
84	/* just die for now, later try to return status */
85	}
86
87	#if 0 /* this gets uncommented after apr_thread_exit() fixes are commited */
88	apr_thread_exit(thd, APR_SUCCESS);
89	#endif
90	return NULL;
91	}
92	#else
93	static void farmer_worker(void data)
94	{
95	apr_status_t stat;
96	apr_pool_t *pool;
97	farmer_worker_info_t *info;
98
99	info = (farmer_worker_info_t *)data;
100	apr_pool_create(&pool, NULL);
101
102	/* should we create a subpool here? */
103	#ifdef FARM_DEBUG
104	apr_file_printf(local_stdout, "Starting farmer_worker child '%s'.\n",
105	info->farmer_name);
106	#endif
107
108	if ((stat = run_farmer(info->config, info->farmer_name,
109	pool)) != APR_SUCCESS) {
110	char buf[256];
111	apr_strerror(stat, (char*) &buf, 256);
112	apr_file_printf(local_stderr, "Error running farmer '%s': %s.\n",
113	info->farmer_name, (char*) &buf);
114	/* just die for now, later try to return status */
115	}
116
117	return NULL;
118	}
119	#endif
120
121	apr_status_t run_farm(config_t config, const char farm_name, apr_pool_t *pool)
122	{
123	#if APR_HAS_THREADS
124	apr_status_t child_stat;
125	#endif
126	apr_status_t stat;
127	int usefarmer_count, i, j;
128	long farmer_start_count = 1;
129	apr_time_t farmer_start_delay;
130	char xml_farm, *usefarmer_names;
131	struct apr_xml_elem e, root_elem, *farm_elem;
132	struct apr_xml_attr *use_elem;
133	farm_t *farm;
134	farmer_worker_info_t *infovec;
135
136	farmer_start_delay = 0;
137
138	xml_farm = apr_pstrdup(pool, XML_FARM);
139
140	/* get the root config node */
141	if ((stat = retrieve_root_xml_elem(&root_elem, config)) != APR_SUCCESS) {
142	return stat;
143	}
144
145	/* get farmer node from config */
146	if ((stat = retrieve_xml_elem_with_childmatch(
147	&farm_elem, root_elem,
148	xml_farm, XML_FARM_NAME, farm_name)) != APR_SUCCESS)
149	return stat;
150
151	/* count the number of "usefarmer" children */
152	usefarmer_count = 0;
153	for (e = farm_elem->first_child; e; e = e->next) {
154	if (strncasecmp(e->name, XML_FARM_USEFARMER, FLOOD_STRLEN_MAX) == 0) {
155	int found_count = 0;
156	for (use_elem = e->attr; use_elem; use_elem = use_elem->next) {
157	if (use_elem->name &&
158	strncasecmp(use_elem->name, XML_FARM_USEFARMER_COUNT,
159	FLOOD_STRLEN_MAX) == 0) {
160	found_count = 1;
161	usefarmer_count += strtol(use_elem->value, NULL, 10);
162	}
163	}
164	if (!found_count) {
165	usefarmer_count++;
166	}
167	}
168	}
169
170	if (!usefarmer_count)
171	usefarmer_count = 1;
172
173	/* create each of the children and put their names in an array */
174	usefarmer_names = apr_palloc(pool, sizeof(char) (usefarmer_count + 1));
175	/* set the sentinel (no need for pcalloc()) */
176	usefarmer_names[usefarmer_count] = NULL;
177	i = 0;
178	for (e = farm_elem->first_child; e; e = e->next) {
179	int handled_usefarmers = 0;
180
181	if (strncasecmp(e->name, XML_FARM_USEFARMER, FLOOD_STRLEN_MAX) == 0) {
182	for (use_elem = e->attr; use_elem; use_elem = use_elem->next) {
183	if (use_elem->name && strncasecmp(use_elem->name,
184	XML_FARM_USEFARMER_COUNT, FLOOD_STRLEN_MAX) == 0) {
185	for (j = strtol(use_elem->value, NULL, 10); j > 0; j--) {
186	usefarmer_names[i++] = apr_pstrdup(pool,
187	e->first_cdata.first->text);
188	}
189	handled_usefarmers = 1;
190	}
191	else if (use_elem->name &&
192	strncasecmp(use_elem->name, XML_FARM_USEFARMER_DELAY,
193	FLOOD_STRLEN_MAX) == 0) {
194	char *endptr;
195	farmer_start_delay = strtoll(use_elem->value, &endptr, 10);
196	if (*endptr != '\0')
197	{
198	apr_file_printf(local_stderr,
199	"Attribute %s has invalid value %s.\n",
200	XML_FARM_USEFARMER_DELAY,
201	use_elem->value);
202	return APR_EGENERAL;
203	}
204	farmer_start_delay *= APR_USEC_PER_SEC;
205	}
206	else if (use_elem->name &&
207	strncasecmp(use_elem->name, XML_FARM_USEFARMER_START,
208	FLOOD_STRLEN_MAX) == 0) {
209	char *endptr;
210	farmer_start_count = strtol(use_elem->value, &endptr, 10);
211	if (*endptr != '\0')
212	{
213	apr_file_printf(local_stderr,
214	"Attribute %s has invalid value %s.\n",
215	XML_FARM_USEFARMER_START,
216	use_elem->value);
217	return APR_EGENERAL;
218	}
219	}
220	}
221	if (!handled_usefarmers)
222	usefarmer_names[i++] = apr_pstrdup(pool,
223	e->first_cdata.first->text);
224	}
225	}
226
227	/* create the farm object */
228	farm = apr_pcalloc(pool, sizeof(farm_t));
229	farm->name = apr_pstrdup(pool, farm_name);
230	farm->n_farmers = usefarmer_count;
231	#if APR_HAS_THREADS
232	farm->farmers = apr_pcalloc(pool,
233	sizeof(apr_thread_t) (usefarmer_count + 1));
234	apr_thread_mutex_create(&config->mutex, APR_THREAD_MUTEX_DEFAULT, pool);
235	#else
236	farm->farmers = apr_pcalloc(pool,
237	sizeof(apr_proc_t) (usefarmer_count + 1));
238
239	for (i = 0; i < usefarmer_count + 1; i++)
240	farm->farmers[i] = apr_pcalloc(pool, sizeof(apr_proc_t*));
241
242	#endif
243
244	infovec = apr_pcalloc(pool, sizeof(farmer_worker_info_t) * usefarmer_count);
245
246	/* for each of my farmers, start them */
247	for (i = 0; i < usefarmer_count; i++) {
248	infovec[i].farmer_name = usefarmer_names[i];
249	infovec[i].config = config;
250	#if APR_HAS_THREADS
251	if ((stat = apr_thread_create(&farm->farmers[i],
252	NULL,
253	farmer_worker,
254	(void*) &infovec[i],
255	pool)) != APR_SUCCESS) {
256	/* error, perhaps shutdown other threads then exit? */
257	return stat;
258	}
259	#else
260	if (apr_proc_fork(farm->farmers[i], pool) == APR_INCHILD)
261	{
262	farmer_worker(&infovec[i]);
263	exit(0);
264	}
265	#endif
266	if (farmer_start_delay && (i+1) % farmer_start_count == 0)
267	apr_sleep(farmer_start_delay);
268	}
269
270	for (i = 0; i < usefarmer_count; i++) {
271	#if APR_HAS_THREADS
272	stat = apr_thread_join(&child_stat, farm->farmers[i]);
273	if (stat != APR_SUCCESS && stat != APR_INCOMPLETE) {
274	#else
275	if ((stat = apr_proc_wait(farm->farmers[i], NULL, NULL, APR_WAIT)) != APR_CHILD_DONE) {
276	#endif
277
278	apr_file_printf(local_stderr, "Error joining farmer thread '%d' ('%s').\n",
279	i, usefarmer_names[i]);
280	return stat;
281	} else {
282	#ifdef FARM_DEBUG
283	apr_file_printf(local_stdout, "Farmer '%d' ('%s') completed successfully.\n",
284	i, usefarmer_names[i]);
285	#endif
286	}
287	}
288
289	return APR_SUCCESS;
290	}
291
infrastructure at apache.org	ViewVC Help
Powered by ViewVC 1.1.26