691  * pipeline with the oldest event.
694  * pipeline will process exactly one of those logfiles.  Distinct pipelines
700  * one pipeline for each known log type - one for error, one for fault, etc.
701  * Each pipeline will process first the rotated logs of that type and then
706  * The output state of a pipeline is one of:
708  *	- record available: the next record from this pipeline is available
711  *	- done: this pipeline will produce no more records
713  *	- polling: this pipeline is polling for new records and will
718  * A pipeline iterates over each file queued to it using fmd_log_xiter.
719  * We do this in a separate thread for each pipeline.  The callback on
727  * A pipeline descriptor.  The pl_cv condition variable is used together
729  * lock for the serializer for pausing and continuing this pipeline.
732 	pthread_mutex_t pl_lock;	/* used only in pipeline startup */
740 	int pl_srlzeridx;		/* serializer index for this pipeline */
755  * Each pipeline has an associated output slot in the serializer.  This
773 	struct fmdump_pipeline *ds_pipearr;	/* pipeline array */
928  * Called on each pipeline record iteration to make a new record
930  * the caller must stall the pipeline, or 1 to indicate that the
932  * addition fills the serializer then choose a pipeline that must
1143 	struct fmdump_pipeline *pipeline, *pl;  in aggregate()  local
1151 		pipeline = calloc(npipe, sizeof (struct fmdump_pipeline));  in aggregate()
1152 		if (!pipeline)  in aggregate()
1156 			pipeline[i].pl_logpath = ifiles[i];  in aggregate()
1158 		pipeline = calloc(sizeof (logtypes) / sizeof (logtypes[0]),  in aggregate()
1160 		if (!pipeline)  in aggregate()
1180 			pipeline[npipe].pl_rotated =  in aggregate()
1183 			pipeline[npipe++].pl_logpath = logpath;  in aggregate()
1196 	srlzer.ds_pipearr = pipeline;  in aggregate()
1203 	for (i = 0, pl = &pipeline[0]; i < npipe; i++, pl++) {  in aggregate()
1219 			fmdump_fatal("pthread_create for pipeline %d failed",  in aggregate()
1223 	for (i = 0, pl = &pipeline[0]; i < npipe; i++, pl++) {  in aggregate()
1230 	for (i = 0, pl = &pipeline[0]; i < npipe; i++, pl++)  in aggregate()
1235 			free(pipeline[i].pl_logpath);  in aggregate()
1240 	free(pipeline);  in aggregate()