Re: [PATCH] operf main process improperly killing conversion process

Maynard:

I looked through the patch.  It looks OK to me but then I am not really
strong when it comes to C++ and pipes.  I was not able to get OProfile
to fail without the patch.  The patch does apply and compile fine.
Under "normal" conditions, it seems to work fine.  I tried to adjust the
sample freqency and workload run time to see if I could get OProfile to
print the progress dots.  I wasn't able to get the sample rate versus
the data processing such that it generated the progress dots. 

                 Carl Love

On Thu, 2014-07-31 at 17:34 -0500, Maynard Johnson wrote:
> operf main process improperly killing conversion process
> 
> Unless operf is run with --lazy-conversion, a separate process (the
> 'operf_read_pid' as referred to in the code) is forked to read
> perf_events-formatted kernel data from a pipe.  The pipe writer is
> another oprofile process  -- operf_record_pid -- which is reading the
> actual kernel data.  The operf_read_pid converts this data to oprofile
> sample files.  This conversion process is fairly expensive, so it's not
> unusual for it to take a little while after profiling stops for it to
> complete.  And the more samples there are (i.e., the longer the profile runs),
> the longer while it will take.  When profiling stops, we call
> _kill_operf_read_pid and a 5 second timeout begins, waiting for operf_read_pid
> to finish its conversion. If it has not completed in 5 seconds, we kill the
> operf_convert_pid!
> 
> Having a hard-coded timeout was a bad idea; we should always allow the
> conversion process to finish. It should eventually read all the data from the
> pipe and then detect that the write has closed the pipe.  At that point,
> it will exit on its own.
> 
> This patch changes the name of the _kill_operf_read_pid() function to
> _waitfor_operf_read_pid(), and modifies its behavior to simply wait for
> the read process to finish and check its exit status.  The conversion
> process has been slightly modified so that, when it's in post-profiling
> conversion phase, it will print a "." every 1 million records processed
> to show its progress.
> 
> Signed-off-by: Maynard Johnson <[email protected]>
> Reported-by:   Andrew Haley <[email protected]>
> ---
>  libperf_events/operf_counter.cpp |   33 +++++++++---
>  libperf_events/operf_counter.h   |    4 +-
>  pe_profiling/operf.cpp           |  112 +++++++++++++++-----------------------
>  3 files changed, 73 insertions(+), 76 deletions(-)
> 
> diff --git a/libperf_events/operf_counter.cpp b/libperf_events/operf_counter.cpp
> index 936d612..88e1708 100644
> --- a/libperf_events/operf_counter.cpp
> +++ b/libperf_events/operf_counter.cpp
> @@ -60,6 +60,15 @@ static const char *__op_magic = "OPFILE";
> 
>  #define OP_MAGIC	(*(u64 *)__op_magic)
> 
> +static bool _print_pp_progress(int fd)
> +{
> +	int msg;
> +	if (read(fd, &msg, sizeof(msg)) > 0)
> +		return true;
> +	else
> +		return false;
> +}
> +
>  /* This function for reading an event from the sample data pipe must
>   * be robust enough to handle the situation where the operf_record process
>   * writes an event record to the pipe in multiple chunks.
> @@ -868,11 +877,13 @@ void operf_record::recordPerfData(void)
>  }
> 
>  void operf_read::init(int sample_data_pipe_fd, string input_filename, string samples_loc, op_cpu cputype,
> -                      bool systemwide, int _record_write_pipe, int _record_read_pipe)
> +                      bool systemwide, int _record_write_pipe, int _record_read_pipe,
> +                      int _post_profiling_pipe)
>  {
>  	sample_data_fd = sample_data_pipe_fd;
>  	read_comm_pipe = _record_read_pipe;
>  	write_comm_pipe = _record_write_pipe;
> +	post_profiling_pipe = _post_profiling_pipe;
>  	inputFname = input_filename;
>  	sampledir = samples_loc;
>  	cpu_type = cputype;
> @@ -1079,6 +1090,11 @@ unsigned int operf_read::convertPerfData(void)
>  	bool error = false;
>  	event_t * event = NULL;
> 
> +	if (fcntl(post_profiling_pipe, F_SETFL, O_NONBLOCK) < 0) {
> +		cerr << "Error: fcntl failed with errno:\n\t" << strerror(errno) << endl;
> +		throw runtime_error("Error: Unable to set post_profiling_pipe to non blocking");
> +	}
> +
>  	if (!inputFname.empty()) {
>  		info.file_data_offset = opHeader.data_offset;
>  		info.file_data_size = opHeader.data_size;
> @@ -1112,9 +1128,8 @@ unsigned int operf_read::convertPerfData(void)
>  	first_time_processing = true;
>  	int num_recs = 0;
>  	struct perf_event_header last_header;
> -	bool print_progress = !inputFname.empty() && syswide;
> -	if (print_progress)
> -		cerr << "Converting profile data to OProfile format" << endl;
> +        bool print_progress = !inputFname.empty() && syswide;
> +        bool printed_progress_msg = false;
>  	while (1) {
>  		streamsize rec_size = 0;
>  		if (!inputFname.empty()) {
> @@ -1135,10 +1150,14 @@ unsigned int operf_read::convertPerfData(void)
>  		}
>  		num_bytes += rec_size;
>  		num_recs++;
> -		if ((num_recs % 1000000 == 0) && print_progress)
> +		if ((num_recs % 1000000 == 0) && (print_progress || _print_pp_progress(post_profiling_pipe))) {
> +			if (!printed_progress_msg) {
> +				cerr << "\nConverting profile data to OProfile format " << endl;
> +				printed_progress_msg = true;
> +			}
>  			cerr << ".";
> +		}
>  	}
> -
>  	if (unlikely(error)) {
>  		if (!inputFname.empty()) {
>  			cerr << "ERROR: operf_read::convertPerfData quitting. Bad data read from file." << endl;
> @@ -1156,7 +1175,7 @@ unsigned int operf_read::convertPerfData(void)
>  	if (!error)
>  		op_reprocess_unresolved_events(opHeader.h_attrs[0].attr.sample_type, print_progress);
> 
> -	if (print_progress)
> +	if (printed_progress_msg)
>  		cerr << endl;
> 
>  	op_release_resources();
> diff --git a/libperf_events/operf_counter.h b/libperf_events/operf_counter.h
> index dfb4336..70e7aac 100644
> --- a/libperf_events/operf_counter.h
> +++ b/libperf_events/operf_counter.h
> @@ -139,7 +139,8 @@ public:
>  	: sample_data_fd(-1), inputFname(""), evts(_evts), cpu_type(CPU_NO_GOOD)
>  	  { valid = syswide = false;}
>  	void init(int sample_data_pipe_fd, std::string input_filename, std::string samples_dir, op_cpu cputype,
> -	          bool systemwide, int _record_write_pipe, int _record_read_pipe);
> +	          bool systemwide, int _record_write_pipe, int _record_read_pipe,
> +	          int _post_profiling_pipe);
>  	~operf_read();
>  	int readPerfHeader(void);
>  	unsigned int convertPerfData(void);
> @@ -154,6 +155,7 @@ private:
>  	int sample_data_fd;
>  	int write_comm_pipe;
>  	int read_comm_pipe;
> +	int post_profiling_pipe;
>  	std::string inputFname;
>  	std::string sampledir;
>  	std::ifstream istrm;
> diff --git a/pe_profiling/operf.cpp b/pe_profiling/operf.cpp
> index 51f55a5..668f9ea 100644
> --- a/pe_profiling/operf.cpp
> +++ b/pe_profiling/operf.cpp
> @@ -110,6 +110,10 @@ static int operf_convert_record_write_pipe[2];
>  // The operf_record_convert_write_pipe is used for the record process to send
>  // data to the convert process in response to the forked PID data.
>  static int operf_record_convert_write_pipe[2];
> +// The operf_post_profiling_pipe is used by the main process to inform the operf_read_pid
> +// that profiling is done.  The operf_read_pid will then print its progress in
> +// finishing the conversion.
> +static int operf_post_profiling_pipe[2];
> 
>  bool ctl_c = false;
>  bool pipe_closed = false;
> @@ -472,81 +476,47 @@ fail_out:
>  	return 0;
>  }
> 
> -static end_code_t _kill_operf_read_pid(end_code_t rc)
> +static end_code_t _waitfor_operf_read_pid(end_code_t rc)
>  {
> -	// Now stop the operf-read process
> -	int waitpid_status;
> -	struct timeval tv;
> -	long long start_time_sec;
> -	long long usec_timer;
> -	bool keep_trying = true;
> -	waitpid_status = 0;
> -	gettimeofday(&tv, NULL);
> -	start_time_sec = tv.tv_sec;
> -	usec_timer = tv.tv_usec;
> -	/* We'll initially try the waitpid with WNOHANG once every 100,000 usecs.
> -	 * If it hasn't ended within 5 seconds, we'll kill it and do one
> -	 * final wait.
> -	 */
> -	while (keep_trying) {
> -		int option = WNOHANG;
> -		int wait_rc;
> -		gettimeofday(&tv, NULL);
> -		if (tv.tv_sec > start_time_sec + 5) {
> -			keep_trying = false;
> -			option = 0;
> -			cerr << "now trying to kill convert pid..." << endl;
> +	// Now wait for the operf-read process to finish
> +	int wait_rc, waitpid_status, post_processing = 1;
> 
> -			if (kill(operf_read_pid, SIGUSR1) < 0) {
> -				perror("Attempt to stop operf-read process failed");
> -				rc = rc ? PERF_BOTH_ERROR : PERF_READ_ERROR;
> -				break;
> -			}
> -		} else {
> -			/* If we exceed the 100000 usec interval or if the tv_usec
> -			 * value has rolled over to restart at 0, then we reset
> -			 * the usec_timer to current tv_usec and try waitpid.
> -			 */
> -			if ((tv.tv_usec % 1000000) > (usec_timer + 100000)
> -					|| (tv.tv_usec < usec_timer))
> -				usec_timer = tv.tv_usec;
> -			else
> -				continue;
> +	if (write(operf_post_profiling_pipe[1], &post_processing, sizeof(post_processing)) < 0) {
> +		perror("Internal error:  Failed to write to operf_post_profiling_pipe");
> +		rc = rc ? PERF_BOTH_ERROR : PERF_READ_ERROR;
> +		return rc;
> +	}
> +	waitpid_status = 0;
> +	if ((wait_rc = waitpid(operf_read_pid, &waitpid_status, 0)) < 0) {
> +		if (errno != ECHILD) {
> +			perror("waitpid for operf-read process failed");
> +			rc = rc ? PERF_BOTH_ERROR : PERF_READ_ERROR;
>  		}
> -		if ((wait_rc = waitpid(operf_read_pid, &waitpid_status, option)) < 0) {
> -			keep_trying = false;
> -			if (errno != ECHILD) {
> -				perror("waitpid for operf-read process failed");
> -				rc = rc ? PERF_BOTH_ERROR : PERF_READ_ERROR;
> -			}
> -		} else if (wait_rc) {
> -			if (WIFEXITED(waitpid_status)) {
> -				keep_trying = false;
> -				if (!WEXITSTATUS(waitpid_status)) {
> -					cverb << vdebug << "operf-read process returned OK" << endl;
> -				} else if (WIFEXITED(waitpid_status)) {
> -					/* If user did ctl-c, operf-read may get spurious errors, like
> -					 * broken pipe, etc.  We ignore these unless the user asks for
> -					 * debug output.
> -					 */
> -					if (!ctl_c || cverb << vdebug) {
> -						cerr <<  "operf-read process ended abnormally.  Status = "
> -						     << WEXITSTATUS(waitpid_status) << endl;
> -						rc = rc ? PERF_BOTH_ERROR : PERF_READ_ERROR;
> -					}
> -				}
> -			}  else if (WIFSIGNALED(waitpid_status)) {
> -				keep_trying = false;
> +	} else if (wait_rc) {
> +		if (WIFEXITED(waitpid_status)) {
> +			if (!WEXITSTATUS(waitpid_status)) {
> +				cverb << vdebug << "operf-read process returned OK" << endl;
> +			} else {
>  				/* If user did ctl-c, operf-read may get spurious errors, like
>  				 * broken pipe, etc.  We ignore these unless the user asks for
>  				 * debug output.
>  				 */
>  				if (!ctl_c || cverb << vdebug) {
> -					cerr << "operf-read process killed by signal "
> -					     << WTERMSIG(waitpid_status) << endl;
> -					rc = PERF_RECORD_ERROR;
> +					cerr <<  "operf-read process ended abnormally.  Status = "
> +							<< WEXITSTATUS(waitpid_status) << endl;
> +					rc = rc ? PERF_BOTH_ERROR : PERF_READ_ERROR;
>  				}
>  			}
> +		}  else if (WIFSIGNALED(waitpid_status)) {
> +			/* If user did ctl-c, operf-read may get spurious errors, like
> +			 * broken pipe, etc.  We ignore these unless the user asks for
> +			 * debug output.
> +			 */
> +			if (!ctl_c || cverb << vdebug) {
> +				cerr << "operf-read process killed by signal "
> +						<< WTERMSIG(waitpid_status) << endl;
> +				rc = PERF_RECORD_ERROR;
> +			}
>  		}
>  	}
>  	return rc;
> @@ -630,12 +600,17 @@ static end_code_t _run(void)
>  	if (!operf_options::post_conversion) {
>  		if (!(!app_started && !operf_options::system_wide)) {
>  			cverb << vdebug << "Forking read pid" << endl;
> +			if (pipe(operf_post_profiling_pipe) < 0) {
> +				perror("Internal error: operf-record could not create pipe");
> +				_exit(EXIT_FAILURE);
> +			}
>  			operf_read_pid = fork();
>  			if (operf_read_pid < 0) {
>  				perror("Internal error: fork failed");
>  				_exit(EXIT_FAILURE);
>  			} else if (operf_read_pid == 0) { // child process
>  				close(sample_data_pipe[1]);
> +				close(operf_post_profiling_pipe[1]);
>  				_set_basic_SIGINT_handler_for_child();
>  				convert_sample_data();
>  				_exit(EXIT_SUCCESS);
> @@ -647,6 +622,7 @@ static end_code_t _run(void)
>  			close(operf_convert_record_write_pipe[1]);
>  			close(operf_record_convert_write_pipe[0]);
>  			close(operf_record_convert_write_pipe[1]);
> +			close(operf_post_profiling_pipe[0]);
>  		}
>  	}
> 
> @@ -753,10 +729,10 @@ again:
>  		if (operf_options::post_conversion)
>  			rc = _kill_operf_record_pid();
>  		else
> -			rc = _kill_operf_read_pid(_kill_operf_record_pid());
> +			rc = _waitfor_operf_read_pid(_kill_operf_record_pid());
>  	} else {
>  		if (!operf_options::post_conversion)
> -			rc = _kill_operf_read_pid(rc);
> +			rc = _waitfor_operf_read_pid(rc);
>  	}
> 
>  	return rc;
> @@ -945,7 +921,7 @@ static void convert_sample_data(void)
>  	close(operf_convert_record_write_pipe[0]);
>  	operfRead.init(inputfd, inputfname, current_sampledir, cpu_type,
>  	               operf_options::system_wide, operf_convert_record_write_pipe[1],
> -	               operf_record_convert_write_pipe[0]);
> +	               operf_record_convert_write_pipe[0], operf_post_profiling_pipe[0]);
>  	if ((rc = operfRead.readPerfHeader()) < 0) {
>  		if (rc != OP_PERF_HANDLED_ERROR)
>  			cerr << "Error: Cannot create read header info for sample data " << endl;



------------------------------------------------------------------------------
Want fast and easy access to all the code in your enterprise? Index and
search up to 200,000 lines of code with a free copy of Black Duck
Code Sight - the same software that powers the world's largest code
search on Ohloh, the Black Duck Open Hub! Try it now.
http://p.sf.net/sfu/bds