Context Navigation

pipeline.sm @ 7823

Revision 7823, 24.6 KB (checked in by pcarns, 4 years ago)
update printf format for 64 bit values

Line
1	/*
2	* (C) 2001 Clemson University and The University of Chicago
3	*
4	* See COPYING in top-level directory.
5	*/
6
7	/*
8	* PVFS2 server state machine for driving read/write I/O operations.
9	*/
10
11	#include <string.h>
12	#include <assert.h>
13	#include <stdlib.h>
14	#include <math.h> /* log2() */
15
16	#include "server-config.h"
17	#include "pvfs2-server.h"
18	#include "pvfs2-attr.h"
19	#include "pvfs2-request.h"
20	#include "pint-distribution.h"
21	#include "pvfs2-dist-simple-stripe.h"
22	#include "pint-request.h"
23	#include "pvfs2-internal.h"
24	#include "trove.h"
25
26	#define LOOP 101
27	#define UNALIGNED 102
28	#define DO_COMP 103
29	#define DO_ALLREDUCE 104
30	#define CONTINUE_ALLREDUCE 105
31
32	static int s2s_comp_fn(
33	void v_p, struct PVFS_server_resp resp_p, int index);
34
35	%%
36
37	nested machine pvfs2_pipeline_sm
38	{
39	state fetch
40	{
41	run fetch_data;
42	success => dispatch;
43	}
44
45	state dispatch
46	{
47	run dispatch_data;
48	DO_COMP => check_align;
49	success => check_pipeline;
50	}
51
52	state check_align
53	{
54	run check_align_fn;
55	UNALIGNED => setup_s2s;
56	success => do_comp;
57	}
58
59	state setup_s2s
60	{
61	run setup_s2s_msg;
62	success => s2s_exchange;
63	}
64
65	state s2s_exchange
66	{
67	jump pvfs2_msgpairarray_sm;
68	success => do_comp;
69	}
70
71	state do_comp
72	{
73	run do_comp_fn;
74	DO_ALLREDUCE => setup_allreduce;
75	success => check_pipeline;
76	}
77
78	state setup_allreduce
79	{
80	pjmp setup_allreduce_sm
81	{
82	success => pvfs2_allreduce_sm;
83	}
84	success => cleanup_allreduce;
85	}
86
87	state cleanup_allreduce
88	{
89	run cleanup_allreduce_fn;
90	success => check_pipeline;
91	}
92
93	state check_pipeline
94	{
95	run check_pipeline_done;
96	LOOP => fetch;
97	success => return;
98	}
99	}
100
101	%%
102
103	/*
104	* fetch data from either TROVE (in case of READ) or BMI (in case of WRITE)
105	*
106	* PINT_segpool_take_segments()
107	* => READ: job_trove_bstream_read_list()
108	* => WRITE: job_bmi_recv()
109	*/
110	static PINT_sm_action fetch_data(struct PINT_smcb smcb, job_status_s js_p)
111	{
112	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
113	PINT_segpool_handle_t seg_handle = s_op->u.pipeline.seg_handle;
114	PINT_segpool_unit_id id = s_op->u.pipeline.id;
115	struct server_configuration_s *user_opts = get_server_config_struct();
116	int count, ret, i;
117	PVFS_offset *offsets;
118	PVFS_size *sizes;
119	PVFS_size bytes;
120	job_id_t tmp_id;
121
122	gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", smcb->base_frame, smcb->frame_count);
123	gossip_debug(GOSSIP_IO_DEBUG, "smcb=%p, smcb->parent_smcb=%p\n", smcb, smcb->parent_smcb);
124	s_op->u.pipeline.buffer_used = 0;
125	bytes = s_op->u.pipeline.buffer_size;
126
127	PINT_segpool_take_segments(seg_handle, id, &bytes, &count,
128	&offsets, &sizes);
129	gossip_debug(GOSSIP_IO_DEBUG, "%s: %s: bytes=%lld, count=%d\n",
130	__func__,
131	(s_op->u.pipeline.io_type==PVFS_IO_READ?"READ":"WRITE"),
132	lld(bytes), count);
133
134	for(i=0; i<count; i++) {
135	gossip_debug(GOSSIP_IO_DEBUG, "offsets[%d]=%lld, sizes[%d]=%lld\n",
136	i, lld(offsets[i]), i, lld(sizes[i]));
137	}
138
139	if(count == 0) {
140	js_p->error_code = 0;
141	//gossip_debug(GOSSIP_IO_DEBUG, "%s: count==0?\n", __func__);
142	return SM_ACTION_COMPLETE;
143	}
144
145	s_op->u.pipeline.buffer_used = bytes;
146	s_op->u.pipeline.offsets = offsets;
147	s_op->u.pipeline.sizes = sizes;
148	s_op->u.pipeline.segs = count;
149
150	if(s_op->u.pipeline.io_type == PVFS_IO_READ) {
151
152	ret = job_trove_bstream_read_list
153	(s_op->u.pipeline.fs_id,
154	s_op->u.pipeline.handle,
155	(char **)&s_op->u.pipeline.buffer,
156	(PVFS_size *)&s_op->u.pipeline.buffer_used,
157	1,
158	offsets,
159	sizes,
160	count,
161	&s_op->u.pipeline.out_size,
162	s_op->u.pipeline.trove_sync_flag,
163	NULL,
164	smcb,
165	0,
166	js_p,
167	&tmp_id,
168	server_job_context,
169	s_op->u.pipeline.hints);
170	}
171	else if (s_op->u.pipeline.io_type == PVFS_IO_WRITE) {
172	ret = job_bmi_recv(s_op->u.pipeline.address,
173	(void *)s_op->u.pipeline.buffer,
174	s_op->u.pipeline.buffer_size,
175	s_op->u.pipeline.tag,
176	BMI_PRE_ALLOC,
177	smcb,
178	0, /* unsigned long status_user_tag = 0 */
179	js_p,
180	&tmp_id,
181	server_job_context,
182	user_opts->server_job_flow_timeout,
183	(bmi_hint)s_op->u.pipeline.hints);
184	}
185
186	if(ret < 0) {
187	gossip_err("%s: I/O error occurred\n", __func__);
188	/* FIXME */
189	//handle_io_error(ret, q_item, flow_data);
190	js_p->error_code = -PVFS_EIO;
191	return SM_ACTION_COMPLETE;
192	}
193
194	/* immediate return */
195	if(ret == 1) {
196	js_p->error_code = 1;
197	return SM_ACTION_COMPLETE;
198	}
199
200	if(ret == 0) {
201	js_p->error_code = 0;
202	return SM_ACTION_DEFERRED;
203	}
204
205	return SM_ACTION_COMPLETE;
206	}
207
208	/*
209	* Dispatch data to either BMI (in case of READ) or TROVE (in case of WRITE)
210	*
211	* => READ: job_bmi_send()
212	* => WRITE: job_trove_bstream_write_list()
213	*/
214	static PINT_sm_action dispatch_data(struct PINT_smcb smcb, job_status_s js_p)
215	{
216	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
217	int ret;
218	job_id_t tmp_id;
219	struct server_configuration_s *user_opts = get_server_config_struct();
220	struct PINT_server_op *parent_s_op = s_op->u.pipeline.parent;
221
222	if(s_op->u.pipeline.segs == 0) {
223	js_p->error_code = 0;
224	gossip_debug(GOSSIP_IO_DEBUG, "%s: count==0?\n", __func__);
225	return SM_ACTION_COMPLETE;
226	}
227
228	gossip_debug(GOSSIP_IO_DEBUG, "%s: %s: buffer_used=%lld\n", __func__,
229	(s_op->u.pipeline.io_type==PVFS_IO_READ?"READ":"WRITE"),
230	lld(s_op->u.pipeline.buffer_used));
231	#if 0
232	gossip_debug(GOSSIP_IO_DEBUG, "%s: buffer[0]=%s\n", __func__,
233	(char *)s_op->u.pipeline.buffer);
234	#endif
235
236
237	if(s_op->u.pipeline.io_type == PVFS_IO_READ) {
238	assert(s_op->u.pipeline.buffer_used);
239
240	if(parent_s_op->u.io.op != 0) { /* AS: when op is specified */
241	ret = DO_COMP; /* AS: skip sending if op is specified */
242	gossip_debug(GOSSIP_IO_DEBUG, "%s: parent->op != 0\n", __func__);
243	js_p->error_code = ret;
244	return SM_ACTION_COMPLETE;
245	}
246	else
247	ret = job_bmi_send(s_op->u.pipeline.address,
248	s_op->u.pipeline.buffer,
249	js_p->actual_size,
250	s_op->u.pipeline.tag,
251	BMI_PRE_ALLOC,
252	0, /* send_unexpected */
253	smcb, /* user_ptr */
254	0, /* status_user_tag */
255	js_p,
256	&tmp_id,
257	server_job_context,
258	user_opts->server_job_bmi_timeout,
259	(bmi_hint)s_op->u.pipeline.hints);
260
261	}
262	else if(s_op->u.pipeline.io_type == PVFS_IO_WRITE) {
263	ret = job_trove_bstream_write_list
264	(s_op->u.pipeline.fs_id,
265	s_op->u.pipeline.handle,
266	(char **)&s_op->u.pipeline.buffer,
267	(TROVE_size *)&js_p->actual_size,
268	1,
269	s_op->u.pipeline.offsets,
270	s_op->u.pipeline.sizes,
271	s_op->u.pipeline.segs,
272	&s_op->u.pipeline.out_size,
273	s_op->u.pipeline.trove_sync_flag,
274	NULL,
275	smcb,
276	0,
277	js_p,
278	&tmp_id,
279	server_job_context,
280	s_op->u.pipeline.hints);
281	}
282
283	if(ret < 0) {
284	gossip_err("%s: I/O error occurred\n", __func__);
285	/* FIXME !!!!!!! */
286	/* handle_io_error(ret, q_item, flow_data); */
287	js_p->error_code = ret;
288	return SM_ACTION_COMPLETE;
289	}
290
291	/* immediate return */
292	if(ret == 1) {
293	js_p->error_code = ret;
294	return SM_ACTION_COMPLETE;
295	}
296
297	if(ret == 0) {
298	js_p->error_code = ret;
299	return SM_ACTION_DEFERRED;
300	}
301
302	return SM_ACTION_COMPLETE;
303	}
304
305
306	static PINT_sm_action check_align_fn(struct PINT_smcb smcb, job_status_s js_p)
307	{
308	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
309	struct PINT_server_op *parent_s_op = s_op->u.pipeline.parent;
310	js_p->error_code = 0;
311	PVFS_size file_req_offset = parent_s_op->u.io.file_req_offset;
312	PINT_request_file_data fdata = parent_s_op->u.io.file_data;
313	PVFS_simple_stripe_params *dparam =
314	(PVFS_simple_stripe_params*)fdata.dist->params;
315	PVFS_size count;
316	PVFS_offset strip_boundary;
317
318	PVFS_offset loff = fdata.dist->methods->physical_to_logical_offset(fdata.dist->params, &fdata, s_op->u.pipeline.offsets[0]);
319
320	s_op->u.pipeline.loff = loff;
321	gossip_debug(GOSSIP_IO_DEBUG, "loff=%lld, file_req_offset=%lld\n", lld(loff), lld(file_req_offset));
322
323	switch(parent_s_op->u.io.datatype) {
324	case ((int)0x4c000405): /* MPI_INT */
325	count = (PVFS_size)(s_op->u.pipeline.buffer_used -
326	file_req_offset)/((*PVFS_INT).ub);
327	gossip_debug(GOSSIP_IO_DEBUG, "count=%lld\n", lld(count));
328	strip_boundary = ((int)(loff/(dparam->strip_size)))*(dparam->strip_size);
329	s_op->u.pipeline.unaligned_size = loff-strip_boundary;
330
331	if (loff == strip_boundary && file_req_offset != 0) {
332	s_op->u.pipeline.unaligned_size = file_req_offset;
333	}
334
335	if (s_op->u.pipeline.unaligned_size != 0 && count != 0) {
336	js_p->error_code = UNALIGNED;
337	gossip_debug(GOSSIP_IO_DEBUG, "unaligned_size=%lld\n",
338	lld(s_op->u.pipeline.unaligned_size));
339	}
340	case ((int)0x4c00080b): /* MPI_DOUBLE */
341	count = (PVFS_size)(s_op->u.pipeline.buffer_used -
342	file_req_offset)/((*PVFS_DOUBLE).ub);
343	gossip_debug(GOSSIP_IO_DEBUG, "count=%lld\n", lld(count));
344	strip_boundary = ((int)(loff/(dparam->strip_size)))(dparam->strip_size); / FIXME */
345	s_op->u.pipeline.unaligned_size = loff-strip_boundary;
346
347	if (loff == strip_boundary && file_req_offset != 0) {
348	s_op->u.pipeline.unaligned_size = file_req_offset;
349	}
350
351	if (s_op->u.pipeline.unaligned_size != 0 && count != 0) {
352	js_p->error_code = UNALIGNED;
353	gossip_debug(GOSSIP_IO_DEBUG, "unaligned_size=%lld\n",
354	lld(s_op->u.pipeline.unaligned_size));
355	}
356	}
357
358	return SM_ACTION_COMPLETE;
359	}
360
361
362	static PINT_sm_action setup_s2s_msg(struct PINT_smcb smcb, job_status_s js_p)
363	{
364	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
365	struct PINT_server_op *parent_s_op = s_op->u.pipeline.parent;
366	PINT_sm_msgpair_state *msg_p = NULL;
367	struct server_configuration_s *user_opts = get_server_config_struct();
368	int regions;
369	int ret;
370	PVFS_credentials creds;
371	int next_server_index;
372	PVFS_handle next_server_handle;
373
374	/* init msgpair */
375	PINT_msgpair_init(&s_op->msgarray_op);
376	msg_p = &s_op->msgarray_op.msgpair;
377
378	s_op->msgarray_op.params.job_timeout = user_opts->client_job_bmi_timeout;
379	s_op->msgarray_op.params.retry_delay = user_opts->client_retry_delay_ms;
380	s_op->msgarray_op.params.retry_limit = user_opts->client_retry_limit;
381	s_op->msgarray_op.params.quiet_flag = 1;
382
383	PINT_util_gen_credentials(&creds);
384	gossip_debug(GOSSIP_IO_DEBUG, "loff=%lld, buffer_used=%lld\n", lld(s_op->u.pipeline.loff), lld(s_op->u.pipeline.buffer_used));
385
386	/* determine which server we need to talk to */
387	next_server_index = (s_op->u.pipeline.dfile_index + 1)%(s_op->u.pipeline.dfile_count);
388	next_server_handle = parent_s_op->u.io.dfile_array[next_server_index];
389
390	/* build a request */
391	ret = PVFS_Request_contiguous(s_op->u.pipeline.unaligned_size,
392	PVFS_BYTE, &s_op->u.pipeline.file_req);
393
394	s_op->u.pipeline.file_req_offset = (((int)(s_op->u.pipeline.loff/262144))+1)262144; / FIXME */
395
396	regions = 1;
397	gossip_debug(GOSSIP_IO_DEBUG, "s_op->u.pipeline.file_req_offset=%lld\n", lld(s_op->u.pipeline.file_req_offset));
398
399	PINT_SERVREQ_SMALL_IO_FILL(msg_p->req,
400	creds,
401	s_op->u.pipeline.fs_id,
402	next_server_handle,
403	s_op->u.pipeline.io_type,
404	next_server_index,
405	s_op->u.pipeline.dfile_count,
406	s_op->u.pipeline.dist,
407	s_op->u.pipeline.file_req,
408	s_op->u.pipeline.file_req_offset,
409	regions,
410	s_op->u.pipeline.unaligned_size,
411	NULL /* s_op->hints */);
412
413	msg_p->fs_id = s_op->u.pipeline.fs_id;
414	msg_p->handle = next_server_handle;
415	msg_p->retry_flag = PVFS_MSGPAIR_RETRY;
416	msg_p->comp_fn = s2s_comp_fn;
417
418	ret = PINT_cached_config_map_to_server(&msg_p->svr_addr,
419	next_server_handle, //s_op->u.pipeline.handle,
420	s_op->u.pipeline.fs_id);
421	gossip_debug(GOSSIP_IO_DEBUG, "%s: msg_p->svr_addr=%llu\n", __func__,
422	llu(msg_p->svr_addr));
423	if(ret < 0) {
424	gossip_err("Failed to map meta server address\n");
425	js_p->error_code = ret;
426	return SM_ACTION_COMPLETE;
427	}
428
429	js_p->error_code = 0;
430
431	PINT_sm_push_frame(smcb, 0, &s_op->msgarray_op);
432	return SM_ACTION_COMPLETE;
433	}
434
435	static int s2s_comp_fn(void v_p, struct PVFS_server_resp resp_p,
436	int index)
437	{
438	PINT_smcb *smcb = v_p;
439	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_MSGPAIR_PARENT_SM);
440
441	gossip_debug(GOSSIP_IO_DEBUG, "%s called\n", __func__);
442	gossip_debug(GOSSIP_IO_DEBUG, "resp_p->status=%d\n", resp_p->status);
443	gossip_debug(GOSSIP_IO_DEBUG, "small_io: result_size=%lld\n",
444	lld(resp_p->u.small_io.result_size));
445
446	assert(resp_p->op == PVFS_SERV_SMALL_IO);
447
448	if (resp_p->status != 0) {
449	return resp_p->status;
450	}
451
452	if(resp_p->u.small_io.result_size != 0) {
453	memcpy(s_op->u.pipeline.tmp_buf, resp_p->u.small_io.buffer,
454	resp_p->u.small_io.result_size);
455	}
456
457	return 0;
458	}
459
460	/* square of Euclid distance between two multi-dimensional points */
461	__inline static
462	float euclid_dist_2(int numdims, /* no. dimensions */
463	float coord1, / [numdims] */
464	float coord2) / [numdims] */
465	{
466	int i;
467	float ans=0.0;
468
469	for (i=0; i<numdims; i++)
470	ans += (coord1[i]-coord2[i]) * (coord1[i]-coord2[i]);
471
472	return(ans);
473	}
474
475	/----< find_nearest_cluster() >---------------------------------------------/
476	__inline static
477	int find_nearest_cluster(int numClusters, /* no. clusters */
478	int numCoords, /* no. coordinates */
479	float object, / [numCoords] */
480	float *clusters) / [numClusters][numCoords] */
481	{
482	int index, i;
483	float dist, min_dist;
484
485	/* find the cluster id that has min distance to object */
486	index = 0;
487	min_dist = euclid_dist_2(numCoords, object, clusters[0]);
488
489	for (i=1; i<numClusters; i++) {
490	dist = euclid_dist_2(numCoords, object, clusters[i]);
491	/* no need square root */
492	if (dist < min_dist) { /* find the min and its array index */
493	min_dist = dist;
494	index = i;
495	}
496	}
497	return(index);
498	}
499
500	static PINT_sm_action do_comp_fn(struct PINT_smcb smcb, job_status_s js_p)
501	{
502	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
503	struct PINT_server_op *parent_s_op = s_op->u.pipeline.parent;
504	js_p->error_code = 0;
505
506	if(s_op->u.pipeline.buffer) {
507	PVFS_size i;
508	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG,
509	"%s: buffer_used=%lld, op=0x%x, datatype=0x%x, actual_size=%lld\n",
510	__func__, lld(s_op->u.pipeline.buffer_used),
511	parent_s_op->u.io.op,
512	parent_s_op->u.io.datatype,
513	lld(js_p->actual_size)); /* AS */
514
515	switch(parent_s_op->u.io.datatype) {
516	case ((int)0x4c000405): /* MPI_INT */
517	{
518	int a = (int)s_op->u.pipeline.buffer;
519	int result;
520	PVFS_size count = (s_op->u.pipeline.buffer_used-s_op->u.pipeline.unaligned_size)/((*PVFS_INT).ub);
521	int *tmp;
522
523	if(count == 0)
524	return SM_ACTION_COMPLETE;
525	/* data is not aligned perfectly, so adjust it */
526	if(s_op->u.pipeline.unaligned_size != 0) {
527	memcpy(((char)&a[count])+(((PVFS_INT).ub)-s_op->u.pipeline.unaligned_size), s_op->u.pipeline.tmp_buf, s_op->u.pipeline.unaligned_size);
528	count++;
529	}
530
531	if (parent_s_op->u.io.total_transferred == 0) {
532	if (parent_s_op->u.io.tmp_buffer == NULL)
533	parent_s_op->u.io.tmp_buffer = (void )malloc(1sizeof(int));
534	memset(parent_s_op->u.io.tmp_buffer, 0, sizeof(int));
535	parent_s_op->u.io.count = 0;
536	}
537	tmp = parent_s_op->u.io.tmp_buffer;
538	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "total_transferred=%lld\n", lld(parent_s_op->u.io.total_transferred));
539	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "count=%lld, tmp=%d\n", lld(count), *tmp);
540
541	switch(parent_s_op->u.io.op) {
542	case 0x58000001: /* MAX */
543	result = *a;
544	for (i=1; i<count; i++ ) {
545	if (a[i] > result) {
546	result = a[i];
547	}
548	}
549	a[0] = result;
550	if (parent_s_op->u.io.total_transferred == 0 \|\|
551	result > *tmp)
552	*tmp = result;
553	break;
554	case 0x58000002: /* MIN */
555	result = *a;
556	for (i=1; i<count; i++ ) {
557	if (a[i] < result) {
558	result = a[i];
559	}
560	}
561	a[0] = result;
562	if (parent_s_op->u.io.total_transferred == 0 \|\|
563	result < *tmp)
564	*tmp = result;
565	break;
566	case 0x58000003: /* SUM */
567	result = 0;
568	for (i=0; i<count; i++ ) {
569	if (i<10)
570	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "a[%lld]=%d\n",
571	lld(i), a[i]);
572	result += a[i];
573	}
574	a[0] = result;
575	*tmp += result;
576	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "sum=%d\n",
577	*tmp);
578	break;
579	case (0x5800000e): /* MEAN */
580	result = 0;
581	for (i=0; i<count; i++) {
582	result += a[i];
583	}
584	result = result/count;
585	a[0] = result;
586
587	if (parent_s_op->u.io.count == 0)
588	*tmp = result;
589	else {
590	int tmp_sum = (tmp)(parent_s_op->u.io.count);
591	tmp_sum = tmp_sum + result*count;
592	*tmp = (tmp_sum)/(parent_s_op->u.io.count+count);
593	}
594	parent_s_op->u.io.count += count;
595	gossip_debug(GOSSIP_IO_DEBUG, "mean=%d\n", *tmp);
596	break;
597	default:
598	break;
599	}
600	s_op->u.pipeline.buffer = (void *)a;
601	parent_s_op->u.io.tmp_buffer = (void *)tmp;
602	}
603	break;
604
605	case ((int)0x4c00080b): /* MPI_DOUBLE */
606	{
607	double a = (double)s_op->u.pipeline.buffer;
608	double result;
609	PVFS_size count = (s_op->u.pipeline.buffer_used-s_op->u.pipeline.unaligned_size)/((*PVFS_DOUBLE).ub);
610	double *tmp;
611	PVFS_offset strip_boundary = (int)(s_op->u.pipeline.loff/262144)262144; / FIXME */
612
613	if (s_op->u.pipeline.buffer_used < 262144 &&
614	s_op->u.pipeline.loff != strip_boundary) { /* FIXME */
615	count = (s_op->u.pipeline.buffer_used)/((*PVFS_DOUBLE).ub);
616	}
617	gossip_debug(GOSSIP_IO_DEBUG, "count=%lld\n", lld(count));
618	if(count < 1)
619	return SM_ACTION_COMPLETE;
620
621	/* data is not aligned perfectly, so adjust it within the memory*/
622	if(s_op->u.pipeline.unaligned_size != 0) {
623	PVFS_size adj_sz = s_op->u.pipeline.unaligned_size;
624	PVFS_size tmp_sz = ((*PVFS_DOUBLE).ub) - adj_sz;
625
626	if(s_op->u.pipeline.loff == strip_boundary) {
627	memcpy(a, (char*)&a[0]+adj_sz,
628	s_op->u.pipeline.buffer_used-adj_sz);
629	}
630
631	memcpy(((char*)&a[count])+tmp_sz, s_op->u.pipeline.tmp_buf,
632	s_op->u.pipeline.unaligned_size);
633
634
635	count++;
636
637	if(s_op->u.pipeline.buffer_used < (262144-((*PVFS_DOUBLE).ub)))
638	count--;
639	}
640
641	if (parent_s_op->u.io.total_transferred == 0) {
642	if (parent_s_op->u.io.tmp_buffer == NULL)
643	parent_s_op->u.io.tmp_buffer = (void )malloc(1sizeof(double));
644	memset( parent_s_op->u.io.tmp_buffer, 0, sizeof(double));
645	parent_s_op->u.io.count = 0;
646	}
647	tmp = parent_s_op->u.io.tmp_buffer;
648	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG,
649	"total_transferred=%lld\n",
650	lld(parent_s_op->u.io.total_transferred));
651	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "count=%lld, tmp=%lf\n",
652	lld(count), *tmp);
653	switch(parent_s_op->u.io.op) {
654	case 0x58000001: /* MAX */
655	result = *a;
656	for (i=1; i<count; i++ ) {
657	if (a[i] > result) {
658	result = a[i];
659	}
660	}
661	a[0] = result;
662	if (parent_s_op->u.io.total_transferred == 0 \|\|
663	result > *tmp)
664	*tmp = result;
665	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "max=%lf\n", *tmp);
666	break;
667	case 0x58000002: /* MIN */
668	result = *a;
669	for (i=1; i<count; i++ ) {
670	if (a[i] < result) {
671	result = a[i];
672	}
673	}
674
675	a[0] = result;
676	if (parent_s_op->u.io.total_transferred == 0 \|\|
677	result < *tmp)
678	*tmp = result;
679	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "min=%lf\n",
680	*tmp);
681	break;
682	case 0x58000003: /* SUM */
683	result = 0;
684	for (i=0; i<count; i++ ) {
685	if (i<10 \|\| i== (count-1) \|\| i == count) gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "a[%lld]=%lf\n", lld(i), a[i]);
686	result += a[i];
687	}
688
689	a[0] = result;
690	*tmp += result;
691	gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "sum=%lf\n",
692	*tmp);
693	break;
694	case (0x5800000e): /* MEAN */
695	result = 0;
696	for (i=0; i<count; i++) {
697	result += a[i];
698	}
699	result = result/count;
700	a[0] = result;
701
702	gossip_debug(GOSSIP_IO_DEBUG, "count=%lld\n", lld(parent_s_op->u.io.count));
703
704	if (parent_s_op->u.io.count == 0)
705	*tmp = result;
706	else {
707	double tmp_sum = (tmp)(parent_s_op->u.io.count);
708	tmp_sum = tmp_sum + result*count;
709	*tmp = (tmp_sum)/(parent_s_op->u.io.count+count);
710	}
711	parent_s_op->u.io.count += count;
712	gossip_debug(GOSSIP_IO_DEBUG, "mean=%lf\n", result);
713	break;
714	case (0x5800000f): /* KMEANS */ {
715	int index, numObjs, j;
716	int numClusters=2, numCoords=4;
717	double delta;
718
719	#if 0
720	for(i=0; i<numObjs; i++) {
721	//result = a[i];
722	index = find_nearest_cluster(numClusters, numCoords, a[i], clusters);
723	if(membership[i] != index) delta += 1.0;
724	membership[i] = index;
725
726	newClusterSize[index]++;
727	for(j=0; j<numCoords; j++)
728	newClusters[index][j] += objects[i][j];
729	}
730	#endif
731	js_p->error_code = DO_ALLREDUCE;
732	break;
733	}
734	default:
735	break;
736	} /* end inner switch */
737	s_op->u.pipeline.buffer = (void *)a;
738	parent_s_op->u.io.tmp_buffer = (void *)tmp;
739	}
740
741	break;
742	default:
743	break;
744	} /* end switch() */
745	} /* end if() */
746
747	return SM_ACTION_COMPLETE;
748	}
749
750	static PINT_sm_action setup_allreduce_sm(struct PINT_smcb smcb, job_status_s js_p)
751	{
752	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
753	struct PINT_server_op *parent_s_op = s_op->u.pipeline.parent;
754	int ret, tmp_id;
755	struct server_configuration_s *user_opts = get_server_config_struct();
756	PVFS_handle new_rank_handle;
757	PVFS_BMI_addr_t svr_addr;
758	struct PINT_server_op *allreduce_op;
759	int i;
760
761	allreduce_op = malloc(sizeof(*allreduce_op));
762	memset(allreduce_op, 0, sizeof(*allreduce_op));
763	allreduce_op->u.allreduce.fs_id = s_op->u.pipeline.fs_id;
764	allreduce_op->u.allreduce.myRank = s_op->u.pipeline.dfile_index;
765	allreduce_op->u.allreduce.recv_buf = (void)malloc(5sizeof(double));
766	memset(allreduce_op->u.allreduce.recv_buf, 0, 5*sizeof(double));
767	allreduce_op->u.allreduce.tree_depth = log2(s_op->u.pipeline.dfile_count); /* FIXME !!!! */
768	gossip_debug(GOSSIP_IO_DEBUG, "tree_depth=%d\n",
769	allreduce_op->u.allreduce.tree_depth);
770	allreduce_op->u.allreduce.current_depth = 0;
771	allreduce_op->u.allreduce.mask = 0x1;
772	allreduce_op->u.allreduce.dfile_array = parent_s_op->u.io.dfile_array;
773	allreduce_op->u.allreduce.send_buf = (void)malloc(5sizeof(double));
774	memset(allreduce_op->u.allreduce.send_buf, 0, 5*sizeof(double));
775
776	ret = PINT_sm_push_frame(smcb, 0, allreduce_op);
777
778	js_p->error_code = 0;
779	return SM_ACTION_COMPLETE;
780	}
781
782	static PINT_sm_action cleanup_allreduce_fn(struct PINT_smcb smcb, job_status_s js_p)
783	{
784	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
785	int task_id;
786	int remaining;
787	PVFS_error tmp_err;
788	struct PINT_server_op *allreduce_op;
789
790	allreduce_op = PINT_sm_pop_frame(smcb, &task_id, &tmp_err,
791	&remaining);
792	gossip_debug(GOSSIP_SERVER_DEBUG,
793	"pipeline: nested sm returned error code: %d\n", tmp_err);
794	memcpy(s_op->u.pipeline.buffer, allreduce_op->u.allreduce.send_buf,
795	5*sizeof(double));
796	free(allreduce_op->u.allreduce.send_buf);
797	free(allreduce_op->u.allreduce.recv_buf);
798	free(allreduce_op);
799
800	js_p->error_code = 0;
801	return SM_ACTION_COMPLETE;
802	}
803
804	static PINT_sm_action check_pipeline_done(struct PINT_smcb smcb, job_status_s js_p)
805	{
806	struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
807	struct PINT_server_op *parent_s_op = s_op->u.pipeline.parent;
808	PINT_segpool_handle_t h = s_op->u.pipeline.seg_handle;
809	js_p->error_code = 0;
810
811	/* FIMXE: do we really need this lock? */
812	gen_mutex_lock(&parent_s_op->u.io.mutex);
813	//parent_s_op->u.io.total_transferred += js_p->actual_size;
814	parent_s_op->u.io.total_transferred += s_op->u.pipeline.buffer_used;
815	gen_mutex_unlock(&parent_s_op->u.io.mutex);
816
817	gossip_debug(GOSSIP_IO_DEBUG, "%s: total_transferred=%lld\n", __func__,
818	lld(parent_s_op->u.io.total_transferred));
819	gossip_debug(GOSSIP_IO_DEBUG, "%s: actual_size=%lld\n", __func__,
820	lld(js_p->actual_size));
821
822	/* FIXME: */
823	/* unless the second condition is set, the server starts
824	a new read request to one already done, and falls into
825	infinite trove_read->bmi_send->check_done loop */
826	if(!segpool_done(h) && s_op->u.pipeline.segs != 0) {
827	gossip_debug(GOSSIP_IO_DEBUG, "%s: LOOP\n", __func__);
828	js_p->error_code = LOOP;
829	}
830	else {
831	if(parent_s_op->u.io.op == (0x5800000f)) { /* KMEANS */
832	parent_s_op->u.io.tmp_buffer = (void *)s_op->u.pipeline.buffer;
833	}
834	gossip_debug(GOSSIP_IO_DEBUG, "%s: DONE\n", __func__);
835	}
836
837	return SM_ACTION_COMPLETE;
838	}
839
840	/*
841	* Local variables:
842	* mode: c
843	* c-indent-level: 4
844	* c-basic-offset: 4
845	* End:
846	*
847	* vim: ft=c ts=8 sts=4 sw=4 expandtab
848	*/

Note: See TracBrowser for help on using the browser.

Download in other formats:

Original Format